Базовые классы¶

BaseEntity¶

Абстрактный базовый класс для всех устройств Sber Smart Home.

Base entity class for Sber Smart Home device representations.

All device types (light, relay, climate, etc.) inherit from BaseEntity. It defines the contract for converting between HA states and Sber JSON protocol.

ROLE_BATTERY `module-attribute` ¶

ROLE_BATTERY = LinkableRole('battery', frozenset({'sensor'}), frozenset({'battery'}))

Battery percentage sensor (sensor domain, battery device_class).

ROLE_BATTERY_LOW `module-attribute` ¶

ROLE_BATTERY_LOW = LinkableRole('battery_low', frozenset({'binary_sensor'}), frozenset({'battery'}))

Low-battery binary sensor (binary_sensor domain, battery device_class).

ROLE_SIGNAL `module-attribute` ¶

ROLE_SIGNAL = LinkableRole('signal_strength', frozenset({'sensor'}), frozenset({'signal_strength'}))

Signal strength sensor (sensor domain, signal_strength device_class).

ROLE_TEMPERATURE `module-attribute` ¶

ROLE_TEMPERATURE = LinkableRole('temperature', frozenset({'sensor'}), frozenset({'temperature'}))

Temperature sensor (sensor domain, temperature device_class).

ROLE_HUMIDITY `module-attribute` ¶

ROLE_HUMIDITY = LinkableRole('humidity', frozenset({'sensor'}), frozenset({'humidity'}))

Humidity sensor (sensor domain, humidity device_class).

SENSOR_LINK_ROLES `module-attribute` ¶

SENSOR_LINK_ROLES = (ROLE_BATTERY, ROLE_BATTERY_LOW, ROLE_SIGNAL)

Common linkable roles for battery-powered devices (sensors, covers, valves).

ALL_LINKABLE_ROLES `module-attribute` ¶

ALL_LINKABLE_ROLES = (ROLE_BATTERY, ROLE_BATTERY_LOW, ROLE_SIGNAL, ROLE_TEMPERATURE, ROLE_HUMIDITY)

Global registry of all known linkable roles for display in UI.

LinkableRole `dataclass` ¶

LinkableRole(role, domains, device_classes)

Describes a linkable sensor role that a device class accepts.

Each role declares which HA domain + device_class combinations it matches. Device classes declare which roles they accept via LINKABLE_ROLES. This eliminates the need for separate mapping dicts and domain overrides.

Attributes:

Name	Type	Description
`role`	`str`	Link role name (e.g. `"battery"`, `"humidity"`).
`domains`	`frozenset[str]`	Accepted HA entity domains (e.g. `{"sensor"}`).
`device_classes`	`frozenset[str]`	Accepted HA device_class values (e.g. `{"humidity"}`).

matches ¶

matches(domain, device_class)

Check if an HA entity matches this role.

Parameters:

Name	Type	Description	Default
`domain`	`str`	HA entity domain (e.g. `"sensor"`).	required
`device_class`	`str`	HA original_device_class (e.g. `"humidity"`).	required

Returns:

Type	Description
`bool`	True if both domain and device_class match.

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

def matches(self, domain: str, device_class: str) -> bool:
    """Check if an HA entity matches this role.

    Args:
        domain: HA entity domain (e.g. ``"sensor"``).
        device_class: HA original_device_class (e.g. ``"humidity"``).

    Returns:
        True if both domain and device_class match.
    """
    return domain in self.domains and device_class in self.device_classes

BaseEntity ¶

BaseEntity(category, entity_data)

Bases: ABC

Abstract base class for all Sber device entities.

Defines the interface that all device types must implement: - fill_by_ha_state: Parse HA state into internal representation - create_features_list: Return Sber feature names - to_sber_state: Build Sber device config JSON - to_sber_current_state: Build Sber current state JSON - process_cmd: Handle Sber commands, return HA service calls - process_state_change: Handle HA state change events

Initialize base entity from HA entity registry data.

Parameters:

Name	Type	Description	Default
`category`	`str`	Sber device category (e.g., 'light', 'relay', 'sensor_temp').	required
`entity_data`	`dict`	Dict with HA entity registry fields.	required

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

