Jyotisha/scripts/ayanamsa_utils.py

#!/usr/bin/env python3
"""Shared Swiss Ephemeris ayanamsa helpers.

Swiss Ephemeris stores sidereal mode globally inside the process. Any helper
that calls calc_ut(..., FLG_SIDEREAL) must set the intended mode explicitly.

Unknown names fail; they never silently collapse to Lahiri. VedAstro / oracle
comparison fixtures must pass an explicit `lahiri` name.
"""

from __future__ import annotations

import os

from contextlib import contextmanager
from typing import Any, Iterator

try:
    import swisseph as swe
except ImportError:  # pragma: no cover
    swe = None


AYANAMSA_MODES = {
    'lahiri': getattr(swe, 'SIDM_LAHIRI', 1) if swe else 1,
    'raman': getattr(swe, 'SIDM_RAMAN', 3) if swe else 3,
    'kp': getattr(swe, 'SIDM_KRISHNAMURTI', 5) if swe else 5,
    'krishnamurti': getattr(swe, 'SIDM_KRISHNAMURTI', 5) if swe else 5,
    'fagan_bradley': getattr(swe, 'SIDM_FAGAN_BRADLEY', 0) if swe else 0,
    'djwhal_khul': getattr(swe, 'SIDM_DJWHAL_KHUL', 6) if swe else 6,
    'sassanian': getattr(swe, 'SIDM_SASSANIAN', 16) if swe else 16,
    'true_citra': getattr(swe, 'SIDM_TRUE_CITRA', 27) if swe else 27,
    'true_pushya': getattr(swe, 'SIDM_TRUE_PUSHYA', 29) if swe else 29,
}

AYANAMSA_DISPLAY_NAMES = {
    'lahiri': 'Lahiri',
    'raman': 'Raman',
    'kp': 'Krishnamurti/KP',
    'krishnamurti': 'Krishnamurti/KP',
    'fagan_bradley': 'Fagan-Bradley',
    'djwhal_khul': 'Djwhal Khul',
    'sassanian': 'Sassanian',
    'true_citra': 'True Citra',
    'true_pushya': 'True Pushya',
}

DEFAULT_AYANAMSA_NAME = 'raman'
ACTIVE_AYANAMSA_NAME = DEFAULT_AYANAMSA_NAME
# Flipped the first time any caller applies a mode. Import bootstraps read this
# so they do not replace a mode that was already selected. This module is
# imported once; jyotish_engine can be loaded a second time as its own module.
AYANAMSA_APPLIED_IN_PROCESS = False


class UnsupportedAyanamsaError(ValueError):
    """Raised when a caller asks for an ayanamsa that is not in AYANAMSA_MODES."""


def _canonical_ayanamsa_key(name: Any) -> str:
    key = str(name or '').strip().lower().replace('-', '_')
    if key in ('krishnamurti_paddhati', 'krishnamurti/kp'):
        key = 'kp'
    if key == 'true_chitra':
        key = 'true_citra'
    return key


def is_supported_ayanamsa_name(name: Any = None) -> bool:
    key = _canonical_ayanamsa_key(name if name not in (None, '') else DEFAULT_AYANAMSA_NAME)
    return key in AYANAMSA_MODES


def supported_ayanamsa_names() -> list[str]:
    return sorted(AYANAMSA_MODES)


def normalize_ayanamsa_name(name: Any = None) -> str:
    if name is None or (isinstance(name, str) and not name.strip()):
        return DEFAULT_AYANAMSA_NAME
    key = _canonical_ayanamsa_key(name)
    if key not in AYANAMSA_MODES:
        raise UnsupportedAyanamsaError(
            f"unsupported ayanamsa name {name!r}; supported: {', '.join(supported_ayanamsa_names())}"
        )
    return key


def ayanamsa_display_name(name: Any = None) -> str:
    key = normalize_ayanamsa_name(name)
    return AYANAMSA_DISPLAY_NAMES.get(key, key.title())


