Commitments can be scoped to specific sensors, binding their aggregate flow

documentation/changelog.rst

@@ -43,6 +43,7 @@ New features
 * 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>`_]
 * Extended the scheduling job ``result`` field with a ``num-beliefs`` field reporting the total number of beliefs (scheduled values) saved to the database [see `PR #2280 <https://www.github.com/FlexMeasures/flexmeasures/pull/2280>`_]
+* Flex-context commitments can be scoped to a subset of devices, via a new optional ``sensors`` list (any devices, possibly across electrical groups) or ``group`` reference (an electrical group's members); the commitment binds the net signed aggregate flow (consumption positive, production negative) of those devices — flexible and inflexible alike — as one commitment, instead of binding each device separately [see `PR #2295 <https://www.github.com/FlexMeasures/flexmeasures/pull/2295>`_]
 * Migrate the asset tree in the UI's Structure tab from Vega to ECharts, adding interactive pan/zoom navigation and refreshed node styling [see `PR #2025 <https://www.github.com/FlexMeasures/flexmeasures/pull/2025>`_ and `PR #2365 <https://www.github.com/FlexMeasures/flexmeasures/pull/2365>`_]
 
 Infrastructure / Support

flexmeasures/data/models/planning/devices.py

@@ -624,6 +624,18 @@ def by_sensor_id(self, sensor_id: int) -> list[FlexDevice]:
         """Return the flexible devices whose power sensor has the given id."""
         return [device for device in self.devices if device.sensor_id == sensor_id]
 
+    def scheduled_devices_by_sensor_id(self, sensor_id: int) -> list[FlexDevice]:
+        """Return all devices (flexible and inflexible) whose power sensor has the given id.
+
+        Unlike :meth:`by_sensor_id`, this includes inflexible (fixed-power) devices,
+        so a commitment scoped to a sensor list can bind an inflexible device's flow too.
+        """
+        return [
+            device
+            for device in (*self.devices, *self.inflexible_devices)
+            if device.sensor_id == sensor_id
+        ]
+
     @cached_property
     def stock_groups(self) -> dict[int, list[int]]:
         """Map each stock key to the indices of the devices drawing from that stock.

flexmeasures/data/models/planning/storage.py

@@ -22,6 +22,7 @@
     DeviceInventory,
     _resolve_stock_key,
     group_key_label,
+    resolve_group_reference,
 )
 from flexmeasures.data.models.planning.linear_optimization import device_scheduler
 from flexmeasures.data.models.planning.utils import (
@@ -1545,6 +1546,92 @@ def device_list_series(
             commitments,
         )
 
+    def _resolve_commitment_scope(
+        self, scoped_sensors, scoped_group
+    ) -> tuple[list[int], str]:
+        """Resolve a scoped commitment's device set to canonical solver indices.
+
+        Both scopes include a device whether it is flexible or inflexible:
+        a ``group`` scope yields the group's (leaf) members,
+        and a ``sensors`` scope yields the devices recording the listed power sensors.
+        So listing a group's member sensors resolves to the same set as scoping by that group.
+        Canonical indices always come from the device inventory,
+        never from re-enumerating raw flex-model lists.
+
+        The commitment then binds the *net signed* aggregate of these devices' flow (consumption positive, production negative),
+        so consumers add, producers subtract, and any inflexible (fixed) member contributes its fixed signed power.
+
+        :returns: A ``(sorted device indices, human-readable scope description)`` pair.
+        """
+        if scoped_group is not None:
+            group_key = resolve_group_reference(scoped_group)
+            scoped_devices = sorted(
+                self.device_inventory.group_to_devices.get(group_key, [])
+            )
+            description = (
+                f"group {group_key_label(group_key)}"
+                if group_key is not None
+                else "an unresolved group reference"
+            )
+            return scoped_devices, description
+        scoped_sensor_ids = {
+            sensor.id if hasattr(sensor, "id") else sensor for sensor in scoped_sensors
+        }
+        scoped_devices = sorted(
+            device.index
+            for sensor_id in scoped_sensor_ids
+            for device in self.device_inventory.scheduled_devices_by_sensor_id(
+                sensor_id
+            )
+        )
+        return scoped_devices, f"sensors {sorted(scoped_sensor_ids)}"
+
+    def _build_scoped_commitment(
+        self, commitment_spec, scoped_sensors, scoped_group, commitment_index
+    ) -> "FlowCommitment | None":
+        """Build one aggregate-flow FlowCommitment for a scoped commitment.
+
+        Returns None (logged) when the scope matches no device in the flex-model,
+        so the commitment binds nothing rather than failing the whole schedule.
+
+        :raises ValueError: When the scoped devices span more than one commodity.
+        """
+        scoped_devices, scope_description = self._resolve_commitment_scope(
+            scoped_sensors, scoped_group
+        )
+        if not scoped_devices:
+            current_app.logger.warning(
+                f"Commitment '{commitment_spec.get('name')}' is scoped to"
+                f" {scope_description}, none of which appear in the flex-model."
+                " This commitment will not bind any device."
+            )
+            return None
+        commodities = {
+            self.device_inventory.by_index(d).commodity for d in scoped_devices
+        }
+        if len(commodities) > 1:
+            raise ValueError(
+                f"Commitment '{commitment_spec.get('name')}' is scoped to devices of"
+                f" more than one commodity ({sorted(commodities)}); a commitment binds"
+                " the aggregate flow of a single commodity."
+            )
+        # A scoped commitment's commodity is defined by its scope,
+        # so pin it to the scoped devices' (single) commodity;
+        # otherwise the commitment keeps the schema's electricity default (or a mismatching explicit value),
+        # and its cost would be misattributed to the wrong commodity.
+        commitment_spec["commodity"] = next(iter(commodities))
+        index = commitment_spec["index"]
+        # device_group maps device index -> group label;
+        # one shared label makes the engine bind the aggregate flow.
+        # The label is unique per commitment, so two scoped commitments never merge, even if they share a name.
+        group_label = f"scoped-commitment-{commitment_index}"
+        return FlowCommitment(
+            device=pd.Series([scoped_devices] * len(index), index=index),
+            device_group=pd.Series({d: group_label for d in scoped_devices}),
+            provenance="custom",
+            **commitment_spec,
+        )
+
     def convert_to_commitments(
         self,
         flex_model,
@@ -1579,7 +1666,7 @@ def convert_to_commitments(
         commitments = []
         # The specs were copied above, so converting (which pops fields) does not
         # mutate self.flex_context and repeated conversions see the original specs.
-        for commitment_spec in commitment_specs:
+        for commitment_index, commitment_spec in enumerate(commitment_specs):
 
             # Convert baseline, up_price and down_price to pd.Series, then create FlowCommitment
             if "up_price" in commitment_spec:
@@ -1608,6 +1695,22 @@ def convert_to_commitments(
                 start, end, timing_kwargs["resolution"]
             )
             commitment_commodity = commitment_spec.get("commodity", "electricity")
+
+            # A commitment scoped to a subset of devices binds the *aggregate* flow of those devices as one commitment,
+            # rather than each device separately.
+            # The scope is given either as a raw list of power `sensors` (a cherry-pick that may span electrical groups,
+            # e.g. an aFRR band on a site's e-heaters),
+            # or as a `group` reference (the members of an electrical group, reusing its already-resolved membership);
+            # the schema allows at most one of the two.
+            scoped_sensors = commitment_spec.pop("sensors", None)
+            scoped_group = commitment_spec.pop("group", None)
+            if scoped_sensors is not None or scoped_group is not None:
+                scoped = self._build_scoped_commitment(
+                    commitment_spec, scoped_sensors, scoped_group, commitment_index
+                )
+                if scoped is not None:
+                    commitments.append(scoped)
+                continue
             bound_device_count = 0
             for d, flex_model_d in enumerate(flex_model):
                 device_commodity = flex_model_d.get("commodity", "electricity")