feat: commitments can be scoped to specific sensors, binding their aggregate flow

A flex-context commitment gains an optional 'sensors' field: instead of
binding each device of the matching commodity separately, the commitment
binds the aggregate flow of the devices whose power sensors are listed,
as one grouped commitment (device_group machinery). Useful to commit a
band on a subset of devices, e.g. an aFRR upward-regulation band on a
site's e-heaters (aggregate consumption >= band, deviation penalized).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016MLCUiSdXDqDBmg8GbYp1B

Author: Flix6x

flexmeasures/data/models/planning/storage.py

@@ -1351,6 +1351,39 @@ def convert_to_commitments(
                 start, end, timing_kwargs["resolution"]
             )
             commitment_commodity = commitment_spec.get("commodity", "electricity")
+
+            # A commitment scoped to specific sensors binds the *aggregate* flow
+            # of those devices as one commitment, rather than each device separately.
+            scoped_sensors = commitment_spec.pop("sensors", None)
+            if scoped_sensors is not None:
+                scoped_sensor_ids = {
+                    sensor.id if hasattr(sensor, "id") else sensor
+                    for sensor in scoped_sensors
+                }
+                scoped_devices = [
+                    d
+                    for d, flex_model_d in enumerate(flex_model)
+                    if getattr(flex_model_d.get("sensor"), "id", None)
+                    in scoped_sensor_ids
+                ]
+                if not scoped_devices:
+                    current_app.logger.warning(
+                        f"Commitment '{commitment_spec.get('name')}' is scoped to"
+                        f" sensors {sorted(scoped_sensor_ids)}, none of which appear"
+                        " in the flex-model. This commitment will not bind any device."
+                    )
+                    continue
+                index = commitment_spec["index"]
+                group_label = commitment_spec.get("name", "scoped commitment")
+                commitment = FlowCommitment(
+                    device=pd.Series([scoped_devices] * len(index), index=index),
+                    # device_group maps device index -> group label; one shared
+                    # label makes the engine bind the aggregate flow.
+                    device_group=pd.Series({d: group_label for d in scoped_devices}),
+                    **commitment_spec,
+                )
+                commitments.append(commitment)
+                continue
             for d, flex_model_d in enumerate(flex_model):
                 device_commodity = flex_model_d.get("commodity", "electricity")
                 if device_commodity != commitment_commodity:

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

@@ -1782,3 +1782,83 @@ def test_electricity_device_indices_exclude_other_commodities():
     assert mapping["electricity"] == [0, 2, 3, 4]
     assert mapping["gas"] == [1, 5]
     assert scheduler._electricity_device_indices() == [0, 2, 3, 4]
+
+
+def test_sensor_scoped_commitment_binds_aggregate_of_selected_devices(app, db):
+    """A commitment scoped to specific sensors (here: two e-heaters) binds their
+    aggregate flow as one commitment: a baseline of 10 MW with a steep penalty on
+    downward deviation keeps their combined consumption at 10 MW even though a
+    cheaper allocation (0 MW) exists, while an unscoped battery stays unaffected.
+    """
+    heater_type = get_or_create_model(GenericAssetType, name="e-heater")
+    site = GenericAsset(
+        name="Band site (scoped commitment test)", generic_asset_type=heater_type
+    )
+    db.session.add(site)
+    db.session.flush()
+
+    resolution = pd.Timedelta("1h")
+    start = pd.Timestamp("2026-02-01T00:00:00+01:00")
+    end = pd.Timestamp("2026-02-01T04:00:00+01:00")
+
+    def sensor(name):
+        s = Sensor(
+            name=name, unit="MW", event_resolution=resolution, generic_asset=site
+        )
+        db.session.add(s)
+        return s
+
+    heater_1 = sensor("band heater 1")
+    heater_2 = sensor("band heater 2")
+    db.session.flush()
+
+    flex_model = [
+        {
+            # Heaters burn money at the consumption price; without the band
+            # commitment the optimum is to stay off.
+            "sensor": heater_1.id,
+            "power-capacity": "8 MW",
+            "consumption-capacity": "8 MW",
+            "production-capacity": "0 kW",
+        },
+        {
+            "sensor": heater_2.id,
+            "power-capacity": "8 MW",
+            "consumption-capacity": "8 MW",
+            "production-capacity": "0 kW",
+        },
+    ]
+    flex_context = {
+        "consumption-price": "50 EUR/MWh",
+        "production-price": "50 EUR/MWh",
+        "site-power-capacity": "1 GW",
+        "commitments": [
+            {
+                "name": "reserved band",
+                "sensors": [heater_1.id, heater_2.id],
+                "baseline": "10 MW",
+                # Steep penalty for consuming less than the band (negative price
+                # penalizes downward deviation); consuming more is free.
+                "down-price": "-10000 EUR/MWh",
+            }
+        ],
+    }
+
+    scheduler = StorageScheduler(
+        asset_or_sensor=site,
+        start=start,
+        end=end,
+        resolution=resolution,
+        belief_time=start,
+        flex_model=flex_model,
+        flex_context=flex_context,
+        return_multiple=True,
+    )
+    results = scheduler.compute(skip_validation=True)
+    schedules = {
+        r["sensor"]: r["data"] for r in results if r.get("name") == "storage_schedule"
+    }
+    combined = schedules[heater_1] + schedules[heater_2]
+    # The band keeps the aggregate at 10 MW (cheapest way to avoid the penalty),
+    # even though each heater alone (8 MW max) could not carry it.
+    np.testing.assert_allclose(combined.iloc[:-1], 10.0, rtol=1e-4)

flexmeasures/data/schemas/scheduling/__init__.py

@@ -71,6 +71,15 @@ def forbid_time_series_specs(self, data: dict, **kwargs):
 
 class CommitmentSchema(Schema):
     name = fields.Str(required=True, data_key="name")
+    # Optional scoping: bind this commitment to the aggregate flow of the
+    # devices whose power sensors are listed, rather than binding each device
+    # separately. Useful to commit a band on a subset of devices (e.g. an
+    # aFRR band on a site's e-heaters).
+    sensors = fields.List(
+        SensorIdField(),
+        required=False,
+        data_key="sensors",
+    )
     baseline = VariableQuantityField("MW", required=False, data_key="baseline")
     up_price = VariableQuantityField("/MW", required=False, data_key="up-price")
     down_price = VariableQuantityField(

flexmeasures/ui/static/openapi-specs.json

@@ -4575,6 +4575,12 @@
           "name": {
             "type": "string"
           },
+          "sensors": {
+            "type": "array",
+            "items": {
+              "type": "integer"
+            }
+          },
           "baseline": {},
           "up-price": {},
           "down-price": {}