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

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.

ALWAYS_PUBLISHED_FEATURES `module-attribute` ¶

ALWAYS_PUBLISHED_FEATURES = frozenset({'online'})

Feature keys that bypass the declared-features publish filter.

online is obligatory for every Sber category (see _generated/obligatory_features.py — it appears in every entry), and Sber drops a device whose publish lacks it. Filtering it out because a buggy _create_features_list forgot to declare it, or because a user removed it via sber_features_remove, would be strictly worse than publishing an undeclared key, so it is always let through.

NO_SNAPSHOT `module-attribute` ¶

NO_SNAPSHOT = NoSnapshot()

Singleton sentinel for :meth:BaseEntity.mark_state_published.

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

ROLE_CO2 `module-attribute` ¶

ROLE_CO2 = LinkableRole('co2', frozenset({'sensor'}), frozenset({'carbon_dioxide'}))

Carbon-dioxide concentration link (Sber sensor_air.co2, ppm).

ROLE_PM1 `module-attribute` ¶

ROLE_PM1 = LinkableRole('pm1', frozenset({'sensor'}), frozenset({'pm1'}))

PM1.0 particulate matter link (Sber sensor_air.pm1_0, µg/m³).

ROLE_PM25 `module-attribute` ¶

ROLE_PM25 = LinkableRole('pm25', frozenset({'sensor'}), frozenset({'pm25'}))

PM2.5 particulate matter link (Sber sensor_air.pm2_5, µg/m³).

ROLE_PM10 `module-attribute` ¶

ROLE_PM10 = LinkableRole('pm10', frozenset({'sensor'}), frozenset({'pm10'}))

PM10 particulate matter link (Sber sensor_air.pm10, µg/m³).

ROLE_TVOC `module-attribute` ¶

ROLE_TVOC = LinkableRole('tvoc', frozenset({'sensor'}), frozenset({'volatile_organic_compounds'}))

TVOC concentration link (Sber sensor_air.tvoc_float, mg/m³).

ROLE_HCHO `module-attribute` ¶

ROLE_HCHO = LinkableRole('hcho', frozenset({'sensor'}), frozenset({'volatile_organic_compounds_parts'}))

Formaldehyde link (Sber sensor_air.hcho_float, mg/m³). HA has no dedicated device_class for formaldehyde; the closest match is volatile_organic_compounds_parts. Users with a distinct HCHO sensor will link it manually via the wizard.

ROLE_OPEN_STATE `module-attribute` ¶

ROLE_OPEN_STATE = LinkableRole('open_state', frozenset({'binary_sensor'}), frozenset({'garage_door', 'door', 'opening'}))

Reed/contact sensor reporting the real position of an impulse-driven gate.

The primary entity of such a gate is the impulse relay (switch / button / script), whose own HA state is only an echo of the last written relay value and says nothing about the leaf. The position must therefore come from a companion contact sensor linked in this role.

window is deliberately not accepted: window contacts are common and would otherwise become gate candidates in the wizard.

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

GATE_LINK_ROLES `module-attribute` ¶

GATE_LINK_ROLES = (ROLE_OPEN_STATE, ROLE_SIGNAL)

Linkable roles accepted by an impulse gate (the Sber gate spec has no battery).

ALL_LINKABLE_ROLES `module-attribute` ¶

ALL_LINKABLE_ROLES = _collect_declared_roles()

Global registry of all known linkable roles.

Derived automatically from every LinkableRole constant declared above in this module. Device classes are expected to compose their LINKABLE_ROLES from these constants; as long as they do, the wizard path (resolve_link_role) and per-class matching (LINKABLE_ROLES) stay in sync. Nothing in this module enforces that composition — a class that builds its own LinkableRole inline still drifts, which is why test_link_roles_registry walks every class in CATEGORY_DOMAIN_MAP and fails on roles unknown to this registry.

NoSnapshot ¶

Sentinel type marking an omitted snapshot argument.

None is a meaningful snapshot value ("serialization failed, treat as changed"), so :meth:BaseEntity.mark_state_published cannot use it as its default. Public because it appears in that method's public signature — callers reading the annotation must be able to name it.

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 - _build_current_state: Build Sber current state JSON (the public to_sber_current_state wraps it with the declared-features filter) - 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] = {}
    self._undeclared_keys_logged: set[str] = set()

    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.

REQUIRED_LINK_ROLES `class-attribute` ¶

REQUIRED_LINK_ROLES = ()

Role names this device class cannot work without.

