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

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.

CommandResult module-attribute

CommandResult = ServiceCallResult | UpdateStateResult

Union type for all possible process_cmd return items.

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.

ServiceCallUrl

Bases: TypedDict

Descriptor for a single HA service call.

ServiceCallResult

Bases: TypedDict

A process_cmd result instructing the bridge to call a HA service.

UpdateStateResult

Bases: TypedDict

A process_cmd result instructing the bridge to re-publish current state.

AttrSpec dataclass

AttrSpec(field, attr_keys=(), parser=lambda v: v, default=None, preserve_on_missing=False, converter=None)

Declarative spec for parsing a single HA attribute into an instance field.

Subclasses of :class:BaseEntity can declare a class-level ATTR_SPECS tuple and rely on :meth:BaseEntity._apply_attr_specs to do the parsing in one line instead of hand-rolling attrs.get(...) / try-except / int() boilerplate for every attribute.

Attributes:

Name	Type	Description
field	str	Instance attribute name to assign (e.g. "_battery_level").
attr_keys	tuple[str, ...]	HA attribute key(s) to read in fallback order. First non-None match wins. Pass a single string for one key.
parser	Callable[[object], object]	Conversion function applied to the raw value. Defaults to identity. Should raise (TypeError, ValueError) for bad input.
default	object	Value to assign when no key matched or parsing failed.
preserve_on_missing	bool	When True and no attr key matched, leave the existing field value untouched instead of assigning default. Used by sensors that receive values from linked companion entities via update_linked_data — we don't want to clobber those when the primary HA state is refreshed.

converter class-attribute instance-attribute

converter = None

Full-attrs converter. When set, receives the entire HA attributes dict instead of a single value looked up by attr_keys. parser and attr_keys are ignored when converter is provided.

DeviceData

Bases: TypedDict

Typed device registry data linked to an entity.

All keys are optional because linked device data may come from partial HA device registry entries. Missing values fall back to sensible defaults in BaseEntity.to_sber_state.

serial_number instance-attribute

serial_number

Real device serial number from HA device registry (empty string if unknown).

mac instance-attribute

mac

Normalised MAC address pulled from DeviceEntry.connections (empty if unknown).

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.

ATTR_SPECS class-attribute

ATTR_SPECS = ()

Declarative HA-attribute parsing specs.

Subclasses can populate this tuple to drive :meth:_apply_attr_specs instead of hand-rolling per-attribute parsing inside fill_by_ha_state.

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.

register_link

register_link(role, linked_entity_id)

Register a linked companion entity for the given role.

Public API for :class:SberEntityLoader — replaces direct mutation of self._linked_entities to preserve encapsulation.

Parameters:

Name	Type	Description	Default
role	str	The role name (e.g. "battery", "signal_strength").	required
linked_entity_id	str	HA entity_id of the linked companion.	required

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

def register_link(self, role: str, linked_entity_id: str) -> None:
    """Register a linked companion entity for the given role.

    Public API for :class:`SberEntityLoader` — replaces direct mutation
    of ``self._linked_entities`` to preserve encapsulation.

    Args:
        role: The role name (e.g. ``"battery"``, ``"signal_strength"``).
        linked_entity_id: HA entity_id of the linked companion.
    """
    self._linked_entities[role] = linked_entity_id

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_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

update_linked_data

update_linked_data(role, ha_state)

Inject state from a linked companion HA entity (default: no-op).

Device classes that accept linked entities (e.g. a binary battery sensor paired with a valve) override this to apply the foreign state to their own fields. The default implementation does nothing, which is correct for classes that don't advertise :attr:LINKABLE_ROLES.

Providing a universal default also eliminates hasattr checks at every call site -- callers may invoke it unconditionally.

Parameters:

Name	Type	Description	Default
role	str	The link role (e.g. "battery", "humidity").	required
ha_state	dict	HA state dict of the linked entity.	required

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

def update_linked_data(self, role: str, ha_state: dict) -> None:  # noqa: B027 — intentional concrete no-op, not abstract
    """Inject state from a linked companion HA entity (default: no-op).

    Device classes that accept linked entities (e.g. a binary battery
    sensor paired with a valve) override this to apply the foreign
    state to their own fields.  The default implementation does
    nothing, which is correct for classes that don't advertise
    :attr:`LINKABLE_ROLES`.

    Providing a universal default also eliminates ``hasattr`` checks
    at every call site -- callers may invoke it unconditionally.

    Args:
        role: The link role (e.g. ``"battery"``, ``"humidity"``).
        ha_state: HA state dict of the linked entity.
    """

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.

