pybragerone.models.events

Event bus and event classes for pybragerone.

Functions

channel_for_event(event)

Return the BusChannel for a typed gateway event.

Classes

ActivityFeedInvalidate(devid, reason, ], ...)	Signal that module activity lists should be re-fetched from REST.
AlarmFeedInvalidate(devid, reason, ], source, ts)	Signal that module alarm lists should be re-fetched from REST.
AlarmQuantityChanged(devid, quantity, ...)	Per-module alarm count changed (REST prime or Socket.IO push).
BusChannel(*values)	Typed EventBus channels for additive multi-event fan-out (#408 / #409).
CloudSessionConnectivity(up, source, , ], ...)	Library ↔ Brager cloud Socket.IO session (client transport health).
ConnectivityEpisode(layer, started_at, ...)	One completed connectivity outage episode for diagnostics history (#379).
EventBus()	Multicast bus: subscribe() is ParamUpdate-only; channels are additive.
FeatureChanged(devid, feature, value)	Feature changed event representing a change in device feature state.
LivePushHealth(healthy, live_stale_for_s, ...)	Live ParamUpdate push health while the Socket.IO session is up.
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.ActivityFeedInvalidate(devid: str, reason: ~typing.Literal['quantity', 'task'], source: ~typing.Literal['ws'] = 'ws', ts: float = <factory>)

Bases: object

Signal that module activity lists should be re-fetched from REST.

Activity rows stay on REST; Socket.IO only indicates that the feed may have changed (quantity badge or task lifecycle events the SPA Activity page watches).

Parameters:

• devid (str)

• reason (Literal['quantity', 'task'])

• source (Literal['ws'])

• ts (float)

devid: str

reason: Literal['quantity', 'task']

Device identifier whose activity feed should refresh.

source: Literal['ws'] = 'ws'

Which upstream signal triggered the invalidate.

ts: float

Always ws for these invalidate signals.

class pybragerone.models.events.AlarmFeedInvalidate(devid: str, reason: ~typing.Literal['change', 'received'], source: ~typing.Literal['ws'] = 'ws', ts: float = <factory>)

Bases: object

Signal that module alarm lists should be re-fetched from REST.

SPA Alarms UI listens to app:module:alarms:change / …:received and reloads lists — it does not push row payloads on the Socket.IO event.

Parameters:

• devid (str)

• reason (Literal['change', 'received'])

• source (Literal['ws'])

• ts (float)

devid: str

reason: Literal['change', 'received']

Device identifier whose alarm feeds should refresh.

source: Literal['ws'] = 'ws'

Which Socket.IO alarm signal triggered the invalidate.

ts: float

Always ws for these invalidate signals.

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

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.BusChannel(*values)

Bases: StrEnum

Typed EventBus channels for additive multi-event fan-out (#408 / #409).

ACTIVITY_FEED = 'activity_feed'

ALARM_FEED = 'alarm_feed'

ALARM_QUANTITY = 'alarm_quantity'

CLOUD_SESSION = 'cloud_session'

LIVE_PUSH = 'live_push'

MODULE_CONNECTIVITY = 'module_connectivity'

PARAM = 'param'

class pybragerone.models.events.CloudSessionConnectivity(up: bool, source: ~typing.Literal['connect', 'disconnect', 'stop'], changed: bool = True, ts: float = <factory>, down_since: float | None = None, down_for_s: float | None = None, reason: ~typing.Literal['disconnect', 'stop', 'handshake_503', 'connect_error', 'empty_queue', 'server_stop', 'eio_close', 'reconnect_error', 'supervisor_stale', 'force_reconnect', 'hard_reset'] | None = None, last_down_for_s: float | None = None, last_reason: ~typing.Literal['disconnect', 'stop', 'handshake_503', 'connect_error', 'empty_queue', 'server_stop', 'eio_close', 'reconnect_error', 'supervisor_stale', 'force_reconnect', 'hard_reset'] | None = None)

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). Published on BusChannel.CLOUD_SESSION and via BragerOneGateway.on_cloud_session (compat); poll BragerOneGateway.ws_session_up() for a snapshot.

Parameters:

• up (bool)

• source (Literal['connect', 'disconnect', 'stop'])

• changed (bool)

• ts (float)

• down_since (float | None)

• down_for_s (float | None)

• reason (Literal['disconnect', 'stop', 'handshake_503', 'connect_error', 'empty_queue', 'server_stop', 'eio_close', 'reconnect_error', 'supervisor_stale', 'force_reconnect', 'hard_reset'] | None)

• last_down_for_s (float | None)

• last_reason (Literal['disconnect', 'stop', 'handshake_503', 'connect_error', 'empty_queue', 'server_stop', 'eio_close', 'reconnect_error', 'supervisor_stale', 'force_reconnect', 'hard_reset'] | None)

changed: bool = True

Why the session bit was updated.

down_for_s: float | None = None

Wall-clock time.time() when the current session-down outage started (None if up).

down_since: float | None = None

Timestamp when this signal was produced.

last_down_for_s: float | None = None

Client-side reason for the current down (coarse disconnect / stop or finer WS tokens such as handshake_503 / eio_close); not plant diagnostics.

last_reason: Literal['disconnect', 'stop', 'handshake_503', 'connect_error', 'empty_queue', 'server_stop', 'eio_close', 'reconnect_error', 'supervisor_stale', 'force_reconnect', 'hard_reset'] | None = None

Duration of the most recently completed session-down outage.

reason: Literal['disconnect', 'stop', 'handshake_503', 'connect_error', 'empty_queue', 'server_stop', 'eio_close', 'reconnect_error', 'supervisor_stale', 'force_reconnect', 'hard_reset'] | None = None

Seconds down so far while up is false; None when up.

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.ConnectivityEpisode(layer: Literal['cloud', 'module', 'live_stale'], started_at: float, ended_at: float, down_for_s: float, reason: str | None = None, devid: str | None = None, episode_id: str | None = None)

Bases: object

One completed connectivity outage episode for diagnostics history (#379).

Parameters:

• layer (Literal['cloud', 'module', 'live_stale'])

• started_at (float)

• ended_at (float)

• down_for_s (float)

• reason (str | None)

• devid (str | None)

• episode_id (str | None)

devid: str | None = None

Observation/source reason (cloud WS tokens, module source, or live_stale).

down_for_s: float

Wall-clock time.time() when the outage ended.

ended_at: float

Wall-clock time.time() when the outage started.

episode_id: str | None = None

Module id when layer is module; otherwise None.

layer: Literal['cloud', 'module', 'live_stale']

reason: str | None = None

Duration of the outage in seconds.

started_at: float

Which connectivity layer this episode belongs to.

class pybragerone.models.events.EventBus

Bases: object

Multicast bus: subscribe() is ParamUpdate-only; channels are additive.

subscribe() stays typed to ParamUpdate so ParamStore / HA parameter loops stay unbroken. Non-parameter signals are available via subscribe_channel() (and still via gateway on_* callbacks for compat). Alarm/activity row lists are never on the bus — REST remains source of truth.

last_seq()

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)

Publish a ParamUpdate to subscribe() and PARAM channel subscribers.

Parameters:

upd (ParamUpdate) – The parameter update event to publish.

Return type:

None

async publish_event(event: ParamUpdate | ModuleConnectivity | CloudSessionConnectivity | LivePushHealth | AlarmQuantityChanged | AlarmFeedInvalidate | ActivityFeedInvalidate)

Publish a typed event to the matching channel (and ParamUpdate subscribers).

Parameters:

event (ParamUpdate | ModuleConnectivity | CloudSessionConnectivity | LivePushHealth | AlarmQuantityChanged | AlarmFeedInvalidate | ActivityFeedInvalidate) – Gateway event dataclass to fan out.

Return type:

None

async subscribe()

Subscribe to ParamUpdate events only.

Returns:

AsyncGenerator[ParamUpdate] – An async iterator that yields parameter update events.

Return type:

AsyncGenerator[ParamUpdate]

async subscribe_channel(*channels: BusChannel | str)

Subscribe to one or more typed channels.

Parameters:

*channels (BusChannel | str) – Channels to receive (BusChannel or matching string values). At least one is required. Duplicates are ignored.

Yields:

Events published on any of the selected channels (FIFO per subscriber).

Raises:

ValueError – If no channels are provided or a name is not a known channel.

Return type:

AsyncGenerator[ParamUpdate | ModuleConnectivity | CloudSessionConnectivity | LivePushHealth | AlarmQuantityChanged | AlarmFeedInvalidate | ActivityFeedInvalidate]

class pybragerone.models.events.FeatureChanged(devid: str, feature: str, value: bool)

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.LivePushHealth(healthy: bool | None, live_stale_for_s: float | None = None, last_resumed_after_s: float | None = None, changed: bool = True, ts: float = <factory>)

Bases: object

Live ParamUpdate push health while the Socket.IO session is up.

Distinct from CloudSessionConnectivity (transport up/down) and ModuleConnectivity (module connectedAt). A zombie session is up=True with healthy=False / live_stale_for_s set.

Parameters:

• healthy (bool | None)

• live_stale_for_s (float | None)

• last_resumed_after_s (float | None)

• changed (bool)

• ts (float)

changed: bool = True

Duration of the most recently completed live-stale episode (after resume).

healthy: bool | None

last_resumed_after_s: float | None = None

Seconds since the last live ParamUpdate while push is unhealthy.

live_stale_for_s: float | None = None

True when live push is fresh; False when stale; None when N/A (session down) or unknown (session up but no live ParamUpdate yet).

ts: float

True when healthy flipped versus the previous cache.

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>, down_since: float | None = None, down_for_s: float | None = None, reason: ~typing.Literal['rest', 'ws', 'derived'] | None = None, last_down_for_s: float | None = None, last_reason: ~typing.Literal['rest', 'ws', 'derived'] | None = None)

