matt/meshai
1
0
Fork 0
mirror of https://github.com/zvx-echo6/meshai.git synced 2026-06-10 17:04:45 +02:00
Code Issues Projects Releases Packages Wiki Activity

feat(content): v0.5.8-state_511_atis -- central_normalizer with Photon nearest_town + composer bypass + SB->S route normalization

Browse source 
First per-adapter content formatter in the meshai-side central_normalizer library (per Central response to schema-divergence + nearest-town reports). state_511_atis (94% of Idaho 511 work-zone traffic) now produces clean wire strings like "🚧 SH-55, near McCall: both directions, emergency repairs" instead of the previous "🚧 ROADS: Work Zone, US-ID. routine -- roadwork".

Implementation: nearest_town(lat, lon) calls Photon directly at 100.64.0.24:2322/reverse with osm_tag=place + client-side filter for city/town/village/hamlet (Navi passthrough route documented in Central response does not exist on current Navi instance). H3-cell-7 LRU cache. Town fallback chain: _enriched.geocoder.city -> nearest_town(coords) -> drop segment. Composer bypass via event.data["_meshai_precomposed"] flag -- renderer owns full wire string for normalized events. SB->S route normalization. distance<1mi -> "near X".

Tests: 535 passed (was 511, +24 net). Synthetic probe over 25 bucket-B + 8 fixture envelopes confirmed 23/25 + 8/8 produce clean output; 2/25 fell back to None (drop segment) on Photon index gaps near Boise/Cascade. Matt eyeballed and approved.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Matt Johnson 2026-06-04 21:38:40 +00:00
commit 7751a40c6c
15 changed files with 1801 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

26
meshai/central/consumer.py
View file

