diff --git a/drb-c2-core/.env.example b/drb-c2-core/.env.example index b0b1067..1faa287 100644 --- a/drb-c2-core/.env.example +++ b/drb-c2-core/.env.example @@ -26,6 +26,7 @@ NODE_OFFLINE_THRESHOLD=90 # Google Maps — for geocoding location strings extracted from transcripts # Enable "Geocoding API" in Cloud Console for this key GOOGLE_MAPS_API_KEY= +NY511_API_KEY= # OpenAI — for transcription (Whisper), intelligence extraction, embeddings, and summaries OPENAI_API_KEY= diff --git a/drb-c2-core/app/config.py b/drb-c2-core/app/config.py index 8ff7b80..cb5695f 100644 --- a/drb-c2-core/app/config.py +++ b/drb-c2-core/app/config.py @@ -32,6 +32,9 @@ class Settings(BaseSettings): # Google Maps (geocoding) google_maps_api_key: Optional[str] = None + # 511NY developer key (server-26#183). Optional: the API answered without one + # as of 2026-09-27, but its terms require a registered key. + ny511_api_key: Optional[str] = None # Gemini (intelligence extraction, embeddings, incident summaries) gemini_api_key: Optional[str] = None diff --git a/drb-c2-core/app/internal/ny511.py b/drb-c2-core/app/internal/ny511.py new file mode 100644 index 0000000..e8708e7 --- /dev/null +++ b/drb-c2-core/app/internal/ny511.py @@ -0,0 +1,140 @@ +"""511NY (NYSDOT) traffic cameras and events, cached in memory (server-26#183). + +Public data, identical for every org, so it is NOT written to Firestore: the +statewide feeds are ~3k cameras and ~2.3k events, and re-writing them every +poll would be millions of writes a day for data nobody needs history of. The +cache is filled lazily on request and refreshed per TTL, so an idle deploy +makes no 511 calls at all. + +A failed refresh keeps serving the last good data and reports the error and +its age to the caller -- an empty layer must never be the only symptom of a +dead feed (the AI-silent-failures lesson). +""" +import asyncio +import time +from datetime import datetime +from typing import Any, Dict, List, Optional + +import httpx + +from app.config import settings +from app.internal.logger import logger + +_BASE = "https://511ny.org/api" +CAMERAS_TTL_S = 60 * 60 # camera list is near-static +EVENTS_TTL_S = 2 * 60 # accidents/closures change minute to minute +RETRY_AFTER_FAILURE_S = 60 +_DESCRIPTION_MAX = 500 + + +class _Feed: + def __init__(self, path: str, ttl_s: int, normalize): + self.path = path + self.ttl_s = ttl_s + self.normalize = normalize + self.items: List[Dict[str, Any]] = [] + self.fetched_at: Optional[float] = None # epoch s of last SUCCESSFUL fetch + self.error: Optional[str] = None + self._next_attempt = 0.0 + self._lock = asyncio.Lock() + + async def get(self) -> "_Feed": + if time.time() < self._next_attempt: + return self + async with self._lock: + if time.time() < self._next_attempt: + return self # another request refreshed while we waited + try: + self.items = await _fetch(self.path, self.normalize) + self.fetched_at = time.time() + self.error = None + self._next_attempt = time.time() + self.ttl_s + except Exception as e: + self._next_attempt = time.time() + min(self.ttl_s, RETRY_AFTER_FAILURE_S) + self.error = f"{type(e).__name__}: {e}"[:300] + logger.warning(f"511NY {self.path} refresh failed, serving {len(self.items)} cached: {self.error}") + return self + + +async def _fetch(path: str, normalize) -> List[Dict[str, Any]]: + params = {"format": "json"} + if settings.ny511_api_key: + params["key"] = settings.ny511_api_key + async with httpx.AsyncClient(timeout=20.0) as client: + r = await client.get(f"{_BASE}/{path}", params=params) + r.raise_for_status() + raw = r.json() + if not isinstance(raw, list): + raise ValueError(f"expected a JSON list, got {type(raw).__name__}") + out = [n for n in (normalize(x) for x in raw) if n is not None] + if raw and not out: + # every record failed to normalize: the schema changed under us + raise ValueError(f"0 of {len(raw)} records parsed -- 511NY schema change?") + return out + + +def _coords(x: Dict[str, Any]) -> Optional[tuple]: + try: + lat, lon = float(x["Latitude"]), float(x["Longitude"]) + except (KeyError, TypeError, ValueError): + return None + if lat == 0 and lon == 0: + return None + return lat, lon + + +def _local_iso(s: Any) -> Optional[str]: + """511NY stamps are 'DD/MM/YYYY HH:MM:SS' New York local time. Returned as a + naive ISO string (no offset) -- display-only, never compared to UTC.""" + if not s: + return None + try: + return datetime.strptime(s, "%d/%m/%Y %H:%M:%S").isoformat() + except (TypeError, ValueError): + return None + + +def normalize_camera(x: Dict[str, Any]) -> Optional[Dict[str, Any]]: + c = _coords(x) + if c is None or x.get("Disabled") or x.get("Blocked") or not x.get("ID"): + return None + return { + "id": x["ID"], + "lat": c[0], + "lon": c[1], + "name": x.get("Name") or "", + "roadway": x.get("RoadwayName") or "", + "direction": x.get("DirectionOfTravel") or "", + "image_url": x.get("Url"), # 511NY serves the current still at this URL + "video_url": x.get("VideoUrl"), # HLS playlist, when the camera streams + } + + +def normalize_event(x: Dict[str, Any]) -> Optional[Dict[str, Any]]: + c = _coords(x) + if c is None or not x.get("ID"): + return None + desc = x.get("Description") or "" + return { + "id": x["ID"], + "lat": c[0], + "lon": c[1], + "type": x.get("EventType") or "", + "subtype": x.get("EventSubType") or "", + "severity": x.get("Severity") or "", + "roadway": x.get("RoadwayName") or "", + "direction": x.get("DirectionOfTravel") or "", + "county": x.get("CountyName") or "", + "description": desc[:_DESCRIPTION_MAX] + ("…" if len(desc) > _DESCRIPTION_MAX else ""), + "start_local": _local_iso(x.get("StartDate")), + "planned_end_local": _local_iso(x.get("PlannedEndDate")), + "updated_local": _local_iso(x.get("LastUpdated")), + } + + +cameras = _Feed("getcameras", CAMERAS_TTL_S, normalize_camera) +events = _Feed("getevents", EVENTS_TTL_S, normalize_event) + + +def in_bbox(items: List[Dict[str, Any]], south: float, west: float, north: float, east: float) -> List[Dict[str, Any]]: + return [i for i in items if south <= i["lat"] <= north and west <= i["lon"] <= east] diff --git a/drb-c2-core/app/main.py b/drb-c2-core/app/main.py index 2923129..f62b20d 100644 --- a/drb-c2-core/app/main.py +++ b/drb-c2-core/app/main.py @@ -17,7 +17,7 @@ from app.internal.auth import ( require_node_service_or_firebase_token, ) from app.routers import nodes, systems, calls, upload, tokens, incidents, alerts, admin, trips, places, links, users -from app.routers import enrollment, media, org, waitlist, telemetry, replay +from app.routers import enrollment, media, org, waitlist, telemetry, replay, traffic from app.internal import dynsec from app.internal import firestore as fstore @@ -127,6 +127,7 @@ app.include_router(incidents.router, dependencies=[Depends(require_service_or_fi app.include_router(alerts.router, dependencies=[Depends(require_service_or_firebase_token)]) app.include_router(trips.router, dependencies=[Depends(require_service_or_firebase_token)]) app.include_router(places.router, dependencies=[Depends(require_service_or_firebase_token)]) +app.include_router(traffic.router, dependencies=[Depends(require_service_or_firebase_token)]) app.include_router(upload.router) # auth is per-node, handled inline app.include_router(admin.router) # auth is per-endpoint (read: firebase, write: admin) app.include_router(replay.router) # auth: admin only (every route spends or reads a replay run) diff --git a/drb-c2-core/app/routers/traffic.py b/drb-c2-core/app/routers/traffic.py new file mode 100644 index 0000000..da80ffa --- /dev/null +++ b/drb-c2-core/app/routers/traffic.py @@ -0,0 +1,38 @@ +import asyncio +from typing import Optional + +from fastapi import APIRouter, HTTPException, Query + +from app.internal import ny511 + +router = APIRouter(prefix="/traffic", tags=["traffic"]) + +# Per-layer cap per response. A statewide view is ~3k cameras; past this the +# map is unreadable anyway, and the client is told the list was cut. +MAX_ITEMS = 1500 + + +def _feed_status(feed: "ny511._Feed") -> dict: + return {"fetched_at": feed.fetched_at, "error": feed.error} + + +@router.get("/511") +async def get_511( + south: float = Query(..., ge=-90, le=90), + west: float = Query(..., ge=-180, le=180), + north: float = Query(..., ge=-90, le=90), + east: float = Query(..., ge=-180, le=180), + layers: Optional[str] = Query("cameras,events", description="comma list: cameras, events"), +): + if south > north or west > east: + raise HTTPException(400, "bbox must satisfy south<=north and west<=east") + wanted = {s.strip() for s in (layers or "").split(",") if s.strip()} + feeds = {name: getattr(ny511, name) for name in ("cameras", "events") if name in wanted} + await asyncio.gather(*(f.get() for f in feeds.values())) + + out: dict = {} + for name, feed in feeds.items(): + hits = ny511.in_bbox(feed.items, south, west, north, east) + out[name] = hits[:MAX_ITEMS] + out[f"{name}_status"] = {**_feed_status(feed), "total_in_bbox": len(hits), "truncated": len(hits) > MAX_ITEMS} + return out diff --git a/drb-c2-core/tests/test_ny511.py b/drb-c2-core/tests/test_ny511.py new file mode 100644 index 0000000..ec86fdd --- /dev/null +++ b/drb-c2-core/tests/test_ny511.py @@ -0,0 +1,98 @@ +""" +server-26#183 — 511NY cameras/events layer. + +The feed is scraped from a third party, so the tests pin the two failure shapes +that would otherwise look like "no traffic right now": a refresh error must keep +the last good data AND report the error, and a schema change (every record +unparseable) must be an error, not an empty list. +""" +import asyncio +from unittest.mock import AsyncMock, patch + +from fastapi.testclient import TestClient + +from app.main import app +from app.internal import ny511 +from app.internal.auth import require_service_or_firebase_token + +client = TestClient(app) + +CAM = {"Latitude": 41.03, "Longitude": -73.76, "ID": "NYSDOT-1", "Name": "I-287 at Exit 5", + "DirectionOfTravel": "Unknown", "RoadwayName": "I-287", "Url": "https://511ny.org/map/Cctv/1", + "VideoUrl": None, "Disabled": False, "Blocked": False} +EVENT = {"Latitude": 41.019265, "Longitude": -73.797869, "ID": "TRANSCOM-1", "EventType": "roadwork", + "EventSubType": "Gas main repairs", "Severity": "Unknown", "RoadwayName": "NY 100", + "DirectionOfTravel": "Both directions", "CountyName": "Westchester", "Description": "x" * 900, + "StartDate": "28/09/2026 09:00:00", "PlannedEndDate": "", "LastUpdated": "26/09/2026 14:01:12"} + + +def setup_function(): + app.dependency_overrides[require_service_or_firebase_token] = lambda: {"admin": True} + for feed in (ny511.cameras, ny511.events): + feed.items, feed.fetched_at, feed.error, feed._next_attempt = [], None, None, 0.0 + + +def teardown_function(): + app.dependency_overrides.pop(require_service_or_firebase_token, None) + + +def test_normalize_camera_skips_disabled_blocked_and_zero_coords(): + assert ny511.normalize_camera(CAM)["image_url"] == "https://511ny.org/map/Cctv/1" + assert ny511.normalize_camera({**CAM, "Disabled": True}) is None + assert ny511.normalize_camera({**CAM, "Blocked": True}) is None + assert ny511.normalize_camera({**CAM, "Latitude": 0, "Longitude": 0}) is None + + +def test_normalize_event_parses_day_first_dates_and_truncates_description(): + e = ny511.normalize_event(EVENT) + assert e["start_local"] == "2026-09-28T09:00:00" # DD/MM, not MM/DD + assert e["planned_end_local"] is None + assert len(e["description"]) == ny511._DESCRIPTION_MAX + 1 + + +def test_failed_refresh_keeps_last_good_data_and_reports_error(): + feed = ny511._Feed("getcameras", 3600, ny511.normalize_camera) + with patch.object(ny511, "_fetch", AsyncMock(return_value=[ny511.normalize_camera(CAM)])): + asyncio.run(feed.get()) + feed._next_attempt = 0.0 + with patch.object(ny511, "_fetch", AsyncMock(side_effect=RuntimeError("boom"))): + asyncio.run(feed.get()) + assert len(feed.items) == 1 and feed.fetched_at is not None + assert "boom" in feed.error + assert feed._next_attempt - feed.fetched_at <= ny511.RETRY_AFTER_FAILURE_S + 5 # retries soon, not after the full TTL + + +def test_schema_change_is_an_error_not_an_empty_layer(): + class Resp: + def raise_for_status(self): pass + def json(self): return [{"lat": 1, "lng": 2}] # renamed fields -> nothing parses + + class Client: + async def __aenter__(self): return self + async def __aexit__(self, *a): pass + async def get(self, *a, **k): return Resp() + + with patch.object(ny511.httpx, "AsyncClient", lambda **k: Client()): + try: + asyncio.run(ny511._fetch("getcameras", ny511.normalize_camera)) + assert False, "expected a schema-change error" + except ValueError as e: + assert "schema" in str(e) + + +def test_endpoint_filters_to_bbox_and_reports_status(): + far = {**CAM, "ID": "NYSDOT-2", "Latitude": 42.9, "Longitude": -78.8} # Buffalo + for feed, rows, norm in ((ny511.cameras, [CAM, far], ny511.normalize_camera), (ny511.events, [EVENT], ny511.normalize_event)): + feed.items = [norm(r) for r in rows] + feed.fetched_at, feed._next_attempt = 1.0, float("inf") + r = client.get("/traffic/511", params={"south": 40.9, "west": -74.0, "north": 41.4, "east": -73.4}) + assert r.status_code == 200 + body = r.json() + assert [c["id"] for c in body["cameras"]] == ["NYSDOT-1"] + assert body["cameras_status"] == {"fetched_at": 1.0, "error": None, "total_in_bbox": 1, "truncated": False} + assert len(body["events"]) == 1 + + +def test_endpoint_rejects_inverted_bbox(): + r = client.get("/traffic/511", params={"south": 41.4, "west": -74.0, "north": 40.9, "east": -73.4}) + assert r.status_code == 400 diff --git a/drb-frontend/components/MapView.tsx b/drb-frontend/components/MapView.tsx index 9767414..2e15488 100644 --- a/drb-frontend/components/MapView.tsx +++ b/drb-frontend/components/MapView.tsx @@ -21,6 +21,8 @@ import { MachineOutputNotice } from "@/components/ui/MachineOutputNotice"; import { useAircraft } from "@/lib/useAircraft"; import { useAircraftTrail } from "@/lib/useAircraftTrail"; import { useVessels } from "@/lib/useVessels"; +import { use511 } from "@/lib/use511"; +import type { Ny511Event, Ny511FeedStatus } from "@/lib/types"; // ── Leaflet icon fix ────────────────────────────────────────────────────────── delete (L.Icon.Default.prototype as unknown as Record)._getIconUrl; @@ -310,6 +312,130 @@ function VesselLayer() { ); } +// ── 511NY traffic layers (server-26#183) ───────────────────────────────────── +// Overlay names double as the event keys for overlayadd/overlayremove. +const OVERLAY_DOT_CAMERAS = "DOT Cameras"; +const OVERLAY_TRAFFIC_EVENTS = "Traffic Events"; + +/** True while the named LayersControl overlay is checked. Overlays start unchecked. */ +function useOverlayShown(map: L.Map, name: string): boolean { + const [shown, setShown] = useState(false); + useEffect(() => { + const on = (e: L.LayersControlEvent) => e.name === name && setShown(true); + const off = (e: L.LayersControlEvent) => e.name === name && setShown(false); + map.on("overlayadd", on); + map.on("overlayremove", off); + return () => { map.off("overlayadd", on); map.off("overlayremove", off); }; + }, [map, name]); + return shown; +} + +function cameraIcon(): L.DivIcon { + return L.divIcon({ + className: "", + html: ``, + iconSize: [16, 12], + iconAnchor: [8, 6], + }); +} + +// Shape + glyph per 511 event type, so the layer reads without colour alone. +const EVENT_STYLE: Record = { + accidentsAndIncidents: { glyph: "!", fill: "#dc2626", label: "Accident / incident" }, + closures: { glyph: "×", fill: "#ea580c", label: "Closure" }, + roadwork: { glyph: "W", fill: "#ca8a04", label: "Roadwork" }, + specialEvents: { glyph: "E", fill: "#7c3aed", label: "Special event" }, + transitOperations: { glyph: "T", fill: "#0891b2", label: "Transit" }, +}; +const EVENT_STYLE_OTHER = { glyph: "i", fill: "#6b7280", label: "Other" }; + +function trafficEventIcon(type: string): L.DivIcon { + const st = EVENT_STYLE[type] ?? EVENT_STYLE_OTHER; + return L.divIcon({ + className: "", + html: `${st.glyph}`, + iconSize: [16, 16], + iconAnchor: [8, 8], + }); +} + +function fmtLocal(iso: string | null): string | null { + if (!iso) return null; + const d = new Date(iso); // naive ISO parses as browser-local; 511NY stamps are NY local + return isNaN(d.getTime()) ? null : d.toLocaleString([], { month: "short", day: "numeric", hour: "numeric", minute: "2-digit" }); +} + +/** Surfaces a dead or stale 511 feed on the map instead of an unexplained empty layer. */ +function FeedProblem({ map, label, status, fetchError }: { map: L.Map; label: string; status?: Ny511FeedStatus; fetchError: string | null }) { + let msg: string | null = null; + if (fetchError) msg = `${label}: could not reach server`; + else if (status?.error) { + const age = status.fetched_at ? `showing data from ${Math.round((Date.now() / 1000 - status.fetched_at) / 60)} min ago` : "no data yet"; + msg = `${label}: 511NY feed error, ${age}`; + } else if (status?.truncated) msg = `${label}: showing ${1500} of ${status.total_in_bbox} — zoom in`; + if (!msg) return null; + return createPortal( +
{msg}
, + map.getContainer(), + ); +} + +function DotCameraLayer() { + const map = useMap(); + const shown = useOverlayShown(map, OVERLAY_DOT_CAMERAS); + const { data, error } = use511(map, "cameras", shown); + return ( + <> + {shown && } + {(data.cameras ?? []).map((c) => ( + + +
+
{c.name}
+ {c.roadway &&
{c.roadway}{c.direction && c.direction !== "Unknown" ? ` · ${c.direction}` : ""}
} + {c.image_url && ( + // Popup content mounts on open, so the timestamp busts the cache per open. + // eslint-disable-next-line @next/next/no-img-element + {`Camera: + )} +
Snapshot: 511NY / NYSDOT
+
+ + + ))} + + ); +} + +function TrafficEventLayer() { + const map = useMap(); + const shown = useOverlayShown(map, OVERLAY_TRAFFIC_EVENTS); + const { data, error } = use511(map, "events", shown); + return ( + <> + {shown && } + {(data.events ?? []).map((e: Ny511Event) => { + const st = EVENT_STYLE[e.type] ?? EVENT_STYLE_OTHER; + const start = fmtLocal(e.start_local); + const end = fmtLocal(e.planned_end_local); + return ( + + +
+
{st.label}{e.subtype ? `: ${e.subtype}` : ""}
+
{[e.roadway, e.direction, e.county].filter(Boolean).join(" · ")}
+ {(start || end) &&
{start ? `From ${start}` : ""}{end ? ` until ${end}` : ""}
} +
{e.description}
+
511NY{e.updated_local ? ` · updated ${fmtLocal(e.updated_local)}` : ""}
+
+ + + ); + })} + + ); +} + function nodeFanIcon(members: NodeRecord[]): L.DivIcon { const n = members.length; const CARD = 13; @@ -826,6 +952,18 @@ export default function MapView({ nodes, activeCalls, incidents = [], calls = [] + {/* Overlays: 511NY DOT cameras + traffic events (server-26#183), opt-in */} + + + + + + + + + + + {/* Overlay: Weather Radar — NEXRAD via Iowa Env Mesonet; key forces remount on refresh */} request(`/org/enrollment-tokens/${tokenId}`, { method: "DELETE" }), + // 511NY cameras/events inside a map bbox (server-26#183) + get511: (bbox: { south: number; west: number; north: number; east: number }, layers: string) => + request(`/traffic/511?${new URLSearchParams({ + south: String(bbox.south), west: String(bbox.west), north: String(bbox.north), east: String(bbox.east), layers, + })}`), + // Public waitlist — no auth, see routers/waitlist.py joinWaitlist: (body: { email: string; org_name?: string; note?: string }) => request<{ ok: boolean }>("/waitlist", { method: "POST", body: JSON.stringify(body) }), diff --git a/drb-frontend/lib/types.ts b/drb-frontend/lib/types.ts index 2cf9546..932620b 100644 --- a/drb-frontend/lib/types.ts +++ b/drb-frontend/lib/types.ts @@ -98,6 +98,49 @@ export interface VesselTrack { last_seen: string; } +// 511NY traffic layer (server-26#183) — GET /traffic/511, cached server-side, +// not Firestore. *_local times are New York local, no offset: display only. +export interface Ny511Camera { + id: string; + lat: number; + lon: number; + name: string; + roadway: string; + direction: string; + image_url: string | null; + video_url: string | null; +} + +export interface Ny511Event { + id: string; + lat: number; + lon: number; + type: string; + subtype: string; + severity: string; + roadway: string; + direction: string; + county: string; + description: string; + start_local: string | null; + planned_end_local: string | null; + updated_local: string | null; +} + +export interface Ny511FeedStatus { + fetched_at: number | null; // epoch seconds of the last successful 511NY fetch + error: string | null; // set when the latest refresh failed (data is then stale) + total_in_bbox: number; + truncated: boolean; +} + +export interface Ny511Response { + cameras?: Ny511Camera[]; + cameras_status?: Ny511FeedStatus; + events?: Ny511Event[]; + events_status?: Ny511FeedStatus; +} + export interface VocabularyPendingTerm { term: string; source: "induction" | "correction"; diff --git a/drb-frontend/lib/use511.ts b/drb-frontend/lib/use511.ts new file mode 100644 index 0000000..f3e1ee3 --- /dev/null +++ b/drb-frontend/lib/use511.ts @@ -0,0 +1,57 @@ +"use client"; + +import { useEffect, useRef, useState } from "react"; +import type L from "leaflet"; +import { c2api } from "@/lib/c2api"; +import type { Ny511Response } from "@/lib/types"; + +// 511NY cameras/events for the current map view (server-26#183). Unlike +// useAircraft/useVessels this is not a Firestore listener: the data is public +// and statewide, so c2-core caches it in memory and serves a bbox slice. +// Fetches only while `enabled` (the overlay is on), on pan/zoom (debounced), +// and on a poll matching the server's events TTL. +const POLL_MS = 2 * 60 * 1000; +const MOVE_DEBOUNCE_MS = 400; +const BBOX_PAD = 0.2; // fetch a little past the edges so small pans don't blank the layer + +export function use511(map: L.Map, layers: "cameras" | "events", enabled: boolean) { + const [data, setData] = useState({}); + const [error, setError] = useState(null); + const seq = useRef(0); + + useEffect(() => { + if (!enabled) { setData({}); return; } + let debounce: ReturnType | undefined; + + const load = async () => { + const b = map.getBounds().pad(BBOX_PAD); + const mine = ++seq.current; + try { + const res = await c2api.get511( + { south: b.getSouth(), west: b.getWest(), north: b.getNorth(), east: b.getEast() }, + layers, + ); + if (mine !== seq.current) return; // a newer pan's response wins + setData(res); + setError(null); + } catch (e) { + if (mine !== seq.current) return; + console.error("use511:", e); + setError(e instanceof Error ? e.message : String(e)); + } + }; + const onMove = () => { clearTimeout(debounce); debounce = setTimeout(load, MOVE_DEBOUNCE_MS); }; + + load(); + map.on("moveend", onMove); + const poll = setInterval(load, POLL_MS); + return () => { + map.off("moveend", onMove); + clearTimeout(debounce); + clearInterval(poll); + seq.current++; // drop any in-flight response + }; + }, [map, layers, enabled]); + + return { data, error }; +}