Merge branch 'master' into osint-dashboard/t_715a3b7e-osint-map-right-click-place-dossier

Reviewed-on: #36

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] = []

app/live_layers.py

@@ -155,6 +155,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",
+        },
         "conflicts": {
             "id": "conflicts",
             "kind": "points",
@@ -1444,3 +1450,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)

app/main.py

@@ -59,7 +59,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,
 )
 from place import reverse_geocode
 
@@ -1899,6 +1899,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"),

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()

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":[