Add unit field for sensor data upload form

documentation/changelog.rst

@@ -67,6 +67,7 @@ New features
 * More explicitly represent the unit of sensors that record dimensionless data [see `PR #1802 <https://github.com/FlexMeasures/flexmeasures/pull/1802>`_]
 * Allow modifying asset trees with the CLI using ``flexmeasures edit transfer-parenthood`` [see `PR #1773 <https://github.com/FlexMeasures/flexmeasures/pull/1773>`_]
 * Expanded the sorting columns of "latest jobs" table [see `PR #1821 <https://github.com/FlexMeasures/flexmeasures/pull/1821>`_]
+* Let users specify the unit, which uploaded data is based on [see `PR #1836 <https://github.com/FlexMeasures/flexmeasures/pull/1836>`_]
 * Allow to keep legend combined below graphs, even with many plots (useful on narrow screens) [see `PR #1816 <https://github.com/FlexMeasures/flexmeasures/pull/1816>`_]
 
 Infrastructure / Support

flexmeasures/api/common/utils/args_parsing.py

@@ -97,6 +97,9 @@ def combined_sensor_data_upload(request: Request, schema):
     data.update(request.files)
     belief_time = request.form.get("belief-time-measured-instantly")
     data.update({"belief-time-measured-instantly": belief_time})
+    unit = request.form.get("unit")
+    if unit is not None:
+        data.update({"unit": unit})
     return MultiDictProxy(data, schema)
 
 

flexmeasures/api/v3_0/sensors.py

@@ -49,7 +49,6 @@
     SensorSchema,
     SensorIdField,
     SensorDataFileSchema,
-    SensorDataFileDescriptionSchema,
 )
 from flexmeasures.data.schemas.times import AwareDateTimeField, PlanningDurationField
 from flexmeasures.data.schemas import AssetIdField
@@ -366,7 +365,11 @@ def index(
         pass_ctx_to_loader=True,
     )
     def upload_data(
-        self, data: list[tb.BeliefsDataFrame], filenames: list[str], **kwargs
+        self,
+        data: list[tb.BeliefsDataFrame],
+        filenames: list[str],
+        unit: str | None = None,
+        **kwargs,
     ):
         """
         .. :quickref: Data; Upload sensor data by file
@@ -458,6 +461,7 @@ def upload_data(
             - Sensors
         """
         sensor = data[0].sensor
