feat(agents): stars agent — astrological transits via pyswisseph (#121)
Computes natal chart (Sun/Moon/Mercury/Venus/Mars/Jupiter/Saturn) from birth_date and finds active transits (conjunction/sextile/square/trine/ opposition) between today's sky and the user's natal positions. Top 3 most-exact transits are passed to the orchestrator as interpretive themes to colour the tip — grounded and actionable, not predictive. Birth date sourced from agent_prefs (populated by a connected Google data source); requires data:google-health consent. Agent self-silences when birth_date is absent. pyswisseph added to ml/serving/requirements.txt. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
233
ml/agents/stars.py
Normal file
233
ml/agents/stars.py
Normal file
@@ -0,0 +1,233 @@
|
||||
"""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)
|
||||
Reference in New Issue
Block a user