Add optional fields to set custom min/max SOC in UDI Events, and amen…

…d test (#325)

Signed-off-by: F.N. Claessen <felix@seita.nl>

documentation/changelog.rst

@@ -15,6 +15,7 @@ New features
 * Charts with sensor data can be requested in one of the supported  [`vega-lite themes <https://github.com/vega/vega-themes#included-themes>`_] (incl. a dark theme) [see `PR #221 <http://www.github.com/FlexMeasures/flexmeasures/pull/221>`_]
 * Mobile friendly (responsive) charts of sensor data, and such charts can be requested with a custom width and height [see `PR #313 <http://www.github.com/FlexMeasures/flexmeasures/pull/313>`_]
 * Schedulers take into account round-trip efficiency if set [see `PR #291 <http://www.github.com/FlexMeasures/flexmeasures/pull/291>`_]
+* Schedulers take into account min/max state of charge if set [see `PR #325 <http://www.github.com/FlexMeasures/flexmeasures/pull/325>`_]
 * Fallback policies for charging schedules of batteries and Charge Points, in cases where the solver is presented with an infeasible problem [see `PR #267 <http://www.github.com/FlexMeasures/flexmeasures/pull/267>`_ and `PR #270 <http://www.github.com/FlexMeasures/flexmeasures/pull/270>`_]
 
 Deprecations

flexmeasures/api/v1_3/implementations.py

@@ -285,6 +285,14 @@ def post_udi_event_response(unit: str, prior: datetime):
     # get optional efficiency
     roundtrip_efficiency = form.get("roundtrip_efficiency", None)
 
+    # get optional min and max SOC
+    soc_min = form.get("soc_min", None)
+    soc_max = form.get("soc_max", None)
+    if soc_min is not None and unit == "kWh":
+        soc_min = soc_min / 1000.0
+    if soc_max is not None and unit == "kWh":
+        soc_max = soc_max / 1000.0
+
     # set soc targets
     start_of_schedule = datetime
     end_of_schedule = datetime + current_app.config.get("FLEXMEASURES_PLANNING_HORIZON")
@@ -354,6 +362,8 @@ def post_udi_event_response(unit: str, prior: datetime):
         belief_time=prior,  # server time if no prior time was sent
         soc_at_start=value,
         soc_targets=soc_targets,
+        soc_min=soc_min,
+        soc_max=soc_max,
         roundtrip_efficiency=roundtrip_efficiency,
         udi_event_ea=form.get("event"),
         enqueue=True,

flexmeasures/api/v1_3/routes.py

@@ -104,6 +104,7 @@ def post_udi_event():
     This "PostUdiEventRequest" message posts a state of charge (soc) of 12.1 kWh at 10.00am,
     and a target state of charge of 25 kWh at 4.00pm,
     as UDI event 204 of device 10 of owner 7.
+    The minimum and maximum soc are set to 10 and 25 kWh, respectively.
     Roundtrip efficiency for use in scheduling is set to 98%.
 
     .. code-block:: json
@@ -120,6 +121,8 @@ def post_udi_event():
                     "datetime": "2015-06-02T16:00:00+00:00"
                 }
             ],
+            "soc_min": 10,
+            "soc_max": 25,
             "roundtrip_efficiency": 0.98
         }
 

flexmeasures/data/models/planning/battery.py

@@ -21,6 +21,8 @@ def schedule_battery(
     resolution: timedelta,
     soc_at_start: float,
     soc_targets: Optional[pd.Series] = None,
+    soc_min: Optional[float] = None,
+    soc_max: Optional[float] = None,
     roundtrip_efficiency: Optional[float] = None,
     prefer_charging_sooner: bool = True,
 ) -> Union[pd.Series, None]:
