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

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

CurtainEntity

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

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

CURTAIN_ENTITY_CATEGORY module-attribute

CURTAIN_ENTITY_CATEGORY = 'curtain'

Sber device category for curtain/cover entities.

CurtainEntity

CurtainEntity(entity_data, category=CURTAIN_ENTITY_CATEGORY)

Bases: BatteryAndSignalLinkMixin, BaseEntity

Sber curtain entity for cover control with position support.

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

Initialize curtain entity.

Parameters:

Name Type Description Default
entity_data dict

HA entity registry dict containing entity metadata.

required
category str

Sber device category (override in subclasses).

CURTAIN_ENTITY_CATEGORY
Source code in custom_components/sber_mqtt_bridge/devices/curtain.py
def __init__(self, entity_data: dict, category: str = CURTAIN_ENTITY_CATEGORY) -> None:
    """Initialize curtain entity.

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

min_position class-attribute instance-attribute

min_position = 0

Minimum allowed position (0-100%).

max_position class-attribute instance-attribute

max_position = 100

Maximum allowed position (0-100%).

battery_level class-attribute instance-attribute

battery_level = 0

Battery level percentage (0-100%).

current_position class-attribute instance-attribute

current_position = 0

Current cover position (0-100%).

fill_by_ha_state

fill_by_ha_state(ha_state)

Update state from Home Assistant data.

Battery level, tilt position and signal strength are parsed via :class:AttrSpec. current_position and open_rate have custom fallback / mapping logic and stay 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/curtain.py
def fill_by_ha_state(self, ha_state: dict) -> None:
    """Update state from Home Assistant data.

    Battery level, tilt position and signal strength are parsed via
    :class:`AttrSpec`.  ``current_position`` and ``open_rate`` have
    custom fallback / mapping logic and stay 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.current_position = self._parse_current_position(attrs)
    self._open_rate = self._parse_open_rate(attrs)

create_allowed_values_list

create_allowed_values_list()

Return allowed values for the controllable cover features.

Built from the final features list so user overrides (sber_features_add / sber_features_remove) stay in sync with the advertised limits. open_rate values follow the Sber curtain reference example.

Source code in custom_components/sber_mqtt_bridge/devices/curtain.py
def create_allowed_values_list(self) -> dict[str, dict]:
    """Return allowed values for the controllable cover features.

    Built from the final features list so user overrides
    (``sber_features_add`` / ``sber_features_remove``) stay in sync
    with the advertised limits.  ``open_rate`` values follow the
    Sber curtain reference example.
    """
    features = set(self.get_final_features_list())
    allowed: dict[str, dict] = {}
    if "open_set" in features:
        allowed["open_set"] = {
            "type": "ENUM",
            "enum_values": {"values": ["open", "close", "stop"]},
        }
    if "open_percentage" in features:
        allowed["open_percentage"] = {
            "type": "INTEGER",
            "integer_values": {"min": "0", "max": "100", "step": "1"},
        }
    if "open_rate" in features:
        allowed["open_rate"] = {
            "type": "ENUM",
            "enum_values": {"values": ["auto", "low", "high"]},
        }
    if "light_transmission_percentage" in features:
        allowed["light_transmission_percentage"] = {
            "type": "INTEGER",
            "integer_values": {"min": "0", "max": "100", "step": "1"},
        }
    return allowed

WindowBlindEntity

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

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

WINDOW_BLIND_CATEGORY module-attribute

WINDOW_BLIND_CATEGORY = 'window_blind'

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

WindowBlindEntity

WindowBlindEntity(entity_data)

Bases: CurtainEntity

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

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

Initialize window blind entity.

Parameters:

Name Type Description Default
entity_data dict

HA entity registry dict containing entity metadata.

