Jyotisha/scripts/rectification_varga_segments_api.py

"""Minute-exact ascendant segments. This endpoint does not score or interpret."""
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")
VERSION = "rectification-varga-segments-v1"


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")
    profile = {
        "lat": _number(body, "latitude", -90, 90),
        "lon": _number(body, "longitude", -180, 180),
        "tz": _number(body, "timezone_offset", -14, 14),
        "ayanamsa": "raman", "node_mode": "mean",
    }
    key = hashlib.sha256(json.dumps({"version": VERSION, "start": str(start), "end": str(end), "targets": targets, **profile}, 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}
    for offset in range(width):
        moment = start + timedelta(minutes=offset)
        chart = domain_calculation_service.compute_chart({
            **profile, "year": moment.year, "month": moment.month, "day": moment.day,
            "hour": moment.hour, "minute": moment.minute,
        })
        ascendant = float(chart["ascendant"]["lon"])
        signs = {"D1": int(ascendant // 30) % 12} if "D1" in targets else {}
        divisions = [int(target[1:]) for target in targets if target != "D1"]
        if divisions:
            planets = {name: float(item["lon"]) for name, item in chart.get("planets", {}).items() if isinstance(item, dict) and isinstance(item.get("lon"), (int, float))}
            computed = varga.calc_all_vargas(planets, ascendant, divisions=divisions)
            for value in computed.values():
                signs[f"D{value['_meta']['div']}"] = int(value["Ascendant"]["sign_idx"])
        stamp = moment.strftime("%Y-%m-%dT%H:%M")
        minutes.append({"offset": offset, "time": stamp[11:], "date": stamp[:10], "signs": 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": "raman", "node_mode": "mean", "step_minutes": 1}
    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 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)