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

Климат

ClimateEntity (HVAC AC)

Кондиционер: температура, режим, скорость вентилятора, свинг.

Sber Climate (AC) entity -- maps HA climate entities to Sber hvac_ac category.

CLIMATE_CATEGORY module-attribute

CLIMATE_CATEGORY = 'hvac_ac'

Sber device category for air conditioner / HVAC entities.

HA_TO_SBER_WORK_MODE module-attribute

HA_TO_SBER_WORK_MODE = {'cool': 'cooling', 'heat': 'heating', 'dry': 'dehumidification', 'fan_only': 'ventilation', 'heat_cool': 'auto', 'auto': 'auto', 'eco': 'eco'}

Map HA HVAC modes to Sber work mode enum values.

'off' is excluded — use on_off. Sber also supports 'turbo' and 'quiet' work modes; these are mapped from HA preset_modes (boost→turbo, sleep→quiet) in to_sber_current_state.

SBER_TO_HA_WORK_MODE module-attribute

SBER_TO_HA_WORK_MODE = {'cooling': 'cool', 'heating': 'heat', 'dehumidification': 'dry', 'ventilation': 'fan_only', 'auto': 'auto', 'eco': 'eco'}

Reverse mapping: Sber work mode → HA hvac_mode.

HA_TO_SBER_SWING module-attribute

HA_TO_SBER_SWING = {'off': 'no', 'vertical': 'vertical', 'horizontal': 'horizontal', 'both': 'rotation', 'swing': 'swing', 'auto': 'auto'}

Map HA swing modes to Sber air flow direction values.

SBER_TO_HA_SWING module-attribute

SBER_TO_HA_SWING = {v: k for k, v in (items())}

Reverse mapping: Sber swing → HA swing_mode.

HA_TO_SBER_THERMOSTAT_MODE module-attribute

HA_TO_SBER_THERMOSTAT_MODE = {'heat': 'heating', 'auto': 'auto', 'heat_cool': 'auto'}

Map HA HVAC modes to Sber thermostat mode enum values (simpler devices).

SBER_TO_HA_THERMOSTAT_MODE module-attribute

SBER_TO_HA_THERMOSTAT_MODE = {'heating': 'heat', 'auto': 'auto'}

Reverse mapping: Sber thermostat mode → HA hvac_mode.

HA_TO_SBER_FAN_MODE module-attribute

HA_TO_SBER_FAN_MODE = {'auto': 'auto', 'low': 'low', 'medium': 'medium', 'mid': 'medium', 'high': 'high', 'turbo': 'turbo', 'quiet': 'quiet', 'silent': 'quiet', 'sleep': 'quiet', 'strong': 'turbo', 'boost': 'turbo', 'max': 'turbo', 'min': 'low', '1': 'quiet', '2': 'low', '3': 'medium', '4': 'high', '5': 'turbo'}

Map HA fan modes to Sber air flow power enum values.

ClimateEntity

ClimateEntity(entity_data, category=CLIMATE_CATEGORY, min_temp=16.0, max_temp=32.0, temp_step=1)

Bases: BaseEntity

Sber climate entity for air conditioner control.

Maps HA climate entities to the Sber 'hvac_ac' category with support for: - On/off control - Temperature reading and target temperature setting - Fan mode, swing mode, and HVAC work mode selection - Allowed values for dynamic enum features

Subclasses override class-level flags to restrict features per Sber spec: - _supports_fan: include hvac_air_flow_power (default True for AC) - _supports_swing: include hvac_air_flow_direction (default True for AC) - _supports_work_mode: include hvac_work_mode (default True for AC) - _supports_thermostat_mode: include hvac_thermostat_mode (default False)

Initialize climate 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).

CLIMATE_CATEGORY
min_temp float

Minimum temperature default.

16.0
max_temp float

Maximum temperature default.

32.0
temp_step int

Temperature step for allowed_values (Sber spec varies by category).

1
Source code in custom_components/sber_mqtt_bridge/devices/climate.py
def __init__(
    self,
    entity_data: dict,
    category: str = CLIMATE_CATEGORY,
    min_temp: float = 16.0,
    max_temp: float = 32.0,
    temp_step: int = 1,
) -> None:
    """Initialize climate entity.

    Args:
        entity_data: HA entity registry dict containing entity metadata.
        category: Sber device category (override in subclasses).
        min_temp: Minimum temperature default.
        max_temp: Maximum temperature default.
        temp_step: Temperature step for allowed_values (Sber spec varies by category).
    """
    super().__init__(category, entity_data)
    self.temp_step = temp_step
    self.current_state = False
    self.temperature = None
    self.target_temperature = None
    self.fan_modes = []
    self.swing_modes = []
    self.hvac_modes = []
    self.fan_mode = None
    self.swing_mode = None
    self.hvac_mode = None
    self.min_temp = min_temp
    self.max_temp = max_temp
    self._target_humidity: int | None = None
    self._preset_mode: str | None = None
    self._preset_modes: list[str] = []
    self._child_lock: bool | None = None

