The event bus

Contracted domain events are how one app becomes another app's input — versioned in the subject, defined in EVENT_CONTRACTS.md, typed here. Tier-0 is the always-on detector every camera already runs, free to consume.

opennvr_app_sdk.DomainEvent dataclass

DomainEvent(
    id: str,
    schema: str,
    camera_id: str,
    ts: str,
    payload: dict[str, Any],
    producer: str | None = None,
    correlation_id: str | None = None,
    subject: str = "",
    raw: dict[str, Any] = dict(),
)

One EVENT_CONTRACTS.md envelope.

typed

typed() -> Any

The payload as its contract class (PlateRecognized, AccessDecided, …, see event_types) — None when this SDK does not type the schema or the payload is off-contract.

Source code in opennvr_app_sdk/domain_subscriber.py

def typed(self) -> Any:
    """The payload as its contract class (``PlateRecognized``,
    ``AccessDecided``, …, see ``event_types``) — ``None`` when this
    SDK does not type the schema or the payload is off-contract."""
    from .event_types import typed_payload

    return typed_payload(self.schema, self.payload)

opennvr_app_sdk.parse_domain_event

parse_domain_event(
    data: bytes | str | dict, *, subject: str = ""
) -> DomainEvent | None

Decode + validate an envelope; None for anything off-contract.

Source code in opennvr_app_sdk/domain_subscriber.py

def parse_domain_event(data: bytes | str | dict, *, subject: str = "") -> DomainEvent | None:
    """Decode + validate an envelope; ``None`` for anything off-contract."""
    try:
        env = data if isinstance(data, dict) else json.loads(data)
    except (ValueError, UnicodeDecodeError):
        return None
    if not isinstance(env, dict):
        return None
    schema, camera_id, payload = env.get("schema"), env.get("camera_id"), env.get("payload")
    if not isinstance(schema, str) or not isinstance(camera_id, str) or not isinstance(payload, dict):
        return None
    return DomainEvent(
        id=str(env.get("id") or ""), schema=schema, camera_id=camera_id,
        ts=str(env.get("ts") or ""), payload=payload,
        producer=env.get("producer"), correlation_id=env.get("correlation_id"),
        subject=subject, raw=env)

opennvr_app_sdk.DomainEventPublisher

DomainEventPublisher(
    url: str,
    *,
    token: str | None = None,
    producer: str = "app",
)

Publish contracted domain events onto the platform bus.

One instance per app process; publish never raises on bus trouble (logged inside the channel, returns False) — an app's decision loop must not crash because the broker blinked.

Source code in opennvr_app_sdk/domain_events.py

def __init__(self, url: str, *, token: str | None = None,
             producer: str = "app") -> None:
    self._producer = producer
    self._channel = NatsAlertChannel(url, token=token)

publish_typed

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

Publish a typed payload (event_types): the schema is the class's, the wire payload is to_payload().

Source code in opennvr_app_sdk/domain_events.py

def publish_typed(self, payload: Any, *, camera_id: str,
                  correlation_id: str | None = None) -> bool:
    """Publish a typed payload (``event_types``): the schema is the
    class's, the wire payload is ``to_payload()``."""
    from .event_types import is_typed_payload

    if not is_typed_payload(payload):
        raise TypeError(f"publish_typed needs a typed event payload, got {type(payload).__name__}")
    return self.publish(type(payload).SCHEMA, camera_id=camera_id,
                        payload=payload.to_payload(), correlation_id=correlation_id)

publish_overlay

publish_overlay(
    camera_id: str,
    boxes: list[dict[str, Any]],
    *,
    frame: dict[str, int] | None = None,
    seq: int | None = None,
) -> bool

Ask core to draw boxes over camera_id's live video.

Sugar over publish_typed(OverlayBoxes(...)). Drawn only if the operator enabled this app's overlay in the App Catalog; otherwise the event is published and ignored, so an app can call this unconditionally. Never raises on bus trouble.

Source code in opennvr_app_sdk/domain_events.py

def publish_overlay(self, camera_id: str, boxes: list[dict[str, Any]], *,
                    frame: dict[str, int] | None = None,
                    seq: int | None = None) -> bool:
    """Ask core to draw ``boxes`` over ``camera_id``'s live video.

    Sugar over ``publish_typed(OverlayBoxes(...))``. Drawn only if the
    operator enabled this app's overlay in the App Catalog; otherwise
    the event is published and ignored, so an app can call this
    unconditionally. Never raises on bus trouble."""
    from .event_types import OverlayBoxes

    return self.publish_typed(
        OverlayBoxes(boxes=list(boxes), frame=frame, seq=seq),
        camera_id=camera_id)

