feat(tomtom_incidents): TomTom real-time traffic incidents adapter (v0.9.5)

Fourth CENTRAL_TRAFFIC event adapter. Complements wzdx (federal work zones) and
state_511_atis (state-DOT reports) with TomTom commercial vehicle-telematics
coverage. Polls the Orbis incidentDetails endpoint per metro bbox, emits one
event per incident to central.traffic.incident.<state>. Ships disabled.

central-supervisor + central-gui restart only -- adapter row on the EXISTING
CENTRAL_TRAFFIC stream, so NO archive restart and no new stream/dependency.
Reuses the existing "tomtom" api key.

- Bbox limit refutation: incidentDetails rejects bbox > 10,000 km^2, so coverage
  is per-metro bboxes (Treasure Valley / Boise, 8,601 km^2), NOT statewide. One
  bbox @ 1800s = 1,440 calls/mo = 58% of the 2,500/mo free-tier cap. Expansion
  rows must respect N*(43200/cadence_min) <= 2500.
- category="incident.tomtom_incidents" -> GUI event_type "incident" (shared with
  state_511_atis; cross-source overlap is by design = additive coverage, distinct
  dedup ids + categories, no Central-side cross-source dedup).
- Severity from magnitudeOfDelay (0->1,1->1,2->2,3->3,4->4; 4=closure). Never None.
- geo.geometry carries TomTom's Point/LineString directly (already lon/lat GeoJSON;
  the v0.9.3 framework renders the affected road as a polyline). No decode needed.
- Dedup id <state_code>:tomtom:<tomtom_id> (upstream id stable across polls,
  verified 154/154 over 60s). Inherits the v0.9.1 dedup mixin.
- aiohttp params= URL-encodes the fields{} GraphQL braces (no curl-glob issue);
  key redacted from logs; poll skips cleanly without a key.

Full suite: 809 passed, 1 skipped (central and unprivileged zvx, 3x each).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

tests/fixtures/tomtom_incidents_sample.json

@@ -0,0 +1 @@
+{"incidents":[{"type":"Feature","properties":{"id":"TTI-5df75143-312c-45ab-86c9-2b8e4e49d562-TTR14671884444019001","iconCategory":9,"magnitudeOfDelay":0,"startTime":"2026-04-03T03:27:30Z","endTime":null,"from":"Early Road","to":"Slade Road","length":234.054,"delay":null,"roadNumbers":[],"timeValidity":"present","events":[{"code":701,"description":"Roadworks","iconCategory":9}]},"geometry":{"type":"LineString","coordinates":[[-116.7368214523,43.794556286],[-116.7363762056,43.7939957253],[-116.7360181307,43.7935411016],[-116.7357297933,43.793107893],[-116.7354816889,43.7926894443]]}},{"type":"Feature","properties":{"id":"TTI-5df75143-312c-45ab-86c9-2b8e4e49d562-TTR14712278436123000","iconCategory":8,"magnitudeOfDelay":4,"startTime":"2026-05-16T01:22:00Z","endTime":null,"from":"Wagner Road","to":"Farmway Road / West Ustick Road","length":1483.2713788089,"delay":null,"roadNumbers":[],"timeValidity":"present","events":[{"code":401,"description":"Closed","iconCategory":8}]},"geometry":{"type":"LineString","coordinates":[[-116.7331709659,43.6332509025],[-116.7326506173,43.6332536325],[-116.73007972,43.6332683743],[-116.7281579172,43.6332831161],[-116.7265311574,43.6332965232],[-116.726281712,43.6332991925],[-116.724549005,43.633311265],[-116.7240957117,43.633311265],[-116.7231542563,43.6333139343],[-116.7217729186,43.633331406],[-116.7217608487,43.633331406],[-116.7206691896,43.6333434785],[-116.719480971,43.6333448132],[-116.7194689011,43.6333448132],[-116.7188157832,43.6333461478],[-116.7181224322,43.6333514864],[-116.7181103622,43.6333514864],[-116.7168953216,43.6333595549],[-116.7153798735,43.6333689581],[-116.7147415077,43.6333729621]]}}]}

tests/test_events_feed_frontend.py

@@ -1145,6 +1145,7 @@ _SAMPLE_INNER = {
     "wzdx": {"road_names": ["I-80"], "direction": "eastbound"},
     "state_511_atis": {"layer": "Incidents", "roadway_name": "US-95", "location_description": "Ponderosa Mobile Home Park"},
     "tomtom_flow": {"road_category": "primary", "relative_speed": 0.11},
+    "tomtom_incidents": {"description": "Roadworks", "from": "Early Road", "to": "Slade Road"},
 }
 
 # Exact expected subjects for the deterministic adapters. swpc_alerts is omitted
@@ -1165,6 +1166,7 @@ _EXPECTED_SUBJECT = {
     "wfigs_perimeters": "Wildfire perimeter — Carbon, MT",
     "wzdx": "Work zone on I-80 eastbound",
     "tomtom_flow": "Traffic flow (primary) — 11% of free-flow",
+    "tomtom_incidents": "Roadworks on Early Road → Slade Road",
 }
 
 

tests/test_tomtom_incidents.py

