From 8d197fabcd77b990aeabb4d7cbb6d2d6e0f2e260 Mon Sep 17 00:00:00 2001 From: Andrei Buzera Date: Tue, 27 Oct 2026 02:50:56 +0200 Subject: [PATCH] Explain each zone: contributions, forecast steps, coefficients MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Each zone forecast entity now shows the current temperature, the forecast at +15/+30/+60/+120 minutes (what the horizon covers), which model layer is in use, and what drives the zone right now in °C/h: envelope loss, boiler, bias, electrical heat, neighbours, ventilation and recovery, plus each layer's coefficients with standard error, confidence and whether it is identified. These attributes change every poll and are not recorded. Tests for inter-zone exchange on a two-room synthetic apartment (living room and a colder bedroom, one boiler): with a true coupling of 0.01 the learner finds a clearly positive value but, because the difference between rooms is fairly steady and therefore collinear with each room's bias, correctly reports it as not identified; with no coupling it stays near zero. Co-Authored-By: Claude Opus 5.5 --- custom_components/heating_mpc/coordinator.py | 22 +++++++ custom_components/heating_mpc/sensor.py | 16 ++++- tests/ha/test_lifecycle.py | 4 ++ tests/test_coordinator.py | 61 +++++++++++++++++++- 4 files changed, 99 insertions(+), 4 deletions(-) diff --git a/custom_components/heating_mpc/coordinator.py b/custom_components/heating_mpc/coordinator.py index 53da2a0..53f612b 100644 --- a/custom_components/heating_mpc/coordinator.py +++ b/custom_components/heating_mpc/coordinator.py @@ -297,6 +297,27 @@ class HeatingMPCCoordinator(DataUpdateCoordinator[dict[str, Any]]): "ventilation": ventilation[key][k]}) for key, value in state.items()} return step + def _zone_details(self, temps: dict[str, float], outside: float, internal: dict[str, float | None], + vent_inputs: dict[str, Any]) -> dict[str, Any]: + """What drives each zone right now (°C/h per effect) and how sure each layer is.""" + details = {} + lag = self._lag["lag"] + for key, temp in temps.items(): + model = ZoneModel(self.learners[key], self.extended.get(key), self.ventilation.get(key)) + inputs = {"outside": outside, "lag": lag, "internal": internal[key], "temps": temps, + "ventilation": vent_inputs.get(key) or {}} + a, b, c = self.learners[key].theta + coefficients = {"base": {"loss": round(a, 6), "heat": round(b, 6), "bias": round(c, 6)}} + for layer, learner in (("extended", model.extended), ("ventilation", model.ventilation)): + if learner is not None: + coefficients[layer] = {name: {k: info[k] for k in ("value", "std", "confidence", "identified")} + for name, info in learner.coefficients().items()} + details[key] = {"temperature": round(temp, 3), + "model_in_use": "extended" if model.uses_extended else "base", + "contributions_c_per_h": model.contributions(temp, inputs), + "coefficients": coefficients} + return details + def _window_diagnostics(self, now: float, weather: dict[str, Any], temps: dict[str, float | None], vent_inputs: dict[str, Any], internal: dict[str, float | None]) -> dict[str, Any]: windows = {} @@ -581,6 +602,7 @@ class HeatingMPCCoordinator(DataUpdateCoordinator[dict[str, Any]]): for room, model in self.learners.items() }, }) + status["zone_details"] = self._zone_details(vals, outside, internal, vent_inputs) if climate.state != "heat": status["status"] = "paused" return status diff --git a/custom_components/heating_mpc/sensor.py b/custom_components/heating_mpc/sensor.py index 64e8b1f..c48a3c4 100644 --- a/custom_components/heating_mpc/sensor.py +++ b/custom_components/heating_mpc/sensor.py @@ -51,7 +51,8 @@ class MPCStatus(CoordinatorEntity, SensorEntity): data = self.coordinator.data # Forecast details live on their own entities. return {key: value for key, value in data.items() if key not in ( - "status", "prediction_60m", "prediction_end", "prediction_trajectory", "forecast_accuracy")} + "status", "prediction_60m", "prediction_end", "prediction_trajectory", "forecast_accuracy", + "zone_details")} class MPCWeightedTemp(CoordinatorEntity, SensorEntity): @@ -82,7 +83,8 @@ class MPCRoomPrediction(CoordinatorEntity, SensorEntity): self._attr_name = f"MPC {coordinator.zones[room].name} 60min Prediction" # Change every poll; the 60-minute value itself is recorded. - _unrecorded_attributes = frozenset({"forecast_15min", "plan_15min"}) + _unrecorded_attributes = frozenset({"forecast_15min", "plan_15min", "forecast", "contributions_c_per_h", + "coefficients", "current_temperature", "model_in_use"}) @property def native_value(self): @@ -91,10 +93,18 @@ class MPCRoomPrediction(CoordinatorEntity, SensorEntity): @property def extra_state_attributes(self): data = self.coordinator.data + trajectory = data.get("prediction_trajectory", {}).get(self.room) or [] + details = data.get("zone_details", {}).get(self.room, {}) return { - "forecast_15min": data.get("prediction_trajectory", {}).get(self.room), + "current_temperature": details.get("temperature"), + "forecast": {f"{minutes}_min": trajectory[minutes // 15 - 1] + for minutes in (15, 30, 60, 120) if len(trajectory) >= minutes // 15}, + "forecast_15min": trajectory or None, "plan_15min": data.get("plan_15min"), "basis": "proposed MPC plan; counterfactual while Generic Thermostat controls the relay", + "model_in_use": details.get("model_in_use"), + "contributions_c_per_h": details.get("contributions_c_per_h"), + "coefficients": details.get("coefficients"), } class MPCOutdoorTemperature(CoordinatorEntity, SensorEntity): diff --git a/tests/ha/test_lifecycle.py b/tests/ha/test_lifecycle.py index 4bee9cc..3262ce4 100644 --- a/tests/ha/test_lifecycle.py +++ b/tests/ha/test_lifecycle.py @@ -93,6 +93,10 @@ async def test_storage_restore_and_remove(hass, entry, inputs): forecast = hass.states.get(forecast_id) assert len(forecast.attributes["forecast_15min"]) == len(forecast.attributes["plan_15min"]) == 4 assert forecast.attributes["forecast_15min"][3] == float(forecast.state) + assert forecast.attributes["forecast"]["60_min"] == float(forecast.state) + assert set(forecast.attributes["forecast"]) == {"15_min", "30_min", "60_min"} # 60-minute horizon + assert set(forecast.attributes["contributions_c_per_h"]) == {"loss", "heat", "bias"} + assert forecast.attributes["model_in_use"] == "base" assert {"forecast_15min", "plan_15min"} <= forecast.state_info["unrecorded_attributes"] assert await hass.config_entries.async_remove(entry.entry_id) await hass.async_block_till_done() diff --git a/tests/test_coordinator.py b/tests/test_coordinator.py index e820a61..14f8084 100644 --- a/tests/test_coordinator.py +++ b/tests/test_coordinator.py @@ -754,7 +754,7 @@ class BoundaryTests(unittest.IsolatedAsyncioTestCase): alpha = 1 - exp(-300 / 900) self.states['climate.reference'].state = 'off' # learn only temp, lag, on, response = 21.0, 0.0, False, 0.0 - for k in range(2000): + for k in range(2021): # ends 25 minutes into a session (2020 % 120 == 100) outside = 5 + 5 * sin(2 * pi * k / 288) gaming = (k % 120) >= 96 # two hours of every ten power = 600.0 if gaming else 200.0 @@ -779,6 +779,11 @@ class BoundaryTests(unittest.IsolatedAsyncioTestCase): self.assertTrue(info['identified']) source = data['heat_sources']['UPS'] self.assertEqual((source['baseline_w'], source['zone'], source['settled']), (200.0, 'Office', True)) + # Mid-session the zone shows the electrical share. + details = data['zone_details']['office'] + self.assertEqual(details['model_in_use'], 'extended') + self.assertGreater(details['contributions_c_per_h']['internal'], 0.2) + self.assertTrue(details['coefficients']['extended']['internal']['identified']) async def test_unknown_power_pauses_extended_learning_only(self): self.office_with_ups() @@ -963,3 +968,57 @@ class BoundaryTests(unittest.IsolatedAsyncioTestCase): recovering = step({'bed': 19.0}, 0.0, 4)['bed'] - 19.0 # 15 min default: closed by step 3 self.assertLess(open_change, base - 0.1) # 0.01 x (5 - 19) = -0.14 °C/5 min self.assertGreater(recovering, base) # air recovers towards 20 °C + + async def coupled_rooms(self, coupling, bed=(0.006, 0.09, 0.0)): + """Living room and a colder bedroom (two exterior walls), one boiler, a shared wall.""" + self.entry.subentries = { + 'l': types.SimpleNamespace(subentry_id='l', subentry_type='zone', data={ + 'key': 'living', 'name': 'Living', 'sensors': ['sensor.room'], 'neighbors': ['bed']}), + 'b': types.SimpleNamespace(subentry_id='b', subentry_type='zone', data={ + 'key': 'bed', 'name': 'Bedroom', 'sensors': ['sensor.bedroom'], 'comfort_weight': 2})} + self.set_state('sensor.bedroom', '19', unit_of_measurement='°C') + self.c = coordinator.HeatingMPCCoordinator(self.hass, self.entry) + self.states['climate.reference'].state = 'off' + params = {'living': (0.003, 0.12, 0.0005), 'bed': bed} + advance = coordinator.advance_lag + temps, lag, on = {'living': 21.0, 'bed': 19.5}, 0.0, False + for k in range(2400): + outside = 3 + 5 * sin(2 * pi * k / 288) + for entity, value in (('sensor.room', temps['living']), ('sensor.bedroom', temps['bed']), + ('sensor.outside', outside)): + self.states[entity].state = f'{value:.5f}' + self.states[entity].last_reported = self.now + await self.sample() + average = (temps['living'] + 2 * temps['bed']) / 3 + command = average < 20.2 or (on and average < 20.8) + if command != on: + self.switch('on' if command else 'off', 180) + lag = advance(advance(lag, on, 120), command, 180) + else: + lag = advance(lag, on, 300) + on = command + new = {} + for zone, other in (('living', 'bed'), ('bed', 'living')): + a, b, c = params[zone] + new[zone] = (temps[zone] + a * (outside - temps[zone]) + b * lag + c + + coupling * (temps[other] - temps[zone])) + temps = new + return self.c + + async def test_inter_zone_transfer_learned_where_present(self): + c = await self.coupled_rooms(0.01) + self.assertEqual(self.c.zones['bed'].neighbors, ('living',)) # made symmetric + for zone, other in (('living', 'bed'), ('bed', 'living')): + with self.subTest(zone=zone): + info = c.extended[zone].coefficients()[f'couple:{other}'] + self.assertGreater(info['value'], 0.004) # clearly present (true 0.01)... + self.assertLess(info['value'], 0.013) + # ...but a steady difference between rooms is collinear with each room's + # bias, so the split stays uncertain and is not reported as identified. + self.assertFalse(info['identified']) + + async def test_inter_zone_transfer_stays_near_zero_when_absent(self): + c = await self.coupled_rooms(0.0, bed=(0.004, 0.11, 0.0)) + for zone, other in (('living', 'bed'), ('bed', 'living')): + with self.subTest(zone=zone): + self.assertLess(c.extended[zone].coefficient(f'couple:{other}'), 0.002)