buzeraandClaude Opus 5.5
3076beebfb
A single stale, unavailable, missing or invalid room sensor paused learning and forecasts for every room. That is not acceptable once the MPC may control heating. Now such a room is left out: the weighted temperature, learning and the optimizer use the remaining rooms with their weights re-normalized, and a room learns only across poll pairs where it was usable at both ends. Readiness considers only the rooms used. Everything pauses only when no room is usable or a shared input (relay, thermostat, outdoor temperature) is missing. A new setup/Configure checkbox, room_stale_use_last_value (default off), keeps using a stale room sensor's last value instead, for sensors that report only on change. It never applies to unavailable or invalid sensors, or to a sensor silent for more than 24 hours. Status adds excluded_rooms, weight_coverage and a per-room "used" flag. English and Romanian labels and help text added. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
381 lines
20 KiB
Python
381 lines
20 KiB
Python
"""Read existing HA entities, learn thermal response, produce shadow MPC decisions."""
|
|
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 Event, HomeAssistant, callback
|
|
from homeassistant.config_entries import ConfigEntry
|
|
from homeassistant.helpers.storage import Store
|
|
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
|
|
|
|
from .const import (
|
|
CONF_ROOM_MAX_AGE, CONF_ROOM_STALE_KEEP, CONF_WEATHER_MAX_AGE, ROOM_STALE_KEEP_MAX_MINUTES,
|
|
DOMAIN, CONF_CLIMATE, CONF_HEATER, CONF_ROOMS, CONF_WEIGHTS, WEATHER_SOURCES,
|
|
CONF_HORIZON, CONF_MIN_ON, CONF_MIN_OFF, CONF_SWITCH_COST, CONF_HEAT_COST,
|
|
CONF_BAND, DEFAULTS,
|
|
)
|
|
from .data_quality import ReportAgeHistory, freshness
|
|
from .physics import (
|
|
LAG_LEARN_MAX_UNCERTAINTY, LAG_TAU_SECONDS, Learner, advance_lag,
|
|
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
|
|
# Models of rooms removed in Options, kept in case a room is added back.
|
|
MAX_RETIRED_ROOMS = 24
|
|
|
|
|
|
class HeatingMPCCoordinator(DataUpdateCoordinator[dict[str, Any]]):
|
|
"""The existing generic thermostat stays the only actuator owner."""
|
|
|
|
def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
|
|
super().__init__(hass, _LOGGER, config_entry=entry, name=DOMAIN,
|
|
update_interval=timedelta(minutes=5))
|
|
self.entry = entry
|
|
self.config = {**DEFAULTS, **entry.data, **entry.options}
|
|
self.rooms: list[str] = self.config[CONF_ROOMS]
|
|
self.learners: dict[str, Learner] = {room: Learner() for room in self.rooms}
|
|
self._retired: dict[str, dict[str, Any]] = {}
|
|
self._report_ages = ReportAgeHistory()
|
|
self._store = Store(hass, 1, f"{DOMAIN}.{entry.entry_id}")
|
|
self._previous: dict[str, Any] | None = None
|
|
# 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
|
|
|
|
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")
|
|
return
|
|
# Replacing the actual heating system invalidates heating coefficients.
|
|
# Switching among weather providers does not discard months of learning.
|
|
if stored.get("heater_entity", self.config[CONF_HEATER]) != self.config[CONF_HEATER]:
|
|
_LOGGER.warning("Heating MPC heater changed; starting a fresh thermal model")
|
|
return
|
|
# Models carry room identity, so changing the room list is safe. Models of
|
|
# rooms no longer configured are retained, most recently removed last.
|
|
for key in ("retired_learners", "learners"): # current learners take precedence
|
|
saved = stored.get(key)
|
|
if not isinstance(saved, dict):
|
|
continue
|
|
for room, data in saved.items():
|
|
if not isinstance(room, str):
|
|
continue
|
|
learner = Learner.from_dict(data)
|
|
if room in self.learners:
|
|
self.learners[room] = learner
|
|
elif learner.count:
|
|
self._retired.pop(room, None)
|
|
self._retired[room] = learner.as_dict()
|
|
while len(self._retired) > MAX_RETIRED_ROOMS:
|
|
self._retired.pop(next(iter(self._retired)))
|
|
self._update_counts()
|
|
|
|
def _update_counts(self) -> None:
|
|
self._on_samples = min(m.on_count for m in self.learners.values())
|
|
self._off_samples = min(m.off_count for m in self.learners.values())
|
|
|
|
def _sensor_quality(self, entity_id: str, now: datetime, limit: float) -> dict:
|
|
state = self.hass.states.get(entity_id)
|
|
if state is None:
|
|
return {"status": "missing", "age_seconds": None}
|
|
if state.state in ("unknown", "unavailable"):
|
|
return {"status": state.state, "age_seconds": None}
|
|
# last_updated is a conservative compatibility fallback for older HA.
|
|
quality = freshness(getattr(state, "last_reported", getattr(state, "last_updated", None)),
|
|
now, limit * 60)
|
|
if quality["status"] in ("valid", "stale"):
|
|
quality["max_age_seconds_24h"] = round(
|
|
self._report_ages.observe(entity_id, now, quality["age_seconds"]), 1)
|
|
return quality
|
|
|
|
@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.
|
|
|
|
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 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(
|
|
-(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:
|
|
"""Values the prediction depends on, used to detect changes during calculation."""
|
|
heater = self.config[CONF_HEATER]
|
|
entities = sorted(set(self.rooms) | set(sources_for(self.config, "temperature")) |
|
|
{self.config[CONF_CLIMATE], heater})
|
|
values = []
|
|
for entity_id in entities:
|
|
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)
|
|
|
|
async def _async_update_data(self) -> dict[str, Any]:
|
|
now = datetime.now(timezone.utc)
|
|
heater = self.hass.states.get(self.config[CONF_HEATER])
|
|
climate = self.hass.states.get(self.config[CONF_CLIMATE])
|
|
# Choose each weather variable independently from ordered sensor sources.
|
|
# Sensor priorities are editable in Options without touching integration code.
|
|
weather_entity_ids = {
|
|
entity_id for metric in WEATHER_SOURCES
|
|
for entity_id in sources_for(self.config, metric)
|
|
}
|
|
snapshot = {entity: self.hass.states.get(entity)
|
|
for entity in set(self.rooms) | weather_entity_ids}
|
|
inputs_before = self._prediction_inputs()
|
|
room_quality = {}
|
|
temps = {}
|
|
keep_stale = self.config.get(CONF_ROOM_STALE_KEEP) is True
|
|
for room in self.rooms:
|
|
quality = self._sensor_quality(room, now, self.config[CONF_ROOM_MAX_AGE])
|
|
value = None
|
|
# Optionally trust sensors that report only on change, but never a
|
|
# sensor silent for a day: it is most likely dead.
|
|
if quality["status"] == "valid" or (
|
|
keep_stale and quality["status"] == "stale"
|
|
and quality["age_seconds"] <= ROOM_STALE_KEEP_MAX_MINUTES * 60):
|
|
state = snapshot[room]
|
|
value = normalize("temperature", state.state, state.attributes.get("unit_of_measurement"))
|
|
if value is None or not 0 <= value <= 45:
|
|
quality["status"] = "invalid_value_or_unit"
|
|
value = None
|
|
quality["used"] = value is not None
|
|
room_quality[room] = quality
|
|
temps[room] = value
|
|
weather_states = {}
|
|
weather_quality = {}
|
|
for entity_id in weather_entity_ids:
|
|
quality = self._sensor_quality(entity_id, now, self.config[CONF_WEATHER_MAX_AGE])
|
|
weather_quality[entity_id] = quality
|
|
if quality["status"] == "valid":
|
|
state = snapshot[entity_id]
|
|
weather_states[entity_id] = (state.state, state.attributes.get("unit_of_measurement"))
|
|
weather, active_sources, weather_comparisons = choose_weather_sources(
|
|
self.config, weather_states
|
|
)
|
|
metric_quality = {}
|
|
for metric in WEATHER_SOURCES:
|
|
metric_quality[metric] = {}
|
|
for entity_id in sources_for(self.config, metric):
|
|
quality = dict(weather_quality[entity_id])
|
|
if quality["status"] == "valid" and entity_id not in weather_comparisons[metric]:
|
|
quality["status"] = "invalid_value_or_unit"
|
|
metric_quality[metric][entity_id] = quality
|
|
outside = weather["temperature"]
|
|
status: dict[str, Any] = {
|
|
"input_quality": {"rooms": room_quality, "weather": metric_quality},
|
|
"status": "waiting_for_sensors", "proposal": None, "prediction_60m": {},
|
|
"temperatures": {r: t for r, t in temps.items() if t is not None},
|
|
"on_samples": self._on_samples, "off_samples": self._off_samples,
|
|
"samples": min(learner.count for learner in self.learners.values()),
|
|
"mode": climate.state if climate else "unavailable",
|
|
"weather": {k: v for k, v in weather.items() if v is not None},
|
|
"weather_active_sources": {k: v for k, v in active_sources.items() if v},
|
|
"weather_comparison": {k: v for k, v in weather_comparisons.items() if v},
|
|
"weather_model_inputs": ["temperature"],
|
|
"retired_room_models": sorted(self._retired),
|
|
"excluded_rooms": [room for room in self.rooms if temps[room] is None],
|
|
}
|
|
heater_valid = heater is not None and heater.state in ("on", "off")
|
|
heater_on = heater_valid and heater.state == "on"
|
|
steady = False
|
|
if heater_valid:
|
|
steady = self._track_lag(now, heater_on, heater.last_changed)
|
|
status["heat_lag"] = round(self._lag["lag"], 3)
|
|
status["heat_lag_uncertainty"] = round(self._lag["uncertainty"], 3)
|
|
else:
|
|
self._lag = None
|
|
self._relay_events.clear()
|
|
self._relay_events_dropped = False
|
|
# Rooms without a usable reading are left out and the others carry on;
|
|
# only the shared inputs, or losing every room, pause the integration.
|
|
vals: dict[str, float] = {room: float(t) for room, t in temps.items() if t is not None}
|
|
if not heater_valid or climate is None or outside is None or not vals:
|
|
self._previous = None
|
|
return status
|
|
|
|
# A room learns only across a poll pair where it was usable at both ends.
|
|
pairs = [room for room in vals if self._previous and room in self._previous["temps"]]
|
|
if not pairs:
|
|
status["sample_result"] = "first_sample"
|
|
else:
|
|
dt = (now - self._previous["time"]).total_seconds()
|
|
# Do not fit across a relay change, a provider/fallback switch or an
|
|
# uncertain lag estimate (restart, relay outage, several changes).
|
|
if not 270 <= dt <= 330:
|
|
status["sample_result"] = "interval_out_of_range"
|
|
elif not steady or heater_on != self._previous["heater_on"]:
|
|
status["sample_result"] = "relay_changed"
|
|
elif self._previous["outside_source"] != active_sources["temperature"]:
|
|
status["sample_result"] = "outside_source_changed"
|
|
elif self._lag["uncertainty"] > LAG_LEARN_MAX_UNCERTAINTY:
|
|
status["sample_result"] = "lag_uncertain"
|
|
else:
|
|
accepted = [
|
|
self.learners[room].learn(
|
|
self._previous["temps"][room], vals[room],
|
|
self._previous["outside"], self._lag["lag"], dt, heater_on)
|
|
for room in pairs
|
|
]
|
|
status["sample_result"] = "learned" if all(accepted) else "partly_rejected"
|
|
self._update_counts()
|
|
# Persist at each sample so restart does not erase hours of training.
|
|
await self._store.async_save({
|
|
"schema_version": 2,
|
|
"learners": {room: learner.as_dict() for room, learner in self.learners.items()},
|
|
# Optional; older 0.2.x readers ignore it and keep working.
|
|
"retired_learners": self._retired,
|
|
"on_samples": self._on_samples, "off_samples": self._off_samples,
|
|
"heater_entity": self.config[CONF_HEATER],
|
|
"outside_source": active_sources["temperature"],
|
|
})
|
|
|
|
self._previous = {
|
|
"time": now, "temps": vals, "outside": outside, "heater_on": heater_on,
|
|
"outside_source": active_sources["temperature"],
|
|
}
|
|
weights = self.config.get(CONF_WEIGHTS, {})
|
|
try:
|
|
weights = {r: float(weights.get(r, 1.0)) for r in self.rooms}
|
|
if any(not isfinite(w) or w <= 0 for w in weights.values()) or not isfinite(sum(weights.values())):
|
|
raise ValueError
|
|
except (AttributeError, TypeError, ValueError, OverflowError):
|
|
status["status"] = "invalid_weights"
|
|
return status
|
|
# Left-out rooms take their weight with them; the rest are re-normalized.
|
|
used_weights = {room: weights[room] for room in vals}
|
|
weighted_temp = sum(used_weights[r] * vals[r] for r in vals) / sum(used_weights.values())
|
|
status.update({
|
|
"weighted_temperature": round(weighted_temp, 3),
|
|
"weight_coverage": round(sum(used_weights.values()) / sum(weights.values()), 3),
|
|
"outside_temperature": outside,
|
|
"reference_heat": heater_on,
|
|
"on_samples": self._on_samples,
|
|
"off_samples": self._off_samples,
|
|
"samples": min(learner.count for learner in self.learners.values()),
|
|
"models": {
|
|
room: {"loss": round(model.theta[0], 6),
|
|
"heat": round(model.theta[1], 6),
|
|
"bias": round(model.theta[2], 6),
|
|
"samples": model.count,
|
|
"on_samples": model.on_count, "off_samples": model.off_count,
|
|
"mae_5m": round(model.mae, 4) if model.mae is not None else None}
|
|
for room, model in self.learners.items()
|
|
},
|
|
})
|
|
if climate.state != "heat":
|
|
status["status"] = "paused"
|
|
return status
|
|
try:
|
|
target = normalize("temperature", climate.attributes["temperature"],
|
|
climate.attributes.get("temperature_unit",
|
|
self.hass.config.units.temperature_unit))
|
|
if target is None or not 5 <= target <= 35:
|
|
raise ValueError
|
|
except (TypeError, KeyError, ValueError):
|
|
status["status"] = "waiting_for_target"
|
|
return status
|
|
status["target"] = target
|
|
# Do not act on a cold model: every room used needs ~17 hours of samples
|
|
# plus both relay conditions.
|
|
used_models = [self.learners[room] for room in vals]
|
|
if (min(m.count for m in used_models) < 200 or min(m.on_count for m in used_models) < 30
|
|
or min(m.off_count for m in used_models) < 30):
|
|
status["status"] = "learning"
|
|
return status
|
|
try:
|
|
# 180-minute exhaustive search can take several seconds on RPi5;
|
|
# never block Home Assistant's event loop with CPU work.
|
|
job = partial(
|
|
optimize,
|
|
vals, outside, self._lag["lag"], target, used_weights,
|
|
{room: self.learners[room] for room in vals},
|
|
heater_on, (now - heater.last_changed).total_seconds(),
|
|
horizon_minutes=int(self.config[CONF_HORIZON]),
|
|
minimum_on_minutes=int(self.config[CONF_MIN_ON]),
|
|
minimum_off_minutes=int(self.config[CONF_MIN_OFF]),
|
|
switching_cost=float(self.config[CONF_SWITCH_COST]),
|
|
heating_cost=float(self.config[CONF_HEAT_COST]),
|
|
comfort_band=float(self.config[CONF_BAND]),
|
|
)
|
|
# 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:
|
|
status["status"] = "inputs_changed"
|
|
return status
|
|
except (ValueError, RuntimeError, OverflowError) as err:
|
|
_LOGGER.warning("MPC prediction failed: %s", err)
|
|
status["status"] = "prediction_error"
|
|
return status
|
|
status.update({
|
|
"status": "ready", "proposal": output["proposal"],
|
|
"prediction_60m": output["predicted_60m"],
|
|
"prediction_end": output["predicted_end"],
|
|
"plan_15min": output["plan_15min"], "objective": output["cost"],
|
|
"model_quality_note": "5-minute in-sample MAE; not a validated 60-minute forecast error",
|
|
})
|
|
return status
|