Feat: sensor data source type filter

documentation/changelog.rst

@@ -11,6 +11,7 @@ New features
 * Added API support for copying assets and their subtrees [see `PR #2017 <https://www.github.com/FlexMeasures/flexmeasures/pull/2017>`_]
 * 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>`_]
+* 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>`_]
 
 Infrastructure / Support
 ----------------------

flexmeasures/api/common/schemas/sensor_data.py

@@ -174,6 +174,19 @@ class GetSensorDataFilterSchemaMixin:
             example=3,
         ),
     )
+    source_type = fields.Str(
+        required=False,
+        metadata=dict(
+            description="Filter by a specific data source type.",
+            example="forecaster",
+        ),
+    )
+
+    @validates_schema
+    def check_source_type_not_blank(self, data, **kwargs):
+        source_type = data.get("source_type")
+        if source_type is not None and not source_type.strip():
+            raise ValidationError({"source_type": ["Source type must not be blank."]})
 
 
 class GetSensorDataSchema(GetSensorDataFilterSchemaMixin, SensorDataDescriptionSchema):
@@ -228,6 +241,7 @@ def load_data_and_make_response(sensor_data_description: dict) -> dict:
         resolution = sensor_data_description.get("resolution")
         source = sensor_data_description.get("source")
         account = sensor_data_description.get("account")
+        source_type = sensor_data_description.get("source_type")
 
         # Post-load configuration of event frequency
         if resolution is None:
@@ -258,6 +272,7 @@ def load_data_and_make_response(sensor_data_description: dict) -> dict:
                 horizons_at_most=horizons_at_most,
                 source=source,
                 source_account_ids=account.id if account else None,
+                source_types=[source_type] if source_type else None,
                 beliefs_before=sensor_data_description.get("prior", None),
                 one_deterministic_belief_per_event=True,
                 resolution=resolution,

flexmeasures/api/v3_0/sensors.py

@@ -592,6 +592,7 @@ def get_data(self, id: int, **sensor_data_description: dict):
             - "prior" (the belief timing docs also apply here)
             - "source" (filter by data source ID, read [the docs about sources](https://flexmeasures.readthedocs.io/latest/api/notation.html#sources))
             - "account" (filter by the account ID linked to data sources)
+            - "source_type" (filter by data source type)
 
             An example query to fetch data for sensor with ID=1, for one hour starting June 7th 2021 at midnight, in 15 minute intervals, in m³/h:
 

flexmeasures/api/v3_0/tests/test_sensor_data.py

@@ -84,7 +84,7 @@ def test_get_sensor_data(
         None,
         None,
     ]
-    assert all(a == b for a, b in zip(values, expected))
+    assert values == expected
 
 
 @pytest.mark.parametrize(
@@ -166,6 +166,91 @@ def test_get_sensor_data_filtered_by_source_account(
     assert all(a == b for a, b in zip(values, expected))
 
 
+@pytest.mark.parametrize(
+    "source_type, expected",
+    [
+        (
+            "user",
+            [
+                sum(GAS_MEASUREMENTS_10MIN[0:2]) / 2,  # 91.5
+                GAS_MEASUREMENTS_10MIN[2],  # 92.1
+                None,
+                None,
+            ],
+        ),
+        (
+            "demo script",
+            [
+                GAS_MEASUREMENTS_10MIN[0],  # 91.3
+                GAS_MEASUREMENTS_10MIN[2],  # 92.1
+                None,
+                None,
+            ],
+        ),
+        ("scheduler", [None, None, None, None]),
+    ],
+)
+@pytest.mark.parametrize(
+    "requesting_user", ["test_supplier_user_4@seita.nl"], indirect=True
+)
+def test_get_sensor_data_filtered_by_source_type(
+    client,
+    setup_api_test_data: dict[str, Sensor],
+    requesting_user,
+    source_type,
+    expected,
+):
+    """Check that GET /sensors/<id>/data can filter by source type."""
+    sensor = setup_api_test_data["some gas sensor"]
+    message = {
+        "start": "2021-05-02T00:00:00+02:00",
+        "duration": "PT1H20M",
+        "horizon": "PT0H",
+        "unit": "m³/h",
+        "source_type": source_type,
+        "resolution": "PT20M",
+    }
+    response = client.get(
+        url_for("SensorAPI:get_data", id=sensor.id),
+        query_string=message,
+    )
+    print("Server responded with:\n%s" % response.json)
+    assert response.status_code == 200
+    assert response.json["values"] == expected
+
+
+@pytest.mark.parametrize(
+    "requesting_user", ["test_supplier_user_4@seita.nl"], indirect=True
+)
+def test_get_sensor_data_rejects_blank_source_type(
+    client,
+    setup_api_test_data: dict[str, Sensor],
+    requesting_user,
+):
+    """Check that GET /sensors/<id>/data rejects a blank source_type filter."""
+    sensor = setup_api_test_data["some gas sensor"]
+    message = {
+        "start": "2021-05-02T00:00:00+02:00",
+        "duration": "PT1H20M",
+        "horizon": "PT0H",
+        "unit": "m³/h",
+        "source_type": "",
+        "resolution": "PT20M",
+    }
+    response = client.get(
+        url_for("SensorAPI:get_data", id=sensor.id),
+        query_string=message,
+    )
+    print("Server responded with:\n%s" % response.json)
+    assert response.status_code == 422
+    assert (
+        "Source type must not be blank."
+        in response.json["message"]["combined_sensor_data_description"]["source_type"][
+            0
+        ]
+    )
+
+
 @pytest.mark.parametrize(
     "requesting_user, status_code",
     [