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

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

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 
31
32
33
34
35
36
37
38
39
40
41
42
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 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
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)
    state_str = ha_state.get("state", "")
    self.current_state = state_str not in ("off", "idle", "unavailable", "unknown")
    attrs = ha_state.get("attributes", {})

    self._current_temperature = self._safe_int(attrs.get("current_temperature"))
    self._target_temperature = self._safe_int(attrs.get("temperature"))
    self._child_lock = bool(attrs.get("child_lock", False))
    self._water_level = self._safe_int(attrs.get("water_level"))

create_features_list 

create_features_list()

Return Sber feature list for kettle capabilities.

Returns:

Type Description
list[str]

List of Sber feature strings supported by this entity.
Source code in custom_components/sber_mqtt_bridge/devices/kettle.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
def create_features_list(self) -> list[str]:
    """Return Sber feature list for kettle capabilities.

    Returns:
        List of Sber feature strings supported by this entity.
    """
    features = [*super().create_features_list(), "on_off"]
    features.append("kitchen_water_temperature")
    features.append("kitchen_water_temperature_set")
    features.append("kitchen_water_level")
    features.append("kitchen_water_low_level")
    features.append("child_lock")
    return features

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 
74
75
76
77
78
79
80
81
82
83
84
85
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 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
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}}

process_cmd 

process_cmd(cmd_data)

Process Sber kettle commands and produce HA service calls.

Handles the following Sber keys: - on_off: turn_on / turn_off (domain auto-detected from entity_id) - kitchen_water_temperature_set: water_heater.set_temperature

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/kettle.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def process_cmd(self, cmd_data: dict) -> list[dict]:
    """Process Sber kettle commands and produce HA service calls.

    Handles the following Sber keys:
    - ``on_off``: turn_on / turn_off (domain auto-detected from entity_id)
    - ``kitchen_water_temperature_set``: water_heater.set_temperature

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

    Returns:
        List of HA service call dicts to execute.
    """
    results: list[dict] = []
    domain = self.get_entity_domain()

    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)
            results.append(self._build_on_off_service_call(self.entity_id, domain, on))

        elif key == "kitchen_water_temperature_set" and value.get("type") == "INTEGER":
            temp = self._safe_int(value.get("integer_value"))
            if temp is None:
                continue
            results.append(
                {
                    "url": {
                        "type": "call_service",
                        "domain": domain,
                        "service": "set_temperature",
                        "service_data": {"temperature": temp},
                        "target": {"entity_id": self.entity_id},
                    }
                }
            )
    return results

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 
52
53
54
55
56
57
58
59
60
61
62
63
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 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
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)
    ha_status = ha_state.get("state", "")
    self._status = _HA_STATE_TO_SBER_STATUS.get(ha_status, "standby")
    attrs = ha_state.get("attributes", {})
    self._fan_speed = attrs.get("fan_speed")
    self._fan_speed_list = attrs.get("fan_speed_list") or []
    self._battery_level = self._safe_int(attrs.get("battery_level"))
    # cleaning_type from custom attribute (dry/wet/dry_and_wet)
    self._cleaning_type = attrs.get("cleaning_type")

create_features_list 

create_features_list()

Return Sber feature list for vacuum capabilities.

Returns:

Type Description
list[str]

List of Sber feature strings supported by this entity.
Source code in custom_components/sber_mqtt_bridge/devices/vacuum_cleaner.py 
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def create_features_list(self) -> list[str]:
    """Return Sber feature list for vacuum capabilities.

    Returns:
        List of Sber feature strings supported by this entity.
    """
    features = [
        *super().create_features_list(),
        "vacuum_cleaner_command",
        "vacuum_cleaner_status",
    ]
    if self._fan_speed_list:
        features.append("vacuum_cleaner_program")
    if self._cleaning_type is not None:
        features.append("vacuum_cleaner_cleaning_type")
    if self._battery_level is not None:
        features.append("battery_percentage")
    return features

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 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
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 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
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}}

process_cmd 

process_cmd(cmd_data)

Process Sber vacuum commands and produce HA service calls.