Empty for every class whose Sber features are derivable from the primary HA entity alone. A non-empty tuple means the device is composite: without those links it would publish a fabricated state (see :class:~devices.gate.ImpulseGateEntity, whose position exists only in a linked contact sensor). The wizard refuses to add such a device when a required role is unmapped (websocket_api.devices_grouped.ws_add_ha_device → missing_required_role).

ENTITY_OPTION_KEYS `class-attribute` ¶

ENTITY_OPTION_KEYS = ()

Per-entity user option keys this device class understands.

Empty for every class whose behaviour is fully derived from the HA entity itself. A non-empty tuple opts the class into the generic per-entity options mechanism: the values are persisted in entry.options[CONF_ENTITY_OPTIONS] keyed by entity id, applied at load time by :class:~entity_registry.SberEntityLoader, edited live through the update_entity_options WebSocket command, reported to the panel inside device_detail and carried by export / import.

The mechanism is deliberately class-driven: nothing outside the device class knows what an option means, so adding one to a new category needs no change in the loader, the bridge, the WS layer or the config round-trip.

ENTITY_OPTIONS_BLOCK `class-attribute` ¶

ENTITY_OPTIONS_BLOCK = 'entity_options'

Key under which device_detail reports :meth:entity_options_state.

Per class rather than global because the panel renders a different form per option set, and because :class:~devices.gate.ImpulseGateEntity shipped its block as gate_options in v1.42 — renaming it would break every panel that is still cached in a browser.

supports_entity_options `property` ¶

supports_entity_options

Whether this device class accepts per-entity user options.

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.

apply_entity_options ¶

apply_entity_options(options)

Apply persisted per-entity options (default: nothing to apply).

Implementations are deliberately lenient: unknown keys and invalid values are ignored rather than raising, because this runs on the entity-loading path where a hand-edited (or downgraded) config must never take the whole integration down. Strict checking of user input belongs to :meth:validate_entity_options.

Parameters:

Name	Type	Description	Default
`options`	`dict`	Mapping of :attr:`ENTITY_OPTION_KEYS` to values; only the keys present are applied.	required

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

def apply_entity_options(self, options: dict) -> None:  # noqa: B027 — intentional concrete no-op, not abstract
    """Apply persisted per-entity options (default: nothing to apply).

    Implementations are deliberately *lenient*: unknown keys and
    invalid values are ignored rather than raising, because this runs
    on the entity-loading path where a hand-edited (or downgraded)
    config must never take the whole integration down.  Strict
    checking of user input belongs to :meth:`validate_entity_options`.

    Args:
        options: Mapping of :attr:`ENTITY_OPTION_KEYS` to values; only
            the keys present are applied.
    """

validate_entity_options ¶

validate_entity_options(options)

Validate user-submitted option values (raises on bad input).

Called by the update_entity_options WebSocket command before anything is persisted, so the user gets a readable message instead of a silently ignored setting. Subclasses override to add value-level checks and are expected to call super() first.

Parameters:

Name	Type	Description	Default
`options`	`dict`	Mapping submitted by the panel.	required

Raises:

Type	Description
`ValueError`	With a human-readable message when the mapping contains a key this class does not accept.

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

def validate_entity_options(self, options: dict) -> None:
    """Validate user-submitted option values (raises on bad input).

    Called by the ``update_entity_options`` WebSocket command *before*
    anything is persisted, so the user gets a readable message instead
    of a silently ignored setting.  Subclasses override to add
    value-level checks and are expected to call ``super()`` first.

    Args:
        options: Mapping submitted by the panel.

    Raises:
        ValueError: With a human-readable message when the mapping
            contains a key this class does not accept.
    """
    unknown = sorted(set(options) - set(self.ENTITY_OPTION_KEYS))
    if unknown:
        raise ValueError(
            f"{self.entity_id}: unknown option(s) {', '.join(unknown)}; "
            f"accepted: {', '.join(self.ENTITY_OPTION_KEYS) or '(none)'}"
        )

entity_options_state ¶

entity_options_state()

Return the option block the panel renders for this entity.

Returns:

Type	Description
`dict[str, object]`	Mapping of the current option values (plus any read-only
`dict[str, object]`	context the form needs), or an empty dict for a class without
`dict[str, object]`	options — `device_detail` then omits the block entirely.

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

def entity_options_state(self) -> dict[str, object]:
    """Return the option block the panel renders for this entity.

    Returns:
        Mapping of the current option values (plus any read-only
        context the form needs), or an empty dict for a class without
        options — ``device_detail`` then omits the block entirely.
    """
    return {}

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 ¶

