From d73e32a4b9c96886acf5425b2d56b2a4f496f85d Mon Sep 17 00:00:00 2001 From: Andrei Buzera Date: Tue, 27 Oct 2026 03:37:51 +0200 Subject: [PATCH] Never publish a plan computed from inputs that have since changed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Confirmed: the check run after the optimizer compared only zone sensors, outdoor temperature, thermostat and relay. A window opening, a UPS load change or a wind change during the calculation went unnoticed; nothing prevented an older update that finished late from overwriting a newer result; and an update in flight during unload still saved. - Every input a prediction uses is captured as a normalized snapshot (zone and outdoor temperatures, wind speed/direction/gust, heat-source power, window states, relay state and switch time, thermostat mode and target) and compared after the calculation with tolerances (0.25 °C, 50 W, 1 m/s, 2 m/s gusts, 20° direction around the circle; exact for relay, thermostat and windows). A meaningful change discards the plan and lists the inputs in changed_inputs; noise does not. No immediate recalculation: an off-schedule poll would break the 5-minute learning interval; the next regular poll computes a fresh plan. - Updates are numbered; one overtaken by a newer update, or running when the entry unloads, keeps the newer data and skips saving and optimizing. Configuration changes reload the entry, so they are covered. - A per-entry storage lock makes a reloaded coordinator restore only after the previous one's save finished, and integration removal waits for it too, so a late save cannot recreate a deleted model file. Co-Authored-By: Claude Opus 5.5 --- custom_components/heating_mpc/__init__.py | 7 +- custom_components/heating_mpc/coordinator.py | 103 +++++++++++++----- custom_components/heating_mpc/data_quality.py | 34 ++++++ tests/test_coordinator.py | 103 ++++++++++++++++++ tests/test_data_quality.py | 16 +++ 5 files changed, 236 insertions(+), 27 deletions(-) diff --git a/custom_components/heating_mpc/__init__.py b/custom_components/heating_mpc/__init__.py index 0b4cc92..0eb32f4 100644 --- a/custom_components/heating_mpc/__init__.py +++ b/custom_components/heating_mpc/__init__.py @@ -12,7 +12,7 @@ from homeassistant.helpers.event import async_track_state_change_event from .const import CONF_HEATER, CONF_ROOMS, CONF_WEIGHTS, DOMAIN, CONF_OUTSIDE, WEATHER_SOURCES from .weather_sources import sources_for -from .coordinator import HeatingMPCCoordinator +from .coordinator import HeatingMPCCoordinator, storage_lock from .zones import SUBENTRY_ZONE, ZONE_KEY, ZONE_NAME, legacy_zone_data PLATFORMS: list[Platform] = [Platform.SENSOR, Platform.BINARY_SENSOR] @@ -62,7 +62,10 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: async def async_remove_entry(hass: HomeAssistant, entry: ConfigEntry) -> None: """Remove the learned model when this integration instance is deleted.""" store = Store(hass, 1, f"{DOMAIN}.{entry.entry_id}") - await store.async_remove() + # Wait for a save still in progress, which would otherwise recreate the file. + async with storage_lock(hass, entry.entry_id): + await store.async_remove() + hass.data.get(f"{DOMAIN}_storage_locks", {}).pop(entry.entry_id, None) async def async_migrate_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: diff --git a/custom_components/heating_mpc/coordinator.py b/custom_components/heating_mpc/coordinator.py index 460cff8..359f447 100644 --- a/custom_components/heating_mpc/coordinator.py +++ b/custom_components/heating_mpc/coordinator.py @@ -1,6 +1,7 @@ """Read existing HA entities, learn thermal response, produce shadow MPC decisions.""" from __future__ import annotations +import asyncio import logging from collections import deque from datetime import datetime, timedelta, timezone @@ -21,7 +22,7 @@ from .const import ( CONF_VENT_WIND, CONF_WINDOW_OPEN_MINUTES, DEFAULTS, READY_OFF_SAMPLES, READY_ON_SAMPLES, READY_SAMPLES, ) from .compatibility import FINGERPRINT_VERSION, baseline_fingerprints, changed, reason, zone_fingerprints -from .data_quality import ReportAgeHistory, freshness +from .data_quality import ReportAgeHistory, freshness, inputs_differ from .evaluation import ForecastEvaluator, model_snapshot from .heat_sources import SUBENTRY_HEAT_SOURCE, HeatSource, HeatSourceTracker, build_sources, normalize_power from .learning import FeatureLearner @@ -30,7 +31,9 @@ from .physics import ( lag_from_time_in_state, optimize, relay_segments, ) from .weather_sources import choose_weather_sources, sources_for, normalize -from .windows import SUBENTRY_WINDOW, VentilationTracker, Window, build_windows, facing_cos, ventilation_vector +from .windows import ( + SUBENTRY_WINDOW, VentilationTracker, Window, build_windows, contact_open, facing_cos, ventilation_vector, +) from .zone_model import ZoneModel, compile_step_model, extended_features, feature_vector from .zones import SUBENTRY_ZONE, Zone, build_zones, legacy_zone_data, zone_temperature @@ -43,6 +46,14 @@ MAX_MODEL_CHANGES = 20 BASE_TERMS = ("loss", "heat", "bias") +def storage_lock(hass: HomeAssistant, entry_id: str) -> asyncio.Lock: + """One lock per entry, shared by successive coordinators across reloads. + + A reload's new coordinator restores only after the old one's save finished. + """ + return hass.data.setdefault(f"{DOMAIN}_storage_locks", {}).setdefault(entry_id, asyncio.Lock()) + + def entry_zone_list(entry: ConfigEntry, config: dict[str, Any]) -> list[Zone]: """Zones from the entry's zone subentries; one zone per legacy room sensor otherwise.""" subentries = getattr(entry, "subentries", None) or {} @@ -130,8 +141,22 @@ class HeatingMPCCoordinator(DataUpdateCoordinator[dict[str, Any]]): self._relay_events_dropped = False self._on_samples = 0 self._off_samples = 0 + # Increases with every update; an older update finishing late never publishes. + self._run = 0 async def async_restore_model(self) -> None: + async with storage_lock(self.hass, self.entry.entry_id): + await self._async_restore_model() + + async def _async_save(self, outside_source: str | None) -> None: + async with storage_lock(self.hass, self.entry.entry_id): + await self._store.async_save(self._payload(outside_source)) + + def _stale(self, run: int) -> bool: + """True when a newer update started or the entry is unloading.""" + return run != self._run or getattr(self, "_shutdown_requested", False) + + async def _async_restore_model(self) -> None: stored = await self._store.async_load() or {} if not isinstance(stored, dict) or stored.get("schema_version", 1) not in (1, 2): _LOGGER.warning("Ignoring unsupported Heating MPC model data") @@ -189,7 +214,7 @@ class HeatingMPCCoordinator(DataUpdateCoordinator[dict[str, Any]]): # Save at once: a recalibration is not redone after a restart, and data # from before fingerprints gets them now, so the next edit is detected # even if it happens before the next learned sample. - await self._store.async_save(self._payload(stored.get("outside_source"))) + await self._store.async_save(self._payload(stored.get("outside_source"))) # lock held def _check_compatibility(self, stored: dict[str, Any] | None, base_prints: dict[str, str | None]) -> bool: """Recalibrate exactly the components whose configuration changed.""" @@ -318,6 +343,11 @@ class HeatingMPCCoordinator(DataUpdateCoordinator[dict[str, Any]]): self._relay_events_dropped = False return steady + def _superseded(self, status: dict[str, Any]) -> dict[str, Any]: + """Result of an update that a newer one (or unloading) overtook: keep what is shown.""" + current = getattr(self, "data", None) + return current if current is not None else {**status, "status": "inputs_changed", "proposal": None} + def _count_result(self, status: dict[str, Any], result: str) -> None: """Record how this poll was handled for learning, and the learning progress.""" status["sample_result"] = result @@ -435,22 +465,36 @@ class HeatingMPCCoordinator(DataUpdateCoordinator[dict[str, Any]]): return {"windows": windows, "ventilation": zones, "ventilation_events": list(self._ventilation.events)[-5:]} - def _prediction_inputs(self) -> tuple: - """Values the prediction depends on, used to detect changes during calculation.""" - heater = self.config[CONF_HEATER] - entities = sorted(set(self.sensors) | set(sources_for(self.config, "temperature")) | - {self.config[CONF_CLIMATE], heater}) - values = [] - for entity_id in entities: + def _input_snapshot(self) -> dict[tuple[str, str], Any]: + """Normalized value of every input a prediction uses, keyed by (kind, entity). + + Compared (with tolerances, see data_quality.inputs_differ) before publishing, + so a result computed from inputs that have since changed is not shown. + """ + def value(entity_id: str, metric: str): state = self.hass.states.get(entity_id) - if state is None: - values.append((entity_id, None)) - continue - attrs = state.attributes - values.append((entity_id, state.state, attrs.get("unit_of_measurement"), - attrs.get("temperature"), attrs.get("temperature_unit"), - state.last_changed if entity_id == heater else None)) - return tuple(values) + return normalize(metric, state.state, state.attributes.get("unit_of_measurement")) if state else None + + snapshot: dict[tuple[str, str], Any] = {} + for entity_id in set(self.sensors) | set(sources_for(self.config, "temperature")): + snapshot[("temperature", entity_id)] = value(entity_id, "temperature") + for metric in ("wind", "wind_direction", "wind_gust"): + for entity_id in sources_for(self.config, metric): + snapshot[(metric, entity_id)] = value(entity_id, metric) + for source in self.sources: + state = self.hass.states.get(source.sensor) + snapshot[("power", source.sensor)] = normalize_power( + state.state, state.attributes.get("unit_of_measurement")) if state else None + for window in self.windows: + state = self.hass.states.get(window.sensor) + snapshot[("window", window.sensor)] = contact_open(state.state if state else None, window.inverted) + heater = self.hass.states.get(self.config[CONF_HEATER]) + snapshot[("relay", self.config[CONF_HEATER])] = (heater.state, heater.last_changed) if heater else None + climate = self.hass.states.get(self.config[CONF_CLIMATE]) + snapshot[("thermostat", self.config[CONF_CLIMATE])] = ( + climate.state, climate.attributes.get("temperature"), climate.attributes.get("temperature_unit") + ) if climate else None + return snapshot async def _async_update_data(self) -> dict[str, Any]: now = datetime.now(timezone.utc) @@ -464,7 +508,9 @@ class HeatingMPCCoordinator(DataUpdateCoordinator[dict[str, Any]]): } snapshot = {entity: self.hass.states.get(entity) for entity in set(self.sensors) | weather_entity_ids} - inputs_before = self._prediction_inputs() + self._run += 1 + run = self._run + inputs_before = self._input_snapshot() sensor_quality = {} readings: dict[str, float | None] = {} keep_stale = self.config.get(CONF_ROOM_STALE_KEEP) is True @@ -649,9 +695,11 @@ class HeatingMPCCoordinator(DataUpdateCoordinator[dict[str, Any]]): self.ventilation.get(room))) for room in vals}, inputs={room: {"internal": internal[room], "ventilation": vent_inputs.get(room)} for room in vals}) status["forecast_accuracy"] = self._evaluator.summary(self.rooms) - if learned: + if learned and not self._stale(run): # Persist at each sample so restart does not erase hours of training. - await self._store.async_save(self._payload(active_sources["temperature"])) + await self._async_save(active_sources["temperature"]) + if self._stale(run): + return self._superseded(status) # Comfort weights are validated when zones are parsed (invalid -> 1.0). weights = {r: self.zones[r].comfort_weight for r in self.rooms} # Left-out zones take their weight with them; the rest are re-normalized. @@ -718,12 +766,17 @@ class HeatingMPCCoordinator(DataUpdateCoordinator[dict[str, Any]]): step_model=self._step_model(vals, outside, int(self.config[CONF_HORIZON]), weather, now.timestamp()), ) - # Discard a result if a value used by the prediction changed while - # saving or while the worker calculated the hypothetical plan. Optional - # diagnostic weather sensors are not prediction inputs. output = await self.hass.async_add_executor_job(job) - if self._prediction_inputs() != inputs_before: + if self._stale(run): + return self._superseded(status) + # Discard a result if an input it used changed meaningfully while saving + # or while the worker calculated the hypothetical plan. Not recomputed at + # once: an off-schedule poll would break the 5-minute learning interval; + # the next regular poll (at most 5 minutes) computes a fresh plan. + changed = inputs_differ(inputs_before, self._input_snapshot()) + if changed: status["status"] = "inputs_changed" + status["changed_inputs"] = [f"{kind}: {entity}" for kind, entity in changed[:10]] return status except (ValueError, RuntimeError, OverflowError) as err: _LOGGER.warning("MPC prediction failed: %s", err) diff --git a/custom_components/heating_mpc/data_quality.py b/custom_components/heating_mpc/data_quality.py index fefc6cd..917e77f 100644 --- a/custom_components/heating_mpc/data_quality.py +++ b/custom_components/heating_mpc/data_quality.py @@ -37,3 +37,37 @@ class ReportAgeHistory: else: buckets.append([hour, age_seconds]) return max(age for start, age in buckets if hour - 24 < start <= hour) + + +# How much an input may move while a prediction is computed before the result no +# longer describes the current situation. Below these, changes are sensor noise. +INPUT_TOLERANCES = { + "temperature": 0.25, # °C, zone sensors and outdoor sources + "power": 50.0, # W, heat-source sensors + "wind": 1.0, # m/s + "wind_gust": 2.0, # m/s + "wind_direction": 20.0, # degrees, compared around the circle +} + + +def inputs_differ(before: dict, after: dict) -> list: + """Inputs whose change is meaningful: keys are (kind, entity), values normalized. + + Kinds without a tolerance (relay, thermostat mode and target, window states, + configuration revision) must match exactly; a value appearing or disappearing + (for example a sensor becoming unavailable) always counts. + """ + changed = [] + for key in sorted(set(before) | set(after), key=str): + old, new = before.get(key), after.get(key) + tolerance = INPUT_TOLERANCES.get(key[0]) + if old is None or new is None or tolerance is None: + if old != new: + changed.append(key) + continue + difference = abs(new - old) + if key[0] == "wind_direction": + difference = min(difference, 360.0 - difference) + if difference > tolerance: + changed.append(key) + return changed diff --git a/tests/test_coordinator.py b/tests/test_coordinator.py index 8d22a59..7b2525b 100644 --- a/tests/test_coordinator.py +++ b/tests/test_coordinator.py @@ -2,6 +2,7 @@ import importlib.util import sys import types +import asyncio import unittest import unittest.mock from datetime import datetime, timedelta, timezone @@ -1193,3 +1194,105 @@ class BoundaryTests(unittest.IsolatedAsyncioTestCase): saved = without._payload('sensor.outside') back = await self.restored(saved, zones=[{**self.ZONES[0], 'sensors': ['sensor.other']}, self.ZONES[1]]) self.assertEqual((back.learners['living'].theta, back.learners['living'].count), ([0.004, 0.12, 0.001], 0)) + + def ready_office(self): + """Office with a UPS and a window, trained enough to optimize.""" + self.entry.subentries = { + 'z': types.SimpleNamespace(subentry_id='z', subentry_type='zone', data={ + 'key': 'office', 'name': 'Office', 'sensors': ['sensor.room']}), + 'h': types.SimpleNamespace(subentry_id='h', subentry_type='heat_source', data={ + 'key': 'ups', 'name': 'UPS', 'sensor': 'sensor.ups_power', 'zone': 'office'}), + 'w': types.SimpleNamespace(subentry_id='w', subentry_type='window', data={ + 'key': 'w', 'name': 'Window', 'sensor': 'binary_sensor.office_window', 'zone': 'office'})} + self.set_state('sensor.ups_power', '200', unit_of_measurement='W') + self.set_state('binary_sensor.office_window', 'off') + self.c = coordinator.HeatingMPCCoordinator(self.hass, self.entry) + base = self.c.learners['office'] + base.count, base.on_count, base.off_count = 200, 100, 100 + + async def change_during_optimization(self, change): + async def worker(job): + output = job() + change() + return output + self.hass.async_add_executor_job = worker + return await self.sample() + + async def test_meaningful_input_changes_during_optimization_discard_the_plan(self): + cases = { + 'window: binary_sensor.office_window': lambda: setattr(self.states['binary_sensor.office_window'], 'state', 'on'), + 'power: sensor.ups_power': lambda: setattr(self.states['sensor.ups_power'], 'state', '600'), + 'thermostat: climate.reference': lambda: self.states['climate.reference'].attributes.update(temperature=21), + 'temperature: sensor.room': lambda: setattr(self.states['sensor.room'], 'state', '20.6'), + } + for expected, change in cases.items(): + with self.subTest(change=expected): + self.setUp() + self.ready_office() + data = await self.change_during_optimization(change) + self.assertEqual((data['status'], data['proposal'], data['changed_inputs']), + ('inputs_changed', None, [expected])) + + async def test_sensor_noise_during_optimization_does_not_discard_the_plan(self): + self.ready_office() + def noise(): + self.states['sensor.room'].state = '20.1' + self.states['sensor.ups_power'].state = '230' + self.assertEqual((await self.change_during_optimization(noise))['status'], 'ready') + + async def test_an_older_run_finishing_late_never_overwrites_a_newer_result(self): + self.ready_office() + release = asyncio.Event() + async def slow_then_fast(job): + if not release.is_set() and not getattr(self, '_first_started', False): + self._first_started = True + await release.wait() + return job() + self.hass.async_add_executor_job = slow_then_fast + with patch.object(coordinator, 'datetime') as clock: + clock.now.return_value = self.now + older = asyncio.ensure_future(self.c._async_update_data()) + await asyncio.sleep(0) + clock.now.return_value = self.now + timedelta(seconds=60) + newer = await self.c._async_update_data() + self.c.data = newer # what DataUpdateCoordinator publishes + release.set() + result = await older + self.assertEqual(newer['status'], 'ready') + self.assertIs(result, newer) + + async def test_unloading_during_save_or_optimization_publishes_nothing_new(self): + self.ready_office() + await self.sample() # a first sample to learn from + previous = self.c.data = {'status': 'learning', 'proposal': None} + self.fresh('sensor.room', 'sensor.outside', 'sensor.ups_power') + async def unload_while_saving(payload): + self.c._shutdown_requested = True + self.c._store.async_save.side_effect = unload_while_saving + worker = AsyncMock(side_effect=AssertionError('no optimization after unload')) + self.hass.async_add_executor_job = worker + self.assertIs(await self.sample(), previous) + worker.assert_not_called() + + async def test_restore_waits_for_a_save_still_in_progress(self): + self.ready_office() + order = [] + release = asyncio.Event() + async def slow_save(payload): + order.append('save started') + await release.wait() + order.append('save finished') + self.c._store.async_save.side_effect = slow_save + saving = asyncio.ensure_future(self.c._async_save('sensor.outside')) + await asyncio.sleep(0) + successor = coordinator.HeatingMPCCoordinator(self.hass, self.entry) + async def load(): + order.append('load') + return None + successor._store.async_load.side_effect = load + restoring = asyncio.ensure_future(successor.async_restore_model()) + await asyncio.sleep(0) + self.assertEqual(order, ['save started']) # the new coordinator waits + release.set() + await asyncio.gather(saving, restoring) + self.assertEqual(order, ['save started', 'save finished', 'load']) diff --git a/tests/test_data_quality.py b/tests/test_data_quality.py index 41f38bf..84416a2 100644 --- a/tests/test_data_quality.py +++ b/tests/test_data_quality.py @@ -46,6 +46,22 @@ class FreshnessTests(unittest.TestCase): self.assertEqual(history.observe('sensor.a', later, 300), 300) self.assertEqual(history.observe('sensor.b', later, 30), 30) + def test_inputs_differ_ignores_noise_but_not_meaningful_changes(self): + before = {('temperature', 'sensor.room'): 20.0, ('power', 'sensor.ups'): 210.0, + ('wind_direction', 'sensor.dir'): 355.0, ('wind', 'sensor.wind'): 4.0, + ('window', 'binary_sensor.w'): False, ('thermostat', 'climate.t'): ('heat', 22, None)} + noise = {**before, ('temperature', 'sensor.room'): 20.1, ('power', 'sensor.ups'): 240.0, + ('wind_direction', 'sensor.dir'): 5.0, ('wind', 'sensor.wind'): 4.6} + self.assertEqual(quality.inputs_differ(before, noise), []) # 10° across north is small + for key, value in ((('temperature', 'sensor.room'), 20.5), (('power', 'sensor.ups'), 610.0), + (('wind_direction', 'sensor.dir'), 40.0), (('window', 'binary_sensor.w'), True), + (('window', 'binary_sensor.w'), None), (('thermostat', 'climate.t'), ('heat', 21, None)), + (('temperature', 'sensor.room'), None)): + with self.subTest(key=key, value=value): + self.assertEqual(quality.inputs_differ(before, {**before, key: value}), [key]) + self.assertEqual(quality.inputs_differ(before, {**before, ('power', 'sensor.new'): 5.0}), + [('power', 'sensor.new')]) + def test_freshness_settings_translated(self): for name in ('strings.json', 'translations/en.json', 'translations/ro.json'): data = json.loads((ROOT / name).read_text())