From c401e7b1fccdf31c909927ec5cb0a6bc3108820e Mon Sep 17 00:00:00 2001 From: Matt Johnson Date: Sat, 18 Jul 2026 04:03:40 +0000 Subject: [PATCH] chore(fire-ripout 2dii-a): remove dead wildfire_growth cutover path + handle_firms wildfire_growth events are always fully precomposed (env/firms.py sets _meshai_precomposed=True + title=) 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) --- work/meshai/env/fire_fusion.py | 333 ++---------------- .../notifications/formatters/__init__.py | 26 +- 2 files changed, 40 insertions(+), 319 deletions(-) diff --git a/work/meshai/env/fire_fusion.py b/work/meshai/env/fire_fusion.py index 649579b..6f09572 100644 --- a/work/meshai/env/fire_fusion.py +++ b/work/meshai/env/fire_fusion.py @@ -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 (central/ handler retirement, chore/ripout-2d). ``ingest_hotspot_pixel`` is the LIVE fire-fusion engine with one consumer: the native FIRMS adapter -(`meshai.env.firms` -> `_run_fusion`). ``handle_firms`` (the Central -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. +(`meshai.env.firms` -> `_run_fusion`). -Pre-v0.6-1 the v0.5.13 default-deny gate at consumer._normalize() silently -dropped every `central.fire.hotspot.>` envelope because no per-adapter handler -existed (audit doc v0.6-phase1-audit.md finding #2). The `firms_pixels` -table was created in v0.5.8b (v1.sql:98-111) and has been empty ever since. +chore/ripout-2dii: the dead Central NATS-envelope entrypoint ``handle_firms`` +(envelope field-extraction, confidence/FRP/bbox filtering, event_log +accounting) has been REMOVED -- it had zero live production callers (Central's +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 -`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): +Subject pattern (Central v0.10.0, historical -- kept for context): central.fire.hotspot... where is `us.` 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: Unique partial index added in v4.sql on (round(lat,5), round(lon,5), acq_time, satellite) Same satellite pixel observation re-published via NATS reconnect / 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. - -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 "|" for grep. """ from __future__ import annotations from meshai.adapter_config import adapter_config @@ -75,7 +31,6 @@ from meshai.adapter_config import adapter_config import json import logging import math -import time from datetime import datetime, timezone 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 -# names retained as backward-compat aliases for test monkeypatches; the -# 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.. -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...`). - 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) +# Source-agnostic pixel ingest (native env/firms.py; formerly also shared +# with the now-deleted Central handle_firms entrypoint) # ============================================================================ @@ -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], time_s: Optional[Any]) -> Optional[int]: """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 -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 # ============================================================================ @@ -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) # is delegated to gating.firms.decide; the wire is still rendered inline via - # the WFIGS _render for byte-identity. The boundary predicate reproduces the - # legacy guard (`last_pass_id == pass_id or last_pass_id is None or prev is - # None` -> suppress) exactly. Growth has no latch, so there is nothing to - # defer -- only the data stamping differs between the cutover and legacy - # paths. - from meshai.notifications.cutover import is_cutover + # the WFIGS _render (the live renderer for wildfire_growth -- see + # notifications/renderers/composer.py's `_meshai_precomposed` bypass; + # wildfire_growth is not in NATIVE_ALWAYS_DECIDE and the cutover formatter + # path was never invoked live, so it was removed -- chore/ripout-2dii). + # The boundary predicate reproduces the legacy guard (`last_pass_id == + # pass_id or last_pass_id is None or prev is None` -> suppress) exactly. from meshai.notifications.gating.firms import decide as _firms_decide boundary_growth = ((last_pass_id != pass_id) and (last_pass_id is not None) @@ -934,25 +666,14 @@ def _handle_pass_boundary(conn, *, irwin_id, pass_id, lat, lon, normalized = dict(fire) normalized["irwin_id"] = irwin_id if isinstance(data, dict): - if is_cutover("wildfire_growth"): - # NEW PATH: fire formatter re-renders from these hints. Feed ONLY the - # fields _render actually reads on the growth path (incident_name, - # lat/lon for the anchor, movement, is_update) so the wire matches. - data.update(gate.data_patch) - 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["_severity_override"] = "immediate" - data["_cooldown_suffix"] = irwin_id + # wildfire_growth's only live render path (byte-for-byte identical + # live output). The cutover NEW-PATH (fire formatter re-render) was + # removed -- wildfire_growth is not in NATIVE_ALWAYS_DECIDE and events + # are precomposed, so the formatter was never invoked live for this + # category (chore/ripout-2dii). + data["category"] = "wildfire_growth" + data["_severity_override"] = "immediate" + data["_cooldown_suffix"] = irwin_id return _render(normalized, prefix="Update", movement=movement) diff --git a/work/meshai/notifications/formatters/__init__.py b/work/meshai/notifications/formatters/__init__.py index b7100cf..83fc506 100644 --- a/work/meshai/notifications/formatters/__init__.py +++ b/work/meshai/notifications/formatters/__init__.py @@ -122,19 +122,19 @@ register("wildfire_declared", _fire_fmt_mod.format) register("wildfire_incident", _fire_fmt_mod.format) register("wildfire_closed", _fire_fmt_mod.format) -# Phase-3c: FIRMS fusion broadcasts. Three categories the firms_handler emits, -# all under toggle "fire". `wildfire_growth` REUSES the fire formatter — its -# wire is the WFIGS incident render with a movement dict, and feeding the fire -# formatter only {incident_name, is_update, movement, lat, lon} reproduces the -# legacy _render() call byte-for-byte (the growth SELECT's current_* / declared_at -# columns are NOT read by _render, so size/containment/date render as unknown/ -# absent — matched here by omitting those fields). `wildfire_spotting` and -# `wildfire_halted` have their own terse wires -> formatters/firms.py. Registered -# under the explicit strings (not the "fire" toggle) so the still-deferred FIRMS -# native categories (wildfire_hotspot / new_ignition / unattributed_hotspot_cluster) -# are NOT captured by the family fallback. NOT cut over this phase. -register("wildfire_growth", _fire_fmt_mod.format) - +# Phase-3c: FIRMS fusion broadcasts. `wildfire_spotting` / `wildfire_halted` +# have their own terse wires -> formatters/firms.py, registered below. +# `wildfire_growth` is NOT registered here (chore/ripout-2dii): its events are +# always fully precomposed (env/firms.py sets `_meshai_precomposed=True` + +# `title=`), and it is not in `cutover.NATIVE_ALWAYS_DECIDE`, so +# compose_mesh_message's formatter-invocation branch +# (notifications/renderers/composer.py:343-347) can never reach a registered +# formatter for this category live -- the precomposed-title bypass always wins +# first. wildfire_growth's live wire comes from `env.fire_render._render`, +# called directly by `env.fire_fusion._handle_pass_boundary`. A prior +# registration here (reusing the fire formatter) was dead code kept from the +# 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 register("wildfire_spotting", _firms_fmt_mod.format) register("wildfire_halted", _firms_fmt_mod.format)