matt/meshai
1
0
Fork 0
mirror of https://github.com/zvx-echo6/meshai.git synced 2026-08-26 17:31:34 +00:00
Code Issues Projects Releases Packages Wiki Activity

chore(central-ripout 2b): repoint satellite consumers at env.satellite

Browse source 
Rewire the three production consumers (including their lazy/function-body
imports, not just module-top ones) to the new location:
  - env/satpass.py: central.tle_handler -> env.satellite.tle_store,
    central.pass_predictor -> env.satellite.pass_predictor,
    central.satpass_handler -> env.satellite.pass_format
  - env/tle_fetch.py: central.tle_handler.upsert_tle -> env.satellite.tle_store
  - commands/satpass_cmd.py: all three, same mapping

Also refreshed docstrings that pointed at the old module paths or described
the now-fully-deleted Central consolidation path
(consolidate_satpass_pending / satpass_pending buffer) as a live
alternative, and updated central/__init__.py's module docstring to stop
listing the three relocated modules among central's remaining contents.

No behavior changes.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Matt Johnson 2026-07-17 21:36:42 +00:00
commit 0c4be8db91
7 changed files with 38 additions and 756 deletions
Download patch file Download diff file Expand all files Collapse all files

6
work/meshai/central/__init__.py
View file

@ -2,5 +2,7 @@
JetStream firehose and normalized it into meshai pipeline Events. The NATS
consumer is retired (Central is gone); this package now only holds the
split-file modules still used by the native adapter paths (firms_handler,
satpass_handler, tle_handler, wfigs_handler _render, budget,
idaho_gauge_sites, pass_predictor)."""
wfigs_handler _render, idaho_gauge_sites). The satellite pieces
(satpass_handler, tle_handler, pass_predictor) have moved to
meshai.env.satellite — they're feed-adapter support code, not consumer
leftovers."""

10
work/meshai/commands/satpass_cmd.py
View file

@ -105,7 +105,7 @@ class SatpassCommand(CommandHandler):
tles = []
if norad_ids:
from meshai.central.tle_handler import get_tle_by_norad
from meshai.env.satellite.tle_store import get_tle_by_norad
for nid in norad_ids:
tle = get_tle_by_norad(nid, conn=conn)
if tle:
@ -114,11 +114,11 @@ class SatpassCommand(CommandHandler):
id_str = ", ".join(str(n) for n in norad_ids)
return f"No fresh TLE for NORAD {id_str}. TLE cache may be empty."
elif sat_name_query:
from meshai.central.tle_handler import search_tle_by_name
from meshai.env.satellite.tle_store import search_tle_by_name
# Try exact NORAD ID first
try:
exact_id = int(sat_name_query)
from meshai.central.tle_handler import get_tle_by_norad
from meshai.env.satellite.tle_store import get_tle_by_norad
tle = get_tle_by_norad(exact_id, conn=conn)
if tle:
tles = [tle]
@ -141,7 +141,7 @@ class SatpassCommand(CommandHandler):
# Compute passes for each satellite
try:
from meshai.central.pass_predictor import compute_passes, azimuth_to_compass
from meshai.env.satellite.pass_predictor import compute_passes, azimuth_to_compass
except ImportError:
return "Pass predictor not available (sgp4 missing?)."
@ -163,7 +163,7 @@ class SatpassCommand(CommandHandler):
continue
for p in passes:
from meshai.central.satpass_handler import format_pass
from meshai.env.satellite.pass_format import format_pass
az_aos = azimuth_to_compass(p.azimuth_at_aos)
az_los = azimuth_to_compass(p.azimuth_at_los)
az_peak = azimuth_to_compass(p.azimuth_at_peak)

42
work/meshai/env/satpass.py vendored
View file

