210 lines
7.9 KiB
Python
210 lines
7.9 KiB
Python
"""Attach sourced dasha children onto a master-pack copy.
|
|
|
|
The natal modules keep their own rows. This module never imports PyJHora.
|
|
Kalachakra children are the upstream proportional model, labeled simplified,
|
|
and they hang on the majors this repository already calculated.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from copy import deepcopy
|
|
from datetime import datetime, timedelta
|
|
from typing import Any
|
|
|
|
|
|
def _rotate_cycle_rows(cycle_rows: list[dict], start_rashi: str | None) -> list[dict]:
|
|
if not cycle_rows:
|
|
return []
|
|
if not start_rashi:
|
|
return list(cycle_rows)
|
|
start_index = next(
|
|
(index for index, row in enumerate(cycle_rows) if row.get("rashi") == start_rashi),
|
|
0,
|
|
)
|
|
return list(cycle_rows[start_index:]) + list(cycle_rows[:start_index])
|
|
|
|
|
|
def _build_subperiods(
|
|
parent_period: dict,
|
|
cycle_rows: list[dict],
|
|
total_cycle_years: float,
|
|
level_key: str,
|
|
year_days: float = 365.25,
|
|
) -> list[dict]:
|
|
"""Proportional Kalachakra children. Last child ends on the parent end."""
|
|
if not cycle_rows:
|
|
return []
|
|
rotated_rows = _rotate_cycle_rows(cycle_rows, parent_period.get("rashi"))
|
|
cycle_basis_total = sum(float(row.get("years") or 0.0) for row in rotated_rows)
|
|
denominator = cycle_basis_total if cycle_basis_total > 0 else float(total_cycle_years or 0.0)
|
|
if denominator <= 0:
|
|
return []
|
|
start_dt = datetime.fromisoformat(str(parent_period["start_date"]))
|
|
current_dt = start_dt
|
|
child_periods = []
|
|
for row in rotated_rows:
|
|
child_years = float(parent_period["years"]) * float(row["years"]) / denominator
|
|
end_dt = current_dt + timedelta(days=child_years * year_days)
|
|
child_periods.append({
|
|
"level": level_key,
|
|
"lord": row.get("lord"),
|
|
"rashi": row.get("rashi"),
|
|
"sign": row.get("rashi"),
|
|
"rashi_index": row.get("rashi_index"),
|
|
"years": child_years,
|
|
"start_date": current_dt.isoformat(),
|
|
"end_date": end_dt.isoformat(),
|
|
"status": "parameter_sensitive",
|
|
"derivation": "native_kalachakra_recursive_proportional_variant",
|
|
"model": "simplified",
|
|
})
|
|
current_dt = end_dt
|
|
if child_periods:
|
|
child_periods[-1]["end_date"] = parent_period["end_date"]
|
|
return child_periods
|
|
|
|
|
|
def attach_kalachakra_simplified_children(
|
|
major_periods: list[dict],
|
|
total_cycle_years: float | None = None,
|
|
year_days: float = 365.25,
|
|
) -> tuple[list[dict], list[dict], list[dict]]:
|
|
"""Return a copy of the existing majors with simplified AD and PD rows.
|
|
|
|
The caller's major rows are not modified. Replacing those majors is out of
|
|
scope: if a row has no start_date, this raises instead of inventing one.
|
|
"""
|
|
majors = deepcopy(major_periods)
|
|
cycle_rows = [
|
|
{
|
|
"lord": row.get("lord"),
|
|
"rashi": row.get("rashi"),
|
|
"rashi_index": row.get("rashi_index"),
|
|
"years": row.get("years"),
|
|
}
|
|
for row in majors
|
|
if isinstance(row, dict)
|
|
]
|
|
if total_cycle_years is None:
|
|
total_cycle_years = sum(float(row.get("years") or 0.0) for row in cycle_rows)
|
|
all_antardashas: list[dict] = []
|
|
all_pratyantardashas: list[dict] = []
|
|
for period in majors:
|
|
if not isinstance(period, dict):
|
|
continue
|
|
antardasha = _build_subperiods(period, cycle_rows, float(total_cycle_years), "antardasha", year_days)
|
|
for child in antardasha:
|
|
child["parent_lord"] = period.get("lord")
|
|
child["parent_rashi"] = period.get("rashi")
|
|
child["parent_sign"] = period.get("rashi")
|
|
period["antardasha"] = antardasha
|
|
all_antardashas.extend(antardasha)
|
|
for child in antardasha:
|
|
pratyantardasha = _build_subperiods(
|
|
child, cycle_rows, float(total_cycle_years), "pratyantardasha", year_days,
|
|
)
|
|
for grand in pratyantardasha:
|
|
grand["parent_lord"] = child.get("lord")
|
|
grand["parent_rashi"] = child.get("rashi")
|
|
grand["parent_sign"] = child.get("rashi")
|
|
grand["main"] = period.get("rashi") or period.get("lord")
|
|
grand["parent_antardasha"] = child.get("rashi") or child.get("lord")
|
|
child["pratyantardasha"] = pratyantardasha
|
|
all_pratyantardashas.extend(pratyantardasha)
|
|
return majors, all_antardashas, all_pratyantardashas
|
|
|
|
|
|
def _copy_rows(rows: list[dict]) -> list[dict]:
|
|
return deepcopy([row for row in rows if isinstance(row, dict)])
|
|
|
|
|
|
def _narayana_periods(module: dict) -> list[dict]:
|
|
sequence = module.get("mahadasha_sequence")
|
|
if not isinstance(sequence, list):
|
|
return []
|
|
children_by_parent: dict[str, list[dict]] = {}
|
|
boundaries = module.get("dense_boundaries")
|
|
if isinstance(boundaries, list):
|
|
for row in boundaries:
|
|
if not isinstance(row, dict) or row.get("level") != "AD" or not row.get("sign"):
|
|
continue
|
|
parent = str(row.get("parent_period_key") or row.get("parent_md") or "")
|
|
children_by_parent.setdefault(parent, []).append({
|
|
"level": "antardasha",
|
|
"sign": row.get("sign"),
|
|
"lord": row.get("lord"),
|
|
"years": row.get("duration_years"),
|
|
"start": row.get("start_age"),
|
|
"end": row.get("end_age"),
|
|
"parent_sign": row.get("parent_md"),
|
|
"parent_lord": row.get("parent_md"),
|
|
"source": "narayana_dense_boundaries",
|
|
})
|
|
periods = []
|
|
for row in sequence:
|
|
if not isinstance(row, dict):
|
|
continue
|
|
sign = row.get("sign")
|
|
parent_key = str(row.get("period_key") or sign or "")
|
|
periods.append({
|
|
"sign": sign,
|
|
"lord": row.get("lord"),
|
|
"years": row.get("years"),
|
|
"start": row.get("start_age"),
|
|
"end": row.get("end_age"),
|
|
"antardasha": children_by_parent.get(parent_key, []),
|
|
})
|
|
return periods
|
|
|
|
|
|
def _yogini_periods(module: dict, rows: list[dict]) -> list[dict]:
|
|
if any(isinstance(row.get("antardasha"), list) and row.get("antardasha") for row in rows):
|
|
return rows
|
|
try:
|
|
from yogini_dasha import yogini_antardashas
|
|
except ImportError:
|
|
from scripts.yogini_dasha import yogini_antardashas
|
|
year_days = module.get("dasha_year_days") or 365.25
|
|
birth = None
|
|
if rows:
|
|
birth = rows[0].get("start_date")
|
|
if not birth:
|
|
return rows
|
|
attached = []
|
|
for row in rows:
|
|
copied = dict(row)
|
|
try:
|
|
copied["antardasha"] = yogini_antardashas(row, birth_dt=birth, year_days=float(year_days))
|
|
except (KeyError, TypeError, ValueError) as exc:
|
|
raise ValueError(f"yogini antardasha failed for {row.get('yogini')}: {exc}") from exc
|
|
for child in copied["antardasha"]:
|
|
child["parent_lord"] = row.get("planet") or row.get("lord")
|
|
child["main"] = row.get("yogini")
|
|
attached.append(copied)
|
|
return attached
|
|
|
|
|
|
def enrich_master_family(key: str, module: dict | None, periods: list | None) -> list[dict]:
|
|
"""Copy family rows and attach children that already have a sourced rule.
|
|
|
|
Vimshottari is left untouched: its three layers are rendered from the
|
|
dasha timeline, not from this pack.
|
|
"""
|
|
module = module if isinstance(module, dict) else {}
|
|
rows = _copy_rows(periods or [])
|
|
if key == "vimshottari":
|
|
return rows
|
|
if key == "narayana":
|
|
return _narayana_periods(module)
|
|
if key == "yogini":
|
|
return _yogini_periods(module, rows)
|
|
if key == "kala_chakra":
|
|
if not rows:
|
|
return rows
|
|
enriched, _ad, _pd = attach_kalachakra_simplified_children(
|
|
rows,
|
|
module.get("total_cycle"),
|
|
)
|
|
return enriched
|
|
return rows
|