SberCommandDispatcher

Интерпретация входящих Sber MQTT-сообщений и диспатч side-эффектов. Владеет обработчиками handle_command, handle_status_request, handle_config_request, handle_error, handle_change_group, handle_rename_device, handle_global_config.

Извлечён из SberBridge в v1.25.1 для изоляции Sber-протокольной логики от транспорта и HA state forwarding (SRP). Держит ссылку на родительский SberBridge, так как ряд handler'ов мутирует состояние моста (entities, redefinitions, acknowledgements) и инициирует publish.

Sber MQTT command dispatcher.

Handles commands, status/config requests, errors, change_group and rename_device messages from the Sber cloud. Extracted from :class:SberBridge to isolate Sber-protocol command interpretation from transport and HA state forwarding (SRP).

The dispatcher owns no reference to :class:SberBridge. Everything it may touch arrives through :class:DispatcherDeps: the collaborators it drives (publisher, redefinitions store, DevTools hub, ack audit) plus callables for the few bridge-owned operations it triggers. Narrowing this bundle is what keeps the bridge free to reshape its internals.

DispatcherDeps dataclass

DispatcherDeps(hass, stats, ack_audit, publisher, redefinitions, devtools, get_entities, get_enabled_entity_ids, schedule_confirm, note_cloud_reported, refresh_repair_issues)

Everything :class:SberCommandDispatcher is allowed to reach.

hass instance-attribute

hass

HA core — used only to invoke services for Sber commands.

stats instance-attribute

stats

Counter bag bumped per inbound message kind.

ack_audit instance-attribute

ack_audit

Reconnect guard consulted before executing a command.

publisher instance-attribute

publisher

Publish coordinator used for state / config / echo responses.

redefinitions instance-attribute

redefinitions

Store fed by change_group / rename_device payloads.

devtools instance-attribute

devtools

Collector aggregate that records the command correlation trace.

get_entities instance-attribute

get_entities

Returns the live entity_id → BaseEntity map.

get_enabled_entity_ids instance-attribute

get_enabled_entity_ids

Returns the ordered list of exposed entity IDs.

schedule_confirm instance-attribute

schedule_confirm

Asks the bridge to (re)arm the delayed state confirm for one entity.

note_cloud_reported instance-attribute

note_cloud_reported

Records entity ids the cloud named, into the persistent registry.

