Writing a rule

Where the object is, how long it has been there, and what to fire when it matters. Most rules are a zone, a dwell timer and an alert; these are those three, unwrapped.

opennvr_app_sdk.Zone dataclass

Zone(name: str, polygon: list[Point])

A named polygonal zone in pixel coordinates.

Polygon vertices are pixel coords on the camera frame. The polygon is implicitly closed (we connect the last vertex back to the first). A degenerate zone with < 3 vertices is rejected.

contains

contains(point: Point) -> bool

True if the point lies inside the polygon (ray-casting).

Source code in opennvr_app_sdk/geometry.py

def contains(self, point: Point) -> bool:
    """True if the point lies inside the polygon (ray-casting)."""
    return _point_in_polygon(point, self.polygon)

from_config classmethod

from_config(
    name: str, vertices: Sequence[Sequence[float]]
) -> "Zone"

Build a Zone from config-style vertex list [[x, y], [x, y], ...].

Source code in opennvr_app_sdk/geometry.py

@classmethod
def from_config(cls, name: str, vertices: Sequence[Sequence[float]]) -> "Zone":
    """Build a Zone from config-style vertex list ``[[x, y], [x, y], ...]``."""
    return cls(
        name=name,
        polygon=[Point(float(v[0]), float(v[1])) for v in vertices],
    )

opennvr_app_sdk.Tripwire dataclass

Tripwire(
    name: str,
    a: Point,
    b: Point,
    count_direction: str = "both",
)

An oriented line segment A→B with a counted direction.

count_direction is one of "a_to_b", "b_to_a", "both".

side

side(p: Point) -> float

Signed side of point p relative to the oriented line A→B. Positive = left, negative = right, ~0 = on the line.

Source code in opennvr_app_sdk/geometry.py

def side(self, p: Point) -> float:
    """Signed side of point ``p`` relative to the oriented line A→B.
    Positive = left, negative = right, ~0 = on the line."""
    return (self.b.x - self.a.x) * (p.y - self.a.y) - (
        self.b.y - self.a.y
    ) * (p.x - self.a.x)

crossing

crossing(prev: Point, curr: Point) -> str | None

Return the crossing direction ("a_to_b" / "b_to_a") if the movement prev → curr crosses this tripwire in a counted direction, else None.

Two conditions must both hold: 1. the segment prev→curr intersects the segment A→B (so the object physically traversed the wire), and 2. the side sign flipped from prev to curr (so it genuinely changed sides, not merely touched the line).

Source code in opennvr_app_sdk/geometry.py

def crossing(self, prev: Point, curr: Point) -> str | None:
    """Return the crossing direction (``"a_to_b"`` / ``"b_to_a"``) if
    the movement ``prev → curr`` crosses this tripwire in a counted
    direction, else ``None``.

    Two conditions must both hold:
    1. the segment prev→curr intersects the segment A→B
       (so the object physically traversed the wire), and
    2. the side sign flipped from prev to curr (so it genuinely
       changed sides, not merely touched the line).
    """
    side_prev = self.side(prev)
    side_curr = self.side(curr)
    # Must end up on opposite sides (strict) — grazing the line
    # (one side == 0) is not a committed crossing.
    if side_prev == 0 or side_curr == 0:
        return None
    if (side_prev > 0) == (side_curr > 0):
        return None
    if not _segments_intersect(prev, curr, self.a, self.b):
        return None
    # side_prev > 0 means it started on the LEFT of A→B and ended on
    # the right → it moved across in the A→B-rightward sense, which
    # we name "a_to_b". The opposite is "b_to_a".
    direction = "a_to_b" if side_prev > 0 else "b_to_a"
    if self.count_direction in (direction, "both"):
        return direction
    return None

opennvr_app_sdk.Point dataclass

Point(x: float, y: float)

2D point in pixel coordinates (origin top-left, y-down).

opennvr_app_sdk.bbox_center

bbox_center(
    bbox_normalized: dict,
    frame_width: int,
    frame_height: int,
) -> Point

Convert a §5.1 NormalizedBBox (x/y/w/h in [0, 1]) into a pixel-space center point given the camera's actual frame size. Defensive against partial bboxes: missing keys default to 0.

Source code in opennvr_app_sdk/geometry.py

