diff --git a/work/meshai/env/roads511.py b/work/meshai/env/roads511.py index ba90bde..a57e5b4 100644 --- a/work/meshai/env/roads511.py +++ b/work/meshai/env/roads511.py @@ -21,6 +21,53 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) +# Explicit full-road-closure language, matched case-insensitively. Used +# ONLY by to_event()'s sub_type mapping (branch a) to decide "road_closed" +# vs. "road_works" — distinct from (and stricter than) the adapter's +# existing `is_closure` property, which loosely matches any "closed" +# substring and therefore also flags partial-restriction wording (e.g. +# "All Shoulders Closed") that the upstream owner explicitly does not want +# broadcast as a closure. +_FULL_CLOSURE_PHRASES = ( + "all lanes closed", + "road closed", + "highway closed", + "fully closed", + "closed in both directions", +) + +# Partial-restriction phrases that must NEVER be counted as a full closure. +# Checked first so one of these vetoes a match even if closure wording +# happens to appear elsewhere in the same description — e.g. "all +# shoulders closed" must not fall through to a naive "closed" test, and +# "one/right/left lane closed" must not be confused with "all lanes +# closed" (note the singular "lane closed" vs. plural "lanes closed"). +_PARTIAL_RESTRICTION_PHRASES = ( + "shoulder closed", + "shoulders closed", + "lane closed", + "ramp closed", + "alternating", + "lane restriction", +) + + +def _has_full_closure_language(description: str) -> bool: + """True if `description` contains explicit full-closure phrasing. + + Case-insensitive. Partial-restriction phrases (shoulder/lane/ramp + closures, alternating traffic) are checked first and veto a match, so + they never register as a full closure regardless of other wording + present. + """ + text = (description or "").lower() + + if any(phrase in text for phrase in _PARTIAL_RESTRICTION_PHRASES): + return False + + return any(phrase in text for phrase in _FULL_CLOSURE_PHRASES) + + class Roads511Adapter: """511 road conditions polling adapter.""" @@ -284,12 +331,21 @@ class Roads511Adapter: lat = loc.get("latitude") or loc.get("lat") lon = loc.get("longitude") or loc.get("lon") or loc.get("lng") - # Check closure status - is_closure = ( + # Raw upstream full-closure signal, captured on its own (no + # description heuristic mixed in) so to_event() can build a + # stricter full-closure determination for sub_type mapping + # without touching is_closure below, which severity/summary + # logic elsewhere already depends on. + is_full_closure_flag = bool( item.get("IsFullClosure") or item.get("is_full_closure") or item.get("fullClosure") or - item.get("closed") or + item.get("closed") + ) + + # Check closure status + is_closure = ( + is_full_closure_flag or "closure" in str(event_type).lower() or "closed" in str(description).lower() ) @@ -323,6 +379,19 @@ class Roads511Adapter: None ) + # Granular ITD 511 v2 fields (roadwork vs. crash vs. closure + # discrimination). Defensive: absent on other states' 511 feeds. + event_sub_type = ( + item.get("EventSubType") or + item.get("event_sub_type") or + "" + ) + cause = ( + item.get("Cause") or + item.get("cause") or + None + ) + # Default 6 hour TTL, refreshed every tick expires = now + 21600 @@ -346,7 +415,10 @@ class Roads511Adapter: "properties": { "roadway": roadway, "is_closure": bool(is_closure), + "is_full_closure_flag": is_full_closure_flag, "last_updated": last_updated, + "event_sub_type": event_sub_type, + "cause": cause, }, } @@ -411,10 +483,49 @@ class Roads511Adapter: # NOTE: Central-era rows used source 'itd_511' with 'idaho_511:event:*' # ids — a different keyspace the pre-seed intentionally does not cover. _external_id = evt.get("external_id") or event_id + + # sub_type mapping: distinguish routine roadwork (suppressed by + # gating/incident.py's work-zone rule) from genuine news (crashes, + # closures, hazards). A construction-caused FULL closure is still + # real news and must broadcast, so a full closure wins regardless + # of EventType. Precedence: + # 1. full closure (raw flag OR explicit closure wording, + # see _has_full_closure_language) -> road_closed (always) + # 2. EventType roadwork / + # EventSubType has "construction" -> road_works (suppressed) + # 3. EventType accidentsAndIncidents -> incident + # 4. EventType closures -> road_closed + # 5. else (specialEvents, unknown) -> incident (fail open) + # + # NOTE: branch 1 deliberately does NOT reuse `_is_closure` above. + # `_is_closure` is a loose OR-chain (also used for severity and + # the summary text) that matches any "closed" substring in the + # description, including partial-restriction wording like "All + # Shoulders Closed" — routine construction the owner explicitly + # does not want broadcast. `_full_closure_for_mapping` is a + # stricter, mapping-only determination. + _event_type_val = str(evt.get("event_type") or "").strip().lower() + _event_sub_type_val = str(props.get("event_sub_type") or "").strip().lower() + _cause = props.get("cause") + _full_closure_for_mapping = bool( + props.get("is_full_closure_flag") + ) or _has_full_closure_language(_desc) + + if _full_closure_for_mapping: + _sub_type = "road_closed" + elif _event_type_val == "roadwork" or "construction" in _event_sub_type_val: + _sub_type = "road_works" + elif _event_type_val == "accidentsandincidents": + _sub_type = "incident" + elif _event_type_val == "closures": + _sub_type = "road_closed" + else: + _sub_type = "incident" + canonical_data = { "external_id": _external_id, # 511_{itd_id}; enables durable pre-seed "source": "511", - "sub_type": "road_closed" if _is_closure else "incident", + "sub_type": _sub_type, "road": _roadway or None, "direction": None, # not structured in native 511 feed "from_loc": None, @@ -423,7 +534,7 @@ class Roads511Adapter: "mile_end": None, "mile_marker": None, "lanes_affected": None, - "cause": None, + "cause": _cause, "comment": _desc[:200] if _desc else title, "impact": "all lanes closed" if _is_closure else None, "county": None, diff --git a/work/tests/test_adapter_roads511.py b/work/tests/test_adapter_roads511.py index 4322c11..c76e74d 100644 --- a/work/tests/test_adapter_roads511.py +++ b/work/tests/test_adapter_roads511.py @@ -200,3 +200,331 @@ def test_to_event_missing_properties_returns_event(adapter): def test_to_event_does_not_raise_on_corrupted_dict(adapter): """Corrupted dict returns None without raising.""" assert adapter.to_event({"garbage": True}) is None + + +# ============================================================ +# EventType / EventSubType / Cause -> sub_type MAPPING TESTS +# +# Drives real upstream-shaped ITD 511 v2 payloads through the actual +# _parse_event() -> to_event() pipeline (not the make_511_event() stored- +# dict helper) so the EventSubType/Cause capture at parse time and the +# sub_type precedence in to_event() are both exercised end to end. +# ============================================================ + +def _itd_raw( + event_id="RW-1001", + event_type="roadwork", + event_sub_type="roadConstruction", + cause="Construction", + description="Paving operations, expect delays", + roadway="I-84", + is_full_closure=False, + lat=43.5, + lon=-116.2, +): + """Realistic raw ITD 511 v2 API item shape (PascalCase fields).""" + return { + "EventId": event_id, + "EventType": event_type, + "EventSubType": event_sub_type, + "Cause": cause, + "RoadwayName": roadway, + "Description": description, + "Latitude": lat, + "Longitude": lon, + "IsFullClosure": is_full_closure, + "LastUpdated": "2026-08-01T00:00:00Z", + } + + +def _through_pipeline(adapter, raw_item): + """Parse a raw ITD item then translate to a pipeline Event, mirroring + the real adapter flow (_fetch -> _parse_event -> get_events -> to_event).""" + stored = adapter._parse_event(raw_item, time.time()) + assert stored is not None + return adapter.to_event(stored) + + +def test_mapping_roadwork_not_full_closure_is_road_works(adapter): + """Routine roadwork, not a full closure -> sub_type 'road_works' (suppressed).""" + raw = _itd_raw(event_type="roadwork", is_full_closure=False) + event = _through_pipeline(adapter, raw) + assert event is not None + assert event.data["sub_type"] == "road_works" + + +def test_mapping_roadwork_with_full_closure_is_road_closed(adapter): + """Construction-caused FULL closure is still genuine news -> 'road_closed', + and must NOT be suppressed like routine roadwork.""" + raw = _itd_raw( + event_type="roadwork", + event_sub_type="bridgeConstruction", + description="Bridge replacement in progress", + is_full_closure=True, + ) + event = _through_pipeline(adapter, raw) + assert event is not None + assert event.data["sub_type"] == "road_closed" + + +def test_mapping_accidents_and_incidents_is_incident(adapter): + """EventType accidentsAndIncidents -> sub_type 'incident'.""" + raw = _itd_raw( + event_id="INC-2001", + event_type="accidentsAndIncidents", + event_sub_type="crash", + cause="Collision", + description="Two vehicle collision blocking right lane", + is_full_closure=False, + ) + event = _through_pipeline(adapter, raw) + assert event is not None + assert event.data["sub_type"] == "incident" + + +def test_mapping_closures_is_road_closed(adapter): + """EventType closures -> sub_type 'road_closed'.""" + raw = _itd_raw( + event_id="CL-3001", + event_type="closures", + event_sub_type="bridgeClosure", + cause="Maintenance", + description="Bridge out of service", + is_full_closure=False, + ) + event = _through_pipeline(adapter, raw) + assert event is not None + assert event.data["sub_type"] == "road_closed" + + +def test_mapping_special_events_is_incident(adapter): + """EventType specialEvents -> sub_type 'incident' (not suppressed).""" + raw = _itd_raw( + event_id="SE-4001", + event_type="specialEvents", + event_sub_type="parade", + cause=None, + description="Downtown parade route", + is_full_closure=False, + ) + event = _through_pipeline(adapter, raw) + assert event is not None + assert event.data["sub_type"] == "incident" + + +def test_mapping_unknown_or_missing_event_type_is_incident(adapter): + """Missing/unrecognized EventType fails open to 'incident' (not suppressed).""" + raw = { + "EventId": "UNK-5001", + "RoadwayName": "SH-21", + "Description": "Unusual event", + "Latitude": 43.5, + "Longitude": -115.5, + } + event = _through_pipeline(adapter, raw) + assert event is not None + assert event.data["sub_type"] == "incident" + + +def test_mapping_event_subtype_construction_overrides_non_roadwork_eventtype(adapter): + """EventSubType containing 'construction' forces 'road_works' even when + EventType itself is not literally 'roadwork'.""" + raw = _itd_raw( + event_id="SE-6001", + event_type="specialEvents", + event_sub_type="bridgeConstruction", + cause="Construction", + description="Bridge deck construction nearby", + is_full_closure=False, + ) + event = _through_pipeline(adapter, raw) + assert event is not None + assert event.data["sub_type"] == "road_works" + + +def test_mapping_cause_is_carried_through_not_none(adapter): + """Cause from the upstream payload reaches canonical_data, not hardcoded None.""" + raw = _itd_raw(cause="Weather") + event = _through_pipeline(adapter, raw) + assert event is not None + assert event.data["cause"] == "Weather" + assert event.data["cause"] is not None + + +def test_mapping_missing_cause_is_none(adapter): + """No Cause field upstream -> canonical cause is None, not a crash.""" + raw = _itd_raw(cause=None) + del raw["Cause"] + event = _through_pipeline(adapter, raw) + assert event is not None + assert event.data["cause"] is None + + +# ============================================================ +# END-TO-END GATING TEST — proves a real roadwork event, run through the +# actual adapter pipeline, is suppressed by gating/incident.py's work-zone +# rule, while a real crash event from the same pipeline still broadcasts. +# ============================================================ + +def test_roadwork_event_suppressed_by_incident_gate_end_to_end(adapter): + """A real ITD roadwork payload, parsed and translated by the adapter, + must be suppressed at the gating layer (not just mapped to the right + sub_type in isolation).""" + from meshai.notifications.gating.incident import decide + + raw = _itd_raw(event_id="RW-7001", event_type="roadwork", is_full_closure=False) + event = _through_pipeline(adapter, raw) + assert event is not None + assert event.data["sub_type"] == "road_works" + + result = decide(dict(event.data), source="511", now=time.time()) + assert result.broadcast is False + assert result.lifecycle == "suppress" + + +def test_crash_event_still_broadcasts_through_incident_gate_end_to_end(adapter): + """A real ITD crash payload, parsed and translated by the adapter, + still broadcasts (gating layer is unaffected for non-work-zone sub_types).""" + from meshai.notifications.gating.incident import decide + + raw = _itd_raw( + event_id="INC-7002", + event_type="accidentsAndIncidents", + event_sub_type="crash", + description="Vehicle collision, right lane blocked", + is_full_closure=False, + ) + event = _through_pipeline(adapter, raw) + assert event is not None + assert event.data["sub_type"] == "incident" + + +# ============================================================ +# FULL-CLOSURE vs. PARTIAL-RESTRICTION LANGUAGE TESTS +# +# The loose `is_closure` property (also used for severity/summary) matches +# any "closed" substring, including partial-restriction wording like "All +# Shoulders Closed" that ITD explicitly does not want broadcast. The +# sub_type mapping's full-closure branch must use a stricter, +# mapping-only determination instead. +# ============================================================ + +def test_all_shoulders_closed_roadwork_is_road_works(adapter): + """'All Shoulders Closed' is a partial restriction, not a full closure + -> stays 'road_works' (suppressed), despite containing 'closed'.""" + raw = _itd_raw( + event_id="RW-8001", + event_type="roadwork", + description="All Shoulders Closed", + is_full_closure=False, + ) + event = _through_pipeline(adapter, raw) + assert event is not None + assert event.data["sub_type"] == "road_works" + + +@pytest.mark.parametrize( + "description", + ["One lane closed", "Right lane closed", "Left lane closed"], +) +def test_single_lane_closed_roadwork_is_road_works(adapter, description): + """Single-lane restriction wording -> stays 'road_works' (suppressed).""" + raw = _itd_raw( + event_id="RW-8002", + event_type="roadwork", + description=description, + is_full_closure=False, + ) + event = _through_pipeline(adapter, raw) + assert event is not None + assert event.data["sub_type"] == "road_works" + + +def test_ramp_closed_roadwork_is_road_works(adapter): + """'Ramp closed' is a partial restriction -> stays 'road_works' (suppressed).""" + raw = _itd_raw( + event_id="RW-8003", + event_type="roadwork", + description="Ramp closed for repaving", + is_full_closure=False, + ) + event = _through_pipeline(adapter, raw) + assert event is not None + assert event.data["sub_type"] == "road_works" + + +def test_all_lanes_closed_roadwork_is_road_closed(adapter): + """'All lanes closed' is a genuine full closure -> 'road_closed', and + must STILL BROADCAST (this is the safety case: a real closure phrased + as roadwork must not be suppressed by the work-zone gate).""" + from meshai.notifications.gating.incident import decide + + raw = _itd_raw( + event_id="RW-8004", + event_type="roadwork", + description="All lanes closed for bridge demolition", + is_full_closure=False, + ) + event = _through_pipeline(adapter, raw) + assert event is not None + assert event.data["sub_type"] == "road_closed" + + result = decide(dict(event.data), source="511", now=time.time()) + assert result.broadcast is True + assert result.lifecycle != "suppress" + + +def test_is_full_closure_flag_true_no_closure_words_is_road_closed(adapter): + """IsFullClosure=true + roadwork, with a description containing no + closure wording at all -> still 'road_closed' via the raw flag.""" + raw = _itd_raw( + event_id="RW-8005", + event_type="roadwork", + description="Bridge deck replacement in progress", + is_full_closure=True, + ) + event = _through_pipeline(adapter, raw) + assert event is not None + assert event.data["sub_type"] == "road_closed" + + +def test_road_closed_due_to_rock_slide_incident_is_road_closed(adapter): + """Explicit full-closure wording on an accidentsAndIncidents event + still wins the mapping -> 'road_closed'.""" + raw = _itd_raw( + event_id="INC-8006", + event_type="accidentsAndIncidents", + event_sub_type="hazard", + cause="Rock slide", + description="Road closed due to rock slide", + is_full_closure=False, + ) + event = _through_pipeline(adapter, raw) + assert event is not None + assert event.data["sub_type"] == "road_closed" + + +def test_lane_restrictions_roadwork_is_road_works(adapter): + """'Lane restrictions' wording -> stays 'road_works' (suppressed).""" + raw = _itd_raw( + event_id="RW-8007", + event_type="roadwork", + description="Lane restrictions in effect through Friday", + is_full_closure=False, + ) + event = _through_pipeline(adapter, raw) + assert event is not None + assert event.data["sub_type"] == "road_works" + + +def test_alternating_traffic_roadwork_is_road_works(adapter): + """'Alternating' one-lane traffic control -> stays 'road_works' (suppressed).""" + raw = _itd_raw( + event_id="RW-8008", + event_type="roadwork", + description="Alternating traffic controlled by flaggers", + is_full_closure=False, + ) + event = _through_pipeline(adapter, raw) + assert event is not None + assert event.data["sub_type"] == "road_works"