78 lines
2.9 KiB
Python
78 lines
2.9 KiB
Python
"""Device registry synchronization for MB-Secure."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from homeassistant.config_entries import ConfigEntry
|
|
from homeassistant.core import HomeAssistant, callback
|
|
from homeassistant.helpers import device_registry as dr
|
|
|
|
from .const import DOMAIN
|
|
from .coordinator import MBSecureCoordinator
|
|
|
|
|
|
def _identifier(bridge_id: str, object_type: str, object_id: str) -> tuple[str, str]:
|
|
return (DOMAIN, f"{bridge_id}:{object_type}:{object_id}")
|
|
|
|
|
|
def async_setup_devices(
|
|
hass: HomeAssistant, entry: ConfigEntry[MBSecureCoordinator]
|
|
) -> None:
|
|
"""Create and dynamically synchronize normalized bridge devices."""
|
|
coordinator = entry.runtime_data
|
|
assert coordinator.info is not None and coordinator.info.bridge_id is not None
|
|
bridge_id = coordinator.info.bridge_id
|
|
registry = dr.async_get(hass)
|
|
managed = {
|
|
identifier: device.id
|
|
for device in registry.async_get_devices(config_entry_id=entry.entry_id)
|
|
for identifier in device.identifiers
|
|
if identifier[0] == DOMAIN and identifier[1].startswith(f"{bridge_id}:")
|
|
}
|
|
|
|
@callback
|
|
def sync_devices() -> None:
|
|
desired: set[tuple[str, str]] = set()
|
|
for object_type, collection in (
|
|
("controller", coordinator.data.controllers),
|
|
("area", coordinator.data.areas),
|
|
("module", coordinator.data.modules),
|
|
):
|
|
desired.update(
|
|
_identifier(bridge_id, object_type, object_id)
|
|
for object_id in collection
|
|
)
|
|
|
|
controller_ids: dict[str, str] = {}
|
|
for object_id, item in coordinator.data.controllers.items():
|
|
identifier = _identifier(bridge_id, "controller", object_id)
|
|
device = registry.async_get_or_create(
|
|
config_entry_id=entry.entry_id,
|
|
identifiers={identifier},
|
|
name=item.name or item.id,
|
|
model="Controller",
|
|
)
|
|
managed[identifier] = device.id
|
|
controller_ids[object_id] = device.id
|
|
|
|
for object_type, collection in (
|
|
("area", coordinator.data.areas),
|
|
("module", coordinator.data.modules),
|
|
):
|
|
for object_id, item in collection.items():
|
|
assert item.controller_id is not None
|
|
identifier = _identifier(bridge_id, object_type, object_id)
|
|
device = registry.async_get_or_create(
|
|
config_entry_id=entry.entry_id,
|
|
identifiers={identifier},
|
|
name=item.name or item.id,
|
|
model=object_type.title(),
|
|
via_device_id=controller_ids[item.controller_id],
|
|
)
|
|
managed[identifier] = device.id
|
|
|
|
for identifier in managed.keys() - desired:
|
|
registry.async_remove_device(managed.pop(identifier))
|
|
|
|
sync_devices()
|
|
entry.async_on_unload(coordinator.async_add_listener(sync_devices))
|