Реле и розетки¶

RelayEntity¶

Реле: switch, script, button (on/off управление).

Sber Relay entity -- maps HA switch/script/button to Sber relay category.

RELAY_CATEGORY `module-attribute` ¶

RELAY_CATEGORY = 'relay'

Sber device category for relay/switch entities.

RelayEntity ¶

RelayEntity(entity_data, category=RELAY_CATEGORY)

Bases: OnOffEntity

Sber relay entity for on/off control devices.

Maps HA switch, script, and button entities to the Sber 'relay' category. Supports basic on/off toggling via the on_off Sber feature.

Initialize relay entity.

Parameters:

Name	Type	Description	Default
`entity_data`	`dict`	HA entity registry dict containing entity metadata.	required
`category`	`str`	Sber device category (override in subclasses).	`RELAY_CATEGORY`

Source code in custom_components/sber_mqtt_bridge/devices/relay.py

def __init__(self, entity_data: dict, category: str = RELAY_CATEGORY) -> None:
    """Initialize relay entity.

    Args:
        entity_data: HA entity registry dict containing entity metadata.
        category: Sber device category (override in subclasses).
    """
    super().__init__(category, entity_data)

process_cmd ¶

process_cmd(cmd_data)

Process Sber on/off command and produce HA service calls.

Handles the on_off key to generate turn_on/turn_off (or press for button domain) service calls.

State is NOT mutated here — it will be updated when HA fires a state_changed event that is handled by fill_by_ha_state.

Parameters:

Name	Type	Description	Default
`cmd_data`	`dict`	Sber command dict with 'states' list.	required

Returns:

Type	Description
`list[dict]`	List of HA service call dicts to execute.

Source code in custom_components/sber_mqtt_bridge/devices/relay.py

def process_cmd(self, cmd_data: dict) -> list[dict]:
    """Process Sber on/off command and produce HA service calls.

    Handles the ``on_off`` key to generate ``turn_on``/``turn_off`` (or
    ``press`` for button domain) service calls.

    State is NOT mutated here — it will be updated when HA fires a
    ``state_changed`` event that is handled by ``fill_by_ha_state``.

    Args:
        cmd_data: Sber command dict with 'states' list.

    Returns:
        List of HA service call dicts to execute.
    """
    results = []
    for item in cmd_data.get("states", []):
        key = item.get("key")
        value = item.get("value", {})

        if key == "on_off" and value.get("type") == "BOOL":
            on = value.get("bool_value", False)
            domain = self.entity_id.split(".")[0]

            service = "press" if domain == "button" else "turn_on" if on else "turn_off"

            results.append(
                {
                    "url": {
                        "type": "call_service",
                        "domain": domain,
                        "service": service,
                        "target": {"entity_id": self.entity_id},
                    }
                }
            )
    return results

SocketEntity¶

Умная розетка.

Sber Socket entity -- maps HA outlet switches to Sber socket category.

SOCKET_CATEGORY `module-attribute` ¶

SOCKET_CATEGORY = 'socket'

Sber device category for smart socket/outlet entities.

SocketEntity ¶

SocketEntity(entity_data)

Bases: RelayEntity

Sber socket entity for smart outlet/plug devices.

Inherits all relay behavior (on/off control) but registers under the Sber 'socket' category instead of 'relay'.

Initialize socket entity.

Parameters:

Name	Type	Description	Default
`entity_data`	`dict`	HA entity registry dict containing entity metadata.	required

Source code in custom_components/sber_mqtt_bridge/devices/socket_entity.py

def __init__(self, entity_data: dict) -> None:
    """Initialize socket entity.

    Args:
        entity_data: HA entity registry dict containing entity metadata.
    """
    super().__init__(entity_data, category=SOCKET_CATEGORY)

ValveEntity¶

Клапан (водоснабжение, газ).

Sber Valve entity -- maps HA valve entities to Sber valve category.

Uses open_set/open_state features (NOT on_off). Per Sber specification, valve is controlled via ENUM open/close/stop commands. Supports optional battery and signal strength reporting.

VALVE_CATEGORY `module-attribute` ¶

VALVE_CATEGORY = 'valve'

Sber device category for valve entities.

ValveEntity ¶

ValveEntity(entity_data)