required
Source code in custom_components/sber_mqtt_bridge/devices/window_blind.py
def __init__(self, entity_data: dict) -> None:
    """Initialize window blind entity.

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

GateEntity

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

Sber Gate entities -- HA cover gates and impulse-relay gates.

Two very different physical devices share the Sber gate category:

  • :class:GateEntity — a real HA cover with position support (open_cover / close_cover / stop_cover). Unchanged since v1.x, it stays the path for anything in the cover domain.
  • :class:ImpulseGateEntity — the "one button + one reed contact" topology (issue #53): an impulse relay (switch / button / script) that pulses the gate motor, plus a binary_sensor contact that is the only source of truth about the leaf position.

:func:make_gate_entity routes between them by HA domain and is what CATEGORY_DOMAIN_MAP registers for the gate category.

GATE_ENTITY_CATEGORY module-attribute

GATE_ENTITY_CATEGORY = 'gate'

Sber device category for gate/garage door entities.

IMPULSE_COOLDOWN_SECONDS module-attribute

IMPULSE_COOLDOWN_SECONDS = 2.0

Minimum gap (seconds) between two impulses sent to the same gate.

Anti-bounce only: it swallows the burst of back-to-back commands (a double-tap in the app, a repeated voice command, a retry after a lost ack), which is the classic way to make an impulse controller pile up pulses. A command arriving inside the window is acknowledged with a state republish instead of a second pulse.

It is deliberately not a travel-time lock: the leaf needs 15-25 s to move, and a command arriving after the window — but before the leaf has arrived — is a legitimate "stop / reverse" request. Guarding the whole travel is the job of the opt-in :data:GATE_OPTION_TRAVEL_TIME motion emulation, not of this window.

GATE_OPTION_INVERT_CONTACT module-attribute

GATE_OPTION_INVERT_CONTACT = 'invert_contact'

Entity-option key flipping the polarity of the linked reed contact.

GATE_OPTION_IMPULSE_SERVICE module-attribute

GATE_OPTION_IMPULSE_SERVICE = 'impulse_service'

Entity-option key choosing the HA service used to pulse the relay.

GATE_OPTION_TRAVEL_TIME module-attribute

GATE_OPTION_TRAVEL_TIME = 'travel_time'

Entity-option key holding the leaf travel time in seconds.

0 / None (the default) disables the motion emulation entirely and keeps the historical behaviour: the published open_state only ever switches between open and close, driven by the contact alone.

GATE_OPTION_AUTO_CLOSE_TIME module-attribute

GATE_OPTION_AUTO_CLOSE_TIME = 'auto_close_time'

Entity-option key holding the gate board's auto-close delay, in seconds.

0 / None (the default) means "the board does not auto-close" and keeps the behaviour byte-for-byte as it was before the option existed. See :meth:ImpulseGateEntity.auto_close_time for the semantics.

MAX_TRAVEL_TIME_SECONDS module-attribute

MAX_TRAVEL_TIME_SECONDS = 600.0

Upper bound accepted for :data:GATE_OPTION_TRAVEL_TIME (10 minutes).

A hand-edited config with a wild value (or a UI sending milliseconds) would otherwise pin the gate in a fake opening state for hours.

MAX_AUTO_CLOSE_TIME_SECONDS module-attribute

MAX_AUTO_CLOSE_TIME_SECONDS = 3600.0

Upper bound accepted for :data:GATE_OPTION_AUTO_CLOSE_TIME (1 hour).

Deliberately larger than :data:MAX_TRAVEL_TIME_SECONDS: a leaf that takes ten minutes to travel does not exist, but a board configured to close the gate a few minutes after it was opened absolutely does. An hour is where "the board auto-closes" stops being a plausible reading of the setting.

ASSUMED_CLOSE_TRAVEL_SECONDS module-attribute

ASSUMED_CLOSE_TRAVEL_SECONDS = 30.0

Fallback deadline for the auto-close closing phase, in seconds.

Used only when :data:GATE_OPTION_AUTO_CLOSE_TIME is on while :data:GATE_OPTION_TRAVEL_TIME is left at 0, i.e. the user told us when the board closes the gate but not how long that takes. The module's own estimate for a leaf is 15-25 s (see :data:IMPULSE_COOLDOWN_SECONDS), so 30 s covers a slow one with margin while still bounding how long a fabricated closing can survive without the contact confirming it.

A leaf slower than this is reported as open again while it is still closing: :meth:ImpulseGateEntity._expire_travel_if_due falls back to the last known position (with a warning), which un-blocks the button in the Sber app mid-motion. There is no way to guess better — the option's premise is that nothing about the close is observable — so the gate form's auto_close_time description tells the user to set :data:GATE_OPTION_TRAVEL_TIME alongside it, which replaces this constant with their own measurement.

TRAVEL_CONFIRM_MARGIN_SECONDS module-attribute

TRAVEL_CONFIRM_MARGIN_SECONDS = 0.5

Extra delay added to the deadline republish requested from the bridge.

:attr:ImpulseGateEntity.pending_confirm_delay is consumed by SberBridge.schedule_confirm, which sleeps on the event loop while the entity measures the deadline with its own injectable clock. The margin makes sure the republish observes an expired deadline instead of racing it by a few milliseconds and publishing opening one last time.

IMPULSE_SERVICE_AUTO module-attribute

IMPULSE_SERVICE_AUTO = 'auto'

impulse_service option value: pick the service from the HA domain.

IMPULSE_SERVICE_TOGGLE module-attribute

IMPULSE_SERVICE_TOGGLE = 'toggle'

impulse_service option value: force switch.toggle.

IMPULSE_SERVICE_TURN_ON module-attribute

IMPULSE_SERVICE_TURN_ON = 'turn_on'

impulse_service option value: force switch.turn_on.

IMPULSE_SERVICE_OPTIONS module-attribute

Accepted values of the per-entity impulse_service gate option.

OPEN_STATE_OPEN module-attribute

OPEN_STATE_OPEN = 'open'

Sber open_state / open_set enum value for an open gate.

OPEN_STATE_CLOSE module-attribute

OPEN_STATE_CLOSE = 'close'

Sber open_state / open_set enum value for a closed gate.

OPEN_STATE_OPENING module-attribute

OPEN_STATE_OPENING = 'opening'

Sber open_state enum value published while the leaf is opening.

Only ever published when the travel-time emulation is switched on, and then always together with an allowed_values.open_state declaration — Sber silently drops a state value it was not told about (issue #44).

OPEN_STATE_CLOSING module-attribute

OPEN_STATE_CLOSING = 'closing'

Sber open_state enum value published while the leaf is closing.

TRAVEL_OPEN_STATE_VALUES module-attribute

open_state enum values declared while the travel emulation is on.

IMPULSE_DOMAINS module-attribute

IMPULSE_DOMAINS = _PRESS_DOMAINS | _TURN_ON_DOMAINS | _TOGGLE_DOMAINS

Every HA domain an impulse can actually be sent to.

Anything else gets no service call: the gate category can be forced onto an arbitrary entity with set_override (which bypasses :meth:~sber_entity_map.CategorySpec.matches), and inventing <domain>.toggle for, say, a lock would turn every Sber command into a ServiceNotFound.

GateEntity

GateEntity(entity_data)

Bases: CurtainEntity

Sber gate entity for gate/garage door control.

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

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

Initialize gate entity.

Parameters:

Name Type Description Default
entity_data dict

HA entity registry dict containing entity metadata.

required
Source code in custom_components/sber_mqtt_bridge/devices/gate.py
def __init__(self, entity_data: dict) -> None:
    """Initialize gate entity.

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