def __init__(self, category: str, entity_data: dict) -> None:
    """Initialize base entity from HA entity registry data.

    Args:
        category: Sber device category (e.g., 'light', 'relay', 'sensor_temp').
        entity_data: Dict with HA entity registry fields.
    """
    self.category = category
    self.attributes: dict = {}
    self.state = None
    self.is_filled_by_state = False
    self.linked_device = None
    self.nicknames: list[str] = []
    self.groups: list[str] = []
    self.parent_entity_id: str | None = None
    self.partner_meta: dict[str, str] = {}
    self.extra_features: list[str] = []
    self.removed_features: list[str] = []
    self._previous_sber_state: dict | None = None
    self._linked_entities: dict[str, str] = {}

    if entity_data:
        self.area_id = entity_data.get("area_id", "")
        self.categories = entity_data.get("categories", [])
        self.config_entry_id = entity_data.get("config_entry_id")
        self.config_subentry_id = entity_data.get("config_subentry_id")
        self.device_id = entity_data.get("device_id")
        self.disabled_by = entity_data.get("disabled_by")
        self.entity_category = entity_data.get("entity_category")
        self.entity_id = entity_data.get("entity_id")
        self.has_entity_name = entity_data.get("has_entity_name")
        self.hidden_by = entity_data.get("hidden_by")
        self.icon = entity_data.get("icon")
        self.id = entity_data.get("id")
        self.labels = entity_data.get("labels", [])
        self.name = entity_data.get("name")
        self.options = entity_data.get("options", {})
        self.original_name = entity_data.get("original_name")
        self.platform = entity_data.get("platform")
        self.translation_key = entity_data.get("translation_key")
        self.unique_id = entity_data.get("unique_id")

        if not self.name:
            self.name = self.original_name or self.entity_id

        if self.area_id is None:
            self.area_id = ""

LINKABLE_ROLES `class-attribute` ¶

LINKABLE_ROLES = ()

Linkable roles this device class accepts. Override in subclasses.

effective_room `property` ¶

effective_room

Return the best available room name.

Priority: entity area_id → device area_id → empty string.

is_online `property` ¶

is_online

Public accessor for entity online status.

Returns:

Type	Description
`bool`	True if the entity state indicates it is reachable.

fill_by_ha_state ¶

fill_by_ha_state(ha_entity_state)

Parse HA state dict and update internal state.

Parameters:

Name	Type	Description	Default
`ha_entity_state`	`dict`	Dict with 'state' and 'attributes' keys from HA.	required

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

def fill_by_ha_state(self, ha_entity_state: dict) -> None:
    """Parse HA state dict and update internal state.

    Args:
        ha_entity_state: Dict with 'state' and 'attributes' keys from HA.
    """
    self.state = ha_entity_state.get("state")
    self.attributes = copy.deepcopy(ha_entity_state.get("attributes", {}))
    self.is_filled_by_state = True

    # Use friendly_name from HA state when entity name was not customized
    # by the user (still matches original_name or entity_id).
    # This handles has_entity_name=True entities where original_name is
    # just a suffix ("Temperature") but friendly_name is the full name
    # ("Climate Sensor Temperature").
    friendly = self.attributes.get("friendly_name")
    if friendly and self.name in (self.entity_id, self.original_name):
        self.name = friendly

is_group_state ¶

is_group_state()

Check if this entity represents a group of other entities.

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

def is_group_state(self) -> bool:
    """Check if this entity represents a group of other entities."""
    entity_list = self.attributes.get("entity_id")
    return entity_list is not None and len(entity_list) > 0

create_features_list ¶

create_features_list()

Return list of Sber feature names supported by this entity.

Base implementation returns ['online']. Child classes extend this.

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

def create_features_list(self) -> list[str]:
    """Return list of Sber feature names supported by this entity.

    Base implementation returns ['online']. Child classes extend this.
    """
    return ["online"]

create_allowed_values_list ¶

create_allowed_values_list()

Return allowed values map for Sber model descriptor.

Override in subclasses to provide allowed_values for features that require INTEGER ranges or ENUM value lists.

Returns:

Type	Description
`dict[str, dict]`	Dict mapping feature key to its allowed values descriptor,
`dict[str, dict]`	or empty dict if no allowed values needed.

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

def create_allowed_values_list(self) -> dict[str, dict]:
    """Return allowed values map for Sber model descriptor.

    Override in subclasses to provide allowed_values for features
    that require INTEGER ranges or ENUM value lists.

    Returns:
        Dict mapping feature key to its allowed values descriptor,
        or empty dict if no allowed values needed.
    """
    return {}

