From a94bc3bbf2f7fde61be66914113dc39148f97cab Mon Sep 17 00:00:00 2001 From: Franck Nijhof Date: Sun, 30 Aug 2026 15:26:02 +0200 Subject: [PATCH] Follow the Peblar charging session as it changes (#180744) --- homeassistant/components/peblar/__init__.py | 6 + homeassistant/components/peblar/const.py | 6 + homeassistant/components/peblar/websocket.py | 80 +++++++++++ tests/components/peblar/conftest.py | 12 +- tests/components/peblar/test_websocket.py | 133 +++++++++++++++++++ 5 files changed, 236 insertions(+), 1 deletion(-) create mode 100644 homeassistant/components/peblar/websocket.py create mode 100644 tests/components/peblar/test_websocket.py diff --git a/homeassistant/components/peblar/__init__.py b/homeassistant/components/peblar/__init__.py index 1c97f6dec0ca..08c301dd5fc9 100644 --- a/homeassistant/components/peblar/__init__.py +++ b/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) diff --git a/homeassistant/components/peblar/const.py b/homeassistant/components/peblar/const.py index 0982e1ef2352..40bb86c4dadf 100644 --- a/homeassistant/components/peblar/const.py +++ b/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 = { diff --git a/homeassistant/components/peblar/websocket.py b/homeassistant/components/peblar/websocket.py new file mode 100644 index 000000000000..d01558b1bd52 --- /dev/null +++ b/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 + ) diff --git a/tests/components/peblar/conftest.py b/tests/components/peblar/conftest.py index e8fe5fad7f44..0fc3dbe48b5b 100644 --- a/tests/components/peblar/conftest.py +++ b/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) diff --git a/tests/components/peblar/test_websocket.py b/tests/components/peblar/test_websocket.py new file mode 100644 index 000000000000..f106dd53e712 --- /dev/null +++ b/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()