Jyotisha/scripts/rectification_varga_segments_api.py

"""Minute-exact ascendant segments. This endpoint does not score or interpret.

Issue #165: with ``stability_items: true`` the same per-minute charts also
report, item by item, whether the lagna degree, the Sripati house-cusp signs
and the Moon's sign / nakshatra / pada / house stay the same across the window,
and (with ``timezone_id``) whether the zone's historical offset is the same
offset at every minute. Each item is compared on its own; nothing is inferred
from another item.
"""
from __future__ import annotations

import hashlib
import json
import math
import os
import tempfile
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any

from scripts import domain_calculation_service, varga

TARGETS = ("D1", "D9", "D10")
BOUNDARY_CHARTS = ("D1", "D2", "D4", "D5", "D7", "D9", "D10", "D11", "D12", "D24", "D30")
VERSION = "rectification-varga-segments-v1"
SEGMENT_AYANAMSAS = frozenset(("raman", "lahiri", "kp", "true_citra", "true_pushya", "fagan_bradley"))
SEGMENT_NODE_MODES = frozenset(("mean", "true"))
STABILITY_VERSION = "birth-window-stability-v1"
# Moon items and cusps compared per minute; the lagna degree is compared at the
# chart page's display resolution (one arc-minute).
STABILITY_SEGMENT_ITEMS = ("moon_sign", "moon_nakshatra", "moon_pada", "moon_house")


def _segment_ayanamsa(value: Any) -> str:
    """Blank stays Raman. Unknown names raise; they do not fall back."""
    if value is None or (isinstance(value, str) and not value.strip()):
        return "raman"
    if not isinstance(value, str):
        raise ValueError("invalid_ayanamsa")
    key = value.strip().lower().replace("-", "_")
    if key in {"krishnamurti", "krishnamurti_paddhati", "krishnamurti/kp"}:
        key = "kp"
    elif key == "true_chitra":
        key = "true_citra"
    if key not in SEGMENT_AYANAMSAS:
        raise ValueError("invalid_ayanamsa")
    return key


def _segment_node_mode(value: Any) -> str:
    """Blank stays mean (callers without a profile). Anything else must be mean or true."""
    if value is None or (isinstance(value, str) and not value.strip()):
        return "mean"
    if not isinstance(value, str) or value.strip().lower() not in SEGMENT_NODE_MODES:
        raise ValueError("invalid_node_mode")
    return value.strip().lower()


def _stability_request(body: dict[str, Any]) -> tuple[bool, str | None]:
    wanted = body.get("stability_items", False)
    if not isinstance(wanted, bool):
        raise ValueError("invalid_stability_items")
    zone = body.get("timezone_id")
    if zone is None or (isinstance(zone, str) and not zone.strip()):
        return wanted, None
    if not wanted or not isinstance(zone, str) or len(zone) > 64:
        raise ValueError("invalid_timezone_id")
    return wanted, zone.strip()


def _number(body, name, low, high):
    value = body.get(name)
    if isinstance(value, bool):
        raise ValueError(f"invalid_{name}")
    try:
        result = float(value)
    except (ValueError, TypeError) as exc:
        raise ValueError(f"invalid_{name}") from exc
    if not math.isfinite(result) or not low <= result <= high:
        raise ValueError(f"invalid_{name}")
    return result


