pybragerone.models.events¶

Event bus and event classes for pybragerone.

Classes

`AlarmQuantityChanged`(devid, quantity, ...)	Per-module alarm count changed (REST prime or Socket.IO push).
`CloudSessionConnectivity`(up, source, , ], ...)	Library ↔ Brager cloud Socket.IO session (client transport health).
`EventBus`()	Event bus for managing parameter update events.
`FeatureChanged`(devid, feature, value)	Feature changed event representing a change in device feature state.
`ModuleConnectivity`(devid, online, source, , ...)	Module ↔ Brager cloud connectivity (SPA `connectedAt`).
`ParamUpdate`(devid, pool, chan, idx, value, ...)	Parameter update event carrying value and metadata updates.

class pybragerone.models.events.AlarmQuantityChanged(devid: str, quantity: int | None, source: ~typing.Literal['rest', 'ws'], changed: bool = True, ts: float = <factory>)[source]¶

Bases: object

Per-module alarm count changed (REST prime or Socket.IO push).

Parameters:

devid (str)
quantity (int | None)
source (Literal['rest', 'ws'])
changed (bool)
ts (float)

changed: bool = True¶: Where the observation came from.

devid: str¶

quantity: int | None¶: Device identifier.

source: Literal['rest', 'ws']¶: New alarm count when known.

ts: float¶: True when the quantity differs from the previous cache.

class pybragerone.models.events.CloudSessionConnectivity(up: bool, source: ~typing.Literal['connect', 'disconnect', 'stop'], changed: bool = True, ts: float = <factory>)[source]¶

Bases: object

Library ↔ Brager cloud Socket.IO session (client transport health).

Distinct from ModuleConnectivity (module ↔ cloud connectedAt). When this session drops the gateway self-heals: Engine.IO reset on connect timeout, supervisor reconnect, resubscribe + REST prime, REST re-prime on the connectivity poll while the socket is still down, and after repeated stale-ParamUpdate cycles a hard WS restart (SPA parity: connect → ModulesService.connect + REST parameters). Consumers register BragerOneGateway.on_cloud_session or poll BragerOneGateway.ws_session_up() so an outage is detectable without looking like a plant module going offline.

Parameters:

up (bool)
source (Literal['connect', 'disconnect', 'stop'])
changed (bool)
ts (float)

changed: bool = True¶: Why the session bit was updated.

source: Literal['connect', 'disconnect', 'stop']¶: True while this gateway’s Socket.IO client session is connected.

ts: float¶: True when up flipped versus the previous cache.

up: bool¶

class pybragerone.models.events.EventBus[source]¶

Bases: object

Event bus for managing parameter update events.

Provides publish-subscribe functionality for parameter updates with sequence numbering and thread-safe operations.

last_seq()[source]¶

Get the last sequence number.

Returns:: int – The last sequence number that was assigned, or -1 if no events have been published.
Return type:: int

async publish(upd: ParamUpdate)[source]¶

Publish an event to all subscribers.

Parameters:: upd (ParamUpdate) – The parameter update event to publish.
Return type:: None

async subscribe()[source]¶

Subscribe to events.

Returns:: AsyncGenerator[ParamUpdate] – An async iterator that yields parameter update events.
Return type:: AsyncGenerator[ParamUpdate]

class pybragerone.models.events.FeatureChanged(devid: str, feature: str, value: bool)[source]¶

Bases: object

Feature changed event representing a change in device feature state.

Parameters:

devid (str)
feature (str)
value (bool)

devid: str¶

feature: str¶: Device identifier.

value: bool¶: Name of the feature that changed.

class pybragerone.models.events.ModuleConnectivity(devid: str, online: bool, source: ~typing.Literal['rest', 'ws', 'derived'], connected_at: int | None = None, gateway: dict[str, ~typing.Any] | None = None, online_changed: bool = True, metadata_changed: bool = False, ts: float = <factory>)[source]¶

Bases: object

Module ↔ Brager cloud connectivity (SPA connectedAt).

This is not published on EventBus (which stays ParamUpdate-only for Home Assistant compatibility). Consumers register BragerOneGateway.on_module_connectivity or poll BragerOneGateway.module_online().

Mirrors the SPA module card / connection modal: online iff connectedAt is truthy (upstream uses 0 as offline). Live updates arrive on Socket.IO app:module:connection:status:changed with {devid: {connectedAt, gateway}}.

When the module is offline there is nothing the client can repair — observe and wait. The library’s own Socket.IO session is a separate layer (CloudSessionConnectivity) and must never be folded into online.

Parameters:

devid (str)
online (bool)
source (Literal['rest', 'ws', 'derived'])
connected_at (int | None)
gateway (dict[str, Any] | None)
online_changed (bool)
metadata_changed (bool)
ts (float)

connected_at: int | None = None¶: Where the observation came from (REST poll, WS push, or derived absence).

devid: str¶

gateway: dict[str, Any] | None = None¶: Raw connectedAt epoch seconds when known (0 means offline upstream).

metadata_changed: bool = False¶: True when the online bit flipped versus the previous cache.

online: bool¶: Device identifier.

online_changed: bool = True¶: Optional gateway blob from REST/WS (address, interface, version).

source: Literal['rest', 'ws', 'derived']¶: True when upstream connectedAt is truthy (SPA parity).

ts: float¶: True when connected_at and/or gateway changed without an online flip.

class pybragerone.models.events.ParamUpdate(devid: str, pool: str, chan: str, idx: int, value: Any | None, meta: dict[str, ~typing.Any]=<factory>, ts: float = <factory>, seq: int = 0)[source]¶

Bases: object

Parameter update event carrying value and metadata updates.

Parameters:

devid (str)
pool (str)
chan (str)
idx (int)
value (Any | None)
meta (dict[str, Any])
ts (float)
seq (int)

chan: str¶: Parameter pool name.

devid: str¶

idx: int¶: Channel identifier (v, s, u …).

meta: dict[str, Any]¶: New parameter value, can be None for meta-only updates.

pool: str¶: Device identifier.

seq: int = 0¶: Timestamp when the update occurred.

ts: float¶: Additional metadata dictionary.

value: Any | None¶: Parameter index.