diff --git a/work/dashboard-frontend/src/pages/Notifications.tsx b/work/dashboard-frontend/src/pages/Notifications.tsx index dbc1fe0..ac7100b 100644 --- a/work/dashboard-frontend/src/pages/Notifications.tsx +++ b/work/dashboard-frontend/src/pages/Notifications.tsx @@ -8,6 +8,7 @@ import { } from 'lucide-react' import ChannelPicker from '@/components/ChannelPicker' import NodePicker from '@/components/NodePicker' +import { fetchConfig as apiFetchConfig, updateConfig as apiUpdateConfig } from '@/lib/api' // Types interface NotificationRuleConfig { @@ -2200,6 +2201,467 @@ export default function Notifications() { )} + + {/* Danger Zones — self-contained panel (own GET/PUT for the isolated + `danger_zones` config section; never entangled with the notifications + save above). Additive only. */} + + + ) +} + +// ============================================================================ +// Danger Zones panel — fully isolated, additive feature. +// Loads/saves the standalone `danger_zones` config section via the generic +// /api/config helpers. Has its OWN state, fetch (on mount), and save. It is +// intentionally decoupled from the page's `notifications` config so it can +// never be tangled with the existing save logic. +// ============================================================================ + +const DZ_MONITOR_ROLES = ['CLIENT_BASE', 'ROUTER', 'ROUTER_LATE'] as const + +const DZ_SEVERITY_OPTIONS = [ + { value: 'routine', label: 'Routine' }, + { value: 'priority', label: 'Priority' }, + { value: 'immediate', label: 'Immediate' }, +] + +const DZ_DELIVERY_OPTIONS = [ + { value: 'mesh_dm', label: 'Mesh DM (unicast to nodes)' }, + { value: 'mesh_broadcast', label: 'Mesh Broadcast (channel)' }, + { value: 'email', label: 'Email' }, + { value: 'webhook', label: 'Webhook' }, + { value: 'none', label: '(None / log only)' }, +] + +// Per-family rows. snow is a sub-gate of weather; flood a sub-gate of seismic. +const DZ_FAMILIES: { + key: string + label: string + description: string + Icon: typeof Activity + showAcres?: boolean +}[] = [ + { key: 'fire', label: 'Fire', description: 'Active wildfires (radius from fire perimeter).', Icon: Flame, showAcres: true }, + { key: 'weather', label: 'Weather', description: 'Severe weather warnings near a node.', Icon: Cloud }, + { key: 'snow', label: 'Snow (sub-gate of Weather)', description: 'Snow-category weather events.', Icon: Snowflake }, + { key: 'flood', label: 'Flood (sub-gate of Seismic)', description: 'Stream/flood gauge events.', Icon: Activity }, + { key: 'avalanche', label: 'Avalanche', description: 'Avalanche advisories near a node.', Icon: Mountain }, + { key: 'seismic', label: 'Seismic', description: 'Earthquakes and seismic events near a node.', Icon: Mountain }, +] + +interface DangerZoneHazardConfig { + enabled: boolean + buffer_mi: number + min_severity: string + min_acres: number +} + +interface DangerZonesConfig { + enabled: boolean + dry_run: boolean + monitor_roles: string[] + position_max_age_hours: number + default_buffer_mi: number + cooldown_minutes: number + fire: DangerZoneHazardConfig + weather: DangerZoneHazardConfig + snow: DangerZoneHazardConfig + flood: DangerZoneHazardConfig + avalanche: DangerZoneHazardConfig + seismic: DangerZoneHazardConfig + delivery_type: string + node_ids: string[] + broadcast_channel: number | null + webhook_url: string + webhook_headers: Record +} + +function dzDefaultHazard(): DangerZoneHazardConfig { + return { enabled: false, buffer_mi: 5.0, min_severity: 'priority', min_acres: 0 } +} + +// New-object default. NOTE: delivery_type defaults to mesh_dm (do NOT copy the +// page's mesh_broadcast new-rule default). +function dzDefault(): DangerZonesConfig { + return { + enabled: false, + dry_run: true, + monitor_roles: ['ROUTER', 'ROUTER_LATE', 'CLIENT_BASE'], + position_max_age_hours: 72, + default_buffer_mi: 5.0, + cooldown_minutes: 360, + fire: dzDefaultHazard(), + weather: dzDefaultHazard(), + snow: dzDefaultHazard(), + flood: dzDefaultHazard(), + avalanche: dzDefaultHazard(), + seismic: dzDefaultHazard(), + delivery_type: 'mesh_dm', + node_ids: [], + broadcast_channel: null, + webhook_url: '', + webhook_headers: {}, + } +} + +// Minimal local select — SelectInput lives in Config.tsx (which we must not +// touch/entangle), so this panel keeps its own tiny equivalent. +function DZSelect({ label, value, onChange, options, info = '' }: { + label: string + value: string + onChange: (v: string) => void + options: { value: string; label: string }[] + info?: string +}) { + return ( +
+ + +
+ ) +} + +// Per-family row — mirrors the AlertRuleToggle (toggle + thresholds) pattern. +// AlertRuleToggle itself lives in Config.tsx and is NOT exported, so this is a +// local equivalent purpose-built for the per-family hazard config. +function DZFamilyRow({ meta, cfg, onChange }: { + meta: { key: string; label: string; description: string; Icon: typeof Activity; showAcres?: boolean } + cfg: DangerZoneHazardConfig + onChange: (c: DangerZoneHazardConfig) => void +}) { + const { Icon } = meta + return ( +
+
+
+ +
+ {meta.label} +

