From ccd816b5997e11d8039062dd8868dace77c4027f Mon Sep 17 00:00:00 2001 From: Matt Johnson Date: Sun, 5 Jul 2026 04:22:01 +0000 Subject: [PATCH] refactor(phase3a): migrate hydro/nwis to formatter+decider (tier-a) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Move nwis wire rendering into notifications/formatters/hydro.py and the threshold-crossing decision into notifications/gating/hydro.py, behind the registry under the real category `stream_flow`. handle_nwis now builds a canonical dict, calls decide(), keeps the gauge_readings INSERT inline (append-only), and branches on is_cutover("stream_flow") exactly like quake_handler β€” legacy _attach_commit path stays byte-identical while the new path bakes in shadow. No cutover. - gauge_readings INSERT stays inline; decider only reads prior state - hydro has no per-event broadcast-state table β†’ GateResult.commit=None; event_log.handled flip stays handler-owned in the cutover wrapper - tier-a: golden byte-identical wire + gate-sequence parity (12 tests) - native env/usgs.py deferred (different category vocab, no threshold_state); non-cutover so store._emit_event won't run it Suite at 34-failure baseline. Co-Authored-By: Claude Opus 4.8 (1M context) --- work/meshai/central/nwis_handler.py | 111 +++--- .../notifications/formatters/__init__.py | 8 + work/meshai/notifications/formatters/hydro.py | 95 +++++ work/meshai/notifications/gating/__init__.py | 9 + work/meshai/notifications/gating/hydro.py | 152 ++++++++ work/tests/test_hydro_refactor.py | 341 ++++++++++++++++++ 6 files changed, 673 insertions(+), 43 deletions(-) create mode 100644 work/meshai/notifications/formatters/hydro.py create mode 100644 work/meshai/notifications/gating/hydro.py create mode 100644 work/tests/test_hydro_refactor.py diff --git a/work/meshai/central/nwis_handler.py b/work/meshai/central/nwis_handler.py index 6223748..6f52f39 100644 --- a/work/meshai/central/nwis_handler.py +++ b/work/meshai/central/nwis_handler.py @@ -58,7 +58,6 @@ from datetime import datetime from typing import Any, Optional from meshai.central.idaho_gauge_sites import ( - THRESHOLD_RANK, compute_threshold_state, lookup_site, normalize_site_id, @@ -165,6 +164,22 @@ def handle_nwis(envelope: dict, subject: str, lat = d.get("latitude") if isinstance(d.get("latitude"), (int, float)) else site_meta.get("lat") lon = d.get("longitude") if isinstance(d.get("longitude"), (int, float)) else site_meta.get("lon") + # Build the canonical dict the decider + formatter read. parameter_code is + # carried so the decider can perform the 00060 discharge back-look; it is + # not part of the formatter's wire schema. + canonical: dict = { + "site_id": site_id, + "gauge_name": site_meta["gauge_name"], + "stage_ft": stage_ft, + "flow_cfs": flow_cfs, + "unit": unit, + "threshold_state": threshold_state, + "reading_time": reading_time, + "lat": lat, + "lon": lon, + "parameter_code": pc, + } + # Always log the envelope to event_log. Initial handled=0; commit # callback flips to 1 if we actually broadcast. log_id = _log_event_returning_id( @@ -173,31 +188,23 @@ def handle_nwis(envelope: dict, subject: str, subject=subject, handled=0, table_name="gauge_readings", table_pk=site_id) - # SELECT most recent prior reading (for this site, any parameter) to - # detect upward threshold crossing. Use threshold_state column directly. - prior = conn.execute( - "SELECT threshold_state FROM gauge_readings " - "WHERE site_id=? AND reading_time < ? " - "ORDER BY reading_time DESC LIMIT 1", - (site_id, reading_time), - ).fetchone() - prior_state = prior["threshold_state"] if prior else "normal" + # Delegate the threshold-crossing decision (prior-reading SELECT, 00060 + # stage back-look, THRESHOLD_RANK upward-crossing check, broadcast_on_recede + # toggle) to the gating module. It reads gauge_readings for prior state + # BEFORE the inline INSERT below, preserving the original ordering. It + # returns the resolved (back-looked) threshold_state + stage_ft in + # data_patch so the INSERT + render use the same values on every path. + from meshai.notifications.gating.hydro import decide as _gate_decide + gate = _gate_decide(canonical, source="nwis", now=float(now)) + threshold_state = gate.data_patch.get("threshold_state", threshold_state) + stage_ft = gate.data_patch.get("stage_ft", stage_ft) + canonical["threshold_state"] = threshold_state + canonical["stage_ft"] = stage_ft - # If this envelope is a 00060 (discharge) reading, look back for the - # latest 00065 stage reading at this site so the wire string can carry - # both. The threshold_state of THIS row inherits from that prior stage - # reading (discharge alone doesn't define a threshold band). - if pc == "00060": - last_stage = conn.execute( - "SELECT reading_value, threshold_state FROM gauge_readings " - "WHERE site_id=? AND reading_unit='ft' " - "ORDER BY reading_time DESC LIMIT 1", - (site_id,)).fetchone() - if last_stage: - stage_ft = last_stage["reading_value"] - threshold_state = last_stage["threshold_state"] or "normal" - - # INSERT the new reading row. Always persist (time-series semantics). + # INSERT the new reading row INLINE (append-only time-series). Always + # persist, regardless of the broadcast decision, and AFTER the decider's + # reads so they never see the current row. Per the refactor plan this + # write stays handler-owned; the decider only READS gauge_readings. conn.execute( "INSERT INTO gauge_readings(site_id, gauge_name, reading_value, " "reading_unit, threshold_state, flow_cfs, reading_time, lat, lon) " @@ -206,30 +213,48 @@ def handle_nwis(envelope: dict, subject: str, threshold_state, flow_cfs, reading_time, lat, lon), ) - # Upward-crossing check. - try: - prior_rank = THRESHOLD_RANK.index(prior_state) - except ValueError: - prior_rank = 0 # unknown prior -> treat as normal - try: - cur_rank = THRESHOLD_RANK.index(threshold_state) - except ValueError: - cur_rank = 0 - - if cur_rank == prior_rank: - # Unchanged band -- no broadcast. - return None - if cur_rank < prior_rank and not bool(adapter_config.usgs_nwis.broadcast_on_recede): - # Receding without the recede toggle -- silent. + if not gate.broadcast: return None - wire = _render(gauge_name=site_meta["gauge_name"], + # Cutover gate: mirror quake_handler. When "stream_flow" is cut over, write + # gate.data_patch and wrap gate.commit so it also flips event_log.handled. + # Hydro has no per-event broadcast-state table, so gate.commit is None and + # the wrapper carries only the event_log flip. Otherwise old-style + # _attach_commit keeps the live broadcast byte-for-byte identical while the + # new formatter+decider bake in shadow. + from meshai.notifications.cutover import is_cutover + if isinstance(data, dict): + data.update(canonical) + if is_cutover("stream_flow"): + data.update(gate.data_patch) + data["_broadcast_audit"] = {"table": "gauge_readings", "pk": site_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: + c = get_db() + c.execute("UPDATE event_log SET handled=1 WHERE id=?", + (int(_log_row_id),)) + except Exception: + logger.exception("nwis commit: event_log update failed") + + data["_on_broadcast_committed"] = _on_commit + else: + _attach_commit(data, site_id=site_id, event_log_row_id=log_id) + + # Return _render() wire for backward-compat (existing call-sites + tests). + # At dispatch time compose_mesh_message() uses the registered formatter + # (formatters/hydro.py) on cutover, re-rendering byte-identically from data. + return _render(gauge_name=site_meta["gauge_name"], threshold_state=threshold_state, stage_ft=stage_ft, flow_cfs=flow_cfs, unit=unit if pc == "00065" else "ft", lat=lat, lon=lon) - _attach_commit(data, site_id=site_id, event_log_row_id=log_id) - return wire # ---- renderer ------------------------------------------------------------ diff --git a/work/meshai/notifications/formatters/__init__.py b/work/meshai/notifications/formatters/__init__.py index 051ef66..c8c3cd6 100644 --- a/work/meshai/notifications/formatters/__init__.py +++ b/work/meshai/notifications/formatters/__init__.py @@ -86,3 +86,11 @@ register("work_zone", _incident_fmt_mod.format) register("road_incident", _incident_fmt_mod.format) register("road_closure", _incident_fmt_mod.format) register("traffic_congestion", _incident_fmt_mod.format) + +# Phase-3: USGS NWIS stream-gauge hydro. The Central nwis path maps every +# `central.hydro.*` envelope to the flat category `stream_flow` (see +# central.consumer.map_category / category_from_subject), so that is the real +# registry key. Native env/usgs.py categories (stream_flood_warning / +# 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) diff --git a/work/meshai/notifications/formatters/hydro.py b/work/meshai/notifications/formatters/hydro.py new file mode 100644 index 0000000..a7e6214 --- /dev/null +++ b/work/meshai/notifications/formatters/hydro.py @@ -0,0 +1,95 @@ +"""USGS NWIS stream-gauge (hydro) formatter β€” Phase-3 migration. + +Mirrors central.nwis_handler._render EXACTLY (tier-a, byte-identical). Reads +the canonical schema the Central path writes into event.data on broadcast: + + site_id, gauge_name, stage_ft, flow_cfs, unit, threshold_state, + reading_time, lat, lon + +Wire format (single line, MEDIUM): + 🌊 New: {gauge_name}: {label} {stage_ft:.1f} ft, flow {flow_cfs:,} cfs, @ lat,lon + +Where {label} maps threshold_state via _LABEL: + action -> "action stage" + flood_minor -> "minor flooding" + flood_moderate -> "moderate flooding" + flood_major -> "major flooding" + +Segments: + * stage segment: "{label} {stage_ft:.1f} ft" when stage_ft is numeric, else + the bare label. The "ft" unit is hard-coded (matches _render, which + ignores its `unit` parameter for the stage segment). + * flow segment: ", flow {int(round(flow_cfs)):,} cfs" only when flow_cfs is + numeric (companion 00060 discharge reading present). + * coords segment: ", @ {lat:.3f},{lon:.3f}" only when both coords numeric. + +Time contract: `now` is accepted but NOT used for rendering (structural seam; +never read the wall clock here). `budget` is injected β€” the caller supplies +budget_for("usgs_nwis"). fit_to_budget is applied last; for the short hydro +wire it is a no-op under a 140-char budget, so output stays byte-identical to +_render (which did not budget-fit). +""" +from __future__ import annotations + +from typing import TYPE_CHECKING + +# Import directly from the canonical implementation to avoid the circular +# import chain: central.budget β†’ formatters.__init__ β†’ formatters.hydro. +from meshai.notifications.formatters._budget import fit_to_budget + +if TYPE_CHECKING: + from meshai.notifications.events import Event + + +# Human-readable label per threshold_state β€” mirrors nwis_handler._LABEL. +_LABEL = { + "action": "action stage", + "flood_minor": "minor flooding", + "flood_moderate": "moderate flooding", + "flood_major": "major flooding", +} + + +def format(event: "Event", *, now: float, budget: int) -> str: + """Render the hydro 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("usgs_nwis")). + + Returns: + UTF-8 string fitting within *budget* characters. + """ + d = event.data or {} + + gauge_name = d.get("gauge_name") + threshold_state = d.get("threshold_state") + stage_ft = d.get("stage_ft") + flow_cfs = d.get("flow_cfs") + lat = d.get("lat") + lon = d.get("lon") + # `unit` is present in the canonical dict but intentionally unused here β€” + # _render ignores its unit parameter and hard-codes "ft" in the stage + # segment; mirror that exactly for byte-identity. + + label = _LABEL.get(threshold_state, threshold_state) + + # Stage segment. + if isinstance(stage_ft, (int, float)): + stage_seg = f"{label} {stage_ft:.1f} ft" + else: + stage_seg = label + + # Optional flow segment (companion 00060 discharge). + flow_seg = "" + if isinstance(flow_cfs, (int, float)): + flow_seg = f", flow {int(round(flow_cfs)):,} cfs" + + # Optional coords segment. + coords = "" + if isinstance(lat, (int, float)) and isinstance(lon, (int, float)): + coords = f", @ {lat:.3f},{lon:.3f}" + + msg = f"🌊 New: {gauge_name}: {stage_seg}{flow_seg}{coords}" + return fit_to_budget(msg, budget) diff --git a/work/meshai/notifications/gating/__init__.py b/work/meshai/notifications/gating/__init__.py index 502c689..3ac485d 100644 --- a/work/meshai/notifications/gating/__init__.py +++ b/work/meshai/notifications/gating/__init__.py @@ -77,3 +77,12 @@ register("work_zone", _incident_gate_mod.decide) register("road_incident", _incident_gate_mod.decide) register("road_closure", _incident_gate_mod.decide) register("traffic_congestion", _incident_gate_mod.decide) + +# Phase-3: USGS NWIS stream-gauge hydro. Registered under `stream_flow` β€” the +# flat category the Central nwis path produces for every `central.hydro.*` +# envelope (map_category "hydro." -> "stream_flow"). Native usgs categories +# (stream_flood_warning / stream_high_water, emitted only by env/usgs.py) are a +# deferred follow-up: env/usgs.py is NOT migrated this phase and, since hydro is +# 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) diff --git a/work/meshai/notifications/gating/hydro.py b/work/meshai/notifications/gating/hydro.py new file mode 100644 index 0000000..6ac578b --- /dev/null +++ b/work/meshai/notifications/gating/hydro.py @@ -0,0 +1,152 @@ +"""USGS NWIS stream-gauge (hydro) gating decider β€” Phase-3 migration. + +Moves the threshold-crossing decision out of central.nwis_handler.handle_nwis +(the inline prior-reading SELECT, the 00060/00065 stage back-look, the +THRESHOLD_RANK upward-crossing check, and the broadcast_on_recede toggle). + +Broadcast rule (verbatim from the old handler): + Compare the current reading's threshold_state to the most recent PRIOR + reading (strictly earlier reading_time, any parameter) for the same site, + on the ranked scale {normal < action < flood_minor < flood_moderate < + flood_major}. + * cur_rank > prior_rank β†’ upward crossing β†’ broadcast + * cur_rank == prior_rank β†’ unchanged band β†’ suppress + * cur_rank < prior_rank β†’ receding β†’ suppress, UNLESS + adapter_config.usgs_nwis.broadcast_on_recede is set β†’ broadcast + +00060 (discharge) back-look: + A discharge-only envelope carries no stage β†’ its threshold band is + inherited from the latest 00065 (gage-height, reading_unit='ft') reading at + the site. The decider performs that look-back and returns the resolved + (threshold_state, stage_ft) in data_patch so the handler can INSERT the + gauge_readings row and render the wire with the same values. + +IMPORTANT β€” division of labour with the handler: + * The append-only gauge_readings INSERT stays INLINE in the Central handler + (unconditional time-series write). This decider only READS gauge_readings + to determine prior state; it NEVER writes it. + * The decider MUST run BEFORE the handler's INSERT (the old handler did its + prior-SELECT and 00060 back-look before inserting the new row), so the + reads never see the current reading. The prior-SELECT additionally + filters `reading_time < current` for defence-in-depth. + * Hydro has no per-event broadcast-state table (unlike quake_events), so + the decider owns NO deferred state write β†’ commit is None. The + event_log.handled flip stays handler-owned (mirrors the old + _attach_commit, wrapped in the cutover branch like quake_handler). + +data_patch keys (populated on EVERY call, broadcast or suppress, so the +handler's unconditional INSERT + render use the back-looked values): + threshold_state : str β€” resolved band (post-00060 back-look) + stage_ft : float|None β€” resolved stage (post-00060 back-look) +""" +from __future__ import annotations + +import logging +from typing import Optional + +from meshai.adapter_config import adapter_config +from meshai.central.idaho_gauge_sites import THRESHOLD_RANK +from meshai.notifications.gating.base import GateResult +from meshai.persistence import get_db + +logger = logging.getLogger(__name__) + + +def _rank(state: Optional[str]) -> int: + """THRESHOLD_RANK index for *state*, treating unknown as 0 (normal).""" + try: + return THRESHOLD_RANK.index(state) + except ValueError: + return 0 + + +def decide(data: dict, *, source: str, now: float) -> GateResult: + """Upward-threshold-crossing decision for USGS NWIS stream-gauge readings. + + Parameters + ---------- + data: + Canonical Event.data dict. Consumed keys: + site_id, reading_time, parameter_code, threshold_state, stage_ft + (gauge_name / flow_cfs / lat / lon are carried through by the handler + for the formatter but are not needed for the gate decision.) + source: + Adapter source name, e.g. "nwis". + now: + Current epoch (from clock.now()) β€” determinism seam, unused here (hydro + gating is state-comparison only, no time window). + + Returns + ------- + GateResult: + broadcast=True lifecycle="new" upward crossing (or recede+toggle) + broadcast=False lifecycle="suppress" unchanged band or receding + data_patch always carries the resolved threshold_state + stage_ft. + commit is always None (no decider-owned state write). + """ + site_id = data.get("site_id") + reading_time = data.get("reading_time") + pc = data.get("parameter_code") + threshold_state = data.get("threshold_state") or "normal" + stage_ft = data.get("stage_ft") + + try: + conn = get_db() + except Exception: + logger.exception("nwis decide: persistence unavailable") + return GateResult( + broadcast=False, lifecycle="suppress", + reason="persistence unavailable", + ) + + # Most recent PRIOR reading (strictly earlier) β€” same as the old handler. + prior = conn.execute( + "SELECT threshold_state FROM gauge_readings " + "WHERE site_id=? AND reading_time < ? " + "ORDER BY reading_time DESC LIMIT 1", + (site_id, reading_time), + ).fetchone() + prior_state = prior["threshold_state"] if prior else "normal" + + # 00060 (discharge) back-look: inherit stage + band from the latest 00065 + # (gage-height) reading at this site. Discharge alone has no threshold band. + if pc == "00060": + last_stage = conn.execute( + "SELECT reading_value, threshold_state FROM gauge_readings " + "WHERE site_id=? AND reading_unit='ft' " + "ORDER BY reading_time DESC LIMIT 1", + (site_id,), + ).fetchone() + if last_stage: + stage_ft = last_stage["reading_value"] + threshold_state = last_stage["threshold_state"] or "normal" + + # Resolved values returned so the handler's inline INSERT + render match. + patch: dict = {"threshold_state": threshold_state, "stage_ft": stage_ft} + + prior_rank = _rank(prior_state) + cur_rank = _rank(threshold_state) + + # Unchanged band β€” no broadcast. + if cur_rank == prior_rank: + return GateResult( + broadcast=False, lifecycle="suppress", + reason=f"unchanged band {threshold_state}", + data_patch=patch, commit=None, + ) + + # Receding without the recede toggle β€” silent. + if cur_rank < prior_rank and not bool( + adapter_config.usgs_nwis.broadcast_on_recede): + return GateResult( + broadcast=False, lifecycle="suppress", + reason=f"receding {prior_state}->{threshold_state}", + data_patch=patch, commit=None, + ) + + # Upward crossing (or recede with the toggle enabled) β€” broadcast. + return GateResult( + broadcast=True, lifecycle="new", + reason=f"crossing {prior_state}->{threshold_state}", + data_patch=patch, commit=None, + ) diff --git a/work/tests/test_hydro_refactor.py b/work/tests/test_hydro_refactor.py new file mode 100644 index 0000000..0d53cbe --- /dev/null +++ b/work/tests/test_hydro_refactor.py @@ -0,0 +1,341 @@ +"""Phase-3 hydro (USGS NWIS) refactor tests. + +Verifies the source-agnostic formatter+decider migration for the stream-gauge +hazard, mirroring test_quake_refactor.py: + +1. Golden byte-identical: formatters.hydro.format() reproduces the old + nwis_handler._render() wire exactly, for a stage-only crossing, a paired + flow(00060)+stage(00065) reading, and every threshold label. + +2. Gate-sequence parity: an explicit `now`-timeline of readings driven through + the NEW gating.hydro.decide() matches the OLD handle_nwis broadcast/suppress + behavior (upward crossing broadcasts; same-rank + receding suppress unless + broadcast_on_recede). + +The real registry / cutover key is "stream_flow" β€” the flat category the +Central nwis path produces for every central.hydro.* envelope. +""" +from __future__ import annotations + +import pytest + +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, run_gate_sequence + +_AT = 1_783_200_000.0 # pinned epoch (unused by hydro render/gate, kept for parity) + + +# ── DB fixture (same shape as test_nwis_handler.py) ────────────────────────── + +@pytest.fixture +def mem_db(monkeypatch, tmp_path): + db_path = str(tmp_path / "hydro-refactor-test.sqlite") + monkeypatch.setenv("MESHAI_DB_PATH", db_path) + persistence_db._initialised.clear() + close_thread_connection() + conn = init_db() + yield conn + close_thread_connection() + persistence_db._initialised.discard(db_path) + + +def _make_fake_event(data: dict): + class _FakeEvent: + pass + e = _FakeEvent() + e.data = data + return e + + +# ───────────────────────────────────────────────────────────────────────────── +# 1. Golden byte-identical β€” formatter reproduces _render() exactly +# ───────────────────────────────────────────────────────────────────────────── + +class TestFormatterGolden: + """formatters.hydro.format() == nwis_handler._render() for the same inputs.""" + + def _render_old(self, **kw): + from meshai.central.nwis_handler import _render + return _render(**kw) + + def _fmt_new(self, canonical: dict) -> str: + from meshai.notifications.formatters.hydro import format as hfmt + return hfmt(_make_fake_event(canonical), now=_AT, budget=140) + + def test_stage_only_crossing_action(self): + """Stage-only (00065) reading at action stage, no companion flow.""" + canonical = { + "gauge_name": "Snake River at Heise", + "threshold_state": "action", + "stage_ft": 12.5, + "flow_cfs": None, + "unit": "ft", + "lat": 43.612, + "lon": -111.654, + } + old = self._render_old( + gauge_name="Snake River at Heise", threshold_state="action", + stage_ft=12.5, flow_cfs=None, unit="ft", lat=43.612, lon=-111.654, + ) + new = self._fmt_new(canonical) + assert_byte_identical(new, old) + assert new == "🌊 New: Snake River at Heise: action stage 12.5 ft, @ 43.612,-111.654" + + def test_paired_flow_and_stage(self): + """00060 discharge back-looked onto a 00065 stage: flow segment present.""" + canonical = { + "gauge_name": "Boise River", + "threshold_state": "flood_minor", + "stage_ft": 14.5, + "flow_cfs": 8400, + "unit": "ft", + "lat": 43.600, + "lon": -116.200, + } + old = self._render_old( + gauge_name="Boise River", threshold_state="flood_minor", + stage_ft=14.5, flow_cfs=8400, unit="ft", lat=43.600, lon=-116.200, + ) + new = self._fmt_new(canonical) + assert_byte_identical(new, old) + assert "flow 8,400 cfs" in new + assert "minor flooding 14.5 ft" in new + + @pytest.mark.parametrize( + "state,label", + [ + ("action", "action stage"), + ("flood_minor", "minor flooding"), + ("flood_moderate", "moderate flooding"), + ("flood_major", "major flooding"), + ], + ) + def test_every_threshold_label(self, state, label): + """Each threshold_state maps to the correct label β€” byte-identical to _render.""" + canonical = { + "gauge_name": "Test Gauge", + "threshold_state": state, + "stage_ft": 20.0, + "flow_cfs": None, + "unit": "ft", + "lat": 44.0, + "lon": -114.0, + } + old = self._render_old( + gauge_name="Test Gauge", threshold_state=state, stage_ft=20.0, + flow_cfs=None, unit="ft", lat=44.0, lon=-114.0, + ) + new = self._fmt_new(canonical) + assert_byte_identical(new, old) + assert f"{label} 20.0 ft" in new + + def test_missing_coords_drops_at_tail(self): + """No coords β†’ no @ segment (byte-identical to _render).""" + canonical = { + "gauge_name": "No Coords Gauge", + "threshold_state": "action", + "stage_ft": 10.0, + "flow_cfs": None, + "unit": "ft", + "lat": None, + "lon": None, + } + old = self._render_old( + gauge_name="No Coords Gauge", threshold_state="action", + stage_ft=10.0, flow_cfs=None, unit="ft", lat=None, lon=None, + ) + new = self._fmt_new(canonical) + assert_byte_identical(new, old) + assert "@" not in new + + +# ───────────────────────────────────────────────────────────────────────────── +# 2. Gate-sequence parity β€” old handle_nwis vs new gating.hydro.decide() +# ───────────────────────────────────────────────────────────────────────────── + +def _nwis_env(*, site_id="USGS-13186000", parameter_code="00065", value=13.0, + unit="ft", time_iso="2026-06-05T15:00:00Z", + lat=43.612, lon=-111.654, envelope_id=None): + envelope_id = envelope_id or f"nwis_{site_id}_{time_iso}" + return { + "id": envelope_id, + "subject": f"central.hydro.{parameter_code}.usgs.{site_id}.us.id", + "data": { + "id": envelope_id, "adapter": "nwis", + "category": f"hydro.{parameter_code}", "severity": 0, + "geo": {"centroid": [lon, lat], "primary_region": "US-ID"}, + "data": { + "id": envelope_id, + "monitoring_location_id": site_id, + "parameter_code": parameter_code, + "time": time_iso, + "value": value, + "unit_of_measure": unit, + "latitude": lat, "longitude": lon, + }, + }, + } + + +def _parse_iso_epoch(s): + from datetime import datetime + return int(datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp()) + + +class TestGateSequenceParity: + """New decide() decisions match old handle_nwis broadcast/suppress.""" + + @pytest.fixture(autouse=True) + def _db(self, mem_db): + self.db = mem_db + + def _old_gate(self, fixture, *, now): + """OLD path: handle_nwis returning non-None = broadcast. + + handle_nwis owns the append-only gauge_readings INSERT, so replaying + through it advances the persisted time-series exactly as production + would β€” the decider (below) then reads that same state. + """ + from meshai.central.nwis_handler import handle_nwis + env = fixture["envelope"] + wire = handle_nwis(env, env["subject"], data={}, now=int(now)) + return wire is not None + + def _new_gate(self, fixture, *, now): + """NEW path: build canonical (as the handler does) then decide(). + + We do NOT insert here β€” the old-gate replay already advances + gauge_readings; the decider only READS prior state. This mirrors the + production ordering where decide() runs before the inline INSERT. + """ + from meshai.notifications.gating.hydro import decide + from meshai.central.idaho_gauge_sites import ( + compute_threshold_state, lookup_site, normalize_site_id, + ) + env = fixture["envelope"] + d = env["data"]["data"] + raw_site = d.get("monitoring_location_id") + site_id = normalize_site_id(raw_site) + site_meta = lookup_site(raw_site) + pc = d.get("parameter_code") + value = float(d.get("value")) + reading_time = _parse_iso_epoch(d.get("time")) + stage_ft = value if pc == "00065" else None + flow_cfs = value if pc == "00060" else None + threshold_state = "normal" + if pc == "00065": + threshold_state = compute_threshold_state(stage_ft, site_meta) + canonical = { + "site_id": site_id, + "gauge_name": site_meta["gauge_name"], + "stage_ft": stage_ft, + "flow_cfs": flow_cfs, + "unit": d.get("unit_of_measure"), + "threshold_state": threshold_state, + "reading_time": reading_time, + "lat": d.get("latitude"), + "lon": d.get("longitude"), + "parameter_code": pc, + } + return decide(canonical, source="nwis", now=float(now)) + + def test_gate_sequence_matches(self): + """Timeline of Heise readings: old and new gates agree on every step. + + Heise (USGS-13186000): action=12.0ft. + [0] 8.0 ft normal (first reading, no prior) β†’ suppress + [1] 12.5 ft action (normal β†’ action upward) β†’ broadcast + [2] 12.8 ft action (action β†’ action same rank) β†’ suppress + [3] 14.5 ft f_minor (action β†’ flood_minor upward) β†’ broadcast + [4] 8.0 ft normal (flood_minor β†’ normal receding) β†’ suppress + """ + base = _parse_iso_epoch("2026-06-05T10:00:00Z") + specs = [ + (8.0, "2026-06-05T10:00:00Z", "s0"), + (12.5, "2026-06-05T10:15:00Z", "s1"), + (12.8, "2026-06-05T10:30:00Z", "s2"), + (14.5, "2026-06-05T10:45:00Z", "s3"), + (8.0, "2026-06-05T11:00:00Z", "s4"), + ] + ordered = [ + {"envelope": _nwis_env(value=v, time_iso=t, envelope_id=eid)} + for (v, t, eid) in specs + ] + timeline = [float(base + i * 900) for i in range(len(specs))] + + results = run_gate_sequence(self._old_gate, self._new_gate, ordered, + timeline=timeline) + mismatches = [r for r in results if not r["match"]] + assert not mismatches, ( + "Gate sequence mismatch old handle_nwis vs new decide():\n" + + "\n".join( + f" step {r['fixture_n']}: old={r['old_broadcast']} " + f"new={r['new_broadcast']} diffs={r['diffs']}" + for r in mismatches + ) + ) + assert results[0]["old_broadcast"] is False, "normal first reading suppressed" + assert results[1]["old_broadcast"] is True, "normalβ†’action broadcasts" + assert results[2]["old_broadcast"] is False, "actionβ†’action suppressed" + assert results[3]["old_broadcast"] is True, "actionβ†’flood_minor broadcasts" + assert results[4]["old_broadcast"] is False, "receding suppressed (no toggle)" + + def test_00060_backlook_inherits_stage_band(self, mem_db): + """A 00060 discharge reading inherits the last 00065 stage band. + + Seed an action-stage 00065 reading, then feed a 00060 discharge: the + decider's back-look must resolve threshold_state=action + the prior + stage_ft, and (same rank as the seeded action) suppress the discharge. + """ + from meshai.notifications.gating.hydro import decide + # Seed a 00065 stage reading at action via the old handler (writes row). + env_stage = _nwis_env(parameter_code="00065", value=12.5, + time_iso="2026-06-05T10:00:00Z", envelope_id="seed") + self._old_gate({"envelope": env_stage}, now=1_000_000) + + # Now decide on a 00060 discharge β€” should back-look the action band. + env_flow = _nwis_env(parameter_code="00060", value=8400, unit="ft^3/s", + time_iso="2026-06-05T10:05:00Z", envelope_id="q") + gate = self._new_gate({"envelope": env_flow}, now=1_000_300) + assert gate.data_patch["threshold_state"] == "action" + assert gate.data_patch["stage_ft"] == 12.5 + # action β†’ action (same rank) β†’ suppress + assert gate.broadcast is False + + def test_recede_toggle_enables_broadcast(self, mem_db): + """With broadcast_on_recede set, a receding crossing broadcasts.""" + from meshai.adapter_config._accessor import set_runtime_override, _overrides + from meshai.notifications.gating.hydro import decide + # Seed an action reading. + env_high = _nwis_env(parameter_code="00065", value=12.5, + time_iso="2026-06-05T10:00:00Z", envelope_id="hi") + self._old_gate({"envelope": env_high}, now=1_000_000) + + # Force the recede toggle on for the decision only (runtime override, + # since adapter_config accessors are read-only). + set_runtime_override("usgs_nwis", "broadcast_on_recede", True) + try: + env_low = _nwis_env(parameter_code="00065", value=8.0, + time_iso="2026-06-05T11:00:00Z", envelope_id="lo") + gate = self._new_gate({"envelope": env_low}, now=1_003_600) + finally: + _overrides.pop(("usgs_nwis", "broadcast_on_recede"), None) + assert gate.broadcast is True, "receding must broadcast when toggle is on" + assert gate.data_patch["threshold_state"] == "normal" + + +# ───────────────────────────────────────────────────────────────────────────── +# 3. Registration β€” stream_flow resolves to hydro formatter + decider +# ───────────────────────────────────────────────────────────────────────────── + +class TestRegistration: + def test_stream_flow_formatter_registered(self): + from meshai.notifications.formatters import get_formatter + from meshai.notifications.formatters.hydro import format as hfmt + assert get_formatter("stream_flow") is hfmt + + def test_stream_flow_decider_registered(self): + from meshai.notifications.gating import get_decider + from meshai.notifications.gating.hydro import decide + assert get_decider("stream_flow") is decide