Перейти к содержанию

Бытовая техника

KettleEntity

Умный чайник: управление температурой нагрева.

Sber Kettle entity -- maps HA water_heater entities to Sber kettle category.

Supports on/off control, water temperature reading, and target temperature setting.

KETTLE_CATEGORY module-attribute

KETTLE_CATEGORY = 'kettle'

Sber device category for kettle entities.

KettleEntity

KettleEntity(entity_data)

Bases: BaseEntity

Sber kettle entity for smart kettle devices.

Maps HA water_heater entities to the Sber 'kettle' category with support for: - On/off control - Current water temperature reading - Target temperature setting (60-100, step 10) - Child lock (read-only from HA attributes) - Water level and low water level indicators

Initialize kettle 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/kettle.py
def __init__(self, entity_data: dict) -> None:
    """Initialize kettle entity.

    Args:
        entity_data: HA entity registry dict containing entity metadata.
    """
    super().__init__(KETTLE_CATEGORY, entity_data)
    self.current_state: bool = False
    self._current_temperature: int | None = None
    self._target_temperature: int | None = None
    self._child_lock: bool = False
    self._water_level: int | None = None

fill_by_ha_state

fill_by_ha_state(ha_state)

Parse HA state and update kettle attributes.

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/kettle.py
def fill_by_ha_state(self, ha_state: dict) -> None:
    """Parse HA state and update kettle attributes.

    Args:
        ha_state: HA state dict with 'state' and 'attributes' keys.
    """
    super().fill_by_ha_state(ha_state)
    attrs = ha_state.get("attributes", {})
    self._apply_attr_specs(attrs)
    state_str = ha_state.get("state", "")
    self.current_state = state_str not in ("off", "idle", "unavailable", "unknown")

create_allowed_values_list

create_allowed_values_list()

Build allowed values map for temperature setting.

Returns:

Type Description
dict[str, dict]

Dict mapping feature key to its allowed INTEGER values descriptor.

Source code in custom_components/sber_mqtt_bridge/devices/kettle.py
def create_allowed_values_list(self) -> dict[str, dict]:
    """Build allowed values map for temperature setting.

    Returns:
        Dict mapping feature key to its allowed INTEGER values descriptor.
    """
    return {
        "kitchen_water_temperature_set": {
            "type": "INTEGER",
            "integer_values": {"min": "60", "max": "100", "step": "10"},
        }
    }

to_sber_current_state

to_sber_current_state()

Build Sber current state payload with kettle attributes.

Returns:

Type Description
dict[str, dict]

Dict mapping entity_id to its Sber state representation.

Source code in custom_components/sber_mqtt_bridge/devices/kettle.py
def to_sber_current_state(self) -> dict[str, dict]:
    """Build Sber current state payload with kettle attributes.

    Returns:
        Dict mapping entity_id to its Sber state representation.
    """
    states = [
        make_state(SberFeature.ONLINE, make_bool_value(self._is_online)),
        make_state(SberFeature.ON_OFF, make_bool_value(self.current_state)),
    ]
    if self._current_temperature is not None:
        states.append(
            make_state(SberFeature.KITCHEN_WATER_TEMPERATURE, make_integer_value(self._current_temperature))
        )
        # Low water level heuristic: temperature below 30 indicates no/little water
        low_level = self._current_temperature < 30
        states.append(make_state(SberFeature.KITCHEN_WATER_LOW_LEVEL, make_bool_value(low_level)))
    if self._water_level is not None:
        states.append(make_state(SberFeature.KITCHEN_WATER_LEVEL, make_integer_value(self._water_level)))
    if self._target_temperature is not None:
        states.append(
            make_state(SberFeature.KITCHEN_WATER_TEMPERATURE_SET, make_integer_value(self._target_temperature))
        )
    states.append(make_state(SberFeature.CHILD_LOCK, make_bool_value(self._child_lock)))
    return {self.entity_id: {"states": states}}

VacuumCleanerEntity

Робот-пылесос: режимы уборки, управление.

Sber Vacuum Cleaner entity -- maps HA vacuum entities to Sber vacuum_cleaner category.

Supports start/stop/pause/return_to_base commands, status reporting, cleaning program (fan speed), and battery level.

VACUUM_CLEANER_CATEGORY module-attribute

VACUUM_CLEANER_CATEGORY = 'vacuum_cleaner'

Sber device category for vacuum cleaner entities.

VacuumCleanerEntity

VacuumCleanerEntity(entity_data)

Bases: BaseEntity

Sber vacuum cleaner entity for robot vacuum devices.

Maps HA vacuum entities to the Sber 'vacuum_cleaner' category with support for: - Start/stop/pause/return_to_base commands - Status reporting (cleaning, charging, docked, returning, error, paused) - Cleaning program (fan speed) - Battery percentage