create_dependencies ¶

create_dependencies()

Return feature dependencies map for Sber model descriptor.

Override in subclasses to declare feature dependencies (e.g., light_colour depends on light_mode == 'colour').

Returns:

Type	Description
`dict[str, dict]`	Dict mapping feature key to its dependency descriptor,
`dict[str, dict]`	or empty dict if no dependencies needed.

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

def create_dependencies(self) -> dict[str, dict]:
    """Return feature dependencies map for Sber model descriptor.

    Override in subclasses to declare feature dependencies
    (e.g., light_colour depends on light_mode == 'colour').

    Returns:
        Dict mapping feature key to its dependency descriptor,
        or empty dict if no dependencies needed.
    """
    return {}

get_final_features_list ¶

get_final_features_list()

Return features list with user overrides applied.

Removes features from removed_features and appends features from extra_features. Duplicate-safe.

Returns:

Type	Description
`list[str]`	Final list of Sber feature names.

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

def get_final_features_list(self) -> list[str]:
    """Return features list with user overrides applied.

    Removes features from ``removed_features`` and appends features
    from ``extra_features``.  Duplicate-safe.

    Returns:
        Final list of Sber feature names.
    """
    features = self.create_features_list()
    if self.removed_features:
        features = [f for f in features if f not in self.removed_features]
    if self.extra_features:
        existing = set(features)
        features.extend(f for f in self.extra_features if f not in existing)
    return features

link_device ¶

link_device(device_data)

Link this entity to a HA device registry entry.

Parameters:

Name	Type	Description	Default
`device_data`	`DeviceData`	Device registry data dict.	required

Raises:

Type	Description
`ValueError`	If device_id does not match.

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

def link_device(self, device_data: DeviceData) -> None:
    """Link this entity to a HA device registry entry.

    Args:
        device_data: Device registry data dict.

    Raises:
        ValueError: If device_id does not match.
    """
    if self.device_id != device_data.get("id"):
        raise ValueError(f"Device ID mismatch: {self.device_id} != {device_data.get('id')}")
    self.linked_device = device_data

to_sber_state ¶

to_sber_state()

Build Sber device config JSON for MQTT publish.

Returns:

Type	Description
`dict`	Dict with device descriptor for Sber (id, name, room, model, features).
`dict`	Optionally includes nicknames, groups, parent_id, and partner_meta
`dict`	when configured.

Raises:

Type	Description
`RuntimeError`	If fill_by_ha_state was not called first.
`RuntimeError`	If device has device_id but linked_device is not set.

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

def to_sber_state(self) -> dict:
    """Build Sber device config JSON for MQTT publish.

    Returns:
        Dict with device descriptor for Sber (id, name, room, model, features).
        Optionally includes nicknames, groups, parent_id, and partner_meta
        when configured.

    Raises:
        RuntimeError: If fill_by_ha_state was not called first.
        RuntimeError: If device has device_id but linked_device is not set.
    """
    if not self.is_filled_by_state:
        raise RuntimeError(f"Entity {self.entity_id}: fill_by_ha_state must be called before to_sber_state")

    if self.device_id is None:
        res: dict = {
            "id": self.entity_id,
            "name": self.name,
            "default_name": self.entity_id,
            "room": self.area_id,
            "model": {
                "id": f"Mdl_{self.category}",
                "manufacturer": "Unknown",
                "model": "Unknown",
                "description": self.name,
                "category": self.category,
                "features": self.get_final_features_list(),
            },
            "hw_version": "1",
            "sw_version": "1",
        }
    else:
        if self.linked_device is None:
            raise RuntimeError(f"Entity {self.entity_id}: linked_device required when device_id is set")
        # Use overridden name (e.g. sber_name from YAML) if set, otherwise device name
        device_name = self.linked_device.get("name", self.original_name)
        display_name = self.name if self.name != self.original_name else device_name
        # Append category suffix to model_id to prevent Sber cloud from
        # overriding our category based on its own model database.
        raw_model_id = self.linked_device["model_id"]
        model_id = f"{raw_model_id}_{self.category}" if raw_model_id else f"Mdl_{self.category}"
        res = {
            "id": self.entity_id,
            "name": display_name,
            "default_name": self.original_name or self.entity_id,
            "room": self.linked_device.get("area_id", self.area_id),
            "model": {
                "id": model_id,
                "manufacturer": self.linked_device["manufacturer"],
                "model": self.linked_device["model"],
                "description": display_name,
                "category": self.category,
                "features": self.get_final_features_list(),
            },
            "hw_version": self.linked_device["hw_version"],
            "sw_version": self.linked_device["sw_version"],
        }

    # Inject allowed_values and dependencies from subclass hooks
    allowed = self.create_allowed_values_list()
    if allowed:
        res["model"]["allowed_values"] = allowed
    deps = self.create_dependencies()
    if deps:
        res["model"]["dependencies"] = deps

    if self.nicknames:
        res["nicknames"] = self.nicknames
    if self.groups:
        res["groups"] = self.groups
    if self.parent_entity_id:
        res["parent_id"] = self.parent_entity_id
    if self.partner_meta:
        res["partner_meta"] = self.partner_meta

    return res

