The platform

Everything an app reads from the running deployment: the camera roster it was assigned, snapshots, recordings, the timeline and its evidence, durable state, and inference. The app's own credential scopes all of it — an app cannot see the whole site by accident.

opennvr_app_sdk.OpenNVR

OpenNVR(
    url: str | None = None,
    *,
    token: str | None = None,
    kaic_url: str | None = None,
    kaic_api_key: str | None = None,
    timeout: float = DEFAULT_TIMEOUT,
    client_id: str = "opennvr-app",
)

See the module docstring. All arguments fall back to the environment the app overlays already set: OPENNVR_URL, KAIC_URL / OPENNVR_KAIC_URL, OPENNVR_INTERNAL_API_KEY (bootstrap) and the app key from credentials.py.

Source code in opennvr_app_sdk/client.py

def __init__(self, url: str | None = None, *, token: str | None = None,
             kaic_url: str | None = None, kaic_api_key: str | None = None,
             timeout: float = DEFAULT_TIMEOUT, client_id: str = "opennvr-app") -> None:
    base = url or os.environ.get("OPENNVR_URL") or ""
    if not base:
        raise ValueError("OpenNVR(url=...) or OPENNVR_URL is required")
    self.credentials = AppCredentials(token)
    self._http = _Http(base, self.credentials, timeout)
    kaic = kaic_url or os.environ.get("KAIC_URL") or os.environ.get("OPENNVR_KAIC_URL")
    self.ai = AIAPI(kaic, kaic_api_key or os.environ.get("KAIC_API_KEY")
                    or os.environ.get("OPENNVR_INTERNAL_API_KEY"), timeout, client_id)
    self.timeline = TimelineAPI(self._http)
    self.alerts = AlertsAPI(self._http)
    self.state = StateAPI(self._http)

roster

roster() -> list[Camera] | None

The cameras picked for this app in its configuration.

[] means core answered and nothing is picked: the app should do nothing. None means core could not be asked (unreachable, an error, a key it refused) — NOT the same answer. An app that treats None as "nothing picked" tears its work down on every core restart; one that treats [] as "keep going" never stops when an operator unpicks the last camera. Keep what you have on None, stop on [].

Source code in opennvr_app_sdk/client.py

def roster(self) -> list[Camera] | None:
    """The cameras picked for this app in its configuration.

    ``[]`` means core answered and nothing is picked: the app should
    do nothing. ``None`` means core could not be asked (unreachable,
    an error, a key it refused) — NOT the same answer. An app that
    treats ``None`` as "nothing picked" tears its work down on every
    core restart; one that treats ``[]`` as "keep going" never stops
    when an operator unpicks the last camera. Keep what you have on
    ``None``, stop on ``[]``.
    """
    body = self._http.get_json("/api/v1/internal/camera-agent/cameras")
    if not isinstance(body, dict):
        return None
    return parse_cameras(body)

cameras

cameras() -> list[Camera]

The cameras picked for this app, or [] — which here means EITHER nothing is picked OR core could not be reached. Use :meth:roster wherever that difference changes what you do.

Source code in opennvr_app_sdk/client.py

def cameras(self) -> list[Camera]:
    """The cameras picked for this app, or ``[]`` — which here means
    EITHER nothing is picked OR core could not be reached. Use
    :meth:`roster` wherever that difference changes what you do."""
    return self.roster() or []

snapshot

snapshot(camera) -> bytes | None

The camera's current frame as JPEG, or None.

Source code in opennvr_app_sdk/client.py

def snapshot(self, camera) -> bytes | None:
    """The camera's current frame as JPEG, or ``None``."""
    return self._http.get_bytes(
        f"/api/v1/internal/app/cameras/{_camera_id(camera)}/snapshot")

save_evidence

save_evidence(jpeg: bytes) -> str | None

Store a JPEG for an alert to cite; returns its path.

Put photos HERE, then pass the paths as Alert(images=...). An alert is a NATS message with a 1 MB ceiling, so a base64 crop inside the alert is not merely wasteful — past the ceiling the broker drops the publish and the alert never reaches anyone.

None when the upload fails: an app must still be able to raise its alert without the picture.

Source code in opennvr_app_sdk/client.py

def save_evidence(self, jpeg: bytes) -> str | None:
    """Store a JPEG for an alert to cite; returns its path.

    Put photos HERE, then pass the paths as ``Alert(images=...)``.
    An alert is a NATS message with a 1 MB ceiling, so a base64 crop
    inside the alert is not merely wasteful — past the ceiling the
    broker drops the publish and the alert never reaches anyone.

    ``None`` when the upload fails: an app must still be able to
    raise its alert without the picture.
    """
    if not jpeg:
        return None
    try:
        body = self._http.post_bytes("/api/v1/internal/app/evidence",
                                     jpeg, "image/jpeg")
    except PlatformError as exc:
        logger.warning("evidence upload failed: %s", exc)
        return None
    path = (body or {}).get("path")
    return str(path) if path else None

stream_grant

stream_grant(camera) -> dict | None

Core's permission to read this camera's video, plus the URL.

The token in it is scoped to this one camera's path and expires, so apps sharing a network cannot read each other's cameras. None when core will not or cannot grant it.

Source code in opennvr_app_sdk/client.py

def stream_grant(self, camera) -> dict | None:
    """Core's permission to read this camera's video, plus the URL.

    The token in it is scoped to this one camera's path and expires,
    so apps sharing a network cannot read each other's cameras.
    ``None`` when core will not or cannot grant it.
    """
    return self._http.get_json(
        f"/api/v1/internal/app/cameras/{_camera_id(camera)}/stream")

stream

stream(camera, *, width: int = 640, fps: float = 10.0)

A live frame stream for one camera, started and self-renewing.

