201 lines
9.2 KiB
Python
201 lines
9.2 KiB
Python
"""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()
|
|
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)}
|