From 150cc5fdc45dba10eb3b5ca1a82436b33071aebf Mon Sep 17 00:00:00 2001 From: Sirius DevOps Date: Mon, 31 Aug 2026 21:31:31 -0400 Subject: [PATCH 1/5] feat(infra): GET /api/infrastructure Overpass nuclear markers (bbox) Overpass nwr power=plant + plant:source=nuclear, bbox-required, one in-flight per quantized bbox, 25s server query with identifying UA, 24h TTL cache. Normalized to {id,name,lat,lon,type,extra}. overlay_catalog entry infra_nuclear. Missing bbox 400; unknown type 422. --- app/live_layers.py | 107 ++++++++++++++++++++++++++ app/main.py | 35 ++++++++- tests/test_infrastructure.py | 145 +++++++++++++++++++++++++++++++++++ 3 files changed, 286 insertions(+), 1 deletion(-) create mode 100644 tests/test_infrastructure.py diff --git a/app/live_layers.py b/app/live_layers.py index fa632b1..36b4bb5 100644 --- a/app/live_layers.py +++ b/app/live_layers.py @@ -150,6 +150,12 @@ def overlay_catalog() -> dict: "endpoint": "/api/map/gpsjam", "attribution": "GPSJAM / John Wiseman / ADS-B Exchange", }, + "infra_nuclear": { + "id": "infra_nuclear", + "kind": "points", + "endpoint": "/api/infrastructure?types=nuclear", + "attribution": "OpenStreetMap contributors / Overpass API", + }, } @@ -1429,3 +1435,104 @@ async def fetch_gpsjam(date: str) -> dict: return gpsjam_csv_to_geojson(resp.text) return await _ttl_get(f"gpsjam:{date}", GPSJAM_TTL, _load) +# ── Infrastructure (Overpass) ─────────────────────────────────────────────── + + +OVERPASS_INTERPRETER = "https://overpass-api.de/api/interpreter" +# One in-flight query per quantized bbox (the per-key lock in _ttl_get). Overpass +# asks for a 25s server timeout in-band; the client gives it 30s of headroom. +OVERPASS_TIMEOUT = httpx.Timeout(30.0, connect=5.0) +INFRA_TTL = 24 * 3600 # 24h per quantized bbox — static infrastructure + +# `types=` enum. Nuclear ships first; military/hospital slot in behind the same +# query template without touching the transport. Overpass bbox is +# (south, west, north, east), i.e. (minlat, minlon, maxlat, maxlon). +_INFRA_QUERIES: dict[str, str] = { + "nuclear": ( + '[out:json][timeout:25];\n' + 'nwr["power"="plant"]["plant:source"="nuclear"]({bbox});\n' + 'out center;' + ), +} + + +def infra_query(type_: str, minlon: float, minlat: float, maxlon: float, maxlat: float) -> str: + """Render one Overpass query with the bbox substituted in south,west,north,east.""" + bbox = f"{minlat},{minlon},{maxlat},{maxlon}" + return _INFRA_QUERIES[type_].replace("{bbox}", bbox) + + +def normalize_infra_element(elem: dict, type_: str) -> dict | None: + """Map one Overpass element to ``{id, name, lat, lon, type, extra}``. + + ``out center`` gives nodes their own ``lat``/``lon`` and ways/relations a + ``center``. Elements with no usable coordinate are dropped. + """ + etype = elem.get("type") + eid = elem.get("id") + if eid is None: + return None + if etype == "node": + lat, lon = elem.get("lat"), elem.get("lon") + else: + center = elem.get("center") or {} + lat, lon = center.get("lat"), center.get("lon") + if lat is None or lon is None: + return None + tags = elem.get("tags") or {} + name = tags.get("name") or tags.get("ref") or f"{etype}/{eid}" + extra = {k: v for k, v in tags.items() if k != "name"} + return { + "id": f"{etype}/{eid}", + "name": name, + "lat": lat, + "lon": lon, + "type": type_, + "extra": extra, + } + + +def overpass_nuclear_to_markers(data: dict) -> list[dict]: + """Convert an Overpass JSON response to normalized nuclear markers.""" + markers = [] + for elem in data.get("elements") or []: + marker = normalize_infra_element(elem, "nuclear") + if marker is not None: + markers.append(marker) + return markers + + +async def fetch_infrastructure(types: str, bbox: str) -> list[dict]: + """Fetch Overpass infrastructure markers, cached 24h per quantized bbox. + + ``types`` is a single supported enum value (``nuclear`` for now). ``bbox`` + is ``minlon,minlat,maxlon,maxlat``. + """ + requested = [t.strip() for t in types.split(",") if t.strip()] + minlon, minlat, maxlon, maxlat = parse_bbox(bbox) + key = f"infra:{','.join(requested)}:{bbox_cell_key(bbox)}" + + async def _load() -> list[dict]: + # One query per requested type, concatenated. Nuclear is the only type + # today; the loop keeps the shape ready for military/hospital. + out: list[dict] = [] + for type_ in requested: + query = infra_query(type_, minlon, minlat, maxlon, maxlat) + if _http is None: + async with httpx.AsyncClient( + timeout=OVERPASS_TIMEOUT, follow_redirects=True, + headers=_headers(), + ) as client: + resp = await client.post(OVERPASS_INTERPRETER, data={"data": query}) + resp.raise_for_status() + out.extend(overpass_nuclear_to_markers(resp.json())) + else: + resp = await _http.post( + OVERPASS_INTERPRETER, data={"data": query}, + timeout=OVERPASS_TIMEOUT, + ) + resp.raise_for_status() + out.extend(overpass_nuclear_to_markers(resp.json())) + return out + + return await _ttl_get(key, float(INFRA_TTL), _load) diff --git a/app/main.py b/app/main.py index 2d12e25..b4fe5bd 100644 --- a/app/main.py +++ b/app/main.py @@ -57,7 +57,7 @@ from live_layers import ( fetch_aircraft, fetch_fire_incidents, fetch_fire_perimeters, fetch_gpsjam, fetch_planespotters_photo, fetch_radar_meta, fetch_sentinel1, fetch_storms, fetch_trains, fetch_vessels, fetch_weather_alerts, - overlay_catalog, parse_bbox, UpstreamRateLimited, + fetch_infrastructure, overlay_catalog, parse_bbox, UpstreamRateLimited, ) logging.basicConfig(level=logging.INFO) @@ -1764,6 +1764,39 @@ async def map_gpsjam( return overlay_json(fc, 3600) +_INFRA_TYPES = frozenset({"nuclear"}) + + +@app.get("/api/infrastructure") +async def api_infrastructure( + types: str = Query(..., description="comma-separated enum (nuclear)"), + bbox: str | None = Query(None, description="minlon,minlat,maxlon,maxlat"), +): + """Overpass-derived static infrastructure markers (nuclear power plants). + + ``bbox`` is required; ``types`` is a comma-separated subset of ``nuclear``. + Fetched from Overpass (identifying UA, 25s query) and cached 24h per + quantized bbox. Markers are ``{id, name, lat, lon, type, extra}``. + """ + if not bbox: + raise HTTPException(400, "bbox required (minlon,minlat,maxlon,maxlat)") + requested = [t.strip() for t in (types or "").split(",") if t.strip()] + if not requested: + raise HTTPException(422, "types required (e.g. nuclear)") + unknown = [t for t in requested if t not in _INFRA_TYPES] + if unknown: + raise HTTPException( + 422, f"unsupported types: {', '.join(unknown)} (supported: nuclear)" + ) + try: + markers = await fetch_infrastructure(",".join(requested), bbox) + except ValueError as exc: + raise HTTPException(422, str(exc)) from exc + except Exception as exc: + _upstream_or_502(exc, "infrastructure") + return overlay_json(markers, 86400) + + @app.get("/api/map/times") async def map_layer_times( layer: str = Query(..., description="GIBS layer identifier, e.g. VIIRS_SNPP_CorrectedReflectance_TrueColor"), diff --git a/tests/test_infrastructure.py b/tests/test_infrastructure.py new file mode 100644 index 0000000..27dd80f --- /dev/null +++ b/tests/test_infrastructure.py @@ -0,0 +1,145 @@ +"""GET /api/infrastructure — Overpass nuclear markers.""" + +import asyncio + +import httpx + +from live_layers import ( + normalize_infra_element, + overlay_catalog, + overpass_nuclear_to_markers, + _cache, +) +from main import app + +BASE = "http://test" + +OVERPASS = { + "version": 0.6, + "generator": "Overpass API", + "elements": [ + { + "type": "node", + "id": 12345, + "lat": 44.0, + "lon": -1.5, + "tags": {"name": "Test NPP", "operator": "EDF", "plant:source": "nuclear"}, + }, + { + "type": "way", + "id": 67890, + "center": {"lat": 43.5, "lon": -1.25}, + "tags": {"name": "Test Plant Way", "plant:source": "nuclear"}, + }, + { + "type": "relation", + "id": 999, + "center": {"lat": 43.0, "lon": -1.0}, + "tags": {}, + }, + ], +} + + +async def _get(path: str) -> httpx.Response: + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url=BASE) as client: + return await client.get(path) + + +def test_normalize_node_to_marker(): + m = normalize_infra_element(OVERPASS["elements"][0], "nuclear") + assert m["id"] == "node/12345" + assert m["name"] == "Test NPP" + assert m["lat"] == 44.0 + assert m["lon"] == -1.5 + assert m["type"] == "nuclear" + assert m["extra"]["operator"] == "EDF" + assert "name" not in m["extra"] + + +def test_way_center_and_unnamed_fallback(): + way = normalize_infra_element(OVERPASS["elements"][1], "nuclear") + assert way["lat"] == 43.5 + assert way["lon"] == -1.25 + rel = normalize_infra_element(OVERPASS["elements"][2], "nuclear") + assert rel["name"] == "relation/999" + + +def test_overpass_json_to_markers(): + markers = overpass_nuclear_to_markers(OVERPASS) + assert len(markers) == 3 + assert markers[0]["id"] == "node/12345" + + +def test_missing_bbox_400(): + resp = asyncio.run(_get("/api/infrastructure?types=nuclear")) + assert resp.status_code == 400 + + +def test_unknown_type_422(): + resp = asyncio.run(_get("/api/infrastructure?types=military&bbox=-2,43,-1,44")) + assert resp.status_code == 422 + + +def test_map_infrastructure_returns_markers(monkeypatch): + async def fake_fetch(types, bbox): + return [ + {"id": "node/1", "name": "X", "lat": 1.0, "lon": 2.0, + "type": "nuclear", "extra": {}} + ] + + monkeypatch.setattr("main.fetch_infrastructure", fake_fetch) + resp = asyncio.run(_get("/api/infrastructure?types=nuclear&bbox=-2,43,-1,44")) + assert resp.status_code == 200 + body = resp.json() + assert body[0]["name"] == "X" + assert body[0]["type"] == "nuclear" + assert "max-age" in (resp.headers.get("cache-control") or "").lower() + + +def test_overlay_catalog_has_infra_nuclear(): + entry = overlay_catalog()["infra_nuclear"] + assert entry["kind"] == "points" + assert "nuclear" in entry["endpoint"] + + +def test_fetch_infrastructure_cache_hit_no_refetch(monkeypatch): + _cache.clear() + hits = {"n": 0} + + class FakeResp: + def raise_for_status(self): + pass + + def json(self): + return OVERPASS + + class FakeClient: + def __init__(self, **kw): + pass + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + async def post(self, url, data=None, timeout=None): + hits["n"] += 1 + assert "overpass-api.de" in url + assert "plant:source" in data["data"] + assert "nuclear" in data["data"] + return FakeResp() + + monkeypatch.setattr("live_layers.httpx.AsyncClient", FakeClient) + monkeypatch.setattr("live_layers._http", None) + + from live_layers import fetch_infrastructure + + m1 = asyncio.run(fetch_infrastructure("nuclear", "-2,43,-1,44")) + m2 = asyncio.run(fetch_infrastructure("nuclear", "-2,43,-1,44")) + assert len(m1) == 3 + assert m2 == m1 + assert hits["n"] == 1 + _cache.clear() From a9ca98e79107511ef5d8ff6736f19b22a41e0915 Mon Sep 17 00:00:00 2001 From: Sirius DevOps Date: Mon, 31 Aug 2026 21:34:50 -0400 Subject: [PATCH 2/5] feat(cameras): UDOT IBI 511 parser -> cameras table Add parse_udot_ibi_page + scrape_udot_ibi for the UDOT IBI 511 traffic camera feed (prod-ut.ibi511.com, no key). DataTables endpoint is POST form-encoded and caps at 100 rows/page regardless of `length`; page walk uses recordsTotal with a UDOT_IBI_MAX_PAGES (default 40) runaway cap. Skip images[0].blocked/disabled, parse WKT POINT(lng lat) from latLng.geography.wellKnownText, drop outside the Utah bbox (lat 36.9-42.1, lon -114.2--108.9). discovery_source=udot, stable source_url == snapshot_url == /map/Cctv/{id} (never scrape frames), url_hash dedupe, OSINT_USER_AGENT + X-Requested-With header. Unit tests: WKT lng/lat order, blocked/disabled skip, bbox drop, malformed payload, missing WKT. --- app/camera_config.py | 12 ++++ app/camera_scraper.py | 136 ++++++++++++++++++++++++++++++++++++++ tests/test_live_layers.py | 79 +++++++++++++++++++++- 3 files changed, 226 insertions(+), 1 deletion(-) 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")) From 2ec210aaa42972ab87684d80a17903a8d4e32867 Mon Sep 17 00:00:00 2001 From: Sirius DevOps Date: Mon, 31 Aug 2026 21:37:10 -0400 Subject: [PATCH 3/5] feat(map): Conflicts layer toggle from /api/conflicts Default-off LAYERS row paints L.circleMarker theatres colored by severity (war/high/elevated). Popup shows label, description, and eventCount. Catalog is fetched once (not on moveend). A 404 hides the toggle. Null lat/lon rows are skipped so they never pin at 0,0. --- app/static/index.html | 104 ++++++++++++++++++++++++++++++- tests/test_conflicts_frontend.py | 66 ++++++++++++++++++++ 2 files changed, 169 insertions(+), 1 deletion(-) create mode 100644 tests/test_conflicts_frontend.py diff --git a/app/static/index.html b/app/static/index.html index 02122d1..9298d03 100644 --- a/app/static/index.html +++ b/app/static/index.html @@ -341,6 +341,7 @@ .lp-dot.wfigs { background: #ef4444; box-shadow: 0 0 7px #ef4444; } .lp-dot.trains { background: #c084fc; box-shadow: 0 0 7px #c084fc; } .lp-dot.storms { background: #f472b6; box-shadow: 0 0 7px #f472b6; } + .lp-dot.conflicts { background: #ff2a6d; box-shadow: 0 0 7px #ff2a6d; } .lp-count { font-family: 'Share Tech Mono', monospace; font-size: 0.7rem; color: var(--cyan); } .lp-sub { display: flex; justify-content: space-between; align-items: center; gap: 0.5rem; } .lp-opacity { display: flex; align-items: center; gap: 0.4rem; font-size: 0.62rem; color: var(--muted); text-transform: uppercase; letter-spacing: 0.05em; } @@ -396,6 +397,7 @@ .blip-pop { min-width: 200px; max-width: 260px; } .blip-pop .blip-src { font-family: 'Share Tech Mono', monospace; font-size: 0.62rem; color: var(--cyan); text-transform: uppercase; letter-spacing: 0.08em; } .blip-pop .blip-time { font-size: 0.7rem; color: var(--muted); font-family: 'Share Tech Mono', monospace; margin: 0.2rem 0 0.3rem; } + .blip-pop .blip-desc { font-size: 0.72rem; color: var(--text); line-height: 1.35; margin-top: 0.15rem; } /* ── Sub-views (News / Events / Alerts / ... ) ── */ .subview { @@ -922,6 +924,13 @@ 0 +
+
+ + 0 +
+
Curated theatres · color by severity · news counts, not LiveUAMap.
+