to_sber_current_state `abstractmethod` ¶

to_sber_current_state()

Build Sber current state JSON for MQTT publish.

Returns:

Type	Description
`dict`	Dict with entity_id key mapping to {'states': [...]}.

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

@abstractmethod
def to_sber_current_state(self) -> dict:
    """Build Sber current state JSON for MQTT publish.

    Returns:
        Dict with entity_id key mapping to {'states': [...]}.
    """

get_entity_domain ¶

get_entity_domain()

Extract HA domain from entity_id.

Returns:

Type	Description
`str`	Domain string (e.g., 'climate' from 'climate.living_room').

Raises:

Type	Description
`ValueError`	If entity_id has invalid format.

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

def get_entity_domain(self) -> str:
    """Extract HA domain from entity_id.

    Returns:
        Domain string (e.g., 'climate' from 'climate.living_room').

    Raises:
        ValueError: If entity_id has invalid format.
    """
    entity_id = self.entity_id
    if not isinstance(entity_id, str) or "." not in entity_id:
        raise ValueError(f"entity_id '{entity_id}' has invalid format")
    domain, _ = entity_id.split(".", 1)
    return domain

process_cmd `abstractmethod` ¶

process_cmd(cmd_data)

Process a command from Sber cloud.

Parameters:

Name	Type	Description	Default
`cmd_data`	`dict`	Command payload with 'states' list, or None.	required

Returns:

Type	Description
`list[dict]`	List of dicts with 'url' key containing HA service call descriptors,
`list[dict]`	or empty list if no action needed or cmd_data is None.

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

@abstractmethod
def process_cmd(self, cmd_data: dict) -> list[dict]:
    """Process a command from Sber cloud.

    Args:
        cmd_data: Command payload with 'states' list, or None.

    Returns:
        List of dicts with 'url' key containing HA service call descriptors,
        or empty list if no action needed or cmd_data is None.
    """
    return []

process_state_change ¶

process_state_change(old_state, new_state)

Handle a state change event from Home Assistant.

Default implementation refreshes internal state via fill_by_ha_state. Override in subclasses if additional processing is needed.

Parameters:

Name	Type	Description	Default
`old_state`	`dict \| None`	Previous HA state dict (may be None).	required
`new_state`	`dict`	New HA state dict.	required

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

def process_state_change(self, old_state: dict | None, new_state: dict) -> None:
    """Handle a state change event from Home Assistant.

    Default implementation refreshes internal state via fill_by_ha_state.
    Override in subclasses if additional processing is needed.

    Args:
        old_state: Previous HA state dict (may be None).
        new_state: New HA state dict.
    """
    self.fill_by_ha_state(new_state)

has_significant_change ¶

has_significant_change()

Check if current Sber state differs from last published state.

Used to avoid unnecessary MQTT publishes when only non-relevant HA attributes changed (e.g., last_updated, icon, etc.).

Returns:

Type	Description
`bool`	True if the state has changed and should be published.

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

def has_significant_change(self) -> bool:
    """Check if current Sber state differs from last published state.

    Used to avoid unnecessary MQTT publishes when only non-relevant
    HA attributes changed (e.g., last_updated, icon, etc.).

    Returns:
        True if the state has changed and should be published.
    """
    if self._previous_sber_state is None:
        return True
    try:
        current = self.to_sber_current_state()
    except (RuntimeError, TypeError, ValueError):
        return True
    return current != self._previous_sber_state

mark_state_published ¶

