server-26/drb-c2-core/app/internal/ny511.py

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

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 = 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:
        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("<!doctype html") or head.startswith("<html"):
        raise ValueError("511NY answered HTML, not a feed (legacy API dead since the 511x cutover, server-26#198)")
    if len(text) > _XML_MAX_BYTES:
        raise ValueError("511NY XML inventory larger than the 32 MB safety cap")
    if "<!doctype" in text.lower() or "<!entity" in text.lower():
        raise ValueError("511NY XML carries a DTD/entity declaration -- refused")
    import xml.etree.ElementTree as ET
    root = ET.fromstring(text)
    out: List[Dict[str, Any]] = []
    for el in root.iter():
        fields: Dict[str, Any] = {}
        for k, v in el.attrib.items():
            key = _XML_LOOKUP.get(_local(k))
            if key:
                fields.setdefault(key, v)
        for child in el:
            key = _XML_LOOKUP.get(_local(child.tag))
            if key and len(child) == 0 and child.text is not None:
                fields.setdefault(key, child.text.strip())
        if "Latitude" in fields and "Longitude" in fields:
            fields["Disabled"] = _xml_bool(fields.get("Disabled", False))
            fields["Blocked"] = _xml_bool(fields.get("Blocked", False))
            out.append(fields)
    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]