152 lines
5.1 KiB
Python
152 lines
5.1 KiB
Python
"""Shared entity support for MB-Secure."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Callable
|
|
from typing import Any
|
|
|
|
from homeassistant.config_entries import ConfigEntry
|
|
from homeassistant.core import callback
|
|
from homeassistant.helpers import entity_registry as er
|
|
from homeassistant.helpers.device_registry import DeviceInfo
|
|
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
|
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
|
|
|
from .const import DOMAIN
|
|
from .coordinator import MBSecureCoordinator
|
|
from .models import BridgeObject
|
|
|
|
type EntityFactory = Callable[[str], MBSecureEntity]
|
|
|
|
|
|
async def _async_remove_entity(entity: MBSecureEntity) -> None:
|
|
"""Remove a topology entity from runtime and the entity registry."""
|
|
entity_id = entity.entity_id
|
|
await entity.async_remove(force_remove=True)
|
|
if entity_id is not None:
|
|
er.async_get(entity.coordinator.hass).async_remove(entity_id)
|
|
|
|
|
|
class MBSecureEntity(CoordinatorEntity[MBSecureCoordinator]):
|
|
"""Base class for an entity backed by one normalized bridge object."""
|
|
|
|
_attr_has_entity_name = True
|
|
|
|
def __init__(
|
|
self,
|
|
coordinator: MBSecureCoordinator,
|
|
bridge_id: str,
|
|
collection_name: str,
|
|
object_id: str,
|
|
) -> None:
|
|
"""Initialize an entity with stable technical identifiers."""
|
|
super().__init__(coordinator)
|
|
self._bridge_id = bridge_id
|
|
self._collection_name = collection_name
|
|
self._object_id = object_id
|
|
self._attr_unique_id = f"{bridge_id}:{collection_name}:{object_id}"
|
|
|
|
@property
|
|
def bridge_object(self) -> BridgeObject | None:
|
|
"""Return the latest normalized object, if it still exists."""
|
|
return getattr(self.coordinator.data, self._collection_name).get(
|
|
self._object_id
|
|
)
|
|
|
|
@property
|
|
def available(self) -> bool:
|
|
"""Report availability from both transport and topology state."""
|
|
return super().available and self.bridge_object is not None
|
|
|
|
@property
|
|
def name(self) -> str | None:
|
|
"""Return a display name without using it for identity."""
|
|
item = self.bridge_object
|
|
if self._collection_name in {"areas", "controllers", "modules"}:
|
|
return None
|
|
return item.name if item is not None and item.name else self._object_id
|
|
|
|
@property
|
|
def device_info(self) -> DeviceInfo | None:
|
|
"""Attach the entity to the normalized device hierarchy."""
|
|
item = self.bridge_object
|
|
if item is None:
|
|
return None
|
|
return _device_info(self._bridge_id, self._collection_name, item)
|
|
|
|
|
|
def _identifier(bridge_id: str, object_type: str, object_id: str) -> tuple[str, str]:
|
|
return (DOMAIN, f"{bridge_id}:{object_type}:{object_id}")
|
|
|
|
|
|
def _device_info(
|
|
bridge_id: str,
|
|
collection_name: str,
|
|
item: BridgeObject,
|
|
) -> DeviceInfo:
|
|
object_type = collection_name.removesuffix("s")
|
|
if collection_name == "controllers":
|
|
return DeviceInfo(
|
|
identifiers={_identifier(bridge_id, object_type, item.id)},
|
|
name=item.name or item.id,
|
|
model="Controller",
|
|
)
|
|
if collection_name in {"areas", "modules"}:
|
|
return DeviceInfo(
|
|
identifiers={_identifier(bridge_id, object_type, item.id)},
|
|
name=item.name or item.id,
|
|
model=object_type.title(),
|
|
)
|
|
|
|
parent_type = "controller"
|
|
parent_id = item.controller_id
|
|
if item.module_id is not None:
|
|
parent_type = "module"
|
|
parent_id = item.module_id
|
|
elif item.area_id is not None:
|
|
parent_type = "area"
|
|
parent_id = item.area_id
|
|
assert parent_id is not None
|
|
return DeviceInfo(identifiers={_identifier(bridge_id, parent_type, parent_id)})
|
|
|
|
|
|
def async_setup_dynamic_entities(
|
|
entry: ConfigEntry[Any],
|
|
async_add_entities: AddEntitiesCallback,
|
|
*,
|
|
collection_name: str,
|
|
capability: str | frozenset[str],
|
|
factory: EntityFactory,
|
|
) -> None:
|
|
"""Keep a capability-filtered platform synchronized with topology changes."""
|
|
coordinator: MBSecureCoordinator = entry.runtime_data
|
|
entities: dict[str, MBSecureEntity] = {}
|
|
|
|
@callback
|
|
def sync_entities() -> None:
|
|
collection: dict[str, BridgeObject] = getattr(coordinator.data, collection_name)
|
|
required_capabilities = (
|
|
frozenset({capability}) if isinstance(capability, str) else capability
|
|
)
|
|
desired = {
|
|
object_id
|
|
for object_id, item in collection.items()
|
|
if required_capabilities & item.capabilities
|
|
}
|
|
added = [factory(object_id) for object_id in desired - entities.keys()]
|
|
for entity in added:
|
|
entities[entity._object_id] = entity
|
|
if added:
|
|
async_add_entities(added)
|
|
|
|
for object_id in entities.keys() - desired:
|
|
entity = entities.pop(object_id)
|
|
entry.async_create_background_task(
|
|
coordinator.hass,
|
|
_async_remove_entity(entity),
|
|
f"{DOMAIN}-remove-{entity.unique_id}",
|
|
)
|
|
|
|
sync_entities()
|
|
entry.async_on_unload(coordinator.async_add_listener(sync_entities))
|