opennvr_app_sdk.domain_envelope

domain_envelope(
    schema: str,
    *,
    camera_id: str,
    payload: dict[str, Any],
    producer: str,
    correlation_id: str | None = None,
) -> dict[str, Any]

The EVENT_CONTRACTS.md envelope — every field, every time (mirrors the KAI-C normaliser's builder).

Source code in opennvr_app_sdk/domain_events.py

def domain_envelope(
    schema: str,
    *,
    camera_id: str,
    payload: dict[str, Any],
    producer: str,
    correlation_id: str | None = None,
) -> dict[str, Any]:
    """The EVENT_CONTRACTS.md envelope — every field, every time
    (mirrors the KAI-C normaliser's builder)."""
    return {
        "id": "evt_" + uuid.uuid4().hex[:12],
        "schema": schema,
        "correlation_id": correlation_id,
        "camera_id": camera_id,
        "ts": datetime.now(timezone.utc).replace(microsecond=0).isoformat(),
        "producer": producer,
        "payload": payload,
    }

opennvr_app_sdk.domain_subject

domain_subject(schema: str, camera_id: str) -> str

The contracted subject for one event instance.

schema must match <domain>.<event>.v<N> and camera_id must be a single valid NATS token (the platform handle, camN) — both fail loudly, because a malformed subject silently reaches no subscriber.

Source code in opennvr_app_sdk/domain_events.py

def domain_subject(schema: str, camera_id: str) -> str:
    """The contracted subject for one event instance.

    ``schema`` must match ``<domain>.<event>.v<N>`` and ``camera_id``
    must be a single valid NATS token (the platform handle, ``camN``)
    — both fail loudly, because a malformed subject silently reaches
    no subscriber.
    """
    if not _SCHEMA_RE.match(schema):
        raise ValueError(
            f"domain event schema {schema!r} must look like "
            "'<domain>.<event>.v<N>' (see docs/EVENT_CONTRACTS.md)")
    if not camera_id or _CAMERA_TOKEN_BAD.search(camera_id):
        raise ValueError(
            f"camera_id {camera_id!r} is not a valid subject token")
    return f"opennvr.events.{schema}.{camera_id}"

opennvr_app_sdk.TypedPayload dataclass

TypedPayload(*, extra: dict[str, Any] = dict())

Base of every typed payload: SCHEMA, from_payload, to_payload. Subclasses list the contract's fields; extra keeps whatever else the producer sent.

to_payload

to_payload() -> dict[str, Any]

The wire payload: every contract field (nullable ones as null), then extra — never overriding a contract field.

Source code in opennvr_app_sdk/event_types.py

def to_payload(self) -> dict[str, Any]:
    """The wire payload: every contract field (nullable ones as
    ``null``), then ``extra`` — never overriding a contract field."""
    out: dict[str, Any] = {}
    for f in fields(self):
        if f.name == "extra":
            continue
        value = getattr(self, f.name)
        out[f.name] = list(value) if isinstance(value, tuple) else value
    for k, v in self.extra.items():
        out.setdefault(k, v)
    return out

opennvr_app_sdk.typed_payload

typed_payload(
    schema: str, payload: dict[str, Any]
) -> TypedPayload | None

The typed payload for schema, None when the schema is not one this SDK types (a newer contract, or an app's own) or the payload is off-contract (logged).

Source code in opennvr_app_sdk/event_types.py

def typed_payload(schema: str, payload: dict[str, Any]) -> TypedPayload | None:
    """The typed payload for ``schema``, ``None`` when the schema is not
    one this SDK types (a newer contract, or an app's own) or the
    payload is off-contract (logged)."""
    cls = EVENT_TYPES.get(schema)
    if cls is None:
        return None
    try:
        return cls.from_payload(payload)
    except ValueError as exc:
        logger.warning("domain event %s: payload off-contract: %s", schema, exc)
        return None

opennvr_app_sdk.EVENT_TYPES module-attribute

EVENT_TYPES: dict[str, type[TypedPayload]] = {
    cls.SCHEMA: cls
    for cls in (
        DetectionObserved,
        VisitRecorded,
        PlateRecognized,
        AccessDecided,
        OccupancyChanged,
        OccupancyHeatmap,
        OccupancyFootfall,
        OverlayBoxes,
        ScreeningCompleted,
    )
}

opennvr_app_sdk.DetectionObserved dataclass

DetectionObserved(
    frame: dict[str, Any] = dict(),
    tracks: list[dict[str, Any]] = list(),
    calibrating: bool = False,
    *,
    extra: dict[str, Any] = dict(),
)

Bases: TypedPayload

detection.observed.v1 — one Tier-0 frame result with tracks.

opennvr_app_sdk.VisitRecorded dataclass

VisitRecorded(
    event_id: int = 0,
    label: str = "",
    started_at: str = "",
    ended_at: str = "",
    evidence_path: str | None = None,
    *,
    extra: dict[str, Any] = dict(),
)

Bases: TypedPayload

visit.recorded.v1 — a finished timeline visit persisted by core.

opennvr_app_sdk.PlateRecognized dataclass

PlateRecognized(
    plate_text: str = "",
    confidence: float | None = None,
    vehicle_label: str | None = None,
    event_id: int | None = None,
    plate_box: list[float] | None = None,
    plate_box_confidence: float | None = None,
    plate_box_image: list[int] | None = None,
    observed_at: str | None = None,
    *,
    extra: dict[str, Any] = dict(),
)

Bases: TypedPayload

plate.recognized.v1 — one accepted OCR read.

opennvr_app_sdk.AccessDecided dataclass

AccessDecided(
    plate_text: str = "",
    decision: str = "deny",
    reason: str = "unknown",
    owner: str | None = None,
    unit: str | None = None,
    confidence: float | None = None,
    *,
    extra: dict[str, Any] = dict(),
)

Bases: TypedPayload

access.decided.v1 — an admission decision for a plate at a gate. Consumers actuate only on decision == "allow"; anything else, including values this SDK does not know, is "do not actuate".

opennvr_app_sdk.OccupancyChanged dataclass

OccupancyChanged(
    count: int = 0,
    level: str = "normal",
    max_occupancy: int | None = None,
    min_occupancy: int | None = None,
    *,
    extra: dict[str, Any] = dict(),
)

Bases: TypedPayload

occupancy.changed.v1 — a zone's head-count moved.

opennvr_app_sdk.OccupancyHeatmap dataclass

OccupancyHeatmap(
    cols: int = 0,
    rows: int = 0,
    cells: list[list[int]] = list(),
    frames: int = 0,
    period_seconds: int = 60,
    labels: list[str] = list(),
    *,
    extra: dict[str, Any] = dict(),
)

Bases: TypedPayload

occupancy.heatmap.v1 — a sparse delta of a per-camera heat grid.

opennvr_app_sdk.ScreeningCompleted dataclass

ScreeningCompleted(
    session: str = "",
    verdict: str = "",
    score: float = 0.0,
    at: str | None = None,
    ts: float | None = None,
    coverage: float | None = None,
    order_score: float | None = None,
    steps_done: list[str] = list(),
    steps_missing: list[str] = list(),
    flagged: bool = False,
    duration_s: float | None = None,
    engaged_s: float | None = None,
    ended_by: str | None = None,
    guard_key: str | None = None,
    guard_name: str | None = None,
    images: dict[str, str] = dict(),
    alert_id: str | None = None,
    *,
    extra: dict[str, Any] = dict(),
)

Bases: TypedPayload

screening.completed.v1 — one entry screening, ruled on.

Published when an app finishes judging whether a person was scanned properly at an entrance. Every screening is published, the clean ones included: compliance is complete scans over ALL screenings, and a stream of only the failures can count complaints but never state a rate.

steps_done / steps_missing are human labels ("Left arm"), not ids — the set of surfaces is the operator's configuration, not a fixed enum. images maps a role to an evidence path already uploaded through the platform, never to bytes: an alert is a NATS message and a couple of base64 crops exceed the default 1 MB ceiling, which the broker refuses silently.

opennvr_app_sdk.OccupancyFootfall dataclass

OccupancyFootfall(
    entries: int = 0,
    exits: int = 0,
    dwell_count: int = 0,
    dwell_seconds: float = 0.0,
    dwell_max_seconds: float = 0.0,
    period_seconds: int = 60,
    labels: list[str] = list(),
    *,
    extra: dict[str, Any] = dict(),
)

Bases: TypedPayload

occupancy.footfall.v1 — entries, exits and finished stays since the last publish.

opennvr_app_sdk.OverlayBoxes dataclass

OverlayBoxes(
    boxes: list[dict[str, Any]] = list(),
    frame: dict[str, int] | None = None,
    seq: int | None = None,
    *,
    extra: dict[str, Any] = dict(),
)

Bases: TypedPayload

overlay.boxes.v1 — boxes an app wants drawn over the live video.

boxes entries are {"label", "box": [x, y, w, h], "score"?, "id"?} with box normalized to 0..1 of the frame — an app never knows the resolution the operator is watching at, and core draws over streams of every size. Send frame: {w, h} if your boxes are in pixels and core will normalize; send nothing drawable and nothing is drawn.

Whether it IS drawn is the operator's call, per app, in the App Catalog (off by default). Publishing costs nothing when it is off.

opennvr_app_sdk.Tier0Snapshot dataclass

Tier0Snapshot(
    camera_id: str = "",
    tracks: list[dict[str, Any]] = list(),
    ts: float | None = None,
    seq: int | None = None,
)

The latest Tier-0 result for one camera, reduced to what apps ask of it.

Built from a Tier-0 event payload (snapshot_from_event). Defensive by design — the payload is whatever JSON arrived on the bus, so missing fields degrade to empty rather than raising.

counts property

counts: dict[str, int]

Object count per label, e.g. {"person": 1, "car": 2}.

has_best

has_best(track_id: Any) -> bool

Whether a fetchable best frame is advertised for a given track id.

Source code in opennvr_app_sdk/tier0.py

def has_best(self, track_id: Any) -> bool:
    """Whether a fetchable best frame is advertised for a given track id."""
    for t in self.tracks:
        if t.get("id") == track_id:
            return bool(t.get("best"))
    return False

describe

describe(*, limit: int = 8) -> str

A short, human/speakable phrase, e.g. "a person, 2 cars".

Source code in opennvr_app_sdk/tier0.py

def describe(self, *, limit: int = 8) -> str:
    """A short, human/speakable phrase, e.g. ``"a person, 2 cars"``."""
    return describe_counts(self.counts, limit=limit)

opennvr_app_sdk.snapshot_from_event

snapshot_from_event(
    payload: dict[str, Any],
) -> Tier0Snapshot

Parse a Tier-0 event payload (opennvr.tier0.v1) into a snapshot.

Source code in opennvr_app_sdk/tier0.py

def snapshot_from_event(payload: dict[str, Any]) -> Tier0Snapshot:
    """Parse a Tier-0 event payload (``opennvr.tier0.v1``) into a snapshot."""
    payload = payload or {}
    # Keep only dict tracks — the payload is whatever JSON arrived on the bus, so a
    # junk `tracks` (a string, or a list of ints) must degrade, not raise.
    tracks = [t for t in (payload.get("tracks") or []) if isinstance(t, dict)]
    return Tier0Snapshot(
        camera_id=str(payload.get("camera_id") or ""),
        tracks=tracks,
        ts=payload.get("ts"),
        seq=payload.get("seq"),
    )

opennvr_app_sdk.tier0_to_detections

tier0_to_detections(
    payload: dict[str, Any],
) -> list[dict[str, Any]]

Bridge a Tier-0 event's tracks into contract-shaped detections.

Lets any :class:DetectorApp consume the always-on Tier-0 stream with the same on_detections code it already runs on adapter events: each track becomes {label, score, bbox, track_id, stationary, best} where bbox is the contract's NormalizedBBox (x/y/w/h in 0-1), computed from the track's pixel box and the event's frame size.

Defensive: a malformed track is skipped; if the event carries no frame size (older detect-pipeline), bbox is omitted so bbox-free consumers (counting, presence) still work while zone tests skip the track.

Source code in opennvr_app_sdk/tier0.py

def tier0_to_detections(payload: dict[str, Any]) -> list[dict[str, Any]]:
    """Bridge a Tier-0 event's ``tracks`` into contract-shaped detections.

    Lets any :class:`DetectorApp` consume the always-on Tier-0 stream with the
    same ``on_detections`` code it already runs on adapter events: each track
    becomes ``{label, score, bbox, track_id, stationary, best}`` where ``bbox``
    is the contract's NormalizedBBox (x/y/w/h in 0-1), computed from the
    track's pixel box and the event's ``frame`` size.

    Defensive: a malformed track is skipped; if the event carries no frame
    size (older detect-pipeline), ``bbox`` is omitted so bbox-free consumers
    (counting, presence) still work while zone tests skip the track.
    """
    tracks = payload.get("tracks")
    if not isinstance(tracks, list):
        return []
    frame = payload.get("frame") or {}
    fw, fh = frame.get("w"), frame.get("h")
    have_dims = (
        isinstance(fw, (int, float)) and fw > 0
        and isinstance(fh, (int, float)) and fh > 0
    )
    out: list[dict[str, Any]] = []
    for t in tracks:
        if not isinstance(t, dict) or not t.get("label"):
            continue
        det: dict[str, Any] = {
            "label": str(t["label"]),
            "score": t.get("score"),
            "track_id": t.get("id"),
            "stationary": t.get("stationary"),
            "best": t.get("best"),
        }
        box = t.get("box")
        if have_dims and isinstance(box, (list, tuple)) and len(box) == 4:
            try:
                x1, y1, x2, y2 = (float(v) for v in box)
                det["bbox"] = {
                    "x": max(0.0, x1 / fw),
                    "y": max(0.0, y1 / fh),
                    "w": max(0.0, (x2 - x1) / fw),
                    "h": max(0.0, (y2 - y1) / fh),
                }
            except (TypeError, ValueError):
                pass
        out.append(det)
    return out

opennvr_app_sdk.is_tier0_subject

is_tier0_subject(subject: str) -> bool

True if a NATS subject is a Tier-0 inference event.

Source code in opennvr_app_sdk/tier0.py

def is_tier0_subject(subject: str) -> bool:
    """True if a NATS subject is a Tier-0 inference event."""
    return subject.startswith(TIER0_SUBJECT_PREFIX)

opennvr_app_sdk.describe_counts

describe_counts(
    counts: dict[str, int],
    *,
    irregular_plurals: dict[str, str] | None = None,
    limit: int = 8,
) -> str

Turn a label→count map into a short phrase: "a person, 2 cars".

Source code in opennvr_app_sdk/tier0.py

def describe_counts(counts: dict[str, int], *,
                    irregular_plurals: dict[str, str] | None = None,
                    limit: int = 8) -> str:
    """Turn a label→count map into a short phrase: ``"a person, 2 cars"``."""
    plurals = irregular_plurals or _IRREGULAR_PLURALS
    parts: list[str] = []
    for label, count in sorted(counts.items()):
        if count <= 0 or not label:
            continue
        if count == 1:
            article = "an" if label[:1].lower() in "aeiou" else "a"
            parts.append(f"{article} {label}")
        else:
            parts.append(f"{count} {plurals.get(label, f'{label}s')}")
    return ", ".join(parts[:limit])

opennvr_app_sdk.BestFrameClient

BestFrameClient(
    base_url: str,
    *,
    resolve_camera: ResolveCamera | None = None,
    http_get: HttpGet | None = None,
)

Fetch Tier-0's best frame from the detect-pipeline /best_frame endpoint.

base_url is the pipeline metrics origin (e.g. http://tier0:9109). resolve_camera optionally maps an app's camera id → the pipeline's camera id (identity by default). http_get is injectable for tests.

Source code in opennvr_app_sdk/tier0.py

def __init__(self, base_url: str, *, resolve_camera: ResolveCamera | None = None,
             http_get: HttpGet | None = None) -> None:
    self._base = base_url.rstrip("/")
    self._resolve = resolve_camera
    self._get = http_get or _default_http_get

fetch async

fetch(camera_id: str, track_id: Any = None) -> bytes | None

Best frame as JPEG bytes, or None. A specific track_id fetches that track's best; omitting it fetches the camera's most-recent best.

Source code in opennvr_app_sdk/tier0.py

async def fetch(self, camera_id: str, track_id: Any = None) -> bytes | None:
    """Best frame as JPEG bytes, or None. A specific ``track_id`` fetches that
    track's best; omitting it fetches the camera's most-recent best."""
    cam = self._resolve(camera_id) if self._resolve else camera_id
    if not cam:
        return None
    url = f"{self._base}/best_frame?camera={cam}"
    if track_id is not None:
        url += f"&track={track_id}"
    status, body = await self._get(url)
    return body if status == 200 and body else None

opennvr_app_sdk.make_best_frame_fetch

make_best_frame_fetch(
    base_url: str,
    *,
    resolve_camera: ResolveCamera | None = None,
    http_get: HttpGet | None = None,
)

Convenience: a bound async fetch(camera_id) -> bytes | None — the shape a consumer (e.g. the camera-agent's describe path) plugs in directly.

Source code in opennvr_app_sdk/tier0.py

def make_best_frame_fetch(base_url: str, *, resolve_camera: ResolveCamera | None = None,
                          http_get: HttpGet | None = None):
    """Convenience: a bound ``async fetch(camera_id) -> bytes | None`` — the shape
    a consumer (e.g. the camera-agent's describe path) plugs in directly."""
    client = BestFrameClient(base_url, resolve_camera=resolve_camera, http_get=http_get)

    async def fetch(camera_id: str) -> bytes | None:
        return await client.fetch(camera_id)

    return fetch