For rules about a shape in TIME — a scan sweep, a fall, a queue forming — where a snapshot every few seconds has already missed it. Newest frame wins; a camera reboot reconnects on its own.

with nvr.stream(cam) as video:
    for frame in video.frames():
        ...

The grant is re-fetched on every reconnect, so an expiring token renews itself rather than failing mid-session.

Source code in opennvr_app_sdk/client.py

def stream(self, camera, *, width: int = 640, fps: float = 10.0):
    """A live frame stream for one camera, started and self-renewing.

    For rules about a shape in TIME — a scan sweep, a fall, a queue
    forming — where a snapshot every few seconds has already missed
    it. Newest frame wins; a camera reboot reconnects on its own.

        with nvr.stream(cam) as video:
            for frame in video.frames():
                ...

    The grant is re-fetched on every reconnect, so an expiring token
    renews itself rather than failing mid-session.
    """
    from .rtsp import RtspFrameStream

    def url_factory() -> str:
        grant = self.stream_grant(camera)
        if not grant or not grant.get("url"):
            raise FrameStreamUnavailable(
                f"core granted no stream for camera {_camera_id(camera)}")
        return str(grant["url"])

    return RtspFrameStream(url_factory=url_factory, width=width, fps=fps,
                           name=f"cam{_camera_id(camera)}").start()

opennvr_app_sdk.AsyncOpenNVR

AsyncOpenNVR(
    url: str | None = None,
    *,
    token: str | None = None,
    kaic_url: str | None = None,
    kaic_api_key: str | None = None,
    timeout: float = DEFAULT_TIMEOUT,
    http_client: AsyncClient | None = None,
    client_id: str = "opennvr-app",
)

