Протокол Sber Smart Home¶

JSON-сериализация для обмена данными с облаком Sber Smart Home.

Sber Smart Home MQTT protocol serialization.

Handles conversion of internal device entities to Sber JSON formats for MQTT communication (device config, state lists, command parsing).

VERSION `module-attribute` ¶

VERSION = '1.49.0'

Protocol version string included in the hub device descriptor.

build_hub_device ¶

build_hub_device(version=VERSION, home='', room='', ha_serial_prefix=None)

Build the root hub device descriptor for Sber.

Parameters:

Name	Type	Description	Default
`version`	`str`	Protocol version string.	`VERSION`
`home`	`str`	Home name for the hub device.	`''`
`room`	`str`	Room name for the hub device.	`''`
`ha_serial_prefix`	`str \| None`	When set, the hub device gets a `partner_meta.ha_serial_number` value of `f"ha-{prefix}"` (typically the first 8 chars of the HA instance UUID). Used by sister integrations that mirror Sber devices back into HA to detect their own loop. `None` disables the marker.	`None`

Source code in custom_components/sber_mqtt_bridge/sber_protocol.py

def build_hub_device(
    version: str = VERSION,
    home: str = "",
    room: str = "",
    ha_serial_prefix: str | None = None,
) -> dict:
    """Build the root hub device descriptor for Sber.

    Args:
        version: Protocol version string.
        home: Home name for the hub device.
        room: Room name for the hub device.
        ha_serial_prefix: When set, the hub device gets a
            ``partner_meta.ha_serial_number`` value of ``f"ha-{prefix}"``
            (typically the first 8 chars of the HA instance UUID).  Used
            by sister integrations that mirror Sber devices back into HA
            to detect their own loop.  ``None`` disables the marker.
    """
    descriptor: dict = {
        "id": "root",
        "name": "Home Assistant Bridge",
        "default_name": "HA-SberBridge Hub",
        "home": home or "Мой дом",
        "room": room or "Мой дом",
        "hw_version": version,
        "sw_version": version,
        "model": {
            "id": "ID_root_hub",
            "manufacturer": "HA-SberBridge",
            "model": "VHub",
            "description": "HA MQTT Sber Bridge HUB",
            "category": "hub",
            "features": ["online"],
        },
    }
    if ha_serial_prefix:
        descriptor["partner_meta"] = {"ha_serial_number": f"ha-{ha_serial_prefix}"}
    return descriptor

resolve_ha_serial_number ¶

resolve_ha_serial_number(entity, ha_serial_prefix)

Resolve the ha_serial_number marker for one entity.

Priority

Real DeviceEntry.serial_number if linked to a device.
Normalised MAC from DeviceEntry.connections.
Fallback f"ha-{ha_serial_prefix}" (per-HA-instance marker).

Parameters:

Name	Type	Description	Default
`entity`	`BaseEntity`	Entity whose `linked_device` may carry HA registry data.	required
`ha_serial_prefix`	`str`	First N chars of the HA instance UUID (used as fallback when the device has no real serial / MAC).	required

Returns:

Type	Description
`str`	Non-empty string suitable for `partner_meta.ha_serial_number`.

Source code in custom_components/sber_mqtt_bridge/sber_protocol.py

def resolve_ha_serial_number(entity: BaseEntity, ha_serial_prefix: str) -> str:
    """Resolve the ``ha_serial_number`` marker for one entity.

    Priority:
        1. Real ``DeviceEntry.serial_number`` if linked to a device.
        2. Normalised MAC from ``DeviceEntry.connections``.
        3. Fallback ``f"ha-{ha_serial_prefix}"`` (per-HA-instance marker).

    Args:
        entity: Entity whose ``linked_device`` may carry HA registry data.
        ha_serial_prefix: First N chars of the HA instance UUID (used as
            fallback when the device has no real serial / MAC).

    Returns:
        Non-empty string suitable for ``partner_meta.ha_serial_number``.
    """
    device = entity.linked_device or {}
    real_serial = (device.get("serial_number") or "").strip()
    if real_serial:
        return real_serial
    mac = (device.get("mac") or "").strip()
    if mac:
        return mac
    return f"ha-{ha_serial_prefix}"