mark_state_published()

Snapshot current Sber state as the last published state.

Called after successful MQTT publish to enable value diffing.

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

def mark_state_published(self) -> None:
    """Snapshot current Sber state as the last published state.

    Called after successful MQTT publish to enable value diffing.
    """
    try:
        self._previous_sber_state = self.to_sber_current_state()
    except (RuntimeError, TypeError, ValueError):
        self._previous_sber_state = None

resolve_link_role ¶

resolve_link_role(domain, device_class)

Determine the link role for an HA entity based on domain and device_class.

Iterates ALL_LINKABLE_ROLES and returns the role name of the first match. Domain-aware disambiguation is built into the role definitions: e.g. sensor + battery → battery, binary_sensor + battery → battery_low.

Parameters:

Name	Type	Description	Default
`domain`	`str`	HA entity domain.	required
`device_class`	`str`	HA original_device_class.	required

Returns:

Type	Description
`str`	Role name string, or empty string if no match.

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

def resolve_link_role(domain: str, device_class: str) -> str:
    """Determine the link role for an HA entity based on domain and device_class.

    Iterates ``ALL_LINKABLE_ROLES`` and returns the role name of the first match.
    Domain-aware disambiguation is built into the role definitions:
    e.g. ``sensor`` + ``battery`` → ``battery``, ``binary_sensor`` + ``battery``
    → ``battery_low``.

    Args:
        domain: HA entity domain.
        device_class: HA original_device_class.

    Returns:
        Role name string, or empty string if no match.
    """
    for lr in ALL_LINKABLE_ROLES:
        if lr.matches(domain, device_class):
            return lr.role
    return ""

OnOffEntity¶

Базовый класс для устройств с on/off состоянием (реле, розетки, клапаны).

Base class for Sber on/off entities (relay, socket).

Provides shared implementations of fill_by_ha_state, create_features_list, and to_sber_current_state for devices that expose a simple on/off state via the Sber on_off feature.

Supports optional power, voltage, and current features when the HA entity reports those values via attributes.

OnOffEntity ¶

OnOffEntity(category, entity_data)

Bases: BaseEntity

Base class for on/off entities that expose the Sber 'on_off' feature.

Subclasses must implement process_cmd to map Sber on/off commands to the appropriate HA service calls (e.g., turn_on/turn_off for relays).

Subclasses may override _ha_on_state if the HA 'on' state string differs from the default "on".

Optionally reports power, voltage, and current when the HA entity has those attributes.

Initialize on/off entity.

Parameters:

Name	Type	Description	Default
`category`	`str`	Sber device category string.	required
`entity_data`	`dict`	HA entity registry dict containing entity metadata.	required

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

def __init__(self, category: str, entity_data: dict) -> None:
    """Initialize on/off entity.

    Args:
        category: Sber device category string.
        entity_data: HA entity registry dict containing entity metadata.
    """
    super().__init__(category, entity_data)
    self.current_state = False
    self._power: int | None = None
    self._voltage: int | None = None
    self._current: int | None = None
    self._child_lock: bool | None = None

current_state `instance-attribute` ¶

current_state = False

Current on/off state of the entity.

fill_by_ha_state ¶

fill_by_ha_state(ha_state)

Parse HA state and update on/off status, energy, and child_lock 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/on_off_entity.py

