astrololo/services/engine-swisseph/app/main.py

"""engine-swisseph — IZOLOWANA usługa silnika B (AGPL).

UWAGA LICENCYJNA: ta usługa linkuje pyswisseph / Swiss Ephemeris, więc jest
objęta **AGPL-3.0** i jest licencjonowana osobno (patrz ./LICENSE). Jest celowo
wydzielona jako osobny proces i wołana przez HTTP — dzięki temu permisywny
produkt (prezentacja + logika z silnikiem własnym + dane) NIE jest linkowany z
kodem AGPL i nie podlega jego obowiązkom (LOG-27).

Rola: wyrocznia walidacyjna / tryb porównawczy (LOG-25/26). Nie wchodzi do
dystrybucji zamkniętej — uruchamiana tylko w profilu porównawczym/dev/CI.

Udostępnia ten sam kontrakt co RemoteEngine po stronie warstwy logicznej.
"""
from __future__ import annotations

from datetime import datetime

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

import swisseph as swe

app = FastAPI(title="astrololo · engine-swisseph (AGPL, izolowany)")

# Tryb Moshiera: bez plików efemeryd, w pełni samowystarczalny (~0,1\" dokładności).
_FLAGS = swe.FLG_MOSEPH | swe.FLG_SPEED

_PLANETS = {
    "Sun": swe.SUN, "Moon": swe.MOON, "Mercury": swe.MERCURY, "Venus": swe.VENUS,
    "Mars": swe.MARS, "Jupiter": swe.JUPITER, "Saturn": swe.SATURN,
    "Uranus": swe.URANUS, "Neptune": swe.NEPTUNE, "Pluto": swe.PLUTO,
    # punkty wirtualne — mean, jak w silniku własnym (parzystość LOG-28)
    "North Node": swe.MEAN_NODE, "Lilith": swe.MEAN_APOG,
    # "South Node" obsługiwany pochodnie w /positions: NN + 180°
}
DEFAULT_OBJECTS = [
    "Sun", "Moon", "Mercury", "Venus", "Mars", "Jupiter", "Saturn",
    "Uranus", "Neptune", "Pluto", "North Node", "South Node", "Lilith",
]


class PositionsRequest(BaseModel):
    when_utc: datetime
    lat: float = 0.0
    lon: float = 0.0
    objects: list[str] | None = None


@app.post("/positions")
def positions(req: PositionsRequest) -> dict:
    d = req.when_utc
    ut_hours = d.hour + d.minute / 60.0 + d.second / 3600.0
    jd = swe.julday(d.year, d.month, d.day, ut_hours)  # czas uniwersalny

    rows = []
    for name in (req.objects or DEFAULT_OBJECTS):
        lookup = "North Node" if name == "South Node" else name
        xx, _retflag = swe.calc_ut(jd, _PLANETS[lookup], _FLAGS)
        lon, lat, _dist, lon_speed = xx[0], xx[1], xx[2], xx[3]
        if name == "South Node":
            lon, lat = lon + 180.0, -lat
        rows.append({
            "name": name,
            "longitude": lon % 360.0,
            "latitude": lat,
            "speed": lon_speed,
            "retrograde": lon_speed < 0,
        })
    return {"engine": "swisseph", "positions": rows}


# Kody systemów domów w Swiss Ephemeris. Nazwy po LEWEJ są nasze — te same,
# których używa houses.SYSTEMS w warstwie logicznej — żeby wołający nie musiał
# znać liter swissepha. Lista celowo pokrywa się 1:1 z naszą: rozjazd oznaczałby,
# że kontrakt parzystości (LOG-28) przestał obejmować część systemów.
_HOUSE_CODES = {
    "whole_sign": b"W", "whole_sign_aries": b"N",
    "equal": b"E", "equal_mc": b"D",
    "porphyry": b"O", "vehlow": b"V", "morinus": b"M",
    "regiomontanus": b"R", "campanus": b"C", "alcabitus": b"B",
    "topocentric": b"T", "placidus": b"P", "koch": b"K",
}

# Kolejność, w jakiej swe_houses zwraca punkty w tablicy ascmc.
_ASCMC = ("Asc", "MC", "ARMC", "Vertex", "equatorial_asc",
          "co_asc_koch", "co_asc_munkasey", "polar_asc")


class HousesRequest(BaseModel):
    when_utc: datetime
    lat: float = 0.0
    lon: float = 0.0
    system: str = "whole_sign"


@app.post("/houses")
def houses(req: HousesRequest) -> dict:
    """Cuspy domów i osie policzone przez silnik B — do porównania z własnym.

    Domyka kontrakt parzystości (LOG-28) po stronie domów: dotąd obejmował
    wyłącznie pozycje obiektów, więc błąd w podziale na domy przechodził przez
    porównanie silników niezauważony. Błąd w domach jest CICHY — wykres wygląda
    poprawnie, tylko planety siedzą gdzie indziej — więc akurat tu warto mieć
    drugie zdanie.

    Placidus i Koch nie istnieją powyżej koła podbiegunowego i swisseph zgłasza
    tam wyjątek. Oddajemy to jako 422 z czytelnym powodem, a NIE podstawiamy po
    cichu innego systemu: cicha podmiana jest nie do wykrycia po stronie
    wołającego, a to on ma zdecydować, co z tym zrobić.
    """
    code = _HOUSE_CODES.get(req.system)
    if code is None:
        raise HTTPException(422, f"nieznany system domów: {req.system!r} "
                                 f"(znane: {', '.join(sorted(_HOUSE_CODES))})")
    d = req.when_utc
    ut_hours = d.hour + d.minute / 60.0 + d.second / 3600.0
    jd = swe.julday(d.year, d.month, d.day, ut_hours)
    try:
        cusps, ascmc = swe.houses(jd, req.lat, req.lon, code)
    except Exception as e:                      # poza dziedziną systemu
        raise HTTPException(
            422, f"system {req.system!r} nie ma definicji dla φ={req.lat}: {e}") from e

    return {
        "engine": "swisseph",
        "system": req.system,
        "cusps": [{"house": i + 1, "longitude": c % 360.0} for i, c in enumerate(cusps)],
        "angles": {name: ascmc[i] % 360.0 for i, name in enumerate(_ASCMC)
                   if i < len(ascmc) and name in ("Asc", "MC", "ARMC", "Vertex")},
    }


@app.get("/health")
def health() -> dict:
    return {"engine": "swisseph", "status": "ok", "mode": "moshier", "license": "AGPL-3.0"}