matt/meshai
1
0
Fork 0
mirror of https://github.com/zvx-echo6/meshai.git synced 2026-08-26 17:31:34 +00:00
Code Issues Projects Releases Packages Wiki Activity

feat(wzdx): native FHWA WZDx work-zone adapter (standalone work_zone source) (#37)

Browse source 
New env/wzdx.py native adapter so meshai has a LOCAL source for the
work_zone category instead of Central-only. Keyless: discovers per-state
WZDx feeds from the FHWA registry (datahub.transportation.gov), filtered to
configured states (default [ID]), fetches each WZDx v4 GeoJSON feed, and
parses road_events into canonical work_zone Event.data.

- reuses central_normalizer._parse_wzdx_federal for identical field mapping
  → renders through the existing Phase-2 formatters/gating incident path
  (no new formatter/gating); verified end-to-end to a valid work-zone wire
- stable external_id = {data_source_id}:{feature_id} for incident dedup
- registry cached 6h; per-feed failures logged + skipped (never crash)
- WZDxConfig extended (registry_url, registry_ttl, states); registered in
  env/store.py; enabled=false + feed_source=native default (dormant until
  turned on at the coordinated flip)

19 new tests; suite at the 10-failure baseline (1640 passed). Only env/wzdx.py
(new), env/store.py, config.py, tests/test_adapter_wzdx.py touched.

Co-authored-by: Matt Johnson <mj@k7zvx.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
malice 2026-07-05 01:36:52 -06:00 • committed by GitHub
commit f338673f0c
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 820 additions and 6 deletions
Download patch file Download diff file Expand all files Collapse all files

22
work/meshai/config.py
View file

@ -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

4
work/meshai/env/store.py vendored
View file

@ -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)

510
work/meshai/env/wzdx.py vendored Normal file
View file

@ -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,
}

290
work/tests/test_adapter_wzdx.py Normal file
View file

@ -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