fill_by_ha_state

fill_by_ha_state(ha_state)

Parse HA state and update all climate attributes.

Parameters:

Name Type Description Default
ha_state dict

HA state dict with 'state' and 'attributes' keys. Attributes may include current_temperature, temperature, fan_modes, swing_modes, hvac_modes, target_humidity, preset_mode, preset_modes, etc.

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

    Args:
        ha_state: HA state dict with 'state' and 'attributes' keys.
            Attributes may include current_temperature, temperature,
            fan_modes, swing_modes, hvac_modes, target_humidity,
            preset_mode, preset_modes, etc.
    """
    super().fill_by_ha_state(ha_state)
    self.current_state = ha_state.get("state", "off") != "off"
    attrs = ha_state.get("attributes", {})
    raw_temp = self._safe_float(attrs.get("current_temperature"))
    self.temperature = raw_temp if raw_temp is not None and math.isfinite(raw_temp) else None
    raw_target = self._safe_float(attrs.get("temperature"))
    self.target_temperature = raw_target if raw_target is not None and math.isfinite(raw_target) else None
    self.fan_modes = attrs.get("fan_modes") or []
    self.swing_modes = attrs.get("swing_modes") or []
    self.hvac_modes = attrs.get("hvac_modes") or []
    self.fan_mode = attrs.get("fan_mode")
    self.swing_mode = attrs.get("swing_mode")
    self.hvac_mode = ha_state.get("state")
    self.min_temp = self._safe_float(attrs.get("min_temp")) or 16.0
    self.max_temp = self._safe_float(attrs.get("max_temp")) or 32.0
    target_humidity = attrs.get("target_humidity")
    if target_humidity is not None:
        try:
            self._target_humidity = int(target_humidity)
        except (TypeError, ValueError):
            self._target_humidity = None
    else:
        self._target_humidity = None
    self._preset_mode = attrs.get("preset_mode")
    self._preset_modes = attrs.get("preset_modes", [])
    child_lock = attrs.get("child_lock")
    self._child_lock = bool(child_lock) if child_lock is not None else None

create_features_list

create_features_list()

Return Sber feature list based on available climate capabilities.

Dynamically includes fan, swing, HVAC mode, humidity, 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/climate.py
def create_features_list(self) -> list[str]:
    """Return Sber feature list based on available climate capabilities.

    Dynamically includes fan, swing, HVAC mode, humidity, 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", "temperature", "hvac_temp_set"]
    if self._supports_swing and self.swing_modes:
        features.append("hvac_air_flow_direction")
    if self._supports_fan and self.fan_modes:
        features.append("hvac_air_flow_power")
    if self._supports_work_mode and self.hvac_modes:
        features.append("hvac_work_mode")
    if self._supports_thermostat_mode and self.hvac_modes:
        features.append("hvac_thermostat_mode")
    if self._target_humidity is not None:
        features.append("hvac_humidity_set")
    if self._has_night_mode:
        features.append("hvac_night_mode")
    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/climate.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._supports_fan and self.fan_modes:
        sber_fans = [HA_TO_SBER_FAN_MODE.get(m, m) for m in self.fan_modes]
        allowed["hvac_air_flow_power"] = {
            "type": "ENUM",
            "enum_values": {"values": list(dict.fromkeys(sber_fans))},
        }
    if self._supports_swing and self.swing_modes:
        sber_swings = [HA_TO_SBER_SWING.get(m, m) for m in self.swing_modes]
        allowed["hvac_air_flow_direction"] = {
            "type": "ENUM",
            "enum_values": {"values": list(dict.fromkeys(sber_swings))},
        }
    if self._supports_work_mode and self.hvac_modes:
        sber_modes = [HA_TO_SBER_WORK_MODE[m] for m in self.hvac_modes if m in HA_TO_SBER_WORK_MODE]
        if sber_modes:
            allowed["hvac_work_mode"] = {"type": "ENUM", "enum_values": {"values": list(dict.fromkeys(sber_modes))}}
    if self._supports_thermostat_mode and self.hvac_modes:
        sber_modes = [HA_TO_SBER_THERMOSTAT_MODE[m] for m in self.hvac_modes if m in HA_TO_SBER_THERMOSTAT_MODE]
        if sber_modes:
            allowed["hvac_thermostat_mode"] = {
                "type": "ENUM",
                "enum_values": {"values": list(dict.fromkeys(sber_modes))},
            }
    allowed["hvac_temp_set"] = {
        "type": "INTEGER",
        "integer_values": {
            "min": str(int(self.min_temp)),
            "max": str(int(self.max_temp)),
            "step": str(self.temp_step),
        },
    }
    return allowed

