SberEntityLoader

Загрузчик сущностей моста: чтение HA registry, применение YAML-overrides, разрешение linked-сенсоров, детекция конфликтов устройств.

Извлечён из SberBridge._load_exposed_entities в v1.25.1 — метод моста превратился в 30-строчный оркестратор поверх SberEntityLoader.load(), возвращающего EntityLoadResult для атомарного swap-on-replace.

Entity loader / registry for Sber MQTT Bridge.

Extracted from SberBridge to isolate HA entity registry lookups, YAML overrides, link resolution, and conflict detection from the bridge's transport and lifecycle concerns (SRP).

Usage::

loader = SberEntityLoader(hass, entry)
result = loader.load()
# result.entities: dict[entity_id, BaseEntity]
# result.enabled_entity_ids: list[str]
# result.entity_links: dict[primary, dict[role, linked_id]]
# result.linked_reverse: dict[linked_id, (primary, role)]
# result.redefinitions: dict[entity_id, dict]

EntityLoadResult dataclass

EntityLoadResult(entities=dict(), enabled_entity_ids=list(), entity_links=dict(), linked_reverse=dict(), redefinitions=dict())

Result of a full entity reload pass.

Attributes:

Name	Type	Description
entities	dict[str, BaseEntity]	Fresh dict of entity_id → BaseEntity.
enabled_entity_ids	list[str]	Ordered list of exposed entity IDs.
entity_links	dict[str, dict[str, str]]	Mapping primary entity → {role: linked_entity_id}.
linked_reverse	dict[str, tuple[str, str]]	Reverse mapping linked_entity_id → (primary, role).
redefinitions	dict[str, dict]	Updated redefinitions dict (pruned of stale entries).

SberEntityLoader

SberEntityLoader(hass, entry)

Build Sber BaseEntity instances from HA registry and YAML config.

This class owns the read-only lookup logic that turns HA entity registry entries into Sber-friendly entity objects. It does NOT own runtime state (debouncing, MQTT, acknowledgments) — those stay in SberBridge.

Initialize the loader.

Parameters:

Name	Type	Description	Default
hass	HomeAssistant	Home Assistant core instance.	required
entry	ConfigEntry	Config entry providing options (exposed IDs, overrides).	required

Source code in custom_components/sber_mqtt_bridge/entity_registry.py

def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
    """Initialize the loader.

    Args:
        hass: Home Assistant core instance.
        entry: Config entry providing options (exposed IDs, overrides).
    """
    self._hass = hass
    self._entry = entry

load

load(existing_redefinitions=None)

Perform a full entity reload pass.

Uses a swap-on-replace pattern: callers receive a new result set and should atomically replace their previous state to avoid race conditions with concurrent readers.

Parameters:

Name	Type	Description	Default
existing_redefinitions	dict[str, dict] | None	Current in-memory redefinitions; merged with persisted options before pruning stale entries.	None

Returns:

Type	Description
EntityLoadResult	class:EntityLoadResult ready for atomic swap.

Source code in custom_components/sber_mqtt_bridge/entity_registry.py

def load(self, existing_redefinitions: dict[str, dict] | None = None) -> EntityLoadResult:
    """Perform a full entity reload pass.

    Uses a swap-on-replace pattern: callers receive a new result set
    and should atomically replace their previous state to avoid race
    conditions with concurrent readers.

    Args:
        existing_redefinitions: Current in-memory redefinitions; merged
            with persisted options before pruning stale entries.

    Returns:
        :class:`EntityLoadResult` ready for atomic swap.
    """
    result = EntityLoadResult()
    new_enabled = list(dict.fromkeys(self._entry.options.get(CONF_EXPOSED_ENTITIES, [])))
    custom_config = get_custom_config(self._hass)

    # Merge persisted + in-memory redefinitions (persisted wins when conflict
    # because it reflects user's most recent explicit edit).
    merged_redefs: dict[str, dict] = dict(existing_redefinitions or {})
    saved_redefs: dict[str, dict] = self._entry.options.get("redefinitions", {})
    if saved_redefs:
        merged_redefs.update(saved_redefs)
        _LOGGER.debug("Loaded %d persisted redefinitions from options", len(saved_redefs))

    result.entities = self._create_entities(new_enabled, custom_config)
    result.entity_links, result.linked_reverse = self._apply_entity_links(result.entities)
    self._check_device_conflicts(result.entities, result.linked_reverse)
    result.enabled_entity_ids = list(result.entities.keys())
    result.redefinitions = self._apply_room_overrides(merged_redefs, result.enabled_entity_ids, custom_config)
    # Prune stale redefinitions: keep only entries for still-enabled entities
    result.redefinitions = {k: v for k, v in result.redefinitions.items() if k in set(new_enabled)}
    _LOGGER.info(
        "Loaded %d Sber entities from %d exposed: %s",
        len(result.entities),
        len(result.enabled_entity_ids),
        ", ".join(result.enabled_entity_ids) if result.enabled_entity_ids else "(none)",
    )
    return result