mirror of
https://github.com/zvx-echo6/meshai.git
synced 2026-08-26 17:31:34 +00:00
feat(ipaws): add FEMA IPAWS-OPEN civil-alert adapter (disabled by default)
Adds a new native `ipaws` adapter for FEMA IPAWS-OPEN EAS civil alerts. - Two-stage CAP fetch via base_url (direct FEMA or Conduit proxy): Atom index -> per-entry CAP 1.2 documents. - Non-weather civil alerts only (evacuation, Civil Emergency Message, AMBER, 911 outage, law-enforcement, HazMat); NWS/NOAA CAP dropped so weather is never double-broadcast. - Idaho + neighbour statefips scope gate applied before stage-2 fetch. - Own `ipaws_alerts` dedup table (migration v29); reuses the NWS CAP severity + formatter pattern. - Ships enabled=False (no transmit until explicitly enabled). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
1199b3576a
commit
707ce036f0
18 changed files with 1367 additions and 8 deletions
|
|
@ -600,6 +600,22 @@ REGISTRY: dict[tuple[str, str], dict[str, Any]] = {
|
|||
"description": "Maximum characters for a single NWS mesh packet. Budget enforced after all fields are assembled; only the locations town-list is trimmed to fit. main.py overrides this to the live transport max_chars at runtime.",
|
||||
},
|
||||
|
||||
# =================================================================
|
||||
# IPAWS -- FEMA IPAWS-OPEN EAS civil-alert dedup/tombstone tunables
|
||||
# (mirror of the NWS gating knobs; the ipaws decider owns its own
|
||||
# ipaws_alerts table so these are independent of the nws.* values)
|
||||
# =================================================================
|
||||
("ipaws", "tombstone_msgtypes"): {
|
||||
"default": ["Cancel", "Expire"],
|
||||
"type": "json",
|
||||
"description": "CAP msgType values that mark an IPAWS civil alert as gone (suppressed).",
|
||||
},
|
||||
("ipaws", "duplicate_allowed_after_seconds"): {
|
||||
"default": 10800, # 3h, mirrors nws
|
||||
"type": "int",
|
||||
"description": "Allow re-broadcast of the same IPAWS CAP id after this many seconds (dedup-window relaxation; uses an 'Active' prefix past this point).",
|
||||
},
|
||||
|
||||
# =================================================================
|
||||
# AVALANCHE -- 1 setting (min danger level broadcast floor)
|
||||
# =================================================================
|
||||
|
|
@ -700,6 +716,11 @@ ADAPTER_META: dict[str, dict[str, Any]] = {
|
|||
"include_in_llm_context": True,
|
||||
"description": "CAP-formatted severe-weather warnings/watches/advisories.",
|
||||
},
|
||||
"ipaws": {
|
||||
"display_name": "FEMA IPAWS civil alerts",
|
||||
"include_in_llm_context": True,
|
||||
"description": "IPAWS-OPEN EAS non-weather civil alerts: evacuation, Civil Emergency Message, AMBER, 911 outage, law-enforcement, HazMat. Ships DISABLED.",
|
||||
},
|
||||
"usgs_quake": {
|
||||
"display_name": "USGS earthquakes",
|
||||
"include_in_llm_context": True,
|
||||
|
|
|
|||
|
|
@ -550,6 +550,48 @@ class SatpassConfig(_SourcedFeed):
|
|||
broadcast_lead_seconds: int = 3600
|
||||
|
||||
|
||||
@dataclass
|
||||
class IPAWSConfig(_SourcedFeed):
|
||||
"""FEMA IPAWS-OPEN EAS civil-alert feed settings.
|
||||
|
||||
Ingests the public IPAWS-OPEN Atom index + per-entry CAP 1.2 documents and
|
||||
broadcasts NON-weather civil emergency alerts (evacuation orders, Civil
|
||||
Emergency Messages, AMBER alerts, 911 outages, law-enforcement warnings,
|
||||
HazMat, shelter-in-place). NWS/NOAA-originated CAP entries are DROPPED so
|
||||
meshai never double-broadcasts weather the `nws` adapter already carries.
|
||||
|
||||
Keyless: the FEMA IPAWS-OPEN EAS REST service needs no auth/API key.
|
||||
|
||||
Two-stage fetch (both honour `base_url`):
|
||||
1. GET {base_url}/feed -> Atom index (~13 rolling national entries)
|
||||
2. GET {base_url}/eas/<id> -> full CAP 1.2 alert, per in-scope entry
|
||||
The Atom <link href> values are ABSOLUTE FEMA URLs; the adapter extracts the
|
||||
trailing ``eas/<id>`` and rebuilds ``{base_url}/eas/<id>`` so stage-2 routes
|
||||
through whatever base_url points at (in prod: the Conduit proxy).
|
||||
"""
|
||||
|
||||
enabled: bool = False
|
||||
tick_seconds: int = 60
|
||||
base_url: str = "https://apps.fema.gov/IPAWSOPEN_EAS_SERVICE/rest"
|
||||
user_agent: str = ""
|
||||
# Coarse region gate, applied BEFORE the stage-2 CAP fetch (the "reduce
|
||||
# load / don't fetch every linked CAP" gate): keep only index entries whose
|
||||
# statefips category term is in this list. Default = Idaho + neighbours
|
||||
# (ID=16, WA=53, OR=41, NV=32, UT=49, WY=56, MT=30).
|
||||
state_fips: list = field(
|
||||
default_factory=lambda: ["16", "53", "41", "32", "49", "56", "30"])
|
||||
# Fine region gate (optional): SAME county codes (6-digit, e.g. "016021").
|
||||
# Empty = accept every alert within the state_fips states.
|
||||
same_codes: list = field(default_factory=list)
|
||||
# Drop NWS/NOAA-originated CAP so we never double-broadcast weather.
|
||||
exclude_weather: bool = True
|
||||
# Sender substrings (case-insensitive) that mark a CAP as weather-sourced.
|
||||
drop_senders: list = field(
|
||||
default_factory=lambda: ["noaa.gov", "nws", "weather.gov"])
|
||||
# Only broadcast status=Actual (skip Test/Exercise/System) when True.
|
||||
status_actual_only: bool = True
|
||||
|
||||
|
||||
@dataclass
|
||||
class CentralConsumerConfig:
|
||||
"""Connection settings for the Central NATS JetStream consumer (v0.4).
|
||||
|
|
@ -596,6 +638,7 @@ class EnvironmentalConfig:
|
|||
wzdx: WZDxConfig = field(default_factory=WZDxConfig)
|
||||
firms: FIRMSConfig = field(default_factory=FIRMSConfig)
|
||||
satpass: SatpassConfig = field(default_factory=SatpassConfig)
|
||||
ipaws: IPAWSConfig = field(default_factory=IPAWSConfig)
|
||||
central: CentralConsumerConfig = field(default_factory=CentralConsumerConfig)
|
||||
geocoder: GeocoderConfig = field(default_factory=GeocoderConfig)
|
||||
|
||||
|
|
@ -1294,6 +1337,8 @@ def _dict_to_dataclass(cls, data: dict):
|
|||
kwargs[key] = _dict_to_dataclass(FIRMSConfig, value)
|
||||
elif key == "satpass" and isinstance(value, dict):
|
||||
kwargs[key] = _dict_to_dataclass(SatpassConfig, value)
|
||||
elif key == "ipaws" and isinstance(value, dict):
|
||||
kwargs[key] = _dict_to_dataclass(IPAWSConfig, value)
|
||||
elif key == "environmental" and isinstance(value, dict):
|
||||
kwargs[key] = _dict_to_dataclass(EnvironmentalConfig, value)
|
||||
elif key == "dashboard" and isinstance(value, dict):
|
||||
|
|
|
|||
|
|
@ -237,6 +237,34 @@ def states_for_bbox(bbox) -> list[str]:
|
|||
return sorted(set(result))
|
||||
|
||||
|
||||
# USPS 2-letter -> 2-digit state FIPS. Covers the same keys as US_STATE_BBOXES
|
||||
# (50 states + DC + territories) so states_for_bbox output maps cleanly to the
|
||||
# statefips tokens the IPAWS Atom feed uses.
|
||||
STATE_ABBR_TO_FIPS: dict[str, str] = {
|
||||
"AL": "01", "AK": "02", "AZ": "04", "AR": "05", "CA": "06", "CO": "08",
|
||||
"CT": "09", "DE": "10", "DC": "11", "FL": "12", "GA": "13", "HI": "15",
|
||||
"ID": "16", "IL": "17", "IN": "18", "IA": "19", "KS": "20", "KY": "21",
|
||||
"LA": "22", "ME": "23", "MD": "24", "MA": "25", "MI": "26", "MN": "27",
|
||||
"MS": "28", "MO": "29", "MT": "30", "NE": "31", "NV": "32", "NH": "33",
|
||||
"NJ": "34", "NM": "35", "NY": "36", "NC": "37", "ND": "38", "OH": "39",
|
||||
"OK": "40", "OR": "41", "PA": "42", "RI": "44", "SC": "45", "SD": "46",
|
||||
"TN": "47", "TX": "48", "UT": "49", "VT": "50", "VA": "51", "WA": "53",
|
||||
"WV": "54", "WI": "55", "WY": "56", "AS": "60", "GU": "66", "MP": "69",
|
||||
"PR": "72", "VI": "78",
|
||||
}
|
||||
|
||||
|
||||
def state_fips_for_bbox(bbox) -> list[str]:
|
||||
"""Return sorted 2-digit state FIPS codes whose bbox intersects the coverage
|
||||
bbox (states_for_bbox mapped through STATE_ABBR_TO_FIPS). Used by the IPAWS
|
||||
adapter's coarse region gate."""
|
||||
return sorted(
|
||||
STATE_ABBR_TO_FIPS[c]
|
||||
for c in states_for_bbox(bbox)
|
||||
if c in STATE_ABBR_TO_FIPS
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Avalanche center bbox table
|
||||
# ---------------------------------------------------------------------------
|
||||
|
|
@ -372,6 +400,12 @@ def resolve_adapter_coverage(
|
|||
"bbox": bbox,
|
||||
}
|
||||
|
||||
if adapter == "ipaws":
|
||||
return {
|
||||
"state_fips": state_fips_for_bbox(bbox),
|
||||
"bbox": bbox,
|
||||
}
|
||||
|
||||
if adapter in ("usgs_quake", "firms", "roads511", "usgs"):
|
||||
return {"bbox": bbox}
|
||||
|
||||
|
|
|
|||
485
work/meshai/env/ipaws.py
vendored
Normal file
485
work/meshai/env/ipaws.py
vendored
Normal file
|
|
@ -0,0 +1,485 @@
|
|||
"""FEMA IPAWS-OPEN EAS civil-alert adapter.
|
||||
|
||||
Two-stage public CAP pipeline (keyless, no auth):
|
||||
|
||||
Stage 1 GET {base_url}/feed -> Atom index (~13 rolling national
|
||||
entries). Each <entry> carries a statefips <category> and a <link>
|
||||
to the full CAP document.
|
||||
Stage 2 GET {base_url}/eas/<id> -> full CAP 1.2 <alert>, fetched ONLY for
|
||||
entries that pass the coarse statefips gate (load reduction).
|
||||
|
||||
Scope (approved behaviour):
|
||||
* Coarse region: keep only entries whose statefips term is in
|
||||
``config.state_fips`` — applied BEFORE the stage-2 fetch.
|
||||
* Fine region: optional ``config.same_codes`` SAME county filter.
|
||||
* Non-weather only: DROP NWS/NOAA-originated CAP (``exclude_weather``) so we
|
||||
never double-broadcast what the ``nws`` adapter already carries.
|
||||
* Language: en-US <info> blocks only (skip es-US).
|
||||
* Status: only ``status=Actual`` when ``status_actual_only`` (default True).
|
||||
* Severity: mapped via the SHARED ``env.nws.map_cap_severity`` (Extreme ->
|
||||
immediate, Severe -> priority, else routine).
|
||||
|
||||
The adapter is duck-typed identically to ``NWSAlertsAdapter`` (tick/_fetch/
|
||||
get_events/to_event/health_status) and emits the canonical CAP ``data`` dict so
|
||||
the formatter+gater architecture (formatters/ipaws.py, gating/ipaws.py) operates
|
||||
on it exactly like NWS events. Its own dedup table is ``ipaws_alerts`` (NOT the
|
||||
NWS ``nws_alerts`` table).
|
||||
"""
|
||||
|
||||
import logging
|
||||
import time
|
||||
import xml.etree.ElementTree as ET
|
||||
from datetime import datetime
|
||||
from typing import TYPE_CHECKING, Optional
|
||||
from urllib.error import HTTPError, URLError
|
||||
from urllib.request import Request, urlopen
|
||||
|
||||
from meshai.env.nws import _cfg_str, map_cap_severity
|
||||
from meshai.notifications.events import Event, make_event
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from ..config import IPAWSConfig
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
DEFAULT_BASE_URL = "https://apps.fema.gov/IPAWSOPEN_EAS_SERVICE/rest"
|
||||
|
||||
_ATOM_NS = "http://www.w3.org/2005/Atom"
|
||||
_CAP_NS = "urn:oasis:names:tc:emergency:cap:1.2"
|
||||
|
||||
# ── SAME eventCode -> meshai emergency category ──────────────────────────────
|
||||
# Civil / public-safety EAS event codes we broadcast. Weather SAME codes are
|
||||
# handled by the nws adapter and are additionally dropped by the weather-sender
|
||||
# gate, so they are intentionally NOT mapped here.
|
||||
_SAME_CATEGORY = {
|
||||
"EVI": "emergency_evacuation", # Evacuation Immediate
|
||||
"EVA": "emergency_evacuation", # Evacuation (Watch)
|
||||
"CEM": "emergency_civil", # Civil Emergency Message
|
||||
"CDW": "emergency_civil", # Civil Danger Warning
|
||||
"CAE": "emergency_amber", # Child Abduction Emergency (AMBER)
|
||||
"LAE": "emergency_amber", # Local Area Emergency (child/other) — treat as amber-ish
|
||||
"LEW": "emergency_law", # Law Enforcement Warning
|
||||
"SPW": "emergency_law", # Shelter in Place Warning
|
||||
"TOE": "emergency_911_outage", # 911 Telephone Outage Emergency
|
||||
"HMW": "emergency_hazmat", # Hazardous Materials Warning
|
||||
"NUW": "emergency_hazmat", # Nuclear Power Plant Warning
|
||||
"RHW": "emergency_hazmat", # Radiological Hazard Warning
|
||||
"FRW": "emergency_hazmat", # Fire Warning
|
||||
}
|
||||
|
||||
# Keyword fallback on the CAP <event> string when the SAME code is unknown.
|
||||
_EVENT_KEYWORD_CATEGORY = [
|
||||
("evacuat", "emergency_evacuation"),
|
||||
("amber", "emergency_amber"),
|
||||
("child abduction", "emergency_amber"),
|
||||
("shelter in place", "emergency_law"),
|
||||
("law enforcement", "emergency_law"),
|
||||
("911", "emergency_911_outage"),
|
||||
("telephone outage", "emergency_911_outage"),
|
||||
("hazardous material", "emergency_hazmat"),
|
||||
("hazmat", "emergency_hazmat"),
|
||||
("radiolog", "emergency_hazmat"),
|
||||
("nuclear", "emergency_hazmat"),
|
||||
("civil danger", "emergency_civil"),
|
||||
("civil emergency", "emergency_civil"),
|
||||
]
|
||||
|
||||
_DEFAULT_CATEGORY = "emergency_civil"
|
||||
|
||||
|
||||
def _norm_fips(value) -> str:
|
||||
"""Normalise a state FIPS token to a zero-padded 2-char string.
|
||||
|
||||
'16' -> '16', '6' -> '06', 6 -> '06'. Non-numeric tokens are returned
|
||||
stripped so an odd upstream value still compares by identity.
|
||||
"""
|
||||
s = str(value).strip()
|
||||
try:
|
||||
return str(int(s)).zfill(2)
|
||||
except (TypeError, ValueError):
|
||||
return s
|
||||
|
||||
|
||||
class IPAWSAlertsAdapter:
|
||||
"""FEMA IPAWS-OPEN EAS civil alerts — two-stage CAP poller."""
|
||||
|
||||
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"
|
||||
|
||||
# Coarse region scope: prefer the universal-coverage-derived state_fips
|
||||
# (bbox -> states), else the adapter's own config list.
|
||||
derived = None
|
||||
if coverage is not None:
|
||||
derived = coverage.get("state_fips")
|
||||
cfg_states = list(getattr(config, "state_fips", None) or [])
|
||||
states = derived if derived else cfg_states
|
||||
self._state_fips = {_norm_fips(s) for s in (states or [])}
|
||||
|
||||
self._same_codes = {str(c).strip() for c in (getattr(config, "same_codes", None) or [])}
|
||||
self._exclude_weather = bool(getattr(config, "exclude_weather", True))
|
||||
self._drop_senders = [
|
||||
str(s).strip().lower()
|
||||
for s in (getattr(config, "drop_senders", None) or [])
|
||||
if str(s).strip()
|
||||
]
|
||||
self._status_actual_only = bool(getattr(config, "status_actual_only", True))
|
||||
|
||||
self._tick_interval = getattr(config, "tick_seconds", None) or 60
|
||||
self._last_tick = 0.0
|
||||
self._events = []
|
||||
self._consecutive_errors = 0
|
||||
self._last_error = None
|
||||
self._backoff_until = 0.0
|
||||
self._is_loaded = False
|
||||
|
||||
# ── Polling ──────────────────────────────────────────────────────────────
|
||||
|
||||
def tick(self) -> bool:
|
||||
"""Execute one polling tick (self-throttled by tick_seconds)."""
|
||||
now = time.time()
|
||||
if now < self._backoff_until:
|
||||
return False
|
||||
if now - self._last_tick < self._tick_interval:
|
||||
return False
|
||||
self._last_tick = now
|
||||
return self._fetch()
|
||||
|
||||
def _get(self, url: str) -> bytes:
|
||||
headers = {"User-Agent": self._user_agent, "Accept": "application/xml"}
|
||||
req = Request(url, headers=headers)
|
||||
with urlopen(req, timeout=15) as resp:
|
||||
return resp.read()
|
||||
|
||||
def _stage2_url(self, link_href: str) -> Optional[str]:
|
||||
"""Rebuild a base_url-relative stage-2 URL from an absolute FEMA link.
|
||||
|
||||
The Atom <link href> is an ABSOLUTE FEMA URL; we extract the trailing
|
||||
``eas/<id>`` and rebuild ``{base_url}/eas/<id>`` so the fetch routes
|
||||
through whatever base_url points at (direct FEMA or the Conduit proxy).
|
||||
Returns None if the href carries no ``eas/<id>`` segment.
|
||||
"""
|
||||
if not link_href:
|
||||
return None
|
||||
marker = "/eas/"
|
||||
idx = link_href.rfind(marker)
|
||||
if idx == -1:
|
||||
# Some feeds use a bare "eas/<id>" without a leading slash.
|
||||
if link_href.startswith("eas/"):
|
||||
return f"{self._base_url}/{link_href}"
|
||||
return None
|
||||
tail = link_href[idx + 1:] # "eas/<id>"
|
||||
return f"{self._base_url}/{tail}"
|
||||
|
||||
def _fetch(self) -> bool:
|
||||
"""Fetch + filter the IPAWS feed. Returns True if the active set changed."""
|
||||
# ── Stage 1: Atom index ──────────────────────────────────────────────
|
||||
try:
|
||||
raw = self._get(f"{self._base_url}/feed")
|
||||
except HTTPError as e:
|
||||
if e.code == 429:
|
||||
self._backoff_until = time.time() + 5
|
||||
logger.warning("IPAWS rate limited, backing off 5s")
|
||||
else:
|
||||
logger.warning("IPAWS HTTP error: %s", e.code)
|
||||
self._last_error = f"HTTP {e.code}"
|
||||
self._consecutive_errors += 1
|
||||
return False
|
||||
except URLError as e:
|
||||
logger.warning("IPAWS connection error: %s", e.reason)
|
||||
self._last_error = str(e.reason)
|
||||
self._consecutive_errors += 1
|
||||
return False
|
||||
except Exception as e: # noqa: BLE001
|
||||
logger.warning("IPAWS feed fetch error: %s", e)
|
||||
self._last_error = str(e)
|
||||
self._consecutive_errors += 1
|
||||
return False
|
||||
|
||||
try:
|
||||
feed = ET.fromstring(raw)
|
||||
except ET.ParseError as e:
|
||||
logger.warning("IPAWS feed parse error: %s", e)
|
||||
self._last_error = f"feed parse: {e}"
|
||||
self._consecutive_errors += 1
|
||||
return False
|
||||
|
||||
new_events = []
|
||||
for entry in feed.findall(f"{{{_ATOM_NS}}}entry"):
|
||||
statefips = None
|
||||
for cat in entry.findall(f"{{{_ATOM_NS}}}category"):
|
||||
if cat.get("label") == "statefips":
|
||||
statefips = cat.get("term")
|
||||
break
|
||||
# ── Coarse region gate (BEFORE stage-2 fetch) ────────────────────
|
||||
if self._state_fips and _norm_fips(statefips) not in self._state_fips:
|
||||
continue
|
||||
|
||||
link_el = entry.find(f"{{{_ATOM_NS}}}link")
|
||||
href = link_el.get("href") if link_el is not None else None
|
||||
cap_url = self._stage2_url(href)
|
||||
if not cap_url:
|
||||
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)
|
||||
continue
|
||||
|
||||
parsed = self._parse_cap(cap_raw, statefips)
|
||||
if parsed:
|
||||
new_events.append(parsed)
|
||||
|
||||
# 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}
|
||||
changed = old_ids != new_ids
|
||||
|
||||
self._events = new_events
|
||||
self._consecutive_errors = 0
|
||||
self._last_error = None
|
||||
self._is_loaded = True
|
||||
if changed:
|
||||
logger.info("IPAWS alerts updated: %d active", len(new_events))
|
||||
return changed
|
||||
|
||||
# ── CAP parsing / filtering ──────────────────────────────────────────────
|
||||
|
||||
def _parse_cap(self, cap_raw: bytes, statefips: Optional[str]) -> Optional[dict]:
|
||||
"""Parse a CAP 1.2 <alert> into an internal event dict, applying the
|
||||
status / weather-sender / language / SAME filters. Returns None when the
|
||||
alert is filtered out or unparseable."""
|
||||
try:
|
||||
alert = ET.fromstring(cap_raw)
|
||||
except ET.ParseError as e:
|
||||
logger.debug("IPAWS CAP parse error: %s", e)
|
||||
return None
|
||||
|
||||
def _t(parent, tag):
|
||||
el = parent.find(f"{{{_CAP_NS}}}{tag}")
|
||||
return el.text.strip() if el is not None and el.text else ""
|
||||
|
||||
identifier = _t(alert, "identifier")
|
||||
sender = _t(alert, "sender")
|
||||
sent = _t(alert, "sent")
|
||||
status = _t(alert, "status")
|
||||
msg_type = _t(alert, "msgType") or "Alert"
|
||||
|
||||
# ── Status gate ──────────────────────────────────────────────────────
|
||||
if self._status_actual_only and status and status.lower() != "actual":
|
||||
return None
|
||||
|
||||
# ── Weather-sender exclusion (drop NWS/NOAA) ─────────────────────────
|
||||
if self._exclude_weather and sender:
|
||||
s = sender.lower()
|
||||
if any(bad in s for bad in self._drop_senders):
|
||||
return None
|
||||
|
||||
# ── en-US <info> block (skip es-US) ──────────────────────────────────
|
||||
infos = alert.findall(f"{{{_CAP_NS}}}info")
|
||||
info = None
|
||||
for cand in infos:
|
||||
lang_el = cand.find(f"{{{_CAP_NS}}}language")
|
||||
lang = (lang_el.text.strip().lower() if lang_el is not None and lang_el.text
|
||||
else "en-us")
|
||||
if lang.startswith("en"):
|
||||
info = cand
|
||||
break
|
||||
if info is None:
|
||||
return None
|
||||
|
||||
event_type = _t(info, "event") or "Emergency Alert"
|
||||
urgency = _t(info, "urgency")
|
||||
cap_severity = _t(info, "severity") or "Unknown"
|
||||
certainty = _t(info, "certainty")
|
||||
headline = _t(info, "headline")
|
||||
description = _t(info, "description")
|
||||
|
||||
# eventCode SAME value
|
||||
same_value = ""
|
||||
for ec in info.findall(f"{{{_CAP_NS}}}eventCode"):
|
||||
vn = ec.find(f"{{{_CAP_NS}}}valueName")
|
||||
vv = ec.find(f"{{{_CAP_NS}}}value")
|
||||
if vn is not None and vn.text and vn.text.strip().upper() == "SAME":
|
||||
same_value = (vv.text.strip() if vv is not None and vv.text else "")
|
||||
break
|
||||
|
||||
# ── Area: areaDesc + SAME geocodes + optional geometry ───────────────
|
||||
area_descs = []
|
||||
area_same_codes = []
|
||||
geometry = None
|
||||
for area in info.findall(f"{{{_CAP_NS}}}area"):
|
||||
ad = _t(area, "areaDesc")
|
||||
if ad:
|
||||
area_descs.append(ad)
|
||||
for gc in area.findall(f"{{{_CAP_NS}}}geocode"):
|
||||
vn = gc.find(f"{{{_CAP_NS}}}valueName")
|
||||
vv = gc.find(f"{{{_CAP_NS}}}value")
|
||||
if (vn is not None and vn.text and vn.text.strip().upper() == "SAME"
|
||||
and vv is not None and vv.text):
|
||||
area_same_codes.append(vv.text.strip())
|
||||
if geometry is None:
|
||||
geometry = self._polygon_geometry(area)
|
||||
|
||||
area_desc = "; ".join(area_descs)
|
||||
|
||||
# ── Fine region gate (optional SAME county filter) ───────────────────
|
||||
if self._same_codes:
|
||||
if not any(c in self._same_codes for c in area_same_codes):
|
||||
return None
|
||||
|
||||
# Stable per-alert id (also the dedup / group key).
|
||||
event_id = identifier or f"ipaws:{same_value}:{sent}"
|
||||
|
||||
category = self._derive_category(same_value, event_type)
|
||||
centroid = self._centroid(geometry)
|
||||
|
||||
return {
|
||||
"source": "ipaws",
|
||||
"event_id": event_id,
|
||||
"cap_id": event_id,
|
||||
"event_type": event_type,
|
||||
"urgency": urgency,
|
||||
"cap_severity": cap_severity,
|
||||
"certainty": certainty,
|
||||
"sender": sender,
|
||||
"status": status,
|
||||
"msgType": msg_type,
|
||||
"headline": headline,
|
||||
"description": description,
|
||||
"same_code": same_value,
|
||||
"area_desc": area_desc,
|
||||
"area_same_codes": area_same_codes,
|
||||
"statefips": statefips,
|
||||
"category": category,
|
||||
"sent": self._parse_iso(sent),
|
||||
"expires": self._parse_iso(_t(info, "expires")),
|
||||
"geometry": geometry,
|
||||
"lat": centroid[0] if centroid else None,
|
||||
"lon": centroid[1] if centroid else None,
|
||||
"fetched_at": time.time(),
|
||||
}
|
||||
|
||||
def _derive_category(self, same_value: str, event_type: str) -> str:
|
||||
"""Map a CAP SAME eventCode (or the event string) to an emergency_* category."""
|
||||
code = (same_value or "").strip().upper()
|
||||
if code in _SAME_CATEGORY:
|
||||
return _SAME_CATEGORY[code]
|
||||
ev = (event_type or "").lower()
|
||||
for kw, cat in _EVENT_KEYWORD_CATEGORY:
|
||||
if kw in ev:
|
||||
return cat
|
||||
return _DEFAULT_CATEGORY
|
||||
|
||||
@staticmethod
|
||||
def _polygon_geometry(area) -> Optional[dict]:
|
||||
"""Build a GeoJSON Polygon from a CAP <polygon> (space-separated
|
||||
'lat,lon' pairs). Returns None when the area has no polygon."""
|
||||
poly_el = area.find(f"{{{_CAP_NS}}}polygon")
|
||||
if poly_el is None or not (poly_el.text and poly_el.text.strip()):
|
||||
return None
|
||||
ring = []
|
||||
for pair in poly_el.text.strip().split():
|
||||
try:
|
||||
lat_s, lon_s = pair.split(",")
|
||||
ring.append([float(lon_s), float(lat_s)]) # GeoJSON is [lon,lat]
|
||||
except (ValueError, IndexError):
|
||||
continue
|
||||
if len(ring) < 4:
|
||||
return None
|
||||
if ring[0] != ring[-1]:
|
||||
ring.append(ring[0])
|
||||
return {"type": "Polygon", "coordinates": [ring]}
|
||||
|
||||
@staticmethod
|
||||
def _centroid(geometry: Optional[dict]) -> Optional[tuple]:
|
||||
"""Best-effort (lat, lon) centroid of a GeoJSON Polygon (or None)."""
|
||||
if not geometry or geometry.get("type") != "Polygon":
|
||||
return None
|
||||
coords = geometry.get("coordinates") or []
|
||||
if not coords or not coords[0]:
|
||||
return None
|
||||
ring = coords[0]
|
||||
lats = [c[1] for c in ring]
|
||||
lons = [c[0] for c in ring]
|
||||
if not lats:
|
||||
return None
|
||||
return (sum(lats) / len(lats), sum(lons) / len(lons))
|
||||
|
||||
@staticmethod
|
||||
def _parse_iso(iso_str: str) -> float:
|
||||
"""Parse a CAP/ISO-8601 timestamp to epoch float (0.0 on failure)."""
|
||||
if not iso_str:
|
||||
return 0.0
|
||||
try:
|
||||
s = iso_str.strip()
|
||||
if s.endswith("Z"):
|
||||
s = s[:-1] + "+00:00"
|
||||
return datetime.fromisoformat(s).timestamp()
|
||||
except Exception: # noqa: BLE001
|
||||
return 0.0
|
||||
|
||||
# ── Event emission ───────────────────────────────────────────────────────
|
||||
|
||||
def to_event(self, raw: dict) -> Event:
|
||||
"""Convert an internal event dict to a pipeline Event carrying the
|
||||
canonical CAP ``data`` dict (same shape the NWS path emits), so
|
||||
formatters/ipaws.py + gating/ipaws.py render/gate it identically."""
|
||||
event_type = raw.get("event_type", "Emergency Alert")
|
||||
category = raw.get("category") or _DEFAULT_CATEGORY
|
||||
severity = map_cap_severity(raw.get("cap_severity", "Unknown"))
|
||||
area_desc = raw.get("area_desc", "")
|
||||
|
||||
canonical = {
|
||||
"cap_id": raw.get("cap_id") or raw.get("event_id", ""),
|
||||
"event": event_type,
|
||||
"same_code": raw.get("same_code", ""),
|
||||
"cap_severity": raw.get("cap_severity", "Unknown"),
|
||||
"certainty": raw.get("certainty", ""),
|
||||
"urgency": raw.get("urgency", ""),
|
||||
"sender": raw.get("sender", ""),
|
||||
"expires_at": raw.get("expires") or None,
|
||||
"area_desc": area_desc,
|
||||
"geocoder": {"city": None, "county": area_desc, "state": None},
|
||||
"description": raw.get("description", ""),
|
||||
"parameters": {},
|
||||
"msgType": raw.get("msgType", "Alert"),
|
||||
"references": [],
|
||||
"category": category,
|
||||
"headline": raw.get("headline", ""),
|
||||
"geometry": raw.get("geometry"),
|
||||
}
|
||||
|
||||
return make_event(
|
||||
source="ipaws",
|
||||
category=category,
|
||||
severity=severity,
|
||||
title=raw.get("headline") or event_type,
|
||||
summary=raw.get("headline", ""),
|
||||
body=raw.get("description", ""),
|
||||
effective=raw.get("sent") or None,
|
||||
expires=raw.get("expires") or None,
|
||||
lat=raw.get("lat"),
|
||||
lon=raw.get("lon"),
|
||||
group_key=raw.get("event_id", ""),
|
||||
inhibit_keys=[],
|
||||
data=canonical,
|
||||
)
|
||||
|
||||
def get_events(self) -> list:
|
||||
"""Return the current active events."""
|
||||
return self._events
|
||||
|
||||
@property
|
||||
def health_status(self) -> dict:
|
||||
"""Adapter health snapshot (same shape as the NWS adapter)."""
|
||||
return {
|
||||
"source": "ipaws",
|
||||
"is_loaded": self._is_loaded,
|
||||
"last_error": str(self._last_error) if self._last_error else None,
|
||||
"consecutive_errors": self._consecutive_errors,
|
||||
"event_count": len(self._events),
|
||||
"last_fetch": self._last_tick,
|
||||
}
|
||||
32
work/meshai/env/nws.py
vendored
32
work/meshai/env/nws.py
vendored
|
|
@ -25,6 +25,24 @@ def _cfg_str(config, attr: str, default: str) -> str:
|
|||
return value if isinstance(value, str) and value else default
|
||||
|
||||
|
||||
def map_cap_severity(cap_severity: str) -> str:
|
||||
"""Map a CAP severity string to meshai's 3-level system.
|
||||
|
||||
Extreme -> immediate; Severe/Warning -> priority; everything else
|
||||
(Moderate, Minor, Unknown) -> routine. Case-insensitive.
|
||||
|
||||
Shared by the NWS adapter (``NWSAlertsAdapter._map_nws_severity``) and the
|
||||
IPAWS civil-alert adapter (``env/ipaws.py``) so both map CAP severity
|
||||
identically — do NOT reimplement this mapping elsewhere.
|
||||
"""
|
||||
sev = (cap_severity or "").lower()
|
||||
if sev == "extreme":
|
||||
return "immediate"
|
||||
if sev in ("severe", "warning"):
|
||||
return "priority"
|
||||
return "routine"
|
||||
|
||||
|
||||
class NWSAlertsAdapter:
|
||||
"""NWS Active Alerts -- polls api.weather.gov"""
|
||||
|
||||
|
|
@ -57,13 +75,13 @@ class NWSAlertsAdapter:
|
|||
|
||||
|
||||
def _map_nws_severity(self, nws_severity: str) -> str:
|
||||
"""Map NWS severity to 3-level system."""
|
||||
if nws_severity == "extreme":
|
||||
return "immediate"
|
||||
elif nws_severity in ("severe", "warning"):
|
||||
return "priority"
|
||||
else: # moderate, minor, unknown
|
||||
return "routine"
|
||||
"""Map NWS severity to 3-level system.
|
||||
|
||||
Thin wrapper over the shared module-level ``map_cap_severity`` (kept as
|
||||
a method for the existing call sites/tests; behaviour unchanged). The
|
||||
NWS adapter already lower-cases severity before calling this.
|
||||
"""
|
||||
return map_cap_severity(nws_severity)
|
||||
|
||||
def _derive_category(self, event_type: str) -> str:
|
||||
"""Derive notification category from NWS event type suffix.
|
||||
|
|
|
|||
2
work/meshai/env/store.py
vendored
2
work/meshai/env/store.py
vendored
|
|
@ -234,6 +234,8 @@ class EnvironmentalStore:
|
|||
lambda cfg: (cfg, self._coverage_for("roads511"))),
|
||||
("wzdx", "wzdx", ".wzdx", "WZDxAdapter",
|
||||
lambda cfg: (cfg, self._coverage_for("wzdx"))),
|
||||
("ipaws", "ipaws", ".ipaws", "IPAWSAlertsAdapter",
|
||||
lambda cfg: (cfg, self._coverage_for("ipaws"))),
|
||||
# Native satpass TLE fetcher (storage-only: populates sat_tles,
|
||||
# emits no events). Gated on satpass.feed_source=="native" like
|
||||
# the rest.
|
||||
|
|
|
|||
|
|
@ -33,6 +33,11 @@ VALID_TOGGLES = frozenset({
|
|||
"seismic",
|
||||
"tracking",
|
||||
"satpass",
|
||||
# Civil / public-safety alerts from FEMA IPAWS-OPEN (env/ipaws.py):
|
||||
# evacuation orders, Civil Emergency Messages, AMBER, 911 outages,
|
||||
# law-enforcement warnings, HazMat. NON-weather (weather stays on the
|
||||
# `weather` toggle via the nws adapter).
|
||||
"emergency",
|
||||
})
|
||||
|
||||
|
||||
|
|
@ -116,6 +121,8 @@ _TOGGLE_PREFIX_FALLBACK = [
|
|||
("rf_", "rf_propagation"),
|
||||
("avalanche", "avalanche"),
|
||||
("sat", "satpass"),
|
||||
# IPAWS civil-alert categories all share the emergency_ prefix.
|
||||
("emergency", "emergency"),
|
||||
]
|
||||
|
||||
|
||||
|
|
@ -597,6 +604,52 @@ ALERT_CATEGORIES = {
|
|||
"example_message": "🛰️ ISS Pass — 75° max\nAOS 19:32 · LOS 19:38\nBoise · NW→SE",
|
||||
"toggle": "satpass",
|
||||
},
|
||||
|
||||
# Civil / public-safety alerts (FEMA IPAWS-OPEN EAS, env/ipaws.py).
|
||||
# NON-weather only: weather CAP is dropped by the adapter's weather-sender
|
||||
# gate and stays on the `weather` toggle via the nws adapter.
|
||||
"emergency_evacuation": {
|
||||
"name": "Evacuation Order",
|
||||
"description": "IPAWS evacuation order/warning (SAME EVI/EVA) — leave the area now or prepare to.",
|
||||
"default_severity": "immediate",
|
||||
"example_message": "🚨 Evacuation Immediate: Blaine County — leave now via ID-75 north. Wildfire threat to structures.",
|
||||
"toggle": "emergency",
|
||||
},
|
||||
"emergency_civil": {
|
||||
"name": "Civil Emergency",
|
||||
"description": "IPAWS Civil Emergency Message / Civil Danger Warning (SAME CEM/CDW) — significant in-progress or imminent threat to safety.",
|
||||
"default_severity": "priority",
|
||||
"example_message": "⚠️ Civil Emergency: Ada County — boil-water order in effect for the city of Kuna until further notice.",
|
||||
"toggle": "emergency",
|
||||
},
|
||||
"emergency_amber": {
|
||||
"name": "AMBER Alert",
|
||||
"description": "IPAWS Child Abduction Emergency / AMBER alert (SAME CAE/LAE).",
|
||||
"default_severity": "immediate",
|
||||
"example_message": "🚨 AMBER Alert: Canyon County — silver Honda Civic, plate 1A-23456. Call 911 with sightings.",
|
||||
"toggle": "emergency",
|
||||
},
|
||||
"emergency_law": {
|
||||
"name": "Law Enforcement / Shelter-in-Place",
|
||||
"description": "IPAWS Law Enforcement Warning or Shelter in Place Warning (SAME LEW/SPW).",
|
||||
"default_severity": "priority",
|
||||
"example_message": "⚠️ Shelter in Place: Twin Falls — active police incident near Blue Lakes Blvd. Stay indoors, lock doors.",
|
||||
"toggle": "emergency",
|
||||
},
|
||||
"emergency_911_outage": {
|
||||
"name": "911 Outage",
|
||||
"description": "IPAWS 911 Telephone Outage Emergency (SAME TOE) — 911 unreachable; use the published alternate number.",
|
||||
"default_severity": "priority",
|
||||
"example_message": "⚠️ 911 Outage: Jerome County — 911 down. For emergencies call (208) 555-0111 until restored.",
|
||||
"toggle": "emergency",
|
||||
},
|
||||
"emergency_hazmat": {
|
||||
"name": "Hazardous Materials",
|
||||
"description": "IPAWS Hazardous Materials / Radiological / Nuclear / Fire Warning (SAME HMW/RHW/NUW/FRW).",
|
||||
"default_severity": "immediate",
|
||||
"example_message": "🚨 HazMat Warning: Nampa — chlorine leak near rail yard. Evacuate 1-mile radius, avoid downwind areas.",
|
||||
"toggle": "emergency",
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -129,3 +129,13 @@ register("wildfire_growth", _fire_fmt_mod.format)
|
|||
from meshai.notifications.formatters import firms as _firms_fmt_mod # noqa: E402,F401
|
||||
register("wildfire_spotting", _firms_fmt_mod.format)
|
||||
register("wildfire_halted", _firms_fmt_mod.format)
|
||||
|
||||
# IPAWS civil alerts (env/ipaws.py). One headline-forward formatter handles all
|
||||
# six emergency_* categories; event.category selects only the leading emoji.
|
||||
from meshai.notifications.formatters import ipaws as _ipaws_fmt_mod # noqa: E402,F401
|
||||
register("emergency_evacuation", _ipaws_fmt_mod.format)
|
||||
register("emergency_civil", _ipaws_fmt_mod.format)
|
||||
register("emergency_amber", _ipaws_fmt_mod.format)
|
||||
register("emergency_law", _ipaws_fmt_mod.format)
|
||||
register("emergency_911_outage", _ipaws_fmt_mod.format)
|
||||
register("emergency_hazmat", _ipaws_fmt_mod.format)
|
||||
|
|
|
|||
86
work/meshai/notifications/formatters/ipaws.py
Normal file
86
work/meshai/notifications/formatters/ipaws.py
Normal file
|
|
@ -0,0 +1,86 @@
|
|||
"""IPAWS civil-alert formatter.
|
||||
|
||||
Renders NON-weather civil emergency alerts (evacuation, Civil Emergency
|
||||
Message, AMBER, 911 outage, law enforcement, HazMat) from the canonical CAP
|
||||
``event.data`` dict the IPAWS adapter emits (same shape as the NWS path).
|
||||
|
||||
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:
|
||||
|
||||
Line 1: {emoji} {prefix}{event} e.g. "🚨 Evacuation Immediate"
|
||||
Line 2: {area}[ · Until {t} {tz}] areaDesc (first area) + expiry
|
||||
Line 3: {headline} the operator's message
|
||||
|
||||
Reuses ``event.data`` (canonical) + ``_budget.fit_to_budget``; does NOT touch
|
||||
the NWS formatter. ``now`` is a structural seam (not used — expiry is absolute).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import zoneinfo
|
||||
from datetime import datetime
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from meshai.notifications.formatters._budget import fit_to_budget
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from meshai.notifications.events import Event
|
||||
|
||||
|
||||
# Category → leading emoji. Immediate-severity civil hazards get 🚨; the rest ⚠️.
|
||||
_CATEGORY_EMOJI = {
|
||||
"emergency_evacuation": "🚨",
|
||||
"emergency_amber": "🚨",
|
||||
"emergency_hazmat": "🚨",
|
||||
"emergency_911_outage": "⚠️",
|
||||
"emergency_law": "⚠️",
|
||||
"emergency_civil": "⚠️",
|
||||
}
|
||||
|
||||
|
||||
def format(event: "Event", *, now: float, budget: int) -> str:
|
||||
"""Render the IPAWS civil-alert wire string from canonical event.data.
|
||||
|
||||
Args:
|
||||
event: Pipeline Event — reads event.data (canonical CAP schema).
|
||||
now: Frozen-clock epoch (structural seam; expiry is absolute).
|
||||
budget: Mesh-packet character budget.
|
||||
|
||||
Returns:
|
||||
UTF-8 string fitting within *budget* characters.
|
||||
"""
|
||||
d = event.data or {}
|
||||
|
||||
event_type = d.get("event") or "Emergency Alert"
|
||||
area_desc = d.get("area_desc") or ""
|
||||
headline = (d.get("headline") or "").strip()
|
||||
expires_epoch = d.get("expires_at")
|
||||
prefix = d.get("_ipaws_prefix") or ""
|
||||
category = d.get("category") or event.category or "emergency_civil"
|
||||
|
||||
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 2: first area + optional expiry ("Until 4:54 PM MDT")
|
||||
area = (area_desc or "").split(";")[0].strip()
|
||||
if len(area) > 60:
|
||||
cut = area[:60].rsplit(" ", 1)[0] or area[:60]
|
||||
area = cut + "…"
|
||||
time_seg = ""
|
||||
if expires_epoch:
|
||||
tz = zoneinfo.ZoneInfo("America/Boise")
|
||||
exp_local = datetime.fromtimestamp(expires_epoch, tz=tz)
|
||||
time_seg = f"Until {exp_local.strftime('%-I:%M %p %Z')}"
|
||||
if area and time_seg:
|
||||
line2 = f"{area} · {time_seg}"
|
||||
else:
|
||||
line2 = area or time_seg
|
||||
|
||||
# Line 3: the headline (the actual civil message)
|
||||
line3 = headline
|
||||
|
||||
msg = "\n".join(ln for ln in (line1, line2, line3) if ln)
|
||||
return fit_to_budget(msg, budget)
|
||||
|
|
@ -119,3 +119,13 @@ from meshai.notifications.gating import firms as _firms_gate_mod # noqa: E402,F
|
|||
register("wildfire_growth", _firms_gate_mod.decide)
|
||||
register("wildfire_spotting", _firms_gate_mod.decide)
|
||||
register("wildfire_halted", _firms_gate_mod.decide)
|
||||
|
||||
# IPAWS civil alerts (env/ipaws.py). Own dedup table (ipaws_alerts) — mirrors
|
||||
# the NWS decider's first-sighting / Update / dedup-window / tombstone logic.
|
||||
from meshai.notifications.gating import ipaws as _ipaws_gate_mod # noqa: E402,F401
|
||||
register("emergency_evacuation", _ipaws_gate_mod.decide)
|
||||
register("emergency_civil", _ipaws_gate_mod.decide)
|
||||
register("emergency_amber", _ipaws_gate_mod.decide)
|
||||
register("emergency_law", _ipaws_gate_mod.decide)
|
||||
register("emergency_911_outage", _ipaws_gate_mod.decide)
|
||||
register("emergency_hazmat", _ipaws_gate_mod.decide)
|
||||
|
|
|
|||
170
work/meshai/notifications/gating/ipaws.py
Normal file
170
work/meshai/notifications/gating/ipaws.py
Normal file
|
|
@ -0,0 +1,170 @@
|
|||
"""IPAWS civil-alert gating decider.
|
||||
|
||||
Mirrors the NWS decider (gating/nws.py) EXACTLY in shape, but keys on the
|
||||
IPAWS-owned ``ipaws_alerts`` table (NOT ``nws_alerts``) so IPAWS and NWS CAP
|
||||
lifecycles never collide:
|
||||
|
||||
- Tombstone: msgType in {Cancel, Expire} -> suppress
|
||||
- First-sighting (ipaws_alerts row is None): broadcast, prefix=""/"Update"
|
||||
- Cold-start race (row exists, last_broadcast_at IS NULL): broadcast
|
||||
- Dedup-window re-broadcast (>= duplicate_allowed_after_seconds):
|
||||
broadcast, prefix="Active"
|
||||
- Within dedup window: suppress
|
||||
|
||||
decide(data, *, source, now) -> GateResult
|
||||
|
||||
Canonical data schema consumed:
|
||||
cap_id, msgType, references, event, area_desc, geocoder,
|
||||
cap_severity, expires_at, description, category, headline
|
||||
|
||||
Emitted data_patch keys:
|
||||
_ipaws_prefix str — "", "Update", or "Active"
|
||||
|
||||
commit(now: float) -> None:
|
||||
Idempotent UPDATE of last_broadcast_at + first_broadcast_at.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
from meshai.adapter_config import adapter_config
|
||||
from meshai.notifications.gating.base import GateResult
|
||||
from meshai.persistence import get_db
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# ── Helpers ───────────────────────────────────────────────────────────────────
|
||||
|
||||
def _is_update(conn, references: list) -> bool:
|
||||
"""True if any CAP id in `references` was previously broadcast (Update)."""
|
||||
if not references:
|
||||
return False
|
||||
ref_ids = [r["identifier"] for r in references
|
||||
if isinstance(r, dict) and r.get("identifier")]
|
||||
if not ref_ids:
|
||||
return False
|
||||
placeholders = ",".join("?" * len(ref_ids))
|
||||
row = conn.execute(
|
||||
f"SELECT 1 FROM ipaws_alerts WHERE event_id IN ({placeholders}) "
|
||||
"AND last_broadcast_at IS NOT NULL LIMIT 1",
|
||||
ref_ids,
|
||||
).fetchone()
|
||||
return row is not None
|
||||
|
||||
|
||||
def _make_commit(cap_id: str):
|
||||
"""Return an idempotent commit closure that arms last_broadcast_at."""
|
||||
def _commit(committed_at: float) -> None:
|
||||
try:
|
||||
c = get_db()
|
||||
c.execute(
|
||||
"UPDATE ipaws_alerts SET last_broadcast_at=?, "
|
||||
"first_broadcast_at=COALESCE(first_broadcast_at, ?) "
|
||||
"WHERE event_id=?",
|
||||
(int(committed_at), int(committed_at), cap_id),
|
||||
)
|
||||
except Exception:
|
||||
logger.exception("ipaws commit: persistence update failed for %s", cap_id)
|
||||
return _commit
|
||||
|
||||
|
||||
# ── Public API ────────────────────────────────────────────────────────────────
|
||||
|
||||
def decide(data: dict, *, source: str, now: float) -> GateResult:
|
||||
"""Gate + first-sighting decision for IPAWS civil alerts (ipaws_alerts table)."""
|
||||
cap_id = data.get("cap_id")
|
||||
if not cap_id:
|
||||
return GateResult(
|
||||
broadcast=False, lifecycle="suppress",
|
||||
reason="no cap_id in canonical data",
|
||||
)
|
||||
|
||||
msg_type = data.get("msgType") or ""
|
||||
references = data.get("references") or []
|
||||
expires_at = data.get("expires_at")
|
||||
area_desc = data.get("area_desc") or ""
|
||||
cap_severity = data.get("cap_severity") or ""
|
||||
event_type = data.get("event") or ""
|
||||
geocoder = data.get("geocoder") or {}
|
||||
county = geocoder.get("county") or area_desc
|
||||
state = geocoder.get("state") or ""
|
||||
description = data.get("description") or ""
|
||||
headline = data.get("headline") or ""
|
||||
|
||||
# ── Tombstone: Cancel/Expire → suppress ───────────────────────────────────
|
||||
try:
|
||||
tombstone_types = set(adapter_config.ipaws.tombstone_msgtypes)
|
||||
except Exception:
|
||||
tombstone_types = {"Cancel", "Expire"}
|
||||
if msg_type in tombstone_types:
|
||||
return GateResult(
|
||||
broadcast=False, lifecycle="tombstone",
|
||||
reason=f"msgType={msg_type!r} is a tombstone",
|
||||
)
|
||||
|
||||
# ── Persistence ───────────────────────────────────────────────────────────
|
||||
try:
|
||||
conn = get_db()
|
||||
except Exception:
|
||||
logger.exception("ipaws decide: persistence unavailable")
|
||||
return GateResult(
|
||||
broadcast=False, lifecycle="suppress",
|
||||
reason="persistence unavailable",
|
||||
)
|
||||
|
||||
row = conn.execute(
|
||||
"SELECT last_broadcast_at FROM ipaws_alerts WHERE event_id=?",
|
||||
(cap_id,),
|
||||
).fetchone()
|
||||
|
||||
# ── First sighting ────────────────────────────────────────────────────────
|
||||
if row is None:
|
||||
_prefix = "Update" if _is_update(conn, references) else ""
|
||||
conn.execute(
|
||||
"INSERT INTO ipaws_alerts(event_id, alert_type, severity, county, "
|
||||
"state, headline, description, expires_at, first_seen_at, "
|
||||
"last_broadcast_at) VALUES (?,?,?,?,?,?,?,?,?,?)",
|
||||
(cap_id, event_type, cap_severity, county, state,
|
||||
headline, description,
|
||||
int(expires_at) if expires_at is not None else None,
|
||||
int(now), None),
|
||||
)
|
||||
return GateResult(
|
||||
broadcast=True,
|
||||
lifecycle="new",
|
||||
reason=f"first sighting cap_id={cap_id}",
|
||||
data_patch={"_ipaws_prefix": _prefix},
|
||||
commit=_make_commit(cap_id),
|
||||
)
|
||||
|
||||
# ── Cold-start race: row exists but broadcast was previously dropped ───────
|
||||
if row["last_broadcast_at"] is None:
|
||||
_prefix = "Update" if _is_update(conn, references) else ""
|
||||
return GateResult(
|
||||
broadcast=True,
|
||||
lifecycle="cold_start",
|
||||
reason=f"cold-start race cap_id={cap_id}",
|
||||
data_patch={"_ipaws_prefix": _prefix},
|
||||
commit=_make_commit(cap_id),
|
||||
)
|
||||
|
||||
# ── Dedup-window check ────────────────────────────────────────────────────
|
||||
last_bcast = float(row["last_broadcast_at"])
|
||||
try:
|
||||
window_s = int(adapter_config.ipaws.duplicate_allowed_after_seconds)
|
||||
except Exception:
|
||||
window_s = 10800 # 3 hours default
|
||||
if window_s > 0 and (now - last_bcast) >= window_s:
|
||||
return GateResult(
|
||||
broadcast=True,
|
||||
lifecycle="rebroadcast",
|
||||
reason=f"dedup window expired ({window_s}s) for cap_id={cap_id}",
|
||||
data_patch={"_ipaws_prefix": "Active"},
|
||||
commit=_make_commit(cap_id),
|
||||
)
|
||||
|
||||
return GateResult(
|
||||
broadcast=False, lifecycle="suppress",
|
||||
reason=f"within {window_s}s dedup window for cap_id={cap_id}",
|
||||
)
|
||||
31
work/meshai/persistence/migrations/v29.sql
Normal file
31
work/meshai/persistence/migrations/v29.sql
Normal file
|
|
@ -0,0 +1,31 @@
|
|||
-- v29 IPAWS civil-alert dedup table (ipaws adapter).
|
||||
--
|
||||
-- The IPAWS-OPEN EAS adapter (env/ipaws.py) broadcasts NON-weather civil
|
||||
-- alerts (evacuation, Civil Emergency Message, AMBER, 911 outage, law
|
||||
-- enforcement, HazMat). Its gating decider (notifications/gating/ipaws.py)
|
||||
-- mirrors the NWS decider's first-sighting / Update / dedup-window /
|
||||
-- tombstone logic but MUST NOT share the NWS nws_alerts table -- an IPAWS CAP
|
||||
-- id and an NWS CAP id could otherwise collide, and their lifecycles are
|
||||
-- independent. This is the IPAWS-owned equivalent of nws_alerts (v1.sql),
|
||||
-- same column shape so the decider code is a near-verbatim parametrisation.
|
||||
--
|
||||
-- PK is the CAP <identifier> (urn-style), falling back to a synthetic
|
||||
-- ipaws:<same>:<sent> key when a document omits one.
|
||||
-- ============================================================================
|
||||
CREATE TABLE IF NOT EXISTS ipaws_alerts (
|
||||
event_id TEXT PRIMARY KEY,
|
||||
alert_type TEXT,
|
||||
severity TEXT,
|
||||
county TEXT,
|
||||
state TEXT,
|
||||
headline TEXT,
|
||||
description TEXT,
|
||||
expires_at INTEGER,
|
||||
first_seen_at INTEGER NOT NULL,
|
||||
last_broadcast_at INTEGER,
|
||||
-- first_broadcast_at was added to nws_alerts et al. in v11; the ipaws
|
||||
-- decider's commit COALESCEs it, so this fresh table carries it inline.
|
||||
first_broadcast_at REAL
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_ipaws_expires ON ipaws_alerts(expires_at);
|
||||
CREATE INDEX IF NOT EXISTS idx_ipaws_state_severity ON ipaws_alerts(state, severity);
|
||||
1
work/tests/fixtures/ipaws/eas_idaho_cem.xml
vendored
Normal file
1
work/tests/fixtures/ipaws/eas_idaho_cem.xml
vendored
Normal file
File diff suppressed because one or more lines are too long
1
work/tests/fixtures/ipaws/eas_oregon_evi.xml
vendored
Normal file
1
work/tests/fixtures/ipaws/eas_oregon_evi.xml
vendored
Normal file
File diff suppressed because one or more lines are too long
1
work/tests/fixtures/ipaws/eas_weather_noaa.xml
vendored
Normal file
1
work/tests/fixtures/ipaws/eas_weather_noaa.xml
vendored
Normal file
|
|
@ -0,0 +1 @@
|
|||
<?xml version='1.0' encoding='UTF-8'?><alert xmlns="urn:oasis:names:tc:emergency:cap:1.2"><identifier>NWS-IDAHO-SYNTH-0001</identifier><sender>w-nws.webmaster@noaa.gov</sender><sent>2026-07-16T04:00:00-06:00</sent><status>Actual</status><msgType>Alert</msgType><scope>Public</scope><info><language>en-US</language><category>Met</category><event>Severe Thunderstorm Warning</event><urgency>Immediate</urgency><severity>Severe</severity><certainty>Observed</certainty><eventCode><valueName>SAME</valueName><value>SVR</value></eventCode><expires>2026-07-16T05:00:00-06:00</expires><headline>Severe Thunderstorm Warning for Ada County</headline><description>The National Weather Service has issued a severe thunderstorm warning.</description><area><areaDesc>Ada County</areaDesc><geocode><valueName>SAME</valueName><value>016001</value></geocode></area></info></alert>
|
||||
1
work/tests/fixtures/ipaws/feed.xml
vendored
Normal file
1
work/tests/fixtures/ipaws/feed.xml
vendored
Normal file
|
|
@ -0,0 +1 @@
|
|||
<?xml version='1.0' encoding='UTF-8'?><feed xmlns="http://www.w3.org/2005/Atom"><title type="text">IPAWS EAS FEED</title><updated>2026-07-16T05:09:57.311Z</updated><id>https://apps.fema.gov/IPAWSOPEN_EAS_SERVICE/rest/feed</id><entry><title type="text">CEM</title><link href="https://apps.fema.gov/IPAWSOPEN_EAS_SERVICE/rest/eas/300130859542"/><id>https://apps.fema.gov/IPAWSOPEN_EAS_SERVICE/rest/eas/300130859542</id><updated>2026-07-16T04:54:44.000Z</updated><category term="16" label="statefips"/><category term="CEM" label="event"/></entry><entry><title type="text">EVI</title><link href="https://apps.fema.gov/IPAWSOPEN_EAS_SERVICE/rest/eas/300130856756"/><id>https://apps.fema.gov/IPAWSOPEN_EAS_SERVICE/rest/eas/300130856756</id><updated>2026-07-16T03:00:00.000Z</updated><category term="41" label="statefips"/><category term="EVI" label="event"/></entry><entry><title type="text">CEM</title><link href="https://apps.fema.gov/IPAWSOPEN_EAS_SERVICE/rest/eas/999999999999"/><id>https://apps.fema.gov/IPAWSOPEN_EAS_SERVICE/rest/eas/999999999999</id><updated>2026-07-16T02:00:00.000Z</updated><category term="06" label="statefips"/><category term="CEM" label="event"/></entry></feed>
|
||||
|
|
@ -153,7 +153,9 @@ def test_registry_has_no_duplicate_keys():
|
|||
|
||||
|
||||
def test_adapter_meta_at_19(fresh_db):
|
||||
assert len(ADAPTER_META) == 23
|
||||
# Count sentinel — bump when an adapter row is added. 23 -> 24 with the
|
||||
# IPAWS civil-alert adapter (adapter_config/defaults.py ADAPTER_META["ipaws"]).
|
||||
assert len(ADAPTER_META) == 24
|
||||
|
||||
|
||||
# ---------- seed ----------------------------------------------------------
|
||||
|
|
|
|||
388
work/tests/test_adapter_ipaws.py
Normal file
388
work/tests/test_adapter_ipaws.py
Normal file
|
|
@ -0,0 +1,388 @@
|
|||
"""Tests for the FEMA IPAWS-OPEN EAS civil-alert adapter (env/ipaws.py),
|
||||
its gating decider (gating/ipaws.py) and formatter (formatters/ipaws.py).
|
||||
|
||||
Covers: two-stage fetch + base_url rebuild, statefips coarse filter, en-US
|
||||
filter, status filter, NWS/NOAA-sender weather exclusion, SAME same_codes fine
|
||||
filter, severity mapping reuse, category derivation, polygon geometry, to_event
|
||||
canonical dict, the ipaws_alerts dedup/Update/tombstone decider (its OWN table,
|
||||
not nws_alerts), and a dry-run formatter render of a real Idaho alert.
|
||||
|
||||
Fixtures under tests/fixtures/ipaws/ are REAL captured IPAWS documents (Idaho
|
||||
CEM w/ polygon, Oregon EVI evacuation) plus a synthetic index + a synthetic
|
||||
NWS/NOAA weather CAP for the exclusion test.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import pathlib
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from meshai.config import IPAWSConfig
|
||||
from meshai.env.ipaws import IPAWSAlertsAdapter, _norm_fips
|
||||
from meshai.env.nws import map_cap_severity
|
||||
from meshai.notifications.events import Event
|
||||
|
||||
FIX = pathlib.Path(__file__).parent / "fixtures" / "ipaws"
|
||||
|
||||
|
||||
def _fx(name: str) -> bytes:
|
||||
return (FIX / name).read_bytes()
|
||||
|
||||
|
||||
# URL suffix -> fixture bytes. Keyed by the trailing path so the same map works
|
||||
# for direct FEMA and proxied (Conduit) base_urls.
|
||||
_ROUTES = {
|
||||
"/feed": "feed.xml",
|
||||
"/eas/300130859542": "eas_idaho_cem.xml",
|
||||
"/eas/300130856756": "eas_oregon_evi.xml",
|
||||
"/eas/016weather": "eas_weather_noaa.xml",
|
||||
# 999999999999 (CA / statefips 06) intentionally has NO route: it must be
|
||||
# dropped by the coarse statefips gate BEFORE any stage-2 fetch.
|
||||
}
|
||||
|
||||
|
||||
class _FakeResp:
|
||||
def __init__(self, data: bytes):
|
||||
self._data = data
|
||||
|
||||
def read(self):
|
||||
return self._data
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, *a):
|
||||
return False
|
||||
|
||||
|
||||
def _make_urlopen(requested: list):
|
||||
"""Return a fake urlopen that records requested URLs and serves fixtures by
|
||||
trailing-path suffix (raising for any URL not explicitly routed)."""
|
||||
def _urlopen(req, timeout=None):
|
||||
url = req.full_url
|
||||
requested.append(url)
|
||||
for suffix, fname in _ROUTES.items():
|
||||
if url.endswith(suffix):
|
||||
return _FakeResp(_fx(fname))
|
||||
raise AssertionError(f"unexpected stage-2 fetch: {url}")
|
||||
return _urlopen
|
||||
|
||||
|
||||
def _config(**over) -> IPAWSConfig:
|
||||
cfg = IPAWSConfig()
|
||||
for k, v in over.items():
|
||||
setattr(cfg, k, v)
|
||||
return cfg
|
||||
|
||||
|
||||
# ============================================================
|
||||
# helpers / severity / fips
|
||||
# ============================================================
|
||||
|
||||
def test_defaults_disabled_and_native():
|
||||
cfg = IPAWSConfig()
|
||||
assert cfg.enabled is False # HARD CONSTRAINT: ships disabled
|
||||
assert cfg.feed_source == "native"
|
||||
assert cfg.exclude_weather is True
|
||||
assert cfg.status_actual_only is True
|
||||
assert "16" in cfg.state_fips # Idaho in default scope
|
||||
|
||||
|
||||
def test_norm_fips():
|
||||
assert _norm_fips("16") == "16"
|
||||
assert _norm_fips("6") == "06"
|
||||
assert _norm_fips(6) == "06"
|
||||
assert _norm_fips("41") == "41"
|
||||
|
||||
|
||||
def test_severity_mapping_reuses_nws():
|
||||
assert map_cap_severity("Extreme") == "immediate"
|
||||
assert map_cap_severity("Severe") == "priority"
|
||||
assert map_cap_severity("Moderate") == "routine"
|
||||
assert map_cap_severity("Minor") == "routine"
|
||||
assert map_cap_severity("Unknown") == "routine"
|
||||
|
||||
|
||||
# ============================================================
|
||||
# stage-2 URL rebuild (base_url honoured for BOTH stages)
|
||||
# ============================================================
|
||||
|
||||
def test_stage2_url_rebuilds_from_absolute_fema_link():
|
||||
a = IPAWSAlertsAdapter(_config())
|
||||
abs_link = "https://apps.fema.gov/IPAWSOPEN_EAS_SERVICE/rest/eas/300130859542"
|
||||
assert a._stage2_url(abs_link) == (
|
||||
"https://apps.fema.gov/IPAWSOPEN_EAS_SERVICE/rest/eas/300130859542"
|
||||
)
|
||||
|
||||
|
||||
def test_stage2_url_routes_through_proxy_base_url():
|
||||
# In prod base_url points at the Conduit proxy; the absolute FEMA link in
|
||||
# the feed must be rewritten to go through it.
|
||||
a = IPAWSAlertsAdapter(_config(base_url="http://100.64.0.12:8010/up/ipaws"))
|
||||
abs_link = "https://apps.fema.gov/IPAWSOPEN_EAS_SERVICE/rest/eas/300130859542"
|
||||
assert a._stage2_url(abs_link) == "http://100.64.0.12:8010/up/ipaws/eas/300130859542"
|
||||
|
||||
|
||||
def test_stage2_url_none_when_no_eas_segment():
|
||||
a = IPAWSAlertsAdapter(_config())
|
||||
assert a._stage2_url("https://apps.fema.gov/somethingelse") is None
|
||||
assert a._stage2_url("") is None
|
||||
|
||||
|
||||
# ============================================================
|
||||
# category derivation
|
||||
# ============================================================
|
||||
|
||||
def test_derive_category_same_codes():
|
||||
a = IPAWSAlertsAdapter(_config())
|
||||
assert a._derive_category("EVI", "Evacuation Immediate") == "emergency_evacuation"
|
||||
assert a._derive_category("CEM", "Civil Emergency Message") == "emergency_civil"
|
||||
assert a._derive_category("CAE", "Child Abduction Emergency") == "emergency_amber"
|
||||
assert a._derive_category("LEW", "Law Enforcement Warning") == "emergency_law"
|
||||
assert a._derive_category("TOE", "911 Telephone Outage Emergency") == "emergency_911_outage"
|
||||
assert a._derive_category("HMW", "Hazardous Materials Warning") == "emergency_hazmat"
|
||||
|
||||
|
||||
def test_derive_category_keyword_fallback():
|
||||
a = IPAWSAlertsAdapter(_config())
|
||||
# Unknown SAME code -> keyword match on the event string.
|
||||
assert a._derive_category("ZZZ", "Mandatory Evacuation Ordered") == "emergency_evacuation"
|
||||
assert a._derive_category("", "AMBER Alert") == "emergency_amber"
|
||||
assert a._derive_category("", "Something Unclassifiable") == "emergency_civil"
|
||||
|
||||
|
||||
# ============================================================
|
||||
# polygon geometry
|
||||
# ============================================================
|
||||
|
||||
def test_polygon_geometry_from_real_cem():
|
||||
a = IPAWSAlertsAdapter(_config())
|
||||
parsed = a._parse_cap(_fx("eas_idaho_cem.xml"), "16")
|
||||
geom = parsed["geometry"]
|
||||
assert geom is not None
|
||||
assert geom["type"] == "Polygon"
|
||||
ring = geom["coordinates"][0]
|
||||
assert ring[0] == ring[-1] # closed ring
|
||||
# GeoJSON order is [lon, lat]; Boundary County ID is ~ -116, +48.
|
||||
assert ring[0][0] < -115 and ring[0][1] > 48
|
||||
|
||||
|
||||
# ============================================================
|
||||
# CAP parse / filters (via the private single-parse helper)
|
||||
# ============================================================
|
||||
|
||||
def test_status_actual_only_drops_test():
|
||||
a = IPAWSAlertsAdapter(_config())
|
||||
xml = _fx("eas_idaho_cem.xml").replace(b"<status>Actual</status>",
|
||||
b"<status>Test</status>")
|
||||
assert a._parse_cap(xml, "16") is None
|
||||
|
||||
|
||||
def test_weather_sender_excluded():
|
||||
a = IPAWSAlertsAdapter(_config())
|
||||
# NOAA-originated CAP must be dropped so we never double-broadcast weather.
|
||||
assert a._parse_cap(_fx("eas_weather_noaa.xml"), "16") is None
|
||||
|
||||
|
||||
def test_weather_sender_kept_when_exclude_off():
|
||||
a = IPAWSAlertsAdapter(_config(exclude_weather=False))
|
||||
parsed = a._parse_cap(_fx("eas_weather_noaa.xml"), "16")
|
||||
assert parsed is not None
|
||||
assert parsed["sender"].endswith("noaa.gov")
|
||||
|
||||
|
||||
def test_same_codes_fine_filter():
|
||||
# same_codes gate: keep only alerts whose area SAME is in the list.
|
||||
keep = IPAWSAlertsAdapter(_config(same_codes=["016021"]))
|
||||
assert keep._parse_cap(_fx("eas_idaho_cem.xml"), "16") is not None
|
||||
drop = IPAWSAlertsAdapter(_config(same_codes=["016099"]))
|
||||
assert drop._parse_cap(_fx("eas_idaho_cem.xml"), "16") is None
|
||||
|
||||
|
||||
def test_parse_cap_fields_idaho_cem():
|
||||
a = IPAWSAlertsAdapter(_config())
|
||||
p = a._parse_cap(_fx("eas_idaho_cem.xml"), "16")
|
||||
assert p["source"] == "ipaws"
|
||||
assert p["event_id"] == "AS-ID-de78a02d-8bf9-4528-8f84-fd417bbcaa5b"
|
||||
assert p["same_code"] == "CEM"
|
||||
assert p["category"] == "emergency_civil"
|
||||
assert p["cap_severity"] == "Extreme"
|
||||
assert p["msgType"] == "Update"
|
||||
assert "016021" in p["area_same_codes"]
|
||||
assert p["area_desc"] == "Boundary County"
|
||||
assert "evacuation" in p["description"].lower()
|
||||
|
||||
|
||||
# ============================================================
|
||||
# two-stage fetch (mocked HTTP) — coarse statefips gate
|
||||
# ============================================================
|
||||
|
||||
def test_fetch_coarse_statefips_filter_and_base_url():
|
||||
requested = []
|
||||
a = IPAWSAlertsAdapter(_config()) # default state_fips includes 16 & 41, not 06
|
||||
with patch("meshai.env.ipaws.urlopen", _make_urlopen(requested)):
|
||||
changed = a._fetch()
|
||||
assert changed is True
|
||||
events = a.get_events()
|
||||
ids = {e["same_code"] for e in events}
|
||||
assert ids == {"CEM", "EVI"} # Idaho CEM + Oregon EVI kept
|
||||
# CA (statefips 06) was dropped BEFORE stage-2: its eas URL was never fetched.
|
||||
assert not any("999999999999" in u for u in requested)
|
||||
# stage-1 + two in-scope stage-2 fetches only.
|
||||
assert sum(1 for u in requested if "/eas/" in u) == 2
|
||||
|
||||
|
||||
def test_fetch_routes_all_stages_through_proxy_base_url():
|
||||
requested = []
|
||||
a = IPAWSAlertsAdapter(_config(base_url="http://100.64.0.12:8010/up/ipaws"))
|
||||
with patch("meshai.env.ipaws.urlopen", _make_urlopen(requested)):
|
||||
a._fetch()
|
||||
assert all(u.startswith("http://100.64.0.12:8010/up/ipaws") for u in requested)
|
||||
|
||||
|
||||
def test_coverage_derived_state_fips_overrides_config():
|
||||
# A coverage dict (bbox->state_fips) narrows scope; here only Idaho (16),
|
||||
# so the Oregon (41) EVI is dropped by the coarse gate.
|
||||
requested = []
|
||||
a = IPAWSAlertsAdapter(_config(), coverage={"state_fips": ["16"]})
|
||||
with patch("meshai.env.ipaws.urlopen", _make_urlopen(requested)):
|
||||
a._fetch()
|
||||
assert {e["same_code"] for e in a.get_events()} == {"CEM"}
|
||||
|
||||
|
||||
# ============================================================
|
||||
# to_event canonical dict
|
||||
# ============================================================
|
||||
|
||||
def test_to_event_canonical_dict():
|
||||
a = IPAWSAlertsAdapter(_config())
|
||||
raw = a._parse_cap(_fx("eas_oregon_evi.xml"), "41")
|
||||
ev = a.to_event(raw)
|
||||
assert isinstance(ev, Event)
|
||||
assert ev.source == "ipaws"
|
||||
assert ev.category == "emergency_evacuation"
|
||||
assert ev.severity == "immediate" # Extreme -> immediate
|
||||
assert ev.group_key == raw["event_id"]
|
||||
d = ev.data
|
||||
assert d["cap_id"] == raw["event_id"]
|
||||
assert d["same_code"] == "EVI"
|
||||
assert d["category"] == "emergency_evacuation"
|
||||
assert d["geometry"] is not None
|
||||
assert d["geometry"]["type"] == "Polygon"
|
||||
assert d["geocoder"]["county"] == d["area_desc"]
|
||||
|
||||
|
||||
def test_health_status_shape():
|
||||
a = IPAWSAlertsAdapter(_config())
|
||||
hs = a.health_status
|
||||
assert hs["source"] == "ipaws"
|
||||
assert set(hs) >= {"is_loaded", "last_error", "consecutive_errors",
|
||||
"event_count", "last_fetch"}
|
||||
|
||||
|
||||
# ============================================================
|
||||
# gating decider — ipaws_alerts table (NOT nws_alerts)
|
||||
# ============================================================
|
||||
|
||||
def _canonical(**over):
|
||||
base = {
|
||||
"cap_id": "IPAWS-TEST-1",
|
||||
"msgType": "Alert",
|
||||
"references": [],
|
||||
"event": "Civil Emergency Message",
|
||||
"area_desc": "Boundary County",
|
||||
"geocoder": {"city": None, "county": "Boundary County", "state": None},
|
||||
"cap_severity": "Extreme",
|
||||
"expires_at": 1_800_000_000,
|
||||
"description": "test",
|
||||
"category": "emergency_civil",
|
||||
"headline": "Test civil alert",
|
||||
}
|
||||
base.update(over)
|
||||
return base
|
||||
|
||||
|
||||
def test_decider_first_sighting_broadcasts(_isolate_meshai_db):
|
||||
from meshai.notifications.gating.ipaws import decide
|
||||
r = decide(_canonical(), source="ipaws", now=1000.0)
|
||||
assert r.broadcast is True
|
||||
assert r.lifecycle == "new"
|
||||
assert r.data_patch.get("_ipaws_prefix") == ""
|
||||
|
||||
|
||||
def test_decider_dedup_window(_isolate_meshai_db):
|
||||
from meshai.notifications.gating.ipaws import decide
|
||||
r1 = decide(_canonical(), source="ipaws", now=1000.0)
|
||||
r1.commit(1000.0)
|
||||
# within window -> suppress
|
||||
r2 = decide(_canonical(), source="ipaws", now=1000.0 + 60)
|
||||
assert r2.broadcast is False
|
||||
assert r2.lifecycle == "suppress"
|
||||
# past the 3h dedup window -> re-broadcast with Active prefix
|
||||
r3 = decide(_canonical(), source="ipaws", now=1000.0 + 10800 + 1)
|
||||
assert r3.broadcast is True
|
||||
assert r3.data_patch.get("_ipaws_prefix") == "Active"
|
||||
|
||||
|
||||
def test_decider_tombstone_suppresses(_isolate_meshai_db):
|
||||
from meshai.notifications.gating.ipaws import decide
|
||||
for mt in ("Cancel", "Expire"):
|
||||
r = decide(_canonical(msgType=mt), source="ipaws", now=1000.0)
|
||||
assert r.broadcast is False
|
||||
assert r.lifecycle == "tombstone"
|
||||
|
||||
|
||||
def test_decider_uses_own_table_not_nws(_isolate_meshai_db):
|
||||
from meshai.notifications.gating.ipaws import decide
|
||||
from meshai.persistence import get_db
|
||||
decide(_canonical(), source="ipaws", now=1000.0)
|
||||
conn = get_db()
|
||||
ipaws_n = conn.execute("SELECT COUNT(*) FROM ipaws_alerts").fetchone()[0]
|
||||
nws_n = conn.execute("SELECT COUNT(*) FROM nws_alerts").fetchone()[0]
|
||||
assert ipaws_n == 1
|
||||
assert nws_n == 0
|
||||
|
||||
|
||||
# ============================================================
|
||||
# formatter (dry-run render — NO transmit)
|
||||
# ============================================================
|
||||
|
||||
def test_formatter_renders_civil_alert():
|
||||
from meshai.notifications.formatters.ipaws import format as ipaws_format
|
||||
a = IPAWSAlertsAdapter(_config())
|
||||
raw = a._parse_cap(_fx("eas_idaho_cem.xml"), "16")
|
||||
ev = a.to_event(raw)
|
||||
ev.data["_ipaws_prefix"] = ""
|
||||
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 "Boundary County" in wire
|
||||
assert "evacuation" in wire.lower() # headline carried the signal
|
||||
|
||||
|
||||
def test_formatter_evacuation_emoji():
|
||||
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
|
||||
|
||||
|
||||
# ============================================================
|
||||
# coverage bbox -> state_fips
|
||||
# ============================================================
|
||||
|
||||
def test_coverage_state_fips_for_bbox():
|
||||
from meshai.coverage import state_fips_for_bbox, resolve_adapter_coverage
|
||||
# An Idaho bbox must resolve to include FIPS 16.
|
||||
idaho_bbox = [-117.0, 42.0, -111.0, 49.0]
|
||||
fips = state_fips_for_bbox(idaho_bbox)
|
||||
assert "16" in fips
|
||||
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]
|
||||
Loading…
Add table
Add a link
Reference in a new issue