feat: allow setting a power sensor in the flex-model of an asset when patching

documentation/changelog.rst

@@ -23,6 +23,7 @@ New features
 * Improve UX after deleting a child asset through the UI [see `PR #2119 <https://www.github.com/FlexMeasures/flexmeasures/pull/2119>`_]
 * Improve source filtering in the sensor data GET endpoint by exposing the documented query parameters in Swagger and allowing filtering by the account linked to data sources [see `PR #2083 <https://www.github.com/FlexMeasures/flexmeasures/pull/2083>`_ and `PR #2151 <https://www.github.com/FlexMeasures/flexmeasures/pull/2151>`_]
 * Support sensor references for efficiency fields in storage flex-models [see `PR #2142 <https://www.github.com/FlexMeasures/flexmeasures/pull/2142>`_]
+* The ``consumption`` and ``production`` flex-model fields for the ``StorageScheduler`` now act as output sensors: the scheduler writes the resulting power schedule to those sensors, with unit conversion and resampling applied. When only one of the two is defined, the full power profile is written (sign-inverted for the production sensor). When both are defined, the schedule is split into its non-negative (consumption) and non-positive (production) parts [see `PR #XXXX <https://www.github.com/FlexMeasures/flexmeasures/pull/XXXX>`_]
 * Added a unified job status endpoint ``GET /api/v3_0/jobs/<uuid>`` to retrieve the current execution status and result message for any background job [see `PR #2141 <https://www.github.com/FlexMeasures/flexmeasures/pull/2141>`_]
 * New ``GET /api/v3_0/sources`` endpoint to list accessible data sources and defined types, with ``only_latest=true`` by default to return only the most recent version per source [see `PR #2126 <https://www.github.com/FlexMeasures/flexmeasures/pull/2126>`_]
 * Add support for filtering sensor data GET requests by ``source-type`` on ``/api/v3_0/sensors/<id>/data`` [see `PR #2127 <https://www.github.com/FlexMeasures/flexmeasures/pull/2127>`_]

documentation/features/scheduling.rst

