mirror of
https://github.com/home-assistant/core.git
synced 2026-09-24 23:41:48 -05:00
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 <nick@home-assistant.io> Co-authored-by: J. Nick Koston <nick@koston.org>
This commit is contained in:
Jeff Terrace
co-authored by
Copilot
J. Nick Koston
J. Nick Koston
Copilot
J. Nick Koston
J. Nick Koston
parent
3c342c0768
commit
6242ef78c4
@@ -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."""
|
||||
|
||||
@@ -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"
|
||||
]
|
||||
}
|
||||
|
||||
@@ -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",
|
||||
)
|
||||
Generated
+3
@@ -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
|
||||
|
||||
|
||||
Generated
+3
@@ -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
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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"
|
||||
@@ -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")
|
||||
Reference in New Issue
Block a user