astrololo/services/logic/app/engine/stations.py

"""Wykrywanie stacji planet (LOG-03): poprzednia/następna stacja, SD/SR, flaga <7 dni.

Stacja ścisła = moment, w którym prędkość zodiakalna przechodzi przez zero.
Metoda: próbki prędkości co 1 dzień w oknie ± SEARCH_DAYS → zmiana znaku →
bisekcja do dokładności ~1 minuty. Klasyfikacja: prędkość przed<0 i po>0 → SD
(stationary direct), odwrotnie → SR (stationary retrograde).

Pomijamy Słońce/Księżyc (nigdy Rx) i punkty mean (NN/SN/Lilith — ruch jednostajny).
"""
from __future__ import annotations

from datetime import timedelta

from app.engine.formats import in_sign
from app.engine.models import ChartMoment

# obiekty bez stacji
NO_STATIONS = {"Sun", "Moon", "North Node", "South Node", "Lilith"}

# Okno musi pokryć najdłuższą przerwę między stacjami (Mars/Wenus: ~700 dni),
# a krok skanu musi być krótszy niż najkrótsza retrogradacja (Merkury ~21 dni).
SEARCH_DAYS = 800          # okno poszukiwań w każdą stronę
SCAN_STEP_DAYS = 4.0       # krok zgrubnego skanu (potem bisekcja)
STATION_SOON_DAYS = 7.0    # próg flagi "stacja blisko" (konfigurowalny, notes2)


def _speed_fn(engine, name: str):
    """Zwraca funkcję: dni_od_momentu_bazowego -> prędkość [°/dobę]."""
    def speed(base_moment: ChartMoment, offset_days: float) -> float:
        m = ChartMoment(
            when_utc=base_moment.when_utc + timedelta(days=offset_days),
            lat=base_moment.lat, lon=base_moment.lon,
        )
        return engine.positions(m, [name])[0].speed
    return speed


def _bisect_zero(speed, moment: ChartMoment, lo: float, hi: float, iters: int = 20) -> float:
    """Bisekcja miejsca zerowego prędkości między dniami lo i hi."""
    s_lo = speed(moment, lo)
    for _ in range(iters):
        mid = (lo + hi) / 2.0
        s_mid = speed(moment, mid)
        if (s_lo < 0) == (s_mid < 0):
            lo, s_lo = mid, s_mid
        else:
            hi = mid
    return (lo + hi) / 2.0


def _station_info(engine, moment: ChartMoment, name: str, day: float, speed) -> dict:
    """Opis stacji w danym dniu (offset od momentu bazowego)."""
    before = speed(moment, day - 0.5)
    kind = "SD" if before < 0 else "SR"
    when = moment.when_utc + timedelta(days=day)
    m = ChartMoment(when_utc=when, lat=moment.lat, lon=moment.lon)
    lon = engine.positions(m, [name])[0].longitude
    return {
        "type": kind,
        "date": when.strftime("%Y-%m-%d %H:%M"),
        "days": round(day, 1),               # ujemne = w przeszłości
        "degree": in_sign(lon),
    }


def find_stations(engine, moment: ChartMoment, name: str, step_days: float = SCAN_STEP_DAYS) -> dict | None:
    """Poprzednia i następna stacja obiektu względem momentu horoskopu."""
    if name in NO_STATIONS:
        return None
    speed = _speed_fn(engine, name)

    prev_day = next_day = None
    # w przeszłość
    s_right = speed(moment, 0.0)
    d = 0.0
    while d > -SEARCH_DAYS:
        s_left = speed(moment, d - step_days)
        if (s_left < 0) != (s_right < 0):
            prev_day = _bisect_zero(speed, moment, d - step_days, d)
            break
        d, s_right = d - step_days, s_left
    # w przyszłość
    s_left = speed(moment, 0.0)
    d = 0.0
    while d < SEARCH_DAYS:
        s_right = speed(moment, d + step_days)
        if (s_left < 0) != (s_right < 0):
            next_day = _bisect_zero(speed, moment, d, d + step_days)
            break
        d, s_left = d + step_days, s_right

    result: dict = {}
    if prev_day is not None:
        result["prev"] = _station_info(engine, moment, name, prev_day, speed)
    if next_day is not None:
        result["next"] = _station_info(engine, moment, name, next_day, speed)
    result["station_soon"] = any(
        abs(x["days"]) < STATION_SOON_DAYS for x in result.values() if isinstance(x, dict)
    )
    return result or None