build_devices_list_json ¶

build_devices_list_json(entities, enabled_entity_ids, redefinitions=None, default_home='', default_room='', auto_parent_id=False, ha_serial_prefix=None)

Build Sber device config JSON for MQTT publish.

The resulting payload is validated against :class:SberConfigPayload (pydantic) before serialisation. Validation failures are logged as warnings but do not prevent publishing.

Parameters:

Name	Type	Description	Default
`entities`	`dict[str, BaseEntity]`	Dict of entity_id -> BaseEntity instances.	required
`enabled_entity_ids`	`list[str]`	List of entity_ids to include.	required
`redefinitions`	`dict[str, dict] \| None`	Optional dict of entity_id -> {home, room, name} overrides.	`None`
`default_home`	`str`	Fallback home name (from HA location_name) when not set via redefinitions. Sber cloud may reject devices without it.	`''`
`default_room`	`str`	Fallback room name when device has no area assigned. Sber cloud may reject devices without a room.	`''`
`auto_parent_id`	`bool`	When True, automatically set `parent_id` to the hub ID (`"root"`) for all child devices that don't have an explicit parent_id. This creates a proper hierarchy in Sber cloud. Defaults to `False` to match `SETTINGS_DEFAULTS` — a caller that forgets this argument must not silently enable a hierarchy the user turned off (issue #44: the DevTools preview did exactly that and always displayed `parent_id: "root"`).	`False`
`ha_serial_prefix`	`str \| None`	Per-HA-instance serial prefix. When provided, every device (including the hub) receives a `partner_meta.ha_serial_number` marker; sister integrations use it for loop-detection. Pass `None` to omit markers.	`None`

Returns:

Type	Description
`str`	Tuple `(json_string, validation_passed, invalid_entity_ids)`.
`bool`	`invalid_entity_ids` lists entities that failed per-device
`list[str]`	validation and were excluded from the payload.

Source code in custom_components/sber_mqtt_bridge/sber_protocol.py

def build_devices_list_json(
    entities: dict[str, BaseEntity],
    enabled_entity_ids: list[str],
    redefinitions: dict[str, dict] | None = None,
    default_home: str = "",
    default_room: str = "",
    auto_parent_id: bool = False,
    ha_serial_prefix: str | None = None,
) -> tuple[str, bool, list[str]]:
    """Build Sber device config JSON for MQTT publish.

    The resulting payload is validated against :class:`SberConfigPayload`
    (pydantic) before serialisation.  Validation failures are logged as
    warnings but do **not** prevent publishing.

    Args:
        entities: Dict of entity_id -> BaseEntity instances.
        enabled_entity_ids: List of entity_ids to include.
        redefinitions: Optional dict of entity_id -> {home, room, name} overrides.
        default_home: Fallback home name (from HA location_name) when not
            set via redefinitions.  Sber cloud may reject devices without it.
        default_room: Fallback room name when device has no area assigned.
            Sber cloud may reject devices without a room.
        auto_parent_id: When True, automatically set ``parent_id`` to the hub
            ID (``"root"``) for all child devices that don't have an explicit
            parent_id.  This creates a proper hierarchy in Sber cloud.
            Defaults to ``False`` to match ``SETTINGS_DEFAULTS`` — a caller
            that forgets this argument must not silently enable a hierarchy
            the user turned off (issue #44: the DevTools preview did exactly
            that and always displayed ``parent_id: "root"``).
        ha_serial_prefix: Per-HA-instance serial prefix.  When provided,
            every device (including the hub) receives a
            ``partner_meta.ha_serial_number`` marker; sister integrations
            use it for loop-detection.  Pass ``None`` to omit markers.

    Returns:
        Tuple ``(json_string, validation_passed, invalid_entity_ids)``.
        ``invalid_entity_ids`` lists entities that failed per-device
        validation and were excluded from the payload.
    """
    device_list: dict[str, Any] = {
        "devices": [build_hub_device(home=default_home, room=default_room, ha_serial_prefix=ha_serial_prefix)]
    }
    invalid_ids: list[str] = []

    for entity_id in enabled_entity_ids:
        entity = entities.get(entity_id)
        if entity is None or not entity.is_filled_by_state:
            continue

        try:
            device_data = entity.to_sber_state()
        except (TypeError, ValueError, KeyError, AttributeError):
            _LOGGER.exception("Error building Sber state for %s", entity_id)
            invalid_ids.append(entity_id)
            continue

        if device_data is None:
            continue

        if redefinitions and entity_id in redefinitions:
            redef = redefinitions[entity_id]
            if redef.get("home"):
                device_data["home"] = redef["home"]
            if redef.get("room"):
                device_data["room"] = redef["room"]
            if redef.get("name"):
                device_data["name"] = redef["name"]

        if "home" not in device_data and default_home:
            device_data["home"] = default_home

        if not device_data.get("room") and default_room:
            device_data["room"] = default_room

        if auto_parent_id and "parent_id" not in device_data:
            device_data["parent_id"] = "root"

        if ha_serial_prefix:
            _inject_ha_serial(device_data, resolve_ha_serial_number(entity, ha_serial_prefix))

        filtered = {k: v for k, v in device_data.items() if v is not None}

        # Advisory: surface names/ids that Sber is known to silently
        # reject or that Salut voice will not recognise.  Non-fatal,
        # we still publish — user sees a clear WARN in the log.
        warn_if_suspicious_id(filtered.get("id", ""))
        warn_if_suspicious_name(filtered.get("id", ""), filtered.get("name", ""))

        # Per-device strict pydantic validation — exclude invalid devices
        # instead of poisoning the entire config payload.
        device_valid, error_msg = validate_device(filtered)
        if not device_valid:
            _LOGGER.warning(
                "Device %s excluded from config (validation failed): %s",
                entity_id,
                error_msg,
            )
            invalid_ids.append(entity_id)
            continue

        device_list["devices"].append(filtered)

    valid = validate_config_payload(device_list)

    return json.dumps(device_list), valid, invalid_ids

