diff --git a/work/meshai/config.py b/work/meshai/config.py index 7ea02d7..9b9eff0 100644 --- a/work/meshai/config.py +++ b/work/meshai/config.py @@ -449,14 +449,26 @@ class Roads511Config(_SourcedFeed): @dataclass class WZDxConfig(_SourcedFeed): - """WZDx work zone data feed settings.""" + """FHWA WZDx (Work Zone Data Exchange) work-zone feed settings. + + Native adapter discovers state DOT WZDx v4 GeoJSON feeds via the FHWA + WZDx Feed Registry (keyless), filters to `states`, then polls each + matching feed and parses ``work-zone`` road_events into canonical + ``work_zone`` events. Keyless: ``api_key`` is retained but unused. + """ enabled: bool = False - tick_seconds: int = 300 - api_key: str = "" # Supports ${ENV_VAR} - base_url: str = "" # e.g. "https://511.idaho.gov/api/v2" + tick_seconds: int = 300 # per-feed poll interval + api_key: str = "" # unused (keyless); retained for parity / ${ENV_VAR} + base_url: str = "" # optional single-feed override (skips registry when set) endpoints: list = field(default_factory=lambda: ["/get/event"]) - bbox: list = field(default_factory=list) # [west, south, east, north] + bbox: list = field(default_factory=list) # [west, south, east, north] optional filter + # FHWA WZDx Feed Registry (Socrata) — rows describe every state DOT feed. + registry_url: str = ( + "https://datahub.transportation.gov/resource/69qe-yiui.json?$limit=200" + ) + registry_ttl: int = 21600 # re-fetch registry every 6h (it rarely changes) + states: list = field(default_factory=lambda: ["ID"]) # states to include @dataclass diff --git a/work/meshai/env/store.py b/work/meshai/env/store.py index 1d3958d..51d5011 100644 --- a/work/meshai/env/store.py +++ b/work/meshai/env/store.py @@ -48,6 +48,8 @@ class EnvironmentalStore: lambda cfg: (cfg,)) self._register_adapter("roads511", config.roads511, ".roads511", "Roads511Adapter", lambda cfg: (cfg,)) + self._register_adapter("wzdx", config.wzdx, ".wzdx", "WZDxAdapter", + lambda cfg: (cfg,)) # FIRMS needs reference to NIFC adapter for cross-referencing if config.firms.enabled and config.firms.feed_source == "native": @@ -61,7 +63,7 @@ class EnvironmentalStore: logger.warning("Failed to initialize firms adapter: %s", err_msg) self._failed_adapters["firms"] = err_msg - _central = [n for n in ("nws", "swpc", "ducting", "fires", "avalanche", "usgs", "usgs_quake", "traffic", "roads511", "firms") + _central = [n for n in ("nws", "swpc", "ducting", "fires", "avalanche", "usgs", "usgs_quake", "traffic", "roads511", "wzdx", "firms") if getattr(getattr(config, n, None), "feed_source", "native") == "central"] if _central: logger.debug("Adapters sourced from Central (native skipped): %s", _central) diff --git a/work/meshai/env/wzdx.py b/work/meshai/env/wzdx.py new file mode 100644 index 0000000..5e24c7f --- /dev/null +++ b/work/meshai/env/wzdx.py @@ -0,0 +1,510 @@ +"""FHWA WZDx (Work Zone Data Exchange) native adapter. + +Gives meshai a local source for the ``work_zone`` event category, so it no +longer depends on Central for work-zone events. Mirrors the native-adapter +shape used by ``roads511`` / ``traffic``: ``tick()`` polls on an interval and +populates ``self._events``; ``get_events()`` returns them; ``to_event()`` +builds a pipeline ``Event`` with ``category="work_zone"`` and a canonical +``data`` dict; ``health_status`` reports adapter health. + +Discovery + fetch +----------------- +1. Fetch the FHWA WZDx Feed Registry (keyless Socrata JSON) — each row + describes a state DOT feed (``feedname``/``url``/``apiurl``/``state``/ + ``version``/``format``). The registry rarely changes, so it is cached for + ``registry_ttl`` seconds (default 6h). +2. Filter registry rows to the configured ``states`` that publish a GeoJSON + WZDx feed, collect their feed URLs. +3. On each tick, fetch every matching feed (WZDx v4 GeoJSON + ``FeatureCollection``) and parse each ``work-zone`` road_event feature. + +A single ``base_url`` override short-circuits registry discovery (poll that +one feed directly). A down state feed is logged + skipped (never crashes the +adapter). Keyless — no ``api_key`` required. stdlib HTTP only (urllib), same +as the other native adapters. + +Canonical ``work_zone`` data shape +---------------------------------- +Field mapping reuses ``central_normalizer._parse_wzdx_federal`` verbatim so +the wire is identical to what the Central-side parser would have produced +(road, direction, mile posts, folded sub_type, impact, ends_at). ``to_event`` +then converts ``ends_at`` → ``ends_at_epoch`` exactly as the Phase-2 bridge +does (``calendar.timegm`` of the naive datetime) and augments with lat/lon + +a stable ``external_id`` (WZDx ``data_source_id`` + feature id) so the +incident gating decider dedups correctly. +""" + +import calendar +import json +import logging +import os +import time +from datetime import datetime +from typing import TYPE_CHECKING, Optional +from urllib.error import HTTPError, URLError +from urllib.request import Request, urlopen + +from meshai.notifications.events import Event, make_event + +# Reuse the exact WZDx field-mapping logic from the Central normalizer so the +# native path and the Central path produce an identical canonical shape. +from meshai.central_normalizer import _parse_wzdx_federal + +if TYPE_CHECKING: + from ..config import WZDxConfig + +logger = logging.getLogger(__name__) + + +# US state abbreviation ↔ full-name map, so a configured "ID" matches a +# registry row whose ``state`` reads "Idaho" (and vice-versa). +_US_STATES = { + "AL": "alabama", "AK": "alaska", "AZ": "arizona", "AR": "arkansas", + "CA": "california", "CO": "colorado", "CT": "connecticut", "DE": "delaware", + "FL": "florida", "GA": "georgia", "HI": "hawaii", "ID": "idaho", + "IL": "illinois", "IN": "indiana", "IA": "iowa", "KS": "kansas", + "KY": "kentucky", "LA": "louisiana", "ME": "maine", "MD": "maryland", + "MA": "massachusetts", "MI": "michigan", "MN": "minnesota", "MS": "mississippi", + "MO": "missouri", "MT": "montana", "NE": "nebraska", "NV": "nevada", + "NH": "new hampshire", "NJ": "new jersey", "NM": "new mexico", "NY": "new york", + "NC": "north carolina", "ND": "north dakota", "OH": "ohio", "OK": "oklahoma", + "OR": "oregon", "PA": "pennsylvania", "RI": "rhode island", "SC": "south carolina", + "SD": "south dakota", "TN": "tennessee", "TX": "texas", "UT": "utah", + "VT": "vermont", "VA": "virginia", "WA": "washington", "WV": "west virginia", + "WI": "wisconsin", "WY": "wyoming", "DC": "district of columbia", +} + + +class WZDxAdapter: + """FHWA WZDx work-zone polling adapter (native ``work_zone`` source).""" + + def __init__(self, config: "WZDxConfig"): + self._api_key = self._resolve_env(config.api_key or "") # unused (keyless) + self._base_url = (config.base_url or "").strip().rstrip("/") + self._registry_url = (config.registry_url or "").strip() + self._registry_ttl = config.registry_ttl or 21600 + self._tick_interval = config.tick_seconds or 300 + self._states = [str(s).strip() for s in (config.states or ["ID"]) if str(s).strip()] + self._bbox = config.bbox or [] # [west, south, east, north] + + self._last_tick = 0.0 + self._events = [] + self._feed_urls = [] # discovered / overridden feed URLs + self._registry_fetched_at = 0.0 + self._consecutive_errors = 0 + self._last_error = None + self._is_loaded = False + + if not self._base_url and not self._registry_url: + logger.info("WZDx: no base_url and no registry_url configured, adapter idle") + + # ── helpers ──────────────────────────────────────────────────────────── + + def _resolve_env(self, value: str) -> str: + """Resolve ${ENV_VAR} references in value.""" + if value and value.startswith("${") and value.endswith("}"): + return os.environ.get(value[2:-1], "") + return value + + def _wanted_states(self) -> set: + """Lowercased set of accepted state tokens (abbrev + full name).""" + wanted = set() + for s in self._states: + low = s.strip().lower() + wanted.add(low) + if s.upper() in _US_STATES: # "ID" → "idaho" + wanted.add(_US_STATES[s.upper()]) + for ab, name in _US_STATES.items(): # "Idaho" → "id" + if name == low: + wanted.add(ab.lower()) + return wanted + + def _http_get_json(self, url: str, timeout: int = 30): + """GET a URL and JSON-decode. Raises on HTTP/URL/parse error.""" + headers = {"User-Agent": "MeshAI/1.0", "Accept": "application/json"} + req = Request(url, headers=headers) + with urlopen(req, timeout=timeout) as resp: + return json.loads(resp.read().decode("utf-8")) + + # ── tick / polling ─────────────────────────────────────────────────────── + + def tick(self) -> bool: + """Execute one polling tick. Returns True if the event set changed.""" + now = time.time() + + if not self._base_url and not self._registry_url: + return False + + if now - self._last_tick < self._tick_interval: + return False + + self._last_tick = now + + # Refresh the (cached) registry-derived feed URL list if stale. + self._refresh_feed_urls(now) + if not self._feed_urls: + return False + + return self._fetch_all(now) + + def _refresh_feed_urls(self, now: float) -> None: + """Populate ``self._feed_urls`` from base_url override or the registry. + + The registry lookup is cached for ``registry_ttl`` seconds. A registry + fetch failure is logged + swallowed (keeps any previously-discovered + feeds). + """ + # Single-feed override always wins and needs no registry. + if self._base_url: + self._feed_urls = [self._base_url] + return + + if self._feed_urls and (now - self._registry_fetched_at) < self._registry_ttl: + return # cache still warm + + if not self._registry_url: + return + + try: + rows = self._http_get_json(self._registry_url) + except (HTTPError, URLError) as e: + reason = getattr(e, "reason", None) or getattr(e, "code", None) or e + logger.warning("WZDx registry fetch failed (%s); keeping cached feeds", reason) + self._last_error = f"registry: {reason}" + self._consecutive_errors += 1 + return + except Exception as e: + logger.warning("WZDx registry parse failed (%s); keeping cached feeds", e) + self._last_error = f"registry parse: {e}" + self._consecutive_errors += 1 + return + + self._feed_urls = self._select_feeds(rows) + self._registry_fetched_at = now + logger.info("WZDx registry: %d GeoJSON feed(s) for states=%s", + len(self._feed_urls), self._states) + + def _select_feeds(self, rows) -> list: + """Filter registry rows to configured states' GeoJSON WZDx feeds.""" + if not isinstance(rows, list): + return [] + wanted = self._wanted_states() + urls = [] + for row in rows: + if not isinstance(row, dict): + continue + state = str(row.get("state") or row.get("state_full") or "").strip().lower() + if wanted and state not in wanted: + continue + fmt = str(row.get("format") or "").strip().lower() + if fmt and "geojson" not in fmt: + continue # only GeoJSON WZDx feeds; skip xml/other + url = (row.get("url") or row.get("apiurl") or row.get("feed_url") or "").strip() + if url and url not in urls: + urls.append(url) + return urls + + def _fetch_all(self, now: float) -> bool: + """Fetch + parse every configured feed. Returns True if events changed.""" + new_events = [] + any_success = False + + for url in self._feed_urls: + try: + fc = self._http_get_json(url) + except (HTTPError, URLError) as e: + reason = getattr(e, "reason", None) or getattr(e, "code", None) or e + logger.warning("WZDx feed fetch failed for %s (%s); skipping", url, reason) + self._last_error = f"feed: {reason}" + self._consecutive_errors += 1 + continue + except Exception as e: + logger.warning("WZDx feed parse failed for %s (%s); skipping", url, e) + self._last_error = f"feed parse: {e}" + self._consecutive_errors += 1 + continue + + any_success = True + for feat in self._iter_features(fc): + evt = self._parse_feature(feat, now) + if evt: + new_events.append(evt) + + if not any_success: + return False # all feeds down — keep the last known good set + + # Optional bbox filter. + if self._bbox and len(self._bbox) == 4: + west, south, east, north = self._bbox + new_events = [ + e for e in new_events + if e.get("lat") is not None and e.get("lon") is not None + and west <= e["lon"] <= east and south <= e["lat"] <= north + ] + + 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._is_loaded = True + self._consecutive_errors = 0 + self._last_error = None + + if changed: + logger.info("WZDx work zones updated: %d active", len(new_events)) + return changed + + @staticmethod + def _iter_features(fc) -> list: + """Yield features from a WZDx v4 GeoJSON FeatureCollection.""" + if not isinstance(fc, dict): + return [] + feats = fc.get("features") + return feats if isinstance(feats, list) else [] + + @staticmethod + def _geometry_centroid(geometry) -> Optional[list]: + """Return [lon, lat] centroid of a GeoJSON geometry, or None.""" + if not isinstance(geometry, dict): + return None + coords = geometry.get("coordinates") + pts = [] + + def _collect(c): + if (isinstance(c, (list, tuple)) and len(c) >= 2 + and all(isinstance(x, (int, float)) for x in c[:2])): + pts.append((c[0], c[1])) + elif isinstance(c, (list, tuple)): + for sub in c: + _collect(sub) + + _collect(coords) + if not pts: + return None + lon = sum(p[0] for p in pts) / len(pts) + lat = sum(p[1] for p in pts) / len(pts) + return [lon, lat] + + def _parse_feature(self, feat: dict, now: float) -> Optional[dict]: + """Parse one WZDx v4 road_event feature into a stored event dict. + + Only ``work-zone`` road_events are kept. Field mapping delegates to + ``central_normalizer._parse_wzdx_federal`` (which reads either the + raw ``core_details.*`` nesting or a flattened envelope). Returns None + for non-work-zone / malformed / id-less features (never raises). + """ + try: + if not isinstance(feat, dict): + return None + props = feat.get("properties") + if not isinstance(props, dict): + return None + + cd = props.get("core_details") + cd = cd if isinstance(cd, dict) else {} + + # Only work-zone road_events (event_type may live nested or flat). + event_type = str(cd.get("event_type") or props.get("event_type") or "").strip().lower() + if event_type and event_type != "work-zone": + return None + + # Stable external id: data_source_id + feature id. + data_source_id = cd.get("data_source_id") or props.get("data_source_id") + feat_id = feat.get("id") or cd.get("id") or props.get("id") + if not feat_id and not data_source_id: + return None # no stable identity → cannot dedup + external_id = ":".join(str(p) for p in (data_source_id, feat_id) if p) + + # Coordinates: prop lat/lon if present, else geometry centroid. + lat = props.get("latitude") + lon = props.get("longitude") + centroid = self._geometry_centroid(feat.get("geometry")) + if (lat is None or lon is None) and centroid: + lon, lat = centroid[0], centroid[1] + if lat is None or lon is None: + return None # unusable without coords + + # inner_data for _parse_wzdx_federal: pass properties verbatim (its + # field() helper reads core_details first, then top-level). + inner_data = dict(props) + geo = {"centroid": centroid} if centroid else {} + n = _parse_wzdx_federal(inner_data, geo) + + # start/end epochs for the incident-gating keys. + start_at = self._iso_to_epoch(props.get("start_date") or cd.get("start_date")) + end_at = self._iso_to_epoch(props.get("end_date") or cd.get("end_date")) + + return { + "source": "wzdx", + "event_id": f"wzdx_{external_id}", + "event_type": "Work Zone", + "severity": "priority" if n.get("impact") == "full_closure" else "routine", + "lat": float(lat), + "lon": float(lon), + "expires": (end_at + 3600) if end_at else (now + 21600), + "fetched_at": now, + "external_id": external_id, + "start_at": start_at, + "end_at": end_at, + "normalized": n, # output of _parse_wzdx_federal + } + except Exception as e: + logger.debug("WZDx feature parse error: %s", e) + return None + + @staticmethod + def _iso_to_epoch(iso) -> Optional[int]: + """ISO-8601 → int epoch (UTC), or None.""" + if not iso: + return None + try: + s = str(iso).replace("Z", "+00:00") + dt = datetime.fromisoformat(s) + if dt.tzinfo is not None: + return int(dt.timestamp()) + # Naive → treat as UTC (matches ends_at_epoch convention). + return int(calendar.timegm(dt.timetuple())) + except Exception: + return None + + # ── pipeline Event construction ────────────────────────────────────────── + + def to_event(self, evt: dict) -> Optional["Event"]: + """Translate a stored WZDx event dict into a ``work_zone`` Event. + + Emits ``category="work_zone"`` with the canonical work-zone ``data`` + (road, direction, mile_start/end, sub_type, impact, ends_at_epoch, + town, distance_mi, bearing, lat, lon) plus dedup keys (external_id, + source) and the incident-gating fields (county, state, start/end, + lanes_affected, icon_category). Returns None if the dict lacks coords + or a stable id. + """ + try: + lat = evt.get("lat") + lon = evt.get("lon") + if lat is None or lon is None: + return None + + external_id = evt.get("external_id") + event_id = evt.get("event_id") + if not event_id: + return None + + n = evt.get("normalized", {}) or {} + severity = evt.get("severity", "routine") + + # ends_at (datetime) → ends_at_epoch (float), TZ-independent. + # Stored as calendar.timegm(naive.timetuple()); the formatter + # reconstructs via datetime.utcfromtimestamp(). + ends_at = n.get("ends_at") + ends_at_epoch = None + if ends_at is not None: + try: + naive = (ends_at.replace(tzinfo=None) + if ends_at.tzinfo is not None else ends_at) + ends_at_epoch = float(calendar.timegm(naive.timetuple())) + except Exception: + ends_at_epoch = None + + impact = n.get("impact") + sub_type = n.get("sub_type") + road = n.get("road") + direction = n.get("direction") + + # Canonical work_zone data (the 12 keys the formatter reads) plus + # the incident-gating keys (external_id/source/start_at/end_at/…). + canonical_data = { + # ── work_zone renderer keys ────────────────────────────── + "road": road, + "direction": direction, # full form ("northbound"…) + "mile_start": n.get("mile_start"), + "mile_end": n.get("mile_end"), + "sub_type": sub_type, # human-readable, impact-folded + "impact": impact, # "full_closure" | "partial" + "ends_at_epoch": ends_at_epoch, + "town": n.get("town"), + "distance_mi": n.get("distance_mi"), + "bearing": n.get("bearing"), + "lat": lat, + "lon": lon, + # ── incident-gating / dedup keys ───────────────────────── + "external_id": external_id, # stable WZDx id → gating dedup + "source": "wzdx", + "county": None, + "state": None, + "from_loc": None, + "to_loc": None, + "mile_marker": None, + "lanes_affected": self._lanes_phrase(impact), + "cause": None, + "comment": n.get("description") or None, + "geocoder_city": None, + "landclass": None, + "start_at": evt.get("start_at"), + "end_at": evt.get("end_at"), + "magnitude": None, + "delay_seconds": None, + "icon_category": "road_works", + } + + # Title / summary for the LLM + dashboard. + head_road = road or (n.get("town") or "Road event") + title = f"Work Zone: {head_road}" + if direction and direction != "unknown": + title += f" {direction}" + summary_parts = [title] + if sub_type: + summary_parts.append(sub_type) + if impact == "full_closure": + summary_parts.append("all lanes closed") + desc = n.get("description") + if desc and desc.strip(): + summary_parts.append(desc.strip()) + summary = " | ".join(summary_parts)[:300] + + return make_event( + source="wzdx", + category="work_zone", + severity=severity, + title=title, + summary=summary, + timestamp=evt.get("fetched_at"), + expires=evt.get("expires"), + lat=lat, + lon=lon, + group_key=event_id, + inhibit_keys=[event_id], + data=canonical_data, + ) + except Exception: + logger.exception("WZDx to_event failed for evt: %s", evt.get("event_id")) + return None + + @staticmethod + def _lanes_phrase(impact: Optional[str]) -> Optional[str]: + """Human lane-status phrase from the folded impact value.""" + if impact == "full_closure": + return "all lanes closed" + if impact == "partial": + return "lanes affected" + return None + + def get_events(self) -> list: + """Get current work-zone events.""" + return self._events + + @property + def health_status(self) -> dict: + """Get adapter health status.""" + return { + "source": "wzdx", + "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, + "feed_count": len(self._feed_urls), + "registry_age": (time.time() - self._registry_fetched_at) + if self._registry_fetched_at else None, + } diff --git a/work/tests/test_adapter_wzdx.py b/work/tests/test_adapter_wzdx.py new file mode 100644 index 0000000..c1a2b90 --- /dev/null +++ b/work/tests/test_adapter_wzdx.py @@ -0,0 +1,290 @@ +"""Tests for the FHWA WZDx native adapter — to_event() + discovery/fetch. + +Mirrors the style of test_adapter_firms.py: build realistic WZDx v4 GeoJSON +fixtures, drive the adapter's parse + to_event, and assert the canonical +``work_zone`` data dict is populated correctly and renders through the real +Phase-2 incident formatter. nearest_town() is patched off (no Photon +network) for deterministic town-independent assertions. +""" + +import calendar +import time +from types import SimpleNamespace + +import pytest + +import meshai.central_normalizer as cn +from meshai.env.wzdx import WZDxAdapter +from meshai.notifications.events import Event +from meshai.notifications.formatters.incident import format as format_incident + + +# ============================================================ +# FIXTURES +# ============================================================ + +@pytest.fixture(autouse=True) +def _no_photon(monkeypatch): + """Disable the Photon nearest_town lookup so parsing is network-free.""" + monkeypatch.setattr(cn, "nearest_town", lambda *a, **k: None) + + +@pytest.fixture +def mock_config(): + """A WZDx config with registry discovery enabled, keyless.""" + return SimpleNamespace( + enabled=True, + feed_source="native", + api_key="", + base_url="", + registry_url="https://datahub.transportation.gov/resource/69qe-yiui.json?$limit=200", + registry_ttl=21600, + tick_seconds=300, + states=["ID"], + bbox=[], + ) + + +@pytest.fixture +def adapter(mock_config): + return WZDxAdapter(mock_config) + + +def make_wzdx_feature( + feat_id="WZ-0001", + data_source_id="idot-1", + road="US-95", + direction="southbound", + event_type="work-zone", + work_type="surface-work", + vehicle_impact="some-lanes-closed", + description="Paving Operations on US-95 from MM (93) to MM (89).", + start_date="2026-07-18T13:00:00Z", + end_date="2026-07-19T00:59:59Z", + coordinates=None, +): + """Build a realistic WZDx v4 GeoJSON road_event feature.""" + if coordinates is None: + coordinates = [[-116.91, 43.66], [-116.92, 43.60]] + core = {"event_type": event_type, "data_source_id": data_source_id} + if road is not None: + core["road_names"] = [road] + if direction is not None: + core["direction"] = direction + if description is not None: + core["description"] = description + props = {"core_details": core} + if work_type is not None: + props["types_of_work"] = [{"type_name": work_type}] + if vehicle_impact is not None: + props["vehicle_impact"] = vehicle_impact + if start_date is not None: + props["start_date"] = start_date + if end_date is not None: + props["end_date"] = end_date + return { + "id": feat_id, + "type": "Feature", + "properties": props, + "geometry": {"type": "LineString", "coordinates": coordinates}, + } + + +def make_feature_collection(*features): + return {"type": "FeatureCollection", "features": list(features)} + + +# ============================================================ +# REGISTRY DISCOVERY +# ============================================================ + +def test_select_feeds_filters_state_and_geojson(adapter): + """Registry selection keeps configured-state GeoJSON feeds only.""" + rows = [ + {"state": "Idaho", "format": "geojson", + "url": "https://itd.idaho.gov/wzdx.geojson", "version": "4.0"}, + {"state": "ID", "format": "GeoJSON", "apiurl": "https://alt.idaho/wzdx"}, + {"state": "Utah", "format": "geojson", "url": "https://udot/wzdx"}, + {"state": "Idaho", "format": "xml", "url": "https://itd/xml"}, + ] + feeds = adapter._select_feeds(rows) + assert feeds == ["https://itd.idaho.gov/wzdx.geojson", "https://alt.idaho/wzdx"] + + +def test_wanted_states_expands_abbrev_and_name(adapter): + """'ID' matches both the abbreviation and 'idaho'.""" + assert {"id", "idaho"} <= adapter._wanted_states() + + +def test_select_feeds_handles_garbage(adapter): + """Non-list / non-dict rows never crash selection.""" + assert adapter._select_feeds("not a list") == [] + assert adapter._select_feeds([None, 3, {"state": "ID"}]) == [] # no url + + +# ============================================================ +# CATEGORY + CANONICAL DATA +# ============================================================ + +def test_to_event_category_is_work_zone(adapter): + evt = adapter._parse_feature(make_wzdx_feature(), time.time()) + event = adapter.to_event(evt) + assert event is not None + assert event.category == "work_zone" + + +def test_canonical_data_core_fields(adapter): + """road / direction / mile posts / lat/lon map correctly.""" + evt = adapter._parse_feature(make_wzdx_feature(), time.time()) + d = adapter.to_event(evt).data + assert d["road"] == "US-95" + assert d["direction"] == "southbound" # full form the renderer prints + assert d["mile_start"] == 93 + assert d["mile_end"] == 89 + assert d["lat"] == pytest.approx(43.63, abs=0.05) + assert d["lon"] == pytest.approx(-116.915, abs=0.05) + + +def test_canonical_external_id_is_stable(adapter): + """external_id combines data_source_id + feature id for gating dedup.""" + evt = adapter._parse_feature(make_wzdx_feature(), time.time()) + d = adapter.to_event(evt).data + assert d["external_id"] == "idot-1:WZ-0001" + assert d["source"] == "wzdx" + + +def test_canonical_start_end_and_ends_epoch(adapter): + """start_at / end_at epochs and ends_at_epoch are populated.""" + evt = adapter._parse_feature(make_wzdx_feature(), time.time()) + d = adapter.to_event(evt).data + # end_date 2026-07-19T00:59:59Z + expected_end = calendar.timegm((2026, 7, 19, 0, 59, 59, 0, 0, 0)) + assert d["end_at"] == expected_end + assert d["start_at"] == calendar.timegm((2026, 7, 18, 13, 0, 0, 0, 0, 0)) + assert d["ends_at_epoch"] == float(expected_end) + + +def test_canonical_lanes_and_subtype_partial(adapter): + """some-lanes-closed → impact partial + folded sub_type + lanes phrase.""" + evt = adapter._parse_feature(make_wzdx_feature(), time.time()) + d = adapter.to_event(evt).data + assert d["impact"] == "partial" + assert d["sub_type"] == "lanes reduced, surface work" + assert d["lanes_affected"] == "lanes affected" + assert d["icon_category"] == "road_works" + + +def test_full_closure_maps_to_priority(adapter): + """all-lanes-closed → priority severity + full_closure impact.""" + feat = make_wzdx_feature( + vehicle_impact="all-lanes-closed", work_type="bridge-construction", + road="I-84", direction="eastbound", + ) + evt = adapter._parse_feature(feat, time.time()) + assert evt["severity"] == "priority" + event = adapter.to_event(evt) + assert event.severity == "priority" + assert event.data["impact"] == "full_closure" + assert event.data["lanes_affected"] == "all lanes closed" + + +# ============================================================ +# FORMATTER INTEGRATION (real Phase-2 renderer) +# ============================================================ + +def test_renders_through_real_formatter(adapter): + """Emitted data flows through formatters/incident.format() to a wire.""" + now = time.time() + evt = adapter._parse_feature(make_wzdx_feature(), now) + event = adapter.to_event(evt) + wire = format_incident(event, now=now, budget=200) + assert wire.startswith("🚧") + assert "US-95" in wire + assert "mile 93" in wire + assert len(wire.encode("utf-8")) <= 80 # work-zone byte cap + + +def test_full_closure_wire_shows_all_lanes_closed(adapter): + feat = make_wzdx_feature(vehicle_impact="all-lanes-closed", + work_type="bridge-construction", road="I-84") + evt = adapter._parse_feature(feat, time.time()) + wire = format_incident(adapter.to_event(evt), now=time.time(), budget=200) + assert "all lanes closed" in wire + + +# ============================================================ +# FEED FETCH FLOW +# ============================================================ + +def test_tick_fetches_and_populates_events(adapter, monkeypatch): + """A full tick discovers a feed, fetches it, and stores parsed events.""" + registry = [{"state": "Idaho", "format": "geojson", + "url": "https://itd.idaho.gov/wzdx.geojson"}] + fc = make_feature_collection( + make_wzdx_feature(feat_id="A"), + make_wzdx_feature(feat_id="B", road="I-15", description="Road work"), + ) + + def fake_get(url, timeout=30): + return registry if "datahub" in url else fc + + monkeypatch.setattr(adapter, "_http_get_json", fake_get) + changed = adapter.tick() + assert changed is True + events = adapter.get_events() + assert len(events) == 2 + assert {e["external_id"] for e in events} == {"idot-1:A", "idot-1:B"} + assert adapter.health_status["feed_count"] == 1 + + +def test_feed_failure_does_not_crash(adapter, monkeypatch): + """A down feed is skipped; the adapter survives and stays healthy-ish.""" + registry = [{"state": "Idaho", "format": "geojson", "url": "https://itd/wzdx"}] + + def fake_get(url, timeout=30): + if "datahub" in url: + return registry + raise TimeoutError("feed down") + + monkeypatch.setattr(adapter, "_http_get_json", fake_get) + # Should not raise; no events, last_error recorded. + assert adapter.tick() is False + assert adapter.get_events() == [] + assert adapter.health_status["last_error"] is not None + + +# ============================================================ +# DEFENSIVE +# ============================================================ + +def test_empty_feature_returns_none(adapter): + assert adapter._parse_feature({}, time.time()) is None + + +def test_non_workzone_feature_skipped(adapter): + feat = make_wzdx_feature(event_type="special-event") + assert adapter._parse_feature(feat, time.time()) is None + + +def test_missing_coords_returns_none(adapter): + feat = make_wzdx_feature(coordinates=[]) + feat["properties"].pop("core_details", None) # also no lat/lon + feat["geometry"] = {"type": "LineString", "coordinates": []} + assert adapter._parse_feature(feat, time.time()) is None + + +def test_no_stable_id_returns_none(adapter): + feat = make_wzdx_feature(data_source_id=None) + feat.pop("id", None) + feat["properties"]["core_details"].pop("data_source_id", None) + assert adapter._parse_feature(feat, time.time()) is None + + +def test_to_event_corrupted_dict_returns_none(adapter): + assert adapter.to_event({"garbage": True}) is None + + +def test_to_event_missing_coords_returns_none(adapter): + evt = adapter._parse_feature(make_wzdx_feature(), time.time()) + evt["lat"] = None + assert adapter.to_event(evt) is None