diff --git a/.gitea/workflows/deploy.yml b/.gitea/workflows/deploy.yml index 93ab700..e186f80 100644 --- a/.gitea/workflows/deploy.yml +++ b/.gitea/workflows/deploy.yml @@ -69,6 +69,21 @@ jobs: name: Deploy to VM needs: build runs-on: ubuntu-latest + # server-26#167: without this, two merges a few minutes apart had two + # "Deploy to VM" jobs on the VM at once (runs 616/617, 652/656): racing + # `compose pull && up -d`, one run's prune breaking the other's + # prune-and-retry, and one run's health check judging the other run's + # containers. Gitea >= 1.26 honours `concurrency` (go-gitea/gitea#32751; + # older versions silently ignored it, and 1.26 also dropped the old + # implicit auto-cancel), the server runs 1.27.3. Job-level so builds still + # overlap, only the VM part is serialised. cancel-in-progress: false + # queues instead of killing a run mid-compose; a newer run joining the + # group replaces any run still waiting, and the Deploy step below skips a + # run that has stopped being main's tip, so an older SHA never lands after + # a newer one. + concurrency: + group: drb-prod-deploy + cancel-in-progress: false outputs: prev_sha: ${{ steps.deploy.outputs.prev_sha }} rollback_status: ${{ steps.rollback.outputs.status }} @@ -112,6 +127,22 @@ jobs: # block a deploy that doesn't actually need the space this time. docker image prune -af || echo "WARNING: pre-deploy image prune failed (server-26#129) -- disk pressure may persist" + # server-26#167: only main's tip deploys. Serialising (the job's + # concurrency group) stops two runs overlapping, but not an older + # run going AFTER a newer one (its build finished later, or someone + # re-ran it), which would silently roll prod back. A superseded run + # exits clean before pulling or restarting anything; the newer run carries this + # commit's changes anyway. To deliberately put an older build back, + # use TAG= by hand on the VM, not a re-run. Runs AFTER the + # server-26#129 prune above on purpose: on a full disk `git fetch` can + # fail too, and that prune must stay the first thing a deploy does. + git fetch -q origin main + MAIN_TIP=$(git rev-parse origin/main) + if [ "$MAIN_TIP" != "${{ gitea.sha }}" ]; then + echo "SUPERSEDED=$MAIN_TIP" + exit 0 + fi + # Update compose files + mosquitto config git pull origin main @@ -183,6 +214,12 @@ jobs: ENDSSH ) echo "$OUTPUT" + SUPERSEDED=$(printf '%s\n' "$OUTPUT" | grep '^SUPERSEDED=' | tail -n1 | cut -d'=' -f2 || true) + if [ -n "$SUPERSEDED" ]; then + echo "main has moved on to $SUPERSEDED - this run (${{ gitea.sha }}) is superseded, nothing deployed." + echo "superseded=true" >> "$GITHUB_OUTPUT" + exit 0 + fi PREV_TAG=$(printf '%s\n' "$OUTPUT" | grep '^PREV_TAG=' | tail -n1 | cut -d'=' -f2) if [ -z "$PREV_TAG" ]; then echo "Could not determine the previous tag from deploy output - rollback target unknown." @@ -192,6 +229,9 @@ jobs: - name: Health check id: health + # A superseded run deployed nothing, so there is nothing of its own to + # check, and /health legitimately answers with the newer SHA. + if: steps.deploy.outputs.superseded != 'true' run: | # Poll rather than sleep-once: the container has to finish starting, # and a fixed sleep is either too short (flaky red) or wastes time on @@ -295,6 +335,12 @@ jobs: name: Deploy Firestore rules & indexes needs: build runs-on: ubuntu-latest + # server-26#167: same race as the app deploy, smaller blast radius: an + # older run's `firebase deploy` finishing last would put stale rules back. + # Its own group, so it never waits on the VM deploy. + concurrency: + group: drb-prod-firestore-rules + cancel-in-progress: false # Deliberately independent of the `deploy` job (app containers) and its # health-check/rollback chain above: a rules/indexes deploy failure has # nothing to roll back (there is no previous "build" of a ruleset to @@ -310,6 +356,13 @@ jobs: FIREBASE_SA_KEY: ${{ secrets.FIREBASE_SA_KEY }} run: | set -e + # server-26#167: only main's tip deploys rules (see the app Deploy + # step). An unreadable remote falls through to deploying, as before. + MAIN_TIP=$(git ls-remote origin refs/heads/main | cut -f1) + if [ -n "$MAIN_TIP" ] && [ "$MAIN_TIP" != "${{ gitea.sha }}" ]; then + echo "main has moved on to $MAIN_TIP - this run is superseded, rules not deployed." + exit 0 + fi # server-26#51: this used to run over SSH on the deploy VM, gated # on the VM having firebase-tools installed. It never did, so it # silently warned-and-skipped on every single deploy for weeks. diff --git a/drb-c2-core/.env.example b/drb-c2-core/.env.example index 1faa287..33eb3e0 100644 --- a/drb-c2-core/.env.example +++ b/drb-c2-core/.env.example @@ -27,6 +27,10 @@ NODE_OFFLINE_THRESHOLD=90 # Enable "Geocoding API" in Cloud Console for this key GOOGLE_MAPS_API_KEY= NY511_API_KEY= +# 511NY feed base URL from the DAA grant (default: legacy 511ny.org/api, dead since the 511x cutover) +NY511_API_BASE= +# Ontario 511 developer key (camera provider "ontario511"; providers default OFF) +ONTARIO511_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 a4a546a..aa258d3 100644 --- a/drb-c2-core/app/config.py +++ b/drb-c2-core/app/config.py @@ -30,11 +30,35 @@ class Settings(BaseSettings): openai_api_key: Optional[str] = None stt_model: str = "whisper-1" # whisper-1 | gpt-4o-mini-transcribe | gpt-4o-transcribe + # AI request timeouts (server-26#153). Without these the OpenAI SDK waits + # up to 600s per attempt and retries twice, and the google-genai client + # waits forever, so a hung provider call had no upper bound at all — and + # recorrelation_sweep.py's hold-back could only guess at one. Every OpenAI + # client is built by app/internal/ai_clients.py and the one Gemini client + # by app/internal/gemini.py; both read these. A timeout raises like any + # other provider error, so every call site's existing fallback applies + # (rules-only correlation, raw transcript, no summary, ...). + # The timeout bounds each phase of one HTTP attempt (connect/write/read); + # these endpoints are non-streaming, so in practice it bounds the attempt. + stt_timeout_seconds: float = 120.0 # Whisper: uploads up to MAX_RECORDING_SECONDS of audio + openai_timeout_seconds: float = 60.0 # chat completions + embeddings + # Retries per OpenAI request (SDK default 2). The SDK retries timeouts too, + # so each one multiplies the worst case by another full timeout. + openai_max_retries: int = 1 + # google-genai does not retry unless told to, so this is the whole bound. + gemini_timeout_seconds: float = 60.0 + # 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. + # 511NY developer key (server-26#183/#198). Issued with the NYSDOT Developer + # Access Agreement grant; the keyless legacy API is gone since the 511x cutover. ny511_api_key: Optional[str] = None + # Base URL for the 511NY feed. Defaults to the legacy API, which is dead since + # the 511x cutover (server-26#198); set it to whatever the DAA grant documents. + ny511_api_base: Optional[str] = None + # Camera providers (app/internal/cameras). Keys only; every provider is still + # OFF until an admin enables it, and licence-gated ones need an acknowledgement. + ontario511_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/ai_clients.py b/drb-c2-core/app/internal/ai_clients.py new file mode 100644 index 0000000..544ce20 --- /dev/null +++ b/drb-c2-core/app/internal/ai_clients.py @@ -0,0 +1,44 @@ +""" +One place every OpenAI client is built, so none of them runs on SDK-default +timeouts (server-26#153). + +The OpenAI SDK's defaults are a 600s read timeout and 2 retries, which it also +applies to timeouts: one hung Whisper request could hold a call's pipeline for +half an hour, twice over with stt_retry_on_degenerate. That unbounded tail is +what recorrelation_sweep.py's MIN_MINUTES_SINCE_PIPELINE_START had to guess +against. With these settings it is derived instead (see +recorrelation_sweep.pipeline_worst_case_seconds). + +A timeout raises openai.APITimeoutError, an ordinary Exception, so each call +site's existing failure path handles it unchanged: ai_health.classify() reads +"Request timed out." as transient. + +The Gemini client is built in app/internal/gemini.py with the same intent. +""" +from typing import Optional + +from app.config import settings + + +def _timeout(timeout: Optional[float]) -> float: + return settings.openai_timeout_seconds if timeout is None else timeout + + +def openai_client(timeout: Optional[float] = None): + """Synchronous OpenAI client. `timeout` overrides openai_timeout_seconds (STT passes its own).""" + from openai import OpenAI # lazy, like every call site before this module + return OpenAI( + api_key=settings.openai_api_key, + timeout=_timeout(timeout), + max_retries=settings.openai_max_retries, + ) + + +def async_openai_client(timeout: Optional[float] = None): + """Async OpenAI client, same limits as openai_client().""" + from openai import AsyncOpenAI + return AsyncOpenAI( + api_key=settings.openai_api_key, + timeout=_timeout(timeout), + max_retries=settings.openai_max_retries, + ) diff --git a/drb-c2-core/app/internal/cameras/__init__.py b/drb-c2-core/app/internal/cameras/__init__.py new file mode 100644 index 0000000..71f1c4b --- /dev/null +++ b/drb-c2-core/app/internal/cameras/__init__.py @@ -0,0 +1,6 @@ +"""Pluggable public-camera providers (live ops wall). + +Every source in ``design-canvas/camera-sources.md`` is one adapter here, each +carrying its licence status from that catalog. Providers default OFF; the +licence gate and the admin toggle live in ``registry``. +""" diff --git a/drb-c2-core/app/internal/cameras/base.py b/drb-c2-core/app/internal/cameras/base.py new file mode 100644 index 0000000..9db4a3a --- /dev/null +++ b/drb-c2-core/app/internal/cameras/base.py @@ -0,0 +1,262 @@ +"""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 diff --git a/drb-c2-core/app/internal/cameras/general.py b/drb-c2-core/app/internal/cameras/general.py new file mode 100644 index 0000000..3e90d68 --- /dev/null +++ b/drb-c2-core/app/internal/cameras/general.py @@ -0,0 +1,373 @@ +"""General / other-region camera sources -- design-canvas/camera-sources.md §C, +including every God's Eye View (GEV) source. + +Endpoints and record shapes for the implemented adapters were checked +first-hand against each live catalog on 2026-10-04 (Ontario 511 excepted: it now +answers "Invalid Key" without a developer key, so its shape is the legacy CARS +v2 shape, UNVERIFIED). Imagery URLs are pinned to each agency's own hosts. None +of these cover DRB's region; they exist so a non-NY customer has a path, and +every one of them is OFF until an admin enables it. + +GEV is MIT-licensed code, but nothing here is copied from it, and none of GEV's +bundled data (config/*.json curated catalogs) is used -- the MIT grant does not +cover data. +""" +import asyncio +from typing import List + +from app.config import settings +from app.internal.cameras.base import ( + Camera, CameraProvider, Licence, LicenceInfo, ProviderMeta, StubProvider, finite, get_json, https_on, + make_camera, plausible, require_parsed, +) + + +def _unverified(label: str, terms: str, url: str = None) -> LicenceInfo: + return LicenceInfo(status=Licence.UNVERIFIED, label=label, terms=terms, terms_url=url) + + +# ── Caltrans CWWP2 ─────────────────────────────────────────────────────────── +class CaltransProvider(CameraProvider): + DISTRICTS = (4, 7, 11, 3) # SF Bay, LA, San Diego, Sacramento + meta = ProviderMeta( + id="caltrans", name="Caltrans CWWP2", region="California (districts 4, 7, 11, 3)", + coverage=(32.4, -124.6, 42.1, -114.0), + licence=_unverified( + "Licence UNVERIFIED -- likely commercial OK; Conditions of Use not yet read", + "\"Commercial Wholesale Web Portal\", \"no charge\"; Conditions of Use apply and have not been read.", + "https://cwwp2.dot.ca.gov/documentation/cctv/cctv.htm"), + formats="stills (~5 min) + HLS", implemented=True, + notes="Schema verified 2026-10-04 (D4). HLS host CORS unverified -- the player falls back to the still.", + homepage="https://cwwp2.dot.ca.gov", + ) + + async def _district(self, d: int) -> list: + payload = await get_json(f"https://cwwp2.dot.ca.gov/data/d{d}/cctv/cctvStatusD{d:02d}.json") + return payload.get("data") or [] if isinstance(payload, dict) else [] + + async def fetch_catalog(self) -> List[Camera]: + results = await asyncio.gather(*(self._district(d) for d in self.DISTRICTS), return_exceptions=True) + ok = [r for r in results if not isinstance(r, BaseException)] + if not ok: + raise RuntimeError(f"every Caltrans district failed: {results[0]}") + rows = [row for district in ok for row in district] + out: List[Camera] = [] + for row in rows: + c = (row or {}).get("cctv") or {} + if str(c.get("inService", "")).lower() != "true": + continue + loc = c.get("location") or {} + lat, lon = finite(loc.get("latitude")), finite(loc.get("longitude")) + if not plausible(lat, lon, self.meta.coverage): + continue + img = https_on(((c.get("imageData") or {}).get("static") or {}).get("currentImageURL"), "dot.ca.gov") + if not img: + continue + vid = https_on((c.get("imageData") or {}).get("streamingVideoURL"), "dot.ca.gov") + name = str(loc.get("locationName") or "") + code = name.split("--", 1)[0].strip() or str(c.get("index")) + out.append(make_camera("caltrans", f"d{loc.get('district')}-{code}", lat, lon, + name=name.split("--", 1)[-1].strip() or name, roadway=str(loc.get("route") or ""), + direction=str(loc.get("direction") or ""), image_url=img, video_url=vid)) + return require_parsed(len(rows), out, "Caltrans") if rows else out + + +# ── DelDOT ─────────────────────────────────────────────────────────────────── +class DelDotProvider(CameraProvider): + meta = ProviderMeta( + id="deldot", name="DelDOT TMC", region="Delaware", + coverage=(38.4, -75.8, 39.9, -75.0), + licence=_unverified("Licence UNVERIFIED -- \"public data\", no explicit licence found", + "No licence or terms of reuse located for tmc.deldot.gov's camera catalog."), + formats="live HLS only (no stills)", implemented=True, + notes="Schema verified 2026-10-04 (361 cams); https HLS host sends ACAO: * (verified). Video-only, so a " + "camera shows nothing when its stream is down. Played directly from the browser -- no DRB egress.", + homepage="https://deldot.gov/map/", + ) + + async def fetch_catalog(self) -> List[Camera]: + payload = await get_json("https://tmc.deldot.gov/json/videocamera.json") + rows = payload.get("videoCameras") or [] if isinstance(payload, dict) else [] + out: List[Camera] = [] + for r in rows: + if r.get("enabled") is not True or r.get("status") != "Active" or not r.get("id"): + continue + lat, lon = finite(r.get("lat")), finite(r.get("lon")) + vid = https_on((r.get("urls") or {}).get("m3u8s"), "deldot.gov") + if not plausible(lat, lon, self.meta.coverage) or not vid: + continue + out.append(make_camera("deldot", r["id"], lat, lon, name=str(r.get("title") or r["id"]), + roadway=str(r.get("county") or ""), video_url=vid)) + return require_parsed(len(rows), out, "DelDOT") + + +# ── Ontario 511 ────────────────────────────────────────────────────────────── +class Ontario511Provider(CameraProvider): + meta = ProviderMeta( + id="ontario511", name="Ontario 511", region="Ontario, Canada", + coverage=(41.6, -95.2, 57.0, -74.3), + licence=_unverified("Licence UNVERIFIED -- Ontario open-data terms page not read", + "GEV reports the Open Government Licence - Ontario; not confirmed first-hand.", + "https://511on.ca/developers/doc"), + formats="stills", implemented=True, + notes="Needs ONTARIO511_API_KEY (verified 2026-10-04: keyless now answers \"Invalid Key\"); 10 calls/60 s. " + "Record shape is the CARS v2 shape (Id/Latitude/Longitude/Views[]), UNVERIFIED since the key change.", + homepage="https://511on.ca", key_setting="ontario511_api_key", + ) + + async def fetch_catalog(self) -> List[Camera]: + if not settings.ontario511_api_key: + raise RuntimeError("ONTARIO511_API_KEY not set") + rows = await get_json("https://511on.ca/api/v2/get/cameras", + params={"format": "json", "lang": "en", "key": settings.ontario511_api_key}) + rows = rows if isinstance(rows, list) else [] + out: List[Camera] = [] + for r in rows: + lat, lon = finite(r.get("Latitude")), finite(r.get("Longitude")) + if not r.get("Id") or not plausible(lat, lon, self.meta.coverage): + continue + views = [v for v in (r.get("Views") or []) if str(v.get("Status", "")).lower() == "enabled"] + img = next((https_on(v.get("Url"), "511on.ca") for v in views if https_on(v.get("Url"), "511on.ca")), None) + if not img: + continue + out.append(make_camera("ontario511", r["Id"], lat, lon, name=str(r.get("Location") or ""), + roadway=str(r.get("Roadway") or ""), direction=str(r.get("Direction") or ""), + image_url=img)) + return require_parsed(len(rows), out, "Ontario 511") + + +# ── DriveBC ────────────────────────────────────────────────────────────────── +class DriveBcProvider(CameraProvider): + meta = ProviderMeta( + id="drivebc", name="DriveBC", region="British Columbia, Canada", + coverage=(48.2, -139.1, 60.1, -114.0), + licence=LicenceInfo( + status=Licence.ATTRIBUTION_REQUIRED, label="Commercial OK with attribution (OGL - British Columbia)", + terms="Open Government Licence - British Columbia: reuse permitted with attribution.", + terms_url="https://www2.gov.bc.ca/gov/content/data/policy-standards/open-data/open-government-licence-bc", + attribution="DriveBC -- contains information licensed under the Open Government Licence - British Columbia", + attribution_url="https://www.drivebc.ca"), + formats="stills", implemented=True, notes="Schema verified 2026-10-04 (1,066 cams).", + homepage="https://www.drivebc.ca", + ) + + async def fetch_catalog(self) -> List[Camera]: + rows = await get_json("https://www.drivebc.ca/api/webcams/") + rows = rows if isinstance(rows, list) else [] + out: List[Camera] = [] + for r in rows: + if r.get("is_on") is not True or r.get("should_appear") is not True: + continue + cid = r.get("id") + coords = (r.get("location") or {}).get("coordinates") or [] + if not isinstance(cid, int) or len(coords) < 2: + continue + lon, lat = finite(coords[0]), finite(coords[1]) + if not plausible(lat, lon, self.meta.coverage): + continue + out.append(make_camera("drivebc", cid, lat, lon, name=str(r.get("name") or ""), + roadway=str(r.get("highway_description") or ""), + direction=str(r.get("orientation") or ""), + image_url=f"https://www.drivebc.ca/images/{cid}.jpg")) + return require_parsed(len(rows), out, "DriveBC") + + +# ── Open Calgary ───────────────────────────────────────────────────────────── +class CalgaryProvider(CameraProvider): + meta = ProviderMeta( + id="calgary", name="Open Calgary traffic cameras", region="Calgary, Alberta", + coverage=(50.8, -114.5, 51.3, -113.8), + # UNVERIFIED like Ontario 511: the OGL - Calgary grant is per GEV, not read + # first-hand (camera-sources.md: "Yes, attribution (per GEV)"), so it is gated. + licence=LicenceInfo( + status=Licence.UNVERIFIED, + label="Licence UNVERIFIED -- OGL - City of Calgary per GEV, terms page not read first-hand", + terms="Open Government Licence - City of Calgary (reuse with attribution) per GEV; not confirmed first-hand.", + terms_url="https://data.calgary.ca/stories/s/Open-Calgary-Terms-of-Use/u45n-7awa", + attribution="Contains information licensed under the Open Government Licence - City of Calgary", + attribution_url="https://data.calgary.ca"), + formats="stills", implemented=True, + notes="Schema verified 2026-10-04 (216 cams). Catalog lists http:// stills; upgraded to https on the same host.", + homepage="https://data.calgary.ca", + ) + + async def fetch_catalog(self) -> List[Camera]: + rows = await get_json("https://data.calgary.ca/resource/k7p9-kppz.json", params={"$limit": "1000"}) + rows = rows if isinstance(rows, list) else [] + out: List[Camera] = [] + for r in rows: + url = str((r.get("camera_url") or {}).get("url") or "") + if url.startswith("http://"): + url = "https://" + url[len("http://"):] + img = https_on(url, "trafficcam.calgary.ca") + coords = (r.get("point") or {}).get("coordinates") or [] + if not img or len(coords) < 2: + continue + lon, lat = finite(coords[0]), finite(coords[1]) + if not plausible(lat, lon, self.meta.coverage): + continue + cid = img.rsplit("/", 1)[-1].rsplit(".", 1)[0] + out.append(make_camera("calgary", cid, lat, lon, name=str(r.get("camera_location") or cid), + roadway=str(r.get("quadrant") or ""), image_url=img)) + return require_parsed(len(rows), out, "Calgary") + + +# ── TfL JamCams ────────────────────────────────────────────────────────────── +class TflProvider(CameraProvider): + IMAGE_PREFIX = "https://s3-eu-west-1.amazonaws.com/jamcams.tfl.gov.uk/" + meta = ProviderMeta( + id="tfl", name="TfL JamCams", region="London, UK", + coverage=(51.2, -0.7, 51.8, 0.4), + # UNVERIFIED: the TfL terms page 403'd during the survey and the grant is + # per GEV only (camera-sources.md: "re-read"), so it is gated until re-read. + licence=LicenceInfo( + status=Licence.UNVERIFIED, + label="Licence UNVERIFIED -- commercial OK with attribution per GEV; TfL terms page not read first-hand", + terms="TfL transport data terms: free incl. commercial, with attribution (per GEV; the terms page " + "returned 403 during the survey -- re-read before selling).", + terms_url="https://tfl.gov.uk/info-for/open-data-users/our-open-data", + attribution="Powered by TfL Open Data", attribution_url="https://tfl.gov.uk"), + formats="stills (+ short MP4, unused)", implemented=True, notes="Schema verified 2026-10-04 (890 cams).", + homepage="https://api.tfl.gov.uk", + ) + + async def fetch_catalog(self) -> List[Camera]: + rows = await get_json("https://api.tfl.gov.uk/Place/Type/JamCam") + rows = rows if isinstance(rows, list) else [] + out: List[Camera] = [] + for r in rows: + props = {p.get("key"): p.get("value") for p in (r.get("additionalProperties") or []) if isinstance(p, dict)} + if str(props.get("available", "")).lower() != "true": + continue + img = str(props.get("imageUrl") or "") + lat, lon = finite(r.get("lat")), finite(r.get("lon")) + raw_id = str(r.get("id") or "").replace("JamCams_", "") + if not img.startswith(self.IMAGE_PREFIX) or not raw_id or not plausible(lat, lon, self.meta.coverage): + continue + out.append(make_camera("tfl", raw_id, lat, lon, name=str(r.get("commonName") or raw_id), + direction=str(props.get("view") or ""), image_url=img)) + return require_parsed(len(rows), out, "TfL") + + +# ── Live Traffic NSW ───────────────────────────────────────────────────────── +class NswProvider(CameraProvider): + meta = ProviderMeta( + id="nsw", name="Live Traffic NSW", region="New South Wales, Australia", + coverage=(-37.6, 140.9, -28.0, 153.7), + licence=LicenceInfo( + status=Licence.ATTRIBUTION_REQUIRED, label="Commercial OK with attribution (CC BY 4.0)", + terms="Catalog's own rights block: copyright Transport for NSW, licence per the Open Data Hub dataset " + "(CC BY 4.0).", + terms_url="https://opendata.transport.nsw.gov.au/dataset/live-traffic-cameras", + attribution="Transport for NSW, CC BY 4.0", attribution_url="https://www.livetraffic.com"), + formats="stills", implemented=True, notes="Schema verified 2026-10-04 (241 cams).", + homepage="https://www.livetraffic.com", + ) + + async def fetch_catalog(self) -> List[Camera]: + payload = await get_json("https://data.livetraffic.com/cameras/traffic-cam.json") + feats = payload.get("features") or [] if isinstance(payload, dict) else [] + out: List[Camera] = [] + for f in feats: + p = f.get("properties") or {} + coords = (f.get("geometry") or {}).get("coordinates") or [] + img = https_on(p.get("href"), "transport.nsw.gov.au") + if not img or len(coords) < 2 or not f.get("id"): + continue + lon, lat = finite(coords[0]), finite(coords[1]) + if not plausible(lat, lon, self.meta.coverage): + continue + out.append(make_camera("nsw", f["id"], lat, lon, name=str(p.get("title") or ""), + roadway=str(p.get("region") or ""), direction=str(p.get("direction") or ""), + image_url=img)) + return require_parsed(len(feats), out, "NSW") + + +# ── Fintraffic / Digitraffic weathercams ───────────────────────────────────── +class FintrafficProvider(CameraProvider): + meta = ProviderMeta( + id="fintraffic", name="Fintraffic / Digitraffic weathercams", region="Finland", + coverage=(59.5, 19.0, 70.2, 31.7), + licence=LicenceInfo( + status=Licence.ATTRIBUTION_REQUIRED, label="Commercial OK with attribution (CC BY 4.0)", + terms="\"Source: Fintraffic / digitraffic.fi, license CC 4.0 BY\".", + terms_url="https://www.digitraffic.fi/en/terms-of-service/", + attribution="Source: Fintraffic / digitraffic.fi, license CC 4.0 BY", + attribution_url="https://www.digitraffic.fi"), + formats="stills (~10 min)", implemented=True, + notes="Schema verified 2026-10-04 (810 stations, several presets each). Digitraffic asks clients to " + "identify themselves (Digitraffic-User header).", + homepage="https://www.digitraffic.fi", ttl_s=30 * 60, + ) + + async def fetch_catalog(self) -> List[Camera]: + payload = await get_json("https://tie.digitraffic.fi/api/weathercam/v1/stations", + headers={"Digitraffic-User": "DRB/c2-core"}) + feats = payload.get("features") or [] if isinstance(payload, dict) else [] + out: List[Camera] = [] + for f in feats: + p = f.get("properties") or {} + coords = (f.get("geometry") or {}).get("coordinates") or [] + sid = str(p.get("id") or "") + if not sid or str(p.get("collectionStatus", "")).upper() != "GATHERING" or len(coords) < 2: + continue + lon, lat = finite(coords[0]), finite(coords[1]) + if not plausible(lat, lon, self.meta.coverage): + continue + for preset in p.get("presets") or []: + pid = str((preset or {}).get("id") or "") + if preset.get("inCollection") is not True or not pid.startswith(sid) or not pid[1:].isdigit(): + continue + out.append(make_camera("fintraffic", pid, lat, lon, name=f"{p.get('name') or sid} ({pid[-2:]})", + image_url=f"https://weathercam.digitraffic.fi/{pid}.jpg")) + return require_parsed(len(feats), out, "Fintraffic") + + +# ── Stubs: catalogued, no adapter ──────────────────────────────────────────── +def _stub(id, name, region, coverage, licence, formats, notes, homepage=None) -> StubProvider: + return StubProvider(ProviderMeta(id=id, name=name, region=region, coverage=coverage, licence=licence, + formats=formats, implemented=False, notes=notes, homepage=homepage)) + + +STUBS = [ + _stub("austin", "City of Austin open data", "Austin, Texas", (30.0, -98.1, 30.6, -97.5), + _unverified("Licence UNVERIFIED -- city open data, no explicit licence", "No explicit licence per GEV."), + "stills (5 min)", "Stub: Socrata column mapping not built.", "https://data.austintexas.gov"), + _stub("txdot", "TxDOT ITS", "Texas (per district)", (25.8, -106.7, 36.6, -93.5), + _unverified("Licence UNVERIFIED -- \"public data\"", "No explicit licence per GEV."), + "base64 JPEG inside JSON", "Stub: frames are base64 inside JSON, so the browser cannot show them " + "without a DRB frame proxy (video/image egress on DRB's bill -- not built).", "https://its.txdot.gov"), + _stub("md_chart", "Maryland CHART", "Maryland", (37.9, -79.5, 39.8, -75.0), + _unverified("Licence UNVERIFIED -- MD open data portal terms not read", "Not read."), + "live video URLs", "Stub: catalog schema not verified.", "https://chart.maryland.gov"), + _stub("wsdot", "WSDOT Traveler Information API", "Washington State", (45.5, -124.9, 49.1, -116.9), + _unverified("Licence UNVERIFIED -- WSDOT disclaimer only", "Free access code required; terms not read."), + "stills", "Stub: needs a WSDOT access code; schema not verified.", "https://wsdot.wa.gov/traffic/api/"), + _stub("tallinn", "Tallinn city cameras", "Tallinn, Estonia", (59.35, 24.55, 59.5, 24.95), + _unverified("Licence UNVERIFIED", "No licence found."), + "stills", "Stub: GEV uses a curated catalog file; GEV's data is not covered by its MIT licence.", + "https://ristmikud.tallinn.ee"), + _stub("tarktee", "Tarktee (Estonian Transport Administration)", "Estonia", (57.5, 21.7, 59.8, 28.3), + _unverified("Licence UNVERIFIED", "No licence found."), + "DATEX2 stills", "Stub: DATEX2 XML parser not built.", "https://tarktee.transpordiamet.ee"), + _stub("warendorf", "Warendorf municipal webcam", "Warendorf, Germany", (51.9, 7.9, 52.0, 8.1), + _unverified("Licence UNVERIFIED", "No licence found."), + "stills", "Stub: single curated webcam; no catalog API.", "https://www.warendorf.de"), + _stub("trafficland", "TrafficLand", "US (18k cams, 50+ DOTs)", (18.0, -170.0, 72.0, -60.0), + LicenceInfo(status=Licence.PAID, label="Paid commercial licence -- no contract", + terms="Licensed API; contractual redistribution with 50+ DOTs. Price UNVERIFIED.", + terms_url="http://www.trafficland.com/api.html"), + "live video", "Stub: the clean route if DRB ever needs video beyond 511NY; needs a contract."), + _stub("windy", "Windy Webcams API", "Global (scenic/weather)", (-90.0, -180.0, 90.0, 180.0), + LicenceInfo(status=Licence.PAID, label="Paid professional tier for commercial use -- no contract", + terms="Free tier image URLs expire after 10 min; commercial terms UNVERIFIED.", + terms_url="https://api.windy.com/webcams/docs"), + "tokenised stills", "Stub: scenic/weather cams, not traffic."), + _stub("earthcam", "EarthCam", "Global", (-90.0, -180.0, 90.0, 180.0), + LicenceInfo(status=Licence.PAID, label="Paid commercial licence -- no contract", + terms="Commercial licensing exists; terms UNVERIFIED."), + "live video", "Stub: low local value."), +] + +PROVIDERS = [CaltransProvider(), DelDotProvider(), Ontario511Provider(), DriveBcProvider(), CalgaryProvider(), + TflProvider(), NswProvider(), FintrafficProvider(), *STUBS] diff --git a/drb-c2-core/app/internal/cameras/regional.py b/drb-c2-core/app/internal/cameras/regional.py new file mode 100644 index 0000000..49d12a6 --- /dev/null +++ b/drb-c2-core/app/internal/cameras/regional.py @@ -0,0 +1,168 @@ +"""Tri-state (Westchester-first) camera sources -- design-canvas/camera-sources.md §A. + +Only 511NY has a working adapter. Every other regional source is listed so +Admin shows its licence status, and is a stub because DRB has no right to use +it yet (NYC DOT, Westchester DPW), it carries no cameras (CTroads), or no +developer program exists (511NJ). Do not "fix" a stub by scraping: CTroads, +511NJ's app and the new 511ny.org backend all ban robots/reverse engineering. +""" +from typing import List + +from app.internal import ny511 +from app.internal.cameras.base import ( + Camera, CameraProvider, Licence, LicenceInfo, ProviderMeta, StubProvider, make_camera, require_parsed, +) + +NY_BBOX = (40.4, -80.0, 45.1, -71.7) +TRISTATE_BBOX = (40.3, -75.6, 42.1, -71.7) + +NYSDOT_DISCLAIMER = ( + "NYSDOT compiles the data from the best available sources and makes a best effort to disseminate it. " + "Data availability and accuracy are subject to change, and may change as often as daily. Posted " + "restrictions in the field may differ from, and supersede, the information in this feed. NYSDOT and its " + "partner agencies make no warranty regarding the data, express or implied, and expressly disclaim all " + "warranties of merchantability and fitness for any particular purpose. NYSDOT makes no warranty that the " + "data will be provided without interruption or that it will be free of errors. Access is provided on an " + "\"as is\" and \"with all faults\" basis." +) + + +class Ny511Provider(CameraProvider): + meta = ProviderMeta( + id="ny511", + name="511NY (NYSDOT)", + region="New York State (aggregates NYSDOT, Thruway, NYC DOT, MTA B&T, PANYNJ, TRANSCOM partners)", + coverage=NY_BBOX, + licence=LicenceInfo( + status=Licence.AGREEMENT_REQUIRED, + label="Requires NYSDOT Developer Access Agreement -- access to the new 511x feed NOT granted yet", + terms=("Commercial redistribution is allowed under the DAA (\"You may redistribute, enhance, repackage, " + "and add value to the data\"; \"You may not alter the substance of a report\"; must display the " + "time the data was published or last updated). Access is granted per request via the 511NY " + "Data Feed Request Form; DRB's 2026-09-27 request went to the retired system. Overlays, " + "detection and frame archiving are not addressed (server-26#199)."), + terms_url="https://511ny.org/help/24", + # DAA permitted use: tag line + logo + link. The approved logo is not + # in the repo yet (owner action), so only the tag line + link render. + attribution="powered by 511NY", + attribution_url="https://511ny.org", + disclaimer=NYSDOT_DISCLAIMER, + issue="server-26#198", + ), + formats="stills (60 s) + HLS on NYSDOT hosts (legacy shape)", + implemented=True, + notes=("Legacy request/record shape behind settings.ny511_api_base + ny511_api_key, plus a tolerant " + "parser for the 511x XML camera inventory. UNVERIFIED until the DAA grant documents the new " + "endpoints: base URL, query params, XML element names, still/HLS URL shape and CORS. The " + "legacy endpoint has returned HTML since the 511x cutover, so expect a feed error until then."), + homepage="https://511ny.org", + key_setting="ny511_api_key", + ttl_s=ny511.CAMERAS_TTL_S, + ) + + async def fetch_catalog(self) -> List[Camera]: + rows = await ny511._fetch("getcameras", ny511.normalize_camera) + # Hosts for the 511x imagery are unverified, so pin the scheme at least: + # only https URLs reach the browser. + def https(u): + return u if isinstance(u, str) and u.startswith("https://") else None + out = [ + make_camera("ny511", r["id"], r["lat"], r["lon"], name=r["name"], roadway=r["roadway"], + direction=r["direction"], image_url=https(r["image_url"]), video_url=https(r["video_url"])) + for r in rows + ] + return require_parsed(len(rows), out, "511NY") + + +NYC_DOT = StubProvider(ProviderMeta( + id="nyc_dot", + name="NYC DOT TMC (webcams.nyctmc.org)", + region="New York City (~975 cams, ~80 in the Bronx)", + coverage=(40.45, -74.30, 40.95, -73.65), + licence=LicenceInfo( + status=Licence.AGREEMENT_REQUIRED, + label="Requires a signed NYC DOT data-sharing agreement -- NOT signed", + terms=("NYC DOT Data Feeds page: \"Developers must sign a data-sharing agreement\" (tmcdot@dot.nyc.gov). " + "The API answers keyless, but keyless use is outside those terms. Agreement terms unpublished."), + terms_url="https://www.nyc.gov/html/dot/html/about/datafeeds.shtml", + issue="server-26#201", + ), + formats="JPEG stills (~15 s), no CORS", + implemented=False, + notes="Stub on purpose: do not integrate until the agreement is signed (server-26#201). Many of these cams " + "are likely already re-published by 511NY.", + homepage="https://webcams.nyctmc.org", +)) + +WESTCHESTER_DPW = StubProvider(ProviderMeta( + id="westchester_dpw", + name="Westchester County DPW traffic cameras", + region="Westchester County (~21 locations, mostly Bronx River Pkwy)", + coverage=(40.88, -73.99, 41.37, -73.48), + licence=LicenceInfo( + status=Licence.AGREEMENT_REQUIRED, + label="No reuse licence -- written permission from County DPW required, NOT obtained", + terms="Only a disclaimer: data \"provided 'as is' without warranty\". No API, no reuse licence.", + terms_url="https://experience.arcgis.com/experience/a59314e926e94301b9083edd6bb52a11", + issue="server-26#202", + ), + formats="stills (5 s), viewer only", + implemented=False, + notes="Stub: there is no documented API or feed, and no permission. Ask DPW in writing (server-26#202).", + homepage="https://www.westchestergov.com/trafficcamera", +)) + +TRANSCOM = StubProvider(ProviderMeta( + id="transcom_xcm", + name="TRANSCOM XCM data exchange", + region="NY/NJ/CT tri-state", + coverage=TRISTATE_BBOX, + licence=LicenceInfo( + status=Licence.UNVERIFIED, + label="Events/travel times commercial OK; cameras UNVERIFIED", + terms="Free registration, open to \"commercial vendors ... for use in their applications\". Whether the " + "feed carries camera URLs is unverified.", + terms_url="https://data.xcmdata.org/DEWeb/Pages/aboutus", + ), + formats="XML events + link travel times (cameras unverified)", + implemented=False, + notes="Stub: register first, then confirm whether any camera inventory exists. Its events feed would be a " + "separate (event) provider, not a camera one.", + homepage="https://data.xcmdata.org", +)) + +CTROADS = StubProvider(ProviderMeta( + id="ctroads", + name="CTDOT / CTroads", + region="Connecticut", + coverage=(40.95, -73.75, 42.06, -71.78), + licence=LicenceInfo( + status=Licence.PROHIBITED, + label="No sanctioned camera path -- site terms forbid camera reuse", + terms="The keyed developer API covers Message Signs, Events and Advisories only -- no cameras. The site " + "T&C forbid use \"for any personal, public or commercial purpose\" and robots.", + terms_url="https://ctroads.org/termsandconditions", + ), + formats="events/signs only (no cameras)", + implemented=False, + notes="Cannot be enabled. CT cameras may surface via 511NY's \"Connecticut Statewide\" region once the DAA " + "grant arrives.", + homepage="https://ctroads.org/developers/doc", +)) + +NJ511 = StubProvider(ProviderMeta( + id="nj511", + name="511NJ / NJDOT / NJTA / PANYNJ", + region="New Jersey", + coverage=(38.9, -75.6, 41.4, -73.9), + licence=LicenceInfo( + status=Licence.UNVERIFIED, + label="No published developer program -- reuse UNVERIFIED", + terms="511NJ is a JS app with no published developer program; NJDOT only cites RSS/Twitter.", + ), + formats="unknown", + implemented=False, + notes="Stub: ask NJDOT 511 if northern NJ matters. Some cams appear in 511NY.", +)) + +PROVIDERS = [Ny511Provider(), NYC_DOT, WESTCHESTER_DPW, TRANSCOM, CTROADS, NJ511] diff --git a/drb-c2-core/app/internal/cameras/registry.py b/drb-c2-core/app/internal/cameras/registry.py new file mode 100644 index 0000000..e3f761c --- /dev/null +++ b/drb-c2-core/app/internal/cameras/registry.py @@ -0,0 +1,200 @@ +"""Camera provider registry: settings, licence gate, and bbox queries. + +Settings live in Firestore ``config/camera_providers``:: + + {"providers": {"": {"enabled": bool, "updated_at": iso, "updated_by": {...}, + "acknowledgement": {"by_uid", "by_email", "at", "licence_status", "warning"} | None}}} + +Rules (owner brief, 2026-10-04): + * every provider defaults OFF -- a missing doc/entry means off; + * a stub (no adapter) or a PROHIBITED source can never be enabled; + * a GATED licence (agreement required / non-commercial / unverified / paid) + can only be enabled with an explicit acknowledgement, recorded with who and + when in the settings doc AND audit_log. The audit write is a precondition + for a gated enable: no record, no enable; + * an acknowledgement is bound to the licence status it was given for. If a + provider's catalogued status changes in code, the old acknowledgement no + longer counts and the provider stops serving until re-acknowledged. +""" +import asyncio +import time +from datetime import datetime, timezone +from typing import Dict, List, Optional, Tuple + +from app.config import settings +from app.internal import firestore as fstore +from app.internal.cameras import general, regional +from app.internal.cameras.base import ( + BBox, Camera, CameraProvider, Catalog, Licence, bbox_intersects, in_bbox, +) +from app.internal.logger import logger + +COLLECTION, DOC_ID = "config", "camera_providers" +_TTL_S = 30.0 +MAX_ITEMS = 1500 # per response, same cap as /traffic/511 + +PROVIDERS: Dict[str, CameraProvider] = {p.meta.id: p for p in [*regional.PROVIDERS, *general.PROVIDERS]} +CATALOGS: Dict[str, Catalog] = {pid: Catalog(p) for pid, p in PROVIDERS.items() if p.meta.implemented} + +_cache: Dict[str, dict] = {} +_cache_ts = 0.0 + + +class ProviderError(Exception): + """A refused settings change. ``code`` is the HTTP status the router returns.""" + + def __init__(self, code: int, message: str, **extra): + super().__init__(message) + self.code, self.message, self.extra = code, message, extra + + +def invalidate_cache() -> None: + global _cache_ts + _cache_ts = 0.0 + + +async def get_settings() -> Dict[str, dict]: + """Per-provider settings. Unreadable Firestore means everything is OFF (fail closed).""" + global _cache, _cache_ts + if _cache_ts and time.monotonic() - _cache_ts < _TTL_S: + return _cache + try: + doc = await fstore.doc_get(COLLECTION, DOC_ID) or {} + provs = doc.get("providers") if isinstance(doc.get("providers"), dict) else {} + _cache = {k: v for k, v in provs.items() if isinstance(v, dict)} + except Exception as e: + logger.warning(f"camera providers: settings unreadable ({e}); treating every provider as OFF") + _cache = {} + _cache_ts = time.monotonic() + return _cache + + +def effective(provider: CameraProvider, entry: Optional[dict]) -> Tuple[bool, Optional[str]]: + """(serving?, reason-if-not) for a provider given its settings entry.""" + meta = provider.meta + if not meta.implemented: + return False, "adapter is a stub" + if meta.licence.status == Licence.PROHIBITED: + return False, "source terms forbid this use" + if not entry or not entry.get("enabled"): + return False, "disabled (default)" + if meta.licence.gated: + ack = entry.get("acknowledgement") or {} + if ack.get("licence_status") != meta.licence.status.value: + return False, "licence acknowledgement missing or out of date" + return True, None + + +async def is_serving(provider_id: str) -> bool: + p = PROVIDERS.get(provider_id) + if p is None: + return False + return effective(p, (await get_settings()).get(provider_id))[0] + + +async def set_enabled(provider_id: str, enabled: bool, actor: Tuple[str, str], + acknowledge_licence: bool = False) -> dict: + p = PROVIDERS.get(provider_id) + if p is None: + raise ProviderError(404, f"unknown camera provider '{provider_id}'") + meta, lic = p.meta, p.meta.licence + uid, email = actor + now = datetime.now(timezone.utc).isoformat() + entry: dict = {"enabled": bool(enabled), "updated_at": now, "updated_by": {"uid": uid, "email": email}} + + if enabled: + if not meta.implemented: + raise ProviderError(409, f"{meta.name}: no working adapter -- {meta.notes}") + if lic.status == Licence.PROHIBITED: + raise ProviderError(409, lic.warning() or "prohibited") + if lic.gated: + if not acknowledge_licence: + raise ProviderError(409, f"{meta.name}: licence acknowledgement required", + requires_acknowledgement=True, warning=lic.warning(), + licence=lic.as_dict()) + entry["acknowledgement"] = {"by_uid": uid, "by_email": email, "at": now, + "licence_status": lic.status.value, "warning": lic.warning()} + # Precondition, not a side effect: a gated enable with no audit + # record must not happen. A failure here aborts before the write. + try: + await _audit(actor, "camera_provider.enable_acknowledged", provider_id, + {"licence_status": lic.status.value, "licence_label": lic.label, + "warning": lic.warning(), "acknowledged_at": now}) + except Exception as e: + logger.error(f"camera providers: audit write failed, refusing gated enable of {provider_id} ({e})") + raise ProviderError(503, "audit log unavailable -- a licence-gated provider cannot be enabled " + "without an audit record; try again") + else: + entry["acknowledgement"] = None # re-enabling a gated source needs a fresh acknowledgement + + before = (await get_settings()).get(provider_id) or {} + await fstore.doc_set(COLLECTION, DOC_ID, {"providers": {provider_id: entry}}, merge=True) + invalidate_cache() + if not (enabled and lic.gated): + try: + await _audit(actor, "camera_provider.enable" if enabled else "camera_provider.disable", provider_id, + {"from": bool(before.get("enabled")), "to": bool(enabled), + "licence_status": lic.status.value}) + except Exception as e: # the toggle already happened; never 500 a completed write + logger.error(f"camera providers: audit write failed for {provider_id} ({e})") + logger.info(f"camera provider {provider_id} -> enabled={enabled} by {email or uid}") + return await describe(provider_id) + + +async def _audit(actor: Tuple[str, str], action: str, provider_id: str, details: dict) -> None: + from app.internal import audit + await audit.write_audit(actor_uid=actor[0], actor_email=actor[1], action=action, + details={"provider_id": provider_id, **details}) + + +async def describe(provider_id: str) -> dict: + p = PROVIDERS[provider_id] + entry = (await get_settings()).get(provider_id) or {} + serving, reason = effective(p, entry) + m = p.meta + cat = CATALOGS.get(provider_id) + key_missing = bool(m.key_setting) and not getattr(settings, m.key_setting, None) + return { + "id": m.id, "name": m.name, "region": m.region, "formats": m.formats, + "implemented": m.implemented, "notes": m.notes, "homepage": m.homepage, + "licence": m.licence.as_dict(), + "enabled": bool(entry.get("enabled")), + "serving": serving, "not_serving_reason": reason, + "acknowledgement": entry.get("acknowledgement"), + "updated_at": entry.get("updated_at"), "updated_by": entry.get("updated_by"), + "key_setting": m.key_setting, "key_missing": key_missing, + "health": cat.status() if cat else None, + } + + +async def list_providers() -> List[dict]: + await get_settings() + return [await describe(pid) for pid in PROVIDERS] + + +def credit(provider_id: str) -> dict: + lic = PROVIDERS[provider_id].meta.licence + return {"attribution": lic.attribution, "attribution_url": lic.attribution_url, "disclaimer": lic.disclaimer} + + +async def cameras_in_bbox(bbox: BBox) -> dict: + """Cameras from every SERVING provider whose coverage meets ``bbox``. + Providers that are off are never fetched.""" + conf = await get_settings() + serving = [pid for pid in CATALOGS if effective(PROVIDERS[pid], conf.get(pid))[0]] + live = [pid for pid in serving if bbox_intersects(PROVIDERS[pid].meta.coverage, bbox)] + await asyncio.gather(*(CATALOGS[pid].get() for pid in live)) + cams: List[Camera] = [] + providers = [] + # Every SERVING provider is listed (credits included) even when the bbox + # misses it, so a client holding one of its cameras from another view can + # still render the required attribution. Only intersecting ones are fetched. + for pid in serving: + cat = CATALOGS[pid] + hits = in_bbox(cat.items, bbox) if pid in live else [] + room = max(0, MAX_ITEMS - len(cams)) + cams.extend(hits[:room]) + providers.append({"id": pid, "name": PROVIDERS[pid].meta.name, **credit(pid), + "fetched_at": cat.fetched_at, "error": cat.error, + "total_in_bbox": len(hits), "truncated": len(hits) > room}) + return {"cameras": cams, "providers": providers, "serving_count": len(serving)} diff --git a/drb-c2-core/app/internal/dynsec.py b/drb-c2-core/app/internal/dynsec.py index 7b1d93f..64044cd 100644 --- a/drb-c2-core/app/internal/dynsec.py +++ b/drb-c2-core/app/internal/dynsec.py @@ -312,7 +312,8 @@ async def delete_node_client(node_id: str) -> None: async def reconcile_all() -> None: """ Rebuild dynsec state for every approved node from Firestore - (node_keys is the source of truth). Called once at c2-core startup, + (node_keys is the source of truth), and delete the login of every + rejected node. Called once at c2-core startup, after ensure_roles_and_c2core_grant(). Self-heals a lost/corrupted dynamic-security.json (e.g. volume wiped, or a prior approve/reissue's dynsec publish silently failed to persist for some other reason) — @@ -320,6 +321,20 @@ async def reconcile_all() -> None: dynsec store would lock out every previously-approved node until someone noticed and manually re-approved each one. """ + # server-26#195: revoke first. A node rejected before reject_node() + # revoked anything (or whose revoke hit a dynsec failure) still has a + # broker login in dynamic-security.json; startup is where that heals. + # delete_node_client() treats "not found" as success, so this is a no-op + # for every node already revoked. + for node in await fstore.collection_list("nodes", approval_status="rejected"): + node_id = node.get("node_id") + if not node_id: + continue + try: + await delete_node_client(node_id) + except DynsecError as e: + logger.error(f"dynsec reconcile: failed to revoke rejected node {node_id!r}: {e}") + nodes = await fstore.collection_list("nodes", approval_status="approved") if not nodes: return diff --git a/drb-c2-core/app/internal/gemini.py b/drb-c2-core/app/internal/gemini.py index a1c577c..5966d27 100644 --- a/drb-c2-core/app/internal/gemini.py +++ b/drb-c2-core/app/internal/gemini.py @@ -44,7 +44,18 @@ def _get_client(): with _client_lock: if _client is None: from google import genai # lazy — only when a Gemini call is made - _client = genai.Client(api_key=settings.gemini_api_key) + from google.genai import types + # server-26#153: google-genai's default is no timeout at all, so a + # hung call held its pipeline (and the sweep hold-back sized + # against it) open indefinitely. HttpOptions.timeout is in ms. + # The client does not retry unless retry_options is set, so this + # is the whole bound per request. A timeout raises from + # generate_json like any API error, so every caller's fallback + # (rules-only correlation, uncorrected transcript) is unchanged. + _client = genai.Client( + api_key=settings.gemini_api_key, + http_options=types.HttpOptions(timeout=int(settings.gemini_timeout_seconds * 1000)), + ) return _client diff --git a/drb-c2-core/app/internal/intelligence.py b/drb-c2-core/app/internal/intelligence.py index 30daf5e..cef7cbc 100644 --- a/drb-c2-core/app/internal/intelligence.py +++ b/drb-c2-core/app/internal/intelligence.py @@ -938,7 +938,7 @@ def _sync_extract( ) -> list[dict]: """Call GPT-4o-mini and return a list of scene dicts.""" from app.config import settings - from openai import OpenAI + from app.internal.ai_clients import openai_client if not settings.openai_api_key: logger.warning("OPENAI_API_KEY not set — intelligence extraction disabled.") @@ -956,7 +956,7 @@ def _sync_extract( ) try: - client = OpenAI(api_key=settings.openai_api_key) + client = openai_client() response = client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": prompt}], @@ -985,13 +985,13 @@ def _sync_extract( def _sync_embed(text: str) -> Optional[list[float]]: """Generate a text-embedding-3-small vector for semantic similarity.""" from app.config import settings - from openai import OpenAI + from app.internal.ai_clients import openai_client if not settings.openai_api_key: return None try: - client = OpenAI(api_key=settings.openai_api_key) + client = openai_client() result = client.embeddings.create(model="text-embedding-3-small", input=text) return result.data[0].embedding except Exception as e: diff --git a/drb-c2-core/app/internal/ny511.py b/drb-c2-core/app/internal/ny511.py index e8708e7..6a940ca 100644 --- a/drb-c2-core/app/internal/ny511.py +++ b/drb-c2-core/app/internal/ny511.py @@ -9,6 +9,25 @@ 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 @@ -19,12 +38,16 @@ import httpx from app.config import settings from app.internal.logger import logger +from app.internal.cameras.base import safe_error -_BASE = "https://511ny.org/api" +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 -_DESCRIPTION_MAX = 500 +# 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: @@ -51,7 +74,7 @@ class _Feed: 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] + 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 @@ -61,9 +84,22 @@ async def _fetch(path: str, normalize) -> List[Dict[str, Any]]: 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) + base = (settings.ny511_api_base or LEGACY_BASE).rstrip("/") + r = await client.get(f"{base}/{path}", params=params) r.raise_for_status() - raw = r.json() + 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] @@ -73,6 +109,69 @@ async def _fetch(path: str, normalize) -> List[Dict[str, Any]]: 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"]) diff --git a/drb-c2-core/app/internal/recorrelation_sweep.py b/drb-c2-core/app/internal/recorrelation_sweep.py index cdbc5a1..a4d9fed 100644 --- a/drb-c2-core/app/internal/recorrelation_sweep.py +++ b/drb-c2-core/app/internal/recorrelation_sweep.py @@ -14,6 +14,7 @@ Never creates new incidents — link-only. Zero LLM tokens (uses pre-computed talkgroup strings, haversine math, and stored embeddings). """ import asyncio +import math from datetime import datetime, timezone, timedelta from typing import Optional from app.internal.logger import logger @@ -33,16 +34,54 @@ from app.config import settings # itself failed) is NOT held back by this — absence isn't evidence of an # in-flight pipeline, and #131's own bug predates this field existing. # -# 15, not 5: neither OpenAI's Whisper client nor Gemini's call in -# llm_correlator.py sets a request timeout (server-26#153), so a hung call can -# run well past a few minutes on SDK-default retries, and this constant is a -# guess against that unbounded tail, not a measured bound. Raising it costs -# nothing on the recovery side: a call that finished processing (linked OR -# genuinely orphaned) always has corr_path set (_apply_and_log writes it even -# on the orphan action), so it's already excluded by the -# `not c.get("corr_path")` filter below and never reaches this check at all — -# this constant only ever delays calls that are still actually running. -MIN_MINUTES_SINCE_PIPELINE_START = 15 +# Derived, not guessed (server-26#153). Every AI request the real-time +# pipeline makes now has an explicit timeout (config.py: stt_timeout_seconds, +# openai_timeout_seconds, openai_max_retries, gemini_timeout_seconds), so the +# longest a call can stay "started but no corr_path yet" has an upper bound: +# pipeline_worst_case_seconds() below. The hold-back is that bound rounded up, +# never below the old hand-picked 15. +# +# Raising it costs nothing on the recovery side: a call that finished +# processing (linked OR genuinely orphaned) always has corr_path set +# (_apply_and_log writes it even on the orphan action), so it's already +# excluded by the `not c.get("corr_path")` filter below and never reaches this +# check at all — this constant only ever delays calls that are still actually +# running, or whose pipeline died after writing the marker. + +# Scenes whose embeddings are computed before the first scene's correlation +# commits (intelligence.extract_scenes embeds every scene, then upload.py +# correlates them in order; scene 0's commit writes corr_path). Extraction +# rarely returns more than 2; 3 is the budget, not a hard cap. +_BOUND_SCENES = 3 +# Non-AI work on the same path: GCS download, Firestore reads/writes, geocoding +# and place verification (httpx, 5s timeouts each). +_NON_AI_SLACK_SECONDS = 120.0 +_FLOOR_MINUTES = 15 + + +def pipeline_worst_case_seconds(n_scenes: int = _BOUND_SCENES) -> float: + """Upper bound on intelligence_started_at -> first corr_path write, from + the configured AI timeouts. Order follows routers/upload.py + _run_intelligence_pipeline.""" + openai_attempts = 1 + max(0, settings.openai_max_retries) + stt_rounds = 2 if settings.stt_retry_on_degenerate else 1 + stt = stt_rounds * openai_attempts * settings.stt_timeout_seconds + # Transcript correction (inside transcribe_call). gemini.generate_json only + # re-issues a request when the model rejects a thinking level — a fast + # 400, not a timeout — and google-genai itself does not retry. + correction = settings.gemini_timeout_seconds + extraction = openai_attempts * settings.openai_timeout_seconds + embeddings = n_scenes * openai_attempts * settings.openai_timeout_seconds + correlation = 2 * settings.gemini_timeout_seconds # decide + tiebreak, scene 0 + return stt + correction + extraction + embeddings + correlation + _NON_AI_SLACK_SECONDS + + +def _derive_min_minutes() -> int: + return max(_FLOOR_MINUTES, math.ceil(pipeline_worst_case_seconds() / 60)) + + +# 21 with the defaults (stt 120s, openai 60s x2 attempts, gemini 60s, 3 scenes). +MIN_MINUTES_SINCE_PIPELINE_START = _derive_min_minutes() # Standard link-only retry budget before a call is tombstoned corr_path="unlinked". MAX_SWEEP_ATTEMPTS = 3 @@ -66,8 +105,17 @@ async def recorrelation_loop() -> None: logger.info( f"Re-correlation sweep started — " f"interval: {settings.summary_interval_minutes}m, " - f"scan window: {settings.recorrelation_scan_minutes}m" + f"scan window: {settings.recorrelation_scan_minutes}m, " + f"pipeline hold-back: {MIN_MINUTES_SINCE_PIPELINE_START}m" ) + if MIN_MINUTES_SINCE_PIPELINE_START >= settings.recorrelation_scan_minutes: + # A call is held back longer than the window that finds it, so the + # sweep can never recover a call whose pipeline died mid-flight. + logger.warning( + f"Re-correlation sweep: pipeline hold-back ({MIN_MINUTES_SINCE_PIPELINE_START}m) " + f">= scan window ({settings.recorrelation_scan_minutes}m) — lower the AI " + f"timeouts or raise RECORRELATION_SCAN_MINUTES" + ) while True: await asyncio.sleep(interval) try: diff --git a/drb-c2-core/app/internal/summarizer.py b/drb-c2-core/app/internal/summarizer.py index acb6ebe..4361391 100644 --- a/drb-c2-core/app/internal/summarizer.py +++ b/drb-c2-core/app/internal/summarizer.py @@ -213,7 +213,7 @@ async def _resolve_stale_incidents() -> None: def _sync_summarize(inc: dict, transcripts: list[str]) -> Optional[str]: from app.config import settings - from openai import OpenAI + from app.internal.ai_clients import openai_client if not settings.openai_api_key: return None @@ -241,7 +241,7 @@ Write a concise factual summary of this incident in 2-4 sentences. Include: Be factual. Do not speculate beyond what the transcripts say. Do not use bullet points.""" try: - client = OpenAI(api_key=settings.openai_api_key) + client = openai_client() response = client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": prompt}], diff --git a/drb-c2-core/app/internal/transcription.py b/drb-c2-core/app/internal/transcription.py index fc7a604..7218bb6 100644 --- a/drb-c2-core/app/internal/transcription.py +++ b/drb-c2-core/app/internal/transcription.py @@ -271,8 +271,8 @@ def _sync_transcribe( """ from google.cloud import storage as gcs from google.oauth2 import service_account - from openai import OpenAI from app.config import settings + from app.internal.ai_clients import openai_client if not settings.openai_api_key: logger.warning("OPENAI_API_KEY not set — transcription disabled.") @@ -315,9 +315,11 @@ def _sync_transcribe( # gpt-4o-transcribe and gpt-4o-mini-transcribe only support json/text. use_verbose = settings.stt_model == "whisper-1" - openai_client = OpenAI(api_key=settings.openai_api_key) + # Own timeout, not openai_timeout_seconds: this request uploads the + # whole recording (server-26#153). + stt_client = openai_client(timeout=settings.stt_timeout_seconds) with open(tmp_path, "rb") as f: - response = openai_client.audio.transcriptions.create( + response = stt_client.audio.transcriptions.create( model=settings.stt_model, file=f, language="en", diff --git a/drb-c2-core/app/internal/vocabulary_learner.py b/drb-c2-core/app/internal/vocabulary_learner.py index fa90dc1..790f99a 100644 --- a/drb-c2-core/app/internal/vocabulary_learner.py +++ b/drb-c2-core/app/internal/vocabulary_learner.py @@ -470,7 +470,7 @@ def _diff_new_terms(original: str, corrected: str) -> list[str]: def _sync_bootstrap(system_name: str, system_type: str, area_hint: str) -> list[str]: from app.config import settings as cfg - from openai import OpenAI + from app.internal.ai_clients import openai_client if not cfg.openai_api_key: return [] @@ -481,7 +481,7 @@ def _sync_bootstrap(system_name: str, system_type: str, area_hint: str) -> list[ area_hint=area_hint, ) try: - client = OpenAI(api_key=cfg.openai_api_key) + client = openai_client() response = client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": prompt}], @@ -504,7 +504,7 @@ def _sync_induct( ) -> list[dict]: """Returns [{term, meaning}] — a bare string is still accepted from the model.""" from app.config import settings as cfg - from openai import OpenAI + from app.internal.ai_clients import openai_client if not cfg.openai_api_key: return [] @@ -518,7 +518,7 @@ def _sync_induct( transcript_block=transcript_block[:8000], ) try: - client = OpenAI(api_key=cfg.openai_api_key) + client = openai_client() response = client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": prompt}], diff --git a/drb-c2-core/app/main.py b/drb-c2-core/app/main.py index 02ed503..4725942 100644 --- a/drb-c2-core/app/main.py +++ b/drb-c2-core/app/main.py @@ -18,7 +18,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, traffic +from app.routers import enrollment, media, org, waitlist, telemetry, replay, traffic, cameras from app.routers import discord as discord_router from app.internal import dynsec from app.internal import firestore as fstore @@ -142,6 +142,8 @@ app.include_router(alerts.router, dependencies=[Depends(require_service_or_fi 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(cameras.router, dependencies=[Depends(require_service_or_firebase_token)]) +app.include_router(cameras.admin_router) # auth: admin only, per endpoint (needs the principal for audit) 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/cameras.py b/drb-c2-core/app/routers/cameras.py new file mode 100644 index 0000000..8596287 --- /dev/null +++ b/drb-c2-core/app/routers/cameras.py @@ -0,0 +1,51 @@ +"""Public-camera endpoints for the live ops wall. + +``GET /cameras`` -- any signed-in user (auth applied at include time in main.py): +cameras inside a bbox from every provider an admin has enabled, each with the +provider's required attribution, disclaimer and catalog freshness. + +``/admin/camera-providers`` -- platform admin only: list every catalogued +provider with its licence status and health, and toggle it. Gated licences +need ``acknowledge_licence: true``; that acknowledgement is stored with who and +when and written to audit_log (see app/internal/cameras/registry.py). +""" +from fastapi import APIRouter, Depends, HTTPException, Query +from fastapi.responses import JSONResponse +from pydantic import BaseModel + +from app.internal.auth import describe_actor, require_admin_token +from app.internal.cameras import registry + +router = APIRouter(prefix="/cameras", tags=["cameras"]) +admin_router = APIRouter(prefix="/admin/camera-providers", tags=["admin"]) + + +@router.get("") +async def get_cameras( + 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), +): + if south > north or west > east: + raise HTTPException(400, "bbox must satisfy south<=north and west<=east") + return await registry.cameras_in_bbox((south, west, north, east)) + + +class ProviderToggle(BaseModel): + enabled: bool + acknowledge_licence: bool = False + + +@admin_router.get("") +async def list_camera_providers(_=Depends(require_admin_token)): + return {"providers": await registry.list_providers()} + + +@admin_router.put("/{provider_id}") +async def set_camera_provider(provider_id: str, body: ProviderToggle, principal: dict = Depends(require_admin_token)): + try: + return await registry.set_enabled(provider_id, body.enabled, describe_actor(principal), + acknowledge_licence=body.acknowledge_licence) + except registry.ProviderError as e: + return JSONResponse(status_code=e.code, content={"detail": e.message, **e.extra}) diff --git a/drb-c2-core/app/routers/nodes.py b/drb-c2-core/app/routers/nodes.py index e3124ce..84190d5 100644 --- a/drb-c2-core/app/routers/nodes.py +++ b/drb-c2-core/app/routers/nodes.py @@ -119,11 +119,36 @@ async def delete_node(node_id: str, _: dict = Depends(require_admin_token)): @router.post("/{node_id}/reject") async def reject_node(node_id: str, _: dict = Depends(require_admin_token)): + """Reject = revoke (server-26#195). Before this, reject only flipped + ``approval_status``, so an approved-then-rejected node kept its MQTT login + (dynsec client), its ``/upload`` key and ``/systems`` read access (both + keyed off ``node_keys``). + + Order is fail-closed, the opposite of approve's: the status flips to + ``rejected`` FIRST, so even if the broker call fails, ``/upload`` (which + now checks approval_status) and ``GET /nodes/{id}/credentials`` already + refuse the node, and ``dynsec.reconcile_all()`` will no longer re-create + its login on the next c2-core restart (it also deletes rejected nodes' + logins there). Then the broker login is deleted, then the key. A dynsec + failure returns 502 with node_keys left intact, and the admin retries: + every step is idempotent. Re-approving later mints a fresh key, because + node_keys is gone; the revoked key is never handed back. + """ node = await fstore.doc_get("nodes", node_id) if not node: raise HTTPException(404, f"Node '{node_id}' not found.") await fstore.doc_update("nodes", node_id, {"approval_status": "rejected"}) - return {"ok": True} + try: + await dynsec.delete_node_client(node_id) + except dynsec.DynsecError as e: + logger.error(f"Reject {node_id!r}: dynsec deleteClient failed, node_keys NOT deleted: {e}") + raise HTTPException( + 502, + f"Node marked rejected, but its MQTT login could not be revoked: {e}. Retry reject.", + ) + await fstore.doc_delete("node_keys", node_id) + logger.info(f"Reject {node_id!r}: status rejected, MQTT login and API key revoked") + return {"ok": True, "revoked": True} @router.post("/{node_id}/command") diff --git a/drb-c2-core/app/routers/traffic.py b/drb-c2-core/app/routers/traffic.py index da80ffa..e8eb352 100644 --- a/drb-c2-core/app/routers/traffic.py +++ b/drb-c2-core/app/routers/traffic.py @@ -4,6 +4,7 @@ from typing import Optional from fastapi import APIRouter, HTTPException, Query from app.internal import ny511 +from app.internal.cameras import registry router = APIRouter(prefix="/traffic", tags=["traffic"]) @@ -27,6 +28,13 @@ async def get_511( 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()} + # 511NY data (events included) is covered by the same NYSDOT DAA as its + # cameras, so the whole feed follows the ny511 camera provider's switch, + # which defaults OFF and needs a licence acknowledgement (server-26#198). + if not await registry.is_serving("ny511"): + off = {"fetched_at": None, "error": None, "total_in_bbox": 0, "truncated": False, "disabled": True} + return {**{n: [] for n in ("cameras", "events") if n in wanted}, + **{f"{n}_status": off for n in ("cameras", "events") if n in wanted}} feeds = {name: getattr(ny511, name) for name in ("cameras", "events") if name in wanted} await asyncio.gather(*(f.get() for f in feeds.values())) @@ -35,4 +43,5 @@ async def get_511( 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} + out["credit"] = registry.credit("ny511") # DAA: source credit + as-is disclaimer travel with the data return out diff --git a/drb-c2-core/app/routers/trips.py b/drb-c2-core/app/routers/trips.py index 4f72e05..d7eef96 100644 --- a/drb-c2-core/app/routers/trips.py +++ b/drb-c2-core/app/routers/trips.py @@ -531,8 +531,8 @@ async def trip_chat( events = await fstore.collection_list("trip_events", trip_id=trip_id) - from openai import AsyncOpenAI - oai = AsyncOpenAI(api_key=settings.openai_api_key) + from app.internal.ai_clients import async_openai_client + oai = async_openai_client() # Strip history to only user/assistant roles to prevent prompt injection safe_history = [ diff --git a/drb-c2-core/app/routers/upload.py b/drb-c2-core/app/routers/upload.py index e142f66..43b72b5 100644 --- a/drb-c2-core/app/routers/upload.py +++ b/drb-c2-core/app/routers/upload.py @@ -47,6 +47,18 @@ async def upload_call_audio( f"(received prefix: {credentials.credentials[:8]}...)" ) raise HTTPException(401, "Invalid node API key") + # server-26#195: a valid key is not enough, the node must still be + # approved. reject_node() deletes node_keys too, but this closes the + # window where that delete failed or was never run (a node rejected before + # reject revoked anything). Uncached read: revocation must bite on the + # next upload, not after doc_get_cached's 5-minute TTL. + node = await fstore.doc_get("nodes", node_id) + if not node or node.get("approval_status") != "approved": + logger.warning( + f"Upload 403: node_id={node_id!r} has a valid key but approval_status=" + f"{(node or {}).get('approval_status')!r}" + ) + raise HTTPException(403, "Node is not approved") data = await file.read() if not data: @@ -63,10 +75,10 @@ async def upload_call_audio( # org_id is stamped defensively here too (not just in # mqtt_handler.py's call_start/call_end): key_doc above proves this # node_id is real and authenticated, so resolving org_id from the - # node doc here covers a call whose Firestore doc was somehow - # never written by call_start (the upload is otherwise the - # authoritative record of which node this audio came from). - node = await fstore.doc_get_cached("nodes", node_id) + # node doc (fetched by the approval gate above) here covers a call + # whose Firestore doc was somehow never written by call_start (the + # upload is otherwise the authoritative record of which node this + # audio came from). updates = {"audio_gcs_uri": gcs_uri} if node and node.get("org_id"): updates["org_id"] = node["org_id"] diff --git a/drb-c2-core/tests/test_ai_timeouts.py b/drb-c2-core/tests/test_ai_timeouts.py new file mode 100644 index 0000000..ffb4b56 --- /dev/null +++ b/drb-c2-core/tests/test_ai_timeouts.py @@ -0,0 +1,246 @@ +""" +server-26#153 — no OpenAI or Gemini request ran with an explicit timeout. + +OpenAI's SDK waits 600s per attempt and retries twice (timeouts included); +google-genai waits forever. A hung provider call therefore had no upper bound, +and recorrelation_sweep's MIN_MINUTES_SINCE_PIPELINE_START was a guess against +that tail. These tests pin: + - every client is built with the configured timeout / retry budget, + - a hung request really does give up (real SDK, local socket that never answers), + - a timeout takes the same fallback path as any other provider error, + - the sweep hold-back is derived from the timeouts, not hand-picked. +""" +import pathlib +import re +import socket +import threading +import time +from unittest.mock import AsyncMock, patch + +import httpx +import openai +import pytest + +from app.config import settings +from app.internal import ai_clients, ai_health, gemini, llm_correlator, recorrelation_sweep +from app.internal import intelligence, summarizer, transcript_correction + +APP_DIR = pathlib.Path(__file__).resolve().parent.parent / "app" + + +def _timeout_error() -> openai.APITimeoutError: + return openai.APITimeoutError(request=httpx.Request("POST", "https://api.openai.com/v1/x")) + + +@pytest.fixture(autouse=True) +def _reset_ai_health(): + ai_health._state = {t: ai_health._default_state() for t in ai_health.TIERS} + yield + ai_health._state = {t: ai_health._default_state() for t in ai_health.TIERS} + + +@pytest.fixture +def openai_key(): + with patch.object(settings, "openai_api_key", "sk-test"): + yield + + +# ── Every client carries the configured limits ────────────────────────────── + +def test_openai_client_uses_configured_timeout_and_retries(openai_key): + with patch.object(settings, "openai_timeout_seconds", 42.0), \ + patch.object(settings, "openai_max_retries", 1): + client = ai_clients.openai_client() + assert client.timeout == 42.0 + assert client.max_retries == 1 + + +def test_openai_client_timeout_override_for_stt(openai_key): + client = ai_clients.openai_client(timeout=settings.stt_timeout_seconds) + assert client.timeout == settings.stt_timeout_seconds + + +def test_async_openai_client_uses_configured_limits(openai_key): + with patch.object(settings, "openai_timeout_seconds", 7.0), \ + patch.object(settings, "openai_max_retries", 0): + client = ai_clients.async_openai_client() + assert client.timeout == 7.0 + assert client.max_retries == 0 + + +def test_defaults_are_not_the_sdk_defaults(): + # SDK defaults: OpenAI 600s x 3 attempts, Gemini unbounded. + assert settings.stt_timeout_seconds < 600 + assert settings.openai_timeout_seconds < 600 + assert settings.openai_max_retries < openai.DEFAULT_MAX_RETRIES + assert settings.gemini_timeout_seconds > 0 + + +def test_gemini_client_sets_timeout_in_milliseconds(): + saved = gemini._client + gemini._client = None + try: + with patch.object(settings, "gemini_api_key", "test-key"), \ + patch.object(settings, "gemini_timeout_seconds", 45.0): + client = gemini._get_client() + assert client._api_client._http_options.timeout == 45_000 + finally: + gemini._client = saved + + +def test_no_openai_client_is_built_outside_ai_clients(): + """Regression guard: a new `OpenAI(...)` elsewhere would silently run on + the 600s SDK default again.""" + offenders = [] + for path in APP_DIR.rglob("*.py"): + if path.name == "ai_clients.py": + continue + for n, line in enumerate(path.read_text().splitlines(), 1): + if re.search(r"\b(Async)?OpenAI\(", line) or re.search(r"genai\.Client\(", line): + if path.name == "gemini.py" and "genai.Client(" in line: + continue + offenders.append(f"{path.relative_to(APP_DIR)}:{n}: {line.strip()}") + assert offenders == [] + + +# ── A hung request actually gives up ──────────────────────────────────────── + +@pytest.fixture +def hung_server(): + """A TCP server that accepts connections and never sends a byte.""" + srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + srv.bind(("127.0.0.1", 0)) + srv.listen(8) + conns = [] + stop = threading.Event() + + def accept(): + srv.settimeout(0.1) + while not stop.is_set(): + try: + conns.append(srv.accept()[0]) + except OSError: + continue + + t = threading.Thread(target=accept, daemon=True) + t.start() + yield f"http://127.0.0.1:{srv.getsockname()[1]}/v1" + stop.set() + t.join(timeout=2) + for c in conns: + c.close() + srv.close() + + +def test_hung_openai_request_times_out_instead_of_waiting_600s(openai_key, hung_server, monkeypatch): + monkeypatch.setenv("OPENAI_BASE_URL", hung_server) + with patch.object(settings, "openai_timeout_seconds", 0.3), \ + patch.object(settings, "openai_max_retries", 1): + client = ai_clients.openai_client() + t0 = time.monotonic() + with pytest.raises(openai.APITimeoutError): + client.chat.completions.create(model="gpt-4o-mini", messages=[{"role": "user", "content": "x"}]) + elapsed = time.monotonic() - t0 + # 2 attempts x 0.3s + SDK backoff (<~1s); far below the 600s default. + assert elapsed < 5 + + +# ── Timeouts degrade exactly like other provider errors ──────────────────── + +def test_timeout_messages_classify_as_transient(): + assert ai_health.classify(str(_timeout_error())) == "transient" + assert ai_health.classify(str(httpx.ReadTimeout("The read operation timed out"))) == "transient" + + +def _corr_ctx(): + return {"call_id": "c1", "is_thin_call": False, "recent": [{"incident_id": "i1"}], + "all_active": [], "call_units": ["12"], "call_cleared": [], "tags": ["mva"]} + + +@pytest.mark.asyncio +async def test_gemini_timeout_in_decide_falls_back_to_rules(): + def hang(model, prompt): + raise httpx.ReadTimeout("The read operation timed out") + + with patch.object(settings, "gemini_api_key", "test-key"), \ + patch.object(llm_correlator, "_build_decide_prompt", lambda ctx: "p"), \ + patch.object(llm_correlator, "_sync_gemini", hang): + assert await llm_correlator.decide("c1", _corr_ctx()) is None # None == rules_only + state = ai_health.snapshot()["correlation_cheap"] + assert state["consecutive_failures"] == 1 + assert state["permanent"] is False + + +@pytest.mark.asyncio +async def test_gemini_timeout_in_tiebreak_keeps_the_rules_decision(): + def hang(model, prompt): + raise httpx.ReadTimeout("The read operation timed out") + + rules = {"action": "new", "matched_incident": None, "corr_debug": {}} + with patch.object(llm_correlator, "_build_tiebreak_prompt", lambda r, l, c: "p"), \ + patch.object(llm_correlator, "_sync_gemini", hang): + out = await llm_correlator.tiebreak(rules, {"action": "orphan"}, _corr_ctx()) + assert out is rules + assert ai_health.snapshot()["correlation_smart"]["permanent"] is False + + +@pytest.mark.asyncio +async def test_gemini_timeout_in_transcript_correction_keeps_raw_text(): + def hang(model, prompt): + raise httpx.ReadTimeout("The read operation timed out") + + text = "engine 5 respond to a structure fire at 12 main street smoke showing" + with patch.object(settings, "gemini_api_key", "test-key"), \ + patch.object(transcript_correction, "resolve_context", AsyncMock(return_value={})), \ + patch.object(transcript_correction, "_sync_gemini", hang): + assert await transcript_correction.correct("c1", text) == (None, None, False) + + +class _TimingOutOpenAI: + """Stands in for the client ai_clients builds; every endpoint times out.""" + def __init__(self, *a, **k): + self.chat = self.embeddings = self + self.completions = self + + def create(self, **kwargs): + raise _timeout_error() + + +def test_extraction_timeout_propagates_to_the_ai_health_path(openai_key): + # _sync_extract re-raises API failures so extract_scenes reports them to + # ai_health (tier "extraction"); a timeout must not be swallowed as "no scenes". + with patch.object(ai_clients, "openai_client", _TimingOutOpenAI): + with pytest.raises(openai.APITimeoutError): + intelligence._sync_extract("unit 12 on scene", "PD Dispatch", 1, "sys", None) + + +def test_embedding_timeout_returns_none(openai_key): + with patch.object(ai_clients, "openai_client", _TimingOutOpenAI): + assert intelligence._sync_embed("unit 12 on scene") is None + + +def test_summary_timeout_returns_none(openai_key): + with patch.object(ai_clients, "openai_client", _TimingOutOpenAI): + assert summarizer._sync_summarize({"type": "fire"}, ["engine 5 on scene"]) is None + + +# ── The sweep hold-back is derived from the timeouts ──────────────────────── + +def test_worst_case_bound_follows_the_settings(): + with patch.object(settings, "stt_timeout_seconds", 100.0), \ + patch.object(settings, "openai_timeout_seconds", 10.0), \ + patch.object(settings, "openai_max_retries", 1), \ + patch.object(settings, "gemini_timeout_seconds", 20.0), \ + patch.object(settings, "stt_retry_on_degenerate", True): + bound = recorrelation_sweep.pipeline_worst_case_seconds(n_scenes=2) + # stt 2x2x100 + correction 20 + extraction 2x10 + embeds 2x2x10 + decide/tiebreak 2x20 + slack + assert bound == 400 + 20 + 20 + 40 + 40 + recorrelation_sweep._NON_AI_SLACK_SECONDS + + +def test_hold_back_covers_the_worst_case_and_never_drops_below_old_floor(): + bound_min = recorrelation_sweep.pipeline_worst_case_seconds() / 60 + assert recorrelation_sweep.MIN_MINUTES_SINCE_PIPELINE_START >= bound_min + assert recorrelation_sweep.MIN_MINUTES_SINCE_PIPELINE_START >= 15 + # ...and stays inside the scan window, or the sweep could never recover a + # call whose pipeline died after writing intelligence_started_at. + assert recorrelation_sweep.MIN_MINUTES_SINCE_PIPELINE_START < settings.recorrelation_scan_minutes diff --git a/drb-c2-core/tests/test_camera_providers.py b/drb-c2-core/tests/test_camera_providers.py new file mode 100644 index 0000000..dfa91f8 --- /dev/null +++ b/drb-c2-core/tests/test_camera_providers.py @@ -0,0 +1,312 @@ +"""Live ops wall camera providers: registry, licence gate, admin routes, adapters. + +What is held here: + * every provider defaults OFF, and an off provider is never fetched; + * stubs and PROHIBITED sources can never be enabled; + * a gated licence needs ``acknowledge_licence``, which is stored with who/when + and audited -- and an audit failure refuses the enable (fail closed); + * an acknowledgement given for a different licence status no longer counts; + * the admin routes are admin-only; + * each implemented adapter parses its (verified) upstream shape, pins imagery + to the agency's hosts, and treats an all-unparseable payload as an error. +""" +import asyncio +from unittest.mock import AsyncMock, patch + +import pytest +from fastapi import HTTPException +from fastapi.testclient import TestClient + +from app.internal.auth import require_admin_token, require_service_or_firebase_token +from app.internal.cameras import base, general, regional, registry +from app.internal.cameras.base import Licence +from app.main import app + +client = TestClient(app) +ADMIN = {"uid": "admin-1", "email": "admin@example.com", "role": "admin"} + + +class FakeStore: + def __init__(self): + self.docs = {} + self.audit = [] + + async def doc_get(self, col, doc): + return self.docs.get((col, doc)) + + async def doc_set(self, col, doc, data, merge=True): + cur = self.docs.setdefault((col, doc), {}) + if col == "audit_log": + self.audit.append(data) + return + for k, v in data.items(): # one-level deep merge is all registry needs + if merge and isinstance(v, dict) and isinstance(cur.get(k), dict): + cur[k] = {**cur[k], **v} + else: + cur[k] = v + + +@pytest.fixture(autouse=True) +def store(): + s = FakeStore() + registry.invalidate_cache() + for cat in registry.CATALOGS.values(): + cat.items, cat.fetched_at, cat.error, cat._next_attempt = [], None, None, 0.0 + with patch.object(registry.fstore, "doc_get", s.doc_get), patch.object(registry.fstore, "doc_set", s.doc_set), \ + patch("app.internal.audit.fstore.doc_set", s.doc_set): + yield s + registry.invalidate_cache() + app.dependency_overrides.clear() + + +def run(coro): + return asyncio.run(coro) + + +# ── registry / gate ────────────────────────────────────────────────────────── + +def test_every_catalogued_source_is_registered_and_defaults_off(): + listed = run(registry.list_providers()) + ids = {p["id"] for p in listed} + assert {"ny511", "nyc_dot", "westchester_dpw", "caltrans", "deldot", "tfl", "ontario511", "drivebc", + "calgary", "nsw", "fintraffic", "txdot", "austin", "tallinn", "tarktee", "warendorf"} <= ids + assert all(p["enabled"] is False and p["serving"] is False for p in listed) + assert all(p["licence"]["label"] for p in listed) + + +def test_off_provider_is_never_fetched(): + fetch = AsyncMock(return_value=[]) + with patch.object(registry.PROVIDERS["tfl"], "fetch_catalog", fetch): + out = run(registry.cameras_in_bbox((51.4, -0.2, 51.6, 0.1))) + assert out["cameras"] == [] and out["providers"] == [] and fetch.await_count == 0 + + +def test_stub_and_prohibited_cannot_be_enabled(): + for pid in ("nyc_dot", "westchester_dpw", "ctroads", "trafficland"): + with pytest.raises(registry.ProviderError) as e: + run(registry.set_enabled(pid, True, ("u", "e"), acknowledge_licence=True)) + assert e.value.code == 409 + + +def test_attribution_licence_enables_without_acknowledgement_and_is_audited(store): + out = run(registry.set_enabled("nsw", True, ("u1", "a@x"))) + assert out["serving"] is True and out["acknowledgement"] is None + assert store.audit[-1]["action"] == "camera_provider.enable" + assert store.audit[-1]["actor_email"] == "a@x" + + +def test_gated_licence_requires_acknowledgement(store): + with pytest.raises(registry.ProviderError) as e: + run(registry.set_enabled("ny511", True, ("u1", "a@x"))) + assert e.value.code == 409 and e.value.extra["requires_acknowledgement"] is True + assert "Developer Access Agreement" in e.value.extra["warning"] + assert not store.audit and not run(registry.is_serving("ny511")) + + out = run(registry.set_enabled("ny511", True, ("u1", "a@x"), acknowledge_licence=True)) + assert out["serving"] is True + ack = out["acknowledgement"] + assert ack["by_uid"] == "u1" and ack["by_email"] == "a@x" and ack["at"] + assert ack["licence_status"] == Licence.AGREEMENT_REQUIRED.value + assert store.audit[-1]["action"] == "camera_provider.enable_acknowledged" + assert store.audit[-1]["details"]["provider_id"] == "ny511" + + +def test_licences_not_read_first_hand_are_gated_but_keep_attribution(): + # TfL (terms page 403'd) and Calgary (per GEV only) are not confirmed, so an + # admin must acknowledge before they serve; their credit still renders. + for pid in ("tfl", "calgary", "ontario511"): + lic = registry.PROVIDERS[pid].meta.licence + assert lic.status == Licence.UNVERIFIED and lic.gated, pid + assert registry.PROVIDERS["tfl"].meta.licence.attribution == "Powered by TfL Open Data" + with pytest.raises(registry.ProviderError) as e: + run(registry.set_enabled("tfl", True, ("u", "e"))) + assert e.value.extra["requires_acknowledgement"] is True + + +def test_gated_enable_refused_when_audit_cannot_be_written(store): + with patch.object(registry, "_audit", AsyncMock(side_effect=RuntimeError("firestore down"))): + with pytest.raises(registry.ProviderError) as e: + run(registry.set_enabled("deldot", True, ("u", "e"), acknowledge_licence=True)) + assert e.value.code == 503 + assert not run(registry.is_serving("deldot")) + + +def test_disable_clears_acknowledgement(store): + run(registry.set_enabled("deldot", True, ("u", "e"), acknowledge_licence=True)) + out = run(registry.set_enabled("deldot", False, ("u", "e"))) + assert out["serving"] is False and out["acknowledgement"] is None + + +def test_stale_acknowledgement_stops_serving(): + p = registry.PROVIDERS["deldot"] + entry = {"enabled": True, "acknowledgement": {"licence_status": "non_commercial"}} + assert registry.effective(p, entry) == (False, "licence acknowledgement missing or out of date") + entry["acknowledgement"]["licence_status"] = p.meta.licence.status.value + assert registry.effective(p, entry) == (True, None) + + +def test_unreadable_settings_fail_closed(): + with patch.object(registry.fstore, "doc_get", AsyncMock(side_effect=RuntimeError("boom"))): + registry.invalidate_cache() + assert run(registry.get_settings()) == {} + + +def test_query_returns_attribution_and_skips_providers_outside_bbox(store): + run(registry.set_enabled("nsw", True, ("u", "e"))) + run(registry.set_enabled("tfl", True, ("u", "e"), acknowledge_licence=True)) + cam = base.make_camera("nsw", "x", -33.87, 151.2, name="Sydney", image_url="https://webcams.transport.nsw.gov.au/a.jpg") + nsw_fetch, tfl_fetch = AsyncMock(return_value=[cam]), AsyncMock(return_value=[]) + with patch.object(registry.PROVIDERS["nsw"], "fetch_catalog", nsw_fetch), \ + patch.object(registry.PROVIDERS["tfl"], "fetch_catalog", tfl_fetch): + out = run(registry.cameras_in_bbox((-34.0, 151.0, -33.7, 151.4))) + assert [c["id"] for c in out["cameras"]] == ["nsw:x"] + by_id = {p["id"]: p for p in out["providers"]} + assert by_id["nsw"]["attribution"] == "Transport for NSW, CC BY 4.0" + assert tfl_fetch.await_count == 0 # London never fetched for a Sydney bbox + # ...but its credit is still listed, for cameras the client holds from another view + assert by_id["tfl"]["attribution"] == "Powered by TfL Open Data" and by_id["tfl"]["total_in_bbox"] == 0 + assert out["serving_count"] == 2 + + +# ── admin routes ───────────────────────────────────────────────────────────── + +def test_admin_routes_require_admin(): + def deny(): + raise HTTPException(403, "Admin access required") + app.dependency_overrides[require_admin_token] = deny + assert client.get("/admin/camera-providers").status_code == 403 + assert client.put("/admin/camera-providers/nsw", json={"enabled": True}).status_code == 403 + assert client.get("/admin/camera-providers", headers={}).status_code == 403 + + +def test_admin_route_toggle_and_gate(store): + app.dependency_overrides[require_admin_token] = lambda: ADMIN + r = client.put("/admin/camera-providers/caltrans", json={"enabled": True}) + assert r.status_code == 409 and r.json()["requires_acknowledgement"] is True + r = client.put("/admin/camera-providers/caltrans", json={"enabled": True, "acknowledge_licence": True}) + assert r.status_code == 200 and r.json()["acknowledgement"]["by_email"] == "admin@example.com" + assert client.put("/admin/camera-providers/nope", json={"enabled": True}).status_code == 404 + listed = client.get("/admin/camera-providers").json()["providers"] + assert next(p for p in listed if p["id"] == "caltrans")["serving"] is True + + +def test_cameras_route_rejects_inverted_bbox(): + app.dependency_overrides[require_service_or_firebase_token] = lambda: {"uid": "v"} + assert client.get("/cameras", params={"south": 2, "west": 0, "north": 1, "east": 1}).status_code == 400 + r = client.get("/cameras", params={"south": 40.9, "west": -74.0, "north": 41.4, "east": -73.4}) + assert r.status_code == 200 and r.json()["cameras"] == [] + + +# ── adapters ───────────────────────────────────────────────────────────────── + +def _with_payload(payload): + return patch.object(general, "get_json", AsyncMock(return_value=payload)) + + +def test_heading_from_text(): + assert base.heading_from_text("Northbound") == 0 + assert base.heading_from_text("WB") == 270 + assert base.heading_from_text("SE") == 135 + assert base.heading_from_text("Both directions") is None + assert base.heading_from_text("Unknown") is None + + +def test_https_on_pins_hosts(): + assert base.https_on("https://cwwp2.dot.ca.gov/x.jpg", "dot.ca.gov") + assert base.https_on("http://cwwp2.dot.ca.gov/x.jpg", "dot.ca.gov") is None + assert base.https_on("https://dot.ca.gov.evil.com/x.jpg", "dot.ca.gov") is None + + +def test_caltrans_adapter(): + row = {"cctv": {"inService": "true", "index": "1", "location": { + "district": "4", "locationName": "TV102 -- I-580 : West of SR-24", "latitude": "37.82539", + "longitude": "-122.27291", "direction": "West", "route": "I-580"}, + "imageData": {"streamingVideoURL": "https://wzmedia.dot.ca.gov/D4/x.stream/playlist.m3u8", + "static": {"currentImageURL": "https://cwwp2.dot.ca.gov/data/d4/cctv/image/a/a.jpg"}}}} + off = {"cctv": {**row["cctv"], "inService": "false"}} + with _with_payload({"data": [row, off]}): + cams = run(general.CaltransProvider().fetch_catalog()) + assert len(cams) == len(general.CaltransProvider.DISTRICTS) # same row per district fetch + c = cams[0] + assert c["id"] == "caltrans:d4-TV102" and c["heading_deg"] == 270 and c["video_url"].endswith(".m3u8") + + +def test_deldot_adapter_requires_https_hls(): + good = {"id": "KCAM001", "title": "DE 1", "lat": 38.99, "lon": -75.44, "enabled": True, "status": "Active", + "urls": {"m3u8s": "https://video.deldot.gov:443/live/KCAM001.stream/playlist.m3u8", + "m3u8": "http://video.deldot.gov:1935/live/KCAM001.stream/playlist.m3u8"}} + down = {**good, "id": "KCAM002", "status": "Unavailable"} + with _with_payload({"videoCameras": [good, down]}): + cams = run(general.DelDotProvider().fetch_catalog()) + assert [c["id"] for c in cams] == ["deldot:KCAM001"] and cams[0]["image_url"] is None + + +def test_drivebc_calgary_tfl_nsw_fintraffic_adapters(): + with _with_payload([{"id": 1176, "name": "Lougheed", "is_on": True, "should_appear": True, "orientation": "W", + "location": {"coordinates": [-122.5, 49.2]}, "highway_description": "Hwy 7"}]): + assert run(general.DriveBcProvider().fetch_catalog())[0]["image_url"] == "https://www.drivebc.ca/images/1176.jpg" + with _with_payload([{"camera_url": {"url": "http://trafficcam.calgary.ca/loc86.jpg"}, "camera_location": "Stoney", + "point": {"coordinates": [-113.97, 50.9]}, "quadrant": "SE"}]): + c = run(general.CalgaryProvider().fetch_catalog())[0] + assert c["image_url"] == "https://trafficcam.calgary.ca/loc86.jpg" and c["id"] == "calgary:loc86" + with _with_payload([{"id": "JamCams_00002.00865", "commonName": "A406", "lat": 51.6, "lon": -0.016, + "additionalProperties": [{"key": "available", "value": "true"}, + {"key": "imageUrl", "value": general.TflProvider.IMAGE_PREFIX + "a.jpg"}]}]): + assert run(general.TflProvider().fetch_catalog())[0]["id"] == "tfl:00002.00865" + with _with_payload({"features": [{"id": "abc", "geometry": {"coordinates": [151.1, -34.0]}, + "properties": {"title": "5 Ways", "direction": "W", + "href": "https://webcams.transport.nsw.gov.au/x.jpeg"}}]}): + assert run(general.NswProvider().fetch_catalog())[0]["heading_deg"] == 270 + with _with_payload({"features": [{"geometry": {"coordinates": [23.99, 60.05, 0]}, + "properties": {"id": "C01503", "name": "kt51", "collectionStatus": "GATHERING", + "presets": [{"id": "C0150301", "inCollection": True}, + {"id": "C0150302", "inCollection": False}]}}]}): + cams = run(general.FintrafficProvider().fetch_catalog()) + assert [c["image_url"] for c in cams] == ["https://weathercam.digitraffic.fi/C0150301.jpg"] + + +def test_schema_change_is_an_error_not_an_empty_catalog(): + with _with_payload([{"renamed": True}]): + with pytest.raises(ValueError, match="schema"): + run(general.DriveBcProvider().fetch_catalog()) + + +def test_ontario_needs_its_key(monkeypatch): + monkeypatch.setattr(general.settings, "ontario511_api_key", None, raising=False) + with pytest.raises(RuntimeError, match="ONTARIO511_API_KEY"): + run(general.Ontario511Provider().fetch_catalog()) + + +def test_ny511_provider_wraps_legacy_client_and_carries_daa_metadata(): + row = {"id": "NYSDOT-1", "lat": 41.03, "lon": -73.76, "name": "I-287", "roadway": "I-287", + "direction": "Northbound", "image_url": "https://x/1", "video_url": "javascript:alert(1)"} + with patch.object(regional.ny511, "_fetch", AsyncMock(return_value=[row])): + cams = run(regional.Ny511Provider().fetch_catalog()) + assert cams[0]["id"] == "ny511:NYSDOT-1" and cams[0]["heading_deg"] == 0 + assert cams[0]["video_url"] is None # non-https never reaches the browser + lic = regional.Ny511Provider.meta.licence + assert lic.attribution == "powered by 511NY" and lic.attribution_url == "https://511ny.org" + assert "with all faults" in lic.disclaimer and lic.gated + + +def test_refresh_error_never_echoes_an_api_key(): + import httpx + req = httpx.Request("GET", "https://511on.ca/api/v2/get/cameras", params={"format": "json", "key": "SECRET123"}) + with pytest.raises(httpx.HTTPStatusError) as raised: # httpx's own message carries the full URL + httpx.Response(403, request=req).raise_for_status() + err = raised.value + cat = registry.CATALOGS["ontario511"] + with patch.object(cat.provider, "fetch_catalog", AsyncMock(side_effect=err)): + run(cat.get()) + assert "SECRET123" not in cat.error and "403" in cat.error and "511on.ca/api/v2/get/cameras" in cat.error + + +def test_failed_refresh_keeps_last_good_catalog(): + cat = registry.CATALOGS["nsw"] + cam = base.make_camera("nsw", "x", -33.8, 151.2) + with patch.object(cat.provider, "fetch_catalog", AsyncMock(return_value=[cam])): + run(cat.get()) + cat._next_attempt = 0.0 + with patch.object(cat.provider, "fetch_catalog", AsyncMock(side_effect=RuntimeError("down"))): + run(cat.get()) + assert cat.items == [cam] and "down" in cat.error and cat.fetched_at diff --git a/drb-c2-core/tests/test_node_reject_revoke.py b/drb-c2-core/tests/test_node_reject_revoke.py new file mode 100644 index 0000000..a2769cf --- /dev/null +++ b/drb-c2-core/tests/test_node_reject_revoke.py @@ -0,0 +1,203 @@ +""" +server-26#195 — reject must revoke: delete the node's dynsec (MQTT) login and +its node_keys entry, /upload must refuse a node that is not approved, and +dynsec.reconcile_all() must delete rejected nodes' logins on startup. +""" +import asyncio +from unittest.mock import AsyncMock, MagicMock, patch + +from fastapi.testclient import TestClient + +from app.main import app +from app.internal import dynsec +from app.internal.auth import require_admin_token, require_service_or_firebase_token +from app.routers import nodes, upload + +client = TestClient(app) + + +def setup_function(): + app.dependency_overrides[require_admin_token] = lambda: {"admin": True} + app.dependency_overrides[require_service_or_firebase_token] = lambda: {"admin": True} + + +def teardown_function(): + app.dependency_overrides.pop(require_admin_token, None) + app.dependency_overrides.pop(require_service_or_firebase_token, None) + + +def _fake_store(node, key_doc): + store = {"nodes": node, "node_keys": key_doc} + calls = [] + + async def doc_get(c, i): + return store.get(c) + + async def doc_update(c, i, data): + calls.append(("doc_update", c, data)) + store[c] = {**(store[c] or {}), **data} + + async def doc_delete(c, i): + calls.append(("doc_delete", c)) + store[c] = None + + async def doc_set(c, i, data, merge=False): + store[c] = data + + return store, calls, doc_get, doc_update, doc_delete, doc_set + + +# --------------------------------------------------------------------------- +# reject_node +# --------------------------------------------------------------------------- + +def test_reject_revokes_mqtt_login_and_key_in_fail_closed_order(): + store, calls, doc_get, doc_update, doc_delete, _ = _fake_store( + {"node_id": "n1", "approval_status": "approved"}, {"node_id": "n1", "api_key": "live-key"}) + + async def delete_client(node_id): + calls.append(("dynsec_delete", node_id)) + + with patch.object(nodes.fstore, "doc_get", doc_get), \ + patch.object(nodes.fstore, "doc_update", doc_update), \ + patch.object(nodes.fstore, "doc_delete", doc_delete), \ + patch.object(nodes.dynsec, "delete_node_client", delete_client): + resp = client.post("/nodes/n1/reject") + + assert resp.status_code == 200 + assert resp.json() == {"ok": True, "revoked": True} + assert calls == [ + ("doc_update", "nodes", {"approval_status": "rejected"}), + ("dynsec_delete", "n1"), + ("doc_delete", "node_keys"), + ] + assert store["node_keys"] is None + assert store["nodes"]["approval_status"] == "rejected" + + +def test_reject_dynsec_failure_is_502_and_keeps_key_but_status_is_already_rejected(): + store, calls, doc_get, doc_update, doc_delete, _ = _fake_store( + {"node_id": "n1", "approval_status": "approved"}, {"node_id": "n1", "api_key": "live-key"}) + + with patch.object(nodes.fstore, "doc_get", doc_get), \ + patch.object(nodes.fstore, "doc_update", doc_update), \ + patch.object(nodes.fstore, "doc_delete", doc_delete), \ + patch.object(nodes.dynsec, "delete_node_client", + AsyncMock(side_effect=dynsec.DynsecError("timeout"))): + resp = client.post("/nodes/n1/reject") + + assert resp.status_code == 502 + assert "Retry reject" in resp.json()["detail"] + assert store["nodes"]["approval_status"] == "rejected" + assert store["node_keys"] == {"node_id": "n1", "api_key": "live-key"} + assert ("doc_delete", "node_keys") not in calls + + +def test_reject_unknown_node_is_404_and_touches_nothing(): + with patch.object(nodes.fstore, "doc_get", AsyncMock(return_value=None)), \ + patch.object(nodes.fstore, "doc_update", AsyncMock()) as doc_update, \ + patch.object(nodes.fstore, "doc_delete", AsyncMock()) as doc_delete, \ + patch.object(nodes.dynsec, "delete_node_client", AsyncMock()) as delete_client: + resp = client.post("/nodes/ghost/reject") + assert resp.status_code == 404 + doc_update.assert_not_awaited() + doc_delete.assert_not_awaited() + delete_client.assert_not_awaited() + + +def test_reject_requires_admin(): + app.dependency_overrides.pop(require_admin_token, None) + resp = client.post("/nodes/n1/reject") + assert resp.status_code in (401, 403) + + +def test_reapprove_after_reject_mints_a_new_key_never_the_revoked_one(): + store, _, doc_get, doc_update, doc_delete, doc_set = _fake_store( + {"node_id": "n1", "approval_status": "approved"}, {"node_id": "n1", "api_key": "old-key"}) + + with patch.object(nodes.fstore, "doc_get", doc_get), \ + patch.object(nodes.fstore, "doc_update", doc_update), \ + patch.object(nodes.fstore, "doc_delete", doc_delete), \ + patch.object(nodes.fstore, "doc_set", doc_set), \ + patch.object(nodes.dynsec, "delete_node_client", AsyncMock()), \ + patch.object(nodes.dynsec, "upsert_node_client", AsyncMock()) as upsert, \ + patch.object(nodes.mqtt_handler, "publish_node_key", MagicMock()): + assert client.post("/nodes/n1/reject").status_code == 200 + assert client.post("/nodes/n1/approve").status_code == 200 + + new_key = store["node_keys"]["api_key"] + assert new_key != "old-key" and len(new_key) == 64 + upsert.assert_awaited_once_with("n1", new_key) + + +# --------------------------------------------------------------------------- +# /upload approval gate +# --------------------------------------------------------------------------- + +def _upload(node_doc, key="live-key"): + store = {"node_keys": {"node_id": "n1", "api_key": "live-key"}, "nodes": node_doc} + + async def doc_get(c, i): + return store.get(c) + + with patch.object(upload.fstore, "doc_get", doc_get), \ + patch.object(upload, "upload_audio", AsyncMock(return_value=None)) as up: + resp = client.post( + "/upload", + data={"call_id": "c1", "node_id": "n1"}, + files={"file": ("a.mp3", b"\x00\x01", "audio/mpeg")}, + headers={"Authorization": f"Bearer {key}"}, + ) + return resp, up + + +def test_upload_refuses_valid_key_from_rejected_node(): + resp, up = _upload({"node_id": "n1", "approval_status": "rejected"}) + assert resp.status_code == 403 + up.assert_not_awaited() + + +def test_upload_refuses_valid_key_from_pending_or_missing_node(): + for node_doc in ({"node_id": "n1", "approval_status": "pending"}, {"node_id": "n1"}, None): + resp, up = _upload(node_doc) + assert resp.status_code == 403, node_doc + up.assert_not_awaited() + + +def test_upload_accepts_approved_node(): + resp, up = _upload({"node_id": "n1", "approval_status": "approved"}) + assert resp.status_code == 200 + up.assert_awaited_once() + + +def test_upload_wrong_key_still_401_before_approval_check(): + resp, up = _upload({"node_id": "n1", "approval_status": "approved"}, key="wrong") + assert resp.status_code == 401 + up.assert_not_awaited() + + +# --------------------------------------------------------------------------- +# dynsec.reconcile_all +# --------------------------------------------------------------------------- + +def test_reconcile_deletes_rejected_logins_and_still_syncs_approved(): + async def collection_list(c, **filters): + status = filters.get("approval_status") + return { + "rejected": [{"node_id": "bad1"}, {"node_id": "bad2"}], + "approved": [{"node_id": "good"}], + }[status] + + async def doc_get(c, i): + return {"node_id": i, "api_key": "k-" + i} + + delete_client = AsyncMock(side_effect=[dynsec.DynsecError("boom"), None]) + with patch.object(dynsec.fstore, "collection_list", collection_list), \ + patch.object(dynsec.fstore, "doc_get", doc_get), \ + patch.object(dynsec, "delete_node_client", delete_client), \ + patch.object(dynsec, "upsert_node_client", AsyncMock()) as upsert: + asyncio.run(dynsec.reconcile_all()) + + # one failed revoke does not stop the rest, nor the approved sync + assert [c.args[0] for c in delete_client.await_args_list] == ["bad1", "bad2"] + upsert.assert_awaited_once_with("good", "k-good") diff --git a/drb-c2-core/tests/test_ny511.py b/drb-c2-core/tests/test_ny511.py index ec86fdd..dc1d578 100644 --- a/drb-c2-core/tests/test_ny511.py +++ b/drb-c2-core/tests/test_ny511.py @@ -43,11 +43,14 @@ def test_normalize_camera_skips_disabled_blocked_and_zero_coords(): assert ny511.normalize_camera({**CAM, "Latitude": 0, "Longitude": 0}) is None -def test_normalize_event_parses_day_first_dates_and_truncates_description(): +def test_normalize_event_parses_day_first_dates_and_keeps_description_whole(): 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 + # DAA: "You may not alter the substance of a report" -- no 500-char cut any more. + assert e["description"] == EVENT["Description"] + long = ny511.normalize_event({**EVENT, "Description": "y" * (ny511._DESCRIPTION_MAX + 50)}) + assert len(long["description"]) == ny511._DESCRIPTION_MAX + 1 # pathological only def test_failed_refresh_keeps_last_good_data_and_reports_error(): @@ -80,14 +83,29 @@ def test_schema_change_is_an_error_not_an_empty_layer(): assert "schema" in str(e) +def _serving(value: bool): + from app.internal.cameras import registry + return patch.object(registry, "is_serving", AsyncMock(return_value=value)) + + +def test_endpoint_is_off_until_the_ny511_provider_is_enabled(): + with _serving(False): + r = client.get("/traffic/511", params={"south": 40.9, "west": -74.0, "north": 41.4, "east": -73.4}) + body = r.json() + assert r.status_code == 200 and body["cameras"] == [] and body["events"] == [] + assert body["events_status"]["disabled"] is True + + 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}) + with _serving(True): + 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 body["credit"]["attribution"] == "powered by 511NY" and "as is" in body["credit"]["disclaimer"] 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 @@ -96,3 +114,24 @@ def test_endpoint_filters_to_bbox_and_reports_status(): 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 + + +def test_xml_inventory_maps_onto_legacy_fields(): + xml = """ + NYSDOT-941.05-73.78 + I-287 at Exit 4I-287Eastbound + https://example.invalid/still.jpgfalse + true + """ + rows = ny511.parse_inventory_xml(xml) + cams = [ny511.normalize_camera(r) for r in rows] + assert cams[0]["id"] == "NYSDOT-9" and cams[0]["image_url"] == "https://example.invalid/still.jpg" + assert cams[1] is None # disabled flag honoured from XML text + + +def test_xml_parser_refuses_html_and_entities(): + import pytest + with pytest.raises(ValueError, match="HTML"): + ny511.parse_inventory_xml("511NY") + with pytest.raises(ValueError, match="DTD"): + ny511.parse_inventory_xml(']>&a;') diff --git a/drb-c2-core/tests/test_recorrelation_sweep.py b/drb-c2-core/tests/test_recorrelation_sweep.py index eeddfce..c187946 100644 --- a/drb-c2-core/tests/test_recorrelation_sweep.py +++ b/drb-c2-core/tests/test_recorrelation_sweep.py @@ -66,11 +66,14 @@ async def test_sweep_pass_skips_a_call_whose_pipeline_just_started(): "ended_at": _iso(now - timedelta(minutes=1)), "intelligence_started_at": _iso(now - timedelta(seconds=30)), } + # Relative to the hold-back, which is now derived from the AI timeouts + # (server-26#153) rather than a fixed 15 minutes. + past_hold = recorrelation_sweep.MIN_MINUTES_SINCE_PIPELINE_START + 4 genuinely_orphaned_call = { "call_id": "call-genuine-orphan", - "started_at": _iso(now - timedelta(minutes=20)), - "ended_at": _iso(now - timedelta(minutes=19)), - "intelligence_started_at": _iso(now - timedelta(minutes=19)), + "started_at": _iso(now - timedelta(minutes=past_hold + 1)), + "ended_at": _iso(now - timedelta(minutes=past_hold)), + "intelligence_started_at": _iso(now - timedelta(minutes=past_hold)), } async def fake_collection_where(collection, clauses): diff --git a/drb-frontend/app/admin/page.tsx b/drb-frontend/app/admin/page.tsx index 5b2cf44..6a744c9 100644 --- a/drb-frontend/app/admin/page.tsx +++ b/drb-frontend/app/admin/page.tsx @@ -3,6 +3,7 @@ import { useAuth } from "@/components/AuthProvider"; import { c2api } from "@/lib/c2api"; import { ReplayTab } from "@/components/admin/ReplayTab"; +import { CameraProvidersTab } from "@/components/admin/CameraProvidersTab"; import { useEffect, useState, useRef, useCallback } from "react"; import { useRouter } from "next/navigation"; import type { UserRecord, AuditEntry, UserRole, CallRecord } from "@/lib/types"; @@ -1267,10 +1268,11 @@ function SttEvalTab() { // Main admin page // --------------------------------------------------------------------------- -type AdminTab = "features" | "correlation" | "replay" | "users" | "audit" | "calls" | "eval"; +type AdminTab = "features" | "cameras" | "correlation" | "replay" | "users" | "audit" | "calls" | "eval"; const TAB_LABELS: { key: AdminTab; label: string }[] = [ { key: "features", label: "AI Features" }, + { key: "cameras", label: "Camera providers" }, { key: "correlation", label: "Correlation Debug" }, { key: "replay", label: "Replay" }, { key: "calls", label: "Calls" }, @@ -1283,6 +1285,11 @@ export default function AdminPage() { const { user, isAdmin, loading: authLoading } = useAuth(); const router = useRouter(); const [tab, setTab] = useState("features"); + // Deep link to a tab (/admin#cameras), e.g. from the live wall's empty state. + useEffect(() => { + const h = window.location.hash.slice(1); + if (TAB_LABELS.some((t) => t.key === h)) setTab(h as AdminTab); + }, []); // Wait for the claims to resolve before deciding. isAdmin is false for the // first render of every cold load (typed URL, hard refresh, bookmark) while @@ -1297,7 +1304,7 @@ export default function AdminPage() { if (!isAdmin) return null; // Users/Audit tabs benefit from full width; everything else is narrow - const wide = tab === "users" || tab === "audit" || tab === "replay"; + const wide = tab === "users" || tab === "audit" || tab === "replay" || tab === "cameras"; return (
@@ -1318,6 +1325,7 @@ export default function AdminPage() {
{tab === "features" && } + {tab === "cameras" && } {tab === "correlation" && } {tab === "replay" && } {tab === "calls" && } diff --git a/drb-frontend/app/globals.css b/drb-frontend/app/globals.css index 083802f..215b20b 100644 --- a/drb-frontend/app/globals.css +++ b/drb-frontend/app/globals.css @@ -246,3 +246,18 @@ html:not(.dark) .marketing-hero-bg { .skeleton { animation: skeleton-pulse 1.6s ease-in-out infinite; } + +/* ── Live ops wall transcript ticker ───────────────────────────────────────── + * The belt holds the items twice, so translating by -50% loops seamlessly. + * Paused on hover/focus; reduced-motion users get a static, scrollable row. */ +@keyframes drb-ticker { + from { transform: translateX(0); } + to { transform: translateX(-50%); } +} +.drb-ticker-belt { animation: drb-ticker var(--drb-ticker-duration, 90s) linear infinite; } +.drb-ticker:hover .drb-ticker-belt, +.drb-ticker:focus-within .drb-ticker-belt { animation-play-state: paused; } +@media (prefers-reduced-motion: reduce) { + .drb-ticker-belt { animation: none; } + .drb-ticker { overflow-x: auto; } +} diff --git a/drb-frontend/app/nodes/[id]/page.tsx b/drb-frontend/app/nodes/[id]/page.tsx index f8cf95c..dc92948 100644 --- a/drb-frontend/app/nodes/[id]/page.tsx +++ b/drb-frontend/app/nodes/[id]/page.tsx @@ -197,7 +197,7 @@ export default function NodeDetailPage() { } async function handleReject() { - if (!confirm("Reject this node? It will not be able to upload recordings.")) return; + if (!confirm("Reject this node? Any MQTT login and API key it holds are revoked, so it cannot connect or upload. Re-approving later issues a new key.")) return; setApproving(true); try { await c2api.rejectNode(id); diff --git a/drb-frontend/components/CameraFeed.tsx b/drb-frontend/components/CameraFeed.tsx index 8a0d736..11f5ea4 100644 --- a/drb-frontend/components/CameraFeed.tsx +++ b/drb-frontend/components/CameraFeed.tsx @@ -1,41 +1,217 @@ "use client"; -import { useEffect, useRef, useState } from "react"; +import { useEffect, useRef, useState, type RefObject } from "react"; import type HlsType from "hls.js"; -import type { Ny511Camera } from "@/lib/types"; +import type { Camera, CameraCredit } from "@/lib/types"; -// 511NY stills are served with max-age=60, so refreshing faster buys nothing. +// ── Feed hygiene (design-canvas/gods-eye-view-notes.md §2, top-5 #2) ───────── +// A 24/7 wall must not burn bandwidth on feeds nobody can see: +// * hidden tab or off-screen tile → no fetches, live video torn down; +// * at most MAX_LIVE_HLS live decoders page-wide (GEV uses 2); extra tiles +// show their still with a "live cap" badge (the drill-down gets one extra); +// * a failed refresh keeps the last good frame with a "stale Nm" badge; +// * a 64×36 downsample signature detects a frozen camera (same frame for +// FROZEN_AFTER_MS) without repainting it; a stalled HLS stream falls back. +// Video is played straight from the agency's host (no DRB egress). + +// Most stills are cached ~60 s upstream, so refreshing faster buys nothing. const SNAPSHOT_REFRESH_MS = 60_000; -// A failed stream is not given up on for good: 511NY feeds drop out briefly, -// and a permanent fallback left pinned tiles showing a stale still image. const VIDEO_RETRY_MS = 30_000; -// In-place recoveries hls.js gets per session before we fall back to the snapshot. const MAX_HLS_RECOVERIES = 3; +const VIDEO_STALL_MS = 20_000; +const FROZEN_AFTER_MS = 20 * 60_000; // > the slowest provider cadence (Fintraffic ~10 min) +export const MAX_LIVE_HLS = 2; + +// ── Page-wide live-stream leases ───────────────────────────────────────────── +const holders = new Set(); +const leaseListeners = new Set<() => void>(); +let nextLeaseId = 1; + +function tryAcquire(id: number, high: boolean): boolean { + if (holders.has(id)) return true; + if (holders.size < MAX_LIVE_HLS + (high ? 1 : 0)) { + holders.add(id); + return true; + } + return false; +} + +function release(id: number) { + if (holders.delete(id)) leaseListeners.forEach((l) => l()); +} + +function useLiveLease(want: boolean, high: boolean): boolean { + const idRef = useRef(0); + if (!idRef.current) idRef.current = nextLeaseId++; + const [granted, setGranted] = useState(false); + useEffect(() => { + const id = idRef.current; + if (!want) { release(id); setGranted(false); return; } + const attempt = () => setGranted(tryAcquire(id, high)); + attempt(); + leaseListeners.add(attempt); + return () => { leaseListeners.delete(attempt); release(id); }; + }, [want, high]); + return granted; +} + +// ── Visibility: tab shown AND element on screen ────────────────────────────── +function useOnScreen(ref: RefObject): boolean { + const [tabVisible, setTabVisible] = useState(true); + const [inView, setInView] = useState(true); + useEffect(() => { + const onVis = () => setTabVisible(!document.hidden); + onVis(); + document.addEventListener("visibilitychange", onVis); + return () => document.removeEventListener("visibilitychange", onVis); + }, []); + useEffect(() => { + const el = ref.current; + if (!el || typeof IntersectionObserver === "undefined") return; + const io = new IntersectionObserver((entries) => setInView(entries.some((e) => e.isIntersecting)), { rootMargin: "100px" }); + io.observe(el); + return () => io.disconnect(); + }, [ref]); + return tabVisible && inView; +} + +// Origins whose stills refuse CORS: shown without crossOrigin, no signature. +const noCorsOrigins = new Set(); +function originOf(url: string): string { + try { return new URL(url).origin; } catch { return url; } +} +function withBust(url: string, t: number): string { + return `${url}${url.includes("?") ? "&" : "?"}t=${t}`; +} + +function signature(img: HTMLImageElement): number[] | null { + try { + const c = document.createElement("canvas"); + c.width = 64; + c.height = 36; + const ctx = c.getContext("2d", { willReadFrequently: true }); + if (!ctx) return null; + ctx.drawImage(img, 0, 0, 64, 36); + const d = ctx.getImageData(0, 0, 64, 36).data; + const out: number[] = []; + for (let i = 0; i < d.length; i += 4) out.push((d[i] + d[i + 1] + d[i + 2]) / 3); + return out; + } catch { + return null; // tainted canvas + } +} +function sameFrame(a: number[] | null, b: number[] | null): boolean { + if (!a || !b || a.length !== b.length) return false; + let diff = 0; + for (let i = 0; i < a.length; i++) diff += Math.abs(a[i] - b[i]); + return diff / a.length < 1.5; +} + +function ageLabel(ms: number): string { + const m = Math.floor(ms / 60_000); + return m < 1 ? "<1m" : m < 60 ? `${m}m` : `${Math.floor(m / 60)}h${m % 60 ? ` ${m % 60}m` : ""}`; +} +function clock(t: number): string { + return new Date(t).toLocaleTimeString([], { hour: "2-digit", minute: "2-digit", second: "2-digit" }); +} + +interface Props { + camera: Camera; + className?: string; + /** "still" never starts video (popups, thumbnails). */ + mode?: "auto" | "still"; + /** The drill-down camera may take one stream past the page-wide cap. */ + priority?: "normal" | "high"; + /** Required attribution for this provider — rendered on the imagery itself. */ + credit?: CameraCredit | null; +} /** - * One 511NY camera: the live HLS stream when the camera has one (NYSDOT's - * skyvdn hosts send Access-Control-Allow-Origin: *), otherwise its snapshot on - * a timer. A stream that fails falls back to the snapshot rather than a black box, - * and video is retried every VIDEO_RETRY_MS while it stays down. + * One public camera: live HLS when the camera has a stream and a lease is + * free, otherwise its still on a timer — with honest state badges and the + * provider's attribution on the frame. A stream that fails or stalls falls + * back to the still and is retried every VIDEO_RETRY_MS. */ -export function CameraFeed({ camera, className }: { camera: Ny511Camera; className?: string }) { +export function CameraFeed({ camera, className, mode = "auto", priority = "normal", credit }: Props) { + const rootRef = useRef(null); const videoRef = useRef(null); + const onScreen = useOnScreen(rootRef); const [videoFailed, setVideoFailed] = useState(false); - const [tick, setTick] = useState(() => Date.now()); - const useVideo = !!camera.video_url && !videoFailed; + const wantsVideo = mode === "auto" && !!camera.video_url && !videoFailed && onScreen; + const leased = useLiveLease(wantsVideo, priority === "high"); + const useVideo = wantsVideo && leased; + const capped = wantsVideo && !leased; + + // Still state + const [shownSrc, setShownSrc] = useState(null); + const [corsOk, setCorsOk] = useState(true); + const [lastGoodAt, setLastGoodAt] = useState(null); + const [frameSince, setFrameSince] = useState(null); // when the shown frame first appeared + const [stale, setStale] = useState(false); + const [now, setNow] = useState(() => Date.now()); + const sigRef = useRef(null); + + // Reset when the camera changes + useEffect(() => { + setShownSrc(null); setLastGoodAt(null); setFrameSince(null); setStale(false); setVideoFailed(false); + sigRef.current = null; + }, [camera.id]); useEffect(() => { - if (useVideo) return; - const id = setInterval(() => setTick(Date.now()), SNAPSHOT_REFRESH_MS); + const id = setInterval(() => { if (!document.hidden) setNow(Date.now()); }, 15_000); return () => clearInterval(id); - }, [useVideo]); + }, []); + // Still polling — only while on screen and not playing video + useEffect(() => { + const url = camera.image_url; + if (useVideo || !onScreen || !url) return; + let cancelled = false; + let probe: HTMLImageElement | null = null; + const load = (allowCors: boolean) => { + const origin = originOf(url); + const cors = allowCors && !noCorsOrigins.has(origin); + const src = withBust(url, Date.now()); + probe = new Image(); + if (cors) probe.crossOrigin = "anonymous"; + probe.onload = () => { + if (cancelled || !probe) return; + const t = Date.now(); + const sig = cors ? signature(probe) : null; + const unchanged = sameFrame(sig, sigRef.current); + sigRef.current = sig; + setStale(false); + setLastGoodAt(t); + setCorsOk(cors); + if (!unchanged) { + setShownSrc(src); + setFrameSince(t); + } + }; + probe.onerror = () => { + if (cancelled) return; + if (cors) { noCorsOrigins.add(origin); load(false); return; } + setStale(true); // keep the last good frame on screen + }; + probe.src = src; + }; + load(true); + const id = setInterval(() => load(true), SNAPSHOT_REFRESH_MS); + return () => { + cancelled = true; + clearInterval(id); + if (probe) { probe.onload = null; probe.onerror = null; probe.src = ""; } // abort in-flight + }; + }, [camera.image_url, useVideo, onScreen]); + + // Retry a failed stream later useEffect(() => { if (!videoFailed || !camera.video_url) return; const id = setTimeout(() => setVideoFailed(false), VIDEO_RETRY_MS); return () => clearTimeout(id); }, [videoFailed, camera.video_url]); + // HLS player + stall watchdog useEffect(() => { const video = videoRef.current; const src = camera.video_url; @@ -65,20 +241,58 @@ export function CameraFeed({ camera, className }: { camera: Ny511Camera; classNa }) .catch(() => setVideoFailed(true)); } + let lastT = -1; + let lastMove = Date.now(); + const watchdog = setInterval(() => { + if (video.currentTime !== lastT) { lastT = video.currentTime; lastMove = Date.now(); return; } + if (Date.now() - lastMove > VIDEO_STALL_MS) setVideoFailed(true); // frozen stream → still + }, 5_000); return () => { cancelled = true; + clearInterval(watchdog); hls?.destroy(); video.removeAttribute("src"); video.load(); // stop the download, not just the playback }; }, [useVideo, camera.video_url]); - if (useVideo) { - return