Add unit query parameter to GET schedule endpoint for unit conversion

documentation/api/change_log.rst

@@ -5,6 +5,11 @@ API change log
 
 .. note:: The FlexMeasures API follows its own versioning scheme. This is also reflected in the URL (e.g. `/api/v3_0`), allowing developers to upgrade at their own pace.
 
+v3.0-30 | 2026-XX-XX
+""""""""""""""""""""
+- Added ``unit`` field to the `/sensors/<id>/schedules/<uuid>` (GET) endpoint for fetching a schedule, to get the schedule in a different unit still compatible to the sensor unit.
+
+
 v3.0-29 | 2026-02-28
 """"""""""""""""""""
 

documentation/changelog.rst

@@ -2,6 +2,7 @@
 **********************
 FlexMeasures Changelog
 **********************
+* Support fetching a schedule in a different unit still compatible to the sensor unit [see `PR #1993 <https://www.github.com/FlexMeasures/flexmeasures/pull/1993>`_]
 
 
 v0.31.0 | February 28, 2026
@@ -50,7 +51,6 @@ New features
 * Add a documentation section on the concept of ``Commitments`` [see `PR #1849 <https://www.github.com/FlexMeasures/flexmeasures/pull/1849>`_]
 * Add resolution column to sensors list on asset context page [see `PR #1986 <https://www.github.com/FlexMeasures/flexmeasures/pull/1986>`_]
 
-
 Infrastructure / Support
 ----------------------
 * Upgraded dependencies [see `PR #1847 <https://www.github.com/FlexMeasures/flexmeasures/pull/1847>`_]

flexmeasures/api/common/schemas/sensors.py

@@ -1,7 +1,7 @@
 from typing import Any
 
 from flask import abort
-from marshmallow import Schema, fields, ValidationError
+from marshmallow import Schema, fields
 from sqlalchemy import select
 
 from flexmeasures.data import db
@@ -11,7 +11,6 @@
     EntityAddressException,
 )
 from flexmeasures.data.models.time_series import Sensor
-from flexmeasures.utils.unit_utils import is_valid_unit
 
 
 class EntityAddressValidationError(FMValidationError):
@@ -90,15 +89,3 @@ def _serialize(self, value: Sensor, attr, obj, **kwargs):
             return value.entity_address_fm0
         else:
             return value.entity_address
-
-
-class UnitField(fields.Str):
-    """Field that represents a unit."""
-
-    def _deserialize(self, value, attr, data, **kwargs) -> str:
-        if not is_valid_unit(value):
-            raise ValidationError(f"Invalid unit: {value}")
-        return value
-
-    def _serialize(self, value: str, attr, obj, **kwargs) -> str:
-        return value

flexmeasures/api/common/utils/args_parsing.py

@@ -17,9 +17,15 @@
 @parser.error_handler
 def handle_error(error, req, schema, *, error_status_code, error_headers):
     """Replacing webargs's error parser, so we can throw custom Exceptions."""
-    if error.__class__ == ValidationError:
-        # re-package all marshmallow's validation errors as our own kind (see below)
-        raise FMValidationError(message=error.messages)
+    if isinstance(error, ValidationError):
+        messages = error.messages
+
+        # Flatten custom location wrapper
+        if isinstance(messages, dict) and "args_and_json" in messages:
+            messages = messages["args_and_json"]
+
+        raise FMValidationError(message=messages)
+
     raise error
 
 

flexmeasures/api/common/utils/validators.py

@@ -1,30 +1,21 @@
 from __future__ import annotations
 
 from datetime import datetime, timedelta
-from functools import wraps
 import re
 
 import isodate
 from isodate.isoerror import ISO8601Error
-from flask import request, current_app
-from flask_json import as_json
-import marshmallow
 
