feat: search filter on sensor API

flexmeasures/api/common/schemas/generic_assets.py

@@ -1,9 +1,7 @@
 from __future__ import annotations
 
-from shlex import join, split
-
 from flask import abort
-from marshmallow import fields, ValidationError
+from marshmallow import fields
 from sqlalchemy import select
 
 from flexmeasures.data import db
@@ -25,17 +23,3 @@ def _deserialize(self, asset_id: int, attr, obj, **kwargs) -> GenericAsset:
 
     def _serialize(self, asset: GenericAsset, attr, data, **kwargs) -> int:
         return asset.id
-
-
-class SearchFilterField(fields.Str):
-    """Field that represents a search filter."""
-
-    def _deserialize(self, value, attr, data, **kwargs) -> list[str]:
-        try:
-            search_terms = split(value)
-        except ValueError as e:
-            raise ValidationError(str(e))
-        return search_terms
-
-    def _serialize(self, value: list[str], attr, obj, **kwargs) -> str:
-        return join(value)

flexmeasures/api/common/schemas/search.py

@@ -0,0 +1,19 @@
+from __future__ import annotations
+
+from shlex import join, split
+
+from marshmallow import fields, ValidationError
+
+
+class SearchFilterField(fields.Str):
+    """Field that represents a search filter."""
+
+    def _deserialize(self, value, attr, data, **kwargs) -> list[str]:
+        try:
+            search_terms = split(value)
+        except ValueError as e:
+            raise ValidationError(str(e))
+        return search_terms
+
+    def _serialize(self, value: list[str], attr, obj, **kwargs) -> str:
+        return join(value)

flexmeasures/api/common/schemas/sensors.py

@@ -1,5 +1,5 @@
 from flask import abort
-from marshmallow import fields
+from marshmallow import fields, ValidationError
 from sqlalchemy import select
 
 from flexmeasures.data import db
@@ -9,6 +9,7 @@
     EntityAddressException,
 )
 from flexmeasures.data.models.time_series import Sensor
+from flexmeasures.utils.unit_utils import is_valid_unit
 
 
 class EntityAddressValidationError(FMValidationError):
