Jyotisha/scripts/chaturshitisama_dasha.py

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

Chaturshitisama is conditional: BPHS-style sources apply it when the 10th
lord occupies the 10th house.  This module requires the caller to provide the
10th-lord sign explicitly so it does not guess house/lord rules from a partial
chart context.
"""

from __future__ import annotations

from datetime import datetime, timedelta
from typing import Any

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


YEAR_DAYS = 365.25636
NAKSHATRA_SPAN = 360.0 / 27.0
SWATI_INDEX = 14
TOTAL_CYCLE = 84
DASHA_SEQUENCE = (
    ("Sun", 12),
    ("Moon", 12),
    ("Mars", 12),
    ("Mercury", 12),
    ("Jupiter", 12),
    ("Venus", 12),
    ("Saturn", 12),
)
_YEARS = dict(DASHA_SEQUENCE)


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


def _tenth_house_sign_index(ascendant_longitude: float) -> int:
    return (_sign_index(ascendant_longitude) + 9) % 12


def chaturshitisama_applicability(
    *, ascendant_longitude: float, tenth_lord_sign_index: int
) -> dict[str, Any]:
    """Apply the 10th-lord-in-10th Chaturshitisama condition."""
    ascendant_sign = _sign_index(ascendant_longitude)
    tenth_house_sign = _tenth_house_sign_index(ascendant_longitude)
    lord_sign = int(tenth_lord_sign_index) % 12
    applicable = lord_sign == tenth_house_sign
    return {
        "applicable": applicable,
        "ascendant_sign_index": ascendant_sign,
        "tenth_house_sign_index": tenth_house_sign,
        "tenth_lord_sign_index": lord_sign,
        "rule_family": "bphs_chaturshitisama_tenth_lord_in_tenth",
        "reason": (
            "10th lord is placed in the 10th house."
            if applicable
            else "Chaturshitisama requires the 10th lord to occupy the 10th house."
        ),
    }


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]:
    return nakshatra_seeded_birth_balance(
        moon_longitude, anchor_index=SWATI_INDEX, sequence=DASHA_SEQUENCE,
    )


def calculate_chaturshitisama_dasha(birth_info: dict[str, Any]) -> dict[str, Any]:
    """Return birth-forward MD rows; child-period boundaries stay blocked."""
    required = ("birth_datetime", "ascendant_longitude", "tenth_lord_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 = chaturshitisama_applicability(
        ascendant_longitude=float(birth_info["ascendant_longitude"]),
        tenth_lord_sign_index=int(birth_info["tenth_lord_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
    return {
        **applicability,
        "execution_status": "executed",
        "confidence_status": "parameter_sensitive",
        "verification_status": "source_bounded_local_candidate",
        "period_depth_status": "mahadasha_executed_ad_pd_blocked_pending_row_level_replay",
        "total_cycle": TOTAL_CYCLE,
        "dasha_year_days": year_days,
        "moon_nakshatra_index": nakshatra,
        "starting_lord": lord,
        "dasha_balance_at_birth_years": balance,
        "major": rows,
        "reason": "Local source-bounded MD calculation; AD/PD are intentionally blocked pending row-level parity and runtime wiring.",
        "source_locator": "BPHS conditional Dasha passages: 10th-lord-in-10th gate, Swati count, Sun-to-Saturn order, and 12-year allotments.",
    }