def bbox_center(bbox_normalized: dict, frame_width: int, frame_height: int) -> Point:
    """Convert a §5.1 ``NormalizedBBox`` (x/y/w/h in [0, 1]) into a
    pixel-space center point given the camera's actual frame size.
    Defensive against partial bboxes: missing keys default to 0."""
    def _coerce(key: str) -> float:
        value = bbox_normalized.get(key, 0.0)
        try:
            return float(value)
        except (TypeError, ValueError):
            return 0.0
    x = _coerce("x")
    y = _coerce("y")
    w = _coerce("w")
    h = _coerce("h")
    return Point(
        x=(x + w / 2.0) * frame_width,
        y=(y + h / 2.0) * frame_height,
    )

opennvr_app_sdk.full_frame_polygon

full_frame_polygon(
    size: int = UNIT_FRAME,
) -> list[list[int]]

The whole-frame zone, in the unit space auto-derived zones use.

Source code in opennvr_app_sdk/cameras.py

def full_frame_polygon(size: int = UNIT_FRAME) -> list[list[int]]:
    """The whole-frame zone, in the unit space auto-derived zones use."""
    return [[0, 0], [size, 0], [size, size], [0, size]]

opennvr_app_sdk.keyed_state

keyed_state(
    ttl: float,
    *,
    auto_gc: bool = True,
    record_factory: Callable[
        ..., StateRecord
    ] = StateRecord,
    clock: Callable[[], float] = time.time,
) -> KeyedState

Build a :class:KeyedState — TTL + latch + GC per §04 of the App SDK spec. ttl is in seconds of event time (whatever timeline you pass to touch(at=...)).

Source code in opennvr_app_sdk/state.py

def keyed_state(
    ttl: float,
    *,
    auto_gc: bool = True,
    record_factory: Callable[..., StateRecord] = StateRecord,
    clock: Callable[[], float] = time.time,
) -> KeyedState:
    """Build a :class:`KeyedState` — TTL + latch + GC per §04 of the
    App SDK spec. ``ttl`` is in seconds of *event time* (whatever
    timeline you pass to ``touch(at=...)``)."""
    return KeyedState(ttl, auto_gc=auto_gc, record_factory=record_factory, clock=clock)

opennvr_app_sdk.KeyedState

KeyedState(
    ttl: float,
    *,
    auto_gc: bool = True,
    record_factory: Callable[
        ..., StateRecord
    ] = StateRecord,
    clock: Callable[[], float] = time.time,
)

A TTL-pruned mapping of hashable keys to :class:StateRecord.

Build via :func:keyed_state. Dict-like surface: get / pop / items / in / len / [key].

Source code in opennvr_app_sdk/state.py

def __init__(
    self,
    ttl: float,
    *,
    auto_gc: bool = True,
    record_factory: Callable[..., StateRecord] = StateRecord,
    clock: Callable[[], float] = time.time,
) -> None:
    if ttl <= 0:
        raise ValueError(f"keyed_state: ttl must be > 0, got {ttl!r}")
    self.ttl = float(ttl)
    self._auto_gc = auto_gc
    self._factory = record_factory
    self._clock = clock
    self._records: dict[Hashable, StateRecord] = {}

touch

touch(
    key: Hashable, at: float | None = None
) -> StateRecord

Record a presence ping for key at time at (defaults to wall clock). Creates the record on first touch (fresh first_seen, alerted=False); refreshes last_seen on subsequent touches. With auto_gc enabled, prunes OTHER stale keys first — never the touched key itself.

Source code in opennvr_app_sdk/state.py

def touch(self, key: Hashable, at: float | None = None) -> StateRecord:
    """Record a presence ping for ``key`` at time ``at`` (defaults
    to wall clock). Creates the record on first touch (fresh
    ``first_seen``, ``alerted=False``); refreshes ``last_seen`` on
    subsequent touches. With ``auto_gc`` enabled, prunes OTHER
    stale keys first — never the touched key itself."""
    now = self._clock() if at is None else float(at)
    if self._auto_gc:
        self.gc(now, exclude=(key,))
    record = self._records.get(key)
    if record is None:
        record = self._factory(first_seen=now, last_seen=now)
        self._records[key] = record
    else:
        record.last_seen = now
    return record

gc

gc(
    now: float | None = None, *, exclude: Any = ()
) -> list[tuple[Hashable, StateRecord]]

Prune every key whose last_seen is more than ttl seconds before now (strictly older — a record exactly at the TTL boundary survives, matching the grace-period semantics the loitering detector shipped with). Keys in exclude are kept regardless. Returns the pruned (key, record) pairs so lifecycle apps can emit disappearance events.

Source code in opennvr_app_sdk/state.py

