diff --git a/work/meshai/central/wfigs_handler.py b/work/meshai/central/wfigs_handler.py index b892124..4368fb6 100644 --- a/work/meshai/central/wfigs_handler.py +++ b/work/meshai/central/wfigs_handler.py @@ -83,16 +83,55 @@ def _fire_too_old_to_announce(declared_at_epoch, now) -> bool: # ---------- public entry -------------------------------------------------- +def _build_canonical(normalized: dict, kind: str) -> dict: + """Flat canonical dict consumed by gating.fire.decide + formatters.fire. + + Carries the render-ready WFIGS fields the decider (gating decision) and the + formatter (wire) read. ``_kind`` routes the decider between the active-fire + state machine, the tombstone all-clear, and the never-broadcast perimeter. + """ + return { + "_kind": kind, + "irwin_id": normalized.get("irwin_id"), + "incident_name": normalized.get("incident_name"), + "incident_type": normalized.get("incident_type"), + "acres": normalized.get("acres"), + "contained_pct": normalized.get("contained_pct"), + "fire_cause": normalized.get("fire_cause"), + "declared_at_epoch": normalized.get("declared_at_epoch"), + "unique_fire_id": normalized.get("unique_fire_id"), + "lat": normalized.get("lat"), + "lon": normalized.get("lon"), + "county": normalized.get("county"), + "state": normalized.get("state"), + "landclass": normalized.get("landclass"), + "geocoder_city": normalized.get("geocoder_city"), + } + + def handle_wfigs(normalized: dict, envelope: dict, subject: str, data: Optional[dict] = None, now: Optional[int] = None) -> Optional[str]: """Route a normalized WFIGS dict through persistence + change-detection. + Phase-3b refactor: the broadcast DECISION (New/Update/suppress + cooldown + + age-gate + all-clear eligibility) now lives in + ``meshai.notifications.gating.fire.decide``. This handler keeps ownership + of the inline ``fires`` INSERT/UPDATE of ``current_*`` (unconditional state + write), the ``tombstoned_at`` stamp, the ``event_log`` row + handled flip, + and the wire it returns (mirror quake/nwis). + `data` is the mutable dict the caller (consumer._normalize) is composing into the Event. When a broadcast should fire, the handler attaches an `_on_broadcast_committed` callback and `_broadcast_audit` descriptor to it; the dispatcher invokes both AFTER a successful deliver(). + Cutover gate: when the emitted category is explicitly cut over, the handler + writes ``gate.data_patch`` + wraps ``gate.commit`` (new path live); + otherwise it keeps the legacy ``_attach_commit_handles`` / all-clear + stamping VERBATIM so the live broadcast stays byte-for-byte identical while + the new formatter+decider bake in shadow. + Returns a wire string when a broadcast should fire, None otherwise. """ if not isinstance(normalized, dict): @@ -114,6 +153,11 @@ def handle_wfigs(normalized: dict, envelope: dict, subject: str, "deferring to default pipeline") return None + from meshai.notifications.cutover import is_cutover + from meshai.notifications.gating.fire import decide as _gate_decide + + canonical = _build_canonical(normalized, kind) + if kind in ("wfigs_tombstone", "wfigs_perimeter"): source = "wfigs_incidents" if kind == "wfigs_tombstone" else "wfigs_perimeters" log_id = _log_event_returning_id( @@ -124,6 +168,7 @@ def handle_wfigs(normalized: dict, envelope: dict, subject: str, # v0.6-tail item 4: tombstone branch stamps fires.tombstoned_at so # the ReminderScheduler stops re-broadcasting the closed fire. # Only the tombstone kind closes the fire; perimeter polls don t. + # UNCONDITIONAL state write — stays inline, mirror the original. if kind == "wfigs_tombstone" and irwin_id: try: conn.execute( @@ -136,13 +181,17 @@ def handle_wfigs(normalized: dict, envelope: dict, subject: str, # All-clear broadcast: only fires that previously made it to mesh # get a closure message. Silent for fires that were never broadcast. + # The DECISION (row exists AND last_broadcast_at NOT NULL) is delegated + # to the decider; the WIRE is still rendered here from the fire row for + # byte-identity, and the not-cutover data stamping stays verbatim. if kind == "wfigs_tombstone" and irwin_id: - fire_row = conn.execute( - "SELECT incident_name, current_acres, current_contained_pct, " - "last_broadcast_at, county, state, lat, lon " - "FROM fires WHERE irwin_id = ?", (irwin_id,) - ).fetchone() - if fire_row is not None and fire_row["last_broadcast_at"] is not None: + gate = _gate_decide(canonical, source="wfigs", now=float(now)) + if gate.broadcast: + fire_row = conn.execute( + "SELECT incident_name, current_acres, current_contained_pct, " + "last_broadcast_at, county, state, lat, lon " + "FROM fires WHERE irwin_id = ?", (irwin_id,) + ).fetchone() name = fire_row["incident_name"] or "(unnamed fire)" # Build line 2 parts parts = [] @@ -163,15 +212,36 @@ def handle_wfigs(normalized: dict, envelope: dict, subject: str, lines.append(" | ".join(parts)) wire = "\n".join(lines) if isinstance(data, dict): - data["category"] = "wildfire_closed" - data["_severity_override"] = "priority" - _attach_commit_handles( - data, irwin_id=irwin_id, - acres=fire_row["current_acres"], - contained_pct=fire_row["current_contained_pct"], - event_log_row_id=log_id) - if isinstance(data, dict): - data["_dedup_suffix"] = "closed" + if is_cutover("wildfire_closed"): + # NEW PATH: formatter re-renders from data_patch fields. + data.update(gate.data_patch) + data["_broadcast_audit"] = {"table": "fires", "pk": irwin_id} + _raw_commit = gate.commit + _log_row_id = log_id + + def _on_commit(committed_at: float) -> None: + if _raw_commit is not None: + _raw_commit(committed_at) + if _log_row_id is not None: + try: + get_db().execute( + "UPDATE event_log SET handled=1 WHERE id=?", + (int(_log_row_id),)) + except Exception: + logger.exception( + "wfigs closed commit: event_log update failed") + + data["_on_broadcast_committed"] = _on_commit + else: + # LEGACY verbatim (byte-for-byte identical live output). + data["category"] = "wildfire_closed" + data["_severity_override"] = "priority" + _attach_commit_handles( + data, irwin_id=irwin_id, + acres=fire_row["current_acres"], + contained_pct=fire_row["current_contained_pct"], + event_log_row_id=log_id) + data["_dedup_suffix"] = "closed" return wire return None @@ -187,15 +257,19 @@ def handle_wfigs(normalized: dict, envelope: dict, subject: str, subject=subject, handled=0, table_name="fires", table_pk=irwin_id) - row = conn.execute( - "SELECT current_acres, current_contained_pct, last_broadcast_at, " - "last_broadcast_acres, last_broadcast_contained " - "FROM fires WHERE irwin_id = ?", (irwin_id,)).fetchone() - acres = normalized.get("acres") contained_pct = normalized.get("contained_pct") - # ---- (i) row missing -- INSERT, mark "New", but DO NOT set last_broadcast_* + # Delegate the New/Update/suppress + age-gate + cooldown decision. The + # decider READS the fires row (pre-write) exactly as the original inline + # branch did; the inline INSERT/UPDATE below stays handler-owned. + gate = _gate_decide(canonical, source="wfigs", now=float(now)) + + # ---- inline state write (UNCONDITIONAL) -- INSERT or UPDATE current_*. + # Re-read the row (same pre-write state the decider saw) to pick the write. + row = conn.execute( + "SELECT last_broadcast_at FROM fires WHERE irwin_id = ?", + (irwin_id,)).fetchone() if row is None: conn.execute( "INSERT INTO fires(irwin_id, incident_name, incident_type, " @@ -217,27 +291,7 @@ def handle_wfigs(normalized: dict, envelope: dict, subject: str, None, None, None, # last_broadcast_* explicitly NULL ), ) - # Step 3 age-gate: keep the INSERT (so genuine future Updates work) but - # suppress the "New" broadcast for fires whose declared_at is too old. - if _fire_too_old_to_announce(normalized.get("declared_at_epoch"), now): - return None - wire = _render(normalized, prefix="New") - # v0.7-fire-tracker-1: tag first-sight broadcasts with the new - # wildfire_declared category so the dispatcher rules them apart - # from acres/containment updates (wildfire_incident). - if isinstance(data, dict): - data["category"] = "wildfire_declared" - # v0.6-3c: severity override for fire broadcasts (downgraded from - # immediate to priority to prevent cooldown/grouper bypass) - if isinstance(data, dict): - data["_severity_override"] = "priority" - _attach_commit_handles(data, irwin_id=irwin_id, - acres=acres, contained_pct=contained_pct, - event_log_row_id=log_id) - return wire - - # ---- (ii) row exists but never broadcast -- UPDATE current_*, prefix="New" - if row["last_broadcast_at"] is None: + else: conn.execute( "UPDATE fires SET current_acres=?, current_contained_pct=?, " "lat=COALESCE(?, lat), lon=COALESCE(?, lon), last_event_at=? " @@ -245,72 +299,57 @@ def handle_wfigs(normalized: dict, envelope: dict, subject: str, (acres, contained_pct, normalized.get("lat"), normalized.get("lon"), now, irwin_id), ) - # Step 3 age-gate: keep the UPDATE but suppress the "New" broadcast for - # fires whose declared_at is too old (closed/stale-fire resurrection). - if _fire_too_old_to_announce(normalized.get("declared_at_epoch"), now): - return None - wire = _render(normalized, prefix="New") - # v0.7-fire-tracker-1: case-(ii) is also first-sight as far as - # broadcast history goes -- the row exists because some prior - # handler call ran but no actual broadcast went out. - if isinstance(data, dict): - data["category"] = "wildfire_declared" - # v0.6-3c: severity override for fire broadcasts (downgraded from - # immediate to priority to prevent cooldown/grouper bypass) - if isinstance(data, dict): + + if not gate.broadcast: + # Case-(iii) suppress (no forward change OR inside cooldown) ran the + # stale-fire cleanup in the original; the age-gate suppress did not. + if gate.lifecycle == "cooldown": + _cleanup_stale_fires(conn) + return None + + # ---- broadcast: render the wire (back-compat) + stamp data. + prefix = "Update" if gate.lifecycle == "update" else "New" + wire = _render(normalized, prefix=prefix, + last_bcast_acres=gate.data_patch.get("last_bcast_acres"), + last_bcast_contained=gate.data_patch.get("last_bcast_contained")) + + # Cutover gate keys on the category THIS broadcast carries: New first-sight + # is wildfire_declared, growth Update stays wildfire_incident. + _cut = (is_cutover("wildfire_incident") if gate.lifecycle == "update" + else is_cutover("wildfire_declared")) + if isinstance(data, dict): + if _cut: + # NEW PATH: formatter re-renders from canonical + data_patch. + data.update(canonical) + data.update(gate.data_patch) + data["_broadcast_audit"] = {"table": "fires", "pk": irwin_id} + _raw_commit = gate.commit + _log_row_id = log_id + + def _on_commit(committed_at: float) -> None: + if _raw_commit is not None: + _raw_commit(committed_at) + if _log_row_id is not None: + try: + get_db().execute( + "UPDATE event_log SET handled=1 WHERE id=?", + (int(_log_row_id),)) + except Exception: + logger.exception( + "wfigs commit: event_log update failed") + + data["_on_broadcast_committed"] = _on_commit + else: + # LEGACY verbatim (byte-for-byte identical live output). + # New first-sight tags wildfire_declared; Update keeps the + # envelope-derived wildfire_incident (no category override). + if gate.lifecycle != "update": + data["category"] = "wildfire_declared" data["_severity_override"] = "priority" - _attach_commit_handles(data, irwin_id=irwin_id, - acres=acres, contained_pct=contained_pct, - event_log_row_id=log_id) - return wire - - # ---- (iii) row exists AND already broadcast -- gate on change + 8h cooldown - conn.execute( - "UPDATE fires SET current_acres=?, current_contained_pct=?, " - "lat=COALESCE(?, lat), lon=COALESCE(?, lon), last_event_at=? " - "WHERE irwin_id=?", - (acres, contained_pct, normalized.get("lat"), - normalized.get("lon"), now, irwin_id), - ) - - last_bcast_at = row["last_broadcast_at"] - last_bcast_acres = row["last_broadcast_acres"] - last_bcast_contained = row["last_broadcast_contained"] - - # Forward-only change detection: more acres or higher containment counts. - # Downward revisions and unchanged values do not warrant re-broadcast. - # v0.6-3b: each axis can be silenced via adapter_config toggles. - changed_acres = ( - bool(adapter_config.wfigs.broadcast_on_acres) - and acres is not None - and (last_bcast_acres is None or acres > last_bcast_acres) - ) - changed_contained = ( - bool(adapter_config.wfigs.broadcast_on_contained) - and contained_pct is not None - and (last_bcast_contained is None or contained_pct > last_bcast_contained) - ) - cooldown_s = int(adapter_config.wfigs.cooldown_seconds) - eight_hours_passed = ( - last_bcast_at is None - or (now - int(last_bcast_at) >= cooldown_s) - ) - - if (changed_acres or changed_contained) and eight_hours_passed: - wire = _render(normalized, prefix="Update", - last_bcast_acres=last_bcast_acres, - last_bcast_contained=last_bcast_contained) - # v0.6-3c: severity override for fire updates (downgraded from - # immediate to priority to prevent cooldown/grouper bypass) - if isinstance(data, dict): - data["_severity_override"] = "priority" - _attach_commit_handles(data, irwin_id=irwin_id, - acres=acres, contained_pct=contained_pct, - event_log_row_id=log_id) - return wire - - _cleanup_stale_fires(conn) - return None + _attach_commit_handles(data, irwin_id=irwin_id, + acres=acres, contained_pct=contained_pct, + event_log_row_id=log_id) + return wire # ---------- commit-callback factory --------------------------------------- diff --git a/work/meshai/notifications/formatters/__init__.py b/work/meshai/notifications/formatters/__init__.py index c8c3cd6..b65cde6 100644 --- a/work/meshai/notifications/formatters/__init__.py +++ b/work/meshai/notifications/formatters/__init__.py @@ -94,3 +94,17 @@ register("traffic_congestion", _incident_fmt_mod.format) # stream_high_water) are deferred — see gating/__init__.py note. from meshai.notifications.formatters import hydro as _hydro_fmt_mod # noqa: E402,F401 register("stream_flow", _hydro_fmt_mod.format) + +# Phase-3b: WFIGS wildfire. The Central wfigs_handler emits THREE explicit +# categories, all under toggle "fire": `wildfire_declared` (first-sight New, +# cases i/ii), `wildfire_incident` (growth/containment Update, case iii — the +# envelope-derived category with no override), and `wildfire_closed` (tombstone +# all-clear). One formatter handles all three; event.category selects the wire +# shape (wildfire_closed -> all-clear, else incident/growth). Registered under +# the explicit strings — NOT the "fire" toggle — so FIRMS categories +# (wildfire_hotspot / new_ignition / unattributed_hotspot_cluster / +# wildfire_growth) are NOT captured by the family fallback. FIRMS is deferred. +from meshai.notifications.formatters import fire as _fire_fmt_mod # noqa: E402,F401 +register("wildfire_declared", _fire_fmt_mod.format) +register("wildfire_incident", _fire_fmt_mod.format) +register("wildfire_closed", _fire_fmt_mod.format) diff --git a/work/meshai/notifications/formatters/fire.py b/work/meshai/notifications/formatters/fire.py new file mode 100644 index 0000000..c291198 --- /dev/null +++ b/work/meshai/notifications/formatters/fire.py @@ -0,0 +1,207 @@ +"""WFIGS wildfire formatter — Phase-3b migration. + +Reproduces BOTH legacy WFIGS wire shapes byte-identically, reading the canonical +schema the Central path writes into event.data on broadcast: + + (a) Active incident / growth — mirrors ``wfigs_handler._render`` exactly: + Line 1: 🔥 {name} — {New|Update} + Line 2: {acres} ac{ (+delta)} · containment {pct}% + Line 3: {movement line} OR {anchor} + Line 4: Cause: {cause} · Discovered {Mon D} (either/both/neither) + + (b) All-clear ("contained & closed") — mirrors the tombstone branch exactly: + Line 1: ✅ {name} — contained & closed + Line 2: {acres} ac | {pct}% contained | {anchor} (only present parts) + +Branch selection (event.data): + category == "wildfire_closed" OR _kind == "wfigs_tombstone" -> all-clear + otherwise -> incident + +Canonical schema consumed (active incident): + incident_name, acres, contained_pct, fire_cause, declared_at_epoch, + unique_fire_id, lat, lon, county, state, landclass, geocoder_city, + movement (FIRMS-injected {direction, speed_mph}, else absent/None), + is_update, last_bcast_acres, last_bcast_contained (decider render hints) + +Canonical schema consumed (all-clear): + incident_name, acres, contained_pct, lat, lon, county, state + +Anchor resolution (``_fire_anchor``) reproduces ``wfigs_handler._location_anchor`` +tier-for-tier: geocoder_city → curated town_anchors / Photon nearest_town (via +the shared ``resolve_anchor`` helper, re-formatted to the legacy string) → +landclass → "{county} Co {state}" → state → "(location unknown)". The extra +fallback tiers around ``resolve_anchor`` are required for byte-identity because +``resolve_anchor`` covers only the town step. + +Time contract: ``now`` is accepted but unused (the discovery date is rendered +from ``declared_at_epoch`` in a fixed UTC-6 offset, exactly as ``_render``). +``budget`` is injected — the caller supplies ``budget_for("wfigs")``. +""" +from __future__ import annotations + +import datetime as _dt +import logging +from typing import TYPE_CHECKING, Optional + +from meshai.adapter_config import adapter_config +from meshai.notifications.formatters._anchor import resolve_anchor +from meshai.notifications.formatters._budget import fit_to_budget + +if TYPE_CHECKING: + from meshai.notifications.events import Event + +logger = logging.getLogger(__name__) + + +def _fire_anchor(d: dict) -> str: + """Byte-identical replica of ``wfigs_handler._location_anchor``. + + geocoder.city > nearest town (curated town_anchors, then Photon) > + landclass > "{county} Co {state}" > state > "(location unknown)". + """ + city = d.get("geocoder_city") + if city: + return str(city) + + lat = d.get("lat") + lon = d.get("lon") + if isinstance(lat, (int, float)) and isinstance(lon, (int, float)): + try: + max_mi = float(adapter_config.wfigs.anchor_max_mi) + except Exception: + max_mi = 100.0 + try: + res = resolve_anchor(lat, lon, max_mi=max_mi) + except Exception: + logger.debug("fire anchor: resolve_anchor failed; falling through") + res = None + if res and res.get("town"): + # Legacy applies .title() in BOTH the town_anchors and nearest_town + # arms; resolve_anchor titles only the town_anchors arm, so title + # here unconditionally (idempotent for already-titled names). + town = str(res["town"]).title() + dist = res.get("distance_mi") + bearing = res.get("bearing") + if isinstance(dist, (int, float)): + if dist < 1: + return f"near {town}" + return f"{int(round(dist))} mi {bearing or ''} of {town}".strip() + return f"near {town}" + + landclass = d.get("landclass") + if landclass: + return str(landclass) + + county = d.get("county") + state = d.get("state") + if county and state: + return f"{county} Co {state}" + if state: + return str(state) + return "(location unknown)" + + +def _render_incident(d: dict, budget: int) -> str: + """Byte-identical replica of ``wfigs_handler._render`` (active incident).""" + name = d.get("incident_name") or "(unnamed)" + acres = d.get("acres") + contained_pct = d.get("contained_pct") + cause = d.get("fire_cause") + declared_at_epoch = d.get("declared_at_epoch") + movement = d.get("movement") + is_update = bool(d.get("is_update")) + last_bcast_acres = d.get("last_bcast_acres") + prefix = "Update" if is_update else "New" + anchor = _fire_anchor(d) + + lines: list[str] = [] + + # Line 1: header + lines.append(f"🔥 {name} — {prefix}") + + # Line 2: size / containment with delta (plain text — no bold markdown). + acres_str = f"{int(acres):,} ac" if acres is not None else "size unknown" + delta_str = "" + if (prefix == "Update" and last_bcast_acres is not None + and acres is not None and acres > last_bcast_acres): + delta_str = f" (+{int(acres - last_bcast_acres):,})" + contained_str = (f"containment {int(contained_pct)}%" + if contained_pct is not None else "containment unknown") + lines.append(f"{acres_str}{delta_str} · {contained_str}") + + # Line 3: movement or plain anchor. + if (isinstance(movement, dict) + and movement.get("direction") and movement.get("speed_mph") is not None): + lines.append(f"Moving {movement['direction']} {movement['speed_mph']:.1f} mi/h · {anchor}") + else: + lines.append(f"{anchor}") + + # Line 4: cause / discovered (DATE ONLY — no time-of-day). + cause_part = cause if cause else None + disc_part = None + if declared_at_epoch is not None: + try: + dt = _dt.datetime.fromtimestamp( + declared_at_epoch, + tz=_dt.timezone(_dt.timedelta(hours=-6))) + disc_part = dt.strftime("%b %-d") + except Exception: + pass + if cause_part and disc_part: + lines.append(f"Cause: {cause_part} · Discovered {disc_part}") + elif cause_part: + lines.append(f"Cause: {cause_part}") + elif disc_part: + lines.append(f"Discovered {disc_part}") + + return fit_to_budget("\n".join(lines), budget) + + +def _render_allclear(d: dict, budget: int) -> str: + """Byte-identical replica of the ``wfigs_handler`` tombstone all-clear wire.""" + name = d.get("incident_name") or "(unnamed fire)" + parts: list[str] = [] + acres = d.get("acres") + contained_pct = d.get("contained_pct") + if acres is not None: + parts.append(f"{int(acres):,} ac") + if contained_pct is not None: + parts.append(f"{int(contained_pct)}% contained") + # Location via the same anchor chain, but the tombstone branch feeds ONLY + # {lat, lon, county, state} (no geocoder_city / landclass). + loc_dict = { + "lat": d.get("lat"), "lon": d.get("lon"), + "county": d.get("county"), "state": d.get("state"), + } + anchor = _fire_anchor(loc_dict) + if anchor and anchor != "(location unknown)": + parts.append(anchor) + lines = [f"✅ {name} — contained & closed"] + if parts: + lines.append(" | ".join(parts)) + return fit_to_budget("\n".join(lines), budget) + + +def format(event: "Event", *, now: float, budget: int) -> str: + """Render the WFIGS wire string from canonical event.data. + + Args: + event: Pipeline Event — reads from event.data (canonical schema). + now: Frozen-clock epoch (seam; not used in current rendering). + budget: Mesh-packet character budget (from budget_for("wfigs")). + + Returns: + UTF-8 string fitting within *budget* characters. + """ + d = event.data or {} + category = None + try: + category = event.category + except Exception: + category = None + if category is None: + category = d.get("category") + + if category == "wildfire_closed" or d.get("_allclear") or d.get("_kind") == "wfigs_tombstone": + return _render_allclear(d, budget) + return _render_incident(d, budget) diff --git a/work/meshai/notifications/gating/__init__.py b/work/meshai/notifications/gating/__init__.py index 3ac485d..286074b 100644 --- a/work/meshai/notifications/gating/__init__.py +++ b/work/meshai/notifications/gating/__init__.py @@ -86,3 +86,17 @@ register("traffic_congestion", _incident_gate_mod.decide) # NOT cut over, store._emit_event's native decider hook won't run it. from meshai.notifications.gating import hydro as _hydro_gate_mod # noqa: E402,F401 register("stream_flow", _hydro_gate_mod.decide) + +# Phase-3b: WFIGS wildfire. Three explicit categories the wfigs_handler emits: +# `wildfire_declared` (New), `wildfire_incident` (growth Update), and +# `wildfire_closed` (tombstone all-clear). One decider handles all three; it +# keys off canonical `_kind` (wfigs_incident / wfigs_tombstone / wfigs_perimeter) +# to route New/Update/suppress vs the all-clear eligibility. Registered under +# explicit strings (not the "fire" toggle) so FIRMS categories are untouched — +# FIRMS + native env/fires.py are a deferred follow-up (env/fires.py lacks the +# IRWIN/FireCause/landclass fields and has no tombstone concept, and since fire +# is NOT cut over, store._emit_event's native decider hook won't run it). +from meshai.notifications.gating import fire as _fire_gate_mod # noqa: E402,F401 +register("wildfire_declared", _fire_gate_mod.decide) +register("wildfire_incident", _fire_gate_mod.decide) +register("wildfire_closed", _fire_gate_mod.decide) diff --git a/work/meshai/notifications/gating/fire.py b/work/meshai/notifications/gating/fire.py new file mode 100644 index 0000000..0ee5c8e --- /dev/null +++ b/work/meshai/notifications/gating/fire.py @@ -0,0 +1,256 @@ +"""WFIGS wildfire gating decider — Phase-3b migration. + +Moves the broadcast DECISION out of ``central.wfigs_handler.handle_wfigs`` (the +New-vs-Update-vs-suppress state machine, the forward-only acres/containment +growth gate + 8h cooldown, the stale-fire age-gate, and the tombstone all-clear +eligibility) into a source-agnostic decider, mirroring quake/hydro. + +The Central handler keeps ownership of: + * the ``fires`` INSERT / UPDATE of ``current_*`` (unconditional state write), + * the ``tombstoned_at`` stamp, + * the ``event_log`` row + its handled flip, + * the wire rendering (``_render`` / all-clear) it returns for back-compat. + +This decider owns ONLY: + * the read of the ``fires`` row for the gating decision, + * the growth/cooldown/age math, + * the ``data_patch`` stamps every broadcast carries + (``category`` override, ``_severity_override``, ``_dedup_suffix``, + ``_cooldown_suffix``, and the render hints ``is_update`` / + ``last_bcast_acres`` / ``last_bcast_contained``), + * the deferred ``commit`` closure that UPSERTs + ``fires(last_broadcast_at, first_broadcast_at, last_broadcast_acres, + last_broadcast_contained)`` on confirmed delivery. + +Canonical ``data`` dict consumed (built by the handler from the normalized +WFIGS dict): + _kind, irwin_id, incident_name, incident_type, acres, contained_pct, + fire_cause, declared_at_epoch, unique_fire_id, lat, lon, county, state, + landclass, geocoder_city + +Three ``_kind`` values are handled: + wfigs_incident -> active-fire New/Update/suppress state machine + wfigs_tombstone -> all-clear ("contained & closed") eligibility + wfigs_perimeter -> never broadcasts (suppress) + +Lifecycle labels (drive handler-side side effects): + "new" first-sight broadcast (INSERT or row-exists-never-broadcast) + "update" forward-only growth/containment broadcast after cooldown + "closed" tombstone all-clear broadcast + "cooldown" case-(iii) suppress (no change OR inside cooldown) — the handler + runs ``_cleanup_stale_fires`` for THIS suppress only + "suppress" every other suppress (age-gate, never-broadcast tombstone, + perimeter, unknown) — no cleanup +""" +from __future__ import annotations + +import logging +from typing import Optional + +from meshai.adapter_config import adapter_config +from meshai.notifications.gating.base import GateResult +from meshai.persistence import get_db + +logger = logging.getLogger(__name__) + + +def _fire_too_old_to_announce(declared_at_epoch, now) -> bool: + """Stale/closed-fire resurrection guard for first-announce ("New") only. + + Verbatim copy of ``wfigs_handler._fire_too_old_to_announce``: re-reads the + knob at the use-site (GUI-live), fails OPEN (announces) when the gate is + disabled or the fire has no discovery date. + """ + try: + max_age = int(adapter_config.wfigs.max_declare_age_seconds) + except Exception: + return False + if max_age <= 0 or declared_at_epoch is None: + return False + return (now - int(declared_at_epoch)) >= max_age + + +def decide(data: dict, *, source: str, now: float) -> GateResult: + """Broadcast decision for a normalized WFIGS event. + + Parameters + ---------- + data: + Canonical WFIGS dict (see module docstring for schema). + source: + Adapter source name, e.g. "wfigs". + now: + Current epoch (from clock.now()) — determinism seam. + + Returns + ------- + GateResult with broadcast/lifecycle/data_patch/commit populated. The + handler still performs the inline ``fires`` INSERT/UPDATE and renders the + wire; ``data_patch`` supplies the stamps + render hints, ``commit`` arms + the ``last_broadcast_*`` columns on confirmed delivery. + """ + kind = data.get("_kind") + irwin_id = data.get("irwin_id") + + try: + conn = get_db() + except Exception: + logger.exception("fire decide: persistence unavailable") + return GateResult(broadcast=False, lifecycle="suppress", + reason="persistence unavailable") + + # ── Tombstone all-clear ───────────────────────────────────────────────── + if kind == "wfigs_tombstone": + if not irwin_id: + return GateResult(broadcast=False, lifecycle="suppress", + reason="tombstone without irwin_id") + fire_row = conn.execute( + "SELECT incident_name, current_acres, current_contained_pct, " + "last_broadcast_at, county, state, lat, lon " + "FROM fires WHERE irwin_id = ?", (irwin_id,) + ).fetchone() + # Only fires that previously reached the mesh get a closure message. + if fire_row is None or fire_row["last_broadcast_at"] is None: + return GateResult(broadcast=False, lifecycle="suppress", + reason="tombstone for never-broadcast fire") + acres = fire_row["current_acres"] + contained_pct = fire_row["current_contained_pct"] + patch: dict = { + "category": "wildfire_closed", + "_severity_override": "priority", + # legacy _attach_commit_handles sets _cooldown_suffix=irwin_id then + # the handler overwrites _dedup_suffix with "closed". + "_cooldown_suffix": irwin_id, + "_dedup_suffix": "closed", + # render hints for the all-clear formatter (from the fire ROW, not + # the tombstone payload): name/size/loc — no geocoder_city/landclass + # so the anchor falls to resolve_anchor -> county exactly as legacy. + "_allclear": True, + "incident_name": fire_row["incident_name"], + "acres": acres, + "contained_pct": contained_pct, + "lat": fire_row["lat"], + "lon": fire_row["lon"], + "county": fire_row["county"], + "state": fire_row["state"], + } + return GateResult( + broadcast=True, lifecycle="closed", + reason=f"all-clear irwin={irwin_id}", + data_patch=patch, + commit=_make_commit(irwin_id, acres, contained_pct), + ) + + # ── Perimeter / unknown never broadcast ───────────────────────────────── + if kind != "wfigs_incident": + return GateResult(broadcast=False, lifecycle="suppress", + reason=f"non-broadcast kind {kind}") + + # ── Active incident state machine ─────────────────────────────────────── + acres = data.get("acres") + contained_pct = data.get("contained_pct") + declared_at_epoch = data.get("declared_at_epoch") + + row = conn.execute( + "SELECT current_acres, current_contained_pct, last_broadcast_at, " + "last_broadcast_acres, last_broadcast_contained " + "FROM fires WHERE irwin_id = ?", (irwin_id,)).fetchone() + + # Stamps every active broadcast (New + Update) carries. category override + # is added only for New (Update keeps the envelope-derived wildfire_incident). + def _broadcast_patch(*, is_update: bool, + last_bcast_acres=None, last_bcast_contained=None, + new_category: bool) -> dict: + p = { + "_severity_override": "priority", + "_dedup_suffix": f"{acres}|{contained_pct}", + "_cooldown_suffix": irwin_id, + "is_update": is_update, + "last_bcast_acres": last_bcast_acres, + "last_bcast_contained": last_bcast_contained, + } + if new_category: + p["category"] = "wildfire_declared" + return p + + # (i) row missing & (ii) row exists but never broadcast -> "New". + if row is None or row["last_broadcast_at"] is None: + # Age-gate suppresses only the "New" BROADCAST; the handler still writes + # current_* unconditionally. Suppress carries NO stamps (legacy returns + # None before setting category/severity) and does NOT trigger cleanup. + if _fire_too_old_to_announce(declared_at_epoch, now): + return GateResult(broadcast=False, lifecycle="suppress", + reason="new fire too old to announce") + return GateResult( + broadcast=True, lifecycle="new", + reason="first sighting", + data_patch=_broadcast_patch(is_update=False, new_category=True), + commit=_make_commit(irwin_id, acres, contained_pct), + ) + + # (iii) row exists AND already broadcast -> forward-only growth + cooldown. + last_bcast_at = row["last_broadcast_at"] + last_bcast_acres = row["last_broadcast_acres"] + last_bcast_contained = row["last_broadcast_contained"] + + changed_acres = ( + bool(adapter_config.wfigs.broadcast_on_acres) + and acres is not None + and (last_bcast_acres is None or acres > last_bcast_acres) + ) + changed_contained = ( + bool(adapter_config.wfigs.broadcast_on_contained) + and contained_pct is not None + and (last_bcast_contained is None or contained_pct > last_bcast_contained) + ) + cooldown_s = int(adapter_config.wfigs.cooldown_seconds) + eight_hours_passed = ( + last_bcast_at is None + or (now - int(last_bcast_at) >= cooldown_s) + ) + + if (changed_acres or changed_contained) and eight_hours_passed: + return GateResult( + broadcast=True, lifecycle="update", + reason="forward-only growth after cooldown", + data_patch=_broadcast_patch( + is_update=True, + last_bcast_acres=last_bcast_acres, + last_bcast_contained=last_bcast_contained, + new_category=False), + commit=_make_commit(irwin_id, acres, contained_pct), + ) + + # No change OR inside cooldown -> suppress; handler runs _cleanup_stale_fires + # for this case only (legacy did the cleanup in the case-(iii) suppress arm). + return GateResult(broadcast=False, lifecycle="cooldown", + reason="no forward change or inside cooldown") + + +def _make_commit(irwin_id: str, acres, contained_pct): + """Build the deferred commit closure: idempotent UPSERT of the + ``fires`` broadcast-state columns. Mirrors + ``wfigs_handler._attach_commit_handles._on_commit`` exactly (minus the + event_log flip, which the handler wraps back on).""" + + def _commit(committed_at: float) -> None: + try: + conn = get_db() + except Exception: + logger.exception( + "fire commit: persistence unavailable; last_broadcast_* not " + "updated for irwin=%s", irwin_id) + return + try: + conn.execute( + "UPDATE fires SET last_broadcast_at=?, " + "first_broadcast_at=COALESCE(first_broadcast_at, ?), " + "last_broadcast_acres=?, last_broadcast_contained=? " + "WHERE irwin_id=?", + (int(committed_at), int(committed_at), acres, contained_pct, + irwin_id), + ) + except Exception: + logger.exception("fire commit: fires UPSERT failed irwin=%s", irwin_id) + + return _commit diff --git a/work/tests/test_fire_refactor.py b/work/tests/test_fire_refactor.py new file mode 100644 index 0000000..c40c0fa --- /dev/null +++ b/work/tests/test_fire_refactor.py @@ -0,0 +1,320 @@ +"""Phase-3b WFIGS wildfire refactor tests. + +Verifies the source-agnostic formatter+decider migration for the wildfire +hazard, mirroring test_hydro_refactor.py / test_quake_refactor.py: + +1. Golden byte-identical: formatters.fire.format() reproduces the legacy + wfigs_handler._render() wire exactly for a New incident, an Update-with- + growth ((+delta) size line), a movement-line case, an anchor-line case, and + the wildfire_closed all-clear. + +2. Gate-sequence parity: an explicit `now`-timeline of WFIGS events driven + through the NEW gating.fire.decide() matches the OLD handle_wfigs + broadcast/suppress behavior AND the data-dict stamps + (category / _severity_override / _dedup_suffix / _cooldown_suffix). + +3. Registration: the three explicit categories the wfigs_handler emits + (wildfire_declared / wildfire_incident / wildfire_closed) resolve to the + fire formatter + decider, and FIRMS categories do NOT. +""" +from __future__ import annotations + +import pytest + +from meshai import central_normalizer as cn +from meshai.central.budget import budget_for +from meshai.central.wfigs_handler import ( + _build_canonical, + _render as _wfigs_render, + handle_wfigs, +) +from meshai.notifications.formatters.fire import format as fire_format +from meshai.notifications.gating.fire import decide as fire_decide +from meshai.persistence import close_thread_connection, init_db +from meshai.persistence import db as persistence_db +from tests.harness.goldens import assert_byte_identical +from tests.test_wfigs_handler import ( + _IRWIN_A, + _make_active_envelope, + _make_tombstone, +) + +_AT = 1_800_000_000.0 # pinned epoch (unused by fire render; kept for parity) + + +@pytest.fixture +def mem_db(monkeypatch, tmp_path): + db_path = str(tmp_path / "fire-refactor-test.sqlite") + monkeypatch.setenv("MESHAI_DB_PATH", db_path) + persistence_db._initialised.clear() + close_thread_connection() + conn = init_db() + try: + from meshai.adapter_config import adapter_config as _ac + _ac.invalidate() + except Exception: + pass + try: + from meshai.central import wfigs_handler as _wh + _wh._last_cleanup = 0 + except Exception: + pass + yield conn + close_thread_connection() + persistence_db._initialised.discard(db_path) + + +class _FakeEvent: + def __init__(self, data, category=None): + self.data = data + self.category = category + + +# ───────────────────────────────────────────────────────────────────────────── +# 1. Golden byte-identical — formatter reproduces _render()/all-clear exactly +# ───────────────────────────────────────────────────────────────────────────── + +class TestFormatterGolden: + """formatters.fire.format() == wfigs_handler._render() for the same inputs.""" + + def _budget(self) -> int: + return budget_for("wfigs") + + def _incident_data(self, **over) -> dict: + d = { + "incident_name": "Cache Peak Fire", + "acres": 1847.0, + "contained_pct": 23, + "fire_cause": "Lightning", + "declared_at_epoch": 1_781_204_400, + "unique_fire_id": "2026-IDSCF-000987", + "geocoder_city": "Burley", # short-circuits anchor (no DB/Photon) + "lat": 42.197, + "lon": -113.710, + "county": "Cassia", + "state": "ID", + "landclass": None, + } + d.update(over) + return d + + def test_new_incident(self, mem_db): + n = self._incident_data() + old = _wfigs_render(n, prefix="New") + new = fire_format(_FakeEvent({**n, "is_update": False}), + now=_AT, budget=self._budget()) + assert_byte_identical(new, old) + assert new.startswith("🔥 Cache Peak Fire — New") + assert "1,847 ac" in new + assert "containment 23%" in new + + def test_update_with_growth_delta(self, mem_db): + n = self._incident_data(acres=3000.0, contained_pct=35) + old = _wfigs_render(n, prefix="Update", + last_bcast_acres=1847.0, last_bcast_contained=23) + new = fire_format( + _FakeEvent({**n, "is_update": True, + "last_bcast_acres": 1847.0, "last_bcast_contained": 23}), + now=_AT, budget=self._budget()) + assert_byte_identical(new, old) + assert new.startswith("🔥 Cache Peak Fire — Update") + assert "3,000 ac (+1,153)" in new # delta line + assert "containment 35%" in new + + def test_movement_line(self, mem_db): + # movement is FIRMS-injected; the formatter must read it from event.data + # exactly as _render(movement=...) does today. + mv = {"direction": "NE", "speed_mph": 1.2} + n = self._incident_data() + old = _wfigs_render(n, prefix="New", movement=mv) + new = fire_format(_FakeEvent({**n, "is_update": False, "movement": mv}), + now=_AT, budget=self._budget()) + assert_byte_identical(new, old) + assert "Moving NE 1.2 mi/h" in new + + def test_anchor_line(self, mem_db): + # No geocoder_city → line 3 is resolved via the shared resolve_anchor + + # the legacy fallback tiers. Both paths hit the same seeded town_anchors + # table, so the wire must stay byte-identical. + n = self._incident_data(geocoder_city=None) + old = _wfigs_render(n, prefix="New") + new = fire_format(_FakeEvent({**n, "is_update": False}), + now=_AT, budget=self._budget()) + assert_byte_identical(new, old) + # line 3 is NOT a movement line and NOT a raw city name + assert "Moving" not in new + + def test_wildfire_closed_all_clear(self, mem_db): + # Drive the real handler to produce the legacy all-clear wire, then + # assert the formatter reproduces it byte-for-byte from event.data. + env = _make_active_envelope(geocoder_city="Burley") + n0 = cn.normalize(env) + data0 = {} + handle_wfigs(n0, env, env["subject"], data=data0, now=1_000_000) + data0["_on_broadcast_committed"](float(1_000_000)) # arm last_broadcast_* + + tomb = _make_tombstone() + data_t = {} + old_wire = handle_wfigs(cn.normalize(tomb), tomb, tomb["subject"], + data=data_t, now=2_000_000) + assert old_wire is not None + assert old_wire.startswith("✅ Cache Peak Fire — contained & closed") + assert data_t["category"] == "wildfire_closed" + + # Reconstruct the canonical fields the decider's data_patch supplies to + # the formatter for the closed wire, and render. + closed_data = { + "category": "wildfire_closed", + "incident_name": "Cache Peak Fire", + "acres": 1847.0, + "contained_pct": 23, + "lat": env["data"]["data"]["latitude"], + "lon": env["data"]["data"]["longitude"], + "county": "Cassia", + "state": "ID", + } + new_wire = fire_format(_FakeEvent(closed_data, category="wildfire_closed"), + now=_AT, budget=budget_for("wfigs")) + assert_byte_identical(new_wire, old_wire) + + +# ───────────────────────────────────────────────────────────────────────────── +# 2. Gate-sequence parity — new decide() vs old handle_wfigs across a lifecycle +# ───────────────────────────────────────────────────────────────────────────── + +class TestGateSequenceParity: + """A full fire lifecycle agrees between decide() and handle_wfigs().""" + + def _decide(self, env, now): + n = cn.normalize(env) + canonical = _build_canonical(n, n["_kind"]) + return fire_decide(canonical, source="wfigs", now=float(now)) + + def _step(self, env, now, *, expect_broadcast, expect_lifecycle): + """Assert decide() and handle_wfigs() agree at one timeline step. + + decide() (called first) only READS state, so it sees the same pre-write + row the handler's internal decide() sees. On broadcast the handler's + legacy (not-cutover) stamps must equal decide()'s data_patch, and we arm + last_broadcast_* via the commit callback (simulating the dispatcher). + """ + n = cn.normalize(env) + gate = self._decide(env, now) + assert gate.broadcast is expect_broadcast, ( + f"decide broadcast {gate.broadcast} != {expect_broadcast} " + f"@ {now} ({expect_lifecycle})") + assert gate.lifecycle == expect_lifecycle, ( + f"decide lifecycle {gate.lifecycle} != {expect_lifecycle} @ {now}") + + data = {} + wire = handle_wfigs(n, env, env["subject"], data=data, now=now) + assert (wire is not None) is expect_broadcast + + if expect_broadcast: + # Handler's stamped keys (legacy path) match decide()'s data_patch. + for k in ("_severity_override", "_dedup_suffix", "_cooldown_suffix"): + assert data.get(k) == gate.data_patch.get(k), ( + f"stamp {k}: handler={data.get(k)!r} " + f"decide={gate.data_patch.get(k)!r}") + if expect_lifecycle == "new": + assert data.get("category") == "wildfire_declared" + assert gate.data_patch.get("category") == "wildfire_declared" + elif expect_lifecycle == "update": + # Update keeps the envelope-derived wildfire_incident: no override. + assert "category" not in data + assert "category" not in gate.data_patch + elif expect_lifecycle == "closed": + assert data.get("category") == "wildfire_closed" + assert gate.data_patch.get("category") == "wildfire_closed" + assert data.get("_severity_override") == "priority" + # Arm last_broadcast_* for the next cooldown check. + data["_on_broadcast_committed"](float(now)) + return wire + + def test_full_lifecycle(self, mem_db): + irwin = _IRWIN_A + base = 1_800_000_000 + disc_ms = (base - 3600) * 1000 # discovered 1h before first sight (fresh) + + def _active(acres, pct, subject_n="a"): + return _make_active_envelope( + irwin_id=irwin, geocoder_city="Burley", + daily_acres=acres, pct_contained=pct, + fire_discovery_dt_ms=disc_ms) + + # [0] first sight → New broadcast + self._step(_active(250.0, 0), base, + expect_broadcast=True, expect_lifecycle="new") + # [1] small growth 1h later (inside 8h cooldown) → suppress + self._step(_active(300.0, 0), base + 3600, + expect_broadcast=False, expect_lifecycle="cooldown") + # [2] growth after cooldown (8h) → Update + self._step(_active(500.0, 0), base + 28800, + expect_broadcast=True, expect_lifecycle="update") + # [3] containment change after another cooldown → Update + self._step(_active(500.0, 40), base + 28800 * 2, + expect_broadcast=True, expect_lifecycle="update") + # [4] tombstone → all-clear (fire was broadcast earlier) + tomb = _make_tombstone(irwin_id=irwin) + self._step(tomb, base + 100000, + expect_broadcast=True, expect_lifecycle="closed") + + def test_dedup_suffix_tracks_state(self, mem_db): + """_dedup_suffix carries the acres|contained that justified the + broadcast (so unchanged re-deliveries dedup but genuine updates pass).""" + base = 1_800_000_000 + disc_ms = (base - 3600) * 1000 + env_new = _make_active_envelope( + irwin_id=_IRWIN_A, geocoder_city="Burley", + daily_acres=250.0, pct_contained=0, fire_discovery_dt_ms=disc_ms) + gate = self._decide(env_new, base) + n = cn.normalize(env_new) + assert gate.data_patch["_dedup_suffix"] == f"{n['acres']}|{n['contained_pct']}" + assert gate.data_patch["_cooldown_suffix"] == _IRWIN_A + + def test_never_broadcast_tombstone_suppressed(self, mem_db): + """A tombstone for a fire that never reached the mesh is silent.""" + tomb = _make_tombstone(irwin_id=_IRWIN_A) + gate = self._decide(tomb, 1_000_000) + assert gate.broadcast is False + assert gate.lifecycle == "suppress" + # Handler agrees: returns None. + out = handle_wfigs(cn.normalize(tomb), tomb, tomb["subject"], + data={}, now=1_000_000) + assert out is None + + def test_perimeter_never_broadcasts(self, mem_db): + from tests.test_wfigs_handler import _make_perimeter + per = _make_perimeter(irwin_id=_IRWIN_A) + gate = self._decide(per, 1_000_000) + assert gate.broadcast is False + assert gate.lifecycle == "suppress" + + +# ───────────────────────────────────────────────────────────────────────────── +# 3. Registration — the three explicit categories resolve; FIRMS does not +# ───────────────────────────────────────────────────────────────────────────── + +class TestRegistration: + @pytest.mark.parametrize( + "cat", ["wildfire_declared", "wildfire_incident", "wildfire_closed"]) + def test_formatter_registered(self, cat): + from meshai.notifications.formatters import get_formatter + assert get_formatter(cat) is fire_format + + @pytest.mark.parametrize( + "cat", ["wildfire_declared", "wildfire_incident", "wildfire_closed"]) + def test_decider_registered(self, cat): + from meshai.notifications.gating import get_decider + assert get_decider(cat) is fire_decide + + @pytest.mark.parametrize( + "cat", ["wildfire_hotspot", "new_ignition", + "unattributed_hotspot_cluster", "wildfire_growth"]) + def test_firms_categories_not_captured(self, cat): + # FIRMS is deferred; its categories must NOT resolve to the fire + # formatter/decider via the "fire" toggle family fallback. + from meshai.notifications.formatters import get_formatter + from meshai.notifications.gating import get_decider + assert get_formatter(cat) is not fire_format + assert get_decider(cat) is not fire_decide diff --git a/work/tests/test_formatter_scaffold.py b/work/tests/test_formatter_scaffold.py index 551e683..e810d30 100644 --- a/work/tests/test_formatter_scaffold.py +++ b/work/tests/test_formatter_scaffold.py @@ -18,7 +18,8 @@ from meshai.notifications.renderers.composer import compose_mesh_message # earthquake_event removed: Phase-1 registers formatters.quake for it. # weather_warning/weather_statement removed: Phase-2 registers formatters.nws. # road_closure/work_zone/road_incident/traffic_congestion removed: Phase-2 registers formatters.incident. - "wildfire_incident", + # wildfire_declared/wildfire_incident/wildfire_closed removed: Phase-3b registers formatters.fire. + "wildfire_hotspot", # FIRMS — still deferred (not migrated) "battery_critical", ]) def test_get_formatter_returns_none_while_registry_empty(category): diff --git a/work/tests/test_shadow_inert.py b/work/tests/test_shadow_inert.py index 2a6400a..cdd6e7b 100644 --- a/work/tests/test_shadow_inert.py +++ b/work/tests/test_shadow_inert.py @@ -142,11 +142,13 @@ class TestShadowInertWhenNoDecider: class TestShadowRenderInertWhenNoFormatter: """MESHAI_SHADOW_CATEGORIES set but no formatter registered → still no-op. - Note: earthquake_event has a formatter in Phase 1+; this class uses - wildfire_incident which remains un-migrated and has no formatter entry. + Note: earthquake_event has a formatter in Phase 1+; Phase-3b migrated the + WFIGS categories (wildfire_declared / wildfire_incident / wildfire_closed), + so this class uses wildfire_hotspot — a FIRMS category that remains + un-migrated (deferred) and has no formatter entry. """ - _CATEGORY = "wildfire_incident" + _CATEGORY = "wildfire_hotspot" def setup_method(self): os.environ["MESHAI_SHADOW_CATEGORIES"] = self._CATEGORY