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

feat(danger-zones): configurable infra-node hazard correlation + fire age-gate

Browse source 
- danger_zones config section (isolated dataclass + danger_zones.yaml + GUI panel on Notifications page); defaults disabled + dry_run
- DangerZoneCorrelator: distance-based correlation of hazard events vs infra nodes (CLIENT_BASE/ROUTER/ROUTER_LATE), DM delivery, cooldown
- fire age-gate (wfigs max_declare_age_seconds) suppresses stale/closed fires announced as "New"
- tests: correlator + fire-gate boundary; wfigs fixture fix

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ubuntu 2026-06-21 06:33:54 +00:00
commit 95b1a23ef1
14 changed files with 1425 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

352
work/tests/test_danger_zone_correlator.py Normal file
View file

@ -0,0 +1,352 @@
"""Step 2 DangerZoneCorrelator tests (Step 5 verification, correlator part).
The correlator subscribes to the notification EventBus and, for located
hazard events, flags infrastructure nodes inside the hazard's threat radius
(+ buffer) and optionally alerts the operator. Distances are MILES
end-to-end. It delivers via channels.create_channel + MeshDMChannel,
BYPASSING the Dispatcher -- so we stub the CONNECTOR (capture send_message),
NOT a dispatcher (per plan C12).
Most tests use a NON-fire family (avalanche/seismic) so we avoid the fires
DB lookup entirely -- a non-fire hazard is a point at the event lat/lon.
Async delivery is fire-and-forget (asyncio.create_task). Tests that assert a
real send run inside an event loop (pytest.mark.asyncio) and yield once so
the scheduled task runs. dry_run / miss / filter tests assert the provably
send-free paths and don't need the loop.
"""
from __future__ import annotations
import asyncio
import time
import pytest
from meshai.config import Config, DangerZonesConfig, DangerZoneHazardConfig
from meshai.mesh_data_store import UnifiedNode
from meshai.notifications.danger_zone_correlator import DangerZoneCorrelator
from meshai.notifications.events import make_event
# ---------------------------------------------------------------------------
# Test doubles
# ---------------------------------------------------------------------------
class StubConnector:
"""Captures send_message calls (mirrors MeshConnector.send_message)."""
def __init__(self):
self.calls: list[dict] = []
def send_message(self, text=None, destination=None, channel=0, **kw):
self.calls.append(
{"text": text, "destination": destination, "channel": channel})
return True
class FakeDataStore:
"""Minimal stand-in exposing get_nodes_by_roles with the SAME filter
semantics as the real MeshDataStore.get_nodes_by_roles (role membership +
fresh last_heard + non-None position), so freshness/role tests exercise
real filtering rather than a hand-fed list."""
def __init__(self, nodes):
self._nodes = list(nodes)
def get_nodes_by_roles(self, roles, max_age_s):
cutoff = time.time() - max_age_s
return [
n for n in list(self._nodes)
if n.role in roles
and n.latitude is not None
and n.longitude is not None
and n.last_heard
and n.last_heard >= cutoff
]
def _node(*, node_num, role="ROUTER", lat=42.0, lon=-114.0,
last_heard=None, short_name=None):
n = UnifiedNode(node_num=node_num)
n.role = role
n.latitude = lat
n.longitude = lon
n.last_heard = last_heard if last_heard is not None else time.time()
n.node_id_hex = f"!{node_num:08x}"
n.short_name = short_name if short_name is not None else f"N{node_num}"
return n
def _config(*, enabled=True, dry_run=False, delivery_type="mesh_dm",
node_ids=("!aaaa0001",), cooldown_minutes=360,
position_max_age_hours=72, default_buffer_mi=5.0,
**family_overrides):
"""Build a Config whose danger_zones is configured per test.
family_overrides: e.g. avalanche=DangerZoneHazardConfig(...) to tune a
specific family. Unspecified families keep defaults.
"""
dz = DangerZonesConfig(
enabled=enabled,
dry_run=dry_run,
delivery_type=delivery_type,
node_ids=list(node_ids),
cooldown_minutes=cooldown_minutes,
position_max_age_hours=position_max_age_hours,
default_buffer_mi=default_buffer_mi,
monitor_roles=["ROUTER", "ROUTER_LATE", "CLIENT_BASE"],
**family_overrides,
)
cfg = Config()
cfg.danger_zones = dz
return cfg
def _avalanche_event(lat, lon, severity="priority", title="Avalanche Warning"):
"""An avalanche hazard -> family 'avalanche', point hazard (no DB)."""
return make_event(
source="avalanche", category="avalanche_warning", severity=severity,
title=title, summary=title, lat=lat, lon=lon,
)
# ---------------------------------------------------------------------------
# 1. Hit
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_hit_flags_colocated_infra_node():
"""A hazard ON a known infra node (well within buffer) -> flagged; a real
(non-dry_run) send fires exactly once."""
conn = StubConnector()
node = _node(node_num=101, role="ROUTER", lat=42.0, lon=-114.0)
ds = FakeDataStore([node])
cfg = _config(dry_run=False, default_buffer_mi=5.0)
corr = DangerZoneCorrelator(cfg, ds, conn)
corr.handle(_avalanche_event(42.0, -114.0)) # right on the node
await asyncio.sleep(0) # let the create_task'd deliver run
assert len(conn.calls) == 1, f"expected exactly one send, got {conn.calls}"
assert conn.calls[0]["destination"] == "!aaaa0001"
@pytest.mark.asyncio
async def test_hit_within_buffer_edge():
"""A hazard a few miles away but inside default_buffer_mi -> still a hit."""
conn = StubConnector()
# ~3.4 mi north of the hazard (0.05 deg lat ~ 3.45 mi); buffer is 5 mi.
node = _node(node_num=102, lat=42.05, lon=-114.0)
ds = FakeDataStore([node])
cfg = _config(dry_run=False, default_buffer_mi=5.0)
corr = DangerZoneCorrelator(cfg, ds, conn)
corr.handle(_avalanche_event(42.0, -114.0))
await asyncio.sleep(0)
assert len(conn.calls) == 1
# ---------------------------------------------------------------------------
# 2. Miss
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_miss_far_node_not_flagged():
"""Same hazard far beyond radius+buffer -> no flag/send."""
conn = StubConnector()
node = _node(node_num=103, lat=43.0, lon=-114.0) # ~69 mi north
ds = FakeDataStore([node])
cfg = _config(dry_run=False, default_buffer_mi=5.0)
corr = DangerZoneCorrelator(cfg, ds, conn)
corr.handle(_avalanche_event(42.0, -114.0))
await asyncio.sleep(0)
assert conn.calls == []
# ---------------------------------------------------------------------------
# 3. Role filter (CLIENT out, CLIENT_BASE in)
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_client_node_not_flagged_but_client_base_is():
"""A CLIENT node co-located with the hazard is NOT flagged (not in
monitor_roles); a co-located CLIENT_BASE node IS flagged."""
conn = StubConnector()
client = _node(node_num=201, role="CLIENT", lat=42.0, lon=-114.0,
short_name="CLNT")
client_base = _node(node_num=202, role="CLIENT_BASE", lat=42.0, lon=-114.0,
short_name="CBAS")
ds = FakeDataStore([client, client_base])
cfg = _config(dry_run=False, default_buffer_mi=5.0)
corr = DangerZoneCorrelator(cfg, ds, conn)
corr.handle(_avalanche_event(42.0, -114.0))
await asyncio.sleep(0)
# Exactly one send (CLIENT_BASE), CLIENT excluded by role filter.
assert len(conn.calls) == 1
# ---------------------------------------------------------------------------
# 4. Freshness
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_stale_node_skipped():
"""A node whose last_heard is older than position_max_age_hours is
skipped (not scanned)."""
conn = StubConnector()
stale = _node(node_num=301, lat=42.0, lon=-114.0,
last_heard=time.time() - 100 * 3600) # 100h old
ds = FakeDataStore([stale])
cfg = _config(dry_run=False, position_max_age_hours=72)
corr = DangerZoneCorrelator(cfg, ds, conn)
corr.handle(_avalanche_event(42.0, -114.0))
await asyncio.sleep(0)
assert conn.calls == []
# ---------------------------------------------------------------------------
# 5. Cooldown
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_cooldown_suppresses_second_then_allows_after_window():
"""Two events for the same (node, family) within cooldown -> one send;
after the cooldown window elapses -> sends again."""
conn = StubConnector()
node = _node(node_num=401, lat=42.0, lon=-114.0)
ds = FakeDataStore([node])
cfg = _config(dry_run=False, cooldown_minutes=60)
corr = DangerZoneCorrelator(cfg, ds, conn)
corr.handle(_avalanche_event(42.0, -114.0))
await asyncio.sleep(0)
corr.handle(_avalanche_event(42.0, -114.0)) # within cooldown
await asyncio.sleep(0)
assert len(conn.calls) == 1, "second event within cooldown must be suppressed"
# Wind the recorded last-alert ts back beyond the cooldown window.
key = (node.node_num, "avalanche")
corr._last[key] = time.time() - (60 * 60 + 5)
corr.handle(_avalanche_event(42.0, -114.0))
await asyncio.sleep(0)
assert len(conn.calls) == 2, "after cooldown window, a new send must fire"
# ---------------------------------------------------------------------------
# 6. dry_run / disabled
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_dry_run_never_sends_even_on_hit():
"""dry_run=True -> connector.send_message NEVER called (zero RF), even on
a definite hit."""
conn = StubConnector()
node = _node(node_num=501, lat=42.0, lon=-114.0)
ds = FakeDataStore([node])
cfg = _config(dry_run=True)
corr = DangerZoneCorrelator(cfg, ds, conn)
corr.handle(_avalanche_event(42.0, -114.0))
await asyncio.sleep(0)
assert conn.calls == []
# dry_run still records cooldown (proves it proceeded, just send-free).
assert (node.node_num, "avalanche") in corr._last
@pytest.mark.asyncio
async def test_disabled_does_nothing():
"""enabled=False -> nothing happens at all."""
conn = StubConnector()
node = _node(node_num=502, lat=42.0, lon=-114.0)
ds = FakeDataStore([node])
cfg = _config(enabled=False, dry_run=False)
corr = DangerZoneCorrelator(cfg, ds, conn)
corr.handle(_avalanche_event(42.0, -114.0))
await asyncio.sleep(0)
assert conn.calls == []
assert corr._last == {}
# ---------------------------------------------------------------------------
# 7. min_severity / family disabled
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_below_min_severity_not_flagged():
"""An event below the family's min_severity -> no flag."""
conn = StubConnector()
node = _node(node_num=601, lat=42.0, lon=-114.0)
ds = FakeDataStore([node])
# avalanche min_severity=immediate; event is only 'priority' -> gated out.
cfg = _config(dry_run=False,
avalanche=DangerZoneHazardConfig(min_severity="immediate"))
corr = DangerZoneCorrelator(cfg, ds, conn)
corr.handle(_avalanche_event(42.0, -114.0, severity="priority"))
await asyncio.sleep(0)
assert conn.calls == []
@pytest.mark.asyncio
async def test_min_severity_met_is_flagged():
"""Sanity counterpart: an event AT min_severity passes the gate."""
conn = StubConnector()
node = _node(node_num=602, lat=42.0, lon=-114.0)
ds = FakeDataStore([node])
cfg = _config(dry_run=False,
avalanche=DangerZoneHazardConfig(min_severity="priority"))
corr = DangerZoneCorrelator(cfg, ds, conn)
corr.handle(_avalanche_event(42.0, -114.0, severity="priority"))
await asyncio.sleep(0)
assert len(conn.calls) == 1
@pytest.mark.asyncio
async def test_family_disabled_not_flagged():
"""The family sub-config disabled -> no flag even on a clear hit."""
conn = StubConnector()
node = _node(node_num=603, lat=42.0, lon=-114.0)
ds = FakeDataStore([node])
cfg = _config(dry_run=False,
avalanche=DangerZoneHazardConfig(enabled=False))
corr = DangerZoneCorrelator(cfg, ds, conn)
corr.handle(_avalanche_event(42.0, -114.0))
await asyncio.sleep(0)
assert conn.calls == []
# ---------------------------------------------------------------------------
# Extra: seismic family also works as a point hazard (proves family routing
# beyond avalanche).
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_seismic_event_routes_and_flags():
conn = StubConnector()
node = _node(node_num=701, lat=42.0, lon=-114.0)
ds = FakeDataStore([node])
cfg = _config(dry_run=False,
seismic=DangerZoneHazardConfig(min_severity="routine"))
corr = DangerZoneCorrelator(cfg, ds, conn)
ev = make_event(source="usgs", category="earthquake_event",
severity="priority", title="M5.1 quake",
summary="M5.1 quake", lat=42.0, lon=-114.0)
corr.handle(ev)
await asyncio.sleep(0)
assert len(conn.calls) == 1

