#!/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.", }