Front door

Six names. If you are writing your first app, this is the whole API — App declares it, @app.on_detection is the rule, and event.alert() is what reaches an operator. Everything else on this site is for the app that outgrows them.

opennvr_app_sdk.App

App(
    app_id: str,
    *,
    name: str | None = None,
    version: str = "0.1.0",
    category: str = "analytics",
    summary: str = "",
    requires_tasks: Sequence[str] | None = None,
    consume_tier0: bool = True,
    **manifest_kwargs: Any,
)

A whole OpenNVR app: identity, config, rules, surfaces, lifecycle.

Construct one at module scope, decorate handlers on it, and call :meth:run from __main__. The NATS loop, alert dispatch, the contract server, registry self-registration, live config, the CLI and signal handling are inherited from :class:~.detector.Detector, which this compiles to.

Constructor arguments beyond the ones below are passed to :class:~.manifest.AppManifest, so anything the catalog understands (description, use_cases, pricing, requires_scopes, provides, author …) is available without leaving the facade. Fields the decorators derive — params, emits, state_schema, actions, has_ui, entitlement — are refused here, with a pointer to the decorator that owns them.

Source code in opennvr_app_sdk/facade.py

def __init__(
    self,
    app_id: str,
    *,
    name: str | None = None,
    version: str = "0.1.0",
    category: str = "analytics",
    summary: str = "",
    requires_tasks: Sequence[str] | None = None,
    consume_tier0: bool = True,
    **manifest_kwargs: Any,
) -> None:
    if not app_id or not app_id.strip():
        raise ValueError("App(app_id): an app id is required")
    app_id = app_id.strip()
    if not _ID_RE.match(app_id):
        raise ValueError(
            f"App({app_id!r}): the app id must be kebab-case — lowercase "
            f"letters and digits, single hyphens between words "
            f"(e.g. 'gate-watch'). It becomes the app's identity "
            f"everywhere: the catalog entry, the NATS subject, the "
            f"container name."
        )
    for field_name, owner in _DERIVED_FIELDS.items():
        if field_name in manifest_kwargs:
            raise TypeError(
                f"App({app_id!r}, {field_name}=…): the facade derives "
                f"{field_name!r} from your decorators — declare it with "
                f"{owner} instead."
            )
    self.id = app_id
    self._name = name or app_id.replace("-", " ").title()
    self._version = version
    self._category = category
    self._summary = summary
    self._requires_tasks = list(requires_tasks or ["object_detection"])
    #: Consume the always-on Tier-0 detector. True by default,
    #: because on a stock install Tier-0 is the ONLY detection
    #: stream on the bus — an app that ignores it registers, shows a
    #: green dot, and fires nothing, forever. Set False when the app
    #: also subscribes to a heavy adapter and would otherwise see
    #: every object twice.
    self.consume_tier0 = bool(consume_tier0)
    self._manifest_kwargs = manifest_kwargs

    self._rules: list[_Rule] = []
    self._raw_handlers: list[Callable[..., Any]] = []
    self._setup_hooks: list[Callable[[Any], Any]] = []
    self._shutdown_hooks: list[Callable[[], Any]] = []
    self._config_hooks: list[Callable[[dict[str, Any]], Any]] = []
    self._params: list[Param] = []
    self._zones: dict[str, str] = {}
    self._emits: list[AlertType] = []
    self._views: list[StateView] = []
    self._state_fn: Callable[[], dict[str, Any]] | None = None
    self._actions: list[tuple[Action, Callable[..., Any]]] = []
    self._ui_fn: Callable[[], str] | None = None
    self._license_fn: Callable[[str], Any] | None = None
    self._publishes: list[str] = []
    self._last_detector: Any = None

config property

config: Any

The parsed config of the running app. event.config is the same object and is what a rule should normally use.

store property

store: dict[str, Any]

A plain dict the app can keep counters and recent items in. Its contents are merged into GET /state, so declaring app.metric("alerted") and doing app.store["alerted"] += 1 is a complete dashboard.

nvr property

nvr

The platform client, built once from this app's own config and credential (:class:~.client.OpenNVR).

publisher property

publisher

The domain-event publisher for this app (:class:~.domain_events.DomainEventPublisher). event.publish is the usual way in.

param

param(
    name: str,
    type_: Any = str,
    *,
    default: Any = None,
    description: str = "",
    per_camera: bool = False,
    required: bool = False,
    suggestions: Sequence[str] | None = None,
) -> "App"

Declare one operator-settable knob.

It becomes a manifest param (so the catalog renders a form field), a field on the generated config dataclass (so config.yml fills it), and an attribute on event.config. Bind a rule filter to it with "$name". Chainable.

Source code in opennvr_app_sdk/facade.py