Handles both device_id is None (standalone HA entity) and device_id is set (entity linked to a device registry entry) cases through a unified source-resolver approach.

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.

    Handles both ``device_id is None`` (standalone HA entity) and
    ``device_id is set`` (entity linked to a device registry entry)
    cases through a unified source-resolver approach.

    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 not None and self.linked_device is None:
        raise RuntimeError(f"Entity {self.entity_id}: linked_device required when device_id is set")

    device: DeviceData = self.linked_device or {}
    display_name = self._resolve_display_name(device)

    res: dict = {
        "id": self.entity_id,
        "name": display_name,
        "default_name": self._resolve_default_name(),
        "room": device.get("area_id") or self.area_id,
        "model": self._build_model_descriptor(device, display_name),
        "hw_version": device.get("hw_version") or "1",
        "sw_version": device.get("sw_version") or "1",
    }

    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

process_cmd(cmd_data)

Process a Sber command via the _cmd_handlers dispatch table.

Subclasses declare which Sber feature keys they handle by overriding :attr:_cmd_handlers. The base implementation walks cmd_data["states"], routes each entry to its handler, and returns the concatenated service-call list.

Parameters:

Name	Type	Description	Default
cmd_data	dict	Command payload with 'states' list. Always a dict — the dispatcher rejects None before reaching here.	required

Returns:

Type	Description
list[CommandResult]	List of :class:ServiceCallResult or :class:UpdateStateResult
list[CommandResult]	items, or empty list if no action needed.

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

def process_cmd(self, cmd_data: dict) -> list[CommandResult]:
    """Process a Sber command via the ``_cmd_handlers`` dispatch table.

    Subclasses declare which Sber feature keys they handle by
    overriding :attr:`_cmd_handlers`.  The base implementation walks
    ``cmd_data["states"]``, routes each entry to its handler, and
    returns the concatenated service-call list.

    Args:
        cmd_data: Command payload with 'states' list. Always a dict —
            the dispatcher rejects ``None`` before reaching here.

    Returns:
        List of :class:`ServiceCallResult` or :class:`UpdateStateResult`
        items, or empty list if no action needed.
    """
    handlers = self._cmd_handlers
    if not handlers:
        return []
    results: list[CommandResult] = []
    for item in cmd_data.get("states", []):
        handler = handlers.get(item.get("key", ""))
        if handler is None:
            continue
        results.extend(handler(item.get("value", {})))
    return results

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). Reserved for subclass overrides that need to compare old and new state.	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). Reserved for
            subclass overrides that need to compare old and new state.
        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.

Uses :class:BaseEntity.ATTR_SPECS for the declarative attribute parsing of power / voltage / current / child_lock, falling back to the current_state check which depends on the subclass-overridable _ha_on_state.

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.

    Uses :class:`BaseEntity.ATTR_SPECS` for the declarative
    attribute parsing of power / voltage / current / child_lock,
    falling back to the ``current_state`` check which depends on
    the subclass-overridable ``_ha_on_state``.

    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
    self._apply_attr_specs(ha_state.get("attributes", {}))

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: BatteryAndSignalLinkMixin, 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._sensor_sensitive: str | None = None
    self._linked_entities: dict[str, str] = {}
    """Linked entity IDs by role: {role: entity_id}."""

fill_by_ha_state

fill_by_ha_state(ha_state)

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

Battery (battery / battery_level) and signal strength (signal_strength / rssi / linkquality) are parsed via :class:AttrSpec with preserve_on_missing=True so that values injected by linked companion sensors via update_linked_data are not clobbered on every primary state refresh.

Sensor sensitivity (Aqara / Tuya) uses custom mapping logic that is not expressible through AttrSpec — kept imperative.

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.

    Battery (``battery`` / ``battery_level``) and signal strength
    (``signal_strength`` / ``rssi`` / ``linkquality``) are parsed via
    :class:`AttrSpec` with ``preserve_on_missing=True`` so that values
    injected by linked companion sensors via ``update_linked_data``
    are not clobbered on every primary state refresh.

    Sensor sensitivity (Aqara / Tuya) uses custom mapping logic that
    is not expressible through ``AttrSpec`` — kept imperative.

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

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()},
    ]
    self._append_battery_signal_states(states)
    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}}