398 Support optional "source" in /sensors/data [GET]

documentation/api/change_log.rst

@@ -5,11 +5,19 @@ API change log
 
 .. note:: The FlexMeasures API follows its own versioning scheme. This is also reflected in the URL, allowing developers to upgrade at their own pace.
 
-v3.0-3 | 2022-08-28
+v3.0-4 | 2022-11-29
 """""""""""""""""""
 
-- Introduced ``consumption_price_sensor``, ``production_price_sensor`` and ``inflexible_device_sensors`` fields to `/sensors/<id>/schedules/trigger` (POST)
+- Introduced the ``source`` field to `/sensors/data` (GET) to obtain data for a given source (ID).
+- Fixed the JSON wrapping of the return message for `/sensors/data` (GET).
+- Changed the Notation section:
+
+    - Rewrote the section on filtering by source (ID) with a deprecation notice on filtering by account role and user ID.
+
+v3.0-3 | 2022-08-28
+"""""""""""""""""""
 
+- Introduced ``consumption_price_sensor``, ``production_price_sensor`` and ``inflexible_device_sensors`` fields to `/sensors/<id>/schedules/trigger` (POST).
 
 v3.0-2 | 2022-07-08
 """""""""""""""""""
@@ -77,8 +85,6 @@ v2.0-3 | 2021-06-07
 - Updated all entity addresses in documentation according to the fm0 scheme, preserving backwards compatibility.
 - Introduced the fm1 scheme for entity addresses for connections, markets, weather sensors and sensors.
 
-
-
 v2.0-2 | 2021-04-02
 """""""""""""""""""
 
@@ -103,7 +109,6 @@ v2.0-0 | 2020-11-14
 
 - Added REST endpoints for managing assets: `/assets/` (GET, POST) and `/asset/<id>` (GET, PATCH, DELETE).
 
-
 v1.3-11 | 2022-01-05
 """"""""""""""""""""
 
@@ -178,7 +183,6 @@ v1.3-2 | 2020-03-11
 
 - Fixed example entity addresses in simulation section
 
-
 v1.3-1 | 2020-02-08
 """""""""""""""""""
 
@@ -299,7 +303,6 @@ v1.0-1 | 2018-07-10
     - Added sections listing all endpoints per version
     - Documentation includes specifications of **all** supported API versions (supported versions have a registered Flask blueprint)
 
-
 v1.0-0 | 2018-07-10
 """""""""""""""""""
 

documentation/api/notation.rst

@@ -287,17 +287,18 @@ of the sensor's resolution, e.g. hourly or daily values if the sensor's resoluti
 Sources
 -------
 
