Jyotisha/scripts/shodashottari_dasha.py

#!/usr/bin/env python3
"""Independent Shodashottari Dasha producer, bounded to a published BPHS rule.

Source basis: BPHS conditional-Dasha passages reproduced in the public
``bphs.pdf`` locator and the conditional-Dasha reference packet.  The local
formula is independently expressed here; PyJHora is used only as an external
observation in the separate raw-probe ledger.
"""

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
PUSHYA_INDEX = 7
TOTAL_CYCLE = 116
DASHA_SEQUENCE = (
    ("Sun", 11),
    ("Mars", 12),
    ("Jupiter", 13),
    ("Saturn", 14),
    ("Ketu", 15),
    ("Moon", 16),
    ("Mercury", 17),
    ("Venus", 18),
)
_YEARS_BY_LORD = dict(DASHA_SEQUENCE)


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


def _hora_lord(ascendant_longitude: float) -> str:
    """Return Parashari hora lord for the ascendant's 15-degree half-sign."""
    sign = _sign_index(ascendant_longitude)
    first_half = float(ascendant_longitude) % 30.0 < 15.0
    odd_sign = sign % 2 == 0  # Aries is index 0.
    if odd_sign:
        return "Sun" if first_half else "Moon"
    return "Moon" if first_half else "Sun"


def _paksha(sun_longitude: float, moon_longitude: float) -> str:
    return "shukla" if (float(moon_longitude) - float(sun_longitude)) % 360.0 < 180.0 else "krishna"


def shodashottari_applicability(*, ascendant_longitude: float, sun_longitude: float, moon_longitude: float) -> dict[str, Any]:
    """Apply the published Shodashottari hora-plus-paksha gate."""
    hora_lord = _hora_lord(ascendant_longitude)
    paksha = _paksha(sun_longitude, moon_longitude)
    applicable = (hora_lord == "Sun" and paksha == "shukla") or (hora_lord == "Moon" and paksha == "krishna")
    return {
        "applicable": applicable,
        "hora_lord": hora_lord,
        "paksha": paksha,
        "rule_family": "bphs_shodashottari_hora_paksha",
        "reason": (
            "Ascendant is in Sun hora during Shukla paksha."
            if hora_lord == "Sun" and paksha == "shukla"
            else "Ascendant is in Moon hora during Krishna paksha."
            if hora_lord == "Moon" and paksha == "krishna"
            else "The Shodashottari hora-plus-paksha applicability gate is not met."
        ),
    }


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


def _birth_lord_and_balance(moon_longitude: float) -> tuple[str, float, int]:
    moon_longitude = float(moon_longitude) % 360.0
    nakshatra_index = int(moon_longitude // NAKSHATRA_SPAN)
    offset = (nakshatra_index - PUSHYA_INDEX) % 27
    starting_lord = DASHA_SEQUENCE[offset % len(DASHA_SEQUENCE)][0]
    fraction_elapsed = (moon_longitude % NAKSHATRA_SPAN) / NAKSHATRA_SPAN
    remaining_years = _YEARS_BY_LORD[starting_lord] * (1.0 - fraction_elapsed)
    return starting_lord, remaining_years, nakshatra_index


def _iso(value: datetime) -> str:
    return value.isoformat(timespec="seconds")


def calculate_shodashottari_dasha(birth_info: dict[str, Any]) -> dict[str, Any]:
    """Calculate birth-forward Shodashottari MD; keep AD/PD explicitly unclosed."""
    required = ("birth_datetime", "ascendant_longitude", "sun_longitude", "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_dt = birth_info["birth_datetime"]
    if isinstance(birth_dt, str):
        birth_dt = datetime.fromisoformat(birth_dt)
    applicability = shodashottari_applicability(
        ascendant_longitude=float(birth_info["ascendant_longitude"]),
        sun_longitude=float(birth_info["sun_longitude"]),
        moon_longitude=float(birth_info["moon_longitude"]),
    )
    standard_table_profile = birth_info.get("standard_table_profile") == "pl9_reports_all_dasha_tables_v1"
    if not applicability["applicable"]:
        if standard_table_profile:
            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": [],
                "reason": applicability["reason"],
            }
    year_days = float(birth_info.get("dasha_year_days") or YEAR_DAYS)
    starting_lord, balance_years, moon_nakshatra_index = _birth_lord_and_balance(float(birth_info["moon_longitude"]))
    periods = []
    cursor = birth_dt
    for index, (lord, full_years) in enumerate(_sequence_from(starting_lord)):
        years = balance_years if index == 0 else float(full_years)
        end = cursor + timedelta(days=years * year_days)
        period = {
            "level": "mahadasha",
            "planet": lord,
            "lord": lord,
            "years": years,
            "full_years": full_years,
            "is_birth_balance": index == 0,
            "start_date": _iso(cursor),
            "end_date": _iso(end),
        }
        periods.append(period)
        cursor = end
    child_profile = birth_info.get("child_period_profile")
    periods, antardasha, pratyantardasha, period_depth_status = expand_parent_seeded_equal_split_children(
        periods,
        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": moon_nakshatra_index,
        "starting_lord": starting_lord,
        "dasha_balance_at_birth_years": balance_years,
        "child_period_profile": child_profile or None,
        "major": periods,
        "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: Shodashottari applicability, Pushya count, lord order, and 11-18 year allotments.",
    }