buzeraandClaude Opus 5.5
8eef04c385
A flipped contact sensor reverses what every term saw as open, including the other zones' cross-ventilation term, whose fingerprint covered only the window's sensor and opening factor. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
114 lines
5.2 KiB
Python
114 lines
5.2 KiB
Python
"""Which learned coefficients stay valid when the configuration changes.
|
|
|
|
Pure Python; no Home Assistant imports.
|
|
|
|
Every learned component gets a fingerprint of exactly the configuration it was
|
|
learned against. When a fingerprint changes, only that component is recalibrated;
|
|
everything else is kept. What each fingerprint covers:
|
|
|
|
* zone measurement ("base", "loss", "heat", "bias"): the zone's sensors and their
|
|
relative weights. Scaling all weights equally changes nothing; replacing, adding
|
|
or removing a sensor, or changing the ratio between weights, does.
|
|
* "internal": the measurement plus each enabled heat source in the zone: sensor,
|
|
heat share, baseline setting, response time and what it includes or is included in.
|
|
* "couple:<zone>": the measurement plus the neighbour's measurement.
|
|
* ventilation terms: the measurement plus the zone's windows (sensor, polarity,
|
|
opening factor); "vent_wind_dir" also the windows' facing directions;
|
|
"recovery" also the recovery time constant; "vent_cross" also the other zones'
|
|
windows (sensor, polarity, opening factor).
|
|
|
|
Comfort weight, target offset and names are not physical and are never included.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from typing import Any, Iterable, Mapping
|
|
|
|
FINGERPRINT_VERSION = 1
|
|
|
|
|
|
def _hash(value: Any) -> str:
|
|
text = json.dumps(value, sort_keys=True, separators=(",", ":"), default=str)
|
|
return hashlib.sha1(text.encode()).hexdigest()[:12]
|
|
|
|
|
|
def measurement(zone: Any) -> list[tuple[str, float]]:
|
|
"""The zone temperature's definition: sensors with normalized weights."""
|
|
weights = {sensor: zone.sensor_weight(sensor) for sensor in zone.sensors}
|
|
total = sum(weights.values())
|
|
return sorted((sensor, round(weight / total, 4)) for sensor, weight in weights.items())
|
|
|
|
|
|
def source_signature(source: Any, sources: Iterable[Any]) -> list[Any]:
|
|
sources = list(sources)
|
|
parent = next((s.sensor for s in sources if s.key == source.included_in), None)
|
|
children = sorted(s.sensor for s in sources if s.included_in == source.key and s.enabled)
|
|
baseline = source.baseline_w if source.baseline_mode == "manual" else None
|
|
return [source.sensor, source.fraction, source.baseline_mode, baseline, source.response_minutes,
|
|
parent, children]
|
|
|
|
|
|
def baseline_fingerprints(sources: Iterable[Any]) -> dict[str, str]:
|
|
"""A source's baseline samples are its effective power: its sensor minus included ones."""
|
|
sources = list(sources)
|
|
return {s.key: _hash([s.sensor, sorted(c.sensor for c in sources if c.included_in == s.key and c.enabled)])
|
|
for s in sources}
|
|
|
|
|
|
def zone_fingerprints(zones: Mapping[str, Any], sources: Iterable[Any], windows: Iterable[Any],
|
|
recovery_minutes: float) -> dict[str, dict[str, str]]:
|
|
"""Per zone: a fingerprint for "base" and for every extended and ventilation feature."""
|
|
sources, windows = list(sources), list(windows)
|
|
measured = {key: _hash(measurement(zone)) for key, zone in zones.items()}
|
|
|
|
def zone_windows(key: str, with_azimuth: bool = False) -> list[Any]:
|
|
return sorted([w.sensor, w.inverted, w.opening] + ([w.azimuth] if with_azimuth else [])
|
|
for w in windows if w.zone == key and w.enabled)
|
|
|
|
result = {}
|
|
for key, zone in zones.items():
|
|
m = measured[key]
|
|
plain = zone_windows(key)
|
|
# Cross-ventilation scales with the other zones' openings.
|
|
others = sorted([w.sensor, w.inverted, w.opening] for w in windows if w.zone != key and w.enabled)
|
|
prints = {"base": m, "loss": m, "heat": m, "bias": m,
|
|
"internal": _hash([m, sorted(source_signature(s, sources) for s in sources
|
|
if s.zone == key and s.enabled)]),
|
|
"vent_natural": _hash([m, plain]), "vent_wind": _hash([m, plain]),
|
|
"vent_gust": _hash([m, plain]),
|
|
"vent_wind_dir": _hash([m, zone_windows(key, with_azimuth=True)]),
|
|
"recovery": _hash([m, plain, float(recovery_minutes)]),
|
|
"vent_cross": _hash([m, plain, others])}
|
|
for neighbor in zone.neighbors:
|
|
prints[f"couple:{neighbor}"] = _hash([m, measured.get(neighbor)])
|
|
result[key] = prints
|
|
return result
|
|
|
|
|
|
REASONS = {
|
|
"base": "zone sensors or sensor weights changed",
|
|
"internal": "heat sources changed",
|
|
"couple": "a neighbouring zone's sensors changed",
|
|
"vent_wind_dir": "window orientation changed",
|
|
"recovery": "windows or recovery time changed",
|
|
}
|
|
|
|
|
|
def reason(component: str) -> str:
|
|
if component in REASONS:
|
|
return REASONS[component]
|
|
if component.startswith("couple:"):
|
|
return REASONS["couple"]
|
|
return "windows changed"
|
|
|
|
|
|
def changed(old: Mapping[str, str] | None, new: Mapping[str, str], components: Iterable[str]) -> list[str]:
|
|
"""Components whose fingerprint differs. Without stored fingerprints (0.4.0 data)
|
|
nothing is reported: those models were learned before fingerprints existed and
|
|
are assumed to match the configuration they are loaded with."""
|
|
if old is None:
|
|
return []
|
|
# A component missing from the old fingerprints is new, not changed.
|
|
return [c for c in components if c in old and old[c] != new.get(c)]
|