当年和后两年写出月返照、原生 Patyayini 和 Tithi Pravesha。已有年主、Muntha、Mudda、Sahams、Tajika 不改写。出生地八年总览会算,72 年生命周期年返先不做。

Co-Authored-By: Grok (agent loop)
This commit is contained in:
jesse-ux
2026-10-23 10:17:13 +08:00
co-authored by Grok (agent loop)
parent 9b2e6aa0b4
commit 4c405d2a51
11 changed files with 1045 additions and 27 deletions
+367
View File
@@ -0,0 +1,367 @@
"""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)
+8 -2
View File
@@ -41,6 +41,7 @@ def build_annual_whole_sign_houses(chart: dict | None) -> dict[str, Any]:
asc_index = int(asc_lon // 30.0) % 12
placed: dict[int, list[str]] = {index: [] for index in range(12)}
source_longitudes = {}
planet_longitudes = {}
for name in ("Sun", "Moon", "Mars", "Mercury", "Jupiter", "Venus", "Saturn", "Rahu", "Ketu"):
data = planets.get(name)
if not isinstance(data, dict):
@@ -50,8 +51,11 @@ def build_annual_whole_sign_houses(chart: dict | None) -> dict[str, Any]:
longitude = _longitude(data.get("degree"))
if longitude is None:
continue
if name in ("Sun", "Moon") and data.get("degree_raw") is not None and not isinstance(data.get("degree_raw"), bool):
source_longitudes[name] = float(data["degree_raw"]) % 360.0
if data.get("degree_raw") is not None and not isinstance(data.get("degree_raw"), bool):
raw = float(data["degree_raw"]) % 360.0
planet_longitudes[name] = raw
if name in ("Sun", "Moon"):
source_longitudes[name] = raw
placed[int(longitude // 30.0) % 12].append(name)
houses = []
for house in range(1, 13):
@@ -69,5 +73,7 @@ def build_annual_whole_sign_houses(chart: dict | None) -> dict[str, Any]:
"status": "partial_verified",
"house_system": "whole_sign",
"houses": houses,
"ascendant_longitude": asc_lon,
"planet_longitudes": planet_longitudes,
"source_longitudes": source_longitudes,
}
+125
View File
@@ -0,0 +1,125 @@
"""Twelve monthly charts measured from an annual solar-return instant.
The Sun search uses this repository's sidereal Sun. Each row's planets come
from ``compute_chart_data``. No apparent-position flag is passed.
"""
from __future__ import annotations
from datetime import datetime, timedelta
from typing import Any
try:
from annual_native.sidereal_bodies import sun_moon_longitudes
except ImportError: # pragma: no cover - repo root is on sys.path
from scripts.annual_native.sidereal_bodies import sun_moon_longitudes
SCHEMA = "jyotish.tajika_monthly_chart_pack.v1"
TROPICAL_YEAR_DAYS = 365.24219
_PLANETS = ("Sun", "Moon", "Mars", "Mercury", "Jupiter", "Venus", "Saturn", "Rahu", "Ketu")
def build_monthly_charts(
*,
annual_jd: float,
birth_sun_longitude: float,
latitude: float,
longitude: float,
tz_hours: float,
ayanamsa: str,
node_mode: str,
target_year: int,
timezone_id: str | None = None,
) -> dict[str, Any]:
rows = []
for month_index in range(1, 13):
target_sun = (float(birth_sun_longitude) + 30.0 * (month_index - 1)) % 360.0
guess = float(annual_jd) + TROPICAL_YEAR_DAYS * (month_index - 1) / 12.0
monthly_jd = _find_sun(guess, target_sun, ayanamsa, node_mode)
rows.append(_chart_row(
month_index, monthly_jd, target_sun, target_year,
latitude, longitude, tz_hours, ayanamsa, node_mode, timezone_id,
))
ready = [row for row in rows if row.get("status") == "partial_verified"]
return {
"schema": SCHEMA,
"status": "partial_verified" if len(ready) == 12 else "blocked",
"reason": None if len(ready) == 12 else "monthly_root_incomplete",
"target_year": target_year,
"rows": rows,
}
def _find_sun(guess: float, target: float, ayanamsa: str, node_mode: str) -> float | None:
jd = guess
for _ in range(24):
longitude = sun_moon_longitudes(jd, node_mode=node_mode, ayanamsa_name=ayanamsa)[0]
difference = ((longitude - target + 180.0) % 360.0) - 180.0
if abs(difference) < 0.0001:
return jd
future = sun_moon_longitudes(jd + 1.0 / 24.0, node_mode=node_mode, ayanamsa_name=ayanamsa)[0]
speed = (((future - longitude + 180.0) % 360.0) - 180.0) * 24.0
jd -= difference / speed if abs(speed) > 0.01 else difference
final = sun_moon_longitudes(jd, node_mode=node_mode, ayanamsa_name=ayanamsa)[0]
if abs(((final - target + 180.0) % 360.0) - 180.0) < 0.001:
return jd
return None
def _chart_row(
month_index: int,
jd: float | None,
target_sun: float,
target_year: int,
latitude: float,
longitude: float,
tz_hours: float,
ayanamsa: str,
node_mode: str,
timezone_id: str | None,
) -> dict[str, Any]:
if jd is None:
return {
"month_index": month_index,
"target_year": target_year,
"status": "blocked",
"reason": "solar_longitude_root_failed",
}
try:
from solar_return import _localize_return_datetime
from jyotish_engine import compute_chart_data
except ImportError:
from scripts.solar_return import _localize_return_datetime
from scripts.jyotish_engine import compute_chart_data
utc = datetime(2000, 1, 1) + timedelta(days=jd - 2451544.5)
local = _localize_return_datetime(utc, tz_hours, timezone_id)
chart, _asc, _chart_jd, _ayan = compute_chart_data(
utc.year, utc.month, utc.day, utc.hour, utc.minute,
float(latitude), float(longitude), 0,
node_mode, second=utc.second, ayanamsa_name=ayanamsa,
)
if not isinstance(chart, dict):
raise RuntimeError(f"compute_chart_data returned no chart for monthly jd {jd}")
planets = chart.get("planets") if isinstance(chart.get("planets"), dict) else {}
sun = planets.get("Sun") if isinstance(planets.get("Sun"), dict) else {}
ascendant = chart.get("ascendant") if isinstance(chart.get("ascendant"), dict) else {}
return {
"month_index": month_index,
"target_year": target_year,
"jd_ut": jd,
"local_datetime": local.strftime("%Y-%m-%d %H:%M:%S"),
"utc_datetime": utc.strftime("%Y-%m-%d %H:%M:%S"),
"target_sun_longitude": target_sun,
"sun_longitude": sun.get("degree_raw"),
"ascendant": {
"sign": ascendant.get("sign"),
"degree_in_sign": ascendant.get("degree_in_sign", ascendant.get("degree")),
"longitude": ascendant.get("degree_raw", ascendant.get("lon")),
},
"planet_longitudes": {
name: planets[name].get("degree_raw")
for name in _PLANETS
if isinstance(planets.get(name), dict) and planets[name].get("degree_raw") is not None
},
"status": "partial_verified",
}
+96
View File
@@ -0,0 +1,96 @@
"""Native Patyayini from an annual Lagna and the seven classical planets.
Durations follow the degree-within-sign arcs. The longitudes passed in are
not recomputed here.
"""
from __future__ import annotations
from datetime import datetime, timedelta
from typing import Any
SIGNS = (
"Aries", "Taurus", "Gemini", "Cancer", "Leo", "Virgo",
"Libra", "Scorpio", "Sagittarius", "Capricorn", "Aquarius", "Pisces",
)
_TIE = ("Jupiter", "Moon", "Mercury", "Saturn", "Lagna", "Sun", "Mars", "Venus")
def build_native_patyayini(
*,
ascendant_longitude: float,
planet_longitudes: dict[str, float],
start: datetime,
year_days: float = 365.25,
) -> dict[str, Any]:
points = {"Lagna": float(ascendant_longitude) % 360.0}
missing = []
for lord in ("Sun", "Moon", "Mars", "Mercury", "Jupiter", "Venus", "Saturn"):
value = planet_longitudes.get(lord)
if isinstance(value, (int, float)) and not isinstance(value, bool):
points[lord] = float(value) % 360.0
else:
missing.append(lord)
if missing:
raise ValueError("patyayini missing longitudes: " + ", ".join(missing))
ordered = sorted(
(
{
"lord": lord,
"longitude": longitude,
"sign": SIGNS[int(longitude // 30) % 12],
"degree_in_sign": longitude % 30.0,
}
for lord, longitude in points.items()
),
key=lambda item: (item["degree_in_sign"], _TIE.index(item["lord"])),
)
denominator = ordered[-1]["degree_in_sign"]
if denominator <= 0.0:
raise ValueError("patyayini denominator is zero")
arcs = []
previous = 0.0
for item in ordered:
arc = max(0.0, float(item["degree_in_sign"]) - previous)
arcs.append(arc)
previous = float(item["degree_in_sign"])
periods = []
antardasha = []
cursor = start
for index, (item, arc) in enumerate(zip(ordered, arcs), start=1):
end = cursor + timedelta(days=float(year_days) * arc / denominator)
periods.append({
**item,
"order": index,
"start": _format(cursor),
"end": _format(end),
})
sub_cursor = cursor
cycle = ordered[index - 1:] + ordered[:index - 1]
cycle_arcs = arcs[index - 1:] + arcs[:index - 1]
main_seconds = (end - cursor).total_seconds()
for sub_index, (sub_item, sub_arc) in enumerate(zip(cycle, cycle_arcs), start=1):
sub_end = sub_cursor + timedelta(seconds=main_seconds * sub_arc / denominator)
antardasha.append({
"main": item["lord"],
"sub": sub_item["lord"],
"main_order": index,
"sub_order": sub_index,
"start": _format(sub_cursor),
"end": _format(sub_end),
"boundary": _format(sub_end),
"boundary_status": "dasha_ending_dates",
})
sub_cursor = sub_end
cursor = end
return {
"schema": "jyotish.patyayini_patyamsha_dasha.v1",
"status": "parameter_sensitive",
"model": "degree_within_sign",
"periods": periods,
"antardasha_periods": antardasha,
}
def _format(value: datetime) -> str:
return value.strftime("%Y-%m-%d %H:%M:%S")
+30
View File
@@ -0,0 +1,30 @@
"""Sun and Moon on the same sidereal frame as the chart function.
The flags match ``_calc_sidereal_planets_for_jd``: legacy true-precession
profile, then subtract ayanamsa. This is not the apparent-position path.
"""
from __future__ import annotations
def sun_moon_longitudes(jd: float, *, node_mode: str = "mean", ayanamsa_name: str | None = None) -> tuple[float, float]:
try:
from jyotish_engine import _apply_ayanamsa, _sidereal_calculation_profile
import swisseph as swe
except ImportError:
from scripts.jyotish_engine import _apply_ayanamsa, _sidereal_calculation_profile
import swisseph as swe
if ayanamsa_name:
_apply_ayanamsa(ayanamsa_name)
_mode, calc_flags, ayanamsa = _sidereal_calculation_profile(jd, "legacy")
flags = calc_flags | getattr(swe, "FLG_SPEED", 256)
sun = _body(swe, jd, swe.SUN, flags, ayanamsa)
moon = _body(swe, jd, swe.MOON, flags, ayanamsa)
return sun, moon
def _body(swe, jd: float, body: int, flags: int, ayanamsa: float) -> float:
pos, ret = swe.calc_ut(jd, body, flags)
if ret < 0:
raise RuntimeError(f"swisseph calc_ut failed for body {body} at jd {jd}: code {ret}")
return (float(pos[0]) - float(ayanamsa)) % 360.0
+172
View File
@@ -0,0 +1,172 @@
"""Tithi Pravesha instants from our sidereal Sun and Moon.
A missing body raises. Import errors are not stored as status=blocked.
"""
from __future__ import annotations
from datetime import datetime, timedelta
from typing import Any
try:
from annual_native.sidereal_bodies import sun_moon_longitudes
except ImportError: # pragma: no cover
from scripts.annual_native.sidereal_bodies import sun_moon_longitudes
SCHEMA = "jyotish.tithi_pravesh.v1"
_NAMES = (
"Pratipada", "Dvitiya", "Tritiya", "Chaturthi", "Panchami",
"Shashthi", "Saptami", "Ashtami", "Navami", "Dashami",
"Ekadashi", "Dvadashi", "Trayodashi", "Chaturdashi", "Purnima/Amavasya",
)
def build_tithi_pravesh(payload: dict[str, Any]) -> dict[str, Any]:
birth = dict(payload["birth"])
settings = dict(payload.get("settings") or {})
target_year = int(payload["target_year"])
year, month, day = [int(value) for value in str(birth["date"]).split("-")]
time_parts = [int(value) for value in str(birth.get("time") or "00:00:00").split(":")]
while len(time_parts) < 3:
time_parts.append(0)
tz = _offset_hours(birth.get("utc_offset"))
ayanamsa = str(settings.get("ayanamsa") or "lahiri")
node_mode = str(settings.get("node_mode") or "mean")
birth_jd = _jd_local(year, month, day, time_parts[0] + time_parts[1] / 60 + time_parts[2] / 3600, tz)
birth_sun, birth_moon = sun_moon_longitudes(birth_jd, node_mode=node_mode, ayanamsa_name=ayanamsa)
target_phase = (birth_moon - birth_sun) % 360
target_solar_sign = int(birth_sun // 30)
candidate_jds = _find_candidates(
target_year, month, day, tz, target_phase, target_solar_sign, ayanamsa, node_mode,
)
birth_place = _place(birth, tz)
birth_rows = [
_chart_row(jd, birth_place, node_mode, target_phase, ayanamsa) for jd in candidate_jds
]
return {
"schema": SCHEMA,
"status": "partial_verified" if birth_rows else "blocked",
"reason": None if birth_rows else "no_tithi_pravesh_candidate",
"target_year": target_year,
"natal_phase_degrees": target_phase,
"natal_sun_longitude": birth_sun,
"natal_moon_longitude": birth_moon,
"rows": birth_rows,
}
def _find_candidates(target_year, birth_month, birth_day, tz, target_phase, target_solar_sign, ayanamsa, node_mode) -> list[float]:
try:
anchor = datetime(target_year, birth_month, birth_day)
except ValueError:
anchor = datetime(target_year, birth_month, 1)
start_jd = _datetime_to_jd(anchor - timedelta(days=42), tz)
end_jd = _datetime_to_jd(anchor + timedelta(days=82), tz)
roots: list[float] = []
previous_jd = start_jd
previous_value = _phase_delta(previous_jd, target_phase, ayanamsa, node_mode)
while previous_jd < end_jd:
current_jd = min(previous_jd + 0.125, end_jd)
current_value = _phase_delta(current_jd, target_phase, ayanamsa, node_mode)
crossed = previous_value * current_value <= 0 and abs(previous_value) < 170 and abs(current_value) < 170
if crossed:
root = _bisect(previous_jd, current_jd, target_phase, ayanamsa, node_mode)
if root is not None:
sun, _moon = sun_moon_longitudes(root, node_mode=node_mode, ayanamsa_name=ayanamsa)
if int(sun // 30) == target_solar_sign and all(abs(root - existing) > 0.02 for existing in roots):
roots.append(root)
previous_jd, previous_value = current_jd, current_value
return roots
def _bisect(low, high, target_phase, ayanamsa, node_mode) -> float | None:
low_value = _phase_delta(low, target_phase, ayanamsa, node_mode)
for _ in range(40):
mid = (low + high) / 2
mid_value = _phase_delta(mid, target_phase, ayanamsa, node_mode)
if abs(mid_value) < 1e-7:
return mid
if low_value * mid_value <= 0:
high = mid
else:
low, low_value = mid, mid_value
root = (low + high) / 2
if abs(_phase_delta(root, target_phase, ayanamsa, node_mode)) >= 1e-4:
return None
return root
def _phase_delta(jd, target_phase, ayanamsa, node_mode) -> float:
sun, moon = sun_moon_longitudes(jd, node_mode=node_mode, ayanamsa_name=ayanamsa)
return ((moon - sun - target_phase + 180.0) % 360.0) - 180.0
def _chart_row(jd, place, node_mode, target_phase, ayanamsa) -> dict[str, Any]:
try:
from jyotish_engine import compute_chart_data
except ImportError:
from scripts.jyotish_engine import compute_chart_data
local = datetime(2000, 1, 1) + timedelta(days=jd - 2451544.5, hours=float(place["tz"]))
chart, *_rest = compute_chart_data(
local.year, local.month, local.day, local.hour, local.minute,
float(place["latitude"]), float(place["longitude"]), float(place["tz"]),
node_mode, second=local.second, ayanamsa_name=ayanamsa,
)
if not isinstance(chart, dict):
raise RuntimeError(f"compute_chart_data returned no chart for tithi jd {jd}")
planets = chart.get("planets") if isinstance(chart.get("planets"), dict) else {}
sun = float(planets["Sun"]["degree_raw"]) % 360.0
moon = float(planets["Moon"]["degree_raw"]) % 360.0
phase = (moon - sun) % 360.0
return {
"jd_ut": jd,
"local_datetime": local.strftime("%Y-%m-%d %H:%M:%S"),
"tithi": _tithi_payload(sun, moon),
"phase_degrees": phase,
"phase_delta_degrees": ((phase - target_phase + 180.0) % 360.0) - 180.0,
"planet_longitudes": {
name: planets[name].get("degree_raw")
for name in ("Sun", "Moon")
if isinstance(planets.get(name), dict)
},
"status": "partial_verified",
}
def _tithi_payload(sun: float, moon: float) -> dict[str, Any]:
phase = (moon - sun) % 360.0
number = min(30, int(phase // 12) + 1)
paksha = "Shukla" if number <= 15 else "Krishna"
within = number if number <= 15 else number - 15
return {
"tithi_num": number,
"paksha": paksha,
"full_name": f"{paksha} {_NAMES[within - 1]}",
"moon_sun_diff": phase,
}
def _place(data: dict, fallback_tz: float) -> dict[str, Any]:
return {
"latitude": float(data["latitude"]),
"longitude": float(data["longitude"]),
"tz": _offset_hours(data.get("utc_offset")) if data.get("utc_offset") is not None else fallback_tz,
}
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)
def _jd_local(year: int, month: int, day: int, local_hours: float, tz: float) -> float:
import swisseph as swe
return float(swe.julday(year, month, day, local_hours - tz))
def _datetime_to_jd(value: datetime, tz: float) -> float:
return _jd_local(value.year, value.month, value.day, value.hour + value.minute / 60 + value.second / 3600, tz)
+21 -4
View File
@@ -1784,7 +1784,7 @@ def _attach_full_report_pack(packet: dict, args) -> dict:
}
annual = timing.get('annual_tajika_pack') if isinstance(timing.get('annual_tajika_pack'), dict) else {}
if not annual:
annual = _build_pl9_full_annual_section(args, profile)
annual = _build_pl9_full_annual_section(args, profile, include_overview=True)
dasha = timing.get('dasha_master_pack') if isinstance(timing.get('dasha_master_pack'), dict) else {}
if not dasha:
dasha = _build_pl9_full_dasha_section(packet, modules)
@@ -2274,6 +2274,7 @@ def _attach_annual_tajika_pack(packet: dict, args) -> dict:
_build_pl9_full_annual_section(
args,
packet.get('calculation_profile') or {},
include_overview=True,
),
args=args,
)
@@ -2299,6 +2300,7 @@ def _attach_annual_tajika_pack(packet: dict, args) -> dict:
_build_pl9_full_annual_section(
annual_args,
packet.get('calculation_profile') or {},
include_overview=False,
)
)
except Exception as exc:
@@ -3063,7 +3065,10 @@ def _annual_series_surface(annual_pack: dict) -> dict:
keep_keys = (
'schema', 'status', 'reason', 'profile', 'audit', 'report_sections',
'solar_return', 'annual_chart', 'year_lord', 'external_engine_comparison',
'patyayini_dasha', 'sahams', 'tajika_yogas', 'muntha',
'patyayini_dasha', 'sahams', 'tajika_yogas', 'muntha', 'mudda_dasha',
'monthly_chart_pack', 'tithi_pravesh_years', 'native_patyayini',
'pl9_annual_dasha_results', 'annual_houses', 'tripataki_chakra',
'annual_year_pages', 'normalized_tables',
)
surface = {
key: _json_safe_report_snapshot(annual_pack.get(key))
@@ -3262,7 +3267,7 @@ def _attach_profile_id_to_dasha_section(dasha: dict, profile: dict) -> dict:
return dasha
def _build_pl9_full_annual_section(args, profile: dict | None = None) -> dict:
def _build_pl9_full_annual_section(args, profile: dict | None = None, *, include_overview: bool = False) -> dict:
target_year = getattr(args, 'target_year', None)
if target_year is None and getattr(args, 'today', None):
try:
@@ -3301,9 +3306,21 @@ def _build_pl9_full_annual_section(args, profile: dict | None = None) -> dict:
place=place,
age=age,
)
return build_annual_tajika_pack(payload, pyjhora_replay=replay)
pack = build_annual_tajika_pack(payload, pyjhora_replay=replay)
except Exception as exc:
return {'schema': 'jyotish.annual_tajika_pack.v1', 'status': 'blocked', 'reason': str(exc)}
try:
from annual_native.annual_expansion import attach_annual_expansion
except ImportError as exc:
logger.warning('annual expansion import failed: %s', exc)
pack['annual_expansion_error'] = f'ImportError: {exc}'
return pack
try:
attach_annual_expansion(pack, payload, args, include_overview=include_overview)
except Exception as exc:
logger.warning('annual expansion failed: %s', exc)
pack['annual_expansion_error'] = f'{exc.__class__.__name__}: {exc}'
return pack
def _annual_payload_from_args(args, profile: dict | None = None) -> dict:
+103 -21
View File
@@ -843,6 +843,7 @@ def render_pl9_parity_markdown(packet: dict) -> str:
'match': '一致',
'mismatch': '需校对',
'not_applicable': '不适用',
'dasha_ending_dates': '结束日',
'pyjhora_behavior_only': '外部行为参考',
'pyjhora_behavior_only / not_multiengine_parity': '外部行为参考',
'blocked_external_callable': '外部回放待校对',
@@ -856,6 +857,7 @@ def render_pl9_parity_markdown(packet: dict) -> str:
'match': 'matches',
'mismatch': 'needs review',
'not_applicable': 'not applicable',
'dasha_ending_dates': 'ending date',
'pyjhora_behavior_only': 'external behavior note',
'pyjhora_behavior_only / not_multiengine_parity': 'external behavior note',
'blocked_external_callable': 'external replay still to review',
@@ -2411,20 +2413,33 @@ def render_pl9_parity_markdown(packet: dict) -> str:
if not isinstance(row, dict):
continue
boundary = f"{row.get('start')} - {row.get('end')}" if row.get('start') and row.get('end') else row.get('start') or '-'
themes = row.get('themes')
if isinstance(themes, dict):
themes = themes.get(_lang) or themes.get('en') or '[translation missing: themes]'
else:
themes = _brief(themes, 2)
lines.append(
f"| {_cell(row.get('order'))} | {_cell(_token(row.get('lord')))} | {_cell(row.get('annual_whole_sign_house'))} | "
f"{_cell(boundary)} | {_cell(_brief(row.get('themes'), 2))} |"
f"{_cell(boundary)} | {_cell(themes)} |"
)
lines.append('')
normalized = _dict(annual.get('normalized_tables'))
mudda_rows = _list(normalized.get('mudda_rows'))
if mudda_rows:
lines.extend([
'### p132-p135 Mudda / Patyayini 年度 Dasha 表',
'',
'| Balance profile | Lord | Start | End | Boundary |' if _lang == 'en' else '| balance profile | 主星 | 开始 | 结束 | 边界口径 |',
'|-----------------|------|-------|-----|----------|' if _lang == 'en' else '|-----------------|------|------|------|----------|',
])
if _lang == 'en':
lines.extend([
'### Mudda annual dasha',
'',
'| Balance profile | Lord | Start | End | Boundary |',
'|-----------------|------|-------|-----|----------|',
])
else:
lines.extend([
'### Mudda 年度大运',
'',
'| 余额口径 | 主星 | 开始 | 结束 | 边界口径 |',
'|----------|------|------|------|----------|',
])
for row in mudda_rows:
if not isinstance(row, dict):
continue
@@ -2449,12 +2464,20 @@ def render_pl9_parity_markdown(packet: dict) -> str:
lines.append('')
patyayini_rows = _list(normalized.get('patyayini_rows'))
if patyayini_rows:
lines.extend([
'### p134 Patyayini / Patyamsha ending-date 表',
'',
'| 主层 | 子层 | 边界 | 边界语义 |',
'|------|------|------|----------|',
])
if _lang == 'en':
lines.extend([
'### Patyayini ending dates',
'',
'| Main | Sub | Boundary | Boundary meaning |',
'|------|-----|----------|------------------|',
])
else:
lines.extend([
'### Patyayini 结束日',
'',
'| 主层 | 子层 | 边界 | 边界语义 |',
'|------|------|------|----------|',
])
for row in patyayini_rows:
if not isinstance(row, dict):
continue
@@ -2661,32 +2684,65 @@ def render_pl9_parity_markdown(packet: dict) -> str:
lines.extend([
'### Monthly returns within the year',
'',
'| Month | Local time | Annual ascendant | Ascendant degree | Planet longitudes (deg) |',
'|-------|------------|------------------|------------------|--------------------------|',
'| Year | Month | Local time | Annual ascendant | Ascendant degree | Planet longitudes (deg) |',
'|------|-------|------------|------------------|------------------|--------------------------|',
])
else:
lines.extend([
'### 年度月返照图盘数据',
'',
'| 月份 | 本地时刻 | 年盘上升 | 上升度数 | 行星黄经(度) |',
'|------|----------|----------|----------|-----------------|',
'| 年份 | 月份 | 本地时刻 | 年盘上升 | 上升度数 | 行星黄经(度) |',
'|------|------|----------|----------|----------|-----------------|',
])
joiner = ', ' if _lang == 'en' else '、'
for row in monthly_rows:
if not isinstance(row, dict):
continue
asc = _dict(row.get('ascendant'))
longitudes = _dict(row.get('planet_longitudes'))
longitude_text = ';'.join(
longitude_text = joiner.join(
f"{_token(planet)} {_degree(value)}"
for planet, value in longitudes.items()
if isinstance(value, (int, float))
)
lines.append(
f"| {_cell(row.get('month_index'))} | {_cell(row.get('local_datetime'))} | "
f"| {_cell(row.get('target_year'))} | {_cell(row.get('month_index'))} | {_cell(row.get('local_datetime'))} | "
f"{_cell(_token(asc.get('sign')))} | {_cell(_degree(asc.get('degree_in_sign') or asc.get('degree')))} | "
f"{_cell(longitude_text)} |"
)
lines.append('')
tithi_years = _list(annual.get('tithi_pravesh_years'))
tithi_rows = [
row for year in tithi_years if isinstance(year, dict)
for row in _list(year.get('rows')) if isinstance(row, dict)
]
if tithi_rows:
if _lang == 'en':
lines.extend([
'### Tithi Pravesha',
'',
'| Year | Local time | Tithi | Phase (deg) |',
'|------|------------|-------|-------------|',
])
else:
lines.extend([
'### Tithi Pravesha',
'',
'| 年份 | 本地时刻 | Tithi | 月相角(度) |',
'|------|----------|-------|-------------|',
])
for year in tithi_years:
if not isinstance(year, dict):
continue
for row in _list(year.get('rows')):
if not isinstance(row, dict):
continue
tithi = _dict(row.get('tithi'))
lines.append(
f"| {_cell(year.get('target_year'))} | {_cell(row.get('local_datetime'))} | "
f"{_cell(tithi.get('full_name'))} | {_cell(_degree(row.get('phase_degrees')))} |"
)
lines.append('')
lifetime_pack = _dict(annual.get('lifetime_annual_chart_pages'))
lifetime_pages = _list(lifetime_pack.get('pages'))
lifetime_rows = [
@@ -2708,10 +2764,11 @@ def render_pl9_parity_markdown(packet: dict) -> str:
'| 年龄 | 年份 | 年返时刻 | 上升 | 上升度数 | 行星黄经(度) |',
'|------|------|----------|------|----------|-----------------|',
])
lifetime_joiner = ', ' if _lang == 'en' else '、'
for row in lifetime_rows:
asc = _dict(row.get('ascendant'))
longitudes = _dict(row.get('planet_longitudes'))
longitude_text = ';'.join(
longitude_text = lifetime_joiner.join(
f"{_token(planet)} {_degree(value)}"
for planet, value in longitudes.items()
if isinstance(value, (int, float))
@@ -2898,9 +2955,14 @@ def render_pl9_parity_markdown(packet: dict) -> str:
if not isinstance(row, dict):
continue
boundary = f"{row.get('start')} - {row.get('end')}" if row.get('start') and row.get('end') else row.get('start') or '-'
themes = row.get('themes')
if isinstance(themes, dict):
themes = themes.get(_lang) or themes.get('en') or '[translation missing: themes]'
else:
themes = _brief(themes, 2)
lines.append(
f"| {_cell(row.get('order'))} | {_cell(_token(row.get('lord')))} | {_cell(row.get('annual_whole_sign_house'))} | "
f"{_cell(boundary)} | {_cell(_brief(row.get('themes'), 2))} |"
f"{_cell(boundary)} | {_cell(themes)} |"
)
lines.append('')
normalized = _dict(annual_pack.get('normalized_tables'))
@@ -3002,6 +3064,26 @@ def render_pl9_parity_markdown(packet: dict) -> str:
f"{_cell(_token(asc.get('sign')))} | {_cell(_degree(asc.get('degree_in_sign') or asc.get('degree')))} |"
)
lines.append('')
tithi_years = _list(annual_pack.get('tithi_pravesh_years'))
if any(isinstance(year_pack, dict) and _list(year_pack.get('rows')) for year_pack in tithi_years):
lines.extend([
'#### Tithi Pravesha' if _lang == 'en' else '#### Tithi Pravesha',
'',
'| Local time | Tithi | Phase (deg) |' if _lang == 'en' else '| 本地时刻 | Tithi | 月相角(度) |',
'|------------|-------|-------------|' if _lang == 'en' else '|----------|-------|-------------|',
])
for year_pack in tithi_years:
if not isinstance(year_pack, dict):
continue
for row in _list(year_pack.get('rows')):
if not isinstance(row, dict):
continue
tithi = _dict(row.get('tithi'))
lines.append(
f"| {_cell(row.get('local_datetime'))} | {_cell(tithi.get('full_name'))} | "
f"{_cell(_degree(row.get('phase_degrees')))} |"
)
lines.append('')
d1_sheet = worksheets.get('d1_rasi_bhava') if isinstance(worksheets.get('d1_rasi_bhava'), dict) else {}
d1_houses = _dict(d1_sheet.get('houses')) or _dict(core_chart.get('houses'))
d1_planets = _dict(d1_sheet.get('planets')) or _dict(core_chart.get('planets'))