buzeraandClaude Opus 5.5
e62798dc4f
A single 5-minute +700 W spike moves the room's thermal response by at most 0.2 kW and leaves the baseline at the persistent load. In calm air the wind terms vanish while natural ventilation remains. Extended and ventilation coefficients and heat-source baselines survive a restart unchanged (no hour of baseline re-learning), and enabling an extra ventilation term later keeps everything learned for the others. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
147 lines
8.2 KiB
Python
147 lines
8.2 KiB
Python
"""Electrical heat sources: units, overlap, baseline, thermal response, forecasts."""
|
|
import sys
|
|
import types
|
|
import unittest
|
|
from math import exp
|
|
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 heat_sources as hs # noqa: E402
|
|
|
|
UPS = {'key': 'ups', 'name': 'UPS', 'sensor': 'sensor.ups_power', 'zone': 'office'}
|
|
PC = {'key': 'pc', 'name': 'Gaming PC', 'sensor': 'sensor.pc_power', 'zone': 'office', 'included_in': 'ups'}
|
|
|
|
|
|
def sources(*items, zones=('office', 'living')):
|
|
return hs.build_sources(((item, None) for item in items), zones)
|
|
|
|
|
|
def run(tracker, powers, start=0.0, step=300.0):
|
|
"""Feed one reading per poll; return the tracker states after each poll."""
|
|
states = []
|
|
for k, readings in enumerate(powers):
|
|
tracker.update(start + k * step, readings)
|
|
states.append({key: dict(info) for key, info in tracker.state.items()})
|
|
return states
|
|
|
|
|
|
class PowerTests(unittest.TestCase):
|
|
def test_units_are_normalized_and_non_heat_units_rejected(self):
|
|
self.assertEqual(hs.normalize_power('600', 'W'), 600)
|
|
self.assertEqual(hs.normalize_power(0.6, ' kW '), 600)
|
|
self.assertEqual(hs.normalize_power(600000, 'mW'), 600)
|
|
for raw, unit in (('600', 'VA'), ('35', '%'), ('600', None), ('600', ''), ('-5', 'W'),
|
|
('nan', 'W'), ('abc', 'W'), ('30000', 'W'), ('600', 7)):
|
|
with self.subTest(raw=raw, unit=unit):
|
|
self.assertIsNone(hs.normalize_power(raw, unit))
|
|
|
|
def test_build_sources_validates_zones_duplicates_and_inclusion(self):
|
|
built = sources(UPS, PC, {**UPS, 'key': 'ups2'}, {**PC, 'key': 'pc2', 'sensor': 'sensor.x', 'zone': 'gone'},
|
|
{'key': 'nas', 'sensor': 'sensor.nas', 'zone': 'office', 'included_in': 'pc'},
|
|
{'key': 'tv', 'sensor': 'sensor.tv', 'zone': 'living', 'included_in': 'missing'})
|
|
self.assertEqual([(s.key, s.included_in) for s in built], [('ups', None), ('pc', 'ups'), ('nas', None),
|
|
('tv', None)])
|
|
|
|
def test_included_child_is_not_double_counted(self):
|
|
built = sources(UPS, PC)
|
|
self.assertEqual(hs.effective_power(built, {'ups': 600, 'pc': 400}), {'ups': 200, 'pc': 400})
|
|
# Without the child reading the split is unknown; timing noise is clamped.
|
|
self.assertEqual(hs.effective_power(built, {'ups': 600, 'pc': None}), {'ups': None, 'pc': None})
|
|
self.assertEqual(hs.effective_power(built, {'ups': 380, 'pc': 400})['ups'], 0)
|
|
# A disabled child stays part of its parent's measurement.
|
|
built = sources(UPS, {**PC, 'enabled': False})
|
|
self.assertEqual(hs.effective_power(built, {'ups': 600, 'pc': None}), {'ups': 600})
|
|
|
|
def test_baseline_is_the_persistent_load_not_the_mean(self):
|
|
estimator = hs.BaselineEstimator()
|
|
for _ in range(11):
|
|
estimator.add(200)
|
|
self.assertIsNone(estimator.value()) # less than an hour of data
|
|
for k in range(288):
|
|
estimator.add(600 if k % 10 < 3 else 200) # gaming 30 % of the time
|
|
self.assertEqual(estimator.value(), 200)
|
|
self.assertEqual(len(estimator.samples), 288)
|
|
self.assertEqual(hs.BaselineEstimator([200, 'x', -5, float('nan'), 300]).samples, hs.deque([200.0, 300.0]))
|
|
|
|
def test_response_and_forecasts(self):
|
|
filt = hs.ResponseFilter(15)
|
|
self.assertEqual(filt.advance(0.0, 0), 0.0)
|
|
self.assertAlmostEqual(filt.advance(0.4, 900), 0.4 * (1 - exp(-1)))
|
|
self.assertEqual(hs.forecast_internal(0.4, 3, None), [0.4] * 3)
|
|
self.assertEqual(hs.forecast_internal(0.4, 2, 0), [0.0, 0.0])
|
|
decay = hs.forecast_internal(0.4, 12, 30)
|
|
self.assertAlmostEqual(decay[5], 0.2) # one half-life after 30 minutes
|
|
response = hs.forecast_response(0.0, 0.4, 24, 15, 30)
|
|
self.assertLess(max(response), 0.4) # the PC is not assumed to run on at full power
|
|
self.assertLess(response[-1], max(response))
|
|
|
|
|
|
class TrackerTests(unittest.TestCase):
|
|
def test_gaming_session_drives_zone_input_with_lag_and_settling(self):
|
|
tracker = hs.HeatSourceTracker(sources(UPS))
|
|
idle = [{'ups': 200.0}] * 12
|
|
states = run(tracker, idle + [{'ups': 600.0}] * 6 + idle[:6])
|
|
self.assertIsNone(states[10]['ups']['variable_kw']) # baseline needs an hour of data
|
|
self.assertEqual(states[11]['ups']['baseline_w'], 200)
|
|
# The poll that sees 600 W has not yet had any extra heat: power is held over the interval.
|
|
self.assertEqual((states[12]['ups']['variable_kw'], states[12]['ups']['response_kw']), (0.4, 0.0))
|
|
self.assertAlmostEqual(states[13]['ups']['response_kw'], 0.4 * (1 - exp(-1 / 3)))
|
|
self.assertGreater(states[18]['ups']['response_kw'], states[19]['ups']['response_kw']) # PC off: decays
|
|
self.assertGreater(states[19]['ups']['response_kw'], 0.1) # but the room still holds the heat
|
|
self.assertEqual(tracker.zone_input('office'), states[-1]['ups']['response_kw'])
|
|
self.assertEqual(tracker.zone_input('living'), 0.0)
|
|
|
|
def test_short_spike_barely_moves_response_or_baseline(self):
|
|
tracker = hs.HeatSourceTracker(sources(UPS), {'baselines': {'ups': [200.0] * 24}})
|
|
states = run(tracker, [{'ups': 200.0}] * 3 + [{'ups': 900.0}] + [{'ups': 200.0}] * 4)
|
|
peak = max(s['ups']['response_kw'] for s in states)
|
|
self.assertAlmostEqual(peak, 0.7 * (1 - exp(-1 / 3))) # one 5-minute interval of +700 W
|
|
self.assertLess(peak, 0.2)
|
|
self.assertEqual(states[-1]['ups']['baseline_w'], 200)
|
|
|
|
def test_unknown_power_resets_until_settled_again(self):
|
|
tracker = hs.HeatSourceTracker(sources(UPS), {'baselines': {'ups': [200.0] * 24}})
|
|
run(tracker, [{'ups': 200.0}] * 7)
|
|
self.assertEqual(tracker.zone_input('office'), 0.0) # saved baseline: usable immediately
|
|
tracker.update(7 * 300, {'ups': 200.0})
|
|
tracker.update(8 * 300, {'ups': None})
|
|
self.assertFalse(tracker.state['ups']['settled'] and tracker.state['ups']['response_kw'] is None)
|
|
tracker.update(9 * 300, {'ups': 200.0})
|
|
self.assertIsNone(tracker.zone_input('office')) # restarted: 2 time constants to settle
|
|
run(tracker, [{'ups': 200.0}] * 6, start=10 * 300)
|
|
self.assertEqual(tracker.zone_input('office'), 0.0)
|
|
|
|
def test_manual_baseline_fraction_and_round_trip(self):
|
|
built = sources({**UPS, 'baseline_mode': 'manual', 'baseline_w': 150, 'heat_fraction': 0.5})
|
|
tracker = hs.HeatSourceTracker(built)
|
|
tracker.update(0, {'ups': 350.0})
|
|
self.assertEqual(tracker.state['ups']['variable_kw'], 0.1)
|
|
restored = hs.HeatSourceTracker(built, tracker.as_dict())
|
|
self.assertEqual(list(restored._baselines['ups'].samples), [350.0])
|
|
|
|
def test_form_validation(self):
|
|
others = sources(UPS)
|
|
cases = {
|
|
'source_name_required': {'name': ' '},
|
|
'sensor_in_other_source': {'name': 'PC', 'sensor': 'sensor.ups_power'},
|
|
'invalid_parent_source': {'name': 'NAS', 'sensor': 'sensor.nas', 'included_in': 'missing'},
|
|
'baseline_required': {'name': 'PC', 'sensor': 'sensor.pc', 'baseline_mode': 'manual'},
|
|
None: {'name': 'PC', 'sensor': 'sensor.pc', 'included_in': 'ups'},
|
|
}
|
|
for expected, data in cases.items():
|
|
with self.subTest(expected=expected):
|
|
self.assertEqual(hs.validate_source(data, others), expected)
|
|
nested = sources(UPS, PC)
|
|
self.assertEqual(hs.validate_source({'name': 'NAS', 'sensor': 'sensor.nas', 'included_in': 'pc'}, nested),
|
|
'invalid_parent_source')
|
|
self.assertEqual(hs.validate_source({'key': 'ups', 'name': 'UPS', 'sensor': 'sensor.ups_power',
|
|
'included_in': 'other'}, [s for s in nested if s.key != 'ups']
|
|
+ sources({'key': 'other', 'sensor': 'sensor.o', 'zone': 'office'})),
|
|
'source_has_children')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|