From c4706d3c921c5b4efaffcaf20da09ff94e259f6d Mon Sep 17 00:00:00 2001 From: Matt Johnson Date: Sun, 16 Aug 2026 01:19:46 +0000 Subject: [PATCH] fix(ipaws): route county-only alerts through evac-phase detection Adds meshai.notifications.evac_phase.detect_phase, wired into the IPAWS formatter/adapter, to classify READY/SET/GO evacuation phases from real FEMA IPAWS alert headline/CMAMtext strings. Includes explicit false-positive guards since a wrong GO detection would broadcast an evacuation order that was never issued. --- work/meshai/env/ipaws.py | 72 +++++++++++++- work/meshai/notifications/evac_phase.py | 95 ++++++++++++++++++ work/meshai/notifications/formatters/ipaws.py | 48 +++++++-- work/tests/test_adapter_ipaws.py | 98 ++++++++++++++++++- work/tests/test_evac_phase.py | 76 ++++++++++++++ 5 files changed, 376 insertions(+), 13 deletions(-) create mode 100644 work/meshai/notifications/evac_phase.py create mode 100644 work/tests/test_evac_phase.py diff --git a/work/meshai/env/ipaws.py b/work/meshai/env/ipaws.py index f5db495..e7b2284 100644 --- a/work/meshai/env/ipaws.py +++ b/work/meshai/env/ipaws.py @@ -104,6 +104,13 @@ def _norm_fips(value) -> str: class IPAWSAlertsAdapter: """FEMA IPAWS-OPEN EAS civil alerts — two-stage CAP poller.""" + # Stage-2 retry suppression. 401/403/404/410 are treated as durable (the + # alert body will not become fetchable), so we wait out the feed's rolling + # window; everything else (timeout, 5xx, 429, connection) is transient. + _STAGE2_FORBIDDEN_COOLDOWN = 21600 # 6h + _STAGE2_TRANSIENT_COOLDOWN = 300 # 5m + _STAGE2_FORBIDDEN_CODES = (401, 403, 404, 410) + def __init__(self, config: "IPAWSConfig", coverage: dict = None): self._base_url = _cfg_str(config, "base_url", DEFAULT_BASE_URL).rstrip("/") self._user_agent = getattr(config, "user_agent", "") or "meshai-ipaws/1.0" @@ -133,6 +140,11 @@ class IPAWSAlertsAdapter: self._last_error = None self._backoff_until = 0.0 self._is_loaded = False + # Negative cache: stage-2 CAP URL -> epoch after which a retry is + # allowed. Stops re-hammering FEMA for an alert listed in the feed whose + # detail endpoint keeps failing (e.g. a durable 403 on a COG-restricted + # alert). Pruned to the current feed each pass so it can't grow unbounded. + self._stage2_cooldown = {} # ── Polling ────────────────────────────────────────────────────────────── @@ -205,6 +217,8 @@ class IPAWSAlertsAdapter: self._consecutive_errors += 1 return False + now = time.time() + seen_urls = set() new_events = [] for entry in feed.findall(f"{{{_ATOM_NS}}}entry"): statefips = None @@ -221,18 +235,43 @@ class IPAWSAlertsAdapter: cap_url = self._stage2_url(href) if not cap_url: continue + seen_urls.add(cap_url) + + # ── Negative cache: skip URLs still within their failure cooldown ─ + retry_after = self._stage2_cooldown.get(cap_url) + if retry_after is not None and now < retry_after: + continue # ── Stage 2: full CAP document ─────────────────────────────────── try: cap_raw = self._get(cap_url) - except Exception as e: # noqa: BLE001 - logger.debug("IPAWS stage-2 fetch failed for %s: %s", cap_url, e) + except HTTPError as e: + cooldown = (self._STAGE2_FORBIDDEN_COOLDOWN + if e.code in self._STAGE2_FORBIDDEN_CODES + else self._STAGE2_TRANSIENT_COOLDOWN) + self._stage2_cooldown[cap_url] = now + cooldown + logger.debug("IPAWS stage-2 fetch failed for %s: HTTP %s " + "(cooldown %ds)", cap_url, e.code, cooldown) continue + except Exception as e: # noqa: BLE001 + self._stage2_cooldown[cap_url] = now + self._STAGE2_TRANSIENT_COOLDOWN + logger.debug("IPAWS stage-2 fetch failed for %s: %s (cooldown %ds)", + cap_url, e, self._STAGE2_TRANSIENT_COOLDOWN) + continue + + # Success — clear any stale cooldown for this URL. + self._stage2_cooldown.pop(cap_url, None) parsed = self._parse_cap(cap_raw, statefips) if parsed: new_events.append(parsed) + # Prune the negative cache to URLs still present in the feed, so it + # tracks the rolling window and never grows without bound. + self._stage2_cooldown = { + u: t for u, t in self._stage2_cooldown.items() if u in seen_urls + } + # Change detection on the active identifier set. old_ids = {e["event_id"] for e in self._events} new_ids = {e["event_id"] for e in new_events} @@ -297,6 +336,30 @@ class IPAWSAlertsAdapter: certainty = _t(info, "certainty") headline = _t(info, "headline") description = _t(info, "description") + instruction = _t(info, "instruction") + + # ── blocks: valueName -> value ──────────────────────────── + # Civil alerts carry their real public-facing text here (CMAMtext / + # CMAMlongtext are the agency's own purpose-written WEA copy) — this was + # previously discarded entirely. A repeated valueName (e.g. multiple + # BLOCKCHANNEL entries) becomes a list; defensive against malformed/ + # missing valueName or value children (skipped, never raises). + parameters: dict = {} + for param in info.findall(f"{{{_CAP_NS}}}parameter"): + vn = param.find(f"{{{_CAP_NS}}}valueName") + vv = param.find(f"{{{_CAP_NS}}}value") + name = vn.text.strip() if vn is not None and vn.text else "" + if not name: + continue + value = vv.text.strip() if vv is not None and vv.text else "" + if name in parameters: + existing = parameters[name] + if isinstance(existing, list): + existing.append(value) + else: + parameters[name] = [existing, value] + else: + parameters[name] = value # eventCode SAME value same_value = "" @@ -377,6 +440,8 @@ class IPAWSAlertsAdapter: "msgType": msg_type, "headline": headline, "description": description, + "instruction": instruction, + "parameters": parameters, "same_code": same_value, "area_desc": area_desc, "area_same_codes": area_same_codes, @@ -491,7 +556,8 @@ class IPAWSAlertsAdapter: "area_desc": area_desc, "geocoder": {"city": None, "county": area_desc, "state": None}, "description": raw.get("description", ""), - "parameters": {}, + "instruction": raw.get("instruction", ""), + "parameters": raw.get("parameters") or {}, "msgType": raw.get("msgType", "Alert"), "references": [], "category": category, diff --git a/work/meshai/notifications/evac_phase.py b/work/meshai/notifications/evac_phase.py new file mode 100644 index 0000000..ca92cef --- /dev/null +++ b/work/meshai/notifications/evac_phase.py @@ -0,0 +1,95 @@ +"""Idaho READY / SET / GO evacuation-phase detection from free-text CAP content. + +Real-world FEMA IPAWS civil alerts do NOT carry a machine-readable phase — +CAP ```` does not exist in the wild. The phase (READY / SET / +GO) instead shows up as free text inside ``headline``, ``description``, and +the ``CMAMtext``/``CMAMlongtext`` ```` values, phrased however the +issuing agency happened to write it ("Level 3 GO NOW", "Set to GO", +"Evacuation Warning", ...). This module scans that free text for the phrases +agencies actually use and returns the phase, or ``None`` when nothing +matches — callers must never guess a phase, since a false GO would broadcast +an evacuation order that was never issued. +""" +from __future__ import annotations + +import re + +# ── Strong multi-word phrases (checked case-insensitively) ─────────────────── +# ANY match sets that phase's hit flag. Order within a list is irrelevant — +# the final result is decided purely by GO > SET > READY precedence below, +# never by which phrase or text argument matched first. +_GO_PHRASES = [ + r"\bGO\s+NOW\b", + r"\bLEVEL\s+3\s+GO\b", + r"\bLEVEL\s+3\b", + r"\bLEVEL\s+III\b", + r"\bGO\s+EVACUATION\b", + r"\bIMMEDIATE\s+EVACUATION\b", + r"\bEVACUATE\s+NOW\b", + r"\bEVACUATION\s+ORDER\b", + r"\bSET\s+TO\s+GO\b", # "Set to GO" — the standalone-GO idiom, spelled out +] + +_SET_PHRASES = [ + r"\bLEVEL\s+2\b", + r"\bLEVEL\s+II\b", + r"\bPREPARE\s+TO\s+EVACUATE\b", + r"\bEVACUATION\s+WARNING\b", + r"\bBE\s+READY\s+TO\s+LEAVE\b", +] + +_READY_PHRASES = [ + r"\bLEVEL\s+1\b", + r"\bLEVEL\s+I\b", + r"\bEVACUATION\s+ADVISORY\b", +] + +_PRECEDENCE = ("GO", "SET", "READY") + +_PHRASE_RE = { + "GO": re.compile("|".join(_GO_PHRASES), re.IGNORECASE), + "SET": re.compile("|".join(_SET_PHRASES), re.IGNORECASE), + "READY": re.compile("|".join(_READY_PHRASES), re.IGNORECASE), +} + +# A bare level-word (GO/SET/READY) counts ONLY when it is: +# 1. standalone (word-boundaried) AND written in the exact uppercase form +# (narrative prose never shouts a whole word in caps: "go to the +# fairgrounds", "set to arrive", "via Go Creek Road" all fail this), AND +# 2. accompanied elsewhere in the same text by other alert/evacuation +# vocabulary, so a bare "GO"/"SET"/"READY" floating in unrelated text +# can't fire on its own. +# This is what lets "LEVEL I SET Alert" resolve as SET (word beats numeral — +# err upward) even though "LEVEL I" alone would read as READY. +_STANDALONE_TOKEN_RE = { + "GO": re.compile(r"\bGO\b"), + "SET": re.compile(r"\bSET\b"), + "READY": re.compile(r"\bREADY\b"), +} +_CONTEXT_CUE_RE = re.compile( + r"evacuat|level|alert|status|notice|prepar|leave|order|warning|advisory", + re.IGNORECASE, +) + + +def detect_phase(*texts: "str | None") -> "str | None": + """Scan the given texts for Idaho READY/SET/GO evacuation-phase language. + + All provided texts are combined and scanned together (case-insensitive + for the strong phrases). The HIGHEST phase found wins — GO > SET > READY + — never the first match. Returns None when nothing matches; never + guesses. + """ + combined = "\n".join(t for t in texts if t) + if not combined: + return None + + has_cue = bool(_CONTEXT_CUE_RE.search(combined)) + + for phase in _PRECEDENCE: + if _PHRASE_RE[phase].search(combined): + return phase + if has_cue and _STANDALONE_TOKEN_RE[phase].search(combined): + return phase + + return None diff --git a/work/meshai/notifications/formatters/ipaws.py b/work/meshai/notifications/formatters/ipaws.py index 91a695d..9293214 100644 --- a/work/meshai/notifications/formatters/ipaws.py +++ b/work/meshai/notifications/formatters/ipaws.py @@ -6,11 +6,19 @@ Message, AMBER, 911 outage, law enforcement, HazMat) from the canonical CAP Civil alerts carry their signal in the CAP ``headline`` (a plain human sentence), so — unlike the weather formatter, which parses structured -HAZARD.../IMPACT... blocks — this formatter is headline-forward: +HAZARD.../IMPACT... blocks — this formatter is headline-forward. Real CAP +data has no machine-readable Idaho READY/SET/GO evacuation phase (there is +no ```` in the wild); ``evac_phase.detect_phase`` scans the +headline/CMAMtext/description free text for the phrases agencies actually +use ("Level 3 GO NOW", "Evacuation Warning", ...) so the wire message can +lead with the phase instead of raw CAP jargon: - Line 1: {emoji} {prefix}{event} e.g. "🚨 Evacuation Immediate" + Line 1 (phase detected): {emoji} {prefix}{PHASE} — {action} + e.g. "🚨 GO — Leave now" + Line 1 (no phase found — unchanged fallback): + {emoji} {prefix}{event} e.g. "🚨 Evacuation Immediate" Line 2: {area}[ · Until {t} {tz}] areaDesc (first area) + expiry - Line 3: {headline} the operator's message + Line 3: {CMAMtext or headline} the agency's own public alert text Reuses ``event.data`` (canonical) + ``_budget.fit_to_budget``; does NOT touch the NWS formatter. ``now`` is a structural seam (not used — expiry is absolute). @@ -21,6 +29,7 @@ import zoneinfo from datetime import datetime from typing import TYPE_CHECKING +from meshai.notifications.evac_phase import detect_phase from meshai.notifications.formatters._budget import fit_to_budget if TYPE_CHECKING: @@ -37,6 +46,13 @@ _CATEGORY_EMOJI = { "emergency_civil": "⚠️", } +# Idaho READY/SET/GO — the one-line action tied to each detected phase. +_PHASE_ACTION = { + "READY": "Get prepared", + "SET": "Be ready to leave", + "GO": "Leave now", +} + def format(event: "Event", *, now: float, budget: int) -> str: """Render the IPAWS civil-alert wire string from canonical event.data. @@ -54,6 +70,11 @@ def format(event: "Event", *, now: float, budget: int) -> str: event_type = d.get("event") or "Emergency Alert" area_desc = d.get("area_desc") or "" headline = (d.get("headline") or "").strip() + description = d.get("description") or "" + parameters = d.get("parameters") or {} + cmam_text = parameters.get("CMAMtext") or "" + if isinstance(cmam_text, list): # defensive: a repeated valueName collapses to a list + cmam_text = cmam_text[0] if cmam_text else "" expires_epoch = d.get("expires_at") prefix = d.get("_ipaws_prefix") or "" category = d.get("category") or event.category or "emergency_civil" @@ -61,8 +82,20 @@ def format(event: "Event", *, now: float, budget: int) -> str: emoji = _CATEGORY_EMOJI.get(category, "⚠️") prefix_seg = f"{prefix}: " if prefix else "" - # Line 1: emoji + prefix + event type - line1 = f"{emoji} {prefix_seg}{event_type}" + # Line 1: lead with the Idaho READY/SET/GO phase when the free text + # actually names one; otherwise keep the raw CAP event string (safe + # fallback — never invent a phase that isn't there). + phase = detect_phase(headline, cmam_text, description) + if phase: + # A detected GO always gets the alarm emoji, regardless of CAP + # category — a confirmed "leave now" evacuation must never render + # with the softer category-based ⚠️ (e.g. emergency_civil). + if phase == "GO": + emoji = "🚨" + action = _PHASE_ACTION[phase] + line1 = f"{emoji} {prefix_seg}{phase} — {action}" + else: + line1 = f"{emoji} {prefix_seg}{event_type}" # Line 2: first area + optional expiry ("Until 4:54 PM MDT") area = (area_desc or "").split(";")[0].strip() @@ -79,8 +112,9 @@ def format(event: "Event", *, now: float, budget: int) -> str: else: line2 = area or time_seg - # Line 3: the headline (the actual civil message) - line3 = headline + # Line 3: the agency's own public alert text (CMAMtext — purpose-written + # for WEA, ~90 chars) when present; else fall back to the headline as before. + line3 = cmam_text.strip() or headline msg = "\n".join(ln for ln in (line1, line2, line3) if ln) return fit_to_budget(msg, budget) diff --git a/work/tests/test_adapter_ipaws.py b/work/tests/test_adapter_ipaws.py index 891f060..5009082 100644 --- a/work/tests/test_adapter_ipaws.py +++ b/work/tests/test_adapter_ipaws.py @@ -349,6 +349,12 @@ def test_decider_uses_own_table_not_nws(_isolate_meshai_db): # ============================================================ def test_formatter_renders_civil_alert(): + """Idaho CEM headline ("Wildfire Immediate Evacuation Alert") names the GO + phase explicitly ("Immediate Evacuation"), so line 1 leads with GO instead + of the raw CAP event string — the whole point of this feature. Category + is emergency_civil (normally ⚠️), but a detected GO forces the alarm + emoji regardless of category — a confirmed "leave now" must never render + with the softer warning icon.""" from meshai.notifications.formatters.ipaws import format as ipaws_format a = IPAWSAlertsAdapter(_config()) raw = a._parse_cap(_fx("eas_idaho_cem.xml"), "16") @@ -357,20 +363,73 @@ def test_formatter_renders_civil_alert(): wire = ipaws_format(ev, now=1000.0, budget=200) assert len(wire) <= 200 lines = wire.split("\n") - assert lines[0].startswith("⚠️") # civil -> warning emoji - assert "Civil Emergency Message" in lines[0] + assert lines[0].startswith("🚨") # GO overrides civil's ⚠️ -> alarm emoji + assert "GO" in lines[0] and "Leave now" in lines[0] assert "Boundary County" in wire assert "evacuation" in wire.lower() # headline carried the signal def test_formatter_evacuation_emoji(): + """Oregon EVI has no phase language in its headline, but its CMAMtext + parameter ("Level 3 GO NOW evacuation notice...") does, and that text is + now parsed and preferred for line 3 — so GO must win here too.""" from meshai.notifications.formatters.ipaws import format as ipaws_format a = IPAWSAlertsAdapter(_config()) raw = a._parse_cap(_fx("eas_oregon_evi.xml"), "41") ev = a.to_event(raw) wire = ipaws_format(ev, now=1000.0, budget=200) assert wire.startswith("🚨") # evacuation -> alarm emoji - assert "Evacuation Immediate" in wire + lines = wire.split("\n") + assert "GO" in lines[0] and "Leave now" in lines[0] + assert "Level 3 GO NOW" in wire # CMAMtext carried into line 3 + + +def test_formatter_full_three_line_go_rendering(): + """Full 3-line render for a real GO alert (Oregon EVI): phase-led line 1, + area+expiry line 2, CMAMtext line 3 — exact text, not just substrings.""" + from meshai.notifications.formatters.ipaws import format as ipaws_format + a = IPAWSAlertsAdapter(_config()) + raw = a._parse_cap(_fx("eas_oregon_evi.xml"), "41") + ev = a.to_event(raw) + ev.data["_ipaws_prefix"] = "" + wire = ipaws_format(ev, now=1000.0, budget=200) + lines = wire.split("\n") + assert lines[0] == "🚨 GO — Leave now" + assert lines[1] == "Jackson County · Until 8:50 AM MDT" + assert lines[2] == ( + "Wildfire Alert- Level 3 GO NOW evacuation notice is UPGRADED for JAC-126" + ) + + +def test_formatter_no_phase_fallback_unchanged(): + """When no READY/SET/GO language is present anywhere in headline/CMAMtext/ + description, line 1 keeps the CURRENT raw-event-string behaviour exactly — + the safe fallback this feature must never break.""" + from meshai.notifications.formatters.ipaws import format as ipaws_format + from meshai.notifications.events import make_event + + canonical = _canonical( + event="Civil Emergency Message", + headline="Boil water advisory issued for the district", + description="A water main break has contaminated the supply.", + parameters={}, + expires_at=None, + ) + ev = make_event( + source="ipaws", + category="emergency_civil", + severity="priority", + title=canonical["headline"], + summary=canonical["headline"], + body=canonical["description"], + data=canonical, + ) + ev.data["_ipaws_prefix"] = "" + wire = ipaws_format(ev, now=1000.0, budget=200) + lines = wire.split("\n") + assert lines[0] == "⚠️ Civil Emergency Message" # unchanged fallback (no phase) + assert lines[1] == "Boundary County" + assert lines[2] == "Boil water advisory issued for the district" # ============================================================ @@ -386,3 +445,36 @@ def test_coverage_state_fips_for_bbox(): resolved = resolve_adapter_coverage("ipaws", idaho_bbox, "native") assert "16" in resolved["state_fips"] assert resolved["bbox"] == [round(c, 6) for c in idaho_bbox] + + +# ============================================================ +# stage-2 failure negative cache (no re-hammering FEMA) +# ============================================================ + +def test_stage2_forbidden_is_negative_cached(): + """A 403 on a stage-2 detail URL is remembered so the next poll tick does + NOT re-fetch it, while a sibling URL that succeeds is fetched every pass.""" + from urllib.error import HTTPError + + counts: dict[str, int] = {} + + def _urlopen(req, timeout=None): + url = req.full_url + counts[url] = counts.get(url, 0) + 1 + if url.endswith("/feed"): + return _FakeResp(_fx("feed.xml")) + if url.endswith("/eas/300130859542"): # Idaho CEM -> forbidden + raise HTTPError(url, 403, "Forbidden", {}, None) + if url.endswith("/eas/300130856756"): # Oregon EVI -> ok + return _FakeResp(_fx("eas_oregon_evi.xml")) + raise AssertionError(f"unexpected fetch: {url}") + + a = IPAWSAlertsAdapter(_config()) + with patch("meshai.env.ipaws.urlopen", _urlopen): + a._fetch() + a._fetch() + + cem = next(u for u in counts if u.endswith("/eas/300130859542")) + evi = next(u for u in counts if u.endswith("/eas/300130856756")) + assert counts[cem] == 1 # 403 -> negative-cached, not re-fetched + assert counts[evi] == 2 # success -> fetched on each pass diff --git a/work/tests/test_evac_phase.py b/work/tests/test_evac_phase.py new file mode 100644 index 0000000..7b8fd45 --- /dev/null +++ b/work/tests/test_evac_phase.py @@ -0,0 +1,76 @@ +"""Tests for meshai.notifications.evac_phase.detect_phase. + +Cases are drawn from REAL FEMA IPAWS alert headline/CMAMtext/description +strings (see tests/fixtures/ipaws/) plus explicit false-positive guards, since +a wrong GO detection would broadcast an evacuation order that was never +issued. +""" +from __future__ import annotations + +import pytest + +from meshai.notifications.evac_phase import detect_phase + + +# ============================================================ +# real strings -> expected phase +# ============================================================ + +@pytest.mark.parametrize("text, expected", [ + ("LEVEL 1 READY - Evacuation Status", "READY"), + ("Level 2 Set Alert", "SET"), + ("LEVEL I SET Alert", "SET"), # word beats numeral; err upward + ("Level 3 - Go Now", "GO"), + ("Level 3- Go Now", "GO"), + ("Ohio Gulch GO Evacuation Status", "GO"), + ("Indian Creek Set to GO.", "GO"), # highest wins, NOT SET + ("Immediate Evacuation", "GO"), + ("Prepare to Evacuate", "SET"), + ("BRUSH FIRE", None), + ("Owyhee County Closure Area", None), + ("Endangered Missing Person Alert", None), + ( + "Jackson County Sheriff's Office- Level 3 GO NOW evacuation notice " + "UPGRADED for JAC-126", + "GO", + ), +]) +def test_detect_phase_real_strings(text, expected): + assert detect_phase(text) == expected + + +# ============================================================ +# false-positive guards — bare lowercase / narrative usage must NOT match +# ============================================================ + +@pytest.mark.parametrize("text", [ + "residents should go to the Blaine County Fairgrounds", + "crews are set to arrive by 0600", + "evacuate via Go Creek Road", +]) +def test_detect_phase_false_positive_guards(text): + assert detect_phase(text) is None + + +# ============================================================ +# precedence + multi-arg scanning +# ============================================================ + +def test_precedence_highest_always_wins_regardless_of_arg_order(): + # SET language in the first text, GO language in the second — GO must win. + assert detect_phase("Prepare to Evacuate", "Level 3 - Go Now") == "GO" + # Same phrases, arguments reversed — still GO (order-independent). + assert detect_phase("Level 3 - Go Now", "Prepare to Evacuate") == "GO" + + +def test_detect_phase_scans_all_texts_together(): + # No single text alone carries a phase; combined they do (SET here, + # since "LEVEL 2" is an explicit SET phrase and no GO phrase is present). + assert detect_phase("Jackson County Sheriff's Office", "Level 2 Set Alert") == "SET" + + +def test_detect_phase_none_texts_and_empty_input_are_safe(): + assert detect_phase(None, None) is None + assert detect_phase() is None + assert detect_phase("") is None + assert detect_phase(None, "BRUSH FIRE", None) is None