to_sber_current_state

to_sber_current_state()

Build Sber current state payload with all climate attributes.

Includes online, on_off, temperature, target temperature, fan mode, swing mode, HVAC work mode, target humidity, and night mode when values are available.

Per Sber specification: - temperature uses x10 encoding (e.g. 22.0C -> 220) - hvac_temp_set uses whole degrees (e.g. 22.0C -> 22) - All integer_value fields are serialized as strings.

Returns:

Type Description
dict[str, dict]

Dict mapping entity_id to its Sber state representation.

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

    Includes online, on_off, temperature, target temperature, fan mode,
    swing mode, HVAC work mode, target humidity, and night mode
    when values are available.

    Per Sber specification:
    - ``temperature`` uses x10 encoding (e.g. 22.0C -> 220)
    - ``hvac_temp_set`` uses whole degrees (e.g. 22.0C -> 22)
    - All ``integer_value`` fields are serialized as strings.

    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.temperature is not None and math.isfinite(self.temperature):
        states.append(
            make_state(SberFeature.TEMPERATURE, make_integer_value(int(self.temperature * 10)))
        )
    if self.target_temperature is not None:
        states.append(
            make_state(SberFeature.HVAC_TEMP_SET, make_integer_value(round(self.target_temperature)))
        )
    if self._supports_fan and self.fan_mode:
        fan_value = HA_TO_SBER_FAN_MODE.get(self.fan_mode, self.fan_mode)
        # Map HA preset modes to Sber air flow power values
        if self._preset_mode == "boost":
            fan_value = "turbo"
        elif self._preset_mode == "sleep" and "quiet" not in (self.fan_modes or []):
            fan_value = "quiet"
        states.append(make_state(SberFeature.HVAC_AIR_FLOW_POWER, make_enum_value(fan_value)))
    if self._supports_swing and self.swing_mode:
        sber_swing = HA_TO_SBER_SWING.get(self.swing_mode, self.swing_mode)
        states.append(make_state(SberFeature.HVAC_AIR_FLOW_DIRECTION, make_enum_value(sber_swing)))
    if self._supports_work_mode and self.hvac_mode and self.hvac_mode != "off":
        # Map HA preset modes to Sber work modes (turbo/quiet)
        if self._preset_mode == "boost":
            sber_mode = "turbo"
        elif self._preset_mode in ("sleep", "eco"):
            sber_mode = "quiet"
        else:
            sber_mode = HA_TO_SBER_WORK_MODE.get(self.hvac_mode)
        if sber_mode:
            states.append(make_state(SberFeature.HVAC_WORK_MODE, make_enum_value(sber_mode)))
    if self._supports_thermostat_mode and self.hvac_mode and self.hvac_mode != "off":
        sber_mode = HA_TO_SBER_THERMOSTAT_MODE.get(self.hvac_mode)
        if sber_mode:
            states.append(make_state(SberFeature.HVAC_THERMOSTAT_MODE, make_enum_value(sber_mode)))
    if self._target_humidity is not None:
        states.append(
            make_state(SberFeature.HVAC_HUMIDITY_SET, make_integer_value(self._target_humidity))
        )
    if self._has_night_mode:
        is_night = self._preset_mode in ("sleep", "night")
        states.append(make_state(SberFeature.HVAC_NIGHT_MODE, make_bool_value(is_night)))
    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 climate commands and produce HA service calls.

