"""Provider interface, licence model and the per-provider catalog cache. A provider is metadata (who runs the cameras, what their terms say, what region they cover) plus one async ``fetch_catalog`` that returns normalised camera records. Catalogs are public and identical for every org, so -- like ny511.py -- they are cached in memory, fetched lazily on request, and never written to Firestore. A failed refresh keeps serving the last good catalog and reports the error and its age (the AI-silent-failures lesson: an empty layer must never be the only symptom of a dead feed). Camera detection (God's Eye View idea #5) is NOT built: it is blocked on server-26#199 (NYSDOT's written answer on imagery analytics) and #200 (CISO ruling on frame retention). ``CameraSignalSink`` is the seam it would plug into. """ import asyncio import re import time from dataclasses import dataclass from enum import Enum from typing import Any, Dict, List, Optional, Protocol, Tuple from app.internal.logger import logger RETRY_AFTER_FAILURE_S = 60 DEFAULT_CATALOG_TTL_S = 15 * 60 # camera lists change rarely (GEV uses the same 15 min) FETCH_TIMEOUT_S = 20.0 USER_AGENT = "DRB-c2-core/1.0 (+public camera catalog; contact via 511/agency programs)" Camera = Dict[str, Any] BBox = Tuple[float, float, float, float] # south, west, north, east class Licence(str, Enum): """Licence status, as catalogued in design-canvas/camera-sources.md.""" COMMERCIAL_OK = "commercial_ok" ATTRIBUTION_REQUIRED = "attribution_required" # commercial OK, credit must render with imagery AGREEMENT_REQUIRED = "agreement_required" # signed agreement / written permission DRB does not hold NON_COMMERCIAL = "non_commercial" UNVERIFIED = "unverified" # no licence read first-hand PAID = "paid" # needs a commercial contract PROHIBITED = "prohibited" # terms forbid this use outright -- cannot be enabled # Statuses that may only be enabled after an admin explicitly acknowledges the # warning (recorded with who/when in the settings doc and audit_log). GATED = frozenset({Licence.AGREEMENT_REQUIRED, Licence.NON_COMMERCIAL, Licence.UNVERIFIED, Licence.PAID}) @dataclass(frozen=True) class LicenceInfo: status: Licence label: str # one line, shown prominently in Admin terms: str # what the source's own terms say (quoted where read) terms_url: Optional[str] = None attribution: Optional[str] = None # must render wherever this provider's imagery appears attribution_url: Optional[str] = None disclaimer: Optional[str] = None # pass-through disclaimer (e.g. NYSDOT DAA) issue: Optional[str] = None # tracking issue, e.g. "server-26#198" @property def gated(self) -> bool: return self.status in GATED def warning(self) -> Optional[str]: """The blocking warning an admin must acknowledge, or None when ungated.""" if self.status == Licence.PROHIBITED: return f"{self.label}. This source's terms forbid this use; it cannot be enabled." if not self.gated: return None return (f"{self.label}. {self.terms} Enabling it means DRB displays this source's imagery " f"to every signed-in user. Only confirm if DRB actually holds the right to do so.") def as_dict(self) -> dict: return { "status": self.status.value, "label": self.label, "terms": self.terms, "terms_url": self.terms_url, "attribution": self.attribution, "attribution_url": self.attribution_url, "disclaimer": self.disclaimer, "issue": self.issue, "gated": self.gated, "warning": self.warning(), } @dataclass(frozen=True) class ProviderMeta: id: str name: str region: str coverage: BBox # rough bbox; a query outside it never triggers a fetch licence: LicenceInfo formats: str # "stills", "stills + HLS", ... implemented: bool notes: str # what is verified vs unverified about this adapter homepage: Optional[str] = None key_setting: Optional[str] = None # settings attribute holding a required API key ttl_s: int = DEFAULT_CATALOG_TTL_S class CameraSignalSink(Protocol): """Future seam (NOT implemented): aggregate per-class counts from a camera still attached to an incident as corroborating evidence. Blocked on server-26#199 and #200. No frames, plates or faces -- class counts only.""" async def submit(self, camera_id: str, incident_id: str, counts: Dict[str, int], observed_at: float) -> None: ... class CameraProvider: """Base adapter. Subclasses set ``meta`` and implement ``fetch_catalog``.""" meta: ProviderMeta async def fetch_catalog(self) -> List[Camera]: raise NotImplementedError(f"{self.meta.id}: adapter is a stub ({self.meta.notes})") class StubProvider(CameraProvider): """A catalogued source with no working adapter -- listed so Admin shows its licence status, but it can never be enabled.""" def __init__(self, meta: ProviderMeta): assert not meta.implemented self.meta = meta class Catalog: """In-memory cache of one provider's normalised camera list.""" def __init__(self, provider: CameraProvider): self.provider = provider self.items: List[Camera] = [] self.fetched_at: Optional[float] = None # epoch s of last SUCCESSFUL fetch self.last_attempt_at: Optional[float] = None self.error: Optional[str] = None self._next_attempt = 0.0 self._lock = asyncio.Lock() async def get(self) -> "Catalog": if time.time() < self._next_attempt: return self async with self._lock: if time.time() < self._next_attempt: return self ttl = self.provider.meta.ttl_s self.last_attempt_at = time.time() try: items = await self.provider.fetch_catalog() self.items = items self.fetched_at = time.time() self.error = None self._next_attempt = time.time() + ttl except Exception as e: self._next_attempt = time.time() + min(ttl, RETRY_AFTER_FAILURE_S) self.error = safe_error(e) logger.warning(f"cameras/{self.provider.meta.id}: refresh failed, serving {len(self.items)} cached: {self.error}") return self def status(self) -> dict: return {"fetched_at": self.fetched_at, "last_attempt_at": self.last_attempt_at, "error": self.error, "count": len(self.items)} # ── helpers shared by adapters ─────────────────────────────────────────────── # Query strings are where API keys travel (?key=...). httpx puts the full URL in # HTTPStatusError text, and this text reaches every signed-in user through the # status fields of GET /cameras and /traffic/511, so it is dropped here. _URL_QUERY = re.compile(r"(https?://[^\s'\"?#]+)\?[^\s'\"#]*") def safe_error(e: BaseException) -> str: """Error text safe to show users: URL query strings removed, length capped.""" return _URL_QUERY.sub(r"\1?[redacted]", f"{type(e).__name__}: {e}")[:300] def make_camera(provider_id: str, raw_id: Any, lat: float, lon: float, *, name: str = "", roadway: str = "", direction: str = "", image_url: Optional[str] = None, video_url: Optional[str] = None, heading_deg: Optional[float] = None) -> Camera: """One normalised record. ``id`` is provider-prefixed so two catalogs can't collide.""" if heading_deg is None: heading_deg = heading_from_text(direction) return { "id": f"{provider_id}:{raw_id}", "provider": provider_id, "lat": lat, "lon": lon, "name": name, "roadway": roadway, "direction": direction, # Approximate at best: most sources publish the road's direction of # travel, not the camera's true heading. The UI labels viewsheds so. "heading_deg": heading_deg, "image_url": image_url, "video_url": video_url, } _COMPASS = {"N": 0, "NE": 45, "E": 90, "SE": 135, "S": 180, "SW": 225, "W": 270, "NW": 315} _WORDS = {"north": "N", "south": "S", "east": "E", "west": "W", "northeast": "NE", "northwest": "NW", "southeast": "SE", "southwest": "SW"} def heading_from_text(text: Optional[str]) -> Optional[float]: """'Northbound' / 'NB' / 'West' / 'SE' -> degrees; None when it can't tell. 'Both directions', 'Unknown', 'Eastbound/Westbound' etc. are None on purpose.""" if not text: return None t = text.strip().lower() t = re.sub(r"bound$", "", t).strip() if t in _WORDS: return float(_COMPASS[_WORDS[t]]) t2 = re.sub(r"b$", "", t).upper() if re.fullmatch(r"(n|s|e|w|ne|nw|se|sw)b?", t) else t.upper() return float(_COMPASS[t2]) if t2 in _COMPASS else None def finite(v: Any) -> Optional[float]: try: f = float(v) except (TypeError, ValueError): return None return f if f == f and f not in (float("inf"), float("-inf")) else None def plausible(lat: Optional[float], lon: Optional[float], coverage: BBox) -> bool: if lat is None or lon is None or (lat == 0 and lon == 0): return False s, w, n, e = coverage return s <= lat <= n and w <= lon <= e def https_on(url: Any, *hosts: str) -> Optional[str]: """Return ``url`` only when it is https on one of ``hosts`` (or a subdomain). Pins imagery to the agency's own hosts so no catalog field can point the browser somewhere else.""" if not isinstance(url, str) or not url.startswith("https://"): return None host = url[len("https://"):].split("/", 1)[0].split(":", 1)[0].lower() return url if any(host == h or host.endswith("." + h) for h in hosts) else None def bbox_intersects(a: BBox, b: BBox) -> bool: return not (a[2] < b[0] or b[2] < a[0] or a[3] < b[1] or b[3] < a[1]) def in_bbox(items: List[Camera], bbox: BBox) -> List[Camera]: s, w, n, e = bbox return [i for i in items if s <= i["lat"] <= n and w <= i["lon"] <= e] async def get_json(url: str, params: Optional[dict] = None, headers: Optional[dict] = None) -> Any: """GET a JSON catalog. Redirects are refused: a catalog that moves is a change to review, not something to follow to an unknown host.""" import httpx # local import keeps module import cheap for the test stubs h = {"User-Agent": USER_AGENT, "Accept": "application/json", **(headers or {})} async with httpx.AsyncClient(timeout=FETCH_TIMEOUT_S, follow_redirects=False) as client: r = await client.get(url, params=params, headers=h) r.raise_for_status() ctype = r.headers.get("content-type", "") if "json" not in ctype: # 511NY's dead legacy API answers 200 text/html (server-26#198). raise ValueError(f"expected JSON, got {ctype or 'no content-type'} (HTTP {r.status_code})") return r.json() def require_parsed(raw_count: int, out: List[Camera], what: str) -> List[Camera]: """Every record failing to normalise means the schema changed under us -- that is an error, not an empty catalog.""" if raw_count and not out: raise ValueError(f"0 of {raw_count} {what} records parsed -- upstream schema change?") return out