mirror of
https://github.com/zvx-echo6/meshai.git
synced 2026-08-26 17:31:34 +00:00
malice
f3990841f8Move 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: Matt Johnson <mj@k7zvx.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
341 lines
15 KiB
Python
341 lines
15 KiB
Python
"""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
|