Jyotisha/scripts/main_yoga_detectors.py

#!/usr/bin/env python3
"""One main detector per classical yoga (astrologer rulings 2026-10-03).

The same yoga used to be judged by several implementations (local code,
`references/yoga_rules.json`, upstream ports), merged by name, so whichever
arrived first won. Each yoga below now has exactly one detector; the data-driven
rules of the same name are `reference_only` in `yoga_rules.json` and the rule
engine (`yoga_engine.YogaEngine.detect`) appends the rows built here instead, so
the reader report, the consultation sieve and `chart.yogas` carry one verdict.

Rows are shaped like `yoga_engine` rows (Chinese fields plus `*_en` twins for the
English reader edition).

- Neecha Bhanga (BUG-1205, 乙14): `yogas_doshas.calc_nicha_bhanga_raj_yoga`;
  only the cancellation conditions that hold are listed, never called Raja Yoga.
- Kemadruma (BUG-1206, 甲7): `yoga_expansion.detect_kemadruma`; the 2nd / 12th
  from the Moon count only Mars, Mercury, Jupiter, Venus, Saturn, and there is no
  kendra cancellation (planets in kendras, including only Rahu / Ketu there, do
  not cancel it).

Inputs are D1 rows `{sign, house, lon?, retrograde?, navamsa_sign?}`. A missing
`navamsa_sign` is derived from the sidereal longitude (Parashari navamsa: each
3°20' from the movable / fixed / dual starting sign), so every caller that has
longitudes gets the same verdict.
"""

from __future__ import annotations

from typing import Any, Dict, List, Optional

try:  # scripts/ on sys.path
    from yoga_expansion import detect_kemadruma
    from yogas_doshas import calc_nicha_bhanga_raj_yoga
except ImportError:  # package import (scripts.main_yoga_detectors)
    from scripts.yoga_expansion import detect_kemadruma
    from scripts.yogas_doshas import calc_nicha_bhanga_raj_yoga

SIGNS = ['Aries', 'Taurus', 'Gemini', 'Cancer', 'Leo', 'Virgo',
         'Libra', 'Scorpio', 'Sagittarius', 'Capricorn', 'Aquarius', 'Pisces']
SIGN_ZH = {
    'Aries': '白羊座', 'Taurus': '金牛座', 'Gemini': '双子座', 'Cancer': '巨蟹座',
    'Leo': '狮子座', 'Virgo': '处女座', 'Libra': '天秤座', 'Scorpio': '天蝎座',
    'Sagittarius': '射手座', 'Capricorn': '摩羯座', 'Aquarius': '水瓶座', 'Pisces': '双鱼座',
}
PLANET_ZH = {
    'Sun': '太阳', 'Moon': '月亮', 'Mars': '火星', 'Mercury': '水星', 'Jupiter': '木星',
    'Venus': '金星', 'Saturn': '土星', 'Rahu': '罗睺', 'Ketu': '计都',
}

NEECHA_BHANGA_NAME = 'Neecha Bhanga'
NEECHA_BHANGA_NAME_ZH = '落陷取消'
NEECHA_BHANGA_CATEGORY = 'neecha_bhanga'
NEECHA_BHANGA_NOTE_ZH = '只列成立的取消条件，不单独下结论'
NEECHA_BHANGA_NOTE_EN = 'Conditions only; no stand-alone conclusion'


def _lon(row: Any) -> Optional[float]:
    if not isinstance(row, dict):
        return None
    for key in ('lon', 'degree_raw', 'longitude'):
        value = row.get(key)
        if isinstance(value, (int, float)):
            return float(value) % 360.0
    return None


def navamsa_sign(lon: float) -> str:
    """D9 sign of a sidereal longitude (Parashari navamsa)."""
    return SIGNS[int((lon % 360.0) * 9.0 / 30.0) % 12]


def with_navamsa(planets: Dict[str, Any]) -> Dict[str, Dict[str, Any]]:
    """Copy of `planets` with `navamsa_sign` filled from the longitude when missing."""
    rows: Dict[str, Dict[str, Any]] = {}
    for name, row in (planets or {}).items():
        if not isinstance(row, dict):
            continue
        copy = dict(row)
        lon = _lon(row)
        if not copy.get('navamsa_sign') and lon is not None:
            copy['navamsa_sign'] = navamsa_sign(lon)
        rows[name] = copy
    return rows


def _planet_zh(name: Any) -> str:
    return PLANET_ZH.get(str(name), str(name))


def _sign_zh(name: Any) -> str:
    return SIGN_ZH.get(str(name), str(name))