to_sber_current_state()

Build the publish-ready Sber current state JSON.

Calls the subclass hook :meth:_build_current_state and then drops every state whose key is not advertised in :meth:get_final_features_list — see :meth:_filter_undeclared_states.

Returns:

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

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

def to_sber_current_state(self) -> dict:
    """Build the publish-ready Sber current state JSON.

    Calls the subclass hook :meth:`_build_current_state` and then
    drops every state whose key is not advertised in
    :meth:`get_final_features_list` — see
    :meth:`_filter_undeclared_states`.

    Returns:
        Dict with entity_id key mapping to {'states': [...]}.
    """
    return self._filter_undeclared_states(self._build_current_state())

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
        # Sber omits proto3-default fields: {"type": "INTEGER"} means 0.
        # Normalize here so every handler sees a complete value dict.
        results.extend(handler(normalize_sber_value(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=NO_SNAPSHOT)

Record the state that was just published, for value diffing.

Called after a successful MQTT publish so :meth:has_significant_change can suppress redundant publishes.

Parameters:

Name	Type	Description	Default
`snapshot`	`dict \| NoSnapshot \| None`	The exact wire state that went out, captured before the publish awaited. Passing it is the correct form for the publish path: re-serializing here would capture changes that raced in during the network round-trip and silently swallow them (lost update). `None` is a valid snapshot value and means "treat the entity as changed on the next diff". When the argument is omitted entirely the current state is serialized now, which is only safe when no publish await intervened.	`NO_SNAPSHOT`

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

def mark_state_published(self, *, snapshot: dict | NoSnapshot | None = NO_SNAPSHOT) -> None:
    """Record the state that was just published, for value diffing.

    Called after a successful MQTT publish so
    :meth:`has_significant_change` can suppress redundant publishes.

    Args:
        snapshot: The exact wire state that went out, captured
            *before* the publish awaited.  Passing it is the correct
            form for the publish path: re-serializing here would
            capture changes that raced in during the network
            round-trip and silently swallow them (lost update).
            ``None`` is a valid snapshot value and means "treat the
            entity as changed on the next diff".  When the argument
            is omitted entirely the current state is serialized now,
            which is only safe when no publish await intervened.
    """
    if not isinstance(snapshot, NoSnapshot):
        self._previous_sber_state = snapshot
        return
    try:
        self._previous_sber_state = self.to_sber_current_state()
    except (RuntimeError, TypeError, ValueError):
        self._previous_sber_state = None

resolve_link_role_for ¶

resolve_link_role_for(accepted_roles, domain, device_class)

Resolve the link role of an HA entity against a specific role set.

Shared helper for link-role matching, used by the global :func:resolve_link_role and available to per-class validation (matching an entity against a primary's LINKABLE_ROLES) instead of re-implementing the loop. Not yet a hard single source of truth: websocket_api/links.py::ws_auto_link_all still runs its own LinkableRole.matches loop. That is safe only because ALL_LINKABLE_ROLES contains no two roles matching the same (domain, device_class) pair — an invariant locked by test_link_roles_registry.test_registry_has_no_ambiguous_matches.

Parameters:

Name	Type	Description	Default
`accepted_roles`	`Iterable[LinkableRole]`	Roles to match against (e.g. a device class's `LINKABLE_ROLES` or :data:`ALL_LINKABLE_ROLES`).	required
`domain`	`str`	HA entity domain.	required
`device_class`	`str`	HA original_device_class.	required

Returns:

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

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

def resolve_link_role_for(accepted_roles: Iterable[LinkableRole], domain: str, device_class: str) -> str:
    """Resolve the link role of an HA entity against a specific role set.

    Shared helper for link-role matching, used by the global
    :func:`resolve_link_role` and available to per-class validation
    (matching an entity against a primary's ``LINKABLE_ROLES``) instead
    of re-implementing the loop.  Not yet a hard single source of truth:
    ``websocket_api/links.py::ws_auto_link_all`` still runs its own
    ``LinkableRole.matches`` loop.  That is safe only because
    ``ALL_LINKABLE_ROLES`` contains no two roles matching the same
    ``(domain, device_class)`` pair — an invariant locked by
    ``test_link_roles_registry.test_registry_has_no_ambiguous_matches``.

    Args:
        accepted_roles: Roles to match against (e.g. a device class's
            ``LINKABLE_ROLES`` or :data:`ALL_LINKABLE_ROLES`).
        domain: HA entity domain.
        device_class: HA original_device_class.

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

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.
    """
    return resolve_link_role_for(ALL_LINKABLE_ROLES, domain, device_class)

OnOffEntity¶

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

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

Provides shared implementations of fill_by_ha_state, _create_features_list, and _build_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", {}))

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 _build_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)

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

BaseEntity¶

ALWAYS_PUBLISHED_FEATURES module-attribute ¶

NO_SNAPSHOT module-attribute ¶

CommandResult module-attribute ¶

ROLE_BATTERY module-attribute ¶

ROLE_BATTERY_LOW module-attribute ¶

ROLE_SIGNAL module-attribute ¶

ROLE_TEMPERATURE module-attribute ¶

ROLE_HUMIDITY module-attribute ¶

ROLE_CO2 module-attribute ¶

ROLE_PM1 module-attribute ¶

ROLE_PM25 module-attribute ¶

ROLE_PM10 module-attribute ¶

ROLE_TVOC module-attribute ¶

ROLE_HCHO module-attribute ¶

ROLE_OPEN_STATE module-attribute ¶

SENSOR_LINK_ROLES module-attribute ¶

GATE_LINK_ROLES module-attribute ¶

ALL_LINKABLE_ROLES module-attribute ¶

NoSnapshot ¶

ServiceCallUrl ¶

ServiceCallResult ¶

UpdateStateResult ¶

AttrSpec dataclass ¶

converter class-attribute instance-attribute ¶

DeviceData ¶

serial_number instance-attribute ¶

mac instance-attribute ¶

LinkableRole dataclass ¶

matches ¶

BaseEntity ¶

LINKABLE_ROLES class-attribute ¶

REQUIRED_LINK_ROLES class-attribute ¶

ENTITY_OPTION_KEYS class-attribute ¶

ENTITY_OPTIONS_BLOCK class-attribute ¶

supports_entity_options property ¶

ATTR_SPECS class-attribute ¶

effective_room property ¶

is_online property ¶

apply_entity_options ¶

validate_entity_options ¶

entity_options_state ¶

register_link ¶

fill_by_ha_state ¶

is_group_state ¶

create_allowed_values_list ¶

create_dependencies ¶

get_final_features_list ¶

update_linked_data ¶

link_device ¶

to_sber_state ¶

to_sber_current_state ¶

get_entity_domain ¶

process_cmd ¶

process_state_change ¶

has_significant_change ¶

mark_state_published ¶

resolve_link_role_for ¶

resolve_link_role ¶

OnOffEntity¶

OnOffEntity ¶

current_state instance-attribute ¶

fill_by_ha_state ¶

SimpleSensor¶

SimpleReadOnlySensor ¶

fill_by_ha_state ¶

ALWAYS_PUBLISHED_FEATURES `module-attribute` ¶

NO_SNAPSHOT `module-attribute` ¶

CommandResult `module-attribute` ¶

ROLE_BATTERY `module-attribute` ¶

ROLE_BATTERY_LOW `module-attribute` ¶

ROLE_SIGNAL `module-attribute` ¶

ROLE_TEMPERATURE `module-attribute` ¶

ROLE_HUMIDITY `module-attribute` ¶

ROLE_CO2 `module-attribute` ¶

ROLE_PM1 `module-attribute` ¶

ROLE_PM25 `module-attribute` ¶

ROLE_PM10 `module-attribute` ¶

ROLE_TVOC `module-attribute` ¶

ROLE_HCHO `module-attribute` ¶

ROLE_OPEN_STATE `module-attribute` ¶

SENSOR_LINK_ROLES `module-attribute` ¶

GATE_LINK_ROLES `module-attribute` ¶

ALL_LINKABLE_ROLES `module-attribute` ¶

AttrSpec `dataclass` ¶

converter `class-attribute` `instance-attribute` ¶

serial_number `instance-attribute` ¶

mac `instance-attribute` ¶

LinkableRole `dataclass` ¶

LINKABLE_ROLES `class-attribute` ¶

REQUIRED_LINK_ROLES `class-attribute` ¶

ENTITY_OPTION_KEYS `class-attribute` ¶

ENTITY_OPTIONS_BLOCK `class-attribute` ¶

supports_entity_options `property` ¶

ATTR_SPECS `class-attribute` ¶

effective_room `property` ¶

is_online `property` ¶

current_state `instance-attribute` ¶