@@ -182,7 +182,13 @@ For more details on the possible formats for field values, see :ref:`variable_qu
 
    * - Field
      - Example value
-     - Description 
+     - Description
+   * - ``consumption``
+     - |CONSUMPTION.example|
+     - .. include:: ../_autodoc/CONSUMPTION.rst
+   * - ``production``
+     - |PRODUCTION.example|
+     - .. include:: ../_autodoc/PRODUCTION.rst
    * - ``state-of-charge``
      - |STATE_OF_CHARGE.example|
      - .. include:: ../_autodoc/STATE_OF_CHARGE.rst

flexmeasures/api/v3_0/tests/test_sensor_schedules_fresh_db.py

@@ -235,14 +235,26 @@ def test_trigger_schedule_uses_state_of_charge_sensor_for_soc_at_start(
             event_value=50,
         )
     )
-    fresh_db.session.commit()
 
+    # Also add a production output sensor to verify the production schedule is stored.
+    # (Using only a production sensor means the full power profile is stored with sign inverted:
+    # production positive, consumption negative.)
     sensor = (
         Sensor.query.filter(Sensor.name == "power")
         .join(GenericAsset, GenericAsset.id == Sensor.generic_asset_id)
         .filter(GenericAsset.name == "Test battery")
         .one_or_none()
     )
+    production_output_sensor = Sensor(
+        name="production output",
+        generic_asset=sensor.generic_asset,
+        unit="MW",
+        event_resolution=sensor.event_resolution,
+    )
+    fresh_db.session.add(production_output_sensor)
+    fresh_db.session.commit()
+
+    message["flex-model"]["production"] = {"sensor": production_output_sensor.id}
 
     with app.test_client() as client:
         trigger_schedule_response = client.post(
@@ -268,6 +280,15 @@ def test_trigger_schedule_uses_state_of_charge_sensor_for_soc_at_start(
     sensor = fresh_db.session.get(Sensor, sensor.id)
     assert sensor.generic_asset.get_attribute("soc_in_mwh") == pytest.approx(0.02)
 
+    # Verify the production output sensor received schedule data.
+    # Only the production sensor is defined (no consumption sensor), so the full power profile
+    # is stored with inverted sign: production as positive, consumption as negative.
+    production_output_sensor = fresh_db.session.get(Sensor, production_output_sensor.id)
+    production_beliefs = production_output_sensor.search_beliefs(
+        event_starts_after=parse_datetime(message["start"])
+    )
+    assert len(production_beliefs) == 96  # 24h at 15-min resolution
+
 
 @pytest.mark.parametrize(
     "context_sensor, asset_sensor, parent_sensor, expect_sensor",

flexmeasures/cli/tests/test_data_add_fresh_db.py

@@ -488,6 +488,17 @@ def test_add_storage_schedule_uses_state_of_charge_sensor_for_soc_at_start(
             belief_time=datetime.fromisoformat(start),
         )
     )
+
+    # Add a consumption output sensor to verify the full power profile is stored on it
+    # (only the consumption sensor is defined, so the sign convention is consumption positive,
+    # production negative).
+    consumption_output_sensor = Sensor(
+        name="consumption output",
+        generic_asset=charging_station,
+        unit="MW",
+        event_resolution=power_sensor.event_resolution,
+    )
+    fresh_db.session.add(consumption_output_sensor)
     fresh_db.session.commit()
 
     cli_input_params = {
@@ -504,6 +515,7 @@ def test_add_storage_schedule_uses_state_of_charge_sensor_for_soc_at_start(
                 "soc-min": "0 MWh",
                 "soc-max": "5 MWh",
                 "power-capacity": "2 MW",
+                "consumption": {"sensor": consumption_output_sensor.id},
             }
         ),
     }
@@ -513,3 +525,11 @@ def test_add_storage_schedule_uses_state_of_charge_sensor_for_soc_at_start(
     check_command_ran_without_error(result)
     assert len(power_sensor.search_beliefs()) == 48
     assert power_sensor.generic_asset.get_attribute("soc_in_mwh") == 2.5
+
+    # Verify the consumption output sensor received the full power schedule.
+    # A charging station is consumption-only (non-negative), so the full schedule
+    # is non-negative and equals what is stored on the power sensor.
+    consumption_output_sensor = fresh_db.session.get(
+        Sensor, consumption_output_sensor.id
+    )
+    assert len(consumption_output_sensor.search_beliefs()) == 48

flexmeasures/data/models/planning/storage.py

@@ -1567,6 +1567,83 @@ def _build_soc_schedule(
             )
         return soc_schedule
 
+    @staticmethod
+    def _build_consumption_production_schedules(
+        flex_model: list[dict],
+        ems_schedule: pd.DataFrame,
+    ) -> dict:
+        """Build consumption and/or production power schedules for devices that define output sensors.
+
+        Each device's flex model may define a ``consumption`` sensor, a ``production`` sensor, or both.
+        The schedule stored on each sensor depends on which sensors are defined:
+
+        - **Only** ``consumption`` **sensor defined**: the full power schedule is written to that
+          sensor using the standard FlexMeasures sign convention (consumption positive, production
+          negative).
+        - **Only** ``production`` **sensor defined**: the full power schedule is written to that
+          sensor with the sign inverted (production positive, consumption negative).
+        - **Both** ``consumption`` **and** ``production`` **sensors defined**: only the non-negative
+          part of the schedule is written to the consumption sensor, and only the non-positive part
+          (sign-flipped to positive values) is written to the production sensor.
+
+        Because the sign convention is encoded in the sensor key name (``consumption`` vs.
+        ``production``), these sensors do not need a ``consumption_is_positive`` attribute.
+
+        Unit conversion from MW to each sensor's unit is applied.
+
+        :param flex_model:    List of per-device flex models (after deserialization).
+        :param ems_schedule:  DataFrame of per-device power schedules in MW (consumption positive).
+        :returns:             Dict mapping each output sensor to its power schedule.
+        """
+        schedules: dict = {}
+        for d, flex_model_d in enumerate(flex_model):
+            consumption_field = flex_model_d.get("consumption")
+            production_field = flex_model_d.get("production")
+            consumption_sensor = (
+                consumption_field["sensor"]
+                if isinstance(consumption_field, dict) and "sensor" in consumption_field
+                else None
+            )
+            production_sensor = (
+                production_field["sensor"]
+                if isinstance(production_field, dict) and "sensor" in production_field
+                else None
+            )
+            if consumption_sensor is None and production_sensor is None:
+                continue
+            power_series = ems_schedule[d]  # in MW; consumption is positive
+            if consumption_sensor is not None and production_sensor is None:
+                # Full power profile on the consumption sensor (consumption positive, production negative).
+                schedules[consumption_sensor] = convert_units(
+                    power_series,
+                    "MW",
+                    consumption_sensor.unit,
+                    event_resolution=consumption_sensor.event_resolution,
+                )
+            elif production_sensor is not None and consumption_sensor is None:
+                # Full power profile on the production sensor (production positive, consumption negative).
+                schedules[production_sensor] = convert_units(
+                    -power_series,
+                    "MW",
+                    production_sensor.unit,
+                    event_resolution=production_sensor.event_resolution,
+                )
+            else:
+                # Both sensors defined: split into non-negative (consumption) and non-positive (production) parts.
+                schedules[consumption_sensor] = convert_units(
+                    power_series.clip(lower=0),
+                    "MW",
+                    consumption_sensor.unit,
+                    event_resolution=consumption_sensor.event_resolution,
+                )
+                schedules[production_sensor] = convert_units(
+                    (-power_series).clip(lower=0),
+                    "MW",
+                    production_sensor.unit,
+                    event_resolution=production_sensor.event_resolution,
+                )
+        return schedules
+
     def compute(self, skip_validation: bool = False) -> SchedulerOutputType:
         """Schedule a battery or Charge Point based directly on the latest beliefs regarding market prices within the specified time window.
         For the resulting consumption schedule, consumption is defined as positive values.
@@ -1640,6 +1717,10 @@ def compute(self, skip_validation: bool = False) -> SchedulerOutputType:
             flex_model, ems_schedule, soc_at_start, device_constraints, resolution
         )
 
+        consumption_production_schedule = self._build_consumption_production_schedules(
+            flex_model, ems_schedule
+        )
+
         # Resample each device schedule to the resolution of the device's power sensor
         if self.resolution is None:
             storage_schedule = {
@@ -1649,6 +1730,12 @@ def compute(self, skip_validation: bool = False) -> SchedulerOutputType:
                 for sensor in storage_schedule.keys()
                 if sensor is not None
             }
+            consumption_production_schedule = {
+                sensor: consumption_production_schedule[sensor]
+                .resample(sensor.event_resolution)
+                .mean()
+                for sensor in consumption_production_schedule.keys()
+            }
 
         # Round schedule
         if self.round_to_decimals:
@@ -1661,6 +1748,12 @@ def compute(self, skip_validation: bool = False) -> SchedulerOutputType:
                 sensor: soc_schedule[sensor].round(self.round_to_decimals)
                 for sensor in soc_schedule.keys()
             }
+            consumption_production_schedule = {
+                sensor: consumption_production_schedule[sensor].round(
+                    self.round_to_decimals
+                )
+                for sensor in consumption_production_schedule.keys()
+            }
 
         if self.return_multiple:
             storage_schedules = [
@@ -1694,7 +1787,32 @@ def compute(self, skip_validation: bool = False) -> SchedulerOutputType:
                 }
                 for sensor, soc in soc_schedule.items()
             ]
-            return storage_schedules + commitment_costs + soc_schedules
+            # Determine which sensors are consumption vs. production output sensors
+            consumption_output_sensors = {
+                flex_model_d["consumption"]["sensor"]
+                for flex_model_d in flex_model
+                if isinstance(flex_model_d.get("consumption"), dict)
+                and "sensor" in flex_model_d["consumption"]
+            }
+            consumption_production_schedules = [
+                {
+                    "name": (
+                        "consumption_schedule"
+                        if sensor in consumption_output_sensors
+                        else "production_schedule"
+                    ),
+                    "data": data,
+                    "sensor": sensor,
+                    "unit": sensor.unit,
+                }
+                for sensor, data in consumption_production_schedule.items()
+            ]
+            return (
+                storage_schedules
+                + commitment_costs
+                + soc_schedules
+                + consumption_production_schedules
+            )
         else:
             return storage_schedule[sensors[0]]
 

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

@@ -29,6 +29,24 @@ def test_battery_solver_multi_commitment(add_battery_assets, db):
     index = initialize_index(start=start, end=end, resolution=resolution)
     production_prices = pd.Series(90, index=index)
     consumption_prices = pd.Series(100, index=index)
+
+    # Add consumption and production output sensors to the battery asset
+    consumption_output_sensor = Sensor(
+        name="consumption output",
+        generic_asset=battery.generic_asset,
+        unit="kW",
+        event_resolution=resolution,
+    )
+    production_output_sensor = Sensor(
+        name="production output",
+        generic_asset=battery.generic_asset,
+        unit="kW",
+        event_resolution=resolution,
+    )
+    db.session.add(consumption_output_sensor)
+    db.session.add(production_output_sensor)
+    db.session.flush()
+
     scheduler: Scheduler = StorageScheduler(
         battery,
         start,
@@ -37,6 +55,8 @@ def test_battery_solver_multi_commitment(add_battery_assets, db):
         flex_model={
             "soc-at-start": f"{soc_at_start} MWh",
             "soc-min": "0 MWh",
+            "consumption": {"sensor": consumption_output_sensor.id},
+            "production": {"sensor": production_output_sensor.id},
             "soc-max": "1 MWh",
             "power-capacity": "1 MVA",
             "soc-minima": [
@@ -131,6 +151,24 @@ def test_battery_solver_multi_commitment(add_battery_assets, db):
         costs["a sample commitment penalizing demand/supply"], 1 * (1 - 0.4)
     )
 
+    # Check consumption/production output sensor schedules.
+    # The battery charges at a constant rate (all positive values), so the consumption schedule
+    # should match the power schedule in kW, and the production schedule should be all zeros.
+    consumption_result = next(
+        r for r in results if r.get("name") == "consumption_schedule"
+    )
+    production_result = next(
+        r for r in results if r.get("name") == "production_schedule"
+    )
+    assert consumption_result["sensor"] is consumption_output_sensor
+    assert consumption_result["unit"] == "kW"
+    assert production_result["sensor"] is production_output_sensor
+    assert production_result["unit"] == "kW"
+    # Both sensors have the same resolution as the power sensor, so no resampling occurs.
+    expected_kw = (1 - 0.4) / 24 * 1000  # MW -> kW
+    np.testing.assert_allclose(consumption_result["data"], expected_kw)
+    np.testing.assert_allclose(production_result["data"], 0)
+
 
 def test_battery_relaxation(add_battery_assets, db):
     """Check that resolving SoC breaches is more important than resolving device power breaches.

flexmeasures/data/schemas/scheduling/__init__.py

@@ -582,6 +582,24 @@ def _to_currency_per_mwh(price_unit: str) -> str:
 }
 
 UI_FLEX_MODEL_SCHEMA: Dict[str, Dict[str, Any]] = {
+    "consumption": {
+        "default": None,
+        "description": rst_to_openapi(metadata.CONSUMPTION.description),
+        "types": {
+            "backend": "typeTwo",
+            "ui": "A sensor which records the scheduled consumption.",
+        },
+        "example-units": EXAMPLE_UNIT_TYPES["power"],
+    },
+    "production": {
+        "default": None,
+        "description": rst_to_openapi(metadata.PRODUCTION.description),
+        "types": {
+            "backend": "typeTwo",
+            "ui": "A sensor which records the scheduled production.",
+        },
+        "example-units": EXAMPLE_UNIT_TYPES["power"],
+    },
     "soc-min": {
         "default": None,
         "description": rst_to_openapi(metadata.SOC_MIN.description),

flexmeasures/data/schemas/scheduling/metadata.py

@@ -185,6 +185,32 @@ def to_dict(self):
 # FLEX-MODEL
 
 
+CONSUMPTION = MetaData(
+    description="""Sensor used to record the scheduled power as seen from a consumption perspective.
+
+The sign convention is determined by the key name, so the sensor itself does not need a ``consumption_is_positive`` attribute.
+
+Depending on which output sensors are defined:
+
+- **Only** ``consumption`` **defined**: the full power schedule is stored on this sensor using the
+  standard FlexMeasures sign convention (consumption positive, production negative).
+- **Only** ``production`` **defined**: the full power schedule is stored on the production sensor
+  with the sign inverted (production positive, consumption negative).
+- **Both defined**: only the non-negative part of the schedule is stored on this sensor (zero for
+  time steps with net production), and only the non-positive part (sign-flipped) is stored on the
+  production sensor.
+""",
+    example={"sensor": 14},
+)
+PRODUCTION = MetaData(
+    description="""Sensor used to record the scheduled power as seen from a production perspective.
+
+The sign convention is determined by the key name, so the sensor itself does not need a ``consumption_is_positive`` attribute.
+
+See ``consumption`` for the full description of the split logic when both sensors are defined.
+""",
+    example={"sensor": 15},
+)
 STATE_OF_CHARGE = MetaData(
     description="Sensor used to record the scheduled state of charge. If ``soc-at-start`` is omitted, FlexMeasures will also use this field to infer the starting state of charge. For this use case, the field may also contain a time series specification instead. When a sensor is used, its unit may be an energy unit (e.g. MWh or kWh) or a percentage (%). For sensors with a % unit, the ``soc-max`` flex-model field must be set to a non-zero value to allow converting between the energy-based schedule and a percentage.",
     example={"sensor": 12},