@@ -80,3 +81,15 @@ def _serialize(self, value: Sensor, attr, data, **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/v3_0/assets.py

@@ -23,19 +23,20 @@
 from flexmeasures.data.queries.generic_assets import query_assets_by_search_terms
 from flexmeasures.data.schemas import AwareDateTimeField
 from flexmeasures.data.schemas.generic_assets import GenericAssetSchema as AssetSchema
-from flexmeasures.api.common.schemas.generic_assets import (
-    AssetIdField,
-    SearchFilterField,
-)
+from flexmeasures.api.common.schemas.generic_assets import AssetIdField
+from flexmeasures.api.common.schemas.search import SearchFilterField
 from flexmeasures.api.common.schemas.users import AccountIdField
 from flexmeasures.utils.coding_utils import flatten_unique
 from flexmeasures.ui.utils.view_utils import set_session_variables
 from flexmeasures.auth.policy import check_access
 from werkzeug.exceptions import Forbidden, Unauthorized
+from flexmeasures.data.schemas.sensors import SensorSchema
+from flexmeasures.data.models.time_series import Sensor
 
 
 asset_schema = AssetSchema()
 assets_schema = AssetSchema(many=True)
+sensors_schema = SensorSchema(many=True)
 partial_asset_schema = AssetSchema(partial=True, exclude=["account_id"])
 
 
@@ -161,10 +162,6 @@ def index(
         if all_accessible:
             filter_statement = filter_statement | GenericAsset.account_id.is_(None)
 
-        num_records = db.session.scalar(
-            select(func.count(GenericAsset.id)).where(filter_statement)
-        )
-
         query = query_assets_by_search_terms(
             search_terms=filter, filter_statement=filter_statement
         )
@@ -177,6 +174,9 @@ def index(
             select_pagination: SelectPagination = db.paginate(
                 query, per_page=per_page, page=page
             )
+            num_records = db.session.scalar(
+                select(func.count(GenericAsset.id)).filter(filter_statement)
+            )
             response = {
                 "data": asset_schema.dump(select_pagination.items, many=True),
                 "num-records": num_records,
@@ -185,6 +185,102 @@ def index(
 
         return response, 200
 
+    @route(
+        "/<id>/sensors",
+        methods=["GET"],
+    )
+    @use_kwargs(
+        {
+            "asset": AssetIdField(data_key="id"),
+        },
+        location="path",
+    )
+    @use_kwargs(
+        {
+            "page": fields.Int(
+                required=False, validate=validate.Range(min=1), default=1
+            ),
+            "per_page": fields.Int(
+                required=False, validate=validate.Range(min=1), default=10
+            ),
+        },
+        location="query",
+    )
+    @as_json
+    def asset_sensors(
+        self,
+        id: int,
+        asset: GenericAsset | None,
+        page: int | None = None,
+        per_page: int | None = None,
+    ):
+        """
+        List all sensors under an asset.
+
+         .. :quickref: Asset; Return all sensors under an asset.
+
+        This endpoint returns all sensors under an asset.
+
+        The endpoint supports pagination of the asset list using the `page` and `per_page` query parameters.
+
+        - If the `page` parameter is not provided, all sensors are returned, without pagination information. The result will be a list of sensors.
+        - If a `page` parameter is provided, the response will be paginated, showing a specific number of assets per page as defined by `per_page` (default is 10).
+        The response schema for pagination is inspired by https://datatables.net/manual/server-side#Returned-data
+
+
+        **Example response**
+
+        An example of one asset being returned in a paginated response:
+
+        .. sourcecode:: json
+
+            {
+                "data" : [
+                    {
+                      "id": 1,
+                      "name": "Test battery",
+                      "latitude": 10,
+                      "longitude": 100,
+                      "account_id": 2,
+                      "generic_asset_type": {"id": 1, "name": "battery"}
+                    }
+                ],
+                "num-records" : 1,
+                "filtered-records" : 1
+
+            }
+
+        If no pagination is requested, the response only consists of the list under the "data" key.
+
+        :reqheader Authorization: The authentication token
+        :reqheader Content-Type: application/json
+        :resheader Content-Type: application/json
+        :status 200: PROCESSED
+        :status 400: INVALID_REQUEST
+        :status 401: UNAUTHORIZED
+        :status 403: INVALID_SENDER
+        :status 422: UNPROCESSABLE_ENTITY
+        """
+        filter_statement = Sensor.generic_asset_id == asset.id
+
+        query = select(Sensor).filter(filter_statement)
+
+        select_pagination: SelectPagination = db.paginate(
+            query, per_page=per_page, page=page
+        )
+
+        num_records = db.session.scalar(
+            select(func.count(Sensor.id)).where(filter_statement)
+        )
+
+        response = {
+            "data": sensors_schema.dump(select_pagination.items, many=True),
+            "num-records": num_records,
+            "filtered-records": select_pagination.total,
+        }
+
+        return response, 200
+
     @route("/public", methods=["GET"])
     @as_json
     def public(self):

flexmeasures/api/v3_0/sensors.py

@@ -1,17 +1,18 @@
 from __future__ import annotations
 
+import isodate
 from datetime import datetime, timedelta
 
 from flask import current_app, url_for
 from flask_classful import FlaskView, route
 from flask_json import as_json
 from flask_security import auth_required
-import isodate
 from marshmallow import fields, ValidationError
+import marshmallow.validate as validate
 from rq.job import Job, NoSuchJobError
 from timely_beliefs import BeliefsDataFrame
 from webargs.flaskparser import use_args, use_kwargs
-from sqlalchemy import delete
+from sqlalchemy import delete, select, or_
 
 from flexmeasures.api.common.responses import (
     request_processed,
@@ -29,6 +30,7 @@
 )
 from flexmeasures.api.common.schemas.users import AccountIdField
 from flexmeasures.api.common.utils.api_utils import save_and_enqueue
+from flexmeasures.auth.policy import check_access
 from flexmeasures.auth.decorators import permission_required_for_context
 from flexmeasures.data import db
 from flexmeasures.data.models.audit_log import AssetAuditLog
@@ -37,8 +39,11 @@
 from flexmeasures.data.models.time_series import Sensor, TimedBelief
 from flexmeasures.data.queries.utils import simplify_index
 from flexmeasures.data.schemas.sensors import SensorSchema, SensorIdField
+from flexmeasures.api.common.schemas.search import SearchFilterField
+from flexmeasures.api.common.schemas.sensors import UnitField
 from flexmeasures.data.schemas.times import AwareDateTimeField, PlanningDurationField
-from flexmeasures.data.services.sensors import get_sensors, get_sensor_stats
+from flexmeasures.data.queries.sensors import query_sensors_by_search_terms
+from flexmeasures.data.services.sensors import get_sensor_stats
 from flexmeasures.data.services.scheduling import (
     create_scheduling_job,
     get_data_source_for_job,
@@ -65,18 +70,37 @@ class SensorAPI(FlaskView):
             "account": AccountIdField(
                 data_key="account_id", load_default=AccountIdField.load_current
             ),
+            "all_accessible": fields.Boolean(required=False, missing=False),
+            "page": fields.Int(
+                required=False, validate=validate.Range(min=1), default=1
+            ),
+            "per_page": fields.Int(
+                required=False, validate=validate.Range(min=1), default=10
+            ),
+            "filter": SearchFilterField(required=False, default=None),
+            "unit": UnitField(required=False, default=None),
         },
         location="query",
     )
     @permission_required_for_context("read", ctx_arg_name="account")
     @as_json
-    def index(self, account: Account):
+    def index(
+        self,
+        account: Account,
+        all_accessible: bool = False,
+        page: int | None = None,
+        per_page: int | None = None,
+        filter: list[str] | None = None,
+        unit: str | None = None,
+    ):
         """API endpoint to list all sensors of an account.
 
         .. :quickref: Sensor; Download sensor list
 
         This endpoint returns all accessible sensors.
         Accessible sensors are sensors in the same account as the current user.
+        Alternatively, you can use the `all_accessible` query parameter to list sensors from all assets that the `current_user` has read access to, as well as all public assets. The default value is `false`.
+
         Only admins can use this endpoint to fetch sensors from a different account (by using the `account_id` query parameter).
 
         **Example response**
@@ -85,7 +109,8 @@ def index(self, account: Account):
 
         .. sourcecode:: json
 
-            [
+        {
+            "data" : [
                 {
                     "entity_address": "ea1.2021-01.io.flexmeasures.company:fm1.42",
                     "event_resolution": PT15M,
@@ -95,7 +120,12 @@ def index(self, account: Account):
                     "unit": "m\u00b3/h"
                     "id": 2
                 }
-            ]
+            ],
+            "num-records" : 1,
+            "filtered-records" : 1
+        }
+
+        If no pagination is requested, the response only consists of the list under the "data" key.
 
         :reqheader Authorization: The authentication token
         :reqheader Content-Type: application/json
@@ -106,8 +136,61 @@ def index(self, account: Account):
         :status 403: INVALID_SENDER
         :status 422: UNPROCESSABLE_ENTITY
         """
-        sensors = get_sensors(account=account)
-        return sensors_schema.dump(sensors), 200
+        if isinstance(account, list):
+            accounts = account
+        else:
+            accounts: list = [account] if account else []
+        account_ids: list = [acc.id for acc in accounts]
+
+        filter_statement = GenericAsset.account_id.in_(account_ids)
+
+        if all_accessible is not None:
+            consultancy_account_ids: list = [
+                acc.consultancy_account_id for acc in accounts
+            ]
+            account_ids.extend(consultancy_account_ids)
+            filter_statement = or_(
+                filter_statement,
+                GenericAsset.account_id.is_(None),
+            )
+
+        sensor_query = (
+            select(Sensor)
+            .join(GenericAsset, Sensor.generic_asset_id == GenericAsset.id)
+            .join(Account, GenericAsset.owner)
+            .filter(filter_statement)
+        )
+
+        if filter is not None:
+            sensor_query = query_sensors_by_search_terms(
+                sensor_query, search_terms=filter
+            )
+
+        if unit:
+            sensor_query = sensor_query.filter(Sensor.unit == unit)
+
+        sensors = (
+            db.session.scalars(sensor_query).all()
+            if page is None
+            else db.paginate(sensor_query, per_page=per_page, page=page).items
+        )
+
+        sensors = [sensor for sensor in sensors if check_access(sensor, "read") is None]
+
+        sensors_response = sensors_schema.dump(sensors)
+
+        # Return appropriate response for paginated or non-paginated data
+        if page is None:
+            return sensors_response, 200
+        else:
+            num_records = db.session.execute(sensor_query).scalars().count()
+            select_pagination = db.paginate(sensor_query, per_page=per_page, page=page)
+            response = {
+                "data": sensors_response,
+                "num-records": num_records,
+                "filtered-records": select_pagination.total,
+            }
+            return response, 200
 
     @route("/data", methods=["POST"])
     @use_args(