Jyotisha/scripts/satabdika_dasha.py

#!/usr/bin/env python3
"""Source-bounded Satabdika (100-year) Mahadasha producer.

Satabdika is conditional on a vargottama Ascendant.  The caller supplies the
D9 Ascendant sign explicitly so this module never guesses a divisional method.
"""

from __future__ import annotations

from datetime import datetime, timedelta
from typing import Any

try:
    from scripts.source_bounded_child_periods import expand_parent_seeded_equal_split_children
except ModuleNotFoundError:  # pragma: no cover - scripts/ on sys.path
    from source_bounded_child_periods import expand_parent_seeded_equal_split_children


YEAR_DAYS = 365.25636
NAKSHATRA_SPAN = 360.0 / 27.0
REVATI_INDEX = 26
TOTAL_CYCLE = 100
DASHA_SEQUENCE = (
    ("Sun", 5), ("Moon", 5), ("Venus", 10), ("Mercury", 10),
    ("Jupiter", 20), ("Mars", 20), ("Saturn", 30),
)
_YEARS = dict(DASHA_SEQUENCE)


def _sign_index(longitude: float) -> int:
    return int(float(longitude) % 360.0 // 30.0)


def satabdika_applicability(*, ascendant_longitude: float, navamsa_ascendant_sign_index: int) -> dict[str, Any]:
    rasi = _sign_index(ascendant_longitude)
    navamsa = int(navamsa_ascendant_sign_index) % 12
    applicable = rasi == navamsa
    return {
        "applicable": applicable,
        "ascendant_rasi_sign_index": rasi,
        "ascendant_navamsa_sign_index": navamsa,
        "rule_family": "bphs_satabdika_vargottama_lagna",
        "reason": "Ascendant is vargottama." if applicable else "Satabdika requires a vargottama Ascendant (same Rasi and Navamsa sign).",
    }


def _sequence_from(lord: str) -> list[tuple[str, int]]:
    start = next(index for index, (item, _years) in enumerate(DASHA_SEQUENCE) if item == lord)
    return list(DASHA_SEQUENCE[start:] + DASHA_SEQUENCE[:start])


def _birth_lord_and_balance(moon_longitude: float) -> tuple[str, float, int]:
    moon = float(moon_longitude) % 360.0
    nakshatra = int(moon // NAKSHATRA_SPAN)
    lord = DASHA_SEQUENCE[((nakshatra - REVATI_INDEX) % 27) % len(DASHA_SEQUENCE)][0]
    balance = _YEARS[lord] * (1.0 - (moon % NAKSHATRA_SPAN) / NAKSHATRA_SPAN)
    return lord, balance, nakshatra


def calculate_satabdika_dasha(birth_info: dict[str, Any]) -> dict[str, Any]:
    """Return birth-forward MD rows; child-period boundaries stay blocked."""
    required = ("birth_datetime", "ascendant_longitude", "navamsa_ascendant_sign_index", "moon_longitude")
    missing = [field for field in required if birth_info.get(field) is None]
    if missing:
        return {"applicable": False, "execution_status": "blocked", "confidence_status": "blocked", "reason": "Missing required input: " + ", ".join(missing), "major": []}
    birth = birth_info["birth_datetime"]
    if isinstance(birth, str):
        birth = datetime.fromisoformat(birth)
    applicability = satabdika_applicability(
        ascendant_longitude=float(birth_info["ascendant_longitude"]),
        navamsa_ascendant_sign_index=int(birth_info["navamsa_ascendant_sign_index"]),
    )
    if not applicability["applicable"]:
        if birth_info.get("standard_table_profile") == "pl9_reports_all_dasha_tables_v1":
            applicability = {
                **applicability,
                "classical_applicable": False,
                "applicable": True,
                "standard_table_profile": "pl9_reports_all_dasha_tables_v1",
                "applicability_reason": applicability["reason"],
                "reason": (
                    "PL9 standard-table profile requested: generate the report table while preserving "
                    "the unmet classical applicability gate in the audit fields."
                ),
            }
        else:
            return {**applicability, "execution_status": "not_applicable", "confidence_status": "not_applicable", "total_cycle": TOTAL_CYCLE, "major": []}
    year_days = float(birth_info.get("dasha_year_days") or YEAR_DAYS)
    lord, balance, nakshatra = _birth_lord_and_balance(float(birth_info["moon_longitude"]))
    cursor = birth
    rows = []
    for index, (row_lord, full_years) in enumerate(_sequence_from(lord)):
        years = balance if index == 0 else float(full_years)
        end = cursor + timedelta(days=years * year_days)
        rows.append({
            "level": "mahadasha", "lord": row_lord, "planet": row_lord,
            "years": years, "full_years": full_years, "is_birth_balance": index == 0,
            "start_date": cursor.isoformat(timespec="seconds"), "end_date": end.isoformat(timespec="seconds"),
        })
        cursor = end
    child_profile = birth_info.get("child_period_profile")
    rows, antardasha, pratyantardasha, period_depth_status = expand_parent_seeded_equal_split_children(
        rows,
        DASHA_SEQUENCE,
        profile=str(child_profile) if child_profile else None,
    )
    return {
        **applicability, "execution_status": "executed", "confidence_status": "parameter_sensitive",
        "verification_status": "source_bounded_local_candidate",
        "period_depth_status": period_depth_status,
        "total_cycle": TOTAL_CYCLE, "dasha_year_days": year_days, "moon_nakshatra_index": nakshatra,
        "starting_lord": lord, "dasha_balance_at_birth_years": balance, "child_period_profile": child_profile or None,
        "major": rows, "antardasha": antardasha, "pratyantardasha": pratyantardasha,
        "reason": (
            "Local source-bounded MD calculation with optional PyJHora-observed child-period profile."
            if child_profile else
            "Local source-bounded MD calculation; AD/PD are intentionally blocked pending row-level parity and runtime wiring."
        ),
        "source_locator": "BPHS conditional Dasha passages: vargottama condition, Revati count, lord order, and 5/5/10/10/20/20/30 year allotments.",
    }