"""Base class for KNX devices.""" from __future__ import annotations from typing import TYPE_CHECKING, Any from xknx.devices import Device as XknxDevice from homeassistant.const import CONF_ENTITY_CATEGORY, EntityCategory from homeassistant.helpers.device_registry import DeviceInfo from homeassistant.helpers.entity import Entity from homeassistant.helpers.entity_platform import EntityPlatform from homeassistant.helpers.entity_registry import RegistryEntry from .const import DOMAIN from .storage.config_store import PlatformControllerBase from .storage.const import CONF_DEVICE_INFO if TYPE_CHECKING: from . import KNXModule class KnxUiEntityPlatformController(PlatformControllerBase): """Class to manage dynamic adding and reloading of UI entities.""" def __init__( self, knx_module: KNXModule, entity_platform: EntityPlatform, entity_class: type[KnxUiEntity], ) -> None: """Initialize the UI platform.""" self._knx_module = knx_module self._entity_platform = entity_platform self._entity_class = entity_class async def create_entity(self, unique_id: str, config: dict[str, Any]) -> None: """Add a new UI entity.""" await self._entity_platform.async_add_entities( [self._entity_class(self._knx_module, unique_id, config)] ) async def update_entity( self, entity_entry: RegistryEntry, config: dict[str, Any] ) -> None: """Update an existing UI entities configuration.""" await self._entity_platform.async_remove_entity(entity_entry.entity_id) await self.create_entity(unique_id=entity_entry.unique_id, config=config) class _KnxEntityBase(Entity): """Representation of a KNX entity.""" _attr_should_poll = False _knx_module: KNXModule _device: XknxDevice @property def name(self) -> str: """Return the name of the KNX device.""" return self._device.name @property def available(self) -> bool: """Return True if entity is available.""" return self._knx_module.connected async def async_update(self) -> None: """Request a state update from KNX bus.""" await self._device.sync() def after_update_callback(self, _device: XknxDevice) -> None: """Call after device was updated.""" self.async_write_ha_state() async def async_added_to_hass(self) -> None: """Store register state change callback and start device object.""" self._device.register_device_updated_cb(self.after_update_callback) self._device.xknx.devices.async_add(self._device) # super call needed to have methods of multi-inherited classes called # eg. for restoring state (like _KNXSwitch) await super().async_added_to_hass() async def async_will_remove_from_hass(self) -> None: """Disconnect device object when removed.""" self._device.unregister_device_updated_cb(self.after_update_callback) self._device.xknx.devices.async_remove(self._device) class KnxYamlEntity(_KnxEntityBase): """Representation of a KNX entity configured from YAML.""" def __init__(self, knx_module: KNXModule, device: XknxDevice) -> None: """Initialize the YAML entity.""" self._knx_module = knx_module self._device = device class KnxUiEntity(_KnxEntityBase): """Representation of a KNX UI entity.""" _attr_unique_id: str _attr_has_entity_name = True def __init__( self, knx_module: KNXModule, unique_id: str, entity_config: dict[str, Any] ) -> None: """Initialize the UI entity.""" self._knx_module = knx_module self._attr_unique_id = unique_id if entity_category := entity_config.get(CONF_ENTITY_CATEGORY): self._attr_entity_category = EntityCategory(entity_category) if device_info := entity_config.get(CONF_DEVICE_INFO): self._attr_device_info = DeviceInfo(identifiers={(DOMAIN, device_info)})