diff --git a/custom_components/heating_mpc/evaluation.py b/custom_components/heating_mpc/evaluation.py index 2d2e89e..f8138f3 100644 --- a/custom_components/heating_mpc/evaluation.py +++ b/custom_components/heating_mpc/evaluation.py @@ -34,7 +34,7 @@ from math import isfinite, sqrt from typing import Any from .physics import Learner -from .windows import ventilation_vector_for_prediction +from .windows import ventilation_known, ventilation_vector_for_prediction from .zone_model import feature_vector HORIZONS = (15, 30, 60, 120) # minutes @@ -97,19 +97,50 @@ def snapshot_delta(snapshot: dict[str, Any], temp: float, inputs: dict[str, Any] return main -def replay_zones(origin: dict[str, Any], path: list[dict[str, Any]]) -> dict[str, float]: - """All zones advanced together with the origin's models and the actual inputs.""" +def inputs_complete(snapshot: dict[str, Any], temp: float, inputs: dict[str, Any], + incomplete: set[str]) -> bool: + """Whether a step of a zone's model used only observed inputs. + + Not when a window state, or a weather input an open window's terms need, was + unknown; nor when the extended layer is in use and its electrical heat or a + neighbour is unknown, or that neighbour's own replay is already incomplete. + The replay itself still falls back as forecasts do; the result is just not a + measurement of the model. + """ + vent_features = snapshot["ventilation"][0] if snapshot.get("ventilation") else () + if not ventilation_known(vent_features, inputs.get("ventilation")): + return False + if snapshot["uses_extended"] and snapshot["extended"]: + features = snapshot["extended"][0] + if feature_vector(features, temp, inputs) is None: + return False + if any(f.startswith("couple:") and f.split(":", 1)[1] in incomplete for f in features): + return False + return True + + +def replay_zones(origin: dict[str, Any], path: list[dict[str, Any]]) -> tuple[dict[str, float], set[str]]: + """All zones advanced together with the origin's models and the actual inputs. + + Returns the replayed temperatures and the zones whose replay needed an input + that was not observed at some step (see inputs_complete). + """ state = dict(origin["temps"]) + incomplete: set[str] = set() for before, poll in zip(path, path[1:]): seconds = (poll["time"] - before["time"]).total_seconds() new = {} + unknown = set(incomplete) for zone, temp in state.items(): actual = poll["inputs"].get(zone, {}) inputs = {"outside": origin["outside"], "lag": poll["lag"], "temps": state, "internal": actual.get("internal"), "ventilation": actual.get("ventilation")} - new[zone] = temp + snapshot_delta(origin["models"][zone], temp, inputs) * seconds / STEP_SECONDS - state = new - return state + snapshot = origin["models"][zone] + if not inputs_complete(snapshot, temp, inputs, incomplete): + unknown.add(zone) + new[zone] = temp + snapshot_delta(snapshot, temp, inputs) * seconds / STEP_SECONDS + state, incomplete = new, unknown + return state, incomplete class ForecastEvaluator: @@ -121,6 +152,9 @@ class ForecastEvaluator: # UTC day -> horizon -> room -> [count, sum error, sum |error|, sum |persistence error|, # sum |base model error|, sum error^2, count with squares (0 for 0.3/0.4 data)] self._days: dict[str, dict[str, dict[str, list[float]]]] = {} + # UTC day -> horizon -> room -> [incomplete spans, skipped spans, sum |base model error| + # over the incomplete ones]. Neither kind enters the statistics above. + self._excluded: dict[str, dict[str, dict[str, list[float]]]] = {} def reset_window(self) -> None: """Inputs were missing: no forecast may be scored across this poll.""" @@ -147,6 +181,11 @@ class ForecastEvaluator: self._score() self._prune(time) + def _exclude(self, day: str, horizon: int, room: str, kind: int, base_error: float = 0.0) -> None: + bucket = self._excluded.setdefault(day, {}).setdefault(str(horizon), {}).setdefault(room, [0, 0, 0.0]) + bucket[kind] += 1 + bucket[2] += abs(base_error) + def _score(self) -> None: history = list(self._history) latest = history[-1] @@ -156,19 +195,27 @@ class ForecastEvaluator: if len(history) < steps + 1: continue path = history[-(steps + 1):] + origin = path[0] # The actual heat input must be known along the way. if any(poll["lag_uncertainty"] > self._max_lag_uncertainty for poll in path[1:]): + for room in origin["temps"]: + self._exclude(day, horizon, room, SKIPPED) continue - origin = path[0] inputs = [(poll["lag"], (poll["time"] - before["time"]).total_seconds()) for before, poll in zip(path, path[1:])] - full = replay_zones(origin, path) - for room, actual in latest["temps"].items(): - start, theta = origin["temps"].get(room), origin["theta"].get(room) - if start is None or theta is None: + full, incomplete = replay_zones(origin, path) + for room, start in origin["temps"].items(): + actual, theta = latest["temps"].get(room), origin["theta"].get(room) + if actual is None or theta is None: + self._exclude(day, horizon, room, SKIPPED) # no reading at the end continue error = full[room] - actual base_error = replay(start, origin["outside"], theta, inputs) - actual + if room in incomplete: + # The base model needs only the relay and outdoor temperature: its + # error is still a measurement, kept apart from the main figures. + self._exclude(day, horizon, room, INCOMPLETE, base_error) + continue bucket = self._days.setdefault(day, {}).setdefault(str(horizon), {}).setdefault( room, [0, 0.0, 0.0, 0.0, 0.0, 0.0, 0]) bucket[0] += 1 @@ -181,8 +228,9 @@ class ForecastEvaluator: def _prune(self, time: datetime) -> None: oldest = (time.date() - timedelta(days=WINDOW_DAYS - 1)).isoformat() - for day in [day for day in self._days if day < oldest]: - del self._days[day] + for store in (self._days, self._excluded): + for day in [day for day in store if day < oldest]: + del store[day] def summary(self, rooms: list[str]) -> dict[int, dict[str, Any]]: """Error statistics per horizon, overall and per configured room.""" @@ -196,15 +244,27 @@ class ForecastEvaluator: for i, value in enumerate(values): totals[i] += value overall = [sum(values[i] for values in per_room.values()) for i in range(7)] - result[horizon] = {**_stats(overall, horizon, len(per_room)), - "rooms": {room: _stats(values, horizon) for room, values in per_room.items()}} + excluded: dict[str, list[float]] = {} + for day in self._excluded.values(): + for room, values in day.get(str(horizon), {}).items(): + if room in rooms: + totals = excluded.setdefault(room, [0, 0, 0.0]) + for i, value in enumerate(values): + totals[i] += value + all_excluded = [sum(values[i] for values in excluded.values()) for i in range(3)] + result[horizon] = {**_stats(overall, horizon, len(per_room)), **_excluded_stats(all_excluded), + "rooms": {room: {**_stats(per_room.get(room, [0, 0.0, 0.0, 0.0, 0.0, 0.0, 0]), + horizon), + **_excluded_stats(excluded.get(room, [0, 0, 0.0]))} + for room in rooms if room in per_room or room in excluded}} return result def forget(self, room: str) -> None: """Drop a zone's statistics, for example when its temperature is measured differently.""" - for by_horizon in self._days.values(): - for by_room in by_horizon.values(): - by_room.pop(room, None) + for store in (self._days, self._excluded): + for by_horizon in store.values(): + for by_room in by_horizon.values(): + by_room.pop(room, None) for poll in self._history: poll["temps"].pop(room, None) @@ -213,11 +273,20 @@ class ForecastEvaluator: return {"days": {day: {horizon: {room: [v[0], *(round(x, 6) for x in v[1:6]), v[6]] for room, v in by_room.items()} for horizon, by_room in by_horizon.items()} - for day, by_horizon in self._days.items()}} + for day, by_horizon in self._days.items()}, + "excluded": {day: {horizon: {room: [v[0], v[1], round(v[2], 6)] for room, v in by_room.items()} + for horizon, by_room in by_horizon.items()} + for day, by_horizon in self._excluded.items()}} def load(self, data: Any) -> None: """Restore stored statistics, keeping only well-formed entries.""" - self._days = {} + self._days, self._excluded = {}, {} + excluded = data.get("excluded") if isinstance(data, dict) else None + for day, horizon, room, values in _entries(excluded): + if (isinstance(values, list) and len(values) == 3 and all(type(v) is int and v >= 0 for v in values[:2]) + and type(values[2]) in (int, float) and isfinite(values[2]) and values[2] >= 0): + self._excluded.setdefault(day, {}).setdefault(horizon, {})[room] = [ + values[0], values[1], float(values[2])] days = data.get("days") if isinstance(data, dict) else None if not isinstance(days, dict): return @@ -243,6 +312,30 @@ class ForecastEvaluator: *squares] +INCOMPLETE, SKIPPED = 0, 1 # columns of the excluded-span counts + + +def _entries(days: Any): + """(day, horizon, zone, values) of stored per-day statistics with a valid shape.""" + if not isinstance(days, dict): + return + horizons = {str(horizon) for horizon in HORIZONS} + for day, by_horizon in days.items(): + if not _valid_day(day) or not isinstance(by_horizon, dict): + continue + for horizon, by_zone in by_horizon.items(): + if horizon in horizons and isinstance(by_zone, dict): + for zone, values in by_zone.items(): + if isinstance(zone, str): + yield day, horizon, zone, values + + +def _excluded_stats(values: list[float]) -> dict[str, Any]: + incomplete, skipped, base = values + return {"incomplete": int(incomplete), "skipped": int(skipped), + "incomplete_base_model_mae": round(base / incomplete, 3) if incomplete else None} + + def _valid_bucket(room: Any, values: Any) -> bool: if not isinstance(room, str) or not isinstance(values, list) or len(values) not in (4, 5, 7): return False diff --git a/custom_components/heating_mpc/sensor.py b/custom_components/heating_mpc/sensor.py index 738fc7e..6021aa2 100644 --- a/custom_components/heating_mpc/sensor.py +++ b/custom_components/heating_mpc/sensor.py @@ -180,6 +180,11 @@ class MPCForecastError(CoordinatorEntity, SensorEntity): "skill": accuracy.get("skill"), "base_model_mae": accuracy.get("base_model_mae"), "improvement_vs_base": accuracy.get("improvement_vs_base"), + # Spans left out of the figures above: an input the model needed was unknown + # (the base model's error on them is still measured), or no replay was possible. + "incomplete": accuracy.get("incomplete", 0), + "incomplete_base_model_mae": accuracy.get("incomplete_base_model_mae"), + "skipped": accuracy.get("skipped", 0), # Per zone, by name. "zones": {self.coordinator.zones[key].name if key in self.coordinator.zones else key: stats for key, stats in accuracy.get("rooms", {}).items()}, diff --git a/custom_components/heating_mpc/windows.py b/custom_components/heating_mpc/windows.py index 958385d..9c6a757 100644 --- a/custom_components/heating_mpc/windows.py +++ b/custom_components/heating_mpc/windows.py @@ -150,17 +150,22 @@ 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. + `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: - 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", "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: @@ -178,7 +183,10 @@ def ventilation_vector(features: Iterable[str], temp: float, outside: float, 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) + 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") @@ -203,10 +211,22 @@ def ventilation_vector_for_prediction(features: Iterable[str], temp: float, outs 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}) + "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") @@ -266,7 +286,7 @@ class VentilationTracker: 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": []}) + "open_confirmed": True, "unknown_windows": [], "facing_known": True}) if fraction is None: zone["known"] = False else: @@ -274,6 +294,8 @@ class VentilationTracker: 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 @@ -382,7 +404,8 @@ class VentilationTracker: 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"])) + 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 @@ -396,8 +419,10 @@ class VentilationTracker: 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"], + 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"])} diff --git a/tests/ha/test_lifecycle.py b/tests/ha/test_lifecycle.py index e865a86..772fb05 100644 --- a/tests/ha/test_lifecycle.py +++ b/tests/ha/test_lifecycle.py @@ -302,6 +302,7 @@ async def test_forecast_error_sensors_score_scheduled_polls(hass, entry, inputs, assert state.attributes["mae"] == accuracy["mae"] > 0 assert state.attributes["quality"] == "insufficient" assert state.attributes["samples"] == 1 + assert (state.attributes["incomplete"], state.attributes["skipped"]) == (0, 0) assert state.attributes["zones"]["room"]["samples"] == 1 # by zone name assert state.attributes["unit_of_measurement"] == "°C" assert state.attributes["friendly_name"] == "Heating MPC Model Error 15 min" diff --git a/tests/test_evaluation.py b/tests/test_evaluation.py index 2a1ede4..4302c7a 100644 --- a/tests/test_evaluation.py +++ b/tests/test_evaluation.py @@ -116,15 +116,22 @@ class EvaluationTests(unittest.TestCase): day: {'45': {'sensor.room': [1, 0.0, 0.1, 0.1]}, '15': {'sensor.a': [1, 0.0, -0.1, 0.1], 'sensor.b': [True, 0, 0, 0], 'sensor.c': [1, float('nan'), 0, 0], 'sensor.d': [2, 0.1, 0.2, 0.4], - 'sensor.e': [2, 0.1, 0.2, 0.4, 0.3], 'sensor.f': [2, 0.1, 0.2, 0.4, -1]}}}}) + 'sensor.e': [2, 0.1, 0.2, 0.4, 0.3], 'sensor.f': [2, 0.1, 0.2, 0.4, -1]}}}, + 'excluded': {day: {'15': {'sensor.d': [1, 2, 0.3], 'sensor.x': [1, -2, 0.3], 'sensor.y': [1, 2]}, + '45': {'sensor.d': [1, 2, 0.3]}}}}) # A 0.3 bucket has no base-model column; its model was the base model then. # Neither 0.3 nor 0.4 buckets have squared errors: they do not count for RMSE. + # 0.4 data has no excluded-span counts. self.assertEqual(restored.as_dict(), {'days': {day: {'15': { - 'sensor.d': [2, 0.1, 0.2, 0.4, 0.2, 0.0, 0], 'sensor.e': [2, 0.1, 0.2, 0.4, 0.3, 0.0, 0]}}}}) + 'sensor.d': [2, 0.1, 0.2, 0.4, 0.2, 0.0, 0], 'sensor.e': [2, 0.1, 0.2, 0.4, 0.3, 0.0, 0]}}}, + 'excluded': {day: {'15': {'sensor.d': [1, 2, 0.3]}}}}) self.assertIsNone(restored.summary(['sensor.d'])[15]['rmse']) + self.assertEqual({k: restored.summary(['sensor.d'])[15][k] for k in ('incomplete', 'skipped', + 'incomplete_base_model_mae')}, + {'incomplete': 1, 'skipped': 2, 'incomplete_base_model_mae': 0.3}) for garbage in (None, [], {'days': []}): restored.load(garbage) - self.assertEqual(restored.as_dict(), {'days': {}}) + self.assertEqual(restored.as_dict(), {'days': {}, 'excluded': {}}) T0 = START.timestamp() @@ -206,3 +213,89 @@ class OperationalForecastTests(unittest.TestCase): if __name__ == '__main__': unittest.main() + + +BASE = (0.004, 0.12, 0.0005) + + +def ventilated(observed, actual=None, wind=4.0, features=('vent_natural',), zones=('r',), internal=None, + extended=None, lag_uncertain_at=(), missing_end_at=()): + """Zones cooled by real ventilation; `observed[k]` is what the window sensor said + for the interval ending at poll k (None: unknown). Returns the summary at 15 min.""" + actual = actual if actual is not None else [1.0 if o is None else o for o in observed] + snapshot = {'base': BASE, 'uses_extended': extended is not None, 'extended': extended, + 'ventilation': (list(features), (0.02,) * len(features))} + evaluator, temps = ev.ForecastEvaluator(0.1), {zone: 20.0 for zone in zones} + for k, seen in enumerate(observed): + if k: + temps = {zone: t + 0.004 * (5 - t) + 0.12 * 0.5 + 0.0005 + + 0.02 * actual[k] * (5 - t) * (1 + (wind or 0) * ('vent_wind' in features)) + for zone, t in temps.items()} + vent = {'opening': seen, 'facing': 0.0, 'other_opening': 0.0, 'recovery': 0.0, 't_ref': None, + 'wind': wind} + readings = {z: t for z, t in temps.items() if not (k in missing_end_at and z == zones[0])} + evaluator.observe(START + timedelta(seconds=300 * k), readings, 5.0, 0.5, + 0.5 if k in lag_uncertain_at else 0.0, {z: BASE for z in zones}, + models={z: snapshot for z in zones}, + inputs={z: {'internal': internal, 'ventilation': vent if z == zones[0] else + {**vent, 'opening': actual[k]}} for z in zones}) + return evaluator.summary(list(zones))[15] + + +class IncompleteInputTests(unittest.TestCase): + """Unknown ventilation or extended inputs never become Model Error samples.""" + + def test_window_unknown_for_the_whole_span_is_incomplete(self): + stats = ventilated([0.0, None, None, None]) + self.assertEqual((stats['samples'], stats['mae'], stats['incomplete']), (0, None, 1)) + # The base model needs no window input: its error is still kept, apart. + self.assertGreater(stats['incomplete_base_model_mae'], 0.1) + + def test_window_unknown_halfway_or_recovering_during_the_span(self): + self.assertEqual(ventilated([0.0, 1.0, None, 1.0])['incomplete'], 1) # unavailable halfway + stats = ventilated([0.0, None, 1.0, 1.0, 1.0, 1.0, 1.0]) # back after one poll + # Spans ending at polls 3-6: only the first contains the unknown interval. + self.assertEqual((stats['incomplete'], stats['samples']), (1, 3)) + self.assertAlmostEqual(stats['mae'], 0, places=6) # the known spans are exact + + def test_one_zone_unknown_leaves_the_other_zone_scored(self): + stats = ventilated([0.0, None, None, None], zones=('a', 'b')) + self.assertEqual((stats['rooms']['a']['incomplete'], stats['rooms']['a']['samples']), (1, 0)) + self.assertEqual((stats['rooms']['b']['incomplete'], stats['rooms']['b']['samples']), (0, 1)) + + def test_missing_wind_matters_only_while_a_window_is_open(self): + features = ('vent_natural', 'vent_wind') + open_span = ventilated([0.0, 1.0, 1.0, 1.0], wind=None, features=features) + closed_span = ventilated([0.0, 0.0, 0.0, 0.0], wind=None, features=features) + self.assertEqual((open_span['incomplete'], open_span['samples']), (1, 0)) + self.assertEqual((closed_span['incomplete'], closed_span['samples']), (0, 1)) + self.assertEqual(ventilated([0.0, 1.0, 1.0, 1.0], features=features)['samples'], 1) + + def test_unknown_electrical_heat_makes_the_extended_replay_incomplete(self): + extended = (['loss', 'heat', 'bias', 'internal'], BASE + (0.15,)) + stats = ventilated([0.0] * 4, extended=extended, internal=None) + self.assertEqual((stats['incomplete'], stats['samples']), (1, 0)) + self.assertIsNotNone(stats['incomplete_base_model_mae']) + self.assertEqual(ventilated([0.0] * 4, extended=extended, internal=0.0)['samples'], 1) + + def test_a_neighbour_with_unknown_inputs_spoils_a_coupled_zone(self): + coupled = {'base': BASE, 'uses_extended': True, 'ventilation': None, + 'extended': (['loss', 'heat', 'bias', 'couple:a'], BASE + (0.01,))} + plain = {'base': BASE, 'uses_extended': False, 'extended': None, + 'ventilation': (['vent_natural'], (0.02,))} + origin = {'temps': {'a': 20.0, 'b': 20.0}, 'outside': 5.0, 'models': {'a': plain, 'b': coupled}} + path = [origin] + [{'time': START + timedelta(seconds=300 * k), 'lag': 0.5, + 'inputs': {'a': {'ventilation': {'opening': None}}, 'b': {}}} for k in range(4)] + origin['time'] = START - timedelta(seconds=300) + _, incomplete = ev.replay_zones(origin, path) + self.assertEqual(incomplete, {'a', 'b'}) + path = [origin] + [{**poll, 'inputs': {'a': {'ventilation': {'opening': 0.0}}, 'b': {}}} for poll in path[1:]] + self.assertEqual(ev.replay_zones(origin, path)[1], set()) + + def test_skipped_spans_are_counted_and_samples_stay_correct(self): + stats = ventilated([0.0] * 9, lag_uncertain_at=(4,)) + # Spans ending at 3..8: those crossing poll 4 (ending 4, 5, 6) are skipped. + self.assertEqual((stats['samples'], stats['skipped'], stats['incomplete']), (3, 3, 0)) + stats = ventilated([0.0] * 5, missing_end_at=(4,), zones=('a', 'b')) + self.assertEqual((stats['rooms']['a']['samples'], stats['rooms']['a']['skipped']), (1, 1)) + self.assertEqual(stats['rooms']['b']['samples'], 2) diff --git a/tests/test_windows.py b/tests/test_windows.py index 40a6317..ddf6e3c 100644 --- a/tests/test_windows.py +++ b/tests/test_windows.py @@ -392,3 +392,26 @@ class StartupAndRestartTests(unittest.TestCase): self.assertFalse(tracker.changed) # nothing new self.poll(tracker, 'unavailable') self.assertTrue(tracker.changed) + + +class StrictInputTests(unittest.TestCase): + """Weather inputs of an open window are required, never assumed, outside forecasts.""" + + def test_unknown_direction_or_other_zone_while_open(self): + windows = wm.build_windows([(window('a'), None), (window('c', zone='living'), None)], ['bed', 'living']) + tracker = wm.VentilationTracker(windows) + calm = {**WEATHER, 'wind_direction': None} + for t, a, c in ((300.0, 'off', 'off'), (600.0, 'on', 'unavailable'), (900.0, 'on', 'unavailable')): + tracker.update(t, {'a': (a, 450.0), 'c': (c, 450.0)}, {'bed': 20.0, 'living': 21.0}, calm) + bed = tracker.zone_inputs('bed', calm) + self.assertEqual((bed['opening'], bed['facing'], bed['other_opening']), (1.0, None, None)) + features = ['vent_natural', 'vent_wind', 'vent_wind_dir', 'vent_cross'] + self.assertFalse(wm.ventilation_known(features, bed)) + self.assertTrue(wm.ventilation_known(['vent_natural', 'vent_wind'], bed)) + self.assertIsNone(wm.ventilation_vector(features, 20.0, 2.0, bed)) # learning refuses + self.assertIsNotNone(wm.ventilation_vector_for_prediction(features, 20.0, 2.0, bed)) + # Closed, every opening term is 0 whatever the weather: nothing is missing. + closed = {**bed, 'opening': 0.0, 'facing': 0.0, 'wind': None, 'other_opening': None} + self.assertEqual(wm.ventilation_vector(features, 20.0, 2.0, closed), [0.0, 0.0, 0.0, 0.0]) + self.assertTrue(wm.ventilation_known(features, closed)) + self.assertTrue(wm.ventilation_known(features, None)) # no windows