SberBridge

Ядро интеграции: управление MQTT-соединением к Sber Cloud, диспетчеризация команд, отслеживание состояний HA-устройств.

Sber Smart Home MQTT Bridge - core bridge logic.

Manages: - Async MQTT connection to Sber cloud broker (aiomqtt) - HA state change listening and publishing to Sber - Sber command reception and forwarding to HA services - Connection health monitoring and device acknowledgment tracking

RECONNECT_GRACE_TIMEOUT module-attribute

RECONNECT_GRACE_TIMEOUT = 30.0

Maximum seconds to wait for Sber acknowledgment after (re)connect.

After a reconnect, the bridge publishes HA states and waits for Sber to acknowledge them (via status_request or config_request) before accepting commands. This timeout is a fallback in case Sber never sends a request.

DEFERRED_CONFIRM_SLOT_SUFFIX module-attribute

DEFERRED_CONFIRM_SLOT_SUFFIX = '#deferred'

Suffix of the _confirm_tasks slot holding an entity-requested republish.

The default confirm is keyed by the bare entity id (unchanged), so an entity that asks for a second, later publish through pending_confirm_delay gets its own slot instead of cancelling the short one. # cannot occur in an HA entity id, so the two namespaces can never collide.

LOG_PAYLOAD_MAX_CHARS module-attribute

LOG_PAYLOAD_MAX_CHARS = 8192

Maximum characters of a payload stored in the DevTools message log.

Payloads may legally be up to max_payload_size (1 MB by default), but the DevTools ring buffer keeps message_log_size entries and pushes each one synchronously to every WebSocket subscriber. Storing full payloads would bound memory at maxlen * max_payload_size (~50 MB with defaults); truncating each stored copy to this limit bounds it at a few hundred KB. Only the DevTools copy is truncated — real MQTT traffic and command handling always see the full payload.

BridgeStats dataclass

BridgeStats(connected_since=None, messages_received=0, messages_sent=0, commands_received=0, config_requests=0, status_requests=0, errors_from_sber=0, publish_errors=0, reconnect_count=0, acknowledged_entities=set(), collectively_acked_entities=set(), last_error_detail='', validation_failures=list())

Connection statistics and health metrics for the Sber MQTT bridge.

connected_since class-attribute instance-attribute

connected_since = None

Timestamp when the current connection was established.

messages_received class-attribute instance-attribute

messages_received = 0

Total MQTT messages received from Sber.

messages_sent class-attribute instance-attribute

messages_sent = 0

Total MQTT messages published to Sber.

commands_received class-attribute instance-attribute

commands_received = 0

Total Sber commands processed.

config_requests class-attribute instance-attribute

config_requests = 0

Total config requests received from Sber.

status_requests class-attribute instance-attribute

status_requests = 0

Total status requests received from Sber.

errors_from_sber class-attribute instance-attribute

errors_from_sber = 0

Total error messages received from Sber.

publish_errors class-attribute instance-attribute

publish_errors = 0

Total failed publish attempts.

reconnect_count class-attribute instance-attribute

reconnect_count = 0

Total number of reconnections since startup.

acknowledged_entities class-attribute instance-attribute

acknowledged_entities = field(default_factory=set)

Entity IDs that Sber has acknowledged (via status_request or command).

collectively_acked_entities class-attribute instance-attribute

collectively_acked_entities = field(default_factory=set)

Subset of :attr:acknowledged_entities marked without being named.

A status_request carrying no device list means "send me the state of everything you have". It is a real acknowledgement — the cloud is talking to this hub — but it is a collective one: it names nobody, so it cannot vouch for any individual device.

