Document & refactor scheduling specs for storage flexibility model

documentation/changelog.rst

@@ -23,7 +23,7 @@ Infrastructure / Support
 
 * Reduce size of Docker image (from 2GB to 1.4GB) [see `PR #512 <http://www.github.com/FlexMeasures/flexmeasures/pull/512>`_]
 * Remove bokeh dependency and obsolete UI views [see `PR #476 <http://www.github.com/FlexMeasures/flexmeasures/pull/476>`_]
-* Improve documentation and code w.r.t. storage flexibility mnodeling [see `PR #511 <http://www.github.com/FlexMeasures/flexmeasures/pull/511>`_]
+* Improve documentation and code w.r.t. storage flexibility modelling [see `PR #511 <http://www.github.com/FlexMeasures/flexmeasures/pull/511>`_]
 
 
 v0.11.2 | September 6, 2022

documentation/plugin/customisation.rst

@@ -16,7 +16,7 @@ but in the background your custom scheduling algorithm is being used.
 Let's walk through an example!
 
 First, we need to write a function which accepts arguments just like the in-built schedulers (their code is `here <https://github.com/FlexMeasures/flexmeasures/tree/main/flexmeasures/data/models/planning>`_).
-The following minimal example gives you an idea of the inputs and outputs:
+The following minimal example gives you an idea of some meta information you can add for labeling your data, as well as the inputs and outputs of such a scheduling function:
 
 .. code-block:: python
 
@@ -25,6 +25,10 @@ The following minimal example gives you an idea of the inputs and outputs:
     from pandas.tseries.frequencies import to_offset
     from flexmeasures.data.models.time_series import Sensor
 