ImpulseGateEntity

ImpulseGateEntity(entity_data, category=GATE_ENTITY_CATEGORY)

Bases: BatteryAndSignalLinkMixin, BaseEntity

Sber gate device built from an impulse relay + a reed contact.

The primary HA entity is the relay that pulses the gate motor (switch, button, input_button or script). Its own HA state is merely an echo of the last value written to the relay — it "sticks" and must never be read as a position. The position comes exclusively from a binary_sensor linked in the :data:~devices.base_entity.ROLE_OPEN_STATE role, hence :attr:REQUIRED_LINK_ROLES.

Sber features: online (obligatory), open_set (ENUM open/close), open_state (obligatory) and signal_strength when a signal sensor is linked. There is deliberately no open_percentage: an impulse drive has no position, and the Sber spec marks that feature conditional. stop is not declared either — a single button cannot stop the leaf.

With the opt-in :data:GATE_OPTION_TRAVEL_TIME option the entity also emulates the movement itself: after an impulse it publishes opening / closing (declared in allowed_values.open_state for exactly as long as the option is on) until either the contact reports — the contact always wins — or the travel time elapses without confirmation, which logs a warning and falls back to the last known position. With the option off (the default) nothing about the device changes: same features, same allowed_values, same model.id.

The second opt-in, :data:GATE_OPTION_AUTO_CLOSE_TIME, mirrors a setting that lives on the gate board itself (see :meth:auto_close_time). Both are off by default.

