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

chore(fire-ripout 2dii-a): remove dead wildfire_growth cutover path + handle_firms

Browse source 
wildfire_growth events are always fully precomposed (env/firms.py sets
_meshai_precomposed=True + title=<wire from _render()>) and the category is
not in cutover.NATIVE_ALWAYS_DECIDE, so the registered fire formatter could
never be reached live via compose_mesh_message for this category -- the
precomposed-title bypass always won first. Removes the dead
is_cutover("wildfire_growth") NEW-PATH branch in
fire_fusion._handle_pass_boundary (kept the live legacy leg that calls
_render unconditionally) and the now-unreachable wildfire_growth formatter
registration in notifications/formatters/__init__.py.

Also removes handle_firms, the dead Central NATS-envelope entrypoint (zero
live production callers -- Central's consumer that drove it is gone), plus
its envelope-specific filtering helpers (_confidence_passes, _in_bbox,
_coerce_severity, _log_event, FIRMS_CONFIDENCE_FLOOR/FRP_FLOOR/BBOX_OPTIONAL)
that had no live callers left. The shared, LIVE fusion core
(_ingest_pixel_core, attribution, clustering, growth/spotting/halt) and
_parse_acq_epoch are unaffected -- still the engine behind the native
ingest_hotspot_pixel entrypoint.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Matt Johnson 2026-07-18 04:03:40 +00:00
commit c401e7b1fc
2 changed files with 42 additions and 321 deletions
Download patch file Download diff file Expand all files Collapse all files

327
work/meshai/env/fire_fusion.py vendored
View file

@ -1,73 +1,29 @@
"""v0.7-fire-tracker-3 FIRMS handler -- storage + attribution + cluster + growth/halt/spotting. """v0.7-fire-tracker-3 FIRMS handler -- attribution + cluster + growth/halt/spotting.
Relocated from `meshai.central.firms_handler` during the Central ripout Relocated from `meshai.central.firms_handler` during the Central ripout
(central/ handler retirement, chore/ripout-2d). ``ingest_hotspot_pixel`` is (central/ handler retirement, chore/ripout-2d). ``ingest_hotspot_pixel`` is
the LIVE fire-fusion engine with one consumer: the native FIRMS adapter the LIVE fire-fusion engine with one consumer: the native FIRMS adapter
(`meshai.env.firms` -> `_run_fusion`). ``handle_firms`` (the Central (`meshai.env.firms` -> `_run_fusion`).
NATS-envelope entrypoint) has no live production caller -- Central's
consumer that drove it is gone -- but it remains the parity-tested legacy
contract (see `tests/test_firms_handler.py`, `tests/test_firms_refactor.py`,
`tests/test_fire_tracker_phase1/2/3.py`) -- kept verbatim, not deleted, per
that coverage.
Pre-v0.6-1 the v0.5.13 default-deny gate at consumer._normalize() silently chore/ripout-2dii: the dead Central NATS-envelope entrypoint ``handle_firms``
dropped every `central.fire.hotspot.>` envelope because no per-adapter handler (envelope field-extraction, confidence/FRP/bbox filtering, event_log
existed (audit doc v0.6-phase1-audit.md finding #2). The `firms_pixels` accounting) has been REMOVED -- it had zero live production callers (Central's
table was created in v0.5.8b (v1.sql:98-111) and has been empty ever since. consumer that drove it is gone) and its own parity tests (`test_firms_handler.py`,
`TestCentralPathUnchanged` in `test_firms_native_fusion.py`). The shared,
LIVE core (`_ingest_pixel_core`, attribution, clustering, growth/spotting/halt
fusion) is unaffected and remains fully covered via `ingest_hotspot_pixel`
(see `test_firms_native_fusion.py`, `test_fire_tracker_phase1/2/3.py`).
This handler closes that gap: every passing FIRMS pixel lands in Subject pattern (Central v0.10.0, historical -- kept for context):
`firms_pixels`. No mesh broadcasts are emitted -- FIRMS data is for the
LLM context (commit #5: env_reporter) and for the v0.6 fire-tracker
fusion (per v0.6-design-fire-tracker.md). Returning None from the handler
tells the consumer's default-deny clause "no broadcast", which is exactly
the v0.6-1 contract (memory rule 19).
Subject pattern (Central v0.10.0):
central.fire.hotspot.<satellite>.<confidence>.<region> central.fire.hotspot.<satellite>.<confidence>.<region>
where <region> is `us.<state>` or `unknown`. where <region> is `us.<state>` or `unknown`.
Envelope shape (from firms-investigation.md, 250 envelopes 2026-05-28..06-04):
envelope["data"]["adapter"] == "firms"
envelope["data"]["data"]:
latitude (REAL)
longitude (REAL)
frp (REAL, MW -- fire radiative power)
bright_ti4 (REAL, K -- VIIRS brightness temperature)
bright_ti5 (REAL, K, optional)
satellite ("N" Suomi-NPP | "N20" NOAA-20)
instrument ("VIIRS" so far; MODIS would extend this)
confidence ("nominal" | "high" | "low")
acq_date ("YYYY-MM-DD" UTC)
acq_time ("HHMM" UTC, 4-digit)
daynight ("D" | "N")
version (str)
_enriched.geocoder.city/state/county/landclass/elevation_m
Filtering (hardcoded defaults; commit #3 migrates these to adapter_config
GUI rows per Rule 17. Per Matt's lock: defaults become GUI default values
with no behavior change on first deploy.):
FIRMS_CONFIDENCE_FLOOR = "low" -- rank-based; "low" = store every conf
FIRMS_FRP_FLOOR = 0.0 -- 0 = store every FRP value
FIRMS_BBOX_OPTIONAL = None -- None = no spatial filter
Permissive defaults are intentional: storage is cheap and v0.6 fire-tracker
fusion (FIRMS + WFIGS) needs the full pixel stream to detect unattributed
clusters early. Query-time filtering happens in env_reporter (commit #5).
Dedup: Dedup:
Unique partial index added in v4.sql on Unique partial index added in v4.sql on
(round(lat,5), round(lon,5), acq_time, satellite) (round(lat,5), round(lon,5), acq_time, satellite)
Same satellite pixel observation re-published via NATS reconnect / Same satellite pixel observation re-published via NATS reconnect /
JetStream replay is a no-op INSERT OR IGNORE. 5 decimals on lat/lon JetStream replay is a no-op INSERT OR IGNORE. 5 decimals on lat/lon
is ~1.1 m precision -- well inside VIIRS' 375 m pixel. is ~1.1 m precision -- well inside VIIRS' 375 m pixel.
event_log accounting:
handled=1 -> row inserted into firms_pixels (or dedup-hit -- still
"successfully handled" semantically: we know about it)
handled=0 -> dropped (missing coords / outside bbox / below conf
floor / below FRP floor / missing acq timestamp).
Category is suffixed with "|<reason>" for grep.
""" """
from __future__ import annotations from __future__ import annotations
from meshai.adapter_config import adapter_config from meshai.adapter_config import adapter_config
@ -75,7 +31,6 @@ from meshai.adapter_config import adapter_config
import json import json
import logging import logging
import math import math
import time
from datetime import datetime, timezone from datetime import datetime, timezone
from typing import Any, Optional from typing import Any, Optional
@ -86,175 +41,8 @@ logger = logging.getLogger(__name__)
# ============================================================================ # ============================================================================
# v0.6-3b: all four settings now live in adapter_config.firms. Module-level # Source-agnostic pixel ingest (native env/firms.py; formerly also shared
# names retained as backward-compat aliases for test monkeypatches; the # with the now-deleted Central handle_firms entrypoint)
# handler reads via adapter_config so a GUI edit takes effect on the next
# envelope without restart.
# ============================================================================
# VIIRS-FIRMS confidence rank table (CODE -- NOAA-defined vocabulary).
_CONFIDENCE_RANK = {"low": 0, "nominal": 1, "high": 2}
# Back-compat aliases for tests that import these names. New code should
# read via adapter_config.firms.<key>.
FIRMS_CONFIDENCE_FLOOR = "low"
FIRMS_FRP_FLOOR = 0.0
FIRMS_BBOX_OPTIONAL: Optional[tuple[float, float, float, float]] = None
# ============================================================================
# Public entry point
# ============================================================================
def handle_firms(envelope: dict, subject: str,
data: Optional[dict] = None,
now: Optional[int] = None) -> Optional[str]:
"""Storage-only FIRMS handler. ALWAYS returns None.
Args:
envelope: CloudEvents envelope from the Central consumer.
subject: NATS subject (`central.fire.hotspot.<sat>.<conf>.<region>`).
data: Mutable Event.data dict (unused -- no broadcast attached).
now: Override current epoch for tests.
Returns:
None unconditionally. The v0.5.13 default-deny clause at
consumer._normalize() interprets None as "no broadcast", which is
the desired contract for storage-only adapters.
"""
if not isinstance(envelope, dict):
return None
inner = envelope.get("data") or {}
if (inner.get("adapter") or "") != "firms":
return None
d = inner.get("data") or {}
now = now if now is not None else int(time.time())
category_raw = inner.get("category") or ""
severity_word = _coerce_severity(inner.get("severity"))
event_id_external = inner.get("id")
try:
conn = get_db()
except Exception:
logger.exception("firms_handler: persistence unavailable; dropping")
return None
# ---- field extraction + validation -----------------------------------
lat = d.get("latitude")
lon = d.get("longitude")
if not (isinstance(lat, (int, float)) and isinstance(lon, (int, float))):
_log_event(conn, now=now, source="firms",
category=category_raw + "|missing_coords",
severity_word=severity_word,
event_id_external=event_id_external,
subject=subject, handled=0,
table_name=None, table_pk=None)
return None
lat = float(lat); lon = float(lon)
# ---- filter: bbox (optional) -----------------------------------------
if not _in_bbox(lat, lon):
_log_event(conn, now=now, source="firms",
category=category_raw + "|outside_bbox",
severity_word=severity_word,
event_id_external=event_id_external,
subject=subject, handled=0,
table_name=None, table_pk=None)
return None
# ---- filter: confidence floor ----------------------------------------
conf = d.get("confidence")
if not _confidence_passes(conf):
_log_event(conn, now=now, source="firms",
category=category_raw + "|below_confidence_floor",
severity_word=severity_word,
event_id_external=event_id_external,
subject=subject, handled=0,
table_name=None, table_pk=None)
return None
# ---- filter: FRP floor (no-op when FIRMS_FRP_FLOOR <= 0) -------------
frp_raw = d.get("frp")
try:
frp = float(frp_raw) if frp_raw is not None else None
except (TypeError, ValueError):
frp = None
import sys as _sys
_this = _sys.modules[__name__]
frp_floor = float(_this.FIRMS_FRP_FLOOR) if _this.FIRMS_FRP_FLOOR > 0 \
else float(adapter_config.firms.frp_floor)
if frp_floor > 0:
if frp is None or frp < frp_floor:
_log_event(conn, now=now, source="firms",
category=category_raw + "|below_frp_floor",
severity_word=severity_word,
event_id_external=event_id_external,
subject=subject, handled=0,
table_name=None, table_pk=None)
return None
# ---- acquisition timestamp (required for dedup key) ------------------
acq_epoch = _parse_acq_epoch(d.get("acq_date"), d.get("acq_time"))
if acq_epoch is None:
_log_event(conn, now=now, source="firms",
category=category_raw + "|missing_acq_time",
severity_word=severity_word,
event_id_external=event_id_external,
subject=subject, handled=0,
table_name=None, table_pk=None)
return None
# ---- 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 \
else d.get("brightness")
try:
brightness = float(brightness_raw) if brightness_raw is not None else None
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)
# ============================================================================ # ============================================================================
@ -401,41 +189,6 @@ def ingest_hotspot_pixel(pixel: dict, *, now, seed=False) -> list[tuple[str, dic
# ============================================================================ # ============================================================================
def _confidence_passes(conf: Optional[str]) -> bool:
"""Return True iff `conf` is at or above the configured floor.
v0.6-3b: floor read from adapter_config.firms.confidence_floor; the
module-level FIRMS_CONFIDENCE_FLOOR still wins when explicitly
monkeypatched (so existing tests stay one-line).
"""
if conf is None:
return False
rank = _CONFIDENCE_RANK.get(str(conf).lower())
if rank is None:
return False
import sys
_this = sys.modules[__name__]
if _this.FIRMS_CONFIDENCE_FLOOR != "low":
floor_str = _this.FIRMS_CONFIDENCE_FLOOR
else:
floor_str = str(adapter_config.firms.confidence_floor)
floor = _CONFIDENCE_RANK.get(str(floor_str).lower(), 0)
return rank >= floor
def _in_bbox(lat: float, lon: float) -> bool:
import sys
_this = sys.modules[__name__]
if _this.FIRMS_BBOX_OPTIONAL is not None:
bbox = _this.FIRMS_BBOX_OPTIONAL
else:
bbox = adapter_config.firms.bbox
if bbox is None:
return True
min_lat, min_lon, max_lat, max_lon = bbox
return (min_lat <= lat <= max_lat) and (min_lon <= lon <= max_lon)
def _parse_acq_epoch(date_s: Optional[str], def _parse_acq_epoch(date_s: Optional[str],
time_s: Optional[Any]) -> Optional[int]: time_s: Optional[Any]) -> Optional[int]:
"""FIRMS publishes acq_date 'YYYY-MM-DD' + acq_time HHMM (UTC). """FIRMS publishes acq_date 'YYYY-MM-DD' + acq_time HHMM (UTC).
@ -454,27 +207,6 @@ def _parse_acq_epoch(date_s: Optional[str],
return None return None
def _coerce_severity(sev: Any) -> Optional[str]:
if sev is None: return None
if isinstance(sev, str): return sev or None
try: return str(int(sev))
except (TypeError, ValueError): return str(sev)
def _log_event(conn, *, now, source, category, severity_word,
event_id_external, subject, handled,
table_name, table_pk) -> None:
"""event_log writer -- shape matches sibling handlers exactly."""
conn.execute(
"INSERT INTO event_log(received_at, source, category, severity_word, "
"event_id_external, nats_subject, handled, table_name, table_pk) "
"VALUES (?,?,?,?,?,?,?,?,?)",
(now, source, category, severity_word, event_id_external, subject,
int(bool(handled)), table_name, table_pk),
)
# ============================================================================ # ============================================================================
# v0.7-fire-tracker-1: attribution + unattributed-cluster detection # v0.7-fire-tracker-1: attribution + unattributed-cluster detection
# ============================================================================ # ============================================================================
@ -892,12 +624,12 @@ def _handle_pass_boundary(conn, *, irwin_id, pass_id, lat, lon,
# Phase-3c: the growth broadcast DECISION (pass boundary + drift threshold) # Phase-3c: the growth broadcast DECISION (pass boundary + drift threshold)
# is delegated to gating.firms.decide; the wire is still rendered inline via # is delegated to gating.firms.decide; the wire is still rendered inline via
# the WFIGS _render for byte-identity. The boundary predicate reproduces the # the WFIGS _render (the live renderer for wildfire_growth -- see
# legacy guard (`last_pass_id == pass_id or last_pass_id is None or prev is # notifications/renderers/composer.py's `_meshai_precomposed` bypass;
# None` -> suppress) exactly. Growth has no latch, so there is nothing to # wildfire_growth is not in NATIVE_ALWAYS_DECIDE and the cutover formatter
# defer -- only the data stamping differs between the cutover and legacy # path was never invoked live, so it was removed -- chore/ripout-2dii).
# paths. # The boundary predicate reproduces the legacy guard (`last_pass_id ==
from meshai.notifications.cutover import is_cutover # pass_id or last_pass_id is None or prev is None` -> suppress) exactly.
from meshai.notifications.gating.firms import decide as _firms_decide from meshai.notifications.gating.firms import decide as _firms_decide
boundary_growth = ((last_pass_id != pass_id) and (last_pass_id is not None) boundary_growth = ((last_pass_id != pass_id) and (last_pass_id is not None)
@ -934,22 +666,11 @@ def _handle_pass_boundary(conn, *, irwin_id, pass_id, lat, lon,
normalized = dict(fire) normalized = dict(fire)
normalized["irwin_id"] = irwin_id normalized["irwin_id"] = irwin_id
if isinstance(data, dict): if isinstance(data, dict):
if is_cutover("wildfire_growth"): # wildfire_growth's only live render path (byte-for-byte identical
# NEW PATH: fire formatter re-renders from these hints. Feed ONLY the # live output). The cutover NEW-PATH (fire formatter re-render) was
# fields _render actually reads on the growth path (incident_name, # removed -- wildfire_growth is not in NATIVE_ALWAYS_DECIDE and events
# lat/lon for the anchor, movement, is_update) so the wire matches. # are precomposed, so the formatter was never invoked live for this
data.update(gate.data_patch) # category (chore/ripout-2dii).
data["incident_name"] = fire["incident_name"]
data["lat"] = fire["lat"]
data["lon"] = fire["lon"]
data["_broadcast_audit"] = {"table": "fires", "pk": irwin_id}
_raw_commit = gate.commit
if _raw_commit is not None:
def _on_commit(committed_at: float) -> None:
_raw_commit(committed_at)
data["_on_broadcast_committed"] = _on_commit
else:
# LEGACY verbatim (byte-for-byte identical live output).
data["category"] = "wildfire_growth" data["category"] = "wildfire_growth"
data["_severity_override"] = "immediate" data["_severity_override"] = "immediate"
data["_cooldown_suffix"] = irwin_id data["_cooldown_suffix"] = irwin_id

26
work/meshai/notifications/formatters/__init__.py
View file

@ -122,19 +122,19 @@ register("wildfire_declared", _fire_fmt_mod.format)
register("wildfire_incident", _fire_fmt_mod.format) register("wildfire_incident", _fire_fmt_mod.format)
register("wildfire_closed", _fire_fmt_mod.format) register("wildfire_closed", _fire_fmt_mod.format)
# Phase-3c: FIRMS fusion broadcasts. Three categories the firms_handler emits, # Phase-3c: FIRMS fusion broadcasts. `wildfire_spotting` / `wildfire_halted`
# all under toggle "fire". `wildfire_growth` REUSES the fire formatter — its # have their own terse wires -> formatters/firms.py, registered below.
# wire is the WFIGS incident render with a movement dict, and feeding the fire # `wildfire_growth` is NOT registered here (chore/ripout-2dii): its events are
# formatter only {incident_name, is_update, movement, lat, lon} reproduces the # always fully precomposed (env/firms.py sets `_meshai_precomposed=True` +
# legacy _render() call byte-for-byte (the growth SELECT's current_* / declared_at # `title=<wire>`), and it is not in `cutover.NATIVE_ALWAYS_DECIDE`, so
# columns are NOT read by _render, so size/containment/date render as unknown/ # compose_mesh_message's formatter-invocation branch
# absent — matched here by omitting those fields). `wildfire_spotting` and # (notifications/renderers/composer.py:343-347) can never reach a registered
# `wildfire_halted` have their own terse wires -> formatters/firms.py. Registered # formatter for this category live -- the precomposed-title bypass always wins
# under the explicit strings (not the "fire" toggle) so the still-deferred FIRMS # first. wildfire_growth's live wire comes from `env.fire_render._render`,
# native categories (wildfire_hotspot / new_ignition / unattributed_hotspot_cluster) # called directly by `env.fire_fusion._handle_pass_boundary`. A prior
# are NOT captured by the family fallback. NOT cut over this phase. # registration here (reusing the fire formatter) was dead code kept from the
register("wildfire_growth", _fire_fmt_mod.format) # Phase-3c migration bake and has been removed; see TestRegistration in
# tests/test_firms_refactor.py for the regression guard.
from meshai.notifications.formatters import firms as _firms_fmt_mod # noqa: E402,F401 from meshai.notifications.formatters import firms as _firms_fmt_mod # noqa: E402,F401
register("wildfire_spotting", _firms_fmt_mod.format) register("wildfire_spotting", _firms_fmt_mod.format)
register("wildfire_halted", _firms_fmt_mod.format) register("wildfire_halted", _firms_fmt_mod.format)