mirror of
https://github.com/zvx-echo6/meshai.git
synced 2026-08-26 17:31:34 +00:00
feat(firms): curated new-fire cluster broadcasts (no per-pixel, no cold-start dump) (#73)
* feat(firms): curated new-fire cluster broadcasts (no per-pixel, no cold-start dump) Enable the built _maybe_emit_cluster path (was dead-coded) so FIRMS broadcasts curated hotspot clusters as possible new fires — clustered, deduped via cluster_broadcast_at, attributed against known WFIGS fires first (so MORA's hotspots don't false-cluster). Give FIRMS a default Idaho bbox so it fetches when coverage is off (coverage bbox still overrides). First-fetch silent-seed prevents a cold-start dump of the day's existing hotspots. Raw pixels stay store-only. Coverage geometry gate filters cluster broadcasts to the region. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * fix(firms): first-fetch silent-seed suppresses fusion wires too (no cold-start) Extend the FIRMS cold-start seed to suppress growth/spotting/halt fusion broadcasts on the first fetch, not just clusters — enabling FIRMS must emit zero broadcasts on the initial hotspot sweep. Persistence, attribution, and dedup baselines still run during seed; only later new activity broadcasts. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> --------- 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:
parent
8d61b16955
commit
d479ca537a
7 changed files with 600 additions and 49 deletions
|
|
@ -248,7 +248,7 @@ def handle_firms(envelope: dict, subject: str,
|
||||||
|
|
||||||
|
|
||||||
def _ingest_pixel_core(conn, *, lat, lon, acq_epoch, frp, confidence,
|
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.
|
"""INSERT one canonical hotspot pixel + run attribution/fusion.
|
||||||
|
|
||||||
This is the source-agnostic heart of the fire-fusion engine, extracted so
|
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
|
2. For a NEWLY-stored pixel, run ``_attribute_or_cluster`` -> writes
|
||||||
``fire_pixels`` / ``fire_passes`` / ``fires`` centroid+cursor and runs
|
``fire_pixels`` / ``fire_passes`` / ``fires`` centroid+cursor and runs
|
||||||
the growth / spotting / halt fusion (honoring the Phase-3c deferred
|
the growth / spotting / halt fusion (honoring the Phase-3c deferred
|
||||||
latches). The dead unattributed-cluster path stays dead -- NO
|
latches). On an attribution MISS it runs the curated
|
||||||
raw-hotspot / cluster broadcast is ever produced here.
|
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.
|
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,
|
lat=lat, lon=lon,
|
||||||
acq_epoch=acq_epoch,
|
acq_epoch=acq_epoch,
|
||||||
frp=frp, satellite=satellite,
|
frp=frp, satellite=satellite,
|
||||||
data=data, now=now,
|
data=data, now=now, seed=seed,
|
||||||
)
|
)
|
||||||
return (True, rowid, wire)
|
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
|
"""Source-agnostic entrypoint: ingest ONE canonical FIRMS pixel into the
|
||||||
fire-fusion engine and return the fusion broadcasts it produced.
|
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,
|
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
|
under cutover, the deferred commit hook). A dedup hit or a pixel that
|
||||||
triggers no fusion returns ``[]``. This NEVER returns a raw-hotspot,
|
triggers no fusion returns ``[]``. This NEVER returns a raw-hotspot,
|
||||||
``wildfire_hotspot``, ``new_ignition``, or cluster broadcast -- the only
|
``wildfire_hotspot``, or ``new_ignition`` broadcast -- the possible outputs
|
||||||
outputs are ``wildfire_growth`` / ``wildfire_spotting`` / ``wildfire_halted``
|
are ``wildfire_growth`` / ``wildfire_spotting`` / ``wildfire_halted`` (an
|
||||||
(the cluster path is dead). Callers must therefore NEVER broadcast the raw
|
attributed pixel grew/moved a known fire) OR a curated
|
||||||
pixel itself -- only these returned fusion wires.
|
``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.
|
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,
|
conn,
|
||||||
lat=float(lat), lon=float(lon), acq_epoch=int(acq_epoch),
|
lat=float(lat), lon=float(lon), acq_epoch=int(acq_epoch),
|
||||||
frp=frp, confidence=pixel.get("confidence"), brightness=brightness,
|
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:
|
if wire is None:
|
||||||
return []
|
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,
|
def _attribute_or_cluster(conn, *, pixel_row_id, lat, lon, acq_epoch,
|
||||||
frp, satellite, data, now):
|
frp, satellite, data, now, seed=False):
|
||||||
"""Try attribution; on miss, run cluster check. Returns wire str | None."""
|
"""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)
|
global_default_mi = float(adapter_config.fires.spread_radius_mi_default)
|
||||||
# Conservative bbox prefilter: take the larger of the global 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
|
# 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(
|
wire = _handle_pass_boundary(
|
||||||
conn, irwin_id=chosen_irwin, pass_id=this_pass_id,
|
conn, irwin_id=chosen_irwin, pass_id=this_pass_id,
|
||||||
lat=lat, lon=lon, acq_epoch=acq_epoch, frp=frp,
|
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:
|
if wire is not None:
|
||||||
return wire
|
return wire
|
||||||
# No growth broadcast; opportunistically run halt detector for
|
# No growth broadcast; opportunistically run halt detector for
|
||||||
# OTHER fires that may have gone idle.
|
# 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.
|
# 0 matches -- run cluster detection.
|
||||||
wire = _maybe_emit_cluster(
|
wire = _maybe_emit_cluster(
|
||||||
conn, lat=lat, lon=lon, acq_epoch=acq_epoch, frp=frp,
|
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:
|
if wire is not None:
|
||||||
return wire
|
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, *,
|
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,
|
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
|
"""Return wire string + set data["category"] when a cluster condition
|
||||||
fires; otherwise return None and leave data alone."""
|
fires; otherwise return None and leave data alone.
|
||||||
return None
|
|
||||||
|
``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)
|
min_pixels = int(adapter_config.firms.cluster_min_pixels)
|
||||||
radius_mi = float(adapter_config.firms.cluster_max_radius_mi)
|
radius_mi = float(adapter_config.firms.cluster_max_radius_mi)
|
||||||
window_s = int(adapter_config.firms.cluster_time_window_minutes) * 60
|
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),
|
(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
|
# Override the FIRMS source category so the dispatcher routes this
|
||||||
# broadcast under unattributed_hotspot_cluster (priority, fire toggle).
|
# broadcast under unattributed_hotspot_cluster (priority, fire toggle).
|
||||||
if isinstance(data, dict):
|
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,
|
def _handle_pass_boundary(conn, *, irwin_id, pass_id, lat, lon,
|
||||||
acq_epoch, frp, data, now):
|
acq_epoch, frp, data, now, seed=False):
|
||||||
"""Maintain fire_passes row, detect boundary, fire growth on drift."""
|
"""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.
|
# (1) Recompute the pass aggregate from fire_pixels.
|
||||||
pass_rows = conn.execute(
|
pass_rows = conn.execute(
|
||||||
"SELECT lat, lon, frp, acq_time FROM fire_pixels "
|
"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,
|
conn, irwin_id=irwin_id, pixel_lat=lat, pixel_lon=lon,
|
||||||
current_pass_id=pass_id,
|
current_pass_id=pass_id,
|
||||||
incident_name=fires_row["incident_name"] or "(unnamed fire)",
|
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:
|
if spotting_wire is not None:
|
||||||
return spotting_wire
|
return spotting_wire
|
||||||
|
|
@ -844,6 +898,16 @@ def _handle_pass_boundary(conn, *, irwin_id, pass_id, lat, lon,
|
||||||
if not gate.broadcast:
|
if not gate.broadcast:
|
||||||
return None
|
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.
|
# Drift exceeded the threshold -- emit wildfire_growth via WFIGS renderer.
|
||||||
fire = conn.execute(
|
fire = conn.execute(
|
||||||
"SELECT incident_name, current_acres, current_contained_pct, "
|
"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.
|
"""Find one fire matching the halt criteria, latch + broadcast.
|
||||||
|
|
||||||
Returns the wire string when a halt event fires; otherwise None.
|
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
|
Phase-3c: the DECISION (the halt-eligibility SELECT) is delegated to
|
||||||
``gating.firms.decide``; the wire is still rendered inline here for
|
``gating.firms.decide``; the wire is still rendered inline here for
|
||||||
byte-identity. On the not-cutover live path the eager
|
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
|
p = gate.data_patch
|
||||||
irwin_id = p["irwin_id"]
|
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"]
|
name = p["incident_name"]
|
||||||
hours = p["hours"]
|
hours = p["hours"]
|
||||||
wire = f"🔥 {name} no growth in {hours}h"
|
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,
|
def _check_spotting(conn, *, irwin_id, pixel_lat, pixel_lon,
|
||||||
current_pass_id, incident_name, data, now):
|
current_pass_id, incident_name, data, now, seed=False):
|
||||||
"""Return spotting wire if criteria met, else None."""
|
"""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)
|
threshold_mi = float(adapter_config.fires.spotting_distance_threshold_mi)
|
||||||
# Phase-3c: the cooldown gate (and its latch) moved into gating.firms.decide;
|
# Phase-3c: the cooldown gate (and its latch) moved into gating.firms.decide;
|
||||||
# the cooldown seconds are read there. The geometry below stays inline.
|
# the cooldown seconds are read there. The geometry below stays inline.
|
||||||
|
|
|
||||||
|
|
@ -444,7 +444,11 @@ class FIRMSConfig(_SourcedFeed):
|
||||||
tick_seconds: int = 1800 # 30 min default
|
tick_seconds: int = 1800 # 30 min default
|
||||||
map_key: str = "" # NASA FIRMS MAP_KEY, get at https://firms.modaps.eosdis.nasa.gov/api/area/
|
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
|
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
|
day_range: int = 1 # 1-10 days of data
|
||||||
confidence_min: str = "nominal" # low, nominal, high
|
confidence_min: str = "nominal" # low, nominal, high
|
||||||
proximity_km: float = 10.0 # km to match known fire
|
proximity_km: float = 10.0 # km to match known fire
|
||||||
|
|
|
||||||
28
work/meshai/env/firms.py
vendored
28
work/meshai/env/firms.py
vendored
|
|
@ -42,6 +42,16 @@ class FIRMSAdapter:
|
||||||
self._last_error = None
|
self._last_error = None
|
||||||
self._is_loaded = False
|
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
|
# For cross-referencing
|
||||||
self._region_anchors = region_anchors or []
|
self._region_anchors = region_anchors or []
|
||||||
self._fires_adapter = fires_adapter # NICFFiresAdapter for cross-ref
|
self._fires_adapter = fires_adapter # NICFFiresAdapter for cross-ref
|
||||||
|
|
@ -392,6 +402,10 @@ class FIRMSAdapter:
|
||||||
|
|
||||||
now = time.time()
|
now = time.time()
|
||||||
satellite = self._satellite_code()
|
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 = []
|
fusion: list = []
|
||||||
for evt in raw_events:
|
for evt in raw_events:
|
||||||
props = evt.get("properties", {}) or {}
|
props = evt.get("properties", {}) or {}
|
||||||
|
|
@ -409,13 +423,25 @@ class FIRMSAdapter:
|
||||||
"acq_epoch": acq_epoch,
|
"acq_epoch": acq_epoch,
|
||||||
}
|
}
|
||||||
try:
|
try:
|
||||||
broadcasts = ingest_hotspot_pixel(pixel, now=int(now))
|
broadcasts = ingest_hotspot_pixel(pixel, now=int(now),
|
||||||
|
seed=cold_start)
|
||||||
except Exception:
|
except Exception:
|
||||||
logger.exception("FIRMS fusion: ingest failed for %s",
|
logger.exception("FIRMS fusion: ingest failed for %s",
|
||||||
evt.get("event_id"))
|
evt.get("event_id"))
|
||||||
continue
|
continue
|
||||||
for wire, data in broadcasts:
|
for wire, data in broadcasts:
|
||||||
fusion.append(self._make_fusion_event(wire, data, evt, now))
|
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
|
return fusion
|
||||||
|
|
||||||
def _make_fusion_event(self, wire: str, data: dict, source_evt: dict,
|
def _make_fusion_event(self, wire: str, data: dict, source_evt: dict,
|
||||||
|
|
|
||||||
|
|
@ -220,10 +220,18 @@ def test_three_unattributed_pixels_fire_cluster_once():
|
||||||
assert data.get("category") == "unattributed_hotspot_cluster"
|
assert data.get("category") == "unattributed_hotspot_cluster"
|
||||||
assert data.get("severity") == "priority"
|
assert data.get("severity") == "priority"
|
||||||
|
|
||||||
# Cluster detection is currently stubbed (_maybe_emit_cluster returns None).
|
# F3: cluster detection is ENABLED. The 3rd pixel reaches cluster_min_pixels
|
||||||
# All three calls return None.
|
# (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]
|
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():
|
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 wire is None
|
||||||
assert "category" not in data4
|
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():
|
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",
|
env, subject="central.fire.hotspot.N20.high.unknown",
|
||||||
data={}, now=1780728000 + 7200 + i,
|
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]
|
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:")
|
||||||
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
|
|
|
||||||
382
work/tests/test_firms_cluster_f3.py
Normal file
382
work/tests/test_firms_cluster_f3.py
Normal file
|
|
@ -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
|
||||||
|
|
@ -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``:
|
Central handler uses, via ``central.firms_handler.ingest_hotspot_pixel``:
|
||||||
|
|
||||||
1. ``ingest_hotspot_pixel`` drives growth / spotting / halt from canonical
|
1. ``ingest_hotspot_pixel`` drives growth / spotting / halt from canonical
|
||||||
pixels and returns ``(wire, data)`` fusion broadcasts — and NEVER a raw
|
pixels (and, F3, curated ``unattributed_hotspot_cluster`` "possible new
|
||||||
hotspot / cluster / new_ignition broadcast.
|
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
|
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
|
the shared engine, emits the fusion Events, and still returns None for raw
|
||||||
hotspots.
|
hotspots.
|
||||||
|
|
@ -25,8 +26,11 @@ from datetime import datetime, timezone
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
_MI_PER_DEG_LAT = 69.0
|
_MI_PER_DEG_LAT = 69.0
|
||||||
_FUSION_CATS = {"wildfire_growth", "wildfire_spotting", "wildfire_halted"}
|
# F3: unattributed_hotspot_cluster is a CURATED fusion output (a "possible new
|
||||||
_RAW_CATS = {"wildfire_hotspot", "new_ignition", "unattributed_hotspot_cluster"}
|
# 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) ─────────────────────────
|
# ── isolation (real-DB, mirrors test_firms_refactor) ─────────────────────────
|
||||||
|
|
@ -201,17 +205,20 @@ class TestIngestHalt:
|
||||||
assert latch == float(now)
|
assert latch == float(now)
|
||||||
|
|
||||||
|
|
||||||
class TestIngestNeverRawOrCluster:
|
class TestIngestNeverRaw:
|
||||||
def test_lone_pixel_no_fire_yields_nothing(self):
|
def test_lone_pixel_no_fire_yields_nothing(self):
|
||||||
# Stored, attributed to nothing, cluster path DEAD, no idle fire ->
|
# Stored, attributed to nothing, below the cluster threshold, no idle
|
||||||
# returns []. A raw hotspot is NEVER emitted on any path.
|
# fire -> returns []. A raw hotspot is NEVER emitted on any path.
|
||||||
out = _feed(_pixel(lat=44.4, lon=-116.2, acq_time="1300"),
|
out = _feed(_pixel(lat=44.4, lon=-116.2, acq_time="1300"),
|
||||||
now=1780750000)
|
now=1780750000)
|
||||||
assert out == []
|
assert out == []
|
||||||
|
|
||||||
def test_dense_unattributed_cluster_stays_silent(self):
|
def test_dense_unattributed_cluster_broadcasts_curated(self):
|
||||||
# Several unattributed pixels close together would have tripped the old
|
# F3: several unattributed pixels close together form a CURATED cluster
|
||||||
# cluster broadcast; that path is dead -> no broadcast, ever.
|
# ("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
|
base_lat, base_lon = 44.0, -116.0
|
||||||
produced = []
|
produced = []
|
||||||
for i in range(6):
|
for i in range(6):
|
||||||
|
|
@ -219,7 +226,11 @@ class TestIngestNeverRawOrCluster:
|
||||||
acq_time=f"13{i:02d}"), now=1780750000 + i)
|
acq_time=f"13{i:02d}"), now=1780750000 + i)
|
||||||
_assert_no_raw(out)
|
_assert_no_raw(out)
|
||||||
produced.extend(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)))
|
_patch_fetch(monkeypatch, _csv(self._growth_rows(center_lat, center_lon)))
|
||||||
|
|
||||||
a = _adapter()
|
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
|
assert a.tick() is True
|
||||||
|
|
||||||
evts = a.get_events()
|
evts = a.get_events()
|
||||||
|
|
@ -341,6 +356,10 @@ class TestAdapterTickFusion:
|
||||||
# ingest would (store._ingest marks every source it touched).
|
# ingest would (store._ingest marks every source it touched).
|
||||||
store._seen = {}
|
store._seen = {}
|
||||||
store._seeded = {"firms", "firms_fusion"}
|
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
|
assert a.tick() is True
|
||||||
store._ingest("firms", a)
|
store._ingest("firms", a)
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -21,7 +21,8 @@ test_hydro_refactor.py:
|
||||||
4. Not-cutover parity: with no category cut over, handle_firms keeps the legacy
|
4. Not-cutover parity: with no category cut over, handle_firms keeps the legacy
|
||||||
eager-latch + stamps VERBATIM (byte-identical live behavior).
|
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
|
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:
|
class TestClusterBelowThreshold:
|
||||||
def test_maybe_emit_cluster_returns_none(self):
|
def test_maybe_emit_cluster_below_threshold_returns_none(self):
|
||||||
from meshai.central.firms_handler import _maybe_emit_cluster
|
from meshai.central.firms_handler import _maybe_emit_cluster
|
||||||
from meshai.persistence import get_db
|
from meshai.persistence import get_db
|
||||||
data = {}
|
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(
|
out = _maybe_emit_cluster(
|
||||||
get_db(), lat=43.0, lon=-115.0, acq_epoch=1780747200,
|
get_db(), lat=43.0, lon=-115.0, acq_epoch=1780747200,
|
||||||
frp=20.0, data=data, now=1780747200, this_pixel_id=1)
|
frp=20.0, data=data, now=1780747200, this_pixel_id=1)
|
||||||
assert out is None
|
assert out is None
|
||||||
# The dead path must not tag the data dict either.
|
|
||||||
assert data == {}
|
assert data == {}
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue