Jyotisha/scripts/trikona_raman.py

"""Raman's numerical Rasi strength and Trikona MD/AD, explicit conventions.

Studies in Jaimini Astrology sections62-72,79-86,92-94. This is Raman's
adaptation, not a universal Jaimini comparator or the Raghava Bhatta method.
"""
from copy import deepcopy
from datetime import datetime, timedelta, timezone
import json
import math

from scripts.jaimini import FRIENDSHIP
from scripts.narayana_dasha import SIGNS, SIGN_LORDS, _NARAYANA_DIGNITY, _rasi_aspects
from scripts.trikona_dasha import RAMAN_AD_PROFILE, RAMAN_SOURCE, build_trikona_raman_antardasa_report

PROFILE = 'trikona_raman_numeric_seed_md_ad_candidate_v1'
STRENGTH_PROFILE = 'raman_studies_62_72_numeric_v1'
DIGNITY_PROFILE = 'raman_manual_degree_mt_exaltation_first_v1'
CLASSICAL = ('Sun', 'Moon', 'Mars', 'Mercury', 'Jupiter', 'Venus', 'Saturn')
SAVYA = {0, 1, 2, 6, 7, 8}
AMSA = (60, 45, 30, 22.5, 15, 7.5, 3.75)
# Raman, A Manual of Hindu Astrology, section29 printed p12. These differ
# from the repository Shadbala table; never silently substitute that table.
MT = {'Sun': (4, 0, 20), 'Moon': (1, 4, 30), 'Mars': (0, 0, 18),
      'Mercury': (5, 16, 20), 'Jupiter': (8, 0, 13), 'Venus': (6, 0, 10),
      'Saturn': (10, 0, 20)}


def _validate_positions(positions):
    if not isinstance(positions, dict) or set(positions) != set(CLASSICAL) | {'Rahu', 'Ketu'}:
        raise ValueError('exactly seven classical planets plus Rahu and Ketu required')
    if any(type(v) not in (int, float) or not math.isfinite(v) or not 0 <= v < 360
           for v in positions.values()):
        raise ValueError('longitudes must be finite numbers in [0,360)')
    if not math.isclose((positions['Rahu'] - positions['Ketu']) % 360, 180, abs_tol=1e-7, rel_tol=0):
        raise ValueError('Rahu and Ketu must be opposite in the same position profile')


