Model electrical heat sources (UPS, gaming PC) per zone

Electrical equipment heats its room, and a gaming session can add
hundreds of watts, which the 3-parameter model attributed to the boiler
or the bias term.

heat_sources.py adds heat sources as config subentries ("Add heat
source"): an active-power sensor (W/kW; VA and % rejected), the zone,
the share that becomes heat, an automatic baseline (20th percentile of
24 h, so the persistent UPS load is not double counted with the bias
term) or a manual one, an optional "included in" parent (the UPS total
minus a separately measured PC, one level deep) and a first-order
thermal response time. The heat during each interval is the power held
from its start, like the radiator lag. After startup or an outage a
source's response must run two time constants before it counts as
settled.

zone_model.py layers each zone: the base model is kept unchanged as the
fallback and comparison, and an extended FeatureLearner adds "internal"
(kW above baseline after the response) and neighbour coupling for
adjacent zones. It learns only intervals where all its inputs are
known, and is used for forecasts once as trained as the base. The
optimizer accepts a multi-zone step model; forecasts let extra power
fade to baseline with a configurable half-life (default 30 min).

On a synthetic office with thermostat cycling and gaming sessions the
extended model recovers the electrical, boiler and loss coefficients
within 10 %. Status lists each source's power, effective power,
baseline, variable and responding heat; models show the extended
coefficients with confidence. Saved models gain optional zone_models
and heat_sources keys.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

tests/test_heat_sources.py

@@ -0,0 +1,138 @@
+"""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_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()