feat: add ProcessScheduler

documentation/api/notation.rst

@@ -204,7 +204,13 @@ Here are the three types of flexibility models you can expect to be built-in:
 
 2) For **shiftable processes**
 
-   .. todo:: A simple and proven algorithm exists, but is awaiting proper integration into FlexMeasures, see `PR 729 <https://github.com/FlexMeasures/flexmeasures/pull/729>`_.
+    - ``consumption_price_sensor``: it defines the utility (economic, environmental, ) in each time period. It has units of quantity/energy, for example, EUR/kWh.
+    - ``power``: nominal power of the load.
+    - ``duration``: time that the load last.
+    - ``optimization_sense``: objective of the scheduler, to maximize or minimize.
+    - ``time_restrictions``: time periods in which the load cannot be schedule to.
+    - ``load_type``: Inflexible, Breakable or Shiftable.
+
 
 3) For **buffer devices** (e.g. thermal energy storage systems connected to heat pumps), use the same flexibility parameters described above for storage devices. Here are some tips to model a buffer with these parameters:
 

documentation/changelog.rst

@@ -17,6 +17,7 @@ New features
 * DataSource table now allows storing arbitrary attributes as a JSON (without content validation), similar to the Sensor and GenericAsset tables [see `PR #750 <https://www.github.com/FlexMeasures/flexmeasures/pull/750>`_]
 * Added API endpoint `/sensor/<id>` for fetching a single sensor. [see `PR #759 <https://www.github.com/FlexMeasures/flexmeasures/pull/759>`_]
 * The CLI now allows to set lists and dicts as asset & sensor attributes (formerly only single values) [see `PR #762 <https://www.github.com/FlexMeasures/flexmeasures/pull/762>`_]
+* Add `ShiftableLoadScheduler` class, which optimizes loads activation using one of the following policies: inflexible, shiftable and breakable [see `PR #729 <https://www.github.com/FlexMeasures/flexmeasures/pull/729>`_]
 
 Bugfixes
 -----------
@@ -28,7 +29,6 @@ Infrastructure / Support
 * The endpoint `[POST] /health/ready <api/v3_0.html#get--api-v3_0-health-ready>`_ returns the status of the Redis connection, if configured [see `PR #699 <https://www.github.com/FlexMeasures/flexmeasures/pull/699>`_]
 * Document the `device_scheduler` linear program [see `PR #764 <https://www.github.com/FlexMeasures/flexmeasures/pull/764>`_].
 
-/api/v3_0/health/ready
 
 v0.14.2 | July 25, 2023
 ============================

flexmeasures/data/models/planning/shiftable_load.py

