diff --git a/work/meshai/config.py b/work/meshai/config.py index c2ca788..56ab8d9 100644 --- a/work/meshai/config.py +++ b/work/meshai/config.py @@ -513,6 +513,11 @@ class SatpassConfig(_SourcedFeed): # Predictor pass filters (used by the next task): min_elevation_deg: float = 10.0 window_hours: int = 24 + # BROADCAST imminence lead (seconds): a predicted pass is broadcast only + # once its AOS is within this near-term window (and still in the future). + # `window_hours` governs how far ahead we PREDICT; this governs when a + # predicted pass is actually announced. Default 60 min. + broadcast_lead_seconds: int = 3600 @dataclass diff --git a/work/meshai/env/satpass.py b/work/meshai/env/satpass.py index b149a03..1cdd691 100644 --- a/work/meshai/env/satpass.py +++ b/work/meshai/env/satpass.py @@ -60,6 +60,18 @@ logger = logging.getLogger(__name__) # pointless). Overridable via SatpassConfig.pass_refresh_seconds if present. DEFAULT_POLL_SECONDS = 900 +# Broadcast imminence lead. Satpass is NOT an API-delta feed — it computes the +# SAME future passes every poll, so a pure "new since last poll" gate would +# seed them all as backlog and then never broadcast. A pass's "newly received" +# analog is IMMINENCE: it becomes broadcast-worthy only once its AOS first +# enters this near-term lead window (and is still in the future). Combined with +# the store's received-delta seen-set (keyed on the pass canonical id), each +# pass then emits exactly once as it crosses into imminence, and the +# first-poll-seed suppresses any startup burst. `window_hours` still governs +# how far ahead we PREDICT (the schedule); this only governs the BROADCAST +# trigger. Overridable via SatpassConfig.broadcast_lead_seconds. +DEFAULT_BROADCAST_LEAD_SECONDS = 3600 + class SatpassAdapter: """Native SGP4 pass predictor — consolidates in-memory, gates synchronously.""" @@ -71,6 +83,9 @@ class SatpassAdapter: 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._lead_s = int( + getattr(config, "broadcast_lead_seconds", DEFAULT_BROADCAST_LEAD_SECONDS) + or DEFAULT_BROADCAST_LEAD_SECONDS) self._norad_ids = self._parse_norad_ids( getattr(config, "norad_ids", None) or []) @@ -84,21 +99,39 @@ class SatpassAdapter: @staticmethod def _parse_norad_ids(raw) -> list[int]: - """Coerce a config norad_ids list (ints or GUI strings) to int list.""" + """Coerce a config norad_ids value to a sorted unique int list. + + Accepts a real list (ints or GUI strings) OR a single comma/space + separated string (as some GUI paths persist it). The string case + MUST be split first — iterating a bare string would walk its + CHARACTERS, turning "25544,33591" into garbage single-digit ids and + silently defeating the norad filter (a cause of non-listed sats + leaking into predictions). + """ + if isinstance(raw, str): + raw = raw.replace(",", " ").split() 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. + + When `norad_ids` is configured the result is ALSO post-filtered to + that set: this is belt-and-suspenders so that only configured NORADs + are ever predicted, regardless of which lookup path ran. Without a + configured list, the fetcher's `tle_groups` (e.g. "weather") stock + sat_tles with GOES/METEOR/FENGYUN etc.; the strict post-filter + guarantees those never leak into predictions once a list is set. """ from meshai.central.tle_handler import get_fresh_tles, get_tle_by_norad if self._norad_ids: + allowed = set(self._norad_ids) out: list[dict] = [] for nid in self._norad_ids: tle = get_tle_by_norad(nid) - if tle is not None: + if tle is not None and int(tle["norad_id"]) in allowed: out.append(tle) return out return get_fresh_tles() @@ -186,6 +219,18 @@ class SatpassAdapter: staged: list[dict] = [] for cid, recs in groups.items(): consolidated = self._consolidate(cid, recs) + # IMMINENCE gate — applied on the CONSOLIDATED pass (earliest AOS + # across observers), so multi-observer consolidation is never + # fragmented. A pass is broadcast-worthy only once its AOS is both + # in the FUTURE and within the near-term lead window. Far-future + # passes are predicted (on the schedule) but not staged until they + # become imminent; a past AOS is never staged. This is satpass's + # "just received" signal — the store's received-delta seen-set then + # emits each pass exactly once as it crosses into the window, and + # the first-poll-seed prevents any startup burst. + aos = consolidated["aos_epoch"] + if aos <= now_epoch or aos > now_epoch + self._lead_s: + continue try: result = sh.gate_consolidated_pass(consolidated, now=now_epoch) except Exception as e: diff --git a/work/meshai/env/store.py b/work/meshai/env/store.py index 74377b2..bdddb2e 100644 --- a/work/meshai/env/store.py +++ b/work/meshai/env/store.py @@ -1,5 +1,7 @@ """Environmental data store with tick-based adapter polling.""" +import hashlib +import json import logging import time from typing import TYPE_CHECKING, Optional @@ -29,6 +31,21 @@ class EnvironmentalStore: self._mesh_zones = config.nws_zones or [] self._region_anchors = region_anchors or [] + # ── Received-delta gate (NATIVE-only) ──────────────────────────── + # Per-adapter set of item keys seen in PRIOR polls, plus the set of + # adapters whose first poll has completed. The model the operator + # demanded: a native adapter broadcasts an item ONLY when it was newly + # RECEIVED from the API this poll — never by scanning an accumulated + # backlog. So the FIRST poll for an adapter records every current key + # as "seen" and emits NOTHING (that batch is pre-existing backlog, not + # ours to announce); every later poll emits only keys not already + # seen. In-memory only (per process) — a restart empties the sets, so + # the next poll is a fresh "first poll" that re-seeds silently. This + # makes broadcasting a backlog item structurally impossible on cold + # start, restart, or after the adapter was disabled for days. + self._seen: dict[str, set] = {} # adapter name -> set of item keys + self._seeded: set[str] = set() # adapters past their first poll + # Create adapter instances with error isolation self._register_adapter("nws", config.nws, ".nws", "NWSAlertsAdapter", lambda cfg: (cfg,)) @@ -112,46 +129,106 @@ class EnvironmentalStore: self._purge_expired() return changed + def _seen_key(self, name: str, raw_evt: dict) -> str: + """Derive a STABLE per-item key for the received-delta gate. + + Stability across polls is the whole point: the SAME real-world item + must produce the SAME key every poll, or it would look "newly + received" forever and re-broadcast on each tick. Preference order, + most→least explicit id: + 1. external_id — the upstream feed's own stable id (e.g. WZDx) + 2. event_id — the adapter's stable per-item id; every native raw + event carries one (the store's own dedup already keys on it, and + the adapters build it from natural ids: USGS quake id, 511 event + id, NWS alert id, ``swpc_``, ``ducting__``, + ``avy_
_``, etc. — all stable across polls). + 3. content hash — last-resort fallback if an item somehow carries no + id at all. + Namespaced by adapter so two feeds never cross-contaminate. + """ + ext = raw_evt.get("external_id") + if ext: + return f"{name}\x1eext:{ext}" + eid = raw_evt.get("event_id") + if eid: + return f"{name}\x1eeid:{eid}" + blob = json.dumps(raw_evt, sort_keys=True, default=str) + return f"{name}\x1ehash:" + hashlib.sha1(blob.encode()).hexdigest()[:16] + + def _delta_emit(self, name: str, adapter, raw_evt: dict, force: bool = False): + """Received-delta gate — emit ONLY items newly received THIS poll. + + - First poll for ``name`` (name not yet in ``self._seeded``): record + the item's key as seen and emit NOTHING. That batch is the + pre-existing backlog. + - Later polls: emit only when the key is not already in the seen-set + (it just appeared upstream = "just received"); then record it. + - ``force=True`` (avalanche danger-level rise) re-emits an already-seen + item on a legitimate CONTENT change — but is IGNORED on the first + poll, so a restart still can never replay the backlog. + + This replaces the native path's reliance on the deciders' + broadcast-state tables for the "is this new" decision. Legitimate + content filtering (severity/threshold/impact, decider gates) still runs + downstream in ``_emit_event``. + """ + seen = self._seen.setdefault(name, set()) + key = self._seen_key(name, raw_evt) + first_poll = name not in self._seeded + + if first_poll: + seen.add(key) # seed silently — this is backlog + return + if key in seen and not force: + return # already received in a prior poll + seen.add(key) + + if self._event_bus is not None and hasattr(adapter, "to_event"): + self._emit_event(adapter, raw_evt) + def _ingest(self, name: str, adapter): - """Ingest data from an adapter after it ticks.""" + """Ingest data from an adapter after it ticks. + + Emission goes through the received-delta gate (``_delta_emit``): the + adapter's FIRST data-bearing poll seeds the seen-set and broadcasts + nothing; only items that newly appear on later polls are broadcast. + ``self._events`` is still maintained for state/queries as before. + """ if name == "swpc": self._swpc_status = adapter.get_status() # Also ingest any alert events (R-scale >= 3) for evt in adapter.get_events(): key = (evt["source"], evt["event_id"]) - is_new = key not in self._events self._events[key] = evt - if is_new and self._event_bus and hasattr(adapter, "to_event"): - self._emit_event(adapter, evt) + self._delta_emit(name, adapter, evt) elif name == "ducting": self._ducting_status = adapter.get_status() for evt in adapter.get_events(): key = (evt["source"], evt["event_id"]) - is_new = key not in self._events self._events[key] = evt - if is_new and self._event_bus and hasattr(adapter, "to_event"): - self._emit_event(adapter, evt) + self._delta_emit(name, adapter, evt) elif name == "avalanche": - # Avalanche: re-emit on danger_level rise (Update:) not just new events. + # Avalanche: re-emit on danger_level rise (Update:) not just new + # events. The rise is a legitimate CONTENT change, so it passes + # `force=True` — but the received-delta gate still suppresses it on + # the first poll (backlog stays silent). for evt in adapter.get_events(): key = (evt["source"], evt["event_id"]) prior = self._events.get(key) - is_new = prior is None prior_level = prior.get("danger_level", -1) if prior else -1 - level_rose = (not is_new) and (evt.get("danger_level", -1) > prior_level) - - if (is_new or level_rose) and self._event_bus and hasattr(adapter, "to_event"): - evt["_is_update"] = level_rose # signal to to_event() - self._emit_event(adapter, evt) - + level_rose = (prior is not None) and ( + evt.get("danger_level", -1) > prior_level) + evt["_is_update"] = level_rose # signal to to_event() + self._delta_emit(name, adapter, evt, force=level_rose) self._events[key] = evt # always update stored state else: for evt in adapter.get_events(): key = (evt["source"], evt["event_id"]) - is_new = key not in self._events self._events[key] = evt - if is_new and self._event_bus and hasattr(adapter, "to_event"): - self._emit_event(adapter, evt) + self._delta_emit(name, adapter, evt) + + # First poll for this adapter is now complete: later polls may emit. + self._seeded.add(name) def _emit_event(self, adapter, raw_evt: dict): """Convert raw event to pipeline Event and emit to bus. diff --git a/work/tests/test_firms_native_fusion.py b/work/tests/test_firms_native_fusion.py index 572a1c1..eec2c6e 100644 --- a/work/tests/test_firms_native_fusion.py +++ b/work/tests/test_firms_native_fusion.py @@ -330,6 +330,13 @@ class TestAdapterTickFusion: store = EnvironmentalStore.__new__(EnvironmentalStore) store._events = {} store._event_bus = bus + # Received-delta gate state. This test exercises STEADY-STATE fusion + # routing (fusion reaches the bus, raw hotspots never do), so mark firms + # as already past its first-poll seed — otherwise the store's + # received-delta gate would (correctly) seed this first batch silently. + # First-poll seed-silence is covered by test_store_received_delta. + store._seen = {} + store._seeded = {"firms"} assert a.tick() is True store._ingest("firms", a) diff --git a/work/tests/test_satpass_native.py b/work/tests/test_satpass_native.py index 69b771e..66dc741 100644 --- a/work/tests/test_satpass_native.py +++ b/work/tests/test_satpass_native.py @@ -162,8 +162,10 @@ def test_second_tick_does_not_rebroadcast_after_commit(monkeypatch): 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 + # Tick 2 (interval elapsed): the pass is STILL imminent (AOS 30 min out, + # inside the 60-min lead) so the imminence gate would stage it — but the + # satpass_events commit from tick 1 must suppress the re-broadcast. + assert adapter.tick(now=T0 - 1800) is False assert adapter.get_events() == [] @@ -176,8 +178,9 @@ def test_second_tick_without_commit_is_not_deduped(monkeypatch): 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 + # No commit fired -> last_broadcast_at still NULL. The pass is still + # imminent at T0-1800 (AOS 30 min out) so it re-stages next tick. + assert adapter.tick(now=T0 - 1800) is True # ══════════════════════════════════════════════════════════════════════ @@ -305,3 +308,107 @@ def test_central_consolidate_feeds_shared_gate_merged(monkeypatch): "SELECT COUNT(*) AS n FROM satpass_pending WHERE consolidated_id=?", (cid,)).fetchone() assert left["n"] == 0 + + +# ══════════════════════════════════════════════════════════════════════ +# 6. IMMINENCE BROADCAST TRIGGER (satpass's "just received" analog) +# ══════════════════════════════════════════════════════════════════════ +# +# Satpass is NOT an API-delta feed — it recomputes the SAME future passes +# every poll. Its "newly received" signal is IMMINENCE: a pass is staged for +# broadcast only once its AOS is both in the FUTURE and within the near-term +# lead window (default 60 min). A far-future pass is predicted (on schedule) +# but not staged; a past AOS is never staged. Combined with the store's +# received-delta seen-set, each pass then emits exactly once as it crosses in. + +# The fixed mock pass sits at AOS=T0 (both observers consolidate to one). + +def test_far_future_pass_not_staged(monkeypatch): + _enable_satpass_db(dry_run=False) + _seed_iss_tle() + _patch_observers(monkeypatch, [_BOISE, _TWIN]) + _patch_predictor(monkeypatch, _two_observer_pass) + + adapter = _adapter() # default broadcast_lead_seconds = 3600 (60 min) + # AOS=T0 is 4 hours ahead of now -> well outside the 60-min lead window. + assert adapter.tick(now=T0 - 4 * 3600) is False + assert adapter.get_events() == [] + + +def test_imminent_pass_is_staged(monkeypatch): + _enable_satpass_db(dry_run=False) + _seed_iss_tle() + _patch_observers(monkeypatch, [_BOISE, _TWIN]) + _patch_predictor(monkeypatch, _two_observer_pass) + + adapter = _adapter() + # AOS=T0 is 30 min ahead of now -> inside the 60-min lead window. + assert adapter.tick(now=T0 - 1800) is True + staged = adapter.get_events() + assert len(staged) == 1 + assert staged[0]["event_id"] == f"25544:{T0 // 3600}" + + +def test_past_aos_never_staged(monkeypatch): + _enable_satpass_db(dry_run=False) + _seed_iss_tle() + _patch_observers(monkeypatch, [_BOISE, _TWIN]) + _patch_predictor(monkeypatch, _two_observer_pass) + + adapter = _adapter() + # now is 10 min AFTER AOS=T0 -> the pass is in the past, never staged. + assert adapter.tick(now=T0 + 600) is False + assert adapter.get_events() == [] + + +def test_configurable_lead_window(monkeypatch): + _enable_satpass_db(dry_run=False) + _seed_iss_tle() + _patch_observers(monkeypatch, [_BOISE, _TWIN]) + _patch_predictor(monkeypatch, _two_observer_pass) + + # With a 3-hour lead, an AOS 2 hours out is now imminent. + adapter = _adapter(broadcast_lead_seconds=3 * 3600) + assert adapter.tick(now=T0 - 2 * 3600) is True + assert len(adapter.get_events()) == 1 + + +# ══════════════════════════════════════════════════════════════════════ +# 7. NORAD_IDS FILTER IS STRICT (non-listed sats never predicted) +# ══════════════════════════════════════════════════════════════════════ + +# A non-ISS TLE (a stand-in for a GOES/METEOR/FENGYUN weather sat the fetcher +# stocks via tle_groups). Different NORAD so a filter break would surface as a +# second consolidated pass with a different canonical id. +OTHER_L1 = "1 43226U 18022A 26182.50000000 .00000000 00000-0 00000-0 0 9999" +OTHER_L2 = "2 43226 0.0300 100.0000 0001000 90.0000 270.0000 1.00270000 12345" + + +def _seed_other_tle(): + conn = get_db() + fresh = datetime.now(timezone.utc).isoformat() + upsert_tle(conn, 43226, "GOES-17", OTHER_L1, OTHER_L2, fresh) + + +def test_norad_ids_filter_excludes_non_listed_sats(monkeypatch): + _enable_satpass_db(dry_run=False) + _seed_iss_tle() # 25544 — configured + _seed_other_tle() # 43226 — fresh but NOT in norad_ids + _patch_observers(monkeypatch, [_BOISE, _TWIN]) + # Predictor would return a pass for ANY satellite (keyed on observer lat), + # so if 43226 leaked through it would produce a second staged pass. + _patch_predictor(monkeypatch, _two_observer_pass) + + adapter = _adapter() # norad_ids=[25544] + assert adapter.tick(now=T0 - 1800) is True + staged = adapter.get_events() + # Only the configured NORAD is predicted, despite 43226 being fresh. + assert len(staged) == 1 + assert staged[0]["norad_id"] == 25544 + + +def test_parse_norad_ids_handles_comma_string(): + # A GUI-persisted comma string must NOT be char-iterated into garbage ids. + assert SatpassAdapter._parse_norad_ids("25544, 33591") == [25544, 33591] + assert SatpassAdapter._parse_norad_ids([25544, "33591"]) == [25544, 33591] + assert SatpassAdapter._parse_norad_ids([]) == [] diff --git a/work/tests/test_store_received_delta.py b/work/tests/test_store_received_delta.py new file mode 100644 index 0000000..73adb2f --- /dev/null +++ b/work/tests/test_store_received_delta.py @@ -0,0 +1,156 @@ +"""Received-delta gate tests for the native EnvironmentalStore path. + +The operator's required model: a native adapter broadcasts an item ONLY when +it was newly RECEIVED from the API this poll — never by scanning an accumulated +backlog. The store enforces this with an in-memory, per-adapter seen-set: + + * the FIRST data-bearing poll for an adapter records every current item key + as "seen" and broadcasts NOTHING (that batch is pre-existing backlog); + * every later poll broadcasts only items whose key is not already seen; + * a fresh store (process restart) has empty sets, so its next poll is again a + "first poll" that re-seeds silently — backlog is never re-broadcast. + +These tests drive the real EnvironmentalStore + EventBus with a fake adapter +whose per-poll batch we control, and assert exactly which events reach the bus. +""" +from __future__ import annotations + +from meshai.env.store import EnvironmentalStore +from meshai.config import EnvironmentalConfig +from meshai.notifications.pipeline.bus import EventBus +from meshai.notifications.events import make_event + + +class _FakeAdapter: + """Native-adapter stand-in: returns a controllable batch of raw events. + + Raw events carry the (source, event_id) shape every native adapter emits, + which is exactly what the store's seen-set keys on. Each poll's batch is + set by the test via `set_batch`. + """ + + def __init__(self, source: str = "roads511"): + self._source = source + self._batch: list[dict] = [] + + def set_batch(self, ids: list[str]) -> None: + self._batch = [ + {"source": self._source, "event_id": eid, "fetched_at": 0} + for eid in ids + ] + + def tick(self) -> bool: + # Data is "fetched" every poll; the store decides what is new. + return True + + def get_events(self) -> list: + return list(self._batch) + + def to_event(self, raw_evt: dict): + eid = raw_evt["event_id"] + return make_event( + source=raw_evt["source"], + category="test_delta", # no decider registered -> emits directly + severity="routine", + title=eid, + summary=eid, + group_key=eid, + ) + + +def _make_store(adapter_name: str = "roads511"): + """Build a store with NO real adapters, then inject one fake adapter.""" + bus = EventBus() + captured: list = [] + bus.subscribe(lambda e: captured.append(e)) + + # All feeds default disabled -> zero native adapters register. + store = EnvironmentalStore(EnvironmentalConfig(), event_bus=bus) + adapter = _FakeAdapter(source="511") + store._adapters[adapter_name] = adapter + return store, adapter, captured + + +def _emitted_ids(captured) -> list[str]: + return [e.title for e in captured] + + +def test_first_poll_seeds_and_broadcasts_nothing(): + store, adapter, captured = _make_store() + adapter.set_batch(["A", "B", "C"]) + + store.refresh() # poll 1 — the backlog + + assert captured == [], "first poll must broadcast NOTHING (backlog seed)" + + +def test_second_poll_emits_only_newly_received(): + store, adapter, captured = _make_store() + + adapter.set_batch(["A", "B", "C"]) + store.refresh() # poll 1: seed + assert _emitted_ids(captured) == [] + + adapter.set_batch(["A", "B", "C", "D"]) + store.refresh() # poll 2: only D is new + assert _emitted_ids(captured) == ["D"] + + +def test_unchanged_poll_emits_nothing(): + store, adapter, captured = _make_store() + + adapter.set_batch(["A", "B", "C"]) + store.refresh() # poll 1: seed + adapter.set_batch(["A", "B", "C", "D"]) + store.refresh() # poll 2: D + adapter.set_batch(["A", "B", "C", "D"]) + store.refresh() # poll 3: nothing new + + assert _emitted_ids(captured) == ["D"], "poll 3 has no new items" + + +def test_restart_reseeds_and_never_rebroadcasts_backlog(): + # Process 1 sees A,B,C,D and broadcasts D. + store1, adapter1, cap1 = _make_store() + adapter1.set_batch(["A", "B", "C"]) + store1.refresh() + adapter1.set_batch(["A", "B", "C", "D"]) + store1.refresh() + assert _emitted_ids(cap1) == ["D"] + + # RESTART: a fresh store has an empty seen-set. The SAME backlog [A,B,C,D] + # arriving on its first poll must be re-seeded silently, not re-broadcast. + store2, adapter2, cap2 = _make_store() + adapter2.set_batch(["A", "B", "C", "D"]) + store2.refresh() + assert cap2 == [], "restart must NEVER re-broadcast the existing backlog" + + # And a genuinely new item after the restart still broadcasts once. + adapter2.set_batch(["A", "B", "C", "D", "E"]) + store2.refresh() + assert _emitted_ids(cap2) == ["E"] + + +def test_stable_key_prevents_reemit_when_batch_reorders(): + # The same real-world items in a different order are NOT "newly received". + store, adapter, captured = _make_store() + adapter.set_batch(["A", "B", "C"]) + store.refresh() # seed + adapter.set_batch(["C", "A", "B"]) # reordered, same items + store.refresh() + assert captured == [], "reordering the same items emits nothing" + + +def test_disabled_for_days_then_backlog_is_not_broadcast(): + # Simulate an adapter that was off for days: its first poll after coming + # back returns a large accumulated backlog. None of it may broadcast. + store, adapter, captured = _make_store() + backlog = [f"evt{i}" for i in range(200)] + adapter.set_batch(backlog) + store.refresh() # first poll after re-enable + assert captured == [], "a days-old backlog is seeded silently, never sent" + + # Only a truly new arrival afterward is announced. + adapter.set_batch(backlog + ["fresh"]) + store.refresh() + assert _emitted_ids(captured) == ["fresh"]