jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-17 18:27:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

Improve deCONZ sensor classes (#79137)

Browse Source
This commit is contained in:
Robert Svensson 2022-09-29 03:45:15 +02:00 committed by GitHub
parent 768b83139f
commit 473d7c484d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 163 additions and 171 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

220
homeassistant/components/deconz/sensor.py
View File

@ -1,12 +1,15 @@
"""Support for deCONZ sensors.""" """Support for deCONZ sensors."""
from __future__ import annotations from __future__ import annotations
from collections.abc import Callable from collections.abc import Callable
from dataclasses import dataclass from dataclasses import dataclass
from datetime import datetime from datetime import datetime
from typing import Generic, TypeVar
from pydeconz.interfaces.sensors import SensorResources from pydeconz.interfaces.sensors import SensorResources
from pydeconz.models.event import EventType from pydeconz.models.event import EventType
from pydeconz.models.sensor import SensorBase as PydeconzSensorBase
from pydeconz.models.sensor.air_quality import AirQuality from pydeconz.models.sensor.air_quality import AirQuality
from pydeconz.models.sensor.consumption import Consumption from pydeconz.models.sensor.consumption import Consumption
from pydeconz.models.sensor.daylight import DAYLIGHT_STATUS, Daylight from pydeconz.models.sensor.daylight import DAYLIGHT_STATUS, Daylight
@ -67,171 +70,163 @@ ATTR_DAYLIGHT = "daylight"
ATTR_EVENT_ID = "event_id" ATTR_EVENT_ID = "event_id"
T = TypeVar(
"T",
AirQuality,
Consumption,
Daylight,
GenericStatus,
Humidity,
LightLevel,
Power,
Pressure,
Temperature,
Time,
PydeconzSensorBase,
)
@dataclass @dataclass
class DeconzSensorDescriptionMixin: class DeconzSensorDescriptionMixin(Generic[T]):
"""Required values when describing secondary sensor attributes.""" """Required values when describing secondary sensor attributes."""
isinstance_fn: Callable[[T], bool]
update_key: str update_key: str
value_fn: Callable[[SensorResources], float | int | str | None] value_fn: Callable[[T], datetime | StateType]
@dataclass @dataclass
class DeconzSensorDescription( class DeconzSensorDescription(
SensorEntityDescription, SensorEntityDescription, DeconzSensorDescriptionMixin[T], Generic[T]
DeconzSensorDescriptionMixin,
): ):
"""Class describing deCONZ binary sensor entities.""" """Class describing deCONZ binary sensor entities."""
suffix: str = "" common: bool = False
name_suffix: str = ""
old_unique_id_suffix: str = ""
ENTITY_DESCRIPTIONS = { ENTITY_DESCRIPTIONS: tuple[DeconzSensorDescription, ...] = (
AirQuality: [ DeconzSensorDescription[AirQuality](
DeconzSensorDescription(
key="air_quality", key="air_quality",
value_fn=lambda device: device.air_quality isinstance_fn=lambda device: isinstance(device, AirQuality),
if isinstance(device, AirQuality) value_fn=lambda device: device.air_quality,
else None,
update_key="airquality", update_key="airquality",
state_class=SensorStateClass.MEASUREMENT, state_class=SensorStateClass.MEASUREMENT,
), ),
DeconzSensorDescription( DeconzSensorDescription[AirQuality](
key="air_quality_ppb", key="air_quality_ppb",
value_fn=lambda device: device.air_quality_ppb isinstance_fn=lambda device: isinstance(device, AirQuality),
if isinstance(device, AirQuality) value_fn=lambda device: device.air_quality_ppb,
else None,
suffix="PPB",
update_key="airqualityppb", update_key="airqualityppb",
name_suffix="PPB",
old_unique_id_suffix="ppb",
device_class=SensorDeviceClass.AQI, device_class=SensorDeviceClass.AQI,
state_class=SensorStateClass.MEASUREMENT, state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=CONCENTRATION_PARTS_PER_BILLION, native_unit_of_measurement=CONCENTRATION_PARTS_PER_BILLION,
), ),
], DeconzSensorDescription[Consumption](
Consumption: [
DeconzSensorDescription(
key="consumption", key="consumption",
value_fn=lambda device: device.scaled_consumption isinstance_fn=lambda device: isinstance(device, Consumption),
if isinstance(device, Consumption) and isinstance(device.consumption, int) value_fn=lambda device: device.scaled_consumption,
else None,
update_key="consumption", update_key="consumption",
device_class=SensorDeviceClass.ENERGY, device_class=SensorDeviceClass.ENERGY,
state_class=SensorStateClass.TOTAL_INCREASING, state_class=SensorStateClass.TOTAL_INCREASING,
native_unit_of_measurement=ENERGY_KILO_WATT_HOUR, native_unit_of_measurement=ENERGY_KILO_WATT_HOUR,
) ),
], DeconzSensorDescription[Daylight](
Daylight: [
DeconzSensorDescription(
key="daylight_status", key="daylight_status",
value_fn=lambda device: DAYLIGHT_STATUS[device.daylight_status] isinstance_fn=lambda device: isinstance(device, Daylight),
if isinstance(device, Daylight) value_fn=lambda device: DAYLIGHT_STATUS[device.daylight_status],
else None,
update_key="status", update_key="status",
icon="mdi:white-balance-sunny", icon="mdi:white-balance-sunny",
entity_registry_enabled_default=False, entity_registry_enabled_default=False,
) ),
], DeconzSensorDescription[GenericStatus](
GenericStatus: [
DeconzSensorDescription(
key="status", key="status",
value_fn=lambda device: device.status isinstance_fn=lambda device: isinstance(device, GenericStatus),
if isinstance(device, GenericStatus) value_fn=lambda device: device.status,
else None,
update_key="status", update_key="status",
) ),
], DeconzSensorDescription[Humidity](
Humidity: [
DeconzSensorDescription(
key="humidity", key="humidity",
value_fn=lambda device: device.scaled_humidity isinstance_fn=lambda device: isinstance(device, Humidity),
if isinstance(device, Humidity) and isinstance(device.humidity, int) value_fn=lambda device: device.scaled_humidity,
else None,
update_key="humidity", update_key="humidity",
device_class=SensorDeviceClass.HUMIDITY, device_class=SensorDeviceClass.HUMIDITY,
state_class=SensorStateClass.MEASUREMENT, state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE, native_unit_of_measurement=PERCENTAGE,
) ),
], DeconzSensorDescription[LightLevel](
LightLevel: [
DeconzSensorDescription(
key="light_level", key="light_level",
value_fn=lambda device: device.scaled_light_level isinstance_fn=lambda device: isinstance(device, LightLevel),
if isinstance(device, LightLevel) and isinstance(device.light_level, int) value_fn=lambda device: device.scaled_light_level,
else None,
update_key="lightlevel", update_key="lightlevel",
device_class=SensorDeviceClass.ILLUMINANCE, device_class=SensorDeviceClass.ILLUMINANCE,
state_class=SensorStateClass.MEASUREMENT, state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=LIGHT_LUX, native_unit_of_measurement=LIGHT_LUX,
) ),
], DeconzSensorDescription[Power](
Power: [
DeconzSensorDescription(
key="power", key="power",
value_fn=lambda device: device.power if isinstance(device, Power) else None, isinstance_fn=lambda device: isinstance(device, Power),
value_fn=lambda device: device.power,
update_key="power", update_key="power",
device_class=SensorDeviceClass.POWER, device_class=SensorDeviceClass.POWER,
state_class=SensorStateClass.MEASUREMENT, state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=POWER_WATT, native_unit_of_measurement=POWER_WATT,
) ),
], DeconzSensorDescription[Pressure](
Pressure: [
DeconzSensorDescription(
key="pressure", key="pressure",
value_fn=lambda device: device.pressure isinstance_fn=lambda device: isinstance(device, Pressure),
if isinstance(device, Pressure) value_fn=lambda device: device.pressure,
else None,
update_key="pressure", update_key="pressure",
device_class=SensorDeviceClass.PRESSURE, device_class=SensorDeviceClass.PRESSURE,
state_class=SensorStateClass.MEASUREMENT, state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PRESSURE_HPA, native_unit_of_measurement=PRESSURE_HPA,
) ),
], DeconzSensorDescription[Temperature](
Temperature: [
DeconzSensorDescription(
key="temperature", key="temperature",
value_fn=lambda device: device.scaled_temperature isinstance_fn=lambda device: isinstance(device, Temperature),
if isinstance(device, Temperature) and isinstance(device.temperature, int) value_fn=lambda device: device.scaled_temperature,
else None,
update_key="temperature", update_key="temperature",
device_class=SensorDeviceClass.TEMPERATURE, device_class=SensorDeviceClass.TEMPERATURE,
state_class=SensorStateClass.MEASUREMENT, state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=TEMP_CELSIUS, native_unit_of_measurement=TEMP_CELSIUS,
) ),
], DeconzSensorDescription[Time](
Time: [
DeconzSensorDescription(
key="last_set", key="last_set",
value_fn=lambda device: device.last_set isinstance_fn=lambda device: isinstance(device, Time),
if isinstance(device, Time) value_fn=lambda device: dt_util.parse_datetime(device.last_set),
else None,
update_key="lastset", update_key="lastset",
device_class=SensorDeviceClass.TIMESTAMP, device_class=SensorDeviceClass.TIMESTAMP,
state_class=SensorStateClass.TOTAL_INCREASING, state_class=SensorStateClass.TOTAL_INCREASING,
) ),
], DeconzSensorDescription[SensorResources](
}
COMMON_SENSOR_DESCRIPTIONS = [
DeconzSensorDescription(
key="battery", key="battery",
isinstance_fn=lambda device: isinstance(device, PydeconzSensorBase),
value_fn=lambda device: device.battery, value_fn=lambda device: device.battery,
suffix="Battery",
update_key="battery", update_key="battery",
common=True,
name_suffix="Battery",
old_unique_id_suffix="battery",
device_class=SensorDeviceClass.BATTERY, device_class=SensorDeviceClass.BATTERY,
state_class=SensorStateClass.MEASUREMENT, state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE, native_unit_of_measurement=PERCENTAGE,
entity_category=EntityCategory.DIAGNOSTIC, entity_category=EntityCategory.DIAGNOSTIC,
), ),
DeconzSensorDescription( DeconzSensorDescription[SensorResources](
key="internal_temperature", key="internal_temperature",
isinstance_fn=lambda device: isinstance(device, PydeconzSensorBase),
value_fn=lambda device: device.internal_temperature, value_fn=lambda device: device.internal_temperature,
suffix="Temperature",
update_key="temperature", update_key="temperature",
common=True,
name_suffix="Temperature",
old_unique_id_suffix="temperature",
device_class=SensorDeviceClass.TEMPERATURE, device_class=SensorDeviceClass.TEMPERATURE,
state_class=SensorStateClass.MEASUREMENT, state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=TEMP_CELSIUS, native_unit_of_measurement=TEMP_CELSIUS,
), ),
] )
@callback @callback
@ -248,8 +243,8 @@ def async_update_unique_id(
if ent_reg.async_get_entity_id(DOMAIN, DECONZ_DOMAIN, new_unique_id): if ent_reg.async_get_entity_id(DOMAIN, DECONZ_DOMAIN, new_unique_id):
return return
if description.suffix: if description.old_unique_id_suffix:
unique_id = f'{unique_id.split("-", 1)[0]}-{description.suffix.lower()}' unique_id = f'{unique_id.split("-", 1)[0]}-{description.old_unique_id_suffix}'
if entity_id := ent_reg.async_get_entity_id(DOMAIN, DECONZ_DOMAIN, unique_id): 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) ent_reg.async_update_entity(entity_id, new_unique_id=new_unique_id)
@ -265,7 +260,9 @@ async def async_setup_entry(
gateway.entities[DOMAIN] = set() gateway.entities[DOMAIN] = set()
known_device_entities: dict[str, set[str]] = { known_device_entities: dict[str, set[str]] = {
description.key: set() for description in COMMON_SENSOR_DESCRIPTIONS description.key: set()
for description in ENTITY_DESCRIPTIONS
if description.common
} }
@callback @callback
@ -274,17 +271,15 @@ async def async_setup_entry(
sensor = gateway.api.sensors[sensor_id] sensor = gateway.api.sensors[sensor_id]
entities: list[DeconzSensor] = [] entities: list[DeconzSensor] = []
for description in ( for description in ENTITY_DESCRIPTIONS:
ENTITY_DESCRIPTIONS.get(type(sensor), []) + COMMON_SENSOR_DESCRIPTIONS if not description.isinstance_fn(sensor):
): continue
no_sensor_data = False no_sensor_data = False
if ( if description.value_fn(sensor) is None:
not hasattr(sensor, description.key)
or description.value_fn(sensor) is None
):
no_sensor_data = True no_sensor_data = True
if description in COMMON_SENSOR_DESCRIPTIONS: if description.common:
if ( if (
sensor.type.startswith("CLIP") sensor.type.startswith("CLIP")
or (no_sensor_data and description.key != "battery") or (no_sensor_data and description.key != "battery")
@ -296,7 +291,10 @@ async def async_setup_entry(
continue continue
known_device_entities[description.key].add(unique_id) known_device_entities[description.key].add(unique_id)
if no_sensor_data and description.key == "battery": if no_sensor_data and description.key == "battery":
DeconzBatteryTracker(sensor_id, gateway, async_add_entities) async_update_unique_id(hass, sensor.unique_id, description)
DeconzBatteryTracker(
sensor_id, gateway, description, async_add_entities
)
continue continue
if no_sensor_data: if no_sensor_data:
@ -327,9 +325,10 @@ class DeconzSensor(DeconzDevice[SensorResources], SensorEntity):
) -> None: ) -> None:
"""Initialize deCONZ sensor.""" """Initialize deCONZ sensor."""
self.entity_description = description self.entity_description = description
self.unique_id_suffix = description.key
self._update_key = description.update_key self._update_key = description.update_key
if description.suffix: if description.name_suffix:
self._name_suffix = description.suffix self._name_suffix = description.name_suffix
super().__init__(device, gateway) super().__init__(device, gateway)
if ( if (
@ -338,18 +337,9 @@ class DeconzSensor(DeconzDevice[SensorResources], SensorEntity):
): ):
self._update_keys.update({"on", "state"}) self._update_keys.update({"on", "state"})
@property
def unique_id(self) -> str:
"""Return a unique identifier for this device."""
return f"{self._device.unique_id}-{self.entity_description.key}"
@property @property
def native_value(self) -> StateType | datetime: def native_value(self) -> StateType | datetime:
"""Return the state of the sensor.""" """Return the state of the sensor."""
if self.entity_description.device_class is SensorDeviceClass.TIMESTAMP:
value = self.entity_description.value_fn(self._device)
assert isinstance(value, str)
return dt_util.parse_datetime(value)
return self.entity_description.value_fn(self._device) return self.entity_description.value_fn(self._device)
@property @property
@ -399,19 +389,21 @@ class DeconzBatteryTracker:
self, self,
sensor_id: str, sensor_id: str,
gateway: DeconzGateway, gateway: DeconzGateway,
description: DeconzSensorDescription,
async_add_entities: AddEntitiesCallback, async_add_entities: AddEntitiesCallback,
) -> None: ) -> None:
"""Set up tracker.""" """Set up tracker."""
self.sensor = gateway.api.sensors[sensor_id] self.sensor = gateway.api.sensors[sensor_id]
self.gateway = gateway self.gateway = gateway
self.description = description
self.async_add_entities = async_add_entities self.async_add_entities = async_add_entities
self.unsubscribe = self.sensor.subscribe(self.async_update_callback) self.unsubscribe = self.sensor.subscribe(self.async_update_callback)
@callback @callback
def async_update_callback(self) -> None: def async_update_callback(self) -> None:
"""Update the device's state.""" """Update the device's state."""
if "battery" in self.sensor.changed_keys: if self.description.update_key in self.sensor.changed_keys:
self.unsubscribe() self.unsubscribe()
desc = COMMON_SENSOR_DESCRIPTIONS[0] self.async_add_entities(
async_update_unique_id(self.gateway.hass, self.sensor.unique_id, desc) [DeconzSensor(self.sensor, self.gateway, self.description)]
self.async_add_entities([DeconzSensor(self.sensor, self.gateway, desc)]) )