def fill_by_ha_state(self, ha_state: dict) -> None:
    """Parse HA state and update on/off status, energy, and child_lock 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") == self._ha_on_state
    attrs = ha_state.get("attributes", {})
    self._power = self._parse_int_attr(attrs, "power")
    self._voltage = self._parse_int_attr(attrs, "voltage")
    self._current = self._parse_int_attr(attrs, "current")
    child_lock = attrs.get("child_lock")
    if child_lock is not None:
        self._child_lock = bool(child_lock)
    else:
        self._child_lock = None

create_features_list ¶

create_features_list()

Return Sber feature list including 'on_off' and optional features.

Returns:

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

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

def create_features_list(self) -> list[str]:
    """Return Sber feature list including 'on_off' and optional features.

    Returns:
        List of Sber feature strings supported by this entity.
    """
    features = [*super().create_features_list(), "on_off"]
    if self._power is not None:
        features.append("power")
    if self._voltage is not None:
        features.append("voltage")
    if self._current is not None:
        features.append("current")
    if self._child_lock is not None:
        features.append("child_lock")
    return features

to_sber_current_state ¶

to_sber_current_state()

Build Sber current state payload with online, on_off, energy, and child_lock.

Returns:

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

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

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

    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._power is not None:
        states.append(make_state(SberFeature.POWER, make_integer_value(self._power)))
    if self._voltage is not None:
        states.append(make_state(SberFeature.VOLTAGE, make_integer_value(self._voltage)))
    if self._current is not None:
        states.append(make_state(SberFeature.CURRENT, make_integer_value(self._current)))
    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}}

SimpleSensor¶

Базовый класс для read-only сенсоров.

Base class for read-only Sber sensors with a single value feature.

Provides shared implementations of process_cmd, create_features_list, and to_sber_current_state so that concrete sensor subclasses only need to define how their value is extracted and formatted for the Sber protocol.

Supports optional battery_percentage feature when the HA entity reports battery level via attributes.

SimpleReadOnlySensor ¶

SimpleReadOnlySensor(category, entity_data)

Bases: BaseEntity

Base class for read-only sensors that expose a single Sber feature.

Subclasses must define the class-level attributes _sber_value_key and _sber_value_type, and implement _get_sber_value to return the current sensor value in the appropriate Sber format.

Optionally reports battery_percentage when the HA entity has a battery or battery_level attribute.

Initialize simple read-only sensor.

Parameters:

Name	Type	Description	Default
`category`	`str`	Sber device category string.	required
`entity_data`	`dict`	HA entity registry dict containing entity metadata.	required

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

def __init__(self, category: str, entity_data: dict) -> None:
    """Initialize simple read-only sensor.

    Args:
        category: Sber device category string.
        entity_data: HA entity registry dict containing entity metadata.
    """
    super().__init__(category, entity_data)
    self._battery_level: int | None = None
    self._battery_low_linked: bool | None = None
    self._signal_strength_raw: int | None = None
    self._sensor_sensitive: str | None = None
    self._linked_entities: dict[str, str] = {}
    """Linked entity IDs by role: {role: entity_id}."""

update_linked_data ¶

update_linked_data(role, ha_state)

Inject data from a linked entity into this sensor.

Parameters:

Name	Type	Description	Default
`role`	`str`	Link role name (battery, battery_low, signal_strength, humidity, temperature).	required
`ha_state`	`dict`	HA state dict with 'state' and 'attributes'.	required

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

def update_linked_data(self, role: str, ha_state: dict) -> None:
    """Inject data from a linked entity into this sensor.

    Args:
        role: Link role name (battery, battery_low, signal_strength, humidity, temperature).
        ha_state: HA state dict with 'state' and 'attributes'.
    """
    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_linked = state_val == "on"
    elif role == "signal_strength":
        with contextlib.suppress(TypeError, ValueError):
            self._signal_strength_raw = int(float(state_val))

fill_by_ha_state ¶

fill_by_ha_state(ha_state)

Parse HA state and update internal state including battery and signal.

Reads battery level from battery or battery_level attribute. Reads signal strength from signal_strength, rssi, or linkquality attribute.

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

def fill_by_ha_state(self, ha_state: dict) -> None:
    """Parse HA state and update internal state including battery and signal.

    Reads battery level from ``battery`` or ``battery_level`` attribute.
    Reads signal strength from ``signal_strength``, ``rssi``, or
    ``linkquality`` attribute.

    Args:
        ha_state: HA state dict with 'state' and 'attributes' keys.
    """
    super().fill_by_ha_state(ha_state)
    attrs = ha_state.get("attributes", {})
    # Only overwrite battery/signal from primary entity attributes when
    # a value IS found. When absent, preserve data injected by linked sensors
    # via update_linked_data() to avoid flip-flopping on every state update.
    battery = attrs.get("battery") or attrs.get("battery_level")
    if battery is not None:
        try:
            self._battery_level = int(battery)
        except (TypeError, ValueError):
            pass  # preserve linked value

    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):
            pass  # preserve linked value

    # Sensor sensitivity (Aqara, Tuya, some Zigbee devices)
    sensitivity = attrs.get("sensitivity") or attrs.get("motion_sensitivity")
    if sensitivity is not None:
        s = str(sensitivity).lower()
        if s in ("auto", "high", "low", "medium"):
            # Sber only accepts auto/high, map medium→auto, low→low
            self._sensor_sensitive = {"medium": "auto", "low": "low"}.get(s, s)
        else:
            self._sensor_sensitive = None
    else:
        self._sensor_sensitive = None

create_features_list ¶

create_features_list()

Return Sber feature list including the sensor's value key.

Adds battery_percentage and battery_low_power if battery level is available. Adds signal_strength if signal data is present.

Returns:

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

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

def create_features_list(self) -> list[str]:
    """Return Sber feature list including the sensor's value key.

    Adds ``battery_percentage`` and ``battery_low_power`` if battery
    level is available. Adds ``signal_strength`` if signal data is present.

    Returns:
        List of Sber feature strings supported by this entity.
    """
    features = [*super().create_features_list(), self._sber_value_key]
    if self._battery_level is not None or self._battery_low_linked 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._sensor_sensitive is not None:
        features.append("sensor_sensitive")
    return features

to_sber_current_state ¶

to_sber_current_state()

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

Returns:

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

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

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

    Returns:
        Dict mapping entity_id to its Sber state representation.
    """
    states = [
        make_state(SberFeature.ONLINE, make_bool_value(self._is_online)),
        {"key": self._sber_value_key, "value": self._build_sber_value_dict()},
    ]
    if self._battery_level is not None:
        states.append(make_state(SberFeature.BATTERY_PERCENTAGE, make_integer_value(self._battery_level)))
        # Use linked binary_sensor if available, otherwise derive from percentage
        battery_low = self._battery_low_linked if self._battery_low_linked 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_linked is not None:
        # Only linked battery_low binary_sensor, no percentage sensor
        states.append(make_state(SberFeature.BATTERY_LOW_POWER, make_bool_value(self._battery_low_linked)))
    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._sensor_sensitive is not None:
        states.append(
            make_state(SberFeature.SENSOR_SENSITIVE, make_enum_value(self._sensor_sensitive))
        )
    return {self.entity_id: {"states": states}}

