diff --git a/work/meshai/central/firms_handler.py b/work/meshai/central/firms_handler.py index be2eb9d..089ceb0 100644 --- a/work/meshai/central/firms_handler.py +++ b/work/meshai/central/firms_handler.py @@ -200,7 +200,7 @@ def handle_firms(envelope: dict, subject: str, table_name=None, table_pk=None) return None - # ---- persist (INSERT OR IGNORE via v4.sql unique partial index) ------ + # ---- persist + attribute + fuse (source-agnostic core) -------------- satellite = d.get("satellite") or "" brightness_raw = d.get("bright_ti4") if d.get("bright_ti4") is not None \ @@ -210,6 +210,69 @@ def handle_firms(envelope: dict, subject: str, except (TypeError, ValueError): brightness = None + # The INSERT-OR-IGNORE into firms_pixels + attribution + growth/spotting/ + # halt fusion now live in the shared _ingest_pixel_core so the native + # env/firms.py adapter feeds the SAME engine. handle_firms keeps the + # envelope-specific concerns (field extraction, filtering, event_log + # accounting) here so the Central path stays byte-identical. `data` is the + # consumer's mutable Event.data dict; core stamps it in place on a fusion + # broadcast exactly as the old inline path did. + stored, rowid, wire = _ingest_pixel_core( + conn, lat=lat, lon=lon, acq_epoch=acq_epoch, frp=frp, + confidence=conf, brightness=brightness, satellite=satellite, + data=data, now=now, + ) + + # event_log row regardless of dedup outcome -- both "stored" and + # "dedup-hit" count as "handled" for accounting; the suffix tells them + # apart for ops grep. (Written after the core call: event_log is a + # distinct table with its own rowid sequence, so ordering it after the + # attribution INSERTs leaves the row content + relative order identical.) + cat_tag = category_raw if stored else category_raw + "|dedup_hit" + _log_event(conn, now=now, source="firms", category=cat_tag, + severity_word=severity_word, + event_id_external=event_id_external, + subject=subject, handled=1, + table_name="firms_pixels" if stored else None, + table_pk=(str(rowid) if stored else None)) + + # Dedup hits skip broadcast -- the original insert already had its chance. + if not stored: + return None + return wire + + +# ============================================================================ +# Source-agnostic pixel ingest (Central + native env/firms.py share this) +# ============================================================================ + + +def _ingest_pixel_core(conn, *, lat, lon, acq_epoch, frp, confidence, + brightness, satellite, data, now): + """INSERT one canonical hotspot pixel + run attribution/fusion. + + This is the source-agnostic heart of the fire-fusion engine, extracted so + the Central NATS handler (``handle_firms``) and the native FIRMS adapter + (``env/firms.py``) drive IDENTICAL attribution/fusion from a bare pixel. + + Steps (exactly what the old inline ``handle_firms`` tail did per pixel): + 1. Compute the v7 meters-quantized ``dedup_key`` and ``INSERT OR IGNORE`` + into ``firms_pixels`` (unique on ``dedup_key, acq_time, satellite``). + A dedup hit is a no-op -- the original insert already ran attribution. + 2. For a NEWLY-stored pixel, run ``_attribute_or_cluster`` -> writes + ``fire_pixels`` / ``fire_passes`` / ``fires`` centroid+cursor and runs + the growth / spotting / halt fusion (honoring the Phase-3c deferred + latches). The dead unattributed-cluster path stays dead -- NO + raw-hotspot / cluster broadcast is ever produced here. + + Determinism: ``now`` is threaded explicitly; there is no hidden clock read. + + Returns ``(stored, rowid, wire)`` where ``wire`` is the fusion broadcast + string (or ``None``). ``data`` is mutated in place with the Phase-3c + broadcast stamps when a fusion wire is produced. + """ + satellite = satellite or "" + # v0.6-3b: dedup_key from meters-based quantization (v7 schema). dedup_distance_m = float(adapter_config.firms.dedup_distance_m) if dedup_distance_m > 0: @@ -224,37 +287,83 @@ def handle_firms(envelope: dict, subject: str, "frp, confidence, satellite, brightness, dedup_key) " "VALUES (?,?,?,?,?,?,?,?,?)", (None, lat, lon, acq_epoch, frp, - (str(conf) if conf is not None else None), + (str(confidence) if confidence is not None else None), satellite, brightness, dedup_key), ) stored = cur.rowcount > 0 - - # event_log row regardless of dedup outcome -- both "stored" and - # "dedup-hit" count as "handled" for accounting; the suffix tells them - # apart for ops grep. - handled = 1 if stored else 1 # dedup hit is still handled-success - cat_tag = category_raw if stored else category_raw + "|dedup_hit" - _log_event(conn, now=now, source="firms", category=cat_tag, - severity_word=severity_word, - event_id_external=event_id_external, - subject=subject, handled=handled, - table_name="firms_pixels" if stored else None, - table_pk=(str(cur.lastrowid) if stored else None)) - - # ---- v0.7-fire-tracker-1: attribution + cluster ------------------- - # Dedup hits skip attribution -- the original insert already had its - # chance. Only newly-stored pixels run through here. + rowid = int(cur.lastrowid) if not stored: - return None + return (False, rowid, None) - return _attribute_or_cluster( + wire = _attribute_or_cluster( conn, - pixel_row_id=int(cur.lastrowid), + pixel_row_id=rowid, lat=lat, lon=lon, acq_epoch=acq_epoch, frp=frp, satellite=satellite, data=data, now=now, ) + return (True, rowid, wire) + + +def ingest_hotspot_pixel(pixel: dict, *, now) -> list[tuple[str, dict]]: + """Source-agnostic entrypoint: ingest ONE canonical FIRMS pixel into the + fire-fusion engine and return the fusion broadcasts it produced. + + ``pixel`` is the canonical FIRMS schema:: + + {lat, lon, frp, confidence, brightness, satellite, acq_epoch} + + (``lat``/``lon``/``acq_epoch`` required; the rest optional.) This INSERTs + the pixel into ``firms_pixels`` (dedup-safe), runs attribution, and runs the + growth / spotting / halt fusion -- the same pipeline the Central handler + runs -- honoring the Phase-3c deferred latches. + + Returns a list of ``(wire, data)`` fusion broadcasts, where ``wire`` is the + mesh text and ``data`` carries the Phase-3c category/severity stamps (and, + under cutover, the deferred commit hook). A dedup hit or a pixel that + triggers no fusion returns ``[]``. This NEVER returns a raw-hotspot, + ``wildfire_hotspot``, ``new_ignition``, or cluster broadcast -- the only + outputs are ``wildfire_growth`` / ``wildfire_spotting`` / ``wildfire_halted`` + (the cluster path is dead). Callers must therefore NEVER broadcast the raw + pixel itself -- only these returned fusion wires. + + Determinism: ``now`` (epoch) is required and threaded straight through. + """ + try: + conn = get_db() + except Exception: + logger.exception("ingest_hotspot_pixel: persistence unavailable; dropping") + return [] + + lat = pixel.get("lat") + lon = pixel.get("lon") + acq_epoch = pixel.get("acq_epoch") + if not (isinstance(lat, (int, float)) and isinstance(lon, (int, float))): + return [] + if acq_epoch is None: + return [] + + try: + frp = float(pixel["frp"]) if pixel.get("frp") is not None else None + except (TypeError, ValueError): + frp = None + try: + brightness = float(pixel["brightness"]) \ + if pixel.get("brightness") is not None else None + except (TypeError, ValueError): + brightness = None + + data: dict = {} + _stored, _rowid, wire = _ingest_pixel_core( + conn, + lat=float(lat), lon=float(lon), acq_epoch=int(acq_epoch), + frp=frp, confidence=pixel.get("confidence"), brightness=brightness, + satellite=pixel.get("satellite") or "", data=data, now=now, + ) + if wire is None: + return [] + return [(wire, data)] # ============================================================================ diff --git a/work/meshai/env/firms.py b/work/meshai/env/firms.py index 04b9974..123dd58 100644 --- a/work/meshai/env/firms.py +++ b/work/meshai/env/firms.py @@ -37,6 +37,7 @@ class FIRMSAdapter: self._last_tick = 0.0 self._events = [] + self._fusion_events = [] # fire-fusion broadcasts (growth/spotting/halt) self._consecutive_errors = 0 self._last_error = None self._is_loaded = False @@ -119,10 +120,19 @@ class FIRMSAdapter: # Parse CSV response new_events = self._parse_csv(csv_data) - # Check if data changed + # Feed every fetched pixel into the SHARED source-agnostic fusion engine + # (central.firms_handler.ingest_hotspot_pixel): attribution + growth / + # spotting / halt. DB-level dedup (firms_pixels unique key) makes this + # idempotent across ticks, so re-fetched pixels never double-count. + # NEVER broadcasts raw hotspots -- only the fusion wires it returns. + self._fusion_events = self._run_fusion(new_events) + + # Check if data changed. Fusion output also counts as "changed" so the + # store ingests + emits the broadcast on this tick (store dedups its own + # emission by event_id, so an already-emitted fusion won't re-fire). old_ids = {e["event_id"] for e in self._events} new_ids = {e["event_id"] for e in new_events} - changed = old_ids != new_ids + changed = (old_ids != new_ids) or bool(self._fusion_events) self._events = new_events self._consecutive_errors = 0 @@ -344,48 +354,163 @@ class FIRMSAdapter: return (None, None) + def _satellite_code(self) -> str: + """Map the FIRMS source product to the satellite code the fusion + engine uses for pass bucketing + the firms_pixels dedup key. + + Consistency across pixels of one fetch is what matters (pass_id groups + an overpass); the exact code just needs to be stable per product.""" + s = (self._source or "").upper() + if "NOAA20" in s or "N20" in s or "J1" in s: + return "N20" + if "NOAA21" in s or "N21" in s or "J2" in s: + return "N21" + if "SNPP" in s or "SUOMI" in s or "NPP" in s: + return "N" + if "MODIS" in s: + return "MODIS" + return self._source or "?" + + def _run_fusion(self, raw_events: list) -> list: + """Feed each fetched hotspot pixel into the SHARED attribution/fusion + engine and collect the fire-fusion broadcasts. + + Every parsed pixel is mapped to the canonical FIRMS schema and passed to + ``central.firms_handler.ingest_hotspot_pixel`` -- the exact same engine + the Central NATS handler drives. The engine INSERTs the pixel, runs + attribution, and runs the growth / spotting / halt fusion; it returns + ONLY fusion wires (never a raw-hotspot / cluster broadcast). DB-level + dedup makes re-fetched pixels no-ops, so nothing double-counts. + """ + try: + from meshai.central.firms_handler import ( + ingest_hotspot_pixel, _parse_acq_epoch, + ) + except Exception: + logger.exception("FIRMS fusion: handler import failed; skipping") + return [] + + now = time.time() + satellite = self._satellite_code() + fusion: list = [] + for evt in raw_events: + props = evt.get("properties", {}) or {} + acq_epoch = _parse_acq_epoch(props.get("acq_date"), + props.get("acq_time")) + if acq_epoch is None: + continue + pixel = { + "lat": evt.get("lat"), + "lon": evt.get("lon"), + "frp": props.get("frp"), + "confidence": props.get("confidence"), + "brightness": props.get("brightness"), + "satellite": satellite, + "acq_epoch": acq_epoch, + } + try: + broadcasts = ingest_hotspot_pixel(pixel, now=int(now)) + except Exception: + logger.exception("FIRMS fusion: ingest failed for %s", + evt.get("event_id")) + continue + for wire, data in broadcasts: + fusion.append(self._make_fusion_event(wire, data, evt, now)) + return fusion + + def _make_fusion_event(self, wire: str, data: dict, source_evt: dict, + now: float) -> dict: + """Wrap a fusion broadcast (wire, data) in an adapter event dict. + + Carries source ``firms_fusion`` (distinct from raw ``firms`` so it never + pollutes the hotspot LLM summary / get_active(source="firms")) and a + ``_fusion`` payload that ``to_event`` turns into a precomposed Event.""" + category = data.get("category") + severity = (data.get("severity") + or data.get("_severity_override") or "routine") + base_id = source_evt.get("event_id") \ + or f"{source_evt.get('lat')}_{source_evt.get('lon')}" + # Precomposed marker: the composer passes event.title through verbatim + # (same contract satpass native + the Central consumer use), so the wire + # the shared engine produced is what the mesh sees. + data["_meshai_precomposed"] = True + return { + "source": "firms_fusion", + "event_id": f"{category}:{base_id}", + "event_type": "Fire Fusion", + "severity": severity, + "headline": wire, + "lat": source_evt.get("lat"), + "lon": source_evt.get("lon"), + "expires": now + 21600, + "fetched_at": now, + "properties": {"new_ignition": False, "category": category}, + "_fusion": {"wire": wire, "data": data, "category": category, + "severity": severity}, + } + def to_event(self, evt: dict) -> Optional["Event"]: - """Attribution-only: the native FIRMS adapter NEVER broadcasts hotspots. + """Emit ONLY fire-fusion broadcasts; NEVER a raw hotspot. Firm rule (Matt): "we do NOT broadcast hotspots." A raw satellite - thermal pixel -- a single-pixel ``wildfire_hotspot`` or - ``new_ignition`` detection -- is noisy and not actionable on its own; - broadcasting it would flood the mesh with unattributed heat. This - mirrors the Central path (``central/firms_handler.py``), which is - storage-only and returns None for every raw pixel. The ONLY FIRMS - signals that ever reach the mesh are the fire-tracker FUSION outputs - (``wildfire_growth`` / ``wildfire_spotting`` / ``wildfire_halted``), - produced by the Central handler's attribution engine -- NOT here. + thermal pixel is noisy and not actionable on its own, so a bare + ``wildfire_hotspot`` / ``new_ignition`` is NEVER emitted -- this returns + None for every raw pixel (identical to the Central storage-only path). - This method therefore ALWAYS returns None. The neutralization is - UNCONDITIONAL (not behind any config flag): the no-hotspots rule is - absolute. Previously this emitted ``make_event(category="new_ignition" - if new_ignition else "wildfire_hotspot", ...)``, and because - ``store._emit_event`` only consults a gating decider for cut-over - categories -- and NO decider is registered for the raw hotspot - categories (see ``notifications/gating/__init__.py``: "native - env/fires.py hotspot broadcasts ... are NOT migrated") -- those Events - went straight to the bus and out to the mesh. Returning None removes - that broadcast on the native path entirely. + What DID change: native FIRMS now feeds the SHARED attribution/fusion + engine (see ``_run_fusion`` -> ``firms_handler.ingest_hotspot_pixel``). + The fire-tracker FUSION outputs (``wildfire_growth`` / + ``wildfire_spotting`` / ``wildfire_halted``) it produces ARE emitted + here, as precomposed Events, so the native/standalone deployment gets + the same fire signals the Central handler produces. Raw hotspots are + distinguished by the absence of a ``_fusion`` payload on the event dict + (they carry source ``firms``; fusion dicts carry source + ``firms_fusion``); the raw dicts remain available via ``get_events()`` / + ``get_new_ignitions()`` for LLM context and health reporting. - The raw hotspot dicts remain available in-memory via ``get_events()`` - / ``get_new_ignitions()`` for LLM context and health reporting; they - just never become a broadcastable Event. - - STANDALONE GAP (not fixed here): unlike the Central handler, the native - adapter has NO fusion wiring -- it does not persist ``firms_pixels`` or - run attribution / clustering / pass-boundary growth. Native FIRMS - therefore feeds NOTHING into the fire tracker today; the cross-ref to - known NIFC fires only sets the (now unused for broadcast) - ``new_ignition`` flag on the in-memory dict. Wiring native pixels into - the attribution engine so growth/spotting/halt can eventually fire is a - separate, larger effort. + The fusion categories render via the EXISTING Phase-3c + formatters/gating; when they are NOT cut over (the default, and FIRMS's + documented state) ``store._emit_event`` emits directly and the decision + already made inside the engine stands. (If those categories were cut + over, ``store._emit_event`` would re-run the decider on event.data -- + which lacks the internal ``_kind`` -- and suppress; FIRMS shadow/cutover + remains a deferred follow-up, unchanged by this wiring.) """ - return None + fusion = evt.get("_fusion") + if not fusion: + return None # raw hotspot -- never broadcast + try: + wire = fusion["wire"] + data = fusion["data"] + category = fusion["category"] + severity = fusion["severity"] + event_id = evt.get("event_id") + return make_event( + source="firms", + category=category, + severity=severity, + title=wire, # precomposed: composer passes it verbatim + summary=wire, + lat=evt.get("lat"), + lon=evt.get("lon"), + timestamp=evt.get("fetched_at"), + expires=evt.get("expires"), + group_key=event_id, + inhibit_keys=[event_id] if event_id else [], + data=data, # carries Phase-3c stamps + precomposed marker + ) + except Exception: + logger.exception("FIRMS fusion: to_event failed") + return None def get_events(self) -> list: - """Get current hotspot events.""" - return self._events + """Get current hotspot events + fire-fusion broadcasts. + + Raw hotspots (source ``firms``) render None from ``to_event`` and are + kept only for LLM context / health; fire-fusion broadcasts (source + ``firms_fusion``) render real Events. The store keys on + ``(source, event_id)`` and emits each once.""" + return self._events + self._fusion_events def get_new_ignitions(self) -> list: """Get only potential new ignitions (not near known fires).""" diff --git a/work/tests/test_firms_native_fusion.py b/work/tests/test_firms_native_fusion.py new file mode 100644 index 0000000..572a1c1 --- /dev/null +++ b/work/tests/test_firms_native_fusion.py @@ -0,0 +1,417 @@ +"""Native FIRMS fire-fusion tests — the source-agnostic ingest entrypoint. + +Covers the wiring that lets the native ``env/firms.py`` adapter feed +locally-fetched NASA FIRMS pixels into the SAME attribution/fusion pipeline the +Central handler uses, via ``central.firms_handler.ingest_hotspot_pixel``: + +1. ``ingest_hotspot_pixel`` drives growth / spotting / halt from canonical + pixels and returns ``(wire, data)`` fusion broadcasts — and NEVER a raw + hotspot / cluster / new_ignition broadcast. +2. ``env/firms.py`` tick() (CSV fetch monkeypatched) feeds those pixels through + the shared engine, emits the fusion Events, and still returns None for raw + hotspots. +3. A guard that the Central ``handle_firms`` path is unchanged by the + extraction (storage side-effects + growth wire/stamps identical). + +Mirrors the driving style of test_firms_refactor.py / test_fire_tracker_phase*. +""" +from __future__ import annotations + +import math +import time +import uuid +from datetime import datetime, timezone + +import pytest + +_MI_PER_DEG_LAT = 69.0 +_FUSION_CATS = {"wildfire_growth", "wildfire_spotting", "wildfire_halted"} +_RAW_CATS = {"wildfire_hotspot", "new_ignition", "unattributed_hotspot_cluster"} + + +# ── isolation (real-DB, mirrors test_firms_refactor) ───────────────────────── + +@pytest.fixture(autouse=True) +def _isolate_db(tmp_path, monkeypatch): + db_path = str(tmp_path / f"meshai-{uuid.uuid4().hex}.sqlite") + monkeypatch.setenv("MESHAI_DB_PATH", db_path) + from meshai.persistence import db as pdb + pdb.close_thread_connection() + pdb._initialised.discard(db_path) + from meshai.persistence import init_db + init_db(db_path) + try: + from meshai.adapter_config import adapter_config as _ac + _ac.invalidate() + except Exception: + pass + yield db_path + pdb.close_thread_connection() + pdb._initialised.discard(db_path) + + +@pytest.fixture(autouse=True) +def _no_cutover(monkeypatch): + """Default deploy state: nothing cut over -> legacy stamps, direct emit.""" + monkeypatch.delenv("MESHAI_CUTOVER_CATEGORIES", raising=False) + from meshai.notifications.cutover import _clear_cache + _clear_cache() + yield + _clear_cache() + + +# ── helpers ────────────────────────────────────────────────────────────────── + +def _seed_fire(*, irwin_id, lat, lon, name="Stub Fire", **cols): + from meshai.persistence import get_db + conn = get_db() + base = {"irwin_id": irwin_id, "incident_name": name, "lat": lat, "lon": lon, + "last_event_at": int(time.time())} + base.update(cols) + keys = ",".join(base) + ph = ",".join("?" * len(base)) + conn.execute(f"INSERT INTO fires({keys}) VALUES ({ph})", tuple(base.values())) + + +def _acq_epoch(acq_date: str, acq_time: str) -> int: + return int(datetime.strptime(f"{acq_date} {acq_time.zfill(4)}", + "%Y-%m-%d %H%M") + .replace(tzinfo=timezone.utc).timestamp()) + + +def _pixel(*, lat, lon, acq_date="2026-06-06", acq_time="1200", + frp=20.0, confidence="high", brightness=320.0, satellite="N20"): + """Build a canonical FIRMS pixel dict.""" + return { + "lat": lat, "lon": lon, "frp": frp, "confidence": confidence, + "brightness": brightness, "satellite": satellite, + "acq_epoch": _acq_epoch(acq_date, acq_time), + } + + +def _offset_mi(lat, lon, north_mi, east_mi): + dlat = north_mi / _MI_PER_DEG_LAT + dlon = east_mi / (_MI_PER_DEG_LAT * math.cos(math.radians(lat))) + return lat + dlat, lon + dlon + + +def _feed(pixel, *, now): + from meshai.central.firms_handler import ingest_hotspot_pixel + return ingest_hotspot_pixel(pixel, now=now) + + +def _assert_no_raw(broadcasts): + """Every returned broadcast must be a fusion category; never a raw one.""" + for wire, data in broadcasts: + cat = data.get("category") + assert cat in _FUSION_CATS, f"unexpected non-fusion broadcast: {cat!r}" + assert cat not in _RAW_CATS + + +# ═════════════════════════════════════════════════════════════════════════════ +# 1. ingest_hotspot_pixel — growth / spotting / halt +# ═════════════════════════════════════════════════════════════════════════════ + +class TestIngestGrowth: + def test_two_pass_growth_broadcasts(self): + center_lat, center_lon = 42.0, -114.0 + _seed_fire(irwin_id="ID-NG", lat=center_lat, lon=center_lon, + name="Pine Gulch") + # Pass A: 5 pixels around the anchor (acq 12:00-12:04 = one pass bucket). + for i in range(5): + out = _feed(_pixel(lat=center_lat + 0.0001 * i, + lon=center_lon + 0.0001 * (i - 2), + acq_time=f"12{i:02d}", frp=20.0 + i), + now=1780747200 + i) + _assert_no_raw(out) + assert out == [], "pass-A pixels must not broadcast (no boundary yet)" + # Pass B: one pixel 1 mi north (18:00 = different pass bucket) -> growth. + pass_b_lat = center_lat + (1.0 / _MI_PER_DEG_LAT) + out = _feed(_pixel(lat=pass_b_lat, lon=center_lon, acq_time="1800", + frp=22.0), now=1780768800) + _assert_no_raw(out) + assert len(out) == 1 + wire, data = out[0] + assert data["category"] == "wildfire_growth" + assert data["_severity_override"] == "immediate" + assert wire.startswith("🔥 Pine Gulch") + assert "Moving N" in wire + + +class TestIngestSpotting: + def _seed_hex_pass_a(self, irwin_id, center_lat, center_lon, + start_now=1780747200): + _seed_fire(irwin_id=irwin_id, lat=center_lat, lon=center_lon, + name=irwin_id) + for i in range(6): + angle = i * math.pi / 3 + la = center_lat + (0.5 / _MI_PER_DEG_LAT) * math.sin(angle) + cos_lat = math.cos(math.radians(center_lat)) + lo = center_lon + (0.5 / (_MI_PER_DEG_LAT * cos_lat)) * math.cos(angle) + out = _feed(_pixel(lat=la, lon=lo, acq_time=f"12{i * 2:02d}"), + now=start_now + i) + _assert_no_raw(out) + + def test_spotting_broadcasts(self): + center_lat, center_lon = 43.0, -115.0 + self._seed_hex_pass_a("ID-NS", center_lat, center_lon) + # Pass B pixel ~2 mi NE of the perimeter -> spotting. + sp_lat, sp_lon = _offset_mi(center_lat, center_lon, + north_mi=2.0 / math.sqrt(2), + east_mi=2.0 / math.sqrt(2)) + out = _feed(_pixel(lat=sp_lat, lon=sp_lon, acq_time="1800"), + now=1780768800) + _assert_no_raw(out) + assert len(out) == 1 + wire, data = out[0] + assert data["category"] == "wildfire_spotting" + assert data["severity"] == "immediate" + assert wire.startswith("🔥 Possible spotting ") + # Eager latch stamped during ingest (not-cutover legacy path). + from meshai.persistence import get_db + latch = get_db().execute( + "SELECT last_spotting_broadcast_at FROM fires WHERE irwin_id=?", + ("ID-NS",)).fetchone()[0] + assert latch == 1780768800.0 + + +class TestIngestHalt: + def test_halt_broadcasts_for_idle_fire(self): + now = 1780768800 + idle_at = now - 14 * 3600 + from meshai.persistence import get_db + get_db().execute( + "INSERT INTO fires(irwin_id, incident_name, lat, lon, " + "last_event_at, last_pass_id, last_pass_at) VALUES (?,?,?,?,?,?,?)", + ("ID-NH", "Cold Fire", 42.5, -114.5, int(idle_at), + "N20-329627", float(idle_at)), + ) + # A fresh pixel FAR from the idle fire -> no attribution, cluster dead, + # halt detector fires for the idle fire. + out = _feed(_pixel(lat=45.0, lon=-118.0, acq_time="1800"), now=now) + _assert_no_raw(out) + assert len(out) == 1 + wire, data = out[0] + assert data["category"] == "wildfire_halted" + assert data["severity"] == "routine" + assert wire == "🔥 Cold Fire no growth in 14h" + latch = get_db().execute( + "SELECT halt_broadcast_at FROM fires WHERE irwin_id=?", + ("ID-NH",)).fetchone()[0] + assert latch == float(now) + + +class TestIngestNeverRawOrCluster: + def test_lone_pixel_no_fire_yields_nothing(self): + # Stored, attributed to nothing, cluster path DEAD, no idle fire -> + # returns []. A raw hotspot is NEVER emitted on any path. + out = _feed(_pixel(lat=44.4, lon=-116.2, acq_time="1300"), + now=1780750000) + assert out == [] + + def test_dense_unattributed_cluster_stays_silent(self): + # Several unattributed pixels close together would have tripped the old + # cluster broadcast; that path is dead -> no broadcast, ever. + base_lat, base_lon = 44.0, -116.0 + produced = [] + for i in range(6): + out = _feed(_pixel(lat=base_lat + 0.001 * i, lon=base_lon, + acq_time=f"13{i:02d}"), now=1780750000 + i) + _assert_no_raw(out) + produced.extend(out) + assert produced == [] + + +# ═════════════════════════════════════════════════════════════════════════════ +# 2. env/firms.py tick() — shared engine + raw-hotspot suppression +# ═════════════════════════════════════════════════════════════════════════════ + +def _adapter(source="VIIRS_NOAA20_NRT"): + from unittest.mock import MagicMock + from meshai.env.firms import FIRMSAdapter + cfg = MagicMock() + cfg.map_key = "test-key" + cfg.source = source + cfg.bbox = [-117, 42, -114, 44] + cfg.day_range = 1 + cfg.tick_seconds = 0 + cfg.confidence_min = "nominal" + cfg.proximity_km = 10.0 + a = FIRMSAdapter(cfg, region_anchors=[], fires_adapter=None) + a._last_tick = 0.0 + return a + + +def _csv(rows): + header = "latitude,longitude,bright_ti4,confidence,frp,acq_date,acq_time" + lines = [header] + for r in rows: + lines.append("{lat},{lon},{b},{c},{frp},{d},{t}".format( + lat=r["lat"], lon=r["lon"], b=r.get("b", 320.0), + c=r.get("c", "high"), frp=r.get("frp", 20.0), + d=r.get("d", "2026-06-06"), t=r.get("t", "1200"))) + return "\n".join(lines) + + +class _FakeResp: + def __init__(self, text): + self._text = text + def read(self): + return self._text.encode("utf-8") + def __enter__(self): + return self + def __exit__(self, *a): + return False + + +def _patch_fetch(monkeypatch, csv_text): + monkeypatch.setattr("meshai.env.firms.urlopen", + lambda *a, **k: _FakeResp(csv_text)) + + +class TestAdapterTickFusion: + def _growth_rows(self, center_lat, center_lon): + # Anchor acq times to NOW (native path uses now=time.time()): pass A + # ~6h ago, pass B ~now. Both are well inside the 12h halt gate, so the + # growing fire never looks idle -- only the growth broadcast fires. + # (6h apart => distinct pass buckets => a real pass boundary.) + now_dt = datetime.now(timezone.utc) + pass_a = now_dt - __import__("datetime").timedelta(hours=6) + rows = [] + for i in range(5): + t = pass_a + __import__("datetime").timedelta(minutes=i) + rows.append({"lat": center_lat + 0.0001 * i, + "lon": center_lon + 0.0001 * (i - 2), + "d": t.strftime("%Y-%m-%d"), "t": t.strftime("%H%M"), + "frp": 20.0 + i}) + rows.append({"lat": center_lat + 1.0 / _MI_PER_DEG_LAT, + "lon": center_lon, + "d": now_dt.strftime("%Y-%m-%d"), + "t": now_dt.strftime("%H%M"), "frp": 22.0}) + return rows + + def test_tick_emits_growth_and_suppresses_raw(self, monkeypatch): + center_lat, center_lon = 42.0, -114.0 + _seed_fire(irwin_id="ID-TG", lat=center_lat, lon=center_lon, + name="Pine Gulch") + _patch_fetch(monkeypatch, _csv(self._growth_rows(center_lat, center_lon))) + + a = _adapter() + assert a.tick() is True + + evts = a.get_events() + raw = [e for e in evts if e.get("source") == "firms"] + fusion = [e for e in evts if e.get("source") == "firms_fusion"] + # Raw pixels present (LLM context) but every one renders None. + assert raw, "raw hotspots should still be cached for LLM context" + for r in raw: + assert a.to_event(r) is None + # Exactly one fusion broadcast, and it renders a real growth Event. + assert len(fusion) == 1 + ev = a.to_event(fusion[0]) + assert ev is not None + assert ev.category == "wildfire_growth" + assert ev.summary.startswith("🔥 Pine Gulch") + assert ev.data.get("_meshai_precomposed") is True + + def test_tick_emits_through_store_emit_event(self, monkeypatch): + """End-to-end: the fusion Event reaches the pipeline bus via + store._emit_event; raw hotspots never do.""" + from unittest.mock import MagicMock + from meshai.env.store import EnvironmentalStore + + center_lat, center_lon = 42.0, -114.0 + _seed_fire(irwin_id="ID-TS", lat=center_lat, lon=center_lon, + name="Pine Gulch") + _patch_fetch(monkeypatch, _csv(self._growth_rows(center_lat, center_lon))) + + a = _adapter() + bus = MagicMock() + store = EnvironmentalStore.__new__(EnvironmentalStore) + store._events = {} + store._event_bus = bus + assert a.tick() is True + store._ingest("firms", a) + + emitted = [c.args[0] for c in bus.emit.call_args_list] + cats = [e.category for e in emitted] + assert cats == ["wildfire_growth"], cats + assert emitted[0].summary.startswith("🔥 Pine Gulch") + + def test_tick_with_no_fire_emits_nothing(self, monkeypatch): + # Pixels with no seeded fire: attribution misses, cluster dead -> + # zero fusion, and raw hotspots still suppressed. + _patch_fetch(monkeypatch, _csv([ + {"lat": 43.5, "lon": -115.5, "t": "1300"}, + {"lat": 43.6, "lon": -115.4, "t": "1305"}, + ])) + a = _adapter() + a.tick() + fusion = [e for e in a.get_events() if e.get("source") == "firms_fusion"] + assert fusion == [] + + +# ═════════════════════════════════════════════════════════════════════════════ +# 3. Central-path guard — extraction did not change handle_firms +# ═════════════════════════════════════════════════════════════════════════════ + +def _envelope(*, lat, lon, acq_date="2026-06-06", acq_time="1200", + frp=20.0, satellite="N20"): + return { + "data": { + "adapter": "firms", "category": "wildfire_hotspot", + "severity": "routine", + "data": { + "latitude": lat, "longitude": lon, "frp": frp, + "bright_ti4": 320.0, "satellite": satellite, + "instrument": "VIIRS", "confidence": "high", + "acq_date": acq_date, "acq_time": acq_time, + "daynight": "D", "version": "2.0NRT", + }, + } + } + + +_SUBJECT = "central.fire.hotspot.N20.high.us.id" + + +class TestCentralPathUnchanged: + def test_storage_only_pixel_still_stores_and_returns_none(self): + from meshai.central.firms_handler import handle_firms + from meshai.persistence import get_db + env = _envelope(lat=42.19664, lon=-113.70981) + out = handle_firms(env, subject=_SUBJECT, data={}, now=1780660000) + assert out is None # no fire seeded -> storage only, no broadcast + n = get_db().execute( + "SELECT COUNT(*) AS n FROM firms_pixels").fetchone()["n"] + assert n == 1 + log = get_db().execute( + "SELECT table_name, table_pk, handled FROM event_log " + "ORDER BY id DESC LIMIT 1").fetchone() + assert log["handled"] == 1 + assert log["table_name"] == "firms_pixels" + row_id = get_db().execute( + "SELECT id FROM firms_pixels").fetchone()["id"] + assert log["table_pk"] == str(row_id) + + def test_central_growth_wire_and_stamps_identical(self): + """The extracted core produces the SAME growth wire/stamps on the + Central envelope path that the inline handler did.""" + from meshai.central.firms_handler import handle_firms + center_lat, center_lon = 42.0, -114.0 + _seed_fire(irwin_id="ID-CG", lat=center_lat, lon=center_lon, + name="Pine Gulch") + for i in range(5): + handle_firms(_envelope(lat=center_lat + 0.0001 * i, + lon=center_lon + 0.0001 * (i - 2), + acq_time=f"12{i:02d}", frp=20.0 + i), + subject=_SUBJECT, data={}, now=1780747200 + i) + data = {} + wire = handle_firms(_envelope(lat=center_lat + 1.0 / _MI_PER_DEG_LAT, + lon=center_lon, acq_time="1800", frp=22.0), + subject=_SUBJECT, data=data, now=1780768800) + assert wire is not None and wire.startswith("🔥 Pine Gulch") + assert "Moving N" in wire + assert data["category"] == "wildfire_growth" + assert data["_severity_override"] == "immediate" + assert data["_cooldown_suffix"] == "ID-CG"