Files
buzeraandClaude Opus 5.5 a44163a57d Report how well the base coefficients are identified
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>
2026-10-27 03:49:13 +02:00

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()