-from webargs.flaskparser import parser
-
-from flexmeasures.data.schemas.times import DurationField
 from flexmeasures.api.common.responses import (  # noqa: F401
     required_info_missing,
     invalid_horizon,
     invalid_method,
     invalid_message_type,
-    invalid_period,
     unapplicable_resolution,
     invalid_resolution_str,
     conflicting_resolutions,
     invalid_source,
     invalid_timezone,
-    invalid_unit,
     no_message_type,
     ptus_incomplete,
     unrecognized_connection_group,
@@ -94,47 +85,3 @@ def parse_duration(
         return duration
     except (ISO8601Error, AttributeError):
         return None
-
-
-def optional_duration_accepted(default_duration: timedelta):
-    """Decorator which specifies that a GET or POST request accepts an optional duration.
-    It parses relevant form data and sets the "duration" keyword param.
-
-    Example:
-
-        @app.route('/getDeviceMessage')
-        @optional_duration_accepted(timedelta(hours=6))
-        def get_device_message(duration):
-            return 'Here is your message'
-
-    The message may specify a duration to overwrite the default duration of 6 hours.
-    """
-
-    def wrapper(fn):
-        @wraps(fn)
-        @as_json
-        def decorated_service(*args, **kwargs):
-            duration_arg = parser.parse(
-                {"duration": DurationField()},
-                request,
-                location="args_and_json",
-                unknown=marshmallow.EXCLUDE,
-            )
-            if "duration" in duration_arg:
-                duration = duration_arg["duration"]
-                duration = DurationField.ground_from(
-                    duration,
-                    kwargs.get("start", kwargs.get("datetime", None)),
-                )
-                if not duration:  # TODO: deprecate
-                    extra_info = "Cannot parse 'duration' value."
-                    current_app.logger.warning(extra_info)
-                    return invalid_period(extra_info)
-                kwargs["duration"] = duration
-            else:
-                kwargs["duration"] = default_duration
-            return fn(*args, **kwargs)
-
-        return decorated_service
-
-    return wrapper

flexmeasures/api/v3_0/sensors.py

@@ -28,9 +28,6 @@
     fallback_schedule_redirect,
 )
 from flexmeasures.api.common.schemas.utils import make_openapi_compatible
-from flexmeasures.api.common.utils.validators import (
-    optional_duration_accepted,
-)
 from flexmeasures.api.common.schemas.sensor_data import (  # noqa F401
     SensorDataDescriptionSchema,
     GetSensorDataSchema,
@@ -62,14 +59,16 @@
 )
 from flexmeasures.data.schemas import AssetIdField
 from flexmeasures.api.common.schemas.search import SearchFilterField
-from flexmeasures.api.common.schemas.sensors import UnitField
+from flexmeasures.data.schemas.scheduling import GetScheduleSchema
+from flexmeasures.data.schemas.units import UnitField
 from flexmeasures.data.services.sensors import get_sensor_stats
 from flexmeasures.data.services.scheduling import (
     create_scheduling_job,
     get_data_source_for_job,
 )
 from flexmeasures.utils.time_utils import duration_isoformat
 from flexmeasures.utils.flexmeasures_inflection import join_words_into_a_list
+from flexmeasures.utils.unit_utils import convert_units
 from flexmeasures.data.models.forecasting import Forecaster
 from flexmeasures.data.services.data_sources import get_data_generator
 from flexmeasures.data.schemas.forecasting.pipeline import (
@@ -856,19 +855,15 @@ def trigger_schedule(
         return dict(**response, **d), s
 
     @route("/<id>/schedules/<uuid>", methods=["GET"])
-    @use_kwargs(
-        {
-            "sensor": SensorIdField(data_key="id"),
-            "job_id": fields.Str(data_key="uuid"),
-        },
-        location="path",
-    )
-    @optional_duration_accepted(
-        timedelta(hours=6)
-    )  # todo: make this a Marshmallow field
+    @use_kwargs(GetScheduleSchema(), location="args_and_json")
     @permission_required_for_context("read", ctx_arg_name="sensor")
     def get_schedule(  # noqa: C901
-        self, sensor: Sensor, job_id: str, duration: timedelta, **kwargs
+        self,
+        sensor: Sensor,
+        job_id: str,
+        duration: timedelta,
+        unit: str | None = None,
+        **kwargs,
     ):
         """
         .. :quickref: Schedules; Get schedule for one device
@@ -881,6 +876,7 @@ def get_schedule(  # noqa: C901
             Optional fields:
 
             - "duration" (6 hours by default; can be increased to plan further into the future)
+            - "unit" (by default, the unit of the schedule is the sensor's unit; a compatible unit can be requested)
           security:
             - ApiKeyAuth: []
           parameters:
@@ -912,6 +908,16 @@ def get_schedule(  # noqa: C901
               example: PT24H
               schema:
                 type: string
+            - in: query
+              name: unit
+              required: false
+              description: |
+                Unit of the schedule values to retrieve.
+                If omitted, the unit of the sensor is used.
+                The unit must be convertible from the sensor's unit.
+              example: kW
+              schema:
+                type: string
           responses:
             200:
               description: PROCESSED
@@ -964,7 +970,7 @@ def get_schedule(  # noqa: C901
                         duration: "PT45M"
                         unit: "MW"
             400:
-              description: INVALID_TIMEZONE, INVALID_DOMAIN, INVALID_UNIT, UNKNOWN_SCHEDULE, UNRECOGNIZED_CONNECTION_GROUP
+              description: INVALID_TIMEZONE, INVALID_DOMAIN, UNKNOWN_SCHEDULE, UNRECOGNIZED_CONNECTION_GROUP
             401:
               description: UNAUTHORIZED
             403:
@@ -1063,11 +1069,21 @@ def get_schedule(  # noqa: C901
         consumption_schedule = consumption_schedule[
             start : start + duration - resolution
         ]
+
+        # Convert to the requested unit if needed
+        if unit != sensor.unit:
+            consumption_schedule = convert_units(
+                consumption_schedule,
+                from_unit=sensor.unit,
+                to_unit=unit,
+                event_resolution=resolution,
+            )
+
         response = dict(
             values=consumption_schedule.tolist(),
             start=isodate.datetime_isoformat(start),
             duration=duration_isoformat(duration),
-            unit=sensor.unit,
+            unit=unit,
         )
 
         d, s = request_processed(scheduler_info_msg)

flexmeasures/api/v3_0/tests/test_sensor_schedules.py

@@ -449,3 +449,76 @@ def test_get_schedule_fallback_not_redirect(
         assert schedule["scheduler_info"]["scheduler"] == "StorageFallbackScheduler"
 
         app.config["FLEXMEASURES_FALLBACK_REDIRECT"] = False
+
+
+@pytest.mark.parametrize(
+    "requesting_user", ["test_prosumer_user@seita.nl"], indirect=True
+)
+def test_get_schedule_with_unit(
+    app,
+    client,
+    add_market_prices,
+    add_battery_assets,
+    battery_soc_sensor,
+    add_charging_station_assets,
+    keep_scheduling_queue_empty,
+    requesting_user,
+    db,
+):
+    """Test that the get_schedule endpoint accepts a unit parameter and converts values accordingly."""
+    sensor = add_battery_assets["Test battery"].sensors[0]
+    assert sensor.unit == "MW"
+
+    message = message_for_trigger_schedule()
+
+    # trigger a schedule
+    trigger_schedule_response = client.post(
+        url_for("SensorAPI:trigger_schedule", id=sensor.id),
+        json=message,
+    )
+    assert trigger_schedule_response.status_code == 200, trigger_schedule_response.json
+    job_id = trigger_schedule_response.json["schedule"]
+
+    # process the scheduling queue
+    work_on_rq(app.queues["scheduling"], exc_handler=handle_scheduling_exception)
+    job = Job.fetch(job_id, connection=app.queues["scheduling"].connection)
+    assert job.is_finished
+
+    # retrieve schedule in the sensor's unit (MW)
+    get_schedule_mw = client.get(
+        url_for("SensorAPI:get_schedule", id=sensor.id, uuid=job_id),
+    )
+    assert get_schedule_mw.status_code == 200, get_schedule_mw.json
+    assert get_schedule_mw.json["unit"] == "MW"
+    mw_values = get_schedule_mw.json["values"]
+
+    # retrieve schedule in a compatible unit (kW) - values should be 1000x
+    get_schedule_kw = client.get(
+        url_for("SensorAPI:get_schedule", id=sensor.id, uuid=job_id),
+        query_string={"unit": "kW"},
+    )
+    assert get_schedule_kw.status_code == 200, get_schedule_kw.json
+    assert get_schedule_kw.json["unit"] == "kW"
+    kw_values = get_schedule_kw.json["values"]
+    assert len(kw_values) == len(mw_values)
+    for mw_val, kw_val in zip(mw_values, kw_values):
+        assert abs(kw_val - mw_val * 1000) < 1e-6
+
+    # retrieve schedule in an invalid unit (bananas) - should return 422
+    get_schedule_incompatible = client.get(
+        url_for("SensorAPI:get_schedule", id=sensor.id, uuid=job_id),
+        query_string={"unit": "bananas"},
+    )
+    assert get_schedule_incompatible.status_code == 422, get_schedule_incompatible.json
+    assert "Invalid unit: bananas" in get_schedule_incompatible.json["message"]["unit"]
+
+    # retrieve schedule in an incompatible unit (°C) - should return 422
+    get_schedule_incompatible = client.get(
+        url_for("SensorAPI:get_schedule", id=sensor.id, uuid=job_id),
+        query_string={"unit": "°C"},
+    )
+    assert get_schedule_incompatible.status_code == 422, get_schedule_incompatible.json
+    assert (
+        "Incompatible units: MW cannot be converted to °C."
+        in get_schedule_incompatible.json["message"]["unit"]
+    )