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

Merged
sirius merged 2 commits from osint-dashboard/t_93732570-osint-odot-tripcheck-cameras-cameras-tab into master 2026-08-31 22:14:54 -04:00
3 changed files with 95 additions and 1 deletions

View file

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

View file

@ -25,6 +25,7 @@ import hashlib
import ipaddress
import json
import logging
import math
import re
import time
from datetime import datetime, timezone
@ -379,6 +380,51 @@ def parse_caltrans_json(text: str, source_name: str) -> list[dict]:
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
# MDOT MiDrive field extractors (fields carry rendered HTML).
_MDOT_LAT_RE = re.compile(r"lat=(-?\d+(?:\.\d+)?)", re.I)
_MDOT_LON_RE = re.compile(r"lon=(-?\d+(?:\.\d+)?)", re.I)
@ -563,6 +609,8 @@ async def scrape_source(client: RateLimitedClient, geo: Geocoder,
body = resp.text
if "cwwp2.dot.ca.gov" in src_url or "cctvStatus" in src_url:
cams = parse_caltrans_json(body, name)
elif "cctvinventory" in src_url or "tripcheck.com" in src_url:
cams = parse_odot_json(body, name)
elif "mdotjboss.state.mi.us" in src_url or "/MiDrive/camera/list" in src_url:
cams = parse_mdot_json(body, name)
elif ("getCameraDataByLoc" in src_url

View file

@ -27,7 +27,7 @@ from live_layers import (
_wfigs_params,
)
from camera_scraper import parse_caltrans_json, parse_mdot_json
from camera_scraper import parse_caltrans_json, parse_odot_json, parse_mdot_json
def test_parse_bbox_and_radius_clamps_to_150_nm():
@ -259,6 +259,50 @@ def test_parse_caltrans_skips_oos_and_maps_jpeg_hls():
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_parse_mdot_extracts_html_fields_and_bbox_filters():
rows = [
# In-bbox, full fields.