pybragerone.gateway

Gateway package: BragerOneGateway facade and internal helpers.

Public import path stays from pybragerone.gateway import BragerOneGateway. Internal layout: connectivity / recovery / session mixins behind the facade.

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, zombie_quarantine_after: int = 3, zombie_quarantine_s: float = 21600.0, connectivity_episode_limit: int = 20, get_modules_fail_offline_after: int = 3, cloud_session_down_hysteresis_s: float | None = None)

Bases: ConnectivityMixin, SessionMixin, RecoveryMixin

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 (list[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)

• zombie_quarantine_after (int)

• zombie_quarantine_s (float)

• connectivity_episode_limit (int)

• get_modules_fail_offline_after (int)

• cloud_session_down_hysteresis_s (float | None)

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, get_modules_fail_offline_after: int = 3)

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

• get_modules_fail_offline_after (int) – Deprecated no-op; 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_live_push(cb: Callable[[LivePushHealth], Awaitable[None] | None])

Register callback for live ParamUpdate push-health flips.

Callbacks receive LivePushHealth. Distinct from on_cloud_session(): Socket.IO can stay up while push goes stale (zombie). Prefer live_push_health() for polling.

Return type:

None

Parameters:

cb (Callable[[LivePushHealth], 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 start()

Start the whole flow (idempotent).

Return type:

None

async stop()

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

Return type:

None

ws_session_up()

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

Return type:

bool

class pybragerone.gateway.RealtimeManager(token: str, *, origin: str = 'https://one.brager.pl', referer: str = 'https://one.brager.pl/', io_base: str = 'https://io.brager.pl', socket_path: str = '/socket.io', namespace: str = '/ws', token_provider: Callable[[], Awaitable[str]] | None = None, connect_timeout_s: float = 20.0)

Bases: object

Thin Socket.IO wrapper for BragerOne realtime channel.

The manager keeps a single AsyncClient connection, exposes the Engine.IO SID and the namespace SID, and forwards selected events to a user-provided callback (EventHandler). It does not interpret payloads; that is the gateway’s responsibility.

Notes

• Authentication is provided only via HTTP headers (Bearer token).

• We always connect to the :data:`~.constants.WS_NAMESPACE` namespace.

• We listen to: snapshot and the various *:parameters:change events.

• Subscriptions are emitted in a few payload variants (modules / devids) and optionally include group_id.

Parameters:

• token (str)

• origin (str)

• referer (str)

• io_base (str)

• socket_path (str)

• namespace (str)

• token_provider (Callable[[], Awaitable[str]] | None)

• connect_timeout_s (float)

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

Register a callback to be called when the connection is established.

Return type:

None

Parameters:

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

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

Register a callback to be called when the Socket.IO session drops.

Return type:

None

Parameters:

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

async connect()

Open a Socket.IO connection with appropriate headers and wait for join.

Return type:

None

async disconnect()

Close the Socket.IO connection if open.

Return type:

None

engine_sid()

Return the underlying Engine.IO SID (transport-level).

Return type:

str | None

async force_reconnect()

Tear down a still-“connected” Socket.IO session and reconnect.

Used when ParamUpdates go silent while the client still reports up (zombie Engine.IO that skipped disconnect callbacks). The SPA recovers via built-in Socket.IO reconnect then connect → ModulesService.connect + REST /modules/parameters; our supervisor only acts when connected looks down, so this forces that same path (on_connected → gateway resubscribe + prime).

Waits for the namespace join (same as connect()) so callers can safely sid() / modules.connect immediately afterwards.

Return type:

None

property group_id: int | None

Return the optional group_id included in subscription payloads.

async hard_reset()

Abandon the Socket.IO client and open a brand-new transport session.

Stronger than force_reconnect(): stops the supervisor, replaces the AsyncClient (handlers re-bound), then runs a full connect(). Used when disconnect/reconnect on the same client fails to restore live ParamUpdate traffic.

Return type:

None

last_disconnect_reason()

Return the most recent classified disconnect reason, if any.

Return type:

Optional[Literal['disconnect', 'stop', 'handshake_503', 'connect_error', 'empty_queue', 'server_stop', 'eio_close', 'reconnect_error', 'supervisor_stale', 'force_reconnect', 'hard_reset']]

on_event(handler: EventDispatcher)

Register a single event dispatcher (gateway attaches here).

Return type:

None

Parameters:

handler (EventDispatcher)

async resubscribe()

Re-emit subscription events after a reconnect.

Return type:

None

sid()

Return the namespace SID (/ws), if available.

Return type:

str | None

async subscribe(modules: list[str])

Emit listen events for the provided devices (devids/modules).

Return type:

None

Parameters:

modules (list[str])

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

last_disconnect_reason()

Return type:

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