Archetypes

The classes App compiles to. Subclass one directly when the decorators stop fitting — several interacting state machines, a custom event walk, or a rule that needs the whole detection batch at once. Same process, same manifest, same alerts.

opennvr_app_sdk.Detector

Detector(
    config: Any,
    dispatcher: AlertDispatcher,
    *,
    clock: Callable[[], datetime] | None = None,
)

Bases: ContractMixin, NatsSubscriberMixin

Base class for NATS-subscribing detection apps.

Subclasses set a class-level manifest (:class:AppManifest), optionally override :meth:setup to allocate state, and implement :meth:on_detections. cfg is the app-parsed config object; the NATS loop reads cfg.nats_url, cfg.nats_token (optional) and cfg.subject_pattern from it.

clock is a callable returning a UTC datetime; per-event timestamps from the NATS payload are preferred, but timestamp parsing needs a "now" fallback for missing / malformed values. Tests pass a controlled clock for determinism.

Source code in opennvr_app_sdk/detector.py

def __init__(
    self,
    config: Any,
    dispatcher: AlertDispatcher,
    *,
    clock: Callable[[], _dt.datetime] | None = None,
) -> None:
    self.cfg = config
    # Compat alias — pre-SDK detectors (and their tests) used
    # ``self._config``.
    self._config = config
    self._dispatcher = dispatcher
    self._clock = clock or (lambda: _dt.datetime.now(_dt.timezone.utc))
    # Opt-in Tier-0 consumption (see docs/tier0-consumption.md): when
    # True, tier0 events are bridged into contract-shaped detections and
    # flow through on_detections like any adapter event. Off by default:
    # an app also subscribed to a heavy adapter would otherwise process
    # the same object twice (double alerts).
    self.consume_tier0: bool = bool(getattr(config, "consume_tier0", False))
    self._stop_event = asyncio.Event()
    self._nc: Any = None
    # This detector emits alerts AS this app. The identity is scoped
    # around each handler call (not set process-wide) so several
    # detectors can share one process — the camera agent's
    # create_monitor case — without clobbering each other's source.
    self._source_block: dict[str, str] | None = (
        {"kind": "app", "name": self.manifest.id, "version": self.manifest.version}
        if self.manifest is not None
        else None
    )
    self._contract_init()
    self.setup()

setup

setup() -> None

Optional hook — allocate per-app state (keyed_state et al.). Runs once at construction, after cfg is set.

Source code in opennvr_app_sdk/detector.py

def setup(self) -> None:
    """Optional hook — allocate per-app state (``keyed_state`` et
    al.). Runs once at construction, after ``cfg`` is set."""

on_detections

on_detections(
    camera_id: str,
    detections: list[dict[str, Any]],
    event: dict[str, Any],
) -> Iterable[Alert] | None

The rule. Called once per decoded inference event that has a camera_id and a result.detections list. Return or yield the :class:Alert objects to fire (or None / empty).

Source code in opennvr_app_sdk/detector.py

def on_detections(
    self,
    camera_id: str,
    detections: list[dict[str, Any]],
    event: dict[str, Any],
) -> Iterable[Alert] | None:
    """The rule. Called once per decoded inference event that has a
    ``camera_id`` and a ``result.detections`` list. Return or yield
    the :class:`Alert` objects to fire (or ``None`` / empty)."""
    raise NotImplementedError

keyed_state

keyed_state(ttl: float, **kwargs: Any) -> KeyedState

Convenience for setup() — see :func:~.state.keyed_state.

Source code in opennvr_app_sdk/detector.py

def keyed_state(self, ttl: float, **kwargs: Any) -> KeyedState:
    """Convenience for ``setup()`` — see :func:`~.state.keyed_state`."""
    return _keyed_state(ttl, **kwargs)

handle_event

handle_event(event: Any) -> list[Alert]

Process one decoded InferenceCompletedEvent dict: extract camera_id + result.detections (defensively — malformed shapes return []), delegate to :meth:on_detections, and dispatch every returned alert. Returns the list of alerts fired.

Source code in opennvr_app_sdk/detector.py