def scan_varga_segments(body: dict[str, Any], *, cache_dir: Path | None = None) -> dict[str, Any]:
    if not isinstance(body, dict):
        raise ValueError("invalid_segment_request")
    try:
        start = datetime.strptime(str(body["start_at"]), "%Y-%m-%dT%H:%M")
        end = datetime.strptime(str(body["end_at"]), "%Y-%m-%dT%H:%M")
    except (KeyError, ValueError) as exc:
        raise ValueError("invalid_segment_window") from exc
    width = int((end - start).total_seconds() / 60) + 1
    if width < 1 or width > 1441:
        raise ValueError("invalid_segment_window")
    targets = body.get("target_charts", list(TARGETS))
    if (not isinstance(targets, list) or not targets
            or any(not isinstance(target, str) or target not in TARGETS for target in targets)
            or len(set(targets)) != len(targets)):
        raise ValueError("invalid_target_charts")
    boundary_charts, recorded_time = _boundary_request(body)
    ayanamsa = _segment_ayanamsa(body.get("ayanamsa"))
    node_mode = _segment_node_mode(body.get("node_mode"))
    stability, timezone_id = _stability_request(body)
    profile = {
        "lat": _number(body, "latitude", -90, 90),
        "lon": _number(body, "longitude", -180, 180),
        "tz": _number(body, "timezone_offset", -14, 14),
        "ayanamsa": ayanamsa, "node_mode": node_mode,
    }
    if boundary_charts:
        profile["boundary_charts"] = list(boundary_charts)
        profile["recorded_time"] = recorded_time
    key_material = {"version": VERSION, "start": str(start), "end": str(end), "targets": targets, **profile}
    if stability:
        # Only stability requests carry these, so every earlier cache key is unchanged.
        key_material.update(stability=STABILITY_VERSION, timezone_id=timezone_id)
    key = hashlib.sha256(json.dumps(key_material, sort_keys=True).encode()).hexdigest()
    root = cache_dir if cache_dir is not None else Path(os.environ.get("JYOTISH_CACHE_DIR", str(Path(__file__).resolve().parents[1] / "scratch" / "local"))) / "varga_segments"
    path = root / f"{key}.json"
    if path.exists():
        try:
            cached = json.loads(path.read_text(encoding="utf-8"))
            if isinstance(cached, dict) and cached.get("window_key") == key and cached.get("version") == VERSION:
                return cached
        except (ValueError, OSError):
            pass
    minutes = []
    segments = {target: [] for target in targets}
    boundary_signs: list[dict[str, int]] = []
    wanted = list(dict.fromkeys([*targets, *boundary_charts]))
    divisions = [int(chart[1:]) for chart in wanted if chart not in {"D1", "D11"}]
    d11_chart = None
    if "D11" in wanted:
        from scripts.active_rectification_event_engine import _d11_chart as d11_chart
    per_minute_items: list[dict[str, Any]] = []
    for offset in range(width):
        moment = start + timedelta(minutes=offset)
        chart = domain_calculation_service.compute_chart({
            "lat": profile["lat"], "lon": profile["lon"], "tz": profile["tz"],
            "ayanamsa": ayanamsa, "node_mode": node_mode,
            "year": moment.year, "month": moment.month, "day": moment.day,
            "hour": moment.hour, "minute": moment.minute,
        })
        if stability:
            per_minute_items.append(_minute_items(chart, moment, profile))
        ascendant = float(chart["ascendant"]["lon"])
        planets = {name: float(item["lon"]) for name, item in chart.get("planets", {}).items() if isinstance(item, dict) and isinstance(item.get("lon"), (int, float))}
        all_signs = {"D1": int(ascendant // 30) % 12} if "D1" in wanted else {}
        if divisions:
            computed = varga.calc_all_vargas(planets, ascendant, divisions=divisions)
            for value in computed.values():
                all_signs[f"D{value['_meta']['div']}"] = int(value["Ascendant"]["sign_idx"])
        if "D11" in wanted:
            all_signs["D11"] = int(d11_chart(planets, ascendant)["Ascendant"]["sign_idx"])
        signs = {target: all_signs[target] for target in targets}
        stamp = moment.strftime("%Y-%m-%dT%H:%M")
        minutes.append({"offset": offset, "time": stamp[11:], "date": stamp[:10], "signs": signs})
        if boundary_charts:
            boundary_signs.append({chart: all_signs[chart] for chart in boundary_charts if chart in all_signs})
        for target in targets:
            rows = segments[target]
            sign = signs[target]
            if rows and rows[-1]["sign"] == sign:
                rows[-1].update(end_at=stamp, end=offset, length=rows[-1]["length"] + 1)
            else:
                rows.append({"index": len(rows), "start": offset, "end": offset, "start_at": stamp, "end_at": stamp, "sign": sign, "length": 1})
    payload = {"version": VERSION, "window_key": key, "start_at": body["start_at"], "end_at": body["end_at"],
               "window_minutes": width, "target_charts": targets, "minutes": minutes,
               "segments": segments, "full_window_unique": {target: len({row['sign'] for row in rows}) == 1 for target, rows in segments.items()},
               "ayanamsa": ayanamsa, "node_mode": node_mode, "step_minutes": 1}
    if stability:
        payload["stability"] = _stability_summary(per_minute_items, [row["date"] + "T" + row["time"] for row in minutes],
                                                  start, width, profile["tz"], timezone_id)
    if boundary_charts:
        payload["nearest_boundary"] = _nearest_from_signs(boundary_signs, [row["time"] for row in minutes], recorded_time)
    root.mkdir(parents=True, exist_ok=True)
    with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", dir=root, delete=False) as handle:
        json.dump(payload, handle, ensure_ascii=False, sort_keys=True)
        temporary = Path(handle.name)
    try:
        temporary.replace(path)
    finally:
        temporary.unlink(missing_ok=True)
    return payload


def _sidereal_mc(moment: datetime, profile: dict[str, Any]) -> float:
    """Sidereal MC the chart page's Sripati cusps use (``bhava_chalit_mc``)."""
    import swisseph as swe
    from ayanamsa_utils import temporary_ayanamsa

    hour_ut = moment.hour + moment.minute / 60.0 - profile["tz"]
    jd = swe.julday(moment.year, moment.month, moment.day, hour_ut)
    with temporary_ayanamsa(profile["ayanamsa"], swe):
        _, ascmc = swe.houses_ex(jd, profile["lat"], profile["lon"], b"R", swe.FLG_SWIEPH | swe.FLG_SIDEREAL)
    return float(ascmc[1]) % 360


def _minute_items(chart: dict[str, Any], moment: datetime, profile: dict[str, Any]) -> dict[str, Any]:
    from scripts.bhava_chalit import BhavaChalitCalculator

    ascendant = float(chart["ascendant"]["lon"]) % 360
    moon = chart["planets"]["Moon"]
    moon_lon = float(moon["lon"]) % 360
    cusps = BhavaChalitCalculator().calculate_cusps(ascendant, _sidereal_mc(moment, profile), "sripati")
    return {
        "lagna_arcmin": int(math.floor(ascendant * 60 + 1e-9)) % (360 * 60),
        "cusp_signs": [int(cusp // 30) % 12 for cusp in cusps],
        "moon_sign": int(moon_lon // 30) % 12,
        "moon_nakshatra": str(moon["nakshatra"]),
        "moon_pada": int(moon["nakshatra_pada"]),
        "moon_house": int(moon["house"]),
    }


def _segments_of(values: list[Any], stamps: list[str]) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    for value, stamp in zip(values, stamps, strict=True):
        if rows and rows[-1]["value"] == value:
            rows[-1].update(end_at=stamp, length=rows[-1]["length"] + 1)
        else:
            rows.append({"value": value, "start_at": stamp, "end_at": stamp, "length": 1})
    return rows


def _zone_check(start: datetime, width: int, offset_hours: float, timezone_id: str | None) -> dict[str, Any]:
    """Whether the zone's historical rules give the stored offset at every minute of the window."""
    if timezone_id is None:
        return {"status": "unchecked"}
    from zoneinfo import ZoneInfo, ZoneInfoNotFoundError

    try:
        zone = ZoneInfo(timezone_id)
    except (ZoneInfoNotFoundError, ValueError):
        return {"status": "unchecked"}
    seen: set[float] = set()
    ambiguous = nonexistent = False
    for offset in range(width):
        local = start + timedelta(minutes=offset)
        valid: set[float] = set()
        for fold in (0, 1):
            candidate = local.replace(tzinfo=zone, fold=fold)
            utc_offset = candidate.utcoffset()
            if utc_offset is None:
                continue
            # A wall time that does not exist does not survive the round trip.
            back = candidate.astimezone(ZoneInfo("UTC")).astimezone(zone).replace(tzinfo=None)
            if back == local:
                valid.add(utc_offset.total_seconds() / 3600)
        if not valid:
            nonexistent = True
        elif len(valid) > 1:
            ambiguous = True
        seen.update(valid)
    if nonexistent:
        status = "nonexistent"
    elif ambiguous:
        status = "ambiguous"
    elif seen != {float(offset_hours)}:
        status = "offset_differs"
    else:
        status = "consistent"
    return {"status": status, "zone_offsets": sorted(seen)}


def _stability_summary(items: list[dict[str, Any]], stamps: list[str], start: datetime, width: int,
                       offset_hours: float, timezone_id: str | None) -> dict[str, Any]:
    lagna = [row["lagna_arcmin"] for row in items]
    cusp_rows = [row["cusp_signs"] for row in items]
    changed_houses = [house + 1 for house in range(12) if len({row[house] for row in cusp_rows}) > 1]
    summary: dict[str, Any] = {
        "version": STABILITY_VERSION,
        "minutes_compared": len(items),
        "lagna_degree": {
            "changes": len(set(lagna)) > 1,
            "first_arcmin": lagna[0],
            "last_arcmin": lagna[-1],
            "min_arcmin": min(lagna),
            "max_arcmin": max(lagna),
        },
        "house_cusps": {
            "house_system": "sripati",
            "changes": bool(changed_houses),
            "changed_houses": changed_houses,
            "first_signs": cusp_rows[0],
        },
        "timezone_check": _zone_check(start, width, offset_hours, timezone_id),
    }
    for name in STABILITY_SEGMENT_ITEMS:
        segments = _segments_of([row[name] for row in items], stamps)
        summary[name] = {"changes": len(segments) > 1, "segments": segments}
    return summary


def _boundary_request(body: dict[str, Any]) -> tuple[tuple[str, ...], str | None]:
    if "boundary_charts" not in body:
        return (), None
    charts = body.get("boundary_charts")
    if (not isinstance(charts, list) or not charts
            or any(not isinstance(chart, str) or chart not in BOUNDARY_CHARTS for chart in charts)
            or len(set(charts)) != len(charts)):
        raise ValueError("invalid_boundary_charts")
    recorded = body.get("recorded_time")
    if not isinstance(recorded, str) or len(recorded) < 5 or recorded[2] != ":":
        raise ValueError("invalid_recorded_time")
    hour, minute = recorded[:2], recorded[3:5]
    if not (hour.isdigit() and minute.isdigit()):
        raise ValueError("invalid_recorded_time")
    hour_value, minute_value = int(hour), int(minute)
    if hour_value > 23 or minute_value > 59:
        raise ValueError("invalid_recorded_time")
    return tuple(charts), f"{hour_value:02d}:{minute_value:02d}"


def _nearest_from_signs(signs_by_minute: list[dict[str, int]], times: list[str], recorded_time: str | None) -> dict[str, Any]:
    from scripts.jyotish_engine import SIGNS, SIGNS_CN
    from scripts.rectification.nearest_boundary import nearest_chart_boundary

    transitions: list[dict[str, str]] = []
    previous: dict[str, int] | None = None
    for signs, clock in zip(signs_by_minute, times):
        if previous is not None:
            for chart, sign in signs.items():
                before = previous.get(chart)
                if isinstance(before, int) and isinstance(sign, int) and before != sign and 0 <= before <= 11 and 0 <= sign <= 11:
                    transitions.append({
                        "layer": chart.lower(),
                        "at": clock,
                        "from_sign": SIGNS_CN[SIGNS[before]],
                        "to_sign": SIGNS_CN[SIGNS[sign]],
                    })
        previous = signs
    return nearest_chart_boundary({"transitions": transitions}, recorded_time)


def handle_varga_segments(handler, body):
    from scripts.offline_compute_mixins import BadRequest
    try:
        payload = scan_varga_segments(body)
    except ValueError as exc:
        raise BadRequest(str(exc)) from exc
    handler._json(payload)