diff --git a/documentation/api/change_log.rst b/documentation/api/change_log.rst index c75a260f70..4a68e79e37 100644 --- a/documentation/api/change_log.rst +++ b/documentation/api/change_log.rst @@ -8,6 +8,7 @@ API change log v3.0-32 | July XX, 2026 """""""""""""""""""""""" +- Added ``POST /api/v3_0/assets//reports/trigger`` to queue a one-off reporting job (to be picked up by a worker processing the ``reporting`` queue), mirroring how forecasts and schedules are triggered. The job status can be polled via ``GET /api/v3_0/jobs/``. - Extended ``GET /api/v3_0/jobs/`` with a ``result`` field containing ``unresolved`` and ``resolved`` arrays, each keyed by asset ID. For scheduling jobs, this surfaces soft state-of-charge constraint analysis: ``soc-minima`` and ``soc-maxima`` violations (with a ``violation`` magnitude) or satisfied constraints (with a ``margin`` headroom). Both arrays are empty when no SoC constraints were defined. v3.0-31 | 2026-06-01 diff --git a/documentation/changelog.rst b/documentation/changelog.rst index cfa087d796..b487b144a8 100644 --- a/documentation/changelog.rst +++ b/documentation/changelog.rst @@ -16,6 +16,7 @@ New features * Automations can also compute schedules on a recurring basis (``flexmeasures add automation --type schedules``), with the schedule start defaulting to each run's time [see `PR #2293 `_] * Automations can be created, edited and deleted in the UI and through new API endpoints (``[POST|PATCH|DELETE] /assets/(id)/automations``), by account admins and consultants [see `PR #2294 `_] * Reports can run as background jobs (``flexmeasures add report --as-job``, processed by workers of the new ``reporting`` queue) and be computed on a recurring basis by automations, with a rolling report window expressed as Pandas offsets or defaulting to the last cron period [see `PR #2297 `_] +* One-off reporting jobs can be triggered via the API (``POST /assets/(id)/reports/trigger``), mirroring how forecasts and schedules are triggered [see `PR #2298 `_] * Breaking behaviour change: the top-level flex-context's ``relax-constraints`` field now defaults to ``True`` (matching the default already used within each ``commodities`` entry), so constraint violations are softly penalized by default instead of being hard constraints, unless explicitly set to ``False`` [see `PR #2172 `_] * In the UI, asset and sensor lists can be filtered by ID prefix through API-backed search fields [see `PR #2231 `_] * Support configurable lower and upper bounds and snapping for forecast post-processing [see `PR #2273 `_] diff --git a/documentation/features/reporting.rst b/documentation/features/reporting.rst index ab2418ed9c..e88d974a25 100644 --- a/documentation/features/reporting.rst +++ b/documentation/features/reporting.rst @@ -132,6 +132,31 @@ Automating reports Reports can be queued as background jobs (add ``--as-job`` to ``flexmeasures add report``, and let a worker process the ``reporting`` queue, see :ref:`redis-queue`), and computed on a recurring basis by an *automation* defined on the asset (see :ref:`automations` for the full concept, including how to manage and run automations). +One-off reporting jobs can also be triggered via the API, mirroring how forecasts and schedules are triggered. +``POST /api/v3_0/assets//reports/trigger`` takes the reporter class, its configuration and the report parameters (which must reference at least one output sensor, assumed to belong to the asset or one of its (grand)children): + +.. code-block:: json + + { + "reporter": "PandasReporter", + "config": { + "required_input": [{"name": "sensor_1"}, {"name": "sensor_2"}], + "required_output": [{"name": "df_agg"}], + "transformations": [ + {"df_input": "sensor_1", "method": "add", "args": ["@sensor_2"], "df_output": "df_agg"}, + {"method": "resample_events", "args": ["2h"]} + ] + }, + "parameters": { + "input": [{"name": "sensor_1", "sensor": 1}, {"name": "sensor_2", "sensor": 2}], + "output": [{"name": "df_agg", "sensor": 3}], + "start": "2023-04-10T00:00:00+00:00", + "end": "2023-04-10T10:00:00+00:00" + } + } + +The response contains the id of the queued reporting job, whose status can be polled via the generic job status endpoint (``GET /api/v3_0/jobs/``). + The reporter and its configuration are stored on a data source (steady across runs, so all report results attribute to the same source), while the report parameters are stored on the automation itself and their timing is resolved freshly on each run: diff --git a/flexmeasures/api/v3_0/__init__.py b/flexmeasures/api/v3_0/__init__.py index 35fda32151..aed6c8e71b 100644 --- a/flexmeasures/api/v3_0/__init__.py +++ b/flexmeasures/api/v3_0/__init__.py @@ -44,6 +44,7 @@ AutomationUpdateSchema, ) from flexmeasures.data.schemas.generic_assets import GenericAssetSchema as AssetSchema +from flexmeasures.data.schemas.reporting import ReportTriggerSchema from flexmeasures.data.schemas.sensors import QuantitySchema, TimeSeriesSchema from flexmeasures.data.schemas.account import AccountSchema from flexmeasures.api.v3_0.accounts import AccountAPIQuerySchema @@ -163,6 +164,7 @@ def create_openapi_specs(app: Flask): ("AnnotationSchema", AnnotationSchema), ("AutomationCreationSchema", AutomationCreationSchema), ("AutomationUpdateSchema", AutomationUpdateSchema), + ("ReportTriggerSchema", ReportTriggerSchema), ("CopyAssetSchema", CopyAssetSchema), ("DefaultAssetViewJSONSchema", DefaultAssetViewJSONSchema), ("AccountSchema", AccountSchema(partial=True)), diff --git a/flexmeasures/api/v3_0/assets.py b/flexmeasures/api/v3_0/assets.py index d06b8c1784..87f849f8b0 100644 --- a/flexmeasures/api/v3_0/assets.py +++ b/flexmeasures/api/v3_0/assets.py @@ -56,6 +56,12 @@ update_automation, ) from flexmeasures.data.models.generic_assets import GenericAsset, GenericAssetType +from flexmeasures.data.models.reporting import Reporter +from flexmeasures.data.schemas.reporting import ( + ReporterParametersSchema, + ReportTriggerSchema, +) +from flexmeasures.data.services.data_sources import get_data_generator from flexmeasures.data.queries.generic_assets import ( filter_assets_under_root, query_assets_by_search_terms, @@ -2050,6 +2056,158 @@ def trigger_schedule( d, s = request_processed() return dict(**response, **d), s + @route("//reports/trigger", methods=["POST"]) + @use_kwargs( + {"asset": AssetIdField(data_key="id")}, + location="path", + ) + # Simplification of checking for create-children access on each of the output sensors, + # which assumes each of the output sensors belongs to the given asset. + @permission_required_for_context("create-children", ctx_arg_name="asset") + @as_json + def trigger_report(self, id: int, asset: GenericAsset): + """ + .. :quickref: Assets; Trigger a one-off reporting job for this asset. + + --- + post: + summary: Trigger a one-off reporting job for this asset. + description: | + Trigger FlexMeasures to compute a report as a background job + (to be picked up by a worker processing the `reporting` queue), + mirroring how forecasts and schedules are triggered. + + In this request, you can describe: + + - the reporter class to use (see the CLI command `flexmeasures show reporters` for the available reporters) + - the reporter's configuration (static parameters, stored on the reporter's data source, e.g. the transformations of a `PandasReporter`) + - the report parameters (dynamic parameters, such as which `input` data to use, which `output` sensors to save the report to, and the `start` and `end` of the reporting window) + + The asset in the URL path scopes permissions: triggering reports requires + `create-children` access to the asset. The report parameters must reference + at least one output sensor, which is assumed to belong to the given asset + (either directly or indirectly, by being assigned to one of the asset's (grand)children). + + The response contains the id of the queued reporting job, + whose status can be polled via [GET /api/v3_0/jobs/(uuid)](#/Jobs/get_api_v3_0_jobs__uuid_). + Once the job finishes, the report is saved to the database + and can be retrieved from the output sensor(s). + security: + - ApiKeyAuth: [] + parameters: + - in: path + name: id + required: true + $ref: '#/components/parameters/AssetIdPath' + requestBody: + content: + application/json: + schema: ReportTriggerSchema + examples: + aggregation_report: + summary: Sum two sensors into a report sensor + description: | + This message triggers a `PandasReporter` report that adds up the data + of sensors 1 and 2 and saves the result to sensor 3. + value: + reporter: PandasReporter + config: + required_input: + - name: sensor_1 + - name: sensor_2 + required_output: + - name: df_agg + transformations: + - df_input: sensor_1 + method: add + args: ["@sensor_2"] + df_output: df_agg + parameters: + input: + - name: sensor_1 + sensor: 1 + - name: sensor_2 + sensor: 2 + output: + - name: df_agg + sensor: 3 + start: "2023-04-10T00:00:00+00:00" + end: "2023-04-10T10:00:00+00:00" + responses: + 200: + description: PROCESSED + content: + application/json: + schema: + type: object + examples: + successful_response: + description: | + This message indicates that the reporting request has been processed without any error. + A reporting job has been created with some Universally Unique Identifier (UUID), + which will be picked up by a worker. + The given UUID may be used to poll the job status: see [GET /api/v3_0/jobs/(uuid)](#/Jobs/get_api_v3_0_jobs__uuid_). + value: + status: PROCESSED + report: "364bfd06-c1fa-430b-8d25-8f5a547651fb" + message: "Request has been processed." + 400: + description: INVALID_REQUEST, REQUIRED_INFO_MISSING, UNEXPECTED_PARAMS + 401: + description: UNAUTHORIZED + 403: + description: INVALID_SENDER + 422: + description: UNPROCESSABLE_ENTITY + tags: + - Assets + """ + body = request.get_json(silent=True) + if not body: + return unprocessable_entity("No JSON data provided.") + try: + report_data = ReportTriggerSchema().load(body) + except ValidationError as e: + return unprocessable_entity(e.messages) + parameters = report_data["parameters"] + try: + ReporterParametersSchema().load(parameters) + except ValidationError as e: + return unprocessable_entity({"parameters": e.messages}) + if not parameters.get("output"): + return unprocessable_entity( + {"parameters": {"output": ["At least one output sensor is required."]}} + ) + # Guard against referencing sensors outside the caller's reach + # (raises Forbidden, answered with 403) + _check_automation_sensor_access("reports", parameters, {}) + try: + reporter = get_data_generator( + source=None, + model=report_data["reporter"], + config=report_data["config"], + save_config=True, + data_generator_type=Reporter, + ) + except ValidationError as e: + return unprocessable_entity({"config": e.messages}) + if reporter is None: + return unprocessable_entity( + f"Reporter class `{report_data['reporter']}` not available." + " Use the CLI command `flexmeasures show reporters` to list all the available reporters." + ) + reporter.set_job_trigger("API") + try: + returns = reporter.compute(as_job=True, parameters=parameters) + except ValidationError as e: + return unprocessable_entity({"parameters": e.messages}) + except ValueError as e: + return unprocessable_entity(str(e)) + + response = dict(report=returns["job_id"]) + d, s = request_processed() + return dict(**response, **d), s + @route("//kpis", methods=["GET"]) @use_kwargs( { diff --git a/flexmeasures/api/v3_0/tests/test_report_trigger_api.py b/flexmeasures/api/v3_0/tests/test_report_trigger_api.py new file mode 100644 index 0000000000..c6a53be1c5 --- /dev/null +++ b/flexmeasures/api/v3_0/tests/test_report_trigger_api.py @@ -0,0 +1,171 @@ +"""Tests for the report trigger endpoint (POST /api/v3_0/assets//reports/trigger).""" + +from __future__ import annotations + +from datetime import timedelta + +import pytest +from flask import url_for + +from flexmeasures.data.models.time_series import Sensor + + +@pytest.fixture(scope="module") +def setup_report_sensors(db, add_battery_assets): + """Add two (hourly) input sensors and a (two-hourly) report sensor to the battery asset.""" + battery = add_battery_assets["Test battery"] + sensor_1 = Sensor( + "input sensor 1", generic_asset=battery, event_resolution=timedelta(hours=1) + ) + sensor_2 = Sensor( + "input sensor 2", generic_asset=battery, event_resolution=timedelta(hours=1) + ) + report_sensor = Sensor( + "report sensor", generic_asset=battery, event_resolution=timedelta(hours=2) + ) + db.session.add_all([sensor_1, sensor_2, report_sensor]) + db.session.flush() + return sensor_1, sensor_2, report_sensor + + +def make_report_trigger_message( + sensor_1: Sensor, sensor_2: Sensor, report_sensor: Sensor +) -> dict: + """A PandasReporter message that adds up two sensors at a two-hour resolution.""" + return { + "reporter": "PandasReporter", + "config": { + "required_input": [{"name": "sensor_1"}, {"name": "sensor_2"}], + "required_output": [{"name": "df_agg"}], + "transformations": [ + { + "df_input": "sensor_1", + "method": "add", + "args": ["@sensor_2"], + "df_output": "df_agg", + }, + {"method": "resample_events", "args": ["2h"]}, + ], + }, + "parameters": { + "input": [ + {"name": "sensor_1", "sensor": sensor_1.id}, + {"name": "sensor_2", "sensor": sensor_2.id}, + ], + "output": [{"name": "df_agg", "sensor": report_sensor.id}], + "start": "2023-04-10T00:00:00+00:00", + "end": "2023-04-10T10:00:00+00:00", + }, + } + + +@pytest.mark.parametrize( + "requesting_user, expected_status_code", + [ + (None, 401), # not logged in + ("test_prosumer_user@seita.nl", 200), # same account + ("test_dummy_user_3@seita.nl", 403), # different account + ], + indirect=["requesting_user"], +) +def test_trigger_report_auth( + app, + add_battery_assets, + setup_report_sensors, + clean_redis, + requesting_user, + expected_status_code, +): + """Triggering a report requires create-children access to the asset (any account member).""" + battery = add_battery_assets["Test battery"] + message = make_report_trigger_message(*setup_report_sensors) + with app.test_client() as client: + response = client.post( + url_for("AssetAPI:trigger_report", id=battery.id), + json=message, + ) + assert response.status_code == expected_status_code + + +@pytest.mark.parametrize( + "requesting_user", ["test_prosumer_user@seita.nl"], indirect=True +) +def test_trigger_report( + app, + add_battery_assets, + setup_report_sensors, + clean_redis, + requesting_user, +): + """A successful trigger queues a job on the reporting queue, which records how it was triggered.""" + battery = add_battery_assets["Test battery"] + message = make_report_trigger_message(*setup_report_sensors) + with app.test_client() as client: + response = client.post( + url_for("AssetAPI:trigger_report", id=battery.id), + json=message, + ) + assert response.status_code == 200 + assert response.json["status"] == "PROCESSED" + job_id = response.json["report"] + + jobs = app.queues["reporting"].jobs + assert len(jobs) == 1 + job = jobs[0] + assert job.id == job_id + assert job.meta["trigger"] == {"origin": "API"} + assert job.kwargs["parameters"]["output"] == message["parameters"]["output"] + + +@pytest.mark.parametrize( + "requesting_user", ["test_prosumer_user@seita.nl"], indirect=True +) +@pytest.mark.parametrize( + "message_updates, expected_error_field", + [ + # missing required timing fields + ({"parameters": {"start": None, "end": None}}, "start"), + # no output to save the report to + ({"parameters": {"output": None}}, "output"), + # unknown reporter class + ({"reporter": "UnknownReporter"}, "UnknownReporter"), + # invalid config for the reporter class + ({"config": {"invalid_field": 1}}, "invalid_field"), + # missing required fields altogether + ({"reporter": None, "parameters": None}, "reporter"), + ], +) +def test_trigger_report_with_invalid_message( + app, + add_battery_assets, + setup_report_sensors, + clean_redis, + requesting_user, + message_updates, + expected_error_field, +): + """Invalid trigger messages should yield a 422, and no job should be queued.""" + battery = add_battery_assets["Test battery"] + message = make_report_trigger_message(*setup_report_sensors) + + # apply updates to the valid message (None means: remove the field) + for field, update in message_updates.items(): + if update is None: + del message[field] + elif isinstance(update, dict): + for subfield, subupdate in update.items(): + if subupdate is None: + del message[field][subfield] + else: + message[field][subfield] = subupdate + else: + message[field] = update + + with app.test_client() as client: + response = client.post( + url_for("AssetAPI:trigger_report", id=battery.id), + json=message, + ) + assert response.status_code == 422 + assert expected_error_field in str(response.json) + assert len(app.queues["reporting"].jobs) == 0 diff --git a/flexmeasures/data/schemas/reporting/__init__.py b/flexmeasures/data/schemas/reporting/__init__.py index 02e580ae42..cb51eaa7e8 100644 --- a/flexmeasures/data/schemas/reporting/__init__.py +++ b/flexmeasures/data/schemas/reporting/__init__.py @@ -40,6 +40,37 @@ class ReporterParametersSchema(Schema): belief_horizon = DurationField(required=False) +class ReportTriggerSchema(Schema): + """Request schema for triggering a one-off reporting job (the asset comes from the URL path). + + The parameters are validated separately: first against the base ReporterParametersSchema, + and then against the reporter's own parameters schema. + """ + + reporter = fields.Str( + required=True, + metadata={ + "description": "Reporter class registered in flexmeasures.data.models.reporting" + " or in an available FlexMeasures plugin (e.g. PandasReporter)." + " Use the CLI command `flexmeasures show reporters` to list all the available reporters." + }, + ) + config = fields.Dict( + keys=fields.Str(), + load_default=dict, + metadata={ + "description": "Reporter configuration (static parameters, stored on the reporter's data source)." + }, + ) + parameters = fields.Dict( + keys=fields.Str(), + required=True, + metadata={ + "description": "Report parameters (dynamic parameters, such as `input`, `output`, `start` and `end`)." + }, + ) + + class BeliefsSearchConfigSchema(Schema): """ This schema implements the required fields to perform a TimedBeliefs search diff --git a/flexmeasures/ui/static/openapi-specs.json b/flexmeasures/ui/static/openapi-specs.json index 7cf3ac3669..0c6ed152f8 100644 --- a/flexmeasures/ui/static/openapi-specs.json +++ b/flexmeasures/ui/static/openapi-specs.json @@ -4195,6 +4195,125 @@ ] } }, + "/api/v3_0/assets/{id}/reports/trigger": { + "post": { + "summary": "Trigger a one-off reporting job for this asset.", + "description": "Trigger FlexMeasures to compute a report as a background job\n(to be picked up by a worker processing the `reporting` queue),\nmirroring how forecasts and schedules are triggered.\n\nIn this request, you can describe:\n\n- the reporter class to use (see the CLI command `flexmeasures show reporters` for the available reporters)\n- the reporter's configuration (static parameters, stored on the reporter's data source, e.g. the transformations of a `PandasReporter`)\n- the report parameters (dynamic parameters, such as which `input` data to use, which `output` sensors to save the report to, and the `start` and `end` of the reporting window)\n\nThe asset in the URL path scopes permissions: triggering reports requires\n`create-children` access to the asset. The report parameters must reference\nat least one output sensor, which is assumed to belong to the given asset\n(either directly or indirectly, by being assigned to one of the asset's (grand)children).\n\nThe response contains the id of the queued reporting job,\nwhose status can be polled via [GET /api/v3_0/jobs/(uuid)](#/Jobs/get_api_v3_0_jobs__uuid_).\nOnce the job finishes, the report is saved to the database\nand can be retrieved from the output sensor(s).\n", + "security": [ + { + "ApiKeyAuth": [] + } + ], + "parameters": [ + { + "in": "path", + "name": "id", + "required": true, + "$ref": "#/components/parameters/AssetIdPath" + } + ], + "requestBody": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ReportTriggerSchema" + }, + "examples": { + "aggregation_report": { + "summary": "Sum two sensors into a report sensor", + "description": "This message triggers a `PandasReporter` report that adds up the data\nof sensors 1 and 2 and saves the result to sensor 3.\n", + "value": { + "reporter": "PandasReporter", + "config": { + "required_input": [ + { + "name": "sensor_1" + }, + { + "name": "sensor_2" + } + ], + "required_output": [ + { + "name": "df_agg" + } + ], + "transformations": [ + { + "df_input": "sensor_1", + "method": "add", + "args": [ + "@sensor_2" + ], + "df_output": "df_agg" + } + ] + }, + "parameters": { + "input": [ + { + "name": "sensor_1", + "sensor": 1 + }, + { + "name": "sensor_2", + "sensor": 2 + } + ], + "output": [ + { + "name": "df_agg", + "sensor": 3 + } + ], + "start": "2023-04-10T00:00:00+00:00", + "end": "2023-04-10T10:00:00+00:00" + } + } + } + } + } + } + }, + "responses": { + "200": { + "description": "PROCESSED", + "content": { + "application/json": { + "schema": { + "type": "object" + }, + "examples": { + "successful_response": { + "description": "This message indicates that the reporting request has been processed without any error.\nA reporting job has been created with some Universally Unique Identifier (UUID),\nwhich will be picked up by a worker.\nThe given UUID may be used to poll the job status: see [GET /api/v3_0/jobs/(uuid)](#/Jobs/get_api_v3_0_jobs__uuid_).\n", + "value": { + "status": "PROCESSED", + "report": "364bfd06-c1fa-430b-8d25-8f5a547651fb", + "message": "Request has been processed." + } + } + } + } + } + }, + "400": { + "description": "INVALID_REQUEST, REQUIRED_INFO_MISSING, UNEXPECTED_PARAMS" + }, + "401": { + "description": "UNAUTHORIZED" + }, + "403": { + "description": "INVALID_SENDER" + }, + "422": { + "description": "UNPROCESSABLE_ENTITY" + } + }, + "tags": [ + "Assets" + ] + } + }, "/api/v3_0/assets/{id}/schedules/trigger": { "post": { "summary": "Trigger scheduling job for any number of devices", @@ -5616,6 +5735,30 @@ }, "additionalProperties": false }, + "ReportTriggerSchema": { + "type": "object", + "properties": { + "reporter": { + "type": "string", + "description": "Reporter class registered in flexmeasures.data.models.reporting or in an available FlexMeasures plugin (e.g. PandasReporter). Use the CLI command `flexmeasures show reporters` to list all the available reporters." + }, + "config": { + "type": "object", + "description": "Reporter configuration (static parameters, stored on the reporter's data source).", + "additionalProperties": {} + }, + "parameters": { + "type": "object", + "description": "Report parameters (dynamic parameters, such as `input`, `output`, `start` and `end`).", + "additionalProperties": {} + } + }, + "required": [ + "parameters", + "reporter" + ], + "additionalProperties": false + }, "CopyAssetSchema": { "type": "object", "properties": {