Merge pull request 'feat(cameras): UDOT IBI 511 traffic cameras -> cameras table' (#38) from osint-dashboard/t_5bfc52e8-osint-udot-ibi-511-cameras-cameras-table into master

Reviewed-on: #38

app/camera_config.py

@@ -61,3 +61,15 @@ SNAPSHOT_TIMEOUT = float(os.getenv("SNAPSHOT_TIMEOUT", "8.0"))
 
 # NATS subject cameras are published on (consumed by the shared ingester).
 CAMERA_NATS_SUBJECT = os.getenv("CAMERA_NATS_SUBJECT", "events.camera")
+
+
+# ── UDOT IBI 511 traffic cameras ──────────────────────────────────────────
+# DataTables endpoint (POST form-encoded; server caps at 100 rows/page no
+# matter what `length` is sent). No API key. Snapshot stills live at a stable
+# /map/Cctv/{id} URL — same URL always serves the latest frame, so we store
+# the URL and never scrape every frame ourselves.
+UDOT_IBI_URL = "https://prod-ut.ibi511.com/List/GetData/Cameras"
+UDOT_IBI_BASE = "https://prod-ut.ibi511.com"
+UDOT_IBI_PAGE_SIZE = 100
+# Safety cap on pages per cycle so a runaway recordsTotal cannot fan out.
+UDOT_IBI_MAX_PAGES = int(os.getenv("UDOT_IBI_MAX_PAGES", "40"))

app/camera_scraper.py

@@ -38,6 +38,7 @@ from camera_config import (
     CAMERA_SOURCE_URLS, CAMERA_REQUEST_DELAY, CAMERA_MAX_PER_SOURCE,
     NOMINATIM_URL, NOMINATIM_MIN_INTERVAL, USER_AGENT,
     SNAPSHOT_CACHE_DIR, SNAPSHOT_TTL_SECONDS, SNAPSHOT_TIMEOUT,
+    UDOT_IBI_URL, UDOT_IBI_BASE, UDOT_IBI_PAGE_SIZE, UDOT_IBI_MAX_PAGES,
 )
 from camera_models import cameras
 from database import async_session
@@ -115,6 +116,15 @@ class RateLimitedClient:
         self._last[host] = time.monotonic()
         return await self.client.get(url, **kw)
 
+    async def post(self, url: str, **kw) -> httpx.Response:
+        host = urlparse(url).netloc
+        now = time.monotonic()
+        wait = self._last.get(host, 0.0) + self._delay - now
+        if wait > 0:
+            await asyncio.sleep(wait)
+        self._last[host] = time.monotonic()
+        return await self.client.post(url, **kw)
+
     async def aclose(self):
         await self.client.aclose()
 
@@ -380,6 +390,83 @@ def parse_caltrans_json(text: str, source_name: str) -> list[dict]:
     return out
 
 
+# ── UDOT IBI 511 ──────────────────────────────────────────────────────────
+# Utah bbox (lat 36.9–42.1, lon -114.2–-108.9). WKT is `POINT (lng lat)`.
+_UDOT_IBI_MIN_LAT, _UDOT_IBI_MAX_LAT = 36.9, 42.1
+_UDOT_IBI_MIN_LON, _UDOT_IBI_MAX_LON = -114.2, -108.9
+_UDOT_WKT_POINT_RE = re.compile(
+    r"POINT\s*\(\s*(-?\d+(?:\.\d+)?)\s+(-?\d+(?:\.\d+)?)\s*\)", re.I,
+)
+
+
+def parse_udot_ibi_page(text: str, source_name: str = "udot") -> list[dict]:
+    """Parse one UDOT IBI 511 DataTables camera page (`{"data": [...]}`).
+
+    Skips rows whose first image is `blocked` or `disabled`, and drops any
+    point outside the Utah bbox. The `/map/Cctv/{id}` URL is a stable identity
+    (always serves the latest frame), so it is stored as both source_url and
+    snapshot_url — we never scrape frames ourselves.
+    """
+    try:
+        payload = json.loads(text)
+    except (json.JSONDecodeError, ValueError):
+        return []
+    rows = payload.get("data") if isinstance(payload, dict) else None
+    if not isinstance(rows, list):
+        return []
+    out: list[dict] = []
+    for row in rows:
+        if not isinstance(row, dict):
+            continue
+        cam_id = row.get("id")
+        images = row.get("images") or []
+        if cam_id is None or not images:
+            continue
+        img = images[0] or {}
+        if img.get("blocked") or img.get("disabled"):
+            continue
+        lon = lat = None
+        try:
+            wkt = (row.get("latLng") or {}).get("geography") or {}
+            wkt = wkt.get("wellKnownText") or ""
+            m = _UDOT_WKT_POINT_RE.match(str(wkt).strip())
+            if m:
+                lon, lat = float(m.group(1)), float(m.group(2))
+        except (AttributeError, TypeError, ValueError):
+            lon = lat = None
+        if lat is None or lon is None:
+            continue
+        if not (_UDOT_IBI_MIN_LAT <= lat <= _UDOT_IBI_MAX_LAT
+                and _UDOT_IBI_MIN_LON <= lon <= _UDOT_IBI_MAX_LON):
+            continue
+        snap = f"{UDOT_IBI_BASE}/map/Cctv/{cam_id}"
+        roadway, direction, location = (
+            row.get("roadway"), row.get("direction"), row.get("location"),
+        )
+        name = ", ".join(
+            str(b) for b in (roadway, direction, location)
+            if b and str(b).strip() and str(b).strip().lower() != "unknown"
+        ) or None
+        out.append({
+            "source_url": snap,
+            "snapshot_url": snap,
+            "discovery_source": source_name,
+            "location_lat": lat,
+            "location_lon": lon,
+            "location_name": name,
+            "vendor": "UDOT",
+            "device_type": "http",
+            "raw": {
+                "udot_id": cam_id,
+                "agency": row.get("source"),
+                "source_id": row.get("sourceId"),
+                "roadway": roadway,
+                "direction": direction,
+            },
+        })
+    return out
+
+
 # Oregon DOT TripCheck inventory bounding box (approx state extent).
 ODOT_BBOX = (41.9, 46.3, -124.6, -116.4)  # lat_min, lat_max, lon_min, lon_max
 
@@ -672,6 +759,54 @@ async def scrape_source(client: RateLimitedClient, geo: Geocoder,
     return out
 
 
+# ── UDOT IBI 511 paginated fetcher ────────────────────────────────────────
+
+async def scrape_udot_ibi(client: RateLimitedClient) -> list[dict]:
+    """Page through the UDOT IBI 511 DataTables endpoint and normalize.
+
+    POSTs `start`/`length` form fields (server caps at 100 rows/page), walking
+    pages until `recordsTotal` is exhausted or UDOT_IBI_MAX_PAGES is hit.
+    """
+    out: list[dict] = []
+    seen: set[str] = set()
+    start = 0
+    for _ in range(UDOT_IBI_MAX_PAGES):
+        try:
+            resp = await client.post(
+                UDOT_IBI_URL,
+                data={
+                    "start": str(start),
+                    "length": str(UDOT_IBI_PAGE_SIZE),
+                    "lang": "en-US",
+                },
+                headers={"X-Requested-With": "XMLHttpRequest"},
+            )
+            resp.raise_for_status()
+            body = resp.text
+        except Exception:  # noqa: BLE001
+            logger.exception("failed to fetch UDOT IBI page start=%d", start)
+            break
+        try:
+            payload = json.loads(body)
+        except ValueError:
+            logger.warning("UDOT IBI non-JSON response at start=%d", start)
+            break
+        total = int(payload.get("recordsTotal") or 0)
+        rows = payload.get("data") or []
+        if not isinstance(rows, list) or not rows:
+            break
+        for cam in parse_udot_ibi_page(body, "udot"):
+            if cam["source_url"] in seen:
+                continue
+            seen.add(cam["source_url"])
+            out.append(cam)
+        if start + len(rows) >= total:
+            break
+        start += len(rows)
+    logger.info("UDOT IBI yielded %d cameras", len(out))
+    return out
+
+
 # ── Persistence ────────────────────────────────────────────────────────────
 
 async def upsert_cameras(cams: list[dict]) -> int:
@@ -723,6 +858,7 @@ async def run_cycle() -> int:
     try:
         results = await asyncio.gather(
             *(scrape_source(client, geo, s) for s in CAMERA_SOURCE_URLS),
+            scrape_udot_ibi(client),
             return_exceptions=True,
         )
         all_cams: list[dict] = []

tests/test_live_layers.py

@@ -27,7 +27,7 @@ from live_layers import (
     _wfigs_params,
 )
 
-from camera_scraper import parse_caltrans_json, parse_odot_json, parse_mdot_json
+from camera_scraper import parse_caltrans_json, parse_udot_ibi_page, parse_odot_json, parse_mdot_json
 
 
 def test_parse_bbox_and_radius_clamps_to_150_nm():
@@ -259,6 +259,83 @@ def test_parse_caltrans_skips_oos_and_maps_jpeg_hls():
     assert "rtsp://" not in cam["snapshot_url"].lower()
 
 
+# ── UDOT IBI 511 parser ──────────────────────────────────────────────────
+
+def _udot_row(cam_id, lng, lat, **img_overrides):
+    img = {
+        "id": cam_id, "cameraSiteId": cam_id,
+        "imageUrl": f"/map/Cctv/{cam_id}", "disabled": False, "blocked": False,
+    }
+    img.update(img_overrides)
+    return {
+        "id": cam_id, "sourceId": "102771", "source": "ADX",
+        "roadway": "Unknown", "direction": "Unknown",
+        "location": "Freedom Blvd / 200 W @ 1100 N, PVO",
+        "latLng": {"geography": {
+            "coordinateSystemId": 4326,
+            "wellKnownText": f"POINT ({lng} {lat})"}},
+        "images": [img],
+    }
+
+
+def _udot_page(rows):
+    import json
+    return json.dumps({"draw": 0, "recordsTotal": len(rows),
+                       "recordsFiltered": len(rows), "data": rows})
+
+
+def test_parse_udot_wkt_maps_lng_lat():
+    cams = parse_udot_ibi_page(_udot_page([_udot_row(112731, -111.66204, 40.24863)]))
+    assert len(cams) == 1
+    cam = cams[0]
+    # WKT is `POINT (lng lat)` — order must not be swapped.
+    assert cam["location_lat"] == 40.24863
+    assert cam["location_lon"] == -111.66204
+    assert cam["discovery_source"] == "udot"
+    assert cam["vendor"] == "UDOT"
+    assert cam["source_url"] == "https://prod-ut.ibi511.com/map/Cctv/112731"
+    assert cam["snapshot_url"] == cam["source_url"]
+    assert "rtsp://" not in cam["source_url"].lower()
+    assert cam["raw"]["udot_id"] == 112731
+
+
+def test_parse_udot_skips_blocked_and_disabled():
+    rows = [
+        _udot_row(1, -111.0, 40.0),
+        _udot_row(2, -111.1, 40.1, blocked=True),
+        _udot_row(3, -111.2, 40.2, disabled=True),
+    ]
+    rows.append(_udot_row(4, -111.3, 40.3))
+    rows[3]["images"] = []  # no images → drop
+    cams = parse_udot_ibi_page(_udot_page(rows))
+    assert [c["raw"]["udot_id"] for c in cams] == [1]
+
+
+def test_parse_udot_drops_out_of_bbox():
+    rows = [
+        _udot_row(1, -111.0, 40.0),   # inside Utah
+        _udot_row(2, -100.0, 40.0),   # east of -108.9
+        _udot_row(3, -120.0, 40.0),   # west of -114.2
+        _udot_row(4, -111.0, 44.0),   # north of 42.1
+        _udot_row(5, -111.0, 30.0),   # south of 36.9
+    ]
+    cams = parse_udot_ibi_page(_udot_page(rows))
+    assert [c["raw"]["udot_id"] for c in cams] == [1]
+
+
+def test_parse_udot_bad_payload_returns_empty():
+    import json
+    assert parse_udot_ibi_page("not json") == []
+    assert parse_udot_ibi_page(json.dumps({"data": None})) == []
+    assert parse_udot_ibi_page(json.dumps({"data": "nope"})) == []
+
+
+def test_parse_udot_missing_wkt_skipped():
+    row = _udot_row(1, -111.0, 40.0)
+    row["latLng"] = {}
+    assert parse_udot_ibi_page(_udot_page([row])) == []
+
+
 def test_parse_odot_tripcheck_keeps_valid_skips_missing_and_oob():
     payload = """
     {"features":[