Skip to content
Open
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
1 change: 1 addition & 0 deletions documentation/changelog.rst
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ Infrastructure / Support

Bugfixes
-----------
* Let storage scheduling treat missing constant SoC bounds as unconstrained lower or upper bounds [see `PR #2221 <https://www.github.com/FlexMeasures/flexmeasures/pull/2221>`_]


v0.33.0 | June 1, 2026
Expand Down
103 changes: 63 additions & 40 deletions flexmeasures/data/models/planning/storage.py
Original file line number Diff line number Diff line change
Expand Up @@ -1427,25 +1427,26 @@ def get_min_max_soc_from_asset(self) -> tuple[str | None, str | None]:

def ensure_soc_min_max(self):
"""
Make sure we have min and max SOC.
If not passed directly, then get default from asset or targets.
Fill in min and max SOC where fallbacks are available.
If not passed directly, then get defaults from asset or targets.
This happens before deserializing the flex-model.
"""
soc_min_asset, soc_max_asset = self.get_min_max_soc_from_asset()
if "soc-min" not in self.flex_model or self.flex_model["soc-min"] is None:
# Default is 0 - can't drain the storage by more than it contains
self.flex_model["soc-min"] = soc_min_asset if soc_min_asset else 0
if soc_min_asset is not None:
self.flex_model["soc-min"] = soc_min_asset
else:
self.flex_model.pop("soc-min", None)
if "soc-max" not in self.flex_model or self.flex_model["soc-max"] is None:
self.flex_model["soc-max"] = soc_max_asset
# Lacking information about the battery's nominal capacity, we use the highest target value as the maximum state of charge
if self.flex_model["soc-max"] is None:
if soc_max_asset is not None:
self.flex_model["soc-max"] = soc_max_asset
else:
# Lacking information about the battery's nominal capacity, we use the highest target value as the maximum state of charge
_, max_target = self.get_min_max_targets()
if max_target:
if max_target is not None:
self.flex_model["soc-max"] = max_target
else:
raise ValueError(
"Need maximal permitted state of charge, please specify soc-max or some soc-targets."
)
self.flex_model.pop("soc-max", None)

