"""Event bus and event classes for pybragerone."""
from __future__ import annotations
import asyncio
import time
from collections.abc import AsyncGenerator
from contextlib import suppress
from dataclasses import dataclass, field
from typing import Any, Literal
ModuleOutageReason = Literal["rest", "ws", "derived"]
CloudOutageReason = Literal[
"disconnect",
"stop",
"handshake_503",
"connect_error",
"empty_queue",
"server_stop",
"eio_close",
"reconnect_error",
"supervisor_stale",
"force_reconnect",
"hard_reset",
]
ConnectivityEpisodeLayer = Literal["cloud", "module", "live_stale"]
[docs]
@dataclass(frozen=True)
class FeatureChanged:
"""Feature changed event representing a change in device feature state."""
devid: str
#: Device identifier.
feature: str
#: Name of the feature that changed.
value: bool
#: New boolean value of the feature.
[docs]
@dataclass(frozen=True)
class ModuleConnectivity:
"""Module ↔ Brager cloud connectivity (SPA ``connectedAt``).
This is **not** published on :class:`EventBus` (which stays ``ParamUpdate``-only
for Home Assistant compatibility). Consumers register
``BragerOneGateway.on_module_connectivity`` or poll
:meth:`BragerOneGateway.module_online`.
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
(:class:`CloudSessionConnectivity`) and must never be folded into ``online``.
"""
devid: str
#: Device identifier.
online: bool
#: ``True`` when upstream ``connectedAt`` is truthy (SPA parity).
source: Literal["rest", "ws", "derived"]
#: Where the observation came from (REST poll, WS push, or derived absence).
connected_at: int | None = None
#: Raw ``connectedAt`` epoch seconds when known (``0`` means offline upstream).
gateway: dict[str, Any] | None = None
#: Optional gateway blob from REST/WS (``address``, ``interface``, ``version``).
online_changed: bool = True
#: ``True`` when the online bit flipped versus the previous cache.
metadata_changed: bool = False
#: ``True`` when ``connected_at`` and/or ``gateway`` changed without an online flip.
ts: float = field(default_factory=time.time)
#: Timestamp when this signal was produced.
down_since: float | None = None
#: Wall-clock ``time.time()`` when the current offline outage started (``None`` if online).
down_for_s: float | None = None
#: Seconds offline so far while ``online`` is false; ``None`` when online.
reason: ModuleOutageReason | None = None
#: Client-side observation source for the current outage (``rest`` / ``ws`` / ``derived``).
last_down_for_s: float | None = None
#: Duration of the most recently completed offline outage.
last_reason: ModuleOutageReason | None = None
#: Observation source of the most recently completed offline outage.
[docs]
@dataclass(frozen=True)
class AlarmQuantityChanged:
"""Per-module alarm count changed (REST prime or Socket.IO push)."""
devid: str
#: Device identifier.
quantity: int | None
#: New alarm count when known.
source: Literal["rest", "ws"]
#: Where the observation came from.
changed: bool = True
#: ``True`` when the quantity differs from the previous cache.
ts: float = field(default_factory=time.time)
#: Timestamp when this signal was produced.
[docs]
@dataclass(frozen=True)
class CloudSessionConnectivity:
"""Library ↔ Brager cloud Socket.IO session (client transport health).
Distinct from :class:`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). Consumers register
``BragerOneGateway.on_cloud_session`` or poll :meth:`BragerOneGateway.ws_session_up`
so an outage is detectable without looking like a plant module going offline.
"""
up: bool
#: ``True`` while this gateway's Socket.IO client session is connected.
source: Literal["connect", "disconnect", "stop"]
#: Why the session bit was updated.
changed: bool = True
#: ``True`` when ``up`` flipped versus the previous cache.
ts: float = field(default_factory=time.time)
#: Timestamp when this signal was produced.
down_since: float | None = None
#: Wall-clock ``time.time()`` when the current session-down outage started (``None`` if up).
down_for_s: float | None = None
#: Seconds down so far while ``up`` is false; ``None`` when up.
reason: CloudOutageReason | 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_down_for_s: float | None = None
#: Duration of the most recently completed session-down outage.
last_reason: CloudOutageReason | None = None
#: Reason of the most recently completed session-down outage.
[docs]
@dataclass(frozen=True)
class ConnectivityEpisode:
"""One completed connectivity outage episode for diagnostics history (#379)."""
layer: ConnectivityEpisodeLayer
#: Which connectivity layer this episode belongs to.
started_at: float
#: Wall-clock ``time.time()`` when the outage started.
ended_at: float
#: Wall-clock ``time.time()`` when the outage ended.
down_for_s: float
#: Duration of the outage in seconds.
reason: str | None = None
#: Observation/source reason (cloud WS tokens, module source, or ``live_stale``).
devid: str | None = None
#: Module id when ``layer`` is ``module``; otherwise ``None``.
episode_id: str | None = None
#: Opaque id for correlating logs with diagnostics.
[docs]
@dataclass(frozen=True)
class LivePushHealth:
"""Live ``ParamUpdate`` push health while the Socket.IO session is up.
Distinct from :class:`CloudSessionConnectivity` (transport up/down) and
:class:`ModuleConnectivity` (module ``connectedAt``). A zombie session is
``up=True`` with ``healthy=False`` / ``live_stale_for_s`` set.
"""
healthy: bool | 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).
live_stale_for_s: float | None = None
#: Seconds since the last live ``ParamUpdate`` while push is unhealthy.
last_resumed_after_s: float | None = None
#: Duration of the most recently completed live-stale episode (after resume).
changed: bool = True
#: ``True`` when ``healthy`` flipped versus the previous cache.
ts: float = field(default_factory=time.time)
#: Timestamp when this signal was produced.
# Official SPA Layout / ObjectsLayout event name.
MODULE_CONNECTION_STATUS_CHANGED = "app:module:connection:status:changed"
# SPA EventChannel.SIGMA_NETWORK_EVENT_MODULE_MEMORY_UPDATED (0x16). Payload ``{devid}``;
# Layout/ObjectsLayout respond with REST ``POST /modules/parameters`` for that module.
MODULE_MEMORY_UPDATED = "22"
[docs]
@dataclass(frozen=True)
class ParamUpdate:
"""Parameter update event carrying value and metadata updates."""
devid: str
#: Device identifier.
pool: str
#: Parameter pool name.
chan: str
#: Channel identifier (``v``, ``s``, ``u`` ...).
idx: int
#: Parameter index.
value: Any | None
#: New parameter value, can be ``None`` for meta-only updates.
meta: dict[str, Any] = field(default_factory=dict)
#: Additional metadata dictionary.
ts: float = field(default_factory=time.time)
#: Timestamp when the update occurred.
seq: int = 0
#: Sequence number assigned by :class:`EventBus`.
[docs]
class EventBus:
"""Multicast bus for :class:`ParamUpdate` events only.
Connectivity, cloud session, live-push, and alarm quantity use gateway
callbacks — not this bus — so typed ``subscribe()`` loops stay ParamUpdate-only.
"""
def __init__(self) -> None:
"""Initialize the event bus."""
self._subs: list[asyncio.Queue[ParamUpdate]] = []
self._seq = 0
self._lock = asyncio.Lock()
[docs]
def last_seq(self) -> int:
"""Get the last sequence number.
Returns:
The last sequence number that was assigned, or -1 if no events have been published.
"""
return max(self._seq - 1, -1)
[docs]
async def publish(self, upd: ParamUpdate) -> None:
"""Publish an event to all subscribers.
Args:
upd: The parameter update event to publish.
"""
async with self._lock:
upd.__dict__["seq"] = self._seq # safe, despite frozen dataclass
self._seq += 1
# snapshot of subscriber list, so we don't hold lock during put()
targets: tuple[asyncio.Queue[ParamUpdate], ...] = tuple(self._subs)
# broadcast outside of lock
for q in targets:
await q.put(upd)
[docs]
async def subscribe(self) -> AsyncGenerator[ParamUpdate]:
"""Subscribe to events.
Returns:
An async iterator that yields parameter update events.
"""
q: asyncio.Queue[ParamUpdate] = asyncio.Queue()
async with self._lock:
self._subs.append(q)
try:
while True:
yield await q.get()
finally:
# unsubscribe subscriber
async with self._lock:
with suppress(ValueError):
self._subs.remove(q)