Bases: object

Module ↔ Brager cloud connectivity (SPA connectedAt).

Published on EventBus channel BusChannel.MODULE_CONNECTIVITY and via BragerOneGateway.on_module_connectivity (compat). Poll BragerOneGateway.module_online() when a snapshot is enough.

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)

• down_since (float | None)

• down_for_s (float | None)

• reason (Literal['rest', 'ws', 'derived'] | None)

• last_down_for_s (float | None)

• last_reason (Literal['rest', 'ws', 'derived'] | None)

connected_at: int | None = None

Where the observation came from (REST poll, WS push, or derived absence).

devid: str

down_for_s: float | None = None

Wall-clock time.time() when the current offline outage started (None if online).

down_since: float | None = None

Timestamp when this signal was produced.

gateway: dict[str, Any] | None = None

Raw connectedAt epoch seconds when known (0 means offline upstream).

last_down_for_s: float | None = None

Client-side observation source for the current outage (rest / ws / derived).

last_reason: Literal['rest', 'ws', 'derived'] | None = None

Duration of the most recently completed offline outage.

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

reason: Literal['rest', 'ws', 'derived'] | None = None

Seconds offline so far while online is false; None when online.

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)

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.

pybragerone.models.events.channel_for_event(event: ParamUpdate | ModuleConnectivity | CloudSessionConnectivity | LivePushHealth | AlarmQuantityChanged | AlarmFeedInvalidate | ActivityFeedInvalidate)

Return the BusChannel for a typed gateway event.

Exact type matches first; otherwise the nearest registered base class via isinstance (so ParamUpdate subclasses keep working).

Parameters:

event (ParamUpdate | ModuleConnectivity | CloudSessionConnectivity | LivePushHealth | AlarmQuantityChanged | AlarmFeedInvalidate | ActivityFeedInvalidate) – Event instance to route.

Returns:

BusChannel – Matching bus channel.

Raises:

TypeError – If event is not a known GatewayEvent type.

Return type:

BusChannel