From 00247f8dfe75eb0ab47323c2e3f184c0d58adde6 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 8 Jun 2026 18:24:03 +0000 Subject: [PATCH 1/7] Replace night_surplus_wh with pv_start_battery_wh and forecast_min_battery_wh night_surplus_wh was semantically ambiguous and computed a net-sum approximation that ignored battery floor/ceiling clamping across multiple charge/discharge cycles. New metrics, both using slot-by-slot simulation with proper MIN/MAX SOC: pv_start_battery_wh: Battery level (above MIN_SOC) at the next net-charging point, defined as the first forecast slot where net_consumption < 0 (PV exceeds load). This is the relevant indicator for evening/night WP decisions. forecast_min_battery_wh: Minimum usable battery level over the entire forecast horizon. Tracks the trough of the simulation, not the final value. 0 = battery expected to hit MIN_SOC at some point -> be conservative. Both are published via MQTT with HA auto-discovery sensors. https://claude.ai/code/session_01AbZuZ73iEQMiwTkwy3Mt1h --- src/batcontrol/core.py | 111 ++++++------- src/batcontrol/mqtt_api.py | 43 +++-- tests/batcontrol/test_night_surplus.py | 212 +++++++++++++------------ 3 files changed, 186 insertions(+), 180 deletions(-) diff --git a/src/batcontrol/core.py b/src/batcontrol/core.py index 221dad1..7bb9e57 100644 --- a/src/batcontrol/core.py +++ b/src/batcontrol/core.py @@ -688,11 +688,16 @@ def run(self): ) self.mqtt_api.publish_solar_active(solar_active) self.mqtt_api.publish_solar_surplus(surplus_wh) - night_surplus_wh = self._compute_night_surplus( - production, consumption, + pv_start_wh = self._compute_pv_start_battery( + net_consumption, calc_input.stored_usable_energy, calc_input.free_capacity ) - self.mqtt_api.publish_night_surplus(night_surplus_wh) + self.mqtt_api.publish_pv_start_battery(pv_start_wh) + forecast_min_wh = self._compute_forecast_min_battery( + net_consumption, + calc_input.stored_usable_energy, calc_input.free_capacity + ) + self.mqtt_api.publish_forecast_min_battery(forecast_min_wh) if self.discharge_blocked and not \ self.general_logic.is_discharge_always_allowed_soc(self.get_SOC()): @@ -922,77 +927,55 @@ def _compute_solar_active_and_surplus( ) return solar_active, surplus_wh - def _compute_night_surplus( + def _compute_pv_start_battery( self, - production: np.ndarray, - consumption: np.ndarray, + net_consumption: np.ndarray, stored_usable_energy: float, free_capacity: float) -> float: - """Compute expected battery surplus at the start of the next production window. - - Answers the question: after tonight's discharge, how much charge will remain - in the battery when tomorrow's solar production starts? - - The calculation intentionally projects through the entire first production - window (including any solar charging) to obtain the battery level at production - end. From there it subtracts overnight consumption to arrive at the battery - level at the next morning's production start: - - battery_at_production_end - night_consumption - - When solar is currently inactive (e.g. early morning), this means net_delta - covers the bridge discharge AND the upcoming solar charging. This is deliberate: - stopping at production_start would give the battery level at dawn of today, not - at dusk — which is the wrong baseline for the overnight calculation. + """Battery level (Wh above MIN_SOC) at the start of the next net-charging window. - If no second production window exists within the forecast horizon, - night_consumption covers the remaining forecast slots (best available proxy). + Simulates slot-by-slot discharge until the first slot where + net_consumption < 0 (solar production exceeds household consumption). + That crossing point is when the battery transitions from discharging to + charging and is the most meaningful reference for overnight planning. - Returns 0.0 if no solar production window exists in the forecast at all. + Returns 0.0 if battery reaches MIN_SOC before that point, or if no + net-charging slot exists in the forecast at all. """ - net_consumption = consumption - production - - # Find start and end of the first production window - production_start: Optional[int] = None - production_end: Optional[int] = None - for i, p in enumerate(production): - if p > 0: - if production_start is None: - production_start = i - production_end = i - elif production_start is not None: - break - - if production_start is None: - return 0.0 - - end_idx = production_end + 1 # type: ignore[operator] - - # Project battery level at end of first production window (clamped to [0, max]) - net_delta = float(-np.sum(net_consumption[0:end_idx])) - battery_at_end = stored_usable_energy + min( - free_capacity, max(-stored_usable_energy, net_delta) - ) - - # Find the start of the next (second) production window after the night gap - next_production_start: Optional[int] = None - for i in range(end_idx, len(production)): - if production[i] > 0: - next_production_start = i - break - night_end = next_production_start if next_production_start is not None \ - else len(production) - - night_consumption_wh = max(0.0, float(np.sum(net_consumption[end_idx:night_end]))) + battery = stored_usable_energy + max_battery = stored_usable_energy + free_capacity + for net in net_consumption: + if net < 0: + return battery + battery = max(0.0, min(max_battery, battery - net)) + return 0.0 + + def _compute_forecast_min_battery( + self, + net_consumption: np.ndarray, + stored_usable_energy: float, + free_capacity: float) -> float: + """Minimum battery level (Wh above MIN_SOC) over the entire forecast horizon. - night_surplus_wh = max(0.0, battery_at_end - night_consumption_wh) + Simulates slot-by-slot with proper floor (MIN_SOC = 0 usable) and ceiling + (MAX_SOC = stored_usable + free_capacity) clamping at each step. + Returns the lowest point reached during the simulation. + A value of 0 means the battery is expected to hit MIN_SOC at some point + in the forecast — a signal to be conservative with flexible loads. + """ + battery = stored_usable_energy + max_battery = stored_usable_energy + free_capacity + min_battery = stored_usable_energy + for net in net_consumption: + battery = max(0.0, min(max_battery, battery - net)) + if battery < min_battery: + min_battery = battery logger.debug( - 'Night surplus: %.1f Wh (battery_at_production_end=%.1f Wh,' - ' night_consumption=%.1f Wh, night_slots=%d)', - night_surplus_wh, battery_at_end, night_consumption_wh, night_end - end_idx + 'Forecast min battery: %.1f Wh (stored=%.1f Wh, slots=%d)', + min_battery, stored_usable_energy, len(net_consumption) ) - return night_surplus_wh + return min_battery def set_stored_energy(self, stored_energy) -> None: """ Set the stored energy in Wh """ diff --git a/src/batcontrol/mqtt_api.py b/src/batcontrol/mqtt_api.py index b4c511d..a504c9e 100644 --- a/src/batcontrol/mqtt_api.py +++ b/src/batcontrol/mqtt_api.py @@ -28,7 +28,8 @@ - /control_source: source that last selected the current control state (api or optimizer) - /solar_surplus_wh: expected solar surplus energy in Wh (>0 means usable surplus available) - /solar_active: bool indicating whether solar is currently producing (slot 0 > 0) -- /night_surplus_wh: expected battery surplus in Wh at start of next production window (>0 means leftover charge after overnight discharge) +- /pv_start_battery_wh: battery level in Wh (above MIN_SOC) at the next net-charging point (when PV first exceeds consumption) +- /forecast_min_battery_wh: minimum battery level in Wh (above MIN_SOC) over the entire forecast horizon (0 = shortage expected) The following statistical arrays are published as JSON arrays: - /FCST/production: forecasted production in W @@ -499,16 +500,28 @@ def publish_solar_surplus(self, surplus_wh: float) -> None: f'{surplus_wh:.1f}' ) - def publish_night_surplus(self, surplus_wh: float) -> None: - """ Publish the expected battery surplus at the start of the next production window. - /night_surplus_wh - Positive values mean the battery will still hold charge (above MIN_SOC) - when solar production resumes the next morning. + def publish_pv_start_battery(self, battery_wh: float) -> None: + """ Publish the battery level at the next net-charging point. + /pv_start_battery_wh + Energy in Wh above MIN_SOC at the moment PV production first exceeds + household consumption. 0 if battery hits MIN_SOC before that point. """ if self.client.is_connected(): self.client.publish( - self.base_topic + '/night_surplus_wh', - f'{surplus_wh:.1f}' + self.base_topic + '/pv_start_battery_wh', + f'{battery_wh:.1f}' + ) + + def publish_forecast_min_battery(self, battery_wh: float) -> None: + """ Publish the minimum battery level over the entire forecast horizon. + /forecast_min_battery_wh + Energy in Wh above MIN_SOC at the trough of the slot-by-slot simulation. + 0 means the battery is expected to hit MIN_SOC at some point. + """ + if self.client.is_connected(): + self.client.publish( + self.base_topic + '/forecast_min_battery_wh', + f'{battery_wh:.1f}' ) def publish_solar_active(self, active: bool) -> None: @@ -922,12 +935,20 @@ def send_mqtt_discovery_messages(self) -> None: self.base_topic + "/solar_surplus_wh") self.publish_mqtt_discovery_message( - "Night Surplus", - "batcontrol_night_surplus_wh", + "PV Start Battery", + "batcontrol_pv_start_battery_wh", + "sensor", + "energy", + "Wh", + self.base_topic + "/pv_start_battery_wh") + + self.publish_mqtt_discovery_message( + "Forecast Min Battery", + "batcontrol_forecast_min_battery_wh", "sensor", "energy", "Wh", - self.base_topic + "/night_surplus_wh") + self.base_topic + "/forecast_min_battery_wh") self.publish_mqtt_discovery_message( "Solar Active", diff --git a/tests/batcontrol/test_night_surplus.py b/tests/batcontrol/test_night_surplus.py index 82d2b94..7b9b7d8 100644 --- a/tests/batcontrol/test_night_surplus.py +++ b/tests/batcontrol/test_night_surplus.py @@ -1,4 +1,4 @@ -"""Tests for Batcontrol._compute_night_surplus.""" +"""Tests for Batcontrol._compute_pv_start_battery and _compute_forecast_min_battery.""" import numpy as np import pytest from unittest.mock import MagicMock @@ -6,132 +6,134 @@ from batcontrol.core import Batcontrol -def _make_core(time_resolution=60): +def _make_core(): stub = MagicMock(spec=Batcontrol) - stub.time_resolution = time_resolution - stub._compute_night_surplus = ( - Batcontrol._compute_night_surplus.__get__(stub, Batcontrol) + stub._compute_pv_start_battery = ( + Batcontrol._compute_pv_start_battery.__get__(stub, Batcontrol) + ) + stub._compute_forecast_min_battery = ( + Batcontrol._compute_forecast_min_battery.__get__(stub, Batcontrol) ) return stub -def _call(stub, production, consumption, stored_usable=0.0, free_cap=0.0): - return stub._compute_night_surplus( - np.array(production, dtype=float), - np.array(consumption, dtype=float), - stored_usable, - free_cap, - ) +def _net(production, consumption): + return np.array(consumption, dtype=float) - np.array(production, dtype=float) + +# --------------------------------------------------------------------------- +# _compute_pv_start_battery +# --------------------------------------------------------------------------- -class TestNightSurplusNoProduction: - def test_zero_when_no_production_in_forecast(self): +class TestPvStartBattery: + def test_returns_battery_just_before_first_net_charging_slot(self): + # 2 discharge slots (net=+300), then net charging starts + # stored=2000, discharge 2x300=600 → battery=1400 at pv start stub = _make_core() - result = _call(stub, [0, 0, 0, 0], [300, 300, 300, 300]) - assert result == pytest.approx(0.0) + net = _net([0, 0, 1000], [300, 300, 200]) + result = stub._compute_pv_start_battery(net, stored_usable_energy=2000.0, free_capacity=3000.0) + assert result == pytest.approx(1400.0) + def test_returns_zero_when_no_net_charging_in_forecast(self): + stub = _make_core() + net = _net([0, 0, 0], [300, 300, 300]) + result = stub._compute_pv_start_battery(net, stored_usable_energy=1000.0, free_capacity=3000.0) + assert result == 0.0 -class TestNightSurplusSolarActive: - def test_full_battery_exceeds_night_consumption(self): - # Solar active (slot 0), production window slots 0-1 - # net_delta = -(500-1500 + 500-1500) = 2000 Wh net gain - # stored_usable=2000, free_cap=3000 -> battery_at_end = min(2000+3000, 2000+2000) = 4000 - # night: slots 2-3, consumption=500 each -> night_consumption=1000 - # surplus = 4000 - 1000 = 3000 + def test_returns_zero_when_battery_depleted_before_pv_start(self): + # stored=500, but 3 slots of 300 discharge → hits 0 before net<0 stub = _make_core() - production = [1500, 1500, 0, 0] - consumption = [500, 500, 500, 500] - result = _call(stub, production, consumption, stored_usable=2000.0, free_cap=3000.0) - assert result == pytest.approx(3000.0) + net = _net([0, 0, 1000], [300, 300, 200]) + result = stub._compute_pv_start_battery(net, stored_usable_energy=500.0, free_capacity=3000.0) + assert result == 0.0 - def test_battery_just_empty_by_morning(self): - # Solar active, net_delta=2000, stored=0, free=2000 -> battery_at_end=2000 - # night consumption = 2000 -> surplus = 0 + def test_returns_stored_when_first_slot_already_net_charging(self): + # slot 0 already net<0 (solar active, surplus) stub = _make_core() - production = [1500, 1500, 0, 0] - consumption = [500, 500, 1000, 1000] - result = _call(stub, production, consumption, stored_usable=0.0, free_cap=2000.0) - assert result == pytest.approx(0.0) + net = _net([1000, 500, 0], [200, 600, 300]) + result = stub._compute_pv_start_battery(net, stored_usable_energy=3000.0, free_capacity=2000.0) + assert result == pytest.approx(3000.0) - def test_surplus_never_negative(self): - # Battery drains completely during night + def test_floor_clamp_at_zero(self): + # battery drains to 0 halfway, stays at 0, then net charging starts stub = _make_core() - production = [500, 0, 0, 0] - consumption = [400, 1000, 1000, 1000] - result = _call(stub, production, consumption, stored_usable=100.0, free_cap=5000.0) + net = _net([0, 0, 0, 1000], [300, 300, 300, 200]) + result = stub._compute_pv_start_battery(net, stored_usable_energy=500.0, free_capacity=3000.0) assert result == 0.0 - def test_uses_only_first_production_window_not_second_day(self): - # Today solar (slots 0-1), night (slots 2-5), tomorrow solar (slots 6-7) - # battery_at_end should be computed at slot 1, night ends at slot 6 (next production) + def test_works_with_15min_resolution(self): + # 4 night slots at 100 Wh each, then net charging + stub = _make_core() + net = _net([0, 0, 0, 0, 600], [100, 100, 100, 100, 100]) + result = stub._compute_pv_start_battery(net, stored_usable_energy=1000.0, free_capacity=2000.0) + assert result == pytest.approx(600.0) + + +# --------------------------------------------------------------------------- +# _compute_forecast_min_battery +# --------------------------------------------------------------------------- + +class TestForecastMinBattery: + def test_returns_stored_when_always_charging(self): + # All slots are net charging: battery only grows, minimum = stored + stub = _make_core() + net = _net([1000, 1000, 1000], [200, 200, 200]) + result = stub._compute_forecast_min_battery(net, stored_usable_energy=2000.0, free_capacity=3000.0) + assert result == pytest.approx(2000.0) + + def test_returns_zero_when_battery_depleted(self): + # Battery drains to zero at some point stub = _make_core() - today = [1500, 1500] - night = [0] * 4 # 4 slots at 200 Wh each = 800 night consumption - tomorrow = [1500, 1500] - production = today + night + tomorrow - consumption = [200] * len(production) - # net_delta during slots 0-1: -((200-1500)+(200-1500)) = 2600 - # stored=1000, free=2000 -> battery_at_end = 1000 + min(2000, 2600) = 3000 - # night consumption slots 2-5: 4*200=800 - # surplus = 3000 - 800 = 2200 - result = _call(stub, production, consumption, stored_usable=1000.0, free_cap=2000.0) - assert result == pytest.approx(2200.0) - - -class TestNightSurplusSolarInactive: - def test_solar_tomorrow_enough_to_cover_night(self): - # slots 0-1: bridge (200 Wh each = 400 Wh discharge) - # slots 2-3: solar production (net +800 Wh each = 1600 Wh) - # end_idx=4, night slots 4-5: 200 Wh each = 400 Wh night consumption - # net_delta 0-3: -(200+200 - (1000-200) - (1000-200)) = -(400-1600) = 1200 - # stored=500, free=1500 -> battery_at_end = 500 + min(1500, 1200) = 1700 - # surplus = 1700 - 400 = 1300 + net = _net([0, 0, 0, 0], [500, 500, 500, 500]) + result = stub._compute_forecast_min_battery(net, stored_usable_energy=1000.0, free_capacity=3000.0) + assert result == 0.0 + + def test_tracks_trough_not_final_value(self): + # Discharge → hits trough → solar recharges → final value higher than trough + # stored=3000, discharge 4x400=1600 → trough=1400, then solar +3000 → final=4400 stub = _make_core() - production = [0, 0, 1000, 1000, 0, 0] - consumption = [200, 200, 200, 200, 200, 200] - result = _call(stub, production, consumption, stored_usable=500.0, free_cap=1500.0) - assert result == pytest.approx(1300.0) + net = _net([0, 0, 0, 0, 2000, 2000], [400, 400, 400, 400, 200, 200]) + result = stub._compute_forecast_min_battery(net, stored_usable_energy=3000.0, free_capacity=5000.0) + assert result == pytest.approx(1400.0) - def test_no_forecast_after_production_end(self): - # Forecast ends right after production window, no night slots + def test_cap_limits_charging(self): + # Battery can't exceed stored + free_capacity = 3000 + # Large solar: cap kicks in, but we care about minimum which is during discharge + stub = _make_core() + net = _net([0, 0, 3000, 3000, 0, 0], [300, 300, 200, 200, 300, 300]) + # Discharge: 3000-300=2700, 2700-300=2400 (trough) + # Solar: 2400+2800=5000 capped at 5000, 5000+2800→capped + # Discharge again: 5000-300=4700, 4700-300=4400 + result = stub._compute_forecast_min_battery( + net, stored_usable_energy=3000.0, free_capacity=2000.0) + assert result == pytest.approx(2400.0) + + def test_multi_day_tracks_deepest_trough(self): + # Day1 night=1000 discharge, Day1 solar=+2000, Day2 night=3000 discharge (deeper) stub = _make_core() - production = [0, 0, 1000, 1000] - consumption = [200, 200, 200, 200] - # net_delta 0-3: -(200+200-800-800) = 1200 - # stored=500, free=1500 -> battery_at_end=1700 - # night_end = len(production) = 4, no slots after production -> consumption=0 - # surplus = 1700 - result = _call(stub, production, consumption, stored_usable=500.0, free_cap=1500.0) - assert result == pytest.approx(1700.0) - - def test_free_cap_limits_charging(self): - # Large production but very little free capacity - # net_delta would be 3000, but free_cap=100 -> battery_at_end = 300+100 = 400 + night1 = [0] * 2 # 2 slots, 500 each = 1000 discharge + solar1 = [1500] * 2 # 2 slots, net = 1500-200 = +1300 each + night2 = [0] * 4 # 4 slots, 800 each = 3200 discharge + production = night1 + solar1 + night2 + consumption = [500, 500, 200, 200, 800, 800, 800, 800] + net = _net(production, consumption) + # stored=4000, free=4000 + # After night1: 4000-500-500=3000 + # After solar1: 3000+1300+1300=5600 capped at 8000 → 5600 + # After night2: 5600-800-800-800-800=2400 (deepest trough) + result = stub._compute_forecast_min_battery(net, stored_usable_energy=4000.0, free_capacity=4000.0) + assert result == pytest.approx(2400.0) + + def test_returns_zero_not_negative(self): + # Massive consumption, battery should clamp at 0 not go negative stub = _make_core() - production = [0, 2000, 2000, 0, 0] - consumption = [100, 100, 100, 200, 200] - # net_delta 0-2: -(100 + (100-2000) + (100-2000)) = -(100-1900-1900) = 3700 - # stored=300, free=100 -> battery_at_end = 300+min(100, 3700) = 400 - # night slots 3-4: 200+200=400 -> surplus = 0 - result = _call(stub, production, consumption, stored_usable=300.0, free_cap=100.0) - assert result == pytest.approx(0.0) + net = _net([0], [10000]) + result = stub._compute_forecast_min_battery(net, stored_usable_energy=1000.0, free_capacity=500.0) + assert result == 0.0 - def test_works_with_15min_resolution(self): - stub = _make_core(time_resolution=15) - # 4 night slots then 4 solar slots then 4 more night slots - production = [0, 0, 0, 0, 500, 500, 500, 500, 0, 0, 0, 0] - consumption = [100] * 12 - # net_delta slots 0-7: -(4*100 + 4*(100-500)) = -(400 - 1600) = 1200 - # stored=500, free=2000 -> battery_at_end = 500 + min(2000, 1200) = 1700 - # night end at slot 8 (no second production), night_end=12 - # night consumption slots 8-11: 4*100=400 - # surplus = 1700 - 400 = 1300 - result = _call(stub, production, consumption, stored_usable=500.0, free_cap=2000.0) - assert result == pytest.approx(1300.0) - - def test_surplus_never_negative_when_consumption_huge(self): + def test_initial_stored_counts_as_potential_minimum(self): + # If stored is already 0, minimum should be 0 even if solar charges later stub = _make_core() - production = [0, 0, 100, 0] - consumption = [500, 500, 500, 5000] - result = _call(stub, production, consumption, stored_usable=100.0, free_cap=10000.0) + net = _net([1000, 1000], [200, 200]) + result = stub._compute_forecast_min_battery(net, stored_usable_energy=0.0, free_capacity=5000.0) assert result == 0.0 From 705707e88d00bdbfcfc86330d829d2147fea5469 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 8 Jun 2026 18:24:32 +0000 Subject: [PATCH 2/7] Rename test_night_surplus.py to test_pv_battery_metrics.py Reflects the new metric names: pv_start_battery_wh and forecast_min_battery_wh. https://claude.ai/code/session_01AbZuZ73iEQMiwTkwy3Mt1h --- .../{test_night_surplus.py => test_pv_battery_metrics.py} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename tests/batcontrol/{test_night_surplus.py => test_pv_battery_metrics.py} (100%) diff --git a/tests/batcontrol/test_night_surplus.py b/tests/batcontrol/test_pv_battery_metrics.py similarity index 100% rename from tests/batcontrol/test_night_surplus.py rename to tests/batcontrol/test_pv_battery_metrics.py From 5c4e7e5a8402b41a6dee2075a6836b61df605fc9 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 8 Jun 2026 18:30:38 +0000 Subject: [PATCH 3/7] Extract forecast metrics into ForecastMetrics class All three _compute_* methods were pure functions with no dependency on Batcontrol instance state. Moved to a dedicated module: src/batcontrol/forecast_metrics.py (new) ForecastMetrics.solar_active_and_surplus() ForecastMetrics.pv_start_battery() ForecastMetrics.forecast_min_battery() core.py now imports and calls ForecastMetrics directly. Tests updated to import ForecastMetrics instead of binding unbound methods via MagicMock. https://claude.ai/code/session_01AbZuZ73iEQMiwTkwy3Mt1h --- src/batcontrol/core.py | 103 +---------------- src/batcontrol/forecast_metrics.py | 116 ++++++++++++++++++++ tests/batcontrol/test_pv_battery_metrics.py | 96 +++++----------- tests/batcontrol/test_solar_surplus.py | 67 ++++------- 4 files changed, 170 insertions(+), 212 deletions(-) create mode 100644 src/batcontrol/forecast_metrics.py diff --git a/src/batcontrol/core.py b/src/batcontrol/core.py index 7bb9e57..425918d 100644 --- a/src/batcontrol/core.py +++ b/src/batcontrol/core.py @@ -36,6 +36,7 @@ from .forecastsolar import ForecastSolar as solar_factory from .forecastconsumption import Consumption as consumption_factory +from .forecast_metrics import ForecastMetrics ERROR_IGNORE_TIME = 600 # 10 Minutes EVALUATIONS_EVERY_MINUTES = 3 # Every x minutes on the clock @@ -683,17 +684,17 @@ def run(self): if self.mqtt_api is not None: self.mqtt_api.publish_min_dynamic_price_diff( calc_output.min_dynamic_price_difference) - solar_active, surplus_wh = self._compute_solar_active_and_surplus( + solar_active, surplus_wh = ForecastMetrics.solar_active_and_surplus( production, consumption, calc_input.free_capacity ) self.mqtt_api.publish_solar_active(solar_active) self.mqtt_api.publish_solar_surplus(surplus_wh) - pv_start_wh = self._compute_pv_start_battery( + pv_start_wh = ForecastMetrics.pv_start_battery( net_consumption, calc_input.stored_usable_energy, calc_input.free_capacity ) self.mqtt_api.publish_pv_start_battery(pv_start_wh) - forecast_min_wh = self._compute_forecast_min_battery( + forecast_min_wh = ForecastMetrics.forecast_min_battery( net_consumption, calc_input.stored_usable_energy, calc_input.free_capacity ) @@ -881,102 +882,6 @@ def get_reserved_energy(self) -> float: """ Returns the reserved energy in Wh from last calculation """ return self.last_reserved_energy - def _compute_solar_active_and_surplus( - self, - production: np.ndarray, - consumption: np.ndarray, - free_capacity: float) -> tuple: - """Compute solar-active flag and expected surplus energy. - - Returns: - solar_active (bool): True iff solar is producing in slot 0 - surplus_wh (float): Expected solar overflow in Wh (>0 = WP can run) - - When solar is active, surplus is the overflow in the current production - window. Otherwise, surplus is the expected overflow at the end of the - next production window after the battery has bridged consumption until - solar restarts. - """ - net_consumption = consumption - production - - # Find start and end of the FIRST production window only - production_start: Optional[int] = None - production_end_current: Optional[int] = None - for i, p in enumerate(production): - if p > 0: - if production_start is None: - production_start = i - production_end_current = i - elif production_start is not None: - break - - solar_active = production_start == 0 - - if production_start is None: - surplus_wh = 0.0 - else: - bridge_wh = max(0.0, float(np.sum(net_consumption[:production_start]))) - end_idx = (production_end_current + 1) if production_end_current is not None \ - else production_start + 1 - solar_net_wh = float(-np.sum(net_consumption[production_start:end_idx])) - surplus_wh = max(0.0, solar_net_wh - free_capacity - bridge_wh) - - logger.debug( - 'Solar active: %s, surplus: %.1f Wh (free_cap=%.1f Wh)', - solar_active, surplus_wh, free_capacity - ) - return solar_active, surplus_wh - - def _compute_pv_start_battery( - self, - net_consumption: np.ndarray, - stored_usable_energy: float, - free_capacity: float) -> float: - """Battery level (Wh above MIN_SOC) at the start of the next net-charging window. - - Simulates slot-by-slot discharge until the first slot where - net_consumption < 0 (solar production exceeds household consumption). - That crossing point is when the battery transitions from discharging to - charging and is the most meaningful reference for overnight planning. - - Returns 0.0 if battery reaches MIN_SOC before that point, or if no - net-charging slot exists in the forecast at all. - """ - battery = stored_usable_energy - max_battery = stored_usable_energy + free_capacity - for net in net_consumption: - if net < 0: - return battery - battery = max(0.0, min(max_battery, battery - net)) - return 0.0 - - def _compute_forecast_min_battery( - self, - net_consumption: np.ndarray, - stored_usable_energy: float, - free_capacity: float) -> float: - """Minimum battery level (Wh above MIN_SOC) over the entire forecast horizon. - - Simulates slot-by-slot with proper floor (MIN_SOC = 0 usable) and ceiling - (MAX_SOC = stored_usable + free_capacity) clamping at each step. - Returns the lowest point reached during the simulation. - - A value of 0 means the battery is expected to hit MIN_SOC at some point - in the forecast — a signal to be conservative with flexible loads. - """ - battery = stored_usable_energy - max_battery = stored_usable_energy + free_capacity - min_battery = stored_usable_energy - for net in net_consumption: - battery = max(0.0, min(max_battery, battery - net)) - if battery < min_battery: - min_battery = battery - logger.debug( - 'Forecast min battery: %.1f Wh (stored=%.1f Wh, slots=%d)', - min_battery, stored_usable_energy, len(net_consumption) - ) - return min_battery - def set_stored_energy(self, stored_energy) -> None: """ Set the stored energy in Wh """ self.last_stored_energy = stored_energy diff --git a/src/batcontrol/forecast_metrics.py b/src/batcontrol/forecast_metrics.py new file mode 100644 index 0000000..e79637c --- /dev/null +++ b/src/batcontrol/forecast_metrics.py @@ -0,0 +1,116 @@ +""" +Forecast-derived battery metrics for state estimation and load-control decisions. + +ForecastMetrics computes indicators from production/consumption forecast arrays +and current battery state. All methods are pure functions with no side effects. + +Metrics: + solar_active_and_surplus -- solar-active flag + expected PV overflow (Wh) + pv_start_battery -- battery level (Wh) at next net-charging point + forecast_min_battery -- minimum battery level (Wh) over forecast horizon +""" +import logging +from typing import Optional, Tuple + +import numpy as np + +logger = logging.getLogger(__name__) + + +class ForecastMetrics: + """Pure-function metrics derived from forecast arrays and battery state.""" + + @staticmethod + def solar_active_and_surplus( + production: np.ndarray, + consumption: np.ndarray, + free_capacity: float) -> Tuple[bool, float]: + """Compute solar-active flag and expected surplus energy. + + Returns: + solar_active (bool): True iff solar is producing in slot 0 + surplus_wh (float): Expected solar overflow in Wh (>0 = WP can run) + + When solar is active, surplus is the overflow in the current production + window. Otherwise, surplus is the expected overflow at the end of the + next production window after the battery has bridged consumption until + solar restarts. + """ + net_consumption = consumption - production + + production_start: Optional[int] = None + production_end_current: Optional[int] = None + for i, p in enumerate(production): + if p > 0: + if production_start is None: + production_start = i + production_end_current = i + elif production_start is not None: + break + + solar_active = production_start == 0 + + if production_start is None: + surplus_wh = 0.0 + else: + bridge_wh = max(0.0, float(np.sum(net_consumption[:production_start]))) + end_idx = (production_end_current + 1) if production_end_current is not None \ + else production_start + 1 + solar_net_wh = float(-np.sum(net_consumption[production_start:end_idx])) + surplus_wh = max(0.0, solar_net_wh - free_capacity - bridge_wh) + + logger.debug( + 'Solar active: %s, surplus: %.1f Wh (free_cap=%.1f Wh)', + solar_active, surplus_wh, free_capacity + ) + return solar_active, surplus_wh + + @staticmethod + def pv_start_battery( + net_consumption: np.ndarray, + stored_usable_energy: float, + free_capacity: float) -> float: + """Battery level (Wh above MIN_SOC) at the start of the next net-charging window. + + Simulates slot-by-slot discharge until the first slot where + net_consumption < 0 (solar production exceeds household consumption). + That crossing point is when the battery transitions from discharging to + charging and is the most meaningful reference for overnight planning. + + Returns 0.0 if the battery reaches MIN_SOC before that point, or if no + net-charging slot exists in the forecast at all. + """ + battery = stored_usable_energy + max_battery = stored_usable_energy + free_capacity + for net in net_consumption: + if net < 0: + return battery + battery = max(0.0, min(max_battery, battery - net)) + return 0.0 + + @staticmethod + def forecast_min_battery( + net_consumption: np.ndarray, + stored_usable_energy: float, + free_capacity: float) -> float: + """Minimum battery level (Wh above MIN_SOC) over the entire forecast horizon. + + Simulates slot-by-slot with proper floor (MIN_SOC = 0 usable) and ceiling + (MAX_SOC = stored_usable + free_capacity) clamping at each step. + Returns the lowest point reached during the simulation. + + A value of 0 means the battery is expected to hit MIN_SOC at some point + in the forecast -- a signal to be conservative with flexible loads. + """ + battery = stored_usable_energy + max_battery = stored_usable_energy + free_capacity + min_battery = stored_usable_energy + for net in net_consumption: + battery = max(0.0, min(max_battery, battery - net)) + if battery < min_battery: + min_battery = battery + logger.debug( + 'Forecast min battery: %.1f Wh (stored=%.1f Wh, slots=%d)', + min_battery, stored_usable_energy, len(net_consumption) + ) + return min_battery diff --git a/tests/batcontrol/test_pv_battery_metrics.py b/tests/batcontrol/test_pv_battery_metrics.py index 7b9b7d8..5975461 100644 --- a/tests/batcontrol/test_pv_battery_metrics.py +++ b/tests/batcontrol/test_pv_battery_metrics.py @@ -1,20 +1,8 @@ -"""Tests for Batcontrol._compute_pv_start_battery and _compute_forecast_min_battery.""" +"""Tests for ForecastMetrics.pv_start_battery and ForecastMetrics.forecast_min_battery.""" import numpy as np import pytest -from unittest.mock import MagicMock -from batcontrol.core import Batcontrol - - -def _make_core(): - stub = MagicMock(spec=Batcontrol) - stub._compute_pv_start_battery = ( - Batcontrol._compute_pv_start_battery.__get__(stub, Batcontrol) - ) - stub._compute_forecast_min_battery = ( - Batcontrol._compute_forecast_min_battery.__get__(stub, Batcontrol) - ) - return stub +from batcontrol.forecast_metrics import ForecastMetrics def _net(production, consumption): @@ -22,118 +10,94 @@ def _net(production, consumption): # --------------------------------------------------------------------------- -# _compute_pv_start_battery +# pv_start_battery # --------------------------------------------------------------------------- class TestPvStartBattery: def test_returns_battery_just_before_first_net_charging_slot(self): # 2 discharge slots (net=+300), then net charging starts - # stored=2000, discharge 2x300=600 → battery=1400 at pv start - stub = _make_core() + # stored=2000, discharge 2x300=600 -> battery=1400 at pv start net = _net([0, 0, 1000], [300, 300, 200]) - result = stub._compute_pv_start_battery(net, stored_usable_energy=2000.0, free_capacity=3000.0) + result = ForecastMetrics.pv_start_battery(net, stored_usable_energy=2000.0, free_capacity=3000.0) assert result == pytest.approx(1400.0) def test_returns_zero_when_no_net_charging_in_forecast(self): - stub = _make_core() net = _net([0, 0, 0], [300, 300, 300]) - result = stub._compute_pv_start_battery(net, stored_usable_energy=1000.0, free_capacity=3000.0) + result = ForecastMetrics.pv_start_battery(net, stored_usable_energy=1000.0, free_capacity=3000.0) assert result == 0.0 def test_returns_zero_when_battery_depleted_before_pv_start(self): - # stored=500, but 3 slots of 300 discharge → hits 0 before net<0 - stub = _make_core() + # stored=500, 2x300 discharge exhausts it before net<0 net = _net([0, 0, 1000], [300, 300, 200]) - result = stub._compute_pv_start_battery(net, stored_usable_energy=500.0, free_capacity=3000.0) + result = ForecastMetrics.pv_start_battery(net, stored_usable_energy=500.0, free_capacity=3000.0) assert result == 0.0 def test_returns_stored_when_first_slot_already_net_charging(self): - # slot 0 already net<0 (solar active, surplus) - stub = _make_core() + # slot 0 already net<0 (solar active with surplus) net = _net([1000, 500, 0], [200, 600, 300]) - result = stub._compute_pv_start_battery(net, stored_usable_energy=3000.0, free_capacity=2000.0) + result = ForecastMetrics.pv_start_battery(net, stored_usable_energy=3000.0, free_capacity=2000.0) assert result == pytest.approx(3000.0) def test_floor_clamp_at_zero(self): - # battery drains to 0 halfway, stays at 0, then net charging starts - stub = _make_core() + # battery drains to 0, stays there, then net charging starts net = _net([0, 0, 0, 1000], [300, 300, 300, 200]) - result = stub._compute_pv_start_battery(net, stored_usable_energy=500.0, free_capacity=3000.0) + result = ForecastMetrics.pv_start_battery(net, stored_usable_energy=500.0, free_capacity=3000.0) assert result == 0.0 def test_works_with_15min_resolution(self): # 4 night slots at 100 Wh each, then net charging - stub = _make_core() net = _net([0, 0, 0, 0, 600], [100, 100, 100, 100, 100]) - result = stub._compute_pv_start_battery(net, stored_usable_energy=1000.0, free_capacity=2000.0) + result = ForecastMetrics.pv_start_battery(net, stored_usable_energy=1000.0, free_capacity=2000.0) assert result == pytest.approx(600.0) # --------------------------------------------------------------------------- -# _compute_forecast_min_battery +# forecast_min_battery # --------------------------------------------------------------------------- class TestForecastMinBattery: def test_returns_stored_when_always_charging(self): - # All slots are net charging: battery only grows, minimum = stored - stub = _make_core() + # All slots net charging: battery only grows, minimum = stored net = _net([1000, 1000, 1000], [200, 200, 200]) - result = stub._compute_forecast_min_battery(net, stored_usable_energy=2000.0, free_capacity=3000.0) + result = ForecastMetrics.forecast_min_battery(net, stored_usable_energy=2000.0, free_capacity=3000.0) assert result == pytest.approx(2000.0) def test_returns_zero_when_battery_depleted(self): - # Battery drains to zero at some point - stub = _make_core() net = _net([0, 0, 0, 0], [500, 500, 500, 500]) - result = stub._compute_forecast_min_battery(net, stored_usable_energy=1000.0, free_capacity=3000.0) + result = ForecastMetrics.forecast_min_battery(net, stored_usable_energy=1000.0, free_capacity=3000.0) assert result == 0.0 def test_tracks_trough_not_final_value(self): - # Discharge → hits trough → solar recharges → final value higher than trough - # stored=3000, discharge 4x400=1600 → trough=1400, then solar +3000 → final=4400 - stub = _make_core() + # Discharge to trough, then solar recharges above trough + # stored=3000, 4x400 discharge -> trough=1400, then solar restores net = _net([0, 0, 0, 0, 2000, 2000], [400, 400, 400, 400, 200, 200]) - result = stub._compute_forecast_min_battery(net, stored_usable_energy=3000.0, free_capacity=5000.0) + result = ForecastMetrics.forecast_min_battery(net, stored_usable_energy=3000.0, free_capacity=5000.0) assert result == pytest.approx(1400.0) def test_cap_limits_charging(self): - # Battery can't exceed stored + free_capacity = 3000 - # Large solar: cap kicks in, but we care about minimum which is during discharge - stub = _make_core() + # Minimum is during initial discharge; cap is irrelevant for the trough net = _net([0, 0, 3000, 3000, 0, 0], [300, 300, 200, 200, 300, 300]) - # Discharge: 3000-300=2700, 2700-300=2400 (trough) - # Solar: 2400+2800=5000 capped at 5000, 5000+2800→capped - # Discharge again: 5000-300=4700, 4700-300=4400 - result = stub._compute_forecast_min_battery( + # stored=3000, trough after 2x discharge: 3000-300-300=2400 + result = ForecastMetrics.forecast_min_battery( net, stored_usable_energy=3000.0, free_capacity=2000.0) assert result == pytest.approx(2400.0) def test_multi_day_tracks_deepest_trough(self): - # Day1 night=1000 discharge, Day1 solar=+2000, Day2 night=3000 discharge (deeper) - stub = _make_core() - night1 = [0] * 2 # 2 slots, 500 each = 1000 discharge - solar1 = [1500] * 2 # 2 slots, net = 1500-200 = +1300 each - night2 = [0] * 4 # 4 slots, 800 each = 3200 discharge - production = night1 + solar1 + night2 + # Night1 discharges 1000, Solar1 recharges, Night2 discharges 3200 (deeper) + production = [0, 0, 1500, 1500, 0, 0, 0, 0] consumption = [500, 500, 200, 200, 800, 800, 800, 800] net = _net(production, consumption) - # stored=4000, free=4000 - # After night1: 4000-500-500=3000 - # After solar1: 3000+1300+1300=5600 capped at 8000 → 5600 - # After night2: 5600-800-800-800-800=2400 (deepest trough) - result = stub._compute_forecast_min_battery(net, stored_usable_energy=4000.0, free_capacity=4000.0) + # stored=4000: 4000-500-500=3000, +1300+1300=5600, -800x4=2400 (deepest) + result = ForecastMetrics.forecast_min_battery(net, stored_usable_energy=4000.0, free_capacity=4000.0) assert result == pytest.approx(2400.0) def test_returns_zero_not_negative(self): - # Massive consumption, battery should clamp at 0 not go negative - stub = _make_core() net = _net([0], [10000]) - result = stub._compute_forecast_min_battery(net, stored_usable_energy=1000.0, free_capacity=500.0) + result = ForecastMetrics.forecast_min_battery(net, stored_usable_energy=1000.0, free_capacity=500.0) assert result == 0.0 def test_initial_stored_counts_as_potential_minimum(self): - # If stored is already 0, minimum should be 0 even if solar charges later - stub = _make_core() + # stored=0: minimum starts at 0, solar later does not change that net = _net([1000, 1000], [200, 200]) - result = stub._compute_forecast_min_battery(net, stored_usable_energy=0.0, free_capacity=5000.0) + result = ForecastMetrics.forecast_min_battery(net, stored_usable_energy=0.0, free_capacity=5000.0) assert result == 0.0 diff --git a/tests/batcontrol/test_solar_surplus.py b/tests/batcontrol/test_solar_surplus.py index 34ba09d..66231cd 100644 --- a/tests/batcontrol/test_solar_surplus.py +++ b/tests/batcontrol/test_solar_surplus.py @@ -1,23 +1,12 @@ -"""Tests for Batcontrol._compute_solar_active_and_surplus.""" +"""Tests for ForecastMetrics.solar_active_and_surplus.""" import numpy as np import pytest -from unittest.mock import MagicMock -from batcontrol.core import Batcontrol +from batcontrol.forecast_metrics import ForecastMetrics -def _make_core(time_resolution=60): - """Return a minimal stub with only the attributes used by the method.""" - stub = MagicMock(spec=Batcontrol) - stub.time_resolution = time_resolution - stub._compute_solar_active_and_surplus = ( - Batcontrol._compute_solar_active_and_surplus.__get__(stub, Batcontrol) - ) - return stub - - -def _call(stub, production, consumption, free_cap=0.0): - return stub._compute_solar_active_and_surplus( +def _call(production, consumption, free_cap=0.0): + return ForecastMetrics.solar_active_and_surplus( np.array(production, dtype=float), np.array(consumption, dtype=float), free_cap, @@ -26,120 +15,104 @@ def _call(stub, production, consumption, free_cap=0.0): class TestSolarActive: def test_active_when_production_starts_at_slot0(self): - stub = _make_core() - active, _ = _call(stub, [1000, 1500, 500], [400, 400, 400]) + active, _ = _call([1000, 1500, 500], [400, 400, 400]) assert active is True def test_inactive_when_production_starts_later(self): - stub = _make_core() - active, _ = _call(stub, [0, 0, 800, 1200, 0], [300, 300, 300, 300, 300]) + active, _ = _call([0, 0, 800, 1200, 0], [300, 300, 300, 300, 300]) assert active is False def test_inactive_when_no_production_at_all(self): - stub = _make_core() - active, _ = _call(stub, [0, 0, 0, 0], [300, 400, 350, 300]) + active, _ = _call([0, 0, 0, 0], [300, 400, 350, 300]) assert active is False def test_active_when_slot0_producing_even_with_gap_after(self): - stub = _make_core() - active, _ = _call(stub, [800, 0, 600, 0], [300, 300, 300, 300]) + active, _ = _call([800, 0, 600, 0], [300, 300, 300, 300]) assert active is True class TestSurplusActive: def test_surplus_zero_when_net_production_fits_in_battery(self): # net = 1000+1000 = 2000 Wh, free_cap=3000 -> fits, no surplus - stub = _make_core() - _, surplus = _call(stub, [1500, 1500, 0], [500, 500, 500], free_cap=3000.0) + _, surplus = _call([1500, 1500, 0], [500, 500, 500], free_cap=3000.0) assert surplus == pytest.approx(0.0) def test_surplus_positive_when_net_production_exceeds_free_capacity(self): # net = 1000+1000 = 2000 Wh, free_cap=1200 -> surplus=800 - stub = _make_core() - _, surplus = _call(stub, [1500, 1500, 0], [500, 500, 0], free_cap=1200.0) + _, surplus = _call([1500, 1500, 0], [500, 500, 0], free_cap=1200.0) assert surplus == pytest.approx(800.0) def test_surplus_accounts_for_consumption_in_window(self): # slot0: +500 net, slot1: -500 net -> total=0, no surplus - stub = _make_core() - _, surplus = _call(stub, [2000, 2000, 0], [1500, 2500, 400], free_cap=0.0) + _, surplus = _call([2000, 2000, 0], [1500, 2500, 400], free_cap=0.0) assert surplus == pytest.approx(0.0) def test_surplus_never_negative(self): - stub = _make_core() - _, surplus = _call(stub, [100, 100, 0], [800, 800, 800], free_cap=10000.0) + _, surplus = _call([100, 100, 0], [800, 800, 800], free_cap=10000.0) assert surplus == 0.0 def test_active_uses_only_first_production_window(self): # 48h forecast: today's solar then a long break then tomorrow's solar # 'during' must NOT include tomorrow's solar (production_end stops at first zero) - stub = _make_core() today_solar = [1500, 1500] # 2000 Wh net production night = [0] * 12 tomorrow_solar = [1500, 1500] production = today_solar + night + tomorrow_solar consumption = [500] * len(production) # solar_net = -(500-1500 + 500-1500) = 2000 Wh, free_cap=1200 -> surplus=800 - _, surplus = _call(stub, production, consumption, free_cap=1200.0) + _, surplus = _call(production, consumption, free_cap=1200.0) assert surplus == pytest.approx(800.0) class TestSurplusInactive: def test_surplus_zero_when_no_solar_in_forecast(self): - stub = _make_core() - _, surplus = _call(stub, [0, 0, 0, 0], [500, 500, 500, 500], free_cap=0.0) + _, surplus = _call([0, 0, 0, 0], [500, 500, 500, 500], free_cap=0.0) assert surplus == pytest.approx(0.0) def test_surplus_positive_when_solar_overflows(self): # bridge: 2 slots * 200 Wh = 400 Wh # solar_net: 2 slots * (1000-200) = 1600 Wh # surplus = max(0, 1600 - 500 - 400) = 700 Wh - stub = _make_core() production = [0, 0, 1000, 1000, 0] consumption = [200, 200, 200, 200, 200] - _, surplus = _call(stub, production, consumption, free_cap=500.0) + _, surplus = _call(production, consumption, free_cap=500.0) assert surplus == pytest.approx(700.0) def test_surplus_zero_when_solar_fits_in_battery_after_night_discharge(self): # bridge=400, solar_net=800, free_cap=2000 -> 800-2000-400 < 0 -> surplus=0 - stub = _make_core() production = [0, 0, 1000, 0] consumption = [200, 200, 200, 200] - _, surplus = _call(stub, production, consumption, free_cap=2000.0) + _, surplus = _call(production, consumption, free_cap=2000.0) assert surplus == pytest.approx(0.0) def test_night_discharge_creates_room_for_solar(self): # slot0: cons=500 (bridge=500, opens battery room) # slots1-2: 2000W prod, 500W cons -> solar_net=3000 Wh # surplus = max(0, 3000 - 2000 - 500) = 500 - stub = _make_core() production = [0, 2000, 2000, 0] consumption = [500, 500, 500, 500] - _, surplus = _call(stub, production, consumption, free_cap=2000.0) + _, surplus = _call(production, consumption, free_cap=2000.0) assert surplus == pytest.approx(500.0) def test_surplus_never_negative(self): - stub = _make_core() - _, surplus = _call(stub, [0, 0, 100, 0], [500, 500, 500, 500], free_cap=10000.0) + _, surplus = _call([0, 0, 100, 0], [500, 500, 500, 500], free_cap=10000.0) assert surplus == 0.0 def test_inactive_only_uses_first_production_window(self): # 48h: night, tomorrow solar window (slots 2-3), second night, day-after solar - stub = _make_core() production = [0, 0, 1000, 1000, 0, 0, 0, 0, 1000, 1000] consumption = [200] * 10 # bridge=400 (slots 0-1), solar_net=1600 (slots 2-3), free_cap=500 # surplus = max(0, 1600-500-400) = 700 (day-after ignored) - _, surplus = _call(stub, production, consumption, free_cap=500.0) + _, surplus = _call(production, consumption, free_cap=500.0) assert surplus == pytest.approx(700.0) def test_works_with_15min_resolution(self): # Arrays are already Wh/slot independent of resolution - stub = _make_core(time_resolution=15) # 4 slots night (200 Wh each) = 800 bridge # 4 slots solar 500 Wh prod, 200 Wh cons each = 4 * 300 = 1200 Wh solar_net # free_cap=0 -> surplus = max(0, 1200 - 0 - 800) = 400 production = [0, 0, 0, 0, 500, 500, 500, 500, 0] consumption = [200] * 9 - _, surplus = _call(stub, production, consumption, free_cap=0.0) + _, surplus = _call(production, consumption, free_cap=0.0) assert surplus == pytest.approx(400.0) From 3eba156a10b1b40dc28174ac77ca9c7cc65e2d04 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 8 Jun 2026 18:37:41 +0000 Subject: [PATCH 4/7] Fix stale net_consumption and remove legacy MQTT discovery entry core.py: net_consumption was computed before production[0]/consumption[0] were factorized for elapsed time, so pv_start_battery and forecast_min_battery received the full-slot slot-0 value instead of the time-adjusted one. Recompute after the factorization. mqtt_api.py: publish an empty retained payload to the old batcontrol_night_surplus_wh discovery topic on connect so that existing Home Assistant brokers remove the stale entity automatically. https://claude.ai/code/session_01AbZuZ73iEQMiwTkwy3Mt1h --- src/batcontrol/core.py | 1 + src/batcontrol/mqtt_api.py | 9 +++++++++ 2 files changed, 10 insertions(+) diff --git a/src/batcontrol/core.py b/src/batcontrol/core.py index 425918d..894f2a4 100644 --- a/src/batcontrol/core.py +++ b/src/batcontrol/core.py @@ -605,6 +605,7 @@ def run(self): # Factorize [0] to account for elapsed time production[0] *= (1 - elapsed_in_current) consumption[0] *= (1 - elapsed_in_current) + net_consumption = consumption - production logger.debug( 'Current interval factorization: elapsed=%.3f, remaining=%.3f', diff --git a/src/batcontrol/mqtt_api.py b/src/batcontrol/mqtt_api.py index a504c9e..17305cb 100644 --- a/src/batcontrol/mqtt_api.py +++ b/src/batcontrol/mqtt_api.py @@ -960,6 +960,15 @@ def send_mqtt_discovery_messages(self) -> None: entity_category="diagnostic", value_template="{% if value == 'true' %}ON{% else %}OFF{% endif %}") + # Remove legacy retained discovery config for the renamed metric. + # An empty retained payload deletes the HA entity from existing brokers. + if self.client.is_connected(): + self.client.publish( + self.auto_discover_topic + + '/sensor/batcontrol/batcontrol_night_surplus_wh/config', + '', + retain=True) + def send_mqtt_discovery_for_mode(self) -> None: """ Publish Home Assistant MQTT Auto Discovery message for mode""" val_templ = ( From 380d3b1addb5f1fbf671f976c18934f6840d7105 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 8 Jun 2026 18:44:05 +0000 Subject: [PATCH 5/7] Add TODO(0.9.1) to remove night_surplus_wh tombstone block https://claude.ai/code/session_01AbZuZ73iEQMiwTkwy3Mt1h --- src/batcontrol/mqtt_api.py | 1 + 1 file changed, 1 insertion(+) diff --git a/src/batcontrol/mqtt_api.py b/src/batcontrol/mqtt_api.py index 17305cb..82426eb 100644 --- a/src/batcontrol/mqtt_api.py +++ b/src/batcontrol/mqtt_api.py @@ -960,6 +960,7 @@ def send_mqtt_discovery_messages(self) -> None: entity_category="diagnostic", value_template="{% if value == 'true' %}ON{% else %}OFF{% endif %}") + # TODO(0.9.1): remove this tombstone block once brokers have been cleaned up. # Remove legacy retained discovery config for the renamed metric. # An empty retained payload deletes the HA entity from existing brokers. if self.client.is_connected(): From b27ee71005765c4fe23ed27acbdca12bf2de9c4c Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 8 Jun 2026 19:27:38 +0000 Subject: [PATCH 6/7] Fix test_mqtt_api failures from tombstone and update docstring _make_discovery_stub lacked api.client and api.auto_discover_topic, causing AttributeError when the tombstone block in send_mqtt_discovery_messages called self.client.is_connected() directly. Added both to the stub. Also update forecast_metrics.py module docstring to not claim the methods are side-effect-free; they emit logger.debug() calls. https://claude.ai/code/session_01AbZuZ73iEQMiwTkwy3Mt1h --- src/batcontrol/forecast_metrics.py | 3 ++- tests/batcontrol/test_mqtt_api.py | 2 ++ 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/src/batcontrol/forecast_metrics.py b/src/batcontrol/forecast_metrics.py index e79637c..045f2e6 100644 --- a/src/batcontrol/forecast_metrics.py +++ b/src/batcontrol/forecast_metrics.py @@ -2,7 +2,8 @@ Forecast-derived battery metrics for state estimation and load-control decisions. ForecastMetrics computes indicators from production/consumption forecast arrays -and current battery state. All methods are pure functions with no side effects. +and current battery state. All methods are stateless with respect to object +state; they emit debug log messages but do not mutate any shared state. Metrics: solar_active_and_surplus -- solar-active flag + expected PV overflow (Wh) diff --git a/tests/batcontrol/test_mqtt_api.py b/tests/batcontrol/test_mqtt_api.py index b4f4e0c..897282e 100644 --- a/tests/batcontrol/test_mqtt_api.py +++ b/tests/batcontrol/test_mqtt_api.py @@ -40,6 +40,8 @@ def _make_discovery_stub(): api.send_mqtt_discovery_messages = ( MqttApi.send_mqtt_discovery_messages.__get__(api, MqttApi) ) + api.client = MagicMock() + api.auto_discover_topic = 'homeassistant' return api From 3c388d7c49f18c70553e518ef4e2e368585fefea Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 9 Jun 2026 04:46:26 +0000 Subject: [PATCH 7/7] Add WIKI_forecast_metrics.md documenting the three forecast metrics Describes solar_surplus_wh, pv_start_battery_wh and forecast_min_battery_wh: what each value means, when it is useful, a 2x2 decision matrix for flexible load control, MQTT topics, HA auto-discovery entities, and implementation notes. https://claude.ai/code/session_01AbZuZ73iEQMiwTkwy3Mt1h --- docs/WIKI_forecast_metrics.md | 154 ++++++++++++++++++++++++++++++++++ 1 file changed, 154 insertions(+) create mode 100644 docs/WIKI_forecast_metrics.md diff --git a/docs/WIKI_forecast_metrics.md b/docs/WIKI_forecast_metrics.md new file mode 100644 index 0000000..aec0c53 --- /dev/null +++ b/docs/WIKI_forecast_metrics.md @@ -0,0 +1,154 @@ +# Forecast Metrics + +## Overview + +`ForecastMetrics` is a stateless module that derives three battery indicators +from the production/consumption forecast arrays and the current battery state. +These indicators are published via MQTT and are intended to drive downstream +automation decisions such as "should I run the heat pump now or save the battery +for tomorrow's solar charge?" + +All three values are updated once per evaluation cycle (every 3 minutes by +default). They are based on the same forecast window that the main optimizer +uses, so their horizon is `min(max available price hours, max available solar +hours)`. + +--- + +## The Three Metrics + +### `solar_surplus_wh` — Current-Window Solar Overflow + +**MQTT topic:** `{base}/solar_surplus_wh` + +Expected energy in Wh that the solar production in the **current production +window** will generate above what the battery can absorb. + +- **solar_active = true (slot 0 is producing):** surplus is the net solar + production of the ongoing window minus the remaining free battery capacity. + `surplus > 0` means that PV power will be exported to the grid even if the + battery is managed optimally. +- **solar_active = false (nighttime or break before next window):** surplus + accounts for the overnight discharge first — the battery will self-discharge + from consumption before solar starts, which creates room. `surplus > 0` means + even after that extra room is created the next solar window will still + overflow. + +A value of `0` means the battery can absorb everything the upcoming solar +window produces. A value `> 0` means some PV will inevitably be exported; it +is safe to run flexible loads (heat pump, EV charging) from the grid right now +because the solar surplus will offset them. + +**Use case:** "Is running the heat pump now free in terms of grid cost?" — if +`solar_surplus_wh >= estimated_heat_pump_wh`, the WP can run without net +additional grid draw over the forecast horizon. + +--- + +### `pv_start_battery_wh` — Battery Level at Next Charging Point + +**MQTT topic:** `{base}/pv_start_battery_wh` + +Battery level in Wh (above MIN_SOC) at the moment when solar production first +exceeds household consumption (`net_consumption < 0`). This is the point where +the battery transitions from discharging to charging. + +- Simulated slot-by-slot from the current moment forward. +- If the battery hits `0` (MIN_SOC) before solar starts, the value is `0`. +- If the battery has no net-charging slot in the forecast at all, the value + is `0`. +- If slot 0 is already a net-charging slot (solar already exceeds + consumption), the value equals the current stored usable energy. + +**Use case:** "How much charge will the battery have left when it starts +refilling tomorrow morning?" A low value (e.g. < 500 Wh) means the battery +will be nearly flat at dawn; flexible loads tonight should be reduced. A high +value means overnight consumption will not deplete the battery significantly +and flexible loads can run. + +**Important:** `pv_start_battery_wh` depends on `net_consumption < 0`, not +just on `production > 0`. The battery does not switch to charging until solar +output exceeds household consumption — on a partly-cloudy morning the +cross-over can happen later than sunrise. + +--- + +### `forecast_min_battery_wh` — Forecast Minimum Battery Level + +**MQTT topic:** `{base}/forecast_min_battery_wh` + +The lowest battery level in Wh (above MIN_SOC) reached at any point during the +entire forecast horizon, based on slot-by-slot simulation with proper +floor/ceiling clamping. + +- A value of `0` means the battery is expected to hit MIN_SOC at some point in + the forecast — a signal that the system will be energy-constrained. +- The simulation respects both the floor (MIN_SOC = 0 usable Wh) and the + ceiling (MAX_SOC = stored_usable + free_capacity), so multi-day + charge/discharge cycles are tracked correctly. +- The horizon covers the full forecast window (same as the optimizer), not + just the next 24 hours. + +**Use case:** "Will the battery run out at any point in the planning horizon?" +If `forecast_min_battery_wh == 0`, batcontrol may need to grid-charge at some +point; flexible loads should be conservative. If `forecast_min_battery_wh` is +comfortably above zero, there is a buffer and flexible loads can run freely. + +--- + +## Decision Matrix + +The three metrics form a natural 2-D decision space for flexible load control: + +| `solar_surplus_wh` | `forecast_min_battery_wh` | Recommended action | +|--------------------|--------------------------|-------------------| +| > 0 | > 0 | Run flexible loads freely — PV will cover them and battery stays healthy | +| > 0 | = 0 | PV surplus exists but battery will be short later — run light loads only | +| = 0 | > 0 | No surplus but battery OK — use `pv_start_battery_wh` to judge night loads | +| = 0 | = 0 | Constrained — block flexible loads, preserve battery | + +`pv_start_battery_wh` refines the third row: if it is high, the overnight +discharge is gentle and a moderate flexible load (e.g. heat pump one cycle) +is fine. If it is near zero, defer to the next solar window. + +--- + +## MQTT Topics + +| Topic | Unit | Retained | Description | +|-------|------|----------|-------------| +| `{base}/solar_surplus_wh` | Wh | No | PV overflow that cannot be stored in the battery | +| `{base}/solar_active` | bool | No | `true` if solar is producing in slot 0 | +| `{base}/pv_start_battery_wh` | Wh | No | Battery level at next net-charging crossover | +| `{base}/forecast_min_battery_wh` | Wh | No | Minimum battery level over entire forecast horizon | + +All values are published after each evaluation cycle (together with the +inverter control decision). + +### Home Assistant Auto-Discovery + +The following HA entities are created automatically when +`auto_discover_enable: true` is configured: + +- **Solar Surplus** — sensor (energy, Wh) +- **Solar Active** — binary sensor (on/off diagnostic) +- **PV Start Battery** — sensor (energy, Wh) +- **Forecast Min Battery** — sensor (energy, Wh) + +--- + +## Implementation Notes + +- All values are computed in `src/batcontrol/forecast_metrics.py` by the + `ForecastMetrics` class. +- Slot 0 is time-adjusted: the elapsed fraction of the current interval is + subtracted so that a slot already 80% elapsed only contributes 20% of its + forecast energy. +- `net_consumption = consumption - production`; negative = battery charging, + positive = battery discharging / grid draw. +- `stored_usable_energy` is the energy above MIN_SOC. `free_capacity` is the + space between the current level and MAX_SOC. +- The slot-by-slot simulation clamps at both ends: + `battery = max(0, min(stored_usable + free_capacity, battery - net))`. + A simple net-sum over slots would overestimate available energy because it + ignores that the battery cannot go below 0 or above MAX_SOC.