The async twin of :class:opennvr_app_sdk.OpenNVR; same arguments and environment fallbacks. Pass http_client to share one httpx.AsyncClient (a FastAPI app's lifespan pool, a test transport); the client then does not close it.

Source code in opennvr_app_sdk/aio.py

def __init__(self, url: str | None = None, *, token: str | None = None,
             kaic_url: str | None = None, kaic_api_key: str | None = None,
             timeout: float = DEFAULT_TIMEOUT,
             http_client: httpx.AsyncClient | None = None,
             client_id: str = "opennvr-app") -> None:
    base = url or os.environ.get("OPENNVR_URL") or ""
    if not base:
        raise ValueError("AsyncOpenNVR(url=...) or OPENNVR_URL is required")
    self.credentials = AppCredentials(token)
    self._http = _AsyncHttp(base, self.credentials, timeout, http_client)
    kaic = kaic_url or os.environ.get("KAIC_URL") or os.environ.get("OPENNVR_KAIC_URL")
    self.ai = AsyncAIAPI(kaic, kaic_api_key or os.environ.get("KAIC_API_KEY")
                         or os.environ.get("OPENNVR_INTERNAL_API_KEY"),
                         timeout, http_client, client_id)
    self.timeline = AsyncTimelineAPI(self._http)
    self.alerts = AsyncAlertsAPI(self._http)
    self.state = AsyncStateAPI(self._http)

roster async

roster() -> list[Camera] | None

Picked cameras; [] = nothing picked, None = core could not be asked. See :meth:opennvr_app_sdk.client.OpenNVR.roster.

Source code in opennvr_app_sdk/aio.py

async def roster(self) -> list[Camera] | None:
    """Picked cameras; ``[]`` = nothing picked, ``None`` = core could
    not be asked. See :meth:`opennvr_app_sdk.client.OpenNVR.roster`."""
    body = await self._http.get_json("/api/v1/internal/camera-agent/cameras")
    if not isinstance(body, dict):
        return None
    return parse_cameras(body)

save_evidence async

save_evidence(jpeg: bytes) -> str | None

Store a JPEG for an alert to cite; returns its path.

Photos go here, not into the alert: an alert is a NATS message with a 1 MB ceiling, and past it the broker drops the publish — the alarm is never seen at all. None when the upload fails, because an app must still be able to raise its alert.

Source code in opennvr_app_sdk/aio.py

async def save_evidence(self, jpeg: bytes) -> str | None:
    """Store a JPEG for an alert to cite; returns its path.

    Photos go here, not into the alert: an alert is a NATS message
    with a 1 MB ceiling, and past it the broker drops the publish —
    the alarm is never seen at all. ``None`` when the upload fails,
    because an app must still be able to raise its alert.
    """
    if not jpeg:
        return None
    try:
        body = await self._http.post_bytes(
            "/api/v1/internal/app/evidence", jpeg, "image/jpeg")
    except PlatformError as exc:
        logger.warning("evidence upload failed: %s", exc)
        return None
    path = (body or {}).get("path")
    return str(path) if path else None

stream_grant async

stream_grant(camera) -> dict | None

Core's permission to read this camera's video, plus the URL.

Scoped to this one camera's path and short-lived, so apps on a shared network cannot read each other's cameras.

Source code in opennvr_app_sdk/aio.py

async def stream_grant(self, camera) -> dict | None:
    """Core's permission to read this camera's video, plus the URL.

    Scoped to this one camera's path and short-lived, so apps on a
    shared network cannot read each other's cameras.
    """
    return await self._http.get_json(
        f"/api/v1/internal/app/cameras/{_camera_id(camera)}/stream")

stream

stream(camera, *, width: int = 640, fps: float = 10.0)

A live frame stream for one camera, started and self-renewing.

Deliberately NOT a coroutine: the decoder is a background thread feeding a newest-frame slot, so there is nothing to await. Awaiting frames would only add the queue this design exists to avoid.

Source code in opennvr_app_sdk/aio.py

def stream(self, camera, *, width: int = 640, fps: float = 10.0):
    """A live frame stream for one camera, started and self-renewing.

    Deliberately NOT a coroutine: the decoder is a background thread
    feeding a newest-frame slot, so there is nothing to await. Awaiting
    frames would only add the queue this design exists to avoid.
    """
    from .rtsp import RtspFrameStream

    cam_id = _camera_id(camera)
    base_headers = self._http.headers
    base_url = self._http.base

    def url_factory() -> str:
        # Sync fetch on the decoder thread: this runs at reconnect
        # time, not per frame, and the stream thread is not the
        # event loop.
        import httpx

        r = httpx.get(
            f"{base_url}/api/v1/internal/app/cameras/{cam_id}/stream",
            headers=base_headers(), timeout=10.0)
        if r.status_code >= 400:
            raise FrameStreamUnavailable(
                f"core granted no stream for camera {cam_id} "
                f"(HTTP {r.status_code})")
        url = (r.json() or {}).get("url")
        if not url:
            raise FrameStreamUnavailable(
                f"core granted no stream for camera {cam_id}")
        return str(url)

    return RtspFrameStream(url_factory=url_factory, width=width, fps=fps,
                           name=f"cam{cam_id}").start()

opennvr_app_sdk.Camera dataclass

Camera(
    id: int,
    handle: str,
    name: str,
    role: str,
    frame_url: str,
    assignments: list[dict] = list(),
    raw: dict = dict(),
)

One camera in the app's roster (what core assigned to this app).

opennvr_app_sdk.Recording dataclass

Recording(start: str, duration: float, raw: dict = dict())

opennvr_app_sdk.PlatformError

Bases: RuntimeError

A write (state, actions) the platform refused or could not take.

opennvr_app_sdk.EventsClient

EventsClient(
    core_url: str,
    api_key: str | None = None,
    *,
    http_get: HttpGetH | None = None,
)

Query visits and fetch their evidence photos from core.

Source code in opennvr_app_sdk/events.py

def __init__(self, core_url: str, api_key: str | None = None, *,
             http_get: HttpGetH | None = None) -> None:
    self._base = core_url.rstrip("/")
    # The app's own key when it has one (scoped to its cameras), else
    # the explicit/site key — see credentials.py.
    from .credentials import AppCredentials

    key = AppCredentials(api_key).token()
    self._headers = {"X-Internal-Api-Key": key} if key else {}
    self._get = http_get or _default_http_get

search async

search(
    *,
    label: str | None = None,
    camera_id: int | None = None,
    plate: str | None = None,
    start: datetime | str | None = None,
    end: datetime | str | None = None,
    limit: int = 50,
) -> list[StoredEvent] | None

Visits overlapping [start, end), newest first.

Returns [] for a genuinely empty window and None on ANY failure (transport, auth, or a rejected query) — the caller must be able to say "nothing came" and "I couldn't check" differently; in a security product those are different answers.

Source code in opennvr_app_sdk/events.py

async def search(
    self,
    *,
    label: str | None = None,
    camera_id: int | None = None,
    plate: str | None = None,
    start: datetime | str | None = None,
    end: datetime | str | None = None,
    limit: int = 50,
) -> list[StoredEvent] | None:
    """Visits overlapping [start, end), newest first.

    Returns ``[]`` for a genuinely empty window and ``None`` on ANY
    failure (transport, auth, or a rejected query) — the caller must be
    able to say "nothing came" and "I couldn't check" differently; in a
    security product those are different answers."""
    params: dict[str, Any] = {"limit": limit}
    if label:
        params["label"] = label
    if camera_id is not None:
        params["camera_id"] = camera_id
    if plate:
        params["plate"] = plate
    if start is not None:
        params["from"] = start.isoformat() if isinstance(start, datetime) else start
    if end is not None:
        params["to"] = end.isoformat() if isinstance(end, datetime) else end
    url = f"{self._base}/api/v1/internal/camera-agent/events?{urlencode(params)}"
    try:
        status, body = await self._get(url, self._headers)
        if status != 200:
            return None
        import json
        rows = json.loads(body.decode("utf-8")).get("events", [])
    except Exception:
        return None
    out = []
    for r in rows:
        try:
            out.append(StoredEvent(
                id=int(r["id"]), camera_id=int(r["camera_id"]),
                label=r.get("label"), score=r.get("score"),
                started_at=r.get("started_at"), ended_at=r.get("ended_at"),
                stationary=r.get("stationary"),
                plate_text=r.get("plate_text"),
                has_evidence=bool(r.get("has_evidence")),
            ))
        except (KeyError, TypeError, ValueError):
            continue
    return out

evidence async

evidence(event_id: int) -> bytes | None

The visit's best-frame JPEG, or None.

Source code in opennvr_app_sdk/events.py

async def evidence(self, event_id: int) -> bytes | None:
    """The visit's best-frame JPEG, or None."""
    url = f"{self._base}/api/v1/internal/camera-agent/events/{int(event_id)}/evidence"
    try:
        status, body = await self._get(url, self._headers)
    except Exception:
        return None
    return body if status == 200 and body else None

recording_frame async

recording_frame(camera_id: int, at: str) -> bytes | None

One JPEG from recorded footage at instant at (ISO 8601), or None.

Powers the agent's describe_window: sample a few instants across a past window and caption each. Internal-key authed like evidence.

Source code in opennvr_app_sdk/events.py

async def recording_frame(self, camera_id: int, at: str) -> bytes | None:
    """One JPEG from recorded footage at instant ``at`` (ISO 8601), or None.

    Powers the agent's describe_window: sample a few instants across a past
    window and caption each. Internal-key authed like ``evidence``.
    """
    from urllib.parse import urlencode

    q = urlencode({"camera_id": int(camera_id), "at": at})
    url = f"{self._base}/api/v1/internal/camera-agent/recordings/frame?{q}"
    try:
        status, body = await self._get(url, self._headers)
    except Exception:
        return None
    return body if status == 200 and body else None

opennvr_app_sdk.StoredEvent dataclass

StoredEvent(
    id: int,
    camera_id: int,
    label: str | None,
    score: float | None,
    started_at: str | None,
    ended_at: str | None,
    stationary: bool | None,
    plate_text: str | None,
    has_evidence: bool,
)

One remembered visit, as the store serves it.

opennvr_app_sdk.KaiCClient

KaiCClient(
    base_url: str,
    adapter_name: str,
    *,
    api_key: str | None = None,
    timeout_seconds: float = 10.0,
    http_client: Client | None = None,
)

Tiny client for KAI-C's POST /api/v1/infer/{adapter}.

Sends the frame as a base64 JSON body (the contract-v1 convenience path — multipart adds boilerplate without benefit at ~1 fps polling) and threads X-Correlation-Id so every alert traces back through KAI-C's audit log and the adapter's logs alike. The optional API key rides the X-Internal-Api-Key header, matching the intrusion-detection example and KAI-C's internal-auth scheme.

Source code in opennvr_app_sdk/frame_app.py

def __init__(
    self,
    base_url: str,
    adapter_name: str,
    *,
    api_key: str | None = None,
    timeout_seconds: float = 10.0,
    http_client: httpx.Client | None = None,
) -> None:
    self._base_url = base_url.rstrip("/")
    self._adapter_name = adapter_name
    self._api_key = api_key
    self._owns_client = http_client is None
    self._client = http_client or httpx.Client(
        timeout=timeout_seconds, trust_env=False,
    )

infer

infer(
    frame_bytes: bytes,
    *,
    task: str,
    camera_id: str | None = None,
    params: dict[str, Any] | None = None,
    correlation_id: str | None = None,
) -> dict[str, Any]

Send one frame; return the raw §5.1 InferResponse body.

camera_id is optional, as it is for KAI-C itself: with it the call is attributed to the camera (audit, skill budgets, the NATS subject); without it KAI-C treats the call as a one-off probe. Raises :class:KaiCError on transport failure or non-200; the frame loop catches and decides whether to alert / skip / abort.

Source code in opennvr_app_sdk/frame_app.py

def infer(
    self,
    frame_bytes: bytes,
    *,
    task: str,
    camera_id: str | None = None,
    params: dict[str, Any] | None = None,
    correlation_id: str | None = None,
) -> dict[str, Any]:
    """Send one frame; return the raw §5.1 ``InferResponse`` body.

    ``camera_id`` is optional, as it is for KAI-C itself: with it
    the call is attributed to the camera (audit, skill budgets, the
    NATS subject); without it KAI-C treats the call as a one-off
    probe. Raises :class:`KaiCError` on transport failure or
    non-200; the frame loop catches and decides whether to alert /
    skip / abort."""
    url, headers, body = build_infer_request(
        self._base_url, self._adapter_name, frame_bytes, task=task,
        api_key=self._api_key, camera_id=camera_id, params=params,
        correlation_id=correlation_id)
    try:
        response = self._client.post(url, json=body, headers=headers)
    except Exception as exc:
        raise KaiCError(f"KAI-C unreachable at {url}: {exc}") from exc
    if response.status_code != 200:
        raise KaiCError(
            f"KAI-C returned HTTP {response.status_code}: {response.text[:200]}"
        )
    return response.json()

opennvr_app_sdk.KaiCError

Bases: Exception

Raised when KAI-C is unreachable or returns a non-200. Frame loops treat this as a transient skip — alerts don't fire on a comms failure (the failure itself is visible in KAI-C's audit log via the correlation_id we sent).