def handle_event(self, event: Any) -> list[Alert]:
    """Process one decoded ``InferenceCompletedEvent`` dict:
    extract ``camera_id`` + ``result.detections`` (defensively —
    malformed shapes return ``[]``), delegate to
    :meth:`on_detections`, and dispatch every returned alert.
    Returns the list of alerts fired."""
    # Contract counters (spec §03): every decoded event counts as
    # "seen" — /health's last_event_age_s is stall detection for
    # the pipe, not a per-shape metric.
    is_tier0 = (
        isinstance(event, dict)
        and (
            event.get("schema") == "opennvr.tier0.v1"
            or event.get("adapter") == "tier0"
        )
    )
    if is_tier0 and not self.consume_tier0:
        # Not counted as "seen": tier0 publishes per-frame, and letting it
        # refresh last_event_age_s would mask a stalled adapter for apps
        # that don't consume tier0 at all.
        return []
    self._contract_note_event()
    if not isinstance(event, dict):
        return []
    camera_id = event.get("camera_id")
    if not camera_id:
        return []
    if not self.camera_picked(camera_id):
        # Not one of this app's cameras. The bus carries every
        # camera's detections; acting only on picked ones is what
        # "nothing picked = the app does nothing" means for a
        # subscriber.
        return []
    if is_tier0:
        from .tier0 import tier0_to_detections

        detections: Any = tier0_to_detections(event)
    else:
        result = event.get("result") or {}
        detections = (
            result.get("detections") if isinstance(result, dict) else None
        )
    if not isinstance(detections, list):
        return []

    token = scoped_default_source(self._source_block) if self._source_block else None
    try:
        produced = self.on_detections(camera_id, detections, event)
        fired: list[Alert] = []
        if produced is None:
            return fired
        for alert in produced:
            self._dispatcher.fire(alert)
            fired.append(alert)
        self._contract_note_alerts(len(fired))
        return fired
    finally:
        if token is not None:
            reset_default_source(token)

parse_event_ts

parse_event_ts(raw: Any) -> float

Extract a POSIX timestamp from the NATS event bus completed_at ISO string. Falls back to the clock for missing / malformed values so a misbehaving publisher doesn't break app state machines.

Source code in opennvr_app_sdk/detector.py

def parse_event_ts(self, raw: Any) -> float:
    """Extract a POSIX timestamp from the NATS event bus
    ``completed_at`` ISO string. Falls back to the clock for
    missing / malformed values so a misbehaving publisher doesn't
    break app state machines."""
    if isinstance(raw, str):
        try:
            # Pydantic emits ISO with a trailing 'Z' or offset.
            ts = _dt.datetime.fromisoformat(raw.replace("Z", "+00:00"))
            if ts.tzinfo is None:
                ts = ts.replace(tzinfo=_dt.timezone.utc)
            return ts.timestamp()
        except ValueError:
            pass
    return self._clock().timestamp()

run async

run(*, once: bool = False) -> None

Connect to NATS, subscribe, drive the handler on every received event. Returns when stop() is called or when once=True and one message has been processed.

Also owns the app-contract lifecycle (spec §03): starts the /health / /manifest / /state server when cfg.contract_port is set and self-registers with the OpenNVR app registry when cfg.opennvr_url is set — both best-effort no-ops otherwise.

The connect / subscribe / drain machinery itself lives on :class:~.nats_loop.NatsSubscriberMixin, shared with the AlertSubscriber archetype.

Source code in opennvr_app_sdk/detector.py

async def run(self, *, once: bool = False) -> None:
    """Connect to NATS, subscribe, drive the handler on every
    received event. Returns when ``stop()`` is called or when
    ``once=True`` and one message has been processed.

    Also owns the app-contract lifecycle (spec §03): starts the
    ``/health`` / ``/manifest`` / ``/state`` server when
    ``cfg.contract_port`` is set and self-registers with the
    OpenNVR app registry when ``cfg.opennvr_url`` is set — both
    best-effort no-ops otherwise.

    The connect / subscribe / drain machinery itself lives on
    :class:`~.nats_loop.NatsSubscriberMixin`, shared with the
    AlertSubscriber archetype."""
    self.start_contract_server()
    self.register_with_opennvr()
    self.start_config_poll()
    try:
        await self._run_nats_loop(once=once)
    finally:
        self.stop_config_poll()
        self.stop_contract_server()

opennvr_app_sdk.FrameApp

FrameApp(
    config: Any,
    dispatcher: AlertDispatcher,
    *,
    frame_source: FrameSource,
    cameras: Iterable[str] | None = None,
    poll_interval_seconds: float | None = None,
)

