feat: add state-of-charge sensor to flex-model

.gitignore

@@ -50,4 +50,8 @@ poetry.lock
 .gitconfig.*
 
 /postgres-data
-coverage.lcov
+coverage.lcov
+venv*
+logs/
+*.dump
+iframe_figures/

documentation/changelog.rst

@@ -10,6 +10,8 @@ v0.26.0 | May XX, 2025
 New features
 -------------
 
+* Support saving the storage schedule SOC using the ``flex-model`` field ``state-of-charge`` to the ``flex-model`` [see `PR #1392 <https://github.com/FlexMeasures/flexmeasures/pull/1392>`_]
+
 Infrastructure / Support
 ----------------------
 

flexmeasures/api/v3_0/sensors.py

@@ -472,6 +472,7 @@ def trigger_schedule(
                 "duration": "PT24H",
                 "flex-model": {
                     "soc-at-start": "12.1 kWh",
+                    "state-of-charge" : {"sensor" : 24},
                     "soc-targets": [
                         {
                             "value": "25 kWh",

flexmeasures/cli/data_add.py

@@ -1219,6 +1219,14 @@ def create_schedule(ctx):
     required=True,
     help="State of charge (e.g 32.8%, or 0.328) at the start of the schedule.",
 )
+@click.option(
+    "--state-of-charge",
+    "state_of_charge",
+    type=SensorIdField(unit="MWh"),
+    help="State of charge sensor.",
+    required=False,
+    default=None,
+)
 @click.option(
     "--soc-target",
     "soc_target_strings",
@@ -1365,6 +1373,7 @@ def add_schedule_for_storage(  # noqa C901
     soc_max: ur.Quantity | None = None,
     roundtrip_efficiency: ur.Quantity | None = None,
     storage_efficiency: ur.Quantity | Sensor | None = None,
+    state_of_charge: Sensor | None = None,
     as_job: bool = False,
 ):
     """Create a new schedule for a storage asset.
@@ -1443,6 +1452,11 @@ def add_schedule_for_storage(  # noqa C901
         },
     )
 
+    if state_of_charge is not None:
+        scheduling_kwargs["flex_model"]["state-of-charge"] = {
+            "sensor": state_of_charge.id
+        }
+
     quantity_or_sensor_vars = {
         "flex_model": {
             "charging-efficiency": charging_efficiency,

flexmeasures/data/models/planning/storage.py

@@ -34,6 +34,10 @@
 from flexmeasures.utils.time_utils import determine_minimum_resampling_resolution
 from flexmeasures.utils.unit_utils import ur, convert_units
 
+from flexmeasures.utils.calculations import (
+    integrate_time_series,
+)
+
 
 class MetaStorageScheduler(Scheduler):
     """This class defines the constraints of a schedule for a storage device from the
@@ -1002,6 +1006,28 @@ def compute(self, skip_validation: bool = False) -> SchedulerOutputType:
             for sensor in sensors
         }
 
+        flex_model = self.flex_model
+
+        if not isinstance(self.flex_model, list):
+            flex_model["sensor"] = sensors[0]
+            flex_model = [flex_model]
+
+        soc_schedule = {
+            flex_model_d["state_of_charge"]: convert_units(
+                integrate_time_series(
+                    series=ems_schedule[d],
+                    initial_stock=soc_at_start[d],
+                    up_efficiency=device_constraints[d]["derivative up efficiency"],
+                    down_efficiency=device_constraints[d]["derivative down efficiency"],
+                    storage_efficiency=device_constraints[d]["efficiency"].fillna(1),
+                ),
+                from_unit="MWh",
+                to_unit=flex_model_d["state_of_charge"].unit,
+            )
+            for d, flex_model_d in enumerate(flex_model)
+            if isinstance(flex_model_d.get("state_of_charge", None), Sensor)
+        }
+
         # Resample each device schedule to the resolution of the device's power sensor
         if self.resolution is None:
             storage_schedule = {
@@ -1017,26 +1043,37 @@ def compute(self, skip_validation: bool = False) -> SchedulerOutputType:
                 sensor: storage_schedule[sensor].round(self.round_to_decimals)
                 for sensor in sensors
             }
+            soc_schedule = {
+                sensor: soc_schedule[sensor].round(self.round_to_decimals)
+                for sensor in soc_schedule.keys()
+            }
 
         if self.return_multiple:
-            return [
-                {
-                    "name": "storage_schedule",
-                    "sensor": sensor,
-                    "data": storage_schedule[sensor],
-                }
-                for sensor in sensors
-            ] + [
-                {
-                    "name": "commitment_costs",
-                    "data": {
-                        c.name: costs
-                        for c, costs in zip(
-                            commitments, model.commitment_costs.values()
-                        )
+            return (
+                [
+                    {
+                        "name": "storage_schedule",
+                        "sensor": sensor,
+                        "data": storage_schedule[sensor],
+                    }
+                    for sensor in sensors
+                ]
+                + [
+                    {
+                        "name": "commitment_costs",
+                        "data": {
+                            c.name: costs
+                            for c, costs in zip(
+                                commitments, model.commitment_costs.values()
+                            )
+                        },
                     },
-                },
-            ]
+                ]
+                + [
+                    {"name": "state_of_charge", "data": soc, "sensor": sensor}
+                    for sensor, soc in soc_schedule.items()
+                ]
+            )
         else:
             return storage_schedule[sensors[0]]
 

flexmeasures/data/schemas/scheduling/storage.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from datetime import datetime
+from datetime import datetime, timedelta
 
 from flask import current_app
 from marshmallow import (
@@ -123,6 +123,12 @@ class StorageFlexModelSchema(Schema):
         required=False,
     )
 
+    state_of_charge = VariableQuantityField(
+        to_unit="MWh",
+        data_key="state-of-charge",
+        required=False,
+    )
+
     charging_efficiency = VariableQuantityField(
         "%", data_key="charging-efficiency", required=False
     )
@@ -216,6 +222,20 @@ def check_whether_targets_exceed_max_planning_horizon(self, data: dict, **kwargs
                 f"Target datetime exceeds {max_server_datetime}. Maximum scheduling horizon is {max_server_horizon}."
             )
 
+    @validates("state_of_charge")
+    def validate_state_of_charge_is_sensor(
+        self, state_of_charge: Sensor | list[dict] | ur.Quantity
+    ):
+        if not isinstance(state_of_charge, Sensor):
+            raise ValidationError(
+                "The `state-of-charge` field can only be a Sensor. In the future, the state-of-charge field will replace soc-at-start field."
+            )
+
+        if state_of_charge.event_resolution != timedelta(0):
+            raise ValidationError(
+                "The field `state-of-charge` points to a sensor with a non-instantaneous event resolution. Please, use an instantaneous sensor."
+            )
+
     @validates("storage_efficiency")
     def validate_storage_efficiency_resolution(self, unit: Sensor | ur.Quantity):
         if (

flexmeasures/data/services/scheduling.py

@@ -12,6 +12,7 @@
 from typing import Callable, Type
 import inspect
 from copy import deepcopy
+from traceback import print_tb
 
 
 from flask import current_app
@@ -646,8 +647,8 @@ def handle_scheduling_exception(job, exc_type, exc_value, traceback):
     click.echo(
         "HANDLING RQ SCHEDULING WORKER EXCEPTION: %s:%s\n" % (exc_type, exc_value)
     )
-    # from traceback import print_tb
-    # print_tb(traceback)
+
+    print_tb(traceback)
     job.meta["exception"] = exc_value
     job.save_meta()
 

flexmeasures/data/tests/conftest.py

@@ -320,10 +320,11 @@ def smart_building(app, fresh_db, smart_building_types):
     fresh_db.session.add_all(assets)
     fresh_db.session.flush()
 
-    sensors = []
+    power_sensors = []
+    soc_sensors = []
 
-    # Add power sensor
     for asset in assets:
+        # Add power sensor
         sensor = Sensor(
             name="power",
             unit="MW",
@@ -335,12 +336,27 @@ def smart_building(app, fresh_db, smart_building_types):
             generic_asset=asset,
             timezone="Europe/Amsterdam",
         )
-        sensors.append(sensor)
+        power_sensors.append(sensor)
 
-    fresh_db.session.add_all(sensors)
+        # Add SOC sensors
+        sensor = Sensor(
+            "state of charge",
+            unit="MWh",
+            event_resolution=timedelta(hours=0),
+            generic_asset=asset,
+            timezone="Europe/Amsterdam",
+        )
+        soc_sensors.append(sensor)
+
+    fresh_db.session.add_all(power_sensors)
+    fresh_db.session.add_all(soc_sensors)
     fresh_db.session.flush()
     asset_names = [asset.name for asset in assets]
-    return dict(zip(asset_names, assets)), dict(zip(asset_names, sensors))
+    return (
+        dict(zip(asset_names, assets)),
+        dict(zip(asset_names, power_sensors)),
+        dict(zip(asset_names, soc_sensors)),
+    )
 
 
 @pytest.fixture
@@ -351,7 +367,7 @@ def flex_description_sequential(
 
     Specifically, the main flex model is deserialized, while the sensors' individual flex models are still serialized.
     """
-    assets, sensors = smart_building
+    assets, sensors, soc_sensors = smart_building
 
     flex_model = [
         {
@@ -398,6 +414,7 @@ def flex_description_sequential(
                         "value": 0.094,
                     }  # 6 kWh discharge
                 ],
+                "state-of-charge": {"sensor": soc_sensors["Test Battery"].id},
             },
         },
     ]

flexmeasures/data/tests/test_scheduling_sequential.py

@@ -1,3 +1,4 @@
+from datetime import timedelta
 from unittest.mock import patch
 from flexmeasures.data.models.planning.exceptions import InfeasibleProblemException
 
@@ -7,6 +8,7 @@
 from flexmeasures.data.tests.utils import work_on_rq
 from flexmeasures.data.services.scheduling import handle_scheduling_exception
 from flexmeasures.data.models.time_series import Sensor
+from flexmeasures.utils.calculations import integrate_time_series
 
 
 def test_create_sequential_jobs(db, app, flex_description_sequential, smart_building):
@@ -16,7 +18,10 @@ def test_create_sequential_jobs(db, app, flex_description_sequential, smart_buil
     We verify that the pipeline creates the right number of jobs (two), corresponding to the inflexible devices,
     and an extra one which corresponds to the success callback job.
     """
-    assets, sensors = smart_building
+    assets, sensors, soc_sensors = smart_building
+
+    assert len(soc_sensors["Test Battery"].search_beliefs()) == 0
+
     queue = app.queues["scheduling"]
     start = pd.Timestamp("2015-01-03").tz_localize("Europe/Amsterdam")
     end = pd.Timestamp("2015-01-04").tz_localize("Europe/Amsterdam")
@@ -84,7 +89,7 @@ def test_create_sequential_jobs(db, app, flex_description_sequential, smart_buil
 
     # Work on jobs
     queued_jobs[0].perform()
-    work_on_rq(queue)
+    work_on_rq(queue, handle_scheduling_exception)
 
     # Check that the jobs completed successfully
     assert queued_jobs[0].get_status() == "finished"
@@ -135,6 +140,25 @@ def test_create_sequential_jobs(db, app, flex_description_sequential, smart_buil
     ), f"Battery cost should be -4.415 €, got {battery_costs} €"
     assert total_cost == -2.1775, f"Total cost should be -2.1775 €, got {total_cost} €"
 
+    # Check that the SOC data is saved
+    soc_schedule = (
+        soc_sensors["Test Battery"]
+        .search_beliefs(resolution=timedelta(0))
+        .reset_index()
+    )
+    power_schedule = sensors["Test Battery"].search_beliefs().reset_index()
+
+    power_schedule = pd.Series(
+        power_schedule.event_value.tolist(),
+        index=pd.DatetimeIndex(power_schedule.event_start.tolist(), freq="15min"),
+    )
+    soc_schedule_from_power = integrate_time_series(
+        -power_schedule,
+        0.1,
+        decimal_precision=6,
+    )
+    assert all(soc_schedule.event_value.values == soc_schedule_from_power.values)
+
 
 def test_create_sequential_jobs_fallback(
     db, app, flex_description_sequential, smart_building
@@ -144,7 +168,7 @@ def test_create_sequential_jobs_fallback(
     Checks execution of a sequential scheduling job, where 1 of the subjobs is set up to fail and trigger its fallback.
     The deferred subjobs should still succeed after the fallback succeeds, even though the first subjob fails.
     """
-    assets, sensors = smart_building
+    assets, sensors, _ = smart_building
     queue = app.queues["scheduling"]
 
     start = pd.Timestamp("2015-01-03").tz_localize("Europe/Amsterdam")

flexmeasures/data/tests/test_scheduling_simultaneous.py

@@ -10,7 +10,7 @@
 def test_create_simultaneous_jobs(
     db, app, flex_description_sequential, smart_building, use_heterogeneous_resolutions
 ):
-    assets, sensors = smart_building
+    assets, sensors, _ = smart_building
     queue = app.queues["scheduling"]
     start = pd.Timestamp("2015-01-03").tz_localize("Europe/Amsterdam")
     end = pd.Timestamp("2015-01-04").tz_localize("Europe/Amsterdam")