"""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). **Feed status (server-26#198), 2026-10-04:** 511ny.org cut over to the CARS "511x" platform around 2026-09-24..30 and the legacy ``/api/getcameras`` and ``/api/getevents`` endpoints now answer HTTP 200 with the SPA's HTML. The new developer feed is granted only after the NYSDOT Developer Access Agreement request is approved ("If approved for use, you will receive instructions for accessing the data feeds & documentation" -- 511NY "For Developers" page), so its URL shape is NOT public. This client therefore keeps the LEGACY request and record shape behind ``settings.ny511_api_base`` (default: the legacy base) and ``settings.ny511_api_key``. UNVERIFIED until the grant arrives: * the new base URL and whether it still takes ``?key=&format=json``; * whether cameras still come as JSON (the public feed list says "Camera inventory (XML)") and keep the legacy field names; * the still/HLS URL shape and its CORS headers. Never point this at the new site's own backend (api-511x-nysdot.carsprogram.org): the 511NY Terms of Use ban robots and reverse engineering outside the DAA. Access is also gated by the camera-provider registry (``ny511`` provider, default OFF, DAA acknowledgement required) -- see app/internal/cameras/. """ 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 from app.internal.cameras.base import safe_error LEGACY_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 # The DAA forbids altering "the substance of a report", so descriptions are # shown as published. This cap only guards against a pathological record; no # real 511NY description comes near it, and a cut one is marked with an ellipsis. _DESCRIPTION_MAX = 4000 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 = safe_error(e) # never echo ?key= to viewers 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: base = (settings.ny511_api_base or LEGACY_BASE).rstrip("/") r = await client.get(f"{base}/{path}", params=params) r.raise_for_status() text = getattr(r, "text", None) if isinstance(text, str) and text.lstrip()[:1] == "<": # The 511x camera inventory is published as XML; HTML is the dead # legacy endpoint serving the SPA shell. raw = parse_inventory_xml(text) else: try: raw = r.json() except ValueError as e: raise ValueError( "511NY answered neither JSON nor XML (the legacy API has served the 511x " "site's HTML since the cutover, server-26#198)" ) from e 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 # ── XML camera inventory (511x) -- UNVERIFIED shape ────────────────────────── # The 511NY "For Developers" page lists the camera feed as "Camera inventory # (XML)" but publishes no schema (it arrives with the DAA grant). This parser is # deliberately tolerant: any element carrying latitude + longitude children (or # attributes) is a record, and child names are matched case-insensitively onto # the legacy JSON field names so normalize_camera() works unchanged. Replace the # alias table with the real element names once the grant's documentation is in hand. _XML_ALIASES = { "ID": ("id", "cameraid", "camera_id", "deviceid", "device_id"), "Latitude": ("latitude", "lat"), "Longitude": ("longitude", "lon", "lng", "long"), "Name": ("name", "cameraname", "location", "description", "title"), "RoadwayName": ("roadwayname", "roadway", "road", "route"), "DirectionOfTravel": ("directionoftravel", "direction"), "Url": ("url", "imageurl", "image_url", "snapshoturl", "image", "stillurl"), "VideoUrl": ("videourl", "video_url", "streamurl", "video", "hlsurl"), "Disabled": ("disabled",), "Blocked": ("blocked",), } _XML_LOOKUP = {alias: key for key, aliases in _XML_ALIASES.items() for alias in aliases} _XML_MAX_BYTES = 32 * 1024 * 1024 def _local(tag: str) -> str: return tag.rsplit("}", 1)[-1].lower() def _xml_bool(v: Any) -> Any: if isinstance(v, str) and v.strip().lower() in ("true", "false", "1", "0"): return v.strip().lower() in ("true", "1") return v def parse_inventory_xml(text: str) -> List[Dict[str, Any]]: """Map an XML camera inventory onto legacy-shaped dicts. Raises on HTML or on any DTD/entity declaration (stdlib expat would expand entities).""" head = text.lstrip()[:512].lower() if head.startswith(" _XML_MAX_BYTES: raise ValueError("511NY XML inventory larger than the 32 MB safety cap") if " 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]