Handles the following Sber keys: - on_off: turn_on / turn_off - hvac_temp_set: set_temperature (whole degrees, no scaling) - hvac_air_flow_power: set_fan_mode - hvac_air_flow_direction: set_swing_mode - hvac_work_mode: set_hvac_mode - hvac_humidity_set: set_humidity (INTEGER 0-100) - hvac_night_mode: set_preset_mode (sleep/none)

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/climate.py
def process_cmd(self, cmd_data: dict) -> list[dict]:
    """Process Sber climate commands and produce HA service calls.

    Handles the following Sber keys:
    - ``on_off``: turn_on / turn_off
    - ``hvac_temp_set``: set_temperature (whole degrees, no scaling)
    - ``hvac_air_flow_power``: set_fan_mode
    - ``hvac_air_flow_direction``: set_swing_mode
    - ``hvac_work_mode``: set_hvac_mode
    - ``hvac_humidity_set``: set_humidity (INTEGER 0-100)
    - ``hvac_night_mode``: set_preset_mode (sleep/none)

    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, "climate", on))
        elif key == "hvac_temp_set":
            raw_temp = value.get("integer_value")
            temp = self._safe_float(raw_temp)
            if temp is None:
                continue
            results.append(
                {
                    "url": {
                        "type": "call_service",
                        "domain": "climate",
                        "service": "set_temperature",
                        "service_data": {"temperature": temp},
                        "target": {"entity_id": self.entity_id},
                    }
                }
            )
        elif key == "hvac_air_flow_power":
            sber_mode = value.get("enum_value")
            if not sber_mode:
                continue
            # Reverse map: find HA fan_mode that maps to this Sber mode
            ha_fan = sber_mode
            for fm in self.fan_modes:
                if HA_TO_SBER_FAN_MODE.get(fm, fm) == sber_mode:
                    ha_fan = fm
                    break
            if ha_fan and (not self.fan_modes or ha_fan in self.fan_modes):
                results.append(
                    {
                        "url": {
                            "type": "call_service",
                            "domain": "climate",
                            "service": "set_fan_mode",
                            "service_data": {"fan_mode": ha_fan},
                            "target": {"entity_id": self.entity_id},
                        }
                    }
                )
            elif sber_mode == "turbo" and "boost" in (self._preset_modes or []):
                results.append(
                    {
                        "url": {
                            "type": "call_service",
                            "domain": "climate",
                            "service": "set_preset_mode",
                            "service_data": {"preset_mode": "boost"},
                            "target": {"entity_id": self.entity_id},
                        }
                    }
                )
            elif sber_mode == "quiet" and "sleep" in (self._preset_modes or []):
                results.append(
                    {
                        "url": {
                            "type": "call_service",
                            "domain": "climate",
                            "service": "set_preset_mode",
                            "service_data": {"preset_mode": "sleep"},
                            "target": {"entity_id": self.entity_id},
                        }
                    }
                )
        elif key == "hvac_air_flow_direction":
            sber_swing = value.get("enum_value")
            ha_swing = SBER_TO_HA_SWING.get(sber_swing or "") if sber_swing else None
            if ha_swing and (not self.swing_modes or ha_swing in self.swing_modes):
                results.append(
                    {
                        "url": {
                            "type": "call_service",
                            "domain": "climate",
                            "service": "set_swing_mode",
                            "service_data": {"swing_mode": ha_swing},
                            "target": {"entity_id": self.entity_id},
                        }
                    }
                )
        elif key == "hvac_work_mode":
            sber_mode = value.get("enum_value")
            if not sber_mode:
                continue
            # Sber turbo/quiet work modes → HA preset_modes
            if sber_mode == "turbo" and "boost" in (self._preset_modes or []):
                results.append(
                    {
                        "url": {
                            "type": "call_service",
                            "domain": "climate",
                            "service": "set_preset_mode",
                            "service_data": {"preset_mode": "boost"},
                            "target": {"entity_id": self.entity_id},
                        }
                    }
                )
            elif sber_mode == "quiet" and "sleep" in (self._preset_modes or []):
                results.append(
                    {
                        "url": {
                            "type": "call_service",
                            "domain": "climate",
                            "service": "set_preset_mode",
                            "service_data": {"preset_mode": "sleep"},
                            "target": {"entity_id": self.entity_id},
                        }
                    }
                )
            else:
                ha_mode = SBER_TO_HA_WORK_MODE.get(sber_mode)
                if ha_mode and (not self.hvac_modes or ha_mode in self.hvac_modes):
                    results.append(
                        {
                            "url": {
                                "type": "call_service",
                                "domain": "climate",
                                "service": "set_hvac_mode",
                                "service_data": {"hvac_mode": ha_mode},
                                "target": {"entity_id": self.entity_id},
                            }
                        }
                    )
        elif key == "hvac_thermostat_mode":
            sber_mode = value.get("enum_value")
            ha_mode = SBER_TO_HA_THERMOSTAT_MODE.get(sber_mode or "") if sber_mode else None
            if ha_mode and (not self.hvac_modes or ha_mode in self.hvac_modes):
                results.append(
                    {
                        "url": {
                            "type": "call_service",
                            "domain": "climate",
                            "service": "set_hvac_mode",
                            "service_data": {"hvac_mode": ha_mode},
                            "target": {"entity_id": self.entity_id},
                        }
                    }
                )
        elif key == "hvac_humidity_set":
            humidity = self._safe_int(value.get("integer_value"))
            if humidity is None:
                continue
            humidity = max(0, min(100, humidity))
            results.append(
                {
                    "url": {
                        "type": "call_service",
                        "domain": "climate",
                        "service": "set_humidity",
                        "service_data": {"humidity": humidity},
                        "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 preset mode
                preset = "sleep" if "sleep" in self._preset_modes else "night"
                results.append(
                    {
                        "url": {
                            "type": "call_service",
                            "domain": "climate",
                            "service": "set_preset_mode",
                            "service_data": {"preset_mode": preset},
                            "target": {"entity_id": self.entity_id},
                        }
                    }
                )
            else:
                # Find first non-night preset, or use "none" as last resort
                normal_presets = [
                    p for p in self._preset_modes
                    if p not in ("sleep", "night")
                ]
                fallback_preset = normal_presets[0] if normal_presets else "none"
                if "none" in self._preset_modes or normal_presets:
                    results.append(
                        {
                            "url": {
                                "type": "call_service",
                                "domain": "climate",
                                "service": "set_preset_mode",
                                "service_data": {"preset_mode": fallback_preset},
                                "target": {"entity_id": self.entity_id},
                            }
                        }
                    )
                else:
                    _LOGGER.warning(
                        "Cannot turn off night mode for %s: no non-night presets available",
                        self.entity_id,
                    )
    return results

HvacRadiatorEntity

Радиатор отопления.

Sber HVAC Radiator entity -- maps HA radiator climate entities to Sber hvac_radiator.

HVAC_RADIATOR_CATEGORY module-attribute

HVAC_RADIATOR_CATEGORY = 'hvac_radiator'

Sber device category for radiator/heater climate entities.

HvacRadiatorEntity

HvacRadiatorEntity(entity_data)

Bases: ClimateEntity

Sber HVAC radiator entity for heating devices.

Inherits climate behavior but registers under the Sber 'hvac_radiator' category with radiator-appropriate temperature defaults (25-40 C).

Per Sber spec, radiators only support: on_off, online, temperature, hvac_temp_set. Fan, swing, and work mode features are disabled.

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

    Args:
        entity_data: HA entity registry dict containing entity metadata.
    """
    super().__init__(
        entity_data,
        category=HVAC_RADIATOR_CATEGORY,
        min_temp=25.0,
        max_temp=40.0,
        temp_step=5,
    )