opennvr_app_sdk.InferStream

InferStream(
    kaic_url: str,
    api_key: str | None,
    *,
    adapter: str,
    camera_id: str,
    client_id: str = "opennvr-app",
    timeout: float = 10.0,
    websocket_factory: Callable[
        [str, list[tuple[str, str]]], Any
    ]
    | None = None,
)

A streaming inference session against one KAI-C adapter.

KaiCClient.infer is one HTTP round-trip per frame — fine at one frame every few seconds, wasteful at ten a second. This holds a WebSocket session open instead, so the model stays warm and every frame in the session shares one audit correlation_id, which is what makes a sequence of frames traceable as a single episode.

Use it as a context manager; the session reopens itself after a failure, so a dropped frame costs one frame::

with InferStream(url, key, adapter="yolov8", camera_id="cam1") as s:
    for jpeg in frames:
        result = s.infer(jpeg)["result"]

nvr.ai.stream(adapter, camera_id=…) builds one from the app's own credential, which is usually what you want inside an app.

Source code in opennvr_app_sdk/infer_stream.py

def __init__(self, kaic_url: str, api_key: str | None, *, adapter: str,
             camera_id: str, client_id: str = "opennvr-app",
             timeout: float = 10.0,
             websocket_factory: Callable[[str, list[tuple[str, str]]], Any] | None = None,
             ) -> None:
    parsed = urlparse(kaic_url)
    scheme = "wss" if parsed.scheme == "https" else "ws"
    prefix = (parsed.path or "").rstrip("/")
    self.url = urlunparse((scheme, parsed.netloc,
                           f"{prefix}/api/v1/infer/{adapter}/stream", "", "", ""))
    self._api_key = api_key
    self._camera_id = str(camera_id)
    self._client_id = client_id
    self._timeout = timeout
    self._factory = websocket_factory
    self._conn: Any = None
    self._seq = 0
    self.correlation_id: str | None = None

open

open(correlation_id: str | None = None) -> 'InferStream'

Connect + §6.1 handshake (idempotent).

Source code in opennvr_app_sdk/infer_stream.py

