tests: add logic scenario helpers

Author: filiplajszczak

tests/batcontrol/logic/conftest.py

@@ -0,0 +1,12 @@
+"""Shared fixtures for logic tests."""
+import pytest
+
+from batcontrol.logic.common import CommonLogic
+
+
+@pytest.fixture(autouse=True)
+def reset_common_logic():
+    """Keep the CommonLogic singleton from leaking settings between tests."""
+    CommonLogic._instance = None
+    yield
+    CommonLogic._instance = None

tests/batcontrol/logic/helpers.py

@@ -0,0 +1,71 @@
+"""Helpers for logic scenario tests."""
+import datetime
+
+import numpy as np
+
+from batcontrol.logic.common import CommonLogic
+from batcontrol.logic.logic_interface import CalculationInput, CalculationParameters
+
+
+CAPACITY_WH = 10240
+MIN_SOC = 0.10
+MAX_CHARGING_FROM_GRID_LIMIT = 0.89
+MIN_GRID_CHARGE_SOC = 0.55
+CHEAP_PRICE = 0.4635
+EXPENSIVE_PRICE = 0.7018
+
+
+def make_logic(logic_cls, *,
+               timezone=datetime.timezone.utc,
+               capacity_wh=CAPACITY_WH,
+               max_charging_from_grid_limit=MAX_CHARGING_FROM_GRID_LIMIT,
+               min_grid_charge_soc=MIN_GRID_CHARGE_SOC,
+               preserve_min_grid_charge_soc=True,
+               min_price_difference=0.05,
+               min_price_difference_rel=0.10,
+               charge_rate_multiplier=1.1,
+               always_allow_discharge_limit=0.90,
+               min_charge_energy=100,
+               peak_shaving_enabled=False):
+    """Create a logic instance with common scenario defaults."""
+    CommonLogic.get_instance(
+        charge_rate_multiplier=charge_rate_multiplier,
+        always_allow_discharge_limit=always_allow_discharge_limit,
+        max_capacity=capacity_wh,
+        min_charge_energy=min_charge_energy,
+    )
+    logic = logic_cls(timezone=timezone, interval_minutes=60)
+    logic.set_calculation_parameters(CalculationParameters(
+        max_charging_from_grid_limit=max_charging_from_grid_limit,
+        min_price_difference=min_price_difference,
+        min_price_difference_rel=min_price_difference_rel,
+        max_capacity=capacity_wh,
+        min_grid_charge_soc=min_grid_charge_soc,
+        preserve_min_grid_charge_soc=preserve_min_grid_charge_soc,
+        peak_shaving_enabled=peak_shaving_enabled,
+    ))
+    return logic
+
+
+def make_calc_input(production, consumption, prices, soc, *,
+                    capacity_wh=CAPACITY_WH,
+                    min_soc=MIN_SOC):
+    """Build CalculationInput from human-readable SoC and forecast arrays."""
+    stored_energy = capacity_wh * soc / 100
+    min_soc_energy = capacity_wh * min_soc
+    return CalculationInput(
+        production=np.array(production, dtype=float),
+        consumption=np.array(consumption, dtype=float),
+        prices={slot: price for slot, price in enumerate(prices)},
+        stored_energy=stored_energy,
+        stored_usable_energy=max(0.0, stored_energy - min_soc_energy),
+        free_capacity=capacity_wh - stored_energy,
+    )
+
+
+def target_usable_energy(*,
+                         capacity_wh=CAPACITY_WH,
+                         min_soc=MIN_SOC,
+                         min_grid_charge_soc=MIN_GRID_CHARGE_SOC):
+    """Return usable energy reserved by the minimum grid-charge target."""
+    return capacity_wh * (min_grid_charge_soc - min_soc)

tests/batcontrol/logic/test_min_grid_charge_soc_live_cases.py

@@ -1,37 +1,28 @@
 """Regression-style tests based on a real overnight grid-charge scenario."""
 import datetime
 
-import numpy as np
 import pytest
 from pytest import approx
 
-from batcontrol.logic.common import CommonLogic
 from batcontrol.logic.default import DefaultLogic
-from batcontrol.logic.logic_interface import CalculationInput, CalculationParameters
 from batcontrol.logic.next import NextLogic
 
-
-CAPACITY_WH = 10240
-MIN_SOC = 0.10
-MAX_CHARGING_FROM_GRID_LIMIT = 0.89
-MIN_GRID_CHARGE_SOC = 0.55
-CHEAP_PRICE = 0.4635
-EXPENSIVE_PRICE = 0.7018
-
-
-@pytest.fixture(autouse=True)
-def reset_common_logic():
-    """Keep the CommonLogic singleton from leaking settings between cases."""
-    CommonLogic._instance = None
-    yield
-    CommonLogic._instance = None
+from .helpers import (
+    CAPACITY_WH,
+    CHEAP_PRICE,
+    EXPENSIVE_PRICE,
+    MIN_GRID_CHARGE_SOC,
+    make_calc_input,
+    make_logic,
+    target_usable_energy,
+)
 
 
 @pytest.mark.parametrize("logic_cls", [DefaultLogic, NextLogic])
 def test_min_grid_charge_soc_preserves_battery_at_start_of_cheap_window(logic_cls):
     """A fixed target preserves battery instead of discharging through a cheap plateau."""
