diff --git a/work/meshai/adapter_config/defaults.py b/work/meshai/adapter_config/defaults.py index 0318dab..b1dec70 100644 --- a/work/meshai/adapter_config/defaults.py +++ b/work/meshai/adapter_config/defaults.py @@ -155,7 +155,7 @@ REGISTRY: dict[tuple[str, str], dict[str, Any]] = { # INCIDENT -- 2 settings (shared freshness gate + Update-after-New toggle) # ================================================================= ("incident", "freshness_seconds"): { - "default": 1800, # incident_handler.py:49 + central_normalizer.py:917 + "default": 1800, # notifications/pipeline/dispatcher.py freshness gate "type": "int", "description": "Drop incidents older than this many seconds.", }, @@ -319,7 +319,7 @@ REGISTRY: dict[tuple[str, str], dict[str, Any]] = { "description": "OSM place classes that count as a town for the nearest_town anchor.", }, ("geocoder", "h3_cache_max"): { - "default": 10000, # central_normalizer.py:297 + "default": 10000, # geo.py _H3_CACHE_MAX "type": "int", "description": "Max H3 cache entries before LRU eviction.", }, diff --git a/work/meshai/central_normalizer.py b/work/meshai/central_normalizer.py deleted file mode 100644 index dc600de..0000000 --- a/work/meshai/central_normalizer.py +++ /dev/null @@ -1,973 +0,0 @@ -"""Meshai-side Central-envelope normalizer. - -Central is a faithful firehose — it preserves upstream payloads verbatim -(per Central v0.10.0 §README "Central takes it all and gives it all"). -Per-adapter shape normalization is the consumer's job. This module is -where that lives. - -First adapter wired: state_511_atis (Castle Rock ATIS feeds — the source -for Idaho 511 work_zone / closure events). Other adapters will be added -as their renderer formats are approved. - -Design: `normalize(envelope) -> dict | None` returns a flat, render-ready -dict whose shape is described in NORMALIZED_KEYS. Adapter-specific -extraction lives in private parsers dispatched off `inner.adapter`. The -output dict is pure-data; formatting is the renderer's job. -""" - -import json -import logging -import math -import re -import urllib.error -import urllib.parse -import urllib.request -from collections import OrderedDict -from datetime import datetime, timezone -from typing import Any, Optional -# Geocoder config is set via init_geocoder_config() - -logger = logging.getLogger(__name__) - - -# ---------- shared normalized output shape -------------------------------- - -NORMALIZED_KEYS = ( - "source", # str -- inner.adapter - "road", # str | None - "direction", # str | None -- 'northbound'/'southbound'/'eastbound'/ - # 'westbound'/'both'/'unknown' - "mile_start", # int | None - "mile_end", # int | None - "description", # str | None -- upstream prose, cleaned - "sub_type", # str | None -- friendly: 'construction work', 'incident', ... - "impact", # str | None -- 'full_closure'/'partial'/'unknown' - "ends_at", # datetime | None (UTC) -- parsed from description if absent structurally - "town", # str | None -- _enriched.geocoder.city or .name - "distance_mi", # int | None -- haversine from event coords to town - "bearing", # str | None -- 'N'/'NE'/.../'NW' -) - - -# ---------- direction normalization --------------------------------------- - -_DIR_MAP = { - "north": "northbound", "northbound": "northbound", "nb": "northbound", - "south": "southbound", "southbound": "southbound", "sb": "southbound", - "east": "eastbound", "eastbound": "eastbound", "eb": "eastbound", - "west": "westbound", "westbound": "westbound", "wb": "westbound", - "both": "both", "both directions": "both", - "unknown": "unknown", "": "unknown", -} - - -def _norm_direction(raw: Optional[str]) -> Optional[str]: - if raw is None: return None - s = str(raw).strip().lower() - return _DIR_MAP.get(s, "unknown") - - -# ---------- sub_type → friendly label ------------------------------------- - -_SUBTYPE_MAP = { - "roadConstruction": "road construction", - "longTermRoadConstruction": "road construction", - "constructionWork": "construction work", - "bridgeConstruction": "bridge construction", - "bridgeMaintenanceOperations": "bridge maintenance", - "bridgeInspectionWork": "bridge inspection", - "pavingOperations": "paving", - "pavementMarkingOperations": "pavement marking", # also w/ trailing space - "emergencyRepairs": "emergency repairs", - "utilityWork": "utility work", - "guardrailRepairs": "guardrail repairs", - "workOnTheShoulder": "shoulder work", - "brushControl": "brush control", - "flaggingOperation": "flagging", - "singleLineTraffic:AlternatingDirections": "alternating one-way", -} - - -def _norm_sub_type(raw: Optional[str]) -> Optional[str]: - if not raw: return None - s = str(raw).strip() - if s in _SUBTYPE_MAP: - return _SUBTYPE_MAP[s] - # Trailing-space variants - if s.strip() in _SUBTYPE_MAP: - return _SUBTYPE_MAP[s.strip()] - # Fallback: camelCase split, lowercase, drop colon-suffix - s = s.split(":", 1)[0] - parts = re.findall(r"[A-Z]?[a-z]+|[A-Z]+(?=[A-Z]|$)", s) or [s] - return " ".join(p.lower() for p in parts) - - -# ---------- description parsers (state_511_atis-style) -------------------- - -# "from MM (93) to MM (89)" → (93, 89) -# "near MM (495)" → (495, None) -# "at MM (60)" → (60, None) -_MM_RE = re.compile( - r"(?:from\s+)?MM\s*\(?(\d+)\)?(?:\s*to\s+MM\s*\(?(\d+)\)?)?", - re.IGNORECASE, -) - - -def _parse_mile_posts(description: str) -> tuple[Optional[int], Optional[int]]: - if not description: return None, None - m = _MM_RE.search(description) - if not m: return None, None - try: - start = int(m.group(1)) - except (TypeError, ValueError): - return None, None - end = None - if m.group(2): - try: end = int(m.group(2)) - except (TypeError, ValueError): end = None - return start, end - - -# "5/29/2026 10:00 AM to 5/29/2026 3:00 PM" → datetime(2026, 5, 29, 15, 0, tzinfo=UTC) -# (we treat the parsed time as local America/Boise but for the short -# format renderer Boise-relative is what users actually want anyway). -_DATERANGE_RE = re.compile( - r"(\d{1,2}/\d{1,2}/\d{4})\s+(\d{1,2}:\d{2})\s+(AM|PM)\s+to\s+" - r"(\d{1,2}/\d{1,2}/\d{4})\s+(\d{1,2}:\d{2})\s+(AM|PM)", - re.IGNORECASE, -) - - -def _parse_ends_at(description: str) -> Optional[datetime]: - if not description: return None - m = _DATERANGE_RE.search(description) - if not m: return None - end_date, end_time, end_ampm = m.group(4), m.group(5), m.group(6).upper() - try: - dt = datetime.strptime(f"{end_date} {end_time} {end_ampm}", "%m/%d/%Y %I:%M %p") - except ValueError: - return None - return dt # naive; renderer treats as local - - -# ---------- description cleanup ------------------------------------------- - -_HTML_TAG_RE = re.compile(r"<[^>]+>") - - -def _clean_description(raw: Optional[str]) -> Optional[str]: - if not raw: return None - s = _HTML_TAG_RE.sub(" ", str(raw)) - s = re.sub(r"\s+", " ", s).strip() - return s or None - - -# ---------- distance / bearing -------------------------------------------- - -# v0.6-4: town_anchors moved to a GUI-editable SQLite table. Lookups go -# through meshai.persistence.curation.lookup_town_anchor() now. - - -def _haversine_miles(lat1: float, lon1: float, lat2: float, lon2: float) -> float: - R = 3958.8 # Earth radius in miles - phi1, phi2 = math.radians(lat1), math.radians(lat2) - dphi = math.radians(lat2 - lat1) - dl = math.radians(lon2 - lon1) - a = math.sin(dphi / 2) ** 2 + math.cos(phi1) * math.cos(phi2) * math.sin(dl / 2) ** 2 - return 2 * R * math.asin(math.sqrt(a)) - - -def _bearing_compass(lat1: float, lon1: float, lat2: float, lon2: float) -> str: - """Compass bearing FROM (lat2, lon2) TO (lat1, lon1) -- i.e., 'event is - of town'. We orient so the event's bearing relative to the - town reads naturally ("8 mi N of Plummer" = event is north of Plummer).""" - phi1, phi2 = math.radians(lat2), math.radians(lat1) - dl = math.radians(lon1 - lon2) - x = math.sin(dl) * math.cos(phi2) - y = math.cos(phi1) * math.sin(phi2) - math.sin(phi1) * math.cos(phi2) * math.cos(dl) - brng = (math.degrees(math.atan2(x, y)) + 360) % 360 - points = ["N", "NE", "E", "SE", "S", "SW", "W", "NW"] - return points[int((brng + 22.5) // 45) % 8] - - -def _compute_distance_bearing( - event_lat: Optional[float], event_lon: Optional[float], town: Optional[str] -) -> tuple[Optional[int], Optional[str]]: - if event_lat is None or event_lon is None or not town: - return None, None - key = str(town).strip().lower() - from meshai.persistence.curation import lookup_town_anchor - coords = lookup_town_anchor(key) - if coords is None: - return None, None - tlat, tlon = coords - d = _haversine_miles(event_lat, event_lon, tlat, tlon) - b = _bearing_compass(event_lat, event_lon, tlat, tlon) - return int(round(d)), b - - -# ---------- road-name normalization --------------------------------------- - -# SB/NB/EB/WB tokens inside a road name (e.g. "I-15 SB Off Ramp") collapse -# to a single cardinal letter ("I-15 S Off Ramp") for tighter mesh output. -_CARDINAL_TOKEN_RE = re.compile(r"\b(SB|NB|EB|WB)\b") -_CARDINAL_MAP = {"SB": "S", "NB": "N", "EB": "E", "WB": "W"} - - -def normalize_road_name(raw: Optional[str]) -> Optional[str]: - """Tighten a raw roadway_name for mesh output: - 'I-15 SB Off Ramp' -> 'I-15 S Off Ramp' - 'US-95 NB' -> 'US-95 N' - Returns None for empty / None input. - """ - if not raw: - return None - s = str(raw).strip() - if not s: - return None - return _CARDINAL_TOKEN_RE.sub(lambda m: _CARDINAL_MAP[m.group(1)], s) - - -# Uninformative road names (Exit-only ramps with no parent route prefix -# visible) get dropped so the renderer leads with the town instead. -_UNINFORMATIVE_ROAD_RE = re.compile( - r"^Exit\s+\d+.*\b(On|Off)\s+Ramp$", - re.IGNORECASE, -) - - -def _is_uninformative_road(road: Optional[str]) -> bool: - if not road: - return False - return bool(_UNINFORMATIVE_ROAD_RE.match(str(road).strip())) - - -# ---------- nearest_town: Photon /reverse + H3 cache ---------------------- - -# Photon is reachable from CT108 at this Tailscale address (verified -# 2026-06-04). It's the same Echo6-local Photon instance that backs Central's -# NaviBackend reverse-geocoder. Photon takes osm_tag=place (KEY only, not -# key:value with comma-list -- that returns 0 features -- per probe). -# v0.6-3b: photon geocoder config - initialized via init_geocoder_config() -# Defaults to public Komoot Photon; deployments override in config.yaml. - -class _GeocoderSettings: - url: str = "https://photon.komoot.io" - timeout_seconds: float = 2.0 - radius_km: float = 80.0 - limit: int = 10 - -_geocoder = _GeocoderSettings() - - -def init_geocoder_config(url: str = None, timeout: float = None, - radius: float = None, limit: int = None) -> None: - """Initialize geocoder settings from config.yaml values.""" - if url is not None: - _geocoder.url = url - if timeout is not None: - _geocoder.timeout_seconds = timeout - if radius is not None: - _geocoder.radius_km = radius - if limit is not None: - _geocoder.limit = limit - - -# OSM place classes we accept as "town". Suburb included for metro coverage; -# locality is rare but valid for tiny rural places. -_TOWN_OSM_VALUES = frozenset({"city", "town", "village"}) - - -# Process-lifetime LRU cache keyed by H3 cell (resolution 7 ≈ 5km hexagons). -# Cells don't move and Photon's reverse output for a coord is stable, so -# entries never expire within a process lifetime. Cap at 10k entries. -_H3_CACHE_RESOLUTION = 7 -_H3_CACHE_MAX = 10_000 -_h3_cache: "OrderedDict[str, Optional[dict]]" = OrderedDict() - - -def _h3_cell(lat: float, lon: float) -> Optional[str]: - try: - import h3 # local import: keep module-import-time h3-free - return h3.latlng_to_cell(lat, lon, _H3_CACHE_RESOLUTION) - except Exception: - # Fallback: coarse-grain by rounding coords (~1.1 km per 0.01 deg). - return f"fallback:{round(lat, 2)},{round(lon, 2)}" - - -def _photon_reverse_places(lat: float, lon: float) -> list[dict]: - """Call Photon /reverse with osm_tag=place. Return raw feature list.""" - qs = urllib.parse.urlencode({ - "lat": f"{lat:.6f}", - "lon": f"{lon:.6f}", - "radius": _geocoder.radius_km, - "osm_tag": "place", - "limit": _geocoder.limit, - }) - url = f"{_geocoder.url}/reverse?{qs}" - try: - with urllib.request.urlopen(url, timeout=_geocoder.timeout_seconds) as resp: - body = resp.read() - d = json.loads(body) - except (urllib.error.URLError, urllib.error.HTTPError, TimeoutError, - json.JSONDecodeError, ConnectionError) as e: - logger.debug("Photon /reverse failed (%s) for %.4f,%.4f", e, lat, lon) - return [] - feats = d.get("features") or [] - return feats if isinstance(feats, list) else [] - - -def nearest_town(lat: float, lon: float, max_distance_mi: float = 50.0) -> Optional[dict]: - """Return the nearest populated place to (lat, lon) within max_distance_mi. - - Result shape: {name: str, distance_mi: int (rounded), bearing: str} - where bearing is an 8-point compass (N/NE/E/SE/S/SW/W/NW) of the event - location relative to the town -- i.e. "8 mi N of Plummer" means the - event is N of the town. Returns None if no town within range or if - Photon is unreachable. - - Calls Photon /reverse?osm_tag=place at _geocoder.url. Results are - H3-cell-cached (resolution 7 ≈ 5 km cells) so the second event near - the same town is free. - """ - if lat is None or lon is None: - return None - try: - lat, lon = float(lat), float(lon) - except (TypeError, ValueError): - return None - - cell = _h3_cell(lat, lon) - if cell is not None and cell in _h3_cache: - # LRU touch - _h3_cache.move_to_end(cell) - cached = _h3_cache[cell] - if cached is None or cached.get("distance_mi", 999) <= max_distance_mi: - return cached - - feats = _photon_reverse_places(lat, lon) - candidates: list[tuple[float, dict]] = [] - for f in feats: - p = f.get("properties") or {} - # Only accept proper populated places. - if p.get("osm_key") != "place" or p.get("osm_value") not in _TOWN_OSM_VALUES: - continue - coords = (f.get("geometry") or {}).get("coordinates") - if not (isinstance(coords, list) and len(coords) >= 2): - continue - tlon, tlat = coords[0], coords[1] - try: - tlat, tlon = float(tlat), float(tlon) - except (TypeError, ValueError): - continue - d_mi = _haversine_miles(lat, lon, tlat, tlon) - if d_mi > max_distance_mi: - continue - name = p.get("name") - if not name: - continue - candidates.append((d_mi, { - "name": str(name), - "distance_mi": int(round(d_mi)), - "bearing": _bearing_compass(lat, lon, tlat, tlon), - })) - - if not candidates: - if cell is not None: - _h3_cache[cell] = None - _h3_cache.move_to_end(cell) - while len(_h3_cache) > _H3_CACHE_MAX: - _h3_cache.popitem(last=False) - return None - - candidates.sort(key=lambda kv: kv[0]) - result = candidates[0][1] - if cell is not None: - _h3_cache[cell] = result - _h3_cache.move_to_end(cell) - while len(_h3_cache) > _H3_CACHE_MAX: - _h3_cache.popitem(last=False) - return result - - -# ---------- per-adapter parsers ------------------------------------------- - -def _parse_state_511_atis(inner_data: dict, geo: dict) -> dict: - desc = _clean_description(inner_data.get("description")) - mile_start, mile_end = _parse_mile_posts(desc or "") - ends_at = _parse_ends_at(desc or "") - is_full = bool(inner_data.get("is_full_closure")) - impact = "full_closure" if is_full else "partial" - enriched = (inner_data.get("_enriched") or {}).get("geocoder") or {} - - # Road name normalization + uninformative drop. - road = normalize_road_name(inner_data.get("roadway_name")) - if _is_uninformative_road(road): - road = None - - # Coordinates: prefer flat lat/lon, fall back to geo.centroid. - event_lat = inner_data.get("latitude") - event_lon = inner_data.get("longitude") - if event_lat is None and geo.get("centroid"): - try: event_lon, event_lat = geo["centroid"][0], geo["centroid"][1] - except (IndexError, TypeError): pass - - # Town selection (Matt's locked plan, post-parse-everything decision): - # PRIMARY: _enriched.geocoder.city (Navi/Photon already chose it for us) - # SECONDARY: nearest_town(lat, lon) -- direct Photon nearest-place hit - # TERTIARY: None -- renderer drops the town segment - # NEVER fall back to _enriched.geocoder.name -- that's nearest-feature - # data (forest-service road numbers, generic street names) not town data. - town = (enriched.get("city") or "").strip() or None - distance_mi: Optional[int] = None - bearing: Optional[str] = None - if town: - distance_mi, bearing = _compute_distance_bearing(event_lat, event_lon, town) - else: - # SECONDARY: ask Photon directly for the nearest populated place. - nt = nearest_town(event_lat, event_lon) if event_lat is not None else None - if nt: - town = nt.get("name") - distance_mi = nt.get("distance_mi") - bearing = nt.get("bearing") - - return { - "source": "state_511_atis", - "road": road, - "direction": _norm_direction(inner_data.get("direction")), - "mile_start": mile_start, - "mile_end": mile_end, - "description": desc, - "sub_type": _norm_sub_type(inner_data.get("event_sub_type")), - "impact": impact, - "ends_at": ends_at, - "town": town, - "distance_mi": distance_mi, - "bearing": bearing, - } - - -# ---------- wzdx federal vocabulary maps ---------------------------------- - -# FHWA WZDx v4 + custom-feed vocabulary observed in the wild. Unknown values -# fall through to lowercased + hyphens→spaces (see _norm_wzdx_sub_type). -_WZDX_WORK_TYPE_MAP: dict[str, Optional[str]] = { - # WZDx v4 spec types_of_work.type_name enum: - "maintenance": "maintenance", - "minor-road-defect-repair": "minor repair", - "roadside-work": "roadside work", - "overhead-work": "overhead work", - "below-road-work": "subsurface work", - "barrier-work": "barrier work", - "surface-work": "surface work", - "painting": "painting", - "roadway-relocation": "roadway relocation", - "roadway-creation": "new construction", - # Common informal values seen in upstream feeds (ID, WA): - "road-work": "road work", - "paving": "paving", - "bridge-construction": "bridge construction", - "bridge-maintenance": "bridge maintenance", - "utility-work": "utility work", - "road-construction": "road construction", - "construction": "construction", - "emergency-repairs": "emergency repairs", - # event_type values (drop the too-generic ones): - "work-zone": None, - "detour": "detour", -} - - -# vehicle_impact taxonomy (WZDx v4). Maps to mesh-friendly phrase. -# Returns None for values the renderer should drop entirely. -_WZDX_IMPACT_MAP: dict[str, Optional[str]] = { - "all-lanes-closed": "all lanes closed", - "some-lanes-closed": "lanes reduced", - "alternating-one-way": "one-way alternating", - "unknown": None, - "all-lanes-open": None, # informational only; nothing to do -} - - -def _norm_wzdx_sub_type(raw) -> Optional[str]: - if not raw: return None - s = str(raw).strip().lower() - if not s: return None - if s in _WZDX_WORK_TYPE_MAP: - return _WZDX_WORK_TYPE_MAP[s] - # Unknown value — keep lowercased, hyphens → spaces, single-line. - return re.sub(r"\s+", " ", s.replace("-", " ")).strip() or None - - -# ---------- per-adapter parser: wzdx federal ------------------------------ - -def _parse_wzdx_federal(inner_data: dict, geo: dict) -> dict: - """Normalize a wzdx-adapter envelope (FHWA WZDx federal spec). - - Central flattens the upstream payload in practice (the FHWA-spec - `core_details.*` nesting is not preserved), but we defensively check - nested keys too so any future Central change doesn't silently regress. - - sub_type uses types_of_work[0].type_name when present, else event_type, - each normalized via _WZDX_WORK_TYPE_MAP. impact_phrase is folded INTO - the sub_type slot for the renderer (so the description-slot reads e.g. - 'lanes reduced, paving' or 'one-way alternating' or 'road work'). - 'all lanes closed' is set on impact='full_closure' so the renderer's - existing full-closure promotion handles it -- avoids double-printing. - """ - cd = inner_data.get("core_details") - if not isinstance(cd, dict): cd = {} - def field(key): - v = cd.get(key) - if v is None or (isinstance(v, str) and not v.strip()): - v = inner_data.get(key) - return v - - # --- road (raw, verbatim per Matt's spec) ----------------------------- - road_names = field("road_names") - road = None - if isinstance(road_names, list) and road_names: - road = str(road_names[0]).strip() or None - elif isinstance(road_names, str) and road_names.strip(): - road = road_names.strip() - if _is_uninformative_road(road): - road = None - - # --- direction -------------------------------------------------------- - direction = _norm_direction(field("direction")) - - # --- sub_type (types_of_work[0] | event_type) ------------------------- - work_type: Optional[str] = None - tow = field("types_of_work") - if isinstance(tow, list) and tow: - first = tow[0] - if isinstance(first, dict): - work_type = _norm_wzdx_sub_type(first.get("type_name")) - elif isinstance(first, str): - work_type = _norm_wzdx_sub_type(first) - if not work_type: - work_type = _norm_wzdx_sub_type(field("event_type")) - - # --- vehicle_impact --------------------------------------------------- - vi_raw = (inner_data.get("vehicle_impact") or cd.get("vehicle_impact") or "") - impact_phrase: Optional[str] = _WZDX_IMPACT_MAP.get(str(vi_raw).strip().lower()) - is_full_closure = (str(vi_raw).strip().lower() == "all-lanes-closed") - - # Fold impact_phrase + work_type into the renderer's sub_type slot. - # For full-closure, exclude impact_phrase here -- the renderer prepends - # "all lanes closed" itself via the impact='full_closure' branch. - parts: list[str] = [] - if impact_phrase and not is_full_closure: - parts.append(impact_phrase) - if work_type: - parts.append(work_type) - sub_type = ", ".join(parts) if parts else None - impact = "full_closure" if is_full_closure else "partial" - - # --- ends_at: structured end_date ISO-8601 --------------------------- - ends_at: Optional[datetime] = None - end_date = inner_data.get("end_date") or cd.get("end_date") - if end_date: - try: - s = str(end_date).replace("Z", "+00:00") - ends_at = datetime.fromisoformat(s) - # Strip tzinfo so _format_end_short compares naive-to-naive. - if ends_at.tzinfo is not None: - ends_at = ends_at.astimezone().replace(tzinfo=None) - except Exception: - ends_at = None - - # --- mile_start/_end: regex on description, fall back to structured -- - desc = _clean_description(field("description")) - mile_start, mile_end = _parse_mile_posts(desc or "") - if mile_start is None: - ms = inner_data.get("road_mile_post_start") - if ms is not None: - try: mile_start = int(ms) - except (TypeError, ValueError): pass - if mile_end is None: - me = inner_data.get("road_mile_post_end") - if me is not None: - try: mile_end = int(me) - except (TypeError, ValueError): pass - - # --- coordinates ----------------------------------------------------- - event_lat = inner_data.get("latitude") - event_lon = inner_data.get("longitude") - if event_lat is None and geo.get("centroid"): - try: event_lon, event_lat = geo["centroid"][0], geo["centroid"][1] - except (IndexError, TypeError): pass - - # --- town fallback chain (same as state_511_atis) -------------------- - enriched = (inner_data.get("_enriched") or {}).get("geocoder") or {} - town = (enriched.get("city") or "").strip() or None - distance_mi: Optional[int] = None - bearing: Optional[str] = None - if town: - distance_mi, bearing = _compute_distance_bearing(event_lat, event_lon, town) - elif event_lat is not None: - nt = nearest_town(event_lat, event_lon) - if nt: - town = nt.get("name") - distance_mi = nt.get("distance_mi") - bearing = nt.get("bearing") - - return { - "source": "wzdx", - "road": road, - "direction": direction, - "mile_start": mile_start, - "mile_end": mile_end, - "description": desc, - "sub_type": sub_type, - "impact": impact, - "ends_at": ends_at, - "town": town, - "distance_mi": distance_mi, - "bearing": bearing, - } - - - -# ---------- WFIGS incidents (wildfire+prescribed) ------------------------- - -# IncidentName values like "IA 1", "IA 27" are auto-numbered Initial-Attack -# placeholders that WFIGS issues before a fire gets a proper name. We pass -# them through verbatim per Matt's call -- they at least signal "new fire -# in " even without an interesting name. -_WFIGS_ACRES_KEYS = ("DailyAcres", "IncidentSize") -_WFIGS_ACRES_RAW_KEYS = ("IncidentSize", "DiscoveryAcres", "FinalAcres") -_WFIGS_CONTAINED_KEYS = ("PercentContained",) -_WFIGS_CONTAINED_RAW_KEYS = ("PercentContained",) - - -def _first_non_null(d: dict, keys) -> Any: - """Return d[k] for the first k in keys with a non-null value, else None.""" - for k in keys: - v = d.get(k) - if v is not None and v != "": - return v - return None - - -def _parse_wfigs_acres(inner_data: dict) -> Optional[float]: - """Acres fallback chain: top-level DailyAcres/IncidentSize -> raw.* -> None.""" - val = _first_non_null(inner_data, _WFIGS_ACRES_KEYS) - if val is None: - raw = inner_data.get("raw") or {} - if isinstance(raw, dict): - val = _first_non_null(raw, _WFIGS_ACRES_RAW_KEYS) - if val is None: - return None - try: return float(val) - except (TypeError, ValueError): return None - - -def _parse_wfigs_contained(inner_data: dict) -> Optional[int]: - """Containment fallback chain: top-level PercentContained -> raw.* -> None.""" - val = _first_non_null(inner_data, _WFIGS_CONTAINED_KEYS) - if val is None: - raw = inner_data.get("raw") or {} - if isinstance(raw, dict): - val = _first_non_null(raw, _WFIGS_CONTAINED_RAW_KEYS) - if val is None: - return None - try: return int(round(float(val))) - except (TypeError, ValueError): return None - - -def _parse_wfigs_incidents(inner_data: dict, geo: dict) -> dict: - """Normalize a WFIGS-incidents payload into a flat render-ready dict. - - Field shapes per Central v0.10.0 guide (see /OneDrive/.../wfigs-investigation.md): - Top-level (incident): IrwinID, IncidentName, IncidentTypeCategory, - latitude, longitude, FireDiscoveryDateTime (epoch-ms), POOState, - POOCounty, DailyAcres, IncidentSize, PercentContained. - Nested raw dict (97-key): DiscoveryAcres, FinalAcres, PercentContained - (often the place where real values live in early season when the - top-level fields haven't populated yet). - _enriched.geocoder.landclass: optional ("Sawtooth National Forest", etc). - - Returns the normalized dict. Caller layers on "_kind": "wfigs_incident". - """ - geocoder = geo.get("geocoder") or {} - irwin_id = inner_data.get("IrwinID") or inner_data.get("irwin_id") - name = inner_data.get("IncidentName") - itype = inner_data.get("IncidentTypeCategory") - if itype is not None and itype not in ("WF", "wildfire"): - return None - lat = inner_data.get("latitude") - lon = inner_data.get("longitude") - county = inner_data.get("POOCounty") - state = inner_data.get("POOState") - landclass = geocoder.get("landclass") - - # FireDiscoveryDateTime is epoch-ms in WFIGS; convert to epoch-s. - declared_at_epoch = None - fdt = inner_data.get("FireDiscoveryDateTime") - if isinstance(fdt, (int, float)): - # Heuristic: anything >1e12 is ms (post-2001 in ms is ~1.4e12). - declared_at_epoch = int(fdt / 1000) if fdt > 1e12 else int(fdt) - - acres = _parse_wfigs_acres(inner_data) - contained_pct = _parse_wfigs_contained(inner_data) - - # Geocoder-side anchor enrichment for the renderer. - city = geocoder.get("city") - raw = inner_data.get("raw") or {} - - return { - "irwin_id": irwin_id, - "incident_name": name, - "incident_type": itype, - "acres": acres, - "contained_pct": contained_pct, - "lat": lat, - "lon": lon, - "county": county, - "state": state, - "landclass": landclass, - "geocoder_city": city, - "declared_at_epoch": declared_at_epoch, - "fire_cause": raw.get("FireCause"), - "agency": raw.get("POOJurisdictionalAgency"), - "personnel": raw.get("TotalIncidentPersonnel"), - "unique_fire_id": raw.get("UniqueFireIdentifier"), - } - - - -# ---------- itd_511 work_zone parser (v0.5.9 GAMMA) ---------------------- - -def _itd_ends_at(planned_end_epoch) -> Optional[datetime]: - """itd_511 stores planned_end_epoch as a Unix int (or None).""" - if not isinstance(planned_end_epoch, (int, float)) or planned_end_epoch <= 0: - return None - try: - return datetime.fromtimestamp(int(planned_end_epoch), tz=timezone.utc) - except (ValueError, OSError): - return None - - -def _parse_itd_511_work_zone(inner_data: dict, geo: dict) -> dict: - """Normalize an itd_511 work_zone (or closure-acting-as-work-zone) - envelope into the work_zone renderer's flat dict shape. - - Mirrors _parse_state_511_atis output: same keys, same town/distance - fallback chain. The renderer consumes both via format_work_zone_mesh. - """ - desc_raw = inner_data.get("description") or "" - desc = _clean_description(desc_raw) - mile_start, mile_end = _parse_mile_posts(desc or "") - - ends_at = _itd_ends_at(inner_data.get("planned_end_epoch")) - is_full = bool(inner_data.get("is_full_closure")) - impact = "full_closure" if is_full else "partial" - - road = normalize_road_name(inner_data.get("roadway_name")) - if _is_uninformative_road(road): - road = None - - event_lat = inner_data.get("latitude") - event_lon = inner_data.get("longitude") - if event_lat is None and geo.get("centroid"): - try: event_lon, event_lat = geo["centroid"][0], geo["centroid"][1] - except (IndexError, TypeError): pass - - enriched = (inner_data.get("_enriched") or {}).get("geocoder") or {} - town = (enriched.get("city") or "").strip() or None - distance_mi: Optional[int] = None - bearing: Optional[str] = None - if town: - distance_mi, bearing = _compute_distance_bearing(event_lat, event_lon, town) - else: - nt = nearest_town(event_lat, event_lon) if event_lat is not None else None - if nt: - town = nt.get("name") - distance_mi = nt.get("distance_mi") - bearing = nt.get("bearing") - - return { - "source": "itd_511", - "road": road, - "direction": _norm_direction(inner_data.get("direction")), - "mile_start": mile_start, - "mile_end": mile_end, - "description": desc, - "sub_type": _norm_sub_type(inner_data.get("event_sub_type")), - "impact": impact, - "ends_at": ends_at, - "town": town, - "distance_mi": distance_mi, - "bearing": bearing, - } - - -# ---------- public entry point -------------------------------------------- - -def normalize(envelope: dict) -> Optional[dict]: - """Normalize a Central CloudEvents envelope into a flat render-ready dict. - - Returns None if the adapter has no normalizer wired yet (caller falls - back to the existing meshai title path). - """ - if not isinstance(envelope, dict): return None - inner = envelope.get("data") or {} - adapter = inner.get("adapter") or "" - inner_data = inner.get("data") or {} - geo = inner.get("geo") or {} - - if adapter == "state_511_atis": - # Parser stays pure: returns parsed dict for ALL states. The - # v0.5.9 GAMMA Idaho-cutover decision lives in the consumer - # (skip + event_log handled=0 before dispatching here). See - # should_skip_state_511_atis_id() below for the test-friendly - # helper that the consumer uses. - return _parse_state_511_atis(inner_data, geo) - if adapter == "wzdx": - return _parse_wzdx_federal(inner_data, geo) - # v0.5.9 GAMMA: itd_511 work_zone parser (incident/closure/special_event - # still route through incident_handler per v0.5.9; work_zone is the - # only EventType that uses the work_zone renderer + Format). - if adapter == "itd_511": - if (inner.get("category") or "").startswith("work_zone."): - return _parse_itd_511_work_zone(inner_data, geo) - - # v0.5.8 WFIGS dispatch -- incidents + tombstones + perimeters. - # The handler downstream uses _kind to route to change-detection - # (active incidents) or to event_log-only logging (tombstones, - # perimeters). Tombstones carry only irwin_id + state + county; - # perimeters share the IrwinID with their parent incident. - category_raw = inner.get("category") or "" - if adapter == "wfigs_incidents": - if category_raw.startswith("fire.incident.removed"): - return { - "_kind": "wfigs_tombstone", - "irwin_id": inner_data.get("irwin_id") or inner_data.get("IrwinID"), - "state": inner_data.get("state") or inner_data.get("POOState"), - "county": inner_data.get("county") or inner_data.get("POOCounty"), - } - if category_raw.startswith("fire.incident"): - n = _parse_wfigs_incidents(inner_data, geo) - if n is None: - return None - n["_kind"] = "wfigs_incident" - return n - if adapter == "wfigs_perimeters": - return { - "_kind": "wfigs_perimeter", - "irwin_id": inner_data.get("irwin_id") or inner_data.get("IrwinID"), - "state": inner_data.get("state") or inner_data.get("POOState"), - "county": inner_data.get("county") or inner_data.get("POOCounty"), - } - - # Other adapters await per-adapter parsers; return None to defer. - return None - - -def should_skip_state_511_atis_id(envelope: dict) -> bool: - """v0.5.9 GAMMA decision helper: True when this envelope is a - state_511_atis publish for an Idaho event (state_code='ID' or - primary_region='US-ID'). - - Used by the consumer to decide 'skip + event_log handled=0' before - dispatching to either the work_zone renderer or the incident_handler. - Kept out of the parser so test_central_normalizer's existing ID - fixtures continue to exercise _parse_state_511_atis directly. - """ - if not isinstance(envelope, dict): - return False - inner = envelope.get("data") or {} - if (inner.get("adapter") or "") != "state_511_atis": - return False - d = inner.get("data") or {} - geo = inner.get("geo") or {} - return (d.get("state_code") == "ID" - or geo.get("primary_region") == "US-ID") - - - -# ---------- v0.5.9 GAMMA universal freshness helper ----------------------- - - -def _parse_iso_epoch_freshness(s: Optional[str]) -> Optional[int]: - """Local copy of the ISO parser used by the universal freshness gate. - Duplicated rather than imported from incident_handler so the dependency - graph stays one-directional (consumer -> central_normalizer).""" - if not s: return None - try: - from datetime import datetime as _dt - return int(_dt.fromisoformat(s.replace("Z", "+00:00")).timestamp()) - except Exception: - return None - - -def _parse_511_date_epoch_freshness(s: Optional[str]) -> Optional[int]: - if not s: return None - try: - from datetime import datetime as _dt, timezone as _tz - return int(_dt.strptime(s, "%m/%d/%y, %I:%M %p").replace( - tzinfo=_tz.utc).timestamp()) - except Exception: - return None - - -def is_incident_envelope_stale(envelope: dict, now: int, - max_age_s: int = 1800) -> bool: - """v0.5.9 GAMMA universal freshness gate. Returns True iff the envelope - should be DROPPED on freshness grounds. - - Per-source start-time fields: - tomtom_incidents -> inner.data.start_time (ISO-8601) - state_511_atis -> inner.data.start_date ("5/28/26, 10:45 PM") - itd_511 -> inner.data.start_epoch (Unix int) - other adapters -> None (default-allow; the gate has nothing to do) - - Two-sided check: 0 <= age <= max_age_s. Negative ages reject future- - scheduled events (e.g. itd_511 work_zone planned to start days from - now); ages > max_age_s reject stale events. None / missing start time - defaults to ALLOW so we err on the side of broadcasting potentially- - fresh data with incomplete metadata. - - Pure (no side effects); caller decides to log + skip when this returns - True. - """ - if not isinstance(envelope, dict): return False - inner = envelope.get("data") or {} - adapter = inner.get("adapter") or "" - d = inner.get("data") or {} - - se: Optional[int] = None - if adapter == "tomtom_incidents": - se = _parse_iso_epoch_freshness(d.get("start_time")) - elif adapter == "state_511_atis": - se = _parse_511_date_epoch_freshness(d.get("start_date")) - elif adapter == "itd_511": - val = d.get("start_epoch") - if isinstance(val, (int, float)) and val > 0: - se = int(val) - elif adapter == "nws": - # NWS CAP: prefer `sent` (issuance), fall back to `effective`. - se = (_parse_iso_epoch_freshness(d.get("sent")) - or _parse_iso_epoch_freshness(d.get("effective"))) - elif adapter == "usgs_quake": - val = d.get("time_ms") - if isinstance(val, (int, float)) and val > 0: - se = int(val / 1000) if val > 1e12 else int(val) - elif adapter in ("swpc_alerts", "swpc_kindex", "swpc_protons"): - # Generic time / issued_at field. - for k in ("time", "issued_at", "issue_time"): - v = d.get(k) - if isinstance(v, str): - se = _parse_iso_epoch_freshness(v) - if se is not None: break - elif isinstance(v, (int, float)) and v > 0: - se = int(v / 1000) if v > 1e12 else int(v) - break - else: - return False # adapter not in scope of this gate - - if se is None: - return False # default-allow on missing start time - age = now - se - return age < 0 or age > max_age_s diff --git a/work/meshai/env/fire_fusion.py b/work/meshai/env/fire_fusion.py index 3e835e3..649579b 100644 --- a/work/meshai/env/fire_fusion.py +++ b/work/meshai/env/fire_fusion.py @@ -968,7 +968,7 @@ def _render_growth_wire(*, incident_name, direction, speed_mph, """ near_part = "" try: - from meshai.central_normalizer import nearest_town + from meshai.geo import nearest_town nt = nearest_town(lat, lon, max_distance_mi=100.0) if nt and nt.get("name"): town = nt["name"] diff --git a/work/meshai/env/fire_render.py b/work/meshai/env/fire_render.py index d0937ee..968087b 100644 --- a/work/meshai/env/fire_render.py +++ b/work/meshai/env/fire_render.py @@ -39,6 +39,7 @@ inside that connection's autocommit mode. from __future__ import annotations from meshai.adapter_config import adapter_config +from meshai.geo import haversine_distance, _bearing_compass from meshai.notifications.formatters._budget import budget_for, fit_to_budget import logging @@ -531,15 +532,13 @@ def _location_anchor(n: dict) -> str: # Try curated town_anchors first try: from meshai.persistence import get_db - from meshai.central_normalizer import _haversine_miles as _haversine_mi - from meshai.central_normalizer import _bearing_compass rows = get_db().execute( "SELECT name, lat, lon FROM town_anchors WHERE lat IS NOT NULL AND lon IS NOT NULL" ).fetchall() best = None best_d = float("inf") for row in rows: - d = _haversine_mi(lat, lon, row["lat"], row["lon"]) + d = haversine_distance(lat, lon, row["lat"], row["lon"]) if d < best_d: best_d = d best = row @@ -553,7 +552,11 @@ def _location_anchor(n: dict) -> str: logger.exception("town_anchors lookup failed; falling back to Photon") try: - from meshai.central_normalizer import nearest_town + # Kept as a lazy, per-call import (not hoisted to the top-level + # geo import above) because several tests monkeypatch + # `meshai.geo.nearest_town` and rely on this function doing a + # fresh attribute lookup on every call to pick that up. + from meshai.geo import nearest_town nt = nearest_town(lat, lon, max_distance_mi=float(adapter_config.wfigs.anchor_max_mi)) except Exception: logger.exception("nearest_town failed; falling through") diff --git a/work/meshai/env/wzdx.py b/work/meshai/env/wzdx.py index 2818d41..1a03cf4 100644 --- a/work/meshai/env/wzdx.py +++ b/work/meshai/env/wzdx.py @@ -25,12 +25,11 @@ 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``. +Field mapping delegates to ``env.wzdx_parse._parse_wzdx_federal`` (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``. Coalescing ---------- @@ -58,10 +57,7 @@ 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 +from meshai.env.wzdx_parse import _parse_wzdx_federal if TYPE_CHECKING: from ..config import WZDxConfig @@ -372,7 +368,7 @@ class WZDxAdapter: """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 + ``env.wzdx_parse._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). diff --git a/work/meshai/env/wzdx_parse.py b/work/meshai/env/wzdx_parse.py new file mode 100644 index 0000000..276c1c0 --- /dev/null +++ b/work/meshai/env/wzdx_parse.py @@ -0,0 +1,273 @@ +"""WZDx (Work Zone Data Exchange) field-mapping/parsing helpers. + +Relocated from the deleted Central-envelope adapter-normalizer module (a +per-adapter Central-envelope shaper with no live production caller, removed +in chore/ripout-2e-geo-normalizer) — this module now owns the FHWA WZDx v4 + +work-zone-description parsing logic. Its sole consumer is +`meshai.env.wzdx.WZDxAdapter._parse_feature`, so it lives right next to the +adapter that uses it rather than in a generic normalizer module. + +`_parse_wzdx_federal(inner_data, geo) -> dict` is the entry point: it reads +either the raw WZDx `core_details.*` nesting or a flattened envelope and +returns a flat, render-ready dict (see the keys built at the bottom of +`_parse_wzdx_federal`) — road, direction, mile posts, folded sub_type, +impact, ends_at, and a town/distance/bearing anchor resolved via +`meshai.geo.nearest_town`. +""" + +import re +from datetime import datetime +from typing import Optional + +# ---------- direction normalization --------------------------------------- + +_DIR_MAP = { + "north": "northbound", "northbound": "northbound", "nb": "northbound", + "south": "southbound", "southbound": "southbound", "sb": "southbound", + "east": "eastbound", "eastbound": "eastbound", "eb": "eastbound", + "west": "westbound", "westbound": "westbound", "wb": "westbound", + "both": "both", "both directions": "both", + "unknown": "unknown", "": "unknown", +} + + +def _norm_direction(raw: Optional[str]) -> Optional[str]: + if raw is None: return None + s = str(raw).strip().lower() + return _DIR_MAP.get(s, "unknown") + + +# ---------- description parsers -------------------------------------------- + +# "from MM (93) to MM (89)" → (93, 89) +# "near MM (495)" → (495, None) +# "at MM (60)" → (60, None) +_MM_RE = re.compile( + r"(?:from\s+)?MM\s*\(?(\d+)\)?(?:\s*to\s+MM\s*\(?(\d+)\)?)?", + re.IGNORECASE, +) + + +def _parse_mile_posts(description: str) -> tuple[Optional[int], Optional[int]]: + if not description: return None, None + m = _MM_RE.search(description) + if not m: return None, None + try: + start = int(m.group(1)) + except (TypeError, ValueError): + return None, None + end = None + if m.group(2): + try: end = int(m.group(2)) + except (TypeError, ValueError): end = None + return start, end + + +# ---------- description cleanup ------------------------------------------- + +_HTML_TAG_RE = re.compile(r"<[^>]+>") + + +def _clean_description(raw: Optional[str]) -> Optional[str]: + if not raw: return None + s = _HTML_TAG_RE.sub(" ", str(raw)) + s = re.sub(r"\s+", " ", s).strip() + return s or None + + +# ---------- uninformative road-name detection ------------------------------ + +# Uninformative road names (Exit-only ramps with no parent route prefix +# visible) get dropped so the renderer leads with the town instead. +_UNINFORMATIVE_ROAD_RE = re.compile( + r"^Exit\s+\d+.*\b(On|Off)\s+Ramp$", + re.IGNORECASE, +) + + +def _is_uninformative_road(road: Optional[str]) -> bool: + if not road: + return False + return bool(_UNINFORMATIVE_ROAD_RE.match(str(road).strip())) + + +# ---------- wzdx federal vocabulary maps ---------------------------------- + +# FHWA WZDx v4 + custom-feed vocabulary observed in the wild. Unknown values +# fall through to lowercased + hyphens→spaces (see _norm_wzdx_sub_type). +_WZDX_WORK_TYPE_MAP: dict[str, Optional[str]] = { + # WZDx v4 spec types_of_work.type_name enum: + "maintenance": "maintenance", + "minor-road-defect-repair": "minor repair", + "roadside-work": "roadside work", + "overhead-work": "overhead work", + "below-road-work": "subsurface work", + "barrier-work": "barrier work", + "surface-work": "surface work", + "painting": "painting", + "roadway-relocation": "roadway relocation", + "roadway-creation": "new construction", + # Common informal values seen in upstream feeds (ID, WA): + "road-work": "road work", + "paving": "paving", + "bridge-construction": "bridge construction", + "bridge-maintenance": "bridge maintenance", + "utility-work": "utility work", + "road-construction": "road construction", + "construction": "construction", + "emergency-repairs": "emergency repairs", + # event_type values (drop the too-generic ones): + "work-zone": None, + "detour": "detour", +} + + +# vehicle_impact taxonomy (WZDx v4). Maps to mesh-friendly phrase. +# Returns None for values the renderer should drop entirely. +_WZDX_IMPACT_MAP: dict[str, Optional[str]] = { + "all-lanes-closed": "all lanes closed", + "some-lanes-closed": "lanes reduced", + "alternating-one-way": "one-way alternating", + "unknown": None, + "all-lanes-open": None, # informational only; nothing to do +} + + +def _norm_wzdx_sub_type(raw) -> Optional[str]: + if not raw: return None + s = str(raw).strip().lower() + if not s: return None + if s in _WZDX_WORK_TYPE_MAP: + return _WZDX_WORK_TYPE_MAP[s] + # Unknown value — keep lowercased, hyphens → spaces, single-line. + return re.sub(r"\s+", " ", s.replace("-", " ")).strip() or None + + +# ---------- entry point: wzdx federal ------------------------------------- + +def _parse_wzdx_federal(inner_data: dict, geo: dict) -> dict: + """Normalize a wzdx-adapter envelope (FHWA WZDx federal spec). + + Some feeds flatten the upstream payload in practice (the FHWA-spec + `core_details.*` nesting is not always preserved), so this defensively + checks nested keys too via the `field()` helper below. + + sub_type uses types_of_work[0].type_name when present, else event_type, + each normalized via _WZDX_WORK_TYPE_MAP. impact_phrase is folded INTO + the sub_type slot for the renderer (so the description-slot reads e.g. + 'lanes reduced, paving' or 'one-way alternating' or 'road work'). + 'all lanes closed' is set on impact='full_closure' so the renderer's + existing full-closure promotion handles it -- avoids double-printing. + """ + cd = inner_data.get("core_details") + if not isinstance(cd, dict): cd = {} + def field(key): + v = cd.get(key) + if v is None or (isinstance(v, str) and not v.strip()): + v = inner_data.get(key) + return v + + # --- road (raw, verbatim per Matt's spec) ----------------------------- + road_names = field("road_names") + road = None + if isinstance(road_names, list) and road_names: + road = str(road_names[0]).strip() or None + elif isinstance(road_names, str) and road_names.strip(): + road = road_names.strip() + if _is_uninformative_road(road): + road = None + + # --- direction -------------------------------------------------------- + direction = _norm_direction(field("direction")) + + # --- sub_type (types_of_work[0] | event_type) ------------------------- + work_type: Optional[str] = None + tow = field("types_of_work") + if isinstance(tow, list) and tow: + first = tow[0] + if isinstance(first, dict): + work_type = _norm_wzdx_sub_type(first.get("type_name")) + elif isinstance(first, str): + work_type = _norm_wzdx_sub_type(first) + if not work_type: + work_type = _norm_wzdx_sub_type(field("event_type")) + + # --- vehicle_impact --------------------------------------------------- + vi_raw = (inner_data.get("vehicle_impact") or cd.get("vehicle_impact") or "") + impact_phrase: Optional[str] = _WZDX_IMPACT_MAP.get(str(vi_raw).strip().lower()) + is_full_closure = (str(vi_raw).strip().lower() == "all-lanes-closed") + + # Fold impact_phrase + work_type into the renderer's sub_type slot. + # For full-closure, exclude impact_phrase here -- the renderer prepends + # "all lanes closed" itself via the impact='full_closure' branch. + parts: list[str] = [] + if impact_phrase and not is_full_closure: + parts.append(impact_phrase) + if work_type: + parts.append(work_type) + sub_type = ", ".join(parts) if parts else None + impact = "full_closure" if is_full_closure else "partial" + + # --- ends_at: structured end_date ISO-8601 --------------------------- + ends_at: Optional[datetime] = None + end_date = inner_data.get("end_date") or cd.get("end_date") + if end_date: + try: + s = str(end_date).replace("Z", "+00:00") + ends_at = datetime.fromisoformat(s) + # Strip tzinfo so _format_end_short compares naive-to-naive. + if ends_at.tzinfo is not None: + ends_at = ends_at.astimezone().replace(tzinfo=None) + except Exception: + ends_at = None + + # --- mile_start/_end: regex on description, fall back to structured -- + desc = _clean_description(field("description")) + mile_start, mile_end = _parse_mile_posts(desc or "") + if mile_start is None: + ms = inner_data.get("road_mile_post_start") + if ms is not None: + try: mile_start = int(ms) + except (TypeError, ValueError): pass + if mile_end is None: + me = inner_data.get("road_mile_post_end") + if me is not None: + try: mile_end = int(me) + except (TypeError, ValueError): pass + + # --- coordinates ----------------------------------------------------- + event_lat = inner_data.get("latitude") + event_lon = inner_data.get("longitude") + if event_lat is None and geo.get("centroid"): + try: event_lon, event_lat = geo["centroid"][0], geo["centroid"][1] + except (IndexError, TypeError): pass + + # --- town fallback chain: geocoder.city, else Photon nearest_town ---- + enriched = (inner_data.get("_enriched") or {}).get("geocoder") or {} + town = (enriched.get("city") or "").strip() or None + distance_mi: Optional[int] = None + bearing: Optional[str] = None + from meshai.geo import nearest_town, _compute_distance_bearing + if town: + distance_mi, bearing = _compute_distance_bearing(event_lat, event_lon, town) + elif event_lat is not None: + nt = nearest_town(event_lat, event_lon) + if nt: + town = nt.get("name") + distance_mi = nt.get("distance_mi") + bearing = nt.get("bearing") + + return { + "source": "wzdx", + "road": road, + "direction": direction, + "mile_start": mile_start, + "mile_end": mile_end, + "description": desc, + "sub_type": sub_type, + "impact": impact, + "ends_at": ends_at, + "town": town, + "distance_mi": distance_mi, + "bearing": bearing, + } diff --git a/work/meshai/geo.py b/work/meshai/geo.py index 6f39457..1aead2d 100644 --- a/work/meshai/geo.py +++ b/work/meshai/geo.py @@ -1,7 +1,22 @@ -"""Geographic utilities for mesh clustering and naming.""" +"""Geographic utilities: mesh clustering/naming, plus event-to-town +resolution (haversine distance/bearing, Photon reverse-geocoding, H3-cached +nearest-populated-place lookup). +The nearest_town()/Photon section below was relocated from the now-deleted +Central-envelope adapter-normalizer module (a per-adapter Central-envelope +shaper with no live production caller, removed in chore/ripout-2e-geo- +normalizer) — it has no adapter-shape logic of its own, just generic +lat/lon → place-name resolution, so it belongs here next to the existing +`nearest_city()` hardcoded-table lookup. +""" + +import json import logging import math +import urllib.error +import urllib.parse +import urllib.request +from collections import OrderedDict from typing import Optional logger = logging.getLogger(__name__) @@ -295,3 +310,199 @@ def assign_to_nearest_cluster( nearest_idx = i return nearest_idx + + +# ── event → town resolution (distance/bearing + Photon reverse-geocoding) ── +# +# Relocated verbatim from the deleted Central-envelope adapter-normalizer +# module. Distinct from nearest_city() above: nearest_city() picks the +# closest entry in the small hardcoded CITY_LOOKUP table; nearest_town() +# below calls the live Photon reverse-geocoder for ANY populated place +# worldwide, H3-cell-cached. Different implementations for different jobs — +# both kept. +# +# NOTE on de-duplication: that module had its own `_haversine_miles` +# helper, algebraically identical to `haversine_distance` above (same +# haversine formula; `_haversine_miles` closed via `asin`, `haversine_distance` +# via the equivalent, more numerically-stable `atan2` form — same distances +# to floating-point precision for every realistic input). That was a TRUE +# duplicate, so it was dropped in the move: the functions below call +# `haversine_distance` directly instead of carrying a second copy. + + +def _bearing_compass(lat1: float, lon1: float, lat2: float, lon2: float) -> str: + """Compass bearing FROM (lat2, lon2) TO (lat1, lon1) -- i.e., 'event is + of town'. We orient so the event's bearing relative to the + town reads naturally ("8 mi N of Plummer" = event is north of Plummer).""" + phi1, phi2 = math.radians(lat2), math.radians(lat1) + dl = math.radians(lon1 - lon2) + x = math.sin(dl) * math.cos(phi2) + y = math.cos(phi1) * math.sin(phi2) - math.sin(phi1) * math.cos(phi2) * math.cos(dl) + brng = (math.degrees(math.atan2(x, y)) + 360) % 360 + points = ["N", "NE", "E", "SE", "S", "SW", "W", "NW"] + return points[int((brng + 22.5) // 45) % 8] + + +def _compute_distance_bearing( + event_lat: Optional[float], event_lon: Optional[float], town: Optional[str] +) -> tuple[Optional[int], Optional[str]]: + """Distance (rounded mi) + compass bearing from a named, curated town + anchor (meshai.persistence.curation.lookup_town_anchor) to the event + coords. Returns (None, None) if the town isn't in the curated table.""" + if event_lat is None or event_lon is None or not town: + return None, None + key = str(town).strip().lower() + from meshai.persistence.curation import lookup_town_anchor + coords = lookup_town_anchor(key) + if coords is None: + return None, None + tlat, tlon = coords + d = haversine_distance(event_lat, event_lon, tlat, tlon) + b = _bearing_compass(event_lat, event_lon, tlat, tlon) + return int(round(d)), b + + +# Photon is reachable from CT108 at this Tailscale address (verified +# 2026-06-04). It's the same Echo6-local Photon instance that backs Central's +# (now-retired) NaviBackend reverse-geocoder. Photon takes osm_tag=place (KEY +# only, not key:value with comma-list -- that returns 0 features -- per probe). +# Geocoder config is set via init_geocoder_config(); defaults to public +# Komoot Photon, deployments override in config.yaml. + +class _GeocoderSettings: + url: str = "https://photon.komoot.io" + timeout_seconds: float = 2.0 + radius_km: float = 80.0 + limit: int = 10 + +_geocoder = _GeocoderSettings() + + +def init_geocoder_config(url: str = None, timeout: float = None, + radius: float = None, limit: int = None) -> None: + """Initialize geocoder settings from config.yaml values.""" + if url is not None: + _geocoder.url = url + if timeout is not None: + _geocoder.timeout_seconds = timeout + if radius is not None: + _geocoder.radius_km = radius + if limit is not None: + _geocoder.limit = limit + + +# OSM place classes we accept as "town". Suburb included for metro coverage; +# locality is rare but valid for tiny rural places. +_TOWN_OSM_VALUES = frozenset({"city", "town", "village"}) + + +# Process-lifetime LRU cache keyed by H3 cell (resolution 7 ≈ 5km hexagons). +# Cells don't move and Photon's reverse output for a coord is stable, so +# entries never expire within a process lifetime. Cap at 10k entries. +_H3_CACHE_RESOLUTION = 7 +_H3_CACHE_MAX = 10_000 +_h3_cache: "OrderedDict[str, Optional[dict]]" = OrderedDict() + + +def _h3_cell(lat: float, lon: float) -> Optional[str]: + try: + import h3 # local import: keep module-import-time h3-free + return h3.latlng_to_cell(lat, lon, _H3_CACHE_RESOLUTION) + except Exception: + # Fallback: coarse-grain by rounding coords (~1.1 km per 0.01 deg). + return f"fallback:{round(lat, 2)},{round(lon, 2)}" + + +def _photon_reverse_places(lat: float, lon: float) -> list[dict]: + """Call Photon /reverse with osm_tag=place. Return raw feature list.""" + qs = urllib.parse.urlencode({ + "lat": f"{lat:.6f}", + "lon": f"{lon:.6f}", + "radius": _geocoder.radius_km, + "osm_tag": "place", + "limit": _geocoder.limit, + }) + url = f"{_geocoder.url}/reverse?{qs}" + try: + with urllib.request.urlopen(url, timeout=_geocoder.timeout_seconds) as resp: + body = resp.read() + d = json.loads(body) + except (urllib.error.URLError, urllib.error.HTTPError, TimeoutError, + json.JSONDecodeError, ConnectionError) as e: + logger.debug("Photon /reverse failed (%s) for %.4f,%.4f", e, lat, lon) + return [] + feats = d.get("features") or [] + return feats if isinstance(feats, list) else [] + + +def nearest_town(lat: float, lon: float, max_distance_mi: float = 50.0) -> Optional[dict]: + """Return the nearest populated place to (lat, lon) within max_distance_mi. + + Result shape: {name: str, distance_mi: int (rounded), bearing: str} + where bearing is an 8-point compass (N/NE/E/SE/S/SW/W/NW) of the event + location relative to the town -- i.e. "8 mi N of Plummer" means the + event is N of the town. Returns None if no town within range or if + Photon is unreachable. + + Calls Photon /reverse?osm_tag=place at _geocoder.url. Results are + H3-cell-cached (resolution 7 ≈ 5 km cells) so the second event near + the same town is free. + """ + if lat is None or lon is None: + return None + try: + lat, lon = float(lat), float(lon) + except (TypeError, ValueError): + return None + + cell = _h3_cell(lat, lon) + if cell is not None and cell in _h3_cache: + # LRU touch + _h3_cache.move_to_end(cell) + cached = _h3_cache[cell] + if cached is None or cached.get("distance_mi", 999) <= max_distance_mi: + return cached + + feats = _photon_reverse_places(lat, lon) + candidates: list[tuple[float, dict]] = [] + for f in feats: + p = f.get("properties") or {} + # Only accept proper populated places. + if p.get("osm_key") != "place" or p.get("osm_value") not in _TOWN_OSM_VALUES: + continue + coords = (f.get("geometry") or {}).get("coordinates") + if not (isinstance(coords, list) and len(coords) >= 2): + continue + tlon, tlat = coords[0], coords[1] + try: + tlat, tlon = float(tlat), float(tlon) + except (TypeError, ValueError): + continue + d_mi = haversine_distance(lat, lon, tlat, tlon) + if d_mi > max_distance_mi: + continue + name = p.get("name") + if not name: + continue + candidates.append((d_mi, { + "name": str(name), + "distance_mi": int(round(d_mi)), + "bearing": _bearing_compass(lat, lon, tlat, tlon), + })) + + if not candidates: + if cell is not None: + _h3_cache[cell] = None + _h3_cache.move_to_end(cell) + while len(_h3_cache) > _H3_CACHE_MAX: + _h3_cache.popitem(last=False) + return None + + candidates.sort(key=lambda kv: kv[0]) + result = candidates[0][1] + if cell is not None: + _h3_cache[cell] = result + _h3_cache.move_to_end(cell) + while len(_h3_cache) > _H3_CACHE_MAX: + _h3_cache.popitem(last=False) + return result diff --git a/work/meshai/main.py b/work/meshai/main.py index 6f30961..33866bf 100644 --- a/work/meshai/main.py +++ b/work/meshai/main.py @@ -19,7 +19,7 @@ from .config import Config from .config_loader import load_config, get_config_dir_from_path from .connector import MeshConnector, MeshMessage from .transport.factory import build_transport -from .central_normalizer import init_geocoder_config +from .geo import init_geocoder_config from .context import MeshContext from .history import ConversationHistory from .memory import ConversationSummary diff --git a/work/meshai/notifications/formatters/_anchor.py b/work/meshai/notifications/formatters/_anchor.py index 3a367ed..10960e1 100644 --- a/work/meshai/notifications/formatters/_anchor.py +++ b/work/meshai/notifications/formatters/_anchor.py @@ -6,15 +6,15 @@ Returns {town: str, distance_mi: int, bearing: str} or None. Priority: 1. Curated ``town_anchors`` SQLite table (GUI-managed, haversine nearest). - 2. Photon reverse geocoder via central_normalizer.nearest_town() fallback. + 2. Photon reverse geocoder via geo.nearest_town() fallback. The function is PURE (no side-effects beyond the LRU cache inside nearest_town) and safe to call from any formatter. Both the incident formatter (Phase 2) and the WFIGS fire formatter (Phase 3) use it. Haversine and bearing implementations are local copies so the module -has no runtime dependency on central_normalizer (avoids the circular -import chain formatters → central_normalizer → persistence → formatters). +has no runtime dependency on meshai.geo at import time (avoids the +circular import chain formatters → geo → persistence → formatters). """ from __future__ import annotations @@ -25,7 +25,7 @@ from typing import Optional logger = logging.getLogger(__name__) -# ── Geometry helpers (mirrors central_normalizer exactly) ───────────────── +# ── Geometry helpers (mirrors meshai.geo exactly) ────────────────────────── def _haversine_miles(lat1: float, lon1: float, lat2: float, lon2: float) -> float: @@ -44,7 +44,7 @@ def _bearing_compass( Result is the direction the event lies relative to the town, so '8 mi N of Plummer' means the event is north of the town. - Mirrors central_normalizer._bearing_compass and wfigs_handler._location_anchor. + Mirrors meshai.geo._bearing_compass and env.fire_render._location_anchor. """ phi1, phi2 = math.radians(lat2), math.radians(lat1) dl = math.radians(lon1 - lon2) @@ -69,8 +69,8 @@ def resolve_anchor( 1. Curated ``town_anchors`` SQLite table — haversine nearest row with ``lat IS NOT NULL AND lon IS NOT NULL``. GUI-managed and always tried first so curated entries beat the Photon fallback. - 2. ``central_normalizer.nearest_town()`` Photon reverse-geocoder — - same H3-cached Photon call used by the normalizer's town selection. + 2. ``geo.nearest_town()`` Photon reverse-geocoder — same H3-cached + Photon call used by the WZDx/state_511_atis town-selection chain. Parameters ---------- @@ -117,7 +117,7 @@ def resolve_anchor( # ── 2. Photon nearest_town fallback ────────────────────────────────── try: - from meshai.central_normalizer import nearest_town + from meshai.geo import nearest_town nt = nearest_town(lat, lon, max_distance_mi=max_mi) if nt and nt.get("name"): return { diff --git a/work/meshai/notifications/renderers/composer.py b/work/meshai/notifications/renderers/composer.py index 7adf1b7..36a484b 100644 --- a/work/meshai/notifications/renderers/composer.py +++ b/work/meshai/notifications/renderers/composer.py @@ -323,10 +323,9 @@ def compose_mesh_message(event: Event) -> str: to fit the budget; never mid-codepoint truncation. OPTION A bypass: if `event.data["_meshai_precomposed"]` is truthy, the - title is already a fully formatted mesh string from the per-adapter - normalizer (meshai/central_normalizer.py + the work_zone renderer). - Return it verbatim -- no family-label prefix, no region tail, no - severity word append. + title is already a fully formatted mesh string from a per-adapter + normalizer/renderer. Return it verbatim -- no family-label prefix, no + region tail, no severity word append. """ # Phase-1+ formatter dispatch — gated on MESHAI_CUTOVER_CATEGORIES. # A formatter registered for an event's category is only called in the diff --git a/work/meshai/persistence/curation.py b/work/meshai/persistence/curation.py index 98e7346..0c69430 100644 --- a/work/meshai/persistence/curation.py +++ b/work/meshai/persistence/curation.py @@ -5,7 +5,8 @@ adapter_config: created by a migration, seeded from Python data on first boot, then runtime reads from SQLite via cached accessors. gauge_sites replaces idaho_gauge_sites.IDAHO_CURATED_SITES. -town_anchors replaces central_normalizer._TOWN_COORDS. +town_anchors replaces a hardcoded _TOWN_COORDS table that used to live in +the (since-deleted) Central-envelope adapter-normalizer module. """ from __future__ import annotations @@ -81,7 +82,8 @@ _GAUGE_SITES_SEED: dict[str, dict[str, Any]] = { }, } -# Idaho + neighbor towns originally from central_normalizer._TOWN_COORDS. +# Idaho + neighbor towns, originally from a hardcoded _TOWN_COORDS table in +# the (since-deleted) Central-envelope adapter-normalizer module. _TOWN_ANCHORS_SEED: dict[str, dict[str, Any]] = { "boise": {"lat": 43.6150, "lon": -116.2023, "state": "ID"}, "meridian": {"lat": 43.6121, "lon": -116.3915, "state": "ID"}, diff --git a/work/tests/test_adapter_wzdx.py b/work/tests/test_adapter_wzdx.py index b347324..4a2a229 100644 --- a/work/tests/test_adapter_wzdx.py +++ b/work/tests/test_adapter_wzdx.py @@ -13,7 +13,7 @@ from types import SimpleNamespace import pytest -import meshai.central_normalizer as cn +from meshai import geo from meshai.env.wzdx import WZDxAdapter from meshai.notifications.events import Event from meshai.notifications.formatters.incident import format as format_incident @@ -26,7 +26,7 @@ from meshai.notifications.formatters.incident import format as format_incident @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) + monkeypatch.setattr(geo, "nearest_town", lambda *a, **k: None) @pytest.fixture diff --git a/work/tests/test_central_normalizer.py b/work/tests/test_central_normalizer.py deleted file mode 100644 index 6d329e1..0000000 --- a/work/tests/test_central_normalizer.py +++ /dev/null @@ -1,804 +0,0 @@ -"""Tests for meshai/central_normalizer.py — adapter-specific envelope -normalization. First adapter wired: state_511_atis.""" - -import json -from datetime import datetime -from pathlib import Path - -import pytest - -from meshai.central_normalizer import normalize - - -FIXTURES = Path(__file__).parent / "fixtures" / "central_envelopes" - - -def _load(name: str) -> dict: - return json.loads((FIXTURES / name).read_text()) - - -def _norm_fixture(name: str) -> dict: - n = normalize(_load(name)) - assert n is not None, f"normalize({name}) returned None" - return n - - -# ---------- adapter dispatch ----------------------------------------------- - - -def test_normalize_returns_none_for_unknown_adapter(): - env = {"data": {"adapter": "totally_made_up", "data": {}}} - assert normalize(env) is None - - -def test_normalize_returns_none_for_non_envelope(): - assert normalize(None) is None - assert normalize("not-a-dict") is None - assert normalize([]) is None - - -# ---------- state_511_atis: MM-range fixture (I-15 SB 93→89) -------------- - - -def test_mm_range_extracted_high_to_low(): - n = _norm_fixture("state_511_atis_01_I-15.json") - assert n["source"] == "state_511_atis" - assert n["road"] == "I-15" - assert n["direction"] == "southbound" - assert n["mile_start"] == 93 - assert n["mile_end"] == 89 # decreasing range is valid for SB I-15 - assert n["impact"] == "partial" - assert n["sub_type"] == "road construction" - assert isinstance(n["description"], str) and "MM (93)" in n["description"] - - -def test_mm_range_extracted_low_to_high(): - n = _norm_fixture("state_511_atis_03_I-15.json") - assert n["road"] == "I-15" - assert n["direction"] == "northbound" - assert n["mile_start"] == 89 - assert n["mile_end"] == 93 - assert n["sub_type"] == "bridge construction" - - -# ---------- state_511_atis: MM-near (single mile post) -------------------- - - -def test_mm_near_single_mile_post(): - n = _norm_fixture("state_511_atis_04_US-95.json") - assert n["road"] == "US-95" - assert n["direction"] == "southbound" - assert n["mile_start"] == 495 - assert n["mile_end"] is None - assert n["sub_type"] == "utility work" - - -# ---------- state_511_atis: no MM (cross-street / landmark) --------------- - - -def test_no_mm_in_description_yields_none_mile_posts(): - n = _norm_fixture("state_511_atis_05_W_Prairie_Ave.json") - assert n["mile_start"] is None - assert n["mile_end"] is None - assert n["road"] == "W Prairie Ave" - assert n["direction"] == "both" - - -def test_no_mm_emergency_repairs_landmark(): - n = _norm_fixture("state_511_atis_06_SH-55.json") - assert n["mile_start"] is None - assert n["road"] == "SH-55" - assert n["direction"] == "both" - assert n["sub_type"] == "emergency repairs" - - -# ---------- impact (full_closure vs partial) ------------------------------ - - -def test_partial_impact_for_lane_restriction(): - n = _norm_fixture("state_511_atis_01_I-15.json") - assert n["impact"] == "partial" - - -def test_full_closure_impact(): - # Synthetic — we didn't capture a full closure in the 60-sample probe, - # so build one inline to exercise the branch. - env = { - "data": { - "adapter": "state_511_atis", - "category": "closure.state_511_atis", - "data": { - "roadway_name": "I-15", - "direction": "South", - "description": "Road construction on I-15 Southbound near Northgate Pkwy. " - "All lanes closed. 6/1/2026 7:00 AM to 6/10/2026 5:00 PM.", - "event_sub_type": "roadConstruction", - "is_full_closure": True, - "county": "Bannock", - "latitude": 42.8713, - "longitude": -112.4455, - }, - }, - } - n = normalize(env) - assert n["impact"] == "full_closure" - - -# ---------- direction normalization --------------------------------------- - - -@pytest.mark.parametrize("raw,expected", [ - ("North", "northbound"), - ("south", "southbound"), - ("Both", "both"), - ("East", "eastbound"), - ("West", "westbound"), - ("Unknown", "unknown"), - ("", "unknown"), - ("NB", "northbound"), - (None, None), -]) -def test_direction_normalization(raw, expected): - env = {"data": {"adapter": "state_511_atis", "category": "work_zone.state_511_atis", - "data": {"roadway_name": "X", "direction": raw, "description": ""}}} - n = normalize(env) - assert n["direction"] == expected - - -# ---------- ends_at parsing ----------------------------------------------- - - -def test_ends_at_parsed_from_description(): - n = _norm_fixture("state_511_atis_04_US-95.json") - assert isinstance(n["ends_at"], datetime) - assert n["ends_at"].month == 6 and n["ends_at"].day == 2 - assert n["ends_at"].hour == 17 # 5 PM - - -def test_ends_at_missing_when_no_date_range(): - env = {"data": {"adapter": "state_511_atis", "category": "work_zone.state_511_atis", - "data": {"roadway_name": "X", "direction": "Both", - "description": "Just some text with no date."}}} - n = normalize(env) - assert n["ends_at"] is None - - -# ---------- _enriched geocoder + town ------------------------------------- - - -def test_town_from_geocoder_city(): - # Use a fixture and check town came from geocoder city/name. - n = _norm_fixture("state_511_atis_01_I-15.json") - assert isinstance(n["town"], str) and n["town"] - - -def test_town_missing_when_no_enriched(): - env = {"data": {"adapter": "state_511_atis", "category": "work_zone.state_511_atis", - "data": {"roadway_name": "X", "direction": "Both", "description": ""}}} - n = normalize(env) - assert n["town"] is None - assert n["distance_mi"] is None - assert n["bearing"] is None - - -def test_distance_bearing_when_town_in_lookup(): - # A known town (Idaho Falls) at known coords; event placed 8 mi north. - env = {"data": {"adapter": "state_511_atis", "category": "work_zone.state_511_atis", - "data": {"roadway_name": "US-20", "direction": "Both", - "description": "Test event", - "_enriched": {"geocoder": {"city": "Idaho Falls"}}, - "latitude": 43.4666 + 8.0 / 69.0, # ~8 mi north - "longitude": -112.0340}}} - n = normalize(env) - assert n["town"] == "Idaho Falls" - assert n["distance_mi"] is not None - assert 7 <= n["distance_mi"] <= 9 # ~8 mi - assert n["bearing"] == "N" - - -def test_distance_none_when_town_not_in_lookup(): - env = {"data": {"adapter": "state_511_atis", "category": "work_zone.state_511_atis", - "data": {"roadway_name": "X", "direction": "Both", - "description": "Test event", - "_enriched": {"geocoder": {"city": "Unknownsville"}}, - "latitude": 43.0, "longitude": -116.0}}} - n = normalize(env) - assert n["town"] == "Unknownsville" - assert n["distance_mi"] is None - assert n["bearing"] is None - - -# ---------- v0.5.8 normalize_road_name (SB/NB/EB/WB → S/N/E/W) ------------ - -from meshai.central_normalizer import normalize_road_name, nearest_town - -@pytest.mark.parametrize("raw,expected", [ - ("I-15 SB Off Ramp", "I-15 S Off Ramp"), - ("I-15 NB Off Ramp", "I-15 N Off Ramp"), - ("US-95 NB", "US-95 N"), - ("SH-55 EB", "SH-55 E"), - ("Exit 80 WB On Ramp", "Exit 80 W On Ramp"), - ("I-86-BL", "I-86-BL"), # no SB/NB token; untouched - ("I-15", "I-15"), - ("", None), - (None, None), -]) -def test_normalize_road_name(raw, expected): - assert normalize_road_name(raw) == expected - - -# ---------- v0.5.8 nearest_town: Photon + H3 cache ------------------------ - -# Photon /reverse?osm_tag=place returns features like: -_PHOTON_STANLEY = { - "features": [ - {"geometry": {"coordinates": [-114.9378523, 44.2161414]}, - "properties": {"name": "Stanley", "osm_key": "place", "osm_value": "city"}}, - ], -} -_PHOTON_MULTI = { - "features": [ - # Closer but a "natural" feature -- must NOT be picked (not a place). - {"geometry": {"coordinates": [-114.93, 44.2155]}, - "properties": {"name": "Mountain Village Restaurant", "osm_key": "amenity", "osm_value": "restaurant"}}, - # Town (~1km away). - {"geometry": {"coordinates": [-114.9378523, 44.2161414]}, - "properties": {"name": "Stanley", "osm_key": "place", "osm_value": "city"}}, - # Town further out. - {"geometry": {"coordinates": [-115.0588585, 44.2436215]}, - "properties": {"name": "Lake Town", "osm_key": "place", "osm_value": "village"}}, - ], -} - - -def _clear_h3_cache(): - from meshai.central_normalizer import _h3_cache - _h3_cache.clear() - - -def test_nearest_town_returns_dict_for_known_coord(monkeypatch): - _clear_h3_cache() - from meshai import central_normalizer as cn - monkeypatch.setattr(cn, "_photon_reverse_places", - lambda lat, lon: _PHOTON_STANLEY["features"]) - n = nearest_town(44.2160, -114.9311) - assert n is not None - assert n["name"] == "Stanley" - assert n["distance_mi"] >= 0 and n["distance_mi"] <= 1 - assert n["bearing"] in {"N", "NE", "E", "SE", "S", "SW", "W", "NW"} - - -def test_nearest_town_filters_non_place_osm_values(monkeypatch): - _clear_h3_cache() - from meshai import central_normalizer as cn - # Only the restaurant; no place tag at all. - monkeypatch.setattr(cn, "_photon_reverse_places", - lambda lat, lon: [ - {"geometry": {"coordinates": [-114.93, 44.2155]}, - "properties": {"name": "Restaurant", - "osm_key": "amenity", "osm_value": "restaurant"}}, - ]) - assert nearest_town(44.2160, -114.9311) is None - - -def test_nearest_town_picks_closest_place(monkeypatch): - _clear_h3_cache() - from meshai import central_normalizer as cn - monkeypatch.setattr(cn, "_photon_reverse_places", - lambda lat, lon: _PHOTON_MULTI["features"]) - n = nearest_town(44.2160, -114.9311) - assert n is not None - assert n["name"] == "Stanley" # closer than Lake Town - - -def test_nearest_town_returns_none_beyond_max_distance(monkeypatch): - _clear_h3_cache() - from meshai import central_normalizer as cn - monkeypatch.setattr(cn, "_photon_reverse_places", - lambda lat, lon: _PHOTON_STANLEY["features"]) - # Event 200 mi from Stanley; max_distance_mi=50 by default. - far_lat = 44.2160 + 200 / 69.0 - n = nearest_town(far_lat, -114.9311) - assert n is None - - -def test_nearest_town_returns_none_on_photon_failure(monkeypatch): - _clear_h3_cache() - from meshai import central_normalizer as cn - monkeypatch.setattr(cn, "_photon_reverse_places", lambda lat, lon: []) - assert nearest_town(44.2160, -114.9311) is None - - -def test_nearest_town_caches_via_h3(monkeypatch): - _clear_h3_cache() - from meshai import central_normalizer as cn - calls = [] - def stub(lat, lon): - calls.append((lat, lon)) - return _PHOTON_STANLEY["features"] - monkeypatch.setattr(cn, "_photon_reverse_places", stub) - # Two calls at the same coord → only one Photon hit. - nearest_town(44.2160, -114.9311) - nearest_town(44.2160, -114.9311) - assert len(calls) == 1 - - -def test_nearest_town_handles_none_inputs(): - _clear_h3_cache() - assert nearest_town(None, -114.9311) is None - assert nearest_town(44.2160, None) is None - - -# ---------- v0.5.8 town fallback chain in _parse_state_511_atis ------------ - -def test_town_uses_geocoder_city_when_present(monkeypatch): - _clear_h3_cache() - from meshai import central_normalizer as cn - photon_calls = [] - monkeypatch.setattr(cn, "_photon_reverse_places", - lambda lat, lon: photon_calls.append("called") or []) - env = {"data": {"adapter": "state_511_atis", "category": "work_zone.state_511_atis", - "data": {"roadway_name": "I-15", "direction": "South", - "description": "construction", - "_enriched": {"geocoder": {"city": "Idaho Falls"}}, - "latitude": 43.4666, "longitude": -112.0340}}} - n = normalize(env) - assert n["town"] == "Idaho Falls" - # When city is present, nearest_town should NOT be called. - assert photon_calls == [] - - -def test_town_falls_back_to_nearest_town_when_city_null(monkeypatch): - _clear_h3_cache() - from meshai import central_normalizer as cn - monkeypatch.setattr(cn, "_photon_reverse_places", - lambda lat, lon: _PHOTON_STANLEY["features"]) - env = {"data": {"adapter": "state_511_atis", "category": "work_zone.state_511_atis", - "data": {"roadway_name": "ID 21", "direction": "Both", - "description": "construction", - "_enriched": {"geocoder": {"city": None, "name": "Some Trail"}}, - "latitude": 44.2160, "longitude": -114.9311}}} - n = normalize(env) - assert n["town"] == "Stanley" - - -def test_town_is_none_when_city_and_photon_both_fail(monkeypatch): - _clear_h3_cache() - from meshai import central_normalizer as cn - monkeypatch.setattr(cn, "_photon_reverse_places", lambda lat, lon: []) - env = {"data": {"adapter": "state_511_atis", "category": "work_zone.state_511_atis", - "data": {"roadway_name": "X", "direction": "Both", - "description": "x", - "_enriched": {"geocoder": {"city": None, "name": "Old Road"}}, - "latitude": 44.2160, "longitude": -114.9311}}} - n = normalize(env) - assert n["town"] is None - assert n["distance_mi"] is None - assert n["bearing"] is None - - -def test_geocoder_name_is_never_used_as_town_fallback(monkeypatch): - """Per Matt's locked plan: geocoder.name is forbidden as a town fallback. - Only geocoder.city (PRIMARY) or nearest_town() (SECONDARY) populate it.""" - _clear_h3_cache() - from meshai import central_normalizer as cn - monkeypatch.setattr(cn, "_photon_reverse_places", lambda lat, lon: []) - env = {"data": {"adapter": "state_511_atis", "category": "work_zone.state_511_atis", - "data": {"roadway_name": "SH-3", "direction": "Both", - "description": "x", - "_enriched": {"geocoder": {"city": None, - "name": "Cache Nf Road 444"}}, - "latitude": 42.2, "longitude": -113.7}}} - n = normalize(env) - # Must NOT pick up "Cache Nf Road 444" from geocoder.name. - assert n["town"] is None - - -# ============================================================================ -# v0.5.8-wzdx federal parser tests -# ============================================================================ - -# --- representative envelopes (flat shape, as Central actually publishes) --- - -_WZDX_ID_FULL = { - "data": { - "adapter": "wzdx", - "category": "work_zone.wzdx", - "time": "2026-06-01T13:00:00Z", - "severity": 3, - "geo": {"centroid": [-112.408309608311, 43.0208066348276], - "primary_region": "US-ID", "regions": ["US-ID"]}, - "data": { - "road_names": ["Exit 80 On Ramp"], - "direction": "southbound", - "description": " Road construction on Exit 80 On Ramp Southbound near MM (80)." - " All lanes closed. 6/1/2026 7:00 AM to 6/10/2026 6:00 PM Mon, Tue ...", - "vehicle_impact": "all-lanes-closed", - "event_status": None, - "start_date": "2026-06-01T13:00:00Z", - "end_date": "2026-06-11T00:00:00Z", - "data_source_id": "ERS", - "feed_name": "iddot", - "feed_state_code": "ID", - "latitude": 43.0208066348276, - "longitude": -112.408309608311, - "_enriched": {"geocoder": {"city": None, "name": "Ross Fork Creek", - "county": "Bannock", "state": "Idaho"}}, - }, - }, -} - -_WZDX_WA = { - "data": { - "adapter": "wzdx", - "category": "work_zone.wzdx", - "time": "2026-06-01T00:00:00+00:00", - "severity": 1, - "geo": {"centroid": [-117.33633, 46.433365], "primary_region": "US-WA"}, - "data": { - "road_names": ["012"], - "direction": "westbound", - "description": "Contract - XE3608 SR 12", - "vehicle_impact": "unknown", - "event_status": "pending", - "start_date": "2026-06-01T00:00:00+00:00", - "end_date": "2026-06-05T00:00:00+00:00", - "data_source_id": "WSDOT-WZDB", - "feed_name": "wsdot", - "feed_state_code": "WA", - "latitude": 46.433365, - "longitude": -117.33633, - "_enriched": {"geocoder": {"city": None, "name": "US Highway 12", - "county": "Garfield", "state": "Washington"}}, - }, - }, -} - -_WZDX_MCCALL = { - "data": { - "adapter": "wzdx", "category": "work_zone.wzdx", - "time": "2026-05-28T23:00:00Z", "severity": 1, - "geo": {"centroid": [-116.09759, 44.9065083834611], "primary_region": "US-ID"}, - "data": { - "road_names": ["SH-55"], - "direction": "unknown", - "description": " Emergency repairs on SH-55 Both Directions near Washington St." - " 5/28/2026 5:00 PM to 5/29/2026 8:00 AM Thu, Fri: ...", - "vehicle_impact": "all-lanes-open", - "start_date": "2026-05-28T23:00:00Z", - "end_date": "2026-05-29T14:00:00Z", - "feed_state_code": "ID", - "latitude": 44.9065083834611, - "longitude": -116.09759, - "_enriched": {"geocoder": {"city": "McCall", "county": "Valley", "state": "ID"}}, - }, - }, -} - - -def _normalize_wzdx(env): - n = normalize(env) - assert n is not None - assert n["source"] == "wzdx" - return n - - -# --- (a) Idaho wzdx full-field parse --------------------------------------- - -def test_wzdx_idaho_full_fields_normalized(monkeypatch): - _clear_h3_cache() - from meshai import central_normalizer as cn - # Mock Photon for the SECONDARY town path (city is null in this envelope). - monkeypatch.setattr(cn, "_photon_reverse_places", - lambda lat, lon: [ - {"geometry": {"coordinates": [-112.4373, 43.0299]}, - "properties": {"name": "Fort Hall", - "osm_key": "place", "osm_value": "village"}}, - ]) - n = _normalize_wzdx(_WZDX_ID_FULL) - assert n["road"] is None # Exit-ramp pattern → uninformative-road drop - assert n["direction"] == "southbound" - # sub_type combines impact-phrase (suppressed under full-closure) + work_type - # (None here — types_of_work absent). With full-closure, sub_type stays None - # and the renderer prepends "all lanes closed". - assert n["sub_type"] is None - assert n["impact"] == "full_closure" - assert n["mile_start"] == 80 and n["mile_end"] is None - assert n["town"] == "Fort Hall" # via Photon nearest_town - assert isinstance(n["ends_at"], datetime) - assert n["ends_at"].year == 2026 and n["ends_at"].month == 6 and n["ends_at"].day == 11 - - -def test_wzdx_wa_road_passes_through_verbatim(monkeypatch): - _clear_h3_cache() - from meshai import central_normalizer as cn - monkeypatch.setattr(cn, "_photon_reverse_places", lambda lat, lon: []) - n = _normalize_wzdx(_WZDX_WA) - # Per spec: "honor upstream verbatim, no expansion" -- raw '012' passes through. - assert n["road"] == "012" - assert n["direction"] == "westbound" - # vehicle_impact='unknown' → impact_phrase=None; sub_type stays None. - assert n["sub_type"] is None - assert n["impact"] == "partial" - # No MM in WA descriptions; mile_start stays None. - assert n["mile_start"] is None - assert isinstance(n["ends_at"], datetime) - assert n["town"] is None # city null + Photon returned no places - - -# --- (c) vehicle_impact mapping for each main value ------------------------ - -@pytest.mark.parametrize("vi_raw,expected_sub_type,expected_impact", [ - ("all-lanes-closed", None, "full_closure"), - ("some-lanes-closed", "lanes reduced", "partial"), - ("alternating-one-way", "one-way alternating", "partial"), - ("unknown", None, "partial"), - ("all-lanes-open", None, "partial"), - ("totally-made-up", None, "partial"), -]) -def test_wzdx_vehicle_impact_mapping(vi_raw, expected_sub_type, expected_impact, monkeypatch): - _clear_h3_cache() - from meshai import central_normalizer as cn - monkeypatch.setattr(cn, "_photon_reverse_places", lambda lat, lon: []) - env = {"data": {"adapter": "wzdx", "category": "work_zone.wzdx", "time": "2026-06-01T00:00:00Z", - "geo": {"centroid": [-116.0, 44.0]}, - "data": {"road_names": ["SH-1"], "direction": "northbound", - "description": "X", "vehicle_impact": vi_raw, - "end_date": "2026-06-05T17:00:00Z", - "latitude": 44.0, "longitude": -116.0, - "_enriched": {"geocoder": {"city": "Boise"}}}}} - n = normalize(env) - assert n["sub_type"] == expected_sub_type - assert n["impact"] == expected_impact - - -# --- (d) structured end_date parses to friendly format -------------------- - -def test_wzdx_end_date_iso_parsed_to_datetime(monkeypatch): - _clear_h3_cache() - from meshai import central_normalizer as cn - monkeypatch.setattr(cn, "_photon_reverse_places", lambda lat, lon: []) - env = {"data": {"adapter": "wzdx", "category": "work_zone.wzdx", - "geo": {"centroid": [-116.0, 44.0]}, - "data": {"road_names": ["SH-1"], "direction": "northbound", - "description": "x", "vehicle_impact": "unknown", - "end_date": "2026-06-15T18:30:00+00:00", - "latitude": 44.0, "longitude": -116.0, - "_enriched": {"geocoder": {"city": "Boise"}}}}} - n = normalize(env) - assert isinstance(n["ends_at"], datetime) - assert n["ends_at"].month == 6 and n["ends_at"].day == 15 - assert n["ends_at"].hour in (18, 11, 12) # depending on local-tz coercion - - -# --- (e) MM regex extraction on ID description ---------------------------- - -def test_wzdx_mile_post_regex_from_description(monkeypatch): - _clear_h3_cache() - from meshai import central_normalizer as cn - monkeypatch.setattr(cn, "_photon_reverse_places", lambda lat, lon: []) - env = {"data": {"adapter": "wzdx", "category": "work_zone.wzdx", - "geo": {"centroid": [-116.0, 44.0]}, - "data": {"road_names": ["I-15"], "direction": "southbound", - "description": "Bridge work on I-15 SB from MM (89) to MM (93). 6/1/2026 7:00 AM to 6/3/2026 5:00 PM", - "vehicle_impact": "some-lanes-closed", - "end_date": "2026-06-03T22:00:00Z", - "latitude": 44.0, "longitude": -116.0, - "_enriched": {"geocoder": {"city": "Blackfoot"}}}}} - n = normalize(env) - assert n["mile_start"] == 89 - assert n["mile_end"] == 93 - - -# --- (f) WA event without MM yields mile_start=None ----------------------- - -def test_wzdx_wa_no_mm_in_description(monkeypatch): - _clear_h3_cache() - from meshai import central_normalizer as cn - monkeypatch.setattr(cn, "_photon_reverse_places", lambda lat, lon: []) - n = _normalize_wzdx(_WZDX_WA) - assert n["mile_start"] is None - assert n["mile_end"] is None - - -# --- (g) town fallback chain ---------------------------------------------- - -def test_wzdx_town_uses_geocoder_city_when_present(monkeypatch): - _clear_h3_cache() - from meshai import central_normalizer as cn - calls = [] - monkeypatch.setattr(cn, "_photon_reverse_places", - lambda lat, lon: calls.append("called") or []) - n = _normalize_wzdx(_WZDX_MCCALL) - assert n["town"] == "McCall" - assert calls == [] # city present → Photon NOT called - - -def test_wzdx_town_falls_back_to_nearest_town_when_city_null(monkeypatch): - _clear_h3_cache() - from meshai import central_normalizer as cn - monkeypatch.setattr(cn, "_photon_reverse_places", - lambda lat, lon: [ - {"geometry": {"coordinates": [-117.293, 46.475]}, - "properties": {"name": "Pomeroy", - "osm_key": "place", "osm_value": "city"}}, - ]) - n = _normalize_wzdx(_WZDX_WA) - assert n["town"] == "Pomeroy" - - -# --- adapter dispatch routes wzdx → _parse_wzdx_federal ------------------- - -def test_wzdx_adapter_routes_to_wzdx_parser(monkeypatch): - _clear_h3_cache() - from meshai import central_normalizer as cn - monkeypatch.setattr(cn, "_photon_reverse_places", lambda lat, lon: []) - n = normalize(_WZDX_WA) - assert n is not None - assert n["source"] == "wzdx" - - -# --- work_type from types_of_work or event_type --------------------------- - -def test_wzdx_sub_type_from_types_of_work(monkeypatch): - _clear_h3_cache() - from meshai import central_normalizer as cn - monkeypatch.setattr(cn, "_photon_reverse_places", lambda lat, lon: []) - env = {"data": {"adapter": "wzdx", "category": "work_zone.wzdx", - "geo": {"centroid": [-116.0, 44.0]}, - "data": {"road_names": ["SH-1"], "direction": "both", - "description": "x", - "types_of_work": [{"type_name": "paving"}], - "vehicle_impact": "some-lanes-closed", - "end_date": "2026-06-05T17:00:00Z", - "latitude": 44.0, "longitude": -116.0, - "_enriched": {"geocoder": {"city": "Boise"}}}}} - n = normalize(env) - # Folded form: impact_phrase + work_type (paving) - assert n["sub_type"] == "lanes reduced, paving" - - -def test_wzdx_sub_type_unknown_vocab_is_lowercased_with_spaces(monkeypatch): - _clear_h3_cache() - from meshai import central_normalizer as cn - monkeypatch.setattr(cn, "_photon_reverse_places", lambda lat, lon: []) - env = {"data": {"adapter": "wzdx", "category": "work_zone.wzdx", - "geo": {"centroid": [-116.0, 44.0]}, - "data": {"road_names": ["SH-1"], "direction": "northbound", - "description": "x", - "types_of_work": [{"type_name": "Some-Custom-Work"}], - "vehicle_impact": "all-lanes-open", - "end_date": "2026-06-05T17:00:00Z", - "latitude": 44.0, "longitude": -116.0, - "_enriched": {"geocoder": {"city": "Boise"}}}}} - n = normalize(env) - assert n["sub_type"] == "some custom work" # lowercased + hyphens→spaces - - - -# ============================================================================ -# v0.5.9 GAMMA -- state_511_atis Idaho cutover -# ============================================================================ - - -def _state_511_envelope(state_code="ID", primary_region="US-ID"): - return { - "subject": "central.traffic.work_zone.id", - "id": "ID:Construction:33333", - "data": { - "id": "ID:Construction:33333", "adapter": "state_511_atis", - "category": "work_zone.state_511_atis", "severity": 1, - "geo": {"centroid": [-116.79, 47.70], - "primary_region": primary_region}, - "data": { - "roadway_name": "US-95", "direction": "Both", - "event_sub_type": "brushControl", - "description": "Minor Brush control on US-95.", - "is_full_closure": False, "layer": "Construction", - "county": "Kootenai", "state": "Idaho", - "state_code": state_code, - "start_date": "6/1/26, 5:00 AM", - "last_updated": "5/28/26, 12:54 PM", - "latitude": 47.7, "longitude": -116.79, - "_enriched": {"geocoder": { - "city": "Coeur d'Alene", "county": "Kootenai", - }}, - }, - }, - } - - -def test_gamma_should_skip_state_511_atis_id_via_state_code(): - """Helper returns True when state_code='ID'.""" - from meshai.central_normalizer import should_skip_state_511_atis_id - env = _state_511_envelope(state_code="ID") - assert should_skip_state_511_atis_id(env) is True - - -def test_gamma_should_skip_state_511_atis_id_via_primary_region(): - """Helper returns True when only primary_region='US-ID' is set.""" - from meshai.central_normalizer import should_skip_state_511_atis_id - env = _state_511_envelope(state_code="", primary_region="US-ID") - assert should_skip_state_511_atis_id(env) is True - - -def test_gamma_should_skip_state_511_atis_id_false_for_non_id(): - """Helper returns False for neighbor states.""" - from meshai.central_normalizer import should_skip_state_511_atis_id - env = _state_511_envelope(state_code="WA", primary_region="US-WA") - assert should_skip_state_511_atis_id(env) is False - - -def test_gamma_state_511_non_id_still_parses(): - """state_511_atis with state_code='WA' continues to be parsed -- - neighbor-state coverage remains active after the Idaho cutover. - With the v0.5.9 GAMMA fixup, the parser is unconditionally pure; - this test guards against a future regression that would put the - skip back into normalize().""" - env = _state_511_envelope(state_code="WA", primary_region="US-WA") - n = normalize(env) - assert n is not None - assert n.get("source") == "state_511_atis" - assert n.get("road") == "US-95" - - -def test_gamma_itd_511_work_zone_dispatch(): - """itd_511 + category=work_zone.* routes to _parse_itd_511_work_zone.""" - env = { - "subject": "central.traffic.work_zone.us.id", - "id": "ITD:469:99", - "data": { - "id": "ITD:469:99", "adapter": "itd_511", - "category": "work_zone.itd_511", "severity": 1, - "geo": {"centroid": [-116.79, 47.70], - "primary_region": "US-ID"}, - "data": { - "event_type_short": "work_zone", - "event_sub_type": "roadConstruction", - "roadway_name": "I-90", "direction": "Both", - "description": "Road construction on I-90.", - "is_full_closure": False, - "comment": "", "cause": "roadwork", - "organization": "ERS", - "start_epoch": 1780600000, - "planned_end_epoch": 1781000000, - "latitude": 47.7, "longitude": -116.79, - "_enriched": {"geocoder": { - "city": "Coeur d'Alene", "county": "Kootenai", - }}, - }, - }, - } - n = normalize(env) - assert n is not None - assert n.get("source") == "itd_511" - assert n.get("road") == "I-90" - - -def test_gamma_itd_511_non_work_zone_returns_none_or_marker(): - """itd_511 + category=incident.* should NOT go through work_zone parser.""" - env = { - "subject": "central.traffic.incident.us.id", - "id": "ITD:469:100", - "data": { - "id": "ITD:469:100", "adapter": "itd_511", - "category": "incident.itd_511", "severity": 1, - "geo": {"primary_region": "US-ID"}, - "data": { - "event_type_short": "incident", - "event_sub_type": "crash", - "roadway_name": "I-84", "direction": "East", - "description": "Crash on I-84.", - "is_full_closure": False, "comment": "", - "cause": "crash", "organization": "ERS", - "start_epoch": 1780600000, - "planned_end_epoch": None, - "latitude": 43.5, "longitude": -116.5, - "_enriched": {"geocoder": {"city": "Boise"}}, - }, - }, - } - n = normalize(env) - # Either None or not a work_zone shape -- never a parsed work_zone. - if n is not None: - assert n.get("source") != "itd_511" or "road" not in n diff --git a/work/tests/test_fire_refactor.py b/work/tests/test_fire_refactor.py index f90de02..b7359c5 100644 --- a/work/tests/test_fire_refactor.py +++ b/work/tests/test_fire_refactor.py @@ -21,7 +21,6 @@ from __future__ import annotations import pytest -from meshai import central_normalizer as cn from meshai.notifications.formatters._budget import budget_for from meshai.env.fire_render import ( _build_canonical, @@ -37,6 +36,7 @@ from tests.test_wfigs_handler import ( _IRWIN_A, _make_active_envelope, _make_tombstone, + _normalize_wfigs, ) _AT = 1_800_000_000.0 # pinned epoch (unused by fire render; kept for parity) @@ -148,14 +148,14 @@ class TestFormatterGolden: # Drive the real handler to produce the legacy all-clear wire, then # assert the formatter reproduces it byte-for-byte from event.data. env = _make_active_envelope(geocoder_city="Burley") - n0 = cn.normalize(env) + n0 = _normalize_wfigs(env) data0 = {} handle_wfigs(n0, env, env["subject"], data=data0, now=1_000_000) data0["_on_broadcast_committed"](float(1_000_000)) # arm last_broadcast_* tomb = _make_tombstone() data_t = {} - old_wire = handle_wfigs(cn.normalize(tomb), tomb, tomb["subject"], + old_wire = handle_wfigs(_normalize_wfigs(tomb), tomb, tomb["subject"], data=data_t, now=2_000_000) assert old_wire is not None assert old_wire.startswith("✅ Cache Peak Fire — contained & closed") @@ -186,7 +186,7 @@ class TestGateSequenceParity: """A full fire lifecycle agrees between decide() and handle_wfigs().""" def _decide(self, env, now): - n = cn.normalize(env) + n = _normalize_wfigs(env) canonical = _build_canonical(n, n["_kind"]) return fire_decide(canonical, source="wfigs", now=float(now)) @@ -198,7 +198,7 @@ class TestGateSequenceParity: legacy (not-cutover) stamps must equal decide()'s data_patch, and we arm last_broadcast_* via the commit callback (simulating the dispatcher). """ - n = cn.normalize(env) + n = _normalize_wfigs(env) gate = self._decide(env, now) assert gate.broadcast is expect_broadcast, ( f"decide broadcast {gate.broadcast} != {expect_broadcast} " @@ -268,7 +268,7 @@ class TestGateSequenceParity: irwin_id=_IRWIN_A, geocoder_city="Burley", daily_acres=250.0, pct_contained=0, fire_discovery_dt_ms=disc_ms) gate = self._decide(env_new, base) - n = cn.normalize(env_new) + n = _normalize_wfigs(env_new) assert gate.data_patch["_dedup_suffix"] == f"{n['acres']}|{n['contained_pct']}" assert gate.data_patch["_cooldown_suffix"] == _IRWIN_A @@ -279,7 +279,7 @@ class TestGateSequenceParity: assert gate.broadcast is False assert gate.lifecycle == "suppress" # Handler agrees: returns None. - out = handle_wfigs(cn.normalize(tomb), tomb, tomb["subject"], + out = handle_wfigs(_normalize_wfigs(tomb), tomb, tomb["subject"], data={}, now=1_000_000) assert out is None diff --git a/work/tests/test_geo_wzdx_parse.py b/work/tests/test_geo_wzdx_parse.py new file mode 100644 index 0000000..89d1d3c --- /dev/null +++ b/work/tests/test_geo_wzdx_parse.py @@ -0,0 +1,407 @@ +"""Tests for the live geo/WZDx-parsing code that used to live in the +Central-envelope adapter-normalizer module (deleted, name and all, in this +PR). + +That module (its `normalize()` Central-envelope dispatcher, +`_parse_state_511_atis`, `_parse_itd_511_work_zone`, +`should_skip_state_511_atis_id`, `normalize_road_name`, and their fixtures/ +tests) was deleted in chore/ripout-2e-geo-normalizer: it had zero live +production callers (Central's NATS consumer that dispatched envelopes to it +was deleted in an earlier ripout pass), and unlike env.fire_render +.handle_wfigs it wasn't kept as a parity-tested legacy contract for anything +else, so it -- and the tests that existed solely to exercise IT -- went with +it. See that PR's report for the full per-symbol accounting. + +What remains here: the WZDx-federal-parser tests (now exercising +`meshai.env.wzdx_parse._parse_wzdx_federal` directly instead of routing +through the deleted `normalize()` dispatcher -- same assertions, same +fixtures, just called one layer closer to the code under test) and the +nearest_town()/Photon/H3-cache tests (now exercising `meshai.geo`, where +that code lives now). +""" + +from datetime import datetime + +import pytest + +from meshai.env.wzdx_parse import _parse_wzdx_federal +from meshai.geo import nearest_town + + +def _parse_wzdx_envelope(env: dict) -> dict: + """Unwrap a Central-envelope-shaped wzdx fixture and call the live + parser directly (`normalize()`'s old wzdx dispatch was just this).""" + inner = env["data"] + return _parse_wzdx_federal(inner["data"], inner.get("geo") or {}) + + +# ---------- v0.5.8 nearest_town: Photon + H3 cache ------------------------ + +# Photon /reverse?osm_tag=place returns features like: +_PHOTON_STANLEY = { + "features": [ + {"geometry": {"coordinates": [-114.9378523, 44.2161414]}, + "properties": {"name": "Stanley", "osm_key": "place", "osm_value": "city"}}, + ], +} +_PHOTON_MULTI = { + "features": [ + # Closer but a "natural" feature -- must NOT be picked (not a place). + {"geometry": {"coordinates": [-114.93, 44.2155]}, + "properties": {"name": "Mountain Village Restaurant", "osm_key": "amenity", "osm_value": "restaurant"}}, + # Town (~1km away). + {"geometry": {"coordinates": [-114.9378523, 44.2161414]}, + "properties": {"name": "Stanley", "osm_key": "place", "osm_value": "city"}}, + # Town further out. + {"geometry": {"coordinates": [-115.0588585, 44.2436215]}, + "properties": {"name": "Lake Town", "osm_key": "place", "osm_value": "village"}}, + ], +} + + +def _clear_h3_cache(): + from meshai.geo import _h3_cache + _h3_cache.clear() + + +def test_nearest_town_returns_dict_for_known_coord(monkeypatch): + _clear_h3_cache() + from meshai import geo + monkeypatch.setattr(geo, "_photon_reverse_places", + lambda lat, lon: _PHOTON_STANLEY["features"]) + n = nearest_town(44.2160, -114.9311) + assert n is not None + assert n["name"] == "Stanley" + assert n["distance_mi"] >= 0 and n["distance_mi"] <= 1 + assert n["bearing"] in {"N", "NE", "E", "SE", "S", "SW", "W", "NW"} + + +def test_nearest_town_filters_non_place_osm_values(monkeypatch): + _clear_h3_cache() + from meshai import geo + # Only the restaurant; no place tag at all. + monkeypatch.setattr(geo, "_photon_reverse_places", + lambda lat, lon: [ + {"geometry": {"coordinates": [-114.93, 44.2155]}, + "properties": {"name": "Restaurant", + "osm_key": "amenity", "osm_value": "restaurant"}}, + ]) + assert nearest_town(44.2160, -114.9311) is None + + +def test_nearest_town_picks_closest_place(monkeypatch): + _clear_h3_cache() + from meshai import geo + monkeypatch.setattr(geo, "_photon_reverse_places", + lambda lat, lon: _PHOTON_MULTI["features"]) + n = nearest_town(44.2160, -114.9311) + assert n is not None + assert n["name"] == "Stanley" # closer than Lake Town + + +def test_nearest_town_returns_none_beyond_max_distance(monkeypatch): + _clear_h3_cache() + from meshai import geo + monkeypatch.setattr(geo, "_photon_reverse_places", + lambda lat, lon: _PHOTON_STANLEY["features"]) + # Event 200 mi from Stanley; max_distance_mi=50 by default. + far_lat = 44.2160 + 200 / 69.0 + n = nearest_town(far_lat, -114.9311) + assert n is None + + +def test_nearest_town_returns_none_on_photon_failure(monkeypatch): + _clear_h3_cache() + from meshai import geo + monkeypatch.setattr(geo, "_photon_reverse_places", lambda lat, lon: []) + assert nearest_town(44.2160, -114.9311) is None + + +def test_nearest_town_caches_via_h3(monkeypatch): + _clear_h3_cache() + from meshai import geo + calls = [] + def stub(lat, lon): + calls.append((lat, lon)) + return _PHOTON_STANLEY["features"] + monkeypatch.setattr(geo, "_photon_reverse_places", stub) + # Two calls at the same coord → only one Photon hit. + nearest_town(44.2160, -114.9311) + nearest_town(44.2160, -114.9311) + assert len(calls) == 1 + + +def test_nearest_town_handles_none_inputs(): + _clear_h3_cache() + assert nearest_town(None, -114.9311) is None + assert nearest_town(44.2160, None) is None + + +# ============================================================================ +# v0.5.8-wzdx federal parser tests +# ============================================================================ + +# --- representative envelopes (flat shape, as Central actually published) --- + +_WZDX_ID_FULL = { + "data": { + "adapter": "wzdx", + "category": "work_zone.wzdx", + "time": "2026-06-01T13:00:00Z", + "severity": 3, + "geo": {"centroid": [-112.408309608311, 43.0208066348276], + "primary_region": "US-ID", "regions": ["US-ID"]}, + "data": { + "road_names": ["Exit 80 On Ramp"], + "direction": "southbound", + "description": " Road construction on Exit 80 On Ramp Southbound near MM (80)." + " All lanes closed. 6/1/2026 7:00 AM to 6/10/2026 6:00 PM Mon, Tue ...", + "vehicle_impact": "all-lanes-closed", + "event_status": None, + "start_date": "2026-06-01T13:00:00Z", + "end_date": "2026-06-11T00:00:00Z", + "data_source_id": "ERS", + "feed_name": "iddot", + "feed_state_code": "ID", + "latitude": 43.0208066348276, + "longitude": -112.408309608311, + "_enriched": {"geocoder": {"city": None, "name": "Ross Fork Creek", + "county": "Bannock", "state": "Idaho"}}, + }, + }, +} + +_WZDX_WA = { + "data": { + "adapter": "wzdx", + "category": "work_zone.wzdx", + "time": "2026-06-01T00:00:00+00:00", + "severity": 1, + "geo": {"centroid": [-117.33633, 46.433365], "primary_region": "US-WA"}, + "data": { + "road_names": ["012"], + "direction": "westbound", + "description": "Contract - XE3608 SR 12", + "vehicle_impact": "unknown", + "event_status": "pending", + "start_date": "2026-06-01T00:00:00+00:00", + "end_date": "2026-06-05T00:00:00+00:00", + "data_source_id": "WSDOT-WZDB", + "feed_name": "wsdot", + "feed_state_code": "WA", + "latitude": 46.433365, + "longitude": -117.33633, + "_enriched": {"geocoder": {"city": None, "name": "US Highway 12", + "county": "Garfield", "state": "Washington"}}, + }, + }, +} + +_WZDX_MCCALL = { + "data": { + "adapter": "wzdx", "category": "work_zone.wzdx", + "time": "2026-05-28T23:00:00Z", "severity": 1, + "geo": {"centroid": [-116.09759, 44.9065083834611], "primary_region": "US-ID"}, + "data": { + "road_names": ["SH-55"], + "direction": "unknown", + "description": " Emergency repairs on SH-55 Both Directions near Washington St." + " 5/28/2026 5:00 PM to 5/29/2026 8:00 AM Thu, Fri: ...", + "vehicle_impact": "all-lanes-open", + "start_date": "2026-05-28T23:00:00Z", + "end_date": "2026-05-29T14:00:00Z", + "feed_state_code": "ID", + "latitude": 44.9065083834611, + "longitude": -116.09759, + "_enriched": {"geocoder": {"city": "McCall", "county": "Valley", "state": "ID"}}, + }, + }, +} + + +def _normalize_wzdx(env): + n = _parse_wzdx_envelope(env) + assert n is not None + assert n["source"] == "wzdx" + return n + + +# --- (a) Idaho wzdx full-field parse --------------------------------------- + +def test_wzdx_idaho_full_fields_normalized(monkeypatch): + _clear_h3_cache() + from meshai import geo + # Mock Photon for the SECONDARY town path (city is null in this envelope). + monkeypatch.setattr(geo, "_photon_reverse_places", + lambda lat, lon: [ + {"geometry": {"coordinates": [-112.4373, 43.0299]}, + "properties": {"name": "Fort Hall", + "osm_key": "place", "osm_value": "village"}}, + ]) + n = _normalize_wzdx(_WZDX_ID_FULL) + assert n["road"] is None # Exit-ramp pattern → uninformative-road drop + assert n["direction"] == "southbound" + # sub_type combines impact-phrase (suppressed under full-closure) + work_type + # (None here — types_of_work absent). With full-closure, sub_type stays None + # and the renderer prepends "all lanes closed". + assert n["sub_type"] is None + assert n["impact"] == "full_closure" + assert n["mile_start"] == 80 and n["mile_end"] is None + assert n["town"] == "Fort Hall" # via Photon nearest_town + assert isinstance(n["ends_at"], datetime) + assert n["ends_at"].year == 2026 and n["ends_at"].month == 6 and n["ends_at"].day == 11 + + +def test_wzdx_wa_road_passes_through_verbatim(monkeypatch): + _clear_h3_cache() + from meshai import geo + monkeypatch.setattr(geo, "_photon_reverse_places", lambda lat, lon: []) + n = _normalize_wzdx(_WZDX_WA) + # Per spec: "honor upstream verbatim, no expansion" -- raw '012' passes through. + assert n["road"] == "012" + assert n["direction"] == "westbound" + # vehicle_impact='unknown' → impact_phrase=None; sub_type stays None. + assert n["sub_type"] is None + assert n["impact"] == "partial" + # No MM in WA descriptions; mile_start stays None. + assert n["mile_start"] is None + assert isinstance(n["ends_at"], datetime) + assert n["town"] is None # city null + Photon returned no places + + +# --- (c) vehicle_impact mapping for each main value ------------------------ + +@pytest.mark.parametrize("vi_raw,expected_sub_type,expected_impact", [ + ("all-lanes-closed", None, "full_closure"), + ("some-lanes-closed", "lanes reduced", "partial"), + ("alternating-one-way", "one-way alternating", "partial"), + ("unknown", None, "partial"), + ("all-lanes-open", None, "partial"), + ("totally-made-up", None, "partial"), +]) +def test_wzdx_vehicle_impact_mapping(vi_raw, expected_sub_type, expected_impact, monkeypatch): + _clear_h3_cache() + from meshai import geo + monkeypatch.setattr(geo, "_photon_reverse_places", lambda lat, lon: []) + env = {"data": {"adapter": "wzdx", "category": "work_zone.wzdx", "time": "2026-06-01T00:00:00Z", + "geo": {"centroid": [-116.0, 44.0]}, + "data": {"road_names": ["SH-1"], "direction": "northbound", + "description": "X", "vehicle_impact": vi_raw, + "end_date": "2026-06-05T17:00:00Z", + "latitude": 44.0, "longitude": -116.0, + "_enriched": {"geocoder": {"city": "Boise"}}}}} + n = _parse_wzdx_envelope(env) + assert n["sub_type"] == expected_sub_type + assert n["impact"] == expected_impact + + +# --- (d) structured end_date parses to friendly format -------------------- + +def test_wzdx_end_date_iso_parsed_to_datetime(monkeypatch): + _clear_h3_cache() + from meshai import geo + monkeypatch.setattr(geo, "_photon_reverse_places", lambda lat, lon: []) + env = {"data": {"adapter": "wzdx", "category": "work_zone.wzdx", + "geo": {"centroid": [-116.0, 44.0]}, + "data": {"road_names": ["SH-1"], "direction": "northbound", + "description": "x", "vehicle_impact": "unknown", + "end_date": "2026-06-15T18:30:00+00:00", + "latitude": 44.0, "longitude": -116.0, + "_enriched": {"geocoder": {"city": "Boise"}}}}} + n = _parse_wzdx_envelope(env) + assert isinstance(n["ends_at"], datetime) + assert n["ends_at"].month == 6 and n["ends_at"].day == 15 + assert n["ends_at"].hour in (18, 11, 12) # depending on local-tz coercion + + +# --- (e) MM regex extraction on ID description ---------------------------- + +def test_wzdx_mile_post_regex_from_description(monkeypatch): + _clear_h3_cache() + from meshai import geo + monkeypatch.setattr(geo, "_photon_reverse_places", lambda lat, lon: []) + env = {"data": {"adapter": "wzdx", "category": "work_zone.wzdx", + "geo": {"centroid": [-116.0, 44.0]}, + "data": {"road_names": ["I-15"], "direction": "southbound", + "description": "Bridge work on I-15 SB from MM (89) to MM (93). 6/1/2026 7:00 AM to 6/3/2026 5:00 PM", + "vehicle_impact": "some-lanes-closed", + "end_date": "2026-06-03T22:00:00Z", + "latitude": 44.0, "longitude": -116.0, + "_enriched": {"geocoder": {"city": "Blackfoot"}}}}} + n = _parse_wzdx_envelope(env) + assert n["mile_start"] == 89 + assert n["mile_end"] == 93 + + +# --- (f) WA event without MM yields mile_start=None ----------------------- + +def test_wzdx_wa_no_mm_in_description(monkeypatch): + _clear_h3_cache() + from meshai import geo + monkeypatch.setattr(geo, "_photon_reverse_places", lambda lat, lon: []) + n = _normalize_wzdx(_WZDX_WA) + assert n["mile_start"] is None + assert n["mile_end"] is None + + +# --- (g) town fallback chain ---------------------------------------------- + +def test_wzdx_town_uses_geocoder_city_when_present(monkeypatch): + _clear_h3_cache() + from meshai import geo + calls = [] + monkeypatch.setattr(geo, "_photon_reverse_places", + lambda lat, lon: calls.append("called") or []) + n = _normalize_wzdx(_WZDX_MCCALL) + assert n["town"] == "McCall" + assert calls == [] # city present → Photon NOT called + + +def test_wzdx_town_falls_back_to_nearest_town_when_city_null(monkeypatch): + _clear_h3_cache() + from meshai import geo + monkeypatch.setattr(geo, "_photon_reverse_places", + lambda lat, lon: [ + {"geometry": {"coordinates": [-117.293, 46.475]}, + "properties": {"name": "Pomeroy", + "osm_key": "place", "osm_value": "city"}}, + ]) + n = _normalize_wzdx(_WZDX_WA) + assert n["town"] == "Pomeroy" + + +# --- work_type from types_of_work or event_type --------------------------- + +def test_wzdx_sub_type_from_types_of_work(monkeypatch): + _clear_h3_cache() + from meshai import geo + monkeypatch.setattr(geo, "_photon_reverse_places", lambda lat, lon: []) + env = {"data": {"adapter": "wzdx", "category": "work_zone.wzdx", + "geo": {"centroid": [-116.0, 44.0]}, + "data": {"road_names": ["SH-1"], "direction": "both", + "description": "x", + "types_of_work": [{"type_name": "paving"}], + "vehicle_impact": "some-lanes-closed", + "end_date": "2026-06-05T17:00:00Z", + "latitude": 44.0, "longitude": -116.0, + "_enriched": {"geocoder": {"city": "Boise"}}}}} + n = _parse_wzdx_envelope(env) + # Folded form: impact_phrase + work_type (paving) + assert n["sub_type"] == "lanes reduced, paving" + + +def test_wzdx_sub_type_unknown_vocab_is_lowercased_with_spaces(monkeypatch): + _clear_h3_cache() + from meshai import geo + monkeypatch.setattr(geo, "_photon_reverse_places", lambda lat, lon: []) + env = {"data": {"adapter": "wzdx", "category": "work_zone.wzdx", + "geo": {"centroid": [-116.0, 44.0]}, + "data": {"road_names": ["SH-1"], "direction": "northbound", + "description": "x", + "types_of_work": [{"type_name": "Some-Custom-Work"}], + "vehicle_impact": "all-lanes-open", + "end_date": "2026-06-05T17:00:00Z", + "latitude": 44.0, "longitude": -116.0, + "_enriched": {"geocoder": {"city": "Boise"}}}}} + n = _parse_wzdx_envelope(env) + assert n["sub_type"] == "some custom work" # lowercased + hyphens→spaces diff --git a/work/tests/test_incident_refactor.py b/work/tests/test_incident_refactor.py index dee0802..8f6d1ca 100644 --- a/work/tests/test_incident_refactor.py +++ b/work/tests/test_incident_refactor.py @@ -18,18 +18,25 @@ Original diffs are preserved in git history. This is a real production gap flagged for Matt: TomTom road-incident ingestion in particular has no native replacement. -Work-zone parity: `meshai.central_normalizer` (a top-level, non-Central-NATS -module; note the name is legacy) was never part of the deleted consumer path -and remains live. `meshai.notifications.renderers.work_zone` was ALSO never -part of the deleted consumer path, but was itself dead code (zero production -callers) once formatters.incident absorbed it as `_render_work_zone()`; it -was removed in a later ripout pass. See TestWorkZoneGolden below for how its -golden-parity coverage was preserved as pinned literals. +Work-zone parity: the (now fully deleted) Central-envelope adapter-normalizer +module was never part of the deleted Central consumer path, but +chore/ripout-2e-geo-normalizer found it had no live production +caller of its OWN either (env/roads511.py, the real live itd_511 adapter, +never used it) and deleted it in turn -- name and all. Its one live part, +the wzdx federal parser, moved to `meshai.env.wzdx_parse` next to its real +consumer `env.wzdx`. `meshai.notifications.renderers.work_zone` was ALSO +never part of the deleted Central consumer path, but was itself dead code +(zero production callers) once formatters.incident absorbed it as +`_render_work_zone()`; it was removed in a later ripout pass. See +TestWorkZoneGolden below for how both dead parsers' golden-parity coverage +was preserved as pinned literals (the itd_511 canonical dict) or a direct +call to the surviving live parser (wzdx). Groups ------ -1. Work-zone golden byte-parity (traffic_last/0002 itd_511, traffic_last/0003 - wzdx) — calls normalize() directly, same as the production consumer. +1. Work-zone golden byte-parity (traffic_last/0002 itd_511 -- pinned + canonical-dict literal, the parser is dead; traffic_last/0003 wzdx -- + calls the live _parse_wzdx_federal directly). 2. Gate sequence: decide() lifecycle transitions (new → cold-dup → suppress-on-update-False → magnitude-up → suppress-no-change). 3. _anchor.resolve_anchor: DB hit and Photon fallback path. @@ -161,6 +168,22 @@ class TestWorkZoneGolden: now is pinned to captured_epoch (1783206522) for both paths so the ends-at segment is deterministic. + + chore/ripout-2e-geo-normalizer update: the deleted normalizer's + normalize() (and the itd_511 work_zone parser it dispatched to, + _parse_itd_511_work_zone) is now gone -- it had no live production + caller (env.roads511, the actual live itd_511 adapter, never used it). + For 0002.json (itd_511) the CANONICAL DICT that used to be computed live + via normalize() is now, by the same "pin it before the source goes away" + methodology already used for the wire-string goldens above, captured as + a literal (`_ITD511_0002_CANONICAL` below -- captured by running + normalize() against this exact fixture immediately before that module's + deletion, confirmed byte-identical to the pinned wire golden at that + time). + For 0003.json (wzdx) the parser (_parse_wzdx_federal) IS still live -- + it moved to meshai.env.wzdx_parse, next to its real consumer env.wzdx -- + so that fixture keeps calling the real parser directly instead of a + pinned literal. """ _GOLDEN = { @@ -168,8 +191,22 @@ class TestWorkZoneGolden: "0003.json": "🚧 US-95, near Wilder: southbound, ends Jul 19", } + # Captured from the deleted normalizer's normalize() + _n_to_canonical_workzone() + # against traffic_last/0002.json immediately before that module's + # deletion. town="Chubbuck"/bearing="NW"/distance_mi=0 came from a live + # Photon nearest_town() call at capture time (this fixture's geocoder.city + # is null) -- the same live dependency the pinned wire golden above already + # implicitly baked in ("near Chubbuck"). + _ITD511_0002_CANONICAL = { + "road": "US-91", "direction": "southbound", + "mile_start": None, "mile_end": None, + "sub_type": "road construction", "impact": "partial", + "ends_at_epoch": 1786968000.0, + "town": "Chubbuck", "distance_mi": 0, "bearing": "NW", + "lat": None, "lon": None, + } + def _run_wz(self, fixture_name: str, adapter_expected: str): - from meshai.central_normalizer import normalize from meshai.notifications.formatters.incident import format as fmt # Find the fixture by name @@ -184,13 +221,18 @@ class TestWorkZoneGolden: envelope = fx["envelope"] now_epoch = float(fx.get("captured_epoch", time.time())) - - n = normalize(envelope) - assert n is not None, f"normalize() returned None for {fixture_name!r}" golden = self._GOLDEN[fixture_name] - # New formatter - canonical = _n_to_canonical_workzone(n) + if adapter == "wzdx": + from meshai.env.wzdx_parse import _parse_wzdx_federal + inner = envelope["data"] + n = _parse_wzdx_federal(inner["data"], inner.get("geo") or {}) + assert n is not None, f"_parse_wzdx_federal returned None for {fixture_name!r}" + canonical = _n_to_canonical_workzone(n) + else: + # itd_511: dead parser, pinned canonical dict (see class docstring). + canonical = dict(self._ITD511_0002_CANONICAL) + event = _make_event("work_zone", canonical) new_out = fmt(event, now=now_epoch, budget=140) @@ -366,7 +408,7 @@ class TestAnchorResolve: import time as _time from meshai.persistence import get_db from meshai.notifications.formatters._anchor import resolve_anchor - import meshai.central_normalizer as cn_mod + from meshai import geo # Clear all seeded town_anchors so the DB step finds nothing. conn = get_db() @@ -378,7 +420,7 @@ class TestAnchorResolve: called.append((lat, lon)) return {"name": "Photon City", "distance_mi": 5, "bearing": "NW"} - monkeypatch.setattr(cn_mod, "nearest_town", fake_nearest_town) + monkeypatch.setattr(geo, "nearest_town", fake_nearest_town) result = resolve_anchor(43.615, -116.205, max_mi=50.0) assert result is not None @@ -400,12 +442,17 @@ class TestSchemaConformance: contain all expected keys.""" def test_workzone_canonical_keys_from_normalize(self): - """All work-zone canonical keys present in extraction from normalize().""" - from meshai.central_normalizer import normalize + """All work-zone canonical keys present in extraction from the live + wzdx parser. (Was the deleted normalizer's normalize() against + 0002.json (itd_511) -- that parser is dead and the module is gone; + 0003.json (wzdx) exercises the same _n_to_canonical_workzone() key + shape via the still-live _parse_wzdx_federal.)""" + from meshai.env.wzdx_parse import _parse_wzdx_federal - with open(_FIXTURE_DIR / "traffic_last" / "0002.json", encoding="utf-8") as f: + with open(_FIXTURE_DIR / "traffic_last" / "0003.json", encoding="utf-8") as f: fx = json.load(f) - n = normalize(fx["envelope"]) + inner = fx["envelope"]["data"] + n = _parse_wzdx_federal(inner["data"], inner.get("geo") or {}) assert n is not None canonical = _n_to_canonical_workzone(n) assert _WZ_CANONICAL_KEYS == set(canonical.keys()) diff --git a/work/tests/test_itd_511_work_zone.py b/work/tests/test_itd_511_work_zone.py deleted file mode 100644 index 4c8cf49..0000000 --- a/work/tests/test_itd_511_work_zone.py +++ /dev/null @@ -1,127 +0,0 @@ -"""Tests for the v0.5.9 GAMMA itd_511 work_zone parser in central_normalizer. - -Covers the cutover path: itd_511 supplies all Idaho work_zone broadcasts now -(state_511_atis ID is skipped). The parser must produce the same flat dict -shape as _parse_state_511_atis so the work-zone renderer (now -formatters.incident._render_work_zone(), via _n_to_canonical_workzone()) -works unchanged. -""" - -from datetime import datetime, timezone - -import pytest - -from meshai import central_normalizer as cn - - -# ---------- envelope builder --------------------------------------------- - - -def _itd_work_zone_env(*, roadway="SH-55", direction="North", - event_sub_type="roadConstruction", - is_full_closure=False, - external_id="ITD:469:42", - lat=44.103, lon=-116.110, - geocoder_city="McCall", - start_epoch=1780600000, - planned_end_epoch=1781200000, - description="Road construction on SH-55 Northbound from MM (140) to MM (145). 6/4/2026 7:00 AM to 6/11/2026 5:00 PM."): - return { - "id": external_id, - "subject": "central.traffic.work_zone.us.id", - "data": { - "id": external_id, "adapter": "itd_511", - "category": "work_zone.itd_511", "severity": 1, - "geo": {"centroid": [lon, lat], "primary_region": "US-ID"}, - "data": { - "event_type_short": "work_zone", - "event_sub_type": event_sub_type, - "roadway_name": roadway, "direction": direction, - "description": description, - "lanes_affected": "All lanes affected", - "is_full_closure": is_full_closure, - "itd_severity": "None", - "comment": "", - "cause": "roadwork", - "organization": "ERS", - "recurrence_text": "", - "recurrence_schedules": [], - "restrictions": {}, - "encoded_polyline": "", - "id_internal": 42, "source_id": "999", - "reported_epoch": start_epoch, - "last_updated_epoch": start_epoch, - "start_epoch": start_epoch, - "planned_end_epoch": planned_end_epoch, - "latitude": lat, "longitude": lon, - "_enriched": {"geocoder": { - "name": None, "city": geocoder_city, - "county": "Valley", "state": "ID", - "country": "United States", - "landclass": None, "elevation_m": 1530.0, - }}, - }, - }, - } - - -@pytest.fixture -def no_photon(monkeypatch): - monkeypatch.setattr(cn, "_photon_reverse_places", lambda lat, lon: []) - if hasattr(cn, "_H3_NEAREST_CACHE"): - cn._H3_NEAREST_CACHE.clear() - - -# ---------- test (a): all fields populate ----------------------------------- - - -def test_itd_511_work_zone_parses_with_all_fields(no_photon): - env = _itd_work_zone_env() - n = cn.normalize(env) - assert n is not None - assert n["source"] == "itd_511" - assert n["road"] == "SH-55" - # _norm_direction returns 'northbound' (matches state_511 convention) - assert n["direction"] == "northbound" - assert n["mile_start"] == 140 - assert n["mile_end"] == 145 - assert n["sub_type"] is not None - # is_full_closure=False -> impact 'partial' matches state_511 convention - assert n["impact"] == "partial" - assert n["town"] == "McCall" - # ends_at is a datetime - assert isinstance(n["ends_at"], datetime) - - -def test_itd_511_work_zone_full_closure_impact(no_photon): - env = _itd_work_zone_env(is_full_closure=True) - n = cn.normalize(env) - assert n["impact"] == "full_closure" - - -def test_itd_511_work_zone_end_date_formatting(no_photon): - """planned_end_epoch should serialize to a datetime that the renderer - can format consistently with state_511.""" - env = _itd_work_zone_env(planned_end_epoch=1781200000) - n = cn.normalize(env) - assert isinstance(n["ends_at"], datetime) - assert n["ends_at"].tzinfo is not None # UTC-aware - - -def test_itd_511_work_zone_no_end_date(no_photon): - """planned_end_epoch == None or 0 -> ends_at is None.""" - env = _itd_work_zone_env(planned_end_epoch=None) - n = cn.normalize(env) - assert n["ends_at"] is None - - -def test_itd_511_incident_does_not_go_through_work_zone_parser(no_photon): - """The work_zone parser should NOT be invoked for category=incident.itd_511 - -- those still route through the incident_handler. normalize() returns - None (defer) for non-work_zone itd_511 categories.""" - env = _itd_work_zone_env() - env["data"]["category"] = "incident.itd_511" - n = cn.normalize(env) - # Either None (defer) or not a work_zone dict (no road/direction/etc). - if n is not None: - assert "_kind" in n # marker, not a parsed work_zone dict diff --git a/work/tests/test_wfigs_handler.py b/work/tests/test_wfigs_handler.py index dbe46bb..4e3b4f6 100644 --- a/work/tests/test_wfigs_handler.py +++ b/work/tests/test_wfigs_handler.py @@ -1,4 +1,4 @@ -"""Tests for meshai/central/wfigs_handler.py -- WFIGS persistence wire-up. +"""Tests for env/fire_render.py's handle_wfigs() -- WFIGS persistence wire-up. Covers: (a) parse clean active-incident envelope (all fields populated) @@ -12,14 +12,27 @@ Covers: (i) known IRWIN acres up but <8h elapsed -> drop, last_broadcast_* unchanged (j) known IRWIN acres up + >=8h elapsed -> "Update:" prefix + audit row (k) location anchor priority: geocoder.city > nearest_town > landclass > county + +handle_wfigs (env/fire_render.py) has no live production caller -- Central's +NATS consumer that drove it is gone -- but it remains the parity-tested +legacy contract for the WFIGS wire format (see that module's docstring), and +`_location_anchor`, which it shares with the LIVE `_render` (used on the +FIRMS fire-growth path), is real regression-tested surface here. It expects +its input pre-shaped into a flat "normalized" dict; that shaping used to be +done by the WFIGS dispatch branch of the deleted Central-envelope adapter- +normalizer module's `normalize()` + its `_parse_wfigs_incidents` helper. +`_normalize_wfigs()` below is a verbatim-logic local replica of that +dispatch, scoped to the wfigs_incidents/wfigs_perimeters envelope shapes the +fixtures in this file build -- it exists so this test file has no dependency +on that (now fully removed) module. """ import os import time +from typing import Any, Optional import pytest -from meshai import central_normalizer as cn from meshai.env.fire_render import ( WFIGS_BROADCAST_COOLDOWN_S, handle_wfigs, @@ -29,6 +42,123 @@ from meshai.persistence import close_thread_connection, init_db from meshai.persistence import db as persistence_db +# ---------- local replica of the deleted normalize()/_parse_wfigs_incidents +# ---------- WFIGS dispatch (see module docstring) -------------------------- + +_WFIGS_ACRES_KEYS = ("DailyAcres", "IncidentSize") +_WFIGS_ACRES_RAW_KEYS = ("IncidentSize", "DiscoveryAcres", "FinalAcres") +_WFIGS_CONTAINED_KEYS = ("PercentContained",) +_WFIGS_CONTAINED_RAW_KEYS = ("PercentContained",) + + +def _first_non_null(d: dict, keys) -> Any: + for k in keys: + v = d.get(k) + if v is not None and v != "": + return v + return None + + +def _parse_wfigs_acres(inner_data: dict) -> Optional[float]: + val = _first_non_null(inner_data, _WFIGS_ACRES_KEYS) + if val is None: + raw = inner_data.get("raw") or {} + if isinstance(raw, dict): + val = _first_non_null(raw, _WFIGS_ACRES_RAW_KEYS) + if val is None: + return None + try: return float(val) + except (TypeError, ValueError): return None + + +def _parse_wfigs_contained(inner_data: dict) -> Optional[int]: + val = _first_non_null(inner_data, _WFIGS_CONTAINED_KEYS) + if val is None: + raw = inner_data.get("raw") or {} + if isinstance(raw, dict): + val = _first_non_null(raw, _WFIGS_CONTAINED_RAW_KEYS) + if val is None: + return None + try: return int(round(float(val))) + except (TypeError, ValueError): return None + + +def _parse_wfigs_incidents(inner_data: dict, geo: dict) -> Optional[dict]: + geocoder = geo.get("geocoder") or {} + irwin_id = inner_data.get("IrwinID") or inner_data.get("irwin_id") + name = inner_data.get("IncidentName") + itype = inner_data.get("IncidentTypeCategory") + if itype is not None and itype not in ("WF", "wildfire"): + return None + lat = inner_data.get("latitude") + lon = inner_data.get("longitude") + county = inner_data.get("POOCounty") + state = inner_data.get("POOState") + landclass = geocoder.get("landclass") + + declared_at_epoch = None + fdt = inner_data.get("FireDiscoveryDateTime") + if isinstance(fdt, (int, float)): + declared_at_epoch = int(fdt / 1000) if fdt > 1e12 else int(fdt) + + acres = _parse_wfigs_acres(inner_data) + contained_pct = _parse_wfigs_contained(inner_data) + city = geocoder.get("city") + raw = inner_data.get("raw") or {} + + return { + "irwin_id": irwin_id, + "incident_name": name, + "incident_type": itype, + "acres": acres, + "contained_pct": contained_pct, + "lat": lat, + "lon": lon, + "county": county, + "state": state, + "landclass": landclass, + "geocoder_city": city, + "declared_at_epoch": declared_at_epoch, + "fire_cause": raw.get("FireCause"), + "agency": raw.get("POOJurisdictionalAgency"), + "personnel": raw.get("TotalIncidentPersonnel"), + "unique_fire_id": raw.get("UniqueFireIdentifier"), + } + + +def _normalize_wfigs(envelope: dict) -> Optional[dict]: + """wfigs_incidents/wfigs_perimeters envelope -> flat normalized dict, + same shape the deleted normalizer's normalize() used to produce.""" + inner = envelope.get("data") or {} + adapter = inner.get("adapter") or "" + inner_data = inner.get("data") or {} + geo = inner.get("geo") or {} + category_raw = inner.get("category") or "" + + if adapter == "wfigs_incidents": + if category_raw.startswith("fire.incident.removed"): + return { + "_kind": "wfigs_tombstone", + "irwin_id": inner_data.get("irwin_id") or inner_data.get("IrwinID"), + "state": inner_data.get("state") or inner_data.get("POOState"), + "county": inner_data.get("county") or inner_data.get("POOCounty"), + } + if category_raw.startswith("fire.incident"): + n = _parse_wfigs_incidents(inner_data, geo) + if n is None: + return None + n["_kind"] = "wfigs_incident" + return n + if adapter == "wfigs_perimeters": + return { + "_kind": "wfigs_perimeter", + "irwin_id": inner_data.get("irwin_id") or inner_data.get("IrwinID"), + "state": inner_data.get("state") or inner_data.get("POOState"), + "county": inner_data.get("county") or inner_data.get("POOCounty"), + } + return None + + # ---------- fixtures ------------------------------------------------------ @@ -60,10 +190,11 @@ def mem_db(monkeypatch, tmp_path): def no_photon(monkeypatch): """Force nearest_town to return None so anchor falls through deterministically. Tests that exercise nearest_town wire it in directly.""" - monkeypatch.setattr(cn, "_photon_reverse_places", lambda lat, lon: []) + from meshai import geo + monkeypatch.setattr(geo, "_photon_reverse_places", lambda lat, lon: []) # Also reset the H3 LRU so cache state doesn't leak across tests. - if hasattr(cn, "_H3_NEAREST_CACHE"): - cn._H3_NEAREST_CACHE.clear() + if hasattr(geo, "_H3_NEAREST_CACHE"): + geo._H3_NEAREST_CACHE.clear() # ---------- envelope builders -------------------------------------------- @@ -175,7 +306,7 @@ def _make_perimeter(irwin_id=_IRWIN_A, state="ID", county="Cassia", # ============================================================================ def test_a_parse_clean_active_envelope(mem_db, no_photon): env = _make_active_envelope() - n = cn.normalize(env) + n = _normalize_wfigs(env) assert n is not None assert n["_kind"] == "wfigs_incident" assert n["irwin_id"] == _IRWIN_A @@ -196,7 +327,7 @@ def test_b_acres_fallback_to_raw_discovery_acres(mem_db, no_photon): env = _make_active_envelope(daily_acres=None, pct_contained=None, raw_discovery_acres=0.1, raw_pct_contained=0) - n = cn.normalize(env) + n = _normalize_wfigs(env) assert n["acres"] == 0.1 assert n["contained_pct"] == 0 @@ -209,7 +340,7 @@ def test_c_acres_missing_renders_na(mem_db, no_photon): pct_contained=None, irwin_id=_IRWIN_C, landclass="Sawtooth National Forest") - wire = handle_wfigs(cn.normalize(env), env, env["subject"], now=1_000_000) + wire = handle_wfigs(_normalize_wfigs(env), env, env["subject"], now=1_000_000) assert wire is not None assert "size unknown" in wire assert "containment unknown" in wire @@ -223,7 +354,7 @@ def test_d_ia_placeholder_passthrough(mem_db, no_photon): daily_acres=None, pct_contained=None, landclass="Sawtooth National Forest", irwin_id=_IRWIN_B) - wire = handle_wfigs(cn.normalize(env), env, env["subject"], now=1_000_000) + wire = handle_wfigs(_normalize_wfigs(env), env, env["subject"], now=1_000_000) assert wire is not None assert "IA 1" in wire @@ -233,7 +364,7 @@ def test_d_ia_placeholder_passthrough(mem_db, no_photon): # ============================================================================ def test_e_tombstone_returns_none_and_logs(mem_db, no_photon): env = _make_tombstone() - n = cn.normalize(env) + n = _normalize_wfigs(env) assert n["_kind"] == "wfigs_tombstone" out = handle_wfigs(n, env, env["subject"], now=2_000_000) assert out is None @@ -257,7 +388,7 @@ def test_e_tombstone_returns_none_and_logs(mem_db, no_photon): # ============================================================================ def test_f_perimeter_returns_none_and_logs(mem_db, no_photon): env = _make_perimeter() - n = cn.normalize(env) + n = _normalize_wfigs(env) assert n["_kind"] == "wfigs_perimeter" out = handle_wfigs(n, env, env["subject"], now=3_000_000) assert out is None @@ -278,7 +409,7 @@ def test_g_new_irwin_inserts_and_broadcasts(mem_db, no_photon): env = _make_active_envelope(geocoder_city="Burley") # avoids Photon path now = 5_000_000 data = {} - wire = handle_wfigs(cn.normalize(env), env, env["subject"], + wire = handle_wfigs(_normalize_wfigs(env), env, env["subject"], data=data, now=now) assert wire is not None assert wire.startswith("🔥 Cache Peak Fire — New") @@ -336,14 +467,14 @@ def test_h_known_irwin_no_change_drops(mem_db, no_photon): env = _make_active_envelope(geocoder_city="Burley", fire_discovery_dt_ms=(first_now - 2 * 86400) * 1000) data0 = {} - handle_wfigs(cn.normalize(env), env, env["subject"], + handle_wfigs(_normalize_wfigs(env), env, env["subject"], data=data0, now=first_now) # v0.5.8b: dispatcher commit closes the broadcast. data0["_on_broadcast_committed"](float(first_now)) # Re-publish the same incident exactly 30 min later: same acres + contained. later = first_now + 1800 - out = handle_wfigs(cn.normalize(env), env, env["subject"], now=later) + out = handle_wfigs(_normalize_wfigs(env), env, env["subject"], now=later) assert out is None fr = mem_db.execute( @@ -377,7 +508,7 @@ def test_i_known_irwin_change_inside_cooldown_drops(mem_db, no_photon): geocoder_city="Burley", fire_discovery_dt_ms=(_base - 2 * 86400) * 1000) data0 = {} - handle_wfigs(cn.normalize(env_initial), env_initial, + handle_wfigs(_normalize_wfigs(env_initial), env_initial, env_initial["subject"], data=data0, now=_base) data0["_on_broadcast_committed"](float(_base)) @@ -387,7 +518,7 @@ def test_i_known_irwin_change_inside_cooldown_drops(mem_db, no_photon): geocoder_city="Burley", daily_acres=3000.0, pct_contained=23, fire_discovery_dt_ms=(_base - 2 * 86400) * 1000) later = _base + 4 * 3600 - out = handle_wfigs(cn.normalize(env_grown), env_grown, + out = handle_wfigs(_normalize_wfigs(env_grown), env_grown, env_grown["subject"], now=later) assert out is None @@ -407,7 +538,7 @@ def test_i_known_irwin_change_inside_cooldown_drops(mem_db, no_photon): def test_j_known_irwin_change_after_cooldown_broadcasts(mem_db, no_photon): env_initial = _make_active_envelope(geocoder_city="Burley") data_j0 = {} - handle_wfigs(cn.normalize(env_initial), env_initial, + handle_wfigs(_normalize_wfigs(env_initial), env_initial, env_initial["subject"], data=data_j0, now=5_000_000) data_j0["_on_broadcast_committed"](float(5_000_000)) @@ -415,7 +546,7 @@ def test_j_known_irwin_change_after_cooldown_broadcasts(mem_db, no_photon): daily_acres=3000.0, pct_contained=35) later = 5_000_000 + WFIGS_BROADCAST_COOLDOWN_S data2 = {} - out = handle_wfigs(cn.normalize(env_grown), env_grown, + out = handle_wfigs(_normalize_wfigs(env_grown), env_grown, env_grown["subject"], data=data2, now=later) assert out is not None assert out.startswith("🔥 Cache Peak Fire — Update") @@ -439,7 +570,7 @@ def test_k_anchor_geocoder_city_wins(mem_db, no_photon): env = _make_active_envelope(geocoder_city="Twin Falls", landclass="Sawtooth NF", county="Cassia") - wire = handle_wfigs(cn.normalize(env), env, env["subject"], now=1) + wire = handle_wfigs(_normalize_wfigs(env), env, env["subject"], now=1) assert "Twin Falls" in wire assert "Sawtooth NF" not in wire assert "Cassia Co" not in wire @@ -449,38 +580,38 @@ def test_k_anchor_falls_to_nearest_town(monkeypatch, mem_db): """When city missing, nearest_town(distance, bearing) feeds the anchor.""" fake = {"name": "Boise", "distance_mi": 47.0, "bearing": "S"} monkeypatch.setattr( - "meshai.central_normalizer.nearest_town", + "meshai.geo.nearest_town", lambda lat, lon, max_distance_mi=50.0: fake, ) env = _make_active_envelope(geocoder_city=None, landclass="Sawtooth NF", county="Cassia") - wire = handle_wfigs(cn.normalize(env), env, env["subject"], now=1) + wire = handle_wfigs(_normalize_wfigs(env), env, env["subject"], now=1) # Handler now resolves anchor via town_anchors table (Burley @ 42.536, -113.793) assert "Burley" in wire def test_k_anchor_falls_to_landclass(monkeypatch, mem_db): monkeypatch.setattr( - "meshai.central_normalizer.nearest_town", + "meshai.geo.nearest_town", lambda lat, lon, max_distance_mi=50.0: None, ) env = _make_active_envelope(geocoder_city=None, landclass="Sawtooth National Forest", county="Cassia") - wire = handle_wfigs(cn.normalize(env), env, env["subject"], now=1) + wire = handle_wfigs(_normalize_wfigs(env), env, env["subject"], now=1) # Handler resolves nearest town from town_anchors table, overriding landclass assert "Burley" in wire def test_k_anchor_falls_to_county(monkeypatch, mem_db): monkeypatch.setattr( - "meshai.central_normalizer.nearest_town", + "meshai.geo.nearest_town", lambda lat, lon, max_distance_mi=50.0: None, ) env = _make_active_envelope(geocoder_city=None, landclass=None, county="Cassia", state="ID") - wire = handle_wfigs(cn.normalize(env), env, env["subject"], now=1) + wire = handle_wfigs(_normalize_wfigs(env), env, env["subject"], now=1) # Handler resolves nearest town from town_anchors table assert "Burley" in wire @@ -488,11 +619,11 @@ def test_k_anchor_falls_to_county(monkeypatch, mem_db): def test_k_anchor_nearest_town_under_one_mile_says_near(monkeypatch, mem_db): fake = {"name": "Burley", "distance_mi": 0.3, "bearing": "N"} monkeypatch.setattr( - "meshai.central_normalizer.nearest_town", + "meshai.geo.nearest_town", lambda lat, lon, max_distance_mi=50.0: fake, ) env = _make_active_envelope(geocoder_city=None) - wire = handle_wfigs(cn.normalize(env), env, env["subject"], now=1) + wire = handle_wfigs(_normalize_wfigs(env), env, env["subject"], now=1) # Handler resolves anchor via town_anchors; exact format depends on distance assert "Burley" in wire @@ -506,7 +637,7 @@ def _run_handler_only(env, data=None, now=None): """Run normalize + handler WITHOUT invoking any commit callback. Simulates the dispatcher dropping the broadcast (grace/cooldown/stale) after the handler has already written persistence rows.""" - n = cn.normalize(env) + n = _normalize_wfigs(env) if data is None: data = {} wire = handle_wfigs(n, env, env["subject"], data=data, now=now) @@ -613,7 +744,7 @@ def test_h_handler_attaches_audit_descriptor_and_callback(mem_db, no_photon): dispatcher hooks attached.""" env = _make_active_envelope(geocoder_city="Burley", irwin_id=_IRWIN_B) data = {} - wire = handle_wfigs(cn.normalize(env), env, env["subject"], + wire = handle_wfigs(_normalize_wfigs(env), env, env["subject"], data=data, now=50_000) assert wire is not None assert callable(data["_on_broadcast_committed"]) @@ -677,7 +808,7 @@ def test_wfigs_discovery_is_date_only(): def test_per_fire_wfigs_broadcasts_new_fire(mem_db, no_photon): env = _make_active_envelope(geocoder_city="Burley") data = {} - wire = handle_wfigs(cn.normalize(env), env, env["subject"], + wire = handle_wfigs(_normalize_wfigs(env), env, env["subject"], data=data, now=6_000_000) assert wire is not None assert wire.startswith("🔥 Cache Peak Fire — New") diff --git a/work/tests/test_wzdx_coalescing.py b/work/tests/test_wzdx_coalescing.py index 0957043..c58cd09 100644 --- a/work/tests/test_wzdx_coalescing.py +++ b/work/tests/test_wzdx_coalescing.py @@ -11,13 +11,13 @@ from types import SimpleNamespace import pytest -import meshai.central_normalizer as cn +from meshai import geo from meshai.env.wzdx import WZDxAdapter @pytest.fixture(autouse=True) def _no_photon(monkeypatch): - monkeypatch.setattr(cn, "nearest_town", lambda *a, **k: None) + monkeypatch.setattr(geo, "nearest_town", lambda *a, **k: None) @pytest.fixture