matt/meshai
1
0
Fork 0
mirror of https://github.com/zvx-echo6/meshai.git synced 2026-08-26 17:31:34 +00:00
Code Issues Projects Releases Packages Wiki Activity

feat(firms): source-agnostic fire-fusion — native FIRMS feeds growth/spotting/halt (#41)

Browse source 
Closes the last standalone gap. Extract ingest_hotspot_pixel(pixel, *, now)
from firms_handler so the FIRMS attribution/fusion engine (firms_pixels ->
_attribute_or_cluster -> fire_pixels/fire_passes/centroid -> growth/spotting/
halt) is source-agnostic. Both the Central NATS path and native env/firms.py
drive one identical engine.

- shared _ingest_pixel_core(conn, ...) called by both ingest_hotspot_pixel
  and handle_firms; Central path byte-identical (its tests pass unchanged)
- env/firms.py _fetch() feeds each fetched pixel into ingest_hotspot_pixel;
  DB-level dedup makes re-fetched pixels no-ops (no double count); to_event()
  returns None for raw hotspots, precomposed Event for fusion outputs
  (wildfire_growth/spotting/halted via the Phase-3c formatters/gating)
- raw hotspots / new_ignition / cluster NEVER broadcast (cluster stays dead)

FLIP NOTE: keep wildfire_growth/spotting/halted OUT of cutover — native
emits precomposed with gating done inside the engine; cutover would re-run
the _kind-keyed decider on data lacking _kind and suppress.

10 new tests; Central firms/fire-tracker suites unchanged; full suite
10 failed/1682 passed (baseline 10, +10 new).

Co-authored-by: Matt Johnson <mj@k7zvx.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
malice 2026-07-05 09:39:05 -06:00 • committed by GitHub
commit 0a75930ade
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 710 additions and 59 deletions
Download patch file Download diff file Expand all files Collapse all files

151
work/meshai/central/firms_handler.py
View file

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

197
work/meshai/env/firms.py vendored
View file

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

417
work/tests/test_firms_native_fusion.py Normal file
View file

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