Axis use entity descripton binary sensor platform (#113705)

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 = (
+@dataclass(frozen=True, kw_only=True)
-    EventTopic.DAY_NIGHT_VISION,
+class AxisBinarySensorDescription(BinarySensorEntityDescription):
-    EventTopic.FENCE_GUARD,
+    """Axis binary sensor entity description."""
-    EventTopic.LOITERING_GUARD,
+
-    EventTopic.MOTION_DETECTION,
+    event_topic: tuple[EventTopic, ...] | EventTopic
-    EventTopic.MOTION_DETECTION_3,
+    """Event topic that provides state updates."""
-    EventTopic.MOTION_DETECTION_4,
+    name_fn: Callable[[AxisHub, Event], str] = lambda hub, event: ""
-    EventTopic.MOTION_GUARD,
+    """Function providing the corresponding name to the event ID."""
-    EventTopic.OBJECT_ANALYTICS,
+    supported_fn: Callable[[AxisHub, Event], bool] = lambda hub, event: True
-    EventTopic.PIR,
+    """Function validating if event is supported."""
-    EventTopic.PORT_INPUT,
+
-    EventTopic.PORT_SUPERVISED_INPUT,
+
-    EventTopic.SOUND_TRIGGER_LEVEL,
+@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:
+    def register_platform(descriptions: Iterable[AxisBinarySensorDescription]) -> None:
-        """Create Axis binary sensor entity."""
+        """Register entity platform to create entities on event initialized signal."""
-        async_add_entities([AxisBinarySensor(event, hub)])
 
-    hub.api.event.subscribe(
+        @callback
-        async_create_entity,
+        def create_entity(
-        topic_filter=EVENT_TOPICS,
+            description: AxisBinarySensorDescription, event: Event
-        operation_filter=EventOperation.INITIALIZED,
+        ) -> 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.entity_description = description
-
+        self._attr_name = description.name_fn(hub, event) or self._attr_name
-        self._attr_device_class = DEVICE_CLASS.get(event.group)
         self._attr_is_on = event.is_tripped
-
+        self._attr_device_class = description.device_class  # temporary
-        self._set_name(event)
+        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

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)