build_states_list_json ¶

build_states_list_json(entities, entity_ids=None, enabled_entity_ids=None)

Build Sber state list JSON for MQTT publish.

The resulting payload is validated against :class:SberStatusPayload (pydantic) before serialisation. Validation failures are logged as warnings but do not prevent publishing — Sber may still accept a partially valid payload.

Parameters:

Name	Type	Description	Default
`entities`	`dict[str, BaseEntity]`	Dict of entity_id -> BaseEntity instances.	required
`entity_ids`	`list[str] \| None`	Specific entity_ids to include (None = all enabled).	`None`
`enabled_entity_ids`	`list[str] \| None`	List of enabled entity_ids (used when entity_ids is None).	`None`

Returns:

Type	Description
`tuple[str, bool]`	Tuple of (JSON string, validation_passed bool).

Source code in custom_components/sber_mqtt_bridge/sber_protocol.py

def build_states_list_json(
    entities: dict[str, BaseEntity],
    entity_ids: list[str] | None = None,
    enabled_entity_ids: list[str] | None = None,
) -> tuple[str, bool]:
    """Build Sber state list JSON for MQTT publish.

    The resulting payload is validated against :class:`SberStatusPayload`
    (pydantic) before serialisation.  Validation failures are logged as
    warnings but do **not** prevent publishing — Sber may still accept
    a partially valid payload.

    Args:
        entities: Dict of entity_id -> BaseEntity instances.
        entity_ids: Specific entity_ids to include (None = all enabled).
        enabled_entity_ids: List of enabled entity_ids (used when entity_ids is None).

    Returns:
        Tuple of (JSON string, validation_passed bool).
    """
    states: dict[str, Any] = {"devices": {}}

    if entity_ids is None or len(entity_ids) == 0:
        entity_ids = enabled_entity_ids or list(entities.keys())

    for entity_id in entity_ids:
        entity = entities.get(entity_id)
        if entity is None:
            continue

        if enabled_entity_ids and entity_id not in enabled_entity_ids:
            continue

        try:
            entity_state = entity.to_sber_current_state()
            if entity_state is not None:
                states["devices"] |= entity_state
        except (TypeError, ValueError, KeyError, AttributeError):
            _LOGGER.exception("Error building Sber current state for %s", entity_id)

    if not states["devices"]:
        # Fallback: report hub as online but with no device states.
        # This happens when no entities have state yet (e.g. all unavailable).
        has_entities = bool(entities)
        states["devices"] = {
            "root": {"states": [{"key": "online", "value": {"type": "BOOL", "bool_value": has_entities}}]}
        }

    valid = validate_status_payload(states)

    return json.dumps(states), valid