+
         AssetAuditLog.add_record(
             sensor.generic_asset,
             f"Data from {join_words_into_a_list(filenames)} uploaded to sensor '{sensor.name}': {sensor.id}",

flexmeasures/api/v3_0/tests/test_assets_api_fresh_db.py

@@ -1,10 +1,16 @@
+import io
+
 from flask import url_for
 import pytest
 from sqlalchemy import select
+from datetime import timedelta
+
+from timely_beliefs import BeliefsDataFrame
 
 from flexmeasures.api.tests.utils import AccountContext
 from flexmeasures.data.models.generic_assets import GenericAsset
-from flexmeasures.api.v3_0.tests.utils import get_asset_post_data
+from flexmeasures.data.models.time_series import TimedBelief
+from flexmeasures.api.v3_0.tests.utils import get_asset_post_data, generate_csv_content
 
 
 @pytest.mark.parametrize(
@@ -73,3 +79,227 @@ def test_delete_an_asset(client, setup_api_fresh_test_data, requesting_user, db)
         select(GenericAsset).filter_by(id=existing_asset_id)
     ).scalar_one_or_none()
     assert deleted_asset is None
+
+
+@pytest.mark.parametrize(
+    "requesting_user, sensor_index, data_unit, data_resolution, data_values, expected_event_values, expected_status",
+    [
+        (
+            "test_prosumer_user_2@seita.nl",
+            1,  # this sensor has unit=kW, res=00:15
+            "m/s",
+            timedelta(hours=1),  # Upsampling
+            [45.3, 45.3],
+            "Provided unit 'm/s' is not convertible to sensor unit 'kW'",
+            422,  # units not convertible
+        ),
+        (
+            "test_prosumer_user_2@seita.nl",
+            2,  # this sensor has unit=kWh, res=01:00
+            "kWh",  # No Conversion needed - kWh to kWh
+            timedelta(hours=1),  # No resampling
+            [45.3] * 4,
+            [45.3] * 4,  # same unit and resolution - values stay the same
+            200,
+        ),
+        (
+            "test_prosumer_user_2@seita.nl",
+            0,  # this sensor has unit=MW, res=00:15
+            "kWh",  # Conversion needed - kWh to MW
+            timedelta(hours=1),  # Upsampling
+            [45.3] * 4,
+            [45.3 / 1000.0]
+            * 4
+            * 4,  # values: / 1000 due to kW(h)->MW, number *4 due to h->15min
+            200,
+        ),
+        (
+            "test_prosumer_user_2@seita.nl",
+            1,  # this sensor has unit=kW, res=00:15
+            "MW",  # Conversion needed - MW to kW
+            timedelta(hours=1),  # Upsampling
+            [2] * 6,
+            [2 * 1000]
+            * 6
+            * 4,  # both power units, so 2 MW = 2000 kW, number *4 due to h->15min
+            200,
+        ),
+        (
+            "test_prosumer_user_2@seita.nl",
+            1,  # this sensor has unit=kW, res=00:15
+            "kWh",  # Conversion needed - kWh to kW
+            timedelta(minutes=30),  # Upsampling
+            [10] * 12,
+            [10 * 2]
+            * 12
+            * 2,  # 10 kWh per half hour = 20 kW power, number *2 due to 30min->15min
+            200,
+        ),
+        (
+            "test_prosumer_user_2@seita.nl",
+            2,  # this sensor has unit=kWh, res=01:00
+            "kWh",  # No Conversion needed - kWh to kWh
+            timedelta(minutes=30),  # Downsampling
+            [10, 20, 20, 40],
+            [
+                15,
+                30,
+            ],  # we make (10/2 + 20/2) the first hour, and (20/2 + 40/2) the second hour
+            200,
+        ),
+        (
+            "test_prosumer_user_2@seita.nl",
+            2,  # this sensor has unit=kWh, res=01:00
+            "kW",  # Conversion needed - kW to kWh
+            timedelta(minutes=30),  # Downsampling
+            [20, 40, 40, 80],
+            "Provided unit 'kW' is not convertible to sensor unit 'kWh'",
+            422,  # we don't support this case yet
+        ),
+        (
+            "test_prosumer_user_2@seita.nl",
+            1,  # this sensor has unit=kW, res=00:15
+            "kWh",  # Conversion needed - kWh to kW
+            timedelta(minutes=7, seconds=30),  # Downsampling
+            [20, 40, 40, 80],
+            [
+                240,
+                480,
+            ],
+            200,
+        ),
+        (
+            "test_prosumer_user_2@seita.nl",
+            1,  # this sensor has unit=kW, res=00:15
+            "MW",  # Conversion needed - MW to kW
+            timedelta(minutes=7, seconds=30),  # Downsampling
+            [20, 40, 40, 80, 30, 60],
+            [
+                30000,
+                60000,
+                45000,
+            ],
+            200,
+        ),
+        (
+            "test_prosumer_user_2@seita.nl",
+            3,  # this sensor has unit=kWh, res=00:00
+            "MWh",  # Conversion needed - MWh to kWh
+            timedelta(minutes=7, seconds=30),  # No resampling
+            [10, 20, 40, 80],
+            [10000, 20000, 40000, 80000],
+            200,
+        ),
+        (
+            "test_prosumer_user_2@seita.nl",
+            3,  # this sensor has unit=kWh, res=00:00
+            "kW",  # Conversion needed - kW to kWh
+            timedelta(minutes=7, seconds=30),  # No resampling
+            [20, 40, 40, 80],
+            "Provided unit 'kW' is not convertible to sensor unit 'kWh'",
+            422,
+        ),
+        (
+            "test_prosumer_user_2@seita.nl",
+            4,  # this sensor has unit=EUR/kWh, res=01:00
+            "EUR/MWh",  # Conversion needed - EUR/MWh to EUR/kWh
+            timedelta(minutes=30),  # Downsampling
+            [200, 300, 400, 500],
+            [0.25, 0.45],
+            200,
+        ),
+        (
+            "test_prosumer_user_2@seita.nl",
+            4,  # this sensor has unit=EUR/kWh, res=01:00
+            "EUR/kWh",  # Conversion needed - EUR/kWh to EUR/kWh
+            timedelta(hours=2),  # Upsampling
+            [200, 300, 400],
+            [200, 200, 300, 300, 400, 400],
+            200,
+        ),
+        (
+            "test_prosumer_user_2@seita.nl",
+            5,  # this sensor has unit=EUR, res=01:00
+            "kEUR",  # Conversion needed - kEUR to EUR
+            timedelta(minutes=30),  # Downsampling
+            [2, 3, 4, 2],
+            [2500, 3000],
+            200,
+        ),
+    ],
+    indirect=["requesting_user"],
+)
+def test_upload_sensor_data_with_distinct_to_from_units_and_target_resolutions(
+    fresh_db,
+    client,
+    add_battery_assets_fresh_db,
+    requesting_user,
+    sensor_index,
+    data_unit,
+    data_resolution,
+    data_values,
+    expected_event_values,
+    expected_status,
+):
+    """
+    Check if unit validation works fine for sensor data upload.
+    The target sensors can have different units and resolution,
+    and the incoming data can also have differing resolutions and declared unit.
+    This test needs to check if the resulting data matches expectations.
+    """
+
+    start_date = (
+        "2025-01-01T10:00:00+00:00"  # This date would be used to generate CSV content
+    )
+    test_battery = add_battery_assets_fresh_db["Test battery"]
+    sensor = test_battery.sensors[sensor_index]
+    num_test_intervals = len(data_values)
+    print(
+        f"Uploading data to sensor '{sensor.name}' with unit={sensor.unit} and resolution={sensor.event_resolution}."
+    )
+    print(f"Data unit is {data_unit} and resolution is {data_resolution}")
+
+    csv_content = generate_csv_content(
+        start_time_str=start_date,
+        interval=data_resolution,
+        values=data_values,
+    )
+    print("Generated CSV content:")
+    print(csv_content)
+    file_obj = io.BytesIO(csv_content.encode("utf-8"))
+
+    response = client.post(
+        url_for("SensorAPI:upload_data", id=sensor.id),
+        data={"uploaded-files": (file_obj, "data.csv"), "unit": data_unit},
+        content_type="multipart/form-data",
+    )
+    print("Response:\n%s" % response.status_code, expected_status)
+    print("Server responded with:\n%s" % response.json)
+    assert response.status_code == expected_status
+
+    # fetch the save timedBeliefs and check if they have the right values
+    if response.status_code == 200:
+        timed_beliefs = fresh_db.session.execute(
+            select(TimedBelief)
+            .filter(TimedBelief.sensor_id == sensor.id)
+            .order_by(TimedBelief.event_start)
+        ).scalars()
+
+        beliefs = timed_beliefs.all()
+        bdf = BeliefsDataFrame(beliefs)
+        print("Stored beliefs: ==============================")
+        print(bdf)
+
+        expected_num_beliefs = num_test_intervals
+        if sensor.event_resolution != timedelta(0):
+            expected_num_beliefs *= data_resolution / sensor.event_resolution
+        assert (
+            len(beliefs) == expected_num_beliefs
+        ), f"Fetched {len(beliefs)} beliefs from the database, expecting {expected_num_beliefs}."
+
+        assert [b.event_value for b in beliefs] == expected_event_values
+    elif response.status_code == 422:
+        assert (
+            expected_event_values
+            in response.json["message"]["combined_sensor_data_upload"]["_schema"]
+        )

flexmeasures/api/v3_0/tests/test_sensors_api.py

@@ -44,7 +44,7 @@
             None,
             None,
             "power",
-            2,
+            4,
             False,
             False,
             200,
@@ -205,7 +205,10 @@ def test_fetch_sensors(
             assert isinstance(response.json, list)
             assert is_valid_unit(response.json[0]["unit"])
             assert response.json[0]["name"] == exp_sensor_name
-            assert len(response.json) == exp_num_results
+            assert len(response.json) == exp_num_results, (
+                f"If this line fails, a conftest may have added another sensor "
+                f"accessible to {requesting_user}. Update the exp_num_results in the test parameters accordingly."
+            )
 
             if asset_id_of_of_first_sensor_result is not None:
                 assert (
@@ -326,6 +329,7 @@ def test_upload_csv_file(client, db, setup_api_test_data, sensor_name, requestin
         content_type="multipart/form-data",
         headers={"Authorization": auth_token},
     )
+    print("Server responded with:\n%s" % response.json)
     assert response.status_code == 200 or response.status_code == 400
 
     check_audit_log_event(
@@ -359,6 +363,7 @@ def test_upload_excel_file(client, requesting_user):
         content_type="multipart/form-data",
         headers={"Authorization": auth_token},
     )
+    print("Server responded with:\n%s" % response.json)
     assert response.status_code == 200 or response.status_code == 400
 
 

flexmeasures/api/v3_0/tests/utils.py

@@ -1,3 +1,5 @@
+from datetime import timedelta, datetime
+
 from sqlalchemy import select
 
 from flexmeasures import Asset, User
@@ -126,3 +128,62 @@ def check_audit_log_event(
         )
     ).scalar()
     assert logs, f"expected audit log event: {event}"
+
+
+def parse_resolution(resolution_str):
+    """
+    Parses a resolution string (e.g., '10m', '30min', '1h') into a timedelta object.
+    """
+    import re
+
+    # Regular expression to capture the number and the unit (m/min/h)
+    match = re.match(r"(\d+)\s*(m|min|h)", resolution_str, re.I)
+    if not match:
+        raise ValueError(
+            f"Invalid resolution format: {resolution_str}. Use formats like '10m', '30min', '1h'."
+        )
+
+    value = int(match.group(1))
+    unit = match.group(2).lower()
+
+    if unit in ("m", "min"):
+        return timedelta(minutes=value)
+    elif unit == "h":
+        return timedelta(hours=value)
+    else:
+        # This would probably not be reached due to the regex, but just in case
+        raise ValueError(f"Unsupported time unit: {unit}")
+
+
+def generate_csv_content(
+    start_time_str: str, interval: timedelta, values: list[float]
+) -> str:
+    """
+    Generates a CSV-formatted string with a specified time resolution.
+
+    Args:
+        start_time_str (str): The starting timestamp (e.g., '2021-01-01T00:10:00+00:00').
+        resolution_str (str): The interval length (e.g., '10m', '30min', '1h').
+        values (list of floats): The values to use.
+
+    Returns:
+        str: The generated CSV content.
+    """
+    # Convert the starting time string to a datetime object
+    current_time = datetime.fromisoformat(start_time_str)
+
+    # Build the CSV content
+    csv_rows = ["Hour,price"]  # Header row
+
+    for value in values:
+        # Format the timestamp back into the required string format
+        timestamp_str = current_time.strftime("%Y-%m-%dT%H:%M:%S%z")
+
+        # Add new row to CSV content
+        csv_rows.append(f"{timestamp_str},{value}")
+
+        # Increment the time for the next interval
+        current_time += interval
+
+    # Join all rows
+    return "\n".join(csv_rows)