def open(self, correlation_id: str | None = None) -> "InferStream":
    """Connect + §6.1 handshake (idempotent)."""
    if self._conn is not None:
        return self
    cid = correlation_id or f"app-{uuid.uuid4().hex[:12]}"
    headers = [("X-Correlation-Id", cid)]
    if self._api_key:
        headers.append(("X-Internal-Api-Key", self._api_key))
    conn = self._connect(headers)
    try:
        conn.send(json.dumps({"type": "handshake", "client_id": self._client_id,
                              "camera_id": self._camera_id,
                              "frame_transport": "websocket"}))
        ack = _loads(conn.recv(timeout=self._timeout))
        if not isinstance(ack, dict) or ack.get("type") != "handshake_ack":
            raise KaiCError(f"unexpected handshake response: {ack!r}")
    except KaiCError:
        _close(conn)
        raise
    except Exception as exc:  # noqa: BLE001
        _close(conn)
        raise KaiCError(f"WS handshake failed: {exc}") from exc
    self._conn = conn
    self.correlation_id = cid
    self._seq = 0                     # §6.3: monotonic PER SESSION
    return self

infer

infer(jpeg: bytes) -> dict[str, Any]

Send one frame, return a §5.1-shaped result. Raises KaiCError and closes the session on any failure.

Source code in opennvr_app_sdk/infer_stream.py

def infer(self, jpeg: bytes) -> dict[str, Any]:
    """Send one frame, return a §5.1-shaped result. Raises
    ``KaiCError`` and closes the session on any failure."""
    self.open()
    self._seq += 1
    try:
        self._conn.send(json.dumps({"type": "frame", "seq": self._seq,
                                    "ts_ms": int(time.monotonic() * 1000),
                                    "content_type": "image/jpeg"}))
        self._conn.send(jpeg)
        raw = self._conn.recv(timeout=self._timeout)
    except Exception as exc:  # noqa: BLE001
        self.close()
        raise KaiCError(f"WS infer failed: {exc}") from exc
    try:
        payload = _loads(raw)
    except Exception as exc:  # noqa: BLE001
        raise KaiCError(f"WS recv: non-JSON payload: {exc}") from exc
    if not isinstance(payload, dict) or payload.get("type") != "result":
        raise KaiCError(f"WS recv: unexpected message {payload!r}")
    return {
        "status": "ok",
        "model_name": "", "model_version": "",
        "inference_ms": int(payload.get("inference_ms", 0) or 0),
        "result": payload.get("result") or {},
        # All frames of a session share KAI-C's audit correlation id.
        "correlation_id": self.correlation_id,
    }

opennvr_app_sdk.discover_cameras

discover_cameras(
    opennvr_url: str,
    *,
    api_key: str | None = None,
    timeout: float = 5.0,
) -> list[dict[str, Any]]

Return OpenNVR's configured cameras, or [] if they can't be read.

Never raises: discovery runs at app startup, and an app that refuses to boot because core was still starting is worse than one that boots with no cameras and says so. Callers should log the empty result.

[] deliberately conflates "core said zero cameras" with "core could not be reached" — for a boot-time listing they are the same non-answer. Callers that must tell them apart (assignment, where the two mean opposite things) use :func:_fetch_cameras.

Source code in opennvr_app_sdk/cameras.py

def discover_cameras(
    opennvr_url: str,
    *,
    api_key: str | None = None,
    timeout: float = 5.0,
) -> list[dict[str, Any]]:
    """Return OpenNVR's configured cameras, or ``[]`` if they can't be read.

    Never raises: discovery runs at app startup, and an app that refuses
    to boot because core was still starting is worse than one that boots
    with no cameras and says so. Callers should log the empty result.

    ``[]`` deliberately conflates "core said zero cameras" with "core
    could not be reached" — for a boot-time listing they are the same
    non-answer. Callers that must tell them apart (assignment, where
    the two mean opposite things) use :func:`_fetch_cameras`.
    """
    return _fetch_cameras(opennvr_url, api_key=api_key, timeout=timeout) or []

opennvr_app_sdk.camera_key

camera_key(value: Any) -> int | None

3 / "3" / "cam3" / "cam-3" → 3; anything else → None.

Never raises: it is used to look things up, and an unparseable key should simply match nothing.

Source code in opennvr_app_sdk/cameras.py

def camera_key(value: Any) -> int | None:
    """``3`` / ``"3"`` / ``"cam3"`` / ``"cam-3"`` → ``3``; anything else → None.

    Never raises: it is used to look things up, and an unparseable key
    should simply match nothing."""
    if isinstance(value, bool) or value is None:
        return None
    if isinstance(value, int):
        return value
    cam_id = getattr(value, "id", None)
    if isinstance(cam_id, int) and not isinstance(cam_id, bool):
        return cam_id
    text = str(value).strip().lower()
    if text.startswith("cam-"):
        text = text[4:]
    elif text.startswith("cam"):
        text = text[3:]
    return int(text) if text.isdigit() else None

opennvr_app_sdk.per_camera_value

per_camera_value(
    mapping: Any, camera: Any, default: Any = None
) -> Any

mapping[camera] for a per-camera config value, whichever spelling of the camera either side used.

Source code in opennvr_app_sdk/cameras.py

def per_camera_value(mapping: Any, camera: Any, default: Any = None) -> Any:
    """``mapping[camera]`` for a per-camera config value, whichever
    spelling of the camera either side used."""
    if not isinstance(mapping, dict):
        return default
    want = camera_key(camera)
    if want is None:
        return mapping.get(camera, default)
    for key, value in mapping.items():
        if camera_key(key) == want:
            return value
    return default

opennvr_app_sdk.cameras_for_skill

cameras_for_skill(
    opennvr_url: str,
    skill: str,
    *,
    api_key: str | None = None,
    timeout: float = 5.0,
) -> list[str] | None

Fetch-and-filter convenience: the camera ids assigned skill.

