"""Windows, wind exposure, ventilation inputs and recovery, without Home Assistant.""" import sys import types import unittest from math import cos, exp, radians from pathlib import Path COMPONENTS = Path(__file__).resolve().parents[1] / 'custom_components' / 'heating_mpc' sys.modules.setdefault('heating_mpc', types.ModuleType('heating_mpc')) sys.modules['heating_mpc'].__path__ = [str(COMPONENTS)] from heating_mpc import windows as wm # noqa: E402 WEATHER = {'temperature': 2.0, 'wind': 4.0, 'wind_direction': 204.0, 'wind_gust': 9.0} def window(key, zone='bed', **extra): return {'key': key, 'name': key.title(), 'sensor': f'binary_sensor.{key}', 'zone': zone, **extra} class GeometryTests(unittest.TestCase): def test_wind_relative_to_facade(self): # The apartment's façade faces about 345° (NNW); OpenWeatherMap wind from 204°. self.assertAlmostEqual(wm.angle_difference(204, 345), -141) self.assertAlmostEqual(wm.facing_cos(204, 345), cos(radians(141))) self.assertAlmostEqual(wm.facing_cos(345, 345), 1) # blowing straight at the window self.assertAlmostEqual(wm.facing_cos(165, 345), -1) # from behind the building self.assertAlmostEqual(wm.facing_cos(75, 345), 0, places=9) # crosswind # Around north, differences wrap instead of jumping by 360°. self.assertAlmostEqual(wm.angle_difference(5, 355), 10) self.assertAlmostEqual(wm.angle_difference(355, 5), -10) self.assertAlmostEqual(wm.facing_cos(5, 355), cos(radians(10))) def test_circular_mean(self): self.assertAlmostEqual(wm.circular_mean([350, 10]) % 360 if wm.circular_mean([350, 10]) < 359 else 0, 0) self.assertAlmostEqual(wm.circular_mean([350, 20], [1, 1]), 5) self.assertIsNone(wm.circular_mean([90, 270])) self.assertAlmostEqual(wm.circular_mean([0, 90], [3, 1]), 18.4349488, places=5) def test_contact_states_and_open_fraction(self): self.assertEqual([wm.contact_open(s) for s in ('on', 'off', 'unknown', 'unavailable', None)], [True, False, None, None, None]) self.assertEqual([wm.contact_open(s, inverted=True) for s in ('on', 'off')], [False, True]) self.assertEqual(wm.open_fraction(True, True, 0, 100, 400), 1.0) self.assertEqual(wm.open_fraction(False, False, 0, 100, 400), 0.0) self.assertAlmostEqual(wm.open_fraction(False, True, 220, 100, 400), 0.6) # opened 120 s in self.assertAlmostEqual(wm.open_fraction(True, False, 220, 100, 400), 0.4) # closed 120 s in self.assertIsNone(wm.open_fraction(False, False, 220, 100, 400)) # round trip: unknown self.assertIsNone(wm.open_fraction(None, True, 0, 100, 400)) def test_ventilation_vector(self): features = list(wm.VENTILATION_FEATURES) inputs = {'opening': 0.5, 'facing': -0.3, 'other_opening': 1.0, 'recovery': 0.0, 't_ref': None, 'wind': 4.0, 'gust': 9.0} for value, expected in zip(wm.ventilation_vector(features, 20.0, 2.0, inputs), [-9.0, -36.0, 21.6, -45.0, -9.0, 0.0]): self.assertAlmostEqual(value, expected) self.assertIsNone(wm.ventilation_vector(['vent_wind'], 20, 2, {**inputs, 'wind': None})) # Calm air: wind terms vanish, natural ventilation stays. calm = wm.ventilation_vector(features, 20.0, 2.0, {**inputs, 'wind': 0.0, 'gust': 0.0}) self.assertEqual([round(v, 9) for v in calm[1:4]], [0.0, 0.0, 0.0]) self.assertEqual(calm[0], -9.0) self.assertEqual(wm.ventilation_vector(['vent_natural'], 20, 2, {**inputs, 'wind': None}), [-9.0]) self.assertIsNone(wm.ventilation_vector(['recovery'], 20, 2, {**inputs, 'recovery': 0.5})) self.assertEqual(wm.ventilation_vector(['recovery'], 19, 2, {**inputs, 'recovery': 0.5, 't_ref': 21}), [1.0]) def test_window_config_validation(self): built = wm.build_windows([ (window('a', azimuth=705), None), (window('b', opening_factor=9), None), (window('c', zone='gone'), None), (window('a'), None), ({**window('d'), 'sensor': 'sensor.d'}, None), (window('e', inverted=True), 's5')], ['bed']) self.assertEqual([(w.key, w.azimuth, w.inverted) for w in built], [('a', 345.0, False), ('e', 345.0, True)]) self.assertEqual(wm.validate_window({'name': ' '}, built), 'window_name_required') self.assertEqual(wm.validate_window({'name': 'x', 'sensor': 'binary_sensor.a'}, built), 'sensor_in_other_window') self.assertIsNone(wm.validate_window({'name': 'x', 'sensor': 'binary_sensor.x'}, built)) class TrackerTests(unittest.TestCase): def setUp(self): self.tracker = wm.VentilationTracker(wm.build_windows([(window('bed_window'), None)], ['bed']), recovery_minutes=20) self.t = 1_000_000.0 self.changed = 0.0 def poll(self, state, temp, changed_offset=None, weather=WEATHER): if changed_offset is not None: self.changed = self.t - changed_offset self.tracker.update(self.t, {'bed_window': (state, self.changed)}, {'bed': temp}, weather) inputs = self.tracker.zone_inputs('bed', weather) self.t += 300 return inputs def test_ventilation_event_and_recovery(self): self.assertFalse(self.poll('off', 21.0)['known']) # no previous poll yet closed = self.poll('off', 21.0) self.assertEqual((closed['opening'], closed['affected']), (0.0, False)) opened = self.poll('on', 20.6, changed_offset=180) # opened 3 minutes ago self.assertAlmostEqual(opened['opening'], 0.6) self.assertAlmostEqual(opened['facing'], 0.6 * cos(radians(141))) self.assertTrue(opened['affected']) self.assertEqual(opened['t_ref'], 21.0) # temperature at the poll before opening for temp in (20.2, 19.9): self.assertEqual(self.poll('on', temp)['opening'], 1.0) closing = self.poll('off', 20.0, changed_offset=60) # closed a minute ago self.assertAlmostEqual(closing['opening'], 0.8) self.assertAlmostEqual(closing['recovery'], exp(-60 / 1200)) after = [self.poll('off', 20.0 + 0.1 * k) for k in range(12)] self.assertTrue(after[0]['affected'] and after[0]['recovery'] > 0.7) self.assertEqual((after[-1]['recovery'], after[-1]['affected']), (0.0, False)) # ~3 time constants event = self.tracker.events[-1] self.assertEqual((event['temperature_before'], event['lowest_temperature'], event['temperature_at_close']), (21.0, 19.9, 20.0)) self.assertEqual((event['duration_minutes'], event['wind_direction'], event['windows']), (17.0, 204.0, ['Bed_Window'])) # 3 min + 15 min - 1 min self.assertEqual(event['temperature_after_minutes'], 30) restored = wm.VentilationTracker(self.tracker.windows, saved=self.tracker.as_dict()) self.assertEqual(list(restored.events), list(self.tracker.events)) def test_unknown_state_blocks_learning_and_multiple_windows_add(self): tracker = wm.VentilationTracker(wm.build_windows( [(window('w1'), None), (window('w2', opening_factor=0.5), None), (window('w3', zone='living'), None)], ['bed', 'living'])) for t in (0.0, 300.0): tracker.update(t, {'w1': ('on', -10), 'w2': ('on', -10), 'w3': ('on', -10)}, {'bed': 20, 'living': 21}, WEATHER) bed = tracker.zone_inputs('bed', WEATHER) self.assertEqual((bed['opening'], bed['other_opening']), (1.5, 1.0)) # living's window is open too tracker.update(600.0, {'w1': ('unavailable', 550), 'w2': ('on', -10), 'w3': ('off', 400)}, {'bed': 20, 'living': 21}, WEATHER) bed = tracker.zone_inputs('bed', WEATHER) self.assertEqual((bed['opening'], bed['known'], bed['affected']), (None, False, True)) self.assertIsNone(tracker.zone_inputs('kitchen', WEATHER)) def test_forecast_assumes_closing_then_recovery(self): self.poll('off', 21.0) self.poll('off', 21.0) self.poll('on', 20.5, changed_offset=100) steps = self.tracker.forecast('bed', 12, open_minutes=15, weather=WEATHER, now=self.t) self.assertEqual([s['opening'] for s in steps[:4]], [1.0, 1.0, 1.0, 0.0]) self.assertAlmostEqual(steps[0]['facing'], cos(radians(141))) self.assertAlmostEqual(steps[3]['recovery'], exp(-300 / 1200)) self.assertGreater(steps[3]['recovery'], steps[6]['recovery']) self.assertEqual(steps[5]['t_ref'], 21.0) if __name__ == '__main__': unittest.main() class UnknownStateTests(unittest.TestCase): """OPEN, CLOSED and UNKNOWN are three states: unknown never means closed.""" def setUp(self): self.tracker = wm.VentilationTracker(wm.build_windows([(window('w'), None)], ['bed']), recovery_minutes=20) self.t = 1_000_000.0 def poll(self, state, changed_ago=None, temp=20.0): changed = self.t - changed_ago if changed_ago is not None else self.t - 100_000 self.tracker.update(self.t, {'w': (state, changed)}, {'bed': temp}, WEATHER) inputs = self.tracker.zone_inputs('bed', WEATHER) forecast = self.tracker.forecast('bed', 6, 15, WEATHER, self.t) self.t += 300 return inputs, forecast, dict(self.tracker.window_state['w']) def open_for_a_while(self): self.poll('off') self.poll('off', temp=21.0) self.poll('on', changed_ago=100, temp=20.4) self.poll('on', changed_ago=400, temp=20.0) def test_open_to_unknown_keeps_the_event_and_assumes_still_open(self): self.open_for_a_while() inputs, forecast, state = self.poll('unavailable', changed_ago=60, temp=19.8) self.assertEqual(len(self.tracker.events), 0) # no confirmed closing without evidence self.assertIn('bed', self.tracker._open_event) self.assertEqual((inputs['known'], inputs['affected'], inputs['recovery']), (False, True, 0.0)) self.assertEqual(forecast[0]['opening'], 1.0) # last seen open: still treated as open self.assertEqual((state['open'], state['last_known_open']), (None, True)) self.assertIsNotNone(state['unknown_since']) def test_unknown_to_open_continues_the_same_event(self): self.open_for_a_while() self.poll('unavailable', changed_ago=60) self.poll('unavailable', changed_ago=360) inputs, _, state = self.poll('on', changed_ago=30) # sensor back, window still open self.assertEqual((len(self.tracker.events), state['unknown_since']), (0, None)) self.assertFalse(inputs['known']) # the interval itself was unknown inputs, _, _ = self.poll('on', changed_ago=330) self.assertEqual((inputs['known'], inputs['opening']), (True, 1.0)) self.poll('off', changed_ago=120, temp=20.2) # confirmed close event = self.tracker.events[-1] self.assertEqual((event['closed_confirmed'], event['sensor_unknown_minutes']), (True, 10.0)) # two polls self.assertIsNotNone(event['duration_minutes']) def test_unknown_to_closed_records_an_unconfirmed_close(self): self.open_for_a_while() self.poll('unavailable', changed_ago=60) inputs, forecast, state = self.poll('off', changed_ago=20, temp=20.1) event = self.tracker.events[-1] self.assertEqual((event['closed'], event['closed_confirmed'], event['duration_minutes']), (None, False, None)) self.assertGreater(event['duration_max_minutes'], 0) self.assertEqual((state['open'], state['last_known_open'], state['unknown_since']), (False, False, None)) # Recovery is assumed from when the sensor reported closed, but its timing is # unknown, so neither the ventilation nor the envelope model learns from it. self.assertGreater(inputs['recovery'], 0.9) self.assertEqual((inputs['known'], inputs['affected'], inputs['opening']), (False, True, None)) self.assertEqual(forecast[0]['opening'], 0.0) self.assertGreater(forecast[0]['recovery'], 0.0) def test_closed_to_unknown_assumes_closed_but_blocks_learning(self): self.poll('off') self.poll('off') inputs, forecast, state = self.poll('unavailable', changed_ago=120) self.assertEqual((inputs['known'], inputs['affected']), (False, True)) self.assertEqual(forecast[0]['opening'], 0.0) self.assertEqual((state['open'], state['last_known_open']), (None, False)) self.assertEqual(len(self.tracker.events), 0) def test_unknown_to_open_after_a_closed_window_starts_an_unconfirmed_event(self): self.poll('off') self.poll('off', temp=21.0) self.poll('unavailable', changed_ago=60, temp=20.6) self.poll('on', changed_ago=30, temp=20.0) # opened at some point during the outage event = self.tracker._open_event['bed'] self.assertEqual((event['opened_confirmed'], event['temperature_before']), (False, None)) self.poll('on', changed_ago=330, temp=19.8) inputs, _, _ = self.poll('off', changed_ago=100, temp=19.9) self.assertTrue(self.tracker.events[-1]['closed_confirmed']) # Without the temperature before ventilation, recovery cannot be learned. self.assertFalse(inputs['known']) def test_recovery_after_prolonged_unavailability(self): self.open_for_a_while() for _ in range(72): # six hours without the sensor self.poll('unavailable', changed_ago=60) self.assertEqual(len(self.tracker.events), 0) inputs, _, _ = self.poll('off', changed_ago=60) event = self.tracker.events[-1] self.assertFalse(event['closed_confirmed']) self.assertGreaterEqual(event['duration_max_minutes'], 360) self.assertEqual(event['sensor_unknown_minutes'], 360.0) for _ in range(12): # recovery decays, learning resumes inputs, _, _ = self.poll('off') self.assertEqual((inputs['known'], inputs['affected'], inputs['recovery']), (True, False, 0.0)) def test_multiple_windows_with_one_unknown(self): tracker = wm.VentilationTracker(wm.build_windows([(window('a'), None), (window('b'), None)], ['bed'])) for t, a, b in ((0, 'off', 'off'), (300, 'on', 'off'), (600, 'on', 'unavailable'), (900, 'off', 'unavailable')): tracker.update(float(t), {'a': (a, t - 60.0), 'b': (b, t - 60.0)}, {'bed': 20.0}, WEATHER) # Window a closed (confirmed); b was last seen closed: the event ends, confirmed, # but the zone stays out of learning while b's state is unknown. self.assertTrue(tracker.events[-1]['closed_confirmed']) inputs = tracker.zone_inputs('bed', WEATHER) self.assertEqual((inputs['known'], inputs['unknown_windows']), (False, ['B'])) class StartupAndRestartTests(unittest.TestCase): """Unknown at startup is not closed; saved state comes back as unknown, never live.""" WINDOWS = wm.build_windows([(window('w'), None)], ['bed']) def setUp(self): self.t = 2_000_000.0 def poll(self, tracker, state, changed_ago=100_000, temp=20.0): tracker.update(self.t, {'w': (state, self.t - changed_ago)}, {'bed': temp}, WEATHER) result = (tracker.zone_inputs('bed', WEATHER), tracker.forecast('bed', 6, 15, WEATHER, self.t), tracker.assumption(self.WINDOWS[0], 15, self.t)) self.t += 300 return result def saved_with(self, *states): tracker = wm.VentilationTracker(self.WINDOWS) for state in states: self.poll(tracker, state, changed_ago=200 if state == 'on' else 100_000, temp=21.0) return tracker, tracker.as_dict() def test_unavailable_from_startup_is_unobserved_not_closed(self): tracker = wm.VentilationTracker(self.WINDOWS) for _ in range(3): inputs, forecast, assumed = self.poll(tracker, 'unavailable') self.assertEqual(assumed, {'basis': 'unobserved', 'open_minutes': 0.0, 'uncertain': True}) self.assertEqual((inputs['known'], inputs['affected'], inputs['unknown_windows']), (False, True, ['W'])) self.assertEqual(forecast[0]['opening'], 0.0) # no ventilation assumed, but flagged uncertain self.assertEqual((len(tracker.events), tracker._open_event), (0, {})) def test_first_reading_open_is_an_unconfirmed_opening(self): tracker = wm.VentilationTracker(self.WINDOWS) self.poll(tracker, 'on', changed_ago=30) # we never saw it closed event = tracker._open_event['bed'] self.assertEqual((event['opened_confirmed'], event['temperature_before']), (False, None)) self.poll(tracker, 'on') self.poll(tracker, 'off', changed_ago=60) event = tracker.events[-1] self.assertTrue(event['closed_confirmed']) # Opened at some unknown time before it was first seen: a minimum duration only. self.assertEqual((event['duration_minutes'], event['duration_max_minutes']), (None, None)) self.assertEqual(event['duration_min_minutes'], 9.5) def test_restart_with_saved_open_window_and_sensor_unavailable(self): before, saved = self.saved_with('off', 'off', 'on', 'on') self.assertEqual(saved['state']['windows']['w']['last_known'], True) self.assertIn('bed', saved['state']['zones']) self.t += 120 # restart takes two minutes tracker = wm.VentilationTracker(self.WINDOWS, saved=saved) self.assertEqual(tracker.window_state, {}) # nothing is live before the first reading inputs, forecast, assumed = self.poll(tracker, 'unavailable') state = tracker.window_state['w'] self.assertEqual((state['open'], state['last_known_open']), (None, True)) self.assertEqual(state['unknown_since'], saved['state']['windows']['w']['seen']) # Assumed open only until 15 minutes after it was last seen (7 minutes ago). self.assertEqual((assumed['basis'], assumed['open_minutes'], assumed['uncertain']), ('last_known_open', 8.0, True)) self.assertEqual([s['opening'] for s in forecast[:3]], [1.0, 1.0, 0.0]) self.assertEqual((inputs['known'], inputs['affected']), (False, True)) # The event from before the restart continues; the downtime counts as unknown. self.assertEqual(len(tracker.events), 0) event = tracker._open_event['bed'] self.assertEqual((event['opened'], event['opened_confirmed']), (before._open_event['bed']['opened'], True)) self.assertEqual(event['sensor_unknown_minutes'], 12.0) # 7 minutes down + one unknown poll # Not assumed open forever: past the bound the forecast assumes it closed. for _ in range(3): _, forecast, assumed = self.poll(tracker, 'unavailable') self.assertEqual((assumed['open_minutes'], forecast[0]['opening']), (0.0, 0.0)) self.assertGreater(forecast[0]['recovery'], 0.0) # Back as closed: closed at an unknown time, so an unconfirmed close and recovery. inputs, _, _ = self.poll(tracker, 'off', changed_ago=30) event = tracker.events[-1] self.assertEqual((event['closed_confirmed'], event['duration_minutes']), (False, None)) self.assertEqual(event['duration_min_minutes'], 8.3) # opened 3.3 min before a poll, seen open 5 min later self.assertEqual((inputs['known'], inputs['affected']), (False, True)) def test_restart_with_saved_closed_window(self): _, saved = self.saved_with('off', 'off') tracker = wm.VentilationTracker(self.WINDOWS, saved=saved) inputs, forecast, assumed = self.poll(tracker, 'unavailable') self.assertEqual((assumed['basis'], assumed['uncertain'], forecast[0]['opening']), ('last_known_closed', True, 0.0)) self.assertEqual((inputs['known'], len(tracker.events)), (False, 0)) # It reports open: opened during the unknown period, so not a confirmed opening. self.poll(tracker, 'on', changed_ago=60) self.assertFalse(tracker._open_event['bed']['opened_confirmed']) def test_restart_with_the_same_state_again_is_a_known_state(self): _, saved = self.saved_with('off', 'off') tracker = wm.VentilationTracker(self.WINDOWS, saved=saved) self.poll(tracker, 'off') # first interval after restart: unknown inputs, _, assumed = self.poll(tracker, 'off') self.assertEqual((inputs['known'], inputs['affected'], assumed['basis']), (True, False, 'observed')) def test_saved_state_of_another_sensor_or_polarity_is_ignored(self): _, saved = self.saved_with('off', 'off', 'on', 'on') for changed in ({'sensor': 'binary_sensor.other'}, {'inverted': True}): with self.subTest(changed=changed): windows = wm.build_windows([(window('w', **changed), None)], ['bed']) tracker = wm.VentilationTracker(windows, saved=saved) self.assertEqual((tracker._memory, tracker._open_event), ({}, {})) corrupt = {'events': [], 'state': {'windows': {'w': {'sensor': 'binary_sensor.w', 'last_known': 'yes', 'seen': 'x'}}, 'zones': {'bed': 5}}} self.assertEqual(wm.VentilationTracker(self.WINDOWS, saved=corrupt)._memory, {}) def test_mixed_windows_each_get_their_own_assumption(self): windows = wm.build_windows([(window('a'), None), (window('b', opening_factor=0.5), None), (window('c', zone='living'), None)], ['bed', 'living']) tracker = wm.VentilationTracker(windows) t = 0.0 for a, b, c in (('off', 'off', 'off'), ('off', 'on', 'off'), ('on', 'on', 'unavailable'), ('on', 'unavailable', 'unavailable'), ('on', 'unavailable', 'unavailable'), ('on', 'unavailable', 'unavailable'), ('on', 'unavailable', 'unavailable')): t += 300 tracker.update(t, {'a': (a, t - 200), 'b': (b, t - 200), 'c': (c, t - 200)}, {'bed': 20.0, 'living': 21.0}, WEATHER) # b was last seen open 20 minutes ago: no longer assumed open; a is observed open. self.assertEqual(tracker.assumption(windows[1], 15, t)['open_minutes'], 0.0) self.assertEqual(tracker.forecast('bed', 4, 15, WEATHER, t)[0]['opening'], 1.0) self.assertEqual(tracker.assumption(windows[2], 15, t)['basis'], 'last_known_closed') # An unknown window in one zone does not stop another zone with known windows. tracker2 = wm.VentilationTracker(windows) for t in (300.0, 600.0): tracker2.update(t, {'a': ('off', 0), 'b': ('off', 0), 'c': ('unavailable', 0)}, {'bed': 20.0, 'living': 21.0}, WEATHER) self.assertEqual((tracker2.zone_inputs('bed', WEATHER)['known'], tracker2.zone_inputs('living', WEATHER)['known']), (True, False)) def test_changes_worth_saving_are_flagged(self): tracker = wm.VentilationTracker(self.WINDOWS) self.poll(tracker, 'off') self.assertTrue(tracker.changed) # first reading tracker.changed = False self.poll(tracker, 'off') 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