Bases: BaseEntity

Sber valve entity for open/close valve control.

Maps HA valve entities to the Sber 'valve' category. Uses open_set (command) and open_state (state) features per Sber specification. Does NOT use on_off.

Optionally reports battery_percentage, battery_low_power, and signal_strength when the HA entity provides these attributes.

Initialize valve entity.

Parameters:

Name	Type	Description	Default
`entity_data`	`dict`	HA entity registry dict containing entity metadata.	required

Source code in custom_components/sber_mqtt_bridge/devices/valve.py

def __init__(self, entity_data: dict) -> None:
    """Initialize valve entity.

    Args:
        entity_data: HA entity registry dict containing entity metadata.
    """
    super().__init__(VALVE_CATEGORY, entity_data)
    self.is_open: bool = False
    self._battery_level: int | None = None
    self._battery_low: bool | None = None
    self._signal_strength_raw: int | None = None

fill_by_ha_state ¶

fill_by_ha_state(ha_state)

Parse HA state and update open/close status, battery, and signal.

Parameters:

Name	Type	Description	Default
`ha_state`	`dict`	HA state dict with 'state' and 'attributes' keys.	required

Source code in custom_components/sber_mqtt_bridge/devices/valve.py

def fill_by_ha_state(self, ha_state: dict) -> None:
    """Parse HA state and update open/close status, battery, and signal.

    Args:
        ha_state: HA state dict with 'state' and 'attributes' keys.
    """
    super().fill_by_ha_state(ha_state)
    self.is_open = ha_state.get("state") == "open"
    attrs = ha_state.get("attributes", {})

    battery = attrs.get("battery") or attrs.get("battery_level")
    if battery is not None:
        try:
            self._battery_level = int(battery)
        except (TypeError, ValueError):
            self._battery_level = None
    else:
        self._battery_level = None

    rssi = attrs.get("signal_strength") or attrs.get("rssi") or attrs.get("linkquality")
    if rssi is not None:
        try:
            self._signal_strength_raw = int(rssi)
        except (TypeError, ValueError):
            self._signal_strength_raw = None
    else:
        self._signal_strength_raw = None

update_linked_data ¶

update_linked_data(role, ha_state)

Inject data from a linked entity (battery, battery_low, signal).

Parameters:

Name	Type	Description	Default
`role`	`str`	Link role name.	required
`ha_state`	`dict`	HA state dict with 'state'.	required

Source code in custom_components/sber_mqtt_bridge/devices/valve.py

def update_linked_data(self, role: str, ha_state: dict) -> None:
    """Inject data from a linked entity (battery, battery_low, signal).

    Args:
        role: Link role name.
        ha_state: HA state dict with 'state'.
    """
    state_val = ha_state.get("state")
    if state_val in (None, "unknown", "unavailable"):
        return
    if role == "battery":
        with contextlib.suppress(TypeError, ValueError):
            self._battery_level = int(float(state_val))
    elif role == "battery_low":
        self._battery_low = state_val == "on"
    elif role == "signal_strength":
        with contextlib.suppress(TypeError, ValueError):
            self._signal_strength_raw = int(float(state_val))

create_features_list ¶

create_features_list()

Return Sber feature list with open_set, open_state, and optional battery/signal.

Returns:

Type	Description
`list[str]`	List of Sber feature strings supported by this entity.

Source code in custom_components/sber_mqtt_bridge/devices/valve.py

def create_features_list(self) -> list[str]:
    """Return Sber feature list with open_set, open_state, and optional battery/signal.

    Returns:
        List of Sber feature strings supported by this entity.
    """
    features = [*super().create_features_list(), "open_set", "open_state"]
    if self._battery_level is not None or self._battery_low is not None:
        features.append("battery_percentage")
        features.append("battery_low_power")
    if self._signal_strength_raw is not None:
        features.append("signal_strength")
    return features

create_allowed_values_list ¶

create_allowed_values_list()

Return allowed values for the open_set feature.

Source code in custom_components/sber_mqtt_bridge/devices/valve.py

def create_allowed_values_list(self) -> dict[str, dict]:
    """Return allowed values for the open_set feature."""
    return {
        "open_set": {
            "type": "ENUM",
            "enum_values": {"values": ["open", "close", "stop"]},
        },
    }