[] means core answered and no camera carries the skill: watch nothing. None means core could not be ASKED — unknown, and unknown must not be acted on in either direction: keep whatever roster you already had rather than dropping to nothing on a restart that raced core, or widening to everything on a network blip.

Use :func:filter_cameras_for_skill when you already hold a :func:discover_cameras result.

Source code in opennvr_app_sdk/cameras.py

def cameras_for_skill(
    opennvr_url: str,
    skill: str,
    *,
    api_key: str | None = None,
    timeout: float = 5.0,
) -> list[str] | None:
    """Fetch-and-filter convenience: the camera ids assigned ``skill``.

    ``[]`` means core answered and no camera carries the skill: watch
    nothing. ``None`` means core could not be ASKED — unknown, and
    unknown must not be acted on in either direction: keep whatever
    roster you already had rather than dropping to nothing on a restart
    that raced core, or widening to everything on a network blip.

    Use :func:`filter_cameras_for_skill` when you already hold a
    :func:`discover_cameras` result.
    """
    cameras = _fetch_cameras(opennvr_url, api_key=api_key, timeout=timeout)
    if cameras is None:
        return None
    return filter_cameras_for_skill(cameras, skill)

opennvr_app_sdk.filter_cameras_for_skill

filter_cameras_for_skill(
    cameras: list[dict[str, Any]], skill: str
) -> list[str] | None

Which of cameras (a :func:discover_cameras payload) are assigned skill.

Returns the camera-id list, EMPTY when no camera carries the skill. An empty list means watch nothing: the operator has not pointed this skill at anything yet. It used to return None there, meaning "watch everything", which is why an unconfigured app saw the fleet.

The return type stays | None for callers that still branch on it; None now only means "could not ask" (an empty skill name), never "no restriction".

