feat(cameras): ODOT TripCheck parser -> cameras table

Add parse_odot_json (Esri-style cctvinventory.js) next to parse_caltrans_json:
JPEG snapshot_url, discovery_source=odot, Oregon bbox + filename guard,
no RTSP. Wire dispatch + default CAMERA_SOURCE_URLS entry. Unit tests.
This commit is contained in:
Sirius DevOps 2026-08-31 21:37:32 -04:00
parent 47c726d68d
commit 41433be574
3 changed files with 95 additions and 1 deletions

View file

@ -25,6 +25,8 @@ _DEFAULT_SOURCE_URL = ",".join((
"https://raw.githubusercontent.com/willytop8/Live-Environment-Streams/main/streams.geojson", "https://raw.githubusercontent.com/willytop8/Live-Environment-Streams/main/streams.geojson",
# Official Caltrans CWWP2 JPEG + HLS CCTV (districts 112). # Official Caltrans CWWP2 JPEG + HLS CCTV (districts 112).
*CALTRANS_CCTV_URLS, *CALTRANS_CCTV_URLS,
# Oregon DOT TripCheck public CCTV JPEG inventory (Esri JSON).
"https://www.tripcheck.com/Scripts/map/data/cctvinventory.js",
)) ))
CAMERA_SOURCE_URLS = [ CAMERA_SOURCE_URLS = [
u.strip() u.strip()

View file

@ -25,6 +25,7 @@ import hashlib
import ipaddress import ipaddress
import json import json
import logging import logging
import math
import re import re
import time import time
from datetime import datetime, timezone from datetime import datetime, timezone
@ -379,6 +380,51 @@ def parse_caltrans_json(text: str, source_name: str) -> list[dict]:
return out 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
def parse_odot_json(text: str, source_name: str) -> list[dict]:
"""Parse Oregon DOT TripCheck cctvinventory Esri-style JSON.
Store the JPEG still as snapshot_url (map thumbs); never RTSP. Keep only
rows with finite coordinates inside Oregon and a usable filename.
"""
try:
payload = json.loads(text)
except (json.JSONDecodeError, ValueError):
return []
lat_min, lat_max, lon_min, lon_max = ODOT_BBOX
out: list[dict] = []
for feat in payload.get("features") or []:
attrs = (feat or {}).get("attributes") or {}
filename = (attrs.get("filename") or "").strip()
if not filename:
continue
try:
lat = float(attrs.get("latitude"))
lon = float(attrs.get("longitude"))
except (TypeError, ValueError):
continue
if not (math.isfinite(lat) and math.isfinite(lon)):
continue
if not (lat_min <= lat <= lat_max and lon_min <= lon <= lon_max):
continue
jpeg = f"https://tripcheck.com/RoadCams/cams/{filename}"
title = (attrs.get("title") or "").strip()
out.append({
"source_url": jpeg,
"snapshot_url": jpeg,
"discovery_source": "odot",
"location_lat": lat,
"location_lon": lon,
"location_name": title or None,
"vendor": "ODOT",
"device_type": "http",
})
return out
def parse_live_streams_geojson(text: str, source_name: str) -> list[dict]: def parse_live_streams_geojson(text: str, source_name: str) -> list[dict]:
"""Parse willytop8/Live-Environment-Streams GeoJSON. """Parse willytop8/Live-Environment-Streams GeoJSON.
@ -490,6 +536,8 @@ async def scrape_source(client: RateLimitedClient, geo: Geocoder,
body = resp.text body = resp.text
if "cwwp2.dot.ca.gov" in src_url or "cctvStatus" in src_url: if "cwwp2.dot.ca.gov" in src_url or "cctvStatus" in src_url:
cams = parse_caltrans_json(body, name) cams = parse_caltrans_json(body, name)
elif "cctvinventory" in src_url or "tripcheck.com" in src_url:
cams = parse_odot_json(body, name)
elif ("getCameraDataByLoc" in src_url elif ("getCameraDataByLoc" in src_url
or ("json" in ctype and '"locs"' in body[:4000] and '"cams"' in body[:8000])): or ("json" in ctype and '"locs"' in body[:4000] and '"cams"' in body[:8000])):
cams = parse_alertwest_json(body, name) cams = parse_alertwest_json(body, name)

View file

@ -25,7 +25,7 @@ from live_layers import (
_wfigs_params, _wfigs_params,
) )
from camera_scraper import parse_caltrans_json from camera_scraper import parse_caltrans_json, parse_odot_json
def test_parse_bbox_and_radius_clamps_to_150_nm(): def test_parse_bbox_and_radius_clamps_to_150_nm():
@ -257,6 +257,50 @@ def test_parse_caltrans_skips_oos_and_maps_jpeg_hls():
assert "rtsp://" not in cam["snapshot_url"].lower() assert "rtsp://" not in cam["snapshot_url"].lower()
def test_parse_odot_tripcheck_keeps_valid_skips_missing_and_oob():
payload = """
{"features":[
{"attributes":{
"cameraId":277,"filename":"AstoriaUS101_pid392.jpg",
"latitude":46.18785,"longitude":-123.85347,
"route":"US101 ","title":"US101 at Astoria"
}},
{"attributes":{
"cameraId":200,"filename":"","latitude":45.0,"longitude":-122.0,
"route":"I-5","title":"missing filename"
}},
{"attributes":{
"cameraId":300,"filename":"nocal_pid1.jpg",
"latitude":40.0,"longitude":-122.0,
"route":"US97","title":"out of bbox"
}},
{"attributes":{
"cameraId":400,"filename":"badcoord_pid2.jpg",
"latitude":null,"longitude":-122.0,
"route":"OR22","title":"null coord"
}}
]}
"""
cams = parse_odot_json(payload, "www.tripcheck.com")
assert len(cams) == 1
cam = cams[0]
assert cam["discovery_source"] == "odot"
assert cam["snapshot_url"] == (
"https://tripcheck.com/RoadCams/cams/AstoriaUS101_pid392.jpg")
assert cam["source_url"] == cam["snapshot_url"]
assert cam["location_lat"] == 46.18785
assert cam["location_lon"] == -123.85347
assert "US101 at Astoria" in cam["location_name"]
assert cam["vendor"] == "ODOT"
assert cam["device_type"] == "http"
assert "rtsp://" not in cam["snapshot_url"].lower()
def test_parse_odot_tripcheck_handles_malformed():
assert parse_odot_json("not json", "www.tripcheck.com") == []
assert parse_odot_json('{"features":null}', "www.tripcheck.com") == []
def test_quantize_bbox_stable_under_jitter(): def test_quantize_bbox_stable_under_jitter():
a = quantize_bbox(*parse_bbox("-78.7912,35.7711,-78.6101,35.9102")) 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")) b = quantize_bbox(*parse_bbox("-78.7900,35.7700,-78.6110,35.9090"))