From 955c679f5eb86e9d1aa1816803a070d618778508 Mon Sep 17 00:00:00 2001 From: Andrei Buzera Date: Mon, 26 Oct 2026 21:49:34 +0200 Subject: [PATCH] Timestamp relay changes with a read-only state listener Polling only sees the relay's current state and last change, so two or more changes between polls (e.g. an on/off round trip) made the lag estimate unknown and paused learning for about half an hour. Subscribe to the relay's state changes (read-only) and integrate every recorded change between polls. Because HA delivers these events one loop iteration late, the current state stays authoritative and fills in a change not yet recorded. Unknown periods (non on/off states, missed transitions, event overflow) still restart the estimate with an explicit error bound. Without the listener, behavior is as before. HA skips unload callbacks after unexpected setup errors, so setup now removes the listener itself on any failure and registers it for unload only once setup succeeds. Tested on HA Core 2025.12.0 and 2026.2.3. Co-Authored-By: Claude Opus 5.5 --- custom_components/heating_mpc/__init__.py | 17 +++-- custom_components/heating_mpc/coordinator.py | 63 +++++++++++----- custom_components/heating_mpc/physics.py | 20 ++++++ tests/ha/test_lifecycle.py | 21 ++++++ tests/test_coordinator.py | 75 +++++++++++++++++++- tests/test_physics.py | 11 +++ 6 files changed, 185 insertions(+), 22 deletions(-) diff --git a/custom_components/heating_mpc/__init__.py b/custom_components/heating_mpc/__init__.py index 2e01191..a9065d6 100644 --- a/custom_components/heating_mpc/__init__.py +++ b/custom_components/heating_mpc/__init__.py @@ -6,8 +6,9 @@ from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant from homeassistant.helpers.storage import Store from homeassistant.helpers import entity_registry as er +from homeassistant.helpers.event import async_track_state_change_event -from .const import DOMAIN, CONF_OUTSIDE, WEATHER_SOURCES +from .const import CONF_HEATER, DOMAIN, CONF_OUTSIDE, WEATHER_SOURCES from .weather_sources import sources_for from .coordinator import HeatingMPCCoordinator @@ -17,13 +18,19 @@ PLATFORMS: list[Platform] = [Platform.SENSOR, Platform.BINARY_SENSOR] async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: coordinator = HeatingMPCCoordinator(hass, entry) await coordinator.async_restore_model() - await coordinator.async_config_entry_first_refresh() - hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator + # Read-only: timestamps relay changes between polls for the radiator-lag estimate. + unsub_relay = async_track_state_change_event( + hass, [coordinator.config[CONF_HEATER]], coordinator.async_relay_changed) try: + await coordinator.async_config_entry_first_refresh() + hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) - except Exception: - hass.data[DOMAIN].pop(entry.entry_id, None) + except BaseException: + # HA skips unload callbacks for unexpected setup errors, so clean up here. + unsub_relay() + hass.data.get(DOMAIN, {}).pop(entry.entry_id, None) raise + entry.async_on_unload(unsub_relay) # Remove only obsolete room forecasts belonging to this entry/platform. registry = er.async_get(hass) expected = {f"{entry.entry_id}_{room}_predicted_60min" for room in coordinator.rooms} diff --git a/custom_components/heating_mpc/coordinator.py b/custom_components/heating_mpc/coordinator.py index b661d85..8db4c74 100644 --- a/custom_components/heating_mpc/coordinator.py +++ b/custom_components/heating_mpc/coordinator.py @@ -2,12 +2,13 @@ from __future__ import annotations import logging +from collections import deque from datetime import datetime, timedelta, timezone from functools import partial from math import exp, isfinite from typing import Any -from homeassistant.core import HomeAssistant +from homeassistant.core import Event, HomeAssistant, callback from homeassistant.config_entries import ConfigEntry from homeassistant.helpers.storage import Store from homeassistant.helpers.update_coordinator import DataUpdateCoordinator @@ -21,11 +22,13 @@ from .const import ( from .data_quality import freshness from .physics import ( LAG_LEARN_MAX_UNCERTAINTY, LAG_TAU_SECONDS, Learner, advance_lag, - lag_from_time_in_state, optimize, + lag_from_time_in_state, optimize, relay_segments, ) from .weather_sources import choose_weather_sources, sources_for, normalize _LOGGER = logging.getLogger(__name__) +# Relay changes kept between polls; more than this in five minutes is chatter. +MAX_RELAY_EVENTS = 256 class HeatingMPCCoordinator(DataUpdateCoordinator[dict[str, Any]]): @@ -43,6 +46,9 @@ class HeatingMPCCoordinator(DataUpdateCoordinator[dict[str, Any]]): # Radiator-lag proxy at the last poll with a valid relay state. Tracked # independently of room sensors so lag survives room outages and transitions. self._lag: dict[str, Any] | None = None + # (time, new state) of relay changes since the last poll, from a read-only listener. + self._relay_events: deque[tuple[datetime, str]] = deque(maxlen=MAX_RELAY_EVENTS) + self._relay_events_dropped = False self._on_samples = 0 self._off_samples = 0 @@ -79,32 +85,55 @@ class HeatingMPCCoordinator(DataUpdateCoordinator[dict[str, Any]]): return freshness(getattr(state, "last_reported", getattr(state, "last_updated", None)), now, limit * 60) + @callback + def async_relay_changed(self, event: Event) -> None: + """Record relay state changes as they happen; read-only state listener.""" + old, new = event.data.get("old_state"), event.data.get("new_state") + if new is None: + self.record_relay_state(event.time_fired, "removed") + elif old is None or old.state != new.state: + self.record_relay_state(new.last_changed, new.state) + + def record_relay_state(self, when: datetime, state: str) -> None: + if len(self._relay_events) == self._relay_events.maxlen: + self._relay_events_dropped = True + self._relay_events.append((when, state)) + def _track_lag(self, now: datetime, heater_on: bool, last_changed: datetime) -> bool: """Advance the lag proxy to `now`; return True if the relay held one state throughout. - A single relay change since the last poll is integrated at its timestamp. - Anything ambiguous (several changes, an outage, clock jumps) restarts the - estimate from the time spent in the current state, with an explicit error bound. + Relay changes since the last poll are integrated at their timestamps. Anything + ambiguous (missed changes, an outage, clock jumps) restarts the estimate from + the time spent in the current state, with an explicit error bound. """ tracker = self._lag steady = False lag = uncertainty = None - if tracker is not None: - dt = (now - tracker["time"]).total_seconds() - if dt > 0 and heater_on == tracker["heater_on"] and last_changed <= tracker["time"]: - lag = advance_lag(tracker["lag"], heater_on, dt) - steady = True - elif dt > 0 and heater_on != tracker["heater_on"] and tracker["time"] < last_changed <= now: - before = (last_changed - tracker["time"]).total_seconds() - lag = advance_lag(tracker["lag"], tracker["heater_on"], before) - lag = advance_lag(lag, heater_on, (now - last_changed).total_seconds()) - if lag is not None: + if tracker is not None and now > tracker["time"]: + start = tracker["time"] + changes = [(when, state) for when, state in self._relay_events if start < when <= now] + # The current state is authoritative; its change event may still be queued, + # and without a listener it is the only change visible. + if last_changed > start and (not changes or changes[-1][0] < last_changed): + changes.append((last_changed, "on" if heater_on else "off")) + segments = None if self._relay_events_dropped else relay_segments( + tracker["heater_on"], start.timestamp(), + [(when.timestamp(), state) for when, state in changes], now.timestamp()) + if segments is not None and segments[-1][0] == heater_on: + lag = tracker["lag"] + for command, seconds in segments: + lag = advance_lag(lag, command, seconds) # The lag update is a contraction, so an initial error decays the same way. - uncertainty = tracker["uncertainty"] * exp(-dt / LAG_TAU_SECONDS) + uncertainty = tracker["uncertainty"] * exp( + -(now - start).total_seconds() / LAG_TAU_SECONDS) + steady = len(segments) == 1 if lag is None: lag, uncertainty = lag_from_time_in_state( heater_on, (now - last_changed).total_seconds()) self._lag = {"time": now, "heater_on": heater_on, "lag": lag, "uncertainty": uncertainty} + self._relay_events = deque((e for e in self._relay_events if e[0] > now), + maxlen=MAX_RELAY_EVENTS) + self._relay_events_dropped = False return steady def _prediction_inputs(self) -> tuple: @@ -191,6 +220,8 @@ class HeatingMPCCoordinator(DataUpdateCoordinator[dict[str, Any]]): status["heat_lag_uncertainty"] = round(self._lag["uncertainty"], 3) else: self._lag = None + self._relay_events.clear() + self._relay_events_dropped = False if (not heater_valid or climate is None or outside is None or any(temp is None for temp in temps.values())): self._previous = None diff --git a/custom_components/heating_mpc/physics.py b/custom_components/heating_mpc/physics.py index 33f4a69..d765dd3 100644 --- a/custom_components/heating_mpc/physics.py +++ b/custom_components/heating_mpc/physics.py @@ -29,6 +29,26 @@ def advance_lag(lag: float, command: bool, seconds: float) -> float: return lag + (1.0 - exp(-seconds / LAG_TAU_SECONDS)) * (float(command) - lag) +def relay_segments( + start_on: bool, start: float, changes: list[tuple[float, str]], end: float, +) -> list[tuple[bool, float]] | None: + """Split [start, end] into (command, seconds) segments from ordered relay changes. + + `changes` holds (timestamp, new_state) pairs. Return None when the command is + unknown for part of the interval: a state other than on/off, a change to the + state already in force (so transitions were missed), or out-of-order times. + """ + segments: list[tuple[bool, float]] = [] + command, since = start_on, start + for when, state in changes: + if state not in ("on", "off") or not since <= when <= end or (state == "on") == command: + return None + segments.append((command, when - since)) + command, since = state == "on", when + segments.append((command, end - since)) + return segments + + def lag_from_time_in_state(command: bool, seconds_in_state: float) -> tuple[float, float]: """Estimate the lag when only the time since the last relay change is known. diff --git a/tests/ha/test_lifecycle.py b/tests/ha/test_lifecycle.py index f12aeba..c0d8805 100644 --- a/tests/ha/test_lifecycle.py +++ b/tests/ha/test_lifecycle.py @@ -161,3 +161,24 @@ async def test_freshly_written_relay_state_defers_learning(hass, entry, inputs, await hass.async_block_till_done() assert coordinator.data["sample_result"] == "lag_uncertain" assert coordinator.learners["sensor.room"].count == 0 + + +async def test_relay_listener_integrates_changes_between_polls(hass, entry, inputs, freezer): + coordinator = await setup(hass, entry) + start = dt_util.utcnow() + for offset, state in ((60, "on"), (200, "off")): # on/off round trip inside one poll + freezer.move_to(start + timedelta(seconds=offset)) + hass.states.async_set("switch.boiler", state) + await hass.async_block_till_done() + for minutes, expected in ((5, "relay_changed"), (10, "learned")): + future = start + timedelta(minutes=minutes, seconds=1) + freezer.move_to(future) + async_fire_time_changed(hass, future) + await hass.async_block_till_done() + assert coordinator.data["sample_result"] == expected + # Without the listener this round trip would restart the estimate (about 0.46). + assert coordinator.data["heat_lag_uncertainty"] < 0.05 + assert await hass.config_entries.async_unload(entry.entry_id) + hass.states.async_set("switch.boiler", "on") + await hass.async_block_till_done() + assert not coordinator._relay_events diff --git a/tests/test_coordinator.py b/tests/test_coordinator.py index 9afa5b2..b328139 100644 --- a/tests/test_coordinator.py +++ b/tests/test_coordinator.py @@ -3,6 +3,7 @@ import importlib.util import sys import types import unittest +import unittest.mock from datetime import datetime, timedelta, timezone from math import exp, pi, sin from pathlib import Path @@ -35,9 +36,13 @@ def load_integration(): 'homeassistant', 'homeassistant.const', 'homeassistant.core', 'homeassistant.config_entries', 'homeassistant.helpers', 'homeassistant.helpers.storage', 'homeassistant.helpers.update_coordinator', - 'homeassistant.helpers.entity_registry')} + 'homeassistant.helpers.entity_registry', 'homeassistant.helpers.event')} modules['homeassistant.const'].Platform = types.SimpleNamespace(SENSOR='sensor', BINARY_SENSOR='binary_sensor') modules['homeassistant.core'].HomeAssistant = object + modules['homeassistant.core'].Event = object + modules['homeassistant.core'].callback = lambda func: func + modules['homeassistant.helpers.event'].async_track_state_change_event = ( + lambda hass, entity_ids, action: hass.track_state(entity_ids, action)) modules['homeassistant.config_entries'].ConfigEntry = object modules['homeassistant.helpers.storage'].Store = Store modules['homeassistant.helpers.update_coordinator'].DataUpdateCoordinator = CoordinatorBase @@ -64,6 +69,7 @@ class BoundaryTests(unittest.IsolatedAsyncioTestCase): data={}, states=types.SimpleNamespace(get=self.states.get), config=types.SimpleNamespace(units=types.SimpleNamespace(temperature_unit='°C')), services=types.SimpleNamespace(async_call=Mock(side_effect=AssertionError('No service calls permitted'))), + track_state=Mock(return_value=Mock(name='unsub_relay')), registry=types.SimpleNamespace(entries=[], async_remove=Mock()), config_entries=types.SimpleNamespace( async_forward_entry_setups=AsyncMock(), async_unload_platforms=AsyncMock(return_value=True), @@ -205,6 +211,10 @@ class BoundaryTests(unittest.IsolatedAsyncioTestCase): kept = types.SimpleNamespace(platform='heating_mpc', unique_id='entry_sensor.room_predicted_60min', entity_id='sensor.forecast') self.hass.registry.entries = [obsolete, kept] self.assertTrue(await integration.async_setup_entry(self.hass, self.entry)) + self.hass.track_state.assert_called_once_with(['switch.boiler'], unittest.mock.ANY) + unsub = self.hass.track_state.return_value + self.entry.async_on_unload.assert_any_call(unsub) + unsub.assert_not_called() self.hass.registry.async_remove.assert_called_once_with('sensor.old_forecast') await integration.async_reload_entry(self.hass, self.entry) self.hass.config_entries.async_reload.assert_awaited_once_with('entry') @@ -221,6 +231,7 @@ class BoundaryTests(unittest.IsolatedAsyncioTestCase): with self.assertRaises(RuntimeError): await integration.async_setup_entry(self.hass, self.entry) self.assertNotIn('entry', self.hass.data['heating_mpc']) + self.hass.track_state.return_value.assert_called_once_with() async def test_uninstall_removes_storage(self): store = Store() @@ -425,3 +436,65 @@ class BoundaryTests(unittest.IsolatedAsyncioTestCase): self.assertGreater(model.on_count, 100) self.assertAlmostEqual(model.theta[1], b, delta=0.1 * b) self.assertAlmostEqual(model.theta[0], a, delta=0.25 * a) + + async def test_several_relay_changes_between_polls_are_integrated(self): + advance = coordinator.advance_lag + await self.sample() + lag, uncertainty = self.c._lag['lag'], self.c._lag['uncertainty'] + start = self.now - timedelta(seconds=300) + # The listener saw on at +60 s and off at +200 s; the state reads off again. + self.c.record_relay_state(start + timedelta(seconds=60), 'on') + self.c.record_relay_state(start + timedelta(seconds=200), 'off') + self.switch('off', 100) + data = await self.sample() + lag = advance(advance(advance(lag, False, 60), True, 140), False, 100) + self.assertEqual(data['sample_result'], 'relay_changed') + self.assertAlmostEqual(self.c._lag['lag'], lag) + self.assertAlmostEqual(self.c._lag['uncertainty'], uncertainty * exp(-300 / 1200)) + self.assertEqual(len(self.c._relay_events), 0) + self.assertEqual((await self.sample())['sample_result'], 'learned') + + async def test_queued_relay_event_is_reconciled_from_current_state(self): + advance = coordinator.advance_lag + await self.sample() + lag = self.c._lag['lag'] + start = self.now - timedelta(seconds=300) + # Only the first change reached the listener before this poll. + self.c.record_relay_state(start + timedelta(seconds=60), 'on') + self.switch('off', 100) + await self.sample() + lag = advance(advance(advance(lag, False, 60), True, 140), False, 100) + self.assertAlmostEqual(self.c._lag['lag'], lag) + # The late event for the change already accounted for is ignored next time. + self.c.record_relay_state(start + timedelta(seconds=200), 'off') + self.assertEqual((await self.sample())['sample_result'], 'learned') + + async def test_relay_outage_or_dropped_events_restart_lag_estimate(self): + for events in ((('unavailable', 60), ('off', 120)), ((None, 0),)): + with self.subTest(events=events): + self.c._lag = self.c._previous = None + await self.sample() + start = self.now - timedelta(seconds=300) + if events[0][0] is None: + self.c._relay_events_dropped = True + for state, offset in events: + if state: + self.c.record_relay_state(start + timedelta(seconds=offset), state) + self.switch('off', 180) + data = await self.sample() + self.assertEqual(data['sample_result'], 'relay_changed') + self.assertAlmostEqual(data['heat_lag_uncertainty'], round(exp(-180 / 1200) / 2, 3)) + self.assertFalse(self.c._relay_events_dropped) + + async def test_relay_listener_records_only_state_changes(self): + def event(old, new, when): + return types.SimpleNamespace(time_fired=when, data={ + 'old_state': old and types.SimpleNamespace(state=old), + 'new_state': new and types.SimpleNamespace(state=new, last_changed=when)}) + self.c.async_relay_changed(event('off', 'on', self.now)) + self.c.async_relay_changed(event('on', 'on', self.now)) # attribute-only update + self.c.async_relay_changed(event('on', None, self.now)) # entity removed + self.assertEqual(list(self.c._relay_events), [(self.now, 'on'), (self.now, 'removed')]) + for _ in range(coordinator.MAX_RELAY_EVENTS): + self.c.record_relay_state(self.now, 'on') + self.assertTrue(self.c._relay_events_dropped) diff --git a/tests/test_physics.py b/tests/test_physics.py index dd3836f..520725e 100644 --- a/tests/test_physics.py +++ b/tests/test_physics.py @@ -148,6 +148,17 @@ class ModelTests(unittest.TestCase): lag, half_width = m.lag_from_time_in_state(command, 900) self.assertLessEqual(abs(m.advance_lag(start, command, 900) - lag), half_width + 1e-12) + def test_relay_segments(self): + seg = m.relay_segments + self.assertEqual(seg(False, 0, [], 300), [(False, 300)]) + self.assertEqual(seg(False, 0, [(60, 'on'), (200, 'off')], 300), + [(False, 60), (True, 140), (False, 100)]) + self.assertEqual(seg(True, 0, [(300, 'off')], 300), [(True, 300), (False, 0)]) + for changes in ([(60, 'unavailable')], [(60, 'off')], [(200, 'on'), (60, 'off')], + [(-1, 'on')], [(301, 'on')]): + with self.subTest(changes=changes): + self.assertIsNone(seg(False, 0, changes, 300)) + if __name__ == '__main__': unittest.main()