Pure — feed it the list you already fetched instead of fetching twice (the occupancy example's refresh loop does exactly this).

Source code in opennvr_app_sdk/cameras.py

def filter_cameras_for_skill(
    cameras: list[dict[str, Any]], skill: str
) -> list[str] | None:
    """Which of ``cameras`` (a :func:`discover_cameras` payload) are
    assigned ``skill``.

    Returns the camera-id list, EMPTY when no camera carries the skill.
    An empty list means watch nothing: the operator has not pointed this
    skill at anything yet. It used to return ``None`` there, meaning
    "watch everything", which is why an unconfigured app saw the fleet.

    The return type stays ``| None`` for callers that still branch on
    it; ``None`` now only means "could not ask" (an empty skill name),
    never "no restriction".

    Pure — feed it the list you already fetched instead of fetching
    twice (the occupancy example's refresh loop does exactly this).
    """
    want = str(skill).strip().lower()
    if not want:
        return None
    out: list[str] = []
    for cam in cameras:
        if not isinstance(cam, dict):
            continue
        for a in cam.get("assignments") or []:
            if isinstance(a, dict) and str(a.get("skill", "")).lower() == want:
                out.append(str(cam["camera_id"]))
                break
    return out

opennvr_app_sdk.AppCredentials

AppCredentials(explicit: str | None = None)

Resolve, remember and rotate the app's credential.

explicit is the config's opennvr_token (either kind).

Source code in opennvr_app_sdk/credentials.py

def __init__(self, explicit: str | None = None) -> None:
    self._explicit = str(explicit) if explicit else None
    self._app_key: str | None = None
    if is_app_key(self._explicit):
        self._app_key = self._explicit
    else:
        self._app_key = stored_app_key()

bus_url property

bus_url: str | None

Where to join the event bus with the app key (core told us).

token

token() -> str | None

What to send: the app key when we have one, else the site key.

Source code in opennvr_app_sdk/credentials.py

def token(self) -> str | None:
    """What to send: the app key when we have one, else the site key."""
    return self._app_key or site_key(
        None if is_app_key(self._explicit) else self._explicit)

headers

headers() -> dict[str, str]

Both header shapes core accepts, so one value works whether it is an app key, the site key or a user JWT.

Source code in opennvr_app_sdk/credentials.py

def headers(self) -> dict[str, str]:
    """Both header shapes core accepts, so one value works whether
    it is an app key, the site key or a user JWT."""
    tok = self.token()
    if not tok:
        return {}
    return {"Authorization": f"Bearer {tok}", "X-Internal-Api-Key": tok}

adopt

adopt(key: str) -> None

A key just issued by core: use it from now on and persist it.

Source code in opennvr_app_sdk/credentials.py

def adopt(self, key: str) -> None:
    """A key just issued by core: use it from now on and persist it."""
    self._app_key = key.strip()
    store_app_key(self._app_key)
    logger.info("app key issued by OpenNVR — using it for every core call")

invalidate

invalidate() -> None

Core refused our app key (rotated/revoked): fall back to the site key so the next registration can ask for a new one.

Source code in opennvr_app_sdk/credentials.py

def invalidate(self) -> None:
    """Core refused our app key (rotated/revoked): fall back to the
    site key so the next registration can ask for a new one."""
    if self._app_key:
        logger.warning("OpenNVR rejected the app key — discarding it; "
                       "will request a new one at next registration")
    self._app_key = None
    forget_app_key()
    forget_bus_url()

opennvr_app_sdk.auth_headers

auth_headers(explicit: str | None = None) -> dict[str, str]

One-shot helper for clients that don't hold an AppCredentials.

Source code in opennvr_app_sdk/credentials.py

def auth_headers(explicit: str | None = None) -> dict[str, str]:
    """One-shot helper for clients that don't hold an AppCredentials."""
    return AppCredentials(explicit).headers()

opennvr_app_sdk.FrameSource

Bases: Protocol

Anything that can produce the latest frame for a camera.

Returns encoded image bytes (JPEG/PNG — whatever the adapter's contract accepts) or None when no frame is available right now (camera offline, snapshot endpoint empty). Raising is also fine — the poll loop isolates per-camera fetch failures.

opennvr_app_sdk.CameraFrameSource

Bases: Protocol

Anything with fetch() -> bytes and a stable camera_id is a per-camera frame source.

opennvr_app_sdk.FileFrameSource

FileFrameSource(*, camera_id: str, path: str)

Read a JPEG/PNG from disk. camera_id is operator-supplied.

Path-traversal protected: we resolve the configured path once at init time and reject any subsequent change. (Operators shouldn't be passing user-controlled paths anyway, but the example sets the pattern for future sources.)

Source code in opennvr_app_sdk/frame_sources.py

def __init__(self, *, camera_id: str, path: str) -> None:
    resolved = pathlib.Path(path).expanduser().resolve()
    if not resolved.is_file():
        raise FrameSourceError(
            f"file frame source: {path!r} does not exist or is not a file"
        )
    self.camera_id = camera_id
    self._path = resolved

opennvr_app_sdk.HttpSnapshotSource

HttpSnapshotSource(
    *,
    camera_id: str,
    url: str,
    timeout_seconds: float = 5.0,
    verify_tls: bool = True,
)

GET an HTTP snapshot URL. Supports basic-auth via the URL (http://user:pass@host/snapshot.jpg) — standard pattern for consumer-grade cameras.

Timeout is intentionally low (default 5s): a slow snapshot in the polling loop blocks every camera. If the camera is consistently slow, the operator should lower the poll interval or move to RTSP.

Source code in opennvr_app_sdk/frame_sources.py

def __init__(
    self,
    *,
    camera_id: str,
    url: str,
    timeout_seconds: float = 5.0,
    verify_tls: bool = True,
) -> None:
    parsed = urlparse(url)
    if parsed.scheme not in ("http", "https"):
        raise FrameSourceError(
            f"http snapshot source: expected http(s) URL, got {parsed.scheme!r}"
        )
    self.camera_id = camera_id
    self._url = url
    self._timeout = timeout_seconds
    self._verify_tls = verify_tls

opennvr_app_sdk.CoreSnapshotSource

CoreSnapshotSource(nvr=None)

Frames from OpenNVR itself, for any camera picked for this app.

get_frame("cam3") fetches the camera's current JPEG through core's app snapshot route, which serves only the app's own picks — so an app connected to core needs no hand-written frame_url per camera, and cannot read a camera it wasn't given. Returns None when core has no frame (camera offline, not picked, core unreachable); the poll loop simply skips that camera for the tick.

Source code in opennvr_app_sdk/frame_sources.py

def __init__(self, nvr=None) -> None:
    self._nvr = nvr

opennvr_app_sdk.DictFrameSource

DictFrameSource(sources: Mapping[str, CameraFrameSource])

Adapt a {camera_id: CameraFrameSource} mapping to the :class:~.frame_app.FrameSource shape the FrameApp loop polls.

Holds the mapping by reference — swapping an entry (a test stub, a reconfigured camera) is picked up on the next tick. Unknown camera ids raise KeyError; the poll loop isolates that like any other per-camera fetch failure.

Source code in opennvr_app_sdk/frame_sources.py

def __init__(self, sources: Mapping[str, CameraFrameSource]) -> None:
    self._sources = sources

opennvr_app_sdk.build_frame_source

build_frame_source(
    *, camera_id: str, url: str
) -> CameraFrameSource

Pick the right source class based on the URL scheme. Anything unrecognised raises FrameSourceError — fail fast at config-load time rather than mid-loop.

Source code in opennvr_app_sdk/frame_sources.py

def build_frame_source(*, camera_id: str, url: str) -> CameraFrameSource:
    """Pick the right source class based on the URL scheme. Anything
    unrecognised raises ``FrameSourceError`` — fail fast at config-load
    time rather than mid-loop."""
    parsed = urlparse(url)
    scheme = parsed.scheme.lower()
    if scheme == "file":
        # urlparse("file:///path") → path is in parsed.path
        return FileFrameSource(camera_id=camera_id, path=parsed.path)
    if scheme in ("http", "https"):
        return HttpSnapshotSource(camera_id=camera_id, url=url)
    if scheme == "opennvr":
        raise FrameSourceError(
            "opennvr:// scheme is reserved for a future OpenNVR-backend snapshot "
            "endpoint and is not yet implemented. Use http(s):// against the "
            "camera's snapshot URL directly for now."
        )
    if scheme == "rtsp":
        # Continuous video is a different shape from a snapshot poll —
        # one long-lived decoder, newest-frame-wins — so it lives in
        # `rtsp.py` and is used directly, not fetched per tick. This
        # adapter exists so a FrameApp built around polling can still be
        # pointed at a stream and get the latest frame each tick.
        from .rtsp import RtspStillSource

        return RtspStillSource(camera_id=camera_id, url=url)
    raise FrameSourceError(
        f"unsupported frame source scheme {scheme!r}; expected file/http/https."
    )

opennvr_app_sdk.dict_frame_source

dict_frame_source(
    sources: Mapping[str, CameraFrameSource],
) -> FrameSource

Convenience constructor for :class:DictFrameSource.

Source code in opennvr_app_sdk/frame_sources.py

def dict_frame_source(sources: Mapping[str, CameraFrameSource]) -> FrameSource:
    """Convenience constructor for :class:`DictFrameSource`."""
    return DictFrameSource(sources)

opennvr_app_sdk.FrameSourceError

Bases: Exception

Raised when a frame source cannot produce a frame this cycle. Caller (the detector loop) decides whether to skip or abort — transient failures are normal (network blips, camera offline).

opennvr_app_sdk.RtspFrameStream

RtspFrameStream(
    url: str | None = None,
    *,
    url_factory: Callable[[], str] | None = None,
    width: int = DEFAULT_WIDTH,
    fps: float = DEFAULT_FPS,
    spawn: SpawnFn | None = None,
    size: tuple[int, int] | None = None,
    name: str = "rtsp",
)

Newest-frame-wins reader for one RTSP URL.

Start it, then either poll :meth:latest or iterate :meth:frames. Both hand back the most recent decoded frame; neither ever hands back a backlog.

url_factory is called for each (re)connect, so a caller whose URL carries a short-lived token — every app reading through the platform's scoped stream grant — renews simply by returning a fresh one.

Source code in opennvr_app_sdk/rtsp.py

def __init__(
    self,
    url: str | None = None,
    *,
    url_factory: Callable[[], str] | None = None,
    width: int = DEFAULT_WIDTH,
    fps: float = DEFAULT_FPS,
    spawn: SpawnFn | None = None,
    size: tuple[int, int] | None = None,
    name: str = "rtsp",
) -> None:
    if not url and not url_factory:
        raise ValueError("RtspFrameStream needs url or url_factory")
    self._url_factory = url_factory or (lambda: str(url))
    self.width = width
    self.fps = fps
    self.name = name
    self._spawn = spawn or self._default_spawn
    self._size = size
    self._frame: Frame | None = None
    self._new = threading.Event()
    self._stop = threading.Event()
    self._lock = threading.Lock()
    self._thread: threading.Thread | None = None
    #: The ffmpeg process of the session running right now. close()
    #: and the stall watchdog both need to reach it: the reader is
    #: parked in a blocking read on its stdout, and killing the
    #: process is the only thing that unblocks it.
    self._proc = None
    self._seq = 0
    self.restarts = 0
    #: Frames the app never asked for before the next arrived. Not a
    #: fault — it is the honest measure of how far inference is
    #: behind the camera, and worth logging when it climbs.
    self.dropped = 0

healthy property

healthy: bool

Whether a frame arrived recently enough to call this live.

latest

latest(*, timeout: float | None = None) -> Frame | None

The newest frame, waiting up to timeout for a new one.

Returns None on timeout — which is a fact about the camera, not an error, and the caller decides what it means.

Source code in opennvr_app_sdk/rtsp.py

def latest(self, *, timeout: float | None = None) -> Frame | None:
    """The newest frame, waiting up to ``timeout`` for a new one.

    Returns None on timeout — which is a fact about the camera, not
    an error, and the caller decides what it means.
    """
    if timeout is not None and not self._new.wait(timeout):
        return None
    self._new.clear()
    with self._lock:
        return self._frame

frames

frames(*, timeout: float = 5.0) -> Iterator[Frame]

Frames as they arrive, skipping any the caller was too slow to collect. Ends when the stream is closed.

Source code in opennvr_app_sdk/rtsp.py

def frames(self, *, timeout: float = 5.0) -> Iterator[Frame]:
    """Frames as they arrive, skipping any the caller was too slow
    to collect. Ends when the stream is closed."""
    while not self._stop.is_set():
        frame = self.latest(timeout=timeout)
        if frame is not None:
            yield frame

opennvr_app_sdk.RtspStillSource

RtspStillSource(
    *,
    camera_id: str,
    url: str,
    width: int = DEFAULT_WIDTH,
    fps: float = 4.0,
    quality: int = 85,
)

A stream dressed as a snapshot source.

The polling FrameApp asks for "a frame now" every few seconds and expects encoded bytes. Pointing it at a stream would otherwise mean spawning ffmpeg per tick (what the camera-agent does, and it costs a second each time), so this keeps ONE decoder warm and hands over the newest frame, JPEG-encoded on demand.

Needs opencv for the encode — an app that only wants stills and has no opencv should use the camera's HTTP snapshot URL instead.

Source code in opennvr_app_sdk/rtsp.py

def __init__(self, *, camera_id: str, url: str, width: int = DEFAULT_WIDTH,
             fps: float = 4.0, quality: int = 85) -> None:
    self.camera_id = camera_id
    self.quality = quality
    self._stream = RtspFrameStream(url, width=width, fps=fps,
                                   name=f"still-{camera_id}")
    self._started = False

fetch

fetch() -> bytes | None

The newest frame as JPEG, or None if the stream has nothing.

Source code in opennvr_app_sdk/rtsp.py

def fetch(self) -> bytes | None:
    """The newest frame as JPEG, or None if the stream has nothing."""
    import cv2

    if not self._started:
        self._stream.start()
        self._started = True
    # First call waits for the decoder to come up; later calls take
    # whatever is there, because a poll loop must not block.
    frame = self._stream.latest(timeout=10.0 if self._started else 2.0)
    if frame is None:
        return None
    ok, buf = cv2.imencode(".jpg", frame.to_ndarray(),
                           [cv2.IMWRITE_JPEG_QUALITY, self.quality])
    return buf.tobytes() if ok else None

opennvr_app_sdk.Frame dataclass

Frame(
    data: bytes,
    width: int,
    height: int,
    seq: int,
    mono_ts: float,
    wall_ts: float = 0.0,
    restarted: bool = False,
)

One decoded frame: raw BGR bytes plus when it was taken.

to_ndarray

to_ndarray()

The frame as an (h, w, 3) BGR array (needs numpy).

Source code in opennvr_app_sdk/rtsp.py

def to_ndarray(self):
    """The frame as an ``(h, w, 3)`` BGR array (needs numpy)."""
    import numpy as np

    return np.frombuffer(self.data, dtype=np.uint8).reshape(
        self.height, self.width, 3)

opennvr_app_sdk.FrameStreamError

Bases: RuntimeError

The stream could not be opened or understood.

opennvr_app_sdk.FrameStreamUnavailable

Bases: PlatformError

Core would not grant a stream for this camera.