Bases: ContractMixin

Base class for frame-polling apps.

Subclasses set manifest, optionally override :meth:setup, and implement :meth:on_frame. The base owns the interval loop and alert dispatch; per-camera fetch and rule failures are isolated so one bad camera never stalls the rest.

cameras / poll_interval_seconds default from the config object (cfg.cameras may be a list of ids or a dict keyed by id; cfg.poll_interval_seconds defaults to 5.0).

Source code in opennvr_app_sdk/frame_app.py

def __init__(
    self,
    config: Any,
    dispatcher: AlertDispatcher,
    *,
    frame_source: FrameSource,
    cameras: Iterable[str] | None = None,
    poll_interval_seconds: float | None = None,
) -> None:
    self.cfg = config
    self._dispatcher = dispatcher
    self._source = frame_source
    if cameras is None:
        cameras = list(getattr(config, "cameras", None) or [])
    self._cameras: list[str] = [str(c) for c in cameras]
    if poll_interval_seconds is None:
        poll_interval_seconds = float(
            getattr(config, "poll_interval_seconds", 5.0)
        )
    if poll_interval_seconds <= 0:
        raise ValueError(
            f"poll_interval_seconds must be > 0, got {poll_interval_seconds!r}"
        )
    self._interval = poll_interval_seconds
    self._stop_event = asyncio.Event()
    self._contract_init()
    self.setup()

setup

setup() -> None

Optional hook — allocate per-app state.

Source code in opennvr_app_sdk/frame_app.py

def setup(self) -> None:
    """Optional hook — allocate per-app state."""

on_frame

on_frame(
    camera_id: str, frame_bytes: bytes
) -> Iterable[Alert] | None

The rule. Called once per fetched frame. Return or yield Alerts to fire (or None / empty).

Source code in opennvr_app_sdk/frame_app.py

def on_frame(
    self, camera_id: str, frame_bytes: bytes
) -> Iterable[Alert] | None:
    """The rule. Called once per fetched frame. Return or yield
    Alerts to fire (or ``None`` / empty)."""
    raise NotImplementedError

on_cameras_update

on_cameras_update(camera_ids: frozenset[int]) -> None

Follow the cameras picked for this app: the poll loop fetches frames for exactly those, as cam<id> handles.

Only runs when the app is connected (the live config poll is what delivers picks). A standalone run keeps the camera list it was constructed with. Nothing picked → an empty list → the loop ticks and does nothing, which is the point.

Source code in opennvr_app_sdk/frame_app.py

def on_cameras_update(self, camera_ids: frozenset[int]) -> None:
    """Follow the cameras picked for this app: the poll loop fetches
    frames for exactly those, as ``cam<id>`` handles.

    Only runs when the app is connected (the live config poll is
    what delivers picks). A standalone run keeps the camera list it
    was constructed with. Nothing picked → an empty list → the loop
    ticks and does nothing, which is the point.
    """
    if getattr(self.manifest, "camera_picker", True) is False:
        return
    self._cameras = [f"cam{i}" for i in sorted(camera_ids)]
    logger.info("%s: cameras now %s",
                self.manifest.id if self.manifest else type(self).__name__,
                ", ".join(self._cameras) or "none (nothing picked)")

handle_tick

handle_tick() -> list[Alert]

One poll cycle: fetch a frame per camera, run the rule, dispatch whatever it produced. Fetch / rule failures are logged per camera and never propagate.

Source code in opennvr_app_sdk/frame_app.py

def handle_tick(self) -> list[Alert]:
    """One poll cycle: fetch a frame per camera, run the rule,
    dispatch whatever it produced. Fetch / rule failures are
    logged per camera and never propagate."""
    fired: list[Alert] = []
    for camera_id in self._cameras:
        try:
            frame = self._source.get_frame(camera_id)
        except Exception:
            logger.exception("frame fetch failed for camera=%s", camera_id)
            continue
        if not frame:
            continue
        # Contract counters (spec §03): for a FrameApp, one fetched
        # frame is one "event" — /health's last_event_age_s then
        # doubles as camera-stall detection.
        self._contract_note_event()
        try:
            produced = self.on_frame(camera_id, frame)
        except Exception:
            logger.exception("on_frame failed for camera=%s", camera_id)
            continue
        for alert in produced or []:
            self._dispatcher.fire(alert)
            fired.append(alert)
    self._contract_note_alerts(len(fired))
    return fired

