Jyotisha/scripts/dwadashottari_dasha.py

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

from __future__ import annotations

from datetime import datetime, timedelta
from typing import Any


YEAR_DAYS = 365.25636
NAKSHATRA_SPAN = 360.0 / 27.0
REVATI_INDEX = 26
TOTAL_CYCLE = 112
VENUS_SIGNS = {1, 6}  # Taurus, Libra, zero-based.
DASHA_SEQUENCE = (
    ("Sun", 7),
    ("Jupiter", 9),
    ("Ketu", 11),
    ("Mercury", 13),
    ("Rahu", 15),
    ("Mars", 17),
    ("Saturn", 19),
    ("Moon", 21),
)
_YEARS = dict(DASHA_SEQUENCE)


def _nakshatra_index(moon_longitude: float) -> int:
    return int(float(moon_longitude) % 360.0 // NAKSHATRA_SPAN)


def dwadashottari_applicability(birth_info: dict[str, Any]) -> dict[str, Any]:
    """Resolve Sukramsaka through an explicit caller-selected profile."""
    profile = str(birth_info.get("sukramsaka_profile") or "")
    if not profile:
        return {
            "applicable": False,
            "execution_status": "blocked",
            "rule_family": "bphs_dwadashottari_sukramsaka_profile_required",
            "reason": "Dwadashottari requires an explicit sukramsaka_profile because sources disagree on the exact applicability meaning.",
        }
    if profile == "navamsa_lagna_taurus_or_libra":
        value = birth_info.get("navamsa_ascendant_sign_index")
        if value is None:
            return {
                "applicable": False,
                "execution_status": "blocked",
                "rule_family": "bphs_dwadashottari_navamsa_lagna_venus_signs",
                "reason": "Missing required input for navamsa_lagna_taurus_or_libra: navamsa_ascendant_sign_index",
            }
        sign = int(value) % 12
        return {
            "applicable": sign in VENUS_SIGNS,
            "sukramsaka_profile": profile,
            "navamsa_ascendant_sign_index": sign,
            "rule_family": "bphs_dwadashottari_navamsa_lagna_venus_signs",
            "reason": (
                "Navamsa Lagna is Taurus or Libra."
                if sign in VENUS_SIGNS
                else "This profile requires Navamsa Lagna in Taurus or Libra."
            ),
        }
    if profile == "lagna_navamsa_matches_venus_rasi":
        required = ("navamsa_ascendant_sign_index", "venus_rasi_sign_index")
        missing = [field for field in required if birth_info.get(field) is None]
        if missing:
            return {
                "applicable": False,
                "execution_status": "blocked",
                "rule_family": "bphs_dwadashottari_lagna_navamsa_matches_venus_rasi",
                "reason": "Missing required input for lagna_navamsa_matches_venus_rasi: " + ", ".join(missing),
            }
        lagna_navamsa = int(birth_info["navamsa_ascendant_sign_index"]) % 12
        venus_rasi = int(birth_info["venus_rasi_sign_index"]) % 12
        return {
            "applicable": lagna_navamsa == venus_rasi,
            "sukramsaka_profile": profile,
            "navamsa_ascendant_sign_index": lagna_navamsa,
            "venus_rasi_sign_index": venus_rasi,
            "rule_family": "bphs_dwadashottari_lagna_navamsa_matches_venus_rasi",
            "reason": (
                "Navamsa Lagna matches Venus's Rasi sign."
                if lagna_navamsa == venus_rasi
                else "This profile requires Navamsa Lagna to match Venus's Rasi sign."
            ),
        }
    if profile == "lagna_rasi_matches_venus_navamsa":
        required = ("ascendant_sign_index", "venus_navamsa_sign_index")
        missing = [field for field in required if birth_info.get(field) is None]
        if missing:
            return {
                "applicable": False,
                "execution_status": "blocked",
                "rule_family": "bphs_dwadashottari_lagna_rasi_matches_venus_navamsa",
                "reason": "Missing required input for lagna_rasi_matches_venus_navamsa: " + ", ".join(missing),
            }
        lagna_rasi = int(birth_info["ascendant_sign_index"]) % 12
        venus_navamsa = int(birth_info["venus_navamsa_sign_index"]) % 12
        return {
            "applicable": lagna_rasi == venus_navamsa,
            "sukramsaka_profile": profile,
            "ascendant_sign_index": lagna_rasi,
            "venus_navamsa_sign_index": venus_navamsa,
            "rule_family": "bphs_dwadashottari_lagna_rasi_matches_venus_navamsa",
            "reason": (
                "Lagna Rasi matches Venus's Navamsa sign."
                if lagna_rasi == venus_navamsa
                else "This profile requires Lagna Rasi to match Venus's Navamsa sign."
            ),
        }
    return {
        "applicable": False,
        "execution_status": "blocked",
        "rule_family": "bphs_dwadashottari_sukramsaka_profile_unknown",
        "reason": "Unknown sukramsaka_profile: " + profile,
    }


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, int]:
    moon = float(moon_longitude) % 360.0
    nakshatra = _nakshatra_index(moon)
    count_to_revati = (REVATI_INDEX - nakshatra) % 27 + 1
    remainder = count_to_revati % len(DASHA_SEQUENCE)
    sequence_index = (remainder - 1) if remainder else len(DASHA_SEQUENCE) - 1
    lord = DASHA_SEQUENCE[sequence_index][0]
    balance = _YEARS[lord] * (1.0 - (moon % NAKSHATRA_SPAN) / NAKSHATRA_SPAN)
    return lord, balance, nakshatra, count_to_revati


def calculate_dwadashottari_dasha(birth_info: dict[str, Any]) -> dict[str, Any]:
    """Return birth-forward MD rows for an explicit Sukramsaka profile."""
    required = ("birth_datetime", "moon_longitude", "sukramsaka_profile")
    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 = dwadashottari_applicability(birth_info)
    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 or disputed classical applicability gate in the audit fields."
                ),
            }
        else:
            return {
                **applicability,
                "execution_status": applicability.get("execution_status", "not_applicable"),
                "confidence_status": "blocked" if applicability.get("execution_status") == "blocked" else "not_applicable",
                "total_cycle": TOTAL_CYCLE,
                "major": [],
            }
    year_days = float(birth_info.get("dasha_year_days") or YEAR_DAYS)
    lord, balance, nakshatra, count_to_revati = _birth_lord_and_balance(float(birth_info["moon_longitude"]))
    rows = []
    cursor = birth
    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,
        "count_to_revati_inclusive": count_to_revati,
        "starting_lord": lord,
        "dasha_balance_at_birth_years": balance,
        "major": rows,
        "reason": "Local source-bounded MD calculation for an explicit Sukramsaka profile; AD/PD and profile arbitration remain gated.",
        "source_locator": "Dwadashottari source passages: Sukramsaka condition variants, count birth nakshatra to Revati, eight-lord order, and 7-21 year allotments.",
    }