HvacBoilerEntity

Бойлер/котёл.

Sber HVAC Boiler entity -- maps HA water heater climate entities to Sber hvac_boiler.

HVAC_BOILER_CATEGORY module-attribute

HVAC_BOILER_CATEGORY = 'hvac_boiler'

Sber device category for boiler/water heater entities.

HvacBoilerEntity

HvacBoilerEntity(entity_data)

Bases: ClimateEntity

Sber HVAC boiler entity for water heater devices.

Inherits climate behavior but registers under the Sber 'hvac_boiler' category with boiler-appropriate temperature defaults (25-80 C).

Per Sber spec, boilers use hvac_thermostat_mode (NOT hvac_work_mode). Fan and swing features are disabled.

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

    Args:
        entity_data: HA entity registry dict containing entity metadata.
    """
    super().__init__(
        entity_data,
        category=HVAC_BOILER_CATEGORY,
        min_temp=25.0,
        max_temp=80.0,
        temp_step=5,
    )

HvacFanEntity

Вентилятор.

Sber HVAC Fan entity -- maps HA fan entities to Sber hvac_fan category.

Supports on/off control and fan speed via the hvac_air_flow_power feature.

HVAC_FAN_CATEGORY module-attribute

HVAC_FAN_CATEGORY = 'hvac_fan'

Sber device category for fan entities.

SBER_SPEED_VALUES module-attribute

SBER_SPEED_VALUES = ['auto', 'high', 'low', 'medium', 'quiet', 'turbo']

Allowed Sber ENUM values for hvac_air_flow_power (per Sber C2C spec).

HvacFanEntity

HvacFanEntity(entity_data)

Bases: BaseEntity

Sber fan entity for ventilator control.

Maps HA fan entities to the Sber 'hvac_fan' category with support for: - On/off control - Fan speed via preset_mode or percentage-based speed mapping

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

    Args:
        entity_data: HA entity registry dict containing entity metadata.
    """
    super().__init__(HVAC_FAN_CATEGORY, entity_data)
    self.current_state: bool = False
    self.preset_mode: str | None = None
    self.preset_modes: list[str] = []
    self.percentage: int | None = None