{meta.description}

+
+
+ +
+ {cfg.enabled && ( +
+ onChange({ ...cfg, buffer_mi: v })} + min={0} + step={0.5} + /> + onChange({ ...cfg, min_severity: v })} + options={DZ_SEVERITY_OPTIONS} + /> + {meta.showAcres && ( + onChange({ ...cfg, min_acres: v })} + min={0} + step={1} + /> + )} +
+ )} +
+ ) +} + +function DangerZonesPanel() { + const [expanded, setExpanded] = useState(false) + const [cfg, setCfg] = useState(null) + const [loading, setLoading] = useState(true) + const [saving, setSaving] = useState(false) + const [error, setError] = useState(null) + const [success, setSuccess] = useState(null) + + const load = useCallback(async () => { + setLoading(true) + setError(null) + try { + const raw = (await apiFetchConfig('danger_zones')) as Partial + // Merge over defaults so missing/new fields are always present. + const d = dzDefault() + setCfg({ + ...d, + ...raw, + fire: { ...d.fire, ...(raw.fire || {}) }, + weather: { ...d.weather, ...(raw.weather || {}) }, + snow: { ...d.snow, ...(raw.snow || {}) }, + flood: { ...d.flood, ...(raw.flood || {}) }, + avalanche: { ...d.avalanche, ...(raw.avalanche || {}) }, + seismic: { ...d.seismic, ...(raw.seismic || {}) }, + monitor_roles: raw.monitor_roles ?? d.monitor_roles, + node_ids: raw.node_ids ?? d.node_ids, + webhook_headers: raw.webhook_headers ?? d.webhook_headers, + }) + } catch (err) { + setError(err instanceof Error ? err.message : 'Failed to load danger zones config') + setCfg(dzDefault()) + } finally { + setLoading(false) + } + }, []) + + useEffect(() => { load() }, [load]) + + const save = async () => { + if (!cfg) return + setSaving(true) + setError(null) + setSuccess(null) + try { + await apiUpdateConfig('danger_zones', cfg) + setSuccess('Danger Zones config saved') + setTimeout(() => setSuccess(null), 3000) + } catch (err) { + setError(err instanceof Error ? err.message : 'Save failed') + } finally { + setSaving(false) + } + } + + const upd = (patch: Partial) => setCfg(c => (c ? { ...c, ...patch } : c)) + + const toggleRole = (role: string) => { + if (!cfg) return + const cur = cfg.monitor_roles || [] + upd({ monitor_roles: cur.includes(role) ? cur.filter(r => r !== role) : [...cur, role] }) + } + + return ( +
+ {/* Collapsible header */} + + + {expanded && ( +
+ {/* Safety copy */} +
+ +
+ Ships disabled; when enabled, defaults to dry-run / log-only — no mesh traffic + until you turn dry-run off. Requires Enable Notifications (above) + and environmental feeds to be on, since hazard events only flow when those are active. +
+
+ + {/* Status messages */} + {error && ( +
{error}
+ )} + {success && ( +
+ {success} +
+ )} + + {loading || !cfg ? ( +
Loading danger zones config...
+ ) : ( + <> + upd({ enabled: v })} + helper="Master switch for the infrastructure danger-zone correlator" + /> + upd({ dry_run: v })} + helper="When on, matches are logged but nothing is sent to the mesh. Turn off only after verifying dry-run output." + /> + + {/* Monitored roles */} +
+ +
+ {DZ_MONITOR_ROLES.map(role => { + const on = (cfg.monitor_roles || []).includes(role) + return ( + + ) + })} +
+
+ + {/* Global numeric settings */} +
+ upd({ position_max_age_hours: v })} + min={1} + helper="Skip nodes whose last position is older than this" + /> + upd({ default_buffer_mi: v })} + min={0} + step={0.5} + helper="Buffer used when a family has none set" + /> + upd({ cooldown_minutes: v })} + min={0} + helper="Min time between repeat alerts per node+family" + /> +
+ + {/* Per-family hazard config */} +
+ + {DZ_FAMILIES.map(meta => ( + upd({ [meta.key]: c } as Partial)} + /> + ))} +
+ + {/* Delivery */} +
+
+ + DELIVERY +
+ upd({ delivery_type: v })} + options={DZ_DELIVERY_OPTIONS} + info="Where danger-zone alerts get delivered. Mesh DM unicasts to specific nodes; broadcast sends to a channel. Has no effect while dry-run is on." + /> + + {cfg.delivery_type === 'mesh_dm' && ( + upd({ node_ids: v })} + helper="Nodes that receive direct messages" + valueType="node_id_hex" + /> + )} + + {cfg.delivery_type === 'mesh_broadcast' && ( + upd({ broadcast_channel: v })} + helper="Select the mesh radio channel" + mode="single" + /> + )} + + {cfg.delivery_type === 'webhook' && ( + upd({ webhook_url: v })} + placeholder="https://discord.com/api/webhooks/..." + helper="POST alert as JSON" + /> + )} + + {cfg.delivery_type === 'email' && ( +

+ Email delivery uses the SMTP settings configured for notification rules. +

+ )} +
+ + {/* Save */} +
+ +
+ + )} +
+ )}
) } diff --git a/work/meshai/adapter_config/defaults.py b/work/meshai/adapter_config/defaults.py index 1d8d6e4..06a9603 100644 --- a/work/meshai/adapter_config/defaults.py +++ b/work/meshai/adapter_config/defaults.py @@ -66,6 +66,11 @@ REGISTRY: dict[tuple[str, str], dict[str, Any]] = { "type": "int", "description": "Staleness gate for wfigs events (0 = disabled). Fire events are always relevant regardless of age.", }, + ("wfigs", "max_declare_age_seconds"): { + "default": 1209600, # 14 days + "type": "int", + "description": "Suppress first-announce ('New') for fires whose declared_at is older than this (closed/stale-fire resurrection guard). 0 = disabled. Does NOT affect Updates.", + }, # ================================================================= # NWS -- 3 settings (severity gate, tombstone msgTypes, suffix-promote toggle) diff --git a/work/meshai/central/consumer.py b/work/meshai/central/consumer.py index ed7d91c..703a6fe 100644 --- a/work/meshai/central/consumer.py +++ b/work/meshai/central/consumer.py @@ -835,9 +835,14 @@ class CentralConsumer: logger.info("drain complete: 0 fires touched") return + # Step 3 age-gate: `now` is otherwise undefined in this method (`time` + # is imported at module level). Used by the first-announce age gate. + now = int(time.time()) + from meshai.persistence import get_db from meshai.central.wfigs_handler import ( _render, _location_anchor, _attach_commit_handles, + _fire_too_old_to_announce, ) from meshai.notifications.events import make_event @@ -877,6 +882,11 @@ class CentralConsumer: category = "wildfire_closed" elif not announced and not tombstoned: # Case 2: Never announced + still active -> NEW + # Step 3 age-gate: suppress first-announce for fires whose + # declared_at is too old (closed/stale-fire resurrection guard). + if _fire_too_old_to_announce(row["declared_at"], now): + silenced += 1 + continue wire = _render(_row_to_normalized(row), prefix="New") category = "wildfire_declared" else: diff --git a/work/meshai/central/wfigs_handler.py b/work/meshai/central/wfigs_handler.py index 6e4d16e..3e8a9f9 100644 --- a/work/meshai/central/wfigs_handler.py +++ b/work/meshai/central/wfigs_handler.py @@ -64,6 +64,19 @@ def _now() -> int: return int(time.time()) +def _fire_too_old_to_announce(declared_at_epoch, now) -> bool: + """Stale/closed-fire resurrection guard for first-announce ("New") only. + + Re-reads the knob INSIDE the call (cache-backed + GUI-invalidated, so a + dashboard edit takes effect on the next poll cycle). Fails OPEN (announces) + when the gate is disabled or the fire has no discovery date. + """ + max_age = int(adapter_config.wfigs.max_declare_age_seconds) + if max_age <= 0 or declared_at_epoch is None: + return False # disabled, or no discovery date -> fail OPEN (announce) + return (now - int(declared_at_epoch)) >= max_age + + # ---------- public entry -------------------------------------------------- @@ -201,6 +214,10 @@ def handle_wfigs(normalized: dict, envelope: dict, subject: str, None, None, None, # last_broadcast_* explicitly NULL ), ) + # Step 3 age-gate: keep the INSERT (so genuine future Updates work) but + # suppress the "New" broadcast for fires whose declared_at is too old. + if _fire_too_old_to_announce(normalized.get("declared_at_epoch"), now): + return None wire = _render(normalized, prefix="New") # v0.7-fire-tracker-1: tag first-sight broadcasts with the new # wildfire_declared category so the dispatcher rules them apart @@ -225,6 +242,10 @@ def handle_wfigs(normalized: dict, envelope: dict, subject: str, (acres, contained_pct, normalized.get("lat"), normalized.get("lon"), now, irwin_id), ) + # Step 3 age-gate: keep the UPDATE but suppress the "New" broadcast for + # fires whose declared_at is too old (closed/stale-fire resurrection). + if _fire_too_old_to_announce(normalized.get("declared_at_epoch"), now): + return None wire = _render(normalized, prefix="New") # v0.7-fire-tracker-1: case-(ii) is also first-sight as far as # broadcast history goes -- the row exists because some prior diff --git a/work/meshai/config.py b/work/meshai/config.py index acfda69..d5acfcd 100644 --- a/work/meshai/config.py +++ b/work/meshai/config.py @@ -667,6 +667,106 @@ class DashboardConfig: port: int = 8080 host: str = "0.0.0.0" +# v0.8 danger_zones: infrastructure-node hazard correlation. A standalone, +# isolated config section (its own dataclass tree + danger_zones.yaml + its own +# GET/PUT) so it never touches the complex notifications dataclass. + +# Hardcoded Meshtastic role-name vocabulary (mirror of +# mesh_data_store.MESHTASTIC_ROLE_MAP values). Defined locally to avoid an +# import cycle (config.py must not import mesh_data_store). +_DZ_VALID_ROLES = frozenset({ + "ROUTER", "ROUTER_LATE", "CLIENT_BASE", "ROUTER_CLIENT", "REPEATER", + "CLIENT", "CLIENT_MUTE", "TRACKER", "TAK", +}) + +_DZ_VALID_SEVERITIES = frozenset({"routine", "priority", "immediate"}) +_DZ_VALID_DELIVERY = frozenset({ + "mesh_broadcast", "mesh_dm", "email", "webhook", "none", +}) +# Hazard families that map onto categories.VALID_TOGGLES. snow is a sub-gate of +# weather and flood a sub-gate of seismic (resolved in the correlator), so they +# are NOT validated against VALID_TOGGLES. +_DZ_PARENT_FAMILIES = ("fire", "weather", "avalanche", "seismic") + + +@dataclass +class DangerZoneHazardConfig: + """Per-hazard-family danger-zone tuning (distances in MILES).""" + + enabled: bool = True + buffer_mi: float = 5.0 + min_severity: str = "priority" # routine|priority|immediate + min_acres: float = 0.0 # fire-only; ignored by other families + + +@dataclass +class DangerZonesConfig: + """Infrastructure-node hazard danger-zone subsystem settings. + + Requires notifications.enabled (the EventBus only exists under that guard). + Ships disabled; enabling without turning off dry_run is log-only (no RF). + """ + + enabled: bool = False + dry_run: bool = True + monitor_roles: list[str] = field( + default_factory=lambda: ["ROUTER", "ROUTER_LATE", "CLIENT_BASE"]) + position_max_age_hours: int = 72 + default_buffer_mi: float = 5.0 + cooldown_minutes: int = 360 + + # Per-family sub-configs. snow->weather, flood->seismic resolved in the + # correlator; both still exposed here for distinct GUI tuning. + fire: DangerZoneHazardConfig = field(default_factory=DangerZoneHazardConfig) + weather: DangerZoneHazardConfig = field(default_factory=DangerZoneHazardConfig) + snow: DangerZoneHazardConfig = field(default_factory=DangerZoneHazardConfig) + flood: DangerZoneHazardConfig = field(default_factory=DangerZoneHazardConfig) + avalanche: DangerZoneHazardConfig = field(default_factory=DangerZoneHazardConfig) + seismic: DangerZoneHazardConfig = field(default_factory=DangerZoneHazardConfig) + + # Delivery + delivery_type: str = "mesh_dm" # mesh_broadcast|mesh_dm|email|webhook|none + node_ids: list = field(default_factory=list) + broadcast_channel: Optional[int] = None + webhook_url: str = "" + webhook_headers: dict = field(default_factory=dict) + + def __post_init__(self): + # Lazy import: categories.py imports only `typing`, so no cycle; kept + # function-local per plan to stay defensive against future imports. + from meshai.notifications.categories import VALID_TOGGLES + + if self.delivery_type not in _DZ_VALID_DELIVERY: + raise ValueError( + f"danger_zones.delivery_type must be one of " + f"{sorted(_DZ_VALID_DELIVERY)}, got {self.delivery_type!r}") + + for role in self.monitor_roles: + if role not in _DZ_VALID_ROLES: + raise ValueError( + f"danger_zones.monitor_roles contains invalid role {role!r}; " + f"valid roles: {sorted(_DZ_VALID_ROLES)}") + + # Parent families must exist in the canonical toggle vocabulary. + for fam in _DZ_PARENT_FAMILIES: + if fam not in VALID_TOGGLES: + raise ValueError( + f"danger_zones parent family {fam!r} is not a valid toggle " + f"({sorted(VALID_TOGGLES)})") + + if self.position_max_age_hours <= 0: + raise ValueError( + "danger_zones.position_max_age_hours must be > 0, got " + f"{self.position_max_age_hours}") + + for fam in ("fire", "weather", "snow", "flood", "avalanche", "seismic"): + sub = getattr(self, fam) + if sub.min_severity not in _DZ_VALID_SEVERITIES: + raise ValueError( + f"danger_zones.{fam}.min_severity must be one of " + f"{sorted(_DZ_VALID_SEVERITIES)}, got {sub.min_severity!r}") + + @dataclass class Config: """Main configuration container.""" @@ -690,6 +790,7 @@ class Config: environmental: EnvironmentalConfig = field(default_factory=EnvironmentalConfig) dashboard: DashboardConfig = field(default_factory=DashboardConfig) notifications: NotificationsConfig = field(default_factory=NotificationsConfig) + danger_zones: DangerZonesConfig = field(default_factory=DangerZonesConfig) _config_path: Optional[Path] = field(default=None, repr=False) diff --git a/work/meshai/config_loader.py b/work/meshai/config_loader.py index 169d99e..220c3e8 100644 --- a/work/meshai/config_loader.py +++ b/work/meshai/config_loader.py @@ -54,6 +54,7 @@ SECTION_TO_FILE: dict[str, str] = { "notifications": "notifications.yaml", "llm": "llm.yaml", "dashboard": "dashboard.yaml", + "danger_zones": "danger_zones.yaml", } # Fields that should be written to local.yaml instead of domain files @@ -79,6 +80,7 @@ SECRET_FIELDS: set[str] = { "environmental.firms.map_key", "notifications.rules.*.smtp_password", "notifications.toggles.*.smtp_password", + "danger_zones.webhook_url", } # Secret env var names expected in .env diff --git a/work/meshai/dashboard/api/config_routes.py b/work/meshai/dashboard/api/config_routes.py index e511511..2bd2797 100644 --- a/work/meshai/dashboard/api/config_routes.py +++ b/work/meshai/dashboard/api/config_routes.py @@ -53,6 +53,7 @@ VALID_SECTIONS = { "mesh_sources", "mesh_intelligence", "dashboard", + "danger_zones", } diff --git a/work/meshai/dashboard/server.py b/work/meshai/dashboard/server.py index a5a396f..7139867 100644 --- a/work/meshai/dashboard/server.py +++ b/work/meshai/dashboard/server.py @@ -121,6 +121,7 @@ async def start_dashboard(meshai_instance: "MeshAI") -> DashboardBroadcaster: app.state.notification_router = getattr(meshai_instance, "notification_router", None) app.state.connector = meshai_instance.connector app.state.bus = getattr(meshai_instance, "event_bus", None) + app.state.danger_correlator = getattr(meshai_instance, "danger_correlator", None) # Create broadcaster and attach to app state broadcaster = DashboardBroadcaster() diff --git a/work/meshai/main.py b/work/meshai/main.py index af0c068..5a17582 100644 --- a/work/meshai/main.py +++ b/work/meshai/main.py @@ -511,6 +511,17 @@ class MeshAI: logger.debug('dashboard app.state stash skipped') logger.info("Notification pipeline EventBus initialized") + # v0.8 danger_zones correlator: subscribe to the EventBus to flag + # infra nodes inside a hazard's threat radius. Guarded — needs both + # the bus and a data_store. Reads config.danger_zones fresh per call. + self.danger_correlator = None + if self.event_bus is not None and self.data_store is not None: + from .notifications.danger_zone_correlator import DangerZoneCorrelator + self.danger_correlator = DangerZoneCorrelator( + self.config, self.data_store, self.connector) + self.event_bus.subscribe(self.danger_correlator.handle) + logger.info("Danger-zone correlator subscribed to EventBus") + # Environmental feeds env_cfg = self.config.environmental if env_cfg.enabled: diff --git a/work/meshai/mesh_data_store.py b/work/meshai/mesh_data_store.py index bae95da..22bea59 100644 --- a/work/meshai/mesh_data_store.py +++ b/work/meshai/mesh_data_store.py @@ -2116,6 +2116,24 @@ class MeshDataStore: infra_roles = {"ROUTER", "ROUTER_CLIENT", "ROUTER_LATE", "REPEATER"} return [n for n in self._nodes.values() if n.role in infra_roles] + def get_nodes_by_roles(self, roles: set[str], max_age_s: float) -> list[UnifiedNode]: + """Return a SNAPSHOT of nodes whose role is in `roles`, that have a + non-None GPS position, and whose last_heard is within `max_age_s`. + + Used by the danger-zone correlator. CLIENT_BASE-inclusive (do NOT + reuse get_infrastructure_nodes, which excludes CLIENT_BASE). Returns a + new list so callers can iterate safely across async scheduling. + """ + cutoff = time.time() - max_age_s + return [ + n for n in list(self._nodes.values()) + 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 get_low_battery_nodes(self, threshold: float = 30.0) -> list[UnifiedNode]: """Get nodes with low battery.""" return [ diff --git a/work/meshai/notifications/danger_zone_correlator.py b/work/meshai/notifications/danger_zone_correlator.py new file mode 100644 index 0000000..e1f5583 --- /dev/null +++ b/work/meshai/notifications/danger_zone_correlator.py @@ -0,0 +1,225 @@ +"""Danger-zone correlator (v0.8). + +Subscribes to the notification EventBus and, for hazard events with a +location, flags infrastructure nodes that fall inside the hazard's threat +radius and (optionally) alerts the operator. Distance math is MILES +end-to-end (haversine_distance returns miles). + +EventBus handlers are SYNC; async delivery is fire-and-forget via +asyncio.create_task guarded by try/except RuntimeError (mirrors the +pipeline tee in notifications/pipeline/__init__.py). + +Ships gated: cfg.enabled defaults False; cfg.dry_run defaults True. dry_run +is provably send-free — it logs and returns BEFORE any channel is built. +""" + +import asyncio +import logging +import time +from typing import Optional + +from meshai.geo import haversine_distance +from meshai.notifications import categories +from meshai.notifications.events import make_event, make_payload_from_event +from meshai.notifications import channels +from meshai.config import NotificationRuleConfig +from meshai.notifications.pipeline.dispatcher import Dispatcher + +logger = logging.getLogger(__name__) + +# NWS event_type substrings that indicate a winter/snow hazard. +_SNOW_MARKERS = ("snow", "winter", "blizzard", "ice storm", "freezing") + + +class DangerZoneCorrelator: + """Correlates located hazard events against infrastructure nodes.""" + + def __init__(self, config, data_store, connector): + self.config = config + self.data_store = data_store + self.connector = connector + # (node_num, family) -> last alert epoch + self._last: dict[tuple[int, str], float] = {} + + def _resolve_family(self, event) -> Optional[str]: + """Return the danger_zones sub-config name for this event, or None.""" + fam = categories.get_toggle(event.category) + if fam not in ("fire", "weather", "avalanche", "seismic"): + return None + if fam == "weather": + et = (event.data.get("event_type") or event.title or "").lower() + if any(m in et for m in _SNOW_MARKERS): + return "snow" + return "weather" + if fam == "seismic": + if str(event.category).startswith("stream"): + return "flood" + return "seismic" + return fam # fire | avalanche + + def _fire_radius_and_centroid(self, event): + """For fire events, look up spread_radius_mi + centroid from the fires + table by irwin_id (= event.group_key). Returns (radius_mi, lat, lon) + with lat/lon falling back to the event's own coords. Non-fire -> (0, None, None).""" + irwin_id = event.group_key + if not irwin_id: + return 0.0, event.lat, event.lon + from meshai.persistence import get_db + try: + row = get_db().execute( + "SELECT spread_radius_mi, current_centroid_lat, current_centroid_lon, " + "lat, lon, current_acres FROM fires WHERE irwin_id = ?", + (irwin_id,), + ).fetchone() + except Exception: + logger.exception("danger_zone: fires lookup failed for %s", irwin_id) + return 0.0, event.lat, event.lon + if row is None: + return 0.0, event.lat, event.lon + radius = row["spread_radius_mi"] if row["spread_radius_mi"] is not None else 0.0 + haz_lat = row["current_centroid_lat"] if row["current_centroid_lat"] is not None else ( + row["lat"] if row["lat"] is not None else event.lat) + haz_lon = row["current_centroid_lon"] if row["current_centroid_lon"] is not None else ( + row["lon"] if row["lon"] is not None else event.lon) + return float(radius), haz_lat, haz_lon + + def _fire_acres(self, event) -> Optional[float]: + """Acres for a fire event: prefer the fires DB row, fall back to event.data.""" + irwin_id = event.group_key + if irwin_id: + from meshai.persistence import get_db + try: + row = get_db().execute( + "SELECT current_acres FROM fires WHERE irwin_id = ?", + (irwin_id,), + ).fetchone() + if row is not None and row["current_acres"] is not None: + return float(row["current_acres"]) + except Exception: + pass + v = event.data.get("acres") + try: + return float(v) if v is not None else None + except (TypeError, ValueError): + return None + + def handle(self, event) -> None: + """Sync EventBus handler. Never raises (bus isolates, but be safe).""" + try: + self._handle(event) + except Exception: + logger.exception("danger_zone correlator failed on event %s", getattr(event, "category", "?")) + + def _handle(self, event) -> None: + cfg = self.config.danger_zones + if not cfg.enabled: + return + if self.data_store is None: + return + + # Active hazards only. + if event.category == "wildfire_closed": + return + + fam = self._resolve_family(event) + if fam is None: + return + sub = getattr(cfg, fam, None) + if sub is None or not sub.enabled: + return + + # min_severity gate (use Dispatcher's rank table). + rank = Dispatcher.SEVERITY_RANK + if rank.get(event.severity, 0) < rank.get(sub.min_severity, 0): + return + + # Fire radius + acres; non-fire is a point hazard at the event location. + if fam == "fire": + radius_mi, haz_lat, haz_lon = self._fire_radius_and_centroid(event) + min_acres = sub.min_acres or 0.0 + if min_acres > 0: + acres = self._fire_acres(event) + if acres is None or acres < min_acres: + return + else: + radius_mi, haz_lat, haz_lon = 0.0, event.lat, event.lon + + if haz_lat is None or haz_lon is None: + return # cannot correlate without a location + + effective_mi = radius_mi + (sub.buffer_mi or cfg.default_buffer_mi) + + # Snapshot nodes BEFORE any async scheduling. + try: + nodes = self.data_store.get_nodes_by_roles( + set(cfg.monitor_roles), + cfg.position_max_age_hours * 3600, + ) + except Exception: + logger.exception("danger_zone: get_nodes_by_roles failed") + return + + now = time.time() + cooldown_s = cfg.cooldown_minutes * 60 + hazard_title = event.title or event.summary or event.category + + for node in nodes: + if node.latitude is None or node.longitude is None: + continue + dist_mi = haversine_distance(haz_lat, haz_lon, node.latitude, node.longitude) + if dist_mi > effective_mi: + continue + + key = (node.node_num, fam) + last = self._last.get(key) + if last is not None and (now - last) < cooldown_s: + continue + + summary = ( + f"DANGER ZONE: {node.short_name or node.node_id_hex} " + f"({node.role}) {dist_mi:.1f} mi from {hazard_title}" + ) + ev = make_event( + source="danger_zone", + category="infra_danger_zone", + severity="priority", + title=f"Infra node in danger zone: {node.short_name or node.node_id_hex}", + summary=summary, + lat=node.latitude, + lon=node.longitude, + ) + + if cfg.dry_run: + logger.info( + "[danger_zone dry-run] would alert %s (%s) %.1f mi from %s", + node.short_name or node.node_id_hex, node.role, dist_mi, hazard_title, + ) + self._last[key] = now + continue + + if cfg.delivery_type == "none": + self._last[key] = now + continue + + rule = NotificationRuleConfig( + name="danger_zone", + delivery_type=cfg.delivery_type, + broadcast_channel=cfg.broadcast_channel or 0, + node_ids=list(cfg.node_ids), + webhook_url=cfg.webhook_url, + webhook_headers=dict(cfg.webhook_headers), + ) + try: + ch = channels.create_channel(rule, self.connector) + except Exception: + logger.exception("danger_zone: create_channel failed") + continue + + try: + asyncio.create_task(ch.deliver(make_payload_from_event(ev), rule)) + self._last[key] = now + except RuntimeError: + logger.warning( + "danger_zone: no running event loop; skipping async deliver for %s", + node.short_name or node.node_id_hex, + ) diff --git a/work/tests/test_danger_zone_correlator.py b/work/tests/test_danger_zone_correlator.py new file mode 100644 index 0000000..f8b4334 --- /dev/null +++ b/work/tests/test_danger_zone_correlator.py @@ -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 diff --git a/work/tests/test_fire_age_gate.py b/work/tests/test_fire_age_gate.py new file mode 100644 index 0000000..8c22e23 --- /dev/null +++ b/work/tests/test_fire_age_gate.py @@ -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" diff --git a/work/tests/test_wfigs_handler.py b/work/tests/test_wfigs_handler.py index be210f3..115e5e1 100644 --- a/work/tests/test_wfigs_handler.py +++ b/work/tests/test_wfigs_handler.py @@ -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)