diff --git a/app/camera_config.py b/app/camera_config.py index fd12ab6..b5a95f1 100644 --- a/app/camera_config.py +++ b/app/camera_config.py @@ -55,3 +55,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")) diff --git a/app/camera_scraper.py b/app/camera_scraper.py index ad5dc91..6df326e 100644 --- a/app/camera_scraper.py +++ b/app/camera_scraper.py @@ -37,6 +37,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 @@ -114,6 +115,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() @@ -379,6 +389,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 + + def parse_live_streams_geojson(text: str, source_name: str) -> list[dict]: """Parse willytop8/Live-Environment-Streams GeoJSON. @@ -549,6 +636,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: @@ -600,6 +735,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] = [] diff --git a/tests/test_live_layers.py b/tests/test_live_layers.py index ae0f68e..c32aa7d 100644 --- a/tests/test_live_layers.py +++ b/tests/test_live_layers.py @@ -25,7 +25,7 @@ from live_layers import ( _wfigs_params, ) -from camera_scraper import parse_caltrans_json +from camera_scraper import parse_caltrans_json, parse_udot_ibi_page def test_parse_bbox_and_radius_clamps_to_150_nm(): @@ -257,6 +257,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_quantize_bbox_stable_under_jitter(): a = quantize_bbox(*parse_bbox("-78.7912,35.7711,-78.6101,35.9102")) b = quantize_bbox(*parse_bbox("-78.7900,35.7700,-78.6110,35.9090"))