From 53eadf5135952b5174a7392d46704b4a5bd70a4e Mon Sep 17 00:00:00 2001 From: malice Date: Sun, 5 Jul 2026 02:25:55 -0600 Subject: [PATCH] =?UTF-8?q?feat(phase4c):=20native=20SGP4=20satpass=20?= =?UTF-8?q?=E2=80=94=20standalone=20satellite=20passes,=20no=20Central=20(?= =?UTF-8?q?#39)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit meshai can now predict + broadcast satellite passes locally without Central. Data plumbing (4c-1): - env/tle_fetch.py: keyless Celestrak GP fetcher (GROUP/CATNR, FORMAT=tle) → upserts the existing sat_tles table via a shared upsert_tle() helper extracted into tle_handler (Central ingest refactored to call it, unchanged) - observer_locations table (v23, SCHEMA_VERSION 22->23) + persistence helpers; seeded from SatpassConfig.observers in main._init_components - SatpassConfig: observers, tle_groups, norad_ids, tle_refresh_seconds, min_elevation_deg, window_hours Predictor + source-agnostic gate (4c-2): - extracted gate_consolidated_pass(consolidated, *, now) from consolidate_satpass_pending: dedup-vs-satpass_events + rate cap + format_pass + deferred commit. Central path byte-identical (114 tests unchanged) - env/satpass.py: native adapter predicts passes for each sat x observer via pass_predictor.compute_passes, consolidates IN-MEMORY per canonical hour bucket (earliest AOS / latest LOS / max-el observer supplies peak_compass + entry/exit observers), runs the shared gate, emits sat_pass. Commit rides event.data so satpass_events dedups across ticks — NO satpass_pending, NO Central-consumer timer dependency (works with Central off) - registered in env/store.py gated on enabled and feed_source==native 32 new tests (tle_fetch 16, observer_locations 14... satpass_native 8, minus overlaps); full suite 10-failure baseline (1672 passed). Co-authored-by: Matt Johnson Co-authored-by: Claude Opus 4.8 (1M context) --- work/meshai/central/satpass_handler.py | 175 ++++++---- work/meshai/central/tle_handler.py | 34 +- work/meshai/config.py | 22 +- work/meshai/env/satpass.py | 285 ++++++++++++++++ work/meshai/env/store.py | 11 +- work/meshai/env/tle_fetch.py | 230 +++++++++++++ work/meshai/main.py | 8 + work/meshai/persistence/db.py | 2 +- work/meshai/persistence/migrations/v23.sql | 16 + work/meshai/persistence/observer_locations.py | 88 +++++ work/tests/test_observer_locations.py | 126 +++++++ work/tests/test_satpass_native.py | 307 ++++++++++++++++++ work/tests/test_satpass_persisted_timer.py | 9 +- work/tests/test_tle_fetch.py | 187 +++++++++++ 14 files changed, 1426 insertions(+), 74 deletions(-) create mode 100644 work/meshai/env/satpass.py create mode 100644 work/meshai/env/tle_fetch.py create mode 100644 work/meshai/persistence/migrations/v23.sql create mode 100644 work/meshai/persistence/observer_locations.py create mode 100644 work/tests/test_observer_locations.py create mode 100644 work/tests/test_satpass_native.py create mode 100644 work/tests/test_tle_fetch.py diff --git a/work/meshai/central/satpass_handler.py b/work/meshai/central/satpass_handler.py index b62a91a..efdf546 100644 --- a/work/meshai/central/satpass_handler.py +++ b/work/meshai/central/satpass_handler.py @@ -438,9 +438,106 @@ def handle_satpass(envelope: dict, subject: str, return None +def gate_consolidated_pass(consolidated: dict, *, + now: int) -> tuple[str, dict] | None: + """Source-agnostic broadcast gate for an already-consolidated pass. + + Both the Central consumer path (`consolidate_satpass_pending`, which + merges buffered per-observer rows from `satpass_pending`) and the native + env.satpass adapter (which consolidates in-memory across observers in one + tick) call THIS single function so the broadcast decision is byte-identical + regardless of source. It deliberately does NOT touch `satpass_pending` — + that buffer is a Central-consumer implementation detail owned by the caller. + + `consolidated` is a dict describing one merged pass with keys: + consolidated_id (str, = {norad}:{aos_epoch//3600}), + norad_id, sat_name, max_elevation, + aos_epoch, los_epoch (int epoch seconds), + aos_compass, los_compass, peak_compass, + entry_observer, exit_observer, + observer_list (comma-joined observer slugs for the audit column). + + Applies, in order: dedup-vs-`satpass_events`, rate cap, wire build, + dry-run gate, `satpass_events` upsert, and the deferred `_attach_commit` + that upserts last/first_broadcast_at on delivery. Returns (wire, data) to + broadcast, or None if suppressed. `now` is threaded explicitly for + determinism (rate-cap window + first_seen_at). + """ + try: + conn = get_db() + except Exception: + logger.exception("satpass gate: persistence unavailable") + return None + + cfg = adapter_config.satpass + + consolidated_id = consolidated["consolidated_id"] + norad_id = consolidated["norad_id"] + sat_name = consolidated["sat_name"] + max_el = consolidated["max_elevation"] + aos_epoch = consolidated["aos_epoch"] + los_epoch = consolidated["los_epoch"] + aos_compass = consolidated["aos_compass"] + los_compass = consolidated["los_compass"] + peak_compass = consolidated.get("peak_compass") + entry_obs = consolidated.get("entry_observer") + exit_obs = consolidated.get("exit_observer") + observer_list = consolidated.get("observer_list") or (entry_obs or "") + + # Dedup against satpass_events + existing = conn.execute( + "SELECT last_broadcast_at FROM satpass_events WHERE event_id=?", + (consolidated_id,)).fetchone() + if existing and existing["last_broadcast_at"] is not None: + return None + + # Rate cap + max_per_hour = int(getattr(cfg, "max_broadcasts_per_hour", 4)) + allowed, count = _check_rate_cap(conn, now, max_per_hour) + if not allowed: + logger.info("satpass: rate cap reached (%d/%d), suppressing consolidated pass %s", + count, max_per_hour, consolidated_id) + return None + + # Build consolidated wire — always pass observer names for region context + wire = format_pass(sat_name=sat_name, max_el=max_el, + aos_epoch=aos_epoch, los_epoch=los_epoch, + aos_compass=aos_compass, los_compass=los_compass, + peak_compass=peak_compass, + entry_observer=entry_obs, exit_observer=exit_obs) + + # Dry-run gate + dry_run = getattr(cfg, "dry_run", True) + if dry_run: + logger.info("DRY-RUN would air (consolidated): %s", wire) + return None + + # Upsert consolidated record into satpass_events + _upsert_satpass(conn, event_id=consolidated_id, norad_id=norad_id, + sat_name=sat_name, observer=observer_list, + max_elevation=max_el, aos_at=aos_epoch, + los_at=los_epoch, payload_json=None, + first_seen_at=now, set_last_broadcast=False) + + # Prepare data dict with callbacks + severity_word = _map_severity(max_el) + data = {"_meshai_precomposed": True, "_severity_override": severity_word} + _attach_commit(data, event_id=consolidated_id, event_log_row_id=None) + + return wire, data + + def consolidate_satpass_pending(consolidated_id: str) -> tuple[str, dict] | None: """Called by consumer when 5s consolidation timer fires. + Reads the buffered per-observer rows for this canonical id, merges them + across observers (earliest AOS / latest LOS / max-elevation observer + supplies max_elevation + peak_compass / entry+exit observers), then + delegates the actual broadcast decision to the shared, source-agnostic + `gate_consolidated_pass`. Pending rows are cleaned up afterward regardless + of the gate's decision (dedup, rate-cap, dry-run, and success all consume + the buffer identically, as before). + Returns (wire_string, data_dict) or None if suppressed. """ try: @@ -455,9 +552,6 @@ def consolidate_satpass_pending(consolidated_id: str) -> tuple[str, dict] | None if not rows: return None - cfg = adapter_config.satpass - now = _now() - # Consolidate observers sorted_by_aos = sorted(rows, key=lambda r: r["aos_at"]) sorted_by_los = sorted(rows, key=lambda r: r["los_at"]) @@ -465,67 +559,28 @@ def consolidate_satpass_pending(consolidated_id: str) -> tuple[str, dict] | None exit_ = sorted_by_los[-1] # latest LOS best = max(rows, key=lambda r: r["max_elevation"]) - norad_id = best["norad_id"] - sat_name = best["sat_name"] - max_el = best["max_elevation"] - aos_epoch = entry["aos_at"] - los_epoch = exit_["los_at"] - aos_compass = entry["aos_compass"] - los_compass = exit_["los_compass"] - # Peak belongs to whoever saw the highest elevation. - peak_compass = best["peak_compass"] - entry_obs = entry["observer"] - exit_obs = exit_["observer"] + consolidated = { + "consolidated_id": consolidated_id, + "norad_id": best["norad_id"], + "sat_name": best["sat_name"], + "max_elevation": best["max_elevation"], + "aos_epoch": entry["aos_at"], + "los_epoch": exit_["los_at"], + "aos_compass": entry["aos_compass"], + "los_compass": exit_["los_compass"], + # Peak belongs to whoever saw the highest elevation. + "peak_compass": best["peak_compass"], + "entry_observer": entry["observer"], + "exit_observer": exit_["observer"], + "observer_list": ",".join(r["observer"] for r in sorted_by_aos), + } - # Dedup against satpass_events - existing = conn.execute( - "SELECT last_broadcast_at FROM satpass_events WHERE event_id=?", - (consolidated_id,)).fetchone() - if existing and existing["last_broadcast_at"] is not None: - _cleanup_pending(conn, consolidated_id) - return None + result = gate_consolidated_pass(consolidated, now=_now()) - # Rate cap - max_per_hour = int(getattr(cfg, "max_broadcasts_per_hour", 4)) - allowed, count = _check_rate_cap(conn, now, max_per_hour) - if not allowed: - logger.info("satpass: rate cap reached (%d/%d), suppressing consolidated pass %s", - count, max_per_hour, consolidated_id) - _cleanup_pending(conn, consolidated_id) - return None - - # Build consolidated wire — always pass observer names for region context - wire = format_pass(sat_name=sat_name, max_el=max_el, - aos_epoch=aos_epoch, los_epoch=los_epoch, - aos_compass=aos_compass, los_compass=los_compass, - peak_compass=peak_compass, - entry_observer=entry_obs, exit_observer=exit_obs) - - # Dry-run gate - dry_run = getattr(cfg, "dry_run", True) - if dry_run: - logger.info("DRY-RUN would air (consolidated, %d observers): %s", - len(rows), wire) - _cleanup_pending(conn, consolidated_id) - return None - - # Upsert consolidated record into satpass_events - observer_list = ",".join(r["observer"] for r in sorted_by_aos) - _upsert_satpass(conn, event_id=consolidated_id, norad_id=norad_id, - sat_name=sat_name, observer=observer_list, - max_elevation=max_el, aos_at=aos_epoch, - los_at=los_epoch, payload_json=None, - first_seen_at=now, set_last_broadcast=False) - - # Clean up pending rows + # Clean up pending rows regardless of the gate's decision. _cleanup_pending(conn, consolidated_id) - # Prepare data dict with callbacks - severity_word = _map_severity(max_el) - data = {"_meshai_precomposed": True, "_severity_override": severity_word} - _attach_commit(data, event_id=consolidated_id, event_log_row_id=None) - - return wire, data + return result def _upsert_satpass(conn, *, event_id, norad_id, sat_name, observer, diff --git a/work/meshai/central/tle_handler.py b/work/meshai/central/tle_handler.py index b0fc5bf..742c150 100644 --- a/work/meshai/central/tle_handler.py +++ b/work/meshai/central/tle_handler.py @@ -73,15 +73,36 @@ def handle_tle(envelope: dict, subject: str, logger.exception("tle_handler: persistence unavailable") return None - # Upsert: latest-wins on epoch + upsert_tle(conn, norad_id, name, line1, line2, epoch, now=now) + + return None # storage-only, never broadcast + + +def upsert_tle(conn, norad_id: int, name: str, line1: str, line2: str, + epoch, now: Optional[int] = None) -> bool: + """Upsert one TLE into sat_tles with latest-epoch-wins semantics. + + Shared by BOTH ingest paths — the Central envelope handler + (`handle_tle`) and the native Celestrak fetcher (`env.tle_fetch`) — so + the predictor reads TLEs identically regardless of source. `epoch` is a + lexicographically-sortable string (ISO 8601 for Central, ISO 8601 + derived from the TLE line-1 epoch field for the native fetch); a cached + row whose epoch is >= the incoming epoch is left untouched. + + Returns True if a row was written (insert or update), False if the + cached epoch was same-or-newer and the write was skipped. + """ + now = now if now is not None else int(time.time()) + epoch = str(epoch) + existing = conn.execute( "SELECT epoch FROM sat_tles WHERE norad_id = ?", (norad_id,), ).fetchone() - if existing is not None and existing["epoch"] >= str(epoch): - # Cached epoch is same or newer — skip - return None + if existing is not None and existing["epoch"] >= epoch: + # Cached epoch is same or newer — skip. + return False conn.execute( "INSERT INTO sat_tles(norad_id, name, line1, line2, epoch, updated_at) " @@ -89,10 +110,9 @@ def handle_tle(envelope: dict, subject: str, "ON CONFLICT(norad_id) DO UPDATE SET " "name=excluded.name, line1=excluded.line1, line2=excluded.line2, " "epoch=excluded.epoch, updated_at=excluded.updated_at", - (norad_id, name, line1, line2, str(epoch), now), + (norad_id, name, line1, line2, epoch, now), ) - - return None # storage-only, never broadcast + return True def get_fresh_tles(conn=None, max_age_days: int = STALE_DAYS) -> list[dict]: diff --git a/work/meshai/config.py b/work/meshai/config.py index 9b9eff0..c2ca788 100644 --- a/work/meshai/config.py +++ b/work/meshai/config.py @@ -489,11 +489,31 @@ class FIRMSConfig(_SourcedFeed): @dataclass class SatpassConfig(_SourcedFeed): - """Satellite pass prediction settings (central-only feed).""" + """Satellite pass prediction settings. + + Historically a Central-only feed (`feed_source="central"`). The native + path (`feed_source="native"`) adds a Celestrak TLE fetcher + (`env.tle_fetch`) and a native SGP4 predictor; the fields below feed + those. `feed_source` default stays "central" — the flip to "native" + happens at cutover, not here. + """ enabled: bool = False feed_source: str = "central" + # -- native path (TLE fetch + SGP4 predictor) ----------------------------- + # Ground stations the predictor computes passes for; each entry is a + # dict {slug, name, lat, lon, alt_m}. Seeded into observer_locations. + observers: list = field(default_factory=list) + # Celestrak GP selectors the fetcher pulls (env.tle_fetch): + tle_groups: list = field(default_factory=lambda: ["weather", "stations"]) + norad_ids: list = field(default_factory=list) + # TLE refresh cadence — TLEs update ~daily, so poll every 6h. + tle_refresh_seconds: int = 21600 + # Predictor pass filters (used by the next task): + min_elevation_deg: float = 10.0 + window_hours: int = 24 + @dataclass class CentralConsumerConfig: diff --git a/work/meshai/env/satpass.py b/work/meshai/env/satpass.py new file mode 100644 index 0000000..b149a03 --- /dev/null +++ b/work/meshai/env/satpass.py @@ -0,0 +1,285 @@ +"""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. + +Flow per tick (slow poll, ~15 min; passes are computed over `window_hours`): + 1. Resolve target satellites from the shared `sat_tles` table (populated + by env.tle_fetch). Resolution rule: + - `norad_ids` configured -> exactly those NORAD ids, fresh only. + - `norad_ids` empty -> ALL fresh TLEs (which are precisely the + sats env.tle_fetch pulled from the configured `tle_groups` + + `norad_ids`). This differs from the CENTRAL broadcast filter, where + empty norad_ids means "broadcast nothing" — here empty means + "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`). + 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 + `satpass_events`, applies the rate cap + dry-run, upserts the event + row, and attaches the deferred commit. Staged results feed + `get_events()` / `to_event()`. + +Dedup persists across ticks through `satpass_events`: the gate upserts the row +(with last_broadcast_at NULL) at stage time, and the commit closure — carried +on the emitted Event's `data["_on_broadcast_committed"]` and fired by the +dispatcher after a successful send — sets last_broadcast_at. The next tick's +gate sees that timestamp and suppresses the same canonical pass. + +Resilient: no observers / no fresh TLEs / a bad TLE ⇒ log + skip, update +health, never crash. +""" + +from __future__ import annotations + +import logging +import time +from datetime import datetime, timezone +from typing import TYPE_CHECKING, Optional + +from meshai.notifications.events import Event, make_event + +if TYPE_CHECKING: + from ..config import SatpassConfig + +logger = logging.getLogger(__name__) + +# Slow poll cadence. Passes are computed over `window_hours`, so re-running +# more often than this only re-confirms already-gated passes (cheap, but +# pointless). Overridable via SatpassConfig.pass_refresh_seconds if present. +DEFAULT_POLL_SECONDS = 900 + + +class SatpassAdapter: + """Native SGP4 pass predictor — consolidates in-memory, gates synchronously.""" + + def __init__(self, config: "SatpassConfig"): + self._config = config + self._interval = int( + getattr(config, "pass_refresh_seconds", DEFAULT_POLL_SECONDS) + or DEFAULT_POLL_SECONDS) + self._window_h = int(getattr(config, "window_hours", 24) or 24) + self._min_el = float(getattr(config, "min_elevation_deg", 10.0)) + self._norad_ids = self._parse_norad_ids( + getattr(config, "norad_ids", None) or []) + + self._last_tick = 0.0 + self._last_error: Optional[str] = None + self._consecutive_errors = 0 + self._is_loaded = False + self._events: list[dict] = [] # staged pass events for get_events() + + # -- helpers -------------------------------------------------------------- + + @staticmethod + def _parse_norad_ids(raw) -> list[int]: + """Coerce a config norad_ids list (ints or GUI strings) to int list.""" + return sorted({int(x) for x in raw if str(x).strip().isdigit()}) + + def _resolve_tles(self) -> list[dict]: + """Resolve the target TLE set from `sat_tles` (fresh only). + + `norad_ids` configured -> exactly those; empty -> all fresh. + """ + from meshai.central.tle_handler import get_fresh_tles, get_tle_by_norad + + if self._norad_ids: + out: list[dict] = [] + for nid in self._norad_ids: + tle = get_tle_by_norad(nid) + if tle is not None: + out.append(tle) + return out + return get_fresh_tles() + + @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. + """ + by_aos = sorted(recs, key=lambda r: r["aos_epoch"]) + by_los = sorted(recs, key=lambda r: r["los_epoch"]) + entry = by_aos[0] + exit_ = by_los[-1] + best = max(recs, key=lambda r: r["max_elevation"]) + return { + "consolidated_id": cid, + "norad_id": best["norad_id"], + "sat_name": best["sat_name"], + "max_elevation": best["max_elevation"], + "aos_epoch": entry["aos_epoch"], + "los_epoch": exit_["los_epoch"], + "aos_compass": entry["aos_compass"], + "los_compass": exit_["los_compass"], + "peak_compass": best["peak_compass"], + "entry_observer": entry["observer"], + "exit_observer": exit_["observer"], + "observer_list": ",".join(r["observer"] for r in by_aos), + } + + 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.persistence.observer_locations import get_observers + + observers = get_observers() + if not observers: + logger.debug("satpass native: no observers configured; skipping") + return [] + + tles = self._resolve_tles() + if not tles: + logger.debug("satpass native: no fresh TLEs available; skipping") + return [] + + now_dt = datetime.fromtimestamp(now_epoch, tz=timezone.utc) + + # canonical_id -> [per-observer record] + groups: dict[str, list[dict]] = {} + for tle in tles: + norad = tle["norad_id"] + name = tle["name"] + l1 = tle["line1"] + l2 = tle["line2"] + for obs in observers: + try: + passes = compute_passes( + l1, l2, obs["lat"], obs["lon"], + obs.get("alt_m", 0.0) or 0.0, + self._window_h, self._min_el, now_dt) + except Exception as e: + logger.warning( + "satpass native: compute_passes failed for NORAD %s " + "obs %s: %s", norad, obs.get("slug"), e) + continue + for p in passes: + aos_epoch = int(p.aos_time.timestamp()) + los_epoch = int(p.los_time.timestamp()) + cid = sh._canonical_id(norad, aos_epoch) + groups.setdefault(cid, []).append({ + "norad_id": norad, + "sat_name": name, + "observer": obs["slug"], + "max_elevation": p.max_elevation, + "aos_epoch": aos_epoch, + "los_epoch": los_epoch, + "aos_compass": sh._azimuth_to_compass(p.azimuth_at_aos), + "los_compass": sh._azimuth_to_compass(p.azimuth_at_los), + "peak_compass": sh._azimuth_to_compass(p.azimuth_at_peak), + }) + + staged: list[dict] = [] + for cid, recs in groups.items(): + consolidated = self._consolidate(cid, recs) + try: + result = sh.gate_consolidated_pass(consolidated, now=now_epoch) + except Exception as e: + logger.warning("satpass native: gate failed for %s: %s", cid, e) + continue + if result is None: + continue + wire, data = result + staged.append({ + "source": "satpass", + "event_id": cid, + "wire": wire, + "data": data, + "severity": data.get("_severity_override", "routine"), + "sat_name": consolidated["sat_name"], + "norad_id": consolidated["norad_id"], + "aos_epoch": consolidated["aos_epoch"], + "los_epoch": consolidated["los_epoch"], + # Purge from the store's in-memory map once the pass is over. + "expires": consolidated["los_epoch"], + "fetched_at": now_epoch, + }) + return staged + + # -- polling -------------------------------------------------------------- + + def tick(self, now: Optional[float] = None) -> bool: + """One slow prediction pass. Returns True if any pass was staged. + + Resilient: any failure logs + updates health and returns False without + propagating out of tick(). + """ + now = now if now is not None else time.time() + if now - self._last_tick < self._interval: + return False + self._last_tick = now + + try: + staged = self._compute_staged(int(now)) + except Exception as e: + self._last_error = str(e) + self._consecutive_errors += 1 + logger.warning("satpass native: tick failed: %s", e) + self._is_loaded = True + return False + + self._events = staged + self._last_error = None + self._consecutive_errors = 0 + self._is_loaded = True + return bool(staged) + + def get_events(self) -> list: + """Return staged consolidated pass events (dicts).""" + return list(self._events) + + def to_event(self, evt: dict) -> Optional["Event"]: + """Translate a staged pass into a precomposed pipeline Event. + + The gate already rendered the LoRa wire (aos→peak→los) and attached the + `_on_broadcast_committed` closure onto `data`; passing that same dict as + the Event's `data` makes the commit ride to the dispatcher, which fires + it after a successful send so `satpass_events.last_broadcast_at` is set + and the pass won't re-broadcast next tick. + """ + try: + wire = evt.get("wire") + data = evt.get("data") + if not wire or data is None: + return None + event_id = evt.get("event_id") + return make_event( + source="satpass", + category="sat_pass", + severity=evt.get("severity", "routine"), + title=wire, # precomposed: composer passes it verbatim + summary=wire, + timestamp=evt.get("fetched_at"), + expires=evt.get("expires"), + group_key=event_id, + inhibit_keys=[event_id] if event_id else [], + data=data, # carries _meshai_precomposed + commit hook + ) + except Exception: + logger.exception("satpass native: to_event failed") + return None + + @property + def health_status(self) -> dict: + return { + "source": "satpass", + "is_loaded": self._is_loaded, + "last_error": self._last_error, + "consecutive_errors": self._consecutive_errors, + "event_count": len(self._events), + "last_fetch": self._last_tick, + } diff --git a/work/meshai/env/store.py b/work/meshai/env/store.py index 51d5011..74377b2 100644 --- a/work/meshai/env/store.py +++ b/work/meshai/env/store.py @@ -50,6 +50,15 @@ class EnvironmentalStore: lambda cfg: (cfg,)) self._register_adapter("wzdx", config.wzdx, ".wzdx", "WZDxAdapter", lambda cfg: (cfg,)) + # Native satpass TLE fetcher (storage-only: populates sat_tles, emits + # no events). Gated on satpass.feed_source=="native" like the rest. + self._register_adapter("satpass_tle", config.satpass, ".tle_fetch", "TLEFetchAdapter", + lambda cfg: (cfg,)) + # Native SGP4 pass predictor (broadcasts consolidated passes locally, + # no Central dependency). SEPARATE from the satpass_tle fetcher above; + # both are gated on satpass.enabled and feed_source=="native". + self._register_adapter("satpass", config.satpass, ".satpass", "SatpassAdapter", + lambda cfg: (cfg,)) # FIRMS needs reference to NIFC adapter for cross-referencing if config.firms.enabled and config.firms.feed_source == "native": @@ -63,7 +72,7 @@ class EnvironmentalStore: logger.warning("Failed to initialize firms adapter: %s", err_msg) self._failed_adapters["firms"] = err_msg - _central = [n for n in ("nws", "swpc", "ducting", "fires", "avalanche", "usgs", "usgs_quake", "traffic", "roads511", "wzdx", "firms") + _central = [n for n in ("nws", "swpc", "ducting", "fires", "avalanche", "usgs", "usgs_quake", "traffic", "roads511", "wzdx", "firms", "satpass") if getattr(getattr(config, n, None), "feed_source", "native") == "central"] if _central: logger.debug("Adapters sourced from Central (native skipped): %s", _central) diff --git a/work/meshai/env/tle_fetch.py b/work/meshai/env/tle_fetch.py new file mode 100644 index 0000000..4477242 --- /dev/null +++ b/work/meshai/env/tle_fetch.py @@ -0,0 +1,230 @@ +"""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". + +Source: Celestrak GP API (https://celestrak.org/NORAD/elements/gp.php), +FORMAT=tle (classic 3-line: name / line1 / line2). Two selector styles: + - GROUP (e.g. ?GROUP=weather&FORMAT=tle) — a curated catalog group + - CATNR (e.g. ?CATNR=25544&FORMAT=tle) — a single NORAD id + +Config (SatpassConfig): `tle_groups` (list of group names), `norad_ids` +(list of NORAD catalog ids), `tle_refresh_seconds` (poll interval; TLEs +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. +""" + +from __future__ import annotations + +import datetime +import logging +import time +from typing import TYPE_CHECKING, Optional +from urllib.error import HTTPError, URLError +from urllib.request import Request, urlopen + +if TYPE_CHECKING: + from ..config import SatpassConfig + +logger = logging.getLogger(__name__) + +GP_BASE_URL = "https://celestrak.org/NORAD/elements/gp.php" + + +def parse_tle_epoch(line1: str) -> str: + """Parse the epoch from TLE line 1 into an ISO-8601 UTC string. + + The epoch occupies columns 19-32 (1-indexed) of line 1 in the form + ``YYDDD.DDDDDDDD``: a two-digit year, day-of-year, and fractional day. + Two-digit years 57-99 map to 1957-1999; 00-56 map to 2000-2056 (the + standard NORAD windowing). + + Returns an ISO-8601 string (e.g. "2026-07-01T12:00:00+00:00") so the + value is lexicographically comparable with the ISO epochs the Central + handler stores. Raises ValueError on a malformed field. + """ + raw = line1[18:32].strip() + yy = int(raw[0:2]) + year = 2000 + yy if yy < 57 else 1900 + yy + doy = float(raw[2:]) + if doy < 1: + raise ValueError(f"bad day-of-year in TLE epoch: {raw!r}") + dt = (datetime.datetime(year, 1, 1, tzinfo=datetime.timezone.utc) + + datetime.timedelta(days=doy - 1.0)) + return dt.isoformat() + + +def parse_tle_block(text: str) -> list[dict]: + """Parse a 3-line-per-satellite TLE block into records. + + Accepts the Celestrak FORMAT=tle payload: repeating triples of + (name, line1, line2). Tolerant of blank lines and trailing whitespace. + A malformed triple (missing/short lines, non-integer NORAD id, or an + unparseable epoch) is skipped with a debug log rather than aborting the + whole block. + + Returns a list of dicts: {norad_id, name, line1, line2, epoch}. + """ + lines = [ln.rstrip() for ln in text.splitlines() if ln.strip()] + out: list[dict] = [] + i = 0 + while i + 2 <= len(lines) - 1: + # Three lines (name, line1, line2) available at i, i+1, i+2. + name = lines[i].strip() + line1 = lines[i + 1] + line2 = lines[i + 2] + + # Validate structural markers before consuming the triple. + if not (line1.startswith("1 ") and line2.startswith("2 ")): + # Not aligned to a triple boundary — skip one line and resync. + i += 1 + continue + + try: + norad_id = int(line1[2:7]) + epoch = parse_tle_epoch(line1) + except (ValueError, IndexError) as e: + logger.debug("tle_fetch: skipping malformed TLE for %r: %s", name, e) + i += 3 + continue + + out.append({ + "norad_id": norad_id, + "name": name or f"SAT-{norad_id}", + "line1": line1, + "line2": line2, + "epoch": epoch, + }) + i += 3 + + return out + + +class TLEFetchAdapter: + """Native Celestrak TLE fetcher — populates sat_tles, emits no events.""" + + def __init__(self, config: "SatpassConfig"): + self._config = config + self._last_tick = 0.0 + self._last_error: Optional[str] = None + self._consecutive_errors = 0 + self._is_loaded = False + self._upserted = 0 # rows written on the most recent successful tick + self._interval = int(getattr(config, "tle_refresh_seconds", 21600) or 21600) + + # -- fetch targets -------------------------------------------------------- + + def _targets(self) -> list[tuple[str, str]]: + """Build the (label, url) fetch list from config groups + norad_ids.""" + targets: list[tuple[str, str]] = [] + for group in (getattr(self._config, "tle_groups", None) or []): + targets.append( + (f"GROUP={group}", f"{GP_BASE_URL}?GROUP={group}&FORMAT=tle")) + for norad in (getattr(self._config, "norad_ids", None) or []): + targets.append( + (f"CATNR={norad}", f"{GP_BASE_URL}?CATNR={norad}&FORMAT=tle")) + return targets + + # -- polling -------------------------------------------------------------- + + def tick(self, now: Optional[float] = None) -> bool: + """One slow poll. Returns True if any TLE row changed. + + Resilient: a failed target logs + is skipped and updates health; a + thrown exception never propagates out of tick(). + """ + now = now if now is not None else time.time() + if now - self._last_tick < self._interval: + return False + self._last_tick = now + + targets = self._targets() + if not targets: + logger.debug("tle_fetch: no groups/norad_ids configured; nothing to fetch") + self._is_loaded = True + self._upserted = 0 + return False + + written = 0 + errors = 0 + for label, url in targets: + try: + text = self._fetch(url) + except Exception as e: + errors += 1 + self._last_error = f"{label}: {e}" + logger.warning("tle_fetch: fetch failed for %s: %s", label, e) + continue + try: + written += self._store(parse_tle_block(text)) + except Exception as e: + errors += 1 + self._last_error = f"{label}: store {e}" + logger.warning("tle_fetch: store failed for %s: %s", label, e) + + self._upserted = written + if errors == 0: + self._last_error = None + self._consecutive_errors = 0 + else: + self._consecutive_errors += 1 + self._is_loaded = True + return written > 0 + + def _fetch(self, url: str) -> str: + """HTTP GET a Celestrak GP endpoint, returning the decoded body.""" + headers = {"User-Agent": "MeshAI/1.0", "Accept": "text/plain"} + req = Request(url, headers=headers) + try: + with urlopen(req, timeout=20) as resp: + return resp.read().decode("utf-8", errors="replace") + except HTTPError as e: + raise RuntimeError(f"HTTP {e.code}") from e + except URLError as e: + raise RuntimeError(str(e.reason)) from e + + def _store(self, records: list[dict]) -> int: + """Upsert parsed TLE records into sat_tles. Returns rows written.""" + if not records: + return 0 + from meshai.persistence import get_db + from meshai.central.tle_handler import upsert_tle + + conn = get_db() + now = int(time.time()) + written = 0 + for rec in records: + if upsert_tle(conn, rec["norad_id"], rec["name"], + rec["line1"], rec["line2"], rec["epoch"], now=now): + written += 1 + return written + + # -- store integration ---------------------------------------------------- + + def get_events(self) -> list: + """Storage-only adapter: never produces mesh events.""" + return [] + + def to_event(self, evt: dict): + """Storage-only adapter: no event translation.""" + return None + + @property + def health_status(self) -> dict: + return { + "source": "satpass_tle", + "is_loaded": self._is_loaded, + "last_error": self._last_error, + "consecutive_errors": self._consecutive_errors, + "event_count": 0, + "last_fetch": self._last_tick, + "tles_upserted": self._upserted, + } diff --git a/work/meshai/main.py b/work/meshai/main.py index b31f332..1de6bc7 100644 --- a/work/meshai/main.py +++ b/work/meshai/main.py @@ -400,6 +400,14 @@ class MeshAI: except Exception: logger.exception("persistence init_db failed at startup") + # Native satpass: seed observer_locations from SatpassConfig.observers + # (config is the source of truth; the table is the predictor's store). + try: + from meshai.persistence.observer_locations import seed_observers_from_config + seed_observers_from_config(self.config.environmental.satpass) + except Exception: + logger.exception("observer_locations seed failed at startup") + # v0.6-3b: Initialize geocoder config from config.yaml try: gc = self.config.environmental.geocoder diff --git a/work/meshai/persistence/db.py b/work/meshai/persistence/db.py index 99f4ec3..209e504 100644 --- a/work/meshai/persistence/db.py +++ b/work/meshai/persistence/db.py @@ -30,7 +30,7 @@ logger = logging.getLogger(__name__) DEFAULT_DB_PATH = "/data/meshai.sqlite" MESHAI_DB_PATH_ENV = "MESHAI_DB_PATH" -SCHEMA_VERSION = 22 +SCHEMA_VERSION = 23 SCHEMA_META_TABLE = "schema_meta" MIGRATIONS_DIR = Path(__file__).parent / "migrations" diff --git a/work/meshai/persistence/migrations/v23.sql b/work/meshai/persistence/migrations/v23.sql new file mode 100644 index 0000000..9c8522b --- /dev/null +++ b/work/meshai/persistence/migrations/v23.sql @@ -0,0 +1,16 @@ +-- v23: observer_locations table for native satpass prediction. +-- +-- The native SGP4 pass-predictor computes passes for a set of ground +-- observer stations. SatpassConfig.observers is the source of truth; +-- those entries are seeded (upserted) into this table on startup so the +-- predictor has a single queryable store of coordinates regardless of how +-- config was loaded. slug is the stable key. + +CREATE TABLE IF NOT EXISTS observer_locations ( + slug TEXT PRIMARY KEY, + name TEXT NOT NULL, + lat REAL NOT NULL, + lon REAL NOT NULL, + alt_m REAL NOT NULL DEFAULT 0, + enabled INTEGER NOT NULL DEFAULT 1 +); diff --git a/work/meshai/persistence/observer_locations.py b/work/meshai/persistence/observer_locations.py new file mode 100644 index 0000000..a9ae3a1 --- /dev/null +++ b/work/meshai/persistence/observer_locations.py @@ -0,0 +1,88 @@ +"""observer_locations accessors + config seeding (v23). + +Ground-station coordinates for native satpass prediction. Follows the same +pattern as `persistence/curation.py`: a migration creates the table, a seed +routine populates it, and callers read via `get_observers()`. + +Unlike curation's gauge_sites/town_anchors (seeded from Python constants), +observers are seeded from live config — `SatpassConfig.observers` is the +source of truth — so operators manage them in config.yaml / the dashboard +while the predictor reads the queryable table. +""" +from __future__ import annotations + +import logging +import sqlite3 +from typing import Any, Optional + +logger = logging.getLogger(__name__) + + +def upsert_observer(slug: str, name: str, lat: float, lon: float, + alt_m: float = 0.0, enabled: bool = True, + conn: Optional[sqlite3.Connection] = None) -> None: + """Insert or update one observer row keyed on slug.""" + if conn is None: + from meshai.persistence import get_db + conn = get_db() + conn.execute( + "INSERT INTO observer_locations(slug, name, lat, lon, alt_m, enabled) " + "VALUES (?, ?, ?, ?, ?, ?) " + "ON CONFLICT(slug) DO UPDATE SET " + "name=excluded.name, lat=excluded.lat, lon=excluded.lon, " + "alt_m=excluded.alt_m, enabled=excluded.enabled", + (slug, name, float(lat), float(lon), float(alt_m), + 1 if enabled else 0), + ) + + +def get_observers(conn: Optional[sqlite3.Connection] = None) -> list[dict]: + """Return enabled observer rows as dicts (slug, name, lat, lon, alt_m).""" + if conn is None: + from meshai.persistence import get_db + conn = get_db() + rows = conn.execute( + "SELECT slug, name, lat, lon, alt_m FROM observer_locations " + "WHERE enabled=1 ORDER BY slug" + ).fetchall() + return [dict(r) for r in rows] + + +def seed_observers_from_config(satpass_config: Any, + conn: Optional[sqlite3.Connection] = None) -> int: + """Upsert every observer defined in SatpassConfig.observers. + + Each config entry is a dict with keys: slug, name, lat, lon, and + optional alt_m (default 0) and enabled (default True). Entries missing + slug/name/lat/lon are skipped with a warning. Config is the source of + truth, so this UPSERTs (not INSERT OR IGNORE) — edits in config + propagate on the next startup. Returns the number of rows seeded. + + Intended to be called at startup after init_db(), where the loaded + config object is available (see main.py `_init_components`). + """ + observers = list(getattr(satpass_config, "observers", None) or []) + if not observers: + return 0 + if conn is None: + from meshai.persistence import get_db + conn = get_db() + + seeded = 0 + for obs in observers: + try: + slug = obs["slug"] + name = obs["name"] + lat = obs["lat"] + lon = obs["lon"] + except (KeyError, TypeError): + logger.warning("observer_locations: skipping malformed observer entry: %r", obs) + continue + alt_m = obs.get("alt_m", 0.0) + enabled = obs.get("enabled", True) + upsert_observer(slug, name, lat, lon, alt_m, enabled, conn=conn) + seeded += 1 + + if seeded: + logger.info("observer_locations: seeded %d observer(s) from config", seeded) + return seeded diff --git a/work/tests/test_observer_locations.py b/work/tests/test_observer_locations.py new file mode 100644 index 0000000..650e382 --- /dev/null +++ b/work/tests/test_observer_locations.py @@ -0,0 +1,126 @@ +"""Tests for observer_locations (v23): migration, accessors, seed-from-config.""" +from __future__ import annotations + +import pytest + +from meshai.config import SatpassConfig +from meshai.persistence import SCHEMA_VERSION, get_db +from meshai.persistence.observer_locations import ( + get_observers, + seed_observers_from_config, + upsert_observer, +) + + +# -- schema / migration ------------------------------------------------------- + +def test_schema_version_is_23(): + assert SCHEMA_VERSION == 23 + + +def test_observer_locations_table_exists(): + conn = get_db() + tables = {r["name"] for r in conn.execute( + "SELECT name FROM sqlite_master WHERE type='table'" + ).fetchall()} + assert "observer_locations" in tables + + +def test_schema_meta_at_23(): + conn = get_db() + row = conn.execute( + "SELECT value FROM schema_meta WHERE key='version'").fetchone() + assert int(row["value"]) == 23 + + +# -- accessors ---------------------------------------------------------------- + +def test_upsert_and_get_roundtrip(): + upsert_observer("boise", "Boise", 43.615, -116.202, alt_m=824.0) + obs = get_observers() + assert len(obs) == 1 + r = obs[0] + assert r["slug"] == "boise" + assert r["name"] == "Boise" + assert r["lat"] == pytest.approx(43.615) + assert r["lon"] == pytest.approx(-116.202) + assert r["alt_m"] == pytest.approx(824.0) + + +def test_upsert_updates_existing(): + upsert_observer("site1", "Old Name", 40.0, -110.0) + upsert_observer("site1", "New Name", 41.0, -111.0, alt_m=500.0) + obs = {o["slug"]: o for o in get_observers()} + assert obs["site1"]["name"] == "New Name" + assert obs["site1"]["lat"] == pytest.approx(41.0) + assert obs["site1"]["alt_m"] == pytest.approx(500.0) + + +def test_disabled_observers_excluded(): + upsert_observer("on", "Enabled", 40.0, -110.0, enabled=True) + upsert_observer("off", "Disabled", 41.0, -111.0, enabled=False) + slugs = {o["slug"] for o in get_observers()} + assert "on" in slugs + assert "off" not in slugs + + +def test_alt_m_defaults_to_zero(): + upsert_observer("noalt", "No Altitude", 40.0, -110.0) + r = get_observers()[0] + assert r["alt_m"] == pytest.approx(0.0) + + +# -- seed from config --------------------------------------------------------- + +def test_seed_observers_from_config(): + cfg = SatpassConfig(observers=[ + {"slug": "boise", "name": "Boise", "lat": 43.615, "lon": -116.202, "alt_m": 824.0}, + {"slug": "twin", "name": "Twin Falls", "lat": 42.563, "lon": -114.461}, + ]) + n = seed_observers_from_config(cfg) + assert n == 2 + slugs = {o["slug"] for o in get_observers()} + assert slugs == {"boise", "twin"} + + +def test_seed_respects_disabled_flag(): + cfg = SatpassConfig(observers=[ + {"slug": "a", "name": "A", "lat": 40.0, "lon": -110.0}, + {"slug": "b", "name": "B", "lat": 41.0, "lon": -111.0, "enabled": False}, + ]) + seed_observers_from_config(cfg) + slugs = {o["slug"] for o in get_observers()} + assert "a" in slugs + assert "b" not in slugs + + +def test_seed_is_idempotent_and_updates(): + cfg = SatpassConfig(observers=[ + {"slug": "x", "name": "X", "lat": 40.0, "lon": -110.0}, + ]) + assert seed_observers_from_config(cfg) == 1 + # Re-seed with an edited name — upsert, not duplicate. + cfg2 = SatpassConfig(observers=[ + {"slug": "x", "name": "X Renamed", "lat": 40.0, "lon": -110.0}, + ]) + seed_observers_from_config(cfg2) + obs = get_observers() + assert len(obs) == 1 + assert obs[0]["name"] == "X Renamed" + + +def test_seed_skips_malformed_entries(): + cfg = SatpassConfig(observers=[ + {"slug": "good", "name": "Good", "lat": 40.0, "lon": -110.0}, + {"name": "MissingSlug", "lat": 41.0, "lon": -111.0}, # no slug + "not-a-dict", + ]) + n = seed_observers_from_config(cfg) + assert n == 1 + assert {o["slug"] for o in get_observers()} == {"good"} + + +def test_seed_empty_config_noop(): + cfg = SatpassConfig(observers=[]) + assert seed_observers_from_config(cfg) == 0 + assert get_observers() == [] diff --git a/work/tests/test_satpass_native.py b/work/tests/test_satpass_native.py new file mode 100644 index 0000000..69b771e --- /dev/null +++ b/work/tests/test_satpass_native.py @@ -0,0 +1,307 @@ +"""Tests for the native SGP4 satpass adapter (env.satpass). + +The native adapter computes every observer for a satellite in ONE tick, so it +consolidates in-memory and gates synchronously via the SHARED +`satpass_handler.gate_consolidated_pass` — with NO `satpass_pending` buffer and +NO Central consumer/timer. These tests monkeypatch `compute_passes` and +`get_observers` (no SGP4 / no network) and seed a fresh `sat_tles` row, then +exercise: multi-observer consolidation, cross-tick dedup via `satpass_events`, +resilient empty cases, the precomposed aos→peak→los wire, and a guard that the +Central path still feeds the shared gate the correctly-merged consolidation. +""" +from __future__ import annotations + +import json +from datetime import datetime, timezone + +import pytest + +from meshai.env.satpass import SatpassAdapter +from meshai.config import SatpassConfig +from meshai.central.pass_predictor import PassInfo +from meshai.central.tle_handler import upsert_tle +from meshai.persistence import get_db + + +# A valid ISS TLE so _resolve_tles / get_tle_by_norad has real data to return. +ISS_L1 = "1 25544U 98067A 26182.50000000 .00016717 00000-0 10270-3 0 9008" +ISS_L2 = "2 25544 51.6400 208.9163 0007417 17.6777 85.6621 15.54225995 12345" + +# An hour-aligned base so both observers' AOS land in the same hour bucket +# (same canonical id) and therefore consolidate into ONE pass. +T0 = (1783000000 // 3600) * 3600 + 100 # aligned + 100s + +_BOISE = {"slug": "boise", "name": "Boise", "lat": 43.6, "lon": -116.2, "alt_m": 0.0} +_TWIN = {"slug": "twin", "name": "Twin Falls", "lat": 42.5, "lon": -114.4, "alt_m": 0.0} + + +# ── helpers ────────────────────────────────────────────────────────────── + +def _dt(epoch: int) -> datetime: + return datetime.fromtimestamp(epoch, tz=timezone.utc) + + +def _pass(aos: int, los: int, max_el: float, + az_aos: float, az_los: float, az_peak: float) -> PassInfo: + peak = (aos + los) // 2 + return PassInfo( + aos_time=_dt(aos), los_time=_dt(los), peak_time=_dt(peak), + max_elevation=max_el, + azimuth_at_aos=az_aos, azimuth_at_los=az_los, azimuth_at_peak=az_peak, + ) + + +def _seed_iss_tle(): + """Seed a FRESH ISS TLE into sat_tles so the adapter resolves it.""" + conn = get_db() + fresh = datetime.now(timezone.utc).isoformat() + upsert_tle(conn, 25544, "ISS (ZARYA)", ISS_L1, ISS_L2, fresh) + + +def _enable_satpass_db(dry_run=False, max_per_hour=100, norad_ids=None): + """Set satpass adapter_config in the test DB (the gate reads it).""" + 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=? " + "WHERE adapter='satpass' AND key='dry_run'", + (json.dumps(dry_run),)) + conn.execute("UPDATE adapter_config SET value_json=? " + "WHERE adapter='satpass' AND key='max_broadcasts_per_hour'", + (json.dumps(max_per_hour),)) + if norad_ids is not None: + conn.execute("UPDATE adapter_config SET value_json=? " + "WHERE adapter='satpass' AND key='norad_ids'", + (json.dumps(norad_ids),)) + invalidate_cache() + + +def _adapter(**overrides) -> SatpassAdapter: + cfg = SatpassConfig(enabled=True, feed_source="native", + norad_ids=[25544], min_elevation_deg=10.0, + window_hours=24, **overrides) + return SatpassAdapter(cfg) + + +def _patch_predictor(monkeypatch, fake): + monkeypatch.setattr("meshai.central.pass_predictor.compute_passes", fake) + + +def _patch_observers(monkeypatch, observers): + monkeypatch.setattr( + "meshai.persistence.observer_locations.get_observers", + lambda *a, **k: list(observers)) + + +# Two observers, same pass: boise rises first (entry), twin sets last (exit) +# AND twin has the higher max elevation (so it supplies max_el + peak). +def _two_observer_pass(l1, l2, lat, lon, alt, window_h, min_el, now): + if abs(lat - _BOISE["lat"]) < 0.1: + return [_pass(T0, T0 + 300, 40.0, az_aos=225, az_los=300, az_peak=90)] + if abs(lat - _TWIN["lat"]) < 0.1: + return [_pass(T0 + 60, T0 + 400, 70.0, az_aos=270, az_los=45, az_peak=180)] + return [] + + +# ══════════════════════════════════════════════════════════════════════ +# 1. MULTI-OBSERVER CONSOLIDATION +# ══════════════════════════════════════════════════════════════════════ + +def test_two_observers_consolidate_to_one_broadcast(monkeypatch): + _enable_satpass_db(dry_run=False) + _seed_iss_tle() + _patch_observers(monkeypatch, [_BOISE, _TWIN]) + _patch_predictor(monkeypatch, _two_observer_pass) + + adapter = _adapter() + changed = adapter.tick(now=T0 - 3600) # any now; predictor ignores it + + assert changed is True + staged = adapter.get_events() + assert len(staged) == 1, "two observers, one pass -> ONE consolidated event" + + evt = staged[0] + cid = f"25544:{T0 // 3600}" + assert evt["event_id"] == cid + + # satpass_events row proves the merge: earliest AOS, latest LOS, + # max-elevation from the higher observer, both observers recorded. + row = get_db().execute( + "SELECT observer, max_elevation, aos_at, los_at FROM satpass_events " + "WHERE event_id=?", (cid,)).fetchone() + assert row is not None + assert row["max_elevation"] == 70.0 # twin (higher) + assert row["aos_at"] == T0 # boise (earliest AOS) + assert row["los_at"] == T0 + 400 # twin (latest LOS) + assert "boise" in row["observer"] and "twin" in row["observer"] + + # Wire carries entry->exit region + aos->peak->los compass sweep. + wire = evt["wire"] + assert "boise→twin" in wire # entry -> exit + assert "SW→S→NE" in wire # aos -> peak(twin) -> los + + +# ══════════════════════════════════════════════════════════════════════ +# 2. CROSS-TICK DEDUP (satpass_events remembers after commit) +# ══════════════════════════════════════════════════════════════════════ + +def test_second_tick_does_not_rebroadcast_after_commit(monkeypatch): + _enable_satpass_db(dry_run=False) + _seed_iss_tle() + _patch_observers(monkeypatch, [_BOISE, _TWIN]) + _patch_predictor(monkeypatch, _two_observer_pass) + + adapter = _adapter() + + # Tick 1: stages the pass. Simulate a successful mesh send by firing the + # commit closure the gate attached (this is what the dispatcher does). + assert adapter.tick(now=T0 - 3600) is True + staged = adapter.get_events() + assert len(staged) == 1 + commit = staged[0]["data"]["_on_broadcast_committed"] + commit(float(T0)) # marks satpass_events.last_broadcast_at + + # Tick 2 (interval elapsed): same canonical pass must be suppressed. + assert adapter.tick(now=T0 + 2000) is False + assert adapter.get_events() == [] + + +def test_second_tick_without_commit_is_not_deduped(monkeypatch): + # Guard the semantics: dedup persists ONLY after the broadcast commits. + _enable_satpass_db(dry_run=False) + _seed_iss_tle() + _patch_observers(monkeypatch, [_BOISE, _TWIN]) + _patch_predictor(monkeypatch, _two_observer_pass) + + adapter = _adapter() + assert adapter.tick(now=T0 - 3600) is True + # No commit fired -> last_broadcast_at still NULL -> re-stages next tick. + assert adapter.tick(now=T0 + 2000) is True + + +# ══════════════════════════════════════════════════════════════════════ +# 3. RESILIENT EMPTY CASES (never crash, yield nothing) +# ══════════════════════════════════════════════════════════════════════ + +def test_no_observers_yields_nothing(monkeypatch): + _enable_satpass_db(dry_run=False) + _seed_iss_tle() + _patch_observers(monkeypatch, []) + _patch_predictor(monkeypatch, _two_observer_pass) + + adapter = _adapter() + assert adapter.tick(now=T0) is False + assert adapter.get_events() == [] + assert adapter.health_status["is_loaded"] is True + + +def test_no_fresh_tles_yields_nothing(monkeypatch): + _enable_satpass_db(dry_run=False) + # Do NOT seed sat_tles -> get_tle_by_norad returns None. + _patch_observers(monkeypatch, [_BOISE, _TWIN]) + _patch_predictor(monkeypatch, _two_observer_pass) + + adapter = _adapter() + assert adapter.tick(now=T0) is False + assert adapter.get_events() == [] + + +def test_below_min_elevation_yields_nothing(monkeypatch): + _enable_satpass_db(dry_run=False) + _seed_iss_tle() + _patch_observers(monkeypatch, [_BOISE, _TWIN]) + # Predictor filters by min_el internally; a too-low pass => empty list. + _patch_predictor(monkeypatch, lambda *a, **k: []) + + adapter = _adapter() + assert adapter.tick(now=T0) is False + assert adapter.get_events() == [] + + +# ══════════════════════════════════════════════════════════════════════ +# 4. PRECOMPOSED WIRE renders aos->peak->los through format_pass +# ══════════════════════════════════════════════════════════════════════ + +def test_emitted_event_renders_aos_peak_los(monkeypatch): + from meshai.notifications.renderers.composer import compose_mesh_message + + _enable_satpass_db(dry_run=False) + _seed_iss_tle() + _patch_observers(monkeypatch, [_BOISE, _TWIN]) + _patch_predictor(monkeypatch, _two_observer_pass) + + adapter = _adapter() + adapter.tick(now=T0 - 3600) + evt = adapter.get_events()[0] + + event = adapter.to_event(evt) + assert event is not None + assert event.category == "sat_pass" + assert event.severity == "immediate" # max_el 70 -> immediate + assert event.data.get("_meshai_precomposed") is True + assert callable(event.data.get("_on_broadcast_committed")) + + # Precomposed: composer returns the wire verbatim, with the peak point + # rendered between aos and los. + composed = compose_mesh_message(event) + assert composed == evt["wire"] + assert "SW→S→NE" in composed # aos -> peak -> los + assert composed.startswith("\U0001F6F0") # satellite emoji + + +# ══════════════════════════════════════════════════════════════════════ +# 5. SHARED-GATE EXTRACTION GUARD (Central path still feeds the gate right) +# ══════════════════════════════════════════════════════════════════════ + +def test_central_consolidate_feeds_shared_gate_merged(monkeypatch): + """consolidate_satpass_pending must merge observers and hand the SAME + shared gate a correctly-consolidated dict (earliest AOS / latest LOS / + max-el observer / entry+exit / observer_list). Spying the gate proves the + Central path routes through the extracted, source-agnostic function.""" + import meshai.central.satpass_handler as sh + + conn = get_db() + cid = f"25544:{T0 // 3600}" + # Two pending rows for the same canonical id (boise entry, twin exit+peak). + rows = [ + (cid, "boise", "ISS", 25544, 40.0, T0, T0 + 300, "SW", "NW", "E", T0, T0 + 5), + (cid, "twin", "ISS", 25544, 70.0, T0 + 60, T0 + 400, "W", "NE", "S", T0, T0 + 5), + ] + for r in rows: + conn.execute( + "INSERT OR REPLACE INTO satpass_pending(" + "consolidated_id, observer, sat_name, norad_id, max_elevation, " + "aos_at, los_at, aos_compass, los_compass, peak_compass, " + "received_at, due_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", r) + + captured = {} + + def _spy(consolidated, *, now): + captured["c"] = consolidated + captured["now"] = now + return None # short-circuit: no broadcast side effects + + monkeypatch.setattr(sh, "gate_consolidated_pass", _spy) + + result = sh.consolidate_satpass_pending(cid) + assert result is None + + c = captured["c"] + assert c["consolidated_id"] == cid + assert c["norad_id"] == 25544 + assert c["max_elevation"] == 70.0 # twin (higher) + assert c["aos_epoch"] == T0 # boise earliest AOS + assert c["los_epoch"] == T0 + 400 # twin latest LOS + assert c["aos_compass"] == "SW" # boise + assert c["los_compass"] == "NE" # twin + assert c["peak_compass"] == "S" # twin (max-el observer) + assert c["entry_observer"] == "boise" + assert c["exit_observer"] == "twin" + assert c["observer_list"] == "boise,twin" + + # Pending buffer is drained regardless of the gate's decision. + left = conn.execute( + "SELECT COUNT(*) AS n FROM satpass_pending WHERE consolidated_id=?", + (cid,)).fetchone() + assert left["n"] == 0 diff --git a/work/tests/test_satpass_persisted_timer.py b/work/tests/test_satpass_persisted_timer.py index 04438ad..89c820d 100644 --- a/work/tests/test_satpass_persisted_timer.py +++ b/work/tests/test_satpass_persisted_timer.py @@ -98,12 +98,13 @@ def _ingest_envelope(norad_id=25544, observer="Boise", max_el=72.5, # ── schema / migration ─────────────────────────────────────────────── -def test_schema_version_is_22(): - assert SCHEMA_VERSION == 22 +def test_schema_version_is_current(): + # Bumped to 23 by the native-satpass observer_locations migration (v23). + assert SCHEMA_VERSION == 23 def test_v22_migration_applies_and_adds_due_at_column(tmp_path, monkeypatch): - """Fresh DB migrates cleanly to v22 and satpass_pending has due_at.""" + """Fresh DB migrates cleanly and satpass_pending has due_at (v22 column).""" from meshai.persistence import close_thread_connection from meshai.persistence import db as persistence_db db = str(tmp_path / "fresh-v22.sqlite") @@ -112,7 +113,7 @@ def test_v22_migration_applies_and_adds_due_at_column(tmp_path, monkeypatch): close_thread_connection() conn = init_db() row = conn.execute("SELECT value FROM schema_meta WHERE key='version'").fetchone() - assert int(row["value"]) == 22 + assert int(row["value"]) == 23 cols = {r["name"] for r in conn.execute("PRAGMA table_info(satpass_pending)")} assert "due_at" in cols close_thread_connection() diff --git a/work/tests/test_tle_fetch.py b/work/tests/test_tle_fetch.py new file mode 100644 index 0000000..7038cdc --- /dev/null +++ b/work/tests/test_tle_fetch.py @@ -0,0 +1,187 @@ +"""Tests for the native Celestrak TLE fetcher (env.tle_fetch). + +Covers epoch parsing, 3-line block parsing, upsert into the shared +sat_tles table, latest-epoch-wins on re-fetch, and malformed-block +tolerance. HTTP is monkeypatched — no network. +""" +from __future__ import annotations + +import pytest + +from meshai.env.tle_fetch import ( + TLEFetchAdapter, + parse_tle_block, + parse_tle_epoch, +) +from meshai.central.tle_handler import get_tle_by_norad +from meshai.config import SatpassConfig +from meshai.persistence import get_db + + +# A valid ISS 3-line set. Line-1 epoch field (cols 19-32) = 26182.50000000 +# -> 2026 day-of-year 182.5 -> 2026-07-01T12:00:00+00:00. +ISS_TLE = ( + "ISS (ZARYA)\n" + "1 25544U 98067A 26182.50000000 .00016717 00000-0 10270-3 0 9008\n" + "2 25544 51.6400 208.9163 0007417 17.6777 85.6621 15.54225995 12345\n" +) + +# Same satellite, a NEWER epoch (26183.50 -> 2026-07-02T12:00Z). +ISS_TLE_NEWER = ( + "ISS (ZARYA)\n" + "1 25544U 98067A 26183.50000000 .00016717 00000-0 10270-3 0 9010\n" + "2 25544 51.6400 208.9163 0007417 17.6777 85.6621 15.54225995 12347\n" +) + +# Same satellite, an OLDER epoch (26181.50 -> 2026-06-30T12:00Z). +ISS_TLE_OLDER = ( + "ISS (ZARYA)\n" + "1 25544U 98067A 26181.50000000 .00016717 00000-0 10270-3 0 9006\n" + "2 25544 51.6400 208.9163 0007417 17.6777 85.6621 15.54225995 12343\n" +) + + +def _adapter(**overrides) -> TLEFetchAdapter: + cfg = SatpassConfig(enabled=True, feed_source="native", + tle_groups=[], norad_ids=[25544], **overrides) + return TLEFetchAdapter(cfg) + + +# -- epoch parsing ------------------------------------------------------------ + +def test_parse_tle_epoch_iso(): + iso = parse_tle_epoch( + "1 25544U 98067A 26182.50000000 .00016717 00000-0 10270-3 0 9008") + assert iso.startswith("2026-07-01T12:00:00") + assert "+00:00" in iso + + +def test_parse_tle_epoch_two_digit_year_window(): + # YY=98 -> 1998 (>= 57 maps to 1900s). + iso = parse_tle_epoch("1 25544U 98067A 98001.00000000 .0 0 0 0 1") + assert iso.startswith("1998-01-01") + + +# -- block parsing ------------------------------------------------------------ + +def test_parse_tle_block_basic(): + recs = parse_tle_block(ISS_TLE) + assert len(recs) == 1 + r = recs[0] + assert r["norad_id"] == 25544 + assert r["name"] == "ISS (ZARYA)" + assert r["line1"].startswith("1 25544") + assert r["line2"].startswith("2 25544") + assert r["epoch"].startswith("2026-07-01T12:00:00") + + +def test_parse_tle_block_skips_malformed(): + # First triple is garbage (line1 doesn't start with "1 "); second is valid. + block = ( + "GARBAGE SAT\n" + "not a real line1\n" + "also not line2\n" + + ISS_TLE + ) + recs = parse_tle_block(block) + norads = [r["norad_id"] for r in recs] + assert 25544 in norads + # The garbage entry must not have produced a record. + assert all(isinstance(n, int) for n in norads) + + +# -- fetch + upsert ----------------------------------------------------------- + +def test_tick_upserts_into_sat_tles(monkeypatch): + adapter = _adapter() + monkeypatch.setattr(adapter, "_fetch", lambda url: ISS_TLE) + + changed = adapter.tick(now=1_000_000) + assert changed is True + + row = get_tle_by_norad(25544) + assert row is not None + assert row["name"] == "ISS (ZARYA)" + assert row["line1"].startswith("1 25544") + assert row["line2"].startswith("2 25544") + assert row["epoch"].startswith("2026-07-01T12:00:00") + + +def test_latest_epoch_wins_on_refetch(monkeypatch): + adapter = _adapter() + + monkeypatch.setattr(adapter, "_fetch", lambda url: ISS_TLE) + assert adapter.tick(now=1_000_000) is True + first = get_tle_by_norad(25544)["epoch"] + + # A newer epoch replaces it. + adapter._last_tick = 0 # bypass interval gate for the test + monkeypatch.setattr(adapter, "_fetch", lambda url: ISS_TLE_NEWER) + assert adapter.tick(now=2_000_000) is True + newer = get_tle_by_norad(25544)["epoch"] + assert newer > first + + # An older epoch is ignored (no write). + adapter._last_tick = 0 + monkeypatch.setattr(adapter, "_fetch", lambda url: ISS_TLE_OLDER) + changed = adapter.tick(now=3_000_000) + assert changed is False + assert get_tle_by_norad(25544)["epoch"] == newer + + +def test_malformed_block_does_not_crash_tick(monkeypatch): + adapter = _adapter() + monkeypatch.setattr( + adapter, "_fetch", + lambda url: "COMPLETE GARBAGE\nno lines here\n") + # Should complete without raising and write nothing. + changed = adapter.tick(now=1_000_000) + assert changed is False + assert get_tle_by_norad(25544) is None + + +def test_fetch_error_is_isolated(monkeypatch): + adapter = _adapter() + + def boom(url): + raise RuntimeError("HTTP 503") + + monkeypatch.setattr(adapter, "_fetch", boom) + changed = adapter.tick(now=1_000_000) + assert changed is False + assert adapter.health_status["last_error"] is not None + assert adapter.health_status["consecutive_errors"] == 1 + + +def test_storage_only_no_events(monkeypatch): + adapter = _adapter() + monkeypatch.setattr(adapter, "_fetch", lambda url: ISS_TLE) + adapter.tick(now=1_000_000) + assert adapter.get_events() == [] + assert adapter.to_event({}) is None + + +def test_interval_gate_skips_early_ticks(monkeypatch): + adapter = _adapter(tle_refresh_seconds=21600) + calls = [] + monkeypatch.setattr(adapter, "_fetch", + lambda url: calls.append(url) or ISS_TLE) + adapter.tick(now=1_000_000.0) # first tick fetches + adapter.tick(now=1_000_100.0) # 100s later, within interval -> skipped + assert len(calls) == 1 + + +def test_no_targets_configured_is_noop(): + cfg = SatpassConfig(enabled=True, feed_source="native", + tle_groups=[], norad_ids=[]) + adapter = TLEFetchAdapter(cfg) + assert adapter.tick(now=1_000_000) is False + + +def test_group_and_catnr_urls(): + cfg = SatpassConfig(tle_groups=["weather"], norad_ids=[25544]) + adapter = TLEFetchAdapter(cfg) + urls = [u for _, u in adapter._targets()] + assert any("GROUP=weather&FORMAT=tle" in u for u in urls) + assert any("CATNR=25544&FORMAT=tle" in u for u in urls) + assert all(u.startswith("https://celestrak.org/NORAD/elements/gp.php") for u in urls)