def gc(
    self,
    now: float | None = None,
    *,
    exclude: Any = (),
) -> list[tuple[Hashable, StateRecord]]:
    """Prune every key whose ``last_seen`` is more than ``ttl``
    seconds before ``now`` (strictly older — a record exactly at
    the TTL boundary survives, matching the grace-period semantics
    the loitering detector shipped with). Keys in ``exclude`` are
    kept regardless. Returns the pruned ``(key, record)`` pairs so
    lifecycle apps can emit disappearance events."""
    now = self._clock() if now is None else float(now)
    cutoff = now - self.ttl
    excluded = set(exclude)
    pruned = [
        (key, record)
        for key, record in self._records.items()
        if key not in excluded and record.last_seen < cutoff
    ]
    for key, _record in pruned:
        del self._records[key]
    return pruned

items

items() -> list[tuple[Hashable, StateRecord]]

Snapshot list — safe to pop while iterating.

Source code in opennvr_app_sdk/state.py

def items(self) -> list[tuple[Hashable, StateRecord]]:
    """Snapshot list — safe to ``pop`` while iterating."""
    return list(self._records.items())

opennvr_app_sdk.StateRecord dataclass

StateRecord(
    first_seen: float,
    last_seen: float,
    alerted: bool = False,
    data: dict[str, Any] = dict(),
)

Bookkeeping for one key.

first_seen is the timestamp of the first touch since the record was created (or re-created after a GC) — loitering reads it as present_since. last_seen is the most recent touch. alerted is a caller-settable latch so a threshold-crossing alert fires once per episode. data is a free-form scratchpad for app-specific flags (state-machine phase, counters, …); subclassing and passing record_factory works too when you want typed fields.

age property

age: float

Seconds between the first and the most recent touch — i.e. the dwell time as of the last touch.

opennvr_app_sdk.AlertType dataclass

AlertType(
    name: str,
    severity: str = "medium",
    description: str = "",
)

One alert kind the app can emit — drives catalog documentation and downstream routing defaults.

opennvr_app_sdk.AlertSource dataclass

AlertSource(
    kind: str = (lambda: _DEFAULT_SOURCE.get()["kind"])(),
    name: str = (lambda: _DEFAULT_SOURCE.get()["name"])(),
    version: str = (
        lambda: _DEFAULT_SOURCE.get()["version"]
    )(),
)

The source block of the §11.5 alert envelope.

Field defaults come from the context default set via :func:set_default_source — one of: kind = kai-c / adapter / app.

opennvr_app_sdk.AlertChannel

Bases: Protocol

Anything that can send(alert) is a channel. Stdout + webhook are the v1 implementations; future channels (OpenNVR alerts API) plug in here without touching the detector loop.

opennvr_app_sdk.AlertDispatcher

AlertDispatcher(channels: list[AlertChannel])

Holds an ordered list of channels and fires an alert through all of them, isolating each channel's failures.

fire() returns a per-channel report so the caller can audit delivery outcomes; the detector loop ignores this in v1 but the OpenNVR alerts-API integration (planned follow-up) will record it.

Source code in opennvr_app_sdk/alerts.py

def __init__(self, channels: list[AlertChannel]) -> None:
    if not channels:
        raise ValueError("AlertDispatcher requires at least one channel.")
    self._channels = channels

channels property

channels: tuple[AlertChannel, ...]

The channels, in delivery order — read-only. Handy in tests and in a /state view ("where do my alerts go?").

close

close() -> None

Drain + tear down any channel that has a close method.

Stdout and webhook channels are stateless and have no close() — only NatsAlertChannel does. Wire this into the detector's SIGINT/SIGTERM finally clause so in-flight NATS publishes get drained instead of dropped (peer-review H2). Per-channel failures are logged but never raise, so shutdown stays clean even if one channel's close hangs.

Source code in opennvr_app_sdk/alerts.py

def close(self) -> None:
    """Drain + tear down any channel that has a ``close`` method.

    Stdout and webhook channels are stateless and have no close()
    — only ``NatsAlertChannel`` does. Wire this into the detector's
    SIGINT/SIGTERM finally clause so in-flight NATS publishes get
    drained instead of dropped (peer-review H2). Per-channel
    failures are logged but never raise, so shutdown stays clean
    even if one channel's close hangs.
    """
    for channel in self._channels:
        close = getattr(channel, "close", None)
        if close is None:
            continue
        channel_name = getattr(channel, "name", channel.__class__.__name__)
        try:
            close()
        except Exception as exc:  # noqa: BLE001
            logger.warning(
                "channel %s close failed: %s", channel_name, exc,
            )

