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