"""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)