pybragerone.gateway

Gateway: WS connect → modules.connect → listen → prime.

Maintains the WS connection and emits ParamUpdate events on the EventBus. Does not contain heavy logic (such as mapping) internally — this is the role of ParamStore/HA.

Connectivity is two layers, both off the ParamUpdate EventBus:

• Module ↔ cloud — SPA connectedAt via on_module_connectivity (observe/wait).

• Library ↔ cloud — Socket.IO session via on_cloud_session (detect + self-heal).

Functions

module_connected_at_means_online(connected_at)

Return whether a connectedAt value means the module is online.

Classes

ApiClient(*args, **kwargs)	Protocol for the HTTP client used by the gateway.
BragerOneGateway(*, api, object_id, modules)	High-level orchestrator for BragerOne realtime data.
RealtimeManagerClient(*args, **kwargs)	Protocol for the WS client used by the gateway.

class pybragerone.gateway.ApiClient(*args, **kwargs)

Bases: Protocol

Protocol for the HTTP client used by the gateway.

This makes the gateway easy to test by allowing a lightweight fake.

property access_token: str

async close()

Return type:

None

async get_modules(object_id: int)

Return type:

list[Module]

Parameters:

object_id (int)

async modules_activity_quantity_prime(modules: list[str], *, return_data: bool = False)

Return type:

tuple[int, Any] | bool

Parameters:

• modules (list[str])

• return_data (bool)

async modules_alarms_quantity(modules: list[str], *, return_data: bool = False)

Return type:

tuple[int, Any] | bool

Parameters:

• modules (list[str])

• return_data (bool)

async modules_connect(wsid_ns: str, modules: list[str], group_id: int | None = None, engine_sid: str | None = None)

Return type:

bool

Parameters:

• wsid_ns (str)

• modules (list[str])

• group_id (int | None)

• engine_sid (str | None)

async modules_parameters_prime(modules: list[str], *, return_data: bool = False)

Return type:

tuple[int, Any] | bool

Parameters:

• modules (list[str])

• return_data (bool)

class pybragerone.gateway.BragerOneGateway(*, api: ApiClient, object_id: int, modules: Iterable[str], ws: RealtimeManagerClient | None = None, owns_api: bool = False, connectivity_poll_interval: float = 60.0, stale_prime_after_s: float = 180.0, zombie_hard_restart_after: int = 2, zombie_full_recycle_after: int = 3, zombie_rebuild_after: int = 2, zombie_recovery_cooldown_s: float = 300.0)

Bases: object

High-level orchestrator for BragerOne realtime data.

Flow:

1. ensure_auth (proactive/reactive refresh in HTTP client)

2. Socket.IO connect → modules.connect (binding WS with DEV)

3. subscribe to streams (parameters, activity)

4. “prime” (REST snapshot of parameters + activity quantities)

5. EventBus emits ParamUpdate for consumers (ParamStore/HA/CLI)

6. Background REST poll of get_modules diffs connectedAt and notifies on_module_connectivity (module↔cloud; separate from EventBus)

7. Socket.IO up/down notifies on_cloud_session (library↔cloud; self-healing)

Parameters:

• api (ApiClient)

• object_id (int)

• modules (Iterable[str])

• ws (RealtimeManagerClient | None)

• owns_api (bool)

• connectivity_poll_interval (float)

• stale_prime_after_s (float)

• zombie_hard_restart_after (int)

• zombie_full_recycle_after (int)

• zombie_rebuild_after (int)

• zombie_recovery_cooldown_s (float)

flatten_parameters(payload: dict[str, Any], *, source: str = 'unknown')

Convert WS/REST parameter payload into ParamUpdate events.

Return type:

list[ParamUpdate]

Parameters:

• payload (dict[str, Any])

• source (str)

async classmethod from_credentials(*, email: str, password: str, object_id: int, modules: Iterable[str], server: ServerConfig | None = None, ws: RealtimeManagerClient | None = None, api: TypeAliasForwardRef('pybragerone.api.BragerOneApiClient') | None = None, connectivity_poll_interval: float = 60.0)

Create a gateway from credentials.

This is a convenience helper for CLI/examples.

Parameters:

• email (str) – BragerOne account email.

• password (str) – BragerOne account password.

• object_id (int) – BragerOne object/group ID.

• modules (Iterable[str]) – Modules to subscribe.

• server (ServerConfig | None) – Optional server/platform configuration (e.g. TiSConnect).

• ws (RealtimeManagerClient | None) – Optional WS client instance (testing).

• api (Optional[pybragerone.api.BragerOneApiClient]) – Optional API client instance (testing/customization).

• connectivity_poll_interval (float) – See __init__().

Returns:

BragerOneGateway – An initialized gateway (not started).

Return type:

BragerOneGateway

async ingest_activity_quantity(data: dict[str, Any] | None)

Ingest /modules/activity/quantity prime (optional).

Return type:

None

Parameters:

data (dict[str, Any] | None)

async ingest_alarm_quantity(data: dict[str, Any] | None, *, source: Literal['rest', 'ws'] = 'rest', ws_floor: dict[str, int] | None = None, rest_seq: int | None = None)

Ingest alarm quantity payload and notify on_alarm_quantity listeners.