Keeping the two strengths apart is what lets the same signal answer two different questions honestly. "Confirmed this session" (the panel counter) legitimately counts a collective ack, while the silent-rejection alarm (:attr:~SberBridge.never_confirmed_entities) must not: a device Sber silently rejected is still covered by "state of everything", so folding the two together made the alarm unable to fire at all — the user saw "confirmed: 36 / never confirmed: 0" on a bridge whose registry knew nothing (issue #57).

An id leaves this set as soon as the cloud names it individually (a command, or a status_request listing it): the weak mark is then superseded by real per-device evidence.

last_error_detail class-attribute instance-attribute

last_error_detail = ''

Human-readable detail of the last error message from Sber cloud.

validation_failures class-attribute instance-attribute

validation_failures = field(default_factory=list)

Entity IDs that failed pydantic validation and were excluded from last config.

as_dict

as_dict()

Return stats as a serializable dict.

Source code in custom_components/sber_mqtt_bridge/sber_bridge.py

def as_dict(self) -> dict:
    """Return stats as a serializable dict."""
    now = time.monotonic()
    return {
        "connected_since": self.connected_since,
        "connection_uptime_seconds": (round(now - self.connected_since, 1) if self.connected_since else None),
        "messages_received": self.messages_received,
        "messages_sent": self.messages_sent,
        "commands_received": self.commands_received,
        "config_requests": self.config_requests,
        "status_requests": self.status_requests,
        "errors_from_sber": self.errors_from_sber,
        "publish_errors": self.publish_errors,
        "reconnect_count": self.reconnect_count,
        "acknowledged_entities": sorted(self.acknowledged_entities),
        "last_error_detail": self.last_error_detail,
        "validation_failures": list(self.validation_failures),
    }

SberBridge

SberBridge(hass, entry)

Bridge between Home Assistant and Sber Smart Home MQTT cloud.

Initialize the bridge.

Source code in custom_components/sber_mqtt_bridge/sber_bridge.py

def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
    """Initialize the bridge."""
    self._hass = hass
    self._entry = entry

    self._login: str = entry.data[CONF_SBER_LOGIN]
    self._password: str = entry.data[CONF_SBER_PASSWORD]
    self._broker: str = entry.data[CONF_SBER_BROKER]
    self._port: int = entry.data[CONF_SBER_PORT]
    self._verify_ssl: bool = entry.options.get(CONF_SBER_VERIFY_SSL, entry.data.get(CONF_SBER_VERIFY_SSL, True))

    self._root_topic = f"{SBER_TOPIC_PREFIX}/{self._login}"
    self._down_topic = f"{self._root_topic}/down"

    self._ha_instance_id_prefix: str = ""
    """Cached 8-char prefix of HA instance UUID; populated in ``async_start``."""
    self._entities: dict[str, BaseEntity] = {}
    self._enabled_entity_ids: list[str] = []
    # Persisted redefinitions delegated to RedefinitionsStore (v1.38.4).
    self._redef_store = RedefinitionsStore(hass, entry)
    self._entity_links: dict[str, dict[str, str]] = {}
    """Primary entity → {role: linked_entity_id}."""
    self._linked_reverse: dict[str, tuple[str, str]] = {}
    """Linked entity_id → (primary_entity_id, role)."""
    self._entity_loader = SberEntityLoader(hass, entry)

    # NOTE: connection state (_connected / _mqtt_client) is NOT stored
    # here — MqttClientService is the single owner; the bridge exposes
    # read/write forwarding properties below for compatibility.
    self._connection_task: asyncio.Task | None = None
    self._running = False

    # Configurable operational settings loaded from ``config_entry.options``.
    # All defaults live in ``SETTINGS_DEFAULTS`` (const.py) — this avoids
    # scattered ``opts.get(key, hardcoded_default)`` calls and keeps the
    # canonical values in exactly one place (DRY).
    self._load_settings_from_options(entry.options)

    self._unsub_lifecycle_listeners: list[Callable] = []

    self._stats = BridgeStats()

    # DevTools collector aggregate (message log, traces, diff, validation).
    # Built early: both the publisher and the dispatcher receive it as an
    # explicit dependency rather than reaching back through the bridge.
    self._devtools = DevToolsHub(message_log_size=self._message_log_size)

    # Delayed confirm tasks per entity (dedup: cancel previous on new command)
    self._confirm_tasks: dict[str, asyncio.Task] = {}

    # MQTT transport service: owns reconnect loop + publish + subscribe
    self._mqtt_service = MqttClientService(
        hass=hass,
        credentials=SberMqttCredentials(
            login=self._login,
            password=self._password,
            broker=self._broker,
            port=self._port,
            verify_ssl=self._verify_ssl,
        ),
        hooks=MqttServiceHooks(
            on_message=self._handle_mqtt_message,
            on_connected=self._handle_mqtt_connected,
            on_disconnected=self._handle_mqtt_disconnected,
        ),
        reconnect_min=self._reconnect_min,
        reconnect_max=self._reconnect_max,
    )

    # Ack audit owns the reconnect guard AND the silent-rejection
    # scheduler in one place — see ``ack_audit.py`` for the rationale.
    from .ack_audit import AckAudit

    self._ack_audit = AckAudit(
        hass,
        grace_timeout=RECONNECT_GRACE_TIMEOUT,
        audit_delay=self._ack_audit_delay,
        on_audit=self._run_ack_audit,
    )

    # Publish coordinator owns the three Sber publish flows and the
    # last-config timestamp; bridge keeps thin delegators below.
    self._publisher = SberPublisher(
        PublisherDeps(
            root_topic=self._root_topic,
            stats=self._stats,
            devtools=self._devtools,
            is_connected=self._is_transport_ready,
            publish=self._publish_via_transport,
            log_message=self._log_message,
            get_entities=lambda: self._entities,
            get_enabled_entity_ids=lambda: self._enabled_entity_ids,
            get_redefinitions=lambda: self._redef_store.raw,
            get_config_context=self._build_config_publish_context,
            on_config_published=self._on_config_published,
        )
    )

    # Gate: delay initial MQTT publish until HA is fully started so that
    # entity states (and therefore Sber features) are fully populated.
    self._ha_ready = asyncio.Event()

    # What the Sber cloud currently holds — the floor a publish must not
    # go below.  Fed by our own publishes and by the device list Sber
    # names in every status_request (issue #44).
    self._cloud_devices = CloudDeviceRegistry(hass, entry)

    # Coalescing gate in front of up/config: Sber treats every config
    # payload as the complete device list, so a partial one (entities
    # still loading) makes it drop and later re-create devices, losing
    # their room.  See ConfigPublishGate (issue #44).
    self._config_gate = ConfigPublishGate(
        loop=hass.loop,
        settle_delay=self._config_settle_delay,
        max_wait=self._config_max_wait,
        get_enabled_entity_ids=self._config_relevant_entity_ids,
        get_ready_entity_ids=self._ready_entity_ids,
        get_cloud_known_ids=lambda: self._cloud_devices.known,
        publish=self._publish_config,
        create_task=self._create_safe_task,
    )

    # HA → Sber event forwarder: owns state-change subscription + debouncing
    self._state_forwarder = HaStateForwarder(
        hass=hass,
        debounce_delay=self._debounce_delay,
        get_entities=lambda: self._entities,
        get_linked_reverse=lambda: self._linked_reverse,
        on_publish_states=self._publish_states,
        on_republish_config=self._request_config_publish,
        create_safe_task=self._create_safe_task,
        on_trace_state_change=self._trace_on_state_change,
        on_state_settled=self._sync_deferred_confirm,
    )

    # Sber protocol command dispatcher (commands, status/config request, etc.)
    self._command_dispatcher = SberCommandDispatcher(
        DispatcherDeps(
            hass=hass,
            stats=self._stats,
            ack_audit=self._ack_audit,
            publisher=self._publisher,
            redefinitions=self._redef_store,
            devtools=self._devtools,
            get_entities=lambda: self._entities,
            get_enabled_entity_ids=lambda: self._enabled_entity_ids,
            schedule_confirm=self.schedule_confirm,
            note_cloud_reported=self._cloud_devices.note_cloud_reported,
            refresh_repair_issues=self.refresh_repair_issues,
        )
    )

config_publish_context property

config_publish_context

Return the descriptor context the next config publish will use.

Public so the DevTools "Raw config" preview can render exactly what would go on the wire. Previously the preview called the payload builder without these arguments and silently got the builder's own defaults, so it always showed parent_id: "root" no matter how hub_auto_parent_id was set — reported as "the setting is not applied to the config" (issue #44).

is_connected property

is_connected

Return True if connected to Sber MQTT (owned by MqttClientService).

config_entry property

config_entry

Return the bridge's owning HA config entry (read-only access).

connection_phase property

connection_phase

Return the current connection lifecycle phase.

Phases

starting — HA not fully loaded, waiting for integrations. connecting — MQTT connection in progress. awaiting_ack — connected, published config, waiting for Sber to acknowledge. ready — fully operational, accepting commands. disconnected — not connected to MQTT broker.

entities_count property

entities_count

Return the number of loaded Sber entities.

entities property

entities

Return the dict of loaded Sber entities (read-only view).

enabled_entity_ids property

enabled_entity_ids

Return a copy of the enabled entity ID list.

redefinitions property

redefinitions

Return a copy of the entity redefinitions mapping.

Values are per-entity dicts with optional name / room / home keys (see :class:RedefinitionsStore).

entity_links property

entity_links

Return the current entity link map.

linked_entity_ids property

linked_entity_ids

Return set of all linked entity IDs (not primary).

stats property

stats

Return bridge statistics as a serializable dict.

ha_serial_prefix property

ha_serial_prefix

Return active per-HA serial prefix, or None when feature is off.

unacknowledged_entities property

unacknowledged_entities

Return entity IDs Sber has not spoken about in this session.

Note what this is not: evidence that the cloud rejected the device. The acknowledgement mark is set when Sber sends a command or a status_request for the entity, and it lives in memory, so every restart empties it. The cloud has no idea we restarted and no reason to speak up immediately — it will ask for state when the user opens the Salute app, issues a voice command, or its own poll comes round. Right after a restart this list therefore contains everything, which says nothing about registration (issue #57).

Use :attr:never_confirmed_entities for "something is actually wrong"; this property answers the narrower question of what has been confirmed since the bridge came up.

cloud_known_entities property

cloud_known_entities

Return exposed entity IDs the cloud is believed to hold.

Backed by :class:~cloud_device_registry.CloudDeviceRegistry, which persists into ConfigEntry.options and therefore survives a restart. This is the closest thing to "the cloud accepted it" that the protocol allows: there is no way to ask Sber what it holds — we only publish on up/config / up/status and learn from the ids it names in down/status_request.

cloud_device_registry_state property

cloud_device_registry_state

Return the raw cloud-device registry state, for diagnostics.

Deliberately unfiltered and paired with what is on disk: when the panel reports "known to Sber: 0" the question is which of the two is empty — the live set, the persisted key, or neither (in which case the panel is at fault). Answering that from a diagnostics download is the whole point; issue #57 was diagnosed by guesswork because none of this was in the dump.

Returns:

Type	Description
dict[str, Any]	Mapping with the in-memory set (known), the value persisted
dict[str, Any]	in ConfigEntry.options (persisted), the exposed subset
dict[str, Any]	the panel shows (known_exposed), and whether a config
dict[str, Any]	publish has succeeded since this bridge came up.

never_confirmed_entities property

never_confirmed_entities

Return exposed entities the cloud has never been seen to know.

Neither confirmed in this session nor remembered from an earlier one. Unlike :attr:unacknowledged_entities this does not light up after every restart, so it is the list worth alerting on: a device published repeatedly that the cloud never once asks about is the signature of a silent rejection.

Only named evidence counts here. A bare status_request acknowledges every exposed entity collectively (see :attr:SberStats.collectively_acked_entities), and counting that as per-device confirmation disarmed the alarm permanently: a silently rejected device is still covered by "send me the state of everything", so it looked confirmed forever (issue #57).

entities_missing_required_links property

entities_missing_required_links

Return loaded composite entities whose required links are unmapped.

A class with a non-empty :attr:~devices.base_entity.BaseEntity.REQUIRED_LINK_ROLES cannot publish a truthful state without its companion — an impulse gate without a reed contact reports close forever. The wizard refuses to create such a device, but "add the entity, then set the category by hand" bypasses that check, so the half-configured device has to stay visible instead of silent: this property feeds the HA repair issue, diagnostics and the panel's device dialog.

Returns:

Type	Description
dict[str, list[str]]	entity_id → unmapped role names (declaration order), empty
dict[str, list[str]]	when every composite device is fully linked.

message_log property

message_log

Return the DevTools outbound-message ring buffer (delegates to hub).

trace_collector property

trace_collector

Return the correlation-trace collector (delegates to hub).

diff_collector property

diff_collector

Return the state-diff collector (delegates to hub).

validation_collector property

validation_collector

Return the schema-validation collector (delegates to hub).

schedule_confirm

schedule_confirm(entity_id)

(Re)arm the delayed state confirm(s) for one commanded entity.

Always arms the short confirm that lets HA settle its async attribute updates (:attr:_confirm_delay). An entity may ask for a second, later republish through a pending_confirm_delay attribute — the impulse gate uses it to replace its emulated opening / closing value once the leaf's travel time is over (see :class:~devices.gate.ImpulseGateEntity). Both go through the very same :meth:_delayed_confirm machinery, just in different slots, so there is exactly one timer mechanism to reason about (and to cancel on :meth:async_stop).

Cancels a still-pending confirm in the same slot first, so a rapid command sequence produces exactly one confirmation per slot. An entity that no longer asks for a deferred republish gets its slot cleared by :meth:_sync_deferred_confirm: a timer armed for a movement that has since been cancelled (counter-command, contact arrival, option switched off) would otherwise survive for the whole travel time and fire a redundant forced publish long after the fact.

Parameters:

Name	Type	Description	Default
entity_id	str	HA entity identifier that was just commanded.	required

Source code in custom_components/sber_mqtt_bridge/sber_bridge.py

@callback
def schedule_confirm(self, entity_id: str) -> None:
    """(Re)arm the delayed state confirm(s) for one commanded entity.

    Always arms the short confirm that lets HA settle its async
    attribute updates (:attr:`_confirm_delay`).  An entity may ask for
    a *second*, later republish through a ``pending_confirm_delay``
    attribute — the impulse gate uses it to replace its emulated
    ``opening`` / ``closing`` value once the leaf's travel time is
    over (see :class:`~devices.gate.ImpulseGateEntity`).  Both go
    through the very same :meth:`_delayed_confirm` machinery, just in
    different slots, so there is exactly one timer mechanism to reason
    about (and to cancel on :meth:`async_stop`).

    Cancels a still-pending confirm in the same slot first, so a rapid
    command sequence produces exactly one confirmation per slot.  An
    entity that no longer asks for a deferred republish gets its slot
    *cleared* by :meth:`_sync_deferred_confirm`: a timer armed for a
    movement that has since been cancelled (counter-command, contact
    arrival, option switched off) would otherwise survive for the
    whole travel time and fire a redundant forced publish long after
    the fact.

    Args:
        entity_id: HA entity identifier that was just commanded.
    """
    self._arm_confirm(entity_id, entity_id, self._confirm_delay)
    self._sync_deferred_confirm(entity_id, floor=self._confirm_delay)

forget_cloud_devices

forget_cloud_devices(entity_ids)

Drop entity ids from the persisted "cloud holds it" registry.

Called when the user un-exposes entities. Normally the next config publish mirrors the shorter list on its own, but it cannot when the shorter list is empty: a publish carrying no device is refused as evidence (see :meth:~cloud_device_registry.CloudDeviceRegistry.note_published), so "remove everything" would otherwise leave the registry claiming devices nobody exposes any more.

Parameters:

Name	Type	Description	Default
entity_ids	Iterable[str]	Entity ids the user removed from the bridge.	required

Source code in custom_components/sber_mqtt_bridge/sber_bridge.py

@callback
def forget_cloud_devices(self, entity_ids: Iterable[str]) -> None:
    """Drop entity ids from the persisted "cloud holds it" registry.

    Called when the user un-exposes entities.  Normally the next config
    publish mirrors the shorter list on its own, but it cannot when the
    shorter list is *empty*: a publish carrying no device is refused as
    evidence (see
    :meth:`~cloud_device_registry.CloudDeviceRegistry.note_published`),
    so "remove everything" would otherwise leave the registry claiming
    devices nobody exposes any more.

    Args:
        entity_ids: Entity ids the user removed from the bridge.
    """
    self._cloud_devices.forget(entity_ids)

async_update_redefinition async

async_update_redefinition(entity_id, fields)

Merge redefinition fields for an entity and trigger config republish.

Public API for frontend / WebSocket handlers to update a device's Sber-side name / room / home without reaching into private state. Delegates data mutation and debounced persistence to :meth:RedefinitionsStore.async_update.

Parameters:

Name	Type	Description	Default
entity_id	str	Target Sber entity identifier (must exist in the bridge).	required
fields	dict[str, str | None]	Partial mapping with any of name / room / home. An empty string or None for a key removes that field.	required

Returns:

Type	Description
dict[str, str]	Resulting redefinitions dict for the entity after merge.

Raises:

Type	Description
KeyError	If entity_id is not loaded in the bridge.
HomeAssistantError	If the follow-up config publish fails.

Source code in custom_components/sber_mqtt_bridge/sber_bridge.py

async def async_update_redefinition(self, entity_id: str, fields: dict[str, str | None]) -> dict[str, str]:
    """Merge redefinition fields for an entity and trigger config republish.

    Public API for frontend / WebSocket handlers to update a device's
    Sber-side name / room / home without reaching into private state.
    Delegates data mutation and debounced persistence to
    :meth:`RedefinitionsStore.async_update`.

    Args:
        entity_id: Target Sber entity identifier (must exist in the bridge).
        fields: Partial mapping with any of ``name`` / ``room`` / ``home``.
            An empty string or ``None`` for a key removes that field.

    Returns:
        Resulting redefinitions dict for the entity after merge.

    Raises:
        KeyError: If ``entity_id`` is not loaded in the bridge.
        HomeAssistantError: If the follow-up config publish fails.
    """
    if entity_id not in self._entities:
        raise KeyError(entity_id)
    existing = await self._redef_store.async_update(entity_id, fields)
    await self._publish_config()
    return existing

async_update_entity_options async

async_update_entity_options(entity_id, fields)

Merge per-entity device options for one entity and apply them live.

Persists into entry.options[CONF_ENTITY_OPTIONS] and then pushes the merged values straight into the loaded entity instead of reloading the config entry: a reload tears the MQTT session down and back up, and dropping the bridge for a couple of seconds because someone flipped a checkbox is not a trade the user agreed to. Same approach as :meth:async_update_redefinition.

The entity is re-seeded from HA afterwards because some options change how existing readings are interpreted (a gate's invert_contact flips the meaning of the contact's last value), and both the config and this entity's state are republished because the model descriptor may change too (travel_time / auto_close_time add allowed_values.open_state). The config publish covers every device on purpose: Sber reads each config payload as the complete device list, so a one-device payload would make the cloud drop and re-create everything else (issue #44). Only the state publish is narrowed to the edited entity.

Category-agnostic: which keys an entity accepts, what they mean and whether a value is usable is decided by the device class (BaseEntity.ENTITY_OPTION_KEYS / validate_entity_options / apply_entity_options).

Parameters:

Name	Type	Description	Default
entity_id	str	HA entity identifier.	required
fields	dict[str, Any]	Partial option mapping; only the keys present are changed.	required

Returns:

Type	Description
dict[str, Any]	The merged option dict stored for this entity.

Raises:

Type	Description
KeyError	If entity_id is not loaded in the bridge.
TypeError	If the entity's class accepts no options.
ValueError	If the entity rejects one of the submitted values.
HomeAssistantError	If the follow-up publish fails.

Source code in custom_components/sber_mqtt_bridge/sber_bridge.py

async def async_update_entity_options(self, entity_id: str, fields: dict[str, Any]) -> dict[str, Any]:
    """Merge per-entity device options for one entity and apply them live.

    Persists into ``entry.options[CONF_ENTITY_OPTIONS]`` and then
    pushes the merged values straight into the loaded entity instead
    of reloading the config entry: a reload tears the MQTT session
    down and back up, and dropping the bridge for a couple of seconds
    because someone flipped a checkbox is not a trade the user agreed
    to.  Same approach as :meth:`async_update_redefinition`.

    The entity is re-seeded from HA afterwards because some options
    change how *existing* readings are interpreted (a gate's
    ``invert_contact`` flips the meaning of the contact's last value),
    and both the config and this entity's state are republished
    because the model descriptor may change too (``travel_time`` /
    ``auto_close_time`` add ``allowed_values.open_state``).  The
    config publish covers *every* device on purpose: Sber reads each
    config payload as the complete device list, so a one-device
    payload would make the cloud drop and re-create everything else
    (issue #44).  Only the state publish is narrowed to the edited
    entity.

    Category-agnostic: which keys an entity accepts, what they mean
    and whether a value is usable is decided by the device class
    (``BaseEntity.ENTITY_OPTION_KEYS`` /
    ``validate_entity_options`` / ``apply_entity_options``).

    Args:
        entity_id: HA entity identifier.
        fields: Partial option mapping; only the keys present are
            changed.

    Returns:
        The merged option dict stored for this entity.

    Raises:
        KeyError: If ``entity_id`` is not loaded in the bridge.
        TypeError: If the entity's class accepts no options.
        ValueError: If the entity rejects one of the submitted values.
        HomeAssistantError: If the follow-up publish fails.
    """
    entity = self._entities.get(entity_id)
    if entity is None:
        raise KeyError(entity_id)
    if not entity.supports_entity_options:
        raise TypeError(f"{entity_id} ({entity.category}) has no configurable options")
    entity.validate_entity_options(fields)

    all_options: dict[str, dict] = dict(self._entry.options.get(CONF_ENTITY_OPTIONS, {}))
    merged: dict[str, Any] = {**all_options.get(entity_id, {}), **fields}
    all_options[entity_id] = merged
    new_options = dict(self._entry.options)
    new_options[CONF_ENTITY_OPTIONS] = all_options
    self._hass.config_entries.async_update_entry(self._entry, options=new_options)

    entity.apply_entity_options(merged)
    self._refresh_entity_from_ha(entity_id)
    # An option change both *destroys* and *creates* deadlines the
    # entity wants to be republished at: a gate drops a running
    # auto-close countdown whenever the delay changes (it was armed
    # against the old value), and re-seeding from HA above can arm a
    # fresh one.  Without this resync the slot armed for the previous
    # value survives — up to ``MAX_AUTO_CLOSE_TIME_SECONDS``, an hour
    # — and fires a redundant forced publish for a movement that was
    # cancelled long before.
    self._sync_deferred_confirm(entity_id)
    await self._publish_config()
    await self._publish_states([entity_id], force=True)
    return merged

async_update_gate_options async

async_update_gate_options(entity_id, fields)

Deprecated alias of :meth:async_update_entity_options.

Kept because the per-entity option store shipped for impulse gates first (v1.42) and this name is part of that public surface.

Parameters:

Name	Type	Description	Default
entity_id	str	HA entity identifier of the gate relay.	required
fields	dict[str, Any]	Partial gate-option mapping.	required

Returns:

Type	Description
dict[str, Any]	The merged option dict stored for this entity.

Source code in custom_components/sber_mqtt_bridge/sber_bridge.py

async def async_update_gate_options(self, entity_id: str, fields: dict[str, Any]) -> dict[str, Any]:
    """Deprecated alias of :meth:`async_update_entity_options`.

    Kept because the per-entity option store shipped for impulse gates
    first (v1.42) and this name is part of that public surface.

    Args:
        entity_id: HA entity identifier of the gate relay.
        fields: Partial gate-option mapping.

    Returns:
        The merged option dict stored for this entity.
    """
    return await self.async_update_entity_options(entity_id, fields)

async_republish_config async

async_republish_config()

Public wrapper for forcing a device config republish to Sber.

Explicit user action — bypasses the coalescing gate so the panel's "Re-publish" button is immediate.

Source code in custom_components/sber_mqtt_bridge/sber_bridge.py

async def async_republish_config(self) -> None:
    """Public wrapper for forcing a device config republish to Sber.

    Explicit user action — bypasses the coalescing gate so the panel's
    "Re-publish" button is immediate.
    """
    await self._config_gate.flush_now()

clear_message_log

clear_message_log()

Clear the DevTools message log (delegates to hub).

Source code in custom_components/sber_mqtt_bridge/sber_bridge.py

def clear_message_log(self) -> None:
    """Clear the DevTools message log (delegates to hub)."""
    self._devtools.clear_message_log()

apply_settings

apply_settings(options)

Apply changed operational settings without full bridge restart.

Settings that take effect immediately: debounce_delay, max_mqtt_payload_size, message_log_size. Settings that take effect on next reconnect: reconnect_min, reconnect_max, verify_ssl.

Parameters:

Name	Type	Description	Default
options	dict	Config entry options dict.	required

Source code in custom_components/sber_mqtt_bridge/sber_bridge.py

def apply_settings(self, options: dict) -> None:
    """Apply changed operational settings without full bridge restart.

    Settings that take effect immediately: debounce_delay, max_mqtt_payload_size,
    message_log_size.
    Settings that take effect on next reconnect: reconnect_min, reconnect_max, verify_ssl.

    Args:
        options: Config entry options dict.
    """
    self._load_settings_from_options(options)
    self._state_forwarder.set_debounce_delay(self._debounce_delay)
    self._config_gate.update_delays(settle_delay=self._config_settle_delay, max_wait=self._config_max_wait)
    self._mqtt_service.update_backoff_limits(self._reconnect_min, self._reconnect_max)
    self._mqtt_service.update_verify_ssl(self._verify_ssl)
    self._devtools.resize(self._message_log_size)

    _LOGGER.info(
        "Bridge settings applied (debounce=%.2fs, log=%d)",
        self._debounce_delay,
        self._message_log_size,
    )

async_publish_raw async

async_publish_raw(payload, target)

Publish arbitrary JSON payload to Sber MQTT for debugging.

Parameters:

Name	Type	Description	Default
payload	str	Raw JSON string to publish.	required
target	str	Topic suffix — either "config" or "status".	required

Raises:

Type	Description
RuntimeError	If not connected to MQTT broker.
MqttError	Propagated on transport errors (counted in publish_errors).

Source code in custom_components/sber_mqtt_bridge/sber_bridge.py

async def async_publish_raw(self, payload: str, target: str) -> None:
    """Publish arbitrary JSON payload to Sber MQTT for debugging.

    Args:
        payload: Raw JSON string to publish.
        target: Topic suffix — either "config" or "status".

    Raises:
        RuntimeError: If not connected to MQTT broker.
        aiomqtt.MqttError: Propagated on transport errors (counted in
            ``publish_errors``).
    """
    topic = f"{self._root_topic}/up/{target}"
    try:
        await self._mqtt_service.publish(topic, payload)
    except aiomqtt.MqttError:
        self._stats.publish_errors += 1
        raise
    self._stats.messages_sent += 1
    self._log_message("out", topic, payload)

async_inject_sber_message async

async_inject_sber_message(topic, payload, *, mark_replay=True)

Feed a synthetic message into the dispatcher as if Sber sent it.

Used by DevTools Replay / Inject: takes a topic (full sbdev/.../down/commands or a bare suffix like commands) and runs it through the normal inbound pipeline — :class:SberCommandDispatcher, correlation trace, state diff, ack audit — without going through the MQTT broker. No network round-trip means an injected command flows even when the bridge is offline, which is exactly what users want when debugging.

Parameters:

Name	Type	Description	Default
topic	str	Either the full MQTT topic as it would arrive from Sber cloud, or just the last segment (suffix) which is automatically expanded to {root}/down/{suffix}.	required
payload	str | bytes	Raw JSON body. Bytes pass through as-is; strings are UTF-8 encoded to match the real on-wire shape.	required
mark_replay	bool	When True (default), the DevTools message log records the direction as "replay" instead of "in" so the UI can visually distinguish synthetic traffic from real Sber commands. Set False to make the injection indistinguishable from real MQTT input (e.g. reproducing a bug for screenshot).	True

Returns:

Type	Description
dict[str, Any]	Dict with {"topic": str, "handled": bool, "suffix": str}.
dict[str, Any]	handled is False only when no dispatcher was registered
dict[str, Any]	for the given suffix (unknown topic).

Source code in custom_components/sber_mqtt_bridge/sber_bridge.py

async def async_inject_sber_message(
    self,
    topic: str,
    payload: str | bytes,
    *,
    mark_replay: bool = True,
) -> dict[str, Any]:
    """Feed a synthetic message into the dispatcher as if Sber sent it.

    Used by DevTools Replay / Inject: takes a topic (full
    ``sbdev/.../down/commands`` or a bare suffix like ``commands``)
    and runs it through the normal inbound pipeline —
    :class:`SberCommandDispatcher`, correlation trace, state diff,
    ack audit — without going through the MQTT broker.  No network
    round-trip means an injected command flows even when the bridge
    is offline, which is exactly what users want when debugging.

    Args:
        topic: Either the full MQTT topic as it would arrive from
            Sber cloud, or just the last segment (suffix) which is
            automatically expanded to ``{root}/down/{suffix}``.
        payload: Raw JSON body.  Bytes pass through as-is; strings
            are UTF-8 encoded to match the real on-wire shape.
        mark_replay: When True (default), the DevTools message log
            records the direction as ``"replay"`` instead of
            ``"in"`` so the UI can visually distinguish synthetic
            traffic from real Sber commands.  Set False to make
            the injection indistinguishable from real MQTT input
            (e.g. reproducing a bug for screenshot).

    Returns:
        Dict with ``{"topic": str, "handled": bool, "suffix": str}``.
        ``handled`` is False only when no dispatcher was registered
        for the given suffix (unknown topic).
    """
    full_topic = topic if "/" in topic else f"{self._down_topic}/{topic}"
    body = payload.encode("utf-8") if isinstance(payload, str) else payload

    # Route through the dispatch table used by the real MQTT handler.
    suffix = full_topic.rsplit("/", 1)[-1] if "/" in full_topic else full_topic
    decoded = body.decode("utf-8", errors="replace")
    self._log_message("replay" if mark_replay else "in", full_topic, decoded)

    if full_topic == SBER_GLOBAL_CONFIG_TOPIC:
        self._handle_global_config(body)
        return {"topic": full_topic, "handled": True, "suffix": "(global_config)"}

    handler = self._mqtt_dispatch.get(suffix)
    if handler is None:
        _LOGGER.warning("Inject: unhandled topic suffix %r", suffix)
        return {"topic": full_topic, "handled": False, "suffix": suffix}

    await handler(body)
    return {"topic": full_topic, "handled": True, "suffix": suffix}

subscribe_messages

subscribe_messages(callback_fn)

Subscribe to new MQTT messages in real time (delegates to hub).

Parameters:

Name	Type	Description	Default
callback_fn	Callable[[dict], None]	Called with each new message dict.	required

Returns:

Type	Description
Callable[[], None]	Unsubscribe callable.

Source code in custom_components/sber_mqtt_bridge/sber_bridge.py

def subscribe_messages(self, callback_fn: Callable[[dict], None]) -> Callable[[], None]:
    """Subscribe to new MQTT messages in real time (delegates to hub).

    Args:
        callback_fn: Called with each new message dict.

    Returns:
        Unsubscribe callable.
    """
    return self._devtools.subscribe_messages(callback_fn)

async_start async

async_start()

Start the bridge: load entities, subscribe to HA events, connect MQTT.

HA state events are subscribed immediately (independent of MQTT connectivity) so that no state changes are lost while waiting for the first connection. MQTT connection is established in a background task with exponential backoff.

Source code in custom_components/sber_mqtt_bridge/sber_bridge.py

async def async_start(self) -> None:
    """Start the bridge: load entities, subscribe to HA events, connect MQTT.

    HA state events are subscribed immediately (independent of MQTT connectivity)
    so that no state changes are lost while waiting for the first connection.
    MQTT connection is established in a background task with exponential backoff.
    """
    self._running = True
    # Cache the HA instance UUID prefix so the publish hot-path stays sync.
    # Used for the per-HA ``ha_serial_number`` loop-detection marker.
    from homeassistant.helpers import instance_id

    full_uuid = await instance_id.async_get(self._hass)
    self._ha_instance_id_prefix: str = full_uuid[:8]
    self._load_exposed_entities()
    self._subscribe_ha_events()
    # Daemon, not a tracked task: this loop never returns, so tracking it
    # would make HA bootstrap wait on it until the setup timeout.
    self._connection_task = self._create_daemon_task(
        self._mqtt_connection_loop(),
        name="mqtt_connection_loop",
    )

    # If HA is already running (e.g. integration reload), entities are
    # fully available — mark ready immediately.  Otherwise, wait for
    # EVENT_HOMEASSISTANT_STARTED to reload entities with real states.
    if self._hass.is_running:
        _LOGGER.debug("HA already running — entities loaded, marking ready")
        self._ha_ready.set()
    else:
        self._unsub_lifecycle_listeners.append(
            self._hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STARTED, self._on_homeassistant_started)
        )

async_stop async

async_stop()

Stop the bridge: disconnect MQTT, unsubscribe from HA events.

Idempotent — safe to call multiple times. Cancels every timer and background task the bridge owns (state forwarder debounce, lifecycle listeners, ack-audit timer, delayed-confirm tasks, redefinitions debounce timer, MQTT connection loop) so nothing outlives the entry unload. A pending redefinitions snapshot is flushed synchronously before shutdown so user edits are not lost on reload.

Source code in custom_components/sber_mqtt_bridge/sber_bridge.py

async def async_stop(self) -> None:
    """Stop the bridge: disconnect MQTT, unsubscribe from HA events.

    Idempotent — safe to call multiple times.  Cancels every timer and
    background task the bridge owns (state forwarder debounce, lifecycle
    listeners, ack-audit timer, delayed-confirm tasks, redefinitions
    debounce timer, MQTT connection loop) so nothing outlives the entry
    unload.  A pending redefinitions snapshot is flushed synchronously
    before shutdown so user edits are not lost on reload.
    """
    self._running = False

    # HA state-change listeners + debounced publish live in the forwarder
    self._state_forwarder.unsubscribe_all()

    for unsub in self._unsub_lifecycle_listeners:
        unsub()
    self._unsub_lifecycle_listeners.clear()

    # Cancel any pending ack-audit timer so it can't fire after shutdown
    self._ack_audit.cancel()

    # Cancel delayed-confirm tasks so they don't touch hass after unload
    for task in self._confirm_tasks.values():
        task.cancel()
    self._confirm_tasks.clear()

    # Cancel the redefinitions debounce timer and flush a pending
    # snapshot synchronously so a reload within the debounce window
    # cannot lose (or later overwrite) user edits.
    self._redef_store.shutdown()

    # Stop the MQTT service reconnect loop
    await self._mqtt_service.stop()

    if self._connection_task:
        self._connection_task.cancel()
        try:
            await self._connection_task
        except asyncio.CancelledError:
            pass
        except Exception:  # shutdown must not fail entry unload
            _LOGGER.exception("MQTT connection task raised during shutdown")
        self._connection_task = None

    self._config_gate.cancel()
    self._cloud_devices.shutdown()
    self._connected = False

refresh_repair_issues

refresh_repair_issues()

Recompute HA repair issues without awaiting.

Wraps :func:check_and_create_issues in a safe background task so callers (notably the command dispatcher) can fire-and-forget after an ack arrives. No-op when HA is not yet running so we don't fight the early-startup grace window in :meth:_load_exposed_entities.

Source code in custom_components/sber_mqtt_bridge/sber_bridge.py

@callback
def refresh_repair_issues(self) -> None:
    """Recompute HA repair issues without awaiting.

    Wraps :func:`check_and_create_issues` in a safe background task so
    callers (notably the command dispatcher) can fire-and-forget after
    an ack arrives.  No-op when HA is not yet running so we don't fight
    the early-startup grace window in :meth:`_load_exposed_entities`.
    """
    if not self._hass.is_running:
        return
    self._create_safe_task(
        check_and_create_issues(self._hass, self),
        name="refresh_repair_issues",
    )

async_publish_entity_status async

async_publish_entity_status(entity_id)

Publish the current state of a single entity to Sber cloud.

Parameters:

Name	Type	Description	Default
entity_id	str	HA entity identifier.	required

Source code in custom_components/sber_mqtt_bridge/sber_bridge.py

async def async_publish_entity_status(self, entity_id: str) -> None:
    """Publish the current state of a single entity to Sber cloud.

    Args:
        entity_id: HA entity identifier.
    """
    await self._publish_states([entity_id])