opennvr_app_sdk.StdoutChannel

Always-on channel: human-readable line to stdout.

opennvr_app_sdk.WebhookChannel

WebhookChannel(url: str, *, timeout_seconds: float = 5.0)

POST the alert's JSON wire shape to an operator-configured URL.

Failures are LOGGED but never raise — a dead webhook should not prevent stdout alerts from firing or crash the detector loop. This matches the §11.2 "audit forwarding failures are themselves audited" pattern (though we don't yet have an audit sink here — that lands when the SDK talks to KAI-C's audit log directly).

Source code in opennvr_app_sdk/alerts.py

def __init__(self, url: str, *, timeout_seconds: float = 5.0) -> None:
    self._url = url
    self._timeout = timeout_seconds

opennvr_app_sdk.NatsAlertChannel

NatsAlertChannel(
    url: str,
    *,
    token: str | None = None,
    subject_prefix: str = DEFAULT_ALERT_SUBJECT_PREFIX,
    connect_timeout_seconds: float = _NATS_CONNECT_TIMEOUT_SECONDS,
    publish_timeout_seconds: float = _NATS_PUBLISH_TIMEOUT_SECONDS,
)

Publish each alert as JSON onto a NATS subject derived from the §11.5 source block.

Why this exists

The other two channels are point-to-point: stdout goes to one operator, the webhook goes to one URL. NATS is bus-shaped — N subscribers can fan out off the same publish (operator UI inbox, SIEM bridge, Slack bot, audit forwarder) without the publishing app knowing they exist. Same pattern KAI-C uses for inference events under the NATS event bus.

Implementation notes

AlertChannel.send is synchronous and called from a detector loop that may itself be sync (HTTP poll mode) or async (NATS subscriber / WS streaming mode). To keep the protocol uniform we run a background daemon thread that owns an asyncio event loop; send schedules the publish coroutine onto it via run_coroutine_threadsafe and waits for the result with a hard timeout. This insulates the dispatcher from the async-ness of nats-py and keeps publish failures isolated to this channel.

Failures (broker down, slow connect, bad credentials) are LOGGED but never raise — same contract as WebhookChannel. The detector loop should never crash because the bus is down.

Source code in opennvr_app_sdk/alerts.py

def __init__(
    self,
    url: str,
    *,
    token: str | None = None,
    subject_prefix: str = DEFAULT_ALERT_SUBJECT_PREFIX,
    connect_timeout_seconds: float = _NATS_CONNECT_TIMEOUT_SECONDS,
    publish_timeout_seconds: float = _NATS_PUBLISH_TIMEOUT_SECONDS,
) -> None:
    # Validate the prefix at construction (not at publish-time)
    # so a bogus operator config fails loudly before alerts start
    # flowing. NATS subjects allow dots as token separators (the
    # prefix is multi-token, e.g. ``opennvr.alerts``), so we
    # split-and-validate per-token rather than re-using the
    # single-token sanitizer (peer-review L6).
    if not subject_prefix or not subject_prefix.strip("."):
        raise ValueError(
            f"NatsAlertChannel: subject_prefix must not be empty, "
            f"got {subject_prefix!r}"
        )
    for token_seg in subject_prefix.split("."):
        if not token_seg or _SUBJECT_TOKEN_BAD.search(token_seg):
            raise ValueError(
                f"NatsAlertChannel: subject_prefix {subject_prefix!r} "
                f"contains a NATS-invalid token {token_seg!r}. Each "
                f"dot-separated segment must match [A-Za-z0-9_-]+."
            )
    self._url = url
    self._token = token
    self._subject_prefix = subject_prefix
    self._connect_timeout = connect_timeout_seconds
    self._publish_timeout = publish_timeout_seconds
    # Created lazily on first send so importing this module
    # doesn't spawn a thread for users who didn't enable NATS.
    self._loop: asyncio.AbstractEventLoop | None = None
    self._thread: threading.Thread | None = None
    self._nc: Any = None
    # Guards the (loop, thread) tuple's lifecycle from cross-thread
    # ``send`` calls and ``close`` calls. Distinct from the in-loop
    # ``_connect_lock`` below, which guards against two concurrent
    # ``_publish_once`` coroutines both calling ``nats.connect``.
    self._lock = threading.Lock()
    # Created together with the loop because asyncio.Lock can't be
    # safely constructed outside the loop's thread.
    self._connect_lock: asyncio.Lock | None = None

publish_json