to_sber_current_state ¶

to_sber_current_state()

Build Sber current state payload with online, open_state, battery, and signal.

Returns:

Type	Description
`dict[str, dict]`	Dict mapping entity_id to its Sber state representation.

Source code in custom_components/sber_mqtt_bridge/devices/valve.py

def to_sber_current_state(self) -> dict[str, dict]:
    """Build Sber current state payload with online, open_state, battery, and signal.

    Returns:
        Dict mapping entity_id to its Sber state representation.
    """
    states = [
        make_state(SberFeature.ONLINE, make_bool_value(self._is_online)),
        make_state(SberFeature.OPEN_STATE, make_enum_value("open" if self.is_open else "close")),
    ]
    if self._battery_level is not None:
        states.append(
            make_state(SberFeature.BATTERY_PERCENTAGE, make_integer_value(self._battery_level))
        )
        battery_low = self._battery_low if self._battery_low is not None else self._battery_level < 20
        states.append(
            make_state(SberFeature.BATTERY_LOW_POWER, make_bool_value(battery_low))
        )
    elif self._battery_low is not None:
        states.append(
            make_state(SberFeature.BATTERY_LOW_POWER, make_bool_value(self._battery_low))
        )
    if self._signal_strength_raw is not None:
        states.append(
            make_state(SberFeature.SIGNAL_STRENGTH, make_enum_value(rssi_to_signal_strength(self._signal_strength_raw)))
        )
    return {self.entity_id: {"states": states}}

process_cmd ¶

process_cmd(cmd_data)

Process Sber open_set command and produce HA valve service calls.

Maps open_set ENUM values: - "open" -> open_valve - "close" -> close_valve - "stop" -> stop_valve

State is NOT mutated here -- it will be updated when HA fires a state_changed event that is handled by fill_by_ha_state.

Parameters:

Name	Type	Description	Default
`cmd_data`	`dict`	Sber command dict with 'states' list.	required

Returns:

Type	Description
`list[dict]`	List of HA service call dicts to execute.

Source code in custom_components/sber_mqtt_bridge/devices/valve.py

def process_cmd(self, cmd_data: dict) -> list[dict]:
    """Process Sber open_set command and produce HA valve service calls.

    Maps ``open_set`` ENUM values:
    - ``"open"`` -> ``open_valve``
    - ``"close"`` -> ``close_valve``
    - ``"stop"`` -> ``stop_valve``

    State is NOT mutated here -- it will be updated when HA fires a
    ``state_changed`` event that is handled by ``fill_by_ha_state``.

    Args:
        cmd_data: Sber command dict with 'states' list.

    Returns:
        List of HA service call dicts to execute.
    """
    results: list[dict] = []
    service_map = {
        "open": "open_valve",
        "close": "close_valve",
        "stop": "stop_valve",
    }
    for item in cmd_data.get("states", []):
        key = item.get("key")
        value = item.get("value", {})

        if key == "open_set" and value.get("type") == "ENUM":
            action = value.get("enum_value")
            service = service_map.get(action)
            if service:
                results.append(
                    {
                        "url": {
                            "type": "call_service",
                            "domain": "valve",
                            "service": service,
                            "target": {"entity_id": self.entity_id},
                        }
                    }
                )
    return results

ScenarioButtonEntity¶

Кнопка сценария (input_boolean).

Sber Scenario Button entity -- maps HA input_boolean to Sber scenario_button.

SCENARIO_BUTTON_CATEGORY `module-attribute` ¶

SCENARIO_BUTTON_CATEGORY = 'scenario_button'

Sber device category for scenario button entities.

ScenarioButtonEntity ¶

ScenarioButtonEntity(entity_data)

Bases: BaseEntity

Sber scenario button entity.

Maps HA input_boolean entities to the Sber 'scenario_button' category. Reports button events (click / double_click) based on the boolean state.

Initialize scenario button entity.

Parameters:

Name	Type	Description	Default
`entity_data`	`dict`	HA entity registry dict containing entity metadata.	required

Source code in custom_components/sber_mqtt_bridge/devices/scenario_button.py

