#!/usr/bin/env python3 """Independent Panchottari Dasha producer bounded to the published BPHS rule. The legacy ``extended_dashas`` helper is deliberately not used: it is a generic placeholder and does not model Panchottari's seven lords, ascending period lengths, or Cancer/D12 applicability gate. """ 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 ANURADHA_INDEX = 16 TOTAL_CYCLE = 105 DASHA_SEQUENCE = ( ("Sun", 12), ("Mercury", 13), ("Saturn", 14), ("Mars", 15), ("Venus", 16), ("Moon", 17), ("Jupiter", 18), ) _YEARS_BY_LORD = dict(DASHA_SEQUENCE) def _sign_index(longitude: float) -> int: return int(float(longitude) % 360.0 // 30.0) def dwadasamsa_sign_index(longitude: float) -> int: """Return the standard D12 sign: twelve 2.5-degree parts from the sign.""" longitude = float(longitude) % 360.0 return (_sign_index(longitude) + int((longitude % 30.0) // 2.5)) % 12 def panchottari_applicability(*, ascendant_longitude: float) -> dict[str, Any]: """Apply BPHS's Cancer-rasi plus Cancer-D12 Panchottari condition.""" rasi_sign = _sign_index(ascendant_longitude) d12_sign = dwadasamsa_sign_index(ascendant_longitude) applicable = rasi_sign == 3 and d12_sign == 3 # Cancer is zero-based sign 3. return { "applicable": applicable, "ascendant_rasi_sign_index": rasi_sign, "ascendant_d12_sign_index": d12_sign, "rule_family": "bphs_panchottari_cancer_rasi_and_d12", "reason": ( "Ascendant is Cancer in both Rasi and Dwadasamsa." if applicable else "Panchottari requires a Cancer Ascendant that is also Cancer in Dwadasamsa." ), } 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 - ANURADHA_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_panchottari_dasha(birth_info: dict[str, Any]) -> dict[str, Any]: """Calculate birth-forward Panchottari MD; AD/PD stay explicitly unclosed.""" required = ("birth_datetime", "ascendant_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 = panchottari_applicability(ascendant_longitude=float(birth_info["ascendant_longitude"])) 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) 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: Cancer Rasi/D12 gate, Anuradha count, lord order, and 12-18 year allotments.", }