buzeraandClaude Opus 5.5
2bddc74d37
Confirmed: when an open window's contact sensor became unknown or unavailable, the tracker ended the ventilation event as a close at the poll time, started recovery at full weight and forecasts assumed the window closed. (Learning was already excluded for that zone.) Each window is now OPEN, CLOSED or UNKNOWN with its last known state, since when, and since when it is unknown: - an unknown sensor keeps an open window's event (and the zone's "open" status) going; forecasts use the last known state; - a close observed only after an outage is recorded as unconfirmed: closed=None with the reported time and a maximum duration; recovery is assumed from that report but not used for any learning; - opening seen only after an outage starts an event without a temperature-before, so its recovery is not learned either; - events count minutes with an unknown sensor; Status shows each window's last known state and unknown minutes; - one unknown sensor only blocks learning for its own zone. Predictions with an unavailable wind sensor now keep an open window's natural ventilation (calm air) instead of dropping the term; learning still refuses unknown inputs. Regression tests for every OPEN/CLOSED/UNKNOWN transition, a six-hour outage, multiple windows with one unknown, and missing wind during ventilation. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
429 lines
21 KiB
Python
429 lines
21 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 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
|