From a9e857202de36bc70e906fe5aae6afae74d55905 Mon Sep 17 00:00:00 2001 From: Robert Svensson Date: Fri, 22 Mar 2024 15:49:51 +0100 Subject: [PATCH] Axis use entity descripton binary sensor platform (#113705) --- .../components/axis/binary_sensor.py | 260 ++++++++++++------ tests/components/axis/test_binary_sensor.py | 109 ++++++++ 2 files changed, 292 insertions(+), 77 deletions(-) diff --git a/homeassistant/components/axis/binary_sensor.py b/homeassistant/components/axis/binary_sensor.py index 4e724c804bd..e68487a6bb1 100644 --- a/homeassistant/components/axis/binary_sensor.py +++ b/homeassistant/components/axis/binary_sensor.py @@ -2,10 +2,12 @@ from __future__ import annotations -from collections.abc import Callable +from collections.abc import Callable, Iterable +from dataclasses import dataclass from datetime import datetime, timedelta +from functools import partial -from axis.models.event import Event, EventGroup, EventOperation, EventTopic +from axis.models.event import Event, EventOperation, EventTopic from axis.vapix.interfaces.applications.fence_guard import FenceGuardHandler from axis.vapix.interfaces.applications.loitering_guard import LoiteringGuardHandler from axis.vapix.interfaces.applications.motion_guard import MotionGuardHandler @@ -14,6 +16,7 @@ from axis.vapix.interfaces.applications.vmd4 import Vmd4Handler from homeassistant.components.binary_sensor import ( BinarySensorDeviceClass, BinarySensorEntity, + BinarySensorEntityDescription, ) from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant, callback @@ -23,26 +26,160 @@ from homeassistant.helpers.event import async_call_later from .entity import AxisEventEntity from .hub import AxisHub -DEVICE_CLASS = { - EventGroup.INPUT: BinarySensorDeviceClass.CONNECTIVITY, - EventGroup.LIGHT: BinarySensorDeviceClass.LIGHT, - EventGroup.MOTION: BinarySensorDeviceClass.MOTION, - EventGroup.SOUND: BinarySensorDeviceClass.SOUND, -} -EVENT_TOPICS = ( - EventTopic.DAY_NIGHT_VISION, - EventTopic.FENCE_GUARD, - EventTopic.LOITERING_GUARD, - EventTopic.MOTION_DETECTION, - EventTopic.MOTION_DETECTION_3, - EventTopic.MOTION_DETECTION_4, - EventTopic.MOTION_GUARD, - EventTopic.OBJECT_ANALYTICS, - EventTopic.PIR, - EventTopic.PORT_INPUT, - EventTopic.PORT_SUPERVISED_INPUT, - EventTopic.SOUND_TRIGGER_LEVEL, +@dataclass(frozen=True, kw_only=True) +class AxisBinarySensorDescription(BinarySensorEntityDescription): + """Axis binary sensor entity description.""" + + event_topic: tuple[EventTopic, ...] | EventTopic + """Event topic that provides state updates.""" + name_fn: Callable[[AxisHub, Event], str] = lambda hub, event: "" + """Function providing the corresponding name to the event ID.""" + supported_fn: Callable[[AxisHub, Event], bool] = lambda hub, event: True + """Function validating if event is supported.""" + + +@callback +def event_id_is_int(event_id: str) -> bool: + """Make sure event ID is int.""" + try: + _ = int(event_id) + except ValueError: + return False + return True + + +@callback +def guard_suite_supported_fn(hub: AxisHub, event: Event) -> bool: + """Validate event ID is int.""" + _, _, profile_id = event.id.partition("Profile") + return event_id_is_int(profile_id) + + +@callback +def object_analytics_supported_fn(hub: AxisHub, event: Event) -> bool: + """Validate event ID is int.""" + _, _, profile_id = event.id.partition("Scenario") + return event_id_is_int(profile_id) + + +@callback +def guard_suite_name_fn( + handler: FenceGuardHandler + | LoiteringGuardHandler + | MotionGuardHandler + | Vmd4Handler, + event: Event, + event_type: str, +) -> str: + """Get guard suite item name.""" + if handler.initialized and (profiles := handler["0"].profiles): + for profile_id, profile in profiles.items(): + camera_id = profile.camera + if event.id == f"Camera{camera_id}Profile{profile_id}": + return f"{event_type} {profile.name}" + return "" + + +@callback +def fence_guard_name_fn(hub: AxisHub, event: Event) -> str: + """Fence guard name.""" + return guard_suite_name_fn(hub.api.vapix.fence_guard, event, "Fence Guard") + + +@callback +def loitering_guard_name_fn(hub: AxisHub, event: Event) -> str: + """Loitering guard name.""" + return guard_suite_name_fn(hub.api.vapix.loitering_guard, event, "Loitering Guard") + + +@callback +def motion_guard_name_fn(hub: AxisHub, event: Event) -> str: + """Motion guard name.""" + return guard_suite_name_fn(hub.api.vapix.motion_guard, event, "Motion Guard") + + +@callback +def motion_detection_4_name_fn(hub: AxisHub, event: Event) -> str: + """Motion detection 4 name.""" + return guard_suite_name_fn(hub.api.vapix.vmd4, event, "VMD4") + + +@callback +def object_analytics_name_fn(hub: AxisHub, event: Event) -> str: + """Get object analytics name.""" + if hub.api.vapix.object_analytics.initialized and ( + scenarios := hub.api.vapix.object_analytics["0"].scenarios + ): + for scenario_id, scenario in scenarios.items(): + device_id = scenario.devices[0]["id"] + if event.id == f"Device{device_id}Scenario{scenario_id}": + return f"Object Analytics {scenario.name}" + return "" + + +ENTITY_DESCRIPTIONS = ( + AxisBinarySensorDescription( + key="Input port state", + device_class=BinarySensorDeviceClass.CONNECTIVITY, + event_topic=(EventTopic.PORT_INPUT, EventTopic.PORT_SUPERVISED_INPUT), + name_fn=lambda hub, event: hub.api.vapix.ports[event.id].name, + supported_fn=lambda hub, event: event_id_is_int(event.id), + ), + AxisBinarySensorDescription( + key="Day/Night vision state", + device_class=BinarySensorDeviceClass.LIGHT, + event_topic=EventTopic.DAY_NIGHT_VISION, + ), + AxisBinarySensorDescription( + key="Sound trigger state", + device_class=BinarySensorDeviceClass.SOUND, + event_topic=EventTopic.SOUND_TRIGGER_LEVEL, + ), + AxisBinarySensorDescription( + key="Motion sensors state", + device_class=BinarySensorDeviceClass.MOTION, + event_topic=( + EventTopic.PIR, + EventTopic.MOTION_DETECTION, + EventTopic.MOTION_DETECTION_3, + ), + ), + AxisBinarySensorDescription( + key="Motion detection 4 state", + device_class=BinarySensorDeviceClass.MOTION, + event_topic=EventTopic.MOTION_DETECTION_4, + name_fn=motion_detection_4_name_fn, + supported_fn=guard_suite_supported_fn, + ), + AxisBinarySensorDescription( + key="Fence guard state", + device_class=BinarySensorDeviceClass.MOTION, + event_topic=EventTopic.FENCE_GUARD, + name_fn=fence_guard_name_fn, + supported_fn=guard_suite_supported_fn, + ), + AxisBinarySensorDescription( + key="Loitering guard state", + device_class=BinarySensorDeviceClass.MOTION, + event_topic=EventTopic.LOITERING_GUARD, + name_fn=loitering_guard_name_fn, + supported_fn=guard_suite_supported_fn, + ), + AxisBinarySensorDescription( + key="Motion guard state", + device_class=BinarySensorDeviceClass.MOTION, + event_topic=EventTopic.MOTION_GUARD, + name_fn=motion_guard_name_fn, + supported_fn=guard_suite_supported_fn, + ), + AxisBinarySensorDescription( + key="Object analytics state", + device_class=BinarySensorDeviceClass.MOTION, + event_topic=EventTopic.OBJECT_ANALYTICS, + name_fn=object_analytics_name_fn, + supported_fn=object_analytics_supported_fn, + ), ) @@ -55,29 +192,40 @@ async def async_setup_entry( hub = AxisHub.get_hub(hass, config_entry) @callback - def async_create_entity(event: Event) -> None: - """Create Axis binary sensor entity.""" - async_add_entities([AxisBinarySensor(event, hub)]) + def register_platform(descriptions: Iterable[AxisBinarySensorDescription]) -> None: + """Register entity platform to create entities on event initialized signal.""" - hub.api.event.subscribe( - async_create_entity, - topic_filter=EVENT_TOPICS, - operation_filter=EventOperation.INITIALIZED, - ) + @callback + def create_entity( + description: AxisBinarySensorDescription, event: Event + ) -> None: + """Create Axis entity.""" + if description.supported_fn(hub, event): + async_add_entities([AxisBinarySensor(hub, description, event)]) + + for description in descriptions: + hub.api.event.subscribe( + partial(create_entity, description), + topic_filter=description.event_topic, + operation_filter=EventOperation.INITIALIZED, + ) + + register_platform(ENTITY_DESCRIPTIONS) class AxisBinarySensor(AxisEventEntity, BinarySensorEntity): """Representation of a binary Axis event.""" - def __init__(self, event: Event, hub: AxisHub) -> None: + def __init__( + self, hub: AxisHub, description: AxisBinarySensorDescription, event: Event + ) -> None: """Initialize the Axis binary sensor.""" super().__init__(event, hub) - self.cancel_scheduled_update: Callable[[], None] | None = None - - self._attr_device_class = DEVICE_CLASS.get(event.group) + self.entity_description = description + self._attr_name = description.name_fn(hub, event) or self._attr_name self._attr_is_on = event.is_tripped - - self._set_name(event) + self._attr_device_class = description.device_class # temporary + self.cancel_scheduled_update: Callable[[], None] | None = None @callback def async_event_callback(self, event: Event) -> None: @@ -103,45 +251,3 @@ class AxisBinarySensor(AxisEventEntity, BinarySensorEntity): timedelta(seconds=self.hub.config.trigger_time), scheduled_update, ) - - @callback - def _set_name(self, event: Event) -> None: - """Set binary sensor name.""" - if ( - event.group == EventGroup.INPUT - and event.id in self.hub.api.vapix.ports - and self.hub.api.vapix.ports[event.id].name - ): - self._attr_name = self.hub.api.vapix.ports[event.id].name - - elif event.group == EventGroup.MOTION: - event_data: FenceGuardHandler | LoiteringGuardHandler | MotionGuardHandler | Vmd4Handler | None = None - if event.topic_base == EventTopic.FENCE_GUARD: - event_data = self.hub.api.vapix.fence_guard - elif event.topic_base == EventTopic.LOITERING_GUARD: - event_data = self.hub.api.vapix.loitering_guard - elif event.topic_base == EventTopic.MOTION_GUARD: - event_data = self.hub.api.vapix.motion_guard - elif event.topic_base == EventTopic.MOTION_DETECTION_4: - event_data = self.hub.api.vapix.vmd4 - if ( - event_data - and event_data.initialized - and (profiles := event_data["0"].profiles) - ): - for profile_id, profile in profiles.items(): - camera_id = profile.camera - if event.id == f"Camera{camera_id}Profile{profile_id}": - self._attr_name = f"{self._event_type} {profile.name}" - return - - if ( - event.topic_base == EventTopic.OBJECT_ANALYTICS - and self.hub.api.vapix.object_analytics.initialized - and (scenarios := self.hub.api.vapix.object_analytics["0"].scenarios) - ): - for scenario_id, scenario in scenarios.items(): - device_id = scenario.devices[0]["id"] - if event.id == f"Device{device_id}Scenario{scenario_id}": - self._attr_name = f"{self._event_type} {scenario.name}" - break diff --git a/tests/components/axis/test_binary_sensor.py b/tests/components/axis/test_binary_sensor.py index 64e567889f1..ce387655889 100644 --- a/tests/components/axis/test_binary_sensor.py +++ b/tests/components/axis/test_binary_sensor.py @@ -52,6 +52,52 @@ async def test_unsupported_binary_sensors( @pytest.mark.parametrize( ("event", "entity"), [ + ( + { + "topic": "tns1:VideoSource/tnsaxis:DayNightVision", + "source_name": "VideoSourceConfigurationToken", + "source_idx": "1", + "data_type": "DayNight", + "data_value": "1", + }, + { + "id": f"{BINARY_SENSOR_DOMAIN}.{NAME}_daynight_1", + "state": STATE_ON, + "name": f"{NAME} DayNight 1", + "device_class": BinarySensorDeviceClass.LIGHT, + }, + ), + ( + { + "topic": "tns1:AudioSource/tnsaxis:TriggerLevel", + "source_name": "channel", + "source_idx": "1", + "data_type": "Sound", + "data_value": "0", + }, + { + "id": f"{BINARY_SENSOR_DOMAIN}.{NAME}_sound_1", + "state": STATE_OFF, + "name": f"{NAME} Sound 1", + "device_class": BinarySensorDeviceClass.SOUND, + }, + ), + ( + { + "topic": "tns1:Device/tnsaxis:IO/Port", + "data_type": "state", + "data_value": "0", + "operation": "Initialized", + "source_name": "port", + "source_idx": "0", + }, + { + "id": f"{BINARY_SENSOR_DOMAIN}.{NAME}_pir_sensor", + "state": STATE_OFF, + "name": f"{NAME} PIR sensor", + "device_class": BinarySensorDeviceClass.CONNECTIVITY, + }, + ), ( { "topic": "tns1:Device/tnsaxis:Sensor/PIR", @@ -147,3 +193,66 @@ async def test_binary_sensors( assert state.state == entity["state"] assert state.name == entity["name"] assert state.attributes["device_class"] == entity["device_class"] + + +@pytest.mark.parametrize( + ("event"), + [ + { + "topic": "tns1:Device/tnsaxis:IO/Port", + "data_type": "state", + "data_value": "0", + "operation": "Initialized", + "source_name": "port", + "source_idx": "-1", + }, + { + "topic": "tnsaxis:CameraApplicationPlatform/VMD/Camera1ProfileANY", + "data_type": "active", + "data_value": "1", + }, + { + "topic": "tnsaxis:CameraApplicationPlatform/ObjectAnalytics/Device1ScenarioANY", + "data_type": "active", + "data_value": "1", + }, + ], +) +async def test_unsupported_events( + hass: HomeAssistant, setup_config_entry, mock_rtsp_event, event +) -> None: + """Validate nothing breaks with unsupported events.""" + mock_rtsp_event(**event) + await hass.async_block_till_done() + assert len(hass.states.async_entity_ids(BINARY_SENSOR_DOMAIN)) == 0 + + +@pytest.mark.parametrize( + ("event", "entity_id"), + [ + ( + { + "topic": "tnsaxis:CameraApplicationPlatform/VMD/Camera1Profile9", + "data_type": "active", + "data_value": "1", + }, + "binary_sensor.name_vmd4_camera1profile9", + ), + ( + { + "topic": "tnsaxis:CameraApplicationPlatform/ObjectAnalytics/Device1Scenario8", + "data_type": "active", + "data_value": "1", + }, + "binary_sensor.name_object_analytics_device1scenario8", + ), + ], +) +async def test_no_primary_name_for_event( + hass: HomeAssistant, setup_config_entry, mock_rtsp_event, event, entity_id +) -> None: + """Validate fallback method for getting name works.""" + mock_rtsp_event(**event) + await hass.async_block_till_done() + assert len(hass.states.async_entity_ids(BINARY_SENSOR_DOMAIN)) == 1 + assert hass.states.get(entity_id)