publish_json(subject: str, obj: Any) -> bool

Publish obj as JSON on subject — the channel's machinery (background loop, lazy connect, hard timeouts, log-never-raise) for ANY payload. send is now a thin wrapper; domain_events.DomainEventPublisher composes this to publish contracted opennvr.events.* envelopes.

Source code in opennvr_app_sdk/alerts.py

def publish_json(self, subject: str, obj: Any) -> bool:
    """Publish ``obj`` as JSON on ``subject`` — the channel's
    machinery (background loop, lazy connect, hard timeouts,
    log-never-raise) for ANY payload. ``send`` is now a thin
    wrapper; ``domain_events.DomainEventPublisher`` composes this
    to publish contracted ``opennvr.events.*`` envelopes."""
    payload = json.dumps(obj).encode("utf-8")
    try:
        self._ensure_thread()
    except Exception as exc:  # noqa: BLE001
        logger.warning("NATS channel thread start failed: %s", exc)
        return False
    assert self._loop is not None
    budget = self._connect_timeout + self._publish_timeout
    future = asyncio.run_coroutine_threadsafe(
        self._publish_once(subject, payload),
        self._loop,
    )
    try:
        return future.result(timeout=budget)
    except Exception as exc:  # noqa: BLE001 — includes TimeoutError
        logger.warning(
            "NATS publish to %r timed out / failed: %s", subject, exc,
        )
        return False

close

close() -> None

Drain pending publishes and stop the background thread.

Called from the detector's shutdown path. Safe to call even if send was never invoked (no thread to clean up). After close() the channel is re-init safe: the next send() spins a fresh loop + thread and reconnects from scratch.

Source code in opennvr_app_sdk/alerts.py

def close(self) -> None:
    """Drain pending publishes and stop the background thread.

    Called from the detector's shutdown path. Safe to call even if
    ``send`` was never invoked (no thread to clean up). After
    ``close()`` the channel is re-init safe: the next ``send()``
    spins a fresh loop + thread and reconnects from scratch.
    """
    with self._lock:
        loop = self._loop
        nc = self._nc
        thread = self._thread
        # Reset so a subsequent ``send`` rebuilds cleanly rather
        # than scheduling on a stopped loop and timing out at
        # ``budget`` (peer-review M1).
        self._loop = None
        self._thread = None
        self._nc = None
        self._connect_lock = None

    if loop is None:
        return

    if nc is not None and loop.is_running():
        try:
            asyncio.run_coroutine_threadsafe(
                nc.drain(), loop,
            ).result(timeout=_NATS_DRAIN_TIMEOUT_SECONDS)
        except Exception as exc:  # noqa: BLE001
            logger.warning("NATS drain failed: %s", exc)

    if loop.is_running():
        loop.call_soon_threadsafe(loop.stop)
    if thread is not None:
        thread.join(timeout=3.0)

opennvr_app_sdk.build_dispatcher

build_dispatcher(
    *,
    webhook_url: str | None,
    nats_alerts_url: str | None = None,
    nats_alerts_token: str | None = None,
    nats_alerts_subject_prefix: str = DEFAULT_ALERT_SUBJECT_PREFIX,
) -> AlertDispatcher

Convenience factory used by app config loading. stdout is always included; webhook and NATS are independently opt-in via config.

