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 <noreply@anthropic.com>

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}

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

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.
 

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

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)

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()