def param(
    self,
    name: str,
    type_: Any = str,
    *,
    default: Any = None,
    description: str = "",
    per_camera: bool = False,
    required: bool = False,
    suggestions: Sequence[str] | None = None,
) -> "App":
    """Declare one operator-settable knob.

    It becomes a manifest param (so the catalog renders a form
    field), a field on the generated config dataclass (so
    ``config.yml`` fills it), and an attribute on ``event.config``.
    Bind a rule filter to it with ``"$name"``. Chainable."""
    self._check_name(name, "param")
    self._params.append(Param(
        name=name, type=type_, default=default, per_camera=per_camera,
        description=description, required=required,
        suggestions=list(suggestions or []),
    ))
    return self

zone

zone(name: str, description: str = '') -> 'App'

Declare a zone the operator draws on each camera.

@app.on_detection(..., zone="driveway") declares it too; call this to give it a description the operator will see in the catalog's geometry editor. Chainable.

Source code in opennvr_app_sdk/facade.py

def zone(self, name: str, description: str = "") -> "App":
    """Declare a zone the operator draws on each camera.

    ``@app.on_detection(..., zone="driveway")`` declares it too;
    call this to give it a description the operator will see in the
    catalog's geometry editor. Chainable."""
    self._check_name(name, "zone")
    self._zones[name] = description or self._zones.get(name, "")
    return self

emits

emits(
    name: str,
    *,
    severity: str = "medium",
    description: str = "",
) -> "App"

Declare an alert kind for the catalog. Optional — one is derived per rule otherwise. Chainable.

Source code in opennvr_app_sdk/facade.py

def emits(self, name: str, *, severity: str = "medium",
          description: str = "") -> "App":
    """Declare an alert kind for the catalog. Optional — one is
    derived per rule otherwise. Chainable."""
    self._emits.append(AlertType(name=name, severity=severity,
                                 description=description))
    return self

publishes

publishes(schema: str) -> 'App'

Declare a domain event this app publishes, so it appears in the app's AsyncAPI document. Chainable.

Source code in opennvr_app_sdk/facade.py

def publishes(self, schema: str) -> "App":
    """Declare a domain event this app publishes, so it appears in
    the app's AsyncAPI document. Chainable."""
    if schema not in self._publishes:
        self._publishes.append(schema)
    return self

on_detection

on_detection(
    *labels: str,
    camera: str | Sequence[str] | None = None,
    zone: str | None = None,
    min_confidence: Any = None,
    dwell: Any = 0.0,
    cooldown: Any = 0.0,
    forget: Any = None,
    severity: str = "medium",
    emits: str | None = None,
) -> Callable[
    [Callable[[DetectionEvent], Any]],
    Callable[[DetectionEvent], Any],
]

Register a rule. The handler is called once per detection that passes every filter, with a :class:DetectionEvent.

