368 lines
16 KiB
Python
368 lines
16 KiB
Python
"""Three-year annual pages on top of an already built Tajika pack.
|
|
|
|
The report year's Muntha, year lord, Sahams and Tajika objects are left as
|
|
they were. Extra years call the same solar-return report. New longitudes
|
|
come from that report's chart.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta
|
|
from typing import Any
|
|
|
|
try:
|
|
from annual_native.monthly_charts import build_monthly_charts
|
|
from annual_native.patyayini import build_native_patyayini
|
|
from annual_native.tithi_pravesh import build_tithi_pravesh
|
|
except ImportError: # pragma: no cover
|
|
from scripts.annual_native.monthly_charts import build_monthly_charts
|
|
from scripts.annual_native.patyayini import build_native_patyayini
|
|
from scripts.annual_native.tithi_pravesh import build_tithi_pravesh
|
|
|
|
|
|
def attach_annual_expansion(pack: dict, payload: dict, args: Any, *, include_overview: bool = False) -> dict:
|
|
"""Add one year's monthly returns, Patyayini and Tithi Pravesha.
|
|
|
|
Later years are separate annual packs. This does not recompute the year
|
|
lord, Muntha, Mudda, Sahams or Tajika already stored on ``pack``.
|
|
The 72-year lifetime table is not built here: one report would calculate
|
|
dozens of extra charts and can miss the 180 second wait.
|
|
"""
|
|
if not isinstance(pack, dict):
|
|
raise TypeError("annual pack must be a dict")
|
|
year = int(payload["target_year"])
|
|
place = _place(payload, args)
|
|
solar = _data(pack.get("solar_return"))
|
|
houses = pack.get("annual_houses") if isinstance(pack.get("annual_houses"), dict) else {}
|
|
mudda = _data(pack.get("mudda_dasha"))
|
|
muntha = _data(pack.get("muntha"))
|
|
year_lord = _data(pack.get("year_lord"))
|
|
sahams = _data(pack.get("sahams"))
|
|
jd = solar.get("jd_ut")
|
|
if not isinstance(jd, (int, float)) or isinstance(jd, bool):
|
|
raise ValueError(f"solar return jd_ut missing for {year}")
|
|
longitudes = houses.get("planet_longitudes") if isinstance(houses.get("planet_longitudes"), dict) else {}
|
|
annual_sun = longitudes.get("Sun")
|
|
if not isinstance(annual_sun, (int, float)) or isinstance(annual_sun, bool):
|
|
raise ValueError(f"annual Sun longitude missing for {year}")
|
|
asc = houses.get("ascendant_longitude")
|
|
if not isinstance(asc, (int, float)) or isinstance(asc, bool):
|
|
raise ValueError(f"annual ascendant longitude missing for {year}")
|
|
start = _parse_dt(solar.get("dt_local"))
|
|
if start is None:
|
|
raise ValueError(f"solar return local time missing for {year}")
|
|
monthly = build_monthly_charts(
|
|
annual_jd=float(jd),
|
|
birth_sun_longitude=float(annual_sun),
|
|
latitude=place["latitude"],
|
|
longitude=place["longitude"],
|
|
tz_hours=place["tz"],
|
|
ayanamsa=place["ayanamsa"],
|
|
node_mode=place["node_mode"],
|
|
target_year=year,
|
|
timezone_id=place["timezone_id"],
|
|
)
|
|
monthly_rows = monthly["rows"]
|
|
patyayini = build_native_patyayini(
|
|
ascendant_longitude=float(asc),
|
|
planet_longitudes=longitudes,
|
|
start=start,
|
|
year_days=place["year_days"],
|
|
)
|
|
patyayini_rows = [{**row, "target_year": year} for row in patyayini["antardasha_periods"]]
|
|
tithi = build_tithi_pravesh({**payload, "target_year": year})
|
|
mudda_rows = []
|
|
dasha_rows = []
|
|
for period in _periods(mudda):
|
|
mudda_rows.append({
|
|
"target_year": year,
|
|
"balance_profile": "Mudda",
|
|
"lord": period.get("lord"),
|
|
"start": period.get("start") or period.get("start_date"),
|
|
"end": period.get("end") or period.get("end_date"),
|
|
"boundary_status": "dasha_ending_dates",
|
|
})
|
|
dasha_rows.append({
|
|
"order": period.get("order") or len(dasha_rows) + 1,
|
|
"lord": period.get("lord"),
|
|
"annual_whole_sign_house": _house(longitudes.get(str(period.get("lord"))), asc),
|
|
"start": period.get("start") or period.get("start_date"),
|
|
"end": period.get("end") or period.get("end_date"),
|
|
"themes": {
|
|
"zh": "可能突出这颗星所主的事务",
|
|
"en": "This lord may stand out",
|
|
},
|
|
})
|
|
pack["monthly_chart_pack"] = {
|
|
"schema": "jyotish.tajika_monthly_chart_pack.v1",
|
|
"status": "partial_verified" if len(monthly_rows) == 12 else "blocked",
|
|
"reason": None if len(monthly_rows) == 12 else f"expected 12 monthly rows, got {len(monthly_rows)}",
|
|
"rows": monthly_rows,
|
|
}
|
|
pack["tithi_pravesh_years"] = [{
|
|
"target_year": year,
|
|
"status": tithi.get("status"),
|
|
"reason": tithi.get("reason"),
|
|
"natal_phase_degrees": tithi.get("natal_phase_degrees"),
|
|
"rows": tithi.get("rows") or [],
|
|
}]
|
|
pack["native_patyayini"] = {"status": "parameter_sensitive", "model": "degree_within_sign", "rows": patyayini_rows}
|
|
normalized = pack.get("normalized_tables")
|
|
if not isinstance(normalized, dict):
|
|
normalized = {}
|
|
pack["normalized_tables"] = normalized
|
|
if patyayini_rows and not normalized.get("patyayini_rows"):
|
|
normalized["patyayini_rows"] = patyayini_rows
|
|
# The primary year still lets the language-specific projector fill Mudda,
|
|
# including antardasha rows. Later years are not projected.
|
|
if mudda_rows and not include_overview and not normalized.get("mudda_rows"):
|
|
normalized["mudda_rows"] = mudda_rows
|
|
if dasha_rows and not pack.get("pl9_annual_dasha_results"):
|
|
pack["pl9_annual_dasha_results"] = {"status": "partial_verified", "rows": dasha_rows}
|
|
pack["annual_year_pages"] = [{
|
|
"target_year": year,
|
|
"muntha_sign": muntha.get("muntha_sign") or _nested(muntha, "muntha_sign"),
|
|
"year_lord": year_lord.get("year_lord") or year_lord.get("lord") or _nested(year_lord, "year_lord"),
|
|
"ascendant_longitude": asc,
|
|
"planet_longitudes": {
|
|
name: longitudes.get(name)
|
|
for name in ("Sun", "Moon", "Mars", "Mercury", "Jupiter", "Venus", "Saturn", "Rahu", "Ketu")
|
|
if isinstance(longitudes.get(name), (int, float)) and not isinstance(longitudes.get(name), bool)
|
|
},
|
|
"saham_count": sum(
|
|
1 for item in sahams.values()
|
|
if isinstance(item, dict) and isinstance(item.get("longitude"), (int, float)) and not isinstance(item.get("longitude"), bool)
|
|
),
|
|
"monthly_count": len(monthly_rows),
|
|
"tithi_count": len(tithi.get("rows") or []),
|
|
"local_datetime": start.strftime("%Y-%m-%d %H:%M:%S"),
|
|
}]
|
|
if include_overview:
|
|
try:
|
|
pack["annual_chart_series_pack"] = _eight_year(args, payload, place, year)
|
|
except Exception as exc:
|
|
import logging
|
|
logging.getLogger(__name__).warning("eight-year overview failed: %s", exc)
|
|
pack["annual_chart_series_pack"] = {
|
|
"status": "blocked",
|
|
"reason": f"{exc.__class__.__name__}: {exc}",
|
|
}
|
|
return pack
|
|
|
|
|
|
def _birth_place(args, place) -> dict:
|
|
located = dict(place)
|
|
located["latitude"] = float(args.lat)
|
|
located["longitude"] = float(args.lon)
|
|
located["tz"] = float(args.tz)
|
|
return located
|
|
|
|
|
|
def _lifetime(args, place) -> dict:
|
|
located = _birth_place(args, place)
|
|
birth_sun = _natal_sun(args, located)
|
|
rows = []
|
|
for age in range(72):
|
|
row = _return_row(args, located, birth_sun, int(args.year) + age)
|
|
row["completed_age"] = age
|
|
row["local_datetime"] = row.get("solar_return_local")
|
|
rows.append(row)
|
|
return {"status": "partial_verified", "pages": [{"rows": rows}]}
|
|
|
|
|
|
def _eight_year(args, payload, place, base_year: int) -> dict:
|
|
"""Birthplace overview for eight years. Current residence only when supplied."""
|
|
located = _birth_place(args, place)
|
|
birth_sun = _natal_sun(args, located)
|
|
rows = [_return_row(args, located, birth_sun, base_year + offset) for offset in range(8)]
|
|
series = {
|
|
"birth_place_eight_year_overview": {
|
|
"status": "partial_verified",
|
|
"rows": rows,
|
|
}
|
|
}
|
|
current = (payload.get("settings") or {}).get("current_location")
|
|
if isinstance(current, dict) and current.get("latitude") is not None and current.get("longitude") is not None:
|
|
local_place = dict(place)
|
|
local_place["latitude"] = float(current["latitude"])
|
|
local_place["longitude"] = float(current["longitude"])
|
|
if current.get("utc_offset") is not None:
|
|
local_place["tz"] = _offset_hours(current.get("utc_offset"))
|
|
series["local_eight_year_overview"] = {
|
|
"status": "partial_verified",
|
|
"rows": [_return_row(args, local_place, birth_sun, base_year + offset) for offset in range(8)],
|
|
}
|
|
return series
|
|
|
|
|
|
def _return_row(args, place, birth_sun: float, year: int) -> dict:
|
|
try:
|
|
from solar_return import _datetime_to_jd_ut, _localize_return_datetime
|
|
from jyotish_engine import compute_chart_data
|
|
from annual_native.sidereal_bodies import sun_moon_longitudes
|
|
except ImportError:
|
|
from scripts.solar_return import _datetime_to_jd_ut, _localize_return_datetime
|
|
from scripts.jyotish_engine import compute_chart_data
|
|
from scripts.annual_native.sidereal_bodies import sun_moon_longitudes
|
|
guess = _datetime_to_jd_ut(datetime(int(year), int(args.month), int(args.day), int(args.hour), int(args.minute)))
|
|
jd = guess
|
|
for _ in range(24):
|
|
sun = sun_moon_longitudes(jd, node_mode=place["node_mode"], ayanamsa_name=place["ayanamsa"])[0]
|
|
difference = ((sun - birth_sun + 180.0) % 360.0) - 180.0
|
|
if abs(difference) < 0.0001:
|
|
break
|
|
future = sun_moon_longitudes(jd + 1.0 / 24.0, node_mode=place["node_mode"], ayanamsa_name=place["ayanamsa"])[0]
|
|
speed = (((future - sun + 180.0) % 360.0) - 180.0) * 24.0
|
|
jd -= difference / speed if abs(speed) > 0.01 else difference
|
|
utc = datetime(2000, 1, 1) + timedelta(days=jd - 2451544.5)
|
|
local = _localize_return_datetime(utc, place["tz"], place["timezone_id"])
|
|
chart, *_rest = compute_chart_data(
|
|
utc.year, utc.month, utc.day, utc.hour, utc.minute,
|
|
place["latitude"], place["longitude"], 0,
|
|
place["node_mode"], second=utc.second, ayanamsa_name=place["ayanamsa"],
|
|
)
|
|
if not isinstance(chart, dict):
|
|
raise RuntimeError(f"compute_chart_data returned no chart for return year {year}")
|
|
asc = chart.get("ascendant") if isinstance(chart.get("ascendant"), dict) else {}
|
|
planets = chart.get("planets") if isinstance(chart.get("planets"), dict) else {}
|
|
return {
|
|
"target_year": year,
|
|
"age": int(year) - int(args.year),
|
|
"solar_return_local": local.strftime("%Y-%m-%d %H:%M:%S"),
|
|
"annual_ascendant": asc.get("sign"),
|
|
"ascendant": {
|
|
"sign": asc.get("sign"),
|
|
"degree_in_sign": asc.get("degree_in_sign", asc.get("degree")),
|
|
},
|
|
"planet_longitudes": {
|
|
name: data.get("degree_raw")
|
|
for name, data in planets.items()
|
|
if isinstance(data, dict) and isinstance(data.get("degree_raw"), (int, float)) and not isinstance(data.get("degree_raw"), bool)
|
|
},
|
|
}
|
|
|
|
|
|
def _natal_sun(args, place) -> float:
|
|
try:
|
|
from jyotish_engine import compute_chart_data
|
|
except ImportError:
|
|
from scripts.jyotish_engine import compute_chart_data
|
|
chart, *_rest = compute_chart_data(
|
|
int(args.year), int(args.month), int(args.day), int(args.hour), int(args.minute),
|
|
float(args.lat), float(args.lon), float(args.tz),
|
|
place["node_mode"], second=int(getattr(args, "second", 0) or 0), ayanamsa_name=place["ayanamsa"],
|
|
)
|
|
return float(chart["planets"]["Sun"]["degree_raw"]) % 360.0
|
|
|
|
|
|
def _solar_report(args, year: int) -> dict:
|
|
try:
|
|
from solar_return import solar_return_full_report
|
|
from jyotish_engine import _arg_second, _current_ayanamsa_name, _solar_return_location_kwargs
|
|
except ImportError:
|
|
from scripts.solar_return import solar_return_full_report
|
|
from scripts.jyotish_engine import _arg_second, _current_ayanamsa_name, _solar_return_location_kwargs
|
|
report = solar_return_full_report(
|
|
args.year, args.month, args.day, args.hour, args.minute,
|
|
args.lat, args.lon, args.tz, year,
|
|
ayanamsa_name=_current_ayanamsa_name(args),
|
|
birth_second=_arg_second(args),
|
|
**_solar_return_location_kwargs(args),
|
|
)
|
|
if not isinstance(report, dict) or report.get("error"):
|
|
raise RuntimeError(f"solar return {year} failed: {report.get('error') if isinstance(report, dict) else type(report).__name__}")
|
|
return report
|
|
|
|
|
|
def _place(payload: dict, args: Any) -> dict:
|
|
settings = payload.get("settings") if isinstance(payload.get("settings"), dict) else {}
|
|
birth = payload.get("birth") if isinstance(payload.get("birth"), dict) else {}
|
|
latitude = float(getattr(args, "lat", birth.get("latitude")))
|
|
longitude = float(getattr(args, "lon", birth.get("longitude")))
|
|
tz = float(getattr(args, "tz", 0.0) or 0.0)
|
|
if settings.get("solar_return_location_mode") == "current_location":
|
|
current = settings.get("current_location") if isinstance(settings.get("current_location"), dict) else {}
|
|
latitude = float(current["latitude"])
|
|
longitude = float(current["longitude"])
|
|
tz = _offset_hours(current.get("utc_offset"))
|
|
try:
|
|
year_days = float(settings.get("dasha_year_days") or 365.25)
|
|
except (TypeError, ValueError) as exc:
|
|
raise ValueError(f"dasha_year_days is not a number: {settings.get('dasha_year_days')}") from exc
|
|
return {
|
|
"latitude": latitude,
|
|
"longitude": longitude,
|
|
"tz": tz,
|
|
"ayanamsa": str(settings.get("ayanamsa") or "raman"),
|
|
"node_mode": str(settings.get("node_mode") or "mean"),
|
|
"timezone_id": birth.get("timezone_id") if isinstance(birth.get("timezone_id"), str) else None,
|
|
"year_days": year_days,
|
|
}
|
|
|
|
|
|
def _data(field: Any) -> dict:
|
|
if isinstance(field, dict) and isinstance(field.get("data"), dict):
|
|
return field["data"]
|
|
return field if isinstance(field, dict) else {}
|
|
|
|
|
|
def _periods(mudda: dict) -> list:
|
|
periods = mudda.get("periods")
|
|
if isinstance(periods, list) and periods:
|
|
return [row for row in periods if isinstance(row, dict)]
|
|
major = mudda.get("major")
|
|
if isinstance(major, list):
|
|
return [row for row in major if isinstance(row, dict)]
|
|
return []
|
|
|
|
|
|
def _nested(payload: dict, key: str):
|
|
data = payload.get("data") if isinstance(payload.get("data"), dict) else {}
|
|
return data.get(key)
|
|
|
|
|
|
def _house(longitude, ascendant) -> int | None:
|
|
if not isinstance(longitude, (int, float)) or isinstance(longitude, bool):
|
|
return None
|
|
if not isinstance(ascendant, (int, float)) or isinstance(ascendant, bool):
|
|
return None
|
|
return ((int(float(longitude) // 30) - int(float(ascendant) // 30)) % 12) + 1
|
|
|
|
|
|
def _saham_rows(sahams: dict) -> list:
|
|
rows = []
|
|
for key, item in sahams.items():
|
|
if key == "formula_traces" or not isinstance(item, dict):
|
|
continue
|
|
if not isinstance(item.get("longitude"), (int, float)) or isinstance(item.get("longitude"), bool):
|
|
continue
|
|
rows.append({
|
|
"name": item.get("name") or key,
|
|
"sign": item.get("sign"),
|
|
"degree_in_sign": item.get("degree_in_sign"),
|
|
})
|
|
return rows
|
|
|
|
|
|
def _parse_dt(value: Any) -> datetime | None:
|
|
if isinstance(value, datetime):
|
|
return value.replace(tzinfo=None)
|
|
if not isinstance(value, str) or not value.strip():
|
|
return None
|
|
text = value.strip().replace("T", " ")
|
|
for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M"):
|
|
try:
|
|
return datetime.strptime(text[:19], fmt)
|
|
except ValueError:
|
|
continue
|
|
return None
|
|
|
|
|
|
def _offset_hours(value: Any) -> float:
|
|
if isinstance(value, (int, float)) and not isinstance(value, bool):
|
|
return float(value)
|
|
text = str(value or "+00:00").strip()
|
|
sign = -1 if text.startswith("-") else 1
|
|
hours, minutes = [int(item) for item in text.lstrip("+-").split(":")[:2]]
|
|
return sign * (hours + minutes / 60.0)
|