def _get_device_power_capacity(
self, flex_model: list[dict], assets: list[Asset]
Expand Down Expand Up @@ -2032,8 +2033,8 @@ def add_storage_constraints(
soc_targets: list[dict[str, datetime | float]] | pd.Series | None,
soc_maxima: list[dict[str, datetime | float]] | pd.Series | None,
soc_minima: list[dict[str, datetime | float]] | pd.Series | None,
soc_max: float,
soc_min: float,
soc_max: float | None,
soc_min: float | None,
) -> pd.DataFrame:
"""Collect all constraints for a given storage device in a DataFrame that the device_scheduler can interpret.

Expand All @@ -2044,8 +2045,8 @@ def add_storage_constraints(
:param soc_targets: Exact targets for the state of charge at each time.
:param soc_maxima: Maximum state of charge at each time.
:param soc_minima: Minimum state of charge at each time.
:param soc_max: Maximum state of charge at all times.
:param soc_min: Minimum state of charge at all times.
:param soc_max: Maximum state of charge at all times, if configured.
:param soc_min: Minimum state of charge at all times, if configured.
:returns: Constraints (StorageScheduler.COLUMNS) for a storage device, at each time step (index).
See device_scheduler for possible column names.
"""
Expand All @@ -2062,8 +2063,16 @@ def add_storage_constraints(
soc_targets, soc_at_start, start, end, resolution
)

soc_min_change = (soc_min - soc_at_start) * timedelta(hours=1) / resolution
soc_max_change = (soc_max - soc_at_start) * timedelta(hours=1) / resolution
soc_min_change = (
(soc_min - soc_at_start) * timedelta(hours=1) / resolution
if soc_min is not None
else None
)
soc_max_change = (
(soc_max - soc_at_start) * timedelta(hours=1) / resolution
if soc_max is not None
else None
)

if soc_minima is not None:
storage_device_constraints["min"] = build_device_soc_values(
Expand All @@ -2074,9 +2083,11 @@ def add_storage_constraints(
resolution,
)

storage_device_constraints["min"] = (
storage_device_constraints["min"].astype(float).fillna(soc_min_change)
)
storage_device_constraints["min"] = storage_device_constraints["min"].astype(float)
if soc_min_change is not None:
storage_device_constraints["min"] = storage_device_constraints["min"].fillna(
soc_min_change
)

if soc_maxima is not None:
storage_device_constraints["max"] = build_device_soc_values(
Expand All @@ -2087,13 +2098,19 @@ def add_storage_constraints(
resolution,
)

storage_device_constraints["max"] = (
storage_device_constraints["max"].astype(float).fillna(soc_max_change)
)
storage_device_constraints["max"] = storage_device_constraints["max"].astype(float)
if soc_max_change is not None:
storage_device_constraints["max"] = storage_device_constraints["max"].fillna(
soc_max_change
)

# limiting max and min to be in the range [soc_min, soc_max]
storage_device_constraints["min"] = storage_device_constraints["min"].clip(
lower=soc_min_change, upper=soc_max_change
# Limit max and min to the constant bounds that are configured.
storage_device_constraints["min"] = (
storage_device_constraints["min"].clip(
lower=soc_min_change, upper=soc_max_change
)
if soc_min_change is not None
else storage_device_constraints["min"].clip(upper=soc_max_change)
)
storage_device_constraints["max"] = storage_device_constraints["max"].clip(
lower=soc_min_change, upper=soc_max_change
Expand All @@ -2105,8 +2122,8 @@ def add_storage_constraints(
def validate_storage_constraints(
constraints: pd.DataFrame,
soc_at_start: float,
soc_min: float,
soc_max: float,
soc_min: float | None,
soc_max: float | None,
resolution: timedelta,
) -> list[dict]:
"""Check that the storage constraints are fulfilled, e.g min <= equals <= max.
Expand All @@ -2128,8 +2145,8 @@ def validate_storage_constraints(

:param constraints: dataframe containing the constraints of a storage device
:param soc_at_start: State of charge at the start time.
:param soc_min: Minimum state of charge at all times.
:param soc_max: Maximum state of charge at all times.
:param soc_min: Minimum state of charge at all times, if configured.
:param soc_max: Maximum state of charge at all times, if configured.
:param resolution: Constant duration between the start of each time step.
:returns: List of constraint violations, specifying their time, constraint and violation.
"""
Expand All @@ -2152,18 +2169,24 @@ def validate_storage_constraints(
########################

# 1) min >= soc_min
soc_min = (soc_min - soc_at_start) * timedelta(hours=1) / resolution
_constraints["soc_min(t)"] = soc_min
constraint_violations += validate_constraint(
_constraints, "soc_min(t)", "<=", "min(t)"
)
if soc_min is not None:
soc_min = (soc_min - soc_at_start) * timedelta(hours=1) / resolution
_constraints["soc_min(t)"] = soc_min
constraint_violations += validate_constraint(
_constraints, "soc_min(t)", "<=", "min(t)"
)
else:
soc_min = np.nan

# 2) max <= soc_max
soc_max = (soc_max - soc_at_start) * timedelta(hours=1) / resolution
_constraints["soc_max(t)"] = soc_max
constraint_violations += validate_constraint(
_constraints, "max(t)", "<=", "soc_max(t)"
)
if soc_max is not None:
soc_max = (soc_max - soc_at_start) * timedelta(hours=1) / resolution
_constraints["soc_max(t)"] = soc_max
constraint_violations += validate_constraint(
_constraints, "max(t)", "<=", "soc_max(t)"
)
else:
soc_max = np.nan

########################################
# B. Validation in the same time frame #
Expand Down
207 changes: 207 additions & 0 deletions flexmeasures/data/models/planning/tests/test_solver.py
Original file line number Diff line number Diff line change
Expand Up @@ -931,6 +931,213 @@ def test_add_storage_constraints(
].all()


def test_add_storage_constraints_skips_global_minimum_when_soc_min_is_missing():
"""Missing soc-min should not imply a zero lower bound."""
start = datetime(2023, 5, 18, tzinfo=pytz.utc)
end = datetime(2023, 5, 18, 5, tzinfo=pytz.utc)
resolution = timedelta(hours=1)
soc_at_start = 0.0
soc_max = 10
soc_min = None

storage_device_constraints = add_storage_constraints(
start,
end,
resolution,
soc_at_start,
soc_targets=None,
soc_maxima=None,
soc_minima=None,
soc_max=soc_max,
soc_min=soc_min,
)

assert storage_device_constraints["min"].isna().all()
assert (storage_device_constraints["max"] == soc_max).all()
assert (
validate_storage_constraints(
storage_device_constraints,
soc_at_start=soc_at_start,
soc_min=soc_min,
soc_max=soc_max,
resolution=resolution,
)
== []
)


def test_add_storage_constraints_skips_global_maximum_when_soc_max_is_missing():
"""Missing soc-max should not imply a constant upper bound."""
start = datetime(2023, 5, 18, tzinfo=pytz.utc)
end = datetime(2023, 5, 18, 5, tzinfo=pytz.utc)
resolution = timedelta(hours=1)
soc_at_start = 0.0
soc_min = 0
soc_max = None

storage_device_constraints = add_storage_constraints(
start,
end,
resolution,
soc_at_start,
soc_targets=None,
soc_maxima=None,
soc_minima=None,
soc_max=soc_max,
soc_min=soc_min,
)

assert (storage_device_constraints["min"] == soc_min).all()
assert storage_device_constraints["max"].isna().all()
assert (
validate_storage_constraints(
storage_device_constraints,
soc_at_start=soc_at_start,
soc_min=soc_min,
soc_max=soc_max,
resolution=resolution,
)
== []
)


def test_add_storage_constraints_skips_bounds_when_soc_min_and_soc_max_are_missing():
"""Missing soc-min and soc-max should not imply constant bounds."""
start = datetime(2023, 5, 18, tzinfo=pytz.utc)
end = datetime(2023, 5, 18, 5, tzinfo=pytz.utc)
resolution = timedelta(hours=1)
soc_at_start = 0.0
soc_min = None
soc_max = None

storage_device_constraints = add_storage_constraints(
start,
end,
resolution,
soc_at_start,
soc_targets=None,
soc_maxima=None,
soc_minima=None,
soc_max=soc_max,
soc_min=soc_min,
)

assert storage_device_constraints["min"].isna().all()
assert storage_device_constraints["max"].isna().all()
assert (
validate_storage_constraints(
storage_device_constraints,
soc_at_start=soc_at_start,
soc_min=soc_min,
soc_max=soc_max,
resolution=resolution,
)
== []
)


def test_add_storage_constraints_with_soc_min_missing_and_soc_minima_has_gaps():
"""Timed minima should not turn missing soc-minima into a global lower bound."""
start = datetime(2023, 5, 18, tzinfo=pytz.utc)
end = datetime(2023, 5, 18, 5, tzinfo=pytz.utc)
resolution = timedelta(hours=1)
soc_at_start = 1.0
soc_max = 10
soc_min = None

soc_minima = initialize_series(np.nan, start, end, resolution)
soc_minima[start + resolution] = 4

storage_device_constraints = add_storage_constraints(
start,
end,
resolution,
soc_at_start,
soc_targets=None,
soc_maxima=None,
soc_minima=soc_minima,
soc_max=soc_max,
soc_min=soc_min,
)

assert storage_device_constraints["min"].notna().sum() == 1
assert storage_device_constraints["min"].dropna().iloc[0] == 3
assert storage_device_constraints["min"].isna().any()
assert (
validate_storage_constraints(
storage_device_constraints,
soc_at_start=soc_at_start,
soc_min=soc_min,
soc_max=soc_max,
resolution=resolution,
)
== []
)


def test_add_storage_constraints_with_soc_max_missing_and_soc_maxima_has_gaps():
"""Timed maxima should not turn missing soc-max into a constant upper bound."""
start = datetime(2023, 5, 18, tzinfo=pytz.utc)
end = datetime(2023, 5, 18, 5, tzinfo=pytz.utc)
resolution = timedelta(hours=1)
soc_at_start = 1.0
soc_min = 0
soc_max = None

soc_maxima = initialize_series(np.nan, start, end, resolution)
soc_maxima[start + resolution] = 6

storage_device_constraints = add_storage_constraints(
start,
end,
resolution,
soc_at_start,
soc_targets=None,
soc_maxima=soc_maxima,
soc_minima=None,
soc_max=soc_max,
soc_min=soc_min,
)

assert storage_device_constraints["max"].notna().sum() == 1
assert storage_device_constraints["max"].dropna().iloc[0] == 5
assert storage_device_constraints["max"].isna().any()
assert (
validate_storage_constraints(
storage_device_constraints,
soc_at_start=soc_at_start,
soc_min=soc_min,
soc_max=soc_max,
resolution=resolution,
)
== []
)


def test_ensure_soc_min_max_allows_missing_soc_max():
"""Missing soc-max should remain unset when no fallback is available."""
scheduler = StorageScheduler.__new__(StorageScheduler)
scheduler.flex_model = {"soc-min": "0 MWh"}
scheduler.asset = None
scheduler.sensor = None

scheduler.ensure_soc_min_max()

assert scheduler.flex_model == {"soc-min": "0 MWh"}


def test_ensure_soc_min_max_allows_missing_soc_min_and_soc_max():
"""Missing soc-min and soc-max should remain unset when no fallback is available."""
scheduler = StorageScheduler.__new__(StorageScheduler)
scheduler.flex_model = {}
scheduler.asset = None
scheduler.sensor = None

scheduler.ensure_soc_min_max()

assert scheduler.flex_model == {}


@pytest.mark.parametrize(
"value_min1, value_equals1, value_max1, value_min2, value_equals2, value_max2, expected_constraint_type_violations",
[
Expand Down
Loading
Loading