pybragerone.models.events

Event bus and event classes for pybragerone.

Classes

EventBus()

Event bus for managing parameter update events.

FeatureChanged(devid, feature, value)

Feature changed event representing a change in device feature state.

ParamUpdate(devid, pool, chan, idx, value, ...)

Parameter update event carrying value and metadata updates.

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

Device identifier.

value: bool

Name of the feature that changed.

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