fill_by_ha_state

fill_by_ha_state(ha_state)

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

    Args:
        ha_state: HA state dict with 'state' and 'attributes' keys.
    """
    super().fill_by_ha_state(ha_state)
    self.current_state = ha_state.get("state") != "off"
    attrs = ha_state.get("attributes", {})
    self.preset_modes = attrs.get("preset_modes") or []
    self.preset_mode = attrs.get("preset_mode")
    self.percentage = attrs.get("percentage")

create_features_list

create_features_list()

Return Sber feature list for fan capabilities.

Only includes hvac_air_flow_power when the HA entity supports speed control. Simple on/off fans (e.g. relay overridden as fan) get only on_off.

Returns:

Type Description
list[str]

List of Sber feature strings supported by this entity.

Source code in custom_components/sber_mqtt_bridge/devices/hvac_fan.py
def create_features_list(self) -> list[str]:
    """Return Sber feature list for fan capabilities.

    Only includes hvac_air_flow_power when the HA entity supports speed
    control. Simple on/off fans (e.g. relay overridden as fan) get only
    on_off.

    Returns:
        List of Sber feature strings supported by this entity.
    """
    features = [*super().create_features_list(), "on_off"]
    if self._supports_speed:
        features.append("hvac_air_flow_power")
    return features

create_allowed_values_list

create_allowed_values_list()

Build allowed values map for fan speed feature.

Returns empty dict when the fan has no speed support.

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/hvac_fan.py
def create_allowed_values_list(self) -> dict[str, dict]:
    """Build allowed values map for fan speed feature.

    Returns empty dict when the fan has no speed support.

    Returns:
        Dict mapping feature key to its allowed ENUM values descriptor.
    """
    if not self._supports_speed:
        return {}
    return {
        "hvac_air_flow_power": {
            "type": "ENUM",
            "enum_values": {"values": SBER_SPEED_VALUES},
        }
    }

to_sber_current_state

to_sber_current_state()

Build Sber current state payload with fan 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/hvac_fan.py
def to_sber_current_state(self) -> dict[str, dict]:
    """Build Sber current state payload with fan 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._supports_speed:
        speed = self._get_sber_speed() or "auto"
        states.append(make_state(SberFeature.HVAC_AIR_FLOW_POWER, make_enum_value(speed)))
    return {self.entity_id: {"states": states}}

process_cmd

process_cmd(cmd_data)

Process Sber fan commands and produce HA service calls.

Handles the following Sber keys: - on_off: fan.turn_on / fan.turn_off - hvac_air_flow_power: fan.set_preset_mode (if mode matches) or fan.set_percentage (converted from speed ENUM)

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/hvac_fan.py
def process_cmd(self, cmd_data: dict) -> list[dict]:
    """Process Sber fan commands and produce HA service calls.

    Handles the following Sber keys:
    - ``on_off``: fan.turn_on / fan.turn_off
    - ``hvac_air_flow_power``: fan.set_preset_mode (if mode matches)
      or fan.set_percentage (converted from speed ENUM)

    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 == "on_off" and value.get("type") == "BOOL":
            on = value.get("bool_value", False)
            results.append(self._build_on_off_service_call(self.entity_id, "fan", on))

        elif key == "hvac_air_flow_power" and value.get("type") == "ENUM":
            speed = value.get("enum_value")
            if not speed:
                continue
            # Try preset_mode first if the HA entity supports it
            if speed in self.preset_modes:
                results.append(
                    {
                        "url": {
                            "type": "call_service",
                            "domain": "fan",
                            "service": "set_preset_mode",
                            "service_data": {"preset_mode": speed},
                            "target": {"entity_id": self.entity_id},
                        }
                    }
                )
            else:
                pct = _SBER_SPEED_TO_PERCENTAGE.get(speed)
                if pct is not None:
                    if pct == 0:
                        # 'auto' mode -- turn on without specific speed
                        results.append(
                            {
                                "url": {
                                    "type": "call_service",
                                    "domain": "fan",
                                    "service": "turn_on",
                                    "target": {"entity_id": self.entity_id},
                                }
                            }
                        )
                    else:
                        results.append(
                            {
                                "url": {
                                    "type": "call_service",
                                    "domain": "fan",
                                    "service": "set_percentage",
                                    "service_data": {"percentage": pct},
                                    "target": {"entity_id": self.entity_id},
                                }
                            }
                        )
    return results

HvacHeaterEntity

Обогреватель.

Sber HVAC Heater entity -- maps HA heater climate entities to Sber hvac_heater.

HVAC_HEATER_CATEGORY module-attribute

HVAC_HEATER_CATEGORY = 'hvac_heater'

Sber device category for heater climate entities.

HvacHeaterEntity

HvacHeaterEntity(entity_data)

Bases: ClimateEntity

Sber HVAC heater entity for space heater devices.

Inherits climate behavior but registers under the Sber 'hvac_heater' category with heater-appropriate temperature defaults (5-40 C).

Per Sber spec, heaters support: hvac_air_flow_power, hvac_temp_set, hvac_thermostat_mode, on_off, online, temperature. No swing, no work mode.

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

    Args:
        entity_data: HA entity registry dict containing entity metadata.
    """
    super().__init__(
        entity_data,
        category=HVAC_HEATER_CATEGORY,
        min_temp=5.0,
        max_temp=40.0,
    )