-    logic = _make_logic(logic_cls)
-    calc_input = _make_input(
+    logic = make_logic(logic_cls)
+    calc_input = make_calc_input(
         # Forecast snapshot around 2026-04-28 22:00.
         production=[0, 0, 0, 0, 0, 0, 0, 129, 570, 1361, 2281, 2579, 2406],
         consumption=[854, 765, 635, 691, 571, 708, 912, 1208, 1221, 1469, 1237, 1106, 983],
@@ -45,14 +36,14 @@ def test_min_grid_charge_soc_preserves_battery_at_start_of_cheap_window(logic_cl
     calc_output = logic.get_calculation_output()
 
     assert result.allow_discharge is False
-    assert calc_output.reserved_energy == approx(_target_usable_energy())
+    assert calc_output.reserved_energy == approx(target_usable_energy())
 
 
 @pytest.mark.parametrize("logic_cls", [DefaultLogic, NextLogic])
 def test_min_grid_charge_soc_charges_at_last_cheap_hour_before_expensive_window(logic_cls):
     """At the last cheap hour, a fixed target causes grid charging before high prices."""
-    logic = _make_logic(logic_cls)
-    calc_input = _make_input(
+    logic = make_logic(logic_cls)
+    calc_input = make_calc_input(
         # Forecast snapshot around 2026-04-29 05:00.
         production=[128, 541, 1196, 2022, 2615, 2728],
         consumption=[1208, 1221, 1469, 1237, 1106, 983],
@@ -75,8 +66,8 @@ def test_min_grid_charge_soc_charges_at_last_cheap_hour_before_expensive_window(
 @pytest.mark.parametrize("logic_cls", [DefaultLogic, NextLogic])
 def test_min_grid_charge_soc_stops_grid_charging_when_gap_is_below_threshold(logic_cls):
     """When the fixed target is nearly reached, avoid a tiny final grid-charge burst."""
-    logic = _make_logic(logic_cls)
-    calc_input = _make_input(
+    logic = make_logic(logic_cls)
+    calc_input = make_calc_input(
         # Similar to 2026-04-29 14:54: target gap was below min_recharge_amount.
         production=[0, 0],
         consumption=[1000, 1000],
@@ -97,8 +88,8 @@ def test_min_grid_charge_soc_stops_grid_charging_when_gap_is_below_threshold(log
 @pytest.mark.parametrize("logic_cls", [DefaultLogic, NextLogic])
 def test_min_grid_charge_soc_allows_discharge_after_target_when_price_is_high(logic_cls):
     """After reaching the target, high-price periods can use the preserved battery."""
-    logic = _make_logic(logic_cls)
-    calc_input = _make_input(
+    logic = make_logic(logic_cls)
+    calc_input = make_calc_input(
         # Similar to 2026-04-29 15:00 after the battery reached the 55% target.
         production=[2491, 2118, 1946, 1687, 1234, 664, 179, 0, 0, 0],
         consumption=[746, 986, 989, 1026, 1361, 1316, 1047, 790, 564, 665],
@@ -114,40 +105,3 @@ def test_min_grid_charge_soc_allows_discharge_after_target_when_price_is_high(lo
     assert result.allow_discharge is True
     assert result.charge_from_grid is False
     assert calc_output.reserved_energy == 0.0
-
-
-def _make_logic(logic_cls):
-    CommonLogic.get_instance(
-        charge_rate_multiplier=1.1,
-        always_allow_discharge_limit=0.90,
-        max_capacity=CAPACITY_WH,
-        min_charge_energy=100,
-    )
-    logic = logic_cls(timezone=datetime.timezone.utc, interval_minutes=60)
-    logic.set_calculation_parameters(CalculationParameters(
-        max_charging_from_grid_limit=MAX_CHARGING_FROM_GRID_LIMIT,
-        min_price_difference=0.05,
-        min_price_difference_rel=0.10,
-        max_capacity=CAPACITY_WH,
-        min_grid_charge_soc=MIN_GRID_CHARGE_SOC,
-        preserve_min_grid_charge_soc=True,
-        peak_shaving_enabled=False,
-    ))
-    return logic
-
-
-def _make_input(production, consumption, prices, soc):
-    stored_energy = CAPACITY_WH * soc / 100
-    min_soc_energy = CAPACITY_WH * MIN_SOC
-    return CalculationInput(
-        production=np.array(production, dtype=float),
-        consumption=np.array(consumption, dtype=float),
-        prices={slot: price for slot, price in enumerate(prices)},
-        stored_energy=stored_energy,
-        stored_usable_energy=max(0.0, stored_energy - min_soc_energy),
-        free_capacity=CAPACITY_WH - stored_energy,
-    )
-
-
-def _target_usable_energy():
-    return CAPACITY_WH * (MIN_GRID_CHARGE_SOC - MIN_SOC)