Fix: power capacity does not need to be mandatory

documentation/changelog.rst

@@ -21,6 +21,7 @@ Infrastructure / Support
 
 Bugfixes
 -----------
+* Let storage scheduling infer missing ``power-capacity`` from directional device capacities before falling back to site capacity [see `PR #2222 <https://www.github.com/FlexMeasures/flexmeasures/pull/2222>`_]
 
 
 v0.33.0 | June 1, 2026

flexmeasures/data/models/planning/storage.py

@@ -176,8 +176,14 @@ def _prepare(self, skip_validation: bool = False) -> tuple:  # noqa: C901
             "inflexible_device_sensors", []
         )
 
-        # Fetch the device's power capacity (required Sensor attribute)
-        power_capacity_in_mw = self._get_device_power_capacity(flex_model, assets)
+        # Fetch the device's power capacity (required to keep the optimization problem bounded)
+        power_capacity_in_mw = self._get_device_power_capacity(
+            flex_model,
+            assets,
+            query_window=(start, end),
+            resolution=resolution,
+            beliefs_before=belief_time,
+        )
 
         # Check for known prices or price forecasts
         up_deviation_prices = get_continuous_series_sensor_or_quantity(
@@ -1448,14 +1454,20 @@ def ensure_soc_min_max(self):
                     )
 
     def _get_device_power_capacity(
-        self, flex_model: list[dict], assets: list[Asset]
-    ) -> list[ur.Quantity]:
+        self,
+        flex_model: list[dict],
+        assets: list[Asset],
+        query_window: tuple[datetime, datetime],
+        resolution: timedelta,
+        beliefs_before: datetime | None,
+    ) -> list[Sensor | SensorReference | list[dict] | ur.Quantity | pd.Series]:
         """The device power capacity for each device must be known for the optimization problem to stay bounded.
 
         We search for the power capacity in the following order:
         1. Look for the power_capacity_in_mw field in the deserialized flex-model.
         2. Look for the power-capacity flex-model field of the asset.
-        3. Look for the site-power-capacity attribute of the asset.
+        3. Look for the greatest device consumption-capacity or production-capacity.
+        4. Look for the site-power-capacity attribute of the asset.
         """
         power_capacities = []
         for flex_model_d, asset in zip(flex_model, assets):
