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>
256 lines
15 KiB
Python
256 lines
15 KiB
Python
"""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']))
|