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

chore(central-ripout 2e): split central_normalizer.py; eliminate the "central" name

Browse source 
Final structural step. central_normalizer.py (973 LOC, misnamed — it lived
OUTSIDE central/ but its docstring described "the Central firehose") is DELETED.
No file or symbol named "central" survives (only immutable historical comments
in an already-applied migration .sql remain — those are not edited).

Split per the owner's rule ("next to what uses it"):
- Live geo/place-name utilities (nearest_town, _haversine_miles, _bearing_compass,
  _compute_distance_bearing, _h3_cell, _photon_reverse_places, geocoder config)
  → the EXISTING meshai/geo.py ("Geographic utilities…"), same domain. NOT a
  junk drawer. geo.py's nearest_city (hardcoded table) and central_normalizer's
  nearest_town (live Photon+H3 reverse-geocode) are different implementations —
  both kept, no false consolidation.
- WZDx work-zone parsers (_parse_wzdx_federal, _norm_wzdx_sub_type, _norm_direction,
  _parse_mile_posts, _clean_description, _is_uninformative_road) → env/wzdx_parse.py,
  next to their sole consumer env/wzdx.py.
- ~400 dead lines deleted: the Central-envelope machinery (normalize,
  _parse_state_511_atis, _parse_wfigs_incidents, _parse_itd_511_work_zone,
  should_skip_state_511_atis_id, is_incident_envelope_stale, normalize_road_name,
  _norm_sub_type, _parse_ends_at) — only caller was the deleted Central consumer.

Consumers rewired: env/wzdx.py, env/fire_render.py, formatters/fire.py,
formatters/_anchor.py, main.py, persistence/curation.py, composer.py.

Tests: test_central_normalizer.py RENAMED → test_geo_wzdx_parse.py (live geo/wzdx
tests kept + rewired; dead-parser tests dropped). test_itd_511_work_zone.py
deleted (tested only the dead itd_511 parser — verified gone from source).
Other test files: import-path rewires. Live coverage for nearest_town /
_parse_wzdx_federal / bearing preserved across test_geo_wzdx_parse, test_adapter_wzdx,
test_wfigs_handler, test_incident_refactor.

Suite: 2010 passed, 0 failed. The 49-test drop is the dead Central-envelope
parser tests, correctly removed. Stale "Central firehose" docstrings on the
moved code updated to describe present behavior.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Matt Johnson 2026-07-17 23:11:26 +00:00
commit ef2df5f54b
19 changed files with 1167 additions and 2002 deletions
Download patch file Download diff file Expand all files Collapse all files

4
work/meshai/adapter_config/defaults.py
View file

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

973
work/meshai/central_normalizer.py
View file

@ -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
<bearing> 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 <county>" 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

2
work/meshai/env/fire_fusion.py vendored
View file

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

11
work/meshai/env/fire_render.py vendored
View file

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

18
work/meshai/env/wzdx.py vendored
View file

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

273
work/meshai/env/wzdx_parse.py vendored Normal file
View file

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

213
work/meshai/geo.py
View file

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

2
work/meshai/main.py
View file

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

16
work/meshai/notifications/formatters/_anchor.py
View file

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

7
work/meshai/notifications/renderers/composer.py
View file

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

6
work/meshai/persistence/curation.py
View file

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

4
work/tests/test_adapter_wzdx.py
View file

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

804
work/tests/test_central_normalizer.py
View file

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

14
work/tests/test_fire_refactor.py
View file

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

407
work/tests/test_geo_wzdx_parse.py Normal file
View file

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

89
work/tests/test_incident_refactor.py
View file

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

127
work/tests/test_itd_511_work_zone.py
View file

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

191
work/tests/test_wfigs_handler.py
View file

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

4
work/tests/test_wzdx_coalescing.py
View file

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