Initialize an impulse gate entity.

Parameters:

Name Type Description Default
entity_data dict

HA entity registry dict containing entity metadata.

required
category str

Sber device category (override in subclasses).

GATE_ENTITY_CATEGORY
Source code in custom_components/sber_mqtt_bridge/devices/gate.py
def __init__(self, entity_data: dict, category: str = GATE_ENTITY_CATEGORY) -> None:
    """Initialize an impulse gate entity.

    Args:
        entity_data: HA entity registry dict containing entity metadata.
        category: Sber device category (override in subclasses).
    """
    super().__init__(category, entity_data)
    self._open: bool = False
    self._contact_seen: bool = False
    self._contact_stale: bool = False
    self._invert_contact: bool = False
    self._impulse_service: str = IMPULSE_SERVICE_AUTO
    self._last_impulse_at: float | None = None
    self._missing_link_logged: bool = False
    self._unknown_domain_logged: bool = False
    # Travel-time emulation (opt-in, see ``travel_time``).  Both fields
    # are None while the leaf is considered at rest.
    self._travel_time: float = 0.0
    self._travel_direction: str | None = None
    self._travel_deadline: float | None = None
    self._travel_window: float = 0.0
    """Length of the window the current emulation was started with.

    Not always :attr:`travel_time`: the auto-close ``closing`` phase
    falls back to :data:`ASSUMED_CLOSE_TRAVEL_SECONDS` when no travel
    time is configured, and the expiry warning must name the deadline
    that actually elapsed."""
    # Auto-close emulation (opt-in, see ``auto_close_time``).  The
    # deadline is armed by the contact reporting an *open* gate and is
    # None whenever no countdown is running.
    self._auto_close_time: float = 0.0
    self._auto_close_deadline: float | None = None
    # Anti-bounce window (seconds) between two impulses.
    self.impulse_cooldown: float = IMPULSE_COOLDOWN_SECONDS
    # Injectable monotonic clock — tests replace it; the command logic
    # never calls ``time.monotonic()`` directly.
    self._now: Callable[[], float] = time.monotonic

ENTITY_OPTIONS_BLOCK class-attribute

ENTITY_OPTIONS_BLOCK = 'gate_options'

Historical block name — the panel's gate form reads gate_options.

ATTR_SPECS class-attribute

ATTR_SPECS = BATTERY_SIGNAL_ATTR_SPECS_PRESERVE

Signal strength is preserved across primary refreshes.

The relay toggles constantly and usually carries no linkquality of its own, so a non-preserving spec would wipe the value injected by the linked signal sensor on every pulse (and flip the advertised feature set back and forth with it).

invert_contact property

invert_contact

Whether the linked contact reports on for a closed gate.