def _neecha_condition_text(item: Dict[str, Any], debilitated: str) -> tuple[str, str]:
    code = item.get('code')
    planet = item.get('planet')
    house = item.get('house')
    frame_zh = '上升' if item.get('frame') == 'lagna' else '月亮'
    frame_en = 'the Lagna' if item.get('frame') == 'lagna' else 'the Moon'
    if code in ('A', 'B'):
        return (f'落陷星座主星{_planet_zh(planet)}在{frame_zh}起第{house}宫（角宫）',
                f'dispositor {planet} in house {house} from {frame_en} (kendra)')
    if code in ('C', 'D'):
        return (f'该星入旺星座的主星{_planet_zh(planet)}在{frame_zh}起第{house}宫（角宫）',
                f'lord of its exaltation sign, {planet}, in house {house} from {frame_en} (kendra)')
    if code == 'E':
        return (f'{_planet_zh(debilitated)}与落陷星座主星{_planet_zh(planet)}互换星座',
                f'{debilitated} and its dispositor {planet} exchange signs')
    if code == 'F':
        return (f'{_planet_zh(planet)}在 D9 入庙或入旺（{_sign_zh(item.get("sign"))}）',
                f'{planet} own or exalted in D9 ({item.get("sign")})')
    return (str(item), str(item))


def neecha_bhanga_rows(planets: Dict[str, Any]) -> List[Dict[str, Any]]:
    """Formed Neecha Bhanga rows; conditions from `calc_nicha_bhanga_raj_yoga`."""
    result = calc_nicha_bhanga_raj_yoga(with_navamsa(planets), {})
    rows: List[Dict[str, Any]] = []
    for item in result.get('yogas') or []:
        planet = item.get('planet')
        texts = [_neecha_condition_text(cond, planet) for cond in item.get('cancellation_conditions') or []]
        if not texts:
            continue
        zh = [text[0] for text in texts]
        en = [text[1] for text in texts]
        sign = item.get('debilitated_sign')
        rows.append({
            'name': NEECHA_BHANGA_NAME,
            'name_cn': NEECHA_BHANGA_NAME_ZH,
            'category': NEECHA_BHANGA_CATEGORY,
            'planets': [planet],
            'combination': f'{_planet_zh(planet)}落陷于{_sign_zh(sign)}；成立条件：' + '；'.join(zh),
            'combination_en': f'{planet} debilitated in {sign}; conditions met: ' + '; '.join(en),
            'conditions': zh,
            'conditions_en': en,
            'effects': [NEECHA_BHANGA_NOTE_ZH],
            'effects_en': [NEECHA_BHANGA_NOTE_EN],
            'strength': f'成立条件 {len(zh)} 条',
            'strength_en': f'{len(zh)} condition(s) met',
            'source': 'BPHS',
            'rule_id': 'main_detector:neecha_bhanga',
            'detector': 'yogas_doshas.calc_nicha_bhanga_raj_yoga',
        })
    return rows


def kemadruma_rows(planets: Dict[str, Any]) -> List[Dict[str, Any]]:
    """Kemadruma from `yoga_expansion.detect_kemadruma` (no kendra cancellation)."""
    result = detect_kemadruma(planets or {})
    if not result.get('present'):
        return []
    moon_sign = (planets.get('Moon') or {}).get('sign')
    return [{
        'name': 'Kemadruma Yoga',
        'name_cn': '空劫瑜伽',
        'category': 'durbhaga',
        'planets': ['Moon'],
        'combination': f'月亮在{_sign_zh(moon_sign)}，第2、12宫无火水木金土（太阳与罗计不计；不启用角宫解除）',
        'combination_en': (f'Moon in {moon_sign}; none of Mars to Saturn in the 2nd or 12th from it '
                           '(the Sun and the nodes are not counted; no kendra cancellation)'),
        'effects': ['人生艰辛', '需自力更生', '精神挑战'],
        'effects_en': ['Hard life', 'Must rely on oneself', 'Mental and spiritual challenges'],
        'strength': '弱',
        'strength_en': 'weak',
        'source': 'BPHS',
        'rule_id': 'main_detector:kemadruma',
        'detector': 'yoga_expansion.detect_kemadruma',
    }]


def main_detector_rows(planets: Dict[str, Any], ascendant: Optional[str] = None) -> List[Dict[str, Any]]:
    """Every main-detector row for one D1 chart, in a fixed order."""
    return kemadruma_rows(planets) + neecha_bhanga_rows(planets)