mirror of https://github.com/home-assistant/core
115 lines
3.9 KiB
Python
115 lines
3.9 KiB
Python
"""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)})
|