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

feat(phase4c): native SGP4 satpass — standalone satellite passes, no Central (#39)

Browse source 
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 <mj@k7zvx.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
malice 2026-07-05 02:25:55 -06:00 • committed by GitHub
commit 53eadf5135
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
14 changed files with 1405 additions and 53 deletions
Download patch file Download diff file Expand all files Collapse all files

173
work/meshai/central/satpass_handler.py
View file

@ -438,9 +438,106 @@ def handle_satpass(envelope: dict, subject: str,
return None 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: def consolidate_satpass_pending(consolidated_id: str) -> tuple[str, dict] | None:
"""Called by consumer when 5s consolidation timer fires. """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. Returns (wire_string, data_dict) or None if suppressed.
""" """
try: try:
@ -455,9 +552,6 @@ def consolidate_satpass_pending(consolidated_id: str) -> tuple[str, dict] | None
if not rows: if not rows:
return None return None
cfg = adapter_config.satpass
now = _now()
# Consolidate observers # Consolidate observers
sorted_by_aos = sorted(rows, key=lambda r: r["aos_at"]) sorted_by_aos = sorted(rows, key=lambda r: r["aos_at"])
sorted_by_los = sorted(rows, key=lambda r: r["los_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 exit_ = sorted_by_los[-1] # latest LOS
best = max(rows, key=lambda r: r["max_elevation"]) best = max(rows, key=lambda r: r["max_elevation"])
norad_id = best["norad_id"] consolidated = {
sat_name = best["sat_name"] "consolidated_id": consolidated_id,
max_el = best["max_elevation"] "norad_id": best["norad_id"],
aos_epoch = entry["aos_at"] "sat_name": best["sat_name"],
los_epoch = exit_["los_at"] "max_elevation": best["max_elevation"],
aos_compass = entry["aos_compass"] "aos_epoch": entry["aos_at"],
los_compass = exit_["los_compass"] "los_epoch": exit_["los_at"],
"aos_compass": entry["aos_compass"],
"los_compass": exit_["los_compass"],
# Peak belongs to whoever saw the highest elevation. # Peak belongs to whoever saw the highest elevation.
peak_compass = best["peak_compass"] "peak_compass": best["peak_compass"],
entry_obs = entry["observer"] "entry_observer": entry["observer"],
exit_obs = exit_["observer"] "exit_observer": exit_["observer"],
"observer_list": ",".join(r["observer"] for r in sorted_by_aos),
}
# Dedup against satpass_events result = gate_consolidated_pass(consolidated, now=_now())
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
# Rate cap # Clean up pending rows regardless of the gate's decision.
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
_cleanup_pending(conn, consolidated_id) _cleanup_pending(conn, consolidated_id)
# Prepare data dict with callbacks return result
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 _upsert_satpass(conn, *, event_id, norad_id, sat_name, observer, def _upsert_satpass(conn, *, event_id, norad_id, sat_name, observer,

34
work/meshai/central/tle_handler.py
View file

@ -73,15 +73,36 @@ def handle_tle(envelope: dict, subject: str,
logger.exception("tle_handler: persistence unavailable") logger.exception("tle_handler: persistence unavailable")
return None 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( existing = conn.execute(
"SELECT epoch FROM sat_tles WHERE norad_id = ?", "SELECT epoch FROM sat_tles WHERE norad_id = ?",
(norad_id,), (norad_id,),
).fetchone() ).fetchone()
if existing is not None and existing["epoch"] >= str(epoch): if existing is not None and existing["epoch"] >= epoch:
# Cached epoch is same or newer — skip # Cached epoch is same or newer — skip.
return None return False
conn.execute( conn.execute(
"INSERT INTO sat_tles(norad_id, name, line1, line2, epoch, updated_at) " "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 " "ON CONFLICT(norad_id) DO UPDATE SET "
"name=excluded.name, line1=excluded.line1, line2=excluded.line2, " "name=excluded.name, line1=excluded.line1, line2=excluded.line2, "
"epoch=excluded.epoch, updated_at=excluded.updated_at", "epoch=excluded.epoch, updated_at=excluded.updated_at",
(norad_id, name, line1, line2, str(epoch), now), (norad_id, name, line1, line2, epoch, now),
) )
return True
return None # storage-only, never broadcast
def get_fresh_tles(conn=None, max_age_days: int = STALE_DAYS) -> list[dict]: def get_fresh_tles(conn=None, max_age_days: int = STALE_DAYS) -> list[dict]:

22
work/meshai/config.py
View file

@ -489,11 +489,31 @@ class FIRMSConfig(_SourcedFeed):
@dataclass @dataclass
class SatpassConfig(_SourcedFeed): 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 enabled: bool = False
feed_source: str = "central" 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 @dataclass
class CentralConsumerConfig: class CentralConsumerConfig:

285
work/meshai/env/satpass.py vendored Normal file
View file

@ -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,
}

11
work/meshai/env/store.py vendored
View file

@ -50,6 +50,15 @@ class EnvironmentalStore:
lambda cfg: (cfg,)) lambda cfg: (cfg,))
self._register_adapter("wzdx", config.wzdx, ".wzdx", "WZDxAdapter", self._register_adapter("wzdx", config.wzdx, ".wzdx", "WZDxAdapter",
lambda cfg: (cfg,)) 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 # FIRMS needs reference to NIFC adapter for cross-referencing
if config.firms.enabled and config.firms.feed_source == "native": 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) logger.warning("Failed to initialize firms adapter: %s", err_msg)
self._failed_adapters["firms"] = 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 getattr(getattr(config, n, None), "feed_source", "native") == "central"]
if _central: if _central:
logger.debug("Adapters sourced from Central (native skipped): %s", _central) logger.debug("Adapters sourced from Central (native skipped): %s", _central)

230
work/meshai/env/tle_fetch.py vendored Normal file
View file

@ -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,
}

8
work/meshai/main.py
View file

@ -400,6 +400,14 @@ class MeshAI:
except Exception: except Exception:
logger.exception("persistence init_db failed at startup") 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 # v0.6-3b: Initialize geocoder config from config.yaml
try: try:
gc = self.config.environmental.geocoder gc = self.config.environmental.geocoder

2
work/meshai/persistence/db.py
View file

@ -30,7 +30,7 @@ logger = logging.getLogger(__name__)
DEFAULT_DB_PATH = "/data/meshai.sqlite" DEFAULT_DB_PATH = "/data/meshai.sqlite"
MESHAI_DB_PATH_ENV = "MESHAI_DB_PATH" MESHAI_DB_PATH_ENV = "MESHAI_DB_PATH"
SCHEMA_VERSION = 22 SCHEMA_VERSION = 23
SCHEMA_META_TABLE = "schema_meta" SCHEMA_META_TABLE = "schema_meta"
MIGRATIONS_DIR = Path(__file__).parent / "migrations" MIGRATIONS_DIR = Path(__file__).parent / "migrations"

16
work/meshai/persistence/migrations/v23.sql Normal file
View file

@ -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
);

88
work/meshai/persistence/observer_locations.py Normal file
View file

@ -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

126
work/tests/test_observer_locations.py Normal file
View file

@ -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() == []

307
work/tests/test_satpass_native.py Normal file
View file

@ -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

9
work/tests/test_satpass_persisted_timer.py
View file

@ -98,12 +98,13 @@ def _ingest_envelope(norad_id=25544, observer="Boise", max_el=72.5,
# ── schema / migration ─────────────────────────────────────────────── # ── schema / migration ───────────────────────────────────────────────
def test_schema_version_is_22(): def test_schema_version_is_current():
assert SCHEMA_VERSION == 22 # 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): 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 close_thread_connection
from meshai.persistence import db as persistence_db from meshai.persistence import db as persistence_db
db = str(tmp_path / "fresh-v22.sqlite") 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() close_thread_connection()
conn = init_db() conn = init_db()
row = conn.execute("SELECT value FROM schema_meta WHERE key='version'").fetchone() 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)")} cols = {r["name"] for r in conn.execute("PRAGMA table_info(satpass_pending)")}
assert "due_at" in cols assert "due_at" in cols
close_thread_connection() close_thread_connection()

187
work/tests/test_tle_fetch.py Normal file
View file

@ -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)