196
work/tests/test_fire_age_gate.py Normal file
View file

@ -0,0 +1,196 @@
"""Step 3 fire age-gate tests (Step 5 verification, age-gate part).
Targets `meshai.central.wfigs_handler._fire_too_old_to_announce` and the
handler's "New"-path suppression behaviour.
The gate exists because MeshAI broadcast a 6-week-old, already-closed fire
(OTR 11, declared_at=2026-05-06, ~45d old) to the live mesh as "New". The
gate keys on the fire's OWN declared_at age, not event recency.
Knob: ("wfigs","max_declare_age_seconds"), default 1209600 (14d), 0 = off.
The helper re-reads the knob each call (cache-backed, GUI-invalidated) and
FAILS OPEN (announces) when disabled or declared_at is None.
The autouse conftest fixture seeds adapter_config from the defaults
registry, so max_declare_age_seconds starts at its 14d default. Tests that
need a different value UPDATE the row + invalidate_cache().
"""
from __future__ import annotations
import time
import pytest
from meshai.central.wfigs_handler import _fire_too_old_to_announce
_14D = 14 * 86400
_45D = 45 * 86400
_5D = 5 * 86400
def _set_knob(seconds: int):
"""Override max_declare_age_seconds in adapter_config + drop the cache so
the helper re-reads it on its next call."""
from meshai.persistence import get_db
from meshai.adapter_config import invalidate_cache
get_db().execute(
"UPDATE adapter_config SET value_json=? "
"WHERE adapter='wfigs' AND key='max_declare_age_seconds'",
(str(int(seconds)),),
)
invalidate_cache()
# ---------------------------------------------------------------------------
# Helper-level: the five required cases.
# ---------------------------------------------------------------------------
def test_declared_at_none_fails_open():
"""declared_at_epoch=None -> announce (fail-open). Default 14d knob."""
now = int(time.time())
assert _fire_too_old_to_announce(None, now) is False
def test_old_fire_suppressed_default_knob():
"""~45 days old + default 14d knob -> suppress (models OTR 11)."""
now = int(time.time())
declared = now - _45D
assert _fire_too_old_to_announce(declared, now) is True
def test_recent_fire_announces():
"""~5 days old -> announce (well under the 14d default)."""
now = int(time.time())
declared = now - _5D
assert _fire_too_old_to_announce(declared, now) is False
def test_knob_zero_disables_gate():
"""knob=0 -> gate disabled, even an ancient fire announces (fail-open)."""
_set_knob(0)
now = int(time.time())
declared = now - _45D
assert _fire_too_old_to_announce(declared, now) is False
def test_boundary_exactly_14d_is_suppressed():
"""Boundary: a fire declared exactly the knob age ago -> suppressed (>=)."""
now = int(time.time())
declared = now - _14D # exactly 14 days
assert _fire_too_old_to_announce(declared, now) is True
def test_boundary_one_second_under_14d_announces():
"""Just under the boundary (14d - 1s) -> announce (strict >= cutoff)."""
now = int(time.time())
declared = now - _14D + 1
assert _fire_too_old_to_announce(declared, now) is False
def test_custom_knob_respected():
"""A custom (non-default) knob value is honoured by the helper."""
_set_knob(_5D) # 5-day gate
now = int(time.time())
assert _fire_too_old_to_announce(now - 6 * 86400, now) is True # older than 5d
assert _fire_too_old_to_announce(now - 4 * 86400, now) is False # younger than 5d
# ---------------------------------------------------------------------------
# Handler-path: New is suppressed, Update still emits.
#
# These exercise the real Case (i)/(ii)/(iii) paths in handle_wfigs against
# the isolated tmp DB seeded by conftest.
# ---------------------------------------------------------------------------
def _normalized(*, irwin_id, declared_at_epoch, acres=250.0, contained=0):
return {
"_kind": "wfigs_incident",
"irwin_id": irwin_id,
"incident_name": "Old Town Road",
"incident_type": "WF",
"acres": acres,
"contained_pct": contained,
"lat": 42.93, "lon": -114.45,
"county": "Twin Falls", "state": "ID",
"declared_at_epoch": declared_at_epoch,
}
def _envelope():
return {
"data": {"adapter": "wfigs", "category": "wildfire_incident",
"severity": "priority"}
}
def test_handler_new_path_suppresses_old_fire():
"""Case (i): a brand-new fire whose declared_at is ~45d old is INSERTed
but the 'New' broadcast is suppressed (wire is None), and the New-path
category tag is NOT applied."""
from meshai.central.wfigs_handler import handle_wfigs
from meshai.persistence import get_db
now = int(time.time())
declared = now - _45D # OTR 11 style
data = {}
wire = handle_wfigs(_normalized(irwin_id="ID-OTR-11", declared_at_epoch=declared),
_envelope(),
subject="central.fire.incident.id",
data=data, now=now)
assert wire is None, f"old fire should be silenced, got wire={wire!r}"
assert data.get("category") != "wildfire_declared"
# The row is still INSERTed (so genuine future Updates work).
row = get_db().execute(
"SELECT irwin_id, last_broadcast_at FROM fires WHERE irwin_id=?",
("ID-OTR-11",)).fetchone()
assert row is not None
assert row["last_broadcast_at"] is None
def test_handler_new_path_announces_recent_fire():
"""Case (i): a recent fire (~5d) DOES broadcast 'New' and tags
wildfire_declared."""
from meshai.central.wfigs_handler import handle_wfigs
now = int(time.time())
declared = now - _5D
data = {}
wire = handle_wfigs(_normalized(irwin_id="ID-RECENT-1", declared_at_epoch=declared),
_envelope(),
subject="central.fire.incident.id",
data=data, now=now)
assert wire is not None and "New" in wire
assert data.get("category") == "wildfire_declared"
def test_handler_update_path_still_emits_for_old_fire():
"""An already-broadcast OLD fire that grows acreage still emits an
'Update' (Case (iii) is NOT gated -- genuine old-but-active fires keep
getting containment/acreage updates)."""
from meshai.central.wfigs_handler import handle_wfigs
from meshai.persistence import get_db
now = int(time.time())
declared = now - _45D
# Pre-existing row that has already been broadcast.
get_db().execute(
"INSERT INTO fires(irwin_id, incident_name, current_acres, "
"current_contained_pct, lat, lon, declared_at, last_event_at, "
"last_broadcast_at, last_broadcast_acres, last_broadcast_contained) "
"VALUES (?,?,?,?,?,?,?,?,?,?,?)",
("ID-OLD-ACTIVE", "Old Town Road", 250.0, 0, 42.93, -114.45,
declared, now - 30000, now - 30000, 250.0, 0),
)
data = {}
wire = handle_wfigs(
_normalized(irwin_id="ID-OLD-ACTIVE", declared_at_epoch=declared,
acres=900.0, contained=20),
_envelope(), subject="central.fire.incident.id",
data=data, now=now)
assert wire is not None and "Update" in wire, \
f"old-but-active fire Update must still emit, got {wire!r}"
assert data.get("category") != "wildfire_declared"