Order matters: stdout fires first (operator-visible immediately), then webhook (still typically the fastest external sink), then NATS (bus fan-out for consumers that don't need synchronous delivery). Each channel's failure is isolated by AlertDispatcher.fire.

Source code in opennvr_app_sdk/alerts.py

def build_dispatcher(
    *,
    webhook_url: str | None,
    nats_alerts_url: str | None = None,
    nats_alerts_token: str | None = None,
    nats_alerts_subject_prefix: str = DEFAULT_ALERT_SUBJECT_PREFIX,
) -> AlertDispatcher:
    """Convenience factory used by app config loading. stdout is always
    included; webhook and NATS are independently opt-in via config.

    Order matters: stdout fires first (operator-visible immediately),
    then webhook (still typically the fastest external sink), then NATS
    (bus fan-out for consumers that don't need synchronous delivery).
    Each channel's failure is isolated by ``AlertDispatcher.fire``.
    """
    channels: list[AlertChannel] = [StdoutChannel()]
    if webhook_url:
        channels.append(WebhookChannel(webhook_url))
    if nats_alerts_url:
        channels.append(
            NatsAlertChannel(
                nats_alerts_url,
                token=nats_alerts_token,
                subject_prefix=nats_alerts_subject_prefix,
            )
        )
    return AlertDispatcher(channels)

opennvr_app_sdk.alert_subject

alert_subject(
    alert: "Alert",
    *,
    prefix: str = DEFAULT_ALERT_SUBJECT_PREFIX,
) -> str

Derive the NATS subject for one alert.

Shape: {prefix}.{source.kind}.{source.name}.{camera_id} with each segment sanitized so the result is a valid NATS subject (no spaces, no dots, no NATS reserved * / >).

Sanitization rule: any character outside [A-Za-z0-9_-] becomes _. Empty-after-sanitization segments fall back to "unknown" so a malformed Alert can't produce opennvr.alerts...cam-X.

Pulled out as a module-level function so tests can assert subject derivation without spinning up NATS.

Source code in opennvr_app_sdk/alerts.py

def alert_subject(alert: "Alert", *, prefix: str = DEFAULT_ALERT_SUBJECT_PREFIX) -> str:
    """Derive the NATS subject for one alert.

    Shape: ``{prefix}.{source.kind}.{source.name}.{camera_id}`` with
    each segment sanitized so the result is a valid NATS subject (no
    spaces, no dots, no NATS reserved ``*`` / ``>``).

    Sanitization rule: any character outside ``[A-Za-z0-9_-]`` becomes
    ``_``. Empty-after-sanitization segments fall back to ``"unknown"``
    so a malformed Alert can't produce ``opennvr.alerts...cam-X``.

    Pulled out as a module-level function so tests can assert subject
    derivation without spinning up NATS.
    """
    return (
        f"{prefix}."
        f"{_sanitize_subject_token(alert.source.kind)}."
        f"{_sanitize_subject_token(alert.source.name)}."
        f"{_sanitize_subject_token(alert.camera_id)}"
    )

opennvr_app_sdk.set_default_source

set_default_source(
    *,
    kind: str | None = None,
    name: str | None = None,
    version: str | None = None,
) -> None

Set the default AlertSource identity for the current context.

Call once at app startup for single-app processes. The Detector base instead scopes its identity around each handler call, so multiple detectors in one process don't fight. Only affects AlertSource instances created WITHOUT explicit values.

Source code in opennvr_app_sdk/alerts.py

def set_default_source(
    *,
    kind: str | None = None,
    name: str | None = None,
    version: str | None = None,
) -> None:
    """Set the default ``AlertSource`` identity for the current context.

    Call once at app startup for single-app processes. The ``Detector``
    base instead scopes its identity around each handler call, so
    multiple detectors in one process don't fight. Only affects
    ``AlertSource`` instances created WITHOUT explicit values."""
    current = dict(_DEFAULT_SOURCE.get())
    if kind is not None:
        current["kind"] = kind
    if name is not None:
        current["name"] = name
    if version is not None:
        current["version"] = version
    _DEFAULT_SOURCE.set(current)

opennvr_app_sdk.DEFAULT_ALERT_SUBJECT_PREFIX module-attribute

DEFAULT_ALERT_SUBJECT_PREFIX = 'opennvr.alerts'

opennvr_app_sdk.DETECTION_LABELS module-attribute

DETECTION_LABELS: tuple[str, ...] = (
    "person",
    "bicycle",
    "car",
    "motorcycle",
    "bus",
    "truck",
    "boat",
    "train",
    "airplane",
    "traffic light",
    "fire hydrant",
    "stop sign",
    "bench",
    "backpack",
    "umbrella",
    "handbag",
    "suitcase",
    "dog",
    "cat",
    "bird",
    "horse",
    "sheep",
    "cow",
    "elephant",
    "bear",
    "zebra",
    "giraffe",
    "bottle",
    "cup",
    "knife",
    "laptop",
    "cell phone",
    "chair",
    "couch",
    "potted plant",
    "bed",
    "dining table",
    "tv",
    "book",
    "clock",
)

opennvr_app_sdk.Setting dataclass

Setting(name: str)

A rule filter that reads an operator-set config value.

Decorator arguments are evaluated at import time, so a literal dwell=30 can never follow the config form. Wrap the param name instead — dwell=setting("dwell_s"), or the shorthand dwell="$dwell_s" — and the value is read from the parsed config when the app starts.

opennvr_app_sdk.DEFAULT_ABSENCE_S module-attribute

DEFAULT_ABSENCE_S = 30.0

opennvr_app_sdk.DEFAULT_MIN_CONFIDENCE module-attribute

DEFAULT_MIN_CONFIDENCE = 0.0