-Requests for data may limit the data selection by specifying a source, for example, a specific user.
-Account roles are also valid source selectors.
-For example, to obtain data originating from either a meter data company or user 42, include the following:
+Requests for data may filter by source. FlexMeasures keeps track of the data source (the data's author, for example, a user, forecaster or scheduler belonging to a given organisation) of time series data.
+For example, to obtain data originating from data source 42, include the following:
 
 .. code-block:: json
 
     {
-        "sources": ["MDC", "42"],
+        "source": 42,
     }
 
-Here, "MDC" is the name of the account role for meter data companies.
+Data source IDs can be found by hovering over data in charts.
+
+.. note:: Older API version (< 3) accepted user IDs (integers), account roles (strings) and lists thereof, instead of data source IDs (integers).
 
 
 .. _units:

flexmeasures/api/common/schemas/sensor_data.py

@@ -15,7 +15,7 @@
 from flexmeasures.api.common.schemas.sensors import SensorField
 from flexmeasures.api.common.utils.api_utils import upsample_values
 from flexmeasures.data.models.planning.utils import initialize_index
-from flexmeasures.data.schemas.times import AwareDateTimeField, DurationField
+from flexmeasures.data.schemas import AwareDateTimeField, DurationField, SourceIdField
 from flexmeasures.data.services.time_series import simplify_index
 from flexmeasures.utils.time_utils import duration_isoformat, server_now
 from flexmeasures.utils.unit_utils import (
@@ -99,6 +99,7 @@ def check_schema_unit_against_sensor_unit(self, data, **kwargs):
 class GetSensorDataSchema(SensorDataDescriptionSchema):
 
     resolution = DurationField(required=False)
+    source = SourceIdField(required=False)
 
     # Optional field that can be used for extra validation
     type = fields.Str(
@@ -152,6 +153,7 @@ def dump_bdf(self, sensor_data_description: dict, **kwargs) -> dict:
         end = sensor_data_description["start"] + duration
         unit = sensor_data_description["unit"]
         resolution = sensor_data_description.get("resolution")
+        source = sensor_data_description.get("source")
 
         # Post-load configuration of belief timing against message type
         horizons_at_least = sensor_data_description.get("horizon", None)
@@ -172,6 +174,7 @@ def dump_bdf(self, sensor_data_description: dict, **kwargs) -> dict:
                 event_ends_before=end,
                 horizons_at_least=horizons_at_least,
                 horizons_at_most=horizons_at_most,
+                source=source,
                 beliefs_before=sensor_data_description.get("prior", None),
                 one_deterministic_belief_per_event=True,
                 resolution=resolution,
@@ -190,8 +193,8 @@ def dump_bdf(self, sensor_data_description: dict, **kwargs) -> dict:
             to_unit=unit,
         )
 
-        # Convert NaN to null
-        values = values.where(pd.notnull(values), None)
+        # Convert NaN to None, which JSON dumps as null values
+        values = values.astype(object).where(pd.notnull(values), None)
 
         # Form the response
         response = dict(

flexmeasures/api/v3_0/sensors.py

@@ -1,5 +1,4 @@
 from datetime import datetime, timedelta
-import json
 from typing import List, Optional
 
 from flask import current_app
@@ -182,6 +181,13 @@ def get_data(self, response: dict):
 
         The unit has to be convertible from the sensor's unit.
 
+        **Optional fields**
+
+        - "resolution" (see :ref:`resolutions`)
+        - "horizon" (see :ref:`beliefs`)
+        - "prior" (see :ref:`beliefs`)
+        - "source" (see :ref:`sources`)
+
         :reqheader Authorization: The authentication token
         :reqheader Content-Type: application/json
         :resheader Content-Type: application/json
@@ -191,7 +197,8 @@ def get_data(self, response: dict):
         :status 403: INVALID_SENDER
         :status 422: UNPROCESSABLE_ENTITY
         """
-        return json.dumps(response)
+        d, s = request_processed()
+        return dict(**response, **d), s
 
     @route("/<id>/schedules/trigger", methods=["POST"])
     @use_kwargs(

flexmeasures/api/v3_0/tests/conftest.py

@@ -1,19 +1,27 @@
 from datetime import timedelta
 
+import pandas as pd
 import pytest
 from flask_security import SQLAlchemySessionUserDatastore, hash_password
 
+from flexmeasures import Sensor, Source
 from flexmeasures.data.models.generic_assets import GenericAssetType, GenericAsset
-from flexmeasures.data.models.time_series import Sensor
+from flexmeasures.data.models.time_series import TimedBelief
 
 
 @pytest.fixture(scope="module")
-def setup_api_test_data(db, setup_roles_users, setup_generic_assets):
+def setup_api_test_data(
+    db, setup_roles_users, setup_generic_assets
+) -> dict[str, Sensor]:
     """
     Set up data for API v3.0 tests.
     """
     print("Setting up data for API v3.0 tests on %s" % db.engine)
-    add_gas_sensor(db, setup_roles_users["Test Supplier User"])
+    gas_sensor = add_gas_sensor(db, setup_roles_users["Test Supplier User"])
+    add_gas_measurements(
+        db, setup_roles_users["Test Supplier User"].data_source[0], gas_sensor
+    )
+    return {gas_sensor.name: gas_sensor}
 
 
 @pytest.fixture(scope="function")
@@ -26,7 +34,22 @@ def setup_api_fresh_test_data(
     print("Setting up fresh data for API 3.0 tests on %s" % fresh_db.engine)
     for sensor in Sensor.query.all():
         fresh_db.delete(sensor)
-    add_gas_sensor(fresh_db, setup_roles_users_fresh_db["Test Supplier User"])
+    gas_sensor = add_gas_sensor(
+        fresh_db, setup_roles_users_fresh_db["Test Supplier User"]
+    )
+    return {gas_sensor.name: gas_sensor}
+
+
+@pytest.fixture(scope="function")
+def setup_api_fresh_gas_measurements(
+    fresh_db, setup_api_fresh_test_data, setup_roles_users_fresh_db
+):
+    """Set up some measurements for the gas sensor."""
+    add_gas_measurements(
+        fresh_db,
+        setup_roles_users_fresh_db["Test Supplier User"].data_source[0],
+        setup_api_fresh_test_data["some gas sensor"],
+    )
 
 
 @pytest.fixture(scope="module")
@@ -46,7 +69,7 @@ def setup_inactive_user(db, setup_accounts, setup_roles_users):
     )
 
 
-def add_gas_sensor(db, test_supplier_user):
+def add_gas_sensor(db, test_supplier_user) -> Sensor:
     incineration_type = GenericAssetType(
         name="waste incinerator",
     )
@@ -58,7 +81,6 @@ def add_gas_sensor(db, test_supplier_user):
         account_id=test_supplier_user.account_id,
     )
     db.session.add(incineration_asset)
-    db.session.flush()
     gas_sensor = Sensor(
         name="some gas sensor",
         unit="m³/h",
@@ -67,3 +89,24 @@ def add_gas_sensor(db, test_supplier_user):
     )
     db.session.add(gas_sensor)
     gas_sensor.owner = test_supplier_user.account
+    db.session.flush()  # assign sensor id
+    return gas_sensor
+
+
+def add_gas_measurements(db, source: Source, gas_sensor: Sensor):
+    event_starts = [
+        pd.Timestamp("2021-08-02T00:00:00+02:00") + timedelta(minutes=minutes)
+        for minutes in range(0, 30, 10)
+    ]
+    event_values = [91.3, 91.7, 92.1]
+    beliefs = [
+        TimedBelief(
+            sensor=gas_sensor,
+            source=source,
+            event_start=event_start,
+            belief_horizon=timedelta(0),
+            event_value=event_value,
+        )
+        for event_start, event_value in zip(event_starts, event_values)
+    ]
+    db.session.add_all(beliefs)

flexmeasures/api/v3_0/tests/test_sensor_data_fresh_db.py

@@ -1,9 +1,14 @@
+from __future__ import annotations
+
+from datetime import timedelta
+
 import pytest
 from flask import url_for
 
+from flexmeasures import Sensor, Source, User
 from flexmeasures.api.tests.utils import get_auth_token
 from flexmeasures.api.v3_0.tests.utils import make_sensor_data_request_for_gas_sensor
-from flexmeasures.data.models.time_series import TimedBelief, Sensor
+from flexmeasures.data.models.time_series import TimedBelief
 
 
 @pytest.mark.parametrize(
@@ -50,3 +55,36 @@ def test_post_sensor_data(
     assert len(beliefs) == expected_num_values
     # check that values are scaled to the sensor unit correctly
     assert pytest.approx(beliefs[0].event_value - expected_value) == 0
+
+
+def test_get_sensor_data(
+    client,
+    setup_api_fresh_test_data: dict[str, Sensor],
+    setup_api_fresh_gas_measurements,
+    setup_roles_users_fresh_db: dict[str, User],
+):
+    """Check the /sensors/data endpoint for fetching 1 hour of data of a 10-minute resolution sensor."""
+    sensor = setup_api_fresh_test_data["some gas sensor"]
+    source: Source = setup_roles_users_fresh_db["Test Supplier User"].data_source[0]
+    assert sensor.event_resolution == timedelta(minutes=10)
+    message = {
+        "sensor": f"ea1.2021-01.io.flexmeasures:fm1.{sensor.id}",
+        "start": "2021-08-02T00:00:00+02:00",
+        "duration": "PT1H20M",
+        "horizon": "PT0H",
+        "unit": "m³/h",
+        "source": source.id,
+        "resolution": "PT20M",
+    }
+    auth_token = get_auth_token(client, "test_supplier_user_4@seita.nl", "testtest")
+    response = client.get(
+        url_for("SensorAPI:get_data"),
+        query_string=message,
+        headers={"content-type": "application/json", "Authorization": auth_token},
+    )
+    print("Server responded with:\n%s" % response.json)
+    assert response.status_code == 200
+    values = response.json["values"]
+    # We expect two data point (from conftest) followed by 2 null values (which are converted to None by .json)
+    # The first data point averages 91.3 and 91.7, and the second data point averages 92.1 and None.
+    assert all(a == b for a, b in zip(values, [91.5, 92.1, None, None]))

flexmeasures/data/models/charts/belief_charts.py

@@ -49,7 +49,7 @@ def bar_chart(
                     **event_value_field_definition,
                     **dict(title=f"{capitalize(sensor.sensor_type)}"),
                 },
-                FIELD_DEFINITIONS["source_name"],
+                FIELD_DEFINITIONS["source_name_and_id"],
                 FIELD_DEFINITIONS["source_model"],
             ],
         },
@@ -58,6 +58,10 @@ def bar_chart(
                 "calculate": f"datum.event_start + {resolution_in_ms}",
                 "as": "event_end",
             },
+            {
+                "calculate": "datum.source.name + ' (ID: ' + datum.source.id + ')'",
+                "as": "source_name_and_id",
+            },
         ],
     }
     for k, v in override_chart_specs.items():
@@ -110,7 +114,7 @@ def chart_for_multiple_sensors(
                 **event_value_field_definition,
                 **dict(title=f"{capitalize(sensor.sensor_type)}"),
             },
-            FIELD_DEFINITIONS["source_name"],
+            FIELD_DEFINITIONS["source_name_and_id"],
             FIELD_DEFINITIONS["source_model"],
         ]
         line_layer = {
@@ -226,6 +230,12 @@ def chart_for_multiple_sensors(
     chart_specs = dict(
         description="A vertically concatenated chart showing sensor data.",
         vconcat=[*sensors_specs],
+        transform=[
+            {
+                "calculate": "datum.source.name + ' (ID: ' + datum.source.id + ')'",
+                "as": "source_name_and_id",
+            },
+        ],
         spacing=100,
         bounds="flush",
     )

flexmeasures/data/models/charts/defaults.py

@@ -51,6 +51,11 @@
         type="nominal",
         title="Time and date",
     ),
+    "source_name_and_id": dict(
+        field="source_name_and_id",
+        type="nominal",
+        title="Source",
+    ),
 }
 SHADE_LAYER = {
     "mark": {

flexmeasures/data/schemas/__init__.py

@@ -1,4 +1,5 @@
 from .assets import LatitudeField, LongitudeField  # noqa F401
 from .generic_assets import GenericAssetIdField as AssetIdField  # noqa F401
 from .sensors import SensorIdField  # noqa F401
+from .sources import DataSourceIdField as SourceIdField  # noqa F401
 from .times import AwareDateTimeField, DurationField  # noqa F401

flexmeasures/data/schemas/sources.py

@@ -1,14 +1,17 @@
-from flask.cli import with_appcontext
 from marshmallow import fields
 
 from flexmeasures.data.models.data_sources import DataSource
-from flexmeasures.data.schemas.utils import FMValidationError, MarshmallowClickMixin
+from flexmeasures.data.schemas.utils import (
+    with_appcontext_if_needed,
+    FMValidationError,
+    MarshmallowClickMixin,
+)
 
 
 class DataSourceIdField(fields.Int, MarshmallowClickMixin):
-    """Field that deserializes to a Sensor and serializes back to an integer."""
+    """Field that deserializes to a DataSource and serializes back to an integer."""
 
-    @with_appcontext
+    @with_appcontext_if_needed()
     def _deserialize(self, value, attr, obj, **kwargs) -> DataSource:
         """Turn a source id into a DataSource."""
         source = DataSource.query.get(value)