onvif_parsers package

async onvif_parsers.parse(topic: str, uid: str, msg: Any) → list[EventEntity]

Parse an ONVIF event notification message.

Args:

topic: The topic string of the ONVIF event notification. uid: Unique identifier for the entity. msg: The raw event data. zeep.xsd.ComplexType or zeep.xsd.AnySimpleType.

Returns:

The parsed EventEntity or an empty list if parsing failed.

Raises:

UnknownTopicError: If the topic is not registered in the parser registry. AttributeError: If the message structure is invalid. KeyError: If expected keys are missing in the message.

Submodules

onvif_parsers.errors module

exception onvif_parsers.errors.UnknownTopicError

Bases: Exception

Raised when an unknown topic is encountered in an ONVIF event notification.

onvif_parsers.model module

class onvif_parsers.model.EventEntity(uid: str, name: str, platform: str, device_class: str | None = None, unit_of_measurement: str | None = None, value: Any = None, entity_category: str | None = None, entity_enabled: bool = True)

Bases: object

Represents a ONVIF event entity.

device_class: str | None = None

entity_category: str | None = None

entity_enabled: bool = True

name: str

platform: str

uid: str

unit_of_measurement: str | None = None

value: Any = None

onvif_parsers.registry module

class onvif_parsers.registry.Registry

Bases: object

A registry of parsers.

get(key: str) → Callable[[str, Any], Awaitable[list[EventEntity]]] | None

Get a parser function by key.

register(key: str, f: Callable[[str, Any], Awaitable[list[EventEntity]]]) → None

Register a parser function under a given key.

onvif_parsers.registry.get_parser(topic: str) → Callable[[str, Any], Awaitable[list[EventEntity]]] | None

Get a parser callable for the given topic.

onvif_parsers.registry.register(topic: str) → Callable[[Callable[[str, Any], Awaitable[list[EventEntity]]]], Callable[[str, Any], Awaitable[list[EventEntity]]]]

Register an onvif parser callable with the given topic.

onvif_parsers.util module

onvif_parsers.util.event_to_debug_format(data: Any) → Any

Converts an event to a format for debugging.

This is useful because the default repr for zeep event payload doesn’t include the body of unknown XML elements. This will convert the unknown XML into strings that can then be deserialized back into an event for testing.

onvif_parsers.util.extract_message(msg: Any) → tuple[str, Any]

Extract the message content and the topic.

onvif_parsers.util.normalize_video_source(source: str) → str

Normalize video source.

Some cameras do not set the VideoSourceToken correctly so we get duplicate sensors, so we need to normalize it to the correct value.