Skip to content

Commit

Permalink
Add support for onvif tplink person and vehicle events (home-assistan…
Browse files Browse the repository at this point in the history 
…t#130769)

Co-authored-by: J. Nick Koston <[email protected]>
  • Loading branch information
@jterrace @bdraco
jterrace and bdraco authored Dec 4, 2024
1 parent de0ffea commit 106c5d4
Show file tree
Hide file tree
Showing 2 changed files with 392 additions and 0 deletions.
57 changes: 57 additions & 0 deletions homeassistant/components/onvif/parsers.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
335 changes: 335 additions & 0 deletions tests/components/onvif/test_parsers.py
View file
Original file line number Diff line number Diff line change
@@ -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

0 comments on commit 106c5d4

Please sign in to comment.