Initialize vacuum cleaner 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/vacuum_cleaner.py
def __init__(self, entity_data: dict) -> None:
    """Initialize vacuum cleaner entity.

    Args:
        entity_data: HA entity registry dict containing entity metadata.
    """
    super().__init__(VACUUM_CLEANER_CATEGORY, entity_data)
    self._status: str = "docked"
    self._fan_speed: str | None = None
    self._fan_speed_list: list[str] = []
    self._battery_level: int | None = None
    self._cleaning_type: str | None = None

fill_by_ha_state

fill_by_ha_state(ha_state)

Parse HA state and update vacuum cleaner attributes.

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/vacuum_cleaner.py
def fill_by_ha_state(self, ha_state: dict) -> None:
    """Parse HA state and update vacuum cleaner attributes.

    Args:
        ha_state: HA state dict with 'state' and 'attributes' keys.
    """
    super().fill_by_ha_state(ha_state)
    attrs = ha_state.get("attributes", {})
    self._apply_attr_specs(attrs)
    ha_status = ha_state.get("state", "")
    self._status = _HA_STATE_TO_SBER_STATUS.get(ha_status, "standby")

create_allowed_values_list

create_allowed_values_list()

Build allowed values map for vacuum features.

Returns:

Type Description
dict[str, dict]

Dict mapping feature key to its allowed ENUM values descriptor.

Source code in custom_components/sber_mqtt_bridge/devices/vacuum_cleaner.py
def create_allowed_values_list(self) -> dict[str, dict]:
    """Build allowed values map for vacuum features.

    Returns:
        Dict mapping feature key to its allowed ENUM values descriptor.
    """
    allowed: dict[str, dict] = {
        "vacuum_cleaner_command": {
            "type": "ENUM",
            "enum_values": {"values": list(_SBER_CMD_TO_HA_SERVICE.keys())},
        },
    }
    if self._fan_speed_list:
        allowed["vacuum_cleaner_program"] = {
            "type": "ENUM",
            "enum_values": {"values": self._fan_speed_list},
        }
    # vacuum_cleaner_status and vacuum_cleaner_cleaning_type are read-only:
    # not included in allowed_values to prevent Sber from sending commands
    # for features that have no HA service handler.
    return allowed

to_sber_current_state

to_sber_current_state()

Build Sber current state payload with vacuum attributes.

Returns:

Type Description
dict[str, dict]

Dict mapping entity_id to its Sber state representation.

Source code in custom_components/sber_mqtt_bridge/devices/vacuum_cleaner.py
def to_sber_current_state(self) -> dict[str, dict]:
    """Build Sber current state payload with vacuum attributes.

    Returns:
        Dict mapping entity_id to its Sber state representation.
    """
    states = [
        make_state(SberFeature.ONLINE, make_bool_value(self._is_online)),
        make_state(SberFeature.VACUUM_CLEANER_STATUS, make_enum_value(self._status)),
    ]
    if self._fan_speed:
        states.append(make_state(SberFeature.VACUUM_CLEANER_PROGRAM, make_enum_value(self._fan_speed)))
    if self._cleaning_type:
        states.append(make_state(SberFeature.VACUUM_CLEANER_CLEANING_TYPE, make_enum_value(self._cleaning_type)))
    if self._battery_level is not None:
        states.append(make_state(SberFeature.BATTERY_PERCENTAGE, make_integer_value(self._battery_level)))
    return {self.entity_id: {"states": states}}

HumidifierEntity

Увлажнитель воздуха.

Sber Humidifier entity -- maps HA humidifier entities to Sber hvac_humidifier.

HUMIDIFIER_CATEGORY module-attribute

HUMIDIFIER_CATEGORY = 'hvac_humidifier'

Sber device category for humidifier entities.

HA_TO_SBER_HUMIDIFIER_MODE module-attribute

HA_TO_SBER_HUMIDIFIER_MODE = {'auto': 'auto', 'low': 'low', 'mid': 'medium', 'medium': 'medium', 'high': 'high', 'silent': 'quiet', 'sleep': 'quiet', 'night': 'quiet', 'strong': 'turbo', 'boost': 'turbo'}

Map HA humidifier modes to Sber-standard enum values (case-insensitive lookup).

HumidifierEntity

HumidifierEntity(entity_data)

Bases: BaseEntity

Sber humidifier entity for humidity control devices.

Maps HA humidifier entities to the Sber 'hvac_humidifier' category with support for: - On/off control - Target humidity setting - Work mode selection (when supported by the device)

Initialize humidifier 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/humidifier.py
def __init__(self, entity_data: dict) -> None:
    """Initialize humidifier entity.

    Args:
        entity_data: HA entity registry dict containing entity metadata.
    """
    super().__init__(HUMIDIFIER_CATEGORY, entity_data)
    self.current_state = False
    self.target_humidity = None
    self.current_humidity = None
    self.available_modes: list[str] = []
    self.mode: str | None = None
    self._min_humidity: int = 35
    self._max_humidity: int = 85
    self._water_level: int | None = None
    self._water_low_level: bool | None = None
    self._child_lock: bool | None = None

fill_by_ha_state

fill_by_ha_state(ha_state)

Parse HA state and update all humidifier attributes.

Parameters:

Name Type Description Default
ha_state dict

