jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-25 06:07:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

Remove unnecessary block use of pylint disable in onvif (#100194)

Browse Source
This commit is contained in:
Erik Montnemery 2023-09-12 16:19:26 +02:00 committed by GitHub
parent c178388956
commit d495208995
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 100 additions and 109 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
homeassistant/components/onvif/event.py
View File

@ -142,7 +142,6 @@ class EventManager:
for update_callback in self._listeners: for update_callback in self._listeners:
update_callback() update_callback()
# pylint: disable=protected-access
async def async_parse_messages(self, messages) -> None: async def async_parse_messages(self, messages) -> None:
"""Parse notification message.""" """Parse notification message."""
unique_id = self.unique_id unique_id = self.unique_id
@ -160,7 +159,7 @@ class EventManager:
# #
# Our parser expects the topic to be # Our parser expects the topic to be
# tns1:RuleEngine/CellMotionDetector/Motion # tns1:RuleEngine/CellMotionDetector/Motion
topic = msg.Topic._value_1.rstrip("/.") topic = msg.Topic._value_1.rstrip("/.") # pylint: disable=protected-access
if not (parser := PARSERS.get(topic)): if not (parser := PARSERS.get(topic)):
if topic not in UNHANDLED_TOPICS: if topic not in UNHANDLED_TOPICS:

206
homeassistant/components/onvif/parsers.py
View File

@ -42,21 +42,21 @@ def local_datetime_or_none(value: str) -> datetime.datetime | None:
@PARSERS.register("tns1:VideoSource/MotionAlarm") @PARSERS.register("tns1:VideoSource/MotionAlarm")
# pylint: disable=protected-access
async def async_parse_motion_alarm(uid: str, msg) -> Event | None: async def async_parse_motion_alarm(uid: str, msg) -> Event | None:
"""Handle parsing event message. """Handle parsing event message.
Topic: tns1:VideoSource/MotionAlarm Topic: tns1:VideoSource/MotionAlarm
""" """
try: try:
source = msg.Message._value_1.Source.SimpleItem[0].Value value_1 = msg.Message._value_1 # pylint: disable=protected-access
source = value_1.Source.SimpleItem[0].Value
return Event( return Event(
f"{uid}_{msg.Topic._value_1}_{source}", f"{uid}_{value_1}_{source}",
"Motion Alarm", "Motion Alarm",
"binary_sensor", "binary_sensor",
"motion", "motion",
None, None,
msg.Message._value_1.Data.SimpleItem[0].Value == "true", value_1.Data.SimpleItem[0].Value == "true",
) )
except (AttributeError, KeyError): except (AttributeError, KeyError):
return None return None
@ -65,21 +65,21 @@ async def async_parse_motion_alarm(uid: str, msg) -> Event | None:
@PARSERS.register("tns1:VideoSource/ImageTooBlurry/AnalyticsService") @PARSERS.register("tns1:VideoSource/ImageTooBlurry/AnalyticsService")
@PARSERS.register("tns1:VideoSource/ImageTooBlurry/ImagingService") @PARSERS.register("tns1:VideoSource/ImageTooBlurry/ImagingService")
@PARSERS.register("tns1:VideoSource/ImageTooBlurry/RecordingService") @PARSERS.register("tns1:VideoSource/ImageTooBlurry/RecordingService")
# pylint: disable=protected-access
async def async_parse_image_too_blurry(uid: str, msg) -> Event | None: async def async_parse_image_too_blurry(uid: str, msg) -> Event | None:
"""Handle parsing event message. """Handle parsing event message.
Topic: tns1:VideoSource/ImageTooBlurry/* Topic: tns1:VideoSource/ImageTooBlurry/*
""" """
try: try:
source = msg.Message._value_1.Source.SimpleItem[0].Value value_1 = msg.Message._value_1 # pylint: disable=protected-access
source = value_1.Source.SimpleItem[0].Value
return Event( return Event(
f"{uid}_{msg.Topic._value_1}_{source}", f"{uid}_{value_1}_{source}",
"Image Too Blurry", "Image Too Blurry",
"binary_sensor", "binary_sensor",
"problem", "problem",
None, None,
msg.Message._value_1.Data.SimpleItem[0].Value == "true", value_1.Data.SimpleItem[0].Value == "true",
EntityCategory.DIAGNOSTIC, EntityCategory.DIAGNOSTIC,
) )
except (AttributeError, KeyError): except (AttributeError, KeyError):
@ -89,21 +89,21 @@ async def async_parse_image_too_blurry(uid: str, msg) -> Event | None:
@PARSERS.register("tns1:VideoSource/ImageTooDark/AnalyticsService") @PARSERS.register("tns1:VideoSource/ImageTooDark/AnalyticsService")
@PARSERS.register("tns1:VideoSource/ImageTooDark/ImagingService") @PARSERS.register("tns1:VideoSource/ImageTooDark/ImagingService")
@PARSERS.register("tns1:VideoSource/ImageTooDark/RecordingService") @PARSERS.register("tns1:VideoSource/ImageTooDark/RecordingService")
# pylint: disable=protected-access
async def async_parse_image_too_dark(uid: str, msg) -> Event | None: async def async_parse_image_too_dark(uid: str, msg) -> Event | None:
"""Handle parsing event message. """Handle parsing event message.
Topic: tns1:VideoSource/ImageTooDark/* Topic: tns1:VideoSource/ImageTooDark/*
""" """
try: try:
source = msg.Message._value_1.Source.SimpleItem[0].Value value_1 = msg.Message._value_1 # pylint: disable=protected-access
source = value_1.Source.SimpleItem[0].Value
return Event( return Event(
f"{uid}_{msg.Topic._value_1}_{source}", f"{uid}_{value_1}_{source}",
"Image Too Dark", "Image Too Dark",
"binary_sensor", "binary_sensor",
"problem", "problem",
None, None,
msg.Message._value_1.Data.SimpleItem[0].Value == "true", value_1.Data.SimpleItem[0].Value == "true",
EntityCategory.DIAGNOSTIC, EntityCategory.DIAGNOSTIC,
) )
except (AttributeError, KeyError): except (AttributeError, KeyError):
@ -113,21 +113,21 @@ async def async_parse_image_too_dark(uid: str, msg) -> Event | None:
@PARSERS.register("tns1:VideoSource/ImageTooBright/AnalyticsService") @PARSERS.register("tns1:VideoSource/ImageTooBright/AnalyticsService")
@PARSERS.register("tns1:VideoSource/ImageTooBright/ImagingService") @PARSERS.register("tns1:VideoSource/ImageTooBright/ImagingService")
@PARSERS.register("tns1:VideoSource/ImageTooBright/RecordingService") @PARSERS.register("tns1:VideoSource/ImageTooBright/RecordingService")
# pylint: disable=protected-access
async def async_parse_image_too_bright(uid: str, msg) -> Event | None: async def async_parse_image_too_bright(uid: str, msg) -> Event | None:
"""Handle parsing event message. """Handle parsing event message.
Topic: tns1:VideoSource/ImageTooBright/* Topic: tns1:VideoSource/ImageTooBright/*
""" """
try: try:
source = msg.Message._value_1.Source.SimpleItem[0].Value value_1 = msg.Message._value_1 # pylint: disable=protected-access
source = value_1.Source.SimpleItem[0].Value
return Event( return Event(
f"{uid}_{msg.Topic._value_1}_{source}", f"{uid}_{value_1}_{source}",
"Image Too Bright", "Image Too Bright",
"binary_sensor", "binary_sensor",
"problem", "problem",
None, None,
msg.Message._value_1.Data.SimpleItem[0].Value == "true", value_1.Data.SimpleItem[0].Value == "true",
EntityCategory.DIAGNOSTIC, EntityCategory.DIAGNOSTIC,
) )
except (AttributeError, KeyError): except (AttributeError, KeyError):
@ -137,28 +137,27 @@ async def async_parse_image_too_bright(uid: str, msg) -> Event | None:
@PARSERS.register("tns1:VideoSource/GlobalSceneChange/AnalyticsService") @PARSERS.register("tns1:VideoSource/GlobalSceneChange/AnalyticsService")
@PARSERS.register("tns1:VideoSource/GlobalSceneChange/ImagingService") @PARSERS.register("tns1:VideoSource/GlobalSceneChange/ImagingService")
@PARSERS.register("tns1:VideoSource/GlobalSceneChange/RecordingService") @PARSERS.register("tns1:VideoSource/GlobalSceneChange/RecordingService")
# pylint: disable=protected-access
async def async_parse_scene_change(uid: str, msg) -> Event | None: async def async_parse_scene_change(uid: str, msg) -> Event | None:
"""Handle parsing event message. """Handle parsing event message.
Topic: tns1:VideoSource/GlobalSceneChange/* Topic: tns1:VideoSource/GlobalSceneChange/*
""" """
try: try:
source = msg.Message._value_1.Source.SimpleItem[0].Value value_1 = msg.Message._value_1 # pylint: disable=protected-access
source = value_1.Source.SimpleItem[0].Value
return Event( return Event(
f"{uid}_{msg.Topic._value_1}_{source}", f"{uid}_{value_1}_{source}",
"Global Scene Change", "Global Scene Change",
"binary_sensor", "binary_sensor",
"problem", "problem",
None, None,
msg.Message._value_1.Data.SimpleItem[0].Value == "true", value_1.Data.SimpleItem[0].Value == "true",
) )
except (AttributeError, KeyError): except (AttributeError, KeyError):
return None return None
@PARSERS.register("tns1:AudioAnalytics/Audio/DetectedSound") @PARSERS.register("tns1:AudioAnalytics/Audio/DetectedSound")
# pylint: disable=protected-access
async def async_parse_detected_sound(uid: str, msg) -> Event | None: async def async_parse_detected_sound(uid: str, msg) -> Event | None:
"""Handle parsing event message. """Handle parsing event message.
@ -168,7 +167,8 @@ async def async_parse_detected_sound(uid: str, msg) -> Event | None:
audio_source = "" audio_source = ""
audio_analytics = "" audio_analytics = ""
rule = "" rule = ""
for source in msg.Message._value_1.Source.SimpleItem: value_1 = msg.Message._value_1 # pylint: disable=protected-access
for source in value_1.Source.SimpleItem:
if source.Name == "AudioSourceConfigurationToken": if source.Name == "AudioSourceConfigurationToken":
audio_source = source.Value audio_source = source.Value
if source.Name == "AudioAnalyticsConfigurationToken": if source.Name == "AudioAnalyticsConfigurationToken":
@ -177,19 +177,18 @@ async def async_parse_detected_sound(uid: str, msg) -> Event | None:
rule = source.Value rule = source.Value
return Event( return Event(
f"{uid}_{msg.Topic._value_1}_{audio_source}_{audio_analytics}_{rule}", f"{uid}_{value_1}_{audio_source}_{audio_analytics}_{rule}",
"Detected Sound", "Detected Sound",
"binary_sensor", "binary_sensor",
"sound", "sound",
None, None,
msg.Message._value_1.Data.SimpleItem[0].Value == "true", value_1.Data.SimpleItem[0].Value == "true",
) )
except (AttributeError, KeyError): except (AttributeError, KeyError):
return None return None
@PARSERS.register("tns1:RuleEngine/FieldDetector/ObjectsInside") @PARSERS.register("tns1:RuleEngine/FieldDetector/ObjectsInside")
# pylint: disable=protected-access
async def async_parse_field_detector(uid: str, msg) -> Event | None: async def async_parse_field_detector(uid: str, msg) -> Event | None:
"""Handle parsing event message. """Handle parsing event message.
@ -199,7 +198,8 @@ async def async_parse_field_detector(uid: str, msg) -> Event | None:
video_source = "" video_source = ""
video_analytics = "" video_analytics = ""
rule = "" rule = ""
for source in msg.Message._value_1.Source.SimpleItem: value_1 = msg.Message._value_1 # pylint: disable=protected-access
for source in value_1.Source.SimpleItem:
if source.Name == "VideoSourceConfigurationToken": if source.Name == "VideoSourceConfigurationToken":
video_source = _normalize_video_source(source.Value) video_source = _normalize_video_source(source.Value)
if source.Name == "VideoAnalyticsConfigurationToken": if source.Name == "VideoAnalyticsConfigurationToken":
@ -208,12 +208,12 @@ async def async_parse_field_detector(uid: str, msg) -> Event | None:
rule = source.Value rule = source.Value
evt = Event( evt = Event(
f"{uid}_{msg.Topic._value_1}_{video_source}_{video_analytics}_{rule}", f"{uid}_{value_1}_{video_source}_{video_analytics}_{rule}",
"Field Detection", "Field Detection",
"binary_sensor", "binary_sensor",
"motion", "motion",
None, None,
msg.Message._value_1.Data.SimpleItem[0].Value == "true", value_1.Data.SimpleItem[0].Value == "true",
) )
return evt return evt
except (AttributeError, KeyError): except (AttributeError, KeyError):
@ -221,7 +221,6 @@ async def async_parse_field_detector(uid: str, msg) -> Event | None:
@PARSERS.register("tns1:RuleEngine/CellMotionDetector/Motion") @PARSERS.register("tns1:RuleEngine/CellMotionDetector/Motion")
# pylint: disable=protected-access
async def async_parse_cell_motion_detector(uid: str, msg) -> Event | None: async def async_parse_cell_motion_detector(uid: str, msg) -> Event | None:
"""Handle parsing event message. """Handle parsing event message.
@ -231,7 +230,8 @@ async def async_parse_cell_motion_detector(uid: str, msg) -> Event | None:
video_source = "" video_source = ""
video_analytics = "" video_analytics = ""
rule = "" rule = ""
for source in msg.Message._value_1.Source.SimpleItem: value_1 = msg.Message._value_1 # pylint: disable=protected-access
for source in value_1.Source.SimpleItem:
if source.Name == "VideoSourceConfigurationToken": if source.Name == "VideoSourceConfigurationToken":
video_source = _normalize_video_source(source.Value) video_source = _normalize_video_source(source.Value)
if source.Name == "VideoAnalyticsConfigurationToken": if source.Name == "VideoAnalyticsConfigurationToken":
@ -240,19 +240,18 @@ async def async_parse_cell_motion_detector(uid: str, msg) -> Event | None:
rule = source.Value rule = source.Value
return Event( return Event(
f"{uid}_{msg.Topic._value_1}_{video_source}_{video_analytics}_{rule}", f"{uid}_{value_1}_{video_source}_{video_analytics}_{rule}",
"Cell Motion Detection", "Cell Motion Detection",
"binary_sensor", "binary_sensor",
"motion", "motion",
None, None,
msg.Message._value_1.Data.SimpleItem[0].Value == "true", value_1.Data.SimpleItem[0].Value == "true",
) )
except (AttributeError, KeyError): except (AttributeError, KeyError):
return None return None
@PARSERS.register("tns1:RuleEngine/MotionRegionDetector/Motion") @PARSERS.register("tns1:RuleEngine/MotionRegionDetector/Motion")
# pylint: disable=protected-access
async def async_parse_motion_region_detector(uid: str, msg) -> Event | None: async def async_parse_motion_region_detector(uid: str, msg) -> Event | None:
"""Handle parsing event message. """Handle parsing event message.
@ -262,7 +261,8 @@ async def async_parse_motion_region_detector(uid: str, msg) -> Event | None:
video_source = "" video_source = ""
video_analytics = "" video_analytics = ""
rule = "" rule = ""
for source in msg.Message._value_1.Source.SimpleItem: value_1 = msg.Message._value_1 # pylint: disable=protected-access
for source in value_1.Source.SimpleItem:
if source.Name == "VideoSourceConfigurationToken": if source.Name == "VideoSourceConfigurationToken":
video_source = _normalize_video_source(source.Value) video_source = _normalize_video_source(source.Value)
if source.Name == "VideoAnalyticsConfigurationToken": if source.Name == "VideoAnalyticsConfigurationToken":
@ -271,19 +271,18 @@ async def async_parse_motion_region_detector(uid: str, msg) -> Event | None:
rule = source.Value rule = source.Value
return Event( return Event(
f"{uid}_{msg.Topic._value_1}_{video_source}_{video_analytics}_{rule}", f"{uid}_{value_1}_{video_source}_{video_analytics}_{rule}",
"Motion Region Detection", "Motion Region Detection",
"binary_sensor", "binary_sensor",
"motion", "motion",
None, None,
msg.Message._value_1.Data.SimpleItem[0].Value in ["1", "true"], value_1.Data.SimpleItem[0].Value in ["1", "true"],
) )
except (AttributeError, KeyError): except (AttributeError, KeyError):
return None return None
@PARSERS.register("tns1:RuleEngine/TamperDetector/Tamper") @PARSERS.register("tns1:RuleEngine/TamperDetector/Tamper")
# pylint: disable=protected-access
async def async_parse_tamper_detector(uid: str, msg) -> Event | None: async def async_parse_tamper_detector(uid: str, msg) -> Event | None:
"""Handle parsing event message. """Handle parsing event message.
@ -293,7 +292,8 @@ async def async_parse_tamper_detector(uid: str, msg) -> Event | None:
video_source = "" video_source = ""
video_analytics = "" video_analytics = ""
rule = "" rule = ""
for source in msg.Message._value_1.Source.SimpleItem: value_1 = msg.Message._value_1 # pylint: disable=protected-access
for source in value_1.Source.SimpleItem:
if source.Name == "VideoSourceConfigurationToken": if source.Name == "VideoSourceConfigurationToken":
video_source = _normalize_video_source(source.Value) video_source = _normalize_video_source(source.Value)
if source.Name == "VideoAnalyticsConfigurationToken": if source.Name == "VideoAnalyticsConfigurationToken":
@ -302,12 +302,12 @@ async def async_parse_tamper_detector(uid: str, msg) -> Event | None:
rule = source.Value rule = source.Value
return Event( return Event(
f"{uid}_{msg.Topic._value_1}_{video_source}_{video_analytics}_{rule}", f"{uid}_{value_1}_{video_source}_{video_analytics}_{rule}",
"Tamper Detection", "Tamper Detection",
"binary_sensor", "binary_sensor",
"problem", "problem",
None, None,
msg.Message._value_1.Data.SimpleItem[0].Value == "true", value_1.Data.SimpleItem[0].Value == "true",
EntityCategory.DIAGNOSTIC, EntityCategory.DIAGNOSTIC,
) )
except (AttributeError, KeyError): except (AttributeError, KeyError):
@ -315,7 +315,6 @@ async def async_parse_tamper_detector(uid: str, msg) -> Event | None:
@PARSERS.register("tns1:RuleEngine/MyRuleDetector/DogCatDetect") @PARSERS.register("tns1:RuleEngine/MyRuleDetector/DogCatDetect")
# pylint: disable=protected-access
async def async_parse_dog_cat_detector(uid: str, msg) -> Event | None: async def async_parse_dog_cat_detector(uid: str, msg) -> Event | None:
"""Handle parsing event message. """Handle parsing event message.
@ -323,24 +322,24 @@ async def async_parse_dog_cat_detector(uid: str, msg) -> Event | None:
""" """
try: try:
video_source = "" video_source = ""
for source in msg.Message._value_1.Source.SimpleItem: value_1 = msg.Message._value_1 # pylint: disable=protected-access
for source in value_1.Source.SimpleItem:
if source.Name == "Source": if source.Name == "Source":
video_source = _normalize_video_source(source.Value) video_source = _normalize_video_source(source.Value)
return Event( return Event(
f"{uid}_{msg.Topic._value_1}_{video_source}", f"{uid}_{value_1}_{video_source}",
"Pet Detection", "Pet Detection",
"binary_sensor", "binary_sensor",
"motion", "motion",
None, None,
msg.Message._value_1.Data.SimpleItem[0].Value == "true", value_1.Data.SimpleItem[0].Value == "true",
) )
except (AttributeError, KeyError): except (AttributeError, KeyError):
return None return None
@PARSERS.register("tns1:RuleEngine/MyRuleDetector/VehicleDetect") @PARSERS.register("tns1:RuleEngine/MyRuleDetector/VehicleDetect")
# pylint: disable=protected-access
async def async_parse_vehicle_detector(uid: str, msg) -> Event | None: async def async_parse_vehicle_detector(uid: str, msg) -> Event | None:
"""Handle parsing event message. """Handle parsing event message.
@ -348,24 +347,24 @@ async def async_parse_vehicle_detector(uid: str, msg) -> Event | None:
""" """
try: try:
video_source = "" video_source = ""
for source in msg.Message._value_1.Source.SimpleItem: value_1 = msg.Message._value_1 # pylint: disable=protected-access
for source in value_1.Source.SimpleItem:
if source.Name == "Source": if source.Name == "Source":
video_source = _normalize_video_source(source.Value) video_source = _normalize_video_source(source.Value)
return Event( return Event(
f"{uid}_{msg.Topic._value_1}_{video_source}", f"{uid}_{value_1}_{video_source}",
"Vehicle Detection", "Vehicle Detection",
"binary_sensor", "binary_sensor",
"motion", "motion",
None, None,
msg.Message._value_1.Data.SimpleItem[0].Value == "true", value_1.Data.SimpleItem[0].Value == "true",
) )
except (AttributeError, KeyError): except (AttributeError, KeyError):
return None return None
@PARSERS.register("tns1:RuleEngine/MyRuleDetector/PeopleDetect") @PARSERS.register("tns1:RuleEngine/MyRuleDetector/PeopleDetect")
# pylint: disable=protected-access
async def async_parse_person_detector(uid: str, msg) -> Event | None: async def async_parse_person_detector(uid: str, msg) -> Event | None:
"""Handle parsing event message. """Handle parsing event message.
@ -373,24 +372,24 @@ async def async_parse_person_detector(uid: str, msg) -> Event | None:
""" """
try: try:
video_source = "" video_source = ""
for source in msg.Message._value_1.Source.SimpleItem: value_1 = msg.Message._value_1 # pylint: disable=protected-access
for source in value_1.Source.SimpleItem:
if source.Name == "Source": if source.Name == "Source":
video_source = _normalize_video_source(source.Value) video_source = _normalize_video_source(source.Value)
return Event( return Event(
f"{uid}_{msg.Topic._value_1}_{video_source}", f"{uid}_{value_1}_{video_source}",
"Person Detection", "Person Detection",
"binary_sensor", "binary_sensor",
"motion", "motion",
None, None,
msg.Message._value_1.Data.SimpleItem[0].Value == "true", value_1.Data.SimpleItem[0].Value == "true",
) )
except (AttributeError, KeyError): except (AttributeError, KeyError):
return None return None
@PARSERS.register("tns1:RuleEngine/MyRuleDetector/FaceDetect") @PARSERS.register("tns1:RuleEngine/MyRuleDetector/FaceDetect")
# pylint: disable=protected-access
async def async_parse_face_detector(uid: str, msg) -> Event | None: async def async_parse_face_detector(uid: str, msg) -> Event | None:
"""Handle parsing event message. """Handle parsing event message.
@ -398,24 +397,24 @@ async def async_parse_face_detector(uid: str, msg) -> Event | None:
""" """
try: try:
video_source = "" video_source = ""
for source in msg.Message._value_1.Source.SimpleItem: value_1 = msg.Message._value_1 # pylint: disable=protected-access
for source in value_1.Source.SimpleItem:
if source.Name == "Source": if source.Name == "Source":
video_source = _normalize_video_source(source.Value) video_source = _normalize_video_source(source.Value)
return Event( return Event(
f"{uid}_{msg.Topic._value_1}_{video_source}", f"{uid}_{value_1}_{video_source}",
"Face Detection", "Face Detection",
"binary_sensor", "binary_sensor",
"motion", "motion",
None, None,
msg.Message._value_1.Data.SimpleItem[0].Value == "true", value_1.Data.SimpleItem[0].Value == "true",
) )
except (AttributeError, KeyError): except (AttributeError, KeyError):
return None return None
@PARSERS.register("tns1:RuleEngine/MyRuleDetector/Visitor") @PARSERS.register("tns1:RuleEngine/MyRuleDetector/Visitor")
# pylint: disable=protected-access
async def async_parse_visitor_detector(uid: str, msg) -> Event | None: async def async_parse_visitor_detector(uid: str, msg) -> Event | None:
"""Handle parsing event message. """Handle parsing event message.
@ -423,80 +422,81 @@ async def async_parse_visitor_detector(uid: str, msg) -> Event | None:
""" """
try: try:
video_source = "" video_source = ""
for source in msg.Message._value_1.Source.SimpleItem: value_1 = msg.Message._value_1 # pylint: disable=protected-access
for source in value_1.Source.SimpleItem:
if source.Name == "Source": if source.Name == "Source":
video_source = _normalize_video_source(source.Value) video_source = _normalize_video_source(source.Value)
return Event( return Event(
f"{uid}_{msg.Topic._value_1}_{video_source}", f"{uid}_{value_1}_{video_source}",
"Visitor Detection", "Visitor Detection",
"binary_sensor", "binary_sensor",
"occupancy", "occupancy",
None, None,
msg.Message._value_1.Data.SimpleItem[0].Value == "true", value_1.Data.SimpleItem[0].Value == "true",
) )
except (AttributeError, KeyError): except (AttributeError, KeyError):
return None return None
@PARSERS.register("tns1:Device/Trigger/DigitalInput") @PARSERS.register("tns1:Device/Trigger/DigitalInput")
# pylint: disable=protected-access
async def async_parse_digital_input(uid: str, msg) -> Event | None: async def async_parse_digital_input(uid: str, msg) -> Event | None:
"""Handle parsing event message. """Handle parsing event message.
Topic: tns1:Device/Trigger/DigitalInput Topic: tns1:Device/Trigger/DigitalInput
""" """
try: try:
source = msg.Message._value_1.Source.SimpleItem[0].Value value_1 = msg.Message._value_1 # pylint: disable=protected-access
source = value_1.Source.SimpleItem[0].Value
return Event( return Event(
f"{uid}_{msg.Topic._value_1}_{source}", f"{uid}_{value_1}_{source}",
"Digital Input", "Digital Input",
"binary_sensor", "binary_sensor",
None, None,
None, None,
msg.Message._value_1.Data.SimpleItem[0].Value == "true", value_1.Data.SimpleItem[0].Value == "true",
) )
except (AttributeError, KeyError): except (AttributeError, KeyError):
return None return None
@PARSERS.register("tns1:Device/Trigger/Relay") @PARSERS.register("tns1:Device/Trigger/Relay")
# pylint: disable=protected-access
async def async_parse_relay(uid: str, msg) -> Event | None: async def async_parse_relay(uid: str, msg) -> Event | None:
"""Handle parsing event message. """Handle parsing event message.
Topic: tns1:Device/Trigger/Relay Topic: tns1:Device/Trigger/Relay
""" """
try: try:
source = msg.Message._value_1.Source.SimpleItem[0].Value value_1 = msg.Message._value_1 # pylint: disable=protected-access
source = value_1.Source.SimpleItem[0].Value
return Event( return Event(
f"{uid}_{msg.Topic._value_1}_{source}", f"{uid}_{value_1}_{source}",
"Relay Triggered", "Relay Triggered",
"binary_sensor", "binary_sensor",
None, None,
None, None,
msg.Message._value_1.Data.SimpleItem[0].Value == "active", value_1.Data.SimpleItem[0].Value == "active",
) )
except (AttributeError, KeyError): except (AttributeError, KeyError):
return None return None
@PARSERS.register("tns1:Device/HardwareFailure/StorageFailure") @PARSERS.register("tns1:Device/HardwareFailure/StorageFailure")
# pylint: disable=protected-access
async def async_parse_storage_failure(uid: str, msg) -> Event | None: async def async_parse_storage_failure(uid: str, msg) -> Event | None:
"""Handle parsing event message. """Handle parsing event message.
Topic: tns1:Device/HardwareFailure/StorageFailure Topic: tns1:Device/HardwareFailure/StorageFailure
""" """
try: try:
source = msg.Message._value_1.Source.SimpleItem[0].Value value_1 = msg.Message._value_1 # pylint: disable=protected-access
source = value_1.Source.SimpleItem[0].Value
return Event( return Event(
f"{uid}_{msg.Topic._value_1}_{source}", f"{uid}_{value_1}_{source}",
"Storage Failure", "Storage Failure",
"binary_sensor", "binary_sensor",
"problem", "problem",
None, None,
msg.Message._value_1.Data.SimpleItem[0].Value == "true", value_1.Data.SimpleItem[0].Value == "true",
EntityCategory.DIAGNOSTIC, EntityCategory.DIAGNOSTIC,
) )
except (AttributeError, KeyError): except (AttributeError, KeyError):
@ -504,19 +504,19 @@ async def async_parse_storage_failure(uid: str, msg) -> Event | None:
@PARSERS.register("tns1:Monitoring/ProcessorUsage") @PARSERS.register("tns1:Monitoring/ProcessorUsage")
# pylint: disable=protected-access
async def async_parse_processor_usage(uid: str, msg) -> Event | None: async def async_parse_processor_usage(uid: str, msg) -> Event | None:
"""Handle parsing event message. """Handle parsing event message.
Topic: tns1:Monitoring/ProcessorUsage Topic: tns1:Monitoring/ProcessorUsage
""" """
try: try:
usage = float(msg.Message._value_1.Data.SimpleItem[0].Value) value_1 = msg.Message._value_1 # pylint: disable=protected-access
usage = float(value_1.Data.SimpleItem[0].Value)
if usage <= 1: if usage <= 1:
usage *= 100 usage *= 100
return Event( return Event(
f"{uid}_{msg.Topic._value_1}", f"{uid}_{value_1}",
"Processor Usage", "Processor Usage",
"sensor", "sensor",
None, None,
@ -529,18 +529,16 @@ async def async_parse_processor_usage(uid: str, msg) -> Event | None:
@PARSERS.register("tns1:Monitoring/OperatingTime/LastReboot") @PARSERS.register("tns1:Monitoring/OperatingTime/LastReboot")
# pylint: disable=protected-access
async def async_parse_last_reboot(uid: str, msg) -> Event | None: async def async_parse_last_reboot(uid: str, msg) -> Event | None:
"""Handle parsing event message. """Handle parsing event message.
Topic: tns1:Monitoring/OperatingTime/LastReboot Topic: tns1:Monitoring/OperatingTime/LastReboot
""" """
try: try:
date_time = local_datetime_or_none( value_1 = msg.Message._value_1 # pylint: disable=protected-access
msg.Message._value_1.Data.SimpleItem[0].Value date_time = local_datetime_or_none(value_1.Data.SimpleItem[0].Value)
)
return Event( return Event(
f"{uid}_{msg.Topic._value_1}", f"{uid}_{value_1}",
"Last Reboot", "Last Reboot",
"sensor", "sensor",
"timestamp", "timestamp",
@ -553,18 +551,16 @@ async def async_parse_last_reboot(uid: str, msg) -> Event | None:
@PARSERS.register("tns1:Monitoring/OperatingTime/LastReset") @PARSERS.register("tns1:Monitoring/OperatingTime/LastReset")
# pylint: disable=protected-access
async def async_parse_last_reset(uid: str, msg) -> Event | None: async def async_parse_last_reset(uid: str, msg) -> Event | None:
"""Handle parsing event message. """Handle parsing event message.
Topic: tns1:Monitoring/OperatingTime/LastReset Topic: tns1:Monitoring/OperatingTime/LastReset
""" """
try: try:
date_time = local_datetime_or_none( value_1 = msg.Message._value_1 # pylint: disable=protected-access
msg.Message._value_1.Data.SimpleItem[0].Value date_time = local_datetime_or_none(value_1.Data.SimpleItem[0].Value)
)
return Event( return Event(
f"{uid}_{msg.Topic._value_1}", f"{uid}_{value_1}",
"Last Reset", "Last Reset",
"sensor", "sensor",
"timestamp", "timestamp",
@ -578,7 +574,6 @@ async def async_parse_last_reset(uid: str, msg) -> Event | None:
@PARSERS.register("tns1:Monitoring/Backup/Last") @PARSERS.register("tns1:Monitoring/Backup/Last")
# pylint: disable=protected-access
async def async_parse_backup_last(uid: str, msg) -> Event | None: async def async_parse_backup_last(uid: str, msg) -> Event | None:
"""Handle parsing event message. """Handle parsing event message.
@ -586,11 +581,10 @@ async def async_parse_backup_last(uid: str, msg) -> Event | None:
""" """
try: try:
date_time = local_datetime_or_none( value_1 = msg.Message._value_1 # pylint: disable=protected-access
msg.Message._value_1.Data.SimpleItem[0].Value date_time = local_datetime_or_none(value_1.Data.SimpleItem[0].Value)
)
return Event( return Event(
f"{uid}_{msg.Topic._value_1}", f"{uid}_{value_1}",
"Last Backup", "Last Backup",
"sensor", "sensor",
"timestamp", "timestamp",
@ -604,18 +598,16 @@ async def async_parse_backup_last(uid: str, msg) -> Event | None:
@PARSERS.register("tns1:Monitoring/OperatingTime/LastClockSynchronization") @PARSERS.register("tns1:Monitoring/OperatingTime/LastClockSynchronization")
# pylint: disable=protected-access
async def async_parse_last_clock_sync(uid: str, msg) -> Event | None: async def async_parse_last_clock_sync(uid: str, msg) -> Event | None:
"""Handle parsing event message. """Handle parsing event message.
Topic: tns1:Monitoring/OperatingTime/LastClockSynchronization Topic: tns1:Monitoring/OperatingTime/LastClockSynchronization
""" """
try: try:
date_time = local_datetime_or_none( value_1 = msg.Message._value_1 # pylint: disable=protected-access
msg.Message._value_1.Data.SimpleItem[0].Value date_time = local_datetime_or_none(value_1.Data.SimpleItem[0].Value)
)
return Event( return Event(
f"{uid}_{msg.Topic._value_1}", f"{uid}_{value_1}",
"Last Clock Synchronization", "Last Clock Synchronization",
"sensor", "sensor",
"timestamp", "timestamp",
@ -629,7 +621,6 @@ async def async_parse_last_clock_sync(uid: str, msg) -> Event | None:
@PARSERS.register("tns1:RecordingConfig/JobState") @PARSERS.register("tns1:RecordingConfig/JobState")
# pylint: disable=protected-access
async def async_parse_jobstate(uid: str, msg) -> Event | None: async def async_parse_jobstate(uid: str, msg) -> Event | None:
"""Handle parsing event message. """Handle parsing event message.
@ -637,14 +628,15 @@ async def async_parse_jobstate(uid: str, msg) -> Event | None:
""" """
try: try:
source = msg.Message._value_1.Source.SimpleItem[0].Value value_1 = msg.Message._value_1 # pylint: disable=protected-access
source = value_1.Source.SimpleItem[0].Value
return Event( return Event(
f"{uid}_{msg.Topic._value_1}_{source}", f"{uid}_{value_1}_{source}",
"Recording Job State", "Recording Job State",
"binary_sensor", "binary_sensor",
None, None,
None, None,
msg.Message._value_1.Data.SimpleItem[0].Value == "Active", value_1.Data.SimpleItem[0].Value == "Active",
EntityCategory.DIAGNOSTIC, EntityCategory.DIAGNOSTIC,
) )
except (AttributeError, KeyError): except (AttributeError, KeyError):
@ -652,7 +644,6 @@ async def async_parse_jobstate(uid: str, msg) -> Event | None:
@PARSERS.register("tns1:RuleEngine/LineDetector/Crossed") @PARSERS.register("tns1:RuleEngine/LineDetector/Crossed")
# pylint: disable=protected-access
async def async_parse_linedetector_crossed(uid: str, msg) -> Event | None: async def async_parse_linedetector_crossed(uid: str, msg) -> Event | None:
"""Handle parsing event message. """Handle parsing event message.
@ -662,7 +653,8 @@ async def async_parse_linedetector_crossed(uid: str, msg) -> Event | None:
video_source = "" video_source = ""
video_analytics = "" video_analytics = ""
rule = "" rule = ""
for source in msg.Message._value_1.Source.SimpleItem: value_1 = msg.Message._value_1 # pylint: disable=protected-access
for source in value_1.Source.SimpleItem:
if source.Name == "VideoSourceConfigurationToken": if source.Name == "VideoSourceConfigurationToken":
video_source = source.Value video_source = source.Value
if source.Name == "VideoAnalyticsConfigurationToken": if source.Name == "VideoAnalyticsConfigurationToken":
@ -671,12 +663,12 @@ async def async_parse_linedetector_crossed(uid: str, msg) -> Event | None:
rule = source.Value rule = source.Value
return Event( return Event(
f"{uid}_{msg.Topic._value_1}_{video_source}_{video_analytics}_{rule}", f"{uid}_{value_1}_{video_source}_{video_analytics}_{rule}",
"Line Detector Crossed", "Line Detector Crossed",
"sensor", "sensor",
None, None,
None, None,
msg.Message._value_1.Data.SimpleItem[0].Value, value_1.Data.SimpleItem[0].Value,
EntityCategory.DIAGNOSTIC, EntityCategory.DIAGNOSTIC,
) )
except (AttributeError, KeyError): except (AttributeError, KeyError):
@ -684,7 +676,6 @@ async def async_parse_linedetector_crossed(uid: str, msg) -> Event | None:
@PARSERS.register("tns1:RuleEngine/CountAggregation/Counter") @PARSERS.register("tns1:RuleEngine/CountAggregation/Counter")
# pylint: disable=protected-access
async def async_parse_count_aggregation_counter(uid: str, msg) -> Event | None: async def async_parse_count_aggregation_counter(uid: str, msg) -> Event | None:
"""Handle parsing event message. """Handle parsing event message.
@ -694,7 +685,8 @@ async def async_parse_count_aggregation_counter(uid: str, msg) -> Event | None:
video_source = "" video_source = ""
video_analytics = "" video_analytics = ""
rule = "" rule = ""
for source in msg.Message._value_1.Source.SimpleItem: value_1 = msg.Message._value_1 # pylint: disable=protected-access
for source in value_1.Source.SimpleItem:
if source.Name == "VideoSourceConfigurationToken": if source.Name == "VideoSourceConfigurationToken":
video_source = _normalize_video_source(source.Value) video_source = _normalize_video_source(source.Value)
if source.Name == "VideoAnalyticsConfigurationToken": if source.Name == "VideoAnalyticsConfigurationToken":
@ -703,12 +695,12 @@ async def async_parse_count_aggregation_counter(uid: str, msg) -> Event | None:
rule = source.Value rule = source.Value
return Event( return Event(
f"{uid}_{msg.Topic._value_1}_{video_source}_{video_analytics}_{rule}", f"{uid}_{value_1}_{video_source}_{video_analytics}_{rule}",
"Count Aggregation Counter", "Count Aggregation Counter",
"sensor", "sensor",
None, None,
None, None,
msg.Message._value_1.Data.SimpleItem[0].Value, value_1.Data.SimpleItem[0].Value,
EntityCategory.DIAGNOSTIC, EntityCategory.DIAGNOSTIC,
) )
except (AttributeError, KeyError): except (AttributeError, KeyError):