diff --git a/work/meshai/central/firms_handler.py b/work/meshai/central/firms_handler.py index 089ceb0..e868622 100644 --- a/work/meshai/central/firms_handler.py +++ b/work/meshai/central/firms_handler.py @@ -248,7 +248,7 @@ def handle_firms(envelope: dict, subject: str, def _ingest_pixel_core(conn, *, lat, lon, acq_epoch, frp, confidence, - brightness, satellite, data, now): + brightness, satellite, data, now, seed=False): """INSERT one canonical hotspot pixel + run attribution/fusion. This is the source-agnostic heart of the fire-fusion engine, extracted so @@ -262,8 +262,14 @@ def _ingest_pixel_core(conn, *, lat, lon, acq_epoch, frp, confidence, 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. + latches). On an attribution MISS it runs the curated + unattributed-cluster path (``_maybe_emit_cluster``). ``seed=True`` + (cold-start silent-seed) suppresses ALL broadcast output on the first + fetch -- cluster AND growth/spotting/halt fusion -- while persistence, + attribution, per-pass aggregation, centroid/cursor updates, and the + dedup latches (cluster_broadcast_at, halt_broadcast_at) still run so a + genuinely-new post-seed event broadcasts normally. A raw per-pixel + hotspot is NEVER broadcast. Determinism: ``now`` is threaded explicitly; there is no hidden clock read. @@ -301,12 +307,12 @@ def _ingest_pixel_core(conn, *, lat, lon, acq_epoch, frp, confidence, lat=lat, lon=lon, acq_epoch=acq_epoch, frp=frp, satellite=satellite, - data=data, now=now, + data=data, now=now, seed=seed, ) return (True, rowid, wire) -def ingest_hotspot_pixel(pixel: dict, *, now) -> list[tuple[str, dict]]: +def ingest_hotspot_pixel(pixel: dict, *, now, seed=False) -> list[tuple[str, dict]]: """Source-agnostic entrypoint: ingest ONE canonical FIRMS pixel into the fire-fusion engine and return the fusion broadcasts it produced. @@ -323,10 +329,23 @@ def ingest_hotspot_pixel(pixel: dict, *, now) -> list[tuple[str, dict]]: 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. + ``wildfire_hotspot``, or ``new_ignition`` broadcast -- the possible outputs + are ``wildfire_growth`` / ``wildfire_spotting`` / ``wildfire_halted`` (an + attributed pixel grew/moved a known fire) OR a curated + ``unattributed_hotspot_cluster`` ("possible new fire") when ``cluster_min_pixels`` + unattributed pixels fall within ``cluster_max_radius_mi`` over + ``cluster_time_window_minutes``. Callers must therefore NEVER broadcast the + raw pixel itself -- only these returned fusion/cluster wires. + + ``seed`` (cold-start silent-seed): on the FIRST fetch after boot, pass + ``seed=True`` so the day's pre-existing hotspots emit NOTHING -- no cold-start + dump of ANY kind. This covers BOTH the curated cluster path AND the + attributed growth / spotting / halt fusion: a cluster or idle-fire halt that + forms from pre-existing pixels is stamped (never re-fires); a growth/spotting + that a pre-existing pass boundary would have produced is withheld. Only + genuinely-new activity arriving on a LATER (``seed=False``) fetch broadcasts. + Persistence, attribution, per-pass aggregation, and dedup baselines run + regardless of ``seed``. Determinism: ``now`` (epoch) is required and threaded straight through. """ @@ -359,7 +378,7 @@ def ingest_hotspot_pixel(pixel: dict, *, now) -> list[tuple[str, dict]]: 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, + satellite=pixel.get("satellite") or "", data=data, now=now, seed=seed, ) if wire is None: return [] @@ -465,8 +484,17 @@ def _log_event(conn, *, now, source, category, severity_word, def _attribute_or_cluster(conn, *, pixel_row_id, lat, lon, acq_epoch, - frp, satellite, data, now): - """Try attribution; on miss, run cluster check. Returns wire str | None.""" + frp, satellite, data, now, seed=False): + """Try attribution; on miss, run cluster check. Returns wire str | None. + + Attribution ALWAYS runs first (a pixel inside a known WFIGS fire's spread + radius grows that fire and never forms a "new" cluster). ``seed`` (cold-start + silent-seed) suppresses ALL broadcast output on the first fetch -- the + unattributed-cluster branch (see ``_maybe_emit_cluster``) AND the attributed + fusion path (growth / spotting / halt). Persistence, attribution, per-pass + aggregation, centroid/cursor updates, and any dedup latches that must survive + to gate a LATER broadcast all still run; only the wire emission is withheld. + """ global_default_mi = float(adapter_config.fires.spread_radius_mi_default) # Conservative bbox prefilter: take the larger of the global default # and 10 mi so a per-fire override beyond the default doesn't get @@ -528,22 +556,22 @@ def _attribute_or_cluster(conn, *, pixel_row_id, lat, lon, acq_epoch, wire = _handle_pass_boundary( conn, irwin_id=chosen_irwin, pass_id=this_pass_id, lat=lat, lon=lon, acq_epoch=acq_epoch, frp=frp, - data=data, now=now, + data=data, now=now, seed=seed, ) if wire is not None: return wire # No growth broadcast; opportunistically run halt detector for # OTHER fires that may have gone idle. - return _maybe_emit_halt(conn, data=data, now=now) + return _maybe_emit_halt(conn, data=data, now=now, seed=seed) # 0 matches -- run cluster detection. wire = _maybe_emit_cluster( conn, lat=lat, lon=lon, acq_epoch=acq_epoch, frp=frp, - data=data, now=now, this_pixel_id=pixel_row_id, + data=data, now=now, this_pixel_id=pixel_row_id, seed=seed, ) if wire is not None: return wire - return _maybe_emit_halt(conn, data=data, now=now) + return _maybe_emit_halt(conn, data=data, now=now, seed=seed) def _recompute_centroid_and_stamp(conn, irwin_id: str, *, @@ -572,10 +600,18 @@ def _recompute_centroid_and_stamp(conn, irwin_id: str, *, def _maybe_emit_cluster(conn, *, lat, lon, acq_epoch, frp, data, now, - this_pixel_id): + this_pixel_id, seed=False): """Return wire string + set data["category"] when a cluster condition - fires; otherwise return None and leave data alone.""" - return None + fires; otherwise return None and leave data alone. + + ``seed`` (cold-start silent-seed): when True, a cluster that meets the + threshold is still STAMPED (``cluster_broadcast_at`` on every member) so it + can never re-fire on a later fetch, but NO wire is returned and ``data`` is + left untouched -- i.e. the day's pre-existing hotspots discovered on the + first fetch after boot are absorbed silently. Only clusters that form from + pixels arriving on a LATER (non-seed) fetch broadcast. Persistence + + attribution are unaffected (they run in the caller before this point). + """ min_pixels = int(adapter_config.firms.cluster_min_pixels) radius_mi = float(adapter_config.firms.cluster_max_radius_mi) window_s = int(adapter_config.firms.cluster_time_window_minutes) * 60 @@ -626,6 +662,16 @@ def _maybe_emit_cluster(conn, *, lat, lon, acq_epoch, frp, data, now, (float(now), *member_ids), ) + # Cold-start silent-seed: the cluster is stamped (above) so it can never + # re-fire, but on the first fetch after boot we emit NOTHING and leave + # `data` untouched -- the day's pre-existing hotspots are absorbed silently. + if seed: + logger.info( + "firms cold-start silent-seed: absorbed %d-pixel cluster " + "@ %.3f,%.3f (stamped, no broadcast)", + len(members), centroid_lat, centroid_lon) + return None + # Override the FIRMS source category so the dispatcher routes this # broadcast under unattributed_hotspot_cluster (priority, fire toggle). if isinstance(data, dict): @@ -709,8 +755,16 @@ def _pass_id(satellite, acq_epoch) -> str: def _handle_pass_boundary(conn, *, irwin_id, pass_id, lat, lon, - acq_epoch, frp, data, now): - """Maintain fire_passes row, detect boundary, fire growth on drift.""" + acq_epoch, frp, data, now, seed=False): + """Maintain fire_passes row, detect boundary, fire growth on drift. + + ``seed`` (cold-start silent-seed): all state -- the fire_passes upsert, the + ``fires`` cursor/centroid update, and the prior-pass perimeter close -- runs + unconditionally so the growth/spotting dedup baselines are correct. Only the + wire EMISSION (growth here, spotting in ``_check_spotting``) is withheld on + the first fetch. Growth has no latch, so a genuinely-new post-seed pass + boundary fires normally on a later fetch. + """ # (1) Recompute the pass aggregate from fire_pixels. pass_rows = conn.execute( "SELECT lat, lon, frp, acq_time FROM fire_pixels " @@ -818,7 +872,7 @@ def _handle_pass_boundary(conn, *, irwin_id, pass_id, lat, lon, conn, irwin_id=irwin_id, pixel_lat=lat, pixel_lon=lon, current_pass_id=pass_id, incident_name=fires_row["incident_name"] or "(unnamed fire)", - data=data, now=now, + data=data, now=now, seed=seed, ) if spotting_wire is not None: return spotting_wire @@ -844,6 +898,16 @@ def _handle_pass_boundary(conn, *, irwin_id, pass_id, lat, lon, if not gate.broadcast: return None + # Cold-start silent-seed: the pass aggregate + fires cursor/centroid were + # already updated above (the growth baseline is intact), but on the first + # fetch after boot we emit NOTHING and leave `data` untouched. Growth has no + # latch, so a genuinely-new post-seed pass boundary broadcasts normally. + if seed: + logger.info( + "firms cold-start silent-seed: absorbed growth for %s " + "(state updated, no broadcast)", irwin_id) + return None + # Drift exceeded the threshold -- emit wildfire_growth via WFIGS renderer. fire = conn.execute( "SELECT incident_name, current_acres, current_contained_pct, " @@ -902,11 +966,21 @@ def _render_growth_wire(*, incident_name, direction, speed_mph, ) -def _maybe_emit_halt(conn, *, data, now): +def _maybe_emit_halt(conn, *, data, now, seed=False): """Find one fire matching the halt criteria, latch + broadcast. Returns the wire string when a halt event fires; otherwise None. + ``seed`` (cold-start silent-seed): halt is triggered opportunistically by + ANY pixel arrival (it scans all fires for idleness, independent of the + current pixel), so a fire that was already idle at boot WOULD otherwise + halt-broadcast on the next fetch's unrelated new pixel. To absorb that + pre-existing idle state -- mirroring how the cluster seed stamps + ``cluster_broadcast_at`` -- the first fetch STILL latches + ``halt_broadcast_at`` but emits NOTHING. Re-eligibility is unchanged: once + real post-seed activity advances ``last_pass_at`` past the stamp, the fire + can halt again legitimately. + Phase-3c: the DECISION (the halt-eligibility SELECT) is delegated to ``gating.firms.decide``; the wire is still rendered inline here for byte-identity. On the not-cutover live path the eager @@ -925,6 +999,22 @@ def _maybe_emit_halt(conn, *, data, now): p = gate.data_patch irwin_id = p["irwin_id"] + + # Cold-start silent-seed: stamp the latch (cutover-agnostic, eager) so this + # pre-existing idle fire is absorbed and never halt-broadcasts on a later + # fetch, but emit NOTHING and leave `data` untouched. It becomes halt- + # eligible again once real post-seed activity advances last_pass_at past + # this stamp -- exactly the cluster seed's "stamp-then-stay-silent" model. + if seed: + conn.execute( + "UPDATE fires SET halt_broadcast_at=? WHERE irwin_id=?", + (float(now), irwin_id), + ) + logger.info( + "firms cold-start silent-seed: absorbed halt for %s " + "(latched, no broadcast)", irwin_id) + return None + name = p["incident_name"] hours = p["hours"] wire = f"🔥 {name} no growth in {hours}h" @@ -1028,8 +1118,18 @@ def _close_prev_perimeter(conn, irwin_id: str, prev_pass_id: str) -> None: def _check_spotting(conn, *, irwin_id, pixel_lat, pixel_lon, - current_pass_id, incident_name, data, now): - """Return spotting wire if criteria met, else None.""" + current_pass_id, incident_name, data, now, seed=False): + """Return spotting wire if criteria met, else None. + + ``seed`` (cold-start silent-seed): on the first fetch after boot, suppress + the wire WITHOUT stamping the ``last_spotting_broadcast_at`` cooldown latch. + Spotting is triggered by a specific NEW pixel lying outside the prior-pass + perimeter; a pre-existing pixel is a dedup no-op on later fetches, so it can + never re-trigger. Leaving the latch clear means a genuinely-new post-seed + spotting pixel still fires. Spotting writes no other state to preserve. + """ + if seed: + return None threshold_mi = float(adapter_config.fires.spotting_distance_threshold_mi) # Phase-3c: the cooldown gate (and its latch) moved into gating.firms.decide; # the cooldown seconds are read there. The geometry below stays inline. diff --git a/work/meshai/config.py b/work/meshai/config.py index 26e5710..8e93466 100644 --- a/work/meshai/config.py +++ b/work/meshai/config.py @@ -444,7 +444,11 @@ class FIRMSConfig(_SourcedFeed): tick_seconds: int = 1800 # 30 min default map_key: str = "" # NASA FIRMS MAP_KEY, get at https://firms.modaps.eosdis.nasa.gov/api/area/ source: str = "VIIRS_SNPP_NRT" # VIIRS_SNPP_NRT, VIIRS_NOAA20_NRT, MODIS_NRT - bbox: list = field(default_factory=list) # [west, south, east, north] + # Default Idaho region box [west, south, east, north] so native FIRMS + # fetches/persists even when the universal coverage bbox is OFF. When + # coverage IS enabled the store passes the coverage enclosing bbox to the + # adapter (env/firms.py:32 `coverage["bbox"]`), which OVERRIDES this default. + bbox: list = field(default_factory=lambda: [-115.5, 42.0, -110.0, 45.2]) day_range: int = 1 # 1-10 days of data confidence_min: str = "nominal" # low, nominal, high proximity_km: float = 10.0 # km to match known fire diff --git a/work/meshai/env/firms.py b/work/meshai/env/firms.py index 70c49bc..2eeb8e3 100644 --- a/work/meshai/env/firms.py +++ b/work/meshai/env/firms.py @@ -42,6 +42,16 @@ class FIRMSAdapter: self._last_error = None self._is_loaded = False + # Cold-start silent-seed gate (mirrors store._fires_seeded, F1). The + # FIRST non-empty FIRMS fetch after boot ingests a full day of existing + # hotspots; every unattributed cluster would look "new" and dump to the + # mesh. On that first fetch we ingest + persist + attribute normally but + # pass seed=True so clusters are stamped (never re-fire) yet emit + # NOTHING. Only clusters formed by pixels arriving on a LATER fetch + # (genuinely new ignitions) broadcast. Flipped True after the first + # non-empty fetch below. + self._firms_seeded = False + # For cross-referencing self._region_anchors = region_anchors or [] self._fires_adapter = fires_adapter # NICFFiresAdapter for cross-ref @@ -392,6 +402,10 @@ class FIRMSAdapter: now = time.time() satellite = self._satellite_code() + # Cold-start silent-seed gate: captured once for the whole batch BEFORE + # the flag is flipped, so every pixel in the first full-day sweep seeds + # together (mirrors store._ingest_fires cold_start capture, F1). + cold_start = not self._firms_seeded fusion: list = [] for evt in raw_events: props = evt.get("properties", {}) or {} @@ -409,13 +423,25 @@ class FIRMSAdapter: "acq_epoch": acq_epoch, } try: - broadcasts = ingest_hotspot_pixel(pixel, now=int(now)) + broadcasts = ingest_hotspot_pixel(pixel, now=int(now), + seed=cold_start) 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)) + + # First non-empty fetch complete: later fetches broadcast genuinely-new + # clusters. Only flip on a non-empty batch so an empty first fetch can't + # consume the seed (a later real batch still seeds silently). + if raw_events: + if cold_start: + logger.info( + "FIRMS cold-start silent-seed complete: %d pixels ingested, " + "0 cluster broadcasts (later fetches broadcast new fires)", + len(raw_events)) + self._firms_seeded = True return fusion def _make_fusion_event(self, wire: str, data: dict, source_evt: dict, diff --git a/work/tests/test_fire_tracker_phase1.py b/work/tests/test_fire_tracker_phase1.py index cb57862..71dad76 100644 --- a/work/tests/test_fire_tracker_phase1.py +++ b/work/tests/test_fire_tracker_phase1.py @@ -220,10 +220,18 @@ def test_three_unattributed_pixels_fire_cluster_once(): assert data.get("category") == "unattributed_hotspot_cluster" assert data.get("severity") == "priority" - # Cluster detection is currently stubbed (_maybe_emit_cluster returns None). - # All three calls return None. + # F3: cluster detection is ENABLED. The 3rd pixel reaches cluster_min_pixels + # (3) within 1 mi / 60 min, so exactly ONE cluster wire fires (on pixel 3); + # pixels 1 and 2 return None (below threshold at the time they arrive). fired = [w for w in wires if w is not None] - assert len(fired) == 0, f"expected 0 cluster wires (stub), got {len(fired)}: {wires}" + assert len(fired) == 1, f"expected 1 cluster wire, got {len(fired)}: {wires}" + assert fired[0].startswith("🔥 Possible new fire:") + # All three members must be stamped so a later pixel can't re-fire them. + conn = get_db() + stamped = conn.execute( + "SELECT COUNT(*) FROM firms_pixels WHERE cluster_broadcast_at IS NOT NULL" + ).fetchone()[0] + assert stamped == 3 def test_fourth_pixel_in_same_cluster_does_not_refire(): @@ -256,7 +264,13 @@ def test_fourth_pixel_in_same_cluster_does_not_refire(): assert wire is None assert "category" not in data4 - # Cluster detection is stubbed; no cluster_broadcast_at stamped on any pixel. + # F3: the first 3 members were stamped by the cluster that fired; the 4th + # pixel is unattributed + unstamped (it alone can't re-form the cluster). + conn = get_db() + stamped = conn.execute( + "SELECT COUNT(*) FROM firms_pixels WHERE cluster_broadcast_at IS NOT NULL" + ).fetchone()[0] + assert stamped == 3 def test_fifth_pixel_after_time_window_can_form_new_cluster(): @@ -293,9 +307,12 @@ def test_fifth_pixel_after_time_window_can_form_new_cluster(): env, subject="central.fire.hotspot.N20.high.unknown", data={}, now=1780728000 + 7200 + i, )) - # Cluster detection is stubbed (_maybe_emit_cluster returns None). + # F3: the first cluster's 3 members are stamped (excluded from the query), + # so the second batch forms an independent NEW cluster -- one wire on its + # 3rd pixel. fired = [w for w in wires2 if w is not None] - assert len(fired) == 0, f"expected 0 cluster wires (stub), got: {wires2}" + assert len(fired) == 1, f"expected 1 new cluster wire, got: {wires2}" + assert fired[0].startswith("🔥 Possible new fire:") # --------------------------------------------------------------------------- diff --git a/work/tests/test_firms_cluster_f3.py b/work/tests/test_firms_cluster_f3.py new file mode 100644 index 0000000..770c5a2 --- /dev/null +++ b/work/tests/test_firms_cluster_f3.py @@ -0,0 +1,382 @@ +"""F3 — curated FIRMS new-fire cluster detection + cold-start silent-seed. + +Enables the previously dead ``_maybe_emit_cluster`` path and adds the +cold-start silent-seed discipline so the FIRST FIRMS fetch after boot (a full +day of pre-existing hotspots) never dumps a wall of "possible new fire" +broadcasts to the live mesh. + +Scenarios (task spec): + 1. Cluster forms + broadcasts once on a LATER (non-seed) fetch; a subsequent + pixel in the same cluster is silent (cluster_broadcast_at stamped). + 2. Cold-start silent-seed: first fetch with many unattributed pixels -> 0 + cluster broadcasts, but pixels persisted + cluster_broadcast_at stamped. + 3. Attribution beats clustering: a pixel within a known fire's (MORA) spread + radius attributes (fire_pixels), never forms a "new" cluster. + 4. Per-pixel silent: a raw hotspot -> FIRMSAdapter.to_event returns None. + 5. Below-threshold: < cluster_min_pixels unattributed pixels -> no cluster. + +Mirrors the driving style / isolation fixture of test_firms_native_fusion.py. +""" +from __future__ import annotations + +import uuid +from datetime import datetime, timezone + +import pytest + + +# ── isolation (real-DB, mirrors test_firms_native_fusion) ──────────────────── + +@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): + monkeypatch.delenv("MESHAI_CUTOVER_CATEGORIES", raising=False) + from meshai.notifications.cutover import _clear_cache + _clear_cache() + yield + _clear_cache() + + +# ── helpers ────────────────────────────────────────────────────────────────── + +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"): + return { + "lat": lat, "lon": lon, "frp": frp, "confidence": confidence, + "brightness": brightness, "satellite": satellite, + "acq_epoch": _acq_epoch(acq_date, acq_time), + } + + +def _feed(pixel, *, now, seed=False): + from meshai.central.firms_handler import ingest_hotspot_pixel + return ingest_hotspot_pixel(pixel, now=now, seed=seed) + + +def _stamped_count(): + from meshai.persistence import get_db + return get_db().execute( + "SELECT COUNT(*) FROM firms_pixels WHERE cluster_broadcast_at IS NOT NULL" + ).fetchone()[0] + + +def _pixel_count(): + from meshai.persistence import get_db + return get_db().execute("SELECT COUNT(*) FROM firms_pixels").fetchone()[0] + + +def _seed_fire(*, irwin_id, lat, lon, name="MORA"): + from meshai.persistence import get_db + get_db().execute( + "INSERT INTO fires(irwin_id, incident_name, lat, lon, last_event_at) " + "VALUES (?,?,?,?,?)", (irwin_id, name, lat, lon, 1780747200)) + + +# ═════════════════════════════════════════════════════════════════════════════ +# 1. Cluster forms + broadcasts (after first-fetch seeding); refire is silent +# ═════════════════════════════════════════════════════════════════════════════ + +def test_cluster_broadcasts_after_seed_then_refire_silent(): + base_lat, base_lon = 43.500, -114.500 + + # First fetch (cold start): three pre-existing hotspots -> silent seed. + seed_wires = [] + for i, dt in enumerate([(0.0, 0.0), (0.001, 0.001), (-0.001, -0.002)]): + seed_wires += _feed(_pixel(lat=base_lat + dt[0], lon=base_lon + dt[1], + acq_time=f"12{i:02d}"), + now=1780728000 + i, seed=True) + assert seed_wires == [], "cold-start seed must emit no cluster wire" + assert _stamped_count() == 3, "seeded cluster members must be stamped" + + # Later fetch (seed=False): THREE genuinely-new hotspots elsewhere -> one + # cluster wire on the 3rd pixel. + n2_lat, n2_lon = 44.000, -116.000 + later = [] + for i, dt in enumerate([(0.0, 0.0), (0.001, 0.001), (-0.001, -0.002)]): + later += _feed(_pixel(lat=n2_lat + dt[0], lon=n2_lon + dt[1], + acq_time=f"14{i:02d}"), + now=1780735200 + i, seed=False) + assert len(later) == 1, f"expected exactly one cluster wire: {later}" + wire, data = later[0] + assert wire.startswith("🔥 Possible new fire:") + assert "3 hotspots within 1 mi" in wire + assert data["category"] == "unattributed_hotspot_cluster" + assert data["severity"] == "priority" + + # A 4th pixel inside the just-broadcast cluster -> silent (members stamped). + refire = _feed(_pixel(lat=n2_lat + 0.0005, lon=n2_lon - 0.0005, + acq_time="1403"), now=1780735300, seed=False) + assert refire == [], "a pixel joining an already-broadcast cluster is silent" + + +# ═════════════════════════════════════════════════════════════════════════════ +# 2. Cold-start silent-seed: many pixels -> 0 broadcasts, persisted + stamped +# ═════════════════════════════════════════════════════════════════════════════ + +def test_cold_start_many_pixels_zero_broadcasts_but_persisted(): + base_lat, base_lon = 43.000, -115.000 + produced = [] + # 12 tightly-clustered pre-existing hotspots on the first fetch. + for i in range(12): + produced += _feed( + _pixel(lat=base_lat + 0.0005 * i, lon=base_lon, + acq_time=f"12{i:02d}"), + now=1780728000 + i, seed=True) + assert produced == [], "cold start must emit ZERO cluster broadcasts" + # All 12 persisted. + assert _pixel_count() == 12 + # Every clustered member stamped so none can re-fire later. With 12 pixels + # within 1 mi / 60 min and min_pixels=3, every pixel from the 3rd onward is + # part of a stamped cluster; at minimum a supermajority are stamped and NONE + # is left in a state that would broadcast on a later identical fetch. + assert _stamped_count() >= 10 + + +def test_cold_start_native_adapter_flag_flips_and_second_fetch_broadcasts(): + """Drive the real native-adapter cold-start gate: first _run_fusion call + silent-seeds (flag False->True, zero wires); the second broadcasts.""" + from meshai.config import FIRMSConfig + from meshai.env.firms import FIRMSAdapter + + adapter = FIRMSAdapter(FIRMSConfig(map_key="x")) + assert adapter._firms_seeded is False + + def _raw(lat, lon, acq_time, i): + return { + "source": "firms", "event_id": f"e{i}", "lat": lat, "lon": lon, + "properties": {"frp": 20.0, "confidence": "high", + "brightness": 320.0, "acq_date": "2026-06-06", + "acq_time": acq_time}, + } + + base_lat, base_lon = 43.700, -114.700 + first_batch = [_raw(base_lat + 0.001 * i, base_lon, f"12{i:02d}", i) + for i in range(4)] + out1 = adapter._run_fusion(first_batch) + assert out1 == [], "first fetch (cold start) broadcasts nothing" + assert adapter._firms_seeded is True, "first non-empty fetch flips the flag" + + # Second fetch: three NEW hotspots in a fresh location -> a cluster wire. + n_lat, n_lon = 44.300, -115.300 + second_batch = [_raw(n_lat + 0.001 * i, n_lon, f"14{i:02d}", 100 + i) + for i in range(3)] + out2 = adapter._run_fusion(second_batch) + fusion_cats = [e["properties"].get("category") for e in out2] + assert "unattributed_hotspot_cluster" in fusion_cats, \ + f"second fetch must broadcast a curated cluster: {fusion_cats}" + + +def test_empty_first_fetch_does_not_consume_seed(): + """An empty first fetch must not flip the seed flag -- a later real batch + still seeds silently (mirrors store._fires_seeded non-empty guard).""" + from meshai.config import FIRMSConfig + from meshai.env.firms import FIRMSAdapter + + adapter = FIRMSAdapter(FIRMSConfig(map_key="x")) + assert adapter._run_fusion([]) == [] + assert adapter._firms_seeded is False, "empty fetch must not seed" + + +# ═════════════════════════════════════════════════════════════════════════════ +# 3. Attribution beats clustering (MORA hotspots grow the fire, never cluster) +# ═════════════════════════════════════════════════════════════════════════════ + +def test_attribution_beats_clustering_for_known_fire(): + from meshai.persistence import get_db + + # Seed a known WFIGS fire (MORA) at a fixed point. + mora_lat, mora_lon = 44.100, -115.600 + _seed_fire(irwin_id="ID-MORA", lat=mora_lat, lon=mora_lon) + + # Three hotspots ~0.05 mi apart, well within the 5 mi default spread radius. + out = [] + for i, dt in enumerate([(0.0, 0.0), (0.001, 0.001), (-0.001, -0.001)]): + out += _feed(_pixel(lat=mora_lat + dt[0], lon=mora_lon + dt[1], + acq_time=f"12{i:02d}"), + now=1780728000 + i, seed=False) + + conn = get_db() + # Attributed: fire_pixels rows exist, firms_pixels.attributed_at set. + assert conn.execute("SELECT COUNT(*) FROM fire_pixels").fetchone()[0] == 3 + attributed = conn.execute( + "SELECT COUNT(*) FROM firms_pixels WHERE attributed_at IS NOT NULL" + ).fetchone()[0] + assert attributed == 3 + # NOT clustered: nothing stamped cluster_broadcast_at, and no cluster wire. + assert _stamped_count() == 0 + cats = [d.get("category") for _w, d in out] + assert "unattributed_hotspot_cluster" not in cats + + +# ═════════════════════════════════════════════════════════════════════════════ +# 3b. Cold-start silent-seed suppresses the FUSION path too (growth/spotting/ +# halt), not just clusters — enabling FIRMS must emit ZERO broadcasts on the +# first fetch even for a pre-existing attributed fire (e.g. MORA). +# ═════════════════════════════════════════════════════════════════════════════ + +# One VIIRS pass is a 90-min (5400 s) bucket (see _pass_id). Anchor pass A and +# pass B in DIFFERENT buckets so the second batch crosses a real pass boundary. +_PASS_A = _acq_epoch("2026-06-06", "1200") # bucket N +_PASS_B = _PASS_A + 6 * 3600 # +6h -> a later bucket + + +def _pixel_at(*, lat, lon, acq_epoch, frp=20.0): + return {"lat": lat, "lon": lon, "frp": frp, "confidence": "high", + "brightness": 320.0, "satellite": "N20", "acq_epoch": acq_epoch} + + +def test_cold_start_seed_suppresses_growth_fusion_but_persists(): + """First (seed) fetch: a pre-existing fire (MORA) with pixels that WOULD + produce a wildfire_growth boundary broadcast -> ZERO wires, yet the pixels + are persisted + attributed and the pass/centroid baseline is built.""" + from meshai.persistence import get_db + + mora_lat, mora_lon = 44.100, -115.600 + _seed_fire(irwin_id="ID-MORA", lat=mora_lat, lon=mora_lon) + + produced = [] + # Pass A: 5 pixels around the anchor (same 90-min bucket). + for i in range(5): + produced += _feed( + _pixel_at(lat=mora_lat + 0.0001 * i, lon=mora_lon + 0.0001 * (i - 2), + acq_epoch=_PASS_A + i, frp=20.0 + i), + now=1780728000 + i, seed=True) + # Pass B: one pixel ~1 mi north in a LATER bucket -> a growth boundary that, + # on a non-seed fetch, WOULD broadcast wildfire_growth. + produced += _feed( + _pixel_at(lat=mora_lat + 1.0 / 69.0, lon=mora_lon, acq_epoch=_PASS_B), + now=1780728100, seed=True) + + assert produced == [], "cold-start seed must suppress the growth fusion wire" + + conn = get_db() + # Persistence + attribution still happened. + assert conn.execute("SELECT COUNT(*) FROM firms_pixels").fetchone()[0] == 6 + assert conn.execute( + "SELECT COUNT(*) FROM fire_pixels").fetchone()[0] == 6 + assert conn.execute( + "SELECT COUNT(*) FROM firms_pixels WHERE attributed_at IS NOT NULL" + ).fetchone()[0] == 6 + # Pass baseline built: the fires cursor advanced to pass B's bucket so a + # genuinely-new LATER pass can measure drift against it. + cursor = conn.execute( + "SELECT last_pass_id, current_centroid_lat FROM fires " + "WHERE irwin_id=?", ("ID-MORA",)).fetchone() + assert cursor["last_pass_id"] is not None + assert cursor["current_centroid_lat"] is not None + + +def test_growth_fires_on_later_fetch_after_seed(): + """After the cold-start seed built the pass baseline, a genuinely-new pass + on a LATER (seed=False) fetch DOES broadcast wildfire_growth.""" + mora_lat, mora_lon = 44.100, -115.600 + _seed_fire(irwin_id="ID-MORA", lat=mora_lat, lon=mora_lon) + + # --- Cold-start seed: pass A + pass B, all silent. --- + for i in range(5): + assert _feed( + _pixel_at(lat=mora_lat + 0.0001 * i, lon=mora_lon + 0.0001 * (i - 2), + acq_epoch=_PASS_A + i, frp=20.0 + i), + now=1780728000 + i, seed=True) == [] + assert _feed( + _pixel_at(lat=mora_lat + 1.0 / 69.0, lon=mora_lon, acq_epoch=_PASS_B), + now=1780728100, seed=True) == [] + + # --- Later real fetch (seed=False): a NEW pass C, another ~1 mi north of + # pass B -> a genuinely-new boundary drift -> wildfire_growth broadcasts. --- + pass_c = _PASS_B + 6 * 3600 + out = _feed( + _pixel_at(lat=mora_lat + 2.0 / 69.0, lon=mora_lon, acq_epoch=pass_c), + now=1780750000, seed=False) + assert len(out) == 1, f"genuinely-new post-seed growth must fire: {out}" + wire, data = out[0] + assert data["category"] == "wildfire_growth" + assert data["_severity_override"] == "immediate" + assert wire.startswith("🔥 MORA") + + +def test_cold_start_seed_suppresses_halt_but_latches(): + """First (seed) fetch: an already-idle fire that WOULD halt-broadcast is + suppressed AND its halt latch is stamped, so it stays silent on later + unrelated fetches until real activity resumes.""" + from meshai.persistence import get_db + + now = 1780768800 + idle_at = now - 14 * 3600 + conn = get_db() + conn.execute( + "INSERT INTO fires(irwin_id, incident_name, lat, lon, last_event_at, " + "last_pass_id, last_pass_at) VALUES (?,?,?,?,?,?,?)", + ("ID-IDLE", "Cold Fire", 42.5, -114.5, int(idle_at), + "N20-329627", float(idle_at))) + + # A fresh unattributed pixel FAR from the idle fire on the cold-start fetch. + out = _feed(_pixel_at(lat=45.0, lon=-118.0, acq_epoch=_PASS_A), now=now, + seed=True) + assert out == [], "cold-start seed must suppress the halt wire" + latch = conn.execute( + "SELECT halt_broadcast_at FROM fires WHERE irwin_id=?", + ("ID-IDLE",)).fetchone()[0] + assert latch == float(now), "seed still latches halt so it can't re-fire" + + # A later unrelated fetch: the idle fire is latched -> still silent. + out2 = _feed(_pixel_at(lat=45.1, lon=-118.1, acq_epoch=_PASS_A + 60), + now=now + 120, seed=False) + assert out2 == [], "latched idle fire must not halt on a later fetch" + + +# ═════════════════════════════════════════════════════════════════════════════ +# 4. Per-pixel silent: a raw hotspot never renders an Event +# ═════════════════════════════════════════════════════════════════════════════ + +def test_raw_hotspot_to_event_returns_none(): + from meshai.config import FIRMSConfig + from meshai.env.firms import FIRMSAdapter + + adapter = FIRMSAdapter(FIRMSConfig(map_key="x")) + raw_evt = { + "source": "firms", "event_id": "firms_43.0_-115.0_2026-06-06_1200", + "lat": 43.0, "lon": -115.0, "properties": {"new_ignition": True}, + } + assert adapter.to_event(raw_evt) is None, "raw hotspots must never broadcast" + + +# ═════════════════════════════════════════════════════════════════════════════ +# 5. Below-threshold: fewer than cluster_min_pixels -> no cluster +# ═════════════════════════════════════════════════════════════════════════════ + +def test_below_threshold_no_cluster(): + base_lat, base_lon = 43.900, -116.100 + produced = [] + # Only two unattributed pixels within radius -> below min_pixels (3). + for i, dt in enumerate([(0.0, 0.0), (0.001, 0.001)]): + produced += _feed(_pixel(lat=base_lat + dt[0], lon=base_lon + dt[1], + acq_time=f"12{i:02d}"), + now=1780728000 + i, seed=False) + assert produced == [], "two pixels must not fire a cluster" + assert _stamped_count() == 0 + assert _pixel_count() == 2 diff --git a/work/tests/test_firms_native_fusion.py b/work/tests/test_firms_native_fusion.py index 8fcdbec..344947f 100644 --- a/work/tests/test_firms_native_fusion.py +++ b/work/tests/test_firms_native_fusion.py @@ -5,8 +5,9 @@ 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. + pixels (and, F3, curated ``unattributed_hotspot_cluster`` "possible new + fire" broadcasts) and returns ``(wire, data)`` — and NEVER a raw per-pixel + hotspot / 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. @@ -25,8 +26,11 @@ 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"} +# F3: unattributed_hotspot_cluster is a CURATED fusion output (a "possible new +# fire"), not a raw per-pixel broadcast. Raw hotspots are still never emitted. +_FUSION_CATS = {"wildfire_growth", "wildfire_spotting", "wildfire_halted", + "unattributed_hotspot_cluster"} +_RAW_CATS = {"wildfire_hotspot", "new_ignition"} # ── isolation (real-DB, mirrors test_firms_refactor) ───────────────────────── @@ -201,17 +205,20 @@ class TestIngestHalt: assert latch == float(now) -class TestIngestNeverRawOrCluster: +class TestIngestNeverRaw: 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. + # Stored, attributed to nothing, below the cluster threshold, 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. + def test_dense_unattributed_cluster_broadcasts_curated(self): + # F3: several unattributed pixels close together form a CURATED cluster + # ("possible new fire"). Non-seed path (no cold-start), so it broadcasts: + # a wire fires on the 3rd pixel (min_pixels=3), the first 3 are stamped + # so they can't re-fire, and the next 3 form a fresh cluster -> a 2nd + # wire on the 6th pixel. NEVER a raw per-pixel broadcast. base_lat, base_lon = 44.0, -116.0 produced = [] for i in range(6): @@ -219,7 +226,11 @@ class TestIngestNeverRawOrCluster: acq_time=f"13{i:02d}"), now=1780750000 + i) _assert_no_raw(out) produced.extend(out) - assert produced == [] + assert len(produced) == 2, f"expected 2 curated cluster wires: {produced}" + for wire, data in produced: + assert wire.startswith("🔥 Possible new fire:") + assert data["category"] == "unattributed_hotspot_cluster" + assert data["severity"] == "priority" # ═════════════════════════════════════════════════════════════════════════════ @@ -297,6 +308,10 @@ class TestAdapterTickFusion: _patch_fetch(monkeypatch, _csv(self._growth_rows(center_lat, center_lon))) a = _adapter() + # Steady-state growth routing (not cold start): mark the adapter past its + # first-fetch silent-seed so this tick's growth broadcasts. Cold-start + # fusion suppression is covered in test_firms_cluster_f3. + a._firms_seeded = True assert a.tick() is True evts = a.get_events() @@ -341,6 +356,10 @@ class TestAdapterTickFusion: # ingest would (store._ingest marks every source it touched). store._seen = {} store._seeded = {"firms", "firms_fusion"} + # Adapter-level cold-start seed is also a first-fetch silence gate; this + # steady-state test wants the growth wire, so mark the adapter seeded too + # (cold-start fusion suppression is covered in test_firms_cluster_f3). + a._firms_seeded = True assert a.tick() is True store._ingest("firms", a) diff --git a/work/tests/test_firms_refactor.py b/work/tests/test_firms_refactor.py index 933df8f..790bf52 100644 --- a/work/tests/test_firms_refactor.py +++ b/work/tests/test_firms_refactor.py @@ -21,7 +21,8 @@ test_hydro_refactor.py: 4. Not-cutover parity: with no category cut over, handle_firms keeps the legacy eager-latch + stamps VERBATIM (byte-identical live behavior). -5. The unattributed_hotspot_cluster path stays DEAD (returns None). +5. The unattributed_hotspot_cluster path is curated: below cluster_min_pixels + it stays silent (returns None) -- F3 enabled the real broadcast path. """ from __future__ import annotations @@ -427,17 +428,19 @@ class TestNotCutoverLegacyVerbatim: # ───────────────────────────────────────────────────────────────────────────── -# 5. Cluster path stays DEAD +# 5. Cluster path: below-threshold is silent (F3 enabled the path; a lone +# unattributed pixel with no nearby unstamped neighbors never clusters) # ───────────────────────────────────────────────────────────────────────────── -class TestClusterDead: - def test_maybe_emit_cluster_returns_none(self): +class TestClusterBelowThreshold: + def test_maybe_emit_cluster_below_threshold_returns_none(self): from meshai.central.firms_handler import _maybe_emit_cluster from meshai.persistence import get_db data = {} + # No pixels in firms_pixels -> the cluster query finds < min_pixels + # members, so no wire and no data tagging (curated: needs a real cluster). out = _maybe_emit_cluster( get_db(), lat=43.0, lon=-115.0, acq_epoch=1780747200, frp=20.0, data=data, now=1780747200, this_pixel_id=1) assert out is None - # The dead path must not tag the data dict either. assert data == {}