def __init__(self, entity_data: dict) -> None:
    """Initialize scenario button entity.

    Args:
        entity_data: HA entity registry dict containing entity metadata.
    """
    super().__init__(SCENARIO_BUTTON_CATEGORY, entity_data)
    self.button_event = "click"

fill_by_ha_state ¶

fill_by_ha_state(ha_state)

Parse HA state and update button event type.

Maps 'on' to 'click' and anything else to 'double_click'.

Parameters:

Name	Type	Description	Default
`ha_state`	`dict`	HA state dict with 'state' key.	required

Source code in custom_components/sber_mqtt_bridge/devices/scenario_button.py

def fill_by_ha_state(self, ha_state: dict) -> None:
    """Parse HA state and update button event type.

    Maps 'on' to 'click' and anything else to 'double_click'.

    Args:
        ha_state: HA state dict with 'state' key.
    """
    super().fill_by_ha_state(ha_state)
    state = ha_state.get("state")
    if state in ("unavailable", "unknown"):
        return
    if state == "on":
        self.button_event = "click"
    else:
        self.button_event = "double_click"

create_features_list ¶

create_features_list()

Return Sber feature list including 'button_event'.

Returns:

Type	Description
`list[str]`	List of Sber feature strings supported by this entity.

Source code in custom_components/sber_mqtt_bridge/devices/scenario_button.py

def create_features_list(self) -> list[str]:
    """Return Sber feature list including 'button_event'.

    Returns:
        List of Sber feature strings supported by this entity.
    """
    return [*super().create_features_list(), "button_event"]

create_allowed_values_list ¶

create_allowed_values_list()

Return allowed values for button_event feature.

Source code in custom_components/sber_mqtt_bridge/devices/scenario_button.py

def create_allowed_values_list(self) -> dict[str, dict]:
    """Return allowed values for button_event feature."""
    return {
        "button_event": {
            "type": "ENUM",
            "enum_values": {"values": ["click", "double_click"]},
        },
    }

to_sber_current_state ¶

to_sber_current_state()

Build Sber current state payload with online and button_event keys.

Returns:

Type	Description
`dict[str, dict]`	Dict mapping entity_id to its Sber state representation.

Source code in custom_components/sber_mqtt_bridge/devices/scenario_button.py

def to_sber_current_state(self) -> dict[str, dict]:
    """Build Sber current state payload with online and button_event keys.

    Returns:
        Dict mapping entity_id to its Sber state representation.
    """
    states = [
        make_state(SberFeature.ONLINE, make_bool_value(self._is_online)),
        make_state(SberFeature.BUTTON_EVENT, make_enum_value(self.button_event)),
    ]
    return {self.entity_id: {"states": states}}

process_cmd ¶

process_cmd(cmd_data)

Process Sber command (no-op for scenario button).

Parameters:

Name	Type	Description	Default
`cmd_data`	`dict`	Sber command dict (ignored).	required

Returns:

Type	Description
`list[dict]`	Empty list -- scenario buttons are read-only.

Source code in custom_components/sber_mqtt_bridge/devices/scenario_button.py

def process_cmd(self, cmd_data: dict) -> list[dict]:
    """Process Sber command (no-op for scenario button).

    Args:
        cmd_data: Sber command dict (ignored).

    Returns:
        Empty list -- scenario buttons are read-only.
    """
    return []

Реле и розетки¶

RelayEntity¶

RELAY_CATEGORY module-attribute ¶

RelayEntity ¶

process_cmd ¶

SocketEntity¶

SOCKET_CATEGORY module-attribute ¶

SocketEntity ¶

ValveEntity¶

VALVE_CATEGORY module-attribute ¶

ValveEntity ¶

fill_by_ha_state ¶

update_linked_data ¶

create_features_list ¶

create_allowed_values_list ¶

to_sber_current_state ¶

process_cmd ¶

ScenarioButtonEntity¶

SCENARIO_BUTTON_CATEGORY module-attribute ¶

ScenarioButtonEntity ¶

fill_by_ha_state ¶

create_features_list ¶

create_allowed_values_list ¶

to_sber_current_state ¶

process_cmd ¶

RELAY_CATEGORY `module-attribute` ¶

SOCKET_CATEGORY `module-attribute` ¶

VALVE_CATEGORY `module-attribute` ¶

SCENARIO_BUTTON_CATEGORY `module-attribute` ¶