run async

run(*, once: bool = False) -> None

Poll every poll_interval_seconds until stop() (or one tick with once=True). The inter-tick sleep is interruptible so shutdown is immediate.

Also owns the app-contract lifecycle (spec §03): starts the /health / /manifest / /state server when cfg.contract_port is set and self-registers with the OpenNVR app registry when cfg.opennvr_url is set — both best-effort no-ops otherwise.

Source code in opennvr_app_sdk/frame_app.py

async def run(self, *, once: bool = False) -> None:
    """Poll every ``poll_interval_seconds`` until ``stop()`` (or
    one tick with ``once=True``). The inter-tick sleep is
    interruptible so shutdown is immediate.

    Also owns the app-contract lifecycle (spec §03): starts the
    ``/health`` / ``/manifest`` / ``/state`` server when
    ``cfg.contract_port`` is set and self-registers with the
    OpenNVR app registry when ``cfg.opennvr_url`` is set — both
    best-effort no-ops otherwise."""
    self.start_contract_server()
    self.register_with_opennvr()
    self.start_config_poll()
    try:
        await self._run_poll_loop(once=once)
    finally:
        self.stop_config_poll()
        self.stop_contract_server()

opennvr_app_sdk.AlertSubscriber

AlertSubscriber(config: Any)

Bases: ContractMixin, NatsSubscriberMixin

Base class for NATS alert-consuming apps.

Subclasses optionally set a class-level manifest (:class:AppManifest), optionally override :meth:setup to allocate state, and implement :meth:on_alert. cfg is the app-parsed config object; the NATS loop reads cfg.nats_url, cfg.nats_token (optional) and cfg.subject_pattern from it (apps default the pattern to "opennvr.alerts.>").

Source code in opennvr_app_sdk/alert_subscriber.py

def __init__(self, config: Any) -> None:
    self.cfg = config
    # Compat alias — pre-SDK subscribers (and their tests) used
    # ``self._config``.
    self._config = config
    self._stop_event = asyncio.Event()
    self._nc: Any = None
    self._contract_init()
    self.setup()

setup

setup() -> None

Optional hook — allocate per-app state (counters, HTTP clients, …). Runs once at construction, after cfg is set.

Source code in opennvr_app_sdk/alert_subscriber.py

def setup(self) -> None:
    """Optional hook — allocate per-app state (counters, HTTP
    clients, …). Runs once at construction, after ``cfg`` is set."""

on_alert

on_alert(alert: dict[str, Any], subject: str) -> None

The sink. Called once per JSON-decoded alert envelope with the raw dict and the NATS subject it arrived on. Forward it, store it, page someone — whatever the bridge is for.

Source code in opennvr_app_sdk/alert_subscriber.py

def on_alert(self, alert: dict[str, Any], subject: str) -> None:
    """The sink. Called once per JSON-decoded alert envelope with
    the raw dict and the NATS subject it arrived on. Forward it,
    store it, page someone — whatever the bridge is for."""
    raise NotImplementedError

run async

run(*, once: bool = False) -> None

Connect to NATS, subscribe, drive the sink on every received alert. Returns when stop() is called or when once=True and one message has been processed.

Also owns the app-contract lifecycle (spec §03), same as the other archetypes: starts the /health / /manifest / /state server when cfg.contract_port is set and self-registers when cfg.opennvr_url is set — both best-effort no-ops otherwise. The connect / subscribe / drain machinery lives on :class:~.nats_loop.NatsSubscriberMixin, shared with Detector.

Source code in opennvr_app_sdk/alert_subscriber.py

async def run(self, *, once: bool = False) -> None:
    """Connect to NATS, subscribe, drive the sink on every received
    alert. Returns when ``stop()`` is called or when ``once=True``
    and one message has been processed.

    Also owns the app-contract lifecycle (spec §03), same as the
    other archetypes: starts the ``/health`` / ``/manifest`` /
    ``/state`` server when ``cfg.contract_port`` is set and
    self-registers when ``cfg.opennvr_url`` is set — both
    best-effort no-ops otherwise. The connect / subscribe / drain
    machinery lives on :class:`~.nats_loop.NatsSubscriberMixin`,
    shared with ``Detector``."""
    self.start_contract_server()
    self.register_with_opennvr()
    self.start_config_poll()
    try:
        await self._run_nats_loop(once=once)
    finally:
        self.stop_config_poll()
        self.stop_contract_server()

