"""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 ventilation_vector_for_prediction(features: Iterable[str], temp: float, outside: float, inputs: Mapping[str, Any]) -> list[float] | None: """Like ventilation_vector, but an unknown wind or gust counts as calm air. For predictions only: an open window keeps its natural ventilation when the wind sensor is unavailable, instead of silently being treated as closed. Learning always uses ventilation_vector, which refuses unknown inputs. """ features = list(features) x = ventilation_vector(features, temp, outside, inputs) if x is None and inputs.get("opening") is not None: x = ventilation_vector(features, temp, outside, { **inputs, "wind": inputs.get("wind") or 0.0, "gust": inputs.get("gust") or inputs.get("wind") or 0.0, "facing": inputs.get("facing") or 0.0}) return x 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]] = {} # window -> last known open/closed state, since when, and since when it is unknown. self._memory: dict[str, dict[str, Any]] = {} self._recovery_confirmed: dict[str, bool] = {} 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). Each window is OPEN, CLOSED or UNKNOWN (unknown/unavailable/missing). Unknown never means closed: the last known state is kept, an open window's event stays open, and a close that happened while the sensor was unknown is recorded as unconfirmed. """ 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) memory = self._memory.setdefault(window.key, {"last_known": None, "since": None, "unknown_since": None}) zone = zones.setdefault(window.zone, { "opening": 0.0, "facing_unit": 0.0, "known": True, "open_now": False, "possibly_open": False, "open_windows": [], "last_close": None, "close_confirmed": True, "first_open": None, "open_confirmed": True, "unknown_windows": []}) 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 is None: if memory["unknown_since"] is None: memory["unknown_since"] = previous[0] if previous and previous[1] is not None else now zone["unknown_windows"].append(window.name) if memory["last_known"]: zone["possibly_open"] = True # last seen open: still treated as open else: after_outage = memory["unknown_since"] is not None if memory["last_known"] is not is_open: memory["since"] = changed if is_open and after_outage: zone["open_confirmed"] = False # opened at some point during the outage if is_open is False and memory["last_known"] and after_outage: zone["close_confirmed"] = False # closed at some point during the outage memory["last_known"], memory["unknown_since"] = is_open, None 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 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, "last_known_open": memory["last_known"], "last_known_since": memory["since"], "unknown_since": memory["unknown_since"], "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 ventilating = zone["open_now"] or zone["possibly_open"] 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)), "opened_confirmed": zone["open_confirmed"], "temperature_before": self._t_ref[key] if zone["open_confirmed"] else None, "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, "sensor_unknown_minutes": 0.0, } if not zone["open_confirmed"]: self._t_ref[key] = None # the room may already have cooled before we saw it elif ventilating and was_open: event = self._open_event[key] if temp is not None: event["lowest_temperature"] = min(t for t in (event["lowest_temperature"], temp) if t is not None) if zone["unknown_windows"]: event["sensor_unknown_minutes"] = round(event["sensor_unknown_minutes"] + 5.0, 1) elif was_open and not ventilating: event = self._open_event.pop(key) if zone["close_confirmed"]: closed = zone["last_close"] or now event.update(closed=_iso(closed), closed_confirmed=True, duration_minutes=round((closed - _parse(event["opened"])) / 60, 1)) else: # Closed while the sensor was unknown: no confirmed time, only a range. closed = now event.update(closed=None, closed_confirmed=False, closed_reported=_iso(now), duration_minutes=None, duration_max_minutes=round((now - _parse(event["opened"])) / 60, 1)) self._closed_at[key] = closed self._recovery_confirmed[key] = zone["close_confirmed"] and event.get("opened_confirmed", True) event["temperature_at_close"] = temp self.events.append(event) elif not ventilating 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 ventilating else 0.0 if recovery < RECOVERY_THRESHOLD: recovery = 0.0 zone.update(recovery=recovery, t_ref=self._t_ref.get(key), recovery_confirmed=self._recovery_confirmed.get(key, True), 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 ventilating 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. `known` is False while a window state is unknown, and during a recovery whose start is not confirmed: those intervals are not used for any learning. """ zone = self.zone_state.get(key) if zone is None: return None known = zone["known"] and (zone["recovery"] == 0 or zone["recovery_confirmed"]) return {"opening": zone["opening"] if 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": known, "unknown_windows": list(zone["unknown_windows"])} 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 (or unknown but last seen open) 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") or (self.window_state.get(w.key, {}).get("open") is None and self.window_state.get(w.key, {}).get("last_known_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