From 6242ef78c46886b885a71c88d6df29b8944ceb46 Mon Sep 17 00:00:00 2001 From: Jeff Terrace Date: Mon, 2 Mar 2026 17:18:05 -0500 Subject: [PATCH] Move ONVIF event parsing into a module outside core (#164550) Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> Co-authored-by: J. Nick Koston Co-authored-by: J. Nick Koston --- homeassistant/components/onvif/event.py | 52 +- homeassistant/components/onvif/manifest.json | 6 +- homeassistant/components/onvif/parsers.py | 755 ---------------- requirements_all.txt | 3 + requirements_test_all.txt | 3 + tests/components/onvif/__init__.py | 60 +- tests/components/onvif/test_event.py | 137 +++ tests/components/onvif/test_parsers.py | 881 ------------------- 8 files changed, 249 insertions(+), 1648 deletions(-) delete mode 100644 homeassistant/components/onvif/parsers.py create mode 100644 tests/components/onvif/test_event.py delete mode 100644 tests/components/onvif/test_parsers.py diff --git a/homeassistant/components/onvif/event.py b/homeassistant/components/onvif/event.py index 86ec419f8921..22242432ff8a 100644 --- a/homeassistant/components/onvif/event.py +++ b/homeassistant/components/onvif/event.py @@ -16,23 +16,30 @@ from onvif.client import ( ) from onvif.exceptions import ONVIFError from onvif.util import stringify_onvif_error +import onvif_parsers from zeep.exceptions import Fault, TransportError, ValidationError, XMLParseError from homeassistant.components import webhook from homeassistant.config_entries import ConfigEntry +from homeassistant.const import EntityCategory from homeassistant.core import CALLBACK_TYPE, HassJob, HomeAssistant, callback from homeassistant.helpers.device_registry import format_mac from homeassistant.helpers.event import async_call_later from homeassistant.helpers.network import NoURLAvailableError, get_url +from homeassistant.util import dt as dt_util from .const import DOMAIN, LOGGER from .models import Event, PullPointManagerState, WebHookManagerState -from .parsers import PARSERS # Topics in this list are ignored because we do not want to create # entities for them. UNHANDLED_TOPICS: set[str] = {"tns1:MediaControl/VideoEncoderConfiguration"} +ENTITY_CATEGORY_MAPPING: dict[str, EntityCategory] = { + "diagnostic": EntityCategory.DIAGNOSTIC, + "config": EntityCategory.CONFIG, +} + SUBSCRIPTION_ERRORS = (Fault, TimeoutError, TransportError) CREATE_ERRORS = ( ONVIFError, @@ -81,6 +88,18 @@ PULLPOINT_MESSAGE_LIMIT = 100 PULLPOINT_COOLDOWN_TIME = 0.75 +def _local_datetime_or_none(value: str) -> dt.datetime | None: + """Convert strings to datetimes, if invalid, return None.""" + # Handle cameras that return times like '0000-00-00T00:00:00Z' (e.g. Hikvision) + try: + ret = dt_util.parse_datetime(value) + except ValueError: + return None + if ret is not None: + return dt_util.as_local(ret) + return None + + class EventManager: """ONVIF Event Manager.""" @@ -176,7 +195,10 @@ class EventManager: # tns1:RuleEngine/CellMotionDetector/Motion topic = msg.Topic._value_1.rstrip("/.") # noqa: SLF001 - if not (parser := PARSERS.get(topic)): + try: + event = await onvif_parsers.parse(topic, unique_id, msg) + error = None + except onvif_parsers.errors.UnknownTopicError: if topic not in UNHANDLED_TOPICS: LOGGER.warning( "%s: No registered handler for event from %s: %s", @@ -186,10 +208,6 @@ class EventManager: ) UNHANDLED_TOPICS.add(topic) continue - - try: - event = await parser(unique_id, msg) - error = None except (AttributeError, KeyError) as e: event = None error = e @@ -202,10 +220,26 @@ class EventManager: error, msg, ) - return + continue - self.get_uids_by_platform(event.platform).add(event.uid) - self._events[event.uid] = event + value = event.value + if event.device_class == "timestamp" and isinstance(value, str): + value = _local_datetime_or_none(value) + + ha_event = Event( + uid=event.uid, + name=event.name, + platform=event.platform, + device_class=event.device_class, + unit_of_measurement=event.unit_of_measurement, + value=value, + entity_category=ENTITY_CATEGORY_MAPPING.get( + event.entity_category or "" + ), + entity_enabled=event.entity_enabled, + ) + self.get_uids_by_platform(ha_event.platform).add(ha_event.uid) + self._events[ha_event.uid] = ha_event def get_uid(self, uid: str) -> Event | None: """Retrieve event for given id.""" diff --git a/homeassistant/components/onvif/manifest.json b/homeassistant/components/onvif/manifest.json index e35addf52fe3..5a097c525a34 100644 --- a/homeassistant/components/onvif/manifest.json +++ b/homeassistant/components/onvif/manifest.json @@ -13,5 +13,9 @@ "integration_type": "device", "iot_class": "local_push", "loggers": ["onvif", "wsdiscovery", "zeep"], - "requirements": ["onvif-zeep-async==4.0.4", "WSDiscovery==2.1.2"] + "requirements": [ + "onvif-zeep-async==4.0.4", + "onvif_parsers==1.2.2", + "WSDiscovery==2.1.2" + ] } diff --git a/homeassistant/components/onvif/parsers.py b/homeassistant/components/onvif/parsers.py deleted file mode 100644 index 32adf696bde2..000000000000 --- a/homeassistant/components/onvif/parsers.py +++ /dev/null @@ -1,755 +0,0 @@ -"""ONVIF event parsers.""" - -from __future__ import annotations - -from collections.abc import Callable, Coroutine -import dataclasses -import datetime -from typing import Any - -from homeassistant.const import EntityCategory -from homeassistant.util import dt as dt_util -from homeassistant.util.decorator import Registry - -from .models import Event - -PARSERS: Registry[str, Callable[[str, Any], Coroutine[Any, Any, Event | None]]] = ( - Registry() -) - -VIDEO_SOURCE_MAPPING = { - "vsconf": "VideoSourceToken", -} - - -def extract_message(msg: Any) -> tuple[str, Any]: - """Extract the message content and the topic.""" - return msg.Topic._value_1, msg.Message._value_1 # noqa: SLF001 - - -def _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. - """ - return VIDEO_SOURCE_MAPPING.get(source, source) - - -def local_datetime_or_none(value: str) -> datetime.datetime | None: - """Convert strings to datetimes, if invalid, return None.""" - # To handle cameras that return times like '0000-00-00T00:00:00Z' (e.g. hikvision) - try: - ret = dt_util.parse_datetime(value) - except ValueError: - return None - if ret is not None: - return dt_util.as_local(ret) - return None - - -@PARSERS.register("tns1:VideoSource/MotionAlarm") -@PARSERS.register("tns1:Device/Trigger/tnshik:AlarmIn") -async def async_parse_motion_alarm(uid: str, msg) -> Event | None: - """Handle parsing event message. - - Topic: tns1:VideoSource/MotionAlarm - """ - topic, payload = extract_message(msg) - source = payload.Source.SimpleItem[0].Value - return Event( - f"{uid}_{topic}_{source}", - "Motion Alarm", - "binary_sensor", - "motion", - None, - payload.Data.SimpleItem[0].Value == "true", - ) - - -@PARSERS.register("tns1:VideoSource/ImageTooBlurry/AnalyticsService") -@PARSERS.register("tns1:VideoSource/ImageTooBlurry/ImagingService") -@PARSERS.register("tns1:VideoSource/ImageTooBlurry/RecordingService") -async def async_parse_image_too_blurry(uid: str, msg) -> Event | None: - """Handle parsing event message. - - Topic: tns1:VideoSource/ImageTooBlurry/* - """ - topic, payload = extract_message(msg) - source = payload.Source.SimpleItem[0].Value - return Event( - f"{uid}_{topic}_{source}", - "Image Too Blurry", - "binary_sensor", - "problem", - None, - payload.Data.SimpleItem[0].Value == "true", - EntityCategory.DIAGNOSTIC, - ) - - -@PARSERS.register("tns1:VideoSource/ImageTooDark/AnalyticsService") -@PARSERS.register("tns1:VideoSource/ImageTooDark/ImagingService") -@PARSERS.register("tns1:VideoSource/ImageTooDark/RecordingService") -async def async_parse_image_too_dark(uid: str, msg) -> Event | None: - """Handle parsing event message. - - Topic: tns1:VideoSource/ImageTooDark/* - """ - topic, payload = extract_message(msg) - source = payload.Source.SimpleItem[0].Value - return Event( - f"{uid}_{topic}_{source}", - "Image Too Dark", - "binary_sensor", - "problem", - None, - payload.Data.SimpleItem[0].Value == "true", - EntityCategory.DIAGNOSTIC, - ) - - -@PARSERS.register("tns1:VideoSource/ImageTooBright/AnalyticsService") -@PARSERS.register("tns1:VideoSource/ImageTooBright/ImagingService") -@PARSERS.register("tns1:VideoSource/ImageTooBright/RecordingService") -async def async_parse_image_too_bright(uid: str, msg) -> Event | None: - """Handle parsing event message. - - Topic: tns1:VideoSource/ImageTooBright/* - """ - topic, payload = extract_message(msg) - source = payload.Source.SimpleItem[0].Value - return Event( - f"{uid}_{topic}_{source}", - "Image Too Bright", - "binary_sensor", - "problem", - None, - payload.Data.SimpleItem[0].Value == "true", - EntityCategory.DIAGNOSTIC, - ) - - -@PARSERS.register("tns1:VideoSource/GlobalSceneChange/AnalyticsService") -@PARSERS.register("tns1:VideoSource/GlobalSceneChange/ImagingService") -@PARSERS.register("tns1:VideoSource/GlobalSceneChange/RecordingService") -async def async_parse_scene_change(uid: str, msg) -> Event | None: - """Handle parsing event message. - - Topic: tns1:VideoSource/GlobalSceneChange/* - """ - topic, payload = extract_message(msg) - source = payload.Source.SimpleItem[0].Value - return Event( - f"{uid}_{topic}_{source}", - "Global Scene Change", - "binary_sensor", - "problem", - None, - payload.Data.SimpleItem[0].Value == "true", - ) - - -@PARSERS.register("tns1:AudioAnalytics/Audio/DetectedSound") -async def async_parse_detected_sound(uid: str, msg) -> Event | None: - """Handle parsing event message. - - Topic: tns1:AudioAnalytics/Audio/DetectedSound - """ - audio_source = "" - audio_analytics = "" - rule = "" - topic, payload = extract_message(msg) - for source in payload.Source.SimpleItem: - if source.Name == "AudioSourceConfigurationToken": - audio_source = source.Value - if source.Name == "AudioAnalyticsConfigurationToken": - audio_analytics = source.Value - if source.Name == "Rule": - rule = source.Value - - return Event( - f"{uid}_{topic}_{audio_source}_{audio_analytics}_{rule}", - "Detected Sound", - "binary_sensor", - "sound", - None, - payload.Data.SimpleItem[0].Value == "true", - ) - - -@PARSERS.register("tns1:RuleEngine/FieldDetector/ObjectsInside") -async def async_parse_field_detector(uid: str, msg) -> Event | None: - """Handle parsing event message. - - Topic: tns1:RuleEngine/FieldDetector/ObjectsInside - """ - video_source = "" - video_analytics = "" - rule = "" - topic, payload = extract_message(msg) - for source in payload.Source.SimpleItem: - if source.Name == "VideoSourceConfigurationToken": - video_source = _normalize_video_source(source.Value) - if source.Name == "VideoAnalyticsConfigurationToken": - video_analytics = source.Value - if source.Name == "Rule": - rule = source.Value - - return Event( - f"{uid}_{topic}_{video_source}_{video_analytics}_{rule}", - "Field Detection", - "binary_sensor", - "motion", - None, - payload.Data.SimpleItem[0].Value == "true", - ) - - -@PARSERS.register("tns1:RuleEngine/CellMotionDetector/Motion") -async def async_parse_cell_motion_detector(uid: str, msg) -> Event | None: - """Handle parsing event message. - - Topic: tns1:RuleEngine/CellMotionDetector/Motion - """ - video_source = "" - video_analytics = "" - rule = "" - topic, payload = extract_message(msg) - for source in payload.Source.SimpleItem: - if source.Name == "VideoSourceConfigurationToken": - video_source = _normalize_video_source(source.Value) - if source.Name == "VideoAnalyticsConfigurationToken": - video_analytics = source.Value - if source.Name == "Rule": - rule = source.Value - - return Event( - f"{uid}_{topic}_{video_source}_{video_analytics}_{rule}", - "Cell Motion Detection", - "binary_sensor", - "motion", - None, - payload.Data.SimpleItem[0].Value == "true", - ) - - -@PARSERS.register("tns1:RuleEngine/MotionRegionDetector/Motion") -async def async_parse_motion_region_detector(uid: str, msg) -> Event | None: - """Handle parsing event message. - - Topic: tns1:RuleEngine/MotionRegionDetector/Motion - """ - video_source = "" - video_analytics = "" - rule = "" - topic, payload = extract_message(msg) - for source in payload.Source.SimpleItem: - if source.Name == "VideoSourceConfigurationToken": - video_source = _normalize_video_source(source.Value) - if source.Name == "VideoAnalyticsConfigurationToken": - video_analytics = source.Value - if source.Name == "Rule": - rule = source.Value - - return Event( - f"{uid}_{topic}_{video_source}_{video_analytics}_{rule}", - "Motion Region Detection", - "binary_sensor", - "motion", - None, - payload.Data.SimpleItem[0].Value in ["1", "true"], - ) - - -@PARSERS.register("tns1:RuleEngine/TamperDetector/Tamper") -async def async_parse_tamper_detector(uid: str, msg) -> Event | None: - """Handle parsing event message. - - Topic: tns1:RuleEngine/TamperDetector/Tamper - """ - video_source = "" - video_analytics = "" - rule = "" - topic, payload = extract_message(msg) - for source in payload.Source.SimpleItem: - if source.Name == "VideoSourceConfigurationToken": - video_source = _normalize_video_source(source.Value) - if source.Name == "VideoAnalyticsConfigurationToken": - video_analytics = source.Value - if source.Name == "Rule": - rule = source.Value - - return Event( - f"{uid}_{topic}_{video_source}_{video_analytics}_{rule}", - "Tamper Detection", - "binary_sensor", - "problem", - None, - payload.Data.SimpleItem[0].Value == "true", - EntityCategory.DIAGNOSTIC, - ) - - -@PARSERS.register("tns1:RuleEngine/MyRuleDetector/DogCatDetect") -async def async_parse_dog_cat_detector(uid: str, msg) -> Event | None: - """Handle parsing event message. - - Topic: tns1:RuleEngine/MyRuleDetector/DogCatDetect - """ - video_source = "" - topic, payload = extract_message(msg) - for source in payload.Source.SimpleItem: - if source.Name == "Source": - video_source = _normalize_video_source(source.Value) - - return Event( - f"{uid}_{topic}_{video_source}", - "Pet Detection", - "binary_sensor", - "motion", - None, - payload.Data.SimpleItem[0].Value == "true", - ) - - -@PARSERS.register("tns1:RuleEngine/MyRuleDetector/VehicleDetect") -async def async_parse_vehicle_detector(uid: str, msg) -> Event | None: - """Handle parsing event message. - - Topic: tns1:RuleEngine/MyRuleDetector/VehicleDetect - """ - video_source = "" - topic, payload = extract_message(msg) - for source in payload.Source.SimpleItem: - if source.Name == "Source": - video_source = _normalize_video_source(source.Value) - - return Event( - f"{uid}_{topic}_{video_source}", - "Vehicle Detection", - "binary_sensor", - "motion", - None, - payload.Data.SimpleItem[0].Value == "true", - ) - - -_TAPO_EVENT_TEMPLATES: dict[str, Event] = { - "IsVehicle": Event( - uid="", - name="Vehicle Detection", - platform="binary_sensor", - device_class="motion", - ), - "IsPeople": Event( - uid="", name="Person Detection", platform="binary_sensor", device_class="motion" - ), - "IsPet": Event( - uid="", name="Pet Detection", platform="binary_sensor", device_class="motion" - ), - "IsLineCross": Event( - uid="", - name="Line Detector Crossed", - platform="binary_sensor", - device_class="motion", - ), - "IsTamper": Event( - uid="", name="Tamper Detection", platform="binary_sensor", device_class="tamper" - ), - "IsIntrusion": Event( - uid="", - name="Intrusion Detection", - platform="binary_sensor", - device_class="safety", - ), -} - - -@PARSERS.register("tns1:RuleEngine/CellMotionDetector/Intrusion") -@PARSERS.register("tns1:RuleEngine/CellMotionDetector/LineCross") -@PARSERS.register("tns1:RuleEngine/CellMotionDetector/People") -@PARSERS.register("tns1:RuleEngine/CellMotionDetector/Tamper") -@PARSERS.register("tns1:RuleEngine/CellMotionDetector/TpSmartEvent") -@PARSERS.register("tns1:RuleEngine/PeopleDetector/People") -@PARSERS.register("tns1:RuleEngine/TPSmartEventDetector/TPSmartEvent") -async def async_parse_tplink_detector(uid: str, msg) -> Event | None: - """Handle parsing tplink smart event messages. - - Topic: tns1:RuleEngine/CellMotionDetector/Intrusion - Topic: tns1:RuleEngine/CellMotionDetector/LineCross - Topic: tns1:RuleEngine/CellMotionDetector/People - Topic: tns1:RuleEngine/CellMotionDetector/Tamper - Topic: tns1:RuleEngine/CellMotionDetector/TpSmartEvent - Topic: tns1:RuleEngine/PeopleDetector/People - Topic: tns1:RuleEngine/TPSmartEventDetector/TPSmartEvent - """ - video_source = "" - video_analytics = "" - rule = "" - topic, payload = extract_message(msg) - for source in payload.Source.SimpleItem: - if source.Name == "VideoSourceConfigurationToken": - video_source = _normalize_video_source(source.Value) - if source.Name == "VideoAnalyticsConfigurationToken": - video_analytics = source.Value - if source.Name == "Rule": - rule = source.Value - - for item in payload.Data.SimpleItem: - event_template = _TAPO_EVENT_TEMPLATES.get(item.Name) - if event_template is None: - continue - - return dataclasses.replace( - event_template, - uid=f"{uid}_{topic}_{video_source}_{video_analytics}_{rule}", - value=item.Value == "true", - ) - - return None - - -@PARSERS.register("tns1:RuleEngine/MyRuleDetector/PeopleDetect") -async def async_parse_person_detector(uid: str, msg) -> Event | None: - """Handle parsing event message. - - Topic: tns1:RuleEngine/MyRuleDetector/PeopleDetect - """ - video_source = "" - topic, payload = extract_message(msg) - for source in payload.Source.SimpleItem: - if source.Name == "Source": - video_source = _normalize_video_source(source.Value) - - return Event( - f"{uid}_{topic}_{video_source}", - "Person Detection", - "binary_sensor", - "motion", - None, - payload.Data.SimpleItem[0].Value == "true", - ) - - -@PARSERS.register("tns1:RuleEngine/MyRuleDetector/FaceDetect") -async def async_parse_face_detector(uid: str, msg) -> Event | None: - """Handle parsing event message. - - Topic: tns1:RuleEngine/MyRuleDetector/FaceDetect - """ - video_source = "" - topic, payload = extract_message(msg) - for source in payload.Source.SimpleItem: - if source.Name == "Source": - video_source = _normalize_video_source(source.Value) - - return Event( - f"{uid}_{topic}_{video_source}", - "Face Detection", - "binary_sensor", - "motion", - None, - payload.Data.SimpleItem[0].Value == "true", - ) - - -@PARSERS.register("tns1:RuleEngine/MyRuleDetector/Visitor") -async def async_parse_visitor_detector(uid: str, msg) -> Event | None: - """Handle parsing event message. - - Topic: tns1:RuleEngine/MyRuleDetector/Visitor - """ - video_source = "" - topic, payload = extract_message(msg) - for source in payload.Source.SimpleItem: - if source.Name == "Source": - video_source = _normalize_video_source(source.Value) - - return Event( - f"{uid}_{topic}_{video_source}", - "Visitor Detection", - "binary_sensor", - "occupancy", - None, - payload.Data.SimpleItem[0].Value == "true", - ) - - -@PARSERS.register("tns1:RuleEngine/MyRuleDetector/Package") -async def async_parse_package_detector(uid: str, msg) -> Event | None: - """Handle parsing event message. - - Topic: tns1:RuleEngine/MyRuleDetector/Package - """ - video_source = "" - topic, payload = extract_message(msg) - for source in payload.Source.SimpleItem: - if source.Name == "Source": - video_source = _normalize_video_source(source.Value) - - return Event( - f"{uid}_{topic}_{video_source}", - "Package Detection", - "binary_sensor", - "occupancy", - None, - payload.Data.SimpleItem[0].Value == "true", - ) - - -@PARSERS.register("tns1:Device/Trigger/DigitalInput") -async def async_parse_digital_input(uid: str, msg) -> Event | None: - """Handle parsing event message. - - Topic: tns1:Device/Trigger/DigitalInput - """ - topic, payload = extract_message(msg) - source = payload.Source.SimpleItem[0].Value - return Event( - f"{uid}_{topic}_{source}", - "Digital Input", - "binary_sensor", - None, - None, - payload.Data.SimpleItem[0].Value == "true", - ) - - -@PARSERS.register("tns1:Device/Trigger/Relay") -async def async_parse_relay(uid: str, msg) -> Event | None: - """Handle parsing event message. - - Topic: tns1:Device/Trigger/Relay - """ - topic, payload = extract_message(msg) - source = payload.Source.SimpleItem[0].Value - return Event( - f"{uid}_{topic}_{source}", - "Relay Triggered", - "binary_sensor", - None, - None, - payload.Data.SimpleItem[0].Value == "active", - ) - - -@PARSERS.register("tns1:Device/HardwareFailure/StorageFailure") -async def async_parse_storage_failure(uid: str, msg) -> Event | None: - """Handle parsing event message. - - Topic: tns1:Device/HardwareFailure/StorageFailure - """ - topic, payload = extract_message(msg) - source = payload.Source.SimpleItem[0].Value - return Event( - f"{uid}_{topic}_{source}", - "Storage Failure", - "binary_sensor", - "problem", - None, - payload.Data.SimpleItem[0].Value == "true", - EntityCategory.DIAGNOSTIC, - ) - - -@PARSERS.register("tns1:Monitoring/ProcessorUsage") -async def async_parse_processor_usage(uid: str, msg) -> Event | None: - """Handle parsing event message. - - Topic: tns1:Monitoring/ProcessorUsage - """ - topic, payload = extract_message(msg) - usage = float(payload.Data.SimpleItem[0].Value) - if usage <= 1: - usage *= 100 - - return Event( - f"{uid}_{topic}", - "Processor Usage", - "sensor", - None, - "percent", - int(usage), - EntityCategory.DIAGNOSTIC, - ) - - -@PARSERS.register("tns1:Monitoring/OperatingTime/LastReboot") -async def async_parse_last_reboot(uid: str, msg) -> Event | None: - """Handle parsing event message. - - Topic: tns1:Monitoring/OperatingTime/LastReboot - """ - topic, payload = extract_message(msg) - date_time = local_datetime_or_none(payload.Data.SimpleItem[0].Value) - return Event( - f"{uid}_{topic}", - "Last Reboot", - "sensor", - "timestamp", - None, - date_time, - EntityCategory.DIAGNOSTIC, - ) - - -@PARSERS.register("tns1:Monitoring/OperatingTime/LastReset") -async def async_parse_last_reset(uid: str, msg) -> Event | None: - """Handle parsing event message. - - Topic: tns1:Monitoring/OperatingTime/LastReset - """ - topic, payload = extract_message(msg) - date_time = local_datetime_or_none(payload.Data.SimpleItem[0].Value) - return Event( - f"{uid}_{topic}", - "Last Reset", - "sensor", - "timestamp", - None, - date_time, - EntityCategory.DIAGNOSTIC, - entity_enabled=False, - ) - - -@PARSERS.register("tns1:Monitoring/Backup/Last") -async def async_parse_backup_last(uid: str, msg) -> Event | None: - """Handle parsing event message. - - Topic: tns1:Monitoring/Backup/Last - """ - topic, payload = extract_message(msg) - date_time = local_datetime_or_none(payload.Data.SimpleItem[0].Value) - return Event( - f"{uid}_{topic}", - "Last Backup", - "sensor", - "timestamp", - None, - date_time, - EntityCategory.DIAGNOSTIC, - entity_enabled=False, - ) - - -@PARSERS.register("tns1:Monitoring/OperatingTime/LastClockSynchronization") -async def async_parse_last_clock_sync(uid: str, msg) -> Event | None: - """Handle parsing event message. - - Topic: tns1:Monitoring/OperatingTime/LastClockSynchronization - """ - topic, payload = extract_message(msg) - date_time = local_datetime_or_none(payload.Data.SimpleItem[0].Value) - return Event( - f"{uid}_{topic}", - "Last Clock Synchronization", - "sensor", - "timestamp", - None, - date_time, - EntityCategory.DIAGNOSTIC, - entity_enabled=False, - ) - - -@PARSERS.register("tns1:RecordingConfig/JobState") -async def async_parse_jobstate(uid: str, msg) -> Event | None: - """Handle parsing event message. - - Topic: tns1:RecordingConfig/JobState - """ - - topic, payload = extract_message(msg) - source = payload.Source.SimpleItem[0].Value - return Event( - f"{uid}_{topic}_{source}", - "Recording Job State", - "binary_sensor", - None, - None, - payload.Data.SimpleItem[0].Value == "Active", - EntityCategory.DIAGNOSTIC, - ) - - -@PARSERS.register("tns1:RuleEngine/LineDetector/Crossed") -async def async_parse_linedetector_crossed(uid: str, msg) -> Event | None: - """Handle parsing event message. - - Topic: tns1:RuleEngine/LineDetector/Crossed - """ - video_source = "" - video_analytics = "" - rule = "" - topic, payload = extract_message(msg) - for source in payload.Source.SimpleItem: - if source.Name == "VideoSourceConfigurationToken": - video_source = source.Value - if source.Name == "VideoAnalyticsConfigurationToken": - video_analytics = source.Value - if source.Name == "Rule": - rule = source.Value - - return Event( - f"{uid}_{topic}_{video_source}_{video_analytics}_{rule}", - "Line Detector Crossed", - "sensor", - None, - None, - payload.Data.SimpleItem[0].Value, - EntityCategory.DIAGNOSTIC, - ) - - -@PARSERS.register("tns1:RuleEngine/CountAggregation/Counter") -async def async_parse_count_aggregation_counter(uid: str, msg) -> Event | None: - """Handle parsing event message. - - Topic: tns1:RuleEngine/CountAggregation/Counter - """ - video_source = "" - video_analytics = "" - rule = "" - topic, payload = extract_message(msg) - for source in payload.Source.SimpleItem: - if source.Name == "VideoSourceConfigurationToken": - video_source = _normalize_video_source(source.Value) - if source.Name == "VideoAnalyticsConfigurationToken": - video_analytics = source.Value - if source.Name == "Rule": - rule = source.Value - - return Event( - f"{uid}_{topic}_{video_source}_{video_analytics}_{rule}", - "Count Aggregation Counter", - "sensor", - None, - None, - payload.Data.SimpleItem[0].Value, - EntityCategory.DIAGNOSTIC, - ) - - -@PARSERS.register("tns1:UserAlarm/IVA/HumanShapeDetect") -async def async_parse_human_shape_detect(uid: str, msg) -> Event | None: - """Handle parsing event message. - - Topic: tns1:UserAlarm/IVA/HumanShapeDetect - """ - topic, payload = extract_message(msg) - video_source = "" - for source in payload.Source.SimpleItem: - if source.Name == "VideoSourceConfigurationToken": - video_source = _normalize_video_source(source.Value) - break - - return Event( - f"{uid}_{topic}_{video_source}", - "Human Shape Detect", - "binary_sensor", - "motion", - None, - payload.Data.SimpleItem[0].Value == "true", - ) diff --git a/requirements_all.txt b/requirements_all.txt index e8927d4138ad..41ac53c09240 100644 --- a/requirements_all.txt +++ b/requirements_all.txt @@ -1681,6 +1681,9 @@ onedrive-personal-sdk==0.1.4 # homeassistant.components.onvif onvif-zeep-async==4.0.4 +# homeassistant.components.onvif +onvif_parsers==1.2.2 + # homeassistant.components.opengarage open-garage==0.2.0 diff --git a/requirements_test_all.txt b/requirements_test_all.txt index 1e7b4b775c32..c5be12f823ad 100644 --- a/requirements_test_all.txt +++ b/requirements_test_all.txt @@ -1467,6 +1467,9 @@ onedrive-personal-sdk==0.1.4 # homeassistant.components.onvif onvif-zeep-async==4.0.4 +# homeassistant.components.onvif +onvif_parsers==1.2.2 + # homeassistant.components.opengarage open-garage==0.2.0 diff --git a/tests/components/onvif/__init__.py b/tests/components/onvif/__init__.py index 868624fb2e45..7c833ed0524e 100644 --- a/tests/components/onvif/__init__.py +++ b/tests/components/onvif/__init__.py @@ -1,16 +1,22 @@ """Tests for the ONVIF integration.""" +from __future__ import annotations + +from collections import defaultdict from unittest.mock import AsyncMock, MagicMock, patch from onvif.exceptions import ONVIFError +from onvif_parsers.model import EventEntity from zeep.exceptions import Fault from homeassistant import config_entries from homeassistant.components.onvif import config_flow from homeassistant.components.onvif.const import CONF_SNAPSHOT_AUTH +from homeassistant.components.onvif.event import EventManager from homeassistant.components.onvif.models import ( Capabilities, DeviceInfo, + Event, Profile, PullPointManagerState, Resolution, @@ -123,7 +129,7 @@ def setup_mock_onvif_camera( mock_onvif_camera.side_effect = mock_constructor -def setup_mock_device(mock_device, capabilities=None, profiles=None): +def setup_mock_device(mock_device, capabilities=None, profiles=None, events=None): """Prepare mock ONVIFDevice.""" mock_device.async_setup = AsyncMock(return_value=True) mock_device.port = 80 @@ -149,7 +155,11 @@ def setup_mock_device(mock_device, capabilities=None, profiles=None): mock_device.events = MagicMock( webhook_manager=MagicMock(state=WebHookManagerState.STARTED), pullpoint_manager=MagicMock(state=PullPointManagerState.PAUSED), + async_stop=AsyncMock(), ) + mock_device.device.close = AsyncMock() + if events: + _setup_mock_events(mock_device.events, events) def mock_constructor( hass: HomeAssistant, config: config_entries.ConfigEntry @@ -160,6 +170,23 @@ def setup_mock_device(mock_device, capabilities=None, profiles=None): mock_device.side_effect = mock_constructor +def _setup_mock_events(mock_events: MagicMock, events: list[Event]) -> None: + """Configure mock events to return proper Event objects.""" + events_by_platform: dict[str, list[Event]] = defaultdict(list) + events_by_uid: dict[str, Event] = {} + uids_by_platform: dict[str, set[str]] = defaultdict(set) + for event in events: + events_by_platform[event.platform].append(event) + events_by_uid[event.uid] = event + uids_by_platform[event.platform].add(event.uid) + + mock_events.get_platform.side_effect = lambda p: list(events_by_platform.get(p, [])) + mock_events.get_uid.side_effect = events_by_uid.get + mock_events.get_uids_by_platform.side_effect = lambda p: set( + uids_by_platform.get(p, set()) + ) + + async def setup_onvif_integration( hass: HomeAssistant, config=None, @@ -168,6 +195,8 @@ async def setup_onvif_integration( entry_id="1", source=config_entries.SOURCE_USER, capabilities=None, + events=None, + raw_events: list[tuple[str, EventEntity]] | None = None, ) -> tuple[MockConfigEntry, MagicMock, MagicMock]: """Create an ONVIF config entry.""" if not config: @@ -202,8 +231,35 @@ async def setup_onvif_integration( setup_mock_onvif_camera(mock_onvif_camera, two_profiles=True) # no discovery mock_discovery.return_value = [] - setup_mock_device(mock_device, capabilities=capabilities) + setup_mock_device(mock_device, capabilities=capabilities, events=events) mock_device.device = mock_onvif_camera + + if raw_events: + # Process raw library events through a real EventManager + # to test the full parsing pipeline including conversions + event_manager = EventManager(hass, mock_onvif_camera, config_entry, NAME) + mock_messages = [] + event_by_topic: dict[str, EventEntity] = {} + for topic, raw_event in raw_events: + mock_msg = MagicMock() + mock_msg.Topic._value_1 = topic + mock_messages.append(mock_msg) + event_by_topic[topic] = raw_event + + async def mock_parse(topic, unique_id, msg): + return event_by_topic.get(topic) + + with patch( + "homeassistant.components.onvif.event.onvif_parsers" + ) as mock_parsers: + mock_parsers.parse = mock_parse + mock_parsers.errors.UnknownTopicError = type( + "UnknownTopicError", (Exception,), {} + ) + await event_manager.async_parse_messages(mock_messages) + + mock_device.events = event_manager + await hass.config_entries.async_setup(config_entry.entry_id) await hass.async_block_till_done() return config_entry, mock_onvif_camera, mock_device diff --git a/tests/components/onvif/test_event.py b/tests/components/onvif/test_event.py new file mode 100644 index 000000000000..eefc41c205f9 --- /dev/null +++ b/tests/components/onvif/test_event.py @@ -0,0 +1,137 @@ +"""Test ONVIF event handling end-to-end.""" + +from onvif_parsers.model import EventEntity + +from homeassistant.components.onvif.models import Capabilities, Event +from homeassistant.const import ATTR_DEVICE_CLASS, STATE_OFF, STATE_ON, EntityCategory +from homeassistant.core import HomeAssistant +from homeassistant.helpers import entity_registry as er + +from . import MAC, setup_onvif_integration + +MOTION_ALARM_UID = f"{MAC}_tns1:VideoSource/MotionAlarm_VideoSourceToken" +IMAGE_TOO_BLURRY_UID = ( + f"{MAC}_tns1:VideoSource/ImageTooBlurry/AnalyticsService_VideoSourceToken" +) +LAST_RESET_UID = f"{MAC}_tns1:Monitoring/LastReset_0" + + +async def test_motion_alarm_event(hass: HomeAssistant) -> None: + """Test that a motion alarm event creates a binary sensor.""" + await setup_onvif_integration( + hass, + capabilities=Capabilities(events=True, imaging=True, ptz=True), + events=[ + Event( + uid=MOTION_ALARM_UID, + name="Motion Alarm", + platform="binary_sensor", + device_class="motion", + value=True, + ), + ], + ) + + state = hass.states.get("binary_sensor.testcamera_motion_alarm") + assert state is not None + assert state.state == STATE_ON + assert state.attributes[ATTR_DEVICE_CLASS] == "motion" + + +async def test_motion_alarm_event_off(hass: HomeAssistant) -> None: + """Test that a motion alarm event with false value is off.""" + await setup_onvif_integration( + hass, + capabilities=Capabilities(events=True, imaging=True, ptz=True), + events=[ + Event( + uid=MOTION_ALARM_UID, + name="Motion Alarm", + platform="binary_sensor", + device_class="motion", + value=False, + ), + ], + ) + + state = hass.states.get("binary_sensor.testcamera_motion_alarm") + assert state is not None + assert state.state == STATE_OFF + + +async def test_diagnostic_event_entity_category( + hass: HomeAssistant, entity_registry: er.EntityRegistry +) -> None: + """Test that a diagnostic event gets the correct entity category.""" + await setup_onvif_integration( + hass, + capabilities=Capabilities(events=True, imaging=True, ptz=True), + events=[ + Event( + uid=IMAGE_TOO_BLURRY_UID, + name="Image Too Blurry", + platform="binary_sensor", + device_class="problem", + value=True, + entity_category=EntityCategory.DIAGNOSTIC, + ), + ], + ) + + state = hass.states.get("binary_sensor.testcamera_image_too_blurry") + assert state is not None + assert state.state == STATE_ON + + entry = entity_registry.async_get("binary_sensor.testcamera_image_too_blurry") + assert entry is not None + assert entry.entity_category is EntityCategory.DIAGNOSTIC + + +async def test_timestamp_event_conversion(hass: HomeAssistant) -> None: + """Test that timestamp sensor events get string values converted to datetime.""" + await setup_onvif_integration( + hass, + capabilities=Capabilities(events=True, imaging=True, ptz=True), + raw_events=[ + ( + "tns1:Monitoring/LastReset", + EventEntity( + uid=LAST_RESET_UID, + name="Last Reset", + platform="sensor", + device_class="timestamp", + value="2023-10-01T12:00:00Z", + ), + ), + ], + ) + + state = hass.states.get("sensor.testcamera_last_reset") + assert state is not None + # Verify the string was converted to a datetime (raw string would end + # with "Z", converted datetime rendered by SensorEntity has "+00:00") + assert state.state == "2023-10-01T12:00:00+00:00" + + +async def test_timestamp_event_invalid_value(hass: HomeAssistant) -> None: + """Test that invalid timestamp values result in unknown state.""" + await setup_onvif_integration( + hass, + capabilities=Capabilities(events=True, imaging=True, ptz=True), + raw_events=[ + ( + "tns1:Monitoring/LastReset", + EventEntity( + uid=LAST_RESET_UID, + name="Last Reset", + platform="sensor", + device_class="timestamp", + value="0000-00-00T00:00:00Z", + ), + ), + ], + ) + + state = hass.states.get("sensor.testcamera_last_reset") + assert state is not None + assert state.state == "unknown" diff --git a/tests/components/onvif/test_parsers.py b/tests/components/onvif/test_parsers.py deleted file mode 100644 index 8448a6e81950..000000000000 --- a/tests/components/onvif/test_parsers.py +++ /dev/null @@ -1,881 +0,0 @@ -"""Test ONVIF parsers.""" - -import datetime -import os - -import onvif -import onvif.settings -import pytest -from zeep import Client -from zeep.transports import Transport - -from homeassistant.components.onvif import models, parsers -from homeassistant.core import HomeAssistant - -TEST_UID = "test-unique-id" - - -async def get_event(notification_data: dict) -> models.Event: - """Take in a zeep dict, run it through the parser, and return an Event. - - When the parser encounters an unknown topic that it doesn't know how to parse, - it outputs a message 'No registered handler for event from ...' along with a - print out of the serialized xml message from zeep. If it tries to parse and - can't, it prints out 'Unable to parse event from ...' along with the same - serialized message. This method can take the output directly from these log - messages and run them through the parser, which makes it easy to add new unit - tests that verify the message can now be parsed. - """ - zeep_client = Client( - f"{os.path.dirname(onvif.__file__)}/wsdl/events.wsdl", - wsse=None, - transport=Transport(), - ) - - notif_msg_type = zeep_client.get_type("ns5:NotificationMessageHolderType") - assert notif_msg_type is not None - notif_msg = notif_msg_type(**notification_data) - assert notif_msg is not None - - # The xsd:any type embedded inside the message doesn't parse, so parse it manually. - msg_elem = zeep_client.get_element("ns8:Message") - assert msg_elem is not None - msg_data = msg_elem(**notification_data["Message"]["_value_1"]) - assert msg_data is not None - notif_msg.Message._value_1 = msg_data - - parser = parsers.PARSERS.get(notif_msg.Topic._value_1) - assert parser is not None - - return await parser(TEST_UID, notif_msg) - - -async def test_line_detector_crossed(hass: HomeAssistant) -> None: - """Tests tns1:RuleEngine/LineDetector/Crossed.""" - event = await get_event( - { - "SubscriptionReference": { - "Address": {"_value_1": None, "_attr_1": None}, - "ReferenceParameters": None, - "Metadata": None, - "_value_1": None, - "_attr_1": None, - }, - "Topic": { - "_value_1": "tns1:RuleEngine/LineDetector/Crossed", - "Dialect": "http://www.onvif.org/ver10/tev/topicExpression/ConcreteSet", - "_attr_1": {}, - }, - "ProducerReference": { - "Address": { - "_value_1": "xx.xx.xx.xx/onvif/event/alarm", - "_attr_1": None, - }, - "ReferenceParameters": None, - "Metadata": None, - "_value_1": None, - "_attr_1": None, - }, - "Message": { - "_value_1": { - "Source": { - "SimpleItem": [ - { - "Name": "VideoSourceConfigurationToken", - "Value": "video_source_config1", - }, - { - "Name": "VideoAnalyticsConfigurationToken", - "Value": "analytics_video_source", - }, - {"Name": "Rule", "Value": "MyLineDetectorRule"}, - ], - "ElementItem": [], - "Extension": None, - "_attr_1": None, - }, - "Key": None, - "Data": { - "SimpleItem": [{"Name": "ObjectId", "Value": "0"}], - "ElementItem": [], - "Extension": None, - "_attr_1": None, - }, - "Extension": None, - "UtcTime": datetime.datetime(2020, 5, 24, 7, 24, 47), - "PropertyOperation": "Initialized", - "_attr_1": {}, - } - }, - } - ) - - assert event is not None - assert event.name == "Line Detector Crossed" - assert event.platform == "sensor" - assert event.value == "0" - assert event.uid == ( - f"{TEST_UID}_tns1:RuleEngine/LineDetector/" - "Crossed_video_source_config1_analytics_video_source_MyLineDetectorRule" - ) - - -async def test_tapo_line_crossed(hass: HomeAssistant) -> None: - """Tests tns1:RuleEngine/CellMotionDetector/LineCross.""" - event = await get_event( - { - "SubscriptionReference": { - "Address": { - "_value_1": "http://CAMERA_LOCAL_IP:2020/event-0_2020", - "_attr_1": None, - }, - "ReferenceParameters": None, - "Metadata": None, - "_value_1": None, - "_attr_1": None, - }, - "Topic": { - "_value_1": "tns1:RuleEngine/CellMotionDetector/LineCross", - "Dialect": "http://www.onvif.org/ver10/tev/topicExpression/ConcreteSet", - "_attr_1": {}, - }, - "ProducerReference": { - "Address": { - "_value_1": "http://CAMERA_LOCAL_IP:5656/event", - "_attr_1": None, - }, - "ReferenceParameters": None, - "Metadata": None, - "_value_1": None, - "_attr_1": None, - }, - "Message": { - "_value_1": { - "Source": { - "SimpleItem": [ - { - "Name": "VideoSourceConfigurationToken", - "Value": "vsconf", - }, - { - "Name": "VideoAnalyticsConfigurationToken", - "Value": "VideoAnalyticsToken", - }, - {"Name": "Rule", "Value": "MyLineCrossDetectorRule"}, - ], - "ElementItem": [], - "Extension": None, - "_attr_1": None, - }, - "Key": None, - "Data": { - "SimpleItem": [{"Name": "IsLineCross", "Value": "true"}], - "ElementItem": [], - "Extension": None, - "_attr_1": None, - }, - "Extension": None, - "UtcTime": datetime.datetime( - 2025, 1, 3, 21, 5, 14, tzinfo=datetime.UTC - ), - "PropertyOperation": "Changed", - "_attr_1": {}, - } - }, - } - ) - - assert event is not None - assert event.name == "Line Detector Crossed" - assert event.platform == "binary_sensor" - assert event.device_class == "motion" - assert event.value - assert event.uid == ( - f"{TEST_UID}_tns1:RuleEngine/CellMotionDetector/" - "LineCross_VideoSourceToken_VideoAnalyticsToken_MyLineCrossDetectorRule" - ) - - -async def test_tapo_tpsmartevent_vehicle(hass: HomeAssistant) -> None: - """Tests tns1:RuleEngine/TPSmartEventDetector/TPSmartEvent - vehicle.""" - event = await get_event( - { - "Message": { - "_value_1": { - "Data": { - "ElementItem": [], - "Extension": None, - "SimpleItem": [{"Name": "IsVehicle", "Value": "true"}], - "_attr_1": None, - }, - "Extension": None, - "Key": None, - "PropertyOperation": "Changed", - "Source": { - "ElementItem": [], - "Extension": None, - "SimpleItem": [ - { - "Name": "VideoSourceConfigurationToken", - "Value": "vsconf", - }, - { - "Name": "VideoAnalyticsConfigurationToken", - "Value": "VideoAnalyticsToken", - }, - { - "Name": "Rule", - "Value": "MyTPSmartEventDetectorRule", - }, - ], - "_attr_1": None, - }, - "UtcTime": datetime.datetime( - 2024, 11, 2, 0, 33, 11, tzinfo=datetime.UTC - ), - "_attr_1": {}, - } - }, - "ProducerReference": { - "Address": { - "_attr_1": None, - "_value_1": "http://192.168.56.127:5656/event", - }, - "Metadata": None, - "ReferenceParameters": None, - "_attr_1": None, - "_value_1": None, - }, - "SubscriptionReference": { - "Address": { - "_attr_1": None, - "_value_1": "http://192.168.56.127:2020/event-0_2020", - }, - "Metadata": None, - "ReferenceParameters": None, - "_attr_1": None, - "_value_1": None, - }, - "Topic": { - "Dialect": "http://www.onvif.org/ver10/tev/topicExpression/ConcreteSet", - "_attr_1": {}, - "_value_1": "tns1:RuleEngine/TPSmartEventDetector/TPSmartEvent", - }, - } - ) - - assert event is not None - assert event.name == "Vehicle Detection" - assert event.platform == "binary_sensor" - assert event.device_class == "motion" - assert event.value - assert event.uid == ( - f"{TEST_UID}_tns1:RuleEngine/TPSmartEventDetector/" - "TPSmartEvent_VideoSourceToken_VideoAnalyticsToken_MyTPSmartEventDetectorRule" - ) - - -async def test_tapo_cellmotiondetector_vehicle(hass: HomeAssistant) -> None: - """Tests tns1:RuleEngine/CellMotionDetector/TpSmartEvent - vehicle.""" - event = await get_event( - { - "SubscriptionReference": { - "Address": { - "_value_1": "http://CAMERA_LOCAL_IP:2020/event-0_2020", - "_attr_1": None, - }, - "ReferenceParameters": None, - "Metadata": None, - "_value_1": None, - "_attr_1": None, - }, - "Topic": { - "_value_1": "tns1:RuleEngine/CellMotionDetector/TpSmartEvent", - "Dialect": "http://www.onvif.org/ver10/tev/topicExpression/ConcreteSet", - "_attr_1": {}, - }, - "ProducerReference": { - "Address": { - "_value_1": "http://CAMERA_LOCAL_IP:5656/event", - "_attr_1": None, - }, - "ReferenceParameters": None, - "Metadata": None, - "_value_1": None, - "_attr_1": None, - }, - "Message": { - "_value_1": { - "Source": { - "SimpleItem": [ - { - "Name": "VideoSourceConfigurationToken", - "Value": "vsconf", - }, - { - "Name": "VideoAnalyticsConfigurationToken", - "Value": "VideoAnalyticsToken", - }, - {"Name": "Rule", "Value": "MyTPSmartEventDetectorRule"}, - ], - "ElementItem": [], - "Extension": None, - "_attr_1": None, - }, - "Key": None, - "Data": { - "SimpleItem": [{"Name": "IsVehicle", "Value": "true"}], - "ElementItem": [], - "Extension": None, - "_attr_1": None, - }, - "Extension": None, - "UtcTime": datetime.datetime( - 2025, 1, 5, 14, 2, 9, tzinfo=datetime.UTC - ), - "PropertyOperation": "Changed", - "_attr_1": {}, - } - }, - } - ) - - assert event is not None - assert event.name == "Vehicle Detection" - assert event.platform == "binary_sensor" - assert event.device_class == "motion" - assert event.value - assert event.uid == ( - f"{TEST_UID}_tns1:RuleEngine/CellMotionDetector/" - "TpSmartEvent_VideoSourceToken_VideoAnalyticsToken_MyTPSmartEventDetectorRule" - ) - - -async def test_tapo_tpsmartevent_person(hass: HomeAssistant) -> None: - """Tests tns1:RuleEngine/TPSmartEventDetector/TPSmartEvent - person.""" - event = await get_event( - { - "Message": { - "_value_1": { - "Data": { - "ElementItem": [], - "Extension": None, - "SimpleItem": [{"Name": "IsPeople", "Value": "true"}], - "_attr_1": None, - }, - "Extension": None, - "Key": None, - "PropertyOperation": "Changed", - "Source": { - "ElementItem": [], - "Extension": None, - "SimpleItem": [ - { - "Name": "VideoSourceConfigurationToken", - "Value": "vsconf", - }, - { - "Name": "VideoAnalyticsConfigurationToken", - "Value": "VideoAnalyticsToken", - }, - {"Name": "Rule", "Value": "MyPeopleDetectorRule"}, - ], - "_attr_1": None, - }, - "UtcTime": datetime.datetime( - 2024, 11, 3, 18, 40, 43, tzinfo=datetime.UTC - ), - "_attr_1": {}, - } - }, - "ProducerReference": { - "Address": { - "_attr_1": None, - "_value_1": "http://192.168.56.127:5656/event", - }, - "Metadata": None, - "ReferenceParameters": None, - "_attr_1": None, - "_value_1": None, - }, - "SubscriptionReference": { - "Address": { - "_attr_1": None, - "_value_1": "http://192.168.56.127:2020/event-0_2020", - }, - "Metadata": None, - "ReferenceParameters": None, - "_attr_1": None, - "_value_1": None, - }, - "Topic": { - "Dialect": "http://www.onvif.org/ver10/tev/topicExpression/ConcreteSet", - "_attr_1": {}, - "_value_1": "tns1:RuleEngine/PeopleDetector/People", - }, - } - ) - - assert event is not None - assert event.name == "Person Detection" - assert event.platform == "binary_sensor" - assert event.device_class == "motion" - assert event.value - assert event.uid == ( - f"{TEST_UID}_tns1:RuleEngine/PeopleDetector/" - "People_VideoSourceToken_VideoAnalyticsToken_MyPeopleDetectorRule" - ) - - -async def test_tapo_tpsmartevent_pet(hass: HomeAssistant) -> None: - """Tests tns1:RuleEngine/TPSmartEventDetector/TPSmartEvent - pet.""" - event = await get_event( - { - "SubscriptionReference": { - "Address": { - "_value_1": "http://192.168.56.63:2020/event-0_2020", - "_attr_1": None, - }, - "ReferenceParameters": None, - "Metadata": None, - "_value_1": None, - "_attr_1": None, - }, - "Topic": { - "_value_1": "tns1:RuleEngine/TPSmartEventDetector/TPSmartEvent", - "Dialect": "http://www.onvif.org/ver10/tev/topicExpression/ConcreteSet", - "_attr_1": {}, - }, - "ProducerReference": { - "Address": { - "_value_1": "http://192.168.56.63:5656/event", - "_attr_1": None, - }, - "ReferenceParameters": None, - "Metadata": None, - "_value_1": None, - "_attr_1": None, - }, - "Message": { - "_value_1": { - "Source": { - "SimpleItem": [ - { - "Name": "VideoSourceConfigurationToken", - "Value": "vsconf", - }, - { - "Name": "VideoAnalyticsConfigurationToken", - "Value": "VideoAnalyticsToken", - }, - {"Name": "Rule", "Value": "MyTPSmartEventDetectorRule"}, - ], - "ElementItem": [], - "Extension": None, - "_attr_1": None, - }, - "Key": None, - "Data": { - "SimpleItem": [{"Name": "IsPet", "Value": "true"}], - "ElementItem": [], - "Extension": None, - "_attr_1": None, - }, - "Extension": None, - "UtcTime": datetime.datetime( - 2025, 1, 22, 13, 24, 57, tzinfo=datetime.UTC - ), - "PropertyOperation": "Changed", - "_attr_1": {}, - } - }, - } - ) - - assert event is not None - assert event.name == "Pet Detection" - assert event.platform == "binary_sensor" - assert event.device_class == "motion" - assert event.value - assert event.uid == ( - f"{TEST_UID}_tns1:RuleEngine/TPSmartEventDetector/" - "TPSmartEvent_VideoSourceToken_VideoAnalyticsToken_MyTPSmartEventDetectorRule" - ) - - -async def test_tapo_cellmotiondetector_person(hass: HomeAssistant) -> None: - """Tests tns1:RuleEngine/CellMotionDetector/People - person.""" - event = await get_event( - { - "SubscriptionReference": { - "Address": { - "_value_1": "http://192.168.56.63:2020/event-0_2020", - "_attr_1": None, - }, - "ReferenceParameters": None, - "Metadata": None, - "_value_1": None, - "_attr_1": None, - }, - "Topic": { - "_value_1": "tns1:RuleEngine/CellMotionDetector/People", - "Dialect": "http://www.onvif.org/ver10/tev/topicExpression/ConcreteSet", - "_attr_1": {}, - }, - "ProducerReference": { - "Address": { - "_value_1": "http://192.168.56.63:5656/event", - "_attr_1": None, - }, - "ReferenceParameters": None, - "Metadata": None, - "_value_1": None, - "_attr_1": None, - }, - "Message": { - "_value_1": { - "Source": { - "SimpleItem": [ - { - "Name": "VideoSourceConfigurationToken", - "Value": "vsconf", - }, - { - "Name": "VideoAnalyticsConfigurationToken", - "Value": "VideoAnalyticsToken", - }, - {"Name": "Rule", "Value": "MyPeopleDetectorRule"}, - ], - "ElementItem": [], - "Extension": None, - "_attr_1": None, - }, - "Key": None, - "Data": { - "SimpleItem": [{"Name": "IsPeople", "Value": "true"}], - "ElementItem": [], - "Extension": None, - "_attr_1": None, - }, - "Extension": None, - "UtcTime": datetime.datetime( - 2025, 1, 3, 20, 9, 22, tzinfo=datetime.UTC - ), - "PropertyOperation": "Changed", - "_attr_1": {}, - } - }, - } - ) - - assert event is not None - assert event.name == "Person Detection" - assert event.platform == "binary_sensor" - assert event.device_class == "motion" - assert event.value - assert event.uid == ( - f"{TEST_UID}_tns1:RuleEngine/CellMotionDetector/" - "People_VideoSourceToken_VideoAnalyticsToken_MyPeopleDetectorRule" - ) - - -async def test_tapo_tamper(hass: HomeAssistant) -> None: - """Tests tns1:RuleEngine/CellMotionDetector/Tamper - tamper.""" - event = await get_event( - { - "SubscriptionReference": { - "Address": { - "_value_1": "http://CAMERA_LOCAL_IP:2020/event-0_2020", - "_attr_1": None, - }, - "ReferenceParameters": None, - "Metadata": None, - "_value_1": None, - "_attr_1": None, - }, - "Topic": { - "_value_1": "tns1:RuleEngine/CellMotionDetector/Tamper", - "Dialect": "http://www.onvif.org/ver10/tev/topicExpression/ConcreteSet", - "_attr_1": {}, - }, - "ProducerReference": { - "Address": { - "_value_1": "http://CAMERA_LOCAL_IP:5656/event", - "_attr_1": None, - }, - "ReferenceParameters": None, - "Metadata": None, - "_value_1": None, - "_attr_1": None, - }, - "Message": { - "_value_1": { - "Source": { - "SimpleItem": [ - { - "Name": "VideoSourceConfigurationToken", - "Value": "vsconf", - }, - { - "Name": "VideoAnalyticsConfigurationToken", - "Value": "VideoAnalyticsToken", - }, - {"Name": "Rule", "Value": "MyTamperDetectorRule"}, - ], - "ElementItem": [], - "Extension": None, - "_attr_1": None, - }, - "Key": None, - "Data": { - "SimpleItem": [{"Name": "IsTamper", "Value": "true"}], - "ElementItem": [], - "Extension": None, - "_attr_1": None, - }, - "Extension": None, - "UtcTime": datetime.datetime( - 2025, 1, 5, 21, 1, 5, tzinfo=datetime.UTC - ), - "PropertyOperation": "Changed", - "_attr_1": {}, - } - }, - } - ) - - assert event is not None - assert event.name == "Tamper Detection" - assert event.platform == "binary_sensor" - assert event.device_class == "tamper" - assert event.value - assert event.uid == ( - f"{TEST_UID}_tns1:RuleEngine/CellMotionDetector/" - "Tamper_VideoSourceToken_VideoAnalyticsToken_MyTamperDetectorRule" - ) - - -async def test_tapo_intrusion(hass: HomeAssistant) -> None: - """Tests tns1:RuleEngine/CellMotionDetector/Intrusion - intrusion.""" - event = await get_event( - { - "SubscriptionReference": { - "Address": { - "_value_1": "http://192.168.100.155:2020/event-0_2020", - "_attr_1": None, - }, - "ReferenceParameters": None, - "Metadata": None, - "_value_1": None, - "_attr_1": None, - }, - "Topic": { - "_value_1": "tns1:RuleEngine/CellMotionDetector/Intrusion", - "Dialect": "http://www.onvif.org/ver10/tev/topicExpression/ConcreteSet", - "_attr_1": {}, - }, - "ProducerReference": { - "Address": { - "_value_1": "http://192.168.100.155:5656/event", - "_attr_1": None, - }, - "ReferenceParameters": None, - "Metadata": None, - "_value_1": None, - "_attr_1": None, - }, - "Message": { - "_value_1": { - "Source": { - "SimpleItem": [ - { - "Name": "VideoSourceConfigurationToken", - "Value": "vsconf", - }, - { - "Name": "VideoAnalyticsConfigurationToken", - "Value": "VideoAnalyticsToken", - }, - {"Name": "Rule", "Value": "MyIntrusionDetectorRule"}, - ], - "ElementItem": [], - "Extension": None, - "_attr_1": None, - }, - "Key": None, - "Data": { - "SimpleItem": [{"Name": "IsIntrusion", "Value": "true"}], - "ElementItem": [], - "Extension": None, - "_attr_1": None, - }, - "Extension": None, - "UtcTime": datetime.datetime( - 2025, 1, 11, 10, 40, 45, tzinfo=datetime.UTC - ), - "PropertyOperation": "Changed", - "_attr_1": {}, - } - }, - } - ) - - assert event is not None - assert event.name == "Intrusion Detection" - assert event.platform == "binary_sensor" - assert event.device_class == "safety" - assert event.value - assert event.uid == ( - f"{TEST_UID}_tns1:RuleEngine/CellMotionDetector/" - "Intrusion_VideoSourceToken_VideoAnalyticsToken_MyIntrusionDetectorRule" - ) - - -async def test_tapo_missing_attributes(hass: HomeAssistant) -> None: - """Tests async_parse_tplink_detector with missing fields.""" - with pytest.raises(AttributeError, match="SimpleItem"): - await get_event( - { - "Message": { - "_value_1": { - "Data": { - "ElementItem": [], - "Extension": None, - "SimpleItem": [{"Name": "IsPeople", "Value": "true"}], - "_attr_1": None, - }, - } - }, - "Topic": { - "_value_1": "tns1:RuleEngine/PeopleDetector/People", - }, - } - ) - - -async def test_tapo_unknown_type(hass: HomeAssistant) -> None: - """Tests async_parse_tplink_detector with unknown event type.""" - event = await get_event( - { - "Message": { - "_value_1": { - "Data": { - "ElementItem": [], - "Extension": None, - "SimpleItem": [{"Name": "IsNotPerson", "Value": "true"}], - "_attr_1": None, - }, - "Source": { - "ElementItem": [], - "Extension": None, - "SimpleItem": [ - { - "Name": "VideoSourceConfigurationToken", - "Value": "vsconf", - }, - { - "Name": "VideoAnalyticsConfigurationToken", - "Value": "VideoAnalyticsToken", - }, - {"Name": "Rule", "Value": "MyPeopleDetectorRule"}, - ], - }, - } - }, - "Topic": { - "_value_1": "tns1:RuleEngine/PeopleDetector/People", - }, - } - ) - - assert event is None - - -async def test_reolink_package(hass: HomeAssistant) -> None: - """Tests reolink package event.""" - event = await get_event( - { - "SubscriptionReference": None, - "Topic": { - "_value_1": "tns1:RuleEngine/MyRuleDetector/Package", - "Dialect": "http://www.onvif.org/ver10/tev/topicExpression/ConcreteSet", - "_attr_1": {}, - }, - "ProducerReference": None, - "Message": { - "_value_1": { - "Source": { - "SimpleItem": [{"Name": "Source", "Value": "000"}], - "ElementItem": [], - "Extension": None, - "_attr_1": None, - }, - "Key": None, - "Data": { - "SimpleItem": [{"Name": "State", "Value": "true"}], - "ElementItem": [], - "Extension": None, - "_attr_1": None, - }, - "Extension": None, - "UtcTime": datetime.datetime( - 2025, 3, 12, 9, 54, 27, tzinfo=datetime.UTC - ), - "PropertyOperation": "Initialized", - "_attr_1": {}, - } - }, - } - ) - - assert event is not None - assert event.name == "Package Detection" - assert event.platform == "binary_sensor" - assert event.device_class == "occupancy" - assert event.value - assert event.uid == (f"{TEST_UID}_tns1:RuleEngine/MyRuleDetector/Package_000") - - -async def test_hikvision_alarm(hass: HomeAssistant) -> None: - """Tests hikvision camera alarm event.""" - event = await get_event( - { - "SubscriptionReference": None, - "Topic": { - "_value_1": "tns1:Device/Trigger/tnshik:AlarmIn", - "Dialect": "http://www.onvif.org/ver10/tev/topicExpression/ConcreteSet", - "_attr_1": {}, - }, - "ProducerReference": None, - "Message": { - "_value_1": { - "Source": { - "SimpleItem": [{"Name": "AlarmInToken", "Value": "AlarmIn_1"}], - "ElementItem": [], - "Extension": None, - "_attr_1": None, - }, - "Key": None, - "Data": { - "SimpleItem": [{"Name": "State", "Value": "true"}], - "ElementItem": [], - "Extension": None, - "_attr_1": None, - }, - "Extension": None, - "UtcTime": datetime.datetime( - 2025, 3, 13, 22, 57, 26, tzinfo=datetime.UTC - ), - "PropertyOperation": "Initialized", - "_attr_1": {}, - } - }, - } - ) - - assert event is not None - assert event.name == "Motion Alarm" - assert event.platform == "binary_sensor" - assert event.device_class == "motion" - assert event.value - assert event.uid == (f"{TEST_UID}_tns1:Device/Trigger/tnshik:AlarmIn_AlarmIn_1")