houseplan-card/custom_components/houseplan/registry_snapshot.py

"""Small registry projection shared by import HTTP and WebSocket previews."""
from __future__ import annotations

from typing import Any

from homeassistant.core import HomeAssistant


def import_registry_snapshot(hass: HomeAssistant) -> dict[str, set[str]]:
    """Return non-sensitive target inventory used only for preview counts."""
    from homeassistant.helpers import area_registry as ar
    from homeassistant.helpers import device_registry as dr
    from homeassistant.helpers import entity_registry as er

    entities = list(er.async_get(hass).entities.values())
    active_entity: set[str] = set()
    disabled_entity: set[str] = set()
    entities_by_device: dict[str, list[Any]] = {}
    for entry in entities:
        entity_id = str(entry.entity_id)
        if getattr(entry, "disabled_by", None) is None:
            active_entity.add(entity_id)
        else:
            disabled_entity.add(entity_id)
        if entry.device_id:
            entities_by_device.setdefault(str(entry.device_id), []).append(entry)

    # Synthetic/runtime entities may legitimately have no registry row.
    active_entity.update(str(state.entity_id) for state in hass.states.async_all())

    active_device: set[str] = set()
    disabled_device: set[str] = set()
    for entry in dr.async_get(hass).devices.values():
        device_id = str(entry.id)
        children = entities_by_device.get(device_id, [])
        disabled = getattr(entry, "disabled_by", None) is not None or (
            bool(children)
            and all(getattr(child, "disabled_by", None) is not None for child in children)
        )
        (disabled_device if disabled else active_device).add(device_id)
    return {
        "active_device": active_device,
        "disabled_device": disabled_device,
        "active_entity": active_entity,
        "disabled_entity": disabled_entity - active_entity,
        "areas": {str(entry.id) for entry in ar.async_get(hass).areas.values()},
    }