@@ -0,0 +1,140 @@
+"""Tests for the tomtom_incidents adapter (v0.9.5).
+
+Fixture is a real Orbis incidentDetails capture (2 incidents, varied
+magnitudeOfDelay) from the Treasure Valley bbox:
+  tests/fixtures/tomtom_incidents_sample.json
+
+No conftest entry: dedup uses the supervisor-injected cursors.db (inherited
+mixin); polling is stateless.
+"""
+
+import json
+from datetime import datetime, timezone
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from central.adapter import SourceAdapter
+from central.adapters.tomtom_incidents import (
+    BBox,
+    TomTomIncidentsAdapter,
+    _first_vertex,
+    _MAGNITUDE_SEVERITY,
+)
+from central.config_models import AdapterConfig
+
+INC = json.loads((Path(__file__).parent / "fixtures" / "tomtom_incidents_sample.json").read_text())["incidents"]
+BB = BBox(name="treasure_valley", min_lon=-116.85, min_lat=43.30, max_lon=-115.65, max_lat=44.10, state_code="ID")
+
+
+def _cfg():
+    return AdapterConfig(
+        name="tomtom_incidents", enabled=True, cadence_s=1800,
+        settings={"api_key_alias": "tomtom", "bboxes": [BB.model_dump()]},
+        updated_at=datetime.now(timezone.utc),
+    )
+
+
+@pytest.fixture
+def adapter(tmp_path):
+    return TomTomIncidentsAdapter(_cfg(), MagicMock(), tmp_path / "cursors.db")
+
+
+@pytest.mark.parametrize("mag,sev", [(0, 1), (1, 1), (2, 2), (3, 3), (4, 4), (None, 1), (99, 1)])
+def test_severity_mapping(mag, sev):
+    assert _MAGNITUDE_SEVERITY.get(mag, 1) == sev
+
+
+def test_dedup_key(adapter):
+    e = adapter._build_event(INC[0], BB)
+    assert e.id == f"ID:tomtom:{INC[0]['properties']['id']}"
+
+
+def test_build_event_linestring(adapter):
+    e = adapter._build_event(INC[0], BB)  # mag-0 Roadworks LineString
+    assert e.category == "incident.tomtom_incidents"
+    assert e.severity == 1
+    assert e.data["description"] == "Roadworks"
+    assert e.data["from"] == "Early Road" and e.data["to"] == "Slade Road"
+    assert e.data["state_code"] == "ID"
+    assert e.data["latitude"] is not None and e.data["longitude"] is not None
+
+
+def test_build_event_closure_severity(adapter):
+    e = adapter._build_event(INC[1], BB)  # mag-4 Closed
+    assert e.data["magnitude_of_delay"] == 4
+    assert e.severity == 4
+
+
+def test_geo_geometry_for_linestring(adapter):
+    # v0.9.3 framework: the affected-road LineString rides on geo.geometry.
+    e = adapter._build_event(INC[0], BB)
+    assert e.geo.geometry["type"] == "LineString"
+    assert e.geo.geometry["coordinates"] == INC[0]["geometry"]["coordinates"]
+
+
+def test_build_event_point():
+    a = TomTomIncidentsAdapter(_cfg(), MagicMock(), Path("/tmp/unused.db"))
+    inc = {"geometry": {"type": "Point", "coordinates": [-116.2, 43.6]},
+           "properties": {"id": "TTI-x", "magnitudeOfDelay": 2,
+                          "events": [{"description": "Accident", "code": 1}]}}
+    e = a._build_event(inc, BB)
+    assert e.geo.geometry["type"] == "Point"
+    assert e.severity == 2
+    assert e.data["latitude"] == 43.6 and e.data["longitude"] == -116.2
+
+
+def test_first_vertex():
+    assert _first_vertex({"type": "Point", "coordinates": [-116.2, 43.6]}) == (43.6, -116.2)
+    assert _first_vertex({"type": "LineString", "coordinates": [[-116.2, 43.6], [-116.1, 43.7]]}) == (43.6, -116.2)
+    assert _first_vertex(None) == (None, None)
+    assert _first_vertex({"type": "Polygon", "coordinates": []}) == (None, None)
+
+
+def test_subject_for_idaho(adapter):
+    e = adapter._build_event(INC[0], BB)
+    assert adapter.subject_for(e) == "central.traffic.incident.id"
+
+
+def test_subject_unknown(adapter):
+    e = adapter._build_event(INC[0], BBox(name="x", min_lon=0, min_lat=0, max_lon=1, max_lat=1, state_code=""))
+    assert adapter.subject_for(e) == "central.traffic.incident.unknown"
+
+
+@pytest.mark.asyncio
+async def test_poll_yields_events(tmp_path):
+    cs = MagicMock()
+    cs.get_api_key = AsyncMock(return_value="testkey")
+    adapter = TomTomIncidentsAdapter(_cfg(), cs, tmp_path / "cursors.db")
+    await adapter.startup()
+    adapter._fetch_bbox = AsyncMock(return_value=INC)  # bypass retry + network
+    events = [e async for e in adapter.poll()]
+    await adapter.shutdown()
+    assert len(events) == 2
+    assert all(e.adapter == "tomtom_incidents" for e in events)
+    assert all(e.category == "incident.tomtom_incidents" for e in events)
+
+
+@pytest.mark.asyncio
+async def test_poll_skips_without_key(tmp_path):
+    cs = MagicMock()
+    cs.get_api_key = AsyncMock(return_value=None)
+    adapter = TomTomIncidentsAdapter(_cfg(), cs, tmp_path / "cursors.db")
+    await adapter.startup()
+    events = [e async for e in adapter.poll()]
+    await adapter.shutdown()
+    assert events == []
+
+
+def test_summary_partial_renders():
+    from central.gui.routes import _derive_subject
+    inner = {"description": "Roadworks", "from": "Early Road", "to": "Slade Road"}
+    row = {"adapter": "tomtom_incidents", "data": {"data": {"data": inner}}}
+    assert _derive_subject(row) == "Roadworks on Early Road → Slade Road"
+
+
+def test_inherits_dedup_mixin():
+    for m in ("is_published", "mark_published", "sweep_old_ids"):
+        assert m not in TomTomIncidentsAdapter.__dict__, f"redefines {m}"
+        assert getattr(TomTomIncidentsAdapter, m) is getattr(SourceAdapter, m)