Files
core/homeassistant/components/eheimdigital/sensor.py

118 lines
4.0 KiB
Python

"""EHEIM Digital sensors."""
from collections.abc import Callable
from dataclasses import dataclass
from typing import Any, override
from eheimdigital.classic_vario import EheimDigitalClassicVario
from eheimdigital.device import EheimDigitalDevice
from eheimdigital.types import FilterErrorCode
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
)
from homeassistant.const import PERCENTAGE, EntityCategory, UnitOfTime
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .coordinator import EheimDigitalConfigEntry, EheimDigitalUpdateCoordinator
from .entity import EheimDigitalEntity
# Coordinator is used to centralize the data updates
PARALLEL_UPDATES = 0
@dataclass(frozen=True, kw_only=True)
class EheimDigitalSensorDescription[_DeviceT: EheimDigitalDevice](
SensorEntityDescription
):
"""Class describing EHEIM Digital sensor entities."""
value_fn: Callable[[_DeviceT], float | str | None]
CLASSICVARIO_DESCRIPTIONS: tuple[
EheimDigitalSensorDescription[EheimDigitalClassicVario], ...
] = (
EheimDigitalSensorDescription[EheimDigitalClassicVario](
key="current_speed",
translation_key="current_speed",
value_fn=lambda device: device.current_speed,
native_unit_of_measurement=PERCENTAGE,
),
EheimDigitalSensorDescription[EheimDigitalClassicVario](
key="service_hours",
translation_key="service_hours",
value_fn=lambda device: device.service_hours,
device_class=SensorDeviceClass.DURATION,
native_unit_of_measurement=UnitOfTime.HOURS,
suggested_unit_of_measurement=UnitOfTime.DAYS,
entity_category=EntityCategory.DIAGNOSTIC,
),
EheimDigitalSensorDescription[EheimDigitalClassicVario](
key="error_code",
translation_key="error_code",
value_fn=(
lambda device: device.error_code.name.lower()
if device.error_code is not None
else None
),
device_class=SensorDeviceClass.ENUM,
options=[name.lower() for name in FilterErrorCode._member_names_],
entity_category=EntityCategory.DIAGNOSTIC,
),
)
async def async_setup_entry(
hass: HomeAssistant,
entry: EheimDigitalConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the callbacks for the coordinator so lights can be added as devices are found."""
coordinator = entry.runtime_data
def async_setup_device_entities(
device_address: dict[str, EheimDigitalDevice],
) -> None:
"""Set up the light entities for one or multiple devices."""
entities: list[EheimDigitalSensor[Any]] = []
for device in device_address.values():
if isinstance(device, EheimDigitalClassicVario):
entities += [
EheimDigitalSensor[EheimDigitalClassicVario](
coordinator, device, description
)
for description in CLASSICVARIO_DESCRIPTIONS
]
async_add_entities(entities)
coordinator.add_platform_callback(async_setup_device_entities)
async_setup_device_entities(coordinator.hub.devices)
class EheimDigitalSensor[_DeviceT: EheimDigitalDevice](
EheimDigitalEntity[_DeviceT], SensorEntity
):
"""Represent a EHEIM Digital sensor entity."""
entity_description: EheimDigitalSensorDescription[_DeviceT]
def __init__(
self,
coordinator: EheimDigitalUpdateCoordinator,
device: _DeviceT,
description: EheimDigitalSensorDescription[_DeviceT],
) -> None:
"""Initialize an EHEIM Digital number entity."""
super().__init__(coordinator, device)
self.entity_description = description
self._attr_unique_id = f"{self._device_address}_{description.key}"
@override
def _async_update_attrs(self) -> None:
self._attr_native_value = self.entity_description.value_fn(self._device)