Add QNAP QSW binary sensors for each port (#76522)
Co-authored-by: J. Nick Koston <nick@koston.org>
This commit is contained in:
parent
8a1805fa89
commit
873b903cf0
3 changed files with 174 additions and 17 deletions
|
@ -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()
|
||||
|
|
|
@ -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
|
||||
|
||||
|
|
|
@ -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
|
||||
|
|
Loading…
Add table
Reference in a new issue