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