diff --git a/flexmeasures/data/models/planning/tests/test_solver.py b/flexmeasures/data/models/planning/tests/test_solver.py index fb563f020c..f71c9ac060 100644 --- a/flexmeasures/data/models/planning/tests/test_solver.py +++ b/flexmeasures/data/models/planning/tests/test_solver.py @@ -2986,3 +2986,95 @@ def run_sequential_scheduler(): assert total_cost_all_devices == sum( expected_cost[1] for expected_cost in expected_costs ), "Total cost mismatch." + + +@pytest.mark.parametrize( + ("capacity_in_w"), + [ + 10, + 1.1, + 1.01, + 1.001, + 1.0001, + 1.00001, + 1.000001, + 1.0000001, + 1.00000001, + 1.000000001, + 1.0000000001, + 1.00000000001, + 1.000000000001, + 1.0000000000001, + 1.00000000000001, + 1.000000000000001, + pytest.param( + 1.0000000000000001, marks=pytest.mark.xfail(reason="some rounding bug") + ), + pytest.param( + 1.00000000000000001, marks=pytest.mark.xfail(reason="some rounding bug") + ), + pytest.param( + 1.000000000000000001, marks=pytest.mark.xfail(reason="some rounding bug") + ), + pytest.param( + 1.0000000000000000001, marks=pytest.mark.xfail(reason="some rounding bug") + ), + pytest.param(1, marks=pytest.mark.xfail(reason="some rounding bug")), + ], +) +def test_battery_kW( + add_battery_assets, + db, + capacity_in_w, +): + """ + Scheduling a very small battery of 1Wh (1e-6 MWh) with a power capacity of 1W with a constant + usage of 1W. + """ + + epex_da, battery = get_sensors_from_db( + db, + add_battery_assets, + battery_name="Test battery", + power_sensor_name="power (kW)", + ) + tz = pytz.timezone("Europe/Amsterdam") + start = tz.localize(datetime(2015, 1, 1)) + end = tz.localize(datetime(2015, 1, 2)) + resolution = timedelta(minutes=15) + soc_max = capacity_in_w # assumes 1 hour to fully charge + soc_usage_in_w = capacity_in_w + device_capacity_in_w = capacity_in_w + + scheduler: Scheduler = StorageScheduler( + battery, + start, + end, + resolution, + flex_model={ + "soc-at-start": "0 kWh", + "soc-max": f"{soc_max} Wh", + "soc-min": "0.0 kWh", + "soc-usage": [f"{soc_usage_in_w} W"], + "charging-efficiency": 1, + "discharging-efficiency": 1, + "storage-efficiency": 1, + "power-capacity": f"{device_capacity_in_w} W", + "consumption-capacity": f"{device_capacity_in_w} W", + "production-capacity": f"{device_capacity_in_w} W", + }, + flex_context={ + "site-power-capacity": "1 MW", + "consumption-price": "100 EUR/MWh", + "production-price": "90 EUR/MWh", + }, + ) + schedule = scheduler.compute() + + # Check if constraints were met + soc_usage = ( + pd.Series(soc_usage_in_w / 1000, index=schedule.index) + * resolution + / timedelta(hours=1) + ) + check_constraints(battery, schedule, 0, soc_usage=soc_usage) diff --git a/flexmeasures/data/models/planning/tests/utils.py b/flexmeasures/data/models/planning/tests/utils.py index 9c270dd9ec..c16ae4e4c8 100644 --- a/flexmeasures/data/models/planning/tests/utils.py +++ b/flexmeasures/data/models/planning/tests/utils.py @@ -10,6 +10,7 @@ def check_constraints( sensor: Sensor, schedule: pd.Series, soc_at_start: float, + soc_usage: pd.Series | float = 0, roundtrip_efficiency: float = 1, storage_efficiency: float = 1, tolerance: float = 0.00001, @@ -22,6 +23,8 @@ def check_constraints( storage_efficiency=storage_efficiency, decimal_precision=6, ) + soc_usage = soc_usage.reindex(soc_schedule.index).shift(1).fillna(0).cumsum() + soc_schedule -= soc_usage # todo: move into integrate_time_series with pd.option_context("display.max_rows", None, "display.max_columns", 3): print(soc_schedule) capacity = sensor.get_attribute( @@ -30,9 +33,10 @@ def check_constraints( ) assert min(schedule.values) >= capacity * -1 - tolerance assert max(schedule.values) <= capacity + tolerance + # breakpoint() for soc in soc_schedule.values: - assert soc >= sensor.get_attribute("min_soc_in_mwh") - assert soc <= sensor.get_attribute("max_soc_in_mwh") + assert soc >= sensor.get_attribute("min_soc_in_mwh") - tolerance + assert soc <= sensor.get_attribute("max_soc_in_mwh") + tolerance return soc_schedule