process_cmd ¶

process_cmd(cmd_data)

Process Sber command (no-op for read-only sensor).

Parameters:

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

Returns:

Type	Description
`list[dict]`	Empty list -- sensors do not accept commands.

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

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

    Args:
        cmd_data: Sber command dict (ignored).

    Returns:
        Empty list -- sensors do not accept commands.
    """
    return []

Базовые классы¶

BaseEntity¶

ROLE_BATTERY module-attribute ¶

ROLE_BATTERY_LOW module-attribute ¶

ROLE_SIGNAL module-attribute ¶

ROLE_TEMPERATURE module-attribute ¶

ROLE_HUMIDITY module-attribute ¶

SENSOR_LINK_ROLES module-attribute ¶

ALL_LINKABLE_ROLES module-attribute ¶

LinkableRole dataclass ¶

matches ¶

BaseEntity ¶

LINKABLE_ROLES class-attribute ¶

effective_room property ¶

is_online property ¶

fill_by_ha_state ¶

is_group_state ¶

create_features_list ¶

create_allowed_values_list ¶

create_dependencies ¶

get_final_features_list ¶

link_device ¶

to_sber_state ¶

to_sber_current_state abstractmethod ¶

get_entity_domain ¶

process_cmd abstractmethod ¶

process_state_change ¶

has_significant_change ¶

mark_state_published ¶

resolve_link_role ¶

OnOffEntity¶

OnOffEntity ¶

current_state instance-attribute ¶

fill_by_ha_state ¶

create_features_list ¶

to_sber_current_state ¶

SimpleSensor¶

SimpleReadOnlySensor ¶

update_linked_data ¶

fill_by_ha_state ¶

create_features_list ¶

to_sber_current_state ¶

process_cmd ¶

ROLE_BATTERY `module-attribute` ¶

ROLE_BATTERY_LOW `module-attribute` ¶

ROLE_SIGNAL `module-attribute` ¶

ROLE_TEMPERATURE `module-attribute` ¶

ROLE_HUMIDITY `module-attribute` ¶

SENSOR_LINK_ROLES `module-attribute` ¶

ALL_LINKABLE_ROLES `module-attribute` ¶

LinkableRole `dataclass` ¶

LINKABLE_ROLES `class-attribute` ¶

effective_room `property` ¶

is_online `property` ¶

to_sber_current_state `abstractmethod` ¶

process_cmd `abstractmethod` ¶

current_state `instance-attribute` ¶