Add binary sensor platform to sunricher_dali (#161463)

Co-authored-by: Joostlek <joostlek@outlook.com>

homeassistant/components/sunricher_dali/__init__.py

@@ -26,6 +26,7 @@ from .const import CONF_SERIAL_NUMBER, DOMAIN, MANUFACTURER
 from .types import DaliCenterConfigEntry, DaliCenterData
 
 _PLATFORMS: list[Platform] = [
+    Platform.BINARY_SENSOR,
     Platform.BUTTON,
     Platform.LIGHT,
     Platform.SCENE,

homeassistant/components/sunricher_dali/binary_sensor.py

@@ -0,0 +1,131 @@
+"""Platform for Sunricher DALI binary sensor entities."""
+
+from __future__ import annotations
+
+from PySrDaliGateway import CallbackEventType, Device
+from PySrDaliGateway.helper import is_motion_sensor
+from PySrDaliGateway.types import MotionState, MotionStatus
+
+from homeassistant.components.binary_sensor import (
+    BinarySensorDeviceClass,
+    BinarySensorEntity,
+)
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.device_registry import DeviceInfo
+from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
+
+from .const import DOMAIN, MANUFACTURER
+from .entity import DaliDeviceEntity
+from .types import DaliCenterConfigEntry
+
+_OCCUPANCY_ON_STATES = frozenset(
+    {
+        MotionState.MOTION,
+        MotionState.OCCUPANCY,
+        MotionState.PRESENCE,
+    }
+)
+
+PARALLEL_UPDATES = 0
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: DaliCenterConfigEntry,
+    async_add_entities: AddConfigEntryEntitiesCallback,
+) -> None:
+    """Set up Sunricher DALI binary sensor entities from config entry."""
+    devices = entry.runtime_data.devices
+
+    entities: list[BinarySensorEntity] = []
+    for device in devices:
+        if is_motion_sensor(device.dev_type):
+            entities.append(SunricherDaliMotionSensor(device))
+            entities.append(SunricherDaliOccupancySensor(device))
+
+    if entities:
+        async_add_entities(entities)
+
+
+class SunricherDaliMotionSensor(DaliDeviceEntity, BinarySensorEntity):
+    """Instantaneous motion detection sensor."""
+
+    _attr_device_class = BinarySensorDeviceClass.MOTION
+
+    def __init__(self, device: Device) -> None:
+        """Initialize the motion sensor."""
+        super().__init__(device)
+        self._device = device
+        self._attr_unique_id = f"{device.dev_id}_motion"
+        self._attr_device_info = DeviceInfo(
+            identifiers={(DOMAIN, device.dev_id)},
+            name=device.name,
+            manufacturer=MANUFACTURER,
+            model=device.model,
+            via_device=(DOMAIN, device.gw_sn),
+        )
+
+    async def async_added_to_hass(self) -> None:
+        """Handle entity addition to Home Assistant."""
+        await super().async_added_to_hass()
+
+        self.async_on_remove(
+            self._device.register_listener(
+                CallbackEventType.MOTION_STATUS, self._handle_motion_status
+            )
+        )
+
+        self._device.read_status()
+
+    @callback
+    def _handle_motion_status(self, status: MotionStatus) -> None:
+        """Handle motion status updates."""
+        motion_state = status["motion_state"]
+        if motion_state == MotionState.MOTION:
+            self._attr_is_on = True
+            self.schedule_update_ha_state()
+        elif motion_state == MotionState.NO_MOTION:
+            self._attr_is_on = False
+            self.schedule_update_ha_state()
+
+
+class SunricherDaliOccupancySensor(DaliDeviceEntity, BinarySensorEntity):
+    """Persistent occupancy detection sensor."""
+
+    _attr_device_class = BinarySensorDeviceClass.OCCUPANCY
+
+    def __init__(self, device: Device) -> None:
+        """Initialize the occupancy sensor."""
+        super().__init__(device)
+        self._device = device
+        self._attr_unique_id = f"{device.dev_id}_occupancy"
+        self._attr_device_info = DeviceInfo(
+            identifiers={(DOMAIN, device.dev_id)},
+            name=device.name,
+            manufacturer=MANUFACTURER,
+            model=device.model,
+            via_device=(DOMAIN, device.gw_sn),
+        )
+
+    async def async_added_to_hass(self) -> None:
+        """Handle entity addition to Home Assistant."""
+        await super().async_added_to_hass()
+
+        self.async_on_remove(
+            self._device.register_listener(
+                CallbackEventType.MOTION_STATUS, self._handle_motion_status
+            )
+        )
+
+        self._device.read_status()
+
+    @callback
+    def _handle_motion_status(self, status: MotionStatus) -> None:
+        """Handle motion status updates."""
+        motion_state = status["motion_state"]
+        if motion_state in _OCCUPANCY_ON_STATES:
+            self._attr_is_on = True
+            self.schedule_update_ha_state()
+        elif motion_state == MotionState.VACANT:
+            self._attr_is_on = False
+            self.schedule_update_ha_state()

tests/components/sunricher_dali/__init__.py

@@ -1,6 +1,7 @@
 """Tests for the Sunricher Sunricher DALI integration."""
 
 from collections.abc import Callable
+from typing import Any
 from unittest.mock import MagicMock
 
 from PySrDaliGateway import CallbackEventType
@@ -9,13 +10,26 @@ from PySrDaliGateway import CallbackEventType
 def find_device_listener(
     device: MagicMock, event_type: CallbackEventType
 ) -> Callable[..., None]:
-    """Find the registered listener callback for a specific device and event type."""
-    for call in device.register_listener.call_args_list:
-        if call[0][0] == event_type:
-            return call[0][1]
-    raise AssertionError(
-        f"Listener for event type {event_type} not found on device {device.dev_id}"
-    )
+    """Find the registered listener callback for a specific device and event type.
+
+    Returns a wrapper that calls all registered listeners for the event type.
+    """
+    callbacks: list[Callable[..., None]] = [
+        call[0][1]
+        for call in device.register_listener.call_args_list
+        if call[0][0] == event_type
+    ]
+
+    if not callbacks:
+        raise AssertionError(
+            f"Listener for event type {event_type} not found on device {device.dev_id}"
+        )
+
+    def trigger_all(*args: Any, **kwargs: Any) -> None:
+        for cb in callbacks:
+            cb(*args, **kwargs)
+
+    return trigger_all
 
 
 def trigger_availability_callback(device: MagicMock, available: bool) -> None:

tests/components/sunricher_dali/conftest.py

@@ -73,6 +73,16 @@ ILLUMINANCE_SENSOR_DATA: dict[str, Any] = {
     "channel": 0,
 }
 
+MOTION_SENSOR_DATA: dict[str, Any] = {
+    "dev_id": "02010000106A242121110E",
+    "dev_type": "0201",
+    "name": "Motion Sensor 0000-10",
+    "model": "DALI Motion Sensor",
+    "color_mode": None,
+    "address": 10,
+    "channel": 0,
+}
+
 
 @pytest.fixture
 async def init_integration(
@@ -142,6 +152,12 @@ def mock_illuminance_device() -> MagicMock:
     return _create_mock_device(ILLUMINANCE_SENSOR_DATA)
 
 
+@pytest.fixture
+def mock_motion_sensor_device() -> MagicMock:
+    """Return a mocked motion sensor device."""
+    return _create_mock_device(MOTION_SENSOR_DATA)
+
+
 def _create_scene_device_property(
     dev_type: str, brightness: int = 128, **kwargs: Any
 ) -> dict[str, Any]:

tests/components/sunricher_dali/snapshots/test_binary_sensor.ambr

@@ -0,0 +1,101 @@
+# serializer version: 1
+# name: test_setup_entry[binary_sensor.motion_sensor_0000_10_motion-entry]
+  EntityRegistryEntrySnapshot({
+    'aliases': set({
+    }),
+    'area_id': None,
+    'capabilities': None,
+    'config_entry_id': <ANY>,
+    'config_subentry_id': <ANY>,
+    'device_class': None,
+    'device_id': <ANY>,
+    'disabled_by': None,
+    'domain': 'binary_sensor',
+    'entity_category': None,
+    'entity_id': 'binary_sensor.motion_sensor_0000_10_motion',
+    'has_entity_name': True,
+    'hidden_by': None,
+    'icon': None,
+    'id': <ANY>,
+    'labels': set({
+    }),
+    'name': None,
+    'object_id_base': 'Motion',
+    'options': dict({
+    }),
+    'original_device_class': <BinarySensorDeviceClass.MOTION: 'motion'>,
+    'original_icon': None,
+    'original_name': 'Motion',
+    'platform': 'sunricher_dali',
+    'previous_unique_id': None,
+    'suggested_object_id': None,
+    'supported_features': 0,
+    'translation_key': None,
+    'unique_id': '02010000106A242121110E_motion',
+    'unit_of_measurement': None,
+  })
+# ---
+# name: test_setup_entry[binary_sensor.motion_sensor_0000_10_motion-state]
+  StateSnapshot({
+    'attributes': ReadOnlyDict({
+      'device_class': 'motion',
+      'friendly_name': 'Motion Sensor 0000-10 Motion',
+    }),
+    'context': <ANY>,
+    'entity_id': 'binary_sensor.motion_sensor_0000_10_motion',
+    'last_changed': <ANY>,
+    'last_reported': <ANY>,
+    'last_updated': <ANY>,
+    'state': 'unknown',
+  })
+# ---
+# name: test_setup_entry[binary_sensor.motion_sensor_0000_10_occupancy-entry]
+  EntityRegistryEntrySnapshot({
+    'aliases': set({
+    }),
+    'area_id': None,
+    'capabilities': None,
+    'config_entry_id': <ANY>,
+    'config_subentry_id': <ANY>,
+    'device_class': None,
+    'device_id': <ANY>,
+    'disabled_by': None,
+    'domain': 'binary_sensor',
+    'entity_category': None,
+    'entity_id': 'binary_sensor.motion_sensor_0000_10_occupancy',
+    'has_entity_name': True,
+    'hidden_by': None,
+    'icon': None,
+    'id': <ANY>,
+    'labels': set({
+    }),
+    'name': None,
+    'object_id_base': 'Occupancy',
+    'options': dict({
+    }),
+    'original_device_class': <BinarySensorDeviceClass.OCCUPANCY: 'occupancy'>,
+    'original_icon': None,
+    'original_name': 'Occupancy',
+    'platform': 'sunricher_dali',
+    'previous_unique_id': None,
+    'suggested_object_id': None,
+    'supported_features': 0,
+    'translation_key': None,
+    'unique_id': '02010000106A242121110E_occupancy',
+    'unit_of_measurement': None,
+  })
+# ---
+# name: test_setup_entry[binary_sensor.motion_sensor_0000_10_occupancy-state]
+  StateSnapshot({
+    'attributes': ReadOnlyDict({
+      'device_class': 'occupancy',
+      'friendly_name': 'Motion Sensor 0000-10 Occupancy',
+    }),
+    'context': <ANY>,
+    'entity_id': 'binary_sensor.motion_sensor_0000_10_occupancy',
+    'last_changed': <ANY>,
+    'last_reported': <ANY>,
+    'last_updated': <ANY>,
+    'state': 'unknown',
+  })
+# ---

tests/components/sunricher_dali/test_binary_sensor.py

@@ -0,0 +1,284 @@
+"""Test the Sunricher DALI binary sensor platform."""
+
+from unittest.mock import MagicMock
+
+from PySrDaliGateway import CallbackEventType
+from PySrDaliGateway.types import MotionState
+import pytest
+
+from homeassistant.const import (
+    STATE_OFF,
+    STATE_ON,
+    STATE_UNAVAILABLE,
+    STATE_UNKNOWN,
+    Platform,
+)
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import entity_registry as er
+
+from . import find_device_listener, trigger_availability_callback
+
+from tests.common import MockConfigEntry, SnapshotAssertion, snapshot_platform
+
+TEST_MOTION_ENTITY_ID = "binary_sensor.motion_sensor_0000_10_motion"
+TEST_OCCUPANCY_ENTITY_ID = "binary_sensor.motion_sensor_0000_10_occupancy"
+
+
+@pytest.fixture
+def mock_devices(mock_motion_sensor_device: MagicMock) -> list[MagicMock]:
+    """Override mock_devices to use motion sensor device only."""
+    return [mock_motion_sensor_device]
+
+
+@pytest.fixture
+def platforms() -> list[Platform]:
+    """Fixture to specify which platforms to test."""
+    return [Platform.BINARY_SENSOR]
+
+
+@pytest.mark.usefixtures("init_integration")
+async def test_setup_entry(
+    hass: HomeAssistant,
+    entity_registry: er.EntityRegistry,
+    mock_config_entry: MockConfigEntry,
+    snapshot: SnapshotAssertion,
+) -> None:
+    """Test that async_setup_entry correctly creates binary sensor entities."""
+    await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
+
+    entity_entries = er.async_entries_for_config_entry(
+        entity_registry, mock_config_entry.entry_id
+    )
+
+    assert len(entity_entries) == 2
+
+    entity_ids = [entry.entity_id for entry in entity_entries]
+    assert TEST_MOTION_ENTITY_ID in entity_ids
+    assert TEST_OCCUPANCY_ENTITY_ID in entity_ids
+
+
+@pytest.mark.usefixtures("init_integration")
+async def test_occupancy_sensor_initial_state(
+    hass: HomeAssistant,
+) -> None:
+    """Test occupancy sensor initial state is OFF."""
+    state = hass.states.get(TEST_OCCUPANCY_ENTITY_ID)
+    assert state is not None
+    assert state.state == STATE_UNKNOWN
+
+
+@pytest.mark.usefixtures("init_integration")
+async def test_occupancy_sensor_motion_detected(
+    hass: HomeAssistant,
+    mock_motion_sensor_device: MagicMock,
+) -> None:
+    """Test occupancy sensor turns ON when motion is detected."""
+    motion_device = mock_motion_sensor_device
+
+    callback = find_device_listener(motion_device, CallbackEventType.MOTION_STATUS)
+
+    callback({"motion_state": MotionState.MOTION})
+    await hass.async_block_till_done()
+
+    state = hass.states.get(TEST_OCCUPANCY_ENTITY_ID)
+    assert state is not None
+    assert state.state == STATE_ON
+
+
+@pytest.mark.usefixtures("init_integration")
+async def test_occupancy_sensor_presence_detected(
+    hass: HomeAssistant,
+    mock_motion_sensor_device: MagicMock,
+) -> None:
+    """Test occupancy sensor turns ON when presence is detected."""
+    motion_device = mock_motion_sensor_device
+
+    callback = find_device_listener(motion_device, CallbackEventType.MOTION_STATUS)
+
+    callback({"motion_state": MotionState.PRESENCE})
+    await hass.async_block_till_done()
+
+    state = hass.states.get(TEST_OCCUPANCY_ENTITY_ID)
+    assert state is not None
+    assert state.state == STATE_ON
+
+
+@pytest.mark.usefixtures("init_integration")
+async def test_occupancy_sensor_occupancy_detected(
+    hass: HomeAssistant,
+    mock_motion_sensor_device: MagicMock,
+) -> None:
+    """Test occupancy sensor turns ON when occupancy is detected."""
+    motion_device = mock_motion_sensor_device
+
+    callback = find_device_listener(motion_device, CallbackEventType.MOTION_STATUS)
+
+    callback({"motion_state": MotionState.OCCUPANCY})
+    await hass.async_block_till_done()
+
+    state = hass.states.get(TEST_OCCUPANCY_ENTITY_ID)
+    assert state is not None
+    assert state.state == STATE_ON
+
+
+@pytest.mark.usefixtures("init_integration")
+async def test_occupancy_sensor_ignores_no_motion(
+    hass: HomeAssistant,
+    mock_motion_sensor_device: MagicMock,
+) -> None:
+    """Test occupancy sensor stays ON after NO_MOTION (only VACANT turns it off)."""
+    motion_device = mock_motion_sensor_device
+
+    callback = find_device_listener(motion_device, CallbackEventType.MOTION_STATUS)
+
+    callback({"motion_state": MotionState.MOTION})
+    await hass.async_block_till_done()
+
+    state = hass.states.get(TEST_OCCUPANCY_ENTITY_ID)
+    assert state is not None
+    assert state.state == STATE_ON
+
+    callback({"motion_state": MotionState.NO_MOTION})
+    await hass.async_block_till_done()
+
+    state = hass.states.get(TEST_OCCUPANCY_ENTITY_ID)
+    assert state is not None
+    assert state.state == STATE_ON  # Still ON - NO_MOTION does not affect occupancy
+
+
+@pytest.mark.usefixtures("init_integration")
+async def test_occupancy_sensor_vacant(
+    hass: HomeAssistant,
+    mock_motion_sensor_device: MagicMock,
+) -> None:
+    """Test occupancy sensor turns OFF when vacant."""
+    motion_device = mock_motion_sensor_device
+
+    callback = find_device_listener(motion_device, CallbackEventType.MOTION_STATUS)
+
+    callback({"motion_state": MotionState.OCCUPANCY})
+    await hass.async_block_till_done()
+
+    state = hass.states.get(TEST_OCCUPANCY_ENTITY_ID)
+    assert state is not None
+    assert state.state == STATE_ON
+
+    callback({"motion_state": MotionState.VACANT})
+    await hass.async_block_till_done()
+
+    state = hass.states.get(TEST_OCCUPANCY_ENTITY_ID)
+    assert state is not None
+    assert state.state == STATE_OFF
+
+
+@pytest.mark.usefixtures("init_integration")
+async def test_occupancy_sensor_availability(
+    hass: HomeAssistant,
+    mock_motion_sensor_device: MagicMock,
+) -> None:
+    """Test availability changes are reflected in binary sensor entity state."""
+    motion_device = mock_motion_sensor_device
+
+    trigger_availability_callback(motion_device, False)
+    await hass.async_block_till_done()
+
+    state = hass.states.get(TEST_OCCUPANCY_ENTITY_ID)
+    assert state is not None
+    assert state.state == STATE_UNAVAILABLE
+
+    trigger_availability_callback(motion_device, True)
+    await hass.async_block_till_done()
+
+    state = hass.states.get(TEST_OCCUPANCY_ENTITY_ID)
+    assert state is not None
+    assert state.state != STATE_UNAVAILABLE
+
+
+# MotionSensor tests
+
+
+@pytest.mark.usefixtures("init_integration")
+async def test_motion_sensor_initial_state(
+    hass: HomeAssistant,
+) -> None:
+    """Test motion sensor initial state is OFF."""
+    state = hass.states.get(TEST_MOTION_ENTITY_ID)
+    assert state is not None
+    assert state.state == STATE_UNKNOWN
+
+
+@pytest.mark.usefixtures("init_integration")
+async def test_motion_sensor_on_motion(
+    hass: HomeAssistant,
+    mock_motion_sensor_device: MagicMock,
+) -> None:
+    """Test motion sensor turns ON when motion is detected."""
+    motion_device = mock_motion_sensor_device
+
+    callback = find_device_listener(motion_device, CallbackEventType.MOTION_STATUS)
+
+    callback({"motion_state": MotionState.MOTION})
+    await hass.async_block_till_done()
+
+    state = hass.states.get(TEST_MOTION_ENTITY_ID)
+    assert state is not None
+    assert state.state == STATE_ON
+
+
+@pytest.mark.usefixtures("init_integration")
+async def test_motion_sensor_off_no_motion(
+    hass: HomeAssistant,
+    mock_motion_sensor_device: MagicMock,
+) -> None:
+    """Test motion sensor turns OFF when no motion."""
+    motion_device = mock_motion_sensor_device
+
+    callback = find_device_listener(motion_device, CallbackEventType.MOTION_STATUS)
+
+    callback({"motion_state": MotionState.MOTION})
+    await hass.async_block_till_done()
+
+    state = hass.states.get(TEST_MOTION_ENTITY_ID)
+    assert state.state == STATE_ON
+
+    callback({"motion_state": MotionState.NO_MOTION})
+    await hass.async_block_till_done()
+
+    state = hass.states.get(TEST_MOTION_ENTITY_ID)
+    assert state is not None
+    assert state.state == STATE_OFF
+
+
+@pytest.mark.usefixtures("init_integration")
+async def test_motion_sensor_ignores_occupancy_events(
+    hass: HomeAssistant,
+    mock_motion_sensor_device: MagicMock,
+) -> None:
+    """Test motion sensor ignores OCCUPANCY, PRESENCE, VACANT events."""
+    motion_device = mock_motion_sensor_device
+
+    callback = find_device_listener(motion_device, CallbackEventType.MOTION_STATUS)
+
+    # Start with motion ON
+    callback({"motion_state": MotionState.MOTION})
+    await hass.async_block_till_done()
+    state = hass.states.get(TEST_MOTION_ENTITY_ID)
+    assert state.state == STATE_ON
+
+    # OCCUPANCY should not change motion sensor
+    callback({"motion_state": MotionState.OCCUPANCY})
+    await hass.async_block_till_done()
+    state = hass.states.get(TEST_MOTION_ENTITY_ID)
+    assert state.state == STATE_ON
+
+    # PRESENCE should not change motion sensor
+    callback({"motion_state": MotionState.PRESENCE})
+    await hass.async_block_till_done()
+    state = hass.states.get(TEST_MOTION_ENTITY_ID)
+    assert state.state == STATE_ON
+
+    # VACANT should not change motion sensor
+    callback({"motion_state": MotionState.VACANT})
+    await hass.async_block_till_done()
+    state = hass.states.get(TEST_MOTION_ENTITY_ID)
+    assert state.state == STATE_ON