Jyotisha/scripts/chart_view_points/api.py

"""`/api/chart_points`: the chart page's data tables and point layers (issue #114).

Everything here is an existing engine producer called directly, so the chart
page does not have to run the full reading:

- upagrahas: ``gulika.calculate_upagrahas`` (solar five + six time-segment points)
- special lagnas: ``native_special_lagnas.native_special_lagna_rows`` (Swiss
  Ephemeris sunrise; not ``jaimini.calc_sunrise_utc_hours``, BUG-1462)
- Yogi / Avayogi: ``annual_native.yogi_sphuta.build_yogi_sphuta_local``
- arudha padas A1–A11 and UL: ``jaimini.calc_arudha_padas``
- friendship (natural, temporary, compound): ``jyotish_engine._build_planetary_friendship_snapshot``

Each block fails on its own (``status: blocked`` + reason); one failure never
fails the response.
"""

from __future__ import annotations

from datetime import datetime
from typing import Any, Callable

SIGNS = (
    "Aries", "Taurus", "Gemini", "Cancer", "Leo", "Virgo",
    "Libra", "Scorpio", "Sagittarius", "Capricorn", "Aquarius", "Pisces",
)


class ChartPointsInputError(ValueError):
    """The request is missing the birth moment or the natal chart."""


def _number(body: dict[str, Any], key: str) -> float:
    value = body.get(key)
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise ChartPointsInputError(f"{key} must be a number")
    return float(value)


def _guard(build: Callable[[], Any]) -> Any:
    try:
        return build()
    except Exception as exc:  # each block is independent
        return {"status": "blocked", "reason": f"{exc.__class__.__name__}: {exc}"}


def _longitude(planet: Any) -> float | None:
    if not isinstance(planet, dict):
        return None
    for key in ("lon", "degree_raw", "longitude"):
        value = planet.get(key)
        if isinstance(value, (int, float)) and not isinstance(value, bool):
            return float(value) % 360.0
    return None


def _nakshatra(longitude: float) -> dict[str, Any]:
    from jyotish_engine import _nakshatra_fields

    return _nakshatra_fields(longitude)