parse_sber_command ¶

parse_sber_command(payload)

Parse and structurally validate a Sber MQTT command payload.

Per spec (VR-032), the top level must be an object and devices must be a dict keyed by device_id. Untrusted input is sanitized so downstream consumers can rely on the shape devices: dict[str, dict] with states: list[dict] in every device entry:

non-object top level → empty result;
non-dict devices → empty result;
non-dict values inside devices are dropped;
a non-list states is replaced with [];
non-dict elements inside states are dropped.

Parameters:

Name	Type	Description	Default
`payload`	`bytes \| str`	Raw MQTT payload (bytes or str).	required

Returns:

Type	Description
`dict[str, Any]`	Parsed dict with a `devices` key mapping device_id to command
`dict[str, Any]`	dicts, or `{"devices": {}}` on any parse / validation error.

Source code in custom_components/sber_mqtt_bridge/sber_protocol.py

def parse_sber_command(payload: bytes | str) -> dict[str, Any]:
    """Parse and structurally validate a Sber MQTT command payload.

    Per spec (VR-032), the top level must be an object and ``devices``
    must be a dict keyed by device_id.  Untrusted input is sanitized so
    downstream consumers can rely on the shape ``devices: dict[str, dict]``
    with ``states: list[dict]`` in every device entry:

    * non-object top level → empty result;
    * non-dict ``devices`` → empty result;
    * non-dict values inside ``devices`` are dropped;
    * a non-list ``states`` is replaced with ``[]``;
    * non-dict elements inside ``states`` are dropped.

    Args:
        payload: Raw MQTT payload (bytes or str).

    Returns:
        Parsed dict with a ``devices`` key mapping device_id to command
        dicts, or ``{"devices": {}}`` on any parse / validation error.
    """
    data = _load_json_payload(payload)
    if data is None:
        return {"devices": {}}
    if not isinstance(data, dict):
        _LOGGER.warning(
            "Invalid command payload: top level must be object, got %s",
            type(data).__name__,
        )
        return {"devices": {}}
    devices = data.get("devices")
    if not isinstance(devices, dict):
        _LOGGER.warning(
            "Invalid command payload: 'devices' must be dict, got %s",
            type(devices).__name__,
        )
        return {"devices": {}}
    data["devices"] = {
        device_id: _sanitize_command_entry(device_id, cmd_data)
        for device_id, cmd_data in devices.items()
        if _is_valid_command_entry(device_id, cmd_data)
    }
    return data

parse_sber_status_request ¶

parse_sber_status_request(payload)

Parse Sber status request payload.

Non-string elements in the devices list are dropped so callers can safely use the result for dict lookups / set membership.

Returns:

Type	Description
`list[str]`	List of requested entity_ids (empty = all).

Source code in custom_components/sber_mqtt_bridge/sber_protocol.py

def parse_sber_status_request(payload: bytes | str) -> list[str]:
    """Parse Sber status request payload.

    Non-string elements in the ``devices`` list are dropped so callers
    can safely use the result for dict lookups / set membership.

    Returns:
        List of requested entity_ids (empty = all).
    """
    data = _load_json_payload(payload)
    if not isinstance(data, dict):
        return []
    requested = data.get("devices")
    if not isinstance(requested, list):
        return []
    entity_ids = [item for item in requested if isinstance(item, str)]
    if len(entity_ids) != len(requested):
        _LOGGER.warning(
            "Invalid status request: %d non-string device id(s) dropped",
            len(requested) - len(entity_ids),
        )
    if len(entity_ids) == 1 and entity_ids[0] == "":
        return []
    return entity_ids

Протокол Sber Smart Home¶

VERSION module-attribute ¶

build_hub_device ¶

resolve_ha_serial_number ¶

build_devices_list_json ¶

build_states_list_json ¶

parse_sber_command ¶

parse_sber_status_request ¶

VERSION `module-attribute` ¶