HvacUnderfloorHeatingEntity

Тёплый пол.

Sber HVAC Underfloor Heating entity -- maps HA underfloor heating climate entities.

HVAC_UNDERFLOOR_CATEGORY module-attribute

HVAC_UNDERFLOOR_CATEGORY = 'hvac_underfloor_heating'

Sber device category for underfloor heating entities.

HvacUnderfloorEntity

HvacUnderfloorEntity(entity_data)

Bases: ClimateEntity

Sber HVAC underfloor heating entity.

Inherits climate behavior but registers under the Sber 'hvac_underfloor_heating' category with underfloor-appropriate temperature defaults (25-50 C).

Per Sber spec, underfloor heating uses hvac_thermostat_mode (NOT hvac_work_mode). Fan and swing features are disabled.

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

    Args:
        entity_data: HA entity registry dict containing entity metadata.
    """
    super().__init__(
        entity_data,
        category=HVAC_UNDERFLOOR_CATEGORY,
        min_temp=25.0,
        max_temp=50.0,
        temp_step=5,
    )

HvacAirPurifierEntity

Очиститель воздуха.

Sber HVAC Air Purifier entity -- maps HA fan entities to Sber hvac_air_purifier category.

Supports on/off control, fan speed via hvac_air_flow_power, and read-only features: ionization, night mode, aromatization, filter/ionizer replacement.

HVAC_AIR_PURIFIER_CATEGORY module-attribute

HVAC_AIR_PURIFIER_CATEGORY = 'hvac_air_purifier'

Sber device category for air purifier entities.

HvacAirPurifierEntity

HvacAirPurifierEntity(entity_data)

Bases: BaseEntity

Sber air purifier entity for purifier fan devices.

Maps HA fan entities (with device_class purifier/air_purifier) to the Sber 'hvac_air_purifier' category with support for: - On/off control - Fan speed via preset_mode or percentage - Read-only flags: ionization, night mode, aromatization, filter replacement, ionizer replacement, decontamination

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

    Args:
        entity_data: HA entity registry dict containing entity metadata.
    """
    super().__init__(HVAC_AIR_PURIFIER_CATEGORY, entity_data)
    self.current_state: bool = False
    self.preset_mode: str | None = None
    self.preset_modes: list[str] = []
    self.percentage: int | None = None
    self._ionization: bool = False
    self._night_mode: bool = False
    self._aromatization: bool = False
    self._replace_filter: bool = False
    self._replace_ionizator: bool = False
    self._decontaminate: bool = False

fill_by_ha_state

