mat/homeassistant-comwatt
1
0
Fork 0
Code Issues 1 Pull requests 1 Projects Releases Packages Activity Actions

WIP: ✨ StreamManager + real-time switches via WebSocket (streaming 2/5) #57

Closed
mat wants to merge 2 commits from feat/stream-switches into feat/capacity-map
pull from: feat/stream-switches
merge into: mat:feat/capacity-map
Conversation 1 Commits 2 Files changed 6 +457 -7
mat commented 2026-07-14 14:24:05 +00:00
Owner
Copy link

Slice 2 of 5 (stacked on #56). Adds a per-site WebSocket stream that pushes switch STATE updates in real time (~15s, vs the 2-min poll).

What

  • New custom_components/comwatt/stream.py: ComwattStreamManager runs the blocking stream_measurements generator in an executor thread per site, pushes messages to an asyncio.Queue via loop.call_soon_threadsafe, drained by a consumer task that batches bursts and applies STATE switch updates to coordinator.data["switches"].
  • _apply_switch_updates: module-level pure function — routes STATE measurements via coordinator.capacity_map (slice 1), skips FLOW/QUANTITY/CapacityChanged and non-switch natures, mutates switches_data, returns whether anything changed.
  • _process_batch notifies entities via async_update_listeners() (NOT async_set_updated_data — that would reset the 2-min poll timer and regress power/energy).
  • Consumer has a try/except guard so a bad batch can't silently kill it.
  • Reconnect/backoff in async land (reconnect=False at the client), teardown via asyncio.Event stop flag. Dedicated ComwattClient (never shares the coordinator's requests.Session).
  • Wired into async_setup_entry (start) / async_unload_entry (stop, idempotent).
  • manifest.json: iot_class cloud_polling → cloud_push, +websocket-client==1.9.0.
  • UPDATE_INTERVAL unchanged (2 min) — power/energy still poll until slices 3/4.

Tests

7 new tests (T1-T6 + consumer-error-survival). 51/51 passing, ruff + mypy clean. T6 is a real end-to-end: sets up a switch device, starts the manager via async_setup_entry, manually queues a STATE measurement, asserts the switch flips through the real consumer, then teardown cleans up.

Stacking

Base: feat/capacity-map (PR #56). Rebase to main once #56 merges.

Slice 2 of 5 (stacked on #56). Adds a per-site WebSocket stream that pushes switch STATE updates in real time (~15s, vs the 2-min poll). ## What - New `custom_components/comwatt/stream.py`: `ComwattStreamManager` runs the blocking `stream_measurements` generator in an executor thread per site, pushes messages to an `asyncio.Queue` via `loop.call_soon_threadsafe`, drained by a consumer task that batches bursts and applies STATE switch updates to `coordinator.data["switches"]`. - `_apply_switch_updates`: module-level pure function — routes STATE measurements via `coordinator.capacity_map` (slice 1), skips FLOW/QUANTITY/CapacityChanged and non-switch natures, mutates `switches_data`, returns whether anything changed. - `_process_batch` notifies entities via `async_update_listeners()` (NOT `async_set_updated_data` — that would reset the 2-min poll timer and regress power/energy). - Consumer has a `try/except` guard so a bad batch can't silently kill it. - Reconnect/backoff in async land (`reconnect=False` at the client), teardown via `asyncio.Event` stop flag. Dedicated `ComwattClient` (never shares the coordinator's `requests.Session`). - Wired into `async_setup_entry` (start) / `async_unload_entry` (stop, idempotent). - `manifest.json`: `iot_class` cloud_polling → cloud_push, +`websocket-client==1.9.0`. - `UPDATE_INTERVAL` unchanged (2 min) — power/energy still poll until slices 3/4. ## Tests 7 new tests (T1-T6 + consumer-error-survival). 51/51 passing, ruff + mypy clean. T6 is a real end-to-end: sets up a switch device, starts the manager via `async_setup_entry`, manually queues a STATE measurement, asserts the switch flips through the real consumer, then teardown cleans up. ## Stacking Base: `feat/capacity-map` (PR #56). Rebase to `main` once #56 merges.
Adds ComwattStreamManager: per-site blocking stream_measurements generator
run in an executor thread, messages pushed to an asyncio.Queue via
loop.call_soon_threadsafe, drained by a consumer task that batches bursts
and applies STATE switch updates to coordinator.data. Reconnect/backoff in
async land (reconnect=False at the client), controlled by an asyncio.Event
stop flag for clean teardown. Dedicated ComwattClient (never shares the
coordinator's requests.Session).

_apply_switch_updates is a module-level pure function: routes STATE
measurements via coordinator.capacity_map (built in slice 1), skips FLOW/
QUANTITY/CapacityChanged and non-switch natures, mutates switches_data,
returns whether anything changed. _process_batch notifies the coordinator
only on change.

Wired into async_setup_entry (start) / async_unload_entry (stop, idempotent).
manifest: iot_class cloud_polling -> cloud_push, +websocket-client==1.9.0.
UPDATE_INTERVAL unchanged (2 min) — power/energy still poll until slices 3/4.

6 tests (T1-T6): pure helper edge cases + consumer end-to-end + teardown.
50/50 passing, ruff + mypy clean.
🐛 Guard stream consumer and preserve poll timer on switch updates (review)
All checks were successful
Validate / lint-ruff (push) Successful in 7s
Details
Validate / test-pytest (push) Successful in 3m3s
Details
Validate / type-check-mypy (push) Successful in 3m11s
Details
Validate / lint-ruff (pull_request) Successful in 7s
Details
Validate / test-pytest (pull_request) Successful in 3m1s
Details
Validate / type-check-mypy (pull_request) Successful in 3m8s
Details
278d893987 
Slice 2 code-review fixes for ComwattStreamManager:

- _process_batch: call async_update_listeners() instead of
  async_set_updated_data(data). The latter cancels+reschedules the
  coordinator refresh, so switch STATE updates (~every 15s) perpetually
  pushed out the 2-min poll and power/energy stopped updating while the
  stream was active. async_update_listeners refreshes entities without
  touching the refresh schedule.
- _consume: wrap the drain + _process_batch call in try/except so a
  single raised exception no longer kills the consumer task silently.
  CancelledError is always re-raised; other exceptions are logged and
  the consumer keeps running.

Tests: T5 now asserts on async_update_listeners; add
test_consume_swallows_batch_errors_and_keeps_running.
mat commented 2026-07-14 17:12:43 +00:00
Author
Owner
Copy link

Landed in main via the linear stack fast-forward (commit 9d016f7). This slice is part of the streaming feature merged as #56…#62 — closing (its commits are already in main).

Landed in `main` via the linear stack fast-forward (commit `9d016f7`). This slice is part of the streaming feature merged as #56…#62 — closing (its commits are already in `main`).
mat closed this pull request 2026-07-14 17:12:43 +00:00
All checks were successful
Validate / lint-ruff (push) Successful in 7s
Details
Validate / test-pytest (push) Successful in 3m3s
Details
Validate / type-check-mypy (push) Successful in 3m11s
Details
Validate / lint-ruff (pull_request) Successful in 7s
Details
Validate / test-pytest (pull_request) Successful in 3m1s
Details
Validate / type-check-mypy (pull_request) Successful in 3m8s
Details

Pull request closed

This pull request cannot be reopened because the branch was deleted.
Sign in to join this conversation.
Reviewers
No reviewers
Labels
Clear labels
No items
No labels
Milestone
Clear milestone
No items
No milestone
Projects
Clear projects
No items
No project
Assignees
Clear assignees
No assignees
1 participant
Notifications
Due date
The due date is invalid or out of range. Please use the format "yyyy-mm-dd".

No due date set.

Dependencies

No dependencies set.

Reference
mat/homeassistant-comwatt!57
No description provided.