"""Stars agent — astrological transit predictions via pyswisseph. Requires birth_date in agent_prefs (ISO 8601 date string, e.g. '1990-06-15'). Populated from a connected data source (Google profile / Google Health). If birth_date is absent the agent returns a no-data snippet and the eligibility filter will silence it once the consent / pref check catches up. Computes today's Sun, Moon, Mercury, Venus, Mars, Jupiter, Saturn positions and finds notable transits (conjunctions, oppositions, squares, trines, sextiles) between today's sky and the user's natal chart. Passes a concise prediction + interpretation to the orchestrator. """ from __future__ import annotations import math from datetime import date, datetime, timezone from typing import ClassVar from .base import BaseAgent, AgentInput, AgentOutput from .manifest import AgentManifest, InferredParam try: import swisseph as swe # type: ignore _SWE_AVAILABLE = True except ImportError: # pragma: no cover — present in container, absent in dev _SWE_AVAILABLE = False # --------------------------------------------------------------------------- # Planet catalogue # --------------------------------------------------------------------------- _PLANETS: list[tuple[int, str]] = [] if _SWE_AVAILABLE: _PLANETS = [ (swe.SUN, "Sun"), (swe.MOON, "Moon"), (swe.MERCURY, "Mercury"), (swe.VENUS, "Venus"), (swe.MARS, "Mars"), (swe.JUPITER, "Jupiter"), (swe.SATURN, "Saturn"), ] # Aspect definitions: (angle, orb, name, nature) _ASPECTS: list[tuple[float, float, str, str]] = [ (0.0, 8.0, "conjunction", "intensifying"), (60.0, 6.0, "sextile", "harmonious"), (90.0, 7.0, "square", "challenging"), (120.0, 8.0, "trine", "flowing"), (180.0, 8.0, "opposition", "tension"), ] _ZODIAC = [ "Aries", "Taurus", "Gemini", "Cancer", "Leo", "Virgo", "Libra", "Scorpio", "Sagittarius", "Capricorn", "Aquarius", "Pisces", ] # Interpretive keywords per planet for transit readings _PLANET_THEMES: dict[str, str] = { "Sun": "identity, vitality, core purpose", "Moon": "emotions, intuition, comfort needs", "Mercury": "communication, thinking, decisions", "Venus": "relationships, values, pleasure", "Mars": "energy, drive, conflict", "Jupiter": "growth, opportunity, expansion", "Saturn": "discipline, responsibility, long-term structure", } def _zodiac_sign(lon: float) -> str: return _ZODIAC[int(lon / 30) % 12] def _jd_from_date(d: date) -> float: """Julian Day Number for noon UTC on the given date.""" assert _SWE_AVAILABLE return swe.julday(d.year, d.month, d.day, 12.0) def _planet_positions(jd: float) -> dict[str, float]: assert _SWE_AVAILABLE positions: dict[str, float] = {} for pid, name in _PLANETS: result, _ = swe.calc_ut(jd, pid) positions[name] = result[0] # ecliptic longitude return positions def _angular_diff(a: float, b: float) -> float: """Smallest angle between two ecliptic longitudes (0–180).""" diff = abs(a - b) % 360 return diff if diff <= 180 else 360 - diff def _find_transits(natal: dict[str, float], today: dict[str, float]) -> list[dict]: """Return list of active transits between today's sky and natal chart.""" transits: list[dict] = [] for t_name, t_lon in today.items(): for n_name, n_lon in natal.items(): diff = _angular_diff(t_lon, n_lon) for angle, orb, aspect_name, nature in _ASPECTS: if abs(diff - angle) <= orb: transits.append({ "transit_planet": t_name, "natal_planet": n_name, "aspect": aspect_name, "nature": nature, "orb": round(abs(diff - angle), 2), }) # Sort by tightness of orb transits.sort(key=lambda x: x["orb"]) return transits def _format_transit(t: dict) -> str: tp, np, asp, nat = t["transit_planet"], t["natal_planet"], t["aspect"], t["nature"] tp_theme = _PLANET_THEMES.get(tp, "") np_theme = _PLANET_THEMES.get(np, "") return ( f"Transiting {tp} ({tp_theme}) {asp} natal {np} ({np_theme}) " f"— a {nat} influence" ) # --------------------------------------------------------------------------- # Manifest # --------------------------------------------------------------------------- MANIFEST = AgentManifest( id="stars", version="1.0.0", description="Astrological transit predictions based on the user's birth date and today's planetary positions.", pref_schema={ "type": "object", "additionalProperties": False, "properties": { "birth_date": { "type": "string", "pattern": r"^\d{4}-\d{2}-\d{2}$", "description": "ISO 8601 birth date (YYYY-MM-DD). Populated from connected data source.", }, }, }, context_schema=["profile.birth_date"], # Requires a connected Google source that supplies birth date. # data:google-health is the current carrier; when Google profile is a # separate consent key, add it here. required_consents=["data:core", "data:google-health"], output_contract={"type": "snippet", "format": "free_text"}, ttl_sec=3_600 * 6, # planetary positions change slowly — 6 h is fine silenced_in_contexts=[], inferred_params=[ InferredParam( key="birth_date", ttl_sec=365 * 86_400, # effectively permanent once known cold_start_default=None, min_history=999_999, # never inferred from events — sourced externally infer=None, ), ], ) class StarsAgent(BaseAgent): """Produces astrological transit predictions for the user's birth chart.""" agent_id: ClassVar[str] = MANIFEST.id ttl_seconds: ClassVar[int] = MANIFEST.ttl_sec version: ClassVar[str] = MANIFEST.version def compute(self, inp: AgentInput) -> AgentOutput: birth_date_str: str | None = inp.agent_prefs.get("birth_date") if not birth_date_str: prompt = ( "Birth date is not available — astrological reading skipped. " "(Always write the tip in English.)" ) return self._make_output(inp, prompt, {"no_birth_date": True}) if not _SWE_AVAILABLE: prompt = ( "Astrological library unavailable — reading skipped. " "(Always write the tip in English.)" ) return self._make_output(inp, prompt, {"swe_unavailable": True}) try: birth_date = date.fromisoformat(birth_date_str) except ValueError: prompt = "Birth date format invalid — astrological reading skipped." return self._make_output(inp, prompt, {"invalid_birth_date": birth_date_str}) today_date = inp.now.date() natal_jd = _jd_from_date(birth_date) today_jd = _jd_from_date(today_date) natal_pos = _planet_positions(natal_jd) today_pos = _planet_positions(today_jd) transits = _find_transits(natal_pos, today_pos) top = transits[:3] # most exact transits only today_sun_sign = _zodiac_sign(today_pos["Sun"]) natal_sun_sign = _zodiac_sign(natal_pos["Sun"]) natal_moon_sign = _zodiac_sign(natal_pos["Moon"]) snapshot = { "birth_date": birth_date_str, "today": today_date.isoformat(), "natal_sun": natal_sun_sign, "natal_moon": natal_moon_sign, "today_sun": today_sun_sign, "active_transits": transits[:5], } if not top: prompt = ( f"Natal chart: Sun in {natal_sun_sign}, Moon in {natal_moon_sign}. " f"Today's Sun is in {today_sun_sign}. " "No exact transits today — a quiet, stable day energetically. " "(Always write the tip in English.)" ) else: transit_lines = "; ".join(_format_transit(t) for t in top) prompt = ( f"Natal chart: Sun in {natal_sun_sign}, Moon in {natal_moon_sign}. " f"Today's Sun is in {today_sun_sign}. " f"Active transits: {transit_lines}. " "Use these planetary themes to colour the tip — " "keep it grounded and actionable, not predictive or fatalistic. " "(Always write the tip in English.)" ) return self._make_output(inp, prompt, snapshot)