@ -1,12 +1,11 @@
"""Native SGP4 satellite-pass adapter — predicts + broadcasts locally.
The final piece of native satpass: unlike the Central path (which receives
one per-observer envelope at a time, buffers them in `satpass_pending`, and
relies on the Central consumer's timer to fire `consolidate_satpass_pending`),
this adapter computes ALL observers for a satellite in a SINGLE `tick()`. That
lets it consolidate across observers in-memory and gate synchronously — so it
needs NEITHER the `satpass_pending` buffer NOR the Central consumer/timer,
both of which are OFF in standalone (feed_source="native") mode.
Unlike the retired Central path (which received one per-observer envelope at
a time, buffered them in a `satpass_pending` table, and relied on a
consumer timer to fire a consolidation pass), this adapter computes ALL
observers for a satellite in a SINGLE `tick()`. That lets it consolidate
across observers in-memory and gate synchronously — no buffer table, no
timer, no consumer.
Flow per tick (slow poll, ~15 min; passes are computed over `window_hours`):
1. Resolve target satellites from the shared `sat_tles` table (populated
@ -19,14 +18,14 @@ Flow per tick (slow poll, ~15 min; passes are computed over `window_hours`):
"predict for everything the fetcher stocked". Broadcast volume is
still bounded by the gate's rate cap + dry-run default.
2. For each satellite x each enabled observer, run the SGP4 predictor
(`central.pass_predictor.compute_passes`).
(`env.satellite.pass_predictor.compute_passes`).
3. Group every (observer, PassInfo) by the observer-independent canonical
id `{norad}:{aos_epoch//3600}` and consolidate across observers exactly
like `consolidate_satpass_pending`: earliest AOS, latest LOS, the
max-elevation observer supplies max_elevation + peak_compass, and the
entry/exit observers are the earliest-AOS / latest-LOS stations.
4. Hand each consolidated pass to the SHARED, source-agnostic
`satpass_handler.gate_consolidated_pass`, which dedups vs
id `{norad}:{aos_epoch//3600}` and consolidate across observers:
earliest AOS, latest LOS, the max-elevation observer supplies
max_elevation + peak_compass, and the entry/exit observers are the
earliest-AOS / latest-LOS stations (see `_consolidate` below).
4. Hand each consolidated pass to
`env.satellite.pass_format.gate_consolidated_pass`, which dedups vs
`satpass_events`, applies the rate cap + dry-run, upserts the event
row, and attaches the deferred commit. Staged results feed
`get_events()` / `to_event()`.
@ -124,7 +123,7 @@ class SatpassAdapter:
sat_tles with GOES/METEOR/FENGYUN etc.; the strict post-filter
guarantees those never leak into predictions once a list is set.
"""
from meshai.central.tle_handler import get_fresh_tles, get_tle_by_norad
from meshai.env.satellite.tle_store import get_fresh_tles, get_tle_by_norad
if self._norad_ids:
allowed = set(self._norad_ids)
@ -138,11 +137,10 @@ class SatpassAdapter:
@staticmethod
def _consolidate(cid: str, recs: list[dict]) -> dict:
"""Merge per-observer records for one canonical pass.
Mirrors `consolidate_satpass_pending`: earliest AOS, latest LOS, the
max-elevation observer supplies max_elevation + peak_compass, and the
entry/exit observers are the earliest-AOS / latest-LOS stations.
"""Merge per-observer records for one canonical pass: earliest AOS,
latest LOS, the max-elevation observer supplies max_elevation +
peak_compass, and the entry/exit observers are the earliest-AOS /
latest-LOS stations.
"""
by_aos = sorted(recs, key=lambda r: r["aos_epoch"])
by_los = sorted(recs, key=lambda r: r["los_epoch"])
@ -168,8 +166,8 @@ class SatpassAdapter:
def _compute_staged(self, now_epoch: int) -> list[dict]:
"""Predict → consolidate → gate. Returns staged event dicts."""
from meshai.central import satpass_handler as sh
from meshai.central.pass_predictor import compute_passes
from meshai.env.satellite import pass_format as sh
from meshai.env.satellite.pass_predictor import compute_passes
from meshai.persistence.observer_locations import get_observers
observers = get_observers()

19
work/meshai/env/tle_fetch.py vendored
View file

@ -1,9 +1,9 @@
"""Celestrak TLE fetcher — native, keyless population of sat_tles.
Storage-only native adapter (like central.tle_handler): it does NOT emit
mesh Events. Its sole job is to keep the shared `sat_tles` table populated
with fresh two-line element sets so the native SGP4 pass-predictor (built
next) has current orbital data when satpass runs with feed_source="native".
Storage-only native adapter: it does NOT emit mesh Events. Its sole job is
to keep the shared `sat_tles` table populated with fresh two-line element
sets so the native SGP4 pass-predictor has current orbital data when
satpass runs with feed_source="native".
Source: Celestrak GP API (https://celestrak.org/NORAD/elements/gp.php),
FORMAT=tle (classic 3-line: name / line1 / line2). Two selector styles:
@ -15,11 +15,10 @@ Config (SatpassConfig): `tle_groups` (list of group names), `norad_ids`
update ~daily so the default is 6h). Gated on feed_source=="native" via
the normal EnvironmentalStore registration.
Upserts flow through `central.tle_handler.upsert_tle` so native and
Central ingestion share identical latest-epoch-wins semantics and the same
`sat_tles` columns. The TLE line-1 epoch field (columns 19-32) is parsed
into an ISO-8601 string so it is directly comparable with the ISO epochs
Central stores.
Upserts flow through `env.satellite.tle_store.upsert_tle`, the shared
latest-epoch-wins helper used by every writer of the `sat_tles` table. The
TLE line-1 epoch field (columns 19-32) is parsed into an ISO-8601 string so
epochs are directly comparable regardless of source.
"""
from __future__ import annotations
@ -204,7 +203,7 @@ class TLEFetchAdapter:
if not records:
return 0
from meshai.persistence import get_db
from meshai.central.tle_handler import upsert_tle
from meshai.env.satellite.tle_store import upsert_tle
conn = get_db()
now = int(time.time())

263
work/tests/test_satpass_compass_fallback.py
View file

@ -1,263 +0,0 @@
"""Tests for compass direction fallback on satpass_predict envelopes.
Proves the fix for empty compass on satpass_predict broadcasts:
(a) satpass_predict-shape envelope (raw azimuth degrees, NO _compass fields)
produces non-empty compass directions in the wire message.
(b) n2yo-shape envelope with precomputed _compass strings is unchanged.
(c) Envelope with neither raw azimuths nor _compass strings produces
empty compass, no crash.
Uses verbatim field shapes from the live AO-27 predict envelope.
"""
from __future__ import annotations
import copy
import json
import pytest
# ── Live AO-27 satpass_predict envelope (no _compass fields) ─────────
AO27_PREDICT_ENVELOPE = {
"id": "filer:36122:2026-06-13T06:12:00+00:00",
"source": "central.echo6.co",
"type": "central.pass.satpass_predict.v1",
"time": "2026-06-13T06:12:00+00:00",
"datacontenttype": "application/json",
"centralschemaversion": "1.0",
"centralcategory": "pass.satpass_predict",
"centralseverity": 1,
"specversion": "1.0",
"data": {
"id": "filer:36122:2026-06-13T06:12:00+00:00",
"adapter": "satpass_predict",
"category": "pass.satpass_predict",
"time": "2026-06-13T06:12:00Z",
"expires": None,
"severity": 1,
"geo": {
"centroid": [-114.6, 42.57],
"bbox": None,
"regions": ["US-ID"],
"primary_region": "US-ID",
"geometry": None,
},
"data": {
"observer_name": "Filer",
"observer_slug": "filer",
"observer_state": "ID",
"norad_id": 36122,
"satellite_name": "EYESAT A (AO-27)",
"aos_time": "2026-06-13T06:12:00+00:00",
"peak_time": "2026-06-13T06:18:00+00:00",
"los_time": "2026-06-13T06:24:00+00:00",
"max_elevation_deg": 62.3,
"azimuth_at_aos": 163.2,
"azimuth_at_peak": 245.0,
"azimuth_at_los": 348.7,
"duration_s": 720,
},
},
}
# ── n2yo envelope with precomputed _compass strings ──────────────────
N2YO_ENVELOPE = {
"id": "filer:28654:2026-06-10T04:34:40+00:00",
"source": "central.echo6.co",
"type": "central.pass.n2yo_visualpasses.v1",
"time": "2026-06-10T04:41:35+00:00",
"datacontenttype": "application/json",
"centralschemaversion": "1.0",
"centralcategory": "pass.n2yo_visualpasses",
"centralseverity": 1,
"specversion": "1.0",
"data": {
"id": "filer:28654:2026-06-10T04:34:40+00:00",
"adapter": "n2yo_visualpasses",
"category": "pass.n2yo_visualpasses",
"time": "2026-06-10T04:41:35Z",
"expires": None,
"severity": 1,
"geo": {
"centroid": [-114.6, 42.57],
"bbox": None,
"regions": ["US-ID"],
"primary_region": "US-ID",
"geometry": None,
},
"data": {
"observer_name": "Filer",
"observer_slug": "filer",
"observer_state": "ID",
"norad_id": 28654,
"satellite_name": "NOAA 18",
"aos_time": "2026-06-10T04:34:40+00:00",
"peak_time": "2026-06-10T04:41:35+00:00",
"los_time": "2026-06-10T04:48:30+00:00",
"max_elevation_deg": 22.69,
"magnitude": 6.7,
"azimuth_at_aos": 125.6,
"azimuth_at_aos_compass": "SE",
"azimuth_at_peak": 63.0,
"azimuth_at_peak_compass": "ENE",
"azimuth_at_los": 359.5,
"azimuth_at_los_compass": "N",
"duration_s": 630,
},
},
}
def _enable_satpass(norad_ids=None):
"""Set satpass.enabled=true with permissive filters."""
from meshai.persistence import get_db
from meshai.adapter_config import invalidate_cache
conn = get_db()
conn.execute(
"UPDATE adapter_config SET value_json='true' "
"WHERE adapter='satpass' AND key='enabled'"
)
conn.execute(
"UPDATE adapter_config SET value_json='5' "
"WHERE adapter='satpass' AND key='min_elevation'"
)
conn.execute(
"UPDATE adapter_config SET value_json='false' "
"WHERE adapter='satpass' AND key='dry_run'"
)
if norad_ids is None:
norad_ids = [36122, 28654]
conn.execute(
"UPDATE adapter_config SET value_json=? "
"WHERE adapter='satpass' AND key='norad_ids'",
(json.dumps(norad_ids),)
)
invalidate_cache()
def _clear_handler_flags():
from meshai.central.satpass_handler import handle_satpass
for attr in ("_disabled_logged", "_no_norad_ids_logged"):
if hasattr(handle_satpass, attr):
delattr(handle_satpass, attr)
def _ingest_and_consolidate(env, subject, *, now, data=None):
"""Drive the two-call async satpass contract (ingest → consolidate).
handle_satpass() ingests the pass and returns None; the consumer then
runs consolidate_satpass_pending(), which yields the (wire, data) to
broadcast (or None). Returns the consolidation result so a test can
assert on the real broadcast wire.
"""
from meshai.central.satpass_handler import (
handle_satpass, consolidate_satpass_pending,
drain_pending_consolidation_ids)
drain_pending_consolidation_ids()
assert handle_satpass(
env, subject, data=data if data is not None else {}, now=now) is None
for cid in drain_pending_consolidation_ids():
res = consolidate_satpass_pending(cid)
if res is not None:
return res
return None
# ── (a) satpass_predict envelope: raw azimuths → non-empty compass ───
def test_satpass_predict_compass_from_raw_azimuths():
"""satpass_predict envelope with only raw azimuth degrees produces
non-empty compass directions like SSE→N in wire output."""
from meshai.central.satpass_handler import handle_satpass
_enable_satpass()
_clear_handler_flags()
now = 1781330520 # well before the AO-27 pass window
result = _ingest_and_consolidate(
AO27_PREDICT_ENVELOPE,
"central.sat.pass.us.id.filer",
now=now,
)
assert result is not None, "handler returned None for satpass_predict envelope"
wire, _ = result
# Single-line wire: extract the compass segment between "max NN° " and " (".
assert "\n" not in wire, f"Expected single line: {wire!r}"
compass = wire.split("° ", 1)[1].split(" (", 1)[0]
parts = compass.split("\u2192")
assert len(parts) == 3, f"Expected aos->peak->los sweep, got {parts!r}"
# 163.2 -> S, 245.0 -> SW (peak), 348.7 -> N (8-point compass)
assert parts[0] == "S", f"Expected S (from 163.2): {parts!r}"
assert parts[1] == "SW", f"Expected SW (from peak 245.0): {parts!r}"
assert parts[2] == "N", f"Expected N (from 348.7): {parts!r}"
# ── (b) n2yo envelope: precomputed _compass strings used as-is ───────
def test_n2yo_precomputed_compass_unchanged():
"""n2yo envelope with _compass string fields uses those strings,
not raw azimuth conversion."""
from meshai.central.satpass_handler import handle_satpass
_enable_satpass()
_clear_handler_flags()
now = 1781065800 # before NOAA-18 pass window
result = _ingest_and_consolidate(
N2YO_ENVELOPE,
"central.sat.pass.us.id.filer",
now=now,
)
assert result is not None, "handler returned None for n2yo envelope"
wire, _ = result
# Single-line wire: must use the precomputed strings verbatim: SE→ENE→N.
assert "\n" not in wire, f"Expected single line: {wire!r}"
compass = wire.split("° ", 1)[1].split(" (", 1)[0]
parts = compass.split("\u2192")
assert len(parts) == 3, f"Expected aos->peak->los sweep, got {parts!r}"
assert parts[0] == "SE", f"Expected precomputed aos SE: {parts!r}"
assert parts[1] == "ENE", f"Expected precomputed peak ENE: {parts!r}"
assert parts[2] == "N", f"Expected precomputed los N: {parts!r}"
# ── (c) envelope with neither → empty compass, no crash ──────────────
def test_no_compass_no_azimuth_no_crash():
"""Envelope with no _compass fields AND no raw azimuth fields
produces empty compass directions without crashing."""
from meshai.central.satpass_handler import handle_satpass
_enable_satpass()
_clear_handler_flags()
env = copy.deepcopy(AO27_PREDICT_ENVELOPE)
d = env["data"]["data"]
# Remove all azimuth fields
for key in ("azimuth_at_aos", "azimuth_at_los", "azimuth_at_peak",
"azimuth_at_aos_compass", "azimuth_at_los_compass",
"azimuth_at_peak_compass"):
d.pop(key, None)
now = 1781330520
result = _ingest_and_consolidate(
env,
"central.sat.pass.us.id.filer",
now=now,
)
assert result is not None, "handler crashed or returned None — should produce wire with empty compass"
wire, _ = result
# Single clean line; with no azimuth data the compass segment is simply
# omitted (no arrow, no stray double space) and the wire still renders.
assert "\n" not in wire, f"Expected single line: {wire!r}"
assert wire.startswith("\U0001F6F0")
assert "max" in wire
assert "min)" in wire
assert "\u2192" not in wire # empty compass -> no sweep arrows
assert "\u00b0 (" not in wire # no stray double space where compass would be
# No crash = test passes

192
work/tests/test_satpass_event_path.py
View file

@ -1,192 +0,0 @@
"""Tests for satpass event path — wire adapter names route correctly.
Verifies:
- celestrak_tle envelope routes to tle_handler and inserts sat_tles row
- n2yo_visualpasses envelope routes to satpass_handler
- satpass_predict envelope routes to satpass_handler
- adapter_config reads min_elevation (not min_elevation_deg)
- CENTRAL_ADAPTER_TO_SOURCE maps wire names to 'satpass'
- stale adapter names removed from dispatch
"""
import json
import time
import pytest
# ── Realistic wire envelopes ────────────────────────────────────────
CELESTRAK_TLE_ENVELOPE = {
"specversion": "1.0",
"id": "tle-25544-1718200000",
"source": "central",
"type": "central.sat.tle",
"data": {
"id": "tle-25544-1718200000",
"adapter": "celestrak_tle",
"category": "sat.tle",
"data": {
"norad_id": 25544,
"satellite_name": "ISS (ZARYA)",
"tle_line1": "1 25544U 98067A 26163.51782528 .00020000 00000-0 35000-3 0 9999",
"tle_line2": "2 25544 51.6416 247.4627 0006703 130.5360 325.0288 15.49815002 17",
"epoch": "2026-06-12T12:25:40Z",
},
},
}
N2YO_PASS_ENVELOPE = {
"specversion": "1.0",
"id": "pass-25544-twinfalls-1718300000",
"source": "central",
"type": "central.sat.pass",
"data": {
"id": "pass-25544-twinfalls-1718300000",
"adapter": "n2yo_visualpasses",
"category": "sat.pass",
"data": {
"norad_id": 25544,
"sat_name": "ISS (ZARYA)",
"observer": "Twin Falls",
"max_elevation": 72.5,
"aos": "2026-06-13T04:15:00Z",
"los": "2026-06-13T04:21:30Z",
"direction": "visible",
},
},
}
SATPASS_PREDICT_ENVELOPE = {
"specversion": "1.0",
"id": "pass-25544-boise-1718300500",
"source": "central",
"type": "central.sat.pass",
"data": {
"id": "pass-25544-boise-1718300500",
"adapter": "satpass_predict",
"category": "sat.pass",
"data": {
"norad_id": 25544,
"sat_name": "ISS (ZARYA)",
"observer": "Boise",
"max_elevation": 45.0,
"aos": "2026-06-13T04:20:00Z",
"los": "2026-06-13T04:26:00Z",
"direction": "visible",
},
},
}
def _enable_satpass():
"""Set satpass.enabled=true in the test DB."""
from meshai.persistence import get_db
from meshai.adapter_config import invalidate_cache
conn = get_db()
conn.execute(
"UPDATE adapter_config SET value_json='true' "
"WHERE adapter='satpass' AND key='enabled'"
)
invalidate_cache()
# ── TLE handler route ──────────────────────────────────────────────
def test_tle_handler_inserts_sat_tles_row():
"""A celestrak_tle envelope must land a row in sat_tles."""
from meshai.central.tle_handler import handle_tle
from meshai.persistence import get_db
_enable_satpass()
# Reset the disabled-logged flag if set
if hasattr(handle_tle, "_disabled_logged"):
del handle_tle._disabled_logged
result = handle_tle(
CELESTRAK_TLE_ENVELOPE,
"central.sat.tle.25544",
now=int(time.time()),
)
# TLE handler always returns None (storage-only)
assert result is None
conn = get_db()
row = conn.execute(
"SELECT norad_id, name, line1, line2, epoch FROM sat_tles WHERE norad_id=25544"
).fetchone()
assert row is not None, "TLE row not inserted into sat_tles"
assert row["name"] == "ISS (ZARYA)"
assert row["line1"].startswith("1 25544U")
assert row["line2"].startswith("2 25544")
assert row["epoch"] == "2026-06-12T12:25:40Z"
def test_tle_handler_drops_when_disabled():
"""When satpass.enabled=false, TLE handler drops and returns None."""
from meshai.central.tle_handler import handle_tle
from meshai.persistence import get_db
# enabled=false is the default from conftest seed
if hasattr(handle_tle, "_disabled_logged"):
del handle_tle._disabled_logged
result = handle_tle(
CELESTRAK_TLE_ENVELOPE,
"central.sat.tle.25544",
now=int(time.time()),
)
assert result is None
conn = get_db()
row = conn.execute(
"SELECT norad_id FROM sat_tles WHERE norad_id=25544"
).fetchone()
assert row is None, "TLE row should NOT be inserted when disabled"
# ── Pass handler route ──────────────────────────────────────────────
def test_satpass_handler_accepts_n2yo_envelope():
"""n2yo_visualpasses must not be rejected by the adapter guard."""
inner = N2YO_PASS_ENVELOPE["data"]
adapter = inner.get("adapter")
assert adapter == "n2yo_visualpasses"
assert adapter in ("n2yo_visualpasses", "satpass_predict")
def test_satpass_handler_accepts_satpass_predict_envelope():
"""satpass_predict must not be rejected by the adapter guard."""
inner = SATPASS_PREDICT_ENVELOPE["data"]
adapter = inner.get("adapter")
assert adapter == "satpass_predict"
assert adapter in ("n2yo_visualpasses", "satpass_predict")
def test_satpass_handler_rejects_wrong_adapter():
"""An envelope with adapter='celestrak_tle' must be rejected."""
from meshai.central.satpass_handler import handle_satpass
if hasattr(handle_satpass, "_disabled_logged"):
del handle_satpass._disabled_logged
result = handle_satpass(CELESTRAK_TLE_ENVELOPE, "central.sat.tle.25544")
assert result is None
# ── Config key consistency ──────────────────────────────────────────
def test_registry_has_min_elevation_not_deg():
"""The REGISTRY key must be 'min_elevation', not 'min_elevation_deg'."""
from meshai.adapter_config.defaults import REGISTRY
assert ("satpass", "min_elevation") in REGISTRY
assert ("satpass", "min_elevation_deg") not in REGISTRY
def test_handler_reads_min_elevation():
"""satpass_handler must read cfg.min_elevation (matching REGISTRY key)."""
import inspect
from meshai.central import satpass_handler
src = inspect.getsource(satpass_handler)
assert "min_elevation" in src
assert "min_elevation_deg" not in src

262
work/tests/test_satpass_wire_fields.py
View file

@ -1,262 +0,0 @@
"""Tests for satpass_handler wire field name reads.
Uses the verbatim live NOAA-18 envelope captured from Central NATS
(central.sat.pass.us.id.filer, 2026-06-10). Proves:
1. Handler extracts correct norad_id, satellite_name, observer_name,
max_elevation_deg, aos_time, los_time from the actual wire format.
2. satpass_events row is inserted with correct values.
3. Envelope missing norad_id is rejected (returns None).
4. Wire message format includes the correct extracted values.
5. Category mapping: pass.n2yo_visualpasses -> sat_pass.
"""
from __future__ import annotations
import json
import time
import pytest
# ── Verbatim live NOAA-18 envelope from Central NATS ────────────────
NOAA18_ENVELOPE = {
"id": "filer:28654:2026-06-10T04:34:40+00:00",
"source": "central.echo6.co",
"type": "central.pass.n2yo_visualpasses.v1",
"time": "2026-06-10T04:41:35+00:00",
"datacontenttype": "application/json",
"centralschemaversion": "1.0",
"centralcategory": "pass.n2yo_visualpasses",
"centralseverity": 1,
"specversion": "1.0",
"data": {
"id": "filer:28654:2026-06-10T04:34:40+00:00",
"adapter": "n2yo_visualpasses",
"category": "pass.n2yo_visualpasses",
"time": "2026-06-10T04:41:35Z",
"expires": None,
"severity": 1,
"geo": {
"centroid": [-114.6, 42.57],
"bbox": None,
"regions": ["US-ID"],
"primary_region": "US-ID",
"geometry": None,
},
"data": {
"observer_name": "Filer",
"observer_slug": "filer",
"observer_state": "ID",
"norad_id": 28654,
"satellite_name": "NOAA 18",
"aos_time": "2026-06-10T04:34:40+00:00",
"peak_time": "2026-06-10T04:41:35+00:00",
"los_time": "2026-06-10T04:48:30+00:00",
"max_elevation_deg": 22.69,
"magnitude": 6.7,
"azimuth_at_aos": 125.6,
"azimuth_at_aos_compass": "SE",
"azimuth_at_peak": 63.0,
"azimuth_at_peak_compass": "ENE",
"azimuth_at_los": 359.5,
"azimuth_at_los_compass": "N",
"duration_s": 630,
},
},
}
def _enable_satpass(norad_ids=None):
"""Set satpass.enabled=true in the test DB."""
from meshai.persistence import get_db
from meshai.adapter_config import invalidate_cache
conn = get_db()
conn.execute(
"UPDATE adapter_config SET value_json='true' "
"WHERE adapter='satpass' AND key='enabled'"
)
# Set min_elevation low enough to accept this 22.69 deg pass
conn.execute(
"UPDATE adapter_config SET value_json='5' "
"WHERE adapter='satpass' AND key='min_elevation'"
)
# Disable dry_run for tests that expect wire output
conn.execute(
"UPDATE adapter_config SET value_json='false' "
"WHERE adapter='satpass' AND key='dry_run'"
)
# Set norad_ids (must be non-empty for opt-in)
if norad_ids is None:
norad_ids = [28654]
conn.execute(
"UPDATE adapter_config SET value_json=? "
"WHERE adapter='satpass' AND key='norad_ids'",
(json.dumps(norad_ids),)
)
invalidate_cache()
def _ingest_and_consolidate(env, subject, *, now, data=None):
"""Drive the two-call async satpass contract (ingest -> consolidate).
handle_satpass() ingests the pass and returns None; the consumer then
runs consolidate_satpass_pending(), which returns the (wire, data) to
broadcast (or None). Returns the consolidation result.
"""
from meshai.central.satpass_handler import (
handle_satpass, consolidate_satpass_pending,
drain_pending_consolidation_ids)
drain_pending_consolidation_ids()
assert handle_satpass(
env, subject, data=data if data is not None else {}, now=now) is None
for cid in drain_pending_consolidation_ids():
res = consolidate_satpass_pending(cid)
if res is not None:
return res
return None
# ── Handler produces correct satpass_events row ─────────────────────
def test_noaa18_envelope_produces_satpass_event():
"""Verbatim NOAA-18 envelope inserts row with correct field values."""
from meshai.central.satpass_handler import handle_satpass
from meshai.persistence import get_db
_enable_satpass()
if hasattr(handle_satpass, "_disabled_logged"):
del handle_satpass._disabled_logged
if hasattr(handle_satpass, "_no_norad_ids_logged"):
del handle_satpass._no_norad_ids_logged
now = 1781065800 # before NOAA18 envelope los_time
result = _ingest_and_consolidate(
NOAA18_ENVELOPE,
"central.sat.pass.us.id.filer",
now=now,
)
assert result is not None, "handler returned None -- field extraction failed"
wire, _ = result
conn = get_db()
rows = conn.execute(
"SELECT norad_id, sat_name, observer, max_elevation, aos_at, los_at "
"FROM satpass_events WHERE norad_id=28654"
).fetchall()
assert len(rows) >= 1, "no satpass_events row for norad_id=28654"
row = rows[0]
assert row["norad_id"] == 28654
assert row["sat_name"] == "NOAA 18"
assert row["observer"] == "Filer"
assert abs(row["max_elevation"] - 22.69) < 0.01
# aos_time = 2026-06-10T04:34:40+00:00 -> epoch
assert row["aos_at"] is not None
assert row["los_at"] is not None
# los must be after aos
assert row["los_at"] > row["aos_at"]
def test_noaa18_wire_message_format():
"""Wire message includes satellite name, direction in new 2-line format."""
from meshai.central.satpass_handler import handle_satpass
_enable_satpass()
if hasattr(handle_satpass, "_disabled_logged"):
del handle_satpass._disabled_logged
if hasattr(handle_satpass, "_no_norad_ids_logged"):
del handle_satpass._no_norad_ids_logged
result = _ingest_and_consolidate(
NOAA18_ENVELOPE,
"central.sat.pass.us.id.filer",
now=1781065800, # before NOAA18 envelope los_time
)
assert result is not None
wire, _ = result
# Single clean line: name, numeric elevation, aos->peak->los compass sweep.
assert "\n" not in wire, f"Expected single line: {wire!r}"
assert "NOAA 18" in wire # no mapping -> cleaned catalog name
assert "max 23°" in wire # 22.69 rounds to 23, not "low pass"
assert "low pass" not in wire
assert "min window" not in wire
assert "SE→ENE→N" in wire # aos -> peak -> los
def test_missing_norad_id_rejected():
"""Envelope with norad_id removed returns None."""
from meshai.central.satpass_handler import handle_satpass
_enable_satpass()
if hasattr(handle_satpass, "_disabled_logged"):
del handle_satpass._disabled_logged
if hasattr(handle_satpass, "_no_norad_ids_logged"):
del handle_satpass._no_norad_ids_logged
# Deep copy and remove norad_id
import copy
env = copy.deepcopy(NOAA18_ENVELOPE)
del env["data"]["data"]["norad_id"]
wire = handle_satpass(
env,
"central.sat.pass.us.id.filer",
data={},
now=1781065800, # before NOAA18 envelope los_time
)
assert wire is None, "handler should reject envelope without norad_id"
def test_missing_max_elevation_deg_rejected():
"""Envelope with max_elevation_deg removed returns None."""
from meshai.central.satpass_handler import handle_satpass
_enable_satpass()
if hasattr(handle_satpass, "_disabled_logged"):
del handle_satpass._disabled_logged
if hasattr(handle_satpass, "_no_norad_ids_logged"):
del handle_satpass._no_norad_ids_logged
import copy
env = copy.deepcopy(NOAA18_ENVELOPE)
del env["data"]["data"]["max_elevation_deg"]
wire = handle_satpass(
env,
"central.sat.pass.us.id.filer",
data={},
now=1781065800, # before NOAA18 envelope los_time
)
assert wire is None, "handler should reject envelope without max_elevation_deg"
def test_observer_fallback_to_slug():
"""When observer_name is absent, falls back to observer_slug."""
from meshai.central.satpass_handler import handle_satpass
_enable_satpass()
if hasattr(handle_satpass, "_disabled_logged"):
del handle_satpass._disabled_logged
if hasattr(handle_satpass, "_no_norad_ids_logged"):
del handle_satpass._no_norad_ids_logged
import copy
env = copy.deepcopy(NOAA18_ENVELOPE)
del env["data"]["data"]["observer_name"]
# observer_slug = "filer" still present
result = _ingest_and_consolidate(
env,
"central.sat.pass.us.id.filer",
now=1781065800, # before NOAA18 envelope los_time
)
assert result is not None
# Observer name stored in DB, not in broadcast wire format
from meshai.persistence import get_db
conn = get_db()
row = conn.execute(
"SELECT observer FROM satpass_events WHERE norad_id=28654"
).fetchone()
assert row is not None
assert row["observer"] == "filer"