impulse_service_option property

impulse_service_option

Current impulse_service option (see :data:IMPULSE_SERVICE_OPTIONS).

travel_time property

travel_time

Configured leaf travel time in seconds (0 = emulation off).

auto_close_time property

auto_close_time

Delay after which the gate board closes the leaf on its own.

0 (the default) means the board has no such timer and nothing is emulated. A positive value is a user-entered mirror of a setting that lives on the gate controller: the bridge has no way to read it, and no way to observe the closing impulse the board sends to the motor.

The countdown starts from the moment the contact reports the gate open — never from a command — precisely because the board's own timer starts when the leaf opens, whoever opened it: the Sber app, a 433 MHz remote, a GSM call to the controller, or a hand on the button. Modelling it from our commands would leave every gate opened by a remote stuck at open in the Sber app.

contact_stale property

contact_stale

True when the contact sensor stopped reporting a usable value.

The last known position is kept published in that case (losing control of a gate is worse than showing a slightly stale position); this flag surfaces the fact in diagnostics.

travel_direction property

travel_direction

Direction currently emulated, or None when the leaf is at rest.

One of :data:OPEN_STATE_OPENING / :data:OPEN_STATE_CLOSING. Reading this settles every due deadline first (an elapsed auto-close countdown starts a closing phase, an elapsed travel ends one), so the value never outlives its window.

pending_confirm_delay property

pending_confirm_delay

Seconds after which the bridge should republish this gate's state.

Read by SberBridge after a command and after every state change of this gate or its contact: an emulated opening / closing value has to be replaced by a real one when its deadline passes, and an armed auto-close countdown has to turn into a published closing when it elapses. Only the bridge owns timers, so the entity can do no more than name the next moment it wants to be looked at. None means "nothing to schedule" — the default, both-options-off case.

Returns:

Type Description
float | None

Seconds until the next deadline plus

float | None

data:TRAVEL_CONFIRM_MARGIN_SECONDS, or None when

float | None

neither an emulated movement nor an auto-close countdown is

float | None

pending.

apply_entity_options

apply_entity_options(options)

Apply per-entity gate options from entry.options.

Parameters:

Name Type Description Default
options dict

Mapping with optional invert_contact (bool), impulse_service (one of :data:IMPULSE_SERVICE_OPTIONS), travel_time (seconds, see :meth:travel_time) and auto_close_time (seconds, see :meth:auto_close_time) keys. Unknown keys and invalid values are ignored, so a hand-edited config cannot break entity loading.

required
Source code in custom_components/sber_mqtt_bridge/devices/gate.py
def apply_entity_options(self, options: dict) -> None:
    """Apply per-entity gate options from ``entry.options``.

    Args:
        options: Mapping with optional ``invert_contact`` (bool),
            ``impulse_service`` (one of :data:`IMPULSE_SERVICE_OPTIONS`),
            ``travel_time`` (seconds, see :meth:`travel_time`) and
            ``auto_close_time`` (seconds, see :meth:`auto_close_time`)
            keys.  Unknown keys and invalid values are ignored, so a
            hand-edited config cannot break entity loading.
    """
    if not options:
        return
    invert = options.get(GATE_OPTION_INVERT_CONTACT)
    if isinstance(invert, bool):
        self._invert_contact = invert
    service = options.get(GATE_OPTION_IMPULSE_SERVICE)
    if service in IMPULSE_SERVICE_OPTIONS:
        self._impulse_service = service
    if GATE_OPTION_TRAVEL_TIME in options:
        self._apply_travel_time(options[GATE_OPTION_TRAVEL_TIME])
    if GATE_OPTION_AUTO_CLOSE_TIME in options:
        self._apply_auto_close_time(options[GATE_OPTION_AUTO_CLOSE_TIME])
    _LOGGER.debug(
        "Gate options for %s: invert_contact=%s impulse_service=%s travel_time=%s auto_close_time=%s",
        self.entity_id,
        self._invert_contact,
        self._impulse_service,
        self._travel_time,
        self._auto_close_time,
    )