opennvr_app_sdk.DomainEventSubscriber

DomainEventSubscriber(
    config: Any, dispatcher: AlertDispatcher | None = None
)

Bases: ContractMixin, NatsSubscriberMixin

Base class for apps that react to contracted domain events.

Set subscriptions (schemas or subjects), implement :meth:on_event. Reads cfg.nats_url / cfg.nats_token; a cfg.subject_pattern overrides subscriptions when present. The contract server, self-registration and the config poll run exactly as for the other archetypes.

Source code in opennvr_app_sdk/domain_subscriber.py

def __init__(self, config: Any, dispatcher: AlertDispatcher | None = None) -> None:
    self.cfg = config
    self._stop_event = asyncio.Event()
    self._nc: Any = None
    self._dispatcher = dispatcher
    # Alerts fired from on_event carry this app's identity, scoped
    # per call exactly as Detector does it.
    self._source_block: dict[str, str] | None = (
        {"kind": "app", "name": self.manifest.id, "version": self.manifest.version}
        if self.manifest is not None else None)
    self._contract_init()
    self.setup()

dispatcher property

dispatcher: AlertDispatcher

stdout + the webhook_url / nats_alerts_* channels from cfg (:class:~.config.BaseAppConfig), built on first use. Pass dispatcher= to the constructor to substitute one.

setup

setup() -> None

Optional: allocate state after cfg is set.

Source code in opennvr_app_sdk/domain_subscriber.py

def setup(self) -> None:
    """Optional: allocate state after ``cfg`` is set."""

fire

fire(alert: Alert) -> dict[str, bool]

Dispatch one alert as this app and count it for /health.

Source code in opennvr_app_sdk/domain_subscriber.py

def fire(self, alert: Alert) -> dict[str, bool]:
    """Dispatch one alert as this app and count it for ``/health``."""
    report = self.dispatcher.fire(alert)
    self._contract_note_alerts(1)
    return report

opennvr_app_sdk.app

app(
    detector_cls: type[Detector],
    *,
    load_config: Callable[[str], Any] | None = None,
) -> AppRunner

Wrap a Detector subclass in a CLI runner::

if name == "main": raise SystemExit(app(Loitering, load_config=load_config).run())

Source code in opennvr_app_sdk/detector.py

def app(
    detector_cls: type[Detector],
    *,
    load_config: Callable[[str], Any] | None = None,
) -> AppRunner:
    """Wrap a Detector subclass in a CLI runner::

        if __name__ == "__main__":
            raise SystemExit(app(Loitering, load_config=load_config).run())
    """
    return AppRunner(detector_cls, load_config=load_config)

opennvr_app_sdk.alert_app

alert_app(
    subscriber_cls: type[AlertSubscriber],
    *,
    load_config: Callable[[str], Any] | None = None,
) -> AlertSubscriberRunner

Wrap an AlertSubscriber subclass in a CLI runner::

if name == "main": raise SystemExit( alert_app(MyBridge, load_config=load_config).run() )

Source code in opennvr_app_sdk/alert_subscriber.py

def alert_app(
    subscriber_cls: type[AlertSubscriber],
    *,
    load_config: Callable[[str], Any] | None = None,
) -> AlertSubscriberRunner:
    """Wrap an AlertSubscriber subclass in a CLI runner::

        if __name__ == "__main__":
            raise SystemExit(
                alert_app(MyBridge, load_config=load_config).run()
            )
    """
    return AlertSubscriberRunner(subscriber_cls, load_config=load_config)

opennvr_app_sdk.domain_event_app

domain_event_app(
    app_cls, *, load_config=None
) -> AlertSubscriberRunner

CLI runner factory, the same shape as :func:alert_app::

if name == "main": raise SystemExit(domain_event_app(Gate, load_config=load_config).run())

Source code in opennvr_app_sdk/domain_subscriber.py

def domain_event_app(app_cls, *, load_config=None) -> AlertSubscriberRunner:
    """CLI runner factory, the same shape as :func:`alert_app`::

        if __name__ == "__main__":
            raise SystemExit(domain_event_app(Gate, load_config=load_config).run())
    """
    return AlertSubscriberRunner(app_cls, load_config=load_config)

