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

"""Camera provider registry: settings, licence gate, and bbox queries.

Settings live in Firestore ``config/camera_providers``::

    {"providers": {"<id>": {"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()
    live = [pid for pid, cat in CATALOGS.items()
            if effective(PROVIDERS[pid], conf.get(pid))[0] and bbox_intersects(PROVIDERS[pid].meta.coverage, bbox)]
    await asyncio.gather(*(CATALOGS[pid].get() for pid in live))
    cams: List[Camera] = []
    providers = []
    for pid in live:
        cat = CATALOGS[pid]
        hits = in_bbox(cat.items, bbox)
        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": sum(1 for pid in CATALOGS if effective(PROVIDERS[pid], conf.get(pid))[0])}