fill_by_ha_state(ha_state)

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

    Args:
        ha_state: HA state dict with 'state' and 'attributes' keys.
    """
    super().fill_by_ha_state(ha_state)
    self.current_state = ha_state.get("state") != "off"
    attrs = ha_state.get("attributes", {})
    self.preset_modes = attrs.get("preset_modes") or []
    self.preset_mode = attrs.get("preset_mode")
    self.percentage = attrs.get("percentage")
    self._ionization = bool(attrs.get("ionization", False))
    self._night_mode = bool(attrs.get("night_mode", False))
    self._aromatization = bool(attrs.get("aromatization", False))
    self._replace_filter = bool(attrs.get("replace_filter", False))
    self._replace_ionizator = bool(attrs.get("replace_ionizator", False))
    self._decontaminate = bool(attrs.get("decontaminate", False))

create_features_list

create_features_list()

Return Sber feature list for air purifier capabilities.

Returns:

Type Description
list[str]

List of Sber feature strings supported by this entity.

Source code in custom_components/sber_mqtt_bridge/devices/hvac_air_purifier.py
def create_features_list(self) -> list[str]:
    """Return Sber feature list for air purifier capabilities.

    Returns:
        List of Sber feature strings supported by this entity.
    """
    features = [
        *super().create_features_list(),
        "on_off",
        "hvac_air_flow_power",
        "hvac_ionization",
        "hvac_night_mode",
        "hvac_aromatization",
        "hvac_replace_filter",
        "hvac_replace_ionizator",
    ]
    if self._decontaminate:
        features.append("hvac_decontaminate")
    return features

create_allowed_values_list

create_allowed_values_list()

Build allowed values map for air flow power feature.

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/hvac_air_purifier.py
def create_allowed_values_list(self) -> dict[str, dict]:
    """Build allowed values map for air flow power feature.

    Returns:
        Dict mapping feature key to its allowed ENUM values descriptor.
    """
    return {
        "hvac_air_flow_power": {
            "type": "ENUM",
            "enum_values": {"values": SBER_SPEED_VALUES},
        }
    }

to_sber_current_state

to_sber_current_state()

Build Sber current state payload with air purifier 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/hvac_air_purifier.py
def to_sber_current_state(self) -> dict[str, dict]:
    """Build Sber current state payload with air purifier 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)),
    ]
    speed = self._get_sber_speed()
    if speed:
        states.append(make_state(SberFeature.HVAC_AIR_FLOW_POWER, make_enum_value(speed)))
    states.extend(
        [
            make_state(SberFeature.HVAC_IONIZATION, make_bool_value(self._ionization)),
            make_state(SberFeature.HVAC_NIGHT_MODE, make_bool_value(self._night_mode)),
            make_state(SberFeature.HVAC_AROMATIZATION, make_bool_value(self._aromatization)),
            make_state(SberFeature.HVAC_REPLACE_FILTER, make_bool_value(self._replace_filter)),
            make_state(SberFeature.HVAC_REPLACE_IONIZATOR, make_bool_value(self._replace_ionizator)),
        ]
    )
    if self._decontaminate:
        states.append(make_state(SberFeature.HVAC_DECONTAMINATE, make_bool_value(self._decontaminate)))
    return {self.entity_id: {"states": states}}

process_cmd

process_cmd(cmd_data)

Process Sber air purifier commands and produce HA service calls.

Handles the following Sber keys: - on_off: fan.turn_on / fan.turn_off - hvac_air_flow_power: fan.set_preset_mode (if mode matches) or fan.set_percentage (converted from speed ENUM)

Other features (ionization, night_mode, etc.) are read-only.

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/hvac_air_purifier.py
def process_cmd(self, cmd_data: dict) -> list[dict]:
    """Process Sber air purifier commands and produce HA service calls.

    Handles the following Sber keys:
    - ``on_off``: fan.turn_on / fan.turn_off
    - ``hvac_air_flow_power``: fan.set_preset_mode (if mode matches)
      or fan.set_percentage (converted from speed ENUM)

    Other features (ionization, night_mode, etc.) are read-only.

    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 == "on_off" and value.get("type") == "BOOL":
            on = value.get("bool_value", False)
            results.append(self._build_on_off_service_call(self.entity_id, "fan", on))

        elif key == "hvac_air_flow_power" and value.get("type") == "ENUM":
            speed = value.get("enum_value")
            if not speed:
                continue
            if speed in self.preset_modes:
                results.append(
                    {
                        "url": {
                            "type": "call_service",
                            "domain": "fan",
                            "service": "set_preset_mode",
                            "service_data": {"preset_mode": speed},
                            "target": {"entity_id": self.entity_id},
                        }
                    }
                )
            else:
                pct = _SBER_SPEED_TO_PERCENTAGE.get(speed)
                if pct is not None:
                    if pct == 0:
                        results.append(
                            {
                                "url": {
                                    "type": "call_service",
                                    "domain": "fan",
                                    "service": "turn_on",
                                    "target": {"entity_id": self.entity_id},
                                }
                            }
                        )
                    else:
                        results.append(
                            {
                                "url": {
                                    "type": "call_service",
                                    "domain": "fan",
                                    "service": "set_percentage",
                                    "service_data": {"percentage": pct},
                                    "target": {"entity_id": self.entity_id},
                                }
                            }
                        )
    return results