def _dignity(planet, longitude):
    sign, degree = int(longitude // 30), longitude % 30
    if sign in _NARAYANA_DIGNITY[planet]['exalted']:
        return 'exalted', 60
    if sign in _NARAYANA_DIGNITY[planet]['debilitated']:
        return 'debilitated', 3.75
    mt_sign, lo, hi = MT[planet]
    if sign == mt_sign and lo <= degree < hi:
        return 'moolatrikona', 45
    owner = SIGN_LORDS[SIGNS[sign]]
    if owner == planet:
        return 'own', 30
    if owner in FRIENDSHIP[planet]['friend']:
        return 'friend', 22.5
    if owner in FRIENDSHIP[planet]['enemy']:
        return 'enemy', 7.5
    return 'neutral', 15


def calculate_raman_rasi_strength(*, planet_longitudes, strength_profile,
                                  dignity_profile, druk_role_policy):
    if strength_profile != STRENGTH_PROFILE or dignity_profile != DIGNITY_PROFILE:
        raise ValueError('explicit Raman strength and dignity profiles required')
    if druk_role_policy not in ('distinct_planets', 'distinct_roles'):
        raise ValueError('druk_role_policy must be distinct_planets or distinct_roles')
    _validate_positions(planet_longitudes)
    signs = {p: int(v // 30) for p, v in planet_longitudes.items()}
    degrees = {p: planet_longitudes[p] % 30 for p in CLASSICAL}
    ranked = sorted(CLASSICAL, key=degrees.get, reverse=True)
    ties = [[p for p in CLASSICAL if degrees[p] == value] for value in set(degrees.values())]
    ties = sorted((group for group in ties if len(group) > 1), key=lambda group: degrees[group[0]], reverse=True)
    result = dict(profile=strength_profile, dignity_profile=dignity_profile,
        druk_role_policy=druk_role_policy, source=RAMAN_SOURCE, locator='sections62-72',
        status='blocked' if ties else 'executed', rank_ties=ties, planet_strengths={}, rasi_strengths=[])
    # Section65 does not specify equal-rank scores. Preserve own-sign strengths
    # below, but never fabricate a total/rank/Atmakaraka when a degree tie exists.
    ak = None if ties else ranked[0]
    result['atmakaraka'] = ak
    if not ties:
        for rank, planet in enumerate(ranked):
            label, dignity = _dignity(planet, planet_longitudes[planet])
            kendra = (60, 30, 15)[(signs[planet] - signs[ak]) % 3]
            result['planet_strengths'][planet] = dict(dignity_label=label,
                moolatrikona_bala=dignity, amsa_bala=AMSA[rank], kendra_bala=kendra,
                total=dignity + AMSA[rank] + kendra)
    for sign in range(12):
        lord = SIGN_LORDS[SIGNS[sign]]
        role_planets = ['Jupiter', 'Mercury', lord]
        if druk_role_policy == 'distinct_planets':
            role_planets = list(dict.fromkeys(role_planets))
        hits = [p for p in role_planets if signs[p] == sign or _rasi_aspects(signs[p], sign)]
        occupants = [p for p, position in signs.items() if position == sign]
        chara = (15, 30, 60)[sign % 3]
        sthira = 45 + 15 * len(occupants) if occupants else 0
        own = chara + 60 * len(hits) + sthira
        special = 60 if ak and signs[ak] == sign else (None if ties else 0)
        lord_strength = None if ties else result['planet_strengths'][lord]['total']
        result['rasi_strengths'].append(dict(sign_idx=sign, sign=SIGNS[sign], lord=lord,
            chara_bala=chara, druk_bala=60 * len(hits), druk_contributors=hits,
            sthira_bala=sthira, occupants=occupants, own_strength=own,
            lord_strength=lord_strength, special_strength=special,
            total=None if ties else own + lord_strength + special))
    if ties:
        result['reason'] = 'equal_degree_AMSA_rank_rule_not_established; provide finer positions or a sourced tie profile'
    return result


def _duration(sign, signs, *, node_dignity_profile):
    """Sections79-86, retaining the selected lord and decision trace."""
    direction = 1 if sign in SAVYA else -1
    classical = SIGN_LORDS[SIGNS[sign]]
    lords = [classical] + ({7: ['Ketu'], 10: ['Rahu']}.get(sign, []))
    base = {p: ((signs[p] - sign) * direction) % 12 or 12 for p in lords}
    adjustment = {p: (int(signs[p] in _NARAYANA_DIGNITY[p]['exalted']) -
                      int(signs[p] in _NARAYANA_DIGNITY[p]['debilitated']))
                  if p in CLASSICAL else 0 for p in lords}
    trace = ['section79_savya_count_minus_one']
    selected = lords[:]
    if len(lords) == 2:
        if signs[lords[0]] == signs[lords[1]]:
            trace.append('section82_same_own' if signs[lords[0]] == sign else 'sections83_84_conjunction')
            # Conjunction outside own sign explicitly refers to section79 only.
            years = base[lords[0]]
            return dict(status='executed', years=years, lords=selected, trace=trace,
                        count_direction=direction, candidates=base, node_dignity_profile=node_dignity_profile)
        away = [p for p in lords if signs[p] != sign]
        if len(away) == 1:
            selected = away
            trace.append('section85_prefer_lord_away_from_own')
        else:
            associates = {p: sum(s == signs[p] for name, s in signs.items() if name != p) for p in lords}
            selected = [p for p in selected if associates[p] == max(associates.values())]
            trace.append('section86_associate_count')
            if len(selected) > 1:
                modes = {p: signs[p] % 3 for p in selected}
                selected = [p for p in selected if modes[p] == max(modes.values())]
                trace.append('section86_dual_fixed_movable')
            if len(selected) > 1:
                selected = [p for p in selected if base[p] == max(base[q] for q in selected)]
                trace.append('section86_larger_span')
                if classical in selected and adjustment[classical] == -1:
                    return dict(status='blocked', reason='section86_over_debilitated_and_node_exaltation_exception_unresolved',
                                lords=selected, candidates=base, trace=trace)
    years_by_lord = {p: (12 if signs[p] == sign else base[p] + adjustment[p]) for p in selected}
    if len(set(years_by_lord.values())) != 1:
        return dict(status='blocked', reason='equal_colord_rank_with_different_adjusted_years',
                    lords=selected, candidates=years_by_lord, trace=trace)
    years = next(iter(years_by_lord.values()))
    return dict(status='executed' if years > 0 else 'blocked', years=years,
        reason=None if years > 0 else 'nonpositive_adjusted_span_no_clipping_rule',
        lords=selected, trace=trace, candidates=base, dignity_adjustments=adjustment,
        count_direction=direction, node_dignity_profile=node_dignity_profile)


def build_trikona_raman_report(*, lagna_sign_idx, planet_longitudes, epoch, year_days,
        profile, strength_profile, dignity_profile, druk_role_policy, node_dignity_profile):
    if profile != PROFILE:
        raise ValueError('explicit Raman native profile required')
    if type(lagna_sign_idx) is not int or not 0 <= lagna_sign_idx < 12:
        raise ValueError('lagna_sign_idx must be integer 0..11')
    if node_dignity_profile != 'unassigned_no_adjustment':
        raise ValueError('node_dignity_profile must explicitly select unassigned_no_adjustment')
    if type(year_days) not in (int, float) or not math.isfinite(year_days) or not 300 <= year_days <= 400:
        raise ValueError('year_days must be finite in 300..400')
    if not isinstance(epoch, str):
        raise ValueError('aware ISO epoch required')
    anchor = datetime.fromisoformat(epoch.replace('Z', '+00:00'))
    if anchor.tzinfo is None or anchor.utcoffset() is None:
        raise ValueError('epoch must include UTC offset')
    anchor = anchor.astimezone(timezone.utc)
    strength = calculate_raman_rasi_strength(planet_longitudes=planet_longitudes,
        strength_profile=strength_profile, dignity_profile=dignity_profile, druk_role_policy=druk_role_policy)
    signs = {p: int(v // 30) for p, v in planet_longitudes.items()}
    trines = [(lagna_sign_idx + offset) % 12 for offset in (0, 4, 8)]
    candidates = [strength['rasi_strengths'][s] for s in trines]
    seeds = [] if strength['status'] == 'blocked' else [r['sign_idx'] for r in candidates
                        if r['total'] == max(c['total'] for c in candidates)]
    durations = [dict(sign_idx=s, sign=SIGNS[s], **_duration(s, signs,
                  node_dignity_profile=node_dignity_profile)) for s in range(12)]
    branches = []
    for seed in seeds:
        age = 0
        dated = True
        periods = []
        direction = 1 if seed in SAVYA else -1
        for offset in [group + trine for group in range(4) for trine in (0, 4, 8)]:
            sign = (seed + direction * offset) % 12
            row = deepcopy(durations[sign])
            if row['status'] != 'executed':
                dated = False
            if dated:
                try:
                    start = (anchor + timedelta(days=age * year_days)).isoformat()
                    end = (anchor + timedelta(days=(age + row['years']) * year_days)).isoformat()
                except OverflowError as exc:
                    raise ValueError('calendar boundary outside supported datetime range') from exc
                row.update(start_age=age, end_age=age + row['years'], start_utc=start, end_utc=end,
                           calendar_status='calculated_with_caller_fixed_year')
                child = build_trikona_raman_antardasa_report(md_sign_idx=sign,
                    full_md_years=row['years'], full_md_start=start,
                    parent_source='Native Raman sections79-86/92-93; see duration trace',
                    parent_profile='raman_studies_jaimini_92_93', year_days=year_days, profile=RAMAN_AD_PROFILE)
                row['antardasas'] = child['evidence_appendix'][0]['periods']
                # Parent and child formulas can round sub-microsecond floating
                # days differently; share the exact already-calculated edge.
                row['antardasas'][-1]['end_utc'] = end
                age += row['years']
            else:
                row.update(calendar_status='blocked_by_unresolved_prior_or_current_duration', antardasas=[])
            periods.append(row)
        branches.append(dict(seed_sign_idx=seed, direction=direction,
            status='parameter_sensitive' if dated else 'partial', periods=periods))
    result = dict(status='blocked' if not seeds else 'parameter_sensitive', variant_profile_id=profile,
        tradition_profile='bv_raman_studies_in_jaimini_astrology', source=RAMAN_SOURCE,
        strength=strength, seed_selection=dict(status='blocked' if not seeds else 'parallel_tie' if len(seeds)>1 else 'calculated',
            candidates=candidates, selected_signs=seeds), duration_evidence=durations, branches=branches,
        epoch_utc=anchor.isoformat(), year_days=year_days, node_dignity_profile=node_dignity_profile,
        pratyantardasa=dict(status='blocked', reason='Raman_section94_does_not_establish_PD'),
        limits=['Raman numerical adaptation, not universal Jaimini strength',
            'Druk role deduplication is explicit; source does not resolve duplicate roles',
            'Degree MT follows Raman Manual section29; exaltation checked first by explicit profile',
            'Node dignity left unassigned by caller convention, not a universal no-exaltation doctrine',
            'Full first MD starts at caller epoch; no fractional birth-balance inferred',
            'Fixed year_days not source-date verified; first cycle only; no customer admission'])
    lines = ['# Trikona — Raman numerical seed / MD / AD', '', f'Profile: {profile}',
             f'Source: {RAMAN_SOURCE}', '', '## Strength and seed',
             '| Trine sign | Strength |', '| --- | --- |']
    lines += [f"| {r['sign']} | {r['total']} |" for r in candidates]
    for branch in branches:
        lines += ['', f"## Seed {SIGNS[branch['seed_sign_idx']]}",
                  '| MD | Years | Start UTC | End UTC |', '| --- | --- | --- | --- |']
        lines += [f"| {r['sign']} | {r.get('years', 'blocked')} | {r.get('start_utc', 'blocked')} | {r.get('end_utc', 'blocked')} |"
                  for r in branch['periods']]
    lines += ['', *result['limits'], '', '```json', json.dumps(result, ensure_ascii=False, indent=2), '```']
    return dict(status=result['status'], customer_export_admitted=False, evidence_appendix=[result],
        exports=dict(markdown='\n'.join(lines), ai_evidence_bundle=deepcopy(result)))