diff --git a/.github/agents/axis-review-verify.agent.md b/.github/agents/axis-review-verify.agent.md index 1ea3c589..955448d2 100644 --- a/.github/agents/axis-review-verify.agent.md +++ b/.github/agents/axis-review-verify.agent.md @@ -65,7 +65,7 @@ You are a verification-focused reviewer for the Axis Python repository. Your job ## Review Priorities 1. Correctness and regressions: - Handler initialization phase behavior in `axis/interfaces/` (`API_DISCOVERY`, `PARAM_CGI_FALLBACK`, `APPLICATION`). -- Event/state behavior in `axis/models/event.py` and `axis/models/event_instance.py`. +- Event/state behavior in `axis/models/event.py` and `axis/models/events/event_instance.py`. - Model input normalization and default handling at boundaries. 2. Reliability and compatibility: - Unknown enum and external input handling (`_missing_`, safe defaults). diff --git a/.github/agents/axis-review.agent.md b/.github/agents/axis-review.agent.md index 867b44bb..7085b05f 100644 --- a/.github/agents/axis-review.agent.md +++ b/.github/agents/axis-review.agent.md @@ -67,7 +67,7 @@ user-invocable: true ## Review Priorities 1. Correctness and regressions: - Behavior changes in `axis/interfaces/` handlers and initialization flow (`API_DISCOVERY`, `PARAM_CGI_FALLBACK`, `APPLICATION`). -- State/event semantics in `axis/models/event.py` and `axis/models/event_instance.py`. +- State/event semantics in `axis/models/event.py` and `axis/models/events/event_instance.py`. - Input boundary normalization in model constructors or `__post_init__`. 2. Reliability and compatibility: - Unknown enum and external input handling (`_missing_` fallbacks, safe defaults). diff --git a/axis/device.py b/axis/device.py index 009946c0..68bfc547 100644 --- a/axis/device.py +++ b/axis/device.py @@ -2,7 +2,8 @@ from typing import TYPE_CHECKING -from .interfaces.event_manager import EventManager +from .interfaces.events.event_manager import EventManager +from .interfaces.events.transport_manager import EventTransportManager from .interfaces.vapix import Vapix from .stream_manager import StreamManager @@ -19,6 +20,12 @@ def __init__(self, configuration: Configuration) -> None: self.vapix = Vapix(self) self.stream = StreamManager(self) self.event = EventManager() + self.event_transport = EventTransportManager( + event_instances=self.vapix.event_instances, + stream=self.stream, + event=self.event, + mqtt=self.vapix.mqtt, + ) def enable_events(self) -> None: """Enable events for stream.""" diff --git a/axis/interfaces/event_instances.py b/axis/interfaces/event_instances.py deleted file mode 100644 index 142e88a6..00000000 --- a/axis/interfaces/event_instances.py +++ /dev/null @@ -1,45 +0,0 @@ -"""Event service and action service APIs available in Axis network device.""" - -from typing import TYPE_CHECKING - -from ..models.event_instance import ( - EventInstance, - ListEventInstancesRequest, - ListEventInstancesResponse, -) -from .api_handler import ApiHandler -from .event_manager import BLACK_LISTED_TOPICS - -if TYPE_CHECKING: - from ..models.event import Event - - -class EventInstanceHandler(ApiHandler[EventInstance]): - """Event instances for Axis devices.""" - - async def _api_request(self) -> dict[str, EventInstance]: - """Get default data of API discovery.""" - return await self.get_event_instances() - - async def get_event_instances(self) -> dict[str, EventInstance]: - """List all event instances.""" - bytes_data = await self.vapix.api_request(ListEventInstancesRequest()) - response = ListEventInstancesResponse.decode(bytes_data) - return response.data - - def get_expected_events_per_topic( - self, - include_internal_topics: bool = False, - ) -> dict[str, list[Event]]: - """Return expected startup events grouped by topic. - - Event instances are the protocol-agnostic bootstrap source for startup - predeclaration. Returned events are synthesized from schema data and represent - expected event identity/state (operation=Initialized), not live stream updates. - """ - grouped: dict[str, list[Event]] = {} - for item in self.values(): - if not include_internal_topics and item.topic in BLACK_LISTED_TOPICS: - continue - grouped[item.topic] = item.to_events() - return grouped diff --git a/axis/interfaces/events/__init__.py b/axis/interfaces/events/__init__.py new file mode 100644 index 00000000..3dad02bd --- /dev/null +++ b/axis/interfaces/events/__init__.py @@ -0,0 +1 @@ +"""Event interface handlers for Axis devices.""" diff --git a/axis/interfaces/events/event_instances.py b/axis/interfaces/events/event_instances.py new file mode 100644 index 00000000..a3ccd098 --- /dev/null +++ b/axis/interfaces/events/event_instances.py @@ -0,0 +1,100 @@ +"""Event service and action service APIs available in Axis network device.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +from ...models.events.event_instance import ( + EventInstance, + ListEventInstancesRequest, + ListEventInstancesResponse, +) +from ...models.events.topic_filter import ( + EventTopicFilter, + to_canonical, + to_topic_filter, +) +from ..api_handler import ApiHandler +from .event_manager import BLACK_LISTED_TOPICS + +if TYPE_CHECKING: + from ...models.event import Event + + +class EventInstanceHandler(ApiHandler[EventInstance]): + """Event instances for Axis devices.""" + + async def _api_request(self) -> dict[str, EventInstance]: + """Get default data of API discovery.""" + return await self.get_event_instances() + + async def get_event_instances(self) -> dict[str, EventInstance]: + """List all event instances.""" + bytes_data = await self.vapix.api_request(ListEventInstancesRequest()) + response = ListEventInstancesResponse.decode(bytes_data) + return response.data + + def get_expected_events_per_topic( + self, + include_internal_topics: bool = False, + ) -> dict[str, list[Event]]: + """Return expected startup events grouped by topic. + + Event instances are the protocol-agnostic bootstrap source for startup + predeclaration. Returned events are synthesized from schema data and represent + expected event identity/state (operation=Initialized), not live stream updates. + """ + grouped: dict[str, list[Event]] = {} + for item in self.values(): + if not include_internal_topics and item.topic in BLACK_LISTED_TOPICS: + continue + grouped[item.topic] = item.to_events() + return grouped + + def get_supported_topics( + self, include_internal_topics: bool = False + ) -> tuple[str, ...]: + """Return canonical supported event topics from event instances.""" + expected = self.get_expected_events_per_topic( + include_internal_topics=include_internal_topics + ) + return tuple(sorted(to_canonical(topic) for topic in expected)) + + def get_supported_event_descriptors( + self, include_internal_topics: bool = False + ) -> dict[str, dict[str, Any]]: + """Return normalized supported-event descriptors for extension consumers.""" + descriptors: dict[str, dict[str, Any]] = {} + for topic, events in self.get_expected_events_per_topic( + include_internal_topics=include_internal_topics + ).items(): + canonical_topic = to_canonical(topic) + topic_filter = to_topic_filter(canonical_topic) + descriptors[canonical_topic] = { + "topic": canonical_topic, + "topic_filter": topic_filter, + "count": len(events), + "sources": tuple( + sorted({event.source for event in events if event.source}) + ), + } + return descriptors + + def build_transport_filter_payloads( + self, + event_filter: EventTopicFilter | None = None, + include_internal_topics: bool = False, + ) -> dict[str, list[str]]: + """Build no-op extension payloads for MQTT and WebSocket topic filters.""" + if event_filter is None or event_filter.is_wildcard: + topics = list(self.get_supported_topics(include_internal_topics)) + else: + topics = event_filter.canonical_topics + + topic_filters = [to_topic_filter(topic) for topic in topics] + + return { + "canonical_topics": topics, + "mqtt_topics": topic_filters, + "websocket_topic_filters": topic_filters, + } diff --git a/axis/interfaces/event_manager.py b/axis/interfaces/events/event_manager.py similarity index 84% rename from axis/interfaces/event_manager.py rename to axis/interfaces/events/event_manager.py index 4745d0e7..717317c8 100644 --- a/axis/interfaces/event_manager.py +++ b/axis/interfaces/events/event_manager.py @@ -4,7 +4,8 @@ import logging from typing import Any -from ..models.event import Event, EventOperation, EventTopic +from ...models.event import Event, EventOperation, EventTopic +from ...models.events.topic_filter import to_canonical SubscriptionCallback = Callable[[Event], None] SubscriptionType = tuple[ @@ -32,8 +33,20 @@ def __init__(self) -> None: """Ready information about events.""" self._known_topics: set[str] = set() self._unsupported_topics: set[str] = set() + self._allowed_topics: set[str] | None = None self._subscribers: dict[str, list[SubscriptionType]] = {ID_FILTER_ALL: []} + def set_allowed_topics(self, topics: list[str] | set[str] | None) -> None: + """Set optional canonical topic allow-list for runtime filtering. + + Default behavior remains unchanged until this extension is explicitly used. + """ + if topics is None: + self._allowed_topics = None + return + + self._allowed_topics = {to_canonical(topic) for topic in topics} + def handler(self, data: bytes | dict[str, Any]) -> None: """Create event and pass it along to subscribers.""" event = Event.decode(data) @@ -49,6 +62,12 @@ def handler(self, data: bytes | dict[str, Any]) -> None: if event.topic in BLACK_LISTED_TOPICS: return + if ( + self._allowed_topics is not None + and to_canonical(event.topic) not in self._allowed_topics + ): + return + known = (unique_topic := f"{event.topic}_{event.id}") not in self._known_topics self._known_topics.add(unique_topic) diff --git a/axis/interfaces/events/transport_manager.py b/axis/interfaces/events/transport_manager.py new file mode 100644 index 00000000..2ed3ddce --- /dev/null +++ b/axis/interfaces/events/transport_manager.py @@ -0,0 +1,115 @@ +"""Event transport orchestration manager.""" + +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING, Any + +from ...models.events.topic_filter import EventTopicFilter +from ...models.events.transport_capabilities import ( + TRANSPORT_FILTER_CAPABILITIES, + EventTransport, + TransportFilterCapability, +) + +if TYPE_CHECKING: + from ...stream_manager import StreamManager + from ..mqtt import MqttClientHandler + from .event_instances import EventInstanceHandler + from .event_manager import EventManager + +LOGGER = logging.getLogger(__name__) + + +class EventTransportManager: + """Owns event transport filter orchestration across subsystems.""" + + def __init__( + self, + event_instances: EventInstanceHandler, + stream: StreamManager, + event: EventManager, + mqtt: MqttClientHandler, + ) -> None: + """Store subsystem dependencies used for transport filter application.""" + self._event_instances = event_instances + self._stream = stream + self._event = event + self._mqtt = mqtt + + def get_event_transport_capabilities( + self, + ) -> dict[EventTransport, TransportFilterCapability]: + """Return extension capability matrix for event transports.""" + return dict(TRANSPORT_FILTER_CAPABILITIES) + + def get_supported_event_descriptors( + self, + include_internal_topics: bool = False, + ) -> dict[str, dict[str, Any]]: + """Return normalized supported-event descriptors from event instances. + + This method is extension-oriented and has no side effects. + """ + return self._event_instances.get_supported_event_descriptors( + include_internal_topics=include_internal_topics + ) + + async def apply_event_transport_filters( + self, + event_filter: EventTopicFilter | None = None, + include_internal_topics: bool = False, + ) -> None: + """Apply validated transport filters to websocket, MQTT, and local fallback. + + This method performs no implicit event-instance initialization. Callers must + invoke ``initialize_event_instances`` before applying filters. + """ + if not self._event_instances.initialized: + msg = ( + "Event instances are not initialized. " + "Call initialize_event_instances() first." + ) + raise RuntimeError(msg) + + if event_filter is None: + supported = self._event_instances.get_supported_topics( + include_internal_topics=include_internal_topics + ) + event_filter = ( + EventTopicFilter.from_topics(list(supported)) + if supported + else EventTopicFilter.for_all_events() + ) + + if not event_filter.is_wildcard: + supported_topics = set( + self._event_instances.get_supported_topics( + include_internal_topics=include_internal_topics + ) + ) + unknown_topics = sorted( + set(event_filter.canonical_topics) - supported_topics + ) + if unknown_topics: + message = f"Requested unsupported topics: {', '.join(unknown_topics)}" + raise ValueError(message) + + self._stream.set_event_subscription(event_filter) + + if self._mqtt.supported: + await self._mqtt.configure_event_publication( + event_filter.transport_topic_filters + ) + + self._event.set_allowed_topics(event_filter.canonical_topics or None) + + LOGGER.debug( + "Applied event transport filters: websocket=%s mqtt=%s local=%s topics=%s", + True, + self._mqtt.supported, + True, + len(event_filter.canonical_topics) + if not event_filter.is_wildcard + else "all", + ) diff --git a/axis/interfaces/mqtt.py b/axis/interfaces/mqtt.py index 882a021f..aabbc702 100644 --- a/axis/interfaces/mqtt.py +++ b/axis/interfaces/mqtt.py @@ -3,6 +3,7 @@ from typing import Any from ..models.api_discovery import ApiId +from ..models.events.topic_filter import to_topic_filter from ..models.mqtt import ( API_VERSION, ActivateClientRequest, @@ -66,12 +67,28 @@ async def get_event_publication_config(self) -> EventPublicationConfig: response = GetEventPublicationConfigResponse.decode(bytes_data) return response.data + def build_topic_filters( + self, topics: list[str], normalize_topics: bool = False + ) -> list[str]: + """Build MQTT topic filters. + + Defaults preserve previous behavior unless normalization is requested. + """ + normalized = ( + [to_topic_filter(topic) for topic in topics] if normalize_topics else topics + ) + # Preserve input order while deduplicating. + return list(dict.fromkeys(normalized)) + async def configure_event_publication( - self, topics: list[str] = DEFAULT_TOPICS + self, + topics: list[str] = DEFAULT_TOPICS, + normalize_topics: bool = False, ) -> None: """Configure MQTT Client event publication.""" + topic_filters = self.build_topic_filters(topics, normalize_topics) event_filters = EventFilter.from_list( - [{"topicFilter": topic} for topic in topics] + [{"topicFilter": topic} for topic in topic_filters] ) config = EventPublicationConfig(event_filter_list=event_filters) await self.vapix.api_request( diff --git a/axis/interfaces/vapix.py b/axis/interfaces/vapix.py index 54c2773f..9b0cb971 100644 --- a/axis/interfaces/vapix.py +++ b/axis/interfaces/vapix.py @@ -21,7 +21,7 @@ from .applications.object_analytics import ObjectAnalyticsHandler from .applications.vmd4 import Vmd4Handler from .basic_device_info import BasicDeviceInfoHandler -from .event_instances import EventInstanceHandler +from .events.event_instances import EventInstanceHandler from .light_control import LightHandler from .mqtt import MqttClientHandler from .parameters.param_cgi import Params diff --git a/axis/models/event.py b/axis/models/event.py index 5aaf2302..f08f03bd 100644 --- a/axis/models/event.py +++ b/axis/models/event.py @@ -1,221 +1,3 @@ -"""Python library to enable Axis devices to integrate with Home Assistant.""" +"""Backward-compatible re-export for event models.""" -from dataclasses import dataclass -import enum -from typing import Any, Self - -import xmltodict - - -class EventOperation(enum.StrEnum): - """Possible operations of an event.""" - - INITIALIZED = "Initialized" - CHANGED = "Changed" - DELETED = "Deleted" - UNKNOWN = "Unknown" - - @classmethod - def _missing_(cls, value: object) -> EventOperation: - """Set default enum member if an unknown value is provided.""" - return EventOperation.UNKNOWN - - -class EventTopic(enum.StrEnum): - """Supported event topics.""" - - DAY_NIGHT_VISION = "tns1:VideoSource/tnsaxis:DayNightVision" - FENCE_GUARD = "tnsaxis:CameraApplicationPlatform/FenceGuard" - LIGHT_STATUS = "tns1:Device/tnsaxis:Light/Status" - LOITERING_GUARD = "tnsaxis:CameraApplicationPlatform/LoiteringGuard" - MOTION_DETECTION = "tns1:VideoAnalytics/tnsaxis:MotionDetection" - MOTION_DETECTION_2 = "tns1:RuleEngine/tnsaxis:VideoMotionDetection" - MOTION_DETECTION_3 = "tns1:RuleEngine/tnsaxis:VMD3" - MOTION_DETECTION_4 = "tnsaxis:CameraApplicationPlatform/VMD" - MOTION_GUARD = "tnsaxis:CameraApplicationPlatform/MotionGuard" - OBJECT_ANALYTICS = "tnsaxis:CameraApplicationPlatform/ObjectAnalytics" - PIR = "tns1:Device/tnsaxis:Sensor/PIR" - PORT_INPUT = "tns1:Device/tnsaxis:IO/Port" - PORT_SUPERVISED_INPUT = "tns1:Device/tnsaxis:IO/SupervisedPort" - PTZ_IS_MOVING = "tns1:PTZController/tnsaxis:Move" - PTZ_ON_PRESET = "tns1:PTZController/tnsaxis:PTZPresets" - RELAY = "tns1:Device/Trigger/Relay" - SOUND_TRIGGER_LEVEL = "tns1:AudioSource/tnsaxis:TriggerLevel" - UNKNOWN = "unknown" - - @classmethod - def _missing_(cls, value: object) -> EventTopic: - """Set default enum member if an unknown value is provided.""" - return EventTopic.UNKNOWN - - -TOPIC_TO_STATE = { - EventTopic.LIGHT_STATUS: "ON", - EventTopic.RELAY: "active", -} - -KNOWN_FALSE_STATES = {"", "0", "false", "inactive", "low", "off"} -KNOWN_TRUE_STATES = {"1", "active", "high", "on", "true"} - -TOPIC_TO_INACTIVE_STATE = { - EventTopic.LIGHT_STATUS: "OFF", - EventTopic.RELAY: "inactive", -} - -EVENT_OPERATION = "operation" -EVENT_SOURCE = "source" -EVENT_SOURCE_IDX = "source_idx" -EVENT_TOPIC = "topic" -EVENT_TYPE = "type" -EVENT_VALUE = "value" - -NOTIFICATION_MESSAGE = ("MetadataStream", "Event", "NotificationMessage") -MESSAGE = (*NOTIFICATION_MESSAGE, "Message", "Message") -TOPIC = (*NOTIFICATION_MESSAGE, "Topic", "#text") -OPERATION = (*MESSAGE, "PropertyOperation") -SOURCE = (*MESSAGE, "Source") -DATA = (*MESSAGE, "Data") - -XML_NAMESPACES = { - "http://www.onvif.org/ver10/schema": None, - "http://docs.oasis-open.org/wsn/b-2": None, -} - - -def traverse(data: Any, keys: tuple[str, ...] | list[str]) -> Any: - """Traverse dictionary using keys to retrieve last item.""" - if not isinstance(data, dict): - return {} - - head, *tail = keys - item = data.get(head, {}) - return traverse(item, tail) if tail else item - - -def extract_name_value( - data: dict[str, list[dict[str, str]] | dict[str, str]], prefer: str | None = None -) -> tuple[str, str]: - """Extract name and value from a simple item, take first dictionary if it is a list.""" - item = data.get("SimpleItem", {}) - if isinstance(item, list): - if not item: - return "", "" - if prefer is None: - item = item[0] - else: - item = next( - (item for item in item if item.get("Name", "") == prefer), item[0] - ) - return item.get("Name", ""), item.get("Value", "") - - -def is_tripped(value: object, topic_base: EventTopic, event_type: object) -> bool: - """Return whether an event value should be considered active/tripped.""" - if (expected_state := TOPIC_TO_STATE.get(topic_base)) is not None: - return str(value) == expected_state - - value_text = str(value).strip() - state = value_text.casefold() - - if state in KNOWN_FALSE_STATES: - return False - - if state in KNOWN_TRUE_STATES: - return True - - if str(event_type).casefold() == "active": - return False - - # Non-empty semantic values (for example classification values like "human") - # are treated as stateless event triggers. - return bool(value_text) - - -@dataclass -class Event: - """Event data from Axis device.""" - - id: str - is_tripped: bool - operation: EventOperation - source: str - state: str - topic: str - topic_base: EventTopic - data: dict[str, Any] - - @classmethod - def decode(cls, data: bytes | dict[str, Any]) -> Self: - """Decode data to an event object.""" - if isinstance(data, dict): - return cls._decode_from_dict(data) - return cls._decode_from_bytes(data) - - @classmethod - def _decode_from_dict(cls, data: dict[str, Any]) -> Self: - """Create event instance from dict.""" - operation = EventOperation(data.get(EVENT_OPERATION, "")) - topic = data.get(EVENT_TOPIC, "") - source = data.get(EVENT_SOURCE, "") - source_idx = data.get(EVENT_SOURCE_IDX, "") - event_type = data.get(EVENT_TYPE, "") - value = data.get(EVENT_VALUE, "") - - if (topic_base := EventTopic(topic)) is EventTopic.UNKNOWN: - _topic_base, _, _source_idx = topic.rpartition("/") - topic_base = EventTopic(_topic_base) - if source_idx == "": - source_idx = _source_idx - - if source_idx == "-1": - source_idx = "ANY" if source != "port" else "" - - return cls( - id=source_idx, - is_tripped=is_tripped(value, topic_base, event_type), - operation=operation, - source=source, - state=value, - topic=topic, - topic_base=topic_base, - data=data, - ) - - @classmethod - def _decode_from_bytes(cls, data: bytes) -> Self: - """Parse metadata xml.""" - raw = xmltodict.parse( - data, - attr_prefix="", # Remove "@" prefix from XML attributes for easier access - process_namespaces=True, - namespaces=XML_NAMESPACES, - ) - - # Normalize the ONVIF metadata root: always use a dict, drop any stray - # XML namespace attribute ("xmlns") added by xmltodict, and bail out - # early if the payload is empty. - stream = raw.get("MetadataStream") or {} - if not stream or not any(key != "xmlns" for key in stream): - return cls._decode_from_dict({}) - - topic = traverse(raw, TOPIC) - operation = traverse(raw, OPERATION) - - source = source_idx = "" - if match := traverse(raw, SOURCE): - source, source_idx = extract_name_value(match) - - data_type = data_value = "" - if match := traverse(raw, DATA): - data_type, data_value = extract_name_value(match, "active") - - return cls._decode_from_dict( - { - EVENT_OPERATION: operation, - EVENT_TOPIC: topic, - EVENT_SOURCE: source, - EVENT_SOURCE_IDX: source_idx, - EVENT_TYPE: data_type, - EVENT_VALUE: data_value, - } - ) +from .events.event import * # noqa: F403 diff --git a/axis/models/events/__init__.py b/axis/models/events/__init__.py new file mode 100644 index 00000000..0e5aaae7 --- /dev/null +++ b/axis/models/events/__init__.py @@ -0,0 +1,19 @@ +"""Event-related models namespace.""" + +from .topic_filter import ( + EventTopicFilter, + SegmentFormat, + detect_format, + to_canonical, + to_mqtt, + to_topic_filter, +) + +__all__ = [ + "EventTopicFilter", + "SegmentFormat", + "detect_format", + "to_canonical", + "to_mqtt", + "to_topic_filter", +] diff --git a/axis/models/events/event.py b/axis/models/events/event.py new file mode 100644 index 00000000..5aaf2302 --- /dev/null +++ b/axis/models/events/event.py @@ -0,0 +1,221 @@ +"""Python library to enable Axis devices to integrate with Home Assistant.""" + +from dataclasses import dataclass +import enum +from typing import Any, Self + +import xmltodict + + +class EventOperation(enum.StrEnum): + """Possible operations of an event.""" + + INITIALIZED = "Initialized" + CHANGED = "Changed" + DELETED = "Deleted" + UNKNOWN = "Unknown" + + @classmethod + def _missing_(cls, value: object) -> EventOperation: + """Set default enum member if an unknown value is provided.""" + return EventOperation.UNKNOWN + + +class EventTopic(enum.StrEnum): + """Supported event topics.""" + + DAY_NIGHT_VISION = "tns1:VideoSource/tnsaxis:DayNightVision" + FENCE_GUARD = "tnsaxis:CameraApplicationPlatform/FenceGuard" + LIGHT_STATUS = "tns1:Device/tnsaxis:Light/Status" + LOITERING_GUARD = "tnsaxis:CameraApplicationPlatform/LoiteringGuard" + MOTION_DETECTION = "tns1:VideoAnalytics/tnsaxis:MotionDetection" + MOTION_DETECTION_2 = "tns1:RuleEngine/tnsaxis:VideoMotionDetection" + MOTION_DETECTION_3 = "tns1:RuleEngine/tnsaxis:VMD3" + MOTION_DETECTION_4 = "tnsaxis:CameraApplicationPlatform/VMD" + MOTION_GUARD = "tnsaxis:CameraApplicationPlatform/MotionGuard" + OBJECT_ANALYTICS = "tnsaxis:CameraApplicationPlatform/ObjectAnalytics" + PIR = "tns1:Device/tnsaxis:Sensor/PIR" + PORT_INPUT = "tns1:Device/tnsaxis:IO/Port" + PORT_SUPERVISED_INPUT = "tns1:Device/tnsaxis:IO/SupervisedPort" + PTZ_IS_MOVING = "tns1:PTZController/tnsaxis:Move" + PTZ_ON_PRESET = "tns1:PTZController/tnsaxis:PTZPresets" + RELAY = "tns1:Device/Trigger/Relay" + SOUND_TRIGGER_LEVEL = "tns1:AudioSource/tnsaxis:TriggerLevel" + UNKNOWN = "unknown" + + @classmethod + def _missing_(cls, value: object) -> EventTopic: + """Set default enum member if an unknown value is provided.""" + return EventTopic.UNKNOWN + + +TOPIC_TO_STATE = { + EventTopic.LIGHT_STATUS: "ON", + EventTopic.RELAY: "active", +} + +KNOWN_FALSE_STATES = {"", "0", "false", "inactive", "low", "off"} +KNOWN_TRUE_STATES = {"1", "active", "high", "on", "true"} + +TOPIC_TO_INACTIVE_STATE = { + EventTopic.LIGHT_STATUS: "OFF", + EventTopic.RELAY: "inactive", +} + +EVENT_OPERATION = "operation" +EVENT_SOURCE = "source" +EVENT_SOURCE_IDX = "source_idx" +EVENT_TOPIC = "topic" +EVENT_TYPE = "type" +EVENT_VALUE = "value" + +NOTIFICATION_MESSAGE = ("MetadataStream", "Event", "NotificationMessage") +MESSAGE = (*NOTIFICATION_MESSAGE, "Message", "Message") +TOPIC = (*NOTIFICATION_MESSAGE, "Topic", "#text") +OPERATION = (*MESSAGE, "PropertyOperation") +SOURCE = (*MESSAGE, "Source") +DATA = (*MESSAGE, "Data") + +XML_NAMESPACES = { + "http://www.onvif.org/ver10/schema": None, + "http://docs.oasis-open.org/wsn/b-2": None, +} + + +def traverse(data: Any, keys: tuple[str, ...] | list[str]) -> Any: + """Traverse dictionary using keys to retrieve last item.""" + if not isinstance(data, dict): + return {} + + head, *tail = keys + item = data.get(head, {}) + return traverse(item, tail) if tail else item + + +def extract_name_value( + data: dict[str, list[dict[str, str]] | dict[str, str]], prefer: str | None = None +) -> tuple[str, str]: + """Extract name and value from a simple item, take first dictionary if it is a list.""" + item = data.get("SimpleItem", {}) + if isinstance(item, list): + if not item: + return "", "" + if prefer is None: + item = item[0] + else: + item = next( + (item for item in item if item.get("Name", "") == prefer), item[0] + ) + return item.get("Name", ""), item.get("Value", "") + + +def is_tripped(value: object, topic_base: EventTopic, event_type: object) -> bool: + """Return whether an event value should be considered active/tripped.""" + if (expected_state := TOPIC_TO_STATE.get(topic_base)) is not None: + return str(value) == expected_state + + value_text = str(value).strip() + state = value_text.casefold() + + if state in KNOWN_FALSE_STATES: + return False + + if state in KNOWN_TRUE_STATES: + return True + + if str(event_type).casefold() == "active": + return False + + # Non-empty semantic values (for example classification values like "human") + # are treated as stateless event triggers. + return bool(value_text) + + +@dataclass +class Event: + """Event data from Axis device.""" + + id: str + is_tripped: bool + operation: EventOperation + source: str + state: str + topic: str + topic_base: EventTopic + data: dict[str, Any] + + @classmethod + def decode(cls, data: bytes | dict[str, Any]) -> Self: + """Decode data to an event object.""" + if isinstance(data, dict): + return cls._decode_from_dict(data) + return cls._decode_from_bytes(data) + + @classmethod + def _decode_from_dict(cls, data: dict[str, Any]) -> Self: + """Create event instance from dict.""" + operation = EventOperation(data.get(EVENT_OPERATION, "")) + topic = data.get(EVENT_TOPIC, "") + source = data.get(EVENT_SOURCE, "") + source_idx = data.get(EVENT_SOURCE_IDX, "") + event_type = data.get(EVENT_TYPE, "") + value = data.get(EVENT_VALUE, "") + + if (topic_base := EventTopic(topic)) is EventTopic.UNKNOWN: + _topic_base, _, _source_idx = topic.rpartition("/") + topic_base = EventTopic(_topic_base) + if source_idx == "": + source_idx = _source_idx + + if source_idx == "-1": + source_idx = "ANY" if source != "port" else "" + + return cls( + id=source_idx, + is_tripped=is_tripped(value, topic_base, event_type), + operation=operation, + source=source, + state=value, + topic=topic, + topic_base=topic_base, + data=data, + ) + + @classmethod + def _decode_from_bytes(cls, data: bytes) -> Self: + """Parse metadata xml.""" + raw = xmltodict.parse( + data, + attr_prefix="", # Remove "@" prefix from XML attributes for easier access + process_namespaces=True, + namespaces=XML_NAMESPACES, + ) + + # Normalize the ONVIF metadata root: always use a dict, drop any stray + # XML namespace attribute ("xmlns") added by xmltodict, and bail out + # early if the payload is empty. + stream = raw.get("MetadataStream") or {} + if not stream or not any(key != "xmlns" for key in stream): + return cls._decode_from_dict({}) + + topic = traverse(raw, TOPIC) + operation = traverse(raw, OPERATION) + + source = source_idx = "" + if match := traverse(raw, SOURCE): + source, source_idx = extract_name_value(match) + + data_type = data_value = "" + if match := traverse(raw, DATA): + data_type, data_value = extract_name_value(match, "active") + + return cls._decode_from_dict( + { + EVENT_OPERATION: operation, + EVENT_TOPIC: topic, + EVENT_SOURCE: source, + EVENT_SOURCE_IDX: source_idx, + EVENT_TYPE: data_type, + EVENT_VALUE: data_value, + } + ) diff --git a/axis/models/event_instance.py b/axis/models/events/event_instance.py similarity index 99% rename from axis/models/event_instance.py rename to axis/models/events/event_instance.py index 9152f8d5..69e3654a 100644 --- a/axis/models/event_instance.py +++ b/axis/models/events/event_instance.py @@ -6,7 +6,7 @@ import xmltodict -from .api import ApiItem, ApiRequest, ApiResponse +from ..api import ApiItem, ApiRequest, ApiResponse from .event import ( EVENT_OPERATION, EVENT_SOURCE, diff --git a/axis/models/events/topic_filter.py b/axis/models/events/topic_filter.py new file mode 100644 index 00000000..bf251664 --- /dev/null +++ b/axis/models/events/topic_filter.py @@ -0,0 +1,174 @@ +"""Topic-domain primitives for event filtering. + +This module owns two related concerns: +1) Topic namespace conversion helpers between canonical (tns1/tnsaxis) + and transport-facing MQTT/websocket format (onvif/axis). +2) The EventTopicFilter value object used by callers to describe a + wildcard subscription or a normalized set of explicit topics. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, Literal + +if TYPE_CHECKING: + from collections.abc import Sequence + + from .event import EventTopic + +# --------------------------------------------------------------------------- +# Topic namespace conversion +# --------------------------------------------------------------------------- + +_CANONICAL_PREFIXES = ("tns1", "tnsaxis") +_MQTT_PREFIXES = ("onvif", "axis") + +SegmentFormat = Literal["canonical", "mqtt", "unknown"] + + +def _convert_segment(segment: str, mapping: dict[str, str]) -> str: + for source, target in mapping.items(): + prefix = f"{source}:" + if segment.startswith(prefix): + return f"{target}:{segment[len(prefix) :]}" + return segment + + +def _convert_topic(topic: str, mapping: dict[str, str]) -> str: + return "/".join(_convert_segment(segment, mapping) for segment in topic.split("/")) + + +def to_canonical(topic: str) -> str: + """Convert topic namespaces to canonical (tns1/tnsaxis) representation. + + This is the internal normalization format used across filtering and + local allow-list comparisons. + """ + return _convert_topic(topic, {"onvif": "tns1", "axis": "tnsaxis"}) + + +def to_mqtt(topic: str) -> str: + """Convert topic namespaces to MQTT (onvif/axis) representation. + + Used for outbound transport configuration where the device API expects + onvif/axis prefixes. + """ + return _convert_topic(topic, {"tns1": "onvif", "tnsaxis": "axis"}) + + +def to_topic_filter(topic: str) -> str: + """Convert canonical topic to transport topic-filter representation. + + Today this is equivalent to MQTT representation and is kept as a named + helper to make intent explicit at call sites. + """ + return to_mqtt(topic) + + +def detect_format(topic: str) -> SegmentFormat: + """Detect topic namespace format from segment prefixes. + + Returns canonical, mqtt, or unknown. Unknown means no known namespace + prefixes were found in any topic segment. + """ + segments = topic.split("/") + + if any( + segment.startswith(f"{prefix}:") + for prefix in _CANONICAL_PREFIXES + for segment in segments + ): + return "canonical" + + if any( + segment.startswith(f"{prefix}:") + for prefix in _MQTT_PREFIXES + for segment in segments + ): + return "mqtt" + + return "unknown" + + +# --------------------------------------------------------------------------- +# EventTopicFilter value object +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class EventTopicFilter: + """Immutable, transport-agnostic event topic filter. + + Normalizes topic inputs to canonical form and exposes per-transport + output properties. Wire-format construction (e.g. WebSocket dict shape) + is the responsibility of the transport layer. + + Do not construct directly; prefer for_all_events() and from_topics(). + This keeps wildcard and normalization behavior centralized. + """ + + # None means wildcard (subscribe to all); never empty frozenset. + _topics: frozenset[str] | None = field(default=None, repr=True) + + @classmethod + def for_all_events(cls) -> EventTopicFilter: + """Return a wildcard filter — subscribe to all events. + + On WebSocket, equivalent to eventFilterList [{"topicFilter": "//."}]. + On MQTT, no topic filter is applied. + EventManager allows all topics through. + """ + return cls(_topics=None) + + @classmethod + def from_topics( + cls, + topics: Sequence[str | EventTopic], + ) -> EventTopicFilter: + """Build from a sequence of topic strings and/or EventTopic enum values. + + Accepts topics in any format: canonical (tns1/tnsaxis), MQTT (onvif/axis), + or EventTopic enum. All inputs are normalized to canonical form. + + Duplicate logical topics are collapsed after normalization. + Raises ValueError if topics is empty; use for_all_events() for wildcard. + """ + if not topics: + msg = "topics must not be empty; use EventTopicFilter.for_all_events() to subscribe to all events" + raise ValueError(msg) + + canonical: set[str] = set() + for topic in topics: + raw: str = topic.value if hasattr(topic, "value") else topic + canonical.add(to_canonical(raw) if detect_format(raw) == "mqtt" else raw) + return cls(_topics=frozenset(canonical)) + + @property + def is_wildcard(self) -> bool: + """True when this filter represents 'subscribe to all events'.""" + return self._topics is None + + @property + def canonical_topics(self) -> list[str]: + """Sorted canonical (tns1/tnsaxis) topic strings. + + Consumed by EventManager.set_allowed_topics(). + Returns an empty list for wildcard filters — callers interpret + empty as 'allow all'. + """ + if self._topics is None: + return [] + return sorted(self._topics) + + @property + def transport_topic_filters(self) -> list[str]: + """Sorted MQTT (onvif/axis) topic filter strings. + + Consumed by MqttClientHandler.configure_event_publication(). + Returns an empty list for wildcard filters — callers should skip + setting an MQTT filter when this is empty. + """ + if self._topics is None: + return [] + return sorted(to_topic_filter(t) for t in self._topics) diff --git a/axis/models/events/transport_capabilities.py b/axis/models/events/transport_capabilities.py new file mode 100644 index 00000000..5b288439 --- /dev/null +++ b/axis/models/events/transport_capabilities.py @@ -0,0 +1,51 @@ +"""Transport capability metadata for event filter planning. + +This module defines the transport contract used by extension-facing planning +APIs: supported event transports and whether each transport can apply topic +filters upstream. +""" + +from __future__ import annotations + +from dataclasses import dataclass +import enum + + +class EventTransport(enum.StrEnum): + """Event transport types used by capability matrix and public contracts.""" + + RTSP = "rtsp" + WEBSOCKET = "websocket" + MQTT = "mqtt" + + +@dataclass(frozen=True) +class TransportFilterCapability: + """Describes upstream filter capabilities for a transport. + + supports_upstream_topic_filter indicates whether the transport API can + accept topic filters at source, reducing downstream event volume. + """ + + transport: EventTransport + supports_upstream_topic_filter: bool + notes: str + + +TRANSPORT_FILTER_CAPABILITIES: dict[EventTransport, TransportFilterCapability] = { + EventTransport.RTSP: TransportFilterCapability( + transport=EventTransport.RTSP, + supports_upstream_topic_filter=False, + notes="RTSP can only toggle event stream on/off; no topic-level filter endpoint.", + ), + EventTransport.WEBSOCKET: TransportFilterCapability( + transport=EventTransport.WEBSOCKET, + supports_upstream_topic_filter=True, + notes="WebSocket supports events:configure with eventFilterList topic filters.", + ), + EventTransport.MQTT: TransportFilterCapability( + transport=EventTransport.MQTT, + supports_upstream_topic_filter=True, + notes="MQTT publication setup supports a topicFilter list for upstream filtering.", + ), +} diff --git a/axis/stream_manager.py b/axis/stream_manager.py index 5c329b69..b32fed93 100644 --- a/axis/stream_manager.py +++ b/axis/stream_manager.py @@ -5,6 +5,7 @@ from typing import TYPE_CHECKING, Any from .models.configuration import WebProtocol +from .models.events.topic_filter import EventTopicFilter from .rtsp import RTSPClient, Signal, State from .websocket import WebSocketClient @@ -25,12 +26,16 @@ class StreamManager: """Setup, start, stop and retry stream.""" - def __init__(self, device: AxisDevice) -> None: + def __init__( + self, + device: AxisDevice, + ) -> None: """Initialize stream manager.""" self.device = device self.video = None # Unsupported self.audio = None # Unsupported self.event = False + self._event_subscription: EventTopicFilter = EventTopicFilter.for_all_events() self.stream: StreamTransport | None = None self.connection_status_callback: list[Callable[[Signal], None]] = [] @@ -124,10 +129,13 @@ def _is_stream_stopped(self) -> bool: def _build_stream(self) -> StreamTransport: """Build transport based on device capabilities and manager settings.""" if self.use_websocket: + topic_filters = self._event_subscription.transport_topic_filters return WebSocketClient( self.device, self.websocket_url, self.session_callback, + event_filter_list=[{"topicFilter": t} for t in topic_filters] + or [{"topicFilter": "//."}], ) return RTSPClient( @@ -138,6 +146,10 @@ def _build_stream(self) -> StreamTransport: self.session_callback, ) + def set_event_subscription(self, request: EventTopicFilter) -> None: + """Set the event topic filter for future WebSocket stream sessions.""" + self._event_subscription = request + def session_callback(self, signal: Signal) -> None: """Signalling from stream session. diff --git a/tests/test_event_extension_contracts.py b/tests/test_event_extension_contracts.py new file mode 100644 index 00000000..a1bd2205 --- /dev/null +++ b/tests/test_event_extension_contracts.py @@ -0,0 +1,19 @@ +"""Tests for extension contract capability matrix.""" + +from axis.models.events.transport_capabilities import ( + TRANSPORT_FILTER_CAPABILITIES, + EventTransport, +) + + +def test_transport_filter_capabilities() -> None: + """Capability matrix should reflect expected transport support.""" + assert not TRANSPORT_FILTER_CAPABILITIES[ + EventTransport.RTSP + ].supports_upstream_topic_filter + assert TRANSPORT_FILTER_CAPABILITIES[ + EventTransport.WEBSOCKET + ].supports_upstream_topic_filter + assert TRANSPORT_FILTER_CAPABILITIES[ + EventTransport.MQTT + ].supports_upstream_topic_filter diff --git a/tests/test_event_instances.py b/tests/test_event_instances.py index e2f6e616..f4f9de3b 100644 --- a/tests/test_event_instances.py +++ b/tests/test_event_instances.py @@ -8,13 +8,14 @@ import pytest from axis.models.event import Event -from axis.models.event_instance import ( +from axis.models.events.event_instance import ( EventInstance, EventInstanceData, EventInstanceSimpleItem, EventInstanceSource, get_events, ) +from axis.models.events.topic_filter import EventTopicFilter, to_topic_filter from .event_fixtures import ( EVENT_INSTANCE_PIR_SENSOR, @@ -27,7 +28,7 @@ ) if TYPE_CHECKING: - from axis.interfaces.event_instances import EventInstanceHandler + from axis.interfaces.events.event_instances import EventInstanceHandler @pytest.fixture @@ -357,6 +358,55 @@ async def test_event_instance_synthesizes_unknown_topics( ) +async def test_supported_event_descriptors_and_filter_payloads( + http_route_mock, event_instances +) -> None: + """Extension helpers should expose normalized descriptors and payloads.""" + http_route_mock.post("/vapix/services").respond( + text=EVENT_INSTANCES, + headers={"Content-Type": "application/soap+xml; charset=utf-8"}, + ) + + await event_instances.update() + + descriptors = event_instances.get_supported_event_descriptors() + assert "tns1:Device/tnsaxis:Sensor/PIR" in descriptors + assert ( + descriptors["tns1:Device/tnsaxis:Sensor/PIR"]["topic_filter"] + == "onvif:Device/axis:Sensor/PIR" + ) + + payloads = event_instances.build_transport_filter_payloads( + event_filter=EventTopicFilter.from_topics(["onvif:Device/axis:Sensor/PIR"]) + ) + assert payloads == { + "canonical_topics": ["tns1:Device/tnsaxis:Sensor/PIR"], + "mqtt_topics": ["onvif:Device/axis:Sensor/PIR"], + "websocket_topic_filters": ["onvif:Device/axis:Sensor/PIR"], + } + + +async def test_transport_filter_payloads_default_to_supported_topics( + http_route_mock, event_instances +) -> None: + """No explicit filter should build payloads from supported topics.""" + http_route_mock.post("/vapix/services").respond( + text=EVENT_INSTANCES, + headers={"Content-Type": "application/soap+xml; charset=utf-8"}, + ) + + await event_instances.update() + + expected_topics = list(event_instances.get_supported_topics()) + payloads = event_instances.build_transport_filter_payloads() + + assert payloads["canonical_topics"] == expected_topics + assert payloads["mqtt_topics"] == [ + to_topic_filter(topic) for topic in expected_topics + ] + assert payloads["websocket_topic_filters"] == payloads["mqtt_topics"] + + async def test_expected_events_protocol_normalization(http_route_mock, event_instances): """Expected-event discovery is protocol-agnostic and deterministic.""" http_route_mock.post("/vapix/services").respond( diff --git a/tests/test_event_instances_protocol_parity.py b/tests/test_event_instances_protocol_parity.py index f9fa1b71..a9dc6677 100644 --- a/tests/test_event_instances_protocol_parity.py +++ b/tests/test_event_instances_protocol_parity.py @@ -15,7 +15,7 @@ from .event_fixtures import EVENT_INSTANCES, LIGHT_STATUS_INIT, PIR_INIT, VMD4_C1P1_INIT if TYPE_CHECKING: - from axis.interfaces.event_instances import EventInstanceHandler + from axis.interfaces.events.event_instances import EventInstanceHandler @pytest.fixture diff --git a/tests/test_event_manager_extensions.py b/tests/test_event_manager_extensions.py new file mode 100644 index 00000000..11221a72 --- /dev/null +++ b/tests/test_event_manager_extensions.py @@ -0,0 +1,49 @@ +"""Tests for EventManager extension hooks.""" + +from axis.interfaces.events.event_manager import EventManager + + +def test_allowed_topics_filter_is_opt_in() -> None: + """Allow-list should only apply when explicitly configured.""" + manager = EventManager() + received = [] + manager.subscribe(lambda event: received.append(event.topic)) + + pir_event = { + "topic": "tns1:Device/tnsaxis:Sensor/PIR", + "source": "sensor", + "source_idx": "0", + "type": "state", + "value": "0", + "operation": "Initialized", + } + light_event = { + "topic": "tns1:Device/tnsaxis:Light/Status", + "source": "id", + "source_idx": "0", + "type": "state", + "value": "OFF", + "operation": "Initialized", + } + + manager.handler(pir_event) + manager.handler(light_event) + assert received == [ + "tns1:Device/tnsaxis:Sensor/PIR", + "tns1:Device/tnsaxis:Light/Status", + ] + + manager.set_allowed_topics(["tns1:Device/tnsaxis:Sensor/PIR"]) + manager.handler(light_event) + assert received == [ + "tns1:Device/tnsaxis:Sensor/PIR", + "tns1:Device/tnsaxis:Light/Status", + ] + + manager.set_allowed_topics(None) + manager.handler(light_event) + assert received == [ + "tns1:Device/tnsaxis:Sensor/PIR", + "tns1:Device/tnsaxis:Light/Status", + "tns1:Device/tnsaxis:Light/Status", + ] diff --git a/tests/test_event_stream.py b/tests/test_event_stream.py index 32b01c04..b36f6727 100644 --- a/tests/test_event_stream.py +++ b/tests/test_event_stream.py @@ -43,7 +43,7 @@ if TYPE_CHECKING: from axis.device import AxisDevice - from axis.interfaces.event_manager import EventManager + from axis.interfaces.events.event_manager import EventManager @pytest.fixture diff --git a/tests/test_event_topic_filter.py b/tests/test_event_topic_filter.py new file mode 100644 index 00000000..4a8ebafd --- /dev/null +++ b/tests/test_event_topic_filter.py @@ -0,0 +1,124 @@ +"""Tests for EventTopicFilter value object. + +pytest --cov-report term-missing --cov=axis.models.events.topic_filter tests/test_event_topic_filter.py +""" + +import pytest + +from axis.models.events.event import EventTopic +from axis.models.events.topic_filter import EventTopicFilter + + +class TestForAllEvents: + """Tests for EventTopicFilter.for_all_events().""" + + def test_is_wildcard(self) -> None: + """for_all_events() produces a wildcard filter.""" + f = EventTopicFilter.for_all_events() + assert f.is_wildcard is True + + def test_canonical_topics_empty(self) -> None: + """Wildcard returns empty canonical_topics list.""" + f = EventTopicFilter.for_all_events() + assert f.canonical_topics == [] + + def test_transport_topic_filters_empty(self) -> None: + """Wildcard returns empty transport_topic_filters list.""" + f = EventTopicFilter.for_all_events() + assert f.transport_topic_filters == [] + + def test_equality(self) -> None: + """Two wildcard instances are equal.""" + assert EventTopicFilter.for_all_events() == EventTopicFilter.for_all_events() + + def test_hashable(self) -> None: + """Wildcard filter is hashable (frozen dataclass).""" + f = EventTopicFilter.for_all_events() + assert hash(f) == hash(EventTopicFilter.for_all_events()) + + +class TestFromTopics: + """Tests for EventTopicFilter.from_topics().""" + + def test_empty_raises(self) -> None: + """from_topics([]) raises ValueError.""" + with pytest.raises(ValueError, match="topics must not be empty"): + EventTopicFilter.from_topics([]) + + def test_canonical_string_stored_as_is(self) -> None: + """Canonical-format strings are stored without conversion.""" + topic = "tns1:Device/tnsaxis:Sensor/PIR" + f = EventTopicFilter.from_topics([topic]) + assert f.canonical_topics == [topic] + + def test_mqtt_string_converted_to_canonical(self) -> None: + """MQTT-format strings are converted to canonical.""" + f = EventTopicFilter.from_topics(["onvif:Device/axis:Sensor/PIR"]) + assert f.canonical_topics == ["tns1:Device/tnsaxis:Sensor/PIR"] + + def test_event_topic_enum(self) -> None: + """EventTopic enum values are accepted and stored as canonical.""" + f = EventTopicFilter.from_topics([EventTopic.PIR]) + assert f.canonical_topics == [EventTopic.PIR.value] + + def test_mixed_input(self) -> None: + """Mixed str and EventTopic inputs are normalized together.""" + f = EventTopicFilter.from_topics([EventTopic.PIR, "onvif:Device/axis:IO/Port"]) + assert f.canonical_topics == [ + "tns1:Device/tnsaxis:IO/Port", + "tns1:Device/tnsaxis:Sensor/PIR", + ] + + def test_is_not_wildcard(self) -> None: + """from_topics() result is never a wildcard.""" + f = EventTopicFilter.from_topics(["tns1:Device/tnsaxis:Sensor/PIR"]) + assert f.is_wildcard is False + + def test_deduplication(self) -> None: + """Duplicate topics (after normalization) are deduplicated.""" + f = EventTopicFilter.from_topics( + [ + "tns1:Device/tnsaxis:Sensor/PIR", + "onvif:Device/axis:Sensor/PIR", # same as above in MQTT form + ] + ) + assert f.canonical_topics == ["tns1:Device/tnsaxis:Sensor/PIR"] + + def test_sorted_output(self) -> None: + """canonical_topics and transport_topic_filters return sorted lists.""" + f = EventTopicFilter.from_topics( + [ + "tns1:Device/tnsaxis:Sensor/PIR", + "tns1:Device/tnsaxis:IO/Port", + ] + ) + assert f.canonical_topics == [ + "tns1:Device/tnsaxis:IO/Port", + "tns1:Device/tnsaxis:Sensor/PIR", + ] + assert f.transport_topic_filters == [ + "onvif:Device/axis:IO/Port", + "onvif:Device/axis:Sensor/PIR", + ] + + def test_transport_topic_filters_conversion(self) -> None: + """transport_topic_filters returns MQTT-format strings.""" + f = EventTopicFilter.from_topics(["tns1:Device/tnsaxis:Sensor/PIR"]) + assert f.transport_topic_filters == ["onvif:Device/axis:Sensor/PIR"] + + def test_immutable(self) -> None: + """EventTopicFilter is immutable (frozen dataclass).""" + f = EventTopicFilter.from_topics(["tns1:Device/tnsaxis:Sensor/PIR"]) + with pytest.raises(AttributeError): + f._topics = frozenset() # type: ignore[misc] + + def test_equality(self) -> None: + """Two filters with the same topics are equal.""" + f1 = EventTopicFilter.from_topics(["tns1:Device/tnsaxis:Sensor/PIR"]) + f2 = EventTopicFilter.from_topics(["onvif:Device/axis:Sensor/PIR"]) + assert f1 == f2 + + def test_hashable(self) -> None: + """from_topics result is hashable.""" + f = EventTopicFilter.from_topics(["tns1:Device/tnsaxis:Sensor/PIR"]) + assert isinstance(hash(f), int) diff --git a/tests/test_mqtt.py b/tests/test_mqtt.py index bfd00916..c03cb884 100644 --- a/tests/test_mqtt.py +++ b/tests/test_mqtt.py @@ -355,6 +355,37 @@ async def test_configure_event_publication_specific_topics( } +async def test_build_topic_filters_deduplicates_preserving_order( + mqtt_client: MqttClientHandler, +) -> None: + """Topic filter builder should keep order while removing duplicates.""" + assert mqtt_client.build_topic_filters(["a", "b", "a"]) == ["a", "b"] + + +async def test_configure_event_publication_normalizes_topics( + mock_mqtt_client_request, + mock_mqtt_event_request, + mqtt_client: MqttClientHandler, +): + """Optional normalize flag should convert canonical topics to filter format.""" + route = mock_mqtt_event_request({}) + + await mqtt_client.configure_event_publication( + ["tns1:Device/tnsaxis:Sensor/PIR"], + normalize_topics=True, + ) + + assert route.called + assert json.loads(route.calls.last.request.content) == { + "apiVersion": "1.0", + "context": "Axis library", + "method": "configureEventPublication", + "params": { + "eventFilterList": [{"topicFilter": "onvif:Device/axis:Sensor/PIR"}] + }, + } + + async def test_convert_json_to_event(): """Verify conversion from json to event.""" event = mqtt_json_to_event( diff --git a/tests/test_stream_manager.py b/tests/test_stream_manager.py index fb3e57bd..1795a9bf 100644 --- a/tests/test_stream_manager.py +++ b/tests/test_stream_manager.py @@ -9,6 +9,7 @@ import pytest from axis.models.api_discovery import ApiId +from axis.models.events.topic_filter import EventTopicFilter from axis.rtsp import Signal, State from axis.stream_manager import RETRY_TIMER, StreamManager from axis.websocket import WebSocketClient, WebSocketFailureReason @@ -121,6 +122,83 @@ async def test_start_uses_websocket_when_supported( rtsp_client.assert_not_called() +@patch("axis.stream_manager.RTSPClient") +@patch("axis.stream_manager.WebSocketClient") +async def test_stream_manager_passes_event_filter_list_to_websocket( + websocket_client, rtsp_client, stream_manager +): + """set_event_subscription should forward filter to websocket.""" + websocket_client.return_value.start = AsyncMock() + stream_manager.event = True + stream_manager.device.config.websocket_enabled = True + stream_manager.device.vapix.api_discovery._items[ + ApiId.EVENT_STREAMING_OVER_WEBSOCKET + ] = MagicMock() + stream_manager.set_event_subscription( + EventTopicFilter.from_topics(["onvif:Device//."]) + ) + + stream_manager.start() + + websocket_client.assert_called_once_with( + stream_manager.device, + stream_manager.websocket_url, + stream_manager.session_callback, + event_filter_list=[{"topicFilter": "onvif:Device//."}], + ) + rtsp_client.assert_not_called() + + +@patch("axis.stream_manager.RTSPClient") +@patch("axis.stream_manager.WebSocketClient") +async def test_stream_manager_set_event_subscription( + websocket_client, rtsp_client, stream_manager +): + """set_event_subscription should pass correct filter dicts to websocket.""" + websocket_client.return_value.start = AsyncMock() + stream_manager.event = True + stream_manager.device.config.websocket_enabled = True + stream_manager.device.vapix.api_discovery._items[ + ApiId.EVENT_STREAMING_OVER_WEBSOCKET + ] = MagicMock() + request = EventTopicFilter.from_topics(["tns1:Device/tnsaxis:Sensor/PIR"]) + stream_manager.set_event_subscription(request) + + stream_manager.start() + + websocket_client.assert_called_once_with( + stream_manager.device, + stream_manager.websocket_url, + stream_manager.session_callback, + event_filter_list=[{"topicFilter": "onvif:Device/axis:Sensor/PIR"}], + ) + rtsp_client.assert_not_called() + + +@patch("axis.stream_manager.RTSPClient") +@patch("axis.stream_manager.WebSocketClient") +async def test_stream_manager_wildcard_uses_default_filter( + websocket_client, rtsp_client, stream_manager +): + """Wildcard EventTopicFilter should produce the default //. filter.""" + websocket_client.return_value.start = AsyncMock() + stream_manager.event = True + stream_manager.device.config.websocket_enabled = True + stream_manager.device.vapix.api_discovery._items[ + ApiId.EVENT_STREAMING_OVER_WEBSOCKET + ] = MagicMock() + # Default state is already for_all_events; no explicit set needed + + stream_manager.start() + + websocket_client.assert_called_once_with( + stream_manager.device, + stream_manager.websocket_url, + stream_manager.session_callback, + event_filter_list=[{"topicFilter": "//."}], + ) + + @patch("axis.stream_manager.RTSPClient") @patch("axis.stream_manager.WebSocketClient") async def test_initialize_stream(websocket_client, rtsp_client, stream_manager): diff --git a/tests/test_topic_normalizer.py b/tests/test_topic_normalizer.py new file mode 100644 index 00000000..8b31cdcb --- /dev/null +++ b/tests/test_topic_normalizer.py @@ -0,0 +1,32 @@ +"""Tests for event topic normalization extension contract.""" + +from axis.models.events.topic_filter import ( + detect_format, + to_canonical, + to_mqtt, + to_topic_filter, +) + + +def test_round_trip_topic_conversion() -> None: + """Canonical and MQTT representations should round-trip.""" + canonical = "tns1:Device/tnsaxis:Sensor/PIR" + + mqtt = to_mqtt(canonical) + assert mqtt == "onvif:Device/axis:Sensor/PIR" + assert to_canonical(mqtt) == canonical + + +def test_detect_topic_format() -> None: + """Format detection should classify known topic namespace variants.""" + assert detect_format("tns1:Device/tnsaxis:Sensor/PIR") == "canonical" + assert detect_format("onvif:Device/axis:Sensor/PIR") == "mqtt" + assert detect_format("Device/Sensor/PIR") == "unknown" + + +def test_to_topic_filter_uses_mqtt_form() -> None: + """Topic filter representation should use MQTT namespaces.""" + assert ( + to_topic_filter("tns1:Device/tnsaxis:Light/Status") + == "onvif:Device/axis:Light/Status" + ) diff --git a/tests/test_vapix.py b/tests/test_vapix.py index 4b915cac..e5e7639e 100644 --- a/tests/test_vapix.py +++ b/tests/test_vapix.py @@ -15,12 +15,19 @@ Unauthorized, ) from axis.interfaces.api_handler import HandlerGroup -from axis.models.api_discovery import ListApisRequest +from axis.models.api_discovery import ApiId, ListApisRequest from axis.models.applications.application import ( ApplicationStatus, ListApplicationsRequest, ) from axis.models.basic_device_info import GetAllPropertiesRequest +from axis.models.events.event_instance import ( + EventInstance, + EventInstanceData, + EventInstanceSource, +) +from axis.models.events.topic_filter import EventTopicFilter +from axis.models.events.transport_capabilities import EventTransport from axis.models.light_control import GetLightInformationRequest from axis.models.parameters.param_cgi import ParamRequest from axis.models.port_management import GetPortsRequest @@ -60,6 +67,7 @@ if TYPE_CHECKING: from axis.device import AxisDevice + from axis.interfaces.events.transport_manager import EventTransportManager from axis.interfaces.vapix import Vapix @@ -87,6 +95,12 @@ def vapix(axis_device: AxisDevice) -> Vapix: return axis_device.vapix +@pytest.fixture +def event_transport(axis_device: AxisDevice) -> EventTransportManager: + """Return the event transport manager object.""" + return axis_device.event_transport + + @pytest.fixture def vapix_companion_device(axis_companion_device: AxisDevice) -> Vapix: """Return the vapix object.""" @@ -127,6 +141,211 @@ def test_vapix_not_initialized(vapix: Vapix) -> None: assert not vapix.users.supported +def test_vapix_extension_helpers(event_transport: EventTransportManager) -> None: + """Extension helper APIs should be available without side effects.""" + capabilities = event_transport.get_event_transport_capabilities() + assert EventTransport.RTSP in capabilities + assert EventTransport.WEBSOCKET in capabilities + assert EventTransport.MQTT in capabilities + + assert event_transport.get_supported_event_descriptors() == {} + + +async def test_apply_event_transport_filters_requires_initialized_event_instances( + vapix: Vapix, + event_transport: EventTransportManager, +) -> None: + """Applying filters should require event-instance initialization.""" + with pytest.raises(RuntimeError, match="Event instances are not initialized"): + await event_transport.apply_event_transport_filters( + event_filter=EventTopicFilter.from_topics(["onvif:Device/axis:Sensor/PIR"]) + ) + + +async def test_apply_event_transport_filters_validates_supported_topics( + vapix: Vapix, + event_transport: EventTransportManager, +) -> None: + """Apply should reject requested topics not present in event instances.""" + topic = "tns1:Device/tnsaxis:Sensor/PIR" + vapix.event_instances._items = { + topic: EventInstance( + id=topic, + topic=topic, + topic_filter="onvif:Device/axis:Sensor/PIR", + is_available=True, + is_application_data=False, + name="pir", + stateful=True, + stateless=False, + source=EventInstanceSource(), + data=EventInstanceData(), + ) + } + vapix.event_instances.initialized = True + + payloads = await event_transport.apply_event_transport_filters( + event_filter=EventTopicFilter.from_topics(["onvif:Device/axis:Sensor/PIR"]) + ) + assert payloads is None + + with pytest.raises(ValueError, match="Requested unsupported topics"): + await event_transport.apply_event_transport_filters( + event_filter=EventTopicFilter.from_topics( + ["onvif:Device/axis:Status/SystemReady"] + ) + ) + + +async def test_apply_event_transport_filters_calls_transport_hooks( + vapix: Vapix, + event_transport: EventTransportManager, +) -> None: + """Apply should use websocket, mqtt, and local fallback hooks when enabled.""" + topic = "tns1:Device/tnsaxis:Sensor/PIR" + vapix.event_instances._items = { + topic: EventInstance( + id=topic, + topic=topic, + topic_filter="onvif:Device/axis:Sensor/PIR", + is_available=True, + is_application_data=False, + name="pir", + stateful=True, + stateless=False, + source=EventInstanceSource(), + data=EventInstanceData(), + ) + } + vapix.event_instances.initialized = True + + mqtt_calls: list[list[str]] = [] + + vapix.api_discovery._items[ApiId.MQTT_CLIENT] = type( + "_Api", (), {"version": "1.0"} + )() + + async def _configure_event_publication( + topics: list[str], *_args, **_kwargs + ) -> None: + mqtt_calls.append(topics) + + vapix.mqtt.configure_event_publication = _configure_event_publication # type: ignore[method-assign] + captured_subscriptions: list[object] = [] + + def _capture_subscription(req: object) -> None: + captured_subscriptions.append(req) + + vapix.device.stream.set_event_subscription = _capture_subscription # type: ignore[method-assign] + vapix.device.event.set_allowed_topics = lambda topics: setattr( + vapix.device.event, "_captured_allowed_topics", topics + ) + + await event_transport.apply_event_transport_filters( + event_filter=EventTopicFilter.from_topics(["onvif:Device/axis:Sensor/PIR"]) + ) + + assert len(captured_subscriptions) == 1 + event_filter = captured_subscriptions[0] + assert isinstance(event_filter, EventTopicFilter) + assert event_filter.canonical_topics == [topic] + assert mqtt_calls == [["onvif:Device/axis:Sensor/PIR"]] + assert vapix.device.event._captured_allowed_topics == [topic] + + +async def test_apply_event_transport_filters_builds_default_filter_from_supported( + vapix: Vapix, + event_transport: EventTransportManager, +) -> None: + """Apply should derive filter from supported topics when no filter is provided.""" + topic = "tns1:Device/tnsaxis:Sensor/PIR" + vapix.event_instances._items = { + topic: EventInstance( + id=topic, + topic=topic, + topic_filter="onvif:Device/axis:Sensor/PIR", + is_available=True, + is_application_data=False, + name="pir", + stateful=True, + stateless=False, + source=EventInstanceSource(), + data=EventInstanceData(), + ) + } + vapix.event_instances.initialized = True + + captured_subscriptions: list[object] = [] + + def _capture_subscription(req: object) -> None: + captured_subscriptions.append(req) + + vapix.device.stream.set_event_subscription = _capture_subscription # type: ignore[method-assign] + vapix.device.event.set_allowed_topics = lambda topics: setattr( + vapix.device.event, "_captured_allowed_topics", topics + ) + + await event_transport.apply_event_transport_filters() + + assert len(captured_subscriptions) == 1 + event_filter = captured_subscriptions[0] + assert isinstance(event_filter, EventTopicFilter) + assert event_filter.canonical_topics == [topic] + assert vapix.device.event._captured_allowed_topics == [topic] + + +async def test_apply_event_transport_filters_skips_unsupported_mqtt( + vapix: Vapix, + event_transport: EventTransportManager, +) -> None: + """Apply should not call MQTT hook when MQTT API is unsupported.""" + topic = "tns1:Device/tnsaxis:Sensor/PIR" + vapix.event_instances._items = { + topic: EventInstance( + id=topic, + topic=topic, + topic_filter="onvif:Device/axis:Sensor/PIR", + is_available=True, + is_application_data=False, + name="pir", + stateful=True, + stateless=False, + source=EventInstanceSource(), + data=EventInstanceData(), + ) + } + vapix.event_instances.initialized = True + + mqtt_calls: list[list[str]] = [] + + async def _configure_event_publication( + topics: list[str], *_args, **_kwargs + ) -> None: + mqtt_calls.append(topics) + + vapix.mqtt.configure_event_publication = _configure_event_publication # type: ignore[method-assign] + captured_subscriptions: list[object] = [] + + def _capture_subscription2(req: object) -> None: + captured_subscriptions.append(req) + + vapix.device.stream.set_event_subscription = _capture_subscription2 # type: ignore[method-assign] + vapix.device.event.set_allowed_topics = lambda topics: setattr( + vapix.device.event, "_captured_allowed_topics", topics + ) + + await event_transport.apply_event_transport_filters( + event_filter=EventTopicFilter.from_topics(["onvif:Device/axis:Sensor/PIR"]), + ) + + assert len(captured_subscriptions) == 1 + event_filter = captured_subscriptions[0] + assert isinstance(event_filter, EventTopicFilter) + assert event_filter.canonical_topics == [topic] + assert mqtt_calls == [] + assert vapix.device.event._captured_allowed_topics == [topic] + + def test_api_discovery_handlers_registration(vapix: Vapix) -> None: """Verify grouped API-discovery handlers matches the startup contract.""" handlers = vapix._handlers_by_group(HandlerGroup.API_DISCOVERY)