Improve deCONZ binary sensor classes (#77419)

This commit is contained in:
Robert Svensson 2022-08-29 08:15:10 +02:00 committed by GitHub
parent 779e020dc4
commit 0154a1cecb
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
5 changed files with 232 additions and 202 deletions

View file

@ -83,6 +83,7 @@ async def async_setup_entry(
class DeconzAlarmControlPanel(DeconzDevice[AncillaryControl], AlarmControlPanelEntity):
"""Representation of a deCONZ alarm control panel."""
_update_key = "panel"
TYPE = DOMAIN
_attr_code_format = CodeFormat.NUMBER
@ -105,11 +106,7 @@ class DeconzAlarmControlPanel(DeconzDevice[AncillaryControl], AlarmControlPanelE
@callback
def async_update_callback(self) -> None:
"""Update the control panels state."""
keys = {"panel", "reachable"}
if (
self._device.changed_keys.intersection(keys)
and self._device.panel in DECONZ_TO_ALARM_STATE
):
if self._device.panel in DECONZ_TO_ALARM_STATE:
super().async_update_callback()
@property

View file

@ -1,11 +1,11 @@
"""Support for deCONZ binary sensors."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from typing import TYPE_CHECKING, TypeVar
from pydeconz.interfaces.sensors import SensorResources
from pydeconz.models.event import EventType
from pydeconz.models.sensor import SensorBase as PydeconzSensorBase
from pydeconz.models.sensor.alarm import Alarm
from pydeconz.models.sensor.carbon_monoxide import CarbonMonoxide
from pydeconz.models.sensor.fire import Fire
@ -19,7 +19,6 @@ from homeassistant.components.binary_sensor import (
DOMAIN,
BinarySensorDeviceClass,
BinarySensorEntity,
BinarySensorEntityDescription,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_TEMPERATURE
@ -32,6 +31,8 @@ from .const import ATTR_DARK, ATTR_ON, DOMAIN as DECONZ_DOMAIN
from .deconz_device import DeconzDevice
from .gateway import DeconzGateway, get_gateway_from_config_entry
_SensorDeviceT = TypeVar("_SensorDeviceT", bound=PydeconzSensorBase)
ATTR_ORIENTATION = "orientation"
ATTR_TILTANGLE = "tiltangle"
ATTR_VIBRATIONSTRENGTH = "vibrationstrength"
@ -48,140 +49,9 @@ PROVIDES_EXTRA_ATTRIBUTES = (
)
@dataclass
class DeconzBinarySensorDescriptionMixin:
"""Required values when describing secondary sensor attributes."""
suffix: str
update_key: str
value_fn: Callable[[SensorResources], bool | None]
@dataclass
class DeconzBinarySensorDescription(
BinarySensorEntityDescription,
DeconzBinarySensorDescriptionMixin,
):
"""Class describing deCONZ binary sensor entities."""
ENTITY_DESCRIPTIONS = {
Alarm: [
DeconzBinarySensorDescription(
key="alarm",
value_fn=lambda device: device.alarm if isinstance(device, Alarm) else None,
suffix="",
update_key="alarm",
device_class=BinarySensorDeviceClass.SAFETY,
)
],
CarbonMonoxide: [
DeconzBinarySensorDescription(
key="carbon_monoxide",
value_fn=lambda device: device.carbon_monoxide
if isinstance(device, CarbonMonoxide)
else None,
suffix="",
update_key="carbonmonoxide",
device_class=BinarySensorDeviceClass.CO,
)
],
Fire: [
DeconzBinarySensorDescription(
key="fire",
value_fn=lambda device: device.fire if isinstance(device, Fire) else None,
suffix="",
update_key="fire",
device_class=BinarySensorDeviceClass.SMOKE,
),
DeconzBinarySensorDescription(
key="in_test_mode",
value_fn=lambda device: device.in_test_mode
if isinstance(device, Fire)
else None,
suffix="Test Mode",
update_key="test",
device_class=BinarySensorDeviceClass.SMOKE,
entity_category=EntityCategory.DIAGNOSTIC,
),
],
GenericFlag: [
DeconzBinarySensorDescription(
key="flag",
value_fn=lambda device: device.flag
if isinstance(device, GenericFlag)
else None,
suffix="",
update_key="flag",
)
],
OpenClose: [
DeconzBinarySensorDescription(
key="open",
value_fn=lambda device: device.open
if isinstance(device, OpenClose)
else None,
suffix="",
update_key="open",
device_class=BinarySensorDeviceClass.OPENING,
)
],
Presence: [
DeconzBinarySensorDescription(
key="presence",
value_fn=lambda device: device.presence
if isinstance(device, Presence)
else None,
suffix="",
update_key="presence",
device_class=BinarySensorDeviceClass.MOTION,
)
],
Vibration: [
DeconzBinarySensorDescription(
key="vibration",
value_fn=lambda device: device.vibration
if isinstance(device, Vibration)
else None,
suffix="",
update_key="vibration",
device_class=BinarySensorDeviceClass.VIBRATION,
)
],
Water: [
DeconzBinarySensorDescription(
key="water",
value_fn=lambda device: device.water if isinstance(device, Water) else None,
suffix="",
update_key="water",
device_class=BinarySensorDeviceClass.MOISTURE,
)
],
}
COMMON_BINARY_SENSOR_DESCRIPTIONS = [
DeconzBinarySensorDescription(
key="tampered",
value_fn=lambda device: device.tampered,
suffix="Tampered",
update_key="tampered",
device_class=BinarySensorDeviceClass.TAMPER,
entity_category=EntityCategory.DIAGNOSTIC,
),
DeconzBinarySensorDescription(
key="low_battery",
value_fn=lambda device: device.low_battery,
suffix="Low Battery",
update_key="lowbattery",
device_class=BinarySensorDeviceClass.BATTERY,
entity_category=EntityCategory.DIAGNOSTIC,
),
]
@callback
def async_update_unique_id(
hass: HomeAssistant, unique_id: str, description: DeconzBinarySensorDescription
hass: HomeAssistant, unique_id: str, entity_class: DeconzBinarySensor
) -> None:
"""Update unique ID to always have a suffix.
@ -189,12 +59,12 @@ def async_update_unique_id(
"""
ent_reg = er.async_get(hass)
new_unique_id = f"{unique_id}-{description.key}"
new_unique_id = f"{unique_id}-{entity_class.unique_id_suffix}"
if ent_reg.async_get_entity_id(DOMAIN, DECONZ_DOMAIN, new_unique_id):
return
if description.suffix:
unique_id = f'{unique_id.split("-", 1)[0]}-{description.suffix.lower()}'
if entity_class.old_unique_id_suffix:
unique_id = f'{unique_id.split("-", 1)[0]}-{entity_class.old_unique_id_suffix}'
if entity_id := ent_reg.async_get_entity_id(DOMAIN, DECONZ_DOMAIN, unique_id):
ent_reg.async_update_entity(entity_id, new_unique_id=new_unique_id)
@ -214,19 +84,19 @@ async def async_setup_entry(
"""Add sensor from deCONZ."""
sensor = gateway.api.sensors[sensor_id]
for description in (
ENTITY_DESCRIPTIONS.get(type(sensor), [])
+ COMMON_BINARY_SENSOR_DESCRIPTIONS
):
for sensor_type, entity_class in ENTITY_CLASSES:
if TYPE_CHECKING:
assert isinstance(entity_class, DeconzBinarySensor)
if (
not hasattr(sensor, description.key)
or description.value_fn(sensor) is None
not isinstance(sensor, sensor_type)
or entity_class.unique_id_suffix is not None
and getattr(sensor, entity_class.unique_id_suffix) is None
):
continue
async_update_unique_id(hass, sensor.unique_id, description)
async_update_unique_id(hass, sensor.unique_id, entity_class)
async_add_entities([DeconzBinarySensor(sensor, gateway, description)])
async_add_entities([entity_class(sensor, gateway)])
gateway.register_platform_add_device_callback(
async_add_sensor,
@ -234,51 +104,28 @@ async def async_setup_entry(
)
class DeconzBinarySensor(DeconzDevice[SensorResources], BinarySensorEntity):
class DeconzBinarySensor(DeconzDevice[_SensorDeviceT], BinarySensorEntity):
"""Representation of a deCONZ binary sensor."""
old_unique_id_suffix = ""
TYPE = DOMAIN
entity_description: DeconzBinarySensorDescription
def __init__(
self,
device: SensorResources,
gateway: DeconzGateway,
description: DeconzBinarySensorDescription,
) -> None:
def __init__(self, device: _SensorDeviceT, gateway: DeconzGateway) -> None:
"""Initialize deCONZ binary sensor."""
self.entity_description: DeconzBinarySensorDescription = description
super().__init__(device, gateway)
if description.suffix:
self._attr_name = f"{self._device.name} {description.suffix}"
self._update_keys = {description.update_key, "reachable"}
if self.entity_description.key in PROVIDES_EXTRA_ATTRIBUTES:
if (
self.unique_id_suffix in PROVIDES_EXTRA_ATTRIBUTES
and self._update_keys is not None
):
self._update_keys.update({"on", "state"})
@property
def unique_id(self) -> str:
"""Return a unique identifier for this device."""
return f"{super().unique_id}-{self.entity_description.key}"
@callback
def async_update_callback(self) -> None:
"""Update the sensor's state."""
if self._device.changed_keys.intersection(self._update_keys):
super().async_update_callback()
@property
def is_on(self) -> bool | None:
"""Return the state of the sensor."""
return self.entity_description.value_fn(self._device)
@property
def extra_state_attributes(self) -> dict[str, bool | float | int | list | None]:
"""Return the state attributes of the sensor."""
attr: dict[str, bool | float | int | list | None] = {}
if self.entity_description.key not in PROVIDES_EXTRA_ATTRIBUTES:
if self.unique_id_suffix not in PROVIDES_EXTRA_ATTRIBUTES:
return attr
if self._device.on is not None:
@ -298,3 +145,179 @@ class DeconzBinarySensor(DeconzDevice[SensorResources], BinarySensorEntity):
attr[ATTR_VIBRATIONSTRENGTH] = self._device.vibration_strength
return attr
class DeconzAlarmBinarySensor(DeconzBinarySensor[Alarm]):
"""Representation of a deCONZ alarm binary sensor."""
unique_id_suffix = "alarm"
_update_key = "alarm"
_attr_device_class = BinarySensorDeviceClass.SAFETY
@property
def is_on(self) -> bool:
"""Return the state of the sensor."""
return self._device.alarm
class DeconzCarbonMonoxideBinarySensor(DeconzBinarySensor[CarbonMonoxide]):
"""Representation of a deCONZ carbon monoxide binary sensor."""
unique_id_suffix = "carbon_monoxide"
_update_key = "carbonmonoxide"
_attr_device_class = BinarySensorDeviceClass.CO
@property
def is_on(self) -> bool:
"""Return the state of the sensor."""
return self._device.carbon_monoxide
class DeconzFireBinarySensor(DeconzBinarySensor[Fire]):
"""Representation of a deCONZ fire binary sensor."""
unique_id_suffix = "fire"
_update_key = "fire"
_attr_device_class = BinarySensorDeviceClass.SMOKE
@property
def is_on(self) -> bool:
"""Return the state of the sensor."""
return self._device.fire
class DeconzFireInTestModeBinarySensor(DeconzBinarySensor[Fire]):
"""Representation of a deCONZ fire in-test-mode binary sensor."""
_name_suffix = "Test Mode"
unique_id_suffix = "in_test_mode"
old_unique_id_suffix = "test mode"
_update_key = "test"
_attr_device_class = BinarySensorDeviceClass.SMOKE
_attr_entity_category = EntityCategory.DIAGNOSTIC
@property
def is_on(self) -> bool:
"""Return the state of the sensor."""
return self._device.in_test_mode
class DeconzFlagBinarySensor(DeconzBinarySensor[GenericFlag]):
"""Representation of a deCONZ generic flag binary sensor."""
unique_id_suffix = "flag"
_update_key = "flag"
@property
def is_on(self) -> bool:
"""Return the state of the sensor."""
return self._device.flag
class DeconzOpenCloseBinarySensor(DeconzBinarySensor[OpenClose]):
"""Representation of a deCONZ open/close binary sensor."""
unique_id_suffix = "open"
_update_key = "open"
_attr_device_class = BinarySensorDeviceClass.OPENING
@property
def is_on(self) -> bool:
"""Return the state of the sensor."""
return self._device.open
class DeconzPresenceBinarySensor(DeconzBinarySensor[Presence]):
"""Representation of a deCONZ presence binary sensor."""
unique_id_suffix = "presence"
_update_key = "presence"
_attr_device_class = BinarySensorDeviceClass.MOTION
@property
def is_on(self) -> bool:
"""Return the state of the sensor."""
return self._device.presence
class DeconzVibrationBinarySensor(DeconzBinarySensor[Vibration]):
"""Representation of a deCONZ vibration binary sensor."""
unique_id_suffix = "vibration"
_update_key = "vibration"
_attr_device_class = BinarySensorDeviceClass.VIBRATION
@property
def is_on(self) -> bool:
"""Return the state of the sensor."""
return self._device.vibration
class DeconzWaterBinarySensor(DeconzBinarySensor[Water]):
"""Representation of a deCONZ water binary sensor."""
unique_id_suffix = "water"
_update_key = "water"
_attr_device_class = BinarySensorDeviceClass.MOISTURE
@property
def is_on(self) -> bool:
"""Return the state of the sensor."""
return self._device.water
class DeconzTamperedCommonBinarySensor(DeconzBinarySensor[SensorResources]):
"""Representation of a deCONZ tampered binary sensor."""
_name_suffix = "Tampered"
unique_id_suffix = "tampered"
old_unique_id_suffix = "tampered"
_update_key = "tampered"
_attr_device_class = BinarySensorDeviceClass.TAMPER
_attr_entity_category = EntityCategory.DIAGNOSTIC
@property
def is_on(self) -> bool | None:
"""Return the state of the sensor."""
return self._device.tampered
class DeconzLowBatteryCommonBinarySensor(DeconzBinarySensor[SensorResources]):
"""Representation of a deCONZ low battery binary sensor."""
_name_suffix = "Low Battery"
unique_id_suffix = "low_battery"
old_unique_id_suffix = "low battery"
_update_key = "lowbattery"
_attr_device_class = BinarySensorDeviceClass.BATTERY
_attr_entity_category = EntityCategory.DIAGNOSTIC
@property
def is_on(self) -> bool | None:
"""Return the state of the sensor."""
return self._device.low_battery
ENTITY_CLASSES = (
(Alarm, DeconzAlarmBinarySensor),
(CarbonMonoxide, DeconzCarbonMonoxideBinarySensor),
(Fire, DeconzFireBinarySensor),
(Fire, DeconzFireInTestModeBinarySensor),
(GenericFlag, DeconzFlagBinarySensor),
(OpenClose, DeconzOpenCloseBinarySensor),
(Presence, DeconzPresenceBinarySensor),
(Vibration, DeconzVibrationBinarySensor),
(Water, DeconzWaterBinarySensor),
(PydeconzSensorBase, DeconzTamperedCommonBinarySensor),
(PydeconzSensorBase, DeconzLowBatteryCommonBinarySensor),
)

View file

@ -32,6 +32,8 @@ _DeviceT = TypeVar(
class DeconzBase(Generic[_DeviceT]):
"""Common base for deconz entities and events."""
unique_id_suffix: str | None = None
def __init__(
self,
device: _DeviceT,
@ -45,6 +47,8 @@ class DeconzBase(Generic[_DeviceT]):
def unique_id(self) -> str:
"""Return a unique identifier for this device."""
assert isinstance(self._device, PydeconzDevice)
if self.unique_id_suffix is not None:
return f"{self._device.unique_id}-{self.unique_id_suffix}"
return self._device.unique_id
@property
@ -78,6 +82,10 @@ class DeconzDevice(DeconzBase[_DeviceT], Entity):
_attr_should_poll = False
_name_suffix: str | None = None
_update_key: str | None = None
_update_keys: set[str] | None = None
TYPE = ""
def __init__(
@ -90,6 +98,13 @@ class DeconzDevice(DeconzBase[_DeviceT], Entity):
self.gateway.entities[self.TYPE].add(self.unique_id)
self._attr_name = self._device.name
if self._name_suffix is not None:
self._attr_name += f" {self._name_suffix}"
if self._update_key is not None:
self._update_keys = {self._update_key}
if self._update_keys is not None:
self._update_keys |= {"reachable"}
async def async_added_to_hass(self) -> None:
"""Subscribe to device events."""
@ -120,6 +135,12 @@ class DeconzDevice(DeconzBase[_DeviceT], Entity):
if self.gateway.ignore_state_updates:
return
if (
self._update_keys is not None
and not self._device.changed_keys.intersection(self._update_keys)
):
return
self.async_write_ha_state()
@property

View file

@ -94,17 +94,10 @@ class DeconzNumber(DeconzDevice[Presence], NumberEntity):
) -> None:
"""Initialize deCONZ number entity."""
self.entity_description: DeconzNumberDescription = description
self._update_key = self.entity_description.update_key
self._name_suffix = description.suffix
super().__init__(device, gateway)
self._attr_name = f"{device.name} {description.suffix}"
self._update_keys = {self.entity_description.update_key, "reachable"}
@callback
def async_update_callback(self) -> None:
"""Update the number value."""
if self._device.changed_keys.intersection(self._update_keys):
super().async_update_callback()
@property
def native_value(self) -> float | None:
"""Return the value of the sensor property."""

View file

@ -287,13 +287,15 @@ class DeconzSensor(DeconzDevice[SensorResources], SensorEntity):
) -> None:
"""Initialize deCONZ sensor."""
self.entity_description = description
self._update_key = description.update_key
if description.suffix:
self._name_suffix = description.suffix
super().__init__(device, gateway)
if description.suffix:
self._attr_name = f"{device.name} {description.suffix}"
self._update_keys = {description.update_key, "reachable"}
if self.entity_description.key in PROVIDES_EXTRA_ATTRIBUTES:
if (
self.entity_description.key in PROVIDES_EXTRA_ATTRIBUTES
and self._update_keys is not None
):
self._update_keys.update({"on", "state"})
@property
@ -315,12 +317,6 @@ class DeconzSensor(DeconzDevice[SensorResources], SensorEntity):
return f"{self.serial}-{self.entity_description.suffix.lower()}"
return super().unique_id
@callback
def async_update_callback(self) -> None:
"""Update the sensor's state."""
if self._device.changed_keys.intersection(self._update_keys):
super().async_update_callback()
@property
def native_value(self) -> StateType | datetime:
"""Return the state of the sensor."""