✨ Realtime measurement streaming over STOMP/WebSocket ([stream] extra) #48

Merged

mat merged 3 commits from feat/streaming-websocket into main

2026-07-13 14:00:01 +00:00

mat commented

2026-07-10 16:44:38 +00:00

Owner

Summary

Adds a realtime measurement streaming client over STOMP-over-WebSocket, as an optional comwatt-client[stream] extra. This is slice 1/2 of the streaming work (the remaining item from the client audit); a stacked follow-up PR adds auto-reconnect.

New public API:

ComwattClient.stream_measurements(site: dict) -> Iterator[Measurement | CapacityChanged] — sync generator; opens wss://frontage.energy.comwatt.com/ws reusing the cwt_session cookie, subscribes to the site's siteUid topic, sends /app/streaming/start, and yields live FLOW/STATE events.
Dataclasses Measurement (gateway_uid, capacity_id, measure_kind, value + parsed value_float/value_bool) and CapacityChanged (capacity_id, enable, raw).
Exception ComwattStreamingError.

Design (per grilling sign-off)

Sync generator (consistent with the rest of the client; HA wraps it in an executor).
Optional extra pip install comwatt-client[stream] (websocket-client); base import ComwattClient still works without it (lazy import → ComwattStreamingError if missing).
In-house minimal STOMP framing (no extra STOMP lib).
Takes a single site dict so the two-id pitfall is impossible by construction.
No reconnect in this slice (deferred to the stacked follow-up): on socket drop the generator raises and stops; the caller reconnects.

Load-bearing correctness (live-verified)

SUBSCRIBE uses the siteUid topic; SEND start body uses the numeric siteId. Both are pinned by a test that parses the actual STOMP frames sent.

Test plan

13 new streaming tests; full suite 116 passed, output pristine (filterwarnings=error).
mypy comwatt_client clean (disallow_untyped_defs=true).
Base package imports with websocket unavailable (lazy import verified).
Frame encode/parse round-trip; measurement yield + ping skip + capacity-change; missing cookie → ComwattAuthError; missing site keys → ValueError; missing extra → ComwattStreamingError; handshake failure → ComwattStreamingError; DISCONNECT/close cleanup on exhaustion and GeneratorExit; Cookie: header format.
README: Features bullet + ## Realtime streaming (optional) section.

Notes

Additive + optional → backward compatible. A version bump (→ 0.4.0) is left to the usual release step once merged.
Draft pending review; the stacked auto-reconnect PR follows.

## Summary Adds a **realtime measurement streaming** client over STOMP-over-WebSocket, as an **optional** `comwatt-client[stream]` extra. This is slice 1/2 of the streaming work (the remaining item from the client audit); a stacked follow-up PR adds auto-reconnect. New public API: - `ComwattClient.stream_measurements(site: dict) -> Iterator[Measurement | CapacityChanged]` — sync generator; opens `wss://frontage.energy.comwatt.com/ws` reusing the `cwt_session` cookie, subscribes to the site's `siteUid` topic, sends `/app/streaming/start`, and yields live `FLOW`/`STATE` events. - Dataclasses `Measurement` (gateway_uid, capacity_id, measure_kind, value + parsed `value_float`/`value_bool`) and `CapacityChanged` (capacity_id, enable, raw). - Exception `ComwattStreamingError`. ## Design (per grilling sign-off) - **Sync generator** (consistent with the rest of the client; HA wraps it in an executor). - **Optional extra** `pip install comwatt-client[stream]` (`websocket-client`); base `import ComwattClient` still works without it (lazy import → `ComwattStreamingError` if missing). - In-house minimal STOMP framing (no extra STOMP lib). - Takes a **single site dict** so the two-id pitfall is impossible by construction. - **No reconnect** in this slice (deferred to the stacked follow-up): on socket drop the generator raises and stops; the caller reconnects. ## Load-bearing correctness (live-verified) - SUBSCRIBE uses the **`siteUid`** topic; SEND `start` body uses the **numeric** `siteId`. Both are pinned by a test that parses the actual STOMP frames sent. ## Test plan - [x] 13 new streaming tests; full suite **116 passed**, output pristine (`filterwarnings=error`). - [x] `mypy comwatt_client` clean (`disallow_untyped_defs=true`). - [x] Base package imports with `websocket` unavailable (lazy import verified). - [x] Frame encode/parse round-trip; measurement yield + ping skip + capacity-change; missing cookie → `ComwattAuthError`; missing site keys → `ValueError`; missing extra → `ComwattStreamingError`; handshake failure → `ComwattStreamingError`; DISCONNECT/close cleanup on exhaustion and `GeneratorExit`; `Cookie:` header format. - [x] README: Features bullet + `## Realtime streaming (optional)` section. ## Notes - Additive + optional → backward compatible. A version bump (→ `0.4.0`) is left to the usual release step once merged. - **Draft** pending review; the stacked auto-reconnect PR follows.

mat added 3 commits

2026-07-10 16:44:38 +00:00

✨ Add realtime measurement streaming over STOMP/WebSocket ([stream] extra) 70661b38f0

Add an optional `comwatt-client[stream]` extra (websocket-client) that streams
live measurements for a single site over STOMP-over-WebSocket:

- `StreamingMixin.stream_measurements(site)` yields `Measurement` /
  `CapacityChanged` events from `wss://frontage.energy.comwatt.com/ws`.
- Subscribes to the `siteUid` topic (`/topic/sites/{siteUid}/capacityChanged`)
  and starts the stream with the numeric `site[id]` (pins the two-id pitfall).
- Lazy-imports websocket so the base package imports without the extra; raises
  `ComwattStreamingError` (with install hint) when the extra is missing.
- Best-effort DISCONNECT + close on generator close/socket drop; no reconnect.
- Exports `Measurement`, `CapacityChanged`, `ComwattStreamingError`.

Tests fake the WS via monkeypatched `_open_websocket`; 103 existing tests
unchanged, 8 new streaming tests, mypy clean.

✅ test: add cleanup and missing-id tests; pin websocket-client>=1.6 033d4a4437

✅ test+docs: handshake-failure test, _open_websocket header test, CapacityChanged.raw docstring

Tests / forgejo-pytest (push) Successful in 43s

Details

Tests / forgejo-pytest (pull_request) Successful in 42s

Details

45ad1606f1

mat referenced this pull request

2026-07-10 17:29:10 +00:00

✨ Opt-in auto-reconnect (backoff) for stream_measurements #49

mat changed title from ~~WIP: ✨ Realtime measurement streaming over STOMP/WebSocket ([stream] extra)~~ to ✨ Realtime measurement streaming over STOMP/WebSocket ([stream] extra)

2026-07-13 13:38:11 +00:00

mat merged commit 23b54f8851 into main

2026-07-13 14:00:01 +00:00

mat deleted branch feat/streaming-websocket

2026-07-13 14:00:01 +00:00

mat referenced this pull request from a commit

2026-07-13 14:00:02 +00:00

✨ Realtime measurement streaming over STOMP/WebSocket ([stream] extra) (#48)