Skip to content

Using the platform

One client, the app's own credential, everything core exposes to apps.

From a facade rule the client is already there, built from the app's own config and credential:

@app.on_detection("person")
def rule(event):
    jpeg = event.snapshot()               # this event's camera, right now
    event.nvr.state.set("last_seen", event.camera)

app.nvr is the same object outside a rule (in @app.on_setup, an action, a state function). Constructed by hand it is:

from opennvr_app_sdk import OpenNVR

with OpenNVR() as nvr:          # OPENNVR_URL + the app's key, from the env
    for camera in nvr.cameras():          # only cameras picked for this app
        jpeg = nvr.snapshot(camera)
You want Use
The camera roster this app was given nvr.cameras(), nvr.camera(id)
A frame right now nvr.snapshot(camera)
Clips, and a playable URL nvr.recordings(camera).list(...), .url(...), .frame_at(...)
What was seen, and the proof nvr.timeline.search(...), .evidence(id)
Whether anyone acknowledged your alerts nvr.alerts.inbox(unacked=True)
State that survives a restart nvr.state.get/set/delete/items
To run a model nvr.ai.infer(adapter, jpeg, task=...), nvr.ai.stream(...)

Every non-2xx raises PlatformError, so there is one exception type to catch.

Async

AsyncOpenNVR is the same surface, awaited. Use it inside an archetype's run loop — a slow snapshot on one camera should not block the others:

    """Fetch a frame from every assigned camera at once, rather than
    one after another — on a 30-camera site this is the difference
    between 200ms and six seconds."""
    async with AsyncOpenNVR() as nvr:
        cameras = await nvr.cameras()
        frames = await asyncio.gather(
            *(nvr.snapshot(c) for c in cameras), return_exceptions=True)
        return {
            c.handle: len(f) if isinstance(f, bytes) else 0
            for c, f in zip(cameras, frames)
        }

Fast inference

KaiCClient.infer is one HTTP round-trip per frame. At ten frames a second use InferStream instead: the session stays open, the model stays warm, and every frame shares one audit correlation_id — which is what makes a sequence traceable as one episode.

The past

EventsClient queries the platform's memory — what was seen, when, with the evidence photo that proves it — so an app can answer "when did that van last come?" without keeping an index of its own.

Free answers: Tier-0

The platform runs a lightweight detector on every camera all the time. Consuming it costs nothing: no adapter, no GPU, no poll. For "how many people are at the loading dock?" it is the whole answer, and it already knows which track has a good crop for evidence.

Full examples: 06_platform_client.py, 13_events_client.py, 14_infer_stream.py, 11_tier0.py.