Add binary sensor platform to devolo Home Network (#60301)

Co-authored-by: J. Nick Koston <nick@koston.org>

homeassistant/components/devolo_home_network/binary_sensor.py

@@ -0,0 +1,99 @@
+"""Platform for binary sensor integration."""
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass
+
+from devolo_plc_api.device import Device
+
+from homeassistant.components.binary_sensor import (
+    DEVICE_CLASS_PLUG,
+    BinarySensorEntity,
+    BinarySensorEntityDescription,
+)
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity import EntityCategory
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+
+from .const import CONNECTED_PLC_DEVICES, CONNECTED_TO_ROUTER, DOMAIN
+from .entity import DevoloEntity
+
+
+def _is_connected_to_router(entity: DevoloBinarySensorEntity) -> bool:
+    """Check, if device is attached to the router."""
+    return all(
+        device["attached_to_router"]
+        for device in entity.coordinator.data["network"]["devices"]
+        if device["mac_address"] == entity.device.mac
+    )
+
+
+@dataclass
+class DevoloBinarySensorRequiredKeysMixin:
+    """Mixin for required keys."""
+
+    value_func: Callable[[DevoloBinarySensorEntity], bool]
+
+
+@dataclass
+class DevoloBinarySensorEntityDescription(
+    BinarySensorEntityDescription, DevoloBinarySensorRequiredKeysMixin
+):
+    """Describes devolo sensor entity."""
+
+
+SENSOR_TYPES: dict[str, DevoloBinarySensorEntityDescription] = {
+    CONNECTED_TO_ROUTER: DevoloBinarySensorEntityDescription(
+        key=CONNECTED_TO_ROUTER,
+        device_class=DEVICE_CLASS_PLUG,
+        entity_category=EntityCategory.DIAGNOSTIC,
+        entity_registry_enabled_default=False,
+        icon="mdi:router-network",
+        name="Connected to router",
+        value_func=_is_connected_to_router,
+    ),
+}
+
+
+async def async_setup_entry(
+    hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
+) -> None:
+    """Get all devices and sensors and setup them via config entry."""
+    device: Device = hass.data[DOMAIN][entry.entry_id]["device"]
+    coordinators: dict[str, DataUpdateCoordinator] = hass.data[DOMAIN][entry.entry_id][
+        "coordinators"
+    ]
+
+    entities: list[BinarySensorEntity] = []
+    if device.plcnet:
+        entities.append(
+            DevoloBinarySensorEntity(
+                coordinators[CONNECTED_PLC_DEVICES],
+                SENSOR_TYPES[CONNECTED_TO_ROUTER],
+                device,
+                entry.title,
+            )
+        )
+    async_add_entities(entities)
+
+
+class DevoloBinarySensorEntity(DevoloEntity, BinarySensorEntity):
+    """Representation of a devolo binary sensor."""
+
+    def __init__(
+        self,
+        coordinator: DataUpdateCoordinator,
+        description: DevoloBinarySensorEntityDescription,
+        device: Device,
+        device_name: str,
+    ) -> None:
+        """Initialize entity."""
+        self.entity_description: DevoloBinarySensorEntityDescription = description
+        super().__init__(coordinator, device, device_name)
+
+    @property
+    def is_on(self) -> bool:
+        """State of the binary sensor."""
+        return self.entity_description.value_func(self)

homeassistant/components/devolo_home_network/const.py

@@ -5,7 +5,7 @@ from datetime import timedelta
 from homeassistant.const import Platform
 
 DOMAIN = "devolo_home_network"
-PLATFORMS = [Platform.SENSOR]
+PLATFORMS = [Platform.BINARY_SENSOR, Platform.SENSOR]
 
 PRODUCT = "product"
 SERIAL_NUMBER = "serial_number"
@@ -15,5 +15,6 @@ LONG_UPDATE_INTERVAL = timedelta(minutes=5)
 SHORT_UPDATE_INTERVAL = timedelta(seconds=15)
 
 CONNECTED_PLC_DEVICES = "connected_plc_devices"
+CONNECTED_TO_ROUTER = "connected_to_router"
 CONNECTED_WIFI_CLIENTS = "connected_wifi_clients"
 NEIGHBORING_WIFI_NETWORKS = "neighboring_wifi_networks"

homeassistant/components/devolo_home_network/entity.py

@@ -21,17 +21,14 @@ class DevoloEntity(CoordinatorEntity):
         """Initialize a devolo home network device."""
         super().__init__(coordinator)
 
-        self._device = device
+        self.device = device
-        self._device_name = device_name
 
         self._attr_device_info = DeviceInfo(
-            configuration_url=f"http://{self._device.ip}",
+            configuration_url=f"http://{device.ip}",
-            identifiers={(DOMAIN, str(self._device.serial_number))},
+            identifiers={(DOMAIN, str(device.serial_number))},
             manufacturer="devolo",
-            model=self._device.product,
+            model=device.product,
-            name=self._device_name,
+            name=device_name,
-            sw_version=self._device.firmware_version,
+            sw_version=device.firmware_version,
-        )
-        self._attr_unique_id = (
-            f"{self._device.serial_number}_{self.entity_description.key}"
         )
+        self._attr_unique_id = f"{device.serial_number}_{self.entity_description.key}"

tests/components/devolo_home_network/__init__.py

@@ -28,5 +28,6 @@ def configure_integration(hass: HomeAssistant) -> MockConfigEntry:
 
 async def async_connect(self, session_instance: Any = None):
     """Give a mocked device the needed properties."""
+    self.mac = DISCOVERY_INFO.properties["PlcMacAddress"]
     self.plcnet = PlcNetApi(IP, None, dataclasses.asdict(DISCOVERY_INFO))
     self.device = DeviceApi(IP, None, dataclasses.asdict(DISCOVERY_INFO))

tests/components/devolo_home_network/const.py

@@ -62,6 +62,12 @@ NEIGHBOR_ACCESS_POINTS = {
 
 PLCNET = {
     "network": {
+        "devices": [
+            {
+                "mac_address": "AA:BB:CC:DD:EE:FF",
+                "attached_to_router": False,
+            }
+        ],
         "data_rates": [
             {
                 "mac_address_from": "AA:BB:CC:DD:EE:FF",
@@ -70,6 +76,17 @@ PLCNET = {
                 "tx_rate": 0.0,
             },
         ],
-        "devices": [],
+    }
+}
+
+PLCNET_ATTACHED = {
+    "network": {
+        "devices": [
+            {
+                "mac_address": "AA:BB:CC:DD:EE:FF",
+                "attached_to_router": True,
+            }
+        ],
+        "data_rates": [],
     }
 }

tests/components/devolo_home_network/test_binary_sensor.py

@@ -0,0 +1,83 @@
+"""Tests for the devolo Home Network sensors."""
+from unittest.mock import AsyncMock, patch
+
+from devolo_plc_api.exceptions.device import DeviceUnavailable
+import pytest
+
+from homeassistant.components.binary_sensor import DOMAIN
+from homeassistant.components.devolo_home_network.const import (
+    CONNECTED_TO_ROUTER,
+    LONG_UPDATE_INTERVAL,
+)
+from homeassistant.const import STATE_OFF, STATE_ON, STATE_UNAVAILABLE
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import entity_registry
+from homeassistant.helpers.entity import EntityCategory
+from homeassistant.util import dt
+
+from . import configure_integration
+from .const import PLCNET_ATTACHED
+
+from tests.common import async_fire_time_changed
+
+
+@pytest.mark.usefixtures("mock_device", "mock_zeroconf")
+async def test_binary_sensor_setup(hass: HomeAssistant):
+    """Test default setup of the binary sensor component."""
+    entry = configure_integration(hass)
+    await hass.config_entries.async_setup(entry.entry_id)
+    await hass.async_block_till_done()
+
+    assert hass.states.get(f"{DOMAIN}.{CONNECTED_TO_ROUTER}") is None
+
+    await hass.config_entries.async_unload(entry.entry_id)
+
+
+@pytest.mark.usefixtures("mock_device", "mock_zeroconf")
+async def test_update_attached_to_router(hass: HomeAssistant):
+    """Test state change of a attached_to_router binary sensor device."""
+    state_key = f"{DOMAIN}.{CONNECTED_TO_ROUTER}"
+    entry = configure_integration(hass)
+
+    er = entity_registry.async_get(hass)
+
+    await hass.config_entries.async_setup(entry.entry_id)
+    await hass.async_block_till_done()
+
+    # Enable entity
+    er.async_update_entity(state_key, disabled_by=None)
+    await hass.async_block_till_done()
+    async_fire_time_changed(hass, dt.utcnow() + LONG_UPDATE_INTERVAL)
+    await hass.async_block_till_done()
+
+    state = hass.states.get(state_key)
+    assert state is not None
+    assert state.state == STATE_OFF
+
+    assert er.async_get(state_key).entity_category == EntityCategory.DIAGNOSTIC
+
+    # Emulate device failure
+    with patch(
+        "devolo_plc_api.plcnet_api.plcnetapi.PlcNetApi.async_get_network_overview",
+        side_effect=DeviceUnavailable,
+    ):
+        async_fire_time_changed(hass, dt.utcnow() + LONG_UPDATE_INTERVAL)
+        await hass.async_block_till_done()
+
+        state = hass.states.get(state_key)
+        assert state is not None
+        assert state.state == STATE_UNAVAILABLE
+
+    # Emulate state change
+    with patch(
+        "devolo_plc_api.plcnet_api.plcnetapi.PlcNetApi.async_get_network_overview",
+        new=AsyncMock(return_value=PLCNET_ATTACHED),
+    ):
+        async_fire_time_changed(hass, dt.utcnow() + LONG_UPDATE_INTERVAL)
+        await hass.async_block_till_done()
+
+        state = hass.states.get(state_key)
+        assert state is not None
+        assert state.state == STATE_ON
+
+    await hass.config_entries.async_unload(entry.entry_id)