apply_gate_options

apply_gate_options(options)

Deprecated alias of :meth:apply_entity_options.

Kept because the option store predates the generic mechanism and this name is part of the v1.42 public surface.

Parameters:

Name Type Description Default
options dict

See :meth:apply_entity_options.

required
Source code in custom_components/sber_mqtt_bridge/devices/gate.py
def apply_gate_options(self, options: dict) -> None:
    """Deprecated alias of :meth:`apply_entity_options`.

    Kept because the option store predates the generic mechanism and
    this name is part of the v1.42 public surface.

    Args:
        options: See :meth:`apply_entity_options`.
    """
    self.apply_entity_options(options)

entity_options_state

entity_options_state()

Return the gate option block rendered by the panel.

Every key the panel's gate form reads must be present: the form submits all of its fields at once, so a control left without a value would reset the stored option the next time the user toggles its neighbour. contact_stale is read-only status, not an option — it rides along because the same form displays it.

Returns:

Type Description
dict[str, object]

invert_contact / impulse_service / travel_time /

dict[str, object]

auto_close_time plus the contact_stale indicator.

Source code in custom_components/sber_mqtt_bridge/devices/gate.py
def entity_options_state(self) -> dict[str, object]:
    """Return the gate option block rendered by the panel.

    Every key the panel's gate form reads must be present: the form
    submits all of its fields at once, so a control left without a
    value would reset the stored option the next time the user toggles
    its neighbour.  ``contact_stale`` is read-only status, not an
    option — it rides along because the same form displays it.

    Returns:
        ``invert_contact`` / ``impulse_service`` / ``travel_time`` /
        ``auto_close_time`` plus the ``contact_stale`` indicator.
    """
    return {
        GATE_OPTION_INVERT_CONTACT: self._invert_contact,
        GATE_OPTION_IMPULSE_SERVICE: self._impulse_service,
        "contact_stale": self._contact_stale,
        GATE_OPTION_TRAVEL_TIME: self._travel_time,
        GATE_OPTION_AUTO_CLOSE_TIME: self._auto_close_time,
    }

fill_by_ha_state

fill_by_ha_state(ha_state)

Refresh from the HA state of the impulse relay.

Only the relay's attributes are of interest: its state is an echo of the last written value and is never a position (see :meth:update_linked_data). Applying :attr:ATTR_SPECS here is what lets a Zigbee relay advertise its own linkquality / rssi as signal_strength; the specs preserve on missing, so a value injected by a linked signal sensor survives every relay refresh.

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/gate.py
def fill_by_ha_state(self, ha_state: dict) -> None:
    """Refresh from the HA state of the impulse relay.

    Only the relay's *attributes* are of interest: its ``state`` is an
    echo of the last written value and is never a position (see
    :meth:`update_linked_data`).  Applying :attr:`ATTR_SPECS` here is
    what lets a Zigbee relay advertise its own ``linkquality`` /
    ``rssi`` as ``signal_strength``; the specs preserve on missing, so
    a value injected by a linked signal sensor survives every relay
    refresh.

    Args:
        ha_state: HA state dict with ``state`` and ``attributes`` keys.
    """
    super().fill_by_ha_state(ha_state)
    self._apply_attr_specs(ha_state.get("attributes", {}))

update_linked_data

update_linked_data(role, ha_state)

Apply a linked companion state.

