Confine device power to operation-mode power bands

documentation/changelog.rst

@@ -26,6 +26,7 @@ New features
 * CLI support for adding/editing account attributes [see `PR #2242 <https://www.github.com/FlexMeasures/flexmeasures/pull/2242>`_]
 * Improve chart axis domain for event values not around zero, with a per-sub-chart ``y-axis`` option in ``sensors_to_show`` (default ``zero``, which pads the axis out to include zero) that can be set to ``data`` to fit a sub-chart's y-axis to the values shown, to an explicit ``[min, max]`` domain that the axis will cover at least (expanding to fit data beyond it), or to a strict ``{"min": min, "max": max}`` domain that the axis will never exceed (clamping data beyond it, with a warning when that happens), editable from the graph editor [see `PR #2244 <https://www.github.com/FlexMeasures/flexmeasures/pull/2244>`_]
 * Extended ``GET /api/v3_0/jobs/<uuid>`` with a ``result`` field containing ``unresolved`` and ``resolved`` soft state-of-charge constraint analysis (``soc-minima``/``soc-maxima`` violations or satisfied constraints, keyed by asset ID) for scheduling jobs; both arrays are empty when no SoC constraints were defined [see `PR #2072 <https://www.github.com/FlexMeasures/flexmeasures/pull/2072>`_]
+* New storage flex-model field ``operation-modes`` confines a device's power to one of several power bands, following the S2 standard's operation modes — for example, a device that is either off or running at one fixed power [see `PR #2278 <https://www.github.com/FlexMeasures/flexmeasures/pull/2278>`_]
 * New ``FLEXMEASURES_LP_SOLVER_OPTIONS`` config setting to pass solver options to the scheduling solver, validated against the installed HiGHS build so that unknown or unsupported options raise instead of being silently ignored [see `PR #2283 <https://www.github.com/FlexMeasures/flexmeasures/pull/2283>`_]
 * Add support for intermediate power constraints on groups of devices, via a new ``group`` field in the storage flex-model [see `PR #2276 <https://www.github.com/FlexMeasures/flexmeasures/pull/2276>`_ and `issue #2092 <https://github.com/FlexMeasures/flexmeasures/issues/2092>`_]
 * The ``group`` field now also accepts a ``{"asset": <id>}`` reference (in addition to ``{"sensor": <id>}``), allowing intermediate power constraints to be defined entirely from flex-models stored on the asset tree, with results saved via the group's ``consumption``/``production`` output sensors, without needing any flex-model in the scheduling trigger [see `issue #2092 <https://github.com/FlexMeasures/flexmeasures/issues/2092>`_]

documentation/features/scheduling.rst

