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.

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

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)

Parameters:

• api (ApiClient)

• object_id (int)

• modules (Iterable[str])

• ws (RealtimeManagerClient | None)

• owns_api (bool)

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)

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

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_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])

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_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 resubscribe()

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

Return type:

None

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

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

Bases: Protocol

Protocol for the WS client used by the gateway.

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

Return type:

None

Parameters:

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

async connect()

Return type:

None

async disconnect()

Return type:

None

engine_sid()

Return type:

str | None

property group_id: int | 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])