From 873b903cf0217c48049e32c45f7d5b115078b831 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C3=81lvaro=20Fern=C3=A1ndez=20Rojas?= Date: Sun, 27 Nov 2022 21:34:14 +0100 Subject: [PATCH] Add QNAP QSW binary sensors for each port (#76522) Co-authored-by: J. Nick Koston --- .../components/qnap_qsw/binary_sensor.py | 95 ++++++++++++++++--- homeassistant/components/qnap_qsw/entity.py | 28 +++++- .../components/qnap_qsw/test_binary_sensor.py | 68 ++++++++++++- 3 files changed, 174 insertions(+), 17 deletions(-) diff --git a/homeassistant/components/qnap_qsw/binary_sensor.py b/homeassistant/components/qnap_qsw/binary_sensor.py index 71af89778b8..aeebb6cc055 100644 --- a/homeassistant/components/qnap_qsw/binary_sensor.py +++ b/homeassistant/components/qnap_qsw/binary_sensor.py @@ -1,10 +1,18 @@ """Support for the QNAP QSW binary sensors.""" from __future__ import annotations -from dataclasses import dataclass +from dataclasses import dataclass, replace from typing import Final -from aioqsw.const import QSD_ANOMALY, QSD_FIRMWARE_CONDITION, QSD_MESSAGE +from aioqsw.const import ( + QSD_ANOMALY, + QSD_FIRMWARE_CONDITION, + QSD_LACP_PORTS, + QSD_LINK, + QSD_MESSAGE, + QSD_PORTS, + QSD_PORTS_STATUS, +) from homeassistant.components.binary_sensor import ( BinarySensorDeviceClass, @@ -18,7 +26,7 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback from .const import ATTR_MESSAGE, DOMAIN, QSW_COORD_DATA from .coordinator import QswDataCoordinator -from .entity import QswEntityDescription, QswSensorEntity +from .entity import QswEntityDescription, QswEntityType, QswSensorEntity @dataclass @@ -28,6 +36,8 @@ class QswBinarySensorEntityDescription( """A class that describes QNAP QSW binary sensor entities.""" attributes: dict[str, list[str]] | None = None + qsw_type: QswEntityType | None = None + sep_key: str = "_" BINARY_SENSOR_TYPES: Final[tuple[QswBinarySensorEntityDescription, ...]] = ( @@ -43,20 +53,77 @@ BINARY_SENSOR_TYPES: Final[tuple[QswBinarySensorEntityDescription, ...]] = ( ), ) +LACP_PORT_BINARY_SENSOR_TYPES: Final[tuple[QswBinarySensorEntityDescription, ...]] = ( + QswBinarySensorEntityDescription( + device_class=BinarySensorDeviceClass.CONNECTIVITY, + entity_registry_enabled_default=False, + key=QSD_PORTS_STATUS, + qsw_type=QswEntityType.LACP_PORT, + name="Link", + subkey=QSD_LINK, + ), +) + +PORT_BINARY_SENSOR_TYPES: Final[tuple[QswBinarySensorEntityDescription, ...]] = ( + QswBinarySensorEntityDescription( + device_class=BinarySensorDeviceClass.CONNECTIVITY, + entity_registry_enabled_default=False, + key=QSD_PORTS_STATUS, + qsw_type=QswEntityType.PORT, + name="Link", + subkey=QSD_LINK, + ), +) + async def async_setup_entry( hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback ) -> None: """Add QNAP QSW binary sensors from a config_entry.""" coordinator: QswDataCoordinator = hass.data[DOMAIN][entry.entry_id][QSW_COORD_DATA] - async_add_entities( - QswBinarySensor(coordinator, description, entry) - for description in BINARY_SENSOR_TYPES + + entities: list[QswBinarySensor] = [] + + for description in BINARY_SENSOR_TYPES: if ( description.key in coordinator.data and description.subkey in coordinator.data[description.key] - ) - ) + ): + entities.append(QswBinarySensor(coordinator, description, entry)) + + for description in LACP_PORT_BINARY_SENSOR_TYPES: + if ( + description.key in coordinator.data + and QSD_LACP_PORTS in coordinator.data[description.key] + ): + for port_id, port_values in coordinator.data[description.key][ + QSD_LACP_PORTS + ].items(): + if description.subkey in port_values: + _desc = replace( + description, + sep_key=f"_lacp_port_{port_id}_", + name=f"LACP Port {port_id} {description.name}", + ) + entities.append(QswBinarySensor(coordinator, _desc, entry, port_id)) + + for description in PORT_BINARY_SENSOR_TYPES: + if ( + description.key in coordinator.data + and QSD_PORTS in coordinator.data[description.key] + ): + for port_id, port_values in coordinator.data[description.key][ + QSD_PORTS + ].items(): + if description.subkey in port_values: + _desc = replace( + description, + sep_key=f"_port_{port_id}_", + name=f"Port {port_id} {description.name}", + ) + entities.append(QswBinarySensor(coordinator, _desc, entry, port_id)) + + async_add_entities(entities) class QswBinarySensor(QswSensorEntity, BinarySensorEntity): @@ -69,13 +136,13 @@ class QswBinarySensor(QswSensorEntity, BinarySensorEntity): coordinator: QswDataCoordinator, description: QswBinarySensorEntityDescription, entry: ConfigEntry, + type_id: int | None = None, ) -> None: """Initialize.""" - super().__init__(coordinator, entry) + super().__init__(coordinator, entry, type_id) + self._attr_name = f"{self.product} {description.name}" - self._attr_unique_id = ( - f"{entry.unique_id}_{description.key}_{description.subkey}" - ) + self._attr_unique_id = f"{entry.unique_id}_{description.key}{description.sep_key}{description.subkey}" self.entity_description = description self._async_update_attrs() @@ -83,6 +150,8 @@ class QswBinarySensor(QswSensorEntity, BinarySensorEntity): def _async_update_attrs(self) -> None: """Update binary sensor attributes.""" self._attr_is_on = self.get_device_value( - self.entity_description.key, self.entity_description.subkey + self.entity_description.key, + self.entity_description.subkey, + self.entity_description.qsw_type, ) super()._async_update_attrs() diff --git a/homeassistant/components/qnap_qsw/entity.py b/homeassistant/components/qnap_qsw/entity.py index 7da47f9734f..7d1ec33ee71 100644 --- a/homeassistant/components/qnap_qsw/entity.py +++ b/homeassistant/components/qnap_qsw/entity.py @@ -7,11 +7,14 @@ from typing import Any from aioqsw.const import ( QSD_FIRMWARE, QSD_FIRMWARE_INFO, + QSD_LACP_PORTS, QSD_MAC, + QSD_PORTS, QSD_PRODUCT, QSD_SYSTEM_BOARD, ) +from homeassistant.backports.enum import StrEnum from homeassistant.config_entries import ConfigEntry from homeassistant.const import CONF_URL from homeassistant.core import callback @@ -23,6 +26,13 @@ from .const import MANUFACTURER from .coordinator import QswDataCoordinator, QswFirmwareCoordinator +class QswEntityType(StrEnum): + """QNAP QSW Entity Type.""" + + LACP_PORT = QSD_LACP_PORTS + PORT = QSD_PORTS + + class QswDataEntity(CoordinatorEntity[QswDataCoordinator]): """Define an QNAP QSW entity.""" @@ -30,10 +40,12 @@ class QswDataEntity(CoordinatorEntity[QswDataCoordinator]): self, coordinator: QswDataCoordinator, entry: ConfigEntry, + type_id: int | None = None, ) -> None: """Initialize.""" super().__init__(coordinator) + self.type_id = type_id self.product = self.get_device_value(QSD_SYSTEM_BOARD, QSD_PRODUCT) self._attr_device_info = DeviceInfo( configuration_url=entry.data[CONF_URL], @@ -49,12 +61,24 @@ class QswDataEntity(CoordinatorEntity[QswDataCoordinator]): sw_version=self.get_device_value(QSD_FIRMWARE_INFO, QSD_FIRMWARE), ) - def get_device_value(self, key: str, subkey: str) -> Any: + def get_device_value( + self, + key: str, + subkey: str, + qsw_type: QswEntityType | None = None, + ) -> Any: """Return device value by key.""" value = None if key in self.coordinator.data: data = self.coordinator.data[key] - if subkey in data: + if qsw_type is not None and self.type_id is not None: + if ( + qsw_type in data + and self.type_id in data[qsw_type] + and subkey in data[qsw_type][self.type_id] + ): + value = data[qsw_type][self.type_id][subkey] + elif subkey in data: value = data[subkey] return value diff --git a/tests/components/qnap_qsw/test_binary_sensor.py b/tests/components/qnap_qsw/test_binary_sensor.py index a36a34f02be..a270e78f051 100644 --- a/tests/components/qnap_qsw/test_binary_sensor.py +++ b/tests/components/qnap_qsw/test_binary_sensor.py @@ -1,17 +1,81 @@ """The binary sensor tests for the QNAP QSW platform.""" +from unittest.mock import AsyncMock + from homeassistant.components.qnap_qsw.const import ATTR_MESSAGE -from homeassistant.const import STATE_OFF +from homeassistant.const import STATE_OFF, STATE_ON from homeassistant.core import HomeAssistant +from homeassistant.helpers import entity_registry from .util import async_init_integration -async def test_qnap_qsw_create_binary_sensors(hass: HomeAssistant) -> None: +async def test_qnap_qsw_create_binary_sensors( + hass: HomeAssistant, entity_registry_enabled_by_default: AsyncMock +) -> None: """Test creation of binary sensors.""" await async_init_integration(hass) + er = entity_registry.async_get(hass) state = hass.states.get("binary_sensor.qsw_m408_4c_anomaly") assert state.state == STATE_OFF assert state.attributes.get(ATTR_MESSAGE) is None + + state = hass.states.get("binary_sensor.qsw_m408_4c_lacp_port_1_link") + assert state.state == STATE_OFF + entry = er.async_get(state.entity_id) + assert entry.unique_id == "qsw_unique_id_ports-status_lacp_port_1_link" + + state = hass.states.get("binary_sensor.qsw_m408_4c_lacp_port_2_link") + assert state.state == STATE_OFF + + state = hass.states.get("binary_sensor.qsw_m408_4c_lacp_port_3_link") + assert state.state == STATE_OFF + + state = hass.states.get("binary_sensor.qsw_m408_4c_lacp_port_4_link") + assert state.state == STATE_OFF + + state = hass.states.get("binary_sensor.qsw_m408_4c_lacp_port_5_link") + assert state.state == STATE_OFF + + state = hass.states.get("binary_sensor.qsw_m408_4c_lacp_port_6_link") + assert state.state == STATE_OFF + + state = hass.states.get("binary_sensor.qsw_m408_4c_port_1_link") + assert state.state == STATE_ON + entry = er.async_get(state.entity_id) + assert entry.unique_id == "qsw_unique_id_ports-status_port_1_link" + + state = hass.states.get("binary_sensor.qsw_m408_4c_port_2_link") + assert state.state == STATE_ON + + state = hass.states.get("binary_sensor.qsw_m408_4c_port_3_link") + assert state.state == STATE_ON + + state = hass.states.get("binary_sensor.qsw_m408_4c_port_4_link") + assert state.state == STATE_OFF + + state = hass.states.get("binary_sensor.qsw_m408_4c_port_5_link") + assert state.state == STATE_OFF + + state = hass.states.get("binary_sensor.qsw_m408_4c_port_6_link") + assert state.state == STATE_OFF + + state = hass.states.get("binary_sensor.qsw_m408_4c_port_7_link") + assert state.state == STATE_OFF + + state = hass.states.get("binary_sensor.qsw_m408_4c_port_8_link") + assert state.state == STATE_OFF + + state = hass.states.get("binary_sensor.qsw_m408_4c_port_9_link") + assert state.state == STATE_OFF + + state = hass.states.get("binary_sensor.qsw_m408_4c_port_10_link") + assert state.state == STATE_OFF + + state = hass.states.get("binary_sensor.qsw_m408_4c_port_11_link") + assert state.state == STATE_OFF + + state = hass.states.get("binary_sensor.qsw_m408_4c_port_12_link") + assert state.state == STATE_OFF