matt/meshai
1
0
Fork 0
mirror of https://github.com/zvx-echo6/meshai.git synced 2026-08-26 17:31:34 +00:00
Code Issues Projects Releases Packages Wiki Activity

refactor(phase3a): migrate hydro/nwis to formatter+decider (tier-a)

Browse source 
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) <noreply@anthropic.com>
This commit is contained in:
Matt Johnson 2026-07-05 04:22:01 +00:00
commit ccd816b599
6 changed files with 674 additions and 44 deletions
Download patch file Download diff file Expand all files Collapse all files

111
work/meshai/central/nwis_handler.py
View file

@ -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 ------------------------------------------------------------

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

@ -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)

95
work/meshai/notifications/formatters/hydro.py Normal file
View file

@ -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)

9
work/meshai/notifications/gating/__init__.py
View file

@ -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)

152
work/meshai/notifications/gating/hydro.py Normal file
View file

@ -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,
)

341
work/tests/test_hydro_refactor.py Normal file
View file

@ -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