"""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