@ -411,10 +411,36 @@ class CentralConsumer:
friendly_name = str(ci["name"])
except Exception:
pass
# v0.5.8 (first per-adapter normalizer): state_511_atis work_zone /
# closure / incident events get a rich one-line title synthesized by
# the meshai.central_normalizer module + the work_zone renderer.
# Failures / unmapped adapters fall through to the registry-friendly
# name chain below.
synthesized = None
try:
from meshai.central_normalizer import normalize as _norm_envelope
from meshai.notifications.renderers.work_zone import format_work_zone_mesh
n = _norm_envelope(envelope)
if n is not None and category in ("work_zone", "road_closure", "road_incident"):
synthesized = format_work_zone_mesh(n) or None
except Exception:
logger.exception("normalizer/renderer failed for adapter=%s category=%s",
inner.get("adapter"), category)
synthesized = None
title = (data.get("title") or data.get("headline")
or synthesized
or friendly_name or cat_raw
or f"{inner.get('adapter', 'central')} event")
# v0.5.8 Option A: when the per-adapter normalizer produced a fully
# formatted mesh string, set a marker on event.data so the composer
# at dispatch time can pass it through verbatim (no family prefix,
# no region tail, no severity append).
if synthesized and title == synthesized:
data["_meshai_precomposed"] = True
kwargs = dict(
title=str(title)[:200],
summary="",

479
meshai/central_normalizer.py Normal file
View file

@ -0,0 +1,479 @@
"""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
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 --------------------------------------------
# Small lookup of Idaho (+ a few neighbor) towns -> (lat, lon). Used to
# render "X mi <bearing> of <town>" when the reverse-geocoder picked a
# city we know. Built from the geocoder.city values seen across 60-sample
# probe + major Idaho cities the operator's mesh is most likely to care
# about. Misses fall through silently: distance_mi/bearing stay None.
_TOWN_COORDS: dict[str, tuple[float, float]] = {
"boise": (43.6150, -116.2023),
"meridian": (43.6121, -116.3915),
"nampa": (43.5407, -116.5635),
"caldwell": (43.6629, -116.6874),
"idaho falls": (43.4666, -112.0340),
"pocatello": (42.8713, -112.4455),
"twin falls": (42.5630, -114.4609),
"coeur d'alene": (47.6777, -116.7805),
"lewiston": (46.4165, -117.0177),
"moscow": (46.7324, -117.0002),
"sandpoint": (48.2766, -116.5535),
"post falls": (47.7180, -116.9516),
"hayden": (47.7660, -116.7866),
"rathdrum": (47.8121, -116.8950),
"plummer": (47.3344, -116.8856),
"kellogg": (47.5380, -116.1352),
"bonners ferry": (48.6914, -116.3181),
"rexburg": (43.8260, -111.7897),
"blackfoot": (43.1905, -112.3447),
"burley": (42.5360, -113.7928),
"jerome": (42.7252, -114.5187),
"mountain home": (43.1330, -115.6912),
"stanley": (44.2160, -114.9311),
"salmon": (45.1758, -113.8957),
"mccall": (44.9111, -116.0987),
"weiser": (44.2510, -116.9690),
"soda springs": (42.6543, -111.6047),
"preston": (42.0963, -111.8766),
"montpelier": (42.3232, -111.2980),
}
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()
coords = _TOWN_COORDS.get(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).
PHOTON_BASE_URL = "http://100.64.0.24:2322"
PHOTON_TIMEOUT_S = 2.0
PHOTON_RADIUS_KM = 80 # ≈ 50 miles
PHOTON_LIMIT = 10
# 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", "hamlet", "suburb", "locality"})
# 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": PHOTON_RADIUS_KM,
"osm_tag": "place",
"limit": PHOTON_LIMIT,
})
url = f"{PHOTON_BASE_URL}/reverse?{qs}"
try:
with urllib.request.urlopen(url, timeout=PHOTON_TIMEOUT_S) 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 PHOTON_BASE_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,
}
# ---------- 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":
return _parse_state_511_atis(inner_data, geo)
# Other adapters await per-adapter parsers; return None to defer.
return None

9
meshai/notifications/renderers/composer.py
View file

@ -299,7 +299,16 @@ def compose_mesh_message(event: Event) -> str:
Single line, no newlines. Drops segments wholesale (lowest priority first)
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.
"""
if event.data and event.data.get("_meshai_precomposed") and event.title:
return event.title
emoji = _category_emoji(event)
label = _category_label(event)
head = f"{emoji} {label}:"

205
meshai/notifications/renderers/work_zone.py Normal file
View file

@ -0,0 +1,205 @@
"""work_zone mesh-string renderer.
Consumes a `central_normalizer.normalize()` output dict (the data layer)
and produces a friendly mesh-broadcast string under an 80-byte UTF-8 cap.
Pure formatting; no adapter knowledge.
Format (per Matt-approved spec):
🚧 <road> @ mile <start>[–<end>][, <dist> mi <bearing> of <town>]:
<direction phrase>, <sub_type>[, ends <when>]
Optional segments are dropped wholesale (lowest-priority first) when the
byte budget is over:
1. ends <when> (lowest priority — dropped first)
2. <bearing> of <town> distance segment
3. <sub_type>
4. <direction phrase>
Required: emoji + road. If even those overrun, the road is truncated
codepoint-safe with an ellipsis. Emojis count as 4 bytes each in UTF-8.
"""
from datetime import datetime, timedelta
from typing import Optional
_BYTE_BUDGET = 80
def _bytelen(s: str) -> int:
return len(s.encode("utf-8"))
def _format_end_short(ends_at: Optional[datetime], now: Optional[datetime] = None) -> Optional[str]:
"""Format a future datetime as a tight mesh-friendly string.
- within 24h -> 'today 6pm' or 'tomorrow 9am'
- within 7 days -> 'Fri 6pm'
- 7-365 days -> 'Jun 15'
- past or far future-> None (caller drops the segment)
"""
if ends_at is None: return None
if now is None: now = datetime.now()
# Normalize to naive (renderer doesn't care about tz here; both inputs
# should be Boise-local in practice since the upstream feed uses local).
if ends_at.tzinfo is not None:
ends_at = ends_at.replace(tzinfo=None)
if now.tzinfo is not None:
now = now.replace(tzinfo=None)
delta = ends_at - now
if delta.total_seconds() < 0:
return None
hour = ends_at.hour
minute = ends_at.minute
if hour == 0:
time_part = "12am" if minute == 0 else f"12:{minute:02d}am"
elif hour < 12:
time_part = f"{hour}am" if minute == 0 else f"{hour}:{minute:02d}am"
elif hour == 12:
time_part = "12pm" if minute == 0 else f"12:{minute:02d}pm"
else:
time_part = f"{hour-12}pm" if minute == 0 else f"{hour-12}:{minute:02d}pm"
if delta < timedelta(hours=24) and ends_at.date() == now.date():
return f"today {time_part}"
if delta < timedelta(hours=48) and ends_at.date() == (now + timedelta(days=1)).date():
return f"tomorrow {time_part}"
if delta < timedelta(days=7):
wd = ends_at.strftime("%a") # 'Mon' .. 'Sun'
return f"{wd} {time_part}"
if delta < timedelta(days=365):
return ends_at.strftime("%b %-d") if hasattr(datetime, "now") else ends_at.strftime("%b ") + str(ends_at.day)
return None
def _format_direction_phrase(direction: Optional[str]) -> Optional[str]:
"""Render the normalized direction as a mesh-friendly noun phrase."""
if not direction or direction == "unknown":
return None
if direction == "both":
return "both directions"
return direction
def _format_mile_segment(mile_start: Optional[int], mile_end: Optional[int]) -> Optional[str]:
if mile_start is None: return None
if mile_end is not None and mile_end != mile_start:
return f"@ mile {mile_start}–{mile_end}"
return f"@ mile {mile_start}"
def _format_distance_segment(distance_mi: Optional[int], bearing: Optional[str], town: Optional[str]) -> Optional[str]:
if not town: return None
# Issue 2 polish: distances under 1 mi are uninformative ("0 mi S of
# McCall" reads worse than "near McCall"). Drop the distance/bearing
# pair and fall back to plain "near <town>" form.
if distance_mi is not None and bearing and distance_mi >= 1:
return f"{distance_mi} mi {bearing} of {town}"
return f"near {town}"
def _truncate_road(road: str, budget: int) -> str:
"""Truncate `road` to fit in `budget` UTF-8 bytes, codepoint-safe."""
if _bytelen(road) <= budget:
return road
cut = road
while cut and _bytelen(cut + "…") > budget:
cut = cut[:-1]
return cut + "…" if cut else "…"
def format_work_zone_mesh(n: dict, now: Optional[datetime] = None) -> str:
"""Render a normalized work_zone dict to a mesh-friendly string.
Always returns a string; drops segments to fit the 80-byte cap.
"""
emoji = "🚧"
raw_road = n.get("road")
town = n.get("town")
# Issue 3d: when the road is uninformative (set None by the normalizer
# for "Exit 80 Southbound On Ramp" shapes) AND we have a town, lead
# with the town as the head instead of a useless placeholder.
if raw_road:
road = raw_road
head = f"{emoji} {road}"
suppress_distance_seg = False
elif town:
# "🚧 near <town>" or "🚧 <dist> mi <bearing> of <town>" as head.
# distance/bearing are folded INTO the head; suppress the separate
# distance segment below to avoid duplication.
road = town # used only by the last-resort road-truncation branch
head = f"{emoji} {_format_distance_segment(n.get('distance_mi'), n.get('bearing'), town)}"
suppress_distance_seg = True
else:
road = "Road event"
head = f"{emoji} {road}"
suppress_distance_seg = False
mile_seg = _format_mile_segment(n.get("mile_start"), n.get("mile_end")) if raw_road else None
dist_seg = None if suppress_distance_seg else \
_format_distance_segment(n.get("distance_mi"), n.get("bearing"), town)
dir_phrase = _format_direction_phrase(n.get("direction"))
sub = n.get("sub_type")
impact = n.get("impact")
if impact == "full_closure":
# Promote full-closure into the description slot so it's louder.
sub = f"all lanes closed{' (' + sub + ')' if sub else ''}"
ends_seg = _format_end_short(n.get("ends_at"), now=now)
# Optional segments in build order; each has a drop_priority where
# HIGHER number = dropped FIRST when over budget.
Segment = tuple # (drop_priority, joiner_before, text)
segs: list[tuple[int, str, str]] = []
if mile_seg:
segs.append((10, " ", mile_seg)) # mile segment is high-value, keep longest
if dist_seg:
segs.append((20, ", ", dist_seg))
if dir_phrase or sub:
segs.append((30, ": ", "")) # marker for the colon transition
if dir_phrase:
segs.append((30, "", dir_phrase))
if sub:
segs.append((40, ", " if dir_phrase else "", sub))
if ends_seg:
segs.append((50, ", ends ", ends_seg))
# Iteratively assemble; drop highest-priority segments until under budget.
kept = list(range(len(segs)))
while True:
out = head
last_was_colon = False
for i in kept:
prio, joiner, text = segs[i]
if text == "" and joiner == ": ":
out += ": "
last_was_colon = True
continue
# If we're right after a ": " marker, the first content segment
# joins with no extra delimiter even if its joiner was ", ".
if last_was_colon:
out += text
last_was_colon = False
else:
out += joiner + text
if _bytelen(out) <= _BYTE_BUDGET:
return out
# Find the highest-priority kept segment with non-empty content;
# drop it (and its preceding colon marker if it was the only one
# past the colon).
droppable = [i for i in kept if segs[i][2] != ""]
if not droppable:
# All optional segments gone; truncate road.
budget_for_road = _BYTE_BUDGET - len((emoji + " ").encode("utf-8"))
return f"{emoji} {_truncate_road(road, budget_for_road)}"
worst = max(droppable, key=lambda i: segs[i][0])
kept.remove(worst)
# If we dropped both dir_phrase and sub, remove the ": " marker too.
remaining_after_colon = any(
segs[i][2] != "" for i in kept
if any(segs[j][0] == 30 and segs[j][2] == "" for j in range(len(segs)) if j < i)
)
if not remaining_after_colon:
kept = [i for i in kept if not (segs[i][2] == "" and segs[i][1] == ": ")]

1
requirements.txt
View file

@ -9,6 +9,7 @@ httpx>=0.25.0
fastembed>=0.3.0
sqlite-vec>=0.1.0
numpy
h3>=4.0
fastapi>=0.110.0
uvicorn[standard]>=0.27.0
aiomqtt>=2.0.0

60
tests/fixtures/central_envelopes/state_511_atis_01_I-15.json vendored Normal file
View file

@ -0,0 +1,60 @@
{
"id": "ID:Construction:33902",
"source": "central.echo6.co",
"type": "central.work_zone.state_511_atis.v1",
"time": "2026-05-28T14:54:00+00:00",
"datacontenttype": "application/json",
"centralschemaversion": "1.0",
"centralcategory": "work_zone.state_511_atis",
"centralseverity": 1,
"specversion": "1.0",
"data": {
"id": "ID:Construction:33902",
"adapter": "state_511_atis",
"category": "work_zone.state_511_atis",
"time": "2026-05-28T14:54:00Z",
"expires": "2026-05-29T15:00:00Z",
"severity": 1,
"geo": {
"centroid": [
-112.358931947559,
43.2045296484166
],
"bbox": null,
"regions": [
"US-ID"
],
"primary_region": "US-ID",
"geometry": null
},
"data": {
"roadway_name": "I-15",
"description": "Road construction on I-15 Southbound from MM (93) to MM (89). 1 Right lane closed. 5/29/2026 10:00 AM to 5/29/2026 3:00 PM Fri: 10:00 AM - 3:00 PM Width Restriction: 19ft Speed Restriction: 65mph Activities: Mandatory Speed Limit in Force, Use Caution.<div class='cellSpacer'><i><b>Comments:</b></i> Pile Driving on North West side of the Interstate for bridge foundations. South Bound right lane closed.</div>",
"event_sub_type": "roadConstruction",
"direction": "South",
"location_description": "I-15-BL | I-15-BL",
"county": "Bingham",
"state": "Idaho",
"start_date": "5/29/26, 10:00 AM",
"last_updated": "5/28/26, 2:54 PM",
"is_full_closure": false,
"layer": "Construction",
"state_code": "ID",
"latitude": 43.2045296484166,
"longitude": -112.358931947559,
"_enriched": {
"geocoder": {
"name": "Jensen Grove Disc Golf Park",
"city": "Blackfoot",
"county": "Bingham",
"state": "Idaho",
"country": "United States",
"postal_code": "83221",
"timezone": "America/Boise",
"landclass": null,
"elevation_m": 1366.6015625
}
}
}
}
}

60
tests/fixtures/central_envelopes/state_511_atis_02_SH-36.json vendored Normal file
View file

@ -0,0 +1,60 @@
{
"id": "ID:Construction:33202",
"source": "central.echo6.co",
"type": "central.work_zone.state_511_atis.v1",
"time": "2026-05-21T10:40:00+00:00",
"datacontenttype": "application/json",
"centralschemaversion": "1.0",
"centralcategory": "work_zone.state_511_atis",
"centralseverity": 1,
"specversion": "1.0",
"data": {
"id": "ID:Construction:33202",
"adapter": "state_511_atis",
"category": "work_zone.state_511_atis",
"time": "2026-05-21T10:40:00Z",
"expires": "2026-06-05T16:30:00Z",
"severity": 1,
"geo": {
"centroid": [
-111.622569529543,
42.3143070666171
],
"bbox": null,
"regions": [
"US-ID"
],
"primary_region": "US-ID",
"geometry": null
},
"data": {
"roadway_name": "SH-36",
"description": "Paving Operations on SH-36 from MM (17) to MM (18). 6/1/2026 6:00 AM to 6/5/2026 4:30 PM Mon, Tue, Wed, Thu, Fri: Active all day Width Restriction: 10ft Activities: Pilot Car in Operation, Reduced to Single Lane, Alternating Direction of Travel, Use Caution, Warning.",
"event_sub_type": "pavingOperations",
"direction": "Unknown",
"location_description": "NF-441 | NF-444",
"county": "Franklin",
"state": "Idaho",
"start_date": "6/1/26, 6:00 AM",
"last_updated": "5/21/26, 10:40 AM",
"is_full_closure": false,
"layer": "Construction",
"state_code": "ID",
"latitude": 42.3143070666171,
"longitude": -111.622569529543,
"_enriched": {
"geocoder": {
"name": "Cache Nf Road 444",
"city": null,
"county": "Franklin",
"state": "Idaho",
"country": "United States",
"postal_code": null,
"timezone": "America/Boise",
"landclass": "Cache National Forest",
"elevation_m": 2035.1875
}
}
}
}
}

60
tests/fixtures/central_envelopes/state_511_atis_03_I-15.json vendored Normal file
View file

@ -0,0 +1,60 @@
{
"id": "ID:Construction:33281",
"source": "central.echo6.co",
"type": "central.work_zone.state_511_atis.v1",
"time": "2026-05-29T08:03:00+00:00",
"datacontenttype": "application/json",
"centralschemaversion": "1.0",
"centralcategory": "work_zone.state_511_atis",
"centralseverity": 1,
"specversion": "1.0",
"data": {
"id": "ID:Construction:33281",
"adapter": "state_511_atis",
"category": "work_zone.state_511_atis",
"time": "2026-05-29T08:03:00Z",
"expires": "2026-06-03T17:00:00Z",
"severity": 1,
"geo": {
"centroid": [
-112.388532253628,
43.1524044098836
],
"bbox": null,
"regions": [
"US-ID"
],
"primary_region": "US-ID",
"geometry": null
},
"data": {
"roadway_name": "I-15",
"description": "Bridge construction on I-15 Northbound from MM (89) to MM (93). 1 Right lane closed. 6/1/2026 7:00 AM to 6/3/2026 5:00 PM Mon: Paused all day, Tue, Wed: 7:00 AM - 5:00 PM Speed Restriction: 65mph Activities: Mandatory Speed Limit in Force, Use Caution.<div class='cellSpacer'><i><b>Comments:</b></i> North Bound right lane closed with speed reduction to 65 mph.\n</div>",
"event_sub_type": "bridgeConstruction",
"direction": "North",
"location_description": "I-15-BL | Snake River",
"county": "Bingham",
"state": "Idaho",
"start_date": "6/1/26, 7:00 AM",
"last_updated": "5/29/26, 8:03 AM",
"is_full_closure": false,
"layer": "Construction",
"state_code": "ID",
"latitude": 43.1524044098836,
"longitude": -112.388532253628,
"_enriched": {
"geocoder": {
"name": "North Treaty Highway",
"city": null,
"county": "Bingham",
"state": "Idaho",
"country": "United States",
"postal_code": null,
"timezone": "America/Boise",
"landclass": "Fort Hall Reservation",
"elevation_m": 1367.98828125
}
}
}
}
}

60
tests/fixtures/central_envelopes/state_511_atis_04_US-95.json vendored Normal file
View file

@ -0,0 +1,60 @@
{
"id": "ID:Construction:33897",
"source": "central.echo6.co",
"type": "central.work_zone.state_511_atis.v1",
"time": "2026-05-28T14:38:00+00:00",
"datacontenttype": "application/json",
"centralschemaversion": "1.0",
"centralcategory": "work_zone.state_511_atis",
"centralseverity": 1,
"specversion": "1.0",
"data": {
"id": "ID:Construction:33897",
"adapter": "state_511_atis",
"category": "work_zone.state_511_atis",
"time": "2026-05-28T14:38:00Z",
"expires": "2026-06-02T17:00:00Z",
"severity": 1,
"geo": {
"centroid": [
-116.411904551719,
48.5439764820932
],
"bbox": null,
"regions": [
"US-ID"
],
"primary_region": "US-ID",
"geometry": null
},
"data": {
"roadway_name": "US-95",
"description": "Utility work on US-95 Southbound near MM (495). 6/2/2026 9:00 AM to 6/2/2026 5:00 PM Tue: Active all day Activities: Use Caution.",
"event_sub_type": "utilityWork",
"direction": "South",
"location_description": "Dusty Ln",
"county": "Boundary",
"state": "Idaho",
"start_date": "6/2/26, 9:00 AM",
"last_updated": "5/28/26, 2:38 PM",
"is_full_closure": false,
"layer": "Construction",
"state_code": "ID",
"latitude": 48.5439764820932,
"longitude": -116.411904551719,
"_enriched": {
"geocoder": {
"name": null,
"city": "Naples",
"county": "Boundary",
"state": "ID",
"country": "United States",
"postal_code": "83847",
"timezone": "America/Los_Angeles",
"landclass": null,
"elevation_m": 659.62890625
}
}
}
}
}

60
tests/fixtures/central_envelopes/state_511_atis_05_W_Prairie_Ave.json vendored Normal file
View file

@ -0,0 +1,60 @@
{
"id": "ID:Construction:33908",
"source": "central.echo6.co",
"type": "central.work_zone.state_511_atis.v1",
"time": "2026-05-28T15:37:00+00:00",
"datacontenttype": "application/json",
"centralschemaversion": "1.0",
"centralcategory": "work_zone.state_511_atis",
"centralseverity": 1,
"specversion": "1.0",
"data": {
"id": "ID:Construction:33908",
"adapter": "state_511_atis",
"category": "work_zone.state_511_atis",
"time": "2026-05-28T15:37:00Z",
"expires": "2026-06-13T18:00:00Z",
"severity": 1,
"geo": {
"centroid": [
-116.804061047849,
47.74449
],
"bbox": null,
"regions": [
"US-ID"
],
"primary_region": "US-ID",
"geometry": null
},
"data": {
"roadway_name": "W Prairie Ave",
"description": "Minor Paving Operations on W Prairie Ave Both Directions from N Ramsey Rd to N Government Way. Lanes Alternating. 5/28/2026 7:00 AM to 6/13/2026 6:00 PM Mon, Tue, Wed, Thu, Fri: Active all day, Sat, Sun: Paused all day<div class='cellSpacer'><i><b>Comments:</b></i> Lakes Highway District is performing paving operations. US-95 will have the left turn lanes reduced to one left turn lane in both directions, and there will be alternating lane closures on Prairie Ave. reducing to one lane in both directions. Reduce your speed and lookout for workers on the roadway.</div>",
"event_sub_type": "pavingOperations",
"direction": "Both",
"location_description": "N Ramsey Rd | N Government Way",
"county": "Kootenai",
"state": "Idaho",
"start_date": "5/28/26, 7:00 AM",
"last_updated": "5/28/26, 3:37 PM",
"is_full_closure": false,
"layer": "Construction",
"state_code": "ID",
"latitude": 47.74449,
"longitude": -116.804061047849,
"_enriched": {
"geocoder": {
"name": "Sandpiper Way",
"city": "Hayden",
"county": "Kootenai",
"state": "Idaho",
"country": "United States",
"postal_code": "83835",
"timezone": "America/Los_Angeles",
"landclass": null,
"elevation_m": 695.83984375
}
}
}
}
}

60
tests/fixtures/central_envelopes/state_511_atis_06_SH-55.json vendored Normal file
View file

@ -0,0 +1,60 @@
{
"id": "ID:Construction:33930",
"source": "central.echo6.co",
"type": "central.work_zone.state_511_atis.v1",
"time": "2026-05-28T17:22:00+00:00",
"datacontenttype": "application/json",
"centralschemaversion": "1.0",
"centralcategory": "work_zone.state_511_atis",
"centralseverity": 1,
"specversion": "1.0",
"data": {
"id": "ID:Construction:33930",
"adapter": "state_511_atis",
"category": "work_zone.state_511_atis",
"time": "2026-05-28T17:22:00Z",
"expires": "2026-05-29T08:00:00Z",
"severity": 1,
"geo": {
"centroid": [
-116.09759,
44.9065083834611
],
"bbox": null,
"regions": [
"US-ID"
],
"primary_region": "US-ID",
"geometry": null
},
"data": {
"roadway_name": "SH-55",
"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: Active all day<div class='cellSpacer'><i><b>Comments:</b></i> Emergency fiber repair</div>",
"event_sub_type": "emergencyRepairs",
"direction": "Both",
"location_description": "Washington St",
"county": "Valley",
"state": "Idaho",
"start_date": "5/28/26, 5:00 PM",
"last_updated": "5/28/26, 5:22 PM",
"is_full_closure": false,
"layer": "Construction",
"state_code": "ID",
"latitude": 44.9065083834611,
"longitude": -116.09759,
"_enriched": {
"geocoder": {
"name": "Shell",
"city": "McCall",
"county": "Valley",
"state": "ID",
"country": "United States",
"postal_code": "83638",
"timezone": "America/Boise",
"landclass": null,
"elevation_m": 1537.4609375
}
}
}
}
}

60
tests/fixtures/central_envelopes/state_511_atis_07_US-2.json vendored Normal file
View file

@ -0,0 +1,60 @@
{
"id": "ID:Construction:32196",
"source": "central.echo6.co",
"type": "central.work_zone.state_511_atis.v1",
"time": "2026-05-11T15:13:00+00:00",
"datacontenttype": "application/json",
"centralschemaversion": "1.0",
"centralcategory": "work_zone.state_511_atis",
"centralseverity": 1,
"specversion": "1.0",
"data": {
"id": "ID:Construction:32196",
"adapter": "state_511_atis",
"category": "work_zone.state_511_atis",
"time": "2026-05-11T15:13:00Z",
"expires": "2026-06-02T16:00:00Z",
"severity": 1,
"geo": {
"centroid": [
-116.89192200007,
48.1805200000001
],
"bbox": null,
"regions": [
"US-ID"
],
"primary_region": "US-ID",
"geometry": null
},
"data": {
"roadway_name": "US-2",
"description": "Minor Road construction on US-2 Eastbound from Keyser Ln to N Riley Creek Rd. 6/1/2026 6:30 AM to 6/2/2026 4:00 PM Mon: 5:00 AM - 3:00 PM, Tue: 5:30 AM - 3:00 PM Activities: Reduced to Single Lane, Alternating Direction of Travel. Expect Delays: Under 15 minutes<div class='cellSpacer'><i><b>Comments:</b></i> road work flaggers in area</div>",
"event_sub_type": "roadConstruction",
"direction": "East",
"location_description": "Keyser Ln | N Riley Creek Rd",
"county": "Bonner",
"state": "Idaho",
"start_date": "6/1/26, 6:30 AM",
"last_updated": "5/11/26, 3:13 PM",
"is_full_closure": false,
"layer": "Construction",
"state_code": "ID",
"latitude": 48.1805200000001,
"longitude": -116.89192200007,
"_enriched": {
"geocoder": {
"name": "Priest River Park",
"city": null,
"county": "Bonner",
"state": "Idaho",
"country": "United States",
"postal_code": null,
"timezone": "America/Los_Angeles",
"landclass": null,
"elevation_m": 633.83984375
}
}
}
}
}

60
tests/fixtures/central_envelopes/state_511_atis_08_SH-41.json vendored Normal file
View file

@ -0,0 +1,60 @@
{
"id": "ID:Construction:33655",
"source": "central.echo6.co",
"type": "central.work_zone.state_511_atis.v1",
"time": "2026-05-26T07:32:00+00:00",
"datacontenttype": "application/json",
"centralschemaversion": "1.0",
"centralcategory": "work_zone.state_511_atis",
"centralseverity": 1,
"specversion": "1.0",
"data": {
"id": "ID:Construction:33655",
"adapter": "state_511_atis",
"category": "work_zone.state_511_atis",
"time": "2026-05-26T07:32:00Z",
"expires": "2026-06-01T18:00:00Z",
"severity": 1,
"geo": {
"centroid": [
-116.893994470776,
47.8013806004727
],
"bbox": null,
"regions": [
"US-ID"
],
"primary_region": "US-ID",
"geometry": null
},
"data": {
"roadway_name": "SH-41",
"description": "Minor Utility work on SH-41 Southbound near W Boekel Rd. Lane Shift Left. 6/1/2026 7:00 AM to 6/1/2026 6:00 PM Mon: Active all day<div class='cellSpacer'><i><b>Comments:</b></i> Southbound traffic will be shifted to accommodate for Avista to work on power poles. Reduce your speed and lookout for workers on the roadway.</div>",
"event_sub_type": "utilityWork",
"direction": "South",
"location_description": "W Boekel Rd",
"county": "Kootenai",
"state": "Idaho",
"start_date": "6/1/26, 7:00 AM",
"last_updated": "5/26/26, 7:32 AM",
"is_full_closure": false,
"layer": "Construction",
"state_code": "ID",
"latitude": 47.8013806004727,
"longitude": -116.893994470776,
"_enriched": {
"geocoder": {
"name": "American Eagle Automotive",
"city": "Rathdrum",
"county": "Kootenai",
"state": "Idaho",
"country": "United States",
"postal_code": "83858",
"timezone": "America/Los_Angeles",
"landclass": null,
"elevation_m": 673.84765625
}
}
}
}
}

394
tests/test_central_normalizer.py Normal file
View file

@ -0,0 +1,394 @@
"""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

207
tests/test_work_zone_renderer.py Normal file
View file

@ -0,0 +1,207 @@
"""Tests for the work_zone mesh renderer."""
from datetime import datetime, timedelta
import pytest
from meshai.notifications.renderers.work_zone import format_work_zone_mesh
def _bytelen(s: str) -> int:
return len(s.encode("utf-8"))
# ---------- canonical / fully-populated case ------------------------------
def test_all_fields_present_produces_canonical_format():
n = {
"source": "state_511_atis", "road": "SH-3", "direction": "both",
"mile_start": 60, "mile_end": None, "description": "...",
"sub_type": "construction work", "impact": "partial",
"ends_at": datetime(2026, 6, 2, 17, 0),
"town": "Plummer", "distance_mi": 8, "bearing": "N",
}
out = format_work_zone_mesh(n, now=datetime(2026, 6, 1, 14, 0))
assert out.startswith("🚧 SH-3 @ mile 60")
assert "8 mi N of Plummer" in out
assert "both directions" in out
assert "construction work" in out
assert _bytelen(out) <= 80
# ---------- segment-drop progression --------------------------------------
def test_no_mile_drops_at_mile_segment():
n = {"source": "state_511_atis", "road": "W Prairie Ave", "direction": "both",
"mile_start": None, "mile_end": None, "sub_type": "paving",
"impact": "partial", "town": "Coeur d'Alene", "distance_mi": 5, "bearing": "E",
"ends_at": None, "description": ""}
out = format_work_zone_mesh(n)
assert "@ mile" not in out
assert "W Prairie Ave" in out
assert "5 mi E of Coeur d'Alene" in out
def test_no_town_drops_distance_segment():
n = {"source": "state_511_atis", "road": "SH-55", "direction": "both",
"mile_start": 17, "mile_end": 18, "sub_type": "paving",
"impact": "partial", "town": None, "distance_mi": None, "bearing": None,
"ends_at": None, "description": ""}
out = format_work_zone_mesh(n)
assert " mi " not in out
assert " of " not in out
assert "@ mile 17–18" in out
def test_no_ends_drops_ends_suffix():
n = {"source": "state_511_atis", "road": "I-86", "direction": "both",
"mile_start": 58, "mile_end": 59, "sub_type": "bridge maintenance",
"impact": "partial", "town": "Pocatello", "distance_mi": 15, "bearing": "W",
"ends_at": None, "description": ""}
out = format_work_zone_mesh(n)
assert ", ends" not in out
def test_unknown_direction_drops_direction_phrase():
n = {"source": "state_511_atis", "road": "SH-36", "direction": "unknown",
"mile_start": 17, "mile_end": 18, "sub_type": "paving",
"impact": "partial", "town": None, "distance_mi": None, "bearing": None,
"ends_at": None, "description": ""}
out = format_work_zone_mesh(n)
assert "unknown" not in out.lower().split(":")[-1] # no 'unknown' in tail
def test_full_closure_promoted():
n = {"source": "state_511_atis", "road": "I-15", "direction": "southbound",
"mile_start": None, "mile_end": None, "sub_type": "road construction",
"impact": "full_closure", "town": None, "distance_mi": None, "bearing": None,
"ends_at": None, "description": ""}
out = format_work_zone_mesh(n)
assert "all lanes closed" in out
# ---------- byte budget ---------------------------------------------------
def test_byte_length_under_80_for_canonical():
n = {"source": "state_511_atis", "road": "SH-3", "direction": "both",
"mile_start": 60, "mile_end": None, "sub_type": "construction work",
"impact": "partial", "town": "Plummer", "distance_mi": 8, "bearing": "N",
"ends_at": datetime(2026, 6, 2, 17, 0), "description": ""}
out = format_work_zone_mesh(n, now=datetime(2026, 6, 1, 14, 0))
assert _bytelen(out) <= 80
def test_byte_length_under_80_with_long_road_name():
n = {"source": "state_511_atis",
"road": "SCIENCE CENTER DR / E ANDERSON ST / N THIRD WAY",
"direction": "both",
"mile_start": 100, "mile_end": 200, "sub_type": "construction work",
"impact": "partial", "town": "Idaho Falls", "distance_mi": 12, "bearing": "SE",
"ends_at": datetime(2026, 9, 16, 1, 0), "description": ""}
out = format_work_zone_mesh(n, now=datetime(2026, 6, 1, 14, 0))
assert _bytelen(out) <= 80, f"over budget: {len(out.encode('utf-8'))} = {out!r}"
def test_emoji_counts_as_4_bytes():
# 🚧 is U+1F6A7 → 4 bytes in UTF-8.
assert _bytelen("🚧") == 4
def test_extreme_road_name_truncated():
# Force the renderer into the truncate-road last-resort branch.
long_road = "VERY-LONG-ROAD-NAME-" * 10
n = {"source": "state_511_atis", "road": long_road, "direction": None,
"mile_start": None, "mile_end": None, "sub_type": None,
"impact": "partial", "town": None, "distance_mi": None, "bearing": None,
"ends_at": None, "description": ""}
out = format_work_zone_mesh(n)
assert _bytelen(out) <= 80
assert out.startswith("🚧 ")
assert "…" in out or _bytelen(long_road) <= 80 # truncated with ellipsis
# ---------- ends_at relative-time formatting ------------------------------
def test_ends_today_format():
now = datetime(2026, 6, 1, 9, 0)
ends = datetime(2026, 6, 1, 18, 0)
n = {"source": "state_511_atis", "road": "X", "direction": None,
"mile_start": None, "mile_end": None, "sub_type": None, "impact": "partial",
"town": None, "distance_mi": None, "bearing": None,
"ends_at": ends, "description": ""}
out = format_work_zone_mesh(n, now=now)
assert "today" in out and "6pm" in out
def test_ends_within_week_uses_weekday():
now = datetime(2026, 6, 1, 9, 0) # Monday
ends = datetime(2026, 6, 5, 16, 30)
n = {"source": "state_511_atis", "road": "X", "direction": None,
"mile_start": None, "mile_end": None, "sub_type": None, "impact": "partial",
"town": None, "distance_mi": None, "bearing": None,
"ends_at": ends, "description": ""}
out = format_work_zone_mesh(n, now=now)
assert "Fri" in out
assert "4:30pm" in out
def test_ends_past_drops_segment():
now = datetime(2026, 6, 10, 9, 0)
ends = datetime(2026, 5, 5, 17, 0) # already past
n = {"source": "state_511_atis", "road": "X", "direction": None,
"mile_start": None, "mile_end": None, "sub_type": None, "impact": "partial",
"town": None, "distance_mi": None, "bearing": None,
"ends_at": ends, "description": ""}
out = format_work_zone_mesh(n, now=now)
assert ", ends" not in out
# ---------- v0.5.8 distance < 1 mi → "near X" -----------------------------
def test_distance_zero_drops_to_near_only():
n = {"source": "state_511_atis", "road": "SH-55", "direction": "both",
"mile_start": None, "mile_end": None, "sub_type": "emergency repairs",
"impact": "partial", "town": "McCall", "distance_mi": 0, "bearing": "S",
"ends_at": None, "description": ""}
out = format_work_zone_mesh(n)
assert "near McCall" in out
assert "0 mi" not in out
assert "S of McCall" not in out
def test_distance_one_keeps_bearing_segment():
n = {"source": "state_511_atis", "road": "SH-41", "direction": "southbound",
"mile_start": None, "mile_end": None, "sub_type": "utility work",
"impact": "partial", "town": "Rathdrum", "distance_mi": 1, "bearing": "S",
"ends_at": None, "description": ""}
out = format_work_zone_mesh(n)
assert "1 mi S of Rathdrum" in out
# ---------- v0.5.8 no-road fallback (leads with town) ---------------------
def test_no_road_leads_with_town_distance():
n = {"source": "state_511_atis", "road": None, "direction": "southbound",
"mile_start": None, "mile_end": None, "sub_type": "ramp work",
"impact": "partial", "town": "Stanley", "distance_mi": 3, "bearing": "NE",
"ends_at": None, "description": ""}
out = format_work_zone_mesh(n)
# Head is the distance/town form, not a placeholder.
assert out.startswith("🚧 3 mi NE of Stanley")
assert "Road event" not in out
def test_no_road_no_town_falls_back_to_placeholder():
n = {"source": "state_511_atis", "road": None, "direction": "both",
"mile_start": None, "mile_end": None, "sub_type": None,
"impact": "partial", "town": None, "distance_mi": None, "bearing": None,
"ends_at": None, "description": ""}
out = format_work_zone_mesh(n)
# Placeholder is acceptable when we have literally nothing.
assert out.startswith("🚧 ")