The base layer (loss, heat, bias) showed values only, so a zone with weeks of samples looked equally trustworthy whether or not the data ever varied its inputs. Each base coefficient now reports a confidence from its own posterior variance, as the extended and ventilation layers already did: it stays at 0 for `heat` while the relay never cycles and below 0.8 for `loss` and `bias` while the indoor-outdoor difference is steady. Readiness is unchanged. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
205 lines
10 KiB
Python
205 lines
10 KiB
Python
"""Offline tests: run python -m unittest discover -s tests -v."""
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
import sys
|
|
import types
|
|
|
|
COMPONENTS = Path(__file__).resolve().parents[1] / 'custom_components' / 'heating_mpc'
|
|
# Fake only the package container, to avoid importing Home Assistant from __init__.
|
|
sys.modules.setdefault('heating_mpc', types.ModuleType('heating_mpc'))
|
|
sys.modules['heating_mpc'].__path__ = [str(COMPONENTS)]
|
|
from heating_mpc import physics as m # noqa: E402
|
|
|
|
|
|
class ModelTests(unittest.TestCase):
|
|
def test_fitting_and_restore(self):
|
|
learn = m.Learner()
|
|
t = 20.0
|
|
lag = 0.0
|
|
for step in range(1000):
|
|
heater = int(step % 96 < 35)
|
|
lag = lag + m.HEAT_LAG * (heater - lag)
|
|
value = t + 0.003 * (8 - t) + 0.055 * lag + 0.001
|
|
learn.learn(t, value, 8, lag)
|
|
t = value
|
|
self.assertGreaterEqual(learn.count, 950)
|
|
self.assertAlmostEqual(learn.theta[0], 0.003, delta=0.003)
|
|
self.assertAlmostEqual(learn.theta[1], 0.055, delta=0.03)
|
|
restored = m.Learner.from_dict(learn.as_dict())
|
|
self.assertEqual(restored.count, learn.count)
|
|
self.assertEqual(restored.theta, learn.theta)
|
|
|
|
def test_optimize_warm_room_prefers_off(self):
|
|
zones = {'living': 23.5, 'bedroom': 23.4}
|
|
learners = {'living': m.Learner(), 'bedroom': m.Learner()}
|
|
out = m.optimize(zones, 12, 0, 22, {'living': 2, 'bedroom': 1},
|
|
learners, False, 3600, horizon_minutes=60)
|
|
self.assertFalse(out['proposal'])
|
|
self.assertEqual(len(out['plan_15min']), 4)
|
|
|
|
def test_optimize_cold_room_prefers_heat(self):
|
|
zones = {'living': 19.5, 'bedroom': 19.9}
|
|
learners = {'living': m.Learner(), 'bedroom': m.Learner()}
|
|
out = m.optimize(zones, 4, 0, 22, {'living': 2, 'bedroom': 1},
|
|
learners, False, 3600, horizon_minutes=60)
|
|
self.assertTrue(out['proposal'])
|
|
|
|
def test_zone_target_offsets_shift_comfort(self):
|
|
learners = {'r': m.Learner()}
|
|
args = ({'r': 20.8}, 8, 0, 22, {'r': 1}, learners, False, 3600)
|
|
self.assertTrue(m.optimize(*args, horizon_minutes=60)['proposal'])
|
|
self.assertFalse(m.optimize(*args, horizon_minutes=60, target_offsets={'r': -1.5})['proposal'])
|
|
with self.assertRaises(ValueError):
|
|
m.optimize(*args, horizon_minutes=60, target_offsets={'r': float('nan')})
|
|
|
|
def test_minimum_duration_on_and_off(self):
|
|
learners = {'r': m.Learner()}
|
|
# Could not switch OFF during first 15 minutes if just turned ON.
|
|
out = m.optimize({'r': 24}, 12, 1, 22, {'r': 1}, learners, True, 120,
|
|
minimum_on_minutes=15, horizon_minutes=60)
|
|
self.assertTrue(out['proposal'])
|
|
# Could not switch ON during first 10 minutes if just turned OFF.
|
|
out = m.optimize({'r': 18}, 0, 0, 22, {'r': 1}, learners, False, 120,
|
|
minimum_off_minutes=10, horizon_minutes=60)
|
|
self.assertFalse(out['proposal'])
|
|
|
|
def test_invalid_room_weights_rejected(self):
|
|
with self.assertRaises(ValueError):
|
|
m.optimize({'r': 20}, 10, 0, 22, {'r': 0}, {'r': m.Learner()},
|
|
False, 1000, horizon_minutes=60)
|
|
|
|
def test_short_horizon_does_not_claim_60_minute_prediction(self):
|
|
out = m.optimize({'r': 20}, 5, 0, 22, {'r': 1}, {'r': m.Learner()},
|
|
False, 3600, horizon_minutes=30)
|
|
self.assertEqual(out['predicted_60m'], {})
|
|
self.assertIn('r', out['predicted_end'])
|
|
|
|
def test_prediction_matches_replay_at_60_minutes(self):
|
|
learners = {'r': m.Learner()}
|
|
out = m.optimize({'r': 20}, 5, .4, 22, {'r': 1}, learners, False, 3600,
|
|
horizon_minutes=90)
|
|
temps, lag = {'r': 20}, .4
|
|
for index, bit in enumerate(out['plan_15min']):
|
|
for _ in range(3):
|
|
temps, lag = m.simulate_step(temps, 5, lag, bit, learners)
|
|
if index == 3:
|
|
self.assertEqual(out['predicted_60m']['r'], round(temps['r'], 3))
|
|
self.assertEqual(out['trajectory_15min']['r'][index], round(temps['r'], 3))
|
|
self.assertEqual(out['predicted_end']['r'], round(temps['r'], 3))
|
|
self.assertEqual(len(out['trajectory_15min']['r']), 6)
|
|
|
|
def test_jitter_normalized_and_rejection_counted(self):
|
|
nominal, delayed = m.Learner(), m.Learner()
|
|
self.assertTrue(nominal.learn(20, 20.1, 5, .5, command=True))
|
|
self.assertTrue(delayed.learn(20, 20.11, 5, .5, 330, True))
|
|
for a, b in zip(nominal.theta, delayed.theta):
|
|
self.assertAlmostEqual(a, b)
|
|
self.assertFalse(delayed.learn(20, 25, 5, .5, command=False))
|
|
self.assertFalse(delayed.learn(20, 20.1, 5, 1.5, command=False))
|
|
self.assertEqual((delayed.count, delayed.on_count, delayed.off_count, delayed.rejected_count),
|
|
(1, 1, 0, 2))
|
|
self.assertEqual(m.Learner.from_dict(delayed.as_dict()).rejected_count, 2)
|
|
|
|
def test_corrupt_model_restore_is_atomic(self):
|
|
for changes in ({'theta': [-1, .1, 0]}, {'p': [[-1, 0, 0], [0, 1, 0], [0, 0, 1]]},
|
|
{'count': float('inf')}, {'count': '5'}, {'error_sum': float('nan')},
|
|
{'off_count': 1}, {'theta': [float('nan'), 0, 0]}, {'p': None},
|
|
{'p': [[1, .5, 0], [0, 1, 0], [0, 0, 1]]}, {'rejected_count': -1},
|
|
{'rejected_count': '3'}):
|
|
with self.subTest(changes=changes):
|
|
data = m.Learner().as_dict()
|
|
data.update(changes)
|
|
self.assertEqual(m.Learner.from_dict(data), m.Learner())
|
|
for bad in (None, [], 'broken'):
|
|
self.assertEqual(m.Learner.from_dict(bad), m.Learner())
|
|
|
|
def test_nonfinite_optimizer_inputs_rejected(self):
|
|
for key, value in (('weights', {'r': float('nan')}), ('effective_heat', float('nan')),
|
|
('switching_cost', float('inf')), ('seconds_since_change', -1),
|
|
('minimum_on_minutes', -1), ('comfort_band', float('nan'))):
|
|
args = dict(temps={'r': 20}, outside=5, effective_heat=0, target=22,
|
|
weights={'r': 1}, learners={'r': m.Learner()}, is_on=False,
|
|
seconds_since_change=3600, horizon_minutes=60)
|
|
args[key] = value
|
|
with self.subTest(key=key), self.assertRaises(ValueError):
|
|
m.optimize(**args)
|
|
|
|
def test_covariance_stays_bounded_without_heat_excitation(self):
|
|
learn = m.Learner()
|
|
t = 22.0
|
|
for _ in range(30000): # about 104 days with the relay off
|
|
value = t + 0.003 * (15 - t)
|
|
self.assertTrue(learn.learn(t, value, 15, 0.0, command=False))
|
|
t = value
|
|
for i in range(3):
|
|
self.assertLessEqual(learn.p[i][i], m.PRIOR_P[i][i])
|
|
for j in range(3):
|
|
self.assertEqual(learn.p[i][j], learn.p[j][i])
|
|
restored = m.Learner.from_dict(learn.as_dict())
|
|
self.assertEqual(restored.count, 30000)
|
|
self.assertEqual(restored.p, learn.p)
|
|
|
|
def test_confidence_reflects_excitation_not_sample_count(self):
|
|
# Two weeks of samples, relay always off, a steady 15 degC difference to outside.
|
|
steady = m.Learner()
|
|
for _ in range(4000):
|
|
steady.learn(20.0, 20.0 + 0.003 * -15 + 0.001, 5.0, 0.0, command=False)
|
|
confidence = steady.confidence()
|
|
self.assertEqual(steady.count, 4000)
|
|
self.assertEqual(confidence['heat'], 0) # never exercised
|
|
self.assertLess(confidence['loss'], 0.8) # not separable from bias
|
|
self.assertLess(confidence['bias'], 0.8)
|
|
# Fewer samples, with the relay cycling and a cold and a mild spell each day.
|
|
varied = m.Learner()
|
|
t, lag = 20.0, 0.0
|
|
for step in range(1200):
|
|
heater = int(step % 96 < 35)
|
|
lag = lag + m.HEAT_LAG * (heater - lag)
|
|
outside = -2 + 14 * ((step // 144) % 2)
|
|
value = t + 0.003 * (outside - t) + 0.055 * lag + 0.001
|
|
varied.learn(t, value, outside, lag, command=bool(heater))
|
|
t = value
|
|
self.assertTrue(all(value > 0.8 for value in varied.confidence().values()), varied.confidence())
|
|
# Recalibration starts confidence over.
|
|
varied.recalibrate()
|
|
self.assertEqual(set(varied.confidence().values()), {0})
|
|
|
|
def test_restore_tolerates_round_off_and_bounds_wound_up_covariance(self):
|
|
# Shape of a v0.2.1 model after a summer: huge heat variance, tiny asymmetry.
|
|
data = m.Learner(count=40000, off_count=40000).as_dict()
|
|
data['p'] = [[0.03, 1.0, 1e-3], [1.0 + 1e-7, 8.7e12, 0.0], [1e-3, 0.0, 0.0012]]
|
|
restored = m.Learner.from_dict(data)
|
|
self.assertEqual(restored.count, 40000)
|
|
self.assertEqual(restored.p[1][1], m.PRIOR_P[1][1])
|
|
self.assertEqual(restored.p[0][0], m.PRIOR_P[0][0])
|
|
self.assertEqual(restored.p[0][1], restored.p[1][0])
|
|
|
|
def test_lag_helpers(self):
|
|
self.assertAlmostEqual(m.advance_lag(0, True, 300), m.HEAT_LAG)
|
|
self.assertAlmostEqual(m.advance_lag(1, False, 1e9), 0)
|
|
self.assertEqual(m.lag_from_time_in_state(True, 0), (0.5, 0.5))
|
|
lag, half_width = m.lag_from_time_in_state(False, 3600)
|
|
self.assertAlmostEqual(lag, half_width)
|
|
self.assertLess(half_width, 0.03)
|
|
# Any starting lag in [0, 1] ends inside the stated interval.
|
|
for start in (0.0, 0.3, 1.0):
|
|
for command in (False, True):
|
|
lag, half_width = m.lag_from_time_in_state(command, 900)
|
|
self.assertLessEqual(abs(m.advance_lag(start, command, 900) - lag), half_width + 1e-12)
|
|
|
|
def test_relay_segments(self):
|
|
seg = m.relay_segments
|
|
self.assertEqual(seg(False, 0, [], 300), [(False, 300)])
|
|
self.assertEqual(seg(False, 0, [(60, 'on'), (200, 'off')], 300),
|
|
[(False, 60), (True, 140), (False, 100)])
|
|
self.assertEqual(seg(True, 0, [(300, 'off')], 300), [(True, 300), (False, 0)])
|
|
for changes in ([(60, 'unavailable')], [(60, 'off')], [(200, 'on'), (60, 'off')],
|
|
[(-1, 'on')], [(301, 'on')]):
|
|
with self.subTest(changes=changes):
|
|
self.assertIsNone(seg(False, 0, changes, 300))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|