From 6571ebf15bf5fc10fa03ef692a4ae177c06f88c4 Mon Sep 17 00:00:00 2001 From: Jeff Terrace Date: Sat, 11 Jan 2025 14:52:46 -0500 Subject: [PATCH] Add additional Tapo ONVIF Person/Vehicle/Line/Tamper/Intrusion events (#135399) --- homeassistant/components/onvif/parsers.py | 88 +++-- tests/components/onvif/test_parsers.py | 384 +++++++++++++++++++++- 2 files changed, 436 insertions(+), 36 deletions(-) diff --git a/homeassistant/components/onvif/parsers.py b/homeassistant/components/onvif/parsers.py index d7bbaa4fb3f..9904a4bbfa9 100644 --- a/homeassistant/components/onvif/parsers.py +++ b/homeassistant/components/onvif/parsers.py @@ -3,6 +3,7 @@ from __future__ import annotations from collections.abc import Callable, Coroutine +import dataclasses import datetime from typing import Any @@ -370,22 +371,56 @@ async def async_parse_vehicle_detector(uid: str, msg) -> Event | None: return None -@PARSERS.register("tns1:RuleEngine/TPSmartEventDetector/TPSmartEvent") +_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" + ), + "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/TPSmartEventDetector/TPSmartEvent + 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 = "" - vehicle = False - person = False - enabled = False try: + video_source = "" + video_analytics = "" + rule = "" topic, payload = extract_message(msg) for source in payload.Source.SimpleItem: if source.Name == "VideoSourceConfigurationToken": @@ -396,34 +431,19 @@ async def async_parse_tplink_detector(uid: str, msg) -> Event | None: rule = source.Value for item in payload.Data.SimpleItem: - if item.Name == "IsVehicle": - vehicle = True - enabled = item.Value == "true" - if item.Name == "IsPeople": - person = True - enabled = item.Value == "true" + event_template = _TAPO_EVENT_TEMPLATES.get(item.Name, None) + if event_template is None: + continue + + return dataclasses.replace( + event_template, + uid=f"{uid}_{topic}_{video_source}_{video_analytics}_{rule}", + value=item.Value == "true", + ) + except (AttributeError, KeyError): return None - if vehicle: - return Event( - f"{uid}_{topic}_{video_source}_{video_analytics}_{rule}", - "Vehicle Detection", - "binary_sensor", - "motion", - None, - enabled, - ) - if person: - return Event( - f"{uid}_{topic}_{video_source}_{video_analytics}_{rule}", - "Person Detection", - "binary_sensor", - "motion", - None, - enabled, - ) - return None diff --git a/tests/components/onvif/test_parsers.py b/tests/components/onvif/test_parsers.py index 209e7cbccef..16172112c11 100644 --- a/tests/components/onvif/test_parsers.py +++ b/tests/components/onvif/test_parsers.py @@ -119,7 +119,83 @@ async def test_line_detector_crossed(hass: HomeAssistant) -> None: ) -async def test_tapo_vehicle(hass: HomeAssistant) -> None: +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( { @@ -198,7 +274,83 @@ async def test_tapo_vehicle(hass: HomeAssistant) -> None: ) -async def test_tapo_person(hass: HomeAssistant) -> None: +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( { @@ -274,6 +426,234 @@ async def test_tapo_person(hass: HomeAssistant) -> None: ) +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.""" event = await get_event(