server-26/drb-c2-core/app/internal/cameras/base.py

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