open_state is handled here (it is this device's position); every other role falls through to :class:~devices.battery_signal_mixin.BatteryAndSignalLinkMixin.

A running travel emulation is cancelled only when the contact brings genuinely new knowledge: a position different from the one already known (the leaf arrived), a first-ever reading, or a drop-out that makes confirmation impossible for good. A reading that merely repeats the current position carries no information — while a 20-second leaf is opening, the reed contact legitimately still reads "closed" for the first part of the travel — and it reaches this method for reasons that have nothing to do with the gate at all:

  • HaStateForwarder forwards every state_changed event of a linked entity, including attribute-only ones, so a Zigbee contact reporting battery / linkquality used to kill the emulation seconds after the impulse;
  • an MQTT binding with force_update: true re-fires the same state on every message;
  • SberBridge._refresh_entity_from_ha (used when gate options are saved) feeds the current contact state back in, so saving an unrelated checkbox mid-travel used to cancel the movement.

A gate that really is stuck reports nothing at all, and is settled by the travel deadline with a warning — not by this method.

The same "new knowledge only" rule drives the auto-close countdown (opt-in :attr:auto_close_time): a fresh open reading arms it — whoever opened the gate, remote and GSM call included — a fresh closed reading or a drop-out drops it, and a reading that repeats the current position leaves it alone. Re-arming on every repeat would push the deadline away forever on a chatty sensor, while the board's timer keeps running regardless.

Parameters:

Name Type Description Default
role str

Link role name.

required
ha_state dict

HA state dict of the linked entity.

required
Source code in custom_components/sber_mqtt_bridge/devices/gate.py
def update_linked_data(self, role: str, ha_state: dict) -> None:
    """Apply a linked companion state.

    ``open_state`` is handled here (it *is* this device's position);
    every other role falls through to
    :class:`~devices.battery_signal_mixin.BatteryAndSignalLinkMixin`.

    A running travel emulation is cancelled only when the contact
    brings genuinely **new** knowledge: a position different from the
    one already known (the leaf arrived), a first-ever reading, or a
    drop-out that makes confirmation impossible for good.  A reading
    that merely repeats the current position carries no information —
    while a 20-second leaf is opening, the reed contact legitimately
    still reads "closed" for the first part of the travel — and it
    reaches this method for reasons that have nothing to do with the
    gate at all:

    * ``HaStateForwarder`` forwards *every* ``state_changed`` event of
      a linked entity, including attribute-only ones, so a Zigbee
      contact reporting ``battery`` / ``linkquality`` used to kill the
      emulation seconds after the impulse;
    * an MQTT binding with ``force_update: true`` re-fires the same
      state on every message;
    * ``SberBridge._refresh_entity_from_ha`` (used when gate options
      are saved) feeds the *current* contact state back in, so saving
      an unrelated checkbox mid-travel used to cancel the movement.

    A gate that really is stuck reports nothing at all, and is settled
    by the travel deadline with a warning — not by this method.

    The same "new knowledge only" rule drives the auto-close
    countdown (opt-in :attr:`auto_close_time`): a fresh **open**
    reading arms it — whoever opened the gate, remote and GSM call
    included — a fresh **closed** reading or a drop-out drops it, and
    a reading that repeats the current position leaves it alone.
    Re-arming on every repeat would push the deadline away forever on
    a chatty sensor, while the board's timer keeps running regardless.

    Args:
        role: Link role name.
        ha_state: HA state dict of the linked entity.
    """
    if role != "open_state":
        super().update_linked_data(role, ha_state)
        return
    raw = ha_state.get("state")
    if raw in (None, STATE_UNKNOWN, STATE_UNAVAILABLE):
        # Hold the last known position — see :attr:`contact_stale`.
        self._cancel_travel("contact sensor dropped out")
        self._cancel_auto_close("contact sensor dropped out")
        self._contact_stale = True
        return
    was_known = self._contact_seen and not self._contact_stale
    was_open = self._open
    self._open = (raw == HAState.ON) != self._invert_contact
    self._contact_seen = True
    self._contact_stale = False
    if not was_known or self._open != was_open:
        self._cancel_travel("contact sensor reported a new position")
        if self._open:
            self._arm_auto_close()
        else:
            self._cancel_auto_close("contact sensor reported a closed gate")

create_allowed_values_list

create_allowed_values_list()

Return allowed values for open_set (and open_state when moving).

open_set offers only open and close. stop is omitted on purpose: the hardware has a single button, so a "stop the gate" voice command would be accepted by Sber and then do nothing. The spec explicitly allows shortening the ENUM list.

open_state is declared only while an emulation that can publish a transient value is on — travel_time or auto_close_time, either of them is enough — and then with the two transient values it can publish. Sber silently ignores a state value the device never declared (the root cause behind issue #44), so opening / closing and their declaration must appear and disappear together. Leaving the key out while both features are off keeps the model descriptor — and therefore the capability digest behind model.id — exactly as it was before they existed.

Returns:

Type Description
dict[str, dict]

Allowed-values map for the Sber model descriptor.

Source code in custom_components/sber_mqtt_bridge/devices/gate.py
def create_allowed_values_list(self) -> dict[str, dict]:
    """Return allowed values for ``open_set`` (and ``open_state`` when moving).

    ``open_set`` offers only ``open`` and ``close``.  ``stop`` is
    omitted on purpose: the hardware has a single button, so a "stop
    the gate" voice command would be accepted by Sber and then do
    nothing.  The spec explicitly allows shortening the ENUM list.

    ``open_state`` is declared **only** while an emulation that can
    publish a transient value is on — ``travel_time`` or
    ``auto_close_time``, either of them is enough — and then with the
    two transient values it can publish.  Sber silently ignores a
    state value the device never declared (the root cause behind
    issue #44), so ``opening`` / ``closing`` and their declaration
    must appear and disappear together.  Leaving the key out while
    both features are off keeps the model descriptor — and therefore
    the capability digest behind ``model.id`` — exactly as it was
    before they existed.

    Returns:
        Allowed-values map for the Sber model descriptor.
    """
    features = set(self.get_final_features_list())
    allowed: dict[str, dict] = {}
    if SberFeature.OPEN_SET.value in features:
        allowed[SberFeature.OPEN_SET.value] = {
            "type": SberValueType.ENUM.value,
            "enum_values": {"values": [OPEN_STATE_OPEN, OPEN_STATE_CLOSE]},
        }
    if self._emulates_motion and SberFeature.OPEN_STATE.value in features:
        allowed[SberFeature.OPEN_STATE.value] = {
            "type": SberValueType.ENUM.value,
            "enum_values": {"values": list(TRAVEL_OPEN_STATE_VALUES)},
        }
    return allowed

make_gate_entity

make_gate_entity(entity_data)

Create the right gate entity for an HA entity.

Registered as CategorySpec.cls for the gate category. HA cover entities keep the historical :class:GateEntity (position-aware, unchanged model.id for existing users); every other domain gets the impulse-relay implementation.

Parameters:

Name Type Description Default
entity_data dict

HA entity registry dict containing entity metadata.

required

Returns:

Type Description
BaseEntity

class:GateEntity for cover.*, :class:ImpulseGateEntity

BaseEntity

otherwise.

Source code in custom_components/sber_mqtt_bridge/devices/gate.py
def make_gate_entity(entity_data: dict) -> BaseEntity:
    """Create the right ``gate`` entity for an HA entity.

    Registered as ``CategorySpec.cls`` for the ``gate`` category.  HA
    ``cover`` entities keep the historical :class:`GateEntity`
    (position-aware, unchanged ``model.id`` for existing users); every
    other domain gets the impulse-relay implementation.

    Args:
        entity_data: HA entity registry dict containing entity metadata.

    Returns:
        :class:`GateEntity` for ``cover.*``, :class:`ImpulseGateEntity`
        otherwise.
    """
    entity_id = entity_data.get("entity_id") or ""
    domain = entity_id.split(".", 1)[0] if "." in entity_id else ""
    if domain == "cover":
        return GateEntity(entity_data)
    return ImpulseGateEntity(entity_data)