opennvr_app_sdk.AppRunner

AppRunner(
    detector_cls: type[Detector],
    *,
    load_config: Callable[[str], Any] | None = None,
)

The app(MyDetector) return value — owns argparse, logging setup, dispatcher construction, and the signal-driven lifecycle. Behavior ported from the loitering-detection example's main().

Source code in opennvr_app_sdk/detector.py

def __init__(
    self,
    detector_cls: type[Detector],
    *,
    load_config: Callable[[str], Any] | None = None,
) -> None:
    loader = load_config or getattr(detector_cls, "load_config", None)
    if loader is None:
        raise TypeError(
            f"app({detector_cls.__name__}): pass load_config= or define "
            f"a load_config classmethod on the detector class"
        )
    self._detector_cls = detector_cls
    self._load_config = loader

opennvr_app_sdk.AlertSubscriberRunner

AlertSubscriberRunner(
    subscriber_cls: type[AlertSubscriber],
    *,
    load_config: Callable[[str], Any] | None = None,
)

The alert_app(MySubscriber) return value — owns argparse, logging setup, and the signal-driven lifecycle. Mirrors the Detector's :class:~.detector.AppRunner minus the dispatcher: an AlertSubscriber consumes alerts rather than emitting them.

Source code in opennvr_app_sdk/alert_subscriber.py

def __init__(
    self,
    subscriber_cls: type[AlertSubscriber],
    *,
    load_config: Callable[[str], Any] | None = None,
) -> None:
    loader = load_config or getattr(subscriber_cls, "load_config", None)
    if loader is None:
        raise TypeError(
            f"alert_app({subscriber_cls.__name__}): pass load_config= or "
            f"define a load_config classmethod on the subscriber class"
        )
    self._subscriber_cls = subscriber_cls
    self._load_config = loader

opennvr_app_sdk.BaseAppConfig dataclass

BaseAppConfig(
    nats_url: str = "",
    nats_token: str | None = None,
    subject_pattern: str | None = None,
    webhook_url: str | None = None,
    nats_alerts_url: str | None = None,
    nats_alerts_token: str | None = None,
    nats_alerts_subject_prefix: str = DEFAULT_ALERT_SUBJECT_PREFIX,
    contract_port: int | None = None,
    contract_bind_host: str | None = None,
    contract_host: str | None = None,
    opennvr_url: str | None = None,
    opennvr_token: str | None = None,
    config_poll_seconds: float | None = None,
    consume_tier0: bool = False,
)

What the SDK reads from cfg — the runners, the NATS loop, the alert dispatcher, the contract server and the registry client all look these up by name. Subclass and add your own fields::

@dataclass
class AppConfig(BaseAppConfig):
    watch_labels: list[str] = field(default_factory=lambda: ["person"])
    dwell_s: float = 30.0

cfg = load_app_config("config.yml", AppConfig)

opennvr_app_sdk.load_app_config

load_app_config(
    path: str | Path, cls: type = BaseAppConfig
)

Load path into cls — :class:BaseAppConfig or a dataclass subclass of it. Base keys get the standard validation; every extra field of cls is taken from the file by name when present, from its default otherwise, and a field with no default is required. Put app-specific checks in __post_init__; raise ValueError with a message an operator can act on.

Source code in opennvr_app_sdk/config.py

def load_app_config(path: str | Path, cls: type = BaseAppConfig):
    """Load ``path`` into ``cls`` — :class:`BaseAppConfig` or a dataclass
    subclass of it. Base keys get the standard validation; every extra
    field of ``cls`` is taken from the file by name when present, from
    its default otherwise, and a field with no default is required.
    Put app-specific checks in ``__post_init__``; raise ``ValueError``
    with a message an operator can act on."""
    raw = load_yaml(path)
    label = f"config {str(path)!r}"
    kwargs = parse_base_config(raw, path=label)
    for f in fields(cls):
        if f.name in _BASE_FIELDS:
            continue
        if f.name in raw and raw[f.name] is not None:
            kwargs[f.name] = raw[f.name]
        elif f.default is MISSING and f.default_factory is MISSING:  # type: ignore[attr-defined]
            raise ValueError(f"{label}: {f.name!r} is required")
    try:
        return cls(**kwargs)
    except (TypeError, ValueError) as exc:
        raise ValueError(f"{label}: {exc}") from exc