def _point(key: str, longitude: float, asc_sign: int, **extra: Any) -> dict[str, Any]:
    longitude = longitude % 360.0
    sign = int(longitude // 30) % 12
    return {
        **_nakshatra(longitude),
        "key": key,
        "longitude": round(longitude, 4),
        "sign": SIGNS[sign],
        "degree_in_sign": round(longitude % 30, 4),
        "house": (sign - asc_sign) % 12 + 1,
        **extra,
    }


UPAGRAHA_ORDER = (
    "Dhuma", "Vyatipata", "Parivesha", "Indrachapa", "Upaketu",
    "Kaala", "Mrityu", "Artha_Praharaka", "Yama_Ghantaka", "Gulika", "Maandi",
)
SPECIAL_ORDER = (
    "Bhava_Lagna", "Hora_Lagna", "Ghati_Lagna", "ViGhati_Lagna",
    "Pranapada_Lagna", "Sree_Lagna", "Indu_Lagna", "Varnada_Lagna",
)
ARUDHA_ORDER = ("A1", "A2", "A3", "A4", "A5", "A6", "A7", "A8", "A9", "A10", "A11", "UL")


def compute_chart_points(body: dict[str, Any]) -> dict[str, Any]:
    if not isinstance(body, dict):
        raise ChartPointsInputError("request body must be an object")
    planets = body.get("planets")
    ascendant = body.get("ascendant")
    if not isinstance(planets, dict) or not isinstance(ascendant, dict):
        raise ChartPointsInputError("planets and ascendant from /api/chart are required")
    asc_lon = _longitude(ascendant)
    sun_lon = _longitude(planets.get("Sun"))
    if asc_lon is None or sun_lon is None:
        raise ChartPointsInputError("ascendant and Sun longitudes are required")
    year, month, day = int(_number(body, "year")), int(_number(body, "month")), int(_number(body, "day"))
    hour, minute = int(_number(body, "hour")), int(_number(body, "minute"))
    second = int(body.get("second") or 0)
    lat, lon, tz = _number(body, "lat"), _number(body, "lon"), _number(body, "tz")
    ayanamsa = str(body.get("ayanamsa") or "raman")
    birth = datetime(year, month, day, hour, minute, second)
    asc_sign = int(asc_lon // 30) % 12

    def upagrahas() -> dict[str, Any]:
        from gulika import calculate_upagrahas

        packet = calculate_upagrahas(birth, lat=lat, lon=lon, tz=tz, sun_longitude=sun_lon, ayanamsa=ayanamsa)
        raw = packet.get("raw") or {}
        rows = [
            _point(name, float(raw[name]["longitude"]), asc_sign,
                   kind="solar" if raw[name].get("source") == "solar_longitude_formula" else "time_segment")
            for name in UPAGRAHA_ORDER if isinstance(raw.get(name), dict) and "longitude" in raw[name]
        ]
        return {"status": "computed", "rows": rows, "boundary": packet.get("boundary")}

    def special() -> dict[str, Any]:
        from native_special_lagnas import native_special_lagna_rows

        rows = native_special_lagna_rows(birth, lat, lon, tz, asc_lon, ayanamsa)
        out = [
            _point(key, float(rows[key]["longitude"]), asc_sign)
            for key in SPECIAL_ORDER
            if isinstance(rows.get(key), dict) and rows[key].get("status") == "computed"
        ]
        if not out:
            reasons = {row.get("reason") for row in rows.values() if isinstance(row, dict)}
            return {"status": "blocked", "reason": ";".join(sorted(str(r) for r in reasons if r))}
        return {"status": "computed", "rows": out}

    def yogi() -> dict[str, Any]:
        from annual_native.yogi_sphuta import build_yogi_sphuta_local

        packet = build_yogi_sphuta_local({"planets": planets, "ascendant": ascendant})
        rows = []
        for key, name in (("yogi", "Yogi_Point"), ("ava_yogi", "Avayogi_Point")):
            part = packet.get(key)
            if isinstance(part, dict) and isinstance(part.get("longitude"), (int, float)):
                rows.append(_point(name, float(part["longitude"]), asc_sign))
        if not rows:
            return {"status": "blocked", "reason": str(packet.get("reason") or "yogi_points_missing")}
        return {"status": "computed", "rows": rows, "formula": packet.get("formula")}

    def arudha() -> dict[str, Any]:
        from jaimini import calc_arudha_padas

        longitudes = {name: value for name, value in ((n, _longitude(p)) for n, p in planets.items()) if value is not None}
        padas = calc_arudha_padas(asc_sign, longitudes).get("padas") or {}
        rows = []
        for key in ARUDHA_ORDER:
            pada = padas.get(key)
            if not isinstance(pada, dict) or pada.get("sign") not in SIGNS:
                continue
            sign = SIGNS.index(pada["sign"])
            rows.append({
                "key": key,
                "sign": pada["sign"],
                "house": (sign - asc_sign) % 12 + 1,
                "source_house": int(pada.get("house_num") or ARUDHA_ORDER.index(key) + 1),
                "source_lord": pada.get("source_house_lord"),
                "exception": bool(pada.get("exception_triggered")),
            })
        return {"status": "computed", "rows": rows}

    def friendship() -> dict[str, Any]:
        from jyotish_engine import _build_planetary_friendship_snapshot

        return _build_planetary_friendship_snapshot(planets)

    return {
        "success": True,
        "endpoint": "chart_points",
        "ascendant_sign": SIGNS[asc_sign],
        "upagrahas": _guard(upagrahas),
        "special_lagnas": _guard(special),
        "yogi_points": _guard(yogi),
        "arudha_padas": _guard(arudha),
        "friendship": _guard(friendship),
    }