buzeraandClaude Opus 5.5
43915ba810
Status showed accepted sample counts but not why learning was slow. Each room's model now counts samples it rejected (outliers, implausible inputs), persisted as an optional rejected_count key and shown as models.<room>.rejected_samples. Status also counts every poll outcome since start (sample_results_since_start). Add tests that a model in the exact shape 0.2.1 wrote restores with its coefficients, and that today's saved model still has the keys and types 0.2.1's restore code reads, so rolling back keeps the learned model. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
169 lines
8.3 KiB
Python
169 lines
8.3 KiB
Python
"""Offline tests: run python -m unittest discover -s tests -v."""
|
|
import importlib.util
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
path = Path(__file__).resolve().parents[1] / 'custom_components' / 'heating_mpc' / 'physics.py'
|
|
spec = importlib.util.spec_from_file_location('heating_mpc_physics', path)
|
|
m = importlib.util.module_from_spec(spec)
|
|
import sys
|
|
sys.modules[spec.name] = m
|
|
spec.loader.exec_module(m)
|
|
|
|
|
|
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_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['predicted_end']['r'], round(temps['r'], 3))
|
|
|
|
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_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()
|