@@ -1472,6 +1484,21 @@ def _get_device_power_capacity(
                 continue
 
             # 3
+            fallback_capacity = self._get_largest_device_capacity(
+                flex_model_d=flex_model_d,
+                query_window=query_window,
+                resolution=resolution,
+                beliefs_before=beliefs_before,
+            )
+            if fallback_capacity is not None:
+                current_app.logger.warning(
+                    f"Missing 'power-capacity' on asset {asset.id}. "
+                    "Using the largest of consumption-capacity and production-capacity instead."
+                )
+                power_capacities.append(fallback_capacity)
+                continue
+
+            # 4
             site_power_capacity = asset.get_attribute("site-power-capacity")
             if site_power_capacity is not None:
                 current_app.logger.warning(
@@ -1494,14 +1521,47 @@ def _get_device_power_capacity(
             )
         return power_capacities
 
+    def _get_largest_device_capacity(
+        self,
+        flex_model_d: dict,
+        query_window: tuple[datetime, datetime],
+        resolution: timedelta,
+        beliefs_before: datetime | None,
+    ) -> Sensor | SensorReference | list[dict] | ur.Quantity | pd.Series | None:
+        """Return the largest directional capacity when both directions are configured."""
+        capacity_fields = ("consumption_capacity", "production_capacity")
+        if any(flex_model_d.get(field) is None for field in capacity_fields):
+            return None
+        capacities = [
+            self._ensure_variable_quantity(flex_model_d[field], "MW")
+            for field in capacity_fields
+        ]
+
+        capacity_series = [
+            get_continuous_series_sensor_or_quantity(
+                variable_quantity=capacity,
+                unit="MW",
+                query_window=query_window,
+                resolution=resolution,
+                beliefs_before=beliefs_before,
+                min_value=0,
+                resolve_overlaps="min",
+            )
+            for capacity in capacities
+        ]
+        largest_capacity = pd.concat(capacity_series, axis=1).max(axis=1)
+        if largest_capacity.isna().all():
+            return None
+        return largest_capacity
+
     def _ensure_variable_quantity(
-        self, value: str | int | float | ur.Quantity, unit: str
-    ) -> Sensor | SensorReference | list[dict] | ur.Quantity:
+        self, value: str | int | float | ur.Quantity | pd.Series, unit: str
+    ) -> Sensor | SensorReference | list[dict] | ur.Quantity | pd.Series:
         if isinstance(value, str):
             q = ur.Quantity(value).to(unit)
         elif isinstance(value, (float, int)):
             q = ur.Quantity(f"{value} {unit}")
-        elif isinstance(value, (Sensor, SensorReference, list, ur.Quantity)):
+        elif isinstance(value, (Sensor, SensorReference, list, ur.Quantity, pd.Series)):
             q = value
         else:
             raise TypeError(

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

@@ -1390,6 +1390,51 @@ def set_if_not_none(dictionary, key, value):
     assert all(ems_constraints["derivative max"] == expected_site_consumption_capacity)
 
 
+@pytest.mark.parametrize(
+    "production_capacity, consumption_capacity, expected_capacity",
+    [
+        ("300 kW", "700 kW", 0.7),
+        ("1.1 MW", "200 kW", 1.1),
+    ],
+)
+def test_device_power_capacity_uses_greatest_directional_capacity_before_site_fallback(
+    db,
+    add_battery_assets,
+    production_capacity,
+    consumption_capacity,
+    expected_capacity,
+):
+    _, battery = get_sensors_from_db(db, add_battery_assets)
+
+    start = pytz.timezone("Europe/Amsterdam").localize(datetime(2015, 1, 2))
+    end = pytz.timezone("Europe/Amsterdam").localize(datetime(2015, 1, 3))
+    resolution = timedelta(minutes=15)
+    scheduler = StorageScheduler(
+        asset_or_sensor=battery,
+        start=start,
+        end=end,
+        resolution=resolution,
+        flex_model={
+            "soc-at-start": 0,
+            "soc-min": 0,
+            "soc-max": 5,
+            "production-capacity": production_capacity,
+            "consumption-capacity": consumption_capacity,
+        },
+    )
+    scheduler.deserialize_config()
+
+    power_capacity = scheduler._get_device_power_capacity(
+        [scheduler.flex_model],
+        [battery.generic_asset],
+        query_window=(start, end),
+        resolution=resolution,
+        beliefs_before=scheduler.belief_time,
+    )[0]
+
+    assert np.allclose(power_capacity.values, expected_capacity)
+
+
 @pytest.mark.parametrize(
     ["soc_values", "log_message", "expected_num_targets"],
     [

flexmeasures/data/schemas/scheduling/metadata.py

@@ -336,16 +336,19 @@ def to_dict(self):
     example=True,
 )
 POWER_CAPACITY = MetaData(
-    description="Device-level power constraint. How much power can be applied to this asset. [#minimum_overlap]_",
+    description="""Symmetric device-level power constraint. How much power can be applied to this asset in either direction.
+If omitted, the scheduler infers this limit from the greatest of ``consumption-capacity`` and ``production-capacity`` when both are configured, before falling back to ``site-power-capacity``. [#minimum_overlap]_""",
     example="50 kVA",
 )
 CONSUMPTION_CAPACITY = MetaData(
-    description="Device-level power constraint on consumption. How much power can be drawn by this asset. [#minimum_overlap]_",
+    description="""Device-level power constraint on consumption. How much power can be drawn by this asset.
+If ``power-capacity`` is omitted and ``production-capacity`` is also configured, this field contributes to the inferred symmetric device-level power limit. [#minimum_overlap]_""",
     example={"sensor": 56},
 )
 PRODUCTION_CAPACITY = MetaData(
     description="""Device-level power constraint on production.
 How much power can be supplied by this asset.
+If ``power-capacity`` is omitted and ``consumption-capacity`` is also configured, this field contributes to the inferred symmetric device-level power limit.
 For :abbr:`PV (photovoltaic solar panels)` curtailment, set this to reference your sensor containing PV power forecasts. [#minimum_overlap]_
 """,
     example="0 kW",