|
|
@@ -485,13 +485,18 @@ async def get_transit_preview(
|
|
|
end_date: str,
|
|
|
transit_latitude: float | None = None,
|
|
|
transit_longitude: float | None = None,
|
|
|
- orb_limits: dict[str, float] | None = None,
|
|
|
+ min_significance: float = 0.0,
|
|
|
) -> dict[str, Any]:
|
|
|
- """Preview transit-to-natal aspects for a person over a time range.
|
|
|
+ """Daily transit-to-natal aspect snapshot for a person over a time range.
|
|
|
+
|
|
|
+ For each day, shows which transiting planets are aspecting which natal planets,
|
|
|
+ with orb, applying/separating status, and significance score (0-10).
|
|
|
|
|
|
- Finds when transiting planets make exact aspects to natal planets.
|
|
|
- Uses a two-pass approach: daily scan to find approach windows, then
|
|
|
- refinement to pinpoint the exact day of the aspect.
|
|
|
+ Significance is based on:
|
|
|
+ - Aspect type (conjunction/opposition > square > trine > sextile)
|
|
|
+ - Transiting planet (Saturn/Jupiter/Pluto > Mars > Sun > Venus/Mercury > Moon)
|
|
|
+ - Natal planet (Sun/Moon > Jupiter/Saturn > others)
|
|
|
+ - Orb tightness (tighter = more significant)
|
|
|
|
|
|
Args:
|
|
|
person_id: ID of a person in the persons database.
|
|
|
@@ -499,17 +504,15 @@ async def get_transit_preview(
|
|
|
end_date: ISO date string for the end of the range (YYYY-MM-DD).
|
|
|
transit_latitude: Current location latitude. Defaults to birth latitude.
|
|
|
transit_longitude: Current location longitude. Defaults to birth longitude.
|
|
|
- orb_limits: Optional dict of {aspect_name: max_orb_degrees}.
|
|
|
- Defaults: conjunction 8°, sextile 6°, square 8°, trine 8°, opposition 8°.
|
|
|
- Slow planets (Jupiter+) get 2x orbs.
|
|
|
+ min_significance: Minimum significance score to include (0-10). Default 0 = all.
|
|
|
|
|
|
Returns:
|
|
|
- List of transit events, each with date, transiting planet, natal planet,
|
|
|
- aspect type, and orb at the time of the event. Sorted by date.
|
|
|
+ Daily snapshots with active transit-to-natal aspects, sorted by date.
|
|
|
+ Each day includes the aspects active on that day with their details.
|
|
|
"""
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
|
|
- from . import storage
|
|
|
+ from . import storage, astrology
|
|
|
|
|
|
# Get person data
|
|
|
person = await storage.get_person(person_id=person_id)
|
|
|
@@ -535,7 +538,7 @@ async def get_transit_preview(
|
|
|
if (end - start).days > 365:
|
|
|
return {"error": "Date range too large. Maximum 365 days."}
|
|
|
|
|
|
- # Get natal planet positions
|
|
|
+ # Get natal planet positions (one ephemeris call)
|
|
|
natal_sky = await call_sky_state(
|
|
|
datetime=birth_dt, lat=birth_lat, lon=birth_lon, geocentric=True,
|
|
|
)
|
|
|
@@ -544,37 +547,20 @@ async def get_transit_preview(
|
|
|
|
|
|
natal_bodies = extract_bodies(natal_sky)
|
|
|
natal_lons = {b["body"]: b["ecliptic_lon"] for b in natal_bodies}
|
|
|
- natal_speeds = {b["body"]: b.get("speed_lon", 0.0) for b in natal_bodies}
|
|
|
|
|
|
- # Default orbs
|
|
|
- default_orbs = dict(astrology.DEFAULT_ORBS)
|
|
|
- if orb_limits:
|
|
|
- default_orbs.update(orb_limits)
|
|
|
-
|
|
|
- # Build list of (transiting, natal, aspect_name, target_angle, max_orb)
|
|
|
- # Planet names are the standard set from ephemeris-mcp
|
|
|
- planet_names = ["sun", "moon", "mercury", "venus", "mars", "jupiter", "saturn",
|
|
|
- "uranus", "neptune", "pluto", "chiron", "true_node"]
|
|
|
+ # Build list of all transit-to-natal aspect checks with per-pair orbs
|
|
|
aspect_checks = []
|
|
|
- for t_name in planet_names:
|
|
|
- for n_name, n_lon in natal_lons.items():
|
|
|
- is_slow = t_name in astrology.SLOW_PLANETS
|
|
|
+ for t_name in astrology.TRANSIT_ORB_RADII:
|
|
|
+ for n_name in natal_lons:
|
|
|
for asp_def in astrology.ASPECT_DEFINITIONS:
|
|
|
asp_name = asp_def["name"]
|
|
|
asp_angle = asp_def["angle"]
|
|
|
- max_orb = default_orbs.get(asp_name, asp_def["default_orb"])
|
|
|
- if max_orb is None:
|
|
|
- max_orb = asp_def["default_orb"]
|
|
|
- if is_slow:
|
|
|
- max_orb = max_orb * 2.0
|
|
|
+ max_orb = astrology.get_transit_orb(t_name, n_name, asp_name)
|
|
|
aspect_checks.append((t_name, n_name, asp_name, asp_angle, max_orb))
|
|
|
|
|
|
- # Phase 1: Daily scan to find approach windows
|
|
|
- # For each day, compute transiting planet positions and check orbs
|
|
|
- events = []
|
|
|
+ # Daily scan
|
|
|
+ days = []
|
|
|
current_day = start
|
|
|
- prev_orbs: dict[str, float] = {} # key -> orb on previous day
|
|
|
-
|
|
|
while current_day <= end:
|
|
|
iso = current_day.strftime("%Y-%m-%dT12:00:00Z")
|
|
|
sky = await call_sky_state(
|
|
|
@@ -588,6 +574,7 @@ async def get_transit_preview(
|
|
|
transit_lons = {b["body"]: b.get("ecliptic_lon", 0.0) for b in transit_bodies}
|
|
|
transit_speeds = {b["body"]: b.get("speed_lon", 0.0) for b in transit_bodies}
|
|
|
|
|
|
+ day_aspects = []
|
|
|
for t_name, n_name, asp_name, asp_angle, max_orb in aspect_checks:
|
|
|
if t_name not in transit_lons or n_name not in natal_lons:
|
|
|
continue
|
|
|
@@ -598,28 +585,36 @@ async def get_transit_preview(
|
|
|
diff = min(diff, 360.0 - diff)
|
|
|
orb = abs(diff - asp_angle)
|
|
|
|
|
|
- key = f"{t_name}_{asp_name}_{n_name}"
|
|
|
- prev_orb = prev_orbs.get(key)
|
|
|
-
|
|
|
- # Detect minimum: orb was decreasing, now increasing
|
|
|
- if prev_orb is not None and orb > prev_orb and prev_orb <= max_orb:
|
|
|
- # The exact aspect happened between yesterday and today
|
|
|
- # Refine by checking yesterday's position more precisely
|
|
|
- events.append({
|
|
|
- "date": (current_day - timedelta(days=1)).strftime("%Y-%m-%d"),
|
|
|
- "transiting": t_name,
|
|
|
- "natal": n_name,
|
|
|
- "aspect": asp_name,
|
|
|
- "orb": round(prev_orb, 4),
|
|
|
- "description": f"transit_{t_name}_{asp_name}_natal_{n_name}",
|
|
|
- })
|
|
|
+ if orb > max_orb:
|
|
|
+ continue
|
|
|
|
|
|
- prev_orbs[key] = orb
|
|
|
+ significance = astrology.get_transit_significance(t_name, n_name, asp_name, orb, max_orb)
|
|
|
+ if significance < min_significance:
|
|
|
+ continue
|
|
|
|
|
|
- current_day += timedelta(days=1)
|
|
|
+ # Determine applying/separating
|
|
|
+ t_speed = transit_speeds.get(t_name, 0.0)
|
|
|
+ applying = _is_applying(t_lon, n_lon, t_speed, 0.0, asp_angle)
|
|
|
+
|
|
|
+ day_aspects.append({
|
|
|
+ "transiting": t_name,
|
|
|
+ "natal": n_name,
|
|
|
+ "aspect": asp_name,
|
|
|
+ "orb": round(orb, 4),
|
|
|
+ "applying": applying,
|
|
|
+ "significance": significance,
|
|
|
+ })
|
|
|
+
|
|
|
+ # Sort by significance (highest first)
|
|
|
+ day_aspects.sort(key=lambda a: a["significance"], reverse=True)
|
|
|
|
|
|
- # Sort by date, then by orb (tightest first)
|
|
|
- events.sort(key=lambda e: (e["date"], e["orb"]))
|
|
|
+ days.append({
|
|
|
+ "date": current_day.strftime("%Y-%m-%d"),
|
|
|
+ "aspects": day_aspects,
|
|
|
+ "count": len(day_aspects),
|
|
|
+ })
|
|
|
+
|
|
|
+ current_day += timedelta(days=1)
|
|
|
|
|
|
return {
|
|
|
"input": {
|
|
|
@@ -627,9 +622,10 @@ async def get_transit_preview(
|
|
|
"person_name": person["name"],
|
|
|
"start_date": start_date,
|
|
|
"end_date": end_date,
|
|
|
+ "min_significance": min_significance,
|
|
|
},
|
|
|
- "events": events,
|
|
|
- "count": len(events),
|
|
|
+ "days": days,
|
|
|
+ "total_aspects": sum(d["count"] for d in days),
|
|
|
}
|
|
|
|
|
|
|
|
|
@@ -1049,5 +1045,5 @@ async def get_transit_preview_by_id(
|
|
|
end_date=end_date,
|
|
|
transit_latitude=transit_latitude if transit_latitude is not None else birth["latitude"],
|
|
|
transit_longitude=transit_longitude if transit_longitude is not None else birth["longitude"],
|
|
|
- orb_limits=orb_limits,
|
|
|
+ min_significance=0.0,
|
|
|
)
|