|
|
@@ -425,6 +425,10 @@ async def calculate_synastry_chart(
|
|
|
elevation: float = 0.0,
|
|
|
house_system: str = "placidus",
|
|
|
orb_limits: dict[str, float] | None = None,
|
|
|
+ top_n_aspects: int | None = None,
|
|
|
+ karmic_filter: bool = False,
|
|
|
+ significator_filter: bool = False,
|
|
|
+ include_davison_full: bool = False,
|
|
|
) -> dict[str, Any]:
|
|
|
"""Calculate a synastry (relationship) chart for two people.
|
|
|
|
|
|
@@ -434,9 +438,14 @@ async def calculate_synastry_chart(
|
|
|
elevation: Birth elevation in meters.
|
|
|
house_system: House system (default: Placidus).
|
|
|
orb_limits: Optional orb configuration.
|
|
|
+ top_n_aspects: Limit interaspects to top N by orb.
|
|
|
+ karmic_filter: Only return Saturn/Pluto/Node interaspects.
|
|
|
+ significator_filter: Only return Venus-Mars, Moon-Venus, Sun-Moon, Sun-Saturn.
|
|
|
+ include_davison_full: Compute full Davison chart (planets, houses, aspects).
|
|
|
|
|
|
Returns:
|
|
|
- Synastry chart with interaspects, house overlays, composite, and Davison charts.
|
|
|
+ Synastry chart with interaspects, house overlays, composite, Davison chart,
|
|
|
+ and summary sections.
|
|
|
"""
|
|
|
sky1 = await call_sky_state(datetime=person1_datetime, lat=person1_latitude, lon=person1_longitude, elevation=elevation)
|
|
|
sky2 = await call_sky_state(datetime=person2_datetime, lat=person2_latitude, lon=person2_longitude, elevation=elevation)
|
|
|
@@ -493,6 +502,26 @@ async def calculate_synastry_chart(
|
|
|
|
|
|
interaspects.sort(key=lambda a: a["orb"])
|
|
|
|
|
|
+ # Apply filters
|
|
|
+ filtered_aspects = interaspects
|
|
|
+ if karmic_filter:
|
|
|
+ karmic_planets = {"saturn", "pluto", "true_node"}
|
|
|
+ filtered_aspects = [
|
|
|
+ a for a in filtered_aspects
|
|
|
+ if a["person1_planet"] in karmic_planets or a["person2_planet"] in karmic_planets
|
|
|
+ ]
|
|
|
+ if significator_filter:
|
|
|
+ significator_pairs = {
|
|
|
+ frozenset(["venus", "mars"]), frozenset(["moon", "venus"]),
|
|
|
+ frozenset(["sun", "moon"]), frozenset(["sun", "saturn"]),
|
|
|
+ }
|
|
|
+ filtered_aspects = [
|
|
|
+ a for a in filtered_aspects
|
|
|
+ if frozenset([a["person1_planet"], a["person2_planet"]]) in significator_pairs
|
|
|
+ ]
|
|
|
+ if top_n_aspects is not None:
|
|
|
+ filtered_aspects = filtered_aspects[:top_n_aspects]
|
|
|
+
|
|
|
# House overlays: person2's planets in person1's houses
|
|
|
p2_in_p1_houses = []
|
|
|
for p2 in chart2_planets:
|
|
|
@@ -525,11 +554,73 @@ async def calculate_synastry_chart(
|
|
|
"absolute_lon": z["absolute_lon"],
|
|
|
})
|
|
|
|
|
|
- # Davison chart (date midpoint)
|
|
|
+ # Davison chart
|
|
|
davison = astrology.compute_davison_chart(0.0, 0.0, person1_datetime, person2_datetime)
|
|
|
davison_mid_lat = (person1_latitude + person2_latitude) / 2
|
|
|
davison_mid_lon = (person1_longitude + person2_longitude) / 2
|
|
|
|
|
|
+ davison_result: dict[str, Any] = {
|
|
|
+ "date_midpoint_jd": davison["date_midpoint_jd"],
|
|
|
+ "latitude_midpoint": davison_mid_lat,
|
|
|
+ "longitude_midpoint": davison_mid_lon,
|
|
|
+ }
|
|
|
+
|
|
|
+ # Full Davison chart if requested
|
|
|
+ if include_davison_full:
|
|
|
+ davison_dt = _jd_to_datetime(davison["date_midpoint_jd"])
|
|
|
+ davison_sky = await call_sky_state(
|
|
|
+ datetime=davison_dt, lat=davison_mid_lat, lon=davison_mid_lon,
|
|
|
+ elevation=0.0, geocentric=True,
|
|
|
+ )
|
|
|
+ if "error" not in davison_sky:
|
|
|
+ davison_raw = extract_bodies(davison_sky)
|
|
|
+ davison_sidereal = davison_sky.get("sidereal_time", {})
|
|
|
+ davison_lst = davison_sidereal.get("local_sidereal_time", 0.0)
|
|
|
+ davison_houses = astrology.calculate_houses(davison_lst, davison_mid_lat, house_system)
|
|
|
+
|
|
|
+ davison_planets = []
|
|
|
+ for body in davison_raw:
|
|
|
+ ecl_lon = body.get("ecliptic_lon", 0.0)
|
|
|
+ z = astrology.ecliptic_to_zodiac(ecl_lon)
|
|
|
+ house = astrology.get_house_placement(ecl_lon, davison_houses)
|
|
|
+ davison_planets.append({
|
|
|
+ "body": body["body"],
|
|
|
+ "sign": z["sign"],
|
|
|
+ "degree_within_sign": z["degree"],
|
|
|
+ "absolute_lon": z["absolute_lon"],
|
|
|
+ "house": house,
|
|
|
+ "retrograde": astrology.is_retrograde(body.get("speed_lon")),
|
|
|
+ })
|
|
|
+
|
|
|
+ davison_aspect_bodies = [{"name": p["body"], "lon": p["absolute_lon"]} for p in davison_planets]
|
|
|
+ davison_aspects = astrology.compute_aspects(davison_aspect_bodies, orb_limits)
|
|
|
+ davison_formatted = []
|
|
|
+ for asp in davison_aspects:
|
|
|
+ davison_formatted.append({
|
|
|
+ "body1": asp["body1"],
|
|
|
+ "body2": asp["body2"],
|
|
|
+ "aspect": asp["aspect"],
|
|
|
+ "orb": asp["orb"],
|
|
|
+ "applying": asp["applying"],
|
|
|
+ "exactness": asp["exactness"],
|
|
|
+ })
|
|
|
+
|
|
|
+ davison_angles = astrology.calculate_angles(davison_lst, davison_mid_lat)
|
|
|
+
|
|
|
+ davison_result["planets"] = davison_planets
|
|
|
+ davison_result["houses"] = davison_houses
|
|
|
+ davison_result["aspects"] = davison_formatted
|
|
|
+ davison_result["angles"] = davison_angles
|
|
|
+
|
|
|
+ # Build summary
|
|
|
+ summary: dict[str, Any] = {
|
|
|
+ "top_aspects": interaspects[:15],
|
|
|
+ "saturn_contacts": [a for a in interaspects if a["person1_planet"] == "saturn" or a["person2_planet"] == "saturn"],
|
|
|
+ "node_contacts": [a for a in interaspects if a["person1_planet"] == "true_node" or a["person2_planet"] == "true_node"],
|
|
|
+ "venus_mars_contacts": [a for a in interaspects if frozenset([a["person1_planet"], a["person2_planet"]]) == frozenset(["venus", "mars"])],
|
|
|
+ "sun_moon_contacts": [a for a in interaspects if frozenset([a["person1_planet"], a["person2_planet"]]) == frozenset(["sun", "moon"])],
|
|
|
+ }
|
|
|
+
|
|
|
return {
|
|
|
"input": {
|
|
|
"person1": {"datetime": person1_datetime, "latitude": person1_latitude, "longitude": person1_longitude},
|
|
|
@@ -539,20 +630,26 @@ async def calculate_synastry_chart(
|
|
|
"chart_type": "synastry",
|
|
|
"chart1_natal": {"planets": chart1_planets, "houses": houses1},
|
|
|
"chart2_natal": {"planets": chart2_planets, "houses": houses2},
|
|
|
- "interaspects": interaspects,
|
|
|
+ "interaspects": filtered_aspects,
|
|
|
"house_overlays": {
|
|
|
"person2_in_person1_houses": p2_in_p1_houses,
|
|
|
"person1_in_person2_houses": p1_in_p2_houses,
|
|
|
},
|
|
|
"composite_chart": {"planets": composite_planets},
|
|
|
- "davison_chart": {
|
|
|
- "date_midpoint_jd": davison["date_midpoint_jd"],
|
|
|
- "latitude_midpoint": davison_mid_lat,
|
|
|
- "longitude_midpoint": davison_mid_lon,
|
|
|
- },
|
|
|
+ "davison_chart": davison_result,
|
|
|
+ "summary": summary,
|
|
|
}
|
|
|
|
|
|
|
|
|
+def _jd_to_datetime(jd: float) -> str:
|
|
|
+ """Convert Julian Day to ISO 8601 datetime string."""
|
|
|
+ from datetime import datetime, timezone, timedelta
|
|
|
+ # JD 2440587.5 = 1970-01-01T00:00:00Z
|
|
|
+ unix_seconds = (jd - 2440587.5) * 86400.0
|
|
|
+ dt = datetime.fromtimestamp(unix_seconds, tz=timezone.utc)
|
|
|
+ return dt.strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
|
+
|
|
|
+
|
|
|
# ── Tool: get_transit_preview ────────────────────────────────────────
|
|
|
|
|
|
@mcp.tool()
|
|
|
@@ -935,6 +1032,436 @@ async def calculate_composite_chart(
|
|
|
}
|
|
|
|
|
|
|
|
|
+# ── Tool: calculate_davison_chart ─────────────────────────────────────
|
|
|
+
|
|
|
+@mcp.tool()
|
|
|
+async def calculate_davison_chart(
|
|
|
+ person1_datetime: str,
|
|
|
+ person1_latitude: float,
|
|
|
+ person1_longitude: float,
|
|
|
+ person2_datetime: str,
|
|
|
+ person2_latitude: float,
|
|
|
+ person2_longitude: float,
|
|
|
+ elevation: float = 0.0,
|
|
|
+ house_system: str = "placidus",
|
|
|
+ orb_limits: dict[str, float] | None = None,
|
|
|
+) -> dict[str, Any]:
|
|
|
+ """Calculate a Davison chart (midpoint in time and space) for two people.
|
|
|
+
|
|
|
+ The Davison chart is a real moment in time (unlike the composite which is
|
|
|
+ purely symbolic). It can be progressed and directed like a natal chart.
|
|
|
+
|
|
|
+ Args:
|
|
|
+ person1_datetime, person1_latitude, person1_longitude: Person 1 birth data.
|
|
|
+ person2_datetime, person2_latitude, person2_longitude: Person 2 birth data.
|
|
|
+ elevation: Birth elevation in meters.
|
|
|
+ house_system: House system (default: Placidus).
|
|
|
+ orb_limits: Optional orb configuration.
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ Full Davison chart with planets, houses, aspects, and angles.
|
|
|
+ """
|
|
|
+ davison = astrology.compute_davison_chart(0.0, 0.0, person1_datetime, person2_datetime)
|
|
|
+ mid_lat = (person1_latitude + person2_latitude) / 2
|
|
|
+ mid_lon = (person1_longitude + person2_longitude) / 2
|
|
|
+ davison_dt = _jd_to_datetime(davison["date_midpoint_jd"])
|
|
|
+
|
|
|
+ sky = await call_sky_state(
|
|
|
+ datetime=davison_dt, lat=mid_lat, lon=mid_lon,
|
|
|
+ elevation=elevation, geocentric=True,
|
|
|
+ )
|
|
|
+ if "error" in sky:
|
|
|
+ return {"error": f"davison ephemeris error: {sky['error']}"}
|
|
|
+
|
|
|
+ raw_bodies = extract_bodies(sky)
|
|
|
+ sidereal = sky.get("sidereal_time", {})
|
|
|
+ lst_hours = sidereal.get("local_sidereal_time", 0.0)
|
|
|
+ houses = astrology.calculate_houses(lst_hours, mid_lat, house_system)
|
|
|
+
|
|
|
+ planets = []
|
|
|
+ for body in raw_bodies:
|
|
|
+ ecl_lon = body.get("ecliptic_lon", 0.0)
|
|
|
+ z = astrology.ecliptic_to_zodiac(ecl_lon)
|
|
|
+ house = astrology.get_house_placement(ecl_lon, houses)
|
|
|
+ planets.append({
|
|
|
+ "body": body["body"],
|
|
|
+ "sign": z["sign"],
|
|
|
+ "degree_within_sign": z["degree"],
|
|
|
+ "absolute_lon": z["absolute_lon"],
|
|
|
+ "house": house,
|
|
|
+ "retrograde": astrology.is_retrograde(body.get("speed_lon")),
|
|
|
+ })
|
|
|
+
|
|
|
+ aspect_bodies = [{"name": p["body"], "lon": p["absolute_lon"]} for p in planets]
|
|
|
+ aspects = astrology.compute_aspects(aspect_bodies, orb_limits)
|
|
|
+ formatted_aspects = []
|
|
|
+ for asp in aspects:
|
|
|
+ formatted_aspects.append({
|
|
|
+ "body1": asp["body1"],
|
|
|
+ "body2": asp["body2"],
|
|
|
+ "aspect": asp["aspect"],
|
|
|
+ "orb": asp["orb"],
|
|
|
+ "applying": asp["applying"],
|
|
|
+ "exactness": asp["exactness"],
|
|
|
+ })
|
|
|
+
|
|
|
+ angles = astrology.calculate_angles(lst_hours, mid_lat)
|
|
|
+
|
|
|
+ return {
|
|
|
+ "input": {
|
|
|
+ "person1": {"datetime": person1_datetime, "latitude": person1_latitude, "longitude": person1_longitude},
|
|
|
+ "person2": {"datetime": person2_datetime, "latitude": person2_latitude, "longitude": person2_longitude},
|
|
|
+ "house_system": house_system,
|
|
|
+ },
|
|
|
+ "chart_type": "davison",
|
|
|
+ "date_midpoint_jd": davison["date_midpoint_jd"],
|
|
|
+ "location_midpoint": {"latitude": mid_lat, "longitude": mid_lon},
|
|
|
+ "planets": planets,
|
|
|
+ "houses": houses,
|
|
|
+ "aspects": formatted_aspects,
|
|
|
+ "angles": angles,
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+# ── Tool: get_composite_transit_preview ───────────────────────────────
|
|
|
+
|
|
|
+@mcp.tool()
|
|
|
+async def get_composite_transit_preview(
|
|
|
+ person1_datetime: str,
|
|
|
+ person1_latitude: float,
|
|
|
+ person1_longitude: float,
|
|
|
+ person2_datetime: str,
|
|
|
+ person2_latitude: float,
|
|
|
+ person2_longitude: float,
|
|
|
+ start_date: str,
|
|
|
+ end_date: str,
|
|
|
+ min_significance: float = 0.0,
|
|
|
+) -> dict[str, Any]:
|
|
|
+ """Daily transit-to-composite chart aspect snapshot over a time range.
|
|
|
+
|
|
|
+ Calculates the composite chart for two people, then shows transiting
|
|
|
+ aspects to composite planet positions for each day in the range.
|
|
|
+
|
|
|
+ Args:
|
|
|
+ person1_datetime, person1_latitude, person1_longitude: Person 1 birth data.
|
|
|
+ person2_datetime, person2_latitude, person2_longitude: Person 2 birth data.
|
|
|
+ start_date: ISO date string for the start of the range (YYYY-MM-DD).
|
|
|
+ end_date: ISO date string for the end of the range (YYYY-MM-DD).
|
|
|
+ min_significance: Minimum significance score (0-10). Default 0 = all.
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ Daily snapshots with active transit-to-composite aspects.
|
|
|
+ """
|
|
|
+ from datetime import datetime, timedelta, timezone
|
|
|
+
|
|
|
+ # Calculate composite chart
|
|
|
+ sky1 = await call_sky_state(datetime=person1_datetime, lat=person1_latitude, lon=person1_longitude, geocentric=True)
|
|
|
+ sky2 = await call_sky_state(datetime=person2_datetime, lat=person2_latitude, lon=person2_longitude, geocentric=True)
|
|
|
+ if "error" in sky1:
|
|
|
+ return {"error": f"person1: {sky1['error']}"}
|
|
|
+ if "error" in sky2:
|
|
|
+ return {"error": f"person2: {sky2['error']}"}
|
|
|
+
|
|
|
+ bodies1 = extract_bodies(sky1)
|
|
|
+ bodies2 = extract_bodies(sky2)
|
|
|
+ composite_bodies = astrology.compute_composite_chart(
|
|
|
+ [{"name": b["body"], "lon": b.get("ecliptic_lon", 0.0)} for b in bodies1],
|
|
|
+ [{"name": b["body"], "lon": b.get("ecliptic_lon", 0.0)} for b in bodies2],
|
|
|
+ )
|
|
|
+ composite_lons = {b["name"]: b["lon"] for b in composite_bodies}
|
|
|
+
|
|
|
+ # Parse date range
|
|
|
+ try:
|
|
|
+ start = datetime.fromisoformat(start_date).replace(tzinfo=timezone.utc)
|
|
|
+ end = datetime.fromisoformat(end_date).replace(tzinfo=timezone.utc)
|
|
|
+ except Exception:
|
|
|
+ return {"error": "Invalid date format. Use ISO format (YYYY-MM-DD)."}
|
|
|
+ if end <= start:
|
|
|
+ return {"error": "end_date must be after start_date"}
|
|
|
+ if (end - start).days > 365:
|
|
|
+ return {"error": "Date range too large. Maximum 365 days."}
|
|
|
+
|
|
|
+ # Build aspect checks
|
|
|
+ aspect_checks = []
|
|
|
+ for t_name in astrology.TRANSIT_ORB_RADII:
|
|
|
+ for c_name in composite_lons:
|
|
|
+ for asp_def in astrology.ASPECT_DEFINITIONS:
|
|
|
+ asp_name = asp_def["name"]
|
|
|
+ asp_angle = asp_def["angle"]
|
|
|
+ max_orb = astrology.get_transit_orb(t_name, c_name, asp_name)
|
|
|
+ aspect_checks.append((t_name, c_name, asp_name, asp_angle, max_orb))
|
|
|
+
|
|
|
+ # Daily scan
|
|
|
+ days = []
|
|
|
+ current_day = start
|
|
|
+ while current_day <= end:
|
|
|
+ iso = current_day.strftime("%Y-%m-%dT12:00:00Z")
|
|
|
+ sky = await call_sky_state(datetime=iso, lat=0.0, lon=0.0, geocentric=True)
|
|
|
+ if "error" in sky:
|
|
|
+ current_day += timedelta(days=1)
|
|
|
+ continue
|
|
|
+
|
|
|
+ transit_bodies = extract_bodies(sky)
|
|
|
+ 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, c_name, asp_name, asp_angle, max_orb in aspect_checks:
|
|
|
+ if t_name not in transit_lons or c_name not in composite_lons:
|
|
|
+ continue
|
|
|
+ t_lon = transit_lons[t_name]
|
|
|
+ c_lon = composite_lons[c_name]
|
|
|
+ diff = abs(t_lon - c_lon)
|
|
|
+ diff = min(diff, 360.0 - diff)
|
|
|
+ orb = abs(diff - asp_angle)
|
|
|
+ if orb > max_orb:
|
|
|
+ continue
|
|
|
+ significance = astrology.get_transit_significance(t_name, c_name, asp_name, orb, max_orb)
|
|
|
+ if significance < min_significance:
|
|
|
+ continue
|
|
|
+ t_speed = transit_speeds.get(t_name, 0.0)
|
|
|
+ applying = astrology._is_applying(t_lon, c_lon, t_speed, 0.0, asp_angle)
|
|
|
+ day_aspects.append({
|
|
|
+ "transiting": t_name,
|
|
|
+ "composite": c_name,
|
|
|
+ "aspect": asp_name,
|
|
|
+ "orb": round(orb, 4),
|
|
|
+ "applying": applying,
|
|
|
+ "significance": significance,
|
|
|
+ })
|
|
|
+
|
|
|
+ day_aspects.sort(key=lambda a: a["significance"], reverse=True)
|
|
|
+ days.append({
|
|
|
+ "date": current_day.strftime("%Y-%m-%d"),
|
|
|
+ "aspects": day_aspects,
|
|
|
+ "count": len(day_aspects),
|
|
|
+ })
|
|
|
+ current_day += timedelta(days=1)
|
|
|
+
|
|
|
+ return {
|
|
|
+ "input": {
|
|
|
+ "person1": {"datetime": person1_datetime, "latitude": person1_latitude, "longitude": person1_longitude},
|
|
|
+ "person2": {"datetime": person2_datetime, "latitude": person2_latitude, "longitude": person2_longitude},
|
|
|
+ "start_date": start_date,
|
|
|
+ "end_date": end_date,
|
|
|
+ },
|
|
|
+ "days": days,
|
|
|
+ "total_aspects": sum(d["count"] for d in days),
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+# ── Tool: get_davison_transit_preview ─────────────────────────────────
|
|
|
+
|
|
|
+@mcp.tool()
|
|
|
+async def get_davison_transit_preview(
|
|
|
+ person1_datetime: str,
|
|
|
+ person1_latitude: float,
|
|
|
+ person1_longitude: float,
|
|
|
+ person2_datetime: str,
|
|
|
+ person2_latitude: float,
|
|
|
+ person2_longitude: float,
|
|
|
+ start_date: str,
|
|
|
+ end_date: str,
|
|
|
+ min_significance: float = 0.0,
|
|
|
+) -> dict[str, Any]:
|
|
|
+ """Daily transit-to-Davison chart aspect snapshot over a time range.
|
|
|
+
|
|
|
+ Calculates the Davison chart for two people, then shows transiting
|
|
|
+ aspects to Davison planet positions for each day in the range.
|
|
|
+
|
|
|
+ Args:
|
|
|
+ person1_datetime, person1_latitude, person1_longitude: Person 1 birth data.
|
|
|
+ person2_datetime, person2_latitude, person2_longitude: Person 2 birth data.
|
|
|
+ start_date: ISO date string for the start of the range (YYYY-MM-DD).
|
|
|
+ end_date: ISO date string for the end of the range (YYYY-MM-DD).
|
|
|
+ min_significance: Minimum significance score (0-10). Default 0 = all.
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ Daily snapshots with active transit-to-Davison aspects.
|
|
|
+ """
|
|
|
+ from datetime import datetime, timedelta, timezone
|
|
|
+
|
|
|
+ # Calculate Davison chart
|
|
|
+ davison = astrology.compute_davison_chart(0.0, 0.0, person1_datetime, person2_datetime)
|
|
|
+ mid_lat = (person1_latitude + person2_latitude) / 2
|
|
|
+ mid_lon = (person1_longitude + person2_longitude) / 2
|
|
|
+ davison_dt = _jd_to_datetime(davison["date_midpoint_jd"])
|
|
|
+
|
|
|
+ sky = await call_sky_state(datetime=davison_dt, lat=mid_lat, lon=mid_lon, geocentric=True)
|
|
|
+ if "error" in sky:
|
|
|
+ return {"error": f"davison ephemeris error: {sky['error']}"}
|
|
|
+
|
|
|
+ raw_bodies = extract_bodies(sky)
|
|
|
+ davison_lons = {b["body"]: b.get("ecliptic_lon", 0.0) for b in raw_bodies}
|
|
|
+
|
|
|
+ # Parse date range
|
|
|
+ try:
|
|
|
+ start = datetime.fromisoformat(start_date).replace(tzinfo=timezone.utc)
|
|
|
+ end = datetime.fromisoformat(end_date).replace(tzinfo=timezone.utc)
|
|
|
+ except Exception:
|
|
|
+ return {"error": "Invalid date format. Use ISO format (YYYY-MM-DD)."}
|
|
|
+ if end <= start:
|
|
|
+ return {"error": "end_date must be after start_date"}
|
|
|
+ if (end - start).days > 365:
|
|
|
+ return {"error": "Date range too large. Maximum 365 days."}
|
|
|
+
|
|
|
+ # Build aspect checks
|
|
|
+ aspect_checks = []
|
|
|
+ for t_name in astrology.TRANSIT_ORB_RADII:
|
|
|
+ for d_name in davison_lons:
|
|
|
+ for asp_def in astrology.ASPECT_DEFINITIONS:
|
|
|
+ asp_name = asp_def["name"]
|
|
|
+ asp_angle = asp_def["angle"]
|
|
|
+ max_orb = astrology.get_transit_orb(t_name, d_name, asp_name)
|
|
|
+ aspect_checks.append((t_name, d_name, asp_name, asp_angle, max_orb))
|
|
|
+
|
|
|
+ # Daily scan
|
|
|
+ days = []
|
|
|
+ current_day = start
|
|
|
+ while current_day <= end:
|
|
|
+ iso = current_day.strftime("%Y-%m-%dT12:00:00Z")
|
|
|
+ sky = await call_sky_state(datetime=iso, lat=0.0, lon=0.0, geocentric=True)
|
|
|
+ if "error" in sky:
|
|
|
+ current_day += timedelta(days=1)
|
|
|
+ continue
|
|
|
+
|
|
|
+ transit_bodies = extract_bodies(sky)
|
|
|
+ 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, d_name, asp_name, asp_angle, max_orb in aspect_checks:
|
|
|
+ if t_name not in transit_lons or d_name not in davison_lons:
|
|
|
+ continue
|
|
|
+ t_lon = transit_lons[t_name]
|
|
|
+ d_lon = davison_lons[d_name]
|
|
|
+ diff = abs(t_lon - d_lon)
|
|
|
+ diff = min(diff, 360.0 - diff)
|
|
|
+ orb = abs(diff - asp_angle)
|
|
|
+ if orb > max_orb:
|
|
|
+ continue
|
|
|
+ significance = astrology.get_transit_significance(t_name, d_name, asp_name, orb, max_orb)
|
|
|
+ if significance < min_significance:
|
|
|
+ continue
|
|
|
+ t_speed = transit_speeds.get(t_name, 0.0)
|
|
|
+ applying = astrology._is_applying(t_lon, d_lon, t_speed, 0.0, asp_angle)
|
|
|
+ day_aspects.append({
|
|
|
+ "transiting": t_name,
|
|
|
+ "davison": d_name,
|
|
|
+ "aspect": asp_name,
|
|
|
+ "orb": round(orb, 4),
|
|
|
+ "applying": applying,
|
|
|
+ "significance": significance,
|
|
|
+ })
|
|
|
+
|
|
|
+ day_aspects.sort(key=lambda a: a["significance"], reverse=True)
|
|
|
+ days.append({
|
|
|
+ "date": current_day.strftime("%Y-%m-%d"),
|
|
|
+ "aspects": day_aspects,
|
|
|
+ "count": len(day_aspects),
|
|
|
+ })
|
|
|
+ current_day += timedelta(days=1)
|
|
|
+
|
|
|
+ return {
|
|
|
+ "input": {
|
|
|
+ "person1": {"datetime": person1_datetime, "latitude": person1_latitude, "longitude": person1_longitude},
|
|
|
+ "person2": {"datetime": person2_datetime, "latitude": person2_latitude, "longitude": person2_longitude},
|
|
|
+ "start_date": start_date,
|
|
|
+ "end_date": end_date,
|
|
|
+ },
|
|
|
+ "days": days,
|
|
|
+ "total_aspects": sum(d["count"] for d in days),
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+# ── Tool: get_karmic_relationship_summary ─────────────────────────────
|
|
|
+
|
|
|
+@mcp.tool()
|
|
|
+async def get_karmic_relationship_summary(
|
|
|
+ person1_id: str,
|
|
|
+ person2_id: str,
|
|
|
+ house_system: str = "placidus",
|
|
|
+) -> dict[str, Any]:
|
|
|
+ """Generate a karmic relationship summary from synastry, composite, and Davison charts.
|
|
|
+
|
|
|
+ Combines karmic indicators across all three relationship chart layers:
|
|
|
+ - Synastry: Saturn/Pluto/Node interchart aspects
|
|
|
+ - Composite: Saturn, Pluto, Node positions
|
|
|
+ - Davison: Saturn, Pluto, Node positions
|
|
|
+
|
|
|
+ Args:
|
|
|
+ person1_id: ID or nickname of person 1 in the persons database.
|
|
|
+ person2_id: ID or nickname of person 2 in the persons database.
|
|
|
+ house_system: House system (default: Placidus).
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ Structured karmic summary with indicators from all three layers.
|
|
|
+ """
|
|
|
+ # Get synastry with karmic filter
|
|
|
+ synastry = await calculate_synastry_chart_by_id(
|
|
|
+ person1_id, person2_id,
|
|
|
+ house_system=house_system,
|
|
|
+ karmic_filter=True,
|
|
|
+ )
|
|
|
+ if "error" in synastry:
|
|
|
+ return synastry
|
|
|
+
|
|
|
+ # Get composite chart
|
|
|
+ composite = await calculate_composite_chart_by_id(
|
|
|
+ person1_id, person2_id,
|
|
|
+ house_system=house_system,
|
|
|
+ )
|
|
|
+ if "error" in composite:
|
|
|
+ return composite
|
|
|
+
|
|
|
+ # Get Davison chart
|
|
|
+ davison = await calculate_davison_chart_by_id(
|
|
|
+ person1_id, person2_id,
|
|
|
+ house_system=house_system,
|
|
|
+ )
|
|
|
+ if "error" in davison:
|
|
|
+ return davison
|
|
|
+
|
|
|
+ def _extract_karmic_planets(chart_data: dict, key: str) -> dict:
|
|
|
+ """Extract Saturn, Pluto, Node from a chart's planet list."""
|
|
|
+ result = {}
|
|
|
+ for p in chart_data.get(key, []):
|
|
|
+ if p["body"] in ("saturn", "pluto", "true_node"):
|
|
|
+ result[p["body"]] = {
|
|
|
+ "sign": p.get("sign"),
|
|
|
+ "house": p.get("house"),
|
|
|
+ "retrograde": p.get("retrograde"),
|
|
|
+ }
|
|
|
+ return result
|
|
|
+
|
|
|
+ synastry_karmic_aspects = synastry.get("interaspects", [])
|
|
|
+ composite_karmic = _extract_karmic_planets(composite, "planets")
|
|
|
+ davison_karmic = _extract_karmic_planets(davison, "planets")
|
|
|
+
|
|
|
+ # Count karmic weight
|
|
|
+ karmic_weight = len(synastry_karmic_aspects)
|
|
|
+ if composite_karmic.get("saturn"):
|
|
|
+ karmic_weight += 1
|
|
|
+ if composite_karmic.get("pluto"):
|
|
|
+ karmic_weight += 1
|
|
|
+ if davison_karmic.get("saturn"):
|
|
|
+ karmic_weight += 1
|
|
|
+ if davison_karmic.get("pluto"):
|
|
|
+ karmic_weight += 1
|
|
|
+
|
|
|
+ return {
|
|
|
+ "karmic_weight": karmic_weight,
|
|
|
+ "synastry_karmic_aspects": synastry_karmic_aspects,
|
|
|
+ "composite_karmic_planets": composite_karmic,
|
|
|
+ "davison_karmic_planets": davison_karmic,
|
|
|
+ "summary": {
|
|
|
+ "saturn_contacts": len([a for a in synastry_karmic_aspects if "saturn" in (a["person1_planet"], a["person2_planet"])]),
|
|
|
+ "pluto_contacts": len([a for a in synastry_karmic_aspects if "pluto" in (a["person1_planet"], a["person2_planet"])]),
|
|
|
+ "node_contacts": len([a for a in synastry_karmic_aspects if "true_node" in (a["person1_planet"], a["person2_planet"])]),
|
|
|
+ },
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
# ── Tool: list_house_systems ─────────────────────────────────────────
|
|
|
|
|
|
@mcp.tool()
|
|
|
@@ -1117,6 +1644,43 @@ async def calculate_composite_chart_by_id(
|
|
|
)
|
|
|
|
|
|
|
|
|
+@mcp.tool()
|
|
|
+async def calculate_davison_chart_by_id(
|
|
|
+ person1_id: str,
|
|
|
+ person2_id: str,
|
|
|
+ house_system: str = "placidus",
|
|
|
+ orb_limits: dict[str, float] | None = None,
|
|
|
+) -> dict[str, Any]:
|
|
|
+ """Calculate Davison chart for two persons from the database.
|
|
|
+
|
|
|
+ Args:
|
|
|
+ person1_id: ID of person 1 in the persons database.
|
|
|
+ person2_id: ID of person 2 in the persons database.
|
|
|
+ house_system: House system (default: Placidus).
|
|
|
+ orb_limits: Optional orb configuration.
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ Davison chart structure.
|
|
|
+ """
|
|
|
+ p1 = await _get_person_birth_data(person1_id)
|
|
|
+ if "error" in p1:
|
|
|
+ return p1
|
|
|
+ p2 = await _get_person_birth_data(person2_id)
|
|
|
+ if "error" in p2:
|
|
|
+ return p2
|
|
|
+ return await calculate_davison_chart(
|
|
|
+ person1_datetime=p1["birth_datetime"],
|
|
|
+ person1_latitude=p1["latitude"],
|
|
|
+ person1_longitude=p1["longitude"],
|
|
|
+ person2_datetime=p2["birth_datetime"],
|
|
|
+ person2_latitude=p2["latitude"],
|
|
|
+ person2_longitude=p2["longitude"],
|
|
|
+ elevation=p1.get("elevation", 0.0),
|
|
|
+ house_system=house_system,
|
|
|
+ orb_limits=orb_limits,
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
@mcp.tool()
|
|
|
async def get_transit_preview_by_id(
|
|
|
person_id: str,
|