buzeraandClaude Opus 5.5
ed8f9fe2e2
Opening a window adds heat exchange that the envelope model mistook for
poor insulation or a constant cooling bias.
windows.py adds windows as config subentries ("Add window"): a contact
sensor (optionally inverted; unknown/unavailable stays unknown, never
"closed"), its zone, the direction its outside faces (default 345°,
editable) and an opening factor. Each poll computes the share of the
interval each window was open (one change located by its timestamp, a
round trip is unknown), the wind exposure with proper circular angles
(meteorological "from" direction vs the window's facing) and a recovery
weight after closing that decays towards 0, pulling the air back
towards its pre-ventilation temperature. A bounded event log records
opening/closing, duration, temperatures before/lowest/at close/30 min
after, weather and other open windows.
While a zone's window is open, unknown or recovering, the base and
extended models skip it; a separate ventilation learner fits the
closed-window model's residual with natural exchange, wind speed, a
directional term bounded by the wind term (a leeward window still
ventilates), recovery, and optional gust excess and cross-ventilation
terms. Forecasts assume an open window closes after a configurable time
(default 15 min), then recovers. New poll outcome "ventilation".
Weather gains optional wind direction (degrees, 360 -> 0) and wind gust
sources with the same priorities and fallback; knots and ft/s accepted.
On a synthetic week with ventilation events every four hours and a
daily outdoor swing, the configured zone keeps loss/heat/bias close to
the truth and recovers the combined natural+wind effect within 15 %,
recovery within 15 % and direction within 30 %. The same data without
the window configured produces a -0.013 °C/5 min cooling bias and an
11 % inflated boiler effect.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
359 lines
17 KiB
Python
359 lines
17 KiB
Python
"""Window contact sensors, ventilation, wind exposure and recovery per zone.
|
|
|
|
Pure Python; no Home Assistant imports.
|
|
|
|
Opening a window adds heat exchange with outdoor air on top of the normal
|
|
envelope loss. Mixing the two would make a well-insulated room look leaky, so:
|
|
|
|
* zones with an open window, an unknown window state, or still recovering after
|
|
closing are left out of the closed-window model's learning;
|
|
* a separate ventilation model is fitted to what the closed-window model cannot
|
|
explain during those periods.
|
|
|
|
Its terms, per 5-minute step, with O = sum of open windows' opening factors
|
|
(weighted by the share of the interval they were open):
|
|
|
|
vent_natural x O (T_out - T) buoyancy, always present
|
|
vent_wind x O v (T_out - T) wind speed v (m/s)
|
|
vent_wind_dir x sum_w O_w v cos(a_w) (T_out - T) a_w: wind direction vs the
|
|
window's facing; bounded so a
|
|
leeward window still ventilates
|
|
vent_gust x O max(0, gust - v) (T_out - T) optional
|
|
vent_cross x O O_other (T_out - T) optional: windows open elsewhere
|
|
recovery x R (T_before - T) after closing, R decays 1 -> 0;
|
|
air recovers towards the walls
|
|
|
|
Wind direction is meteorological (where the wind comes FROM, clockwise from north).
|
|
A window's azimuth is the direction its outside face points to, so wind from the
|
|
azimuth blows straight at it (cos = 1) and wind from behind the building gives -1.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from collections import deque
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from math import atan2, cos, degrees, exp, hypot, isfinite, radians, sin
|
|
from typing import Any, Iterable, Mapping
|
|
|
|
SUBENTRY_WINDOW = "window"
|
|
MAX_WINDOWS = 24
|
|
DEFAULT_AZIMUTH = 345.0
|
|
OPENING_RANGE = (0.1, 3.0)
|
|
DEFAULT_RECOVERY_MINUTES = 20.0
|
|
# Recovery below this weight no longer counts as ventilation-affected (~3 time constants).
|
|
RECOVERY_THRESHOLD = 0.05
|
|
MAX_EVENTS = 20
|
|
EVENT_AFTER_MINUTES = 30
|
|
|
|
WINDOW_KEY = "key"
|
|
WINDOW_NAME = "name"
|
|
WINDOW_SENSOR = "sensor"
|
|
WINDOW_ZONE = "zone"
|
|
WINDOW_INVERTED = "inverted"
|
|
WINDOW_AZIMUTH = "azimuth"
|
|
WINDOW_OPENING = "opening_factor"
|
|
WINDOW_ENABLED = "enabled"
|
|
|
|
VENTILATION_FEATURES = ("vent_natural", "vent_wind", "vent_wind_dir", "vent_gust", "vent_cross", "recovery")
|
|
|
|
|
|
def angle_difference(a: float, b: float) -> float:
|
|
"""Signed smallest difference a - b in degrees, in [-180, 180)."""
|
|
return (a - b + 180.0) % 360.0 - 180.0
|
|
|
|
|
|
def facing_cos(wind_from: float, azimuth: float) -> float:
|
|
"""1 when the wind blows straight at a window facing `azimuth`, -1 from behind."""
|
|
return cos(radians(angle_difference(wind_from, azimuth)))
|
|
|
|
|
|
def circular_mean(angles: Iterable[float], weights: Iterable[float] | None = None) -> float | None:
|
|
"""Mean direction in [0, 360); None when the directions cancel out."""
|
|
angles = list(angles)
|
|
weights = list(weights) if weights is not None else [1.0] * len(angles)
|
|
x = sum(w * cos(radians(a)) for a, w in zip(angles, weights))
|
|
y = sum(w * sin(radians(a)) for a, w in zip(angles, weights))
|
|
if hypot(x, y) < 1e-9 * max(1.0, sum(abs(w) for w in weights)):
|
|
return None
|
|
return degrees(atan2(y, x)) % 360.0
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Window:
|
|
key: str
|
|
name: str
|
|
sensor: str
|
|
zone: str
|
|
inverted: bool = False
|
|
azimuth: float = DEFAULT_AZIMUTH
|
|
opening: float = 1.0
|
|
enabled: bool = True
|
|
subentry_id: str | None = None
|
|
|
|
|
|
def window_from_data(data: Mapping[str, Any], subentry_id: str | None = None) -> Window | None:
|
|
key = data.get(WINDOW_KEY) or subentry_id
|
|
sensor, zone = data.get(WINDOW_SENSOR), data.get(WINDOW_ZONE)
|
|
if not (isinstance(key, str) and isinstance(sensor, str) and sensor.startswith("binary_sensor.")
|
|
and isinstance(zone, str)):
|
|
return None
|
|
name = data.get(WINDOW_NAME)
|
|
try:
|
|
azimuth = float(data.get(WINDOW_AZIMUTH, DEFAULT_AZIMUTH)) % 360.0
|
|
opening = float(data.get(WINDOW_OPENING, 1.0))
|
|
except (TypeError, ValueError):
|
|
return None
|
|
if not isfinite(azimuth) or not isfinite(opening) or not OPENING_RANGE[0] <= opening <= OPENING_RANGE[1]:
|
|
return None
|
|
return Window(key=key, name=name if isinstance(name, str) and name.strip() else key, sensor=sensor,
|
|
zone=zone, inverted=data.get(WINDOW_INVERTED) is True, azimuth=azimuth, opening=opening,
|
|
enabled=data.get(WINDOW_ENABLED, True) is not False, subentry_id=subentry_id)
|
|
|
|
|
|
def build_windows(items: Iterable[tuple[Mapping[str, Any], str | None]], zone_keys: Iterable[str]) -> list[Window]:
|
|
zone_keys = set(zone_keys)
|
|
windows: list[Window] = []
|
|
for data, subentry_id in items:
|
|
window = window_from_data(data, subentry_id)
|
|
if (window is None or window.zone not in zone_keys or len(windows) == MAX_WINDOWS
|
|
or any(w.key == window.key or w.sensor == window.sensor for w in windows)):
|
|
continue
|
|
windows.append(window)
|
|
return windows
|
|
|
|
|
|
def contact_open(state: str | None, inverted: bool = False) -> bool | None:
|
|
"""Open/closed from a binary sensor state; unknown and unavailable stay unknown."""
|
|
if state not in ("on", "off"):
|
|
return None
|
|
return (state == "on") != inverted
|
|
|
|
|
|
def open_fraction(was_open: bool | None, is_open: bool | None, last_changed: float,
|
|
start: float, end: float) -> float | None:
|
|
"""Share of [start, end] a window was open; None when that is unknown.
|
|
|
|
One change inside the interval is located by its timestamp; a change to the
|
|
same state (an open/close round trip) or a missing state is unknown.
|
|
"""
|
|
if was_open is None or is_open is None or end <= start:
|
|
return None
|
|
if last_changed <= start:
|
|
return float(is_open) if was_open == is_open else None
|
|
if was_open != is_open and last_changed <= end:
|
|
share = (end - last_changed) / (end - start)
|
|
return share if is_open else 1.0 - share
|
|
return None
|
|
|
|
|
|
def ventilation_vector(features: Iterable[str], temp: float, outside: float,
|
|
inputs: Mapping[str, Any]) -> list[float] | None:
|
|
"""Ventilation feature values for one zone and step; None if an input is unknown.
|
|
|
|
`inputs`: opening, facing (sum of opening x cos), other_opening, recovery,
|
|
t_ref (pre-ventilation temperature), wind, gust.
|
|
"""
|
|
difference = outside - temp
|
|
values = []
|
|
for name in features:
|
|
opening = inputs.get("opening")
|
|
if opening is None:
|
|
return None
|
|
if name == "vent_natural":
|
|
values.append(opening * difference)
|
|
elif name in ("vent_wind", "vent_wind_dir", "vent_gust"):
|
|
wind = inputs.get("wind")
|
|
if wind is None:
|
|
return None
|
|
if name == "vent_wind":
|
|
values.append(opening * wind * difference)
|
|
elif name == "vent_wind_dir":
|
|
facing = inputs.get("facing")
|
|
if facing is None:
|
|
return None
|
|
values.append(facing * wind * difference)
|
|
else:
|
|
gust = inputs.get("gust")
|
|
if gust is None:
|
|
return None
|
|
values.append(opening * max(0.0, gust - wind) * difference)
|
|
elif name == "vent_cross":
|
|
values.append(opening * (inputs.get("other_opening") or 0.0) * difference)
|
|
elif name == "recovery":
|
|
recovery = inputs.get("recovery") or 0.0
|
|
t_ref = inputs.get("t_ref")
|
|
if recovery > 0 and t_ref is None:
|
|
return None
|
|
values.append(recovery * ((t_ref if t_ref is not None else temp) - temp))
|
|
else:
|
|
raise KeyError(name)
|
|
return values
|
|
|
|
|
|
def _iso(timestamp: float) -> str:
|
|
return datetime.fromtimestamp(timestamp, timezone.utc).isoformat(timespec="seconds")
|
|
|
|
|
|
class VentilationTracker:
|
|
"""Window states, ventilation inputs per zone, recovery after closing, event log."""
|
|
|
|
def __init__(self, windows: list[Window], recovery_minutes: float = DEFAULT_RECOVERY_MINUTES,
|
|
saved: Any = None) -> None:
|
|
self.windows = windows
|
|
self.tau = max(1.0, recovery_minutes) * 60.0
|
|
self._previous: dict[str, tuple[float, bool | None]] = {} # window -> (poll time, open)
|
|
self._closed_at: dict[str, float] = {} # zone -> when its last window closed
|
|
self._t_ref: dict[str, float | None] = {} # zone -> temperature before it opened
|
|
self._last_temps: dict[str, float | None] = {}
|
|
self._open_event: dict[str, dict[str, Any]] = {}
|
|
saved = saved if isinstance(saved, Mapping) else {}
|
|
events = saved.get("events") if isinstance(saved.get("events"), list) else []
|
|
self.events: deque[dict[str, Any]] = deque(
|
|
(e for e in events if isinstance(e, dict)), maxlen=MAX_EVENTS)
|
|
self.window_state: dict[str, dict[str, Any]] = {}
|
|
self.zone_state: dict[str, dict[str, Any]] = {}
|
|
|
|
def update(self, now: float, states: Mapping[str, tuple[str | None, float | None]],
|
|
temps: Mapping[str, float | None], weather: Mapping[str, float | None]) -> None:
|
|
"""One poll. `states`: window key -> (binary sensor state, last_changed timestamp)."""
|
|
self.window_state = {}
|
|
zones: dict[str, dict[str, Any]] = {}
|
|
for window in self.windows:
|
|
if not window.enabled:
|
|
continue
|
|
state, changed = states.get(window.key, (None, None))
|
|
is_open = contact_open(state, window.inverted)
|
|
previous = self._previous.get(window.key)
|
|
fraction = None
|
|
if previous and changed is not None and 0 < now - previous[0] <= 900:
|
|
fraction = open_fraction(previous[1], is_open, changed, previous[0], now)
|
|
self._previous[window.key] = (now, is_open)
|
|
zone = zones.setdefault(window.zone, {"opening": 0.0, "facing_unit": 0.0, "known": True,
|
|
"open_now": False, "open_windows": [], "last_close": None,
|
|
"first_open": None})
|
|
if fraction is None:
|
|
zone["known"] = False
|
|
else:
|
|
zone["opening"] += window.opening * fraction
|
|
zone["facing_unit"] += window.opening * fraction * (
|
|
facing_cos(weather["wind_direction"], window.azimuth)
|
|
if weather.get("wind_direction") is not None else 0.0)
|
|
if is_open:
|
|
zone["open_now"] = True
|
|
zone["open_windows"].append(window.name)
|
|
if changed is not None:
|
|
zone["first_open"] = min(zone["first_open"] or changed, changed)
|
|
elif is_open is False and previous and changed is not None and changed > previous[0]:
|
|
zone["last_close"] = max(zone["last_close"] or 0.0, changed)
|
|
self.window_state[window.key] = {
|
|
"open": is_open, "since": changed,
|
|
"relative_wind_direction": (round(angle_difference(weather["wind_direction"], window.azimuth), 1)
|
|
if weather.get("wind_direction") is not None else None)}
|
|
for key, zone in zones.items():
|
|
self._update_zone(now, key, zone, temps.get(key), weather, zones)
|
|
self.zone_state = zones
|
|
self._last_temps = dict(temps)
|
|
|
|
def _update_zone(self, now: float, key: str, zone: dict[str, Any], temp: float | None,
|
|
weather: Mapping[str, float | None], zones: Mapping[str, dict[str, Any]]) -> None:
|
|
was_open = key in self._open_event
|
|
if zone["open_now"] and not was_open:
|
|
# Temperature at the previous poll approximates the room before ventilation.
|
|
self._t_ref[key] = self._last_temps.get(key)
|
|
self._closed_at.pop(key, None)
|
|
self._open_event[key] = {
|
|
"zone": key, "opened": _iso(min(zone["first_open"] or now, now)),
|
|
"temperature_before": self._t_ref[key],
|
|
"outside": weather.get("temperature"), "wind": weather.get("wind"),
|
|
"wind_direction": weather.get("wind_direction"), "wind_gust": weather.get("wind_gust"),
|
|
"windows": list(zone["open_windows"]),
|
|
"other_open_windows": [n for k, z in zones.items() if k != key for n in z["open_windows"]],
|
|
"lowest_temperature": temp,
|
|
}
|
|
elif zone["open_now"] and temp is not None:
|
|
event = self._open_event[key]
|
|
event["lowest_temperature"] = min(t for t in (event["lowest_temperature"], temp) if t is not None)
|
|
elif was_open and not zone["open_now"]:
|
|
event = self._open_event.pop(key)
|
|
closed = zone["last_close"] or now
|
|
self._closed_at[key] = closed
|
|
event.update(closed=_iso(closed), duration_minutes=round((closed - _parse(event["opened"])) / 60, 1),
|
|
temperature_at_close=temp)
|
|
self.events.append(event)
|
|
elif not zone["open_now"] and key in self._closed_at and temp is not None:
|
|
# Fill in the temperature half an hour after closing, for recovery review.
|
|
for event in reversed(self.events):
|
|
if event.get("zone") == key and "temperature_after" not in event:
|
|
if now - self._closed_at[key] >= EVENT_AFTER_MINUTES * 60:
|
|
event["temperature_after"] = temp
|
|
event["temperature_after_minutes"] = EVENT_AFTER_MINUTES
|
|
break
|
|
closed = self._closed_at.get(key)
|
|
recovery = exp(-(now - closed) / self.tau) if closed is not None and not zone["open_now"] else 0.0
|
|
if recovery < RECOVERY_THRESHOLD:
|
|
recovery = 0.0
|
|
zone.update(recovery=recovery, t_ref=self._t_ref.get(key),
|
|
other_opening=sum(z["opening"] for k, z in zones.items() if k != key and z["known"]))
|
|
# Learning of the closed-window model is unsafe while anything ventilation-related goes on.
|
|
zone["affected"] = (not zone["known"]) or zone["opening"] > 0 or zone["open_now"] or recovery > 0
|
|
|
|
def zone_inputs(self, key: str, weather: Mapping[str, float | None]) -> dict[str, Any] | None:
|
|
"""Ventilation inputs of the interval that just ended; None for zones without windows."""
|
|
zone = self.zone_state.get(key)
|
|
if zone is None:
|
|
return None
|
|
return {"opening": zone["opening"] if zone["known"] else None, "facing": zone["facing_unit"],
|
|
"other_opening": zone["other_opening"], "recovery": zone["recovery"],
|
|
"t_ref": zone["t_ref"], "wind": weather.get("wind"), "gust": weather.get("wind_gust"),
|
|
"affected": zone["affected"], "known": zone["known"]}
|
|
|
|
def forecast(self, key: str, steps: int, open_minutes: float, weather: Mapping[str, float | None],
|
|
now: float) -> list[dict[str, Any]]:
|
|
"""Assumed ventilation inputs per forecast step.
|
|
|
|
Windows open now are assumed to close after `open_minutes` (at least one
|
|
step); after that the zone recovers. Weather is held at current values.
|
|
"""
|
|
zone = self.zone_state.get(key)
|
|
if zone is None:
|
|
return [{} for _ in range(steps)]
|
|
windows = [w for w in self.windows if w.zone == key and w.enabled
|
|
and self.window_state.get(w.key, {}).get("open")]
|
|
opening = sum(w.opening for w in windows)
|
|
facing = sum(w.opening * facing_cos(weather["wind_direction"], w.azimuth) for w in windows) \
|
|
if weather.get("wind_direction") is not None else 0.0
|
|
open_steps = max(1, round(open_minutes / 5)) if windows else 0
|
|
closed_at = self._closed_at.get(key)
|
|
result = []
|
|
for k in range(steps):
|
|
t = now + (k + 1) * 300.0
|
|
if k < open_steps:
|
|
result.append({"opening": opening, "facing": facing, "recovery": 0.0, "t_ref": zone["t_ref"],
|
|
"other_opening": zone["other_opening"], "wind": weather.get("wind") or 0.0,
|
|
"gust": weather.get("wind_gust") or weather.get("wind") or 0.0})
|
|
continue
|
|
start = now + open_steps * 300.0 if windows else closed_at
|
|
recovery = exp(-(t - start) / self.tau) if start is not None else 0.0
|
|
result.append({"opening": 0.0, "facing": 0.0, "other_opening": 0.0, "wind": 0.0, "gust": 0.0,
|
|
"recovery": recovery if recovery >= RECOVERY_THRESHOLD else 0.0,
|
|
"t_ref": zone["t_ref"]})
|
|
return result
|
|
|
|
def as_dict(self) -> dict[str, Any]:
|
|
return {"events": list(self.events)}
|
|
|
|
|
|
def _parse(iso: str) -> float:
|
|
return datetime.fromisoformat(iso).timestamp()
|
|
|
|
|
|
def validate_window(data: Mapping[str, Any], others: Iterable[Window]) -> str | None:
|
|
others = list(others)
|
|
if not str(data.get(WINDOW_NAME, "")).strip():
|
|
return "window_name_required"
|
|
if any(w.sensor == data.get(WINDOW_SENSOR) for w in others):
|
|
return "sensor_in_other_window"
|
|
if len(others) >= MAX_WINDOWS:
|
|
return "too_many_windows"
|
|
return None
|