@@ -45,6 +47,12 @@ def schedule_battery(
     if roundtrip_efficiency <= 0 or roundtrip_efficiency > 1:
         raise ValueError("roundtrip_efficiency expected within the interval (0, 1]")
 
+    # Check for min and max SOC, or get default from sensor
+    if soc_min is None:
+        soc_min = sensor.get_attribute("min_soc_in_mwh")
+    if soc_max is None:
+        soc_max = sensor.get_attribute("max_soc_in_mwh")
+
     # Check for known prices or price forecasts, trimming planning window accordingly
     prices, (start, end) = get_prices(
         sensor, (start, end), resolution, allow_trimmed_query_window=True
@@ -89,12 +97,12 @@ def schedule_battery(
         )  # shift "equals" constraint for target SOC by one resolution (the target defines a state at a certain time,
         # while the "equals" constraint defines what the total stock should be at the end of a time slot,
         # where the time slot is indexed by its starting time)
-    device_constraints[0]["min"] = (
-        sensor.get_attribute("min_soc_in_mwh") - soc_at_start
-    ) * (timedelta(hours=1) / resolution)
-    device_constraints[0]["max"] = (
-        sensor.get_attribute("max_soc_in_mwh") - soc_at_start
-    ) * (timedelta(hours=1) / resolution)
+    device_constraints[0]["min"] = (soc_min - soc_at_start) * (
+        timedelta(hours=1) / resolution
+    )
+    device_constraints[0]["max"] = (soc_max - soc_at_start) * (
+        timedelta(hours=1) / resolution
+    )
     device_constraints[0]["derivative min"] = (
         sensor.get_attribute("capacity_in_mw") * -1
     )

flexmeasures/data/models/planning/charging_station.py

@@ -21,6 +21,8 @@ def schedule_charging_station(
     resolution: timedelta,
     soc_at_start: float,
     soc_targets: Series,
+    soc_min: Optional[float] = None,
+    soc_max: Optional[float] = None,
     roundtrip_efficiency: Optional[float] = None,
     prefer_charging_sooner: bool = True,
 ) -> Union[Series, None]:
@@ -40,6 +42,14 @@ def schedule_charging_station(
     if roundtrip_efficiency <= 0 or roundtrip_efficiency > 1:
         raise ValueError("roundtrip_efficiency expected within the interval (0, 1]")
 
+    # Check for min and max SOC, or get default from sensor
+    if soc_min is None:
+        # Can't drain the EV battery by more than it contains
+        soc_min = sensor.get_attribute("min_soc_in_mwh", 0)
+    if soc_max is None:
+        # Lacking information about the battery's nominal capacity, we use the highest target value as the maximum state of charge
+        soc_max = sensor.get_attribute("max_soc_in_mwh", max(soc_targets.values))
+
     # Check for known prices or price forecasts, trimming planning window accordingly
     prices, (start, end) = get_prices(
         sensor, (start, end), resolution, allow_trimmed_query_window=True
@@ -83,14 +93,12 @@ def schedule_charging_station(
     )  # shift "equals" constraint for target SOC by one resolution (the target defines a state at a certain time,
     # while the "equals" constraint defines what the total stock should be at the end of a time slot,
     # where the time slot is indexed by its starting time)
-    device_constraints[0]["min"] = -soc_at_start * (
-        timedelta(hours=1) / resolution
-    )  # Can't drain the EV battery by more than it contains
-    device_constraints[0]["max"] = max(soc_targets.values) * (
+    device_constraints[0]["min"] = (soc_min - soc_at_start) * (
         timedelta(hours=1) / resolution
-    ) - soc_at_start * (
+    )
+    device_constraints[0]["max"] = (soc_max - soc_at_start) * (
         timedelta(hours=1) / resolution
-    )  # Lacking information about the battery's nominal capacity, we use the highest target value as the maximum state of charge
+    )
 
     if sensor.get_attribute("is_strictly_non_positive"):
         device_constraints[0]["derivative min"] = 0

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

@@ -64,12 +64,16 @@ def test_battery_solver_day_2(add_battery_assets, roundtrip_efficiency: float):
     end = as_server_time(datetime(2015, 1, 3))
     resolution = timedelta(minutes=15)
     soc_at_start = battery.get_attribute("soc_in_mwh")
+    soc_min = 0.5
+    soc_max = 4.5
     schedule = schedule_battery(
         battery,
         start,
         end,
         resolution,
         soc_at_start,
+        soc_min=soc_min,
+        soc_max=soc_max,
         roundtrip_efficiency=roundtrip_efficiency,
     )
     soc_schedule = integrate_time_series(schedule, soc_at_start, decimal_precision=6)
@@ -81,21 +85,21 @@ def test_battery_solver_day_2(add_battery_assets, roundtrip_efficiency: float):
     assert min(schedule.values) >= battery.get_attribute("capacity_in_mw") * -1
     assert max(schedule.values) <= battery.get_attribute("capacity_in_mw") + TOLERANCE
     for soc in soc_schedule.values:
-        assert soc >= battery.get_attribute("min_soc_in_mwh")
+        assert soc >= max(soc_min, battery.get_attribute("min_soc_in_mwh"))
         assert soc <= battery.get_attribute("max_soc_in_mwh")
 
     # Check whether the resulting soc schedule follows our expectations for 8 expensive, 8 cheap and 8 expensive hours
-    assert soc_schedule.iloc[-1] == battery.get_attribute(
-        "min_soc_in_mwh"
+    assert soc_schedule.iloc[-1] == max(
+        soc_min, battery.get_attribute("min_soc_in_mwh")
     )  # Battery sold out at the end of its planning horizon
 
     # As long as the roundtrip efficiency isn't too bad (I haven't computed the actual switch point)
     if roundtrip_efficiency > 0.9:
-        assert soc_schedule.loc[start + timedelta(hours=8)] == battery.get_attribute(
-            "min_soc_in_mwh"
+        assert soc_schedule.loc[start + timedelta(hours=8)] == max(
+            soc_min, battery.get_attribute("min_soc_in_mwh")
         )  # Sell what you begin with
-        assert soc_schedule.loc[start + timedelta(hours=16)] == battery.get_attribute(
-            "max_soc_in_mwh"
+        assert soc_schedule.loc[start + timedelta(hours=16)] == min(
+            soc_max, battery.get_attribute("max_soc_in_mwh")
         )  # Buy what you can to sell later
     else:
         # If the roundtrip efficiency is poor, best to stand idle

flexmeasures/data/services/scheduling.py

@@ -36,6 +36,8 @@ def create_scheduling_job(
     resolution: timedelta = DEFAULT_RESOLUTION,
     soc_at_start: Optional[float] = None,
     soc_targets: Optional[pd.Series] = None,
+    soc_min: Optional[float] = None,
+    soc_max: Optional[float] = None,
     roundtrip_efficiency: Optional[float] = None,
     udi_event_ea: Optional[str] = None,
     enqueue: bool = True,
@@ -62,6 +64,8 @@ def create_scheduling_job(
             resolution=resolution,
             soc_at_start=soc_at_start,
             soc_targets=soc_targets,
+            soc_min=soc_min,
+            soc_max=soc_max,
             roundtrip_efficiency=roundtrip_efficiency,
         ),
         id=udi_event_ea,
@@ -90,6 +94,8 @@ def make_schedule(
     resolution: timedelta,
     soc_at_start: Optional[float] = None,
     soc_targets: Optional[pd.Series] = None,
+    soc_min: Optional[float] = None,
+    soc_max: Optional[float] = None,
     roundtrip_efficiency: Optional[float] = None,
 ) -> bool:
     """Preferably, a starting soc is given.
@@ -131,6 +137,8 @@ def make_schedule(
             resolution,
             soc_at_start,
             soc_targets,
+            soc_min,
+            soc_max,
             roundtrip_efficiency,
         )
     elif sensor.generic_asset.generic_asset_type.name in (
@@ -144,6 +152,8 @@ def make_schedule(
             resolution,
             soc_at_start,
             soc_targets,
+            soc_min,
+            soc_max,
             roundtrip_efficiency,
         )
     else: