Feat: scheduler support for device groups

documentation/changelog.rst

@@ -16,6 +16,7 @@ New features
 Infrastructure / Support
 ----------------------
 * Upgraded dependencies [see `PR #1485 <https://www.github.com/FlexMeasures/flexmeasures/pull/1485>`_ and `PR #2215 <https://www.github.com/FlexMeasures/flexmeasures/pull/2215>`_]
+* Prepare the ``device_scheduler`` to deal with commitments per device group [see `PR #1934 <https://www.github.com/FlexMeasures/flexmeasures/pull/1934>`_]
 
 Bugfixes
 -----------

flexmeasures/data/models/planning/__init__.py

@@ -1,9 +1,12 @@
 from __future__ import annotations
 
+from collections.abc import Iterable
 from dataclasses import dataclass, field
 from datetime import datetime, timedelta
+from tabulate import tabulate
 from typing import Any, Type
 
+import numpy as np
 import pandas as pd
 from flask import current_app
 
@@ -279,6 +282,7 @@ class Commitment:
 
     name: str
     device: pd.Series = None
+    device_group: pd.Series = None
     index: pd.DatetimeIndex = field(repr=False, default=None)
     _type: str = field(repr=False, default="each")
     group: pd.Series = field(init=False)
@@ -287,10 +291,12 @@ class Commitment:
     downwards_deviation_price: pd.Series = 0
 
     def __post_init__(self):
+        # device_group is a device→label lookup table, not a time series;
+        # exclude it from automatic time-series index coercion.
         series_attributes = [
             attr
             for attr, _type in self.__annotations__.items()
-            if _type == "pd.Series" and hasattr(self, attr)
+            if _type == "pd.Series" and hasattr(self, attr) and attr != "device_group"
         ]
         for series_attr in series_attributes:
             val = getattr(self, series_attr)
@@ -361,10 +367,67 @@ def __post_init__(self):
             "downwards deviation price"
         )
         self.group = self.group.rename("group")
+        self._init_device_group()
+
+    def _init_device_group(self):
+        # Extract device universe (empty for EMS-level commitments where device was None)
+        if isinstance(self.device, pd.Series):
+            devices = extract_devices(self.device)
+        else:
+            devices = [self.device]
+
+        devices = list(devices)
+
+        # EMS-level commitment: no device assignments, leave device_group empty.
+        if not devices:
+            self.device_group = pd.Series(dtype=object, name="device_group")
+            return
+
+        # Default: one group per device (backwards compatible)
+        if self.device_group is None:
+            self.device_group = pd.Series(
+                range(len(devices)), index=devices, name="device_group"
+            )
+        else:
+            # Validate custom grouping
+            missing = set(devices) - set(self.device_group.index)
+            if missing:
+                raise ValueError(
+                    f"device_group missing assignments for devices: {missing}"
+                )
+            self.device_group = self.device_group.loc[devices]
+            self.device_group.name = "device_group"
+
+    def pretty_print(self):
+        """
+        Pretty-print a list of Commitment objects as tabulated pandas DataFrames.
+
+        For each Commitment, a DataFrame indexed by time is created containing
+        the commitment name, device values, group index, quantity, and any available
+        upward or downward deviation prices. Each commitment is printed separately
+        in a readable table format, making this function suitable for debugging,
+        logging, and interactive inspection.
+        """
+        df = self.to_frame()
+        df = pd.DataFrame(index=df.device.index)
+
+        df["commitment"] = self.name
+        df["device"] = self.device
+        df["group"] = self.group
+        df["quantity"] = self.quantity
+
+        if hasattr(self, "upwards_deviation_price"):
+            df["up_price"] = self.upwards_deviation_price
+
+        if hasattr(self, "downwards_deviation_price"):
+            df["down_price"] = self.downwards_deviation_price
+
+        if not df.empty:
+            print(tabulate(df, headers=df.columns, tablefmt="fancy_grid"))
 
     def to_frame(self) -> pd.DataFrame:
         """Contains all info apart from the name."""
