Silnik efemeryd: EngineProvider + SkyfieldEngine + harness porównawczy

Pierwszy increment implementacji warstwy logicznej (ścieżka A).

LOG-24: interfejs EphemerisEngine z dwoma backendami — SkyfieldEngine
  (własny, permisywny: Skyfield MIT + dane JPL public domain) oraz
  RemoteEngine (klient izolowanej usługi swisseph). Fabryka + leniwa
  inicjalizacja; endpointy /chart/positions i /chart/compare.
LOG-01: pozycje obiektów (długość/szerokość ekliptyczna, prędkość,
  kierunek, formaty: w znaku / absolutny / dziesiętny).
LOG-25/28: harness porównawczy (compare.py) z progami tolerancji oraz
  wspólny kontrakt parzystości; pełen zestaw testów.
LOG-27: services/engine-swisseph — osobna, opcjonalna usługa AGPL
  (pyswisseph, tryb Moshiera), licencjonowana osobno, w compose pod
  profilem "comparison"; nie wchodzi do zamkniętego produktu.

Walidacja: SkyfieldEngine zgadza się ze Swiss Ephemeris co do ~1" dla
wszystkich 10 obiektów na horoskopie referencyjnym (30.04.1984, Warszawa);
12 testów przechodzi (silnik B pomijany gdy nieskonfigurowany).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-06-27 19:48:31 +02:00
parent 9bf297d461
commit e69c0714b9
25 changed files with 805 additions and 0 deletions
+31
View File
@@ -0,0 +1,31 @@
"""Wspólne fixture'y testów silnika.
Horoskop referencyjny = przykład z notatek projektu (notes3): 30.04.1984,
09:35 CEST = 07:35 UTC, Warszawa. Pozycje znane z astro.com/Swiss Ephemeris
służą jako wyrocznia (LOG-25).
"""
from __future__ import annotations
import datetime as dt
import pytest
from app.engine.models import ChartMoment
REFERENCE_UTC = dt.datetime(1984, 4, 30, 7, 35, 0, tzinfo=dt.timezone.utc)
@pytest.fixture(scope="session")
def reference_moment() -> ChartMoment:
return ChartMoment(when_utc=REFERENCE_UTC, lat=52.2333, lon=21.0167)
@pytest.fixture(scope="session")
def own_engine():
pytest.importorskip("skyfield")
from app.engine.skyfield_engine import SkyfieldEngine
try:
return SkyfieldEngine()
except Exception as e: # brak efemeryd / brak sieci
pytest.skip(f"Nie można wczytać efemeryd: {e}")
+35
View File
@@ -0,0 +1,35 @@
"""Testy harnessu porównawczego (LOG-25) — bez efemeryd."""
from app.engine.compare import compare_positions
from app.engine.models import ObjectPosition
def test_identical_positions_have_zero_diff():
a = [ObjectPosition("Sun", 40.0, 0.0, 0.95, False)]
report = compare_positions(a, a, lon_tol_arcsec=1.0)
assert report.ok
assert report.max_arcsec == 0.0
def test_offset_beyond_tolerance_is_flagged():
a = [ObjectPosition("Sun", 40.0, 0.0, 0.95, False)]
b = [ObjectPosition("Sun", 40.1, 0.0, 0.95, False)] # 0.1° = 360"
report = compare_positions(a, b, lon_tol_arcsec=120.0)
assert not report.ok
assert report.diffs[0].over_tolerance
assert abs(abs(report.diffs[0].delta_arcsec) - 360.0) < 1.0
def test_retrograde_mismatch_is_flagged():
a = [ObjectPosition("Mars", 234.0, 0.0, -0.2, True)]
b = [ObjectPosition("Mars", 234.0, 0.0, 0.2, False)]
report = compare_positions(a, b, lon_tol_arcsec=120.0)
assert not report.ok
assert report.diffs[0].speed_sign_mismatch
def test_wraparound_delta_is_small():
a = [ObjectPosition("Sun", 359.99, 0.0, 1.0, False)]
b = [ObjectPosition("Sun", 0.01, 0.0, 1.0, False)]
report = compare_positions(a, b, lon_tol_arcsec=120.0)
assert report.ok # 0.02° ≈ 72" < 120"
assert abs(report.diffs[0].delta_arcsec) < 80.0
@@ -0,0 +1,25 @@
"""Kontrakt parzystości silników (LOG-28).
Ten sam test musi przejść dla KAŻDEGO silnika. Dziś uruchamiamy go dla silnika
własnego; gdy skonfigurowany jest ENGINE_SWISSEPH_URL, ten sam kontrakt sprawdza
też silnik B.
"""
import os
import pytest
from app.engine.compare import check_engine_contract
def test_own_engine_contract(own_engine, reference_moment):
check_engine_contract(own_engine, reference_moment)
@pytest.mark.skipif(
not os.getenv("ENGINE_SWISSEPH_URL"),
reason="silnik B (swisseph) nieskonfigurowany — pomijam",
)
def test_swisseph_engine_contract(reference_moment):
from app.engine.remote_engine import RemoteEngine
check_engine_contract(RemoteEngine(), reference_moment)
+28
View File
@@ -0,0 +1,28 @@
"""Testy formatowania (czyste, bez efemeryd)."""
from app.engine import formats
def test_sign_index():
assert formats.sign_index(0) == 0 # Aries
assert formats.sign_index(35) == 1 # Taurus
assert formats.sign_index(359.9) == 11 # Pisces
def test_in_sign_basic():
# 40°08'21" absolutnie = Taurus 10°08'21"
assert formats.in_sign(40.139166) == "Tau 10°08'21\""
def test_absolute():
assert formats.absolute(40.139166) == "40°08'21\""
def test_wraparound_and_norm():
assert formats.norm360(370.0) == 10.0
assert formats.norm360(-1.0) == 359.0
def test_rounding_carry_into_next_sign():
# tuż przy granicy znaku zaokrąglenie nie może dać "30°"
s = formats.in_sign(59.99999)
assert s.startswith("Gem 0°") or s.startswith("Tau 29°59")
@@ -0,0 +1,44 @@
"""Walidacja silnika własnego względem wyroczni (LOG-01 + LOG-25).
Porównujemy pozycje policzone Skyfieldem ze znanymi wartościami Swiss Ephemeris
(astro.com) dla horoskopu referencyjnego. To dokładnie mechanizm LOG-25 — tyle
że wyrocznią są tu zapisane wcześniej wartości referencyjne zamiast żywej usługi
swisseph. Tolerancja 2' (120") z zapasem na różnicę barycentrum/centrum planety.
"""
from app.engine.compare import compare_positions
from app.engine.models import ObjectPosition
# obiekt -> (długość absolutna w stopniach, retrogradacja) — Swiss Ephemeris / astro.com
REFERENCE = {
"Sun": (40.139167, False), # Tau 10°08'21"
"Moon": (30.551389, False), # Tau 0°33'05"
"Mercury": (27.383889, True), # Ari 27°23'02" Rx
"Venus": (27.676944, False), # Ari 27°40'37"
"Mars": (234.541111, True), # Sco 24°32'28" Rx
"Jupiter": (282.961667, True), # Cap 12°57'42" Rx
"Saturn": (223.310833, True), # Sco 13°18'39" Rx
"Uranus": (252.813333, True), # Sag 12°48'48" Rx
"Neptune": (271.220833, True), # Cap 1°13'15" Rx
"Pluto": (210.478889, True), # Sco 0°28'44" Rx
}
TOLERANCE_ARCSEC = 120.0
def _reference_positions():
return [
ObjectPosition(name, lon, 0.0, (-1.0 if retro else 1.0), retro)
for name, (lon, retro) in REFERENCE.items()
]
def test_positions_match_swisseph_reference(own_engine, reference_moment):
got = own_engine.positions(reference_moment)
report = compare_positions(got, _reference_positions(), lon_tol_arcsec=TOLERANCE_ARCSEC)
assert report.ok, report.summary()
def test_retrograde_flags_match(own_engine, reference_moment):
got = {p.name: p for p in own_engine.positions(reference_moment)}
for name, (_lon, retro) in REFERENCE.items():
assert got[name].retrograde == retro, f"{name}: kierunek niezgodny"