+
+    __author__ = "My Company"
+    __version__ = "v2"
+
     def compute_a_schedule(
         sensor: Sensor,
         start: datetime,
@@ -44,7 +48,7 @@ The following minimal example gives you an idea of the inputs and outputs:
 
 
 .. note:: It's possible to add arguments that describe the asset flexibility and the EMS context in more detail. For example,
-          for storage assets we support various state-of-charge parameters. For now, the existing schedulers are the best documentation.
+          for storage assets we support various state-of-charge parameters. For now, the existing in-built schedulers are the best documentation.
 
 
 Finally, make your scheduler be the one that FlexMeasures will use for certain sensors:
@@ -57,7 +61,6 @@ Finally, make your scheduler be the one that FlexMeasures will use for certain s
     scheduler_specs = {
         "module": "flexmeasures.data.tests.dummy_scheduler",  # or a file path, see note below
         "function": "compute_a_schedule",
-        "source": "My Company"
     }
 
     my_sensor = Sensor.query.filter(Sensor.name == "My power sensor on a flexible asset").one_or_none()

flexmeasures/api/v1_3/implementations.py

@@ -99,6 +99,7 @@ def get_device_message_response(generic_asset_name_groups, duration):
             if event_type not in ("soc", "soc-with-targets"):
                 return unrecognized_event_type(event_type)
             connection = current_app.queues["scheduling"].connection
+            job = None
             try:  # First try the scheduling queue
                 job = Job.fetch(event, connection=connection)
             except NoSuchJobError:  # Then try the most recent event_id (stored as a generic asset attribute)
@@ -144,19 +145,25 @@ def get_device_message_response(generic_asset_name_groups, duration):
                     return unknown_schedule("Scheduling job has an unknown status.")
                 schedule_start = job.kwargs["start"]
 
-            schedule_data_source_name = "Seita"
-            scheduler_source = DataSource.query.filter_by(
-                name=schedule_data_source_name, type="scheduling script"
-            ).one_or_none()
-            if scheduler_source is None:
-                return unknown_schedule(
-                    message + f'no data is known from "{schedule_data_source_name}".'
-                )
+            data_source_info = None
+            if job:
+                data_source_info = job.meta.get("data_source_info")
+            if data_source_info is None:
+                data_source_info = dict(
+                    name="Seita"
+                )  # TODO: change to raise later - all scheduling jobs now get full info
+            scheduler_sources = DataSource.query.filter_by(
+                type="scheduling script",
+                **data_source_info,
+            ).all()  # Might be more than one, e.g. per user
+            if len(scheduler_sources) == 0:
+                s_info = ",".join([f"{k}={v}" for k, v in data_source_info.items()])
+                return unknown_schedule(message + f"no data is known from [{s_info}].")
 
             power_values = sensor.search_beliefs(
                 event_starts_after=schedule_start,
                 event_ends_before=schedule_start + planning_horizon,
-                source=scheduler_source,
+                source=scheduler_sources[-1],
                 most_recent_beliefs_only=True,
                 one_deterministic_belief_per_event=True,
             )

flexmeasures/api/v1_3/tests/test_api_v1_3.py

@@ -88,9 +88,10 @@ def test_post_udi_event_and_get_device_message(
     )
 
     # check results are in the database
-    resolution = timedelta(minutes=15)
+    job.refresh()  # catch meta info that was added on this very instance
+    data_source_info = job.meta.get("data_source_info")
     scheduler_source = DataSource.query.filter_by(
-        name="Seita", type="scheduling script"
+        type="scheduling script", **data_source_info
     ).one_or_none()
     assert (
         scheduler_source is not None
@@ -100,6 +101,7 @@ def test_post_udi_event_and_get_device_message(
         .filter(TimedBelief.source_id == scheduler_source.id)
         .all()
     )
+    resolution = timedelta(minutes=15)
     consumption_schedule = pd.Series(
         [-v.event_value for v in power_values],
         index=pd.DatetimeIndex([v.event_start for v in power_values], freq=resolution),

flexmeasures/api/v3_0/sensors.py

@@ -279,7 +279,7 @@ def trigger_schedule(  # noqa: C901
             - soc-max (defaults to max soc target)
             - soc-targets (defaults to NaN values)
             - roundtrip-efficiency (defaults to 100%)
-            - prefer-charging-sooner (defaults to True)
+            - prefer-charging-sooner (defaults to True, also signals a preference to discharge later)
 
         2) Heat pump sensors are work in progress.
 
@@ -552,21 +552,25 @@ def get_schedule(self, sensor: Sensor, job_id: str, duration: timedelta, **kwarg
             return unknown_schedule("Scheduling job has an unknown status.")
         schedule_start = job.kwargs["start"]
 
-        schedule_data_source_name = "Seita"
-        if "data_source_name" in job.meta:
-            schedule_data_source_name = job.meta["data_source_name"]
-        scheduler_source = DataSource.query.filter_by(
-            name=schedule_data_source_name, type="scheduling script"
-        ).one_or_none()
-        if scheduler_source is None:
+        data_source_info = job.meta.get("data_source_info")
+        if data_source_info is None:
+            data_source_info = dict(
+                name="Seita"
+            )  # TODO: change to raise later - all scheduling jobs now get full info
+        scheduler_sources = DataSource.query.filter_by(
+            type="scheduling script",
+            **data_source_info,
+        ).all()  # there can be more than one, e.g. different users
+        if len(scheduler_sources) == 0:
+            s_info = ",".join([f"{k}={v}" for k, v in data_source_info.items()])
             return unknown_schedule(
-                error_message + f'no data is known from "{schedule_data_source_name}".'
+                error_message + f"no data is known from [{s_info}]."
             )
 
         power_values = sensor.search_beliefs(
             event_starts_after=schedule_start,
             event_ends_before=schedule_start + planning_horizon,
-            source=scheduler_source,
+            source=scheduler_sources[-1],
             most_recent_beliefs_only=True,
             one_deterministic_belief_per_event=True,
         )

flexmeasures/api/v3_0/tests/test_sensor_schedules.py

@@ -66,9 +66,10 @@ def test_trigger_and_get_schedule(
     )
 
     # check results are in the database
-    resolution = timedelta(minutes=15)
+    job.refresh()  # catch meta info that was added on this very instance
+    data_source_info = job.meta.get("data_source_info")
     scheduler_source = DataSource.query.filter_by(
-        name="Seita", type="scheduling script"
+        type="scheduling script", **data_source_info
     ).one_or_none()
     assert (
         scheduler_source is not None
@@ -78,6 +79,7 @@ def test_trigger_and_get_schedule(
         .filter(TimedBelief.source_id == scheduler_source.id)
         .all()
     )
+    resolution = timedelta(minutes=15)
     consumption_schedule = pd.Series(
         [-v.event_value for v in power_values],
         index=pd.DatetimeIndex([v.event_start for v in power_values], freq=resolution),

flexmeasures/data/models/planning/battery.py

@@ -15,6 +15,10 @@
 )
 
 
+__version__ = "1"
+__author__ = "Seita"
+
+
 def schedule_battery(
     sensor: Sensor,
     start: datetime,

flexmeasures/data/models/planning/charging_station.py

@@ -14,6 +14,9 @@
     fallback_charging_policy,
 )
 
+__version__ = "1"
+__author__ = "Seita"
+
 
 def schedule_charging_station(
     sensor: Sensor,

flexmeasures/data/models/planning/utils.py

@@ -46,7 +46,6 @@ def initialize_index(
     resolution: timedelta,
     inclusive: str = "left",
 ) -> pd.DatetimeIndex:
-    assert inclusive == "left" or inclusive == "right"
     i = pd.date_range(
         start=start,
         end=end,
@@ -78,20 +77,14 @@ def ensure_storage_specs(
     if specs is None:
         specs = {}
 
-    sensor: Optional[Sensor] = None
-
-    def ensure_sensor_is_set(sensor) -> Sensor:
-        if sensor is None:
-            sensor = Sensor.query.filter_by(id=sensor_id).one_or_none()
-        return sensor
+    sensor: Optional[Sensor] = Sensor.query.filter_by(id=sensor_id).one_or_none()
 
     # Check state of charge
     # Preferably, a starting soc is given.
     # Otherwise, we try to retrieve the current state of charge from the asset (if that is the valid one at the start).
     # Otherwise, we set the starting soc to 0 (some assets don't use the concept of a state of charge,
     # and without soc targets and limits the starting soc doesn't matter).
     if "soc_at_start" not in specs or specs["soc_at_start"] is None:
-        sensor = ensure_sensor_is_set(sensor)
         if (
             start_of_schedule == sensor.get_attribute("soc_datetime")
             and sensor.get_attribute("soc_in_mwh") is not None
@@ -115,19 +108,16 @@ def ensure_sensor_is_set(sensor) -> Sensor:
 
     # Check for min and max SOC, or get default from sensor
     if "soc_min" not in specs or specs["soc_min"] is None:
-        sensor = ensure_sensor_is_set(sensor)
         # Can't drain the storage by more than it contains
         specs["soc_min"] = sensor.get_attribute("min_soc_in_mwh", 0)
     if "soc_max" not in specs or specs["soc_max"] is None:
-        sensor = ensure_sensor_is_set(sensor)
         # Lacking information about the battery's nominal capacity, we use the highest target value as the maximum state of charge
         specs["soc_max"] = sensor.get_attribute(
             "max_soc_in_mwh", max(specs["soc_targets"].values)
         )
 
     # Check for round-trip efficiency
     if "roundtrip_efficiency" not in specs or specs["roundtrip_efficiency"] is None:
-        sensor = ensure_sensor_is_set(sensor)
         # Get default from sensor, or use 100% otherwise
         specs["roundtrip_efficiency"] = sensor.get_attribute("roundtrip_efficiency", 1)
     if specs["roundtrip_efficiency"] <= 0 or specs["roundtrip_efficiency"] > 1:

flexmeasures/data/services/scheduling.py

@@ -115,7 +115,9 @@ def make_schedule(
     db.engine.dispose()
 
     sensor = Sensor.query.filter_by(id=sensor_id).one_or_none()
-    data_source_name = "Seita"
+    data_source_info = dict(
+        name="Seita", model="Unknown", version="1"
+    )  # will be overwritten by scheduler choice below
 
     rq_job = get_current_job()
     if rq_job:
@@ -127,24 +129,27 @@ def make_schedule(
     # Choose which algorithm to use
     if "custom-scheduler" in sensor.attributes:
         scheduler_specs = sensor.attributes.get("custom-scheduler")
-        scheduler, data_source_name = load_custom_scheduler(scheduler_specs)
-        if rq_job:
-            rq_job.meta["data_source_name"] = data_source_name
-            rq_job.save_meta()
+        scheduler, data_source_info = load_custom_scheduler(scheduler_specs)
     elif sensor.generic_asset.generic_asset_type.name == "battery":
         scheduler = schedule_battery
+        data_source_info["model"] = "schedule_battery"
     elif sensor.generic_asset.generic_asset_type.name in (
         "one-way_evse",
         "two-way_evse",
     ):
         scheduler = schedule_charging_station
-
+        data_source_info["model"] = "schedule_charging_station"
     else:
         raise ValueError(
             "Scheduling is not (yet) supported for asset type %s."
             % sensor.generic_asset.generic_asset_type
         )
 
+    # saving info on the job, so the API for a job can look the data up
+    if rq_job:
+        rq_job.meta["data_source_info"] = data_source_info
+        rq_job.save_meta()
+
     consumption_schedule = scheduler(
         sensor,
         start,
@@ -156,13 +161,15 @@ def make_schedule(
         inflexible_device_sensors=inflexible_device_sensors,
         belief_time=belief_time,
     )
+    if rq_job:
+        click.echo("Job %s made schedule." % rq_job.id)
 
     data_source = get_data_source(
-        data_source_name=data_source_name,
+        data_source_name=data_source_info["name"],
+        data_source_model=data_source_info["model"],
+        data_source_version=data_source_info["version"],
         data_source_type="scheduling script",
     )
-    if rq_job:
-        click.echo("Job %s made schedule." % rq_job.id)
 
     ts_value_schedule = [
         TimedBelief(
@@ -181,17 +188,16 @@ def make_schedule(
     return True
 
 
-def load_custom_scheduler(scheduler_specs: dict) -> Tuple[Callable, str]:
+def load_custom_scheduler(scheduler_specs: dict) -> Tuple[Callable, dict]:
     """
     Read in custom scheduling spec.
-    Attempt to load the Callable, also derive a data source name.
+    Attempt to load the Callable, also derive data source info.
 
     Example specs:
 
     {
         "module": "/path/to/module.py",  # or sthg importable, e.g. "package.module"
         "function": "name_of_function",
-        "source": "source name"
     }
 
     """
@@ -201,10 +207,8 @@ def load_custom_scheduler(scheduler_specs: dict) -> Tuple[Callable, str]:
     assert "module" in scheduler_specs, "scheduler specs have no 'module'."
     assert "function" in scheduler_specs, "scheduler specs have no 'function'"
 
-    source_name = scheduler_specs.get(
-        "source", f"custom scheduler - {scheduler_specs['function']}"
-    )
     scheduler_name = scheduler_specs["function"]
+    source_info = dict(model=scheduler_name, version="1", name="")  # default  # default
 
     # import module
     module_descr = scheduler_specs["module"]
@@ -233,7 +237,12 @@ def load_custom_scheduler(scheduler_specs: dict) -> Tuple[Callable, str]:
         module, scheduler_specs["function"]
     ), "Module at {module_descr} has no function {scheduler_specs['function']}"
 
-    return getattr(module, scheduler_specs["function"]), source_name
+    if hasattr(module, "__version__"):
+        source_info["version"] = str(module.__version__)
+    if hasattr(module, "__author__"):
+        source_info["name"] = str(module.__author__)
+
+    return getattr(module, scheduler_specs["function"]), source_info
 
 
 def handle_scheduling_exception(job, exc_type, exc_value, traceback):

flexmeasures/data/tests/dummy_scheduler.py

@@ -4,6 +4,10 @@
 from flexmeasures.data.models.planning.utils import initialize_series
 
 
+__author__ = "Test Organization"
+__version__ = "v3"
+
+
 def compute_a_schedule(
     sensor: Sensor,
     start: datetime,