Шторы и ворота

CurtainEntity

Шторы/жалюзи с управлением позицией.

Sber Curtain entity -- maps HA cover entities to Sber curtain category.

CURTAIN_ENTITY_CATEGORY module-attribute

CURTAIN_ENTITY_CATEGORY = 'curtain'

Sber device category for curtain/cover entities.

CurtainEntity

CurtainEntity(entity_data, category=CURTAIN_ENTITY_CATEGORY)

Bases: BaseEntity

Sber curtain entity for cover control with position support.

Maps HA cover entities to the Sber 'curtain' category with support for: - Position control (0-100%) - Open/close/stop commands - Open state reporting

Initialize curtain 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).	CURTAIN_ENTITY_CATEGORY

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

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

    Args:
        entity_data: HA entity registry dict containing entity metadata.
        category: Sber device category (override in subclasses).
    """
    super().__init__(category, entity_data)
    self.current_position = 0
    self._battery_level: int | None = None
    self._battery_low: bool | None = None
    self._signal_strength_raw: int | None = None
    self._open_rate: str | None = None
    self._tilt_position: int | None = None

min_position class-attribute instance-attribute

min_position = 0

Minimum allowed position (0-100%).

max_position class-attribute instance-attribute

max_position = 100

Maximum allowed position (0-100%).

battery_level class-attribute instance-attribute

battery_level = 0

Battery level percentage (0-100%).

current_position class-attribute instance-attribute

current_position = 0

Current cover position (0-100%).

fill_by_ha_state

fill_by_ha_state(ha_state)

Update state from Home Assistant data.

Reads current_position from attributes; falls back to 100 if state is 'opened', otherwise 0. Also reads signal strength from signal_strength, rssi, or linkquality.

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/curtain.py

def fill_by_ha_state(self, ha_state: dict) -> None:
    """Update state from Home Assistant data.

    Reads ``current_position`` from attributes; falls back to 100
    if state is 'opened', otherwise 0. Also reads signal strength
    from ``signal_strength``, ``rssi``, or ``linkquality``.

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

    position = attrs.get("current_position")
    if position is not None:
        try:
            self.current_position = max(0, min(100, int(float(position))))
        except (TypeError, ValueError):
            self.current_position = 100 if self.state in ("open", "opening") else 0
    else:
        self.current_position = 100 if self.state in ("open", "opening") else 0

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

    # Tilt position → light_transmission_percentage (blinds)
    tilt = attrs.get("current_tilt_position")
    self._tilt_position = int(tilt) if tilt is not None else None

    # Motor speed (open_rate) — some Tuya/Zigbee covers expose this
    speed = attrs.get("speed") or attrs.get("motor_speed")
    if speed is not None:
        speed_str = str(speed).lower()
        if speed_str in ("auto", "low", "high"):
            self._open_rate = speed_str
        else:
            self._open_rate = None
    else:
        self._open_rate = None

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

update_linked_data

update_linked_data(role, ha_state)

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

Parameters:

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

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

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

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

process_cmd

process_cmd(cmd_data)

Process Sber curtain commands and produce HA service calls.

Handles the following Sber keys: - open_percentage: set_cover_position (INTEGER 0-100) - cover_position: set_cover_position (INTEGER 0-100) - open_set: open_cover / close_cover / stop_cover (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/curtain.py

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

    Handles the following Sber keys:
    - ``open_percentage``: set_cover_position (INTEGER 0-100)
    - ``cover_position``: set_cover_position (INTEGER 0-100)
    - ``open_set``: open_cover / close_cover / stop_cover (ENUM)

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

    Returns:
        List of HA service call dicts to execute.
    """
    processing_result = []

    for data_item in cmd_data.get("states", []):
        key = data_item.get("key")
        value = data_item.get("value", {})

        if key is None:
            continue

        if key in ("open_percentage", "cover_position"):
            ha_position = self._safe_int(value.get("integer_value"))
            if ha_position is None:
                continue
            ha_position = max(0, min(100, ha_position))
            processing_result.append(
                {
                    "url": {
                        "type": "call_service",
                        "domain": "cover",
                        "service": "set_cover_position",
                        "service_data": {"position": ha_position},
                        "target": {"entity_id": self.entity_id},
                    }
                }
            )

        if key == "open_set":
            action = value.get("enum_value", None)
            if action is None:
                continue

            if action == "open":
                processing_result.append(
                    {
                        "url": {
                            "type": "call_service",
                            "domain": "cover",
                            "service": "open_cover",
                            "target": {"entity_id": self.entity_id},
                        }
                    }
                )

            elif action == "close":
                processing_result.append(
                    {
                        "url": {
                            "type": "call_service",
                            "domain": "cover",
                            "service": "close_cover",
                            "target": {"entity_id": self.entity_id},
                        }
                    }
                )

            elif action == "stop":
                processing_result.append(
                    {
                        "url": {
                            "type": "call_service",
                            "domain": "cover",
                            "service": "stop_cover",
                            "target": {"entity_id": self.entity_id},
                        }
                    }
                )

    return processing_result

create_features_list

create_features_list()

Return Sber feature list for curtain capabilities.

Includes open_percentage, open_set, open_state, and optionally signal_strength features.

Returns:

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

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

def create_features_list(self) -> list[str]:
    """Return Sber feature list for curtain capabilities.

    Includes open_percentage, open_set, open_state, and optionally
    signal_strength features.

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

create_allowed_values_list

create_allowed_values_list()

Return allowed values for open_set, open_percentage, and open_rate features.

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

def create_allowed_values_list(self) -> dict[str, dict]:
    """Return allowed values for open_set, open_percentage, and open_rate features."""
    allowed: dict[str, dict] = {
        "open_set": {
            "type": "ENUM",
            "enum_values": {"values": ["open", "close", "stop"]},
        },
        "open_percentage": {
            "type": "INTEGER",
            "integer_values": {"min": "0", "max": "100", "step": "1"},
        },
    }
    # open_rate is read-only (HA cover has no set_speed service)
    # light_transmission_percentage maps to tilt — handled via open_percentage
    return allowed

to_sber_current_state

to_sber_current_state()

Build Sber current state payload with position, open state, and signal.

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/curtain.py

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

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

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

    states = [
        make_state(SberFeature.ONLINE, make_bool_value(True)),
    ]

    states.append(
        make_state(SberFeature.OPEN_PERCENTAGE, make_integer_value(self._convert_position(self.current_position)))
    )

    # Enforce consistency: open_state must match open_percentage
    sber_pos = self._convert_position(self.current_position)
    # Sber supports: open, close, opening, closing
    state_map = {"open": "open", "opening": "opening", "closed": "close", "closing": "closing"}
    open_state = state_map.get(self.state, "close" if sber_pos == 0 else "open")
    # Force alignment for stable states: percentage > 0 must be 'open'; 0 must be 'close'
    if self.state not in ("opening", "closing"):
        if sber_pos > 0 and open_state == "close":
            open_state = "open"
        elif sber_pos == 0 and open_state == "open":
            open_state = "close"
    states.append(
        make_state(SberFeature.OPEN_STATE, make_enum_value(open_state))
    )

    if self._battery_level is not None:
        states.append(
            make_state(SberFeature.BATTERY_PERCENTAGE, make_integer_value(self._battery_level))
        )
        battery_low = self._battery_low if self._battery_low is not None else self._battery_level < 20
        states.append(
            make_state(SberFeature.BATTERY_LOW_POWER, make_bool_value(battery_low))
        )
    elif self._battery_low is not None:
        states.append(
            make_state(SberFeature.BATTERY_LOW_POWER, make_bool_value(self._battery_low))
        )

    if self._signal_strength_raw is not None:
        states.append(
            make_state(SberFeature.SIGNAL_STRENGTH, make_enum_value(rssi_to_signal_strength(self._signal_strength_raw)))
        )
    if self._open_rate is not None:
        states.append(make_state(SberFeature.OPEN_RATE, make_enum_value(self._open_rate)))
    if self._tilt_position is not None:
        states.append(
            make_state(SberFeature.LIGHT_TRANSMISSION_PERCENTAGE, make_integer_value(self._tilt_position))
        )

    return {self.entity_id: {"states": states}}

WindowBlindEntity

Оконные жалюзи с управлением наклоном.

Sber Window Blind entity -- maps HA blind/shade/shutter covers to Sber window_blind.

WINDOW_BLIND_CATEGORY module-attribute

WINDOW_BLIND_CATEGORY = 'window_blind'

Sber device category for window blind/shade/shutter entities.

WindowBlindEntity

WindowBlindEntity(entity_data)

Bases: CurtainEntity

Sber window blind entity for blind/shade/shutter devices.

Inherits all curtain behavior (position control, open/close/stop) but registers under the Sber 'window_blind' category.

Initialize window blind 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/window_blind.py

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

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

GateEntity

Ворота/калитка.

Sber Gate entity -- maps HA cover (gate/garage_door) entities to Sber gate category.

GATE_ENTITY_CATEGORY module-attribute

GATE_ENTITY_CATEGORY = 'gate'

Sber device category for gate/garage door entities.

GateEntity

GateEntity(entity_data)

Bases: CurtainEntity

Sber gate entity for gate/garage door control.

Inherits all curtain functionality (position, open/close/stop) but uses the Sber 'gate' category instead of 'curtain'.

Maps HA cover entities with device_class 'gate' or 'garage_door'.

Initialize gate 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/gate.py

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

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