Jyotisha/scripts/chart_choice/boundaries.py

"""Exact dividing lines inside a birth-time window.

A "line" is the instant the ascendant crosses a degree where at least one of the
tracked divisional charts changes its ascendant sign.  Several charts can flip at
the same degree (for example 15 degrees flips D2, D4, D10, D12 and D24 at once);
those are one line, one choice.
"""
from __future__ import annotations

import sys
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any, Final

_SCRIPTS = Path(__file__).resolve().parents[1]
if str(_SCRIPTS) not in sys.path:
    sys.path.insert(0, str(_SCRIPTS))

import domain_calculation_service  # noqa: E402
import varga  # noqa: E402

TRACKED_DIVISIONS: Final[tuple[int, ...]] = (1, 2, 4, 9, 10, 12, 24)
MAX_WINDOW_MINUTES: Final = 121  # a ±60 min range (the widest quick choice) is compared directly
# Lines closer than this to the edge of a segment are not used as a saved minute.
EDGE_GUARD_SECONDS: Final = 60


class ChartChoiceInputError(ValueError):
    """The request cannot be evaluated (bad date, window, or place)."""


@dataclass(frozen=True)
class ChartRequest:
    birth_date: str  # YYYY-MM-DD
    lat: float
    lon: float
    tz: float
    ayanamsa: str = "raman"
    node_mode: str = "mean"


@dataclass
class Line:
    at: datetime
    degree: float  # ascendant degree inside its sign where the line sits
    flips: dict[str, tuple[int, int]] = field(default_factory=dict)  # "D10": (before_sign, after_sign)


def chart_at(req: ChartRequest, at: datetime) -> dict[str, Any]:
    return domain_calculation_service.compute_chart({
        "year": at.year, "month": at.month, "day": at.day,
        "hour": at.hour, "minute": at.minute, "second": at.second,
        "lat": req.lat, "lon": req.lon, "tz": req.tz,
        "ayanamsa": req.ayanamsa, "node_mode": req.node_mode,
    })


def ascendant_longitude(req: ChartRequest, at: datetime) -> float:
    return float(chart_at(req, at)["ascendant"]["lon"])


def varga_sign(longitude: float, division: int) -> int:
    if division == 1:
        return int(longitude // 30) % 12
    return int(varga.calc_varga(longitude, division)["sign_idx"])


def signature(longitude: float) -> dict[str, int]:
    return {f"D{d}": varga_sign(longitude, d) for d in TRACKED_DIVISIONS}


def _unwrap(prev: float, cur: float) -> float:
    while cur < prev - 180:
        cur += 360
    return cur


def _crossing_degrees(lo: float, hi: float) -> list[float]:
    """Every tracked division boundary strictly inside (lo, hi], unwrapped degrees."""
    found: set[float] = set()
    for d in TRACKED_DIVISIONS:
        part = 30.0 / d
        k = int(lo // part) + 1
        while k * part <= hi + 1e-12:
            found.add(round(k * part, 9))
            k += 1
    return sorted(found)


def _bisect_time(req: ChartRequest, t0: datetime, a0: float, t1: datetime, degree: float) -> datetime:
    """Second-resolution time at which the (unwrapped) ascendant reaches ``degree``."""
    lo, hi = t0, t1
    while (hi - lo).total_seconds() > 1:
        mid = lo + (hi - lo) / 2
        mid = mid.replace(microsecond=0)
        if mid <= lo:
            break
        a = _unwrap(a0, ascendant_longitude(req, mid))
        if a < degree:
            lo = mid
        else:
            hi = mid
    return hi


def find_lines(req: ChartRequest, start: datetime, end: datetime) -> list[Line]:
    """All dividing lines in [start, end], to the second."""
    if end <= start:
        raise ChartChoiceInputError("window end must be after start")
    lines: list[Line] = []
    t = start
    a_prev = ascendant_longitude(req, t)
    while t < end:
        t_next = min(t + timedelta(minutes=1), end)
        a_next = _unwrap(a_prev, ascendant_longitude(req, t_next))
        for deg in _crossing_degrees(a_prev, a_next):
            at = _bisect_time(req, t, a_prev, t_next, deg)
            before = signature((deg - 1e-6) % 360)
            after = signature((deg + 1e-6) % 360)
            flips = {k: (before[k], after[k]) for k in before if before[k] != after[k]}
            if flips:
                lines.append(Line(at=at, degree=round(deg % 30, 6), flips=flips))
        t, a_prev = t_next, a_next
    return lines


def nearest_line(lines: list[Line], reported: datetime) -> Line | None:
    if not lines:
        return None
    return min(lines, key=lambda line: abs((line.at - reported).total_seconds()))


def segment_around(lines: list[Line], line: Line, start: datetime, end: datetime) -> tuple[tuple[datetime, datetime], tuple[datetime, datetime]]:
    """(before_segment, after_segment) bounded by the neighbouring lines and the window."""
    idx = lines.index(line)
    prev_at = lines[idx - 1].at if idx > 0 else start
    next_at = lines[idx + 1].at if idx + 1 < len(lines) else end
    return (max(prev_at, start), line.at), (line.at, min(next_at, end))


def representative_minute(segment: tuple[datetime, datetime], reported: datetime) -> datetime:
    """Whole minute to save for a side.  Keeps the reported time when it lies on that side."""
    lo, hi = segment
    if lo <= reported < hi:
        return reported
    guard = timedelta(seconds=EDGE_GUARD_SECONDS)
    inner_lo, inner_hi = lo + guard, hi - guard
    if inner_hi < inner_lo:  # very short segment: use its middle
        inner_lo = inner_hi = lo + (hi - lo) / 2
    mid = inner_lo + (inner_hi - inner_lo) / 2
    minute = mid.replace(second=0, microsecond=0)
    if minute < lo:
        minute += timedelta(minutes=1)
    if minute >= hi:
        minute = (hi - timedelta(seconds=1)).replace(second=0, microsecond=0)
    return minute