labels Class labels to react to ("person", "car"). None given ⇒ every label. camera One camera id or a list; omitted ⇒ every camera. zone Only fire inside this zone. Declaring one adds a per-camera geometry.polygon param of that name, which is how the operator knows which zones to draw and where each one goes. min_confidence Floor on detector confidence. No hidden default: omit it and every detection reaches the rule. dwell Seconds the object must have satisfied THIS rule's filters continuously before the handler runs, and then only once per presence episode — the loitering pattern without the state machine. The clock starts when the filters first match, so a zone rule times presence in the zone. cooldown Minimum seconds between handler calls for the same object. What stops a parked car alerting on every frame. forget Seconds without a sighting that end a presence episode and re-arm dwell. Defaults to max(DEFAULT_ABSENCE_S, dwell) — long enough that a few dropped frames are not mistaken for the object leaving. severity Default severity for alerts this rule fires, and the severity of the alert type derived for the manifest. emits Name of the alert type this rule fires. Defaults to the handler's name, slugified; declare it when the name matters (it is part of the catalog listing, so a later rename of the function would otherwise change the app's public contract).

The handler may fire alerts with event.alert(...), return an :class:~.alerts.Alert (or a list), or return nothing.

Source code in opennvr_app_sdk/facade.py

def on_detection(
    self,
    *labels: str,
    camera: str | Sequence[str] | None = None,
    zone: str | None = None,
    min_confidence: Any = None,
    dwell: Any = 0.0,
    cooldown: Any = 0.0,
    forget: Any = None,
    severity: str = "medium",
    emits: str | None = None,
) -> Callable[[Callable[[DetectionEvent], Any]], Callable[[DetectionEvent], Any]]:
    """Register a rule. The handler is called once per detection
    that passes every filter, with a :class:`DetectionEvent`.

    ``labels``
        Class labels to react to (``"person"``, ``"car"``). None
        given ⇒ every label.
    ``camera``
        One camera id or a list; omitted ⇒ every camera.
    ``zone``
        Only fire inside this zone. Declaring one adds a per-camera
        ``geometry.polygon`` param of that name, which is how the
        operator knows which zones to draw and where each one goes.
    ``min_confidence``
        Floor on detector confidence. No hidden default: omit it and
        every detection reaches the rule.
    ``dwell``
        Seconds the object must have satisfied THIS rule's filters
        continuously before the handler runs, and then only once per
        presence episode — the loitering pattern without the state
        machine. The clock starts when the filters first match, so a
        zone rule times presence *in the zone*.
    ``cooldown``
        Minimum seconds between handler calls for the same object.
        What stops a parked car alerting on every frame.
    ``forget``
        Seconds without a sighting that end a presence episode and
        re-arm ``dwell``. Defaults to
        ``max(DEFAULT_ABSENCE_S, dwell)`` — long enough that a few
        dropped frames are not mistaken for the object leaving.
    ``severity``
        Default severity for alerts this rule fires, and the
        severity of the alert type derived for the manifest.
    ``emits``
        Name of the alert type this rule fires. Defaults to the
        handler's name, slugified; declare it when the name matters
        (it is part of the catalog listing, so a later rename of the
        function would otherwise change the app's public contract).

    The handler may fire alerts with ``event.alert(...)``, return an
    :class:`~.alerts.Alert` (or a list), or return nothing.
    """
    if zone is not None:
        self.zone(zone) if zone not in self._zones else None
    if camera is None:
        cameras: tuple[str, ...] = ()
    elif isinstance(camera, str):
        cameras = (camera,)
    else:
        cameras = tuple(str(c) for c in camera)

    def decorate(fn: Callable[[DetectionEvent], Any]):
        self._rules.append(_Rule(
            fn=fn,
            index=len(self._rules),
            labels=tuple(str(label).lower() for label in labels),
            cameras=cameras,
            zone=zone,
            min_confidence=_spec(min_confidence),
            dwell=_spec(dwell),
            cooldown=_spec(cooldown),
            forget=_spec(forget),
            severity=severity,
            emits=emits,
        ))
        return fn

    return decorate

on_event

on_event() -> Callable[
    [Callable[..., Any]], Callable[..., Any]
]

Escape hatch: see every inference event whole.

The handler is called as fn(camera_id, detections, event) — the triple :meth:Detector.on_detections receives — and may return alerts to fire. Use it for rules about the frame rather than about one object (crowding, absence, ratios).

Source code in opennvr_app_sdk/facade.py

def on_event(self) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
    """Escape hatch: see every inference event whole.

    The handler is called as ``fn(camera_id, detections, event)`` —
    the triple :meth:`Detector.on_detections` receives — and may
    return alerts to fire. Use it for rules about the frame rather
    than about one object (crowding, absence, ratios)."""

    def decorate(fn: Callable[..., Any]):
        self._raw_handlers.append(fn)
        return fn

    return decorate

state

state() -> Callable[
    [Callable[[], dict[str, Any]]],
    Callable[[], dict[str, Any]],
]

Register what GET /state returns — the live data the catalog renders through the tiles declared below.

Anything in :attr:store is included automatically, so an app that only keeps counters needs no @app.state at all.

Source code in opennvr_app_sdk/facade.py

def state(self) -> Callable[[Callable[[], dict[str, Any]]], Callable[[], dict[str, Any]]]:
    """Register what ``GET /state`` returns — the live data the
    catalog renders through the tiles declared below.

    Anything in :attr:`store` is included automatically, so an app
    that only keeps counters needs no ``@app.state`` at all."""

    def decorate(fn: Callable[[], dict[str, Any]]):
        self._state_fn = fn
        return fn

    return decorate

metric

metric(
    path: str,
    *,
    label: str | None = None,
    description: str = "",
) -> "App"

A single number from /state, shown as a stat chip.

Source code in opennvr_app_sdk/facade.py

def metric(self, path: str, *, label: str | None = None,
           description: str = "") -> "App":
    """A single number from ``/state``, shown as a stat chip."""
    return self._view("metric", path, label, description)

gauge

gauge(
    path: str,
    *,
    label: str | None = None,
    min: float = 0.0,
    max: float = 100.0,
    warn: float | None = None,
    danger: float | None = None,
    unit: str = "",
    description: str = "",
) -> "App"

A number between bounds, shown as a bar — amber past warn, red past danger.

Source code in opennvr_app_sdk/facade.py

def gauge(self, path: str, *, label: str | None = None, min: float = 0.0,
          max: float = 100.0, warn: float | None = None,
          danger: float | None = None, unit: str = "",
          description: str = "") -> "App":
    """A number between bounds, shown as a bar — amber past ``warn``,
    red past ``danger``."""
    return self._view("gauge", path, label, description, min=min, max=max,
                      warn=warn, danger=danger, unit=unit)

table

table(
    path: str,
    *,
    label: str | None = None,
    columns: Sequence[str] = (),
    description: str = "",
) -> "App"

A list from /state, shown as a table.

Source code in opennvr_app_sdk/facade.py

def table(self, path: str, *, label: str | None = None,
          columns: Sequence[str] = (), description: str = "") -> "App":
    """A list from ``/state``, shown as a table."""
    return self._view("table", path, label, description,
                      columns=list(columns))

log

log(
    path: str,
    *,
    label: str | None = None,
    limit: int = 20,
    description: str = "",
) -> "App"

A recent-events feed, newest first.

Source code in opennvr_app_sdk/facade.py

def log(self, path: str, *, label: str | None = None, limit: int = 20,
        description: str = "") -> "App":
    """A recent-events feed, newest first."""
    return self._view("log", path, label, description, limit=limit)

gallery

gallery(
    path: str,
    *,
    label: str | None = None,
    limit: int = 12,
    description: str = "",
) -> "App"

A thumbnail wall — plate crops, doorbell snapshots. Entries are {image|url, label, time}; image may be a data URI.

Source code in opennvr_app_sdk/facade.py

def gallery(self, path: str, *, label: str | None = None, limit: int = 12,
            description: str = "") -> "App":
    """A thumbnail wall — plate crops, doorbell snapshots. Entries
    are ``{image|url, label, time}``; ``image`` may be a data URI."""
    return self._view("gallery", path, label, description, limit=limit)

action

action(
    name: str,
    *,
    label: str | None = None,
    params: Sequence[Param] = (),
    description: str = "",
    confirm: bool = False,
) -> Callable[[Callable[..., Any]], Callable[..., Any]]

Register an operator verb — a button in the catalog with a form generated from params.

The handler is called with the declared params as keyword arguments, defaults filled in. Raise ValueError for params you reject (the operator sees the message); return anything JSON-serializable.

Actions are reached only through core's proxy, which is user-JWT only: an action is always invoked by a person, and :func:~.usercontext.current_user tells you which one.

Source code in opennvr_app_sdk/facade.py

def action(
    self,
    name: str,
    *,
    label: str | None = None,
    params: Sequence[Param] = (),
    description: str = "",
    confirm: bool = False,
) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
    """Register an operator verb — a button in the catalog with a
    form generated from ``params``.

    The handler is called with the declared params as keyword
    arguments, defaults filled in. Raise ``ValueError`` for params
    you reject (the operator sees the message); return anything
    JSON-serializable.

    Actions are reached only through core's proxy, which is
    **user-JWT only**: an action is always invoked by a person, and
    :func:`~.usercontext.current_user` tells you which one."""
    self._check_action_name(name)

    def decorate(fn: Callable[..., Any]):
        self._actions.append((Action(
            name=name, label=label or name.replace("_", " ").capitalize(),
            params=list(params), description=description or (fn.__doc__ or "").strip(),
            confirm=confirm,
        ), fn))
        return fn

    return decorate

ui

ui() -> Callable[[Callable[[], str]], Callable[[], str]]

Register an HTML dashboard, served at GET /ui and rendered sandboxed inside the catalog. Return a string.

Source code in opennvr_app_sdk/facade.py

def ui(self) -> Callable[[Callable[[], str]], Callable[[], str]]:
    """Register an HTML dashboard, served at ``GET /ui`` and
    rendered sandboxed inside the catalog. Return a string."""

    def decorate(fn: Callable[[], str]):
        self._ui_fn = fn
        return fn

    return decorate

on_license

on_license() -> Callable[
    [Callable[[str], Any]], Callable[[str], Any]
]

Register the licence gate for a paid app.

Declaring it sets entitlement="license_key": the catalog collects a key from the administrator and refuses to enable the app until this function says it is good. Return an :class:~.contract.Entitlement, or a bool for the simple case. OpenNVR takes no part in the transaction.

Source code in opennvr_app_sdk/facade.py

def on_license(self) -> Callable[[Callable[[str], Any]], Callable[[str], Any]]:
    """Register the licence gate for a paid app.

    Declaring it sets ``entitlement="license_key"``: the catalog
    collects a key from the administrator and refuses to enable the
    app until this function says it is good. Return an
    :class:`~.contract.Entitlement`, or a bool for the simple case.
    OpenNVR takes no part in the transaction."""

    def decorate(fn: Callable[[str], Any]):
        self._license_fn = fn
        return fn

    return decorate

on_setup

on_setup() -> Callable[
    [Callable[[Any], Any]], Callable[[Any], Any]
]

Run once with the parsed config before any event — open a database, build a client, load a denylist.

Source code in opennvr_app_sdk/facade.py

def on_setup(self) -> Callable[[Callable[[Any], Any]], Callable[[Any], Any]]:
    """Run once with the parsed config before any event — open a
    database, build a client, load a denylist."""

    def decorate(fn: Callable[[Any], Any]):
        self._setup_hooks.append(fn)
        return fn

    return decorate

on_config

on_config() -> Callable[
    [Callable[[dict], Any]], Callable[[dict], Any]
]

Run when an operator changes the app's config, which core re-delivers without a restart. Called with the new config dict; make it idempotent — the first call usually restates what boot already applied. Zones are re-read for you either way.

Source code in opennvr_app_sdk/facade.py

def on_config(self) -> Callable[[Callable[[dict], Any]], Callable[[dict], Any]]:
    """Run when an operator changes the app's config, which core
    re-delivers without a restart. Called with the new config dict;
    make it idempotent — the first call usually restates what boot
    already applied. Zones are re-read for you either way."""

    def decorate(fn: Callable[[dict], Any]):
        self._config_hooks.append(fn)
        return fn

    return decorate

on_shutdown

on_shutdown() -> Callable[
    [Callable[[], Any]], Callable[[], Any]
]

Run on the way out (SIGINT / SIGTERM), after the event loop stops — close a database, flush a buffer. The SDK's own resources are closed for you.

Source code in opennvr_app_sdk/facade.py

def on_shutdown(self) -> Callable[[Callable[[], Any]], Callable[[], Any]]:
    """Run on the way out (SIGINT / SIGTERM), after the event loop
    stops — close a database, flush a buffer. The SDK's own
    resources are closed for you."""

    def decorate(fn: Callable[[], Any]):
        self._shutdown_hooks.append(fn)
        return fn

    return decorate

manifest

manifest() -> AppManifest

The :class:~.manifest.AppManifest this app compiles to — declared fields plus everything the decorators imply.

Source code in opennvr_app_sdk/facade.py

def manifest(self) -> AppManifest:
    """The :class:`~.manifest.AppManifest` this app compiles to —
    declared fields plus everything the decorators imply."""
    params: list[Param] = []
    for zone_name in self._zones:
        params.append(Param(
            zone_name, "geometry.polygon", default=None, per_camera=True,
            description=self._zones[zone_name]
            or f"The {zone_name.replace('_', ' ')} area on each camera.",
        ))
    params.extend(self._params)

    emits = list(self._emits)
    if not emits:
        seen: dict[str, str] = {}
        for rule in self._rules:
            seen.setdefault(rule.emits or _slug(rule.label) or self.id,
                            rule.severity)
        emits = [AlertType(name=n, severity=s) for n, s in seen.items()] or [
            AlertType(name=self.id, severity="medium")]

    kwargs = dict(self._manifest_kwargs)
    kwargs.setdefault("subscribes", "opennvr.inference.>")
    return AppManifest(
        id=self.id,
        name=self._name,
        version=self._version,
        category=self._category,
        summary=self._summary,
        requires_tasks=list(self._requires_tasks),
        params=params,
        emits=emits,
        state_schema=list(self._views),
        actions=[a for a, _ in self._actions],
        has_ui=self._ui_fn is not None,
        entitlement="license_key" if self._license_fn else "none",
        **kwargs,
    )

config_class

config_class() -> type

The dataclass config.yml is loaded into: :class:~.config.BaseAppConfig plus one field per zone and per declared param.

Source code in opennvr_app_sdk/facade.py

def config_class(self) -> type:
    """The dataclass ``config.yml`` is loaded into:
    :class:`~.config.BaseAppConfig` plus one field per zone and per
    declared param."""
    fields_spec: list[tuple[str, Any, Any]] = [
        ("consume_tier0", Any, field(default=self.consume_tier0)),
    ]
    for zone_name in self._zones:
        fields_spec.append((zone_name, Any, field(default_factory=dict)))
    for param in self._params:
        default = param.default
        if isinstance(default, (list, dict, set)):
            spec: Any = field(
                default_factory=lambda frozen=default: copy.deepcopy(frozen))
        else:
            spec = field(default=default)
        fields_spec.append((param.name, Any, spec))
    return make_dataclass(
        f"{_pascal(self.id)}Config", fields_spec, bases=(BaseAppConfig,),
    )

load_config

load_config(path: str) -> Any

Load path into :meth:config_class, defaulting the NATS subject to the inference broadcast.

Source code in opennvr_app_sdk/facade.py

def load_config(self, path: str) -> Any:
    """Load ``path`` into :meth:`config_class`, defaulting the NATS
    subject to the inference broadcast."""
    cfg = load_app_config(path, self.config_class())
    if getattr(cfg, "subject_pattern", None) is None:
        cfg.subject_pattern = "opennvr.inference.>"
    return cfg

detector_class

detector_class() -> type[Detector]

Compile the app to a :class:~.detector.Detector subclass.

This is the whole trick: the facade is a code generator with one output. Anything that accepts a Detector — the test helpers, the camera agent's runtime monitors, a custom runner — accepts this.

Source code in opennvr_app_sdk/facade.py

def detector_class(self) -> type[Detector]:
    """Compile the app to a :class:`~.detector.Detector` subclass.

    This is the whole trick: the facade is a code generator with one
    output. Anything that accepts a Detector — the test helpers, the
    camera agent's runtime monitors, a custom runner — accepts this.
    """
    owner = self
    manifest = self.manifest()
    return _build_detector_class(owner, manifest)

build

build(config: Any, dispatcher: AlertDispatcher) -> Detector

Instantiate the compiled detector directly — for tests and for embedding an app in another process.

Source code in opennvr_app_sdk/facade.py

def build(self, config: Any, dispatcher: AlertDispatcher) -> Detector:
    """Instantiate the compiled detector directly — for tests and
    for embedding an app in another process."""
    return self.detector_class()(config, dispatcher)

run

run(argv: list[str] | None = None) -> int

Parse the CLI, load the config, and run until signalled. The return value is the process exit code::

if __name__ == "__main__":
    raise SystemExit(app.run())

Source code in opennvr_app_sdk/facade.py

def run(self, argv: list[str] | None = None) -> int:
    """Parse the CLI, load the config, and run until signalled.
    The return value is the process exit code::

        if __name__ == "__main__":
            raise SystemExit(app.run())
    """
    if not self._rules and not self._raw_handlers:
        raise RuntimeError(
            f"App({self.id!r}).run(): no handlers registered — decorate "
            f"at least one function with @app.on_detection(...)"
        )
    return AppRunner(
        self.detector_class(), load_config=self.load_config,
    ).run(argv)

opennvr_app_sdk.DetectionEvent

DetectionEvent(
    *,
    detection: dict[str, Any],
    camera: str,
    raw: dict[str, Any],
    config: Any,
    zones: dict[str, Zone],
    owner: "App",
    state: KeyedState | None = None,
    record: Any = None,
    severity: str = "medium",
)

One detection, in context — what a @app.on_detection handler is called with.

The object is deliberately flat: what a rule asks about is an attribute or a one-word method, and the raw envelope stays available as :attr:raw for anything the facade does not model.

Coordinates are NORMALIZED (0–1 of the frame) throughout, matching the platform's NormalizedBBox wire shape, so a rule written against one camera resolution works on all of them.

Source code in opennvr_app_sdk/facade.py

def __init__(
    self,
    *,
    detection: dict[str, Any],
    camera: str,
    raw: dict[str, Any],
    config: Any,
    zones: dict[str, Zone],
    owner: "App",
    state: KeyedState | None = None,
    record: Any = None,
    severity: str = "medium",
) -> None:
    self.detection = detection
    self.camera = camera
    self.raw = raw
    self.config = config
    self.state = state
    self._zones = zones
    self._record = record
    self._app = owner
    self._severity = severity
    self._alerts: list[Alert] = []

label property

label: str

The detection's class label, lowercased ("person").

confidence property

confidence: float

Detector confidence in [0, 1]; 0.0 when absent.

track_id property

track_id: str | None

Tracker identity for this object, when the adapter emits one.

Present ⇒ dwell and cooldown follow the OBJECT. Absent ⇒ they fall back to the (camera, label) pair, which is coarser but still keeps a parked car from re-alerting on every frame.

bbox property

bbox: dict[str, float]

Normalized {x, y, w, h} box; missing keys read as 0.

center property

center: Point

Normalized centre point of the box — what zone tests use.

zone property

zone: str | None

Name of the first zone containing the detection, or None when it is outside every zone.

zones property

zones: list[str]

Every zone the detection falls inside.

dwell_s property

dwell_s: float

Seconds this object has continuously satisfied THIS rule.

The clock starts when the rule's filters first match, so zone="driveway", dwell=30 means thirty seconds in the driveway — not thirty seconds on camera followed by one frame in the driveway. A gap longer than forget= ends the episode.

first_seen property

first_seen: bool

True on the first event of a presence episode.

ts property

ts: float

POSIX timestamp of the inference event.

detections property

detections: list[dict[str, Any]]

Every detection in the same inference event, this one included — for rules that need company ("a person AND a car").

correlation_id property

correlation_id: str

The platform's id for this inference. Thread it onto anything you emit and the whole causal chain — inference, audit line, evidence frame, alert — joins up in the timeline.

adapter property

adapter: str

Which detector produced this — an adapter name, or tier0 for the always-on detector every camera runs.

nvr property

nvr

The platform client, built from this app's own config and credential: event.nvr.cameras(), .timeline.search(...), .state.set(...), .ai.infer(...). See :class:~.client.OpenNVR.

in_zone

in_zone(name: str | None = None) -> bool

True when the detection's centre falls inside a zone the operator drew. With no name, true when it falls inside ANY of this app's zones.

A zone with no polygon drawn yet is False, not an error: zones are operator config and a camera may simply not have one.

Source code in opennvr_app_sdk/facade.py

def in_zone(self, name: str | None = None) -> bool:
    """True when the detection's centre falls inside a zone the
    operator drew. With no ``name``, true when it falls inside ANY
    of this app's zones.

    A zone with no polygon drawn yet is False, not an error: zones
    are operator config and a camera may simply not have one."""
    if name is None:
        return any(z.contains(self.center) for z in self._zones.values())
    zone = self._zones.get(name)
    return bool(zone and zone.contains(self.center))

count

count(label: str | None = None) -> int

How many objects of label (or of any label) the same event carried.

Source code in opennvr_app_sdk/facade.py

def count(self, label: str | None = None) -> int:
    """How many objects of ``label`` (or of any label) the same
    event carried."""
    if label is None:
        return len(self.detections)
    wanted = label.lower()
    return sum(
        1 for d in self.detections
        if isinstance(d, dict) and str(d.get("label", "")).lower() == wanted
    )

snapshot

snapshot() -> bytes | None

The current frame from this event's camera, as JPEG bytes.

Source code in opennvr_app_sdk/facade.py

def snapshot(self) -> bytes | None:
    """The current frame from this event's camera, as JPEG bytes."""
    return self.nvr.snapshot(self.camera)

alert

alert(
    title: str,
    description: str = "",
    *,
    severity: str | None = None,
    evidence: dict[str, Any] | None = None,
    tags: Iterable[str] | None = None,
    camera_id: str | None = None,
) -> Alert

Fire an alert — the thing a human sees.

Everything the platform needs is filled in from the event: camera, correlation id, label, confidence, track, zone and dwell. severity defaults to the rule's, so the manifest's emits block and the alerts actually fired cannot disagree.

The alert is dispatched whether or not you return it; the return value is there so a handler can adjust it first.

Source code in opennvr_app_sdk/facade.py

def alert(
    self,
    title: str,
    description: str = "",
    *,
    severity: str | None = None,
    evidence: dict[str, Any] | None = None,
    tags: Iterable[str] | None = None,
    camera_id: str | None = None,
) -> Alert:
    """Fire an alert — the thing a human sees.

    Everything the platform needs is filled in from the event:
    camera, correlation id, label, confidence, track, zone and
    dwell. ``severity`` defaults to the rule's, so the manifest's
    ``emits`` block and the alerts actually fired cannot disagree.

    The alert is dispatched whether or not you return it; the return
    value is there so a handler can adjust it first."""
    body = dict(evidence or {})
    body.setdefault("label", self.label)
    body.setdefault("confidence", self.confidence)
    if self.adapter:
        body.setdefault("adapter", self.adapter)
    if self.track_id:
        body.setdefault("track_id", self.track_id)
    zone_name = self.zone
    if zone_name:
        body.setdefault("zone", zone_name)
    if self.dwell_s > 0:
        body.setdefault("dwell_s", round(self.dwell_s, 1))
    tag_list = [self._app.id, self.label]
    if zone_name:
        tag_list.append(zone_name)
    if tags:
        tag_list.extend(str(t) for t in tags)
    alert = Alert(
        title=title,
        description=description or title,
        camera_id=camera_id or self.camera,
        severity=severity or self._severity,
        correlation_id=self.correlation_id or None,
        evidence=body,
        tags=list(dict.fromkeys(tag_list)),
    )
    self._alerts.append(alert)
    return alert

publish

publish(
    schema: str,
    payload: dict[str, Any],
    *,
    camera_id: str | None = None,
) -> bool

Publish a contracted domain event — the thing another APP sees. The envelope, the producer (app:<id>), the camera and the correlation id are filled in; schema and payload are yours (docs/EVENT_CONTRACTS.md).

Source code in opennvr_app_sdk/facade.py

def publish(self, schema: str, payload: dict[str, Any], *,
            camera_id: str | None = None) -> bool:
    """Publish a contracted domain event — the thing another APP
    sees. The envelope, the producer (``app:<id>``), the camera and
    the correlation id are filled in; ``schema`` and ``payload`` are
    yours (docs/EVENT_CONTRACTS.md)."""
    return self._app.publisher.publish(
        schema, camera_id=camera_id or self.camera, payload=payload,
        correlation_id=self.correlation_id or None,
    )

publish_typed

publish_typed(
    payload: Any, *, camera_id: str | None = None
) -> bool

As :meth:publish, from a typed payload (PlateRecognized(...)) — the schema comes from the class, so a missing required field fails here rather than in someone else's app.

Source code in opennvr_app_sdk/facade.py

def publish_typed(self, payload: Any, *, camera_id: str | None = None) -> bool:
    """As :meth:`publish`, from a typed payload
    (``PlateRecognized(...)``) — the schema comes from the class, so
    a missing required field fails here rather than in someone
    else's app."""
    return self._app.publisher.publish_typed(
        payload, camera_id=camera_id or self.camera,
        correlation_id=self.correlation_id or None,
    )

remember

remember(**values: Any) -> None

Stash values on this object's presence record — readable on the next event for the same object via :meth:recall.

Source code in opennvr_app_sdk/facade.py

def remember(self, **values: Any) -> None:
    """Stash values on this object's presence record — readable on
    the next event for the same object via :meth:`recall`."""
    if self._record is not None:
        self._record.data.update(values)

recall

recall(name: str, default: Any = None) -> Any

Read back what :meth:remember stored for this object.

Source code in opennvr_app_sdk/facade.py

def recall(self, name: str, default: Any = None) -> Any:
    """Read back what :meth:`remember` stored for this object."""
    if self._record is None:
        return default
    return self._record.data.get(name, default)

opennvr_app_sdk.Alert dataclass

Alert(
    title: str,
    description: str,
    camera_id: str,
    severity: str = "high",
    source: AlertSource = AlertSource(),
    correlation_id: str | None = None,
    evidence: dict[str, Any] = dict(),
    tags: list[str] = list(),
    alert_type: str | None = None,
    images: dict[str, str] = dict(),
    alert_id: str = (
        lambda: f"alrt_{uuid.uuid4().hex[:12]}"
    )(),
    fired_at: str = (lambda: _utcnow_iso())(),
)

A single fired alert.

Maps 1:1 to the §11.5 alert wire shape: alert_id, fired_at, title, description, severity, source, camera_id, correlation_id, evidence, tags.

Severity levels are operator-visible — low / medium / high / critical per the design doc.

to_wire

to_wire() -> dict[str, Any]

Serialize to the §11.5 JSON shape.

Source code in opennvr_app_sdk/alerts.py

def to_wire(self) -> dict[str, Any]:
    """Serialize to the §11.5 JSON shape."""
    evidence = dict(self.evidence)
    if self.images:
        # Inside evidence, so an inbox that predates the images
        # column still receives the paths rather than nothing.
        evidence["images"] = dict(self.images)
    wire = {
        "alert_id": self.alert_id,
        "fired_at": self.fired_at,
        "title": self.title,
        "description": self.description,
        "severity": self.severity,
        "source": asdict(self.source),
        "camera_id": self.camera_id,
        "correlation_id": self.correlation_id,
        "evidence": evidence,
        "tags": list(self.tags),
    }
    if self.alert_type:
        wire["alert_type"] = self.alert_type
    return wire

opennvr_app_sdk.AppManifest dataclass

AppManifest(
    id: str,
    name: str,
    version: str,
    category: str,
    summary: str = "",
    requires_tasks: list[str] = list(),
    requires_adapters: list[str] = list(),
    requires_scopes: list[str] = list(),
    provides: list[str] = list(),
    subscribes: str | None = None,
    params: list[Param] = list(),
    emits: list[AlertType] = list(),
    state_schema: list[StateView] = list(),
    overlay: bool = False,
    camera_picker: bool = True,
    actions: list[Action] = list(),
    has_ui: bool = False,
    ui_mode: str = "internal",
    ui_url: str = "",
    description: str = "",
    author: str = "",
    website: str = "",
    license: str = "",
    use_cases: list[str] = list(),
    contact: str = "",
    pricing: str = "free",
    price_note: str = "",
    entitlement: str = "none",
)

The static identity + schema of one app.

subscribes is the NATS subject pattern for InferenceSubscriber apps (None for FrameApps that drive inference themselves). requires_tasks names adapter task types the app depends on, e.g. ["object_detection"]. requires_adapters (RFC-0002 Phase 3, decision 7) names the specific KAI-C adapters that must be PROVISIONED with the app — the installer ups them alongside the app and refcounts them across apps on uninstall. Adapters the standard stack already ships (yolov8) are reused, not listed.

to_dict

to_dict() -> dict[str, Any]

The GET /manifest payload (and the manifest_json snapshot the app registry stores).

Source code in opennvr_app_sdk/manifest.py

def to_dict(self) -> dict[str, Any]:
    """The ``GET /manifest`` payload (and the ``manifest_json``
    snapshot the app registry stores)."""
    return {
        "id": self.id,
        "name": self.name,
        "version": self.version,
        "category": self.category,
        "summary": self.summary,
        "requires_tasks": list(self.requires_tasks),
        "requires_adapters": list(self.requires_adapters),
        "requires_scopes": list(self.requires_scopes),
        "provides": list(self.provides),
        "subscribes": self.subscribes,
        "params": [p.to_dict() for p in self.params],
        "emits": [a.to_dict() for a in self.emits],
        "state_schema": [v.to_dict() for v in self.state_schema],
        "actions": [a.to_dict() for a in self.actions],
        "overlay": bool(self.overlay),
        "camera_picker": bool(self.camera_picker),
        "has_ui": bool(self.has_ui),
        "ui_mode": self.ui_mode,
        "ui_url": self.ui_url,
        "description": self.description,
        "author": self.author,
        "website": self.website,
        "license": self.license,
        "use_cases": list(self.use_cases),
        "contact": self.contact,
        "pricing": self.pricing,
        "price_note": self.price_note,
        "entitlement": self.entitlement,
    }

opennvr_app_sdk.Param dataclass

Param(
    name: str,
    type: Any,
    default: Any = None,
    per_camera: bool = False,
    description: str = "",
    required: bool = False,
    suggestions: list[str] = list(),
    label: str = "",
    group: str = "",
    advanced: bool = False,
    choices: list[Any] = list(),
)

One typed, declarative config knob.

per_camera=True marks params the catalog collects per camera (zones, tripwires) rather than once per app.

opennvr_app_sdk.setting

setting(name: str) -> Setting

Bind a rule filter to a config value: dwell=setting("dwell_s"). dwell="$dwell_s" means the same thing.

Source code in opennvr_app_sdk/facade.py

def setting(name: str) -> Setting:
    """Bind a rule filter to a config value: ``dwell=setting("dwell_s")``.
    ``dwell="$dwell_s"`` means the same thing."""
    return Setting(name)