From f8dfaf4d40b853bfd5baeb7f1cfec3f6cb5e1e9b Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 27 May 2026 08:47:50 +0000 Subject: [PATCH 1/4] Add solar surplus and production phase to MQTT output After each control cycle, publish two new topics: - /solar_surplus_wh: expected usable solar surplus in Wh - /solar_production_phase: 'before', 'during', or 'after' Phase detection uses a None sentinel to correctly distinguish "no production in forecast" from "production starting at slot 0", fixing the ambiguity in the wp-feedin heuristic. Production window boundaries track the last non-zero slot to tolerate single-slot forecast gaps. Surplus calculation: - during/before: net production surplus over the forecast window minus free battery capacity (what would be exported/wasted) - after: unreserved stored usable energy minus expected consumption until next solar production starts (what the battery can spare) The corrected (elapsed-time-adjusted) production/consumption arrays are used together with calc_output.reserved_energy from the just-completed optimizer run, so the values are consistent with the control decision. HA MQTT auto-discovery entries added for both sensors. https://claude.ai/code/session_01WQKwRRnW55Tr2mBkKao573 --- src/batcontrol/core.py | 74 ++++++++++++++++++++++++++++++++++++++ src/batcontrol/mqtt_api.py | 43 ++++++++++++++++++++++ 2 files changed, 117 insertions(+) diff --git a/src/batcontrol/core.py b/src/batcontrol/core.py index b318711d..ec20e1cb 100644 --- a/src/batcontrol/core.py +++ b/src/batcontrol/core.py @@ -683,6 +683,11 @@ def run(self): if self.mqtt_api is not None: self.mqtt_api.publish_min_dynamic_price_diff( calc_output.min_dynamic_price_difference) + phase, surplus_wh = self._compute_solar_surplus_and_phase( + production, consumption, calc_output.reserved_energy + ) + self.mqtt_api.publish_production_phase(phase) + self.mqtt_api.publish_solar_surplus(surplus_wh) if self.discharge_blocked and not \ self.general_logic.is_discharge_always_allowed_soc(self.get_SOC()): @@ -866,6 +871,75 @@ def get_reserved_energy(self) -> float: """ Returns the reserved energy in Wh from last calculation """ return self.last_reserved_energy + def _compute_solar_surplus_and_phase( + self, + production: np.ndarray, + consumption: np.ndarray, + reserved_energy: float) -> tuple: + """Compute solar production phase and expected surplus energy. + + Uses corrected arrays (elapsed-time-adjusted for slot [0]). + + Returns: + phase (str): 'before', 'during', or 'after' + surplus_wh (float): Expected usable surplus in Wh (>0 = WP can run) + + Phase semantics: + before: production has not started yet (production[0] == 0, future slots > 0) + during: solar production is happening right now (production[0] > 0) + after: no production remaining in forecast window + """ + wh_per_slot = self.time_resolution / 60.0 + net_consumption = consumption - production + + # Locate production window, tolerating single-slot gaps + 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 + + if production_start is None: + phase = 'after' + elif production_start == 0: + phase = 'during' + else: + phase = 'before' + + free_capacity = self.get_free_capacity() + + if phase in ('during', 'before'): + end_idx = (production_end + 1) if production_end is not None else len(net_consumption) + net_surplus_wh = sum( + (-net_consumption[i]) * wh_per_slot + for i in range(0, end_idx) + if net_consumption[i] < 0 + ) + surplus_wh = max(0.0, net_surplus_wh - free_capacity) + else: + # Find when next significant solar production starts (net excess > 100 W) + next_solar_start = len(net_consumption) + for i, nc in enumerate(net_consumption): + if nc < -100: + next_solar_start = i + break + expected_consumption_wh = sum( + nc * wh_per_slot + for nc in net_consumption[:next_solar_start] + if nc > 0 + ) + stored_usable = self.get_stored_usable_energy() + unreserved = max(0.0, stored_usable - reserved_energy) + surplus_wh = max(0.0, unreserved - expected_consumption_wh) + + logger.debug( + 'Solar phase: %s, surplus: %.1f Wh (free_cap=%.1f Wh)', + phase, surplus_wh, free_capacity + ) + return phase, surplus_wh + 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/mqtt_api.py b/src/batcontrol/mqtt_api.py index 1c5c0d97..682e7be6 100644 --- a/src/batcontrol/mqtt_api.py +++ b/src/batcontrol/mqtt_api.py @@ -26,6 +26,8 @@ - /production_offset: production offset percentage (1.0 = 100%, 0.8 = 80%, etc.) - /api_override_active: bool indicating whether a temporary external/API override is active - /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_production_phase: solar production phase ('before', 'during', 'after') The following statistical arrays are published as JSON arrays: - /FCST/production: forecasted production in W @@ -484,6 +486,30 @@ def publish_production_offset(self, production_offset: float) -> None: f'{production_offset:.3f}' ) + def publish_solar_surplus(self, surplus_wh: float) -> None: + """ Publish the expected solar surplus energy to MQTT. + /solar_surplus_wh + Positive values indicate surplus beyond battery free capacity (during/before) + or surplus stored energy above expected consumption until next solar (after). + """ + if self.client.is_connected(): + self.client.publish( + self.base_topic + '/solar_surplus_wh', + f'{surplus_wh:.1f}' + ) + + def publish_production_phase(self, phase: str) -> None: + """ Publish the solar production phase to MQTT. + /solar_production_phase + Values: 'before' (production expected later), 'during' (producing now), + 'after' (no more production in forecast). + """ + if self.client.is_connected(): + self.client.publish( + self.base_topic + '/solar_production_phase', + phase + ) + def publish_api_override_active(self, active: bool) -> None: """ Publish whether a temporary API override is currently active. /api_override_active @@ -875,6 +901,23 @@ def send_mqtt_discovery_messages(self) -> None: "/stored_energy_capacity", entity_category="diagnostic") + self.publish_mqtt_discovery_message( + "Solar Surplus", + "batcontrol_solar_surplus_wh", + "sensor", + "energy", + "Wh", + self.base_topic + "/solar_surplus_wh") + + self.publish_mqtt_discovery_message( + "Solar Production Phase", + "batcontrol_solar_production_phase", + "sensor", + None, + None, + self.base_topic + "/solar_production_phase", + entity_category="diagnostic") + def send_mqtt_discovery_for_mode(self) -> None: """ Publish Home Assistant MQTT Auto Discovery message for mode""" val_templ = ( From 839034540d80f37d200f55f13ab8dbded6c239fe Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 28 May 2026 04:59:50 +0000 Subject: [PATCH 2/4] Address Copilot review: fix surplus calc and add tests Fix surplus calculation for during/before phase: sum the net energy balance over the entire production window (production minus consumption) instead of only counting slots where production exceeds consumption. Positive consumption slots in the window now correctly reduce the reported surplus. Fix free_capacity and stored_usable_energy consistency: pass calc_input.free_capacity and calc_input.stored_usable_energy into the helper so the published values are based on the same inverter snapshot as the optimizer decision, avoiding a second inverter read. Add tests/batcontrol/test_solar_surplus.py: 19 focused tests covering phase detection (during/before/after, gaps, 15-min resolution) and surplus calculation for all three phases including edge cases (fits in battery, exceeds capacity, negative surplus clamped to zero, reserved energy, 15-min scaling). Extend tests/batcontrol/test_mqtt_api.py: 8 tests covering publish_solar_surplus (topic, format, zero, disconnected) and publish_production_phase (all three values, disconnected). https://claude.ai/code/session_01WQKwRRnW55Tr2mBkKao573 --- src/batcontrol/core.py | 27 ++-- tests/batcontrol/test_mqtt_api.py | 65 ++++++++++ tests/batcontrol/test_solar_surplus.py | 163 +++++++++++++++++++++++++ 3 files changed, 242 insertions(+), 13 deletions(-) create mode 100644 tests/batcontrol/test_solar_surplus.py diff --git a/src/batcontrol/core.py b/src/batcontrol/core.py index ec20e1cb..50fd3c2e 100644 --- a/src/batcontrol/core.py +++ b/src/batcontrol/core.py @@ -684,7 +684,8 @@ def run(self): self.mqtt_api.publish_min_dynamic_price_diff( calc_output.min_dynamic_price_difference) phase, surplus_wh = self._compute_solar_surplus_and_phase( - production, consumption, calc_output.reserved_energy + production, consumption, calc_output.reserved_energy, + calc_input.free_capacity, calc_input.stored_usable_energy ) self.mqtt_api.publish_production_phase(phase) self.mqtt_api.publish_solar_surplus(surplus_wh) @@ -875,10 +876,15 @@ def _compute_solar_surplus_and_phase( self, production: np.ndarray, consumption: np.ndarray, - reserved_energy: float) -> tuple: + reserved_energy: float, + free_capacity: float, + stored_usable_energy: float) -> tuple: """Compute solar production phase and expected surplus energy. - Uses corrected arrays (elapsed-time-adjusted for slot [0]). + Uses corrected arrays (elapsed-time-adjusted for slot [0]) and + the optimizer's own input values for free_capacity and + stored_usable_energy so the published values are consistent with + the control decision that was just made. Returns: phase (str): 'before', 'during', or 'after' @@ -908,16 +914,12 @@ def _compute_solar_surplus_and_phase( else: phase = 'before' - free_capacity = self.get_free_capacity() - if phase in ('during', 'before'): end_idx = (production_end + 1) if production_end is not None else len(net_consumption) - net_surplus_wh = sum( - (-net_consumption[i]) * wh_per_slot - for i in range(0, end_idx) - if net_consumption[i] < 0 - ) - surplus_wh = max(0.0, net_surplus_wh - free_capacity) + # Net energy balance over the production window: positive = more production + # than consumption. Only what exceeds free battery capacity is real surplus. + net_window_wh = -np.sum(net_consumption[:end_idx]) * wh_per_slot + surplus_wh = max(0.0, net_window_wh - free_capacity) else: # Find when next significant solar production starts (net excess > 100 W) next_solar_start = len(net_consumption) @@ -930,8 +932,7 @@ def _compute_solar_surplus_and_phase( for nc in net_consumption[:next_solar_start] if nc > 0 ) - stored_usable = self.get_stored_usable_energy() - unreserved = max(0.0, stored_usable - reserved_energy) + unreserved = max(0.0, stored_usable_energy - reserved_energy) surplus_wh = max(0.0, unreserved - expected_consumption_wh) logger.debug( diff --git a/tests/batcontrol/test_mqtt_api.py b/tests/batcontrol/test_mqtt_api.py index 4baa2cb5..49a70db1 100644 --- a/tests/batcontrol/test_mqtt_api.py +++ b/tests/batcontrol/test_mqtt_api.py @@ -425,3 +425,68 @@ def test_invalid_mode_keeps_old_value(self): Batcontrol.api_set_peak_shaving_mode(bc, 'bogus') assert bc.peak_shaving_config is original bc.mqtt_api.publish_peak_shaving_mode.assert_not_called() + + +def _make_solar_publish_stub(): + """Stub for solar surplus / production phase publish tests.""" + api = MagicMock(spec=MqttApi) + api.base_topic = 'batcontrol' + api.client = MagicMock() + api.client.is_connected.return_value = True + api.publish_solar_surplus = MqttApi.publish_solar_surplus.__get__(api, MqttApi) + api.publish_production_phase = MqttApi.publish_production_phase.__get__(api, MqttApi) + return api + + +class TestPublishSolarSurplus: + def test_publishes_to_correct_topic(self): + api = _make_solar_publish_stub() + api.publish_solar_surplus(1234.5) + api.client.publish.assert_called_once_with( + 'batcontrol/solar_surplus_wh', '1234.5' + ) + + def test_formats_with_one_decimal(self): + api = _make_solar_publish_stub() + api.publish_solar_surplus(999.987) + topic, payload = api.client.publish.call_args[0] + assert payload == '1000.0' + + def test_zero_surplus_published(self): + api = _make_solar_publish_stub() + api.publish_solar_surplus(0.0) + _, payload = api.client.publish.call_args[0] + assert payload == '0.0' + + def test_skips_publish_when_disconnected(self): + api = _make_solar_publish_stub() + api.client.is_connected.return_value = False + api.publish_solar_surplus(500.0) + api.client.publish.assert_not_called() + + +class TestPublishProductionPhase: + def test_publishes_during_to_correct_topic(self): + api = _make_solar_publish_stub() + api.publish_production_phase('during') + api.client.publish.assert_called_once_with( + 'batcontrol/solar_production_phase', 'during' + ) + + def test_publishes_before(self): + api = _make_solar_publish_stub() + api.publish_production_phase('before') + _, payload = api.client.publish.call_args[0] + assert payload == 'before' + + def test_publishes_after(self): + api = _make_solar_publish_stub() + api.publish_production_phase('after') + _, payload = api.client.publish.call_args[0] + assert payload == 'after' + + def test_skips_publish_when_disconnected(self): + api = _make_solar_publish_stub() + api.client.is_connected.return_value = False + api.publish_production_phase('during') + api.client.publish.assert_not_called() diff --git a/tests/batcontrol/test_solar_surplus.py b/tests/batcontrol/test_solar_surplus.py new file mode 100644 index 00000000..7119db5d --- /dev/null +++ b/tests/batcontrol/test_solar_surplus.py @@ -0,0 +1,163 @@ +"""Tests for Batcontrol._compute_solar_surplus_and_phase.""" +import numpy as np +import pytest +from unittest.mock import MagicMock + +from batcontrol.core import Batcontrol + + +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_surplus_and_phase = ( + Batcontrol._compute_solar_surplus_and_phase.__get__(stub, Batcontrol) + ) + return stub + + +def _call(stub, production, consumption, reserved=0.0, free_cap=0.0, stored_usable=0.0): + return stub._compute_solar_surplus_and_phase( + np.array(production, dtype=float), + np.array(consumption, dtype=float), + reserved, free_cap, stored_usable, + ) + + +class TestPhaseDetection: + def test_phase_during_when_production_starts_at_slot0(self): + stub = _make_core() + phase, _ = _call(stub, [1000, 1500, 500], [400, 400, 400]) + assert phase == 'during' + + def test_phase_before_when_production_starts_later(self): + stub = _make_core() + phase, _ = _call(stub, [0, 0, 800, 1200, 0], [300, 300, 300, 300, 300]) + assert phase == 'before' + + def test_phase_after_when_no_production_at_all(self): + stub = _make_core() + phase, _ = _call(stub, [0, 0, 0, 0], [300, 400, 350, 300]) + assert phase == 'after' + + def test_phase_after_when_all_production_is_zero_float(self): + stub = _make_core() + phase, _ = _call(stub, [0.0, 0.0, 0.0], [200.0, 200.0, 200.0]) + assert phase == 'after' + + def test_phase_during_tolerates_single_slot_gap(self): + # Slot 0 producing, slot 1 gap, slot 2 producing again + stub = _make_core() + phase, _ = _call(stub, [800, 0, 600, 0], [300, 300, 300, 300]) + assert phase == 'during' + + def test_phase_before_with_15min_resolution(self): + stub = _make_core(time_resolution=15) + # 4 empty slots (1h) then solar + phase, _ = _call(stub, [0, 0, 0, 0, 500, 800, 600], [200] * 7) + assert phase == 'before' + + +class TestSurplusDuringBefore: + def test_surplus_zero_when_net_production_fits_in_battery(self): + # 2h * 1000 W net = 2000 Wh net, free_capacity 3000 Wh -> fits, no surplus + stub = _make_core() + production = [1500, 1500, 0] + consumption = [500, 500, 500] + _, surplus = _call(stub, production, consumption, 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 Wh -> surplus = 800 Wh + stub = _make_core() + production = [1500, 1500, 0] + consumption = [500, 500, 0] + _, surplus = _call(stub, production, consumption, free_cap=1200.0) + assert surplus == pytest.approx(800.0) + + def test_surplus_accounts_for_consumption_in_window(self): + # Slot 0: 2000 W prod, 1500 W cons -> +500 W net + # Slot 1: 2000 W prod, 2500 W cons -> -500 W net + # Total net = 0 Wh -> no surplus regardless of free_cap + stub = _make_core() + production = [2000, 2000, 0] + consumption = [1500, 2500, 400] + _, surplus = _call(stub, production, consumption, free_cap=0.0) + assert surplus == pytest.approx(0.0) + + def test_surplus_never_negative(self): + stub = _make_core() + production = [100, 100, 0] + consumption = [800, 800, 800] + _, surplus = _call(stub, production, consumption, free_cap=10000.0) + assert surplus == 0.0 + + def test_before_phase_uses_full_window_including_pre_production_consumption(self): + # Slot 0: no production, 500 W consumption (eats into net) + # Slot 1-2: 2000 W prod, 500 W cons each -> +1500 W * 2 = 3000 Wh net + # Net window (slots 0-2) = -500 + 1500 + 1500 = 2500 Wh + # free_cap = 2000 -> surplus = 500 + stub = _make_core() + production = [0, 2000, 2000, 0] + consumption = [500, 500, 500, 500] + _, surplus = _call(stub, production, consumption, free_cap=2000.0) + assert surplus == pytest.approx(500.0) + + def test_15min_resolution_scales_wh_correctly(self): + # Each slot is 15 min = 0.25 h + # production=2000W, consumption=0 for 4 slots = 4 * 2000 * 0.25 = 2000 Wh net + # free_cap=1500 -> surplus = 500 Wh + stub = _make_core(time_resolution=15) + production = [2000, 2000, 2000, 2000, 0] + consumption = [0, 0, 0, 0, 0] + _, surplus = _call(stub, production, consumption, free_cap=1500.0) + assert surplus == pytest.approx(500.0) + + +class TestSurplusAfter: + def test_surplus_positive_when_stored_exceeds_expected_consumption(self): + # stored_usable=5000, reserved=1000 -> unreserved=4000 + # next 3 slots all consuming 500 Wh -> expected = 1500 Wh + # surplus = 4000 - 1500 = 2500 Wh + stub = _make_core() + production = [0, 0, 0] + consumption = [500, 500, 500] + _, surplus = _call(stub, production, consumption, + reserved=1000.0, stored_usable=5000.0) + assert surplus == pytest.approx(2500.0) + + def test_surplus_zero_when_stored_less_than_consumption(self): + stub = _make_core() + production = [0, 0, 0] + consumption = [1000, 1000, 1000] + _, surplus = _call(stub, production, consumption, + reserved=0.0, stored_usable=1500.0) + assert surplus == pytest.approx(0.0) + + def test_sums_all_consumption_slots_when_no_future_solar(self): + # No production at all -> next_solar_start = 4 (end of array) + # All 4 slots consume 500 W = 4 * 500 * 1h = 2000 Wh expected + # stored=5000, reserved=0 -> unreserved=5000 -> surplus=3000 + stub = _make_core() + production = [0, 0, 0, 0] + consumption = [500, 500, 500, 500] + _, surplus = _call(stub, production, consumption, + reserved=0.0, stored_usable=5000.0) + assert surplus == pytest.approx(3000.0) + + def test_reserved_energy_reduces_usable(self): + stub = _make_core() + production = [0, 0] + consumption = [0, 0] + # stored=2000, reserved=1000 -> unreserved=1000, consumption=0 -> surplus=1000 + _, surplus = _call(stub, production, consumption, + reserved=1000.0, stored_usable=2000.0) + assert surplus == pytest.approx(1000.0) + + def test_surplus_never_negative_when_reserved_exceeds_stored(self): + stub = _make_core() + production = [0] + consumption = [100] + _, surplus = _call(stub, production, consumption, + reserved=5000.0, stored_usable=1000.0) + assert surplus == 0.0 From b9f1e1fee246b6bb969d12a5e1a896d1fb2bfd8e Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 28 May 2026 07:02:58 +0000 Subject: [PATCH 3/4] Fix solar surplus calculation: forecast-based formula, correct window scoping Three bugs fixed in _compute_solar_surplus_and_phase: - Remove wh_per_slot multiplier (arrays are already Wh/slot, not W) - Use first production window only (not last slot in multi-day forecast) - Replace reserved_energy/stored_usable with forecast-based bridge calculation New formula for 'before'/'after': expected solar overflow at end of next production window = max(0, solar_net_next - free_capacity - bridge_wh), where bridge_wh is forecast consumption until solar starts. Identical formula for both phases; phase label differs only by threshold distance. Add before_phase_threshold_h (default 4h) to battery_control_expert config. --- src/batcontrol/core.py | 69 ++++------ tests/batcontrol/test_solar_surplus.py | 182 ++++++++++++++----------- 2 files changed, 134 insertions(+), 117 deletions(-) diff --git a/src/batcontrol/core.py b/src/batcontrol/core.py index 50fd3c2e..b9e7bc04 100644 --- a/src/batcontrol/core.py +++ b/src/batcontrol/core.py @@ -250,6 +250,7 @@ def __init__(self, configdict: dict): self.round_price_digits = 4 self.production_offset_percent = 1.0 # Default: no offset + self.before_phase_threshold_h = 4.0 self.market_price_refresh_time = "12:30" if self.config.get('battery_control_expert', None) is not None: @@ -264,6 +265,9 @@ def __init__(self, configdict: dict): self.preserve_min_grid_charge_soc = battery_control_expert.get( 'preserve_min_grid_charge_soc', self.preserve_min_grid_charge_soc) + self.before_phase_threshold_h = float(battery_control_expert.get( + 'before_phase_threshold_h', + self.before_phase_threshold_h)) raw_refresh_time = battery_control_expert.get( 'market_price_refresh_time', self.market_price_refresh_time) @@ -684,8 +688,7 @@ def run(self): self.mqtt_api.publish_min_dynamic_price_diff( calc_output.min_dynamic_price_difference) phase, surplus_wh = self._compute_solar_surplus_and_phase( - production, consumption, calc_output.reserved_energy, - calc_input.free_capacity, calc_input.stored_usable_energy + production, consumption, calc_input.free_capacity ) self.mqtt_api.publish_production_phase(phase) self.mqtt_api.publish_solar_surplus(surplus_wh) @@ -876,64 +879,52 @@ def _compute_solar_surplus_and_phase( self, production: np.ndarray, consumption: np.ndarray, - reserved_energy: float, - free_capacity: float, - stored_usable_energy: float) -> tuple: + free_capacity: float) -> tuple: """Compute solar production phase and expected surplus energy. - Uses corrected arrays (elapsed-time-adjusted for slot [0]) and - the optimizer's own input values for free_capacity and - stored_usable_energy so the published values are consistent with - the control decision that was just made. - Returns: - phase (str): 'before', 'during', or 'after' - surplus_wh (float): Expected usable surplus in Wh (>0 = WP can run) + phase (str): 'during', 'before', or 'after' + surplus_wh (float): Expected solar overflow in Wh (>0 = WP can run) Phase semantics: - before: production has not started yet (production[0] == 0, future slots > 0) - during: solar production is happening right now (production[0] > 0) - after: no production remaining in forecast window + during: solar is running now (production[0] > 0) + before: solar starts within before_phase_threshold_h hours + after: solar is beyond threshold or not in forecast """ - wh_per_slot = self.time_resolution / 60.0 net_consumption = consumption - production + threshold_slots = round(self.before_phase_threshold_h * 60 / self.time_resolution) - # Locate production window, tolerating single-slot gaps + # Find start and end of the FIRST production window only production_start: Optional[int] = None - production_end: 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 = i + production_end_current = i + elif production_start is not None: + break # stop at first zero slot after production started - if production_start is None: + if production_start is None or production_start > threshold_slots: phase = 'after' elif production_start == 0: phase = 'during' else: phase = 'before' - if phase in ('during', 'before'): - end_idx = (production_end + 1) if production_end is not None else len(net_consumption) - # Net energy balance over the production window: positive = more production - # than consumption. Only what exceeds free battery capacity is real surplus. - net_window_wh = -np.sum(net_consumption[:end_idx]) * wh_per_slot - surplus_wh = max(0.0, net_window_wh - free_capacity) + if phase == 'during': + end_idx = (production_end_current + 1) if production_end_current is not None else 1 + solar_net_wh = float(-np.sum(net_consumption[:end_idx])) + surplus_wh = max(0.0, solar_net_wh - free_capacity) else: - # Find when next significant solar production starts (net excess > 100 W) - next_solar_start = len(net_consumption) - for i, nc in enumerate(net_consumption): - if nc < -100: - next_solar_start = i - break - expected_consumption_wh = sum( - nc * wh_per_slot - for nc in net_consumption[:next_solar_start] - if nc > 0 - ) - unreserved = max(0.0, stored_usable_energy - reserved_energy) - surplus_wh = max(0.0, unreserved - expected_consumption_wh) + 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 phase: %s, surplus: %.1f Wh (free_cap=%.1f Wh)', diff --git a/tests/batcontrol/test_solar_surplus.py b/tests/batcontrol/test_solar_surplus.py index 7119db5d..f8503fa7 100644 --- a/tests/batcontrol/test_solar_surplus.py +++ b/tests/batcontrol/test_solar_surplus.py @@ -6,21 +6,22 @@ from batcontrol.core import Batcontrol -def _make_core(time_resolution=60): +def _make_core(time_resolution=60, before_phase_threshold_h=4.0): """Return a minimal stub with only the attributes used by the method.""" stub = MagicMock(spec=Batcontrol) stub.time_resolution = time_resolution + stub.before_phase_threshold_h = before_phase_threshold_h stub._compute_solar_surplus_and_phase = ( Batcontrol._compute_solar_surplus_and_phase.__get__(stub, Batcontrol) ) return stub -def _call(stub, production, consumption, reserved=0.0, free_cap=0.0, stored_usable=0.0): +def _call(stub, production, consumption, free_cap=0.0): return stub._compute_solar_surplus_and_phase( np.array(production, dtype=float), np.array(consumption, dtype=float), - reserved, free_cap, stored_usable, + free_cap, ) @@ -30,7 +31,8 @@ def test_phase_during_when_production_starts_at_slot0(self): phase, _ = _call(stub, [1000, 1500, 500], [400, 400, 400]) assert phase == 'during' - def test_phase_before_when_production_starts_later(self): + def test_phase_before_when_production_starts_within_threshold(self): + # production_start=2, threshold=4 slots (4h at 60-min) -> before stub = _make_core() phase, _ = _call(stub, [0, 0, 800, 1200, 0], [300, 300, 300, 300, 300]) assert phase == 'before' @@ -46,118 +48,142 @@ def test_phase_after_when_all_production_is_zero_float(self): assert phase == 'after' def test_phase_during_tolerates_single_slot_gap(self): - # Slot 0 producing, slot 1 gap, slot 2 producing again + # Slot 0 producing, slot 1 gap -> still 'during' because production_start==0 stub = _make_core() phase, _ = _call(stub, [800, 0, 600, 0], [300, 300, 300, 300]) assert phase == 'during' + def test_phase_after_when_production_start_beyond_threshold(self): + # threshold=4 slots, production_start=5 -> after + stub = _make_core() + phase, _ = _call(stub, [0, 0, 0, 0, 0, 1000, 1000, 0], [200] * 8) + assert phase == 'after' + + def test_phase_before_at_threshold_boundary(self): + # production_start == threshold_slots (4) -> before (inclusive) + stub = _make_core() + phase, _ = _call(stub, [0, 0, 0, 0, 1000, 0], [200] * 6) + assert phase == 'before' + + def test_threshold_slots_respect_time_resolution(self): + # 4h threshold at 15-min resolution = 16 slots + stub = _make_core(time_resolution=15) + # production_start=16 -> exactly at threshold -> before + prod = [0] * 16 + [500, 800, 600] + cons = [200] * 19 + phase, _ = _call(stub, prod, cons) + assert phase == 'before' + # production_start=17 -> beyond threshold -> after + prod2 = [0] * 17 + [500, 800] + cons2 = [200] * 19 + phase2, _ = _call(stub, prod2, cons2) + assert phase2 == 'after' + def test_phase_before_with_15min_resolution(self): stub = _make_core(time_resolution=15) - # 4 empty slots (1h) then solar + # 4 empty slots (1h) then solar; threshold=16 slots -> before phase, _ = _call(stub, [0, 0, 0, 0, 500, 800, 600], [200] * 7) assert phase == 'before' -class TestSurplusDuringBefore: +class TestSurplusDuring: def test_surplus_zero_when_net_production_fits_in_battery(self): - # 2h * 1000 W net = 2000 Wh net, free_capacity 3000 Wh -> fits, no surplus + # net = 1000+1000 = 2000 Wh, free_cap=3000 -> fits, no surplus stub = _make_core() - production = [1500, 1500, 0] - consumption = [500, 500, 500] - _, surplus = _call(stub, production, consumption, free_cap=3000.0) + _, surplus = _call(stub, [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 Wh -> surplus = 800 Wh + # net = 1000+1000 = 2000 Wh, free_cap=1200 -> surplus=800 stub = _make_core() - production = [1500, 1500, 0] - consumption = [500, 500, 0] - _, surplus = _call(stub, production, consumption, free_cap=1200.0) + _, surplus = _call(stub, [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): - # Slot 0: 2000 W prod, 1500 W cons -> +500 W net - # Slot 1: 2000 W prod, 2500 W cons -> -500 W net - # Total net = 0 Wh -> no surplus regardless of free_cap + # slot0: +500 net, slot1: -500 net -> total=0, no surplus stub = _make_core() - production = [2000, 2000, 0] - consumption = [1500, 2500, 400] - _, surplus = _call(stub, production, consumption, free_cap=0.0) + _, surplus = _call(stub, [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() - production = [100, 100, 0] - consumption = [800, 800, 800] - _, surplus = _call(stub, production, consumption, free_cap=10000.0) + _, surplus = _call(stub, [100, 100, 0], [800, 800, 800], free_cap=10000.0) assert surplus == 0.0 - def test_before_phase_uses_full_window_including_pre_production_consumption(self): - # Slot 0: no production, 500 W consumption (eats into net) - # Slot 1-2: 2000 W prod, 500 W cons each -> +1500 W * 2 = 3000 Wh net - # Net window (slots 0-2) = -500 + 1500 + 1500 = 2500 Wh - # free_cap = 2000 -> surplus = 500 + def test_during_uses_only_first_production_window(self): + # 48h forecast: today solar slots 0-1, tonight/tomorrow no solar, tomorrow solar again + # Must NOT include tomorrow's solar in 'during' surplus stub = _make_core() - production = [0, 2000, 2000, 0] - consumption = [500, 500, 500, 500] - _, surplus = _call(stub, production, consumption, free_cap=2000.0) - assert surplus == pytest.approx(500.0) + today_solar = [1500, 1500] # 2000 Wh net production + night = [0] * 12 # 12h night + tomorrow_solar = [1500, 1500] + production = today_solar + night + tomorrow_solar + consumption = [500] * len(production) + # production_end_current stops at slot 1 (first zero is slot 2) + # solar_net = -sum(nc[:2]) = -(500-1500 + 500-1500) = 2000 Wh + # free_cap=1200 -> surplus=800 + _, surplus = _call(stub, production, consumption, free_cap=1200.0) + assert surplus == pytest.approx(800.0) - def test_15min_resolution_scales_wh_correctly(self): - # Each slot is 15 min = 0.25 h - # production=2000W, consumption=0 for 4 slots = 4 * 2000 * 0.25 = 2000 Wh net - # free_cap=1500 -> surplus = 500 Wh - stub = _make_core(time_resolution=15) - production = [2000, 2000, 2000, 2000, 0] - consumption = [0, 0, 0, 0, 0] - _, surplus = _call(stub, production, consumption, free_cap=1500.0) - assert surplus == pytest.approx(500.0) +class TestSurplusBeforeAfter: + 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) + assert surplus == pytest.approx(0.0) -class TestSurplusAfter: - def test_surplus_positive_when_stored_exceeds_expected_consumption(self): - # stored_usable=5000, reserved=1000 -> unreserved=4000 - # next 3 slots all consuming 500 Wh -> expected = 1500 Wh - # surplus = 4000 - 1500 = 2500 Wh + 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, 0] - consumption = [500, 500, 500] - _, surplus = _call(stub, production, consumption, - reserved=1000.0, stored_usable=5000.0) - assert surplus == pytest.approx(2500.0) + production = [0, 0, 1000, 1000, 0] + consumption = [200, 200, 200, 200, 200] + _, surplus = _call(stub, production, consumption, free_cap=500.0) + assert surplus == pytest.approx(700.0) - def test_surplus_zero_when_stored_less_than_consumption(self): + 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, 0] - consumption = [1000, 1000, 1000] - _, surplus = _call(stub, production, consumption, - reserved=0.0, stored_usable=1500.0) + production = [0, 0, 1000, 0] + consumption = [200, 200, 200, 200] + _, surplus = _call(stub, production, consumption, free_cap=2000.0) assert surplus == pytest.approx(0.0) - def test_sums_all_consumption_slots_when_no_future_solar(self): - # No production at all -> next_solar_start = 4 (end of array) - # All 4 slots consume 500 W = 4 * 500 * 1h = 2000 Wh expected - # stored=5000, reserved=0 -> unreserved=5000 -> surplus=3000 + 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, 0, 0, 0] + production = [0, 2000, 2000, 0] consumption = [500, 500, 500, 500] - _, surplus = _call(stub, production, consumption, - reserved=0.0, stored_usable=5000.0) - assert surplus == pytest.approx(3000.0) + _, surplus = _call(stub, production, consumption, free_cap=2000.0) + assert surplus == pytest.approx(500.0) - def test_reserved_energy_reduces_usable(self): - stub = _make_core() - production = [0, 0] - consumption = [0, 0] - # stored=2000, reserved=1000 -> unreserved=1000, consumption=0 -> surplus=1000 - _, surplus = _call(stub, production, consumption, - reserved=1000.0, stored_usable=2000.0) - assert surplus == pytest.approx(1000.0) + def test_before_and_after_use_same_formula(self): + # Same inputs, only threshold changes: one gives 'before', other 'after', surplus same + production = [0, 0, 1000, 1000, 0] + consumption = [200, 200, 200, 200, 200] + stub_before = _make_core(before_phase_threshold_h=4.0) # start=2 <= 4 -> before + stub_after = _make_core(before_phase_threshold_h=1.0) # start=2 > 1 -> after + phase_before, surplus_before = _call(stub_before, production, consumption, free_cap=500.0) + phase_after, surplus_after = _call(stub_after, production, consumption, free_cap=500.0) + assert phase_before == 'before' + assert phase_after == 'after' + assert surplus_before == pytest.approx(surplus_after) - def test_surplus_never_negative_when_reserved_exceeds_stored(self): + def test_surplus_never_negative(self): stub = _make_core() - production = [0] - consumption = [100] - _, surplus = _call(stub, production, consumption, - reserved=5000.0, stored_usable=1000.0) + _, surplus = _call(stub, [0, 0, 100, 0], [500, 500, 500, 500], free_cap=10000.0) assert surplus == 0.0 + + def test_before_only_uses_first_production_window(self): + # 48h: night, tomorrow solar window (slots 2-3), second night, day-after solar (slots 8-9) + # Must NOT include day-after solar in 'before' calculation + 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) + assert surplus == pytest.approx(700.0) From 3da9dc491d314e32d6187a4171aba3798c3c21f2 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 28 May 2026 07:08:05 +0000 Subject: [PATCH 4/4] Collapse production_phase to solar_active boolean 'before' and 'after' used identical formulas; only the label differed. Replace the 3-state string with a binary signal: solar_active=True iff production[0] > 0. Drops the unused before_phase_threshold_h config. MQTT topic renamed: solar_production_phase -> solar_active (binary_sensor in HA discovery). Consumers should switch to solar_surplus_wh + solar_active. --- src/batcontrol/core.py | 57 +++++------- src/batcontrol/mqtt_api.py | 26 +++--- tests/batcontrol/test_mqtt_api.py | 26 +++--- tests/batcontrol/test_solar_surplus.py | 120 ++++++++----------------- 4 files changed, 82 insertions(+), 147 deletions(-) diff --git a/src/batcontrol/core.py b/src/batcontrol/core.py index b9e7bc04..279762bb 100644 --- a/src/batcontrol/core.py +++ b/src/batcontrol/core.py @@ -250,7 +250,6 @@ def __init__(self, configdict: dict): self.round_price_digits = 4 self.production_offset_percent = 1.0 # Default: no offset - self.before_phase_threshold_h = 4.0 self.market_price_refresh_time = "12:30" if self.config.get('battery_control_expert', None) is not None: @@ -265,9 +264,6 @@ def __init__(self, configdict: dict): self.preserve_min_grid_charge_soc = battery_control_expert.get( 'preserve_min_grid_charge_soc', self.preserve_min_grid_charge_soc) - self.before_phase_threshold_h = float(battery_control_expert.get( - 'before_phase_threshold_h', - self.before_phase_threshold_h)) raw_refresh_time = battery_control_expert.get( 'market_price_refresh_time', self.market_price_refresh_time) @@ -687,10 +683,10 @@ def run(self): if self.mqtt_api is not None: self.mqtt_api.publish_min_dynamic_price_diff( calc_output.min_dynamic_price_difference) - phase, surplus_wh = self._compute_solar_surplus_and_phase( + solar_active, surplus_wh = self._compute_solar_active_and_surplus( production, consumption, calc_input.free_capacity ) - self.mqtt_api.publish_production_phase(phase) + self.mqtt_api.publish_solar_active(solar_active) self.mqtt_api.publish_solar_surplus(surplus_wh) if self.discharge_blocked and not \ @@ -875,24 +871,23 @@ def get_reserved_energy(self) -> float: """ Returns the reserved energy in Wh from last calculation """ return self.last_reserved_energy - def _compute_solar_surplus_and_phase( + def _compute_solar_active_and_surplus( self, production: np.ndarray, consumption: np.ndarray, free_capacity: float) -> tuple: - """Compute solar production phase and expected surplus energy. + """Compute solar-active flag and expected surplus energy. Returns: - phase (str): 'during', 'before', or 'after' + solar_active (bool): True iff solar is producing in slot 0 surplus_wh (float): Expected solar overflow in Wh (>0 = WP can run) - Phase semantics: - during: solar is running now (production[0] > 0) - before: solar starts within before_phase_threshold_h hours - after: solar is beyond threshold or not in forecast + 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 - threshold_slots = round(self.before_phase_threshold_h * 60 / self.time_resolution) # Find start and end of the FIRST production window only production_start: Optional[int] = None @@ -903,34 +898,24 @@ def _compute_solar_surplus_and_phase( production_start = i production_end_current = i elif production_start is not None: - break # stop at first zero slot after production started + break - if production_start is None or production_start > threshold_slots: - phase = 'after' - elif production_start == 0: - phase = 'during' - else: - phase = 'before' + solar_active = production_start == 0 - if phase == 'during': - end_idx = (production_end_current + 1) if production_end_current is not None else 1 - solar_net_wh = float(-np.sum(net_consumption[:end_idx])) - surplus_wh = max(0.0, solar_net_wh - free_capacity) + if production_start is None: + surplus_wh = 0.0 else: - 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) + 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 phase: %s, surplus: %.1f Wh (free_cap=%.1f Wh)', - phase, surplus_wh, free_capacity + 'Solar active: %s, surplus: %.1f Wh (free_cap=%.1f Wh)', + solar_active, surplus_wh, free_capacity ) - return phase, surplus_wh + return solar_active, surplus_wh 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 682e7be6..bb75b12b 100644 --- a/src/batcontrol/mqtt_api.py +++ b/src/batcontrol/mqtt_api.py @@ -27,7 +27,7 @@ - /api_override_active: bool indicating whether a temporary external/API override is active - /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_production_phase: solar production phase ('before', 'during', 'after') +- /solar_active: bool indicating whether solar is currently producing (slot 0 > 0) The following statistical arrays are published as JSON arrays: - /FCST/production: forecasted production in W @@ -498,16 +498,15 @@ def publish_solar_surplus(self, surplus_wh: float) -> None: f'{surplus_wh:.1f}' ) - def publish_production_phase(self, phase: str) -> None: - """ Publish the solar production phase to MQTT. - /solar_production_phase - Values: 'before' (production expected later), 'during' (producing now), - 'after' (no more production in forecast). + def publish_solar_active(self, active: bool) -> None: + """ Publish whether solar is currently producing. + /solar_active """ if self.client.is_connected(): self.client.publish( - self.base_topic + '/solar_production_phase', - phase + self.base_topic + '/solar_active', + str(active).lower(), + retain=True ) def publish_api_override_active(self, active: bool) -> None: @@ -910,13 +909,14 @@ def send_mqtt_discovery_messages(self) -> None: self.base_topic + "/solar_surplus_wh") self.publish_mqtt_discovery_message( - "Solar Production Phase", - "batcontrol_solar_production_phase", - "sensor", + "Solar Active", + "batcontrol_solar_active", + "binary_sensor", None, None, - self.base_topic + "/solar_production_phase", - entity_category="diagnostic") + self.base_topic + "/solar_active", + entity_category="diagnostic", + value_template="{% if value == 'true' %}ON{% else %}OFF{% endif %}") def send_mqtt_discovery_for_mode(self) -> None: """ Publish Home Assistant MQTT Auto Discovery message for mode""" diff --git a/tests/batcontrol/test_mqtt_api.py b/tests/batcontrol/test_mqtt_api.py index 49a70db1..b4f4e0ca 100644 --- a/tests/batcontrol/test_mqtt_api.py +++ b/tests/batcontrol/test_mqtt_api.py @@ -428,13 +428,13 @@ def test_invalid_mode_keeps_old_value(self): def _make_solar_publish_stub(): - """Stub for solar surplus / production phase publish tests.""" + """Stub for solar surplus / solar_active publish tests.""" api = MagicMock(spec=MqttApi) api.base_topic = 'batcontrol' api.client = MagicMock() api.client.is_connected.return_value = True api.publish_solar_surplus = MqttApi.publish_solar_surplus.__get__(api, MqttApi) - api.publish_production_phase = MqttApi.publish_production_phase.__get__(api, MqttApi) + api.publish_solar_active = MqttApi.publish_solar_active.__get__(api, MqttApi) return api @@ -465,28 +465,22 @@ def test_skips_publish_when_disconnected(self): api.client.publish.assert_not_called() -class TestPublishProductionPhase: - def test_publishes_during_to_correct_topic(self): +class TestPublishSolarActive: + def test_publishes_true_to_correct_topic(self): api = _make_solar_publish_stub() - api.publish_production_phase('during') + api.publish_solar_active(True) api.client.publish.assert_called_once_with( - 'batcontrol/solar_production_phase', 'during' + 'batcontrol/solar_active', 'true', retain=True ) - def test_publishes_before(self): + def test_publishes_false(self): api = _make_solar_publish_stub() - api.publish_production_phase('before') + api.publish_solar_active(False) _, payload = api.client.publish.call_args[0] - assert payload == 'before' - - def test_publishes_after(self): - api = _make_solar_publish_stub() - api.publish_production_phase('after') - _, payload = api.client.publish.call_args[0] - assert payload == 'after' + assert payload == 'false' def test_skips_publish_when_disconnected(self): api = _make_solar_publish_stub() api.client.is_connected.return_value = False - api.publish_production_phase('during') + api.publish_solar_active(True) api.client.publish.assert_not_called() diff --git a/tests/batcontrol/test_solar_surplus.py b/tests/batcontrol/test_solar_surplus.py index f8503fa7..34ba09d2 100644 --- a/tests/batcontrol/test_solar_surplus.py +++ b/tests/batcontrol/test_solar_surplus.py @@ -1,4 +1,4 @@ -"""Tests for Batcontrol._compute_solar_surplus_and_phase.""" +"""Tests for Batcontrol._compute_solar_active_and_surplus.""" import numpy as np import pytest from unittest.mock import MagicMock @@ -6,87 +6,47 @@ from batcontrol.core import Batcontrol -def _make_core(time_resolution=60, before_phase_threshold_h=4.0): +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.before_phase_threshold_h = before_phase_threshold_h - stub._compute_solar_surplus_and_phase = ( - Batcontrol._compute_solar_surplus_and_phase.__get__(stub, Batcontrol) + 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_surplus_and_phase( + return stub._compute_solar_active_and_surplus( np.array(production, dtype=float), np.array(consumption, dtype=float), free_cap, ) -class TestPhaseDetection: - def test_phase_during_when_production_starts_at_slot0(self): +class TestSolarActive: + def test_active_when_production_starts_at_slot0(self): stub = _make_core() - phase, _ = _call(stub, [1000, 1500, 500], [400, 400, 400]) - assert phase == 'during' + active, _ = _call(stub, [1000, 1500, 500], [400, 400, 400]) + assert active is True - def test_phase_before_when_production_starts_within_threshold(self): - # production_start=2, threshold=4 slots (4h at 60-min) -> before + def test_inactive_when_production_starts_later(self): stub = _make_core() - phase, _ = _call(stub, [0, 0, 800, 1200, 0], [300, 300, 300, 300, 300]) - assert phase == 'before' + active, _ = _call(stub, [0, 0, 800, 1200, 0], [300, 300, 300, 300, 300]) + assert active is False - def test_phase_after_when_no_production_at_all(self): + def test_inactive_when_no_production_at_all(self): stub = _make_core() - phase, _ = _call(stub, [0, 0, 0, 0], [300, 400, 350, 300]) - assert phase == 'after' + active, _ = _call(stub, [0, 0, 0, 0], [300, 400, 350, 300]) + assert active is False - def test_phase_after_when_all_production_is_zero_float(self): + def test_active_when_slot0_producing_even_with_gap_after(self): stub = _make_core() - phase, _ = _call(stub, [0.0, 0.0, 0.0], [200.0, 200.0, 200.0]) - assert phase == 'after' - - def test_phase_during_tolerates_single_slot_gap(self): - # Slot 0 producing, slot 1 gap -> still 'during' because production_start==0 - stub = _make_core() - phase, _ = _call(stub, [800, 0, 600, 0], [300, 300, 300, 300]) - assert phase == 'during' - - def test_phase_after_when_production_start_beyond_threshold(self): - # threshold=4 slots, production_start=5 -> after - stub = _make_core() - phase, _ = _call(stub, [0, 0, 0, 0, 0, 1000, 1000, 0], [200] * 8) - assert phase == 'after' - - def test_phase_before_at_threshold_boundary(self): - # production_start == threshold_slots (4) -> before (inclusive) - stub = _make_core() - phase, _ = _call(stub, [0, 0, 0, 0, 1000, 0], [200] * 6) - assert phase == 'before' - - def test_threshold_slots_respect_time_resolution(self): - # 4h threshold at 15-min resolution = 16 slots - stub = _make_core(time_resolution=15) - # production_start=16 -> exactly at threshold -> before - prod = [0] * 16 + [500, 800, 600] - cons = [200] * 19 - phase, _ = _call(stub, prod, cons) - assert phase == 'before' - # production_start=17 -> beyond threshold -> after - prod2 = [0] * 17 + [500, 800] - cons2 = [200] * 19 - phase2, _ = _call(stub, prod2, cons2) - assert phase2 == 'after' - - def test_phase_before_with_15min_resolution(self): - stub = _make_core(time_resolution=15) - # 4 empty slots (1h) then solar; threshold=16 slots -> before - phase, _ = _call(stub, [0, 0, 0, 0, 500, 800, 600], [200] * 7) - assert phase == 'before' + active, _ = _call(stub, [800, 0, 600, 0], [300, 300, 300, 300]) + assert active is True -class TestSurplusDuring: +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() @@ -110,23 +70,21 @@ def test_surplus_never_negative(self): _, surplus = _call(stub, [100, 100, 0], [800, 800, 800], free_cap=10000.0) assert surplus == 0.0 - def test_during_uses_only_first_production_window(self): - # 48h forecast: today solar slots 0-1, tonight/tomorrow no solar, tomorrow solar again - # Must NOT include tomorrow's solar in 'during' surplus + 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 # 12h night + night = [0] * 12 tomorrow_solar = [1500, 1500] production = today_solar + night + tomorrow_solar consumption = [500] * len(production) - # production_end_current stops at slot 1 (first zero is slot 2) - # solar_net = -sum(nc[:2]) = -(500-1500 + 500-1500) = 2000 Wh - # free_cap=1200 -> surplus=800 + # solar_net = -(500-1500 + 500-1500) = 2000 Wh, free_cap=1200 -> surplus=800 _, surplus = _call(stub, production, consumption, free_cap=1200.0) assert surplus == pytest.approx(800.0) -class TestSurplusBeforeAfter: +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) @@ -160,26 +118,13 @@ def test_night_discharge_creates_room_for_solar(self): _, surplus = _call(stub, production, consumption, free_cap=2000.0) assert surplus == pytest.approx(500.0) - def test_before_and_after_use_same_formula(self): - # Same inputs, only threshold changes: one gives 'before', other 'after', surplus same - production = [0, 0, 1000, 1000, 0] - consumption = [200, 200, 200, 200, 200] - stub_before = _make_core(before_phase_threshold_h=4.0) # start=2 <= 4 -> before - stub_after = _make_core(before_phase_threshold_h=1.0) # start=2 > 1 -> after - phase_before, surplus_before = _call(stub_before, production, consumption, free_cap=500.0) - phase_after, surplus_after = _call(stub_after, production, consumption, free_cap=500.0) - assert phase_before == 'before' - assert phase_after == 'after' - assert surplus_before == pytest.approx(surplus_after) - def test_surplus_never_negative(self): stub = _make_core() _, surplus = _call(stub, [0, 0, 100, 0], [500, 500, 500, 500], free_cap=10000.0) assert surplus == 0.0 - def test_before_only_uses_first_production_window(self): - # 48h: night, tomorrow solar window (slots 2-3), second night, day-after solar (slots 8-9) - # Must NOT include day-after solar in 'before' calculation + 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 @@ -187,3 +132,14 @@ def test_before_only_uses_first_production_window(self): # surplus = max(0, 1600-500-400) = 700 (day-after ignored) _, surplus = _call(stub, 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) + assert surplus == pytest.approx(400.0)