Handles the following Sber keys: - vacuum_cleaner_command: start/stop/pause/return_to_base - vacuum_cleaner_program: vacuum.set_fan_speed

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/vacuum_cleaner.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
def process_cmd(self, cmd_data: dict) -> list[dict]:
    """Process Sber vacuum commands and produce HA service calls.

    Handles the following Sber keys:
    - ``vacuum_cleaner_command``: start/stop/pause/return_to_base
    - ``vacuum_cleaner_program``: vacuum.set_fan_speed

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

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

        if key == "vacuum_cleaner_command" and value.get("type") == "ENUM":
            cmd = value.get("enum_value")
            ha_service = _SBER_CMD_TO_HA_SERVICE.get(cmd or "")
            if ha_service is None:
                continue
            results.append(
                {
                    "url": {
                        "type": "call_service",
                        "domain": "vacuum",
                        "service": ha_service,
                        "target": {"entity_id": self.entity_id},
                    }
                }
            )

        elif key == "vacuum_cleaner_program" and value.get("type") == "ENUM":
            fan_speed = value.get("enum_value")
            if not fan_speed:
                continue
            results.append(
                {
                    "url": {
                        "type": "call_service",
                        "domain": "vacuum",
                        "service": "set_fan_speed",
                        "service_data": {"fan_speed": fan_speed},
                        "target": {"entity_id": self.entity_id},
                    }
                }
            )
    return results

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 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
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 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
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)
    self.current_state = ha_state.get("state") == "on"
    attrs = ha_state.get("attributes", {})
    self.target_humidity = attrs.get("humidity")
    self.current_humidity = attrs.get("current_humidity")
    self.available_modes = attrs.get("available_modes") or []
    self.mode = attrs.get("mode")
    self._min_humidity = self._safe_int(attrs.get("min_humidity")) or 35
    self._max_humidity = self._safe_int(attrs.get("max_humidity")) or 85
    water_level = attrs.get("water_level")
    if water_level is not None:
        self._water_level = self._safe_int(water_level)
    else:
        self._water_level = None
    water_low = attrs.get("water_low_level")
    if water_low is not None:
        self._water_low_level = bool(water_low)
    else:
        self._water_low_level = None
    child_lock = attrs.get("child_lock")
    self._child_lock = bool(child_lock) if child_lock is not None else None

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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
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_features_list 

create_features_list()

Return Sber feature list based on available humidifier capabilities.

Dynamically includes work mode and night mode features only when the HA entity supports them.

Returns:

Type Description
list[str]

List of Sber feature strings supported by this entity.
Source code in custom_components/sber_mqtt_bridge/devices/humidifier.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def create_features_list(self) -> list[str]:
    """Return Sber feature list based on available humidifier capabilities.

    Dynamically includes work mode and night mode features only when
    the HA entity supports them.

    Returns:
        List of Sber feature strings supported by this entity.
    """
    features = [*super().create_features_list(), "on_off", "humidity", "hvac_humidity_set"]
    if self.available_modes:
        features.append("hvac_air_flow_power")
    if self._has_night_mode:
        features.append("hvac_night_mode")
    if self._water_level is not None:
        features.append("hvac_water_percentage")
    if self._water_low_level is not None:
        features.append("hvac_water_low_level")
    if self._child_lock is not None:
        features.append("child_lock")
    return features

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 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
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 
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
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}}

process_cmd 

process_cmd(cmd_data)

Process Sber humidifier commands and produce HA service calls.

Handles the following Sber keys: - on_off: turn_on / turn_off - humidity: set_humidity (INTEGER 0-100, plain percentage) - hvac_work_mode: set_mode (ENUM) - hvac_night_mode: set_mode to sleep/normal (BOOL)

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/humidifier.py 
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
def process_cmd(self, cmd_data: dict) -> list[dict]:
    """Process Sber humidifier commands and produce HA service calls.

    Handles the following Sber keys:
    - ``on_off``: turn_on / turn_off
    - ``humidity``: set_humidity (INTEGER 0-100, plain percentage)
    - ``hvac_work_mode``: set_mode (ENUM)
    - ``hvac_night_mode``: set_mode to sleep/normal (BOOL)

    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":
            on = value.get("bool_value", False)
            results.append(self._build_on_off_service_call(self.entity_id, "humidifier", on))
        elif key in ("humidity", "hvac_humidity_set"):
            raw_humidity = value.get("integer_value")
            humidity = self._safe_int(raw_humidity)
            if humidity is None:
                continue
            results.append(
                {
                    "url": {
                        "type": "call_service",
                        "domain": "humidifier",
                        "service": "set_humidity",
                        "service_data": {"humidity": humidity},
                        "target": {"entity_id": self.entity_id},
                    }
                }
            )
        elif key in ("hvac_air_flow_power", "hvac_work_mode"):
            sber_mode = value.get("enum_value")
            if sber_mode is None:
                continue
            # Reverse map: find HA mode that maps to this Sber mode
            ha_mode = sber_mode
            for ha_m in self.available_modes:
                if HA_TO_SBER_HUMIDIFIER_MODE.get(ha_m.lower(), ha_m.lower()) == sber_mode:
                    ha_mode = ha_m
                    break
            results.append(
                {
                    "url": {
                        "type": "call_service",
                        "domain": "humidifier",
                        "service": "set_mode",
                        "service_data": {"mode": ha_mode},
                        "target": {"entity_id": self.entity_id},
                    }
                }
            )
        elif key == "hvac_night_mode":
            night_on = value.get("bool_value", False)
            if night_on:
                # Find the night/sleep mode
                mode = "sleep" if "sleep" in self.available_modes else "night"
                results.append(
                    {
                        "url": {
                            "type": "call_service",
                            "domain": "humidifier",
                            "service": "set_mode",
                            "service_data": {"mode": mode},
                            "target": {"entity_id": self.entity_id},
                        }
                    }
                )
            else:
                # Find the first non-night mode to revert to
                normal_modes = [m for m in self.available_modes if m not in ("sleep", "night")]
                if normal_modes:
                    results.append(
                        {
                            "url": {
                                "type": "call_service",
                                "domain": "humidifier",
                                "service": "set_mode",
                                "service_data": {"mode": normal_modes[0]},
                                "target": {"entity_id": self.entity_id},
                            }
                        }
                    )
    return results