@@ -0,0 +1,261 @@
+from __future__ import annotations
+
+from math import ceil
+from datetime import timedelta
+import pytz
+
+import pandas as pd
+
+from flexmeasures.data.models.planning import Scheduler
+
+from flexmeasures.data.queries.utils import simplify_index
+from flexmeasures.data.models.time_series import Sensor
+from flexmeasures.data.schemas.scheduling.shiftable_load import (
+    ShiftableLoadFlexModelSchema,
+    LoadType,
+    OptimizationSense,
+)
+from flexmeasures.data.schemas.scheduling import FlexContextSchema
+
+
+class ShiftableLoadScheduler(Scheduler):
+
+    __version__ = "1"
+    __author__ = "Seita"
+
+    def compute(self) -> pd.Series | None:
+        """Schedule a fix load, defined as a `power` and a `duration`, within the specified time window.
+        To schedule a battery, please, refer to the StorageScheduler.
+
+        For example, this scheduler can plan the start of a process of type `Shiftable` that lasts 5h and requires a power of 10kW.
+        In that case, the scheduler will find the best (as to minimize/maximize the cost) hour to start the process.
+
+        This scheduler supports three types of `load_types`:
+            - Inflexible: this load requires to be scheduled as soon as possible.
+            - Breakable: this load can be divisible in smaller consumption periods.
+            - Shiftable: this load can start at any time within the specified time window.
+
+        The resulting schedule provides the power flow at each time period.
+
+        Parameters
+        ==========
+
+        consumption_price_sensor: it defines the utility (economic, environmental, ) in each
+                     time period. It has units of quantity/energy, for example, EUR/kWh.
+        power: nominal power of the load.
+        duration: time that the load last.
+
+        optimization_sense: objective of the scheduler, to maximize or minimize.
+        time_restrictions: time periods in which the load cannot be schedule to.
+        load_type: Inflexible, Breakable or Shiftable.
+
+        :returns:               The computed schedule.
+        """
+
+        if not self.config_deserialized:
+            self.deserialize_config()
+
+        start = self.start.astimezone(pytz.utc)
+        end = self.end.astimezone(pytz.utc)
+        resolution = self.resolution
+        belief_time = self.belief_time
+        sensor = self.sensor
+
+        consumption_price_sensor: Sensor = self.flex_context.get(
+            "consumption_price_sensor"
+        )
+        duration: timedelta = self.flex_model.get("duration")
+        power = self.flex_model.get("power")
+        optimization_sense = self.flex_model.get("optimization_sense")
+        load_type: LoadType = self.flex_model.get("load_type")
+        time_restrictions = self.flex_model.get("time_restrictions")
+
+        # get cost data
+        cost = consumption_price_sensor.search_beliefs(
+            event_starts_after=start,
+            event_ends_before=end,
+            resolution=resolution,
+            one_deterministic_belief_per_event=True,
+            beliefs_before=belief_time,
+        )
+        cost = simplify_index(cost)
+
+        # create an empty schedule
+        schedule = pd.Series(
+            index=pd.date_range(
+                start,
+                end,
+                freq=sensor.event_resolution,
+                inclusive="left",
+                name="event_start",
+            ),
+            data=0,
+            name="event_value",
+        )
+
+        # convert power to energy using the resolution of the sensor.
+        # e.g. resolution=15min, power=1kW -> energy=250W
+        energy = power * consumption_price_sensor.event_resolution / timedelta(hours=1)
+
+        # we can fill duration/resolution rows or, if the duration is larger than the schedule
+        # window, fill the entire window.
+        rows_to_fill = min(
+            ceil(duration / consumption_price_sensor.event_resolution), len(schedule)
+        )
+
+        if rows_to_fill == len(schedule):
+            schedule[:] = energy
+            return schedule
+
+        time_restrictions = (
+            self.block_invalid_starting_times_for_whole_process_scheduling(
+                load_type, time_restrictions, duration, rows_to_fill
+            )
+        )
+
+        # create schedule
+        if load_type == LoadType.INFLEXIBLE:
+            self.compute_inflexible(schedule, time_restrictions, rows_to_fill, energy)
+        elif load_type == LoadType.BREAKABLE:
+            self.compute_breakable(
+                schedule,
+                optimization_sense,
+                time_restrictions,
+                cost,
+                rows_to_fill,
+                energy,
+            )
+        elif load_type == LoadType.SHIFTABLE:
+            self.compute_shiftable(
+                schedule,
+                optimization_sense,
+                time_restrictions,
+                cost,
+                rows_to_fill,
+                energy,
+            )
+        else:
+            raise ValueError(f"Unknown load type '{load_type}'")
+
+        return schedule.tz_convert(self.start.tzinfo)
+
+    def block_invalid_starting_times_for_whole_process_scheduling(
+        self,
+        load_type: LoadType,
+        time_restrictions: pd.Series,
+        duration: timedelta,
+        rows_to_fill: int,
+    ) -> pd.Series:
+        """Blocks time periods where the load cannot be schedule into, making
+          sure no other time restrictions runs in the middle of the activation of the load
+
+        More technically, this function applying an erosion of the time_restrictions array with a block of length duration.
+
+        Then, the condition if time_restrictions.sum() == len(time_restrictions):, makes sure that at least we have a spot to place the load.
+
+        For example:
+
+            time_restriction = [1 0 0 1 1 1 0 0 1 0]
+
+            # applying a dilation with duration = 2
+            time_restriction = [1 0 1 1 1 1 0 1 1 1]
+
+        We can only fit a block of duration = 2 in the positions 1 and 6. sum(time_restrictions) == 8,
+        while the len(time_restriction) == 10, which means we have 10-8=2 positions.
+
+        :param load_type: INFLEXIBLE, SHIFTABLE or BREAKABLE
+        :param time_restrictions: boolean time series indicating time periods in which the load cannot be scheduled.
+        :param duration: (datetime) duration of the length
+        :param rows_to_fill: (int) time periods that the load lasts
+        :return: filtered time restrictions
+        """
+
+        if load_type in [LoadType.INFLEXIBLE, LoadType.SHIFTABLE]:
+            # get start time instants that are not feasible, i.e. some time during the ON period goes through
+            # a time restriction interval
+            time_restrictions = (
+                time_restrictions.rolling(duration).max().shift(-rows_to_fill + 1)
+            )
+            time_restrictions = (time_restrictions == 1) | time_restrictions.isna()
+
+            if time_restrictions.sum() == len(time_restrictions):
+                raise ValueError(
+                    "Cannot allocate a block of time {duration} given the time restrictions provided."
+                )
+        else:  # LoadType.BREAKABLE
+            if (~time_restrictions).sum() < rows_to_fill:
+                raise ValueError(
+                    "Cannot allocate a block of time {duration} given the time restrictions provided."
+                )
+        return time_restrictions
+
+    def compute_inflexible(
+        self,
+        schedule: pd.Series,
+        time_restrictions: pd.Series,
+        rows_to_fill: int,
+        energy: float,
+    ) -> None:
+        """Schedule load as early as possible."""
+        start = time_restrictions[~time_restrictions].index[0]
+
+        schedule.loc[start : start + self.resolution * (rows_to_fill - 1)] = energy
+
+    def compute_breakable(
+        self,
+        schedule: pd.Series,
+        optimization_sense: OptimizationSense,
+        time_restrictions: pd.Series,
+        cost: pd.DataFrame,
+        rows_to_fill: int,
+        energy: float,
+    ) -> None:
+        """Break up schedule and divide it over the time slots with the largest utility (max/min cost depending on optimization_sense)."""
+        cost = cost[~time_restrictions].reset_index()
+
+        if optimization_sense == OptimizationSense.MIN:
+            cost_ranking = cost.sort_values(
+                by=["event_value", "event_start"], ascending=[True, True]
+            )
+        else:
+            cost_ranking = cost.sort_values(
+                by=["event_value", "event_start"], ascending=[False, True]
+            )
+
+        schedule.loc[cost_ranking.head(rows_to_fill).event_start] = energy
+
+    def compute_shiftable(
+        self,
+        schedule: pd.Series,
+        optimization_sense: OptimizationSense,
+        time_restrictions: pd.Series,
+        cost: pd.DataFrame,
+        rows_to_fill: int,
+        energy: float,
+    ) -> None:
+        """Schedules a block of consumption/production of `rows_to_fill` periods to maximize a utility."""
+        block_cost = simplify_index(
+            cost.rolling(rows_to_fill).sum().shift(-rows_to_fill + 1)
+        )
+
+        if optimization_sense == OptimizationSense.MIN:
+            start = block_cost[~time_restrictions].idxmin()
+        else:
+            start = block_cost[~time_restrictions].idxmax()
+
+        start = start.iloc[0]
+
+        schedule.loc[start : start + self.resolution * (rows_to_fill - 1)] = energy
+
+    def deserialize_flex_config(self):
+        """Deserialize flex_model using the schema ShiftableLoadFlexModelSchema and
+        flex_context using FlexContextSchema
+        """
+        if self.flex_model is None:
+            self.flex_model = {}
+
+        self.flex_model = ShiftableLoadFlexModelSchema(
+            start=self.start, end=self.end, sensor=self.sensor
+        ).load(self.flex_model)
+
+        self.flex_context = FlexContextSchema().load(self.flex_context)

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

@@ -187,6 +187,22 @@ def add_inflexible_device_forecasts(
     }
 
 
+@pytest.fixture(scope="module")
+def shiftable_load(db, building, setup_sources) -> dict[str, Sensor]:
+    """
+    Set up a shiftable load sensor where the output of the optimization is stored.
+    """
+    _shiftable_load = Sensor(
+        name="Shiftable Load",
+        generic_asset=building,
+        event_resolution=timedelta(hours=1),
+        unit="kWh",
+    )
+    db.session.add(_shiftable_load)
+
+    return _shiftable_load
+
+
 def add_as_beliefs(db, sensor, values, time_slots, source):
     beliefs = [
         TimedBelief(