Follow the Peblar charging session as it changes (#180744)

homeassistant/components/peblar/__init__.py

@@ -27,6 +27,7 @@ from .coordinator import (
     PeblarVersionDataUpdateCoordinator,
 )
 from .services import async_setup_services
+from .websocket import PeblarSessionListener
 
 CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
 
@@ -96,6 +97,11 @@ async def async_setup_entry(hass: HomeAssistant, entry: PeblarConfigEntry) -> bo
         version_coordinator=version_coordinator,
     )
 
+    listener = PeblarSessionListener(hass, entry, peblar, meter_coordinator)
+    entry.async_create_background_task(
+        hass, listener.async_run(), name=f"Peblar {entry.title} event stream"
+    )
+
     # Forward the setup to the platforms
     await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
 

homeassistant/components/peblar/const.py

@@ -1,5 +1,6 @@
 """Constants for the Peblar integration."""
 
+from datetime import timedelta
 import logging
 from typing import Final
 
@@ -10,6 +11,11 @@ DOMAIN: Final = "peblar"
 CONF_EVCC_ID: Final = "evcc_id"
 CONF_UID: Final = "uid"
 
+# The stream only makes the poll quicker, so there is no hurry, and no
+# point hammering a charger that is switched off.
+EVENT_STREAM_RETRY_MINIMUM: Final = timedelta(seconds=5)
+EVENT_STREAM_RETRY_MAXIMUM: Final = timedelta(minutes=5)
+
 LOGGER = logging.getLogger(__package__)
 
 PEBLAR_CHARGE_LIMITER_TO_HOME_ASSISTANT = {

homeassistant/components/peblar/websocket.py

@@ -0,0 +1,80 @@
+"""Live event stream for the Peblar integration."""
+
+import asyncio
+
+from peblar import Peblar, PeblarError, PeblarSessionStatus
+
+from homeassistant.core import HomeAssistant, callback
+
+from .const import EVENT_STREAM_RETRY_MAXIMUM, EVENT_STREAM_RETRY_MINIMUM, LOGGER
+from .coordinator import PeblarConfigEntry, PeblarDataUpdateCoordinator
+
+
+class PeblarSessionListener:
+    """Follows the charging session over the charger's event stream.
+
+    The charger pushes a session change as it happens, which the poll
+    would otherwise take up to its interval to notice. This only tells the
+    poll to catch up early, so a stream that never comes up, or one that
+    falls over, costs nothing beyond going back to the poll on its own.
+    """
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        entry: PeblarConfigEntry,
+        peblar: Peblar,
+        coordinator: PeblarDataUpdateCoordinator,
+    ) -> None:
+        """Initialize the listener."""
+        self._hass = hass
+        self._entry = entry
+        self._peblar = peblar
+        self._coordinator = coordinator
+        self._retry = EVENT_STREAM_RETRY_MINIMUM
+
+    async def async_run(self) -> None:
+        """Keep a subscription up for as long as the entry is loaded."""
+        self._retry = EVENT_STREAM_RETRY_MINIMUM
+
+        while True:
+            try:
+                await self._async_listen()
+            except PeblarError as error:
+                LOGGER.debug(
+                    "Peblar event stream for %s stopped: %s", self._entry.title, error
+                )
+
+            await asyncio.sleep(self._retry.total_seconds())
+            self._retry = min(self._retry * 2, EVENT_STREAM_RETRY_MAXIMUM)
+
+    async def _async_listen(self) -> None:
+        """Open the stream and stay on it until it closes."""
+        websocket = self._peblar.websocket()
+        try:
+            await websocket.connect()
+            await websocket.subscribe_session_status(self._handle_session_status)
+
+            # A charger that was unreachable at startup can leave the wait
+            # at its longest. A subscription that landed settles that, so a
+            # drop hours later is not held against whatever went before.
+            # Taking the socket without ever getting this far is not a
+            # working stream, and keeps backing off.
+            self._retry = EVENT_STREAM_RETRY_MINIMUM
+
+            await websocket.listen()
+        finally:
+            await websocket.disconnect()
+
+    @callback
+    def _handle_session_status(self, status: PeblarSessionStatus) -> None:
+        """Ask the poll to catch up, now the session has moved on.
+
+        The charger sends the current status right after subscribing, so
+        the first call says nothing new. Refreshing anyway is harmless and
+        cheaper than working out which one that was.
+        """
+        LOGGER.debug("Peblar session for %s is %s", self._entry.title, status.state)
+        self._entry.async_create_task(
+            self._hass, self._coordinator.async_request_refresh(), eager_start=False
+        )

tests/components/peblar/conftest.py

@@ -1,9 +1,10 @@
 """Fixtures for the Peblar integration tests."""
 
+import asyncio
 from collections.abc import Generator
 from contextlib import nullcontext
 import json
-from unittest.mock import MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 from peblar import (
     PeblarEVInterface,
@@ -81,6 +82,15 @@ def mock_peblar(request: pytest.FixtureRequest) -> Generator[MagicMock]:
             system_information
         )
 
