buzeraandClaude Opus 5.5
2cfba024ff
Reproduced: a window unknown for a whole 15-minute span was replayed as closed and scored as a Model Error sample (MAE 0.76 °C on one sample). - A zone's replay is complete only if every step's model inputs were observed: window openings (and a confirmed recovery), the wind, direction and other zones' windows that an open window's terms use, and, for the extended layer, electrical heat and neighbours, whose own replay must be complete too. - Incomplete spans are counted, not scored; the base model's error on them is kept separately (it needs only the relay and outdoor temperature). Spans that could not be replayed (uncertain radiator lag, no reading at the end) are counted as skipped. Model Error sensors show incomplete, skipped and incomplete_base_model_mae; stored data gains an "excluded" section and older data loads unchanged. - Window inputs no longer stand in for unknown values: the facing term is unknown while a window is open and the wind direction is not, and the cross-ventilation input is unknown while another zone's window is. With every window closed, opening terms are 0 and need no weather input. Learning refuses such samples; forecasts still assume calm air. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
570 lines
30 KiB
Python
570 lines
30 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; None when the wind direction is
|
|
unknown), other_opening (None when another zone's windows are unknown),
|
|
recovery, t_ref (pre-ventilation temperature), wind, gust. With every window
|
|
closed the opening terms are 0 whatever the weather, so only the opening and,
|
|
while recovering, the reference temperature must be known.
|
|
"""
|
|
difference = outside - temp
|
|
opening = inputs.get("opening")
|
|
if opening is None:
|
|
return None
|
|
values = []
|
|
for name in features:
|
|
if name == "vent_natural":
|
|
values.append(opening * difference)
|
|
elif name in ("vent_wind", "vent_wind_dir", "vent_gust", "vent_cross") and opening == 0:
|
|
values.append(0.0)
|
|
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":
|
|
other = inputs.get("other_opening")
|
|
if other is None:
|
|
return None
|
|
values.append(opening * other * 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, "other_opening": inputs.get("other_opening") or 0.0})
|
|
return x
|
|
|
|
|
|
def ventilation_known(features: Iterable[str], inputs: Mapping[str, Any] | None) -> bool:
|
|
"""True when a zone's ventilation term follows from observed inputs alone.
|
|
|
|
False while a window state is unknown or a recovery's start is unconfirmed,
|
|
or while a window is open and a term it activates lacks its weather input.
|
|
Zones without windows (`inputs` None or empty) need nothing.
|
|
"""
|
|
if not inputs:
|
|
return True
|
|
return ventilation_vector(list(features), 0.0, 0.0, inputs) is not None
|
|
|
|
|
|
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.
|
|
|
|
Each window is OPEN, CLOSED or UNKNOWN (unknown, unavailable or missing).
|
|
Unknown never means closed. The last known state is kept apart from the
|
|
current one, and an opening or closing counts as confirmed only when it was
|
|
seen: the previous reading was the other state, with no unknown period in
|
|
between. Last known states, open events and recovery are saved so a restart
|
|
or reload continues them; they come back as unknown until the sensor reports.
|
|
"""
|
|
|
|
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, when it was last seen
|
|
# and since when it is unknown (None while known).
|
|
self._memory: dict[str, dict[str, Any]] = {}
|
|
self._recovery_confirmed: dict[str, bool] = {}
|
|
self._last_update: float | None = None
|
|
self._restored_at: float | None = None
|
|
self.changed = False # something worth saving happened since the last save
|
|
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]] = {}
|
|
self._restore(saved.get("state"))
|
|
|
|
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)
|
|
memory = self._memory.setdefault(window.key, {"last_known": None, "since": None, "seen": 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": [], "facing_known": True})
|
|
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 fraction > 0 and weather.get("wind_direction") is None:
|
|
zone["facing_known"] = False
|
|
if is_open is None:
|
|
if memory["unknown_since"] is None:
|
|
memory["unknown_since"] = memory["seen"] if memory["seen"] is not None else now
|
|
self.changed = True
|
|
zone["unknown_windows"].append(window.name)
|
|
if memory["last_known"]:
|
|
zone["possibly_open"] = True # last seen open: still treated as open
|
|
else:
|
|
if memory["last_known"] is not is_open: # a change, or the first reading
|
|
memory["since"] = changed
|
|
# Seen only if the previous reading was the other state, with no gap.
|
|
witnessed = memory["last_known"] is not None and memory["unknown_since"] is None
|
|
if not witnessed:
|
|
zone["open_confirmed" if is_open else "close_confirmed"] = False
|
|
self.changed = True
|
|
elif memory["unknown_since"] is not None:
|
|
self.changed = True # the same state again after a gap
|
|
memory.update(last_known=is_open, seen=now, unknown_since=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"], "seen": memory["seen"],
|
|
"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)}
|
|
# Time the integration was not running counts as unknown for events it interrupted.
|
|
gap = now - self._restored_at if self._restored_at is not None else 0.0
|
|
self._restored_at = None
|
|
for key, zone in zones.items():
|
|
if gap > 330 and key in self._open_event:
|
|
event = self._open_event[key]
|
|
event["sensor_unknown_minutes"] = round(event.get("sensor_unknown_minutes", 0.0) + gap / 60, 1)
|
|
self._update_zone(now, key, zone, temps.get(key), weather, zones)
|
|
self.zone_state = zones
|
|
self._last_temps = dict(temps)
|
|
self._last_update = now
|
|
|
|
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, "last_seen_open": _iso(now),
|
|
}
|
|
if not zone["open_confirmed"]:
|
|
self._t_ref[key] = None # the room may already have cooled before we saw it
|
|
self.changed = True
|
|
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)
|
|
if zone["open_now"]:
|
|
event["last_seen_open"] = _iso(now)
|
|
elif was_open and not ventilating:
|
|
event = self._open_event.pop(key)
|
|
opened = _parse(event["opened"]) # the confirmed opening, or when it was first seen open
|
|
seen_open = _parse(event.get("last_seen_open", event["opened"]))
|
|
closed = zone["last_close"] or now if zone["close_confirmed"] else now
|
|
confirmed = zone["close_confirmed"] and event.get("opened_confirmed", True)
|
|
# Uncertain periods give a range instead of a duration.
|
|
event.update(
|
|
closed=_iso(closed) if zone["close_confirmed"] else None,
|
|
closed_confirmed=zone["close_confirmed"],
|
|
duration_minutes=round((closed - opened) / 60, 1) if confirmed else None,
|
|
duration_min_minutes=round(((closed if zone["close_confirmed"] else seen_open) - opened) / 60, 1),
|
|
duration_max_minutes=(round((now - opened) / 60, 1)
|
|
if not zone["close_confirmed"] and event.get("opened_confirmed", True)
|
|
else None))
|
|
if not zone["close_confirmed"]:
|
|
event["closed_reported"] = _iso(now) # closed at some point before this
|
|
self._closed_at[key] = closed
|
|
self._recovery_confirmed[key] = confirmed
|
|
event["temperature_at_close"] = temp
|
|
self.events.append(event)
|
|
self.changed = True
|
|
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"]),
|
|
others_known=all(z["known"] for k, z in zones.items() if k != key))
|
|
# 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"] if zone["facing_known"] else None,
|
|
"other_opening": zone["other_opening"] if zone["others_known"] else None,
|
|
"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 assumption(self, window: Window, open_minutes: float, now: float) -> dict[str, Any]:
|
|
"""What forecasts assume for one window, and on what basis.
|
|
|
|
Observed open: open for `open_minutes` more. Unknown but last seen open: open
|
|
until `open_minutes` after it was last seen, never renewed, so an unavailable
|
|
sensor does not keep a window open forever. Unknown otherwise: no ventilation,
|
|
marked uncertain (`last_known_closed`, or `unobserved` with no reading at all).
|
|
"""
|
|
state = self.window_state.get(window.key, {})
|
|
if state.get("open"):
|
|
return {"basis": "observed", "open_minutes": float(open_minutes), "uncertain": False}
|
|
if state.get("open") is False:
|
|
return {"basis": "observed", "open_minutes": 0.0, "uncertain": False}
|
|
if state.get("last_known_open"):
|
|
seen = state.get("seen")
|
|
left = open_minutes - (now - seen) / 60 if seen is not None else 0.0
|
|
return {"basis": "last_known_open", "open_minutes": round(max(0.0, left), 1), "uncertain": True,
|
|
"assumed_closed_at": seen + open_minutes * 60 if seen is not None else now}
|
|
basis = "last_known_closed" if state.get("last_known_open") is False else "unobserved"
|
|
return {"basis": basis, "open_minutes": 0.0, "uncertain": True}
|
|
|
|
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 (see `assumption`).
|
|
|
|
Each window open now, or assumed open, closes after its assumed time; once
|
|
the last one has closed 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]
|
|
assumed = {w.key: self.assumption(w, open_minutes, now) for w in windows}
|
|
# Steps each window stays open: at least one for an observed open window.
|
|
open_steps = {w.key: (max(1, round(open_minutes / 5)) if self.window_state.get(w.key, {}).get("open")
|
|
else round(assumed[w.key]["open_minutes"] / 5)) for w in windows}
|
|
direction = weather.get("wind_direction")
|
|
last_open = max(open_steps.values(), default=0)
|
|
if last_open:
|
|
start = now + last_open * 300.0
|
|
else:
|
|
# Recovery from the last close: observed, or assumed for a window last seen open.
|
|
closes = [a["assumed_closed_at"] for a in assumed.values() if "assumed_closed_at" in a]
|
|
closed_at = self._closed_at.get(key)
|
|
start = max([t for t in (closed_at, *closes) if t is not None], default=None)
|
|
result = []
|
|
for k in range(steps):
|
|
t = now + (k + 1) * 300.0
|
|
open_now = [w for w in windows if k < open_steps[w.key]]
|
|
if open_now:
|
|
result.append({
|
|
"opening": sum(w.opening for w in open_now),
|
|
"facing": (sum(w.opening * facing_cos(direction, w.azimuth) for w in open_now)
|
|
if direction is not None else 0.0),
|
|
"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
|
|
recovery = exp(-(t - start) / self.tau) if start is not None and t > start 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]:
|
|
"""Events, plus the state a restart or reload needs to continue them."""
|
|
windows = {w.key: w for w in self.windows if w.enabled}
|
|
state = {
|
|
"time": self._last_update,
|
|
"windows": {key: {"sensor": windows[key].sensor, "inverted": windows[key].inverted,
|
|
"last_known": m["last_known"], "since": m["since"], "seen": m["seen"]}
|
|
for key, m in self._memory.items()
|
|
if key in windows and m["last_known"] is not None and m["seen"] is not None},
|
|
"zones": {zone: {"open_event": self._open_event.get(zone), "closed_at": self._closed_at.get(zone),
|
|
"recovery_confirmed": self._recovery_confirmed.get(zone, True),
|
|
"t_ref": self._t_ref.get(zone)}
|
|
for zone in {w.zone for w in windows.values()}
|
|
if zone in self._open_event or zone in self._closed_at},
|
|
}
|
|
return {"events": list(self.events), "state": state}
|
|
|
|
def _restore(self, state: Any) -> None:
|
|
"""Continue from saved state. Nothing restored is live: every window is unknown
|
|
from when it was last seen until its sensor reports again."""
|
|
if not isinstance(state, Mapping):
|
|
return
|
|
windows = {w.key: w for w in self.windows if w.enabled}
|
|
saved_windows = state.get("windows") if isinstance(state.get("windows"), Mapping) else {}
|
|
for key, item in saved_windows.items():
|
|
window = windows.get(key)
|
|
if (window is None or not isinstance(item, Mapping) or item.get("sensor") != window.sensor
|
|
or item.get("inverted") is not window.inverted or not isinstance(item.get("last_known"), bool)
|
|
or not _finite(item.get("seen"))):
|
|
continue # another sensor now, or polarity flipped: nothing carries over
|
|
seen = float(item["seen"])
|
|
self._memory[key] = {"last_known": item["last_known"], "seen": seen, "unknown_since": seen,
|
|
"since": float(item["since"]) if _finite(item.get("since")) else None}
|
|
remembered_open = {windows[key].zone for key, m in self._memory.items() if m["last_known"]}
|
|
saved_zones = state.get("zones") if isinstance(state.get("zones"), Mapping) else {}
|
|
for zone, item in saved_zones.items():
|
|
if zone not in {w.zone for w in windows.values()} or not isinstance(item, Mapping):
|
|
continue
|
|
event = item.get("open_event")
|
|
# An open event continues only while one of its windows is remembered open.
|
|
if isinstance(event, dict) and zone in remembered_open and _valid_iso(event.get("opened")):
|
|
self._open_event[zone] = dict(event)
|
|
if _finite(item.get("closed_at")):
|
|
self._closed_at[zone] = float(item["closed_at"])
|
|
self._recovery_confirmed[zone] = item.get("recovery_confirmed") is True
|
|
if _finite(item.get("t_ref")):
|
|
self._t_ref[zone] = float(item["t_ref"])
|
|
if self._open_event and _finite(state.get("time")):
|
|
self._restored_at = float(state["time"])
|
|
|
|
|
|
def _finite(value: Any) -> bool:
|
|
return isinstance(value, (int, float)) and not isinstance(value, bool) and isfinite(value)
|
|
|
|
|
|
def _valid_iso(value: Any) -> bool:
|
|
try:
|
|
datetime.fromisoformat(value)
|
|
return True
|
|
except (TypeError, ValueError):
|
|
return False
|
|
|
|
|
|
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
|