-        return pd.concat(
+        df = pd.concat(
             [
                 self.device,
                 self.quantity,
@@ -375,6 +438,19 @@ def to_frame(self) -> pd.DataFrame:
             ],
             axis=1,
         )
+        # Map device → device_group.
+        # For EMS-level commitments (device was None, now a NaN series) there are
+        # no device assignments, so device_group is an empty Series and we must not
+        # call map_device_to_group (it would KeyError on the NaN values).
+        devices = extract_devices(self.device)
+        if devices:
+            df["device_group"] = map_device_to_group(self.device, self.device_group)
+        else:
+            df["device_group"] = (
+                np.nan
+            )  # EMS-level: handled by ems_flow_commitment_equalities
+
+        return df
 
 
 class FlowCommitment(Commitment):
@@ -396,3 +472,48 @@ class StockCommitment(Commitment):
 Scheduler.compute_schedule = deprecated(Scheduler.compute, "0.14")(
     Scheduler.compute_schedule
 )
+
+
+def extract_devices(device):
+    """
+    Return a flat list of unique device identifiers from:
+    - scalar device
+    - Series of scalars
+    - Series of iterables (e.g. [0, 1])
+    """
+    if device is None:
+        return []
+
+    if isinstance(device, pd.Series):
+        values = device.dropna().values
+    else:
+        values = [device]
+
+    devices = set()
+    for v in values:
+        if isinstance(v, Iterable) and not isinstance(v, (str, bytes)):
+            devices.update(v)
+        else:
+            devices.add(v)
+
+    return list(devices)
+
+
+def map_device_to_group(device_series, device_group_map):
+    """
+    Map device identifiers to device_group.
+
+    - scalar device → group label
+    - iterable of devices → group label (must be identical)
+    """
+
+    def resolve(v):
+        if isinstance(v, (list, tuple, set)):
+            groups = {device_group_map[d] for d in v}
+            if len(groups) != 1:
+                raise ValueError(f"Devices {v} map to multiple device groups: {groups}")
+            return groups.pop()
+        else:
+            return device_group_map[v]
+
+    return device_series.apply(resolve)

flexmeasures/data/models/planning/linear_optimization.py

@@ -207,6 +207,38 @@ def convert_commitments_to_subcommitments(
 
     commitments, commitment_mapping = convert_commitments_to_subcommitments(commitments)
 
+    device_group_lookup = {}
+
+    for c, df in enumerate(commitments):
+        if "device" not in df.columns:
+            # EMS-level commitment: no device grouping needed here;
+            # handled by ems_flow_commitment_equalities.
+            continue
+
+        has_device_group = "device_group" in df.columns
+        if has_device_group:
+            rows = df[["device", "device_group"]].dropna()
+        else:
+            # Backwards-compatible default: each device is its own group.
+            # This preserves the behaviour of old-style DataFrame commitments that
+            # pre-date the device_group feature (e.g. from initialize_device_commitment).
+            rows = df[["device"]].dropna()
+
+        device_group_lookup[c] = {}
+
+        for _, row in rows.iterrows():
+            d = row["device"]
+            # When no device_group column is present, use the device id itself as
+            # the group label so that each device forms an independent group.
+            g = row["device_group"] if has_device_group else d
+
+            if isinstance(d, (list, tuple, set, np.ndarray)):
+                devices = set(d)
+            else:
+                devices = {d}
+
+            device_group_lookup[c].setdefault(g, set()).update(devices)
+
     # Oversimplified check for a convex cost curve
     df = pd.concat(commitments)[
         ["upwards deviation price", "downwards deviation price"]
@@ -243,13 +275,24 @@ def convert_commitments_to_subcommitments(
     )
     model.c = RangeSet(0, len(commitments) - 1, doc="Set of commitments")
 
-    # Add 2D indices for commitment datetimes (cj)
+    # Add 2D indices for commitment device groups (cg)
+    def commitment_device_groups_init(m):
+        return ((c, g) for c, groups in device_group_lookup.items() for g in groups)
+
+    model.cg = Set(dimen=2, initialize=commitment_device_groups_init)
 
+    # Add 2D indices for commitment datetimes (cj)
     def commitments_init(m):
         return ((c, j) for c in m.c for j in commitments[c]["j"])
 
     model.cj = Set(dimen=2, initialize=commitments_init)
 
+    # Add 3D indices for commitment datetime device groups (cjg)
+    def commitment_time_device_groups_init(m):
+        return ((c, j, g) for (c, j) in m.cj for (_, g) in m.cg if _ == c)
+
+    model.cjg = Set(dimen=3, initialize=commitment_time_device_groups_init)
+
     # Add parameters
     def price_down_select(m, c):
         if "downwards deviation price" not in commitments[c].columns:
@@ -362,6 +405,40 @@ def device_derivative_up_efficiency(m, d, j):
     def device_stock_delta(m, d, j):
         return device_constraints[d]["stock delta"].iloc[j]
 
+    def grouped_commitment_equalities(m, c, j, g):
+        """
+        Enforce a commitment deviation constraint on the aggregate of devices in a group.
+
+        For commitment ``c`` at time index ``j``, this constraint couples the commitment
+        baseline (plus deviation variables) to the summed flow or stock of all devices
+        belonging to device group ``g``. StockCommitments aggregate device stocks, while
+        FlowCommitments aggregate device flows. Constraints are skipped if the commitment
+        is inactive at ``(c, j)`` or if the group contains no devices.
+        """
+        if m.commitment_quantity[c, j] == -infinity:
+            return Constraint.Skip
+
+        devices_in_group = device_group_lookup.get(c, {}).get(g, set())
+        if not devices_in_group:
+            return Constraint.Skip
+
+        center = (
+            m.commitment_quantity[c, j]
+            + m.commitment_downwards_deviation[c]
+            + m.commitment_upwards_deviation[c]
+        )
+
+        if commitments[c]["class"].apply(lambda cl: cl == StockCommitment).all():
+            center -= sum(_get_stock_change(m, d, j) for d in devices_in_group)
+        else:
+            center -= sum(m.ems_power[d, j] for d in devices_in_group)
+
+        return (
+            0 if "upwards deviation price" in commitments[c].columns else None,
+            center,
+            0 if "downwards deviation price" in commitments[c].columns else None,
+        )
+
     model.up_price = Param(model.c, initialize=price_up_select)
     model.down_price = Param(model.c, initialize=price_down_select)
     model.commitment_quantity = Param(
@@ -565,6 +642,10 @@ def device_derivative_equalities(m, d, j):
             0,
         )
 
+    model.grouped_commitment_equalities = Constraint(
+        model.cjg, rule=grouped_commitment_equalities
+    )
+
     model.device_energy_bounds = Constraint(model.d, model.j, rule=device_bounds)
     model.device_power_bounds = Constraint(
         model.d, model.j, rule=device_derivative_bounds
@@ -592,9 +673,7 @@ def device_derivative_equalities(m, d, j):
     model.ems_power_commitment_equalities = Constraint(
         model.cj, rule=ems_flow_commitment_equalities
     )
-    model.device_energy_commitment_equalities = Constraint(
-        model.cj, model.d, rule=device_stock_commitment_equalities
-    )
+
     model.device_power_equalities = Constraint(
         model.d, model.j, rule=device_derivative_equalities
     )