diff --git a/homeassistant/components/onvif/parsers.py b/homeassistant/components/onvif/parsers.py index 57bd8a974db..d7bbaa4fb3f 100644 --- a/homeassistant/components/onvif/parsers.py +++ b/homeassistant/components/onvif/parsers.py @@ -370,6 +370,63 @@ async def async_parse_vehicle_detector(uid: str, msg) -> Event | None: return None +@PARSERS.register("tns1:RuleEngine/TPSmartEventDetector/TPSmartEvent") +@PARSERS.register("tns1:RuleEngine/PeopleDetector/People") +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/PeopleDetector/People + """ + video_source = "" + video_analytics = "" + rule = "" + topic = "" + vehicle = False + person = False + enabled = False + try: + 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: + if item.Name == "IsVehicle": + vehicle = True + enabled = item.Value == "true" + if item.Name == "IsPeople": + person = True + enabled = 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 + + @PARSERS.register("tns1:RuleEngine/MyRuleDetector/PeopleDetect") async def async_parse_person_detector(uid: str, msg) -> Event | None: """Handle parsing event message. diff --git a/tests/components/onvif/test_parsers.py b/tests/components/onvif/test_parsers.py new file mode 100644 index 00000000000..209e7cbccef --- /dev/null +++ b/tests/components/onvif/test_parsers.py @@ -0,0 +1,335 @@ +"""Test ONVIF parsers.""" + +import datetime +import os + +import onvif +import onvif.settings +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_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_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_missing_attributes(hass: HomeAssistant) -> None: + """Tests async_parse_tplink_detector with missing fields.""" + event = 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", + }, + } + ) + + assert event is None + + +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