"""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") BOUNDARY_CHARTS = ("D1", "D2", "D4", "D5", "D7", "D9", "D10", "D11", "D12", "D24", "D30") 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") boundary_charts, recorded_time = _boundary_request(body) 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", } if boundary_charts: profile["boundary_charts"] = list(boundary_charts) profile["recorded_time"] = recorded_time 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} 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 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": "raman", "node_mode": "mean", "year": moment.year, "month": moment.month, "day": moment.day, "hour": moment.hour, "minute": moment.minute, }) 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": "raman", "node_mode": "mean", "step_minutes": 1} 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 _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)