Протокол 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.24.5'

Protocol version string included in the hub device descriptor.

build_hub_device

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

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.	''

Source code in custom_components/sber_mqtt_bridge/sber_protocol.py

def build_hub_device(version: str = VERSION, home: str = "", room: str = "") -> 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.
    """
    return {
        "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"],
        },
    }

build_devices_list_json

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

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.	True

Returns:

Type	Description
str	JSON string with the Sber device list 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 = True,
) -> 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.

    Returns:
        JSON string with the Sber device list payload.
    """
    device_list: dict[str, Any] = {"devices": [build_hub_device(home=default_home, room=default_room)]}

    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)
            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"

        filtered = {k: v for k, v in device_data.items() if v is not None}
        device_list["devices"].append(filtered)

    validate_config_payload(device_list)

    return json.dumps(device_list)

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 Sber MQTT command payload.

Per spec (VR-032), devices must be a dict keyed by device_id.

Parameters:

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

Returns:

Type	Description
dict[str, Any]	Parsed dict with 'devices' key, or empty dict on parse error.

Source code in custom_components/sber_mqtt_bridge/sber_protocol.py

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

    Per spec (VR-032), ``devices`` must be a dict keyed by device_id.

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

    Returns:
        Parsed dict with 'devices' key, or empty dict on parse error.
    """
    try:
        data = json.loads(payload)
    except (json.JSONDecodeError, TypeError):
        _LOGGER.warning(
            "Failed to parse Sber command payload: %s", payload[:200] if isinstance(payload, (str, bytes)) else payload
        )
        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": {}}
    return data

parse_sber_status_request

parse_sber_status_request(payload)

Parse Sber status request payload.

Returns 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.

    Returns list of requested entity_ids (empty = all).
    """
    try:
        data = json.loads(payload).get("devices") or []
    except (json.JSONDecodeError, AttributeError, TypeError):
        return []
    else:
        if not isinstance(data, list):
            return []
        if len(data) == 1 and data[0] == "":
            return []
        return data