Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions tests/batcontrol/logic/conftest.py
Original file line number Diff line number Diff line change
@@ -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
89 changes: 89 additions & 0 deletions tests/batcontrol/logic/helpers.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,89 @@
"""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,
PeakShavingConfig,
)


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.

The CommonLogic singleton is reset so each helper call applies the
requested singleton-backed tuning values independently.
"""
CommonLogic._instance = None
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)
Comment thread
filiplajszczak marked this conversation as resolved.
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=PeakShavingConfig(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 forecast arrays and state of charge.

Args:
production: Forecast production values in Wh for each time slot.
consumption: Forecast consumption values in Wh for each time slot.
prices: Energy prices for each time slot.
soc: Current battery state of charge as a percentage, 0-100.
capacity_wh: Battery capacity in Wh.
min_soc: Minimum battery state of charge as a ratio, 0-1.
"""
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)
81 changes: 18 additions & 63 deletions tests/batcontrol/logic/test_min_grid_charge_soc_live_cases.py
Original file line number Diff line number Diff line change
@@ -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],
Expand All @@ -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],
Expand All @@ -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],
Expand All @@ -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],
Expand All @@ -114,39 +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,
))
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)