Return type:

None

Parameters:

• data (dict[str, Any] | None)

• source (Literal['rest', 'ws'])

• ws_floor (dict[str, int] | None)

• rest_seq (int | None)

async ingest_prime_parameters(data: dict[str, Any])

Treat /modules/parameters prime as “cold snapshot” and publish all pairs.

Return type:

None

Parameters:

data (dict[str, Any])

last_live_param_update_age_s()

Return seconds since the last live (WS) ParamUpdate, or None if never.

Return type:

float | None

last_param_update_age_s()

Return seconds since the last published ParamUpdate, or None if never.

Return type:

float | None

module_connected_at(devid: str)

Return the last connectedAt for devid, or None if unknown.

Return type:

int | None

Parameters:

devid (str)

module_gateway(devid: str)

Return the last gateway blob for devid (address/interface/version).

Return type:

dict[str, Any] | None

Parameters:

devid (str)

module_online(devid: str)

Return current online state for devid, or None if not yet known.

Return type:

bool | None

Parameters:

devid (str)

on_alarm_quantity(cb: Callable[[AlarmQuantityChanged], Awaitable[None] | None])

Register callback for per-module alarm count changes.

Callbacks receive AlarmQuantityChanged when REST prime or Socket.IO app:modules:alarms:quantity:change reports a new count for a subscribed module.

Return type:

None

Parameters:

cb (Callable[[AlarmQuantityChanged], Awaitable[None] | None])

on_any(cb: Callable[[str, Any], Awaitable[None] | None])

Register callback for any WS event for diagnostics.

Return type:

None

Parameters:

cb (Callable[[str, Any], Awaitable[None] | None])

on_cloud_session(cb: Callable[[CloudSessionConnectivity], Awaitable[None] | None])

Register callback for library↔cloud Socket.IO session up/down.

Callbacks receive CloudSessionConnectivity. Distinct from on_module_connectivity(): a dropped client session must self-heal and stay detectable without looking like a module went offline.

Return type:

None

Parameters:

cb (Callable[[CloudSessionConnectivity], Awaitable[None] | None])

on_module_connectivity(cb: Callable[[ModuleConnectivity], Awaitable[None] | None])

Register callback for module↔cloud online/offline (SPA connectedAt).

Callbacks receive ModuleConnectivity. This path is intentionally separate from bus so ParamUpdate subscribers stay unchanged. Offline modules are observed only — the client cannot repair plant↔cloud links.

Return type:

None

Parameters:

cb (Callable[[ModuleConnectivity], Awaitable[None] | None])

on_parameters_change(cb: Callable[[str, dict[str, Any]], Awaitable[None] | None])

Register callback for app:modules:parameters:change.

Return type:

None

Parameters:

cb (Callable[[str, dict[str, Any]], Awaitable[None] | None])

on_snapshot(cb: Callable[[dict[str, Any]], Awaitable[None] | None])

Register callback for snapshot event (full state-like payload).

Return type:

None

Parameters:

cb (Callable[[dict[str, Any]], Awaitable[None] | None])

async refresh_module_connectivity()

Refresh module↔cloud connectivity from REST get_modules.

Return type:

None

async resubscribe()

Call after WS reconnect to re-bind modules + prime again.

Returns:

bool – True when modules.connect succeeded and subscribe/prime ran; False when there is no WS client, no namespace SID, or connect failed. REST prime still runs when connect fails so the store is refreshed from the authoritative snapshot even without a live push binding.

Return type:

bool

async start()

Start the whole flow (idempotent).

Return type:

None

async stop()

Gracefully stop the gateway: drop WS and release HTTP resources.

Return type:

None

async wait_for_prime(timeout: float | None = None)

Wait until the latest prime pass is finished.

Parameters:

timeout (float | None) – Optional timeout in seconds. When None, waits indefinitely.

Returns:

bool – True if prime completion event was observed, False on timeout.

Return type:

bool

ws_session_up()

Return whether this gateway’s Socket.IO (library↔cloud) session is up.

Return type:

bool

class pybragerone.gateway.RealtimeManagerClient(*args, **kwargs)

Bases: Protocol

Protocol for the WS client used by the gateway.

add_on_connected(cb: Callable[[], Awaitable[None] | None])

Return type:

None

Parameters:

cb (Callable[[], Awaitable[None] | None])

add_on_disconnected(cb: Callable[[], Awaitable[None] | None])

Return type:

None

Parameters:

cb (Callable[[], Awaitable[None] | None])

async connect()

Return type:

None

async disconnect()

Return type:

None

engine_sid()

Return type:

str | None

async force_reconnect()

Return type:

None

property group_id: int | None

async hard_reset()

Return type:

None

on_event(handler: Any)

Return type:

None

Parameters:

handler (Any)

sid()

Return type:

str | None

async subscribe(modules: list[str])

Return type:

None

Parameters:

modules (list[str])

pybragerone.gateway.module_connected_at_means_online(connected_at: int)

Return whether a connectedAt value means the module is online.

Mirrors the SPA ternary connectedAt ? 'connected' : 'notConnected'. Upstream uses 0 as the offline sentinel (see fixtures and live payloads).

Return type:

bool

Parameters:

connected_at (int)