27
work/tests/test_wfigs_handler.py
View file

@ -322,10 +322,15 @@ def test_g_new_irwin_inserts_and_broadcasts(mem_db, no_photon):
# (h) known IRWIN no-change -> drop silently, last_broadcast_* unchanged
# ============================================================================
def test_h_known_irwin_no_change_drops(mem_db, no_photon):
env = _make_active_envelope(geocoder_city="Burley")
# Use wall-clock-adjacent timestamps so _cleanup_stale_fires doesn't
# delete the row (it uses real time.time() internally).
# delete the row (it uses real time.time() with a 7d cutoff internally).
# Anchor the fire's discovery date 2d before that `now` so it stays
# inside the 14d fire age-gate -- otherwise the static fixture date
# (2026-06-03) is now stale vs wall-clock and the first-sight "New"
# broadcast this test depends on would be (correctly) suppressed.
first_now = int(time.time())
env = _make_active_envelope(geocoder_city="Burley",
fire_discovery_dt_ms=(first_now - 2 * 86400) * 1000)
data0 = {}
handle_wfigs(cn.normalize(env), env, env["subject"],
data=data0, now=first_now)
@ -359,16 +364,24 @@ def test_h_known_irwin_no_change_drops(mem_db, no_photon):
# (i) known IRWIN acres up but <8h elapsed -> drop, last_broadcast_* unchanged
# ============================================================================
def test_i_known_irwin_change_inside_cooldown_drops(mem_db, no_photon):
env_initial = _make_active_envelope(geocoder_city="Burley")
data0 = {}
# Wall-clock `now` so _cleanup_stale_fires (real time.time(), 7d cutoff)
# keeps the row; anchor discovery 2d earlier so the fire stays inside
# the 14d fire age-gate (the static fixture date is now stale vs
# wall-clock and would suppress the first-sight "New" broadcast).
_base = int(time.time())
env_initial = _make_active_envelope(
geocoder_city="Burley",
fire_discovery_dt_ms=(_base - 2 * 86400) * 1000)
data0 = {}
handle_wfigs(cn.normalize(env_initial), env_initial,
env_initial["subject"], data=data0, now=_base)
data0["_on_broadcast_committed"](float(_base))
# Bigger fire, but only 4h later -- inside cooldown.
env_grown = _make_active_envelope(geocoder_city="Burley",
daily_acres=3000.0, pct_contained=23)
# Bigger fire, but only 4h later -- inside cooldown. Same discovery
# date as the initial envelope (same fire, still inside the age-gate).
env_grown = _make_active_envelope(
geocoder_city="Burley", daily_acres=3000.0, pct_contained=23,
fire_discovery_dt_ms=(_base - 2 * 86400) * 1000)
later = _base + 4 * 3600
out = handle_wfigs(cn.normalize(env_grown), env_grown,
env_grown["subject"], now=later)