A command addressed to a device is the strongest evidence the protocol offers — Sber does not command a device it does not hold — and it used to be thrown away: only the in-memory session mark was set, so a restart forgot it. A bridge driven purely by voice and app commands therefore learned nothing that survived (issue #57).

refresh_repair_issues instance-attribute

refresh_repair_issues

Asks the bridge to recompute its HA repair-issue set.

SberCommandDispatcher

SberCommandDispatcher(deps)

Interprets incoming Sber MQTT payloads and dispatches side effects.

Each handle_* method corresponds to one topic suffix in the Sber down/* namespace. The bridge's _mqtt_dispatch table routes incoming messages to the matching handler.

Initialize the dispatcher bound to its dependency bundle.

Parameters:

Name	Type	Description	Default
deps	DispatcherDeps	Narrow dependency bundle assembled by the bridge.	required

Source code in custom_components/sber_mqtt_bridge/command_dispatcher.py

def __init__(self, deps: DispatcherDeps) -> None:
    """Initialize the dispatcher bound to its dependency bundle.

    Args:
        deps: Narrow dependency bundle assembled by the bridge.
    """
    self._deps = deps

handle_command async

handle_command(payload, context=None)

Handle a command from Sber cloud → execute HA service.

During the reconnect grace period, commands are rejected and current HA states are re-published so that Sber cloud accepts HA as the authoritative source of truth.

Parameters:

Name	Type	Description	Default
payload	bytes	Raw MQTT payload from down/commands.	required
context	Context | None	Optional HA context to attribute the resulting service calls to (e.g. a user-scoped context for WS-initiated replays). A fresh anonymous Context is created when omitted.	None

Source code in custom_components/sber_mqtt_bridge/command_dispatcher.py

async def handle_command(self, payload: bytes, context: Context | None = None) -> None:
    """Handle a command from Sber cloud → execute HA service.

    During the reconnect grace period, commands are rejected and
    current HA states are re-published so that Sber cloud accepts
    HA as the authoritative source of truth.

    Args:
        payload: Raw MQTT payload from ``down/commands``.
        context: Optional HA context to attribute the resulting
            service calls to (e.g. a user-scoped context for
            WS-initiated replays). A fresh anonymous ``Context``
            is created when omitted.
    """
    deps = self._deps
    data = parse_sber_command(payload)
    deps.stats.commands_received += 1
    devices = data.get("devices", {})

    if await self._handle_reconnect_grace(devices):
        return

    _LOGGER.debug("Sber command for %d device(s): %s", len(devices), list(devices.keys()))

    if context is None:
        context = Context()
    self._open_command_trace(devices, context)

    update_state_ids: list[str] = []
    for entity_id, cmd_data in devices.items():
        if await self._process_one_entity(entity_id, cmd_data, context):
            update_state_ids.append(entity_id)

    # Only well-formed (dict) command payloads participate in the echo
    # ack — a single type-confused entry must not break the ack for the
    # rest of the batch.
    valid_devices = {eid: cmd for eid, cmd in devices.items() if isinstance(cmd, dict)}
    entities = deps.get_entities()
    commanded_ids = [eid for eid in valid_devices if eid in entities]

    if update_state_ids:
        await deps.publisher.publish_states(update_state_ids, force=True)

    # Immediate echo ack: publish the received command states back to
    # Sber within milliseconds so its ack timer does not expire before
    # HA propagates the real state change.  Required for integrations
    # that delay/omit ``state_changed`` events on no-op commands (e.g.
    # HA WLED integration with WLED 16.0.0 — see GitHub issue #35 and
    # HA core issue #170435).
    if commanded_ids:
        await deps.publisher.publish_command_echo(valid_devices)

    self._schedule_confirms(commanded_ids)

    # Named, per-device evidence, and the strongest the protocol has:
    # Sber does not send a command to a device it does not hold.  It
    # goes to the *persistent* registry, not just the session mark —
    # a bridge the cloud only ever drives by voice or app command used
    # to forget every one of them on restart and report "known to
    # Sber: 0" while working perfectly (issue #57).  Only ids the
    # bridge actually knows are recorded: a command for a stale id
    # would otherwise hold the publish gate open forever on an entity
    # that no longer exists in HA.
    if commanded_ids:
        for eid in commanded_ids:
            deps.stats.collectively_acked_entities.discard(eid)
        deps.note_cloud_reported(commanded_ids)

    # Receiving any command is positive evidence that Sber accepted at
    # least one entity — re-evaluate the silent-rejection issue so a
    # stale repair tile clears as soon as the user activates the device.
    self._refresh_repair_issues()

handle_status_request async

handle_status_request(payload)

Handle a status request from Sber cloud.

If Sber asks about entities not in our current set, automatically re-publishes the device config so Sber is aware of the correct list. A status_request also counts as Sber acknowledgment.

Source code in custom_components/sber_mqtt_bridge/command_dispatcher.py

async def handle_status_request(self, payload: bytes) -> None:
    """Handle a status request from Sber cloud.

    If Sber asks about entities not in our current set, automatically
    re-publishes the device config so Sber is aware of the correct list.
    A status_request also counts as Sber acknowledgment.
    """
    deps = self._deps
    requested_ids = parse_sber_status_request(payload)
    deps.stats.status_requests += 1

    deps.ack_audit.acknowledge()

    if requested_ids:
        entities = deps.get_entities()
        unknown = [eid for eid in requested_ids if eid not in entities and eid != "root"]
        if unknown:
            _LOGGER.info(
                "Sber asked about unknown entities, re-publishing config: %s",
                unknown,
            )
            await deps.publisher.publish_config()

    if requested_ids:
        # Named evidence: the cloud listed these devices itself, so it
        # holds them.  Promote any weak collective mark they carried.
        for eid in requested_ids:
            deps.stats.acknowledged_entities.add(eid)
            deps.stats.collectively_acked_entities.discard(eid)
        _LOGGER.info(
            "Sber status request for %d specific entities: %s",
            len(requested_ids),
            requested_ids,
        )
    else:
        # Collective evidence: "state of everything" names nobody, so
        # it must not vouch for any individual device.  Recorded as a
        # weak mark alongside the acknowledgement so the panel counter
        # stays truthful while the silent-rejection alarm keeps working
        # (issue #57) — anything already known by name keeps its
        # stronger standing.
        enabled_ids = deps.get_enabled_entity_ids()
        deps.stats.collectively_acked_entities.update(
            eid for eid in enabled_ids if eid not in deps.stats.acknowledged_entities
        )
        deps.stats.acknowledged_entities.update(enabled_ids)
        _LOGGER.info(
            "Sber status request for ALL entities (%d)",
            len(enabled_ids),
        )

    await deps.publisher.publish_states(requested_ids if requested_ids else None, force=True)

    # status_request is the strongest single ack signal we get from
    # Sber (it explicitly enumerates accepted entities or asks for
    # the whole set).  Refresh the repair issues so the silent-
    # rejection tile clears in real time, not only on next reload.
    self._refresh_repair_issues()

handle_config_request async

handle_config_request()

Handle config request from Sber cloud — send device list.

Forced past the unchanged-payload check. The cloud asked us a direct question, and answering it with silence because the answer has not changed is wrong twice over: Sber gets nothing (it asks precisely when its own copy is in doubt), and the publish that records what the cloud holds never happens — leaving the registry empty for the whole session with no second chance (issue #57).

Source code in custom_components/sber_mqtt_bridge/command_dispatcher.py

async def handle_config_request(self) -> None:
    """Handle config request from Sber cloud — send device list.

    Forced past the unchanged-payload check.  The cloud asked us a
    direct question, and answering it with silence because the answer
    has not changed is wrong twice over: Sber gets nothing (it asks
    precisely when its own copy is in doubt), and the publish that
    records what the cloud holds never happens — leaving the registry
    empty for the whole session with no second chance (issue #57).
    """
    deps = self._deps
    deps.stats.config_requests += 1
    deps.ack_audit.acknowledge()
    _LOGGER.info(
        "Sber config request received (will publish %d entities)",
        len(deps.get_enabled_entity_ids()),
    )
    await deps.publisher.publish_config(force=True)

handle_error

handle_error(payload)

Handle error message from Sber cloud.

Parses the error payload, stores the detail in stats for repair issue creation, and logs the error.

Source code in custom_components/sber_mqtt_bridge/command_dispatcher.py

def handle_error(self, payload: bytes) -> None:
    """Handle error message from Sber cloud.

    Parses the error payload, stores the detail in stats for repair
    issue creation, and logs the error.
    """
    stats = self._deps.stats
    stats.errors_from_sber += 1
    try:
        error_data = json.loads(payload)
        detail = json.dumps(error_data, ensure_ascii=False)
        stats.last_error_detail = detail[:500]
        _LOGGER.warning(
            "Sber error (#%d): %s",
            stats.errors_from_sber,
            detail,
        )
    except (json.JSONDecodeError, TypeError):
        raw = payload.decode(errors="replace")[:500]
        stats.last_error_detail = raw
        _LOGGER.warning(
            "Sber error (#%d, raw): %s",
            stats.errors_from_sber,
            raw,
        )

handle_change_group async

handle_change_group(payload)

Handle device group/room change from Sber.

Values are validated (string type, length limit) and stored through :meth:RedefinitionsStore.async_update so cloud input goes through the same normalization as the WS API — invalid or missing values clear the corresponding key instead of persisting arbitrary payloads.

Only stores the redefinition locally. Does NOT re-publish config to avoid an infinite loop: Sber sends change_group → we publish config → Sber sends change_group again → loop forever.

Source code in custom_components/sber_mqtt_bridge/command_dispatcher.py

async def handle_change_group(self, payload: bytes) -> None:
    """Handle device group/room change from Sber.

    Values are validated (string type, length limit) and stored
    through :meth:`RedefinitionsStore.async_update` so cloud input
    goes through the same normalization as the WS API — invalid or
    missing values clear the corresponding key instead of persisting
    arbitrary payloads.

    Only stores the redefinition locally. Does NOT re-publish config
    to avoid an infinite loop: Sber sends change_group → we publish
    config → Sber sends change_group again → loop forever.
    """
    store = self._deps.redefinitions
    data = _parse_json_dict(payload, "change_group_device_request")
    if data is None:
        return
    entity_id = self._extract_redef_target(data, "change_group")
    if entity_id is None:
        return
    fields: dict[str, str | None] = {
        "home": _sanitize_redef_value(data.get("home")),
        "room": _sanitize_redef_value(data.get("room")),
    }
    if all(value is None for value in fields.values()) and not store.has(entity_id):
        # Nothing usable to store and no existing record to clear —
        # avoid creating empty {} entries (and persist churn) for
        # arbitrary cloud-supplied ids.
        _LOGGER.debug("Sber change_group for %s carries no usable values — ignored", entity_id)
        return
    await store.async_update(entity_id, fields)
    _LOGGER.info("Sber group change stored: %s → room=%s", entity_id, fields["room"])

handle_rename_device async

handle_rename_device(payload)

Handle device rename from Sber.

The new name is validated (string type, length limit) and stored through :meth:RedefinitionsStore.async_update; payloads with a non-string or oversized name are rejected without touching the persistent store.

Only stores the redefinition locally. Does NOT re-publish config to avoid potential loops.

Source code in custom_components/sber_mqtt_bridge/command_dispatcher.py

async def handle_rename_device(self, payload: bytes) -> None:
    """Handle device rename from Sber.

    The new name is validated (string type, length limit) and stored
    through :meth:`RedefinitionsStore.async_update`; payloads with a
    non-string or oversized name are rejected without touching the
    persistent store.

    Only stores the redefinition locally. Does NOT re-publish config
    to avoid potential loops.
    """
    store = self._deps.redefinitions
    data = _parse_json_dict(payload, "rename_device_request")
    if data is None:
        return
    entity_id = self._extract_redef_target(data, "rename_device")
    if entity_id is None:
        return
    new_name = _sanitize_redef_value(data.get("new_name"))
    if new_name is None:
        if data.get("new_name") is not None:
            _LOGGER.warning(
                "Ignoring Sber rename for %s with invalid new_name: %.60r",
                entity_id,
                data.get("new_name"),
            )
        return
    await store.async_update(entity_id, {"name": new_name})
    _LOGGER.info("Sber rename stored: %s → %s", entity_id, new_name)

handle_global_config

handle_global_config(payload)

Handle global config from Sber (http_api_endpoint).

Source code in custom_components/sber_mqtt_bridge/command_dispatcher.py

def handle_global_config(self, payload: bytes) -> None:
    """Handle global config from Sber (http_api_endpoint)."""
    data = _parse_json_dict(payload, "global_config")
    if data is None:
        return
    endpoint = data.get("http_api_endpoint", "")
    if endpoint:
        _LOGGER.info("Sber HTTP API endpoint: %s", endpoint)