def current_ayanamsa_name(args: Any = None) -> str:
    name = getattr(args, 'ayanamsa', None) if args is not None else None
    return normalize_ayanamsa_name(name)


def apply_ayanamsa(name: Any = None, swe_module: Any = None) -> bool:
    """Set Swiss Ephemeris sidereal mode explicitly."""
    global ACTIVE_AYANAMSA_NAME, AYANAMSA_APPLIED_IN_PROCESS
    module = swe_module or swe
    if module is None:
        return False
    key = normalize_ayanamsa_name(name)
    try:
        module.set_sid_mode(AYANAMSA_MODES[key])
    except Exception:
        return False
    ACTIVE_AYANAMSA_NAME = key
    AYANAMSA_APPLIED_IN_PROCESS = True
    return True


def ayanamsa_already_applied() -> bool:
    """True once any caller has applied an ayanamsa in this process."""
    return AYANAMSA_APPLIED_IN_PROCESS


@contextmanager
def temporary_ayanamsa(name: Any = None, swe_module: Any = None) -> Iterator[str]:
    """Apply a sidereal mode and restore the previous process-wide mode."""
    previous = ACTIVE_AYANAMSA_NAME
    target = normalize_ayanamsa_name(name)
    if not apply_ayanamsa(target, swe_module):
        raise RuntimeError(f"failed to apply ayanamsa {target}")
    try:
        yield target
    finally:
        if not apply_ayanamsa(previous, swe_module):
            raise RuntimeError(f"failed to restore ayanamsa {previous}")


def sidereal_flags(swe_module: Any, ayanamsa_name: Any = None) -> int:
    apply_ayanamsa(ayanamsa_name, swe_module)
    return getattr(swe_module, 'FLG_SWIEPH', 2) | getattr(swe_module, 'FLG_SIDEREAL', 65536)


def sidereal_ayanamsa_degrees(jd_ut: float, swe_module: Any = None, flags: int | None = None) -> float:
    """Ayanamsa that FLG_SIDEREAL subtracts for these flags.

    ``get_ayanamsa`` and ``get_ayanamsa_ut`` omit nutation. Apparent tropical
    positions include it, so that subtraction shifts every longitude by Δψ.
    """
    module = swe_module or swe
    if module is None:
        raise RuntimeError("swisseph is not available")
    if flags is None:
        flags = getattr(module, "FLG_SWIEPH", 2)
    raw = module.get_ayanamsa_ex_ut(float(jd_ut), int(flags))
    if isinstance(raw, (tuple, list)):
        return float(raw[1])
    return float(raw)

# ---------------------------------------------------------------------------
# Bundled Swiss Ephemeris files (engine sync 2026-11-07).
# Bundled Swiss Ephemeris files (seas/semo/sepl_18.se1).
#
# The files are tracked under ``references/open_source_sources/vedic-astro-skills``
# and copied into the API image with ``references/``. Same candidates and order
# as the upstream engine (yinduzhanxing ``EPHE_PATH_CANDIDATES``). Without a path
# swisseph silently falls back to Moshier (returned flag 260 instead of 258), and
# ``set_ephe_path`` is process-global, so every caller that computes positions
# sets it explicitly instead of inheriting whatever an earlier call left.


_EPHE_ROOT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
EPHE_PATH_CANDIDATES = [
    os.path.join(_EPHE_ROOT_DIR, "references", "open_source_sources", "vedic-astro-skills",
                 variant, "skills", "vedic-calculator", "scripts", "ephe")
    for variant in ("antigravity", "codex", "claude-code")
]


def preferred_ephe_path() -> str:
    for path in EPHE_PATH_CANDIDATES:
        if os.path.isdir(path):
            return path
    return ""


def configure_ephemeris_path(swe) -> str:
    """Point swisseph at the bundled files; return the path used ('' = none found)."""
    path = preferred_ephe_path()
    swe.set_ephe_path(path)
    return path