+        # The event stream parks here until the entry unloads, the way a
+        # real one waits on the charger rather than returning.
+        async def _listen_until_cancelled() -> None:
+            await asyncio.Event().wait()
+
+        websocket = AsyncMock()
+        websocket.listen.side_effect = _listen_until_cancelled
+        peblar.websocket.return_value = websocket
+
         api = peblar.rest_api.return_value
         api.ev_interface.return_value = PeblarEVInterface.from_json(
             load_fixture("ev_interface.json", DOMAIN)

tests/components/peblar/test_websocket.py

@@ -0,0 +1,133 @@
+"""Tests for the Peblar event stream."""
+
+import asyncio
+from unittest.mock import MagicMock, patch
+
+from peblar import PeblarConnectionError, PeblarSessionStatus, SessionState
+import pytest
+
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+
+from tests.common import MockConfigEntry
+
+pytestmark = [
+    pytest.mark.parametrize("init_integration", [Platform.SENSOR], indirect=True),
+    pytest.mark.usefixtures("init_integration"),
+]
+
+
+async def test_the_stream_is_subscribed_to(mock_peblar: MagicMock) -> None:
+    """Test the charger's session is followed as soon as the entry loads."""
+    websocket = mock_peblar.websocket.return_value
+    websocket.connect.assert_awaited_once()
+    websocket.subscribe_session_status.assert_awaited_once()
+
+
+async def test_a_session_change_pulls_the_poll_forward(
+    hass: HomeAssistant,
+    mock_peblar: MagicMock,
+) -> None:
+    """Test an event asks the poll to catch up rather than waiting it out."""
+    meter = mock_peblar.rest_api.return_value.meter
+    meter.reset_mock()
+
+    websocket = mock_peblar.websocket.return_value
+    handle_session_status = websocket.subscribe_session_status.call_args.args[0]
+    handle_session_status(
+        PeblarSessionStatus(state=SessionState.CHARGING, meter_data=None)
+    )
+    await hass.async_block_till_done()
+
+    meter.assert_awaited()
+
+
+async def test_the_wait_backs_off_and_settles_once_the_charger_answers(
+    hass: HomeAssistant,
+    mock_peblar: MagicMock,
+    mock_config_entry: MockConfigEntry,
+) -> None:
+    """Test how long the stream waits between attempts.
+
+    A charger that cannot be reached is given more room each time. Once it
+    answers, that is settled: a drop hours later starts over from the
+    shortest wait rather than the longest one reached at startup.
+    """
+    websocket = mock_peblar.websocket.return_value
+    websocket.connect.side_effect = [
+        PeblarConnectionError("Gone"),
+        PeblarConnectionError("Still gone"),
+        None,
+        None,
+    ]
+
+    hang_ups = 0
+
+    async def _hang_up_once() -> None:
+        nonlocal hang_ups
+        hang_ups += 1
+        if hang_ups == 1:
+            return
+        await asyncio.Event().wait()
+
+    websocket.listen.side_effect = _hang_up_once
+
+    waits: list[float] = []
+
+    async def _record(delay: float) -> None:
+        waits.append(delay)
+
+    with patch("homeassistant.components.peblar.websocket.asyncio.sleep", _record):
+        await hass.config_entries.async_reload(mock_config_entry.entry_id)
+        await hass.async_block_till_done()
+
+    # Five seconds, then ten while the charger stays away. It answers on
+    # the third try and hangs up, and the wait is back to five.
+    assert waits[:3] == [5, 10, 5]
+
+
+async def test_a_subscription_that_never_lands_keeps_backing_off(
+    hass: HomeAssistant,
+    mock_peblar: MagicMock,
+    mock_config_entry: MockConfigEntry,
+) -> None:
+    """Test taking the socket is not the same as having a stream.
+
+    A charger that accepts the connection but never completes the
+    subscription would otherwise be retried every five seconds forever.
+    """
+    websocket = mock_peblar.websocket.return_value
+    subscriptions = 0
+
+    async def _refuse_twice(_callback: object) -> None:
+        nonlocal subscriptions
+        subscriptions += 1
+        if subscriptions <= 2:
+            raise PeblarConnectionError("Not listening")
+
+    websocket.subscribe_session_status.side_effect = _refuse_twice
+
+    waits: list[float] = []
+
+    async def _record(delay: float) -> None:
+        waits.append(delay)
+
+    with patch("homeassistant.components.peblar.websocket.asyncio.sleep", _record):
+        await hass.config_entries.async_reload(mock_config_entry.entry_id)
+        await hass.async_block_till_done()
+
+    # The socket opened every time, so a reset on that alone would have
+    # left both waits at five seconds.
+    assert waits[:2] == [5, 10]
+
+
+async def test_the_stream_is_closed_when_the_entry_unloads(
+    hass: HomeAssistant,
+    mock_peblar: MagicMock,
+    mock_config_entry: MockConfigEntry,
+) -> None:
+    """Test the charger is let go of when the entry goes away."""
+    await hass.config_entries.async_unload(mock_config_entry.entry_id)
+    await hass.async_block_till_done()
+
+    mock_peblar.websocket.return_value.disconnect.assert_awaited()