@@ -286,6 +286,9 @@ For more details on the possible formats for field values, see :ref:`variable_qu
    * - ``production-capacity``
      - |PRODUCTION_CAPACITY.example| (only consumption)
      - .. include:: ../_autodoc/PRODUCTION_CAPACITY.rst
+   * - ``operation-modes``
+     - |OPERATION_MODES.example|
+     - .. include:: ../_autodoc/OPERATION_MODES.rst
    * - ``group``
      - |GROUP.example|
      - .. include:: ../_autodoc/GROUP.rst

flexmeasures/data/models/planning/linear_optimization.py

@@ -83,6 +83,7 @@ def device_scheduler(  # noqa C901
     initial_stock: float | list[float] = 0,
     stock_groups: dict[int, list[int]] | None = None,
     ems_constraint_groups: list[list[int]] | None = None,
+    device_power_bands: list[list[tuple[float, float]] | None] | None = None,
 ) -> tuple[list[pd.Series], float, SolverResults, ConcreteModel]:
     """This generic device scheduler is able to handle an EMS with multiple devices,
     with various types of constraints on the EMS level and on the device level,
@@ -120,6 +121,11 @@ def device_scheduler(  # noqa C901
                                     device:                     0 (corresponds to device d; if not set, commitment is on an EMS level)
     :param initial_stock:       initial stock for each device. Use a list with the same number of devices as device_constraints,
                                 or use a single value to set the initial stock to be the same for all devices.
+    :param device_power_bands:  optional per-device list of signed power bands (min, max), in flow units
+                                (e.g. MW, positive for consumption). A device with bands must operate within
+                                one of its bands at every time step (see S2 operation modes); this introduces
+                                binary variables (one per device per band per time step). Use None (per device
+                                or for the whole argument) for devices without band restrictions.
 
     Potentially deprecated arguments:
         commitment_quantities: amounts of flow specified in commitments (both previously ordered and newly requested)
@@ -412,6 +418,15 @@ def convert_commitments_to_subcommitments(
                 device_constraints[d]["stock delta"].astype(float).fillna(0)
             )
 
+    # Look up power bands (S2 operation modes) per device
+    if device_power_bands is None:
+        device_power_bands = [None] * len(device_constraints)
+    band_lookup: dict[int, list[tuple[float, float]]] = {
+        d: list(bands)
+        for d, bands in enumerate(device_power_bands)
+        if bands is not None and len(bands) > 0
+    }
+
     # Add indices for devices (d), datetimes (j) and commitments (c)
     model.d = RangeSet(0, len(device_constraints) - 1, doc="Set of devices")
     model.j = RangeSet(
@@ -819,6 +834,49 @@ def device_derivative_equalities(m, d, j):
         model.d, model.j, rule=device_derivative_equalities
     )
 
+    # Power bands (S2 operation modes): a banded device must operate within
+    # exactly one of its declared signed power ranges at every time step.
+    model.db = Set(
+        dimen=2,
+        initialize=lambda m: (
+            (d, b) for d, bands in band_lookup.items() for b in range(len(bands))
+        ),
+        doc="Set of (device, band) pairs for devices with power bands",
+    )
+    model.device_band = Var(model.db, model.j, domain=Binary, initialize=0)
+
+    def device_band_choice(m, d, b, j):
+        """Each banded device runs in exactly one band per time step (tied to band 0)."""
+        if b != 0:
+            return Constraint.Skip
+        return sum(m.device_band[d, b_, j] for b_ in range(len(band_lookup[d]))) == 1
+
+    def device_band_power_lower(m, d, b, j):
+        """Device power at least the chosen band's minimum."""
+        if b != 0:
+            return Constraint.Skip
+        return m.device_power_down[d, j] + m.device_power_up[d, j] >= sum(
+            m.device_band[d, b_, j] * band_lookup[d][b_][0]
+            for b_ in range(len(band_lookup[d]))
+        )
+
+    def device_band_power_upper(m, d, b, j):
+        """Device power at most the chosen band's maximum."""
+        if b != 0:
+            return Constraint.Skip
+        return m.device_power_down[d, j] + m.device_power_up[d, j] <= sum(
+            m.device_band[d, b_, j] * band_lookup[d][b_][1]
+            for b_ in range(len(band_lookup[d]))
+        )
+
+    model.device_band_choice = Constraint(model.db, model.j, rule=device_band_choice)
+    model.device_band_power_lower = Constraint(
+        model.db, model.j, rule=device_band_power_lower
+    )
+    model.device_band_power_upper = Constraint(
+        model.db, model.j, rule=device_band_power_upper
+    )
+
     # Add objective
     def cost_function(m):
         costs = 0

flexmeasures/data/models/planning/storage.py

@@ -69,6 +69,32 @@
 SCHEDULING_RESULT_KEY = "scheduling_result"
 
 
+def _operation_mode_signed_band(mode: dict) -> tuple[float, float]:
+    """Convert one operation mode's consumption-/production-range into a signed
+    ``(min, max)`` band in MW (positive is consumption) for the device scheduler.
+
+    ``consumption-range`` maps to the positive side, ``production-range`` to the
+    negative side; combining both (each validated to start at 0) yields one band
+    through zero ``[-production_max, +consumption_max]``.
+    """
+    cons = mode.get("consumption_range")
+    prod = mode.get("production_range")
+    if cons and prod:
+        return (
+            -float(prod[1].to("MW").magnitude),
+            float(cons[1].to("MW").magnitude),
+        )
+    if cons:
+        return (
+            float(cons[0].to("MW").magnitude),
+            float(cons[1].to("MW").magnitude),
+        )
+    return (
+        -float(prod[1].to("MW").magnitude),
+        -float(prod[0].to("MW").magnitude),
+    )
+
+
 class MetaStorageScheduler(Scheduler):
     """This class defines the constraints of a schedule for a storage device from the
     flex-model, flex-context, and sensor and asset attributes"""
@@ -347,6 +373,9 @@ def _prepare(self, skip_validation: bool = False) -> tuple:  # noqa: C901
         production_capacity = [
             flex_model_d.get("production_capacity") for flex_model_d in flex_model
         ]
+        operation_modes = [
+            flex_model_d.get("operation_modes") for flex_model_d in flex_model
+        ]
         charging_efficiency = [
             flex_model_d.get("charging_efficiency") for flex_model_d in flex_model
         ]
@@ -989,6 +1018,16 @@ def device_list_series(
             device_constraints[d]["derivative max"] = power_capacity_in_mw[d]
             device_constraints[d]["derivative min"] = -power_capacity_in_mw[d]
 
+            # Power bands (S2 operation modes): carried on the constraints frame,
+            # in signed MW (positive is consumption), for the device scheduler.
+            # A mode's consumption-range maps to the positive side, its
+            # production-range to the negative side; combining both (each starting
+            # at 0) forms one band through zero [-production_max, +consumption_max].
+            if operation_modes[d]:
+                device_constraints[d].attrs["operation_modes"] = [
+                    _operation_mode_signed_band(mode) for mode in operation_modes[d]
+                ]
+
             if sensor_d is not None and sensor_d.get_attribute(
                 "is_strictly_non_positive"
             ):
@@ -3172,6 +3211,9 @@ def compute(self, skip_validation: bool = False) -> SchedulerOutputType:
             commitments=commitments,
             initial_stock=initial_stock,
             stock_groups=self.stock_groups,
+            device_power_bands=[
+                dc.attrs.get("operation_modes") for dc in device_constraints
+            ],
         )
         if "infeasible" in (tc := scheduler_results.solver.termination_condition):
             raise InfeasibleProblemException(tc)

flexmeasures/data/models/planning/tests/test_operation_modes.py

@@ -0,0 +1,153 @@
+"""Tests for power bands (S2 operation modes) in the device scheduler."""
+
+import numpy as np
+import pandas as pd
+
+from flexmeasures.data.models.planning import FlowCommitment
+from flexmeasures.data.models.planning.linear_optimization import device_scheduler
+from flexmeasures.data.models.planning.utils import initialize_index
+
+
+def _one_device_setup(stock_target: float):
+    """One storage device charging towards a stock target over 4 hourly steps.
+
+    Consumption is priced per step: cheap in steps 1 and 3, expensive in 0 and 2.
+    """
+    start = pd.Timestamp("2026-01-01T00:00+01")
+    end = pd.Timestamp("2026-01-01T04:00+01")
+    resolution = pd.Timedelta("PT1H")
+    index = initialize_index(start=start, end=end, resolution=resolution)
+
+    equals = pd.Series(np.nan, index=index)
+    equals.iloc[-1] = stock_target
+    device_constraints = [
+        pd.DataFrame(
+            {
+                "min": 0,
+                "max": 10,
+                "equals": equals,
+                "derivative min": 0,
+                "derivative max": 0.5,
+                "derivative equals": np.nan,
+            },
+            index=index,
+        )
+    ]
+    ems_constraints = pd.DataFrame(
+        {
+            "derivative min": -10,
+            "derivative max": 10,
+        },
+        index=index,
+    )
+    energy_commitment = FlowCommitment(
+        name="energy",
+        index=index,
+        quantity=0,
+        upwards_deviation_price=pd.Series([10, 1, 10, 1], index=index),
+        downwards_deviation_price=0,
+        device=pd.Series(0, index=index),
+    )
+    return device_constraints, ems_constraints, energy_commitment
+
+
+def _schedule(device_power_bands=None, stock_target: float = 1.2):
+    device_constraints, ems_constraints, energy_commitment = _one_device_setup(
+        stock_target
+    )
+    schedule, costs, results, model = device_scheduler(
+        device_constraints,
+        ems_constraints,
+        commitments=[energy_commitment],
+        device_power_bands=device_power_bands,
+    )
+    assert "optimal" in str(results.solver.termination_condition)
+    return schedule[0].values, costs
+
+
+def test_device_scheduler_without_bands_uses_fractional_power():
+    """Sanity check: without bands, the cheapest plan uses fractional power (0.2)."""
+    values, costs = _schedule(stock_target=1.2)
+    # Cheap steps maxed out (0.5 each); the 0.2 remainder lands in the expensive steps
+    assert np.isclose(values[1], 0.5) and np.isclose(values[3], 0.5)
+    assert np.isclose(values[0] + values[2], 0.2)
+    assert np.isclose(costs, 1.0 + 2.0)
+
+
+def test_device_scheduler_with_on_off_bands():
+    """A device confined to {0} U {0.5} runs full-on where cheap, never fractionally."""
+    values, costs = _schedule(
+        device_power_bands=[[(0, 0), (0.5, 0.5)]], stock_target=1.0
+    )
+    assert np.isclose(values, [0, 0.5, 0, 0.5]).all()
+    assert np.isclose(costs, 1.0)
+
+
+def test_device_scheduler_with_min_power_band():
+    """A device confined to {0} U [0.4, 0.5] cannot run below its minimum power.
+
+    To reach the 1.2 stock target, running the two cheap steps at 0.4 plus one
+    expensive step at 0.4 (cost 4.8) beats maxing out the cheap steps, because
+    the 0.2 remainder would have to be rounded up to the 0.4 band minimum
+    (0.5 + 0.5 + 0.4 sums to 1.4, overshooting the exact stock target).
+    """
+    values, costs = _schedule(
+        device_power_bands=[[(0, 0), (0.4, 0.5)]], stock_target=1.2
+    )
+    for v in values:
+        assert np.isclose(v, 0) or (0.4 - 1e-6 <= v <= 0.5 + 1e-6)
+    assert np.isclose(values.sum(), 1.2)
+    assert np.isclose(sorted(values), [0, 0.4, 0.4, 0.4]).all()
+    assert np.isclose(costs, 4.8)
+
+
+# --- schema: consumption-range / production-range -> signed band -----------------
+
+import pytest  # noqa: E402
+from marshmallow import ValidationError  # noqa: E402
+
+from flexmeasures.data.schemas.scheduling.storage import (  # noqa: E402
+    OperationModeSchema,
+)
+from flexmeasures.data.models.planning.storage import (  # noqa: E402
+    _operation_mode_signed_band,
+)
+
+
+def _band(payload):
+    return _operation_mode_signed_band(OperationModeSchema().load(payload))
+
+
+def test_consumption_range_maps_to_positive_band():
+    # The S2 signed power-range maps to the FM consumption-range.
+    assert _band({"consumption-range": ["0 MW", "10 MW"]}) == (0.0, 10.0)
+
+
+def test_production_range_maps_to_negative_band():
+    assert _band({"production-range": ["4 MW", "55 MW"]}) == (-55.0, -4.0)
+
+
+def test_combined_ranges_form_a_band_through_zero():
+    assert _band(
+        {"consumption-range": ["0 MW", "20 MW"], "production-range": ["0 MW", "55 MW"]}
+    ) == (-55.0, 20.0)
+
+
+def test_operation_mode_requires_at_least_one_range():
+    with pytest.raises(ValidationError):
+        OperationModeSchema().load({})
+
+
+def test_combined_ranges_must_start_at_zero():
+    with pytest.raises(ValidationError, match="contiguous band"):
+        OperationModeSchema().load(
+            {
+                "consumption-range": ["1 MW", "20 MW"],
+                "production-range": ["0 MW", "55 MW"],
+            }
+        )
+
+
+def test_range_min_cannot_exceed_max():
+    with pytest.raises(ValidationError):
+        OperationModeSchema().load({"consumption-range": ["10 MW", "5 MW"]})