Fire heat intensity

@@ -1923,7 +1932,8 @@ let acGroup = null, acOn = true; let trainsGroup = null, trainsOn = true; let vesselsGroup = null, vesselsOn = false; let stormsGroup = null, stormsOn = true; -let overlayReq = {ac:0, trains:0, vessels:0, alerts:0, perim:0, incidents:0, storms:0, sar:0}; +let conflictsGroup = null, conflictsOn = false, conflictsCache = null; +let overlayReq = {ac:0, trains:0, vessels:0, alerts:0, perim:0, incidents:0, storms:0, sar:0, conflicts:0}; let moveDebounce = null; let overlayAbort = null; let lastCell = ''; @@ -2159,8 +2169,10 @@ async function initMap() { trainsOn = document.getElementById('lp-trains-on').checked; vesselsOn = document.getElementById('lp-vessels-on').checked; stormsOn = document.getElementById('lp-storms-on').checked; + conflictsOn = document.getElementById('lp-conflicts-on').checked; connectLiveWs(); loadChokepoints(); + probeConflicts(); requestAnimationFrame(() => { if (firesOn) loadFires(); setTimeout(() => { @@ -3411,6 +3423,96 @@ async function loadStorms() { } } +function conflictSeverityColor(sev) { + const s = String(sev || '').toLowerCase(); + if (s === 'war') return '#ff2a6d'; + if (s === 'high') return '#fb923c'; + if (s === 'elevated') return '#facc15'; + return '#35e0ff'; +} +function hideConflictsToggle() { + const row = document.getElementById('conflicts-layer'); + if (row) row.hidden = true; + const cb = document.getElementById('lp-conflicts-on'); + if (cb) cb.checked = false; + conflictsOn = false; + conflictsCache = null; + conflictsGroup = dropLayer(conflictsGroup); +} +function paintConflicts() { + if (!map || !conflictsOn) return; + const zones = (conflictsCache && Array.isArray(conflictsCache.zones)) ? conflictsCache.zones : []; + conflictsGroup = dropLayer(conflictsGroup); + const markers = []; + for (const z of zones) { + if (z.lat == null || z.lon == null) continue; + const lat = Number(z.lat), lon = Number(z.lon); + if (!Number.isFinite(lat) || !Number.isFinite(lon)) continue; + const col = conflictSeverityColor(z.severity); + const sev = String(z.severity || '').toLowerCase(); + const radius = sev === 'war' ? 10 : sev === 'high' ? 8 : 7; + const m = L.circleMarker([lat, lon], { + radius, + color: col, + fillColor: col, + fillOpacity: 0.28, + weight: 2, + className: 'conflict-zone', + }); + const n = Number(z.eventCount); + const count = Number.isFinite(n) ? n : 0; + m.bindPopup( + `
` + + `
${esc(z.severity || '')} · ${esc(count)} events
` + + `${esc(z.label || '')}` + + `
${esc(z.description || '')}
` + + `
` + ); + markers.push(m); + } + conflictsGroup = L.layerGroup(markers).addTo(map); + const countEl = document.getElementById('lp-conflicts-count'); + if (countEl) countEl.textContent = markers.length.toLocaleString(); + addExtraAttrib('Curated OSINT conflict catalog'); +} +async function probeConflicts() { + await loadConflicts(false); +} +async function toggleConflicts() { + const cb = document.getElementById('lp-conflicts-on'); + conflictsOn = !!(cb && cb.checked); + if (conflictsOn) await loadConflicts(true); + else conflictsGroup = dropLayer(conflictsGroup); +} +async function loadConflicts(paint) { + const shouldPaint = paint === true || conflictsOn; + const req = ++overlayReq.conflicts; + const countEl = document.getElementById('lp-conflicts-count'); + try { + if (!conflictsCache) { + // Own fetch — catalog is viewport-independent; overlayAbort on + // moveend must not cancel this (and we never refetch on pan). + const r = await fetch(`${API}/api/conflicts`); + if (req !== overlayReq.conflicts) return; + if (r.status === 404) { + hideConflictsToggle(); + return; + } + if (!r.ok) throw new Error('conflicts ' + r.status); + const body = await r.json(); + if (req !== overlayReq.conflicts) return; + conflictsCache = body && typeof body === 'object' ? body : { zones: [] }; + } + const zones = Array.isArray(conflictsCache.zones) ? conflictsCache.zones : []; + if (countEl) countEl.textContent = zones.length.toLocaleString(); + if (shouldPaint) paintConflicts(); + } catch (e) { + if (req !== overlayReq.conflicts) return; + console.error('Conflicts load failed', e); + if (countEl) countEl.textContent = 'err'; + } +} + /* ═══════════════ INITIAL LOAD ═══════════════ */ initNav(); initSettings(); diff --git a/tests/test_conflicts_frontend.py b/tests/test_conflicts_frontend.py new file mode 100644 index 0000000..058fd3b --- /dev/null +++ b/tests/test_conflicts_frontend.py @@ -0,0 +1,66 @@ +"""Conflicts Leaflet overlay: default-off toggle, catalog fetch, no jitter.""" + +from __future__ import annotations + +from pathlib import Path + +ROOT = Path(__file__).resolve().parent.parent +HTML = (ROOT / "app/static/index.html").read_text() + + +def _fn(name: str, until: str | None = None) -> str: + chunk = HTML.split(f"function {name}", 1)[1] + if until: + chunk = chunk.split(until, 1)[0] + return chunk + + +def test_conflicts_toggle_default_off(): + assert 'id="lp-conflicts-on"' in HTML + assert 'id="conflicts-layer"' in HTML + assert "> Conflicts<" in HTML or "> Conflicts", 1)[0] + assert "checked" not in on + + +def test_conflicts_fetches_catalog_not_liveuamap(): + js = _fn("loadConflicts", "/* ═══════════════ INITIAL LOAD") + assert "/api/conflicts" in js + assert "liveuamap.com" not in HTML.lower() + assert "Math.random" not in js + assert "jitter" not in js.lower() + + +def test_conflicts_not_refetched_on_moveend(): + refresh = HTML.split("function refreshLiveOverlays", 1)[1].split( + "function addExtraAttrib", 1 + )[0] + assert "loadConflicts" not in refresh + assert "probeConflicts" not in refresh + init = HTML.split("function initMap", 1)[1].split("function readMapPrefs", 1)[0] + assert "probeConflicts()" in init + assert "loadConflicts(true)" not in init + assert "paintConflicts()" not in init + + +def test_conflicts_hides_toggle_on_404(): + js = _fn("loadConflicts", "/* ═══════════════ INITIAL LOAD") + assert "r.status === 404" in js + assert "hideConflictsToggle()" in js + hide = _fn("hideConflictsToggle", "function paintConflicts") + assert "row.hidden = true" in hide + assert "lp-conflicts-on" in hide + + +def test_conflicts_popup_and_severity_colors(): + paint = _fn("paintConflicts", "async function probeConflicts") + assert "z.label" in paint + assert "z.description" in paint + assert "eventCount" in paint + assert "L.circleMarker" in paint + assert "z.lat == null || z.lon == null" in paint + assert "Number.isFinite(lat)" in paint + color = _fn("conflictSeverityColor", "function hideConflictsToggle") + assert "war" in color and "#ff2a6d" in color + assert "high" in color and "#fb923c" in color + assert "elevated" in color and "#facc15" in color From 41433be5740f04dda0e2d5ecaca95a166885f5f3 Mon Sep 17 00:00:00 2001 From: Sirius DevOps Date: Mon, 31 Aug 2026 21:37:32 -0400 Subject: [PATCH 4/5] 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. --- app/camera_config.py | 2 ++ app/camera_scraper.py | 48 +++++++++++++++++++++++++++++++++++++++ tests/test_live_layers.py | 46 ++++++++++++++++++++++++++++++++++++- 3 files changed, 95 insertions(+), 1 deletion(-) diff --git a/app/camera_config.py b/app/camera_config.py index fd12ab6..72b64f7 100644 --- a/app/camera_config.py +++ b/app/camera_config.py @@ -25,6 +25,8 @@ _DEFAULT_SOURCE_URL = ",".join(( "https://raw.githubusercontent.com/willytop8/Live-Environment-Streams/main/streams.geojson", # Official Caltrans CWWP2 JPEG + HLS CCTV (districts 1–12). *CALTRANS_CCTV_URLS, + # Oregon DOT TripCheck public CCTV JPEG inventory (Esri JSON). + "https://www.tripcheck.com/Scripts/map/data/cctvinventory.js", )) CAMERA_SOURCE_URLS = [ u.strip() diff --git a/app/camera_scraper.py b/app/camera_scraper.py index ad5dc91..3989d90 100644 --- a/app/camera_scraper.py +++ b/app/camera_scraper.py @@ -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 + + def parse_live_streams_geojson(text: str, source_name: str) -> list[dict]: """Parse willytop8/Live-Environment-Streams GeoJSON. @@ -490,6 +536,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 ("getCameraDataByLoc" in src_url or ("json" in ctype and '"locs"' in body[:4000] and '"cams"' in body[:8000])): cams = parse_alertwest_json(body, name) diff --git a/tests/test_live_layers.py b/tests/test_live_layers.py index ae0f68e..e79c2de 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_odot_json 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() +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(): 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")) From 92dd9c5803fcff30de780d258403426d4b4da7be Mon Sep 17 00:00:00 2001 From: Sirius DevOps Date: Mon, 31 Aug 2026 21:41:00 -0400 Subject: [PATCH 5/5] feat(api): GET /api/stats HUD counters (counts only, never 500) --- app/live_layers.py | 9 +++++ app/main.py | 62 +++++++++++++++++++++++++++++ tests/test_api_stats.py | 87 +++++++++++++++++++++++++++++++++++++++++ 3 files changed, 158 insertions(+) create mode 100644 tests/test_api_stats.py diff --git a/app/live_layers.py b/app/live_layers.py index fa632b1..c79714b 100644 --- a/app/live_layers.py +++ b/app/live_layers.py @@ -97,6 +97,11 @@ _MAX_VESSELS = 6000 # Last ADS-B snapshot + WFIGS points for fire↔tanker correlation. aircraft_last_known: dict[str, dict] = {} fire_last_known: list[dict] = [] +# Last-known counts for the cheap GET /api/stats HUD counter. Updated by the +# upstream fetchers so the stats endpoint never does its own network/SQL fan-out +# for these layers; reads are O(1) in-process. +train_count: int = 0 +nws_alert_count: int = 0 def overlay_catalog() -> dict: @@ -966,6 +971,8 @@ async def fetch_trains(bbox: str | None, limit: int = DEFAULT_LIMIT) -> list[dic return transform_amtraker(await _get_json(AMTRAKER_TRAINS)) rows = await _ttl_get("amtraker:trains", 20.0, _load) + global train_count + train_count = len(rows) if bbox: minlon, minlat, maxlon, maxlat = parse_bbox(bbox) return filter_points_bbox(rows, minlon, minlat, maxlon, maxlat, limit) @@ -1119,6 +1126,8 @@ async def fetch_weather_alerts(area: str | None, bbox: str | None) -> dict: logger.warning("NWS alerts fetch failed: %s", exc) nws_ok = False nws_fc = {"features": []} + global nws_alert_count + nws_alert_count = len(nws_fc.get("features") or []) sbw_fc = await _ttl_get("iem:sbw", 45.0, _load_iem) features = [] for feat in nws_fc.get("features") or []: diff --git a/app/main.py b/app/main.py index 2d12e25..177e1c7 100644 --- a/app/main.py +++ b/app/main.py @@ -15,6 +15,7 @@ import asyncio import json import logging import re +import time from contextlib import asynccontextmanager from datetime import datetime, timedelta, timezone from decimal import Decimal @@ -272,6 +273,67 @@ def overlay_json(data, max_age: int) -> JSONResponse: return resp +# ── HUD counters ───────────────────────────────────────────────────────── + +# Cheap ~100 B–2 KB counts for the layer rail. Cached in-process so the HUD +# can poll every second without re-hitting SQL or upstream feeds. +_STATS_TTL = 20.0 +_stats_cache: dict[str, tuple[float, dict]] = {} + + +async def _stats_counts() -> dict: + """Fan out to in-memory last-known / cheap SQL counts. Never raises.""" + from live_layers import ( + aircraft_last_known, vessel_last_known, train_count, nws_alert_count, + ) + + counts: dict[str, int | str] = { + "aircraft": len(aircraft_last_known), + "vessels": len(vessel_last_known), + "trains": train_count, + "cameras": 0, + "fires": 0, + "quakes": 0, + "alerts": nws_alert_count, + } + + # SQL counts are best-effort: a down DB or missing table must not 500 the + # rail — the frontend still renders with zeros. + try: + from camera_models import cameras as cam_table + async with async_session() as session: + counts["cameras"] = int( + (await session.execute(select(func.count()).select_from(cam_table))).scalar() or 0 + ) + counts["fires"] = int( + (await session.execute(select(func.count()).select_from(fires))).scalar() or 0 + ) + counts["quakes"] = int( + (await session.execute( + select(func.count()).select_from(events).where( + events.c.source_type == "earthquake" + ) + )).scalar() or 0 + ) + except Exception as exc: # noqa: BLE001 + logger.warning("stats_db_failed", error=str(exc)) + + counts["timestamp"] = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") + return counts + + +@app.get("/api/stats") +async def api_stats(): + """Cheap HUD counters (counts only — no GeoJSON). Cached ~20 s.""" + now = time.monotonic() + cached = _stats_cache.get("stats") + if cached and now - cached[0] < _STATS_TTL: + return cached[1] + payload = await _stats_counts() + _stats_cache["stats"] = (now, payload) + return overlay_json(payload, 15) + + # ── Feed Sources ────────────────────────────────────────────────────────── @app.get("/api/sources", response_model=list[FeedSourceOut]) diff --git a/tests/test_api_stats.py b/tests/test_api_stats.py new file mode 100644 index 0000000..870965d --- /dev/null +++ b/tests/test_api_stats.py @@ -0,0 +1,87 @@ +"""GET /api/stats HUD counter contract (counts only, small, never 500).""" + +from __future__ import annotations + +import asyncio +import re +from datetime import timezone + +import httpx + +from main import app, _stats_counts + +BASE = "http://test" + +EXPECTED_KEYS = ("aircraft", "vessels", "trains", "cameras", + "fires", "quakes", "alerts", "timestamp") + + +async def _get(path: str) -> httpx.Response: + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url=BASE) as client: + return await client.get(path) + + +def test_stats_200_all_keys_present(): + resp = asyncio.run(_get("/api/stats")) + assert resp.status_code == 200 + body = resp.json() + for key in EXPECTED_KEYS: + assert key in body, f"missing key {key}" + assert "max-age" in (resp.headers.get("cache-control") or "").lower() + + +def test_stats_counters_are_ints(): + body = asyncio.run(_get("/api/stats")).json() + for key in EXPECTED_KEYS: + if key == "timestamp": + continue + assert isinstance(body[key], int), f"{key} is not an int: {body[key]!r}" + + +def test_stats_timestamp_is_iso8601_z(): + body = asyncio.run(_get("/api/stats")).json() + ts = body["timestamp"] + # ISO8601 with a trailing Z (we normalize +00:00 -> Z). + assert isinstance(ts, str) and ts.endswith("Z") + assert re.match(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}", ts) + + +def test_stats_payload_is_tiny(): + resp = asyncio.run(_get("/api/stats")) + assert len(resp.content) < 2048, "stats payload must be counts-only, not GeoJSON" + + +def test_stats_counts_reflect_last_known(monkeypatch): + """aircraft/vessels/trains/alerts come from in-memory last-known state.""" + import live_layers + + monkeypatch.setattr(live_layers, "aircraft_last_known", {str(i): {} for i in range(7)}) + monkeypatch.setattr(live_layers, "vessel_last_known", {str(i): {} for i in range(3)}) + monkeypatch.setattr(live_layers, "train_count", 11) + monkeypatch.setattr(live_layers, "nws_alert_count", 5) + + # _stats_counts imports the dicts/counters inside the function from live_layers, + # so monkeypatching the module attributes is what it observes. + from main import _stats_counts as fn + + body = asyncio.run(fn()) + assert body["aircraft"] == 7 + assert body["vessels"] == 3 + assert body["trains"] == 11 + assert body["alerts"] == 5 + + +def test_stats_db_failure_degrades_to_zero(monkeypatch): + """A down DB yields zeros for the SQL-backed counters, never a 500.""" + # Make the session factory raise synchronously so the try/except in + # _stats_counts degrades the SQL counters to zero (no dangling coroutine). + def _raise(*args, **kwargs): + raise RuntimeError("db down") + + monkeypatch.setattr("main.async_session", _raise) + body = asyncio.run(_stats_counts()) + assert body["cameras"] == 0 + assert body["fires"] == 0 + assert body["quakes"] == 0 + assert isinstance(body["timestamp"], str) \ No newline at end of file