HA state dict with 'state' and 'attributes' keys. Attributes may include humidity, current_humidity, available_modes, and mode.

required
Source code in custom_components/sber_mqtt_bridge/devices/humidifier.py
def fill_by_ha_state(self, ha_state: dict) -> None:
    """Parse HA state and update all humidifier attributes.

    Args:
        ha_state: HA state dict with 'state' and 'attributes' keys.
            Attributes may include humidity, current_humidity,
            available_modes, and mode.
    """
    super().fill_by_ha_state(ha_state)
    attrs = ha_state.get("attributes", {})
    self._apply_attr_specs(attrs)
    self.current_state = ha_state.get("state") == "on"

update_linked_data

update_linked_data(role, ha_state)

Inject current humidity from a linked sensor entity.

When the HA humidifier entity does not provide current_humidity in its attributes, an external humidity sensor can be linked to supply the value for the Sber humidity feature.

Parameters:

Name Type Description Default
role str

Link role name (only humidity is handled).

required
ha_state dict

HA state dict with 'state' containing the reading.

required
Source code in custom_components/sber_mqtt_bridge/devices/humidifier.py
def update_linked_data(self, role: str, ha_state: dict) -> None:
    """Inject current humidity from a linked sensor entity.

    When the HA humidifier entity does not provide ``current_humidity``
    in its attributes, an external humidity sensor can be linked to
    supply the value for the Sber ``humidity`` feature.

    Args:
        role: Link role name (only ``humidity`` is handled).
        ha_state: HA state dict with 'state' containing the reading.
    """
    if role == "humidity":
        state_val = ha_state.get("state")
        if state_val not in (None, "unknown", "unavailable"):
            with contextlib.suppress(TypeError, ValueError):
                self.current_humidity = float(state_val)

create_allowed_values_list

create_allowed_values_list()

Build allowed values map for enum-based and integer-based features.

Returns:

Type Description
dict[str, dict]

Dict mapping feature key to its allowed values descriptor.

Source code in custom_components/sber_mqtt_bridge/devices/humidifier.py
def create_allowed_values_list(self) -> dict[str, dict]:
    """Build allowed values map for enum-based and integer-based features.

    Returns:
        Dict mapping feature key to its allowed values descriptor.
    """
    allowed: dict[str, dict] = {}
    if self.available_modes:
        sber_modes = [HA_TO_SBER_HUMIDIFIER_MODE.get(m.lower(), m.lower()) for m in self.available_modes]
        # Deduplicate while preserving order
        allowed["hvac_air_flow_power"] = {
            "type": "ENUM",
            "enum_values": {"values": list(dict.fromkeys(sber_modes))},
        }
    allowed["hvac_humidity_set"] = {
        "type": "INTEGER",
        "integer_values": {
            "min": str(self._min_humidity),
            "max": str(self._max_humidity),
            "step": "5",
        },
    }
    return allowed

to_sber_current_state

to_sber_current_state()

Build Sber current state payload with humidifier attributes.

Includes online, on_off, target humidity, work mode, and night mode when values are available.

Per Sber C2C specification, integer_value is serialized as a string.

Returns:

Type Description
dict[str, dict]

Dict mapping entity_id to its Sber state representation.

Source code in custom_components/sber_mqtt_bridge/devices/humidifier.py
def to_sber_current_state(self) -> dict[str, dict]:
    """Build Sber current state payload with humidifier attributes.

    Includes online, on_off, target humidity, work mode, and night mode
    when values are available.

    Per Sber C2C specification, ``integer_value`` is serialized as a string.

    Returns:
        Dict mapping entity_id to its Sber state representation.
    """
    states = [
        make_state(SberFeature.ONLINE, make_bool_value(self._is_online)),
        make_state(SberFeature.ON_OFF, make_bool_value(self.current_state)),
    ]
    if self.current_humidity is not None:
        states.append(make_state(SberFeature.HUMIDITY, make_integer_value(round(self.current_humidity))))
    if self.target_humidity is not None:
        states.append(make_state(SberFeature.HVAC_HUMIDITY_SET, make_integer_value(round(self.target_humidity))))
    if self.mode:
        sber_mode = HA_TO_SBER_HUMIDIFIER_MODE.get(self.mode.lower(), self.mode.lower())
        states.append(make_state(SberFeature.HVAC_AIR_FLOW_POWER, make_enum_value(sber_mode)))
    if self._has_night_mode:
        is_night = self.mode in ("sleep", "night")
        states.append(make_state(SberFeature.HVAC_NIGHT_MODE, make_bool_value(is_night)))
    if self._water_level is not None:
        states.append(make_state(SberFeature.HVAC_WATER_PERCENTAGE, make_integer_value(self._water_level)))
    if self._water_low_level is not None:
        states.append(make_state(SberFeature.HVAC_WATER_LOW_LEVEL, make_bool_value(self._water_low_level)))
    if self._child_lock is not None:
        states.append(make_state(SberFeature.CHILD_LOCK, make_bool_value(self._child_lock)))
    return {self.entity_id: {"states": states}}