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
meshai/tests/test_central_normalizer.py
K7ZVX 0099d0fd94 feat(v0.5.9): unified incident pipeline + state_511_atis Idaho cutover + two-sided freshness gate 
Coordinated change across the consumer dispatch layer + central_normalizer + new incident_handler + new itd_511 work_zone parser. The integrated story: Central PM filed a heads-up that ITD 511 publishes four EventTypes (work_zone, closure, incident, special_event) under us.id Convention A, and meshais v0.5.8 work focused on work_zone shape only -- meaning incidents (the higher-priority operational signal) were silently rendering as the v0.5.7-regression-style fallback. This v0.5.9 closes that gap by treating incidents + closures + special_events from all three traffic adapters (state_511_atis, itd_511, tomtom_incidents) as a single unified pipeline, while also migrating Idaho coverage from state_511_atis to itd_511 (the direct ITD feed) at the consumer level. Components: (1) new meshai/central/incident_handler.py routes incident/closure/special_event events through per-adapter parsers (tomtom, itd_511, state_511_atis-non-ID) to a canonical incident shape, then a single rendering pipeline with sub_type-aware emoji selection (jam/crash/road_closed/disabled_vehicle/parade/special_event/vehicle_fire/road_works). (2) Universal two-sided freshness gate in the consumer dispatch layer: only events with 0 <= age <= 1800s (default-allow on missing start_time) make it past the gate. Rejects both stale events (more than 30 min old) AND future-scheduled events (negative age -- a real itd_511 case for scheduled work projects). The gate sits ABOVE both incident_handler and the v0.5.8 work_zone formatter so all adapters get gated uniformly. (3) state_511_atis Idaho cutover -- both incident_handler and the v0.5.8 work_zone parser skip state_511_atis events where the state token is ID, deferring to itd_511 as the authoritative source. state_511_atis remains fully active for non-Idaho neighbor coverage (WA/OR/MT/UT/WY/NV) -- verified by Phase 2 WA broadcasts in the synthetic probe. (4) new itd_511 work_zone parser (extension to central_normalizer.py) consumes the itd_511 work_zone EventType and produces the same MEDIUM-style wire format as the existing state_511_atis work_zone parser (road + mile range + town + direction + sub_type + ends-at). (5) No Update: broadcasts in the incident pipeline -- per Matts call, real-time traffic Updates (jam getting worse, delay growing) are not actionable for mesh users. State tracking continues via traffic_events UPSERT but only the first sighting of an external_id ever fires a New: broadcast. WFIGS handler unchanged -- fires keep their 8h-rate-limited Update: behavior since acres growth IS operationally meaningful (evacuation decisions). Forecast: 3-10 mesh broadcasts/day in Idaho, all New:. Cross-check: original raw broadcast count was 623 against a fixed-clock 49-min synthetic window; after v0.5.9 REVISED (no Updates) it dropped to 18; after v0.5.9 GAMMA (two-sided gate + Idaho cutover) it dropped to 9. Test count: was 589 baseline, +45 net new -- 634 passing. Synthetic probe verified all four phases: Phase 1 (replay 3032 captured envelopes) = 0 broadcasts (correctly suppressed); Phase 2 (synthesized fresh non-ID + ID) = 7 broadcasts; Phase 3 (synthesized fresh itd_511 work_zone) = 2 broadcasts; Phase 4 (synthesized fresh ID for explicit ID-skip exercise) = caught by ID-skip 1/1. Master stays off in prod; no toggle flips.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-06-05 06:41:21 +00:00

804 lines
32 KiB
Python
Raw Blame History

"""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
Reference in a new issue View git blame Copy permalink