diff --git a/docker-compose.yml b/docker-compose.yml index 38bcbe6d48..1dd6273238 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -122,7 +122,7 @@ services: if [ -f /usr/var/flexmeasures-instance/requirements.txt ]; then pip install --no-cache-dir -r /usr/var/flexmeasures-instance/requirements.txt fi - flexmeasures jobs run-worker --name flexmeasures-worker --queue forecasting\|scheduling\|ingestion + flexmeasures jobs run-worker --name flexmeasures-worker --queue forecasting\|scheduling\|ingestion\|reporting test-db: image: postgres expose: diff --git a/documentation/api/change_log.rst b/documentation/api/change_log.rst index bb98ba154a..91b348e4a3 100644 --- a/documentation/api/change_log.rst +++ b/documentation/api/change_log.rst @@ -11,6 +11,7 @@ v3.0-34 | September 2, 2026 v3.0-33 | September 1, 2026 """"""""""""""""""""""""""" +- Added ``POST /api/v3_0/assets//reports/trigger`` to queue a one-off report as a background job. It returns ``202 Accepted`` with the canonical ``job`` and ``job-url`` fields, and shares the trigger rate limit with forecast and schedule endpoints. - Added ``GET /api/v3_0/assets//automations`` and ``GET /api/v3_0/assets//automations/`` for listing and inspecting forecast automations, including the sensors an automation reads from and writes to. Each automation shows the IANA ``timezone`` in which its cron expression is interpreted, and a ``cursor``: the offset-aware UTC time of the most recent run it committed to. The cursor advances just before queueing, so it does not indicate that queueing or the forecast itself succeeded. Asset job entries now include ``created_via`` provenance; automation identity is included only when the caller may read that automation. - Added ``GET /api/v3_0/sources/`` to show the full record of one data source, including the attributes in which data generators store their configuration. diff --git a/documentation/changelog.rst b/documentation/changelog.rst index d59bd84fc2..8a30a2a74f 100644 --- a/documentation/changelog.rst +++ b/documentation/changelog.rst @@ -17,6 +17,7 @@ v1.1.0 | September XX, 2026 New features ------------- +* Run one-off reports as background jobs from the CLI or the asset API, with sensor-level authorization and a dedicated reporting worker queue [see `PR #2298 `_] * A single automation can now be run on demand, from the CLI (``flexmeasures jobs run-automation``), the API (``POST /assets//automations//trigger``) and the asset's *Automations* page (a *Run now* button), which is useful to try out a new automation, to re-run one after fixing what made it fail, or to refresh its results after late input data arrived [see `PR #2460 `_] * Changing the selected time range on an asset or sensor chart now only loads the data that is actually new, instead of reloading the whole range, which makes stepping through or extending a long period much faster; reloading the page, or leaving it open for five minutes, still fetches everything afresh [see `PR #2433 `_] diff --git a/documentation/cli/change_log.rst b/documentation/cli/change_log.rst index e77643ded9..d940c74549 100644 --- a/documentation/cli/change_log.rst +++ b/documentation/cli/change_log.rst @@ -4,6 +4,11 @@ FlexMeasures CLI Changelog ********************** +since v1.1.0 | September XX, 2026 +================================= + +* Add ``flexmeasures add report --as-job`` and the ``reporting`` worker queue for asynchronous one-off reports. + since v1.0.0 | August 11, 2026 ================================= diff --git a/documentation/configuration.rst b/documentation/configuration.rst index 471408cb47..37e0d05ae8 100644 --- a/documentation/configuration.rst +++ b/documentation/configuration.rst @@ -346,7 +346,7 @@ Default: ``timedelta(days=1)`` FLEXMEASURES_DEFAULT_JOB_TIMEOUT ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ -Default timeout for jobs (e.g. forecasting, scheduling and ingestion), expressed as a fixed ISO 8601 duration. +Default timeout for jobs (e.g. forecasting, scheduling, ingestion and reporting), expressed as a fixed ISO 8601 duration. Jobs that exceed this timeout are moved to RQ's failed queue. Default: ``timedelta(seconds=180)`` (``"PT180S"``) @@ -356,9 +356,9 @@ FLEXMEASURES_JOB_TIMEOUT Timeouts per queue, expressed as fixed ISO 8601 durations. Queue-specific values override ``FLEXMEASURES_DEFAULT_JOB_TIMEOUT``. -Supported queue names are ``forecasting``, ``scheduling`` and ``ingestion``. +Supported queue names are ``forecasting``, ``scheduling``, ``ingestion`` and ``reporting``. -Example: ``{"forecasting": "PT2M", "scheduling": "PT5M", "ingestion": "PT30S"}`` +Example: ``{"forecasting": "PT2M", "scheduling": "PT5M", "ingestion": "PT30S", "reporting": "PT10M"}`` Default: ``{}`` @@ -925,9 +925,8 @@ Default: ``"500 per minute"`` FLEXMEASURES_API_TRIGGER_RATE_LIMIT ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ -How often a client may trigger a schedule or a forecast. This is the expensive work, so this limit is stricter -than the default one. The trigger endpoints share this budget, so triggering a forecast and triggering a schedule -draw on the same one. +How often a client may trigger a schedule, forecast or report. This is the expensive work, so this limit is stricter +than the default one. The trigger endpoints share this budget, so all three kinds of computation draw on the same one. Default: ``"10 per 5 minutes"`` diff --git a/documentation/features/reporting.rst b/documentation/features/reporting.rst index 9576b035bd..5262af925c 100644 --- a/documentation/features/reporting.rst +++ b/documentation/features/reporting.rst @@ -17,6 +17,12 @@ We added an infrastructure that allows us to define computation pipelines and CL - ``flexmeasures show reporters`` - ``flexmeasures add report`` +Reports can be queued for asynchronous processing with ``flexmeasures add report --as-job``. +Run ``flexmeasures jobs run-worker --queue reporting`` to process these jobs. A one-off report +can also be queued through ``POST /api/v3_0/assets//reports/trigger``. The caller needs read +access to every input and configuration sensor and permission to record data on each output; +outputs are limited to the asset in the URL and its descendants. + The reporter classes we are designing are using pandas under the hood and can be sub-classed, allowing us to build new reporters from stable simpler ones, and even pipelines. Remember: re-use is developer power! We believe this infrastructure will become very powerful and enable FlexMeasures hosts and plugin developers to implement exciting new features. @@ -122,4 +128,4 @@ The input sensor stores the power/energy flow, and the output sensor will store Here, the ``ProfitOrLossReporter`` used as source (with Id 6) is the one we configured above. With the offsets, we control the timing ― we indicate that we want the new report to encompass the day of tomorrow (see Pandas offset strings). -The report sensor will now store all costs which we know will be made tomorrow by the schedule. \ No newline at end of file +The report sensor will now store all costs which we know will be made tomorrow by the schedule. diff --git a/documentation/host/installation.rst b/documentation/host/installation.rst index bc509e744c..466e76116a 100644 --- a/documentation/host/installation.rst +++ b/documentation/host/installation.rst @@ -382,7 +382,7 @@ Then, start workers in a console (or some other method to keep a long-running pr .. code-block:: bash - $ flexmeasures jobs run-worker --queue "scheduling|forecasting|ingestion" + $ flexmeasures jobs run-worker --queue "scheduling|forecasting|ingestion|reporting" You can go to `http://localhost:5000/tasks/` and see the state of job queues and find individual jobs (and investigate why they failed, for instance). diff --git a/documentation/host/queues.rst b/documentation/host/queues.rst index 5454061615..b9180e447b 100644 --- a/documentation/host/queues.rst +++ b/documentation/host/queues.rst @@ -24,7 +24,7 @@ Here is how to run one worker for each kind of job (in separate terminals): .. code-block:: bash - $ flexmeasures jobs run-worker --name our-only-worker --queue forecasting|scheduling|ingestion + $ flexmeasures jobs run-worker --name our-only-worker --queue "forecasting|scheduling|ingestion|reporting" Running multiple workers in parallel might be a great idea. @@ -33,6 +33,7 @@ Running multiple workers in parallel might be a great idea. $ flexmeasures jobs run-worker --name forecaster --queue forecasting $ flexmeasures jobs run-worker --name scheduler --queue scheduling $ flexmeasures jobs run-worker --name ingester --queue ingestion + $ flexmeasures jobs run-worker --name reporter --queue reporting You can also clear the job queues: @@ -41,6 +42,7 @@ You can also clear the job queues: $ flexmeasures jobs clear-queue --queue forecasting $ flexmeasures jobs clear-queue --queue scheduling $ flexmeasures jobs clear-queue --queue ingestion + $ flexmeasures jobs clear-queue --queue reporting When the main FlexMeasures process runs (e.g. by ``flexmeasures run``\ ), the queues of forecasting and scheduling jobs can be visited at ``http://localhost:5000/tasks/forecasting`` and ``http://localhost:5000/tasks/schedules``\ , respectively (by admins). diff --git a/flexmeasures/api/common/schemas/tests/test_sensor_data_schema.py b/flexmeasures/api/common/schemas/tests/test_sensor_data_schema.py index 5d16c50cea..d4d44de45c 100644 --- a/flexmeasures/api/common/schemas/tests/test_sensor_data_schema.py +++ b/flexmeasures/api/common/schemas/tests/test_sensor_data_schema.py @@ -2,6 +2,7 @@ import json import pytest import pytz +from rq.job import JobStatus from marshmallow import ValidationError import pandas as pd @@ -465,7 +466,7 @@ def test_asset_sensors_metadata( def test_build_asset_jobs_data(db, app, add_battery_assets, clean_redis): - """Check that we get both types of jobs for a battery asset.""" + """Check that we get scheduling, forecasting and reporting jobs.""" battery_asset = add_battery_assets["Test battery"] battery = battery_asset.sensors[0] tz = pytz.timezone("Europe/Amsterdam") @@ -495,26 +496,47 @@ def test_build_asset_jobs_data(db, app, add_battery_assets, clean_redis): }, ) forecasting_job = app.queues["forecasting"].fetch_job(pipeline_returns["job_id"]) + reporting_job = app.queues["reporting"].enqueue(sum, [1, 2]) + reporting_job.meta["exception"] = "report failed" + reporting_job.save_meta() + reporting_job.set_status(JobStatus.FAILED) + app.job_cache.add( + battery.id, + reporting_job.id, + queue="reporting", + asset_or_sensor_type="sensor", + ) jobs_data = build_asset_jobs_data(battery_asset) forecasting_jobs_data = [j for j in jobs_data if j["queue"] == "forecasting"] scheduling_jobs_data = [j for j in jobs_data if j["queue"] == "scheduling"] + reporting_jobs_data = [j for j in jobs_data if j["queue"] == "reporting"] assert len(forecasting_jobs_data) == 1 assert scheduling_jobs_data + assert len(reporting_jobs_data) == 1 + assert ( + reporting_jobs_data[0]["err"] == "Reporting job failed with str: report failed" + ) scheduling_job_ids = set() for job_data in jobs_data: metadata = json.loads(job_data["metadata"]) if job_data["queue"] == "forecasting": assert metadata["job_id"] == forecasting_job.id assert job_data["entity"] == f"sensor: {battery.name} (Id: {battery.id})" - else: + assert job_data["status"] == "queued" + elif job_data["queue"] == "scheduling": scheduling_job_ids.add(metadata["job_id"]) - assert job_data["status"] == "queued" + assert job_data["status"] == "queued" + else: + assert metadata["job_id"] == reporting_job.id + assert job_data["status"] == JobStatus.FAILED assert scheduling_job.id in scheduling_job_ids # Clean up queues app.queues["scheduling"].empty() app.queues["forecasting"].empty() + app.queues["reporting"].empty() assert app.queues["scheduling"].count == 0 assert app.queues["forecasting"].count == 0 + assert app.queues["reporting"].count == 0 diff --git a/flexmeasures/api/common/utils/api_utils.py b/flexmeasures/api/common/utils/api_utils.py index 0ec2f8ef26..1ddca6fb60 100644 --- a/flexmeasures/api/common/utils/api_utils.py +++ b/flexmeasures/api/common/utils/api_utils.py @@ -25,8 +25,8 @@ add_beliefs_to_db_and_enqueue_forecasting_jobs, ) from flexmeasures.data.models.generic_assets import GenericAsset -from flexmeasures.data.models.time_series import Sensor from flexmeasures.data.queries.generic_assets import asset_is_in_subtree +from flexmeasures.data.models.time_series import Sensor from flexmeasures.data.utils import ( SAVE_TO_DB_SUCCESS, SAVE_TO_DB_SUCCESS_BUT_NOTHING_NEW, diff --git a/flexmeasures/api/v3_0/__init__.py b/flexmeasures/api/v3_0/__init__.py index d6b73a9cf0..515df4f918 100644 --- a/flexmeasures/api/v3_0/__init__.py +++ b/flexmeasures/api/v3_0/__init__.py @@ -42,6 +42,7 @@ ) from flexmeasures.data.schemas.annotations import AnnotationSchema 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, @@ -222,6 +223,7 @@ def create_openapi_specs(app: Flask): ("AssetAPIQuerySchema", AssetAPIQuerySchema), ("AssetSchema", AssetSchema), ("AnnotationSchema", AnnotationSchema), + ("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 a51cd37cec..28c84dfd79 100644 --- a/flexmeasures/api/v3_0/assets.py +++ b/flexmeasures/api/v3_0/assets.py @@ -57,13 +57,21 @@ run_automation, ) from flexmeasures.data.models.generic_assets import GenericAsset, GenericAssetType +from flexmeasures.data.models.reporting import Reporter from flexmeasures.data.queries.generic_assets import ( + asset_is_in_subtree, filter_assets_under_root, query_assets_by_search_terms, ) from flexmeasures.data.queries.utils import id_prefix_filter from flexmeasures.data.schemas import AwareDateTimeField from flexmeasures.data.schemas.annotations import AnnotationSchema +from flexmeasures.data.schemas.reporting import ReportTriggerSchema +from flexmeasures.data.services.data_generators import ( + check_sensor_access, + resolve_data_generator_sensors, +) +from flexmeasures.data.services.data_sources import get_data_generator from flexmeasures.data.services.annotations import prepare_annotations_for_chart from flexmeasures.data.schemas.generic_assets import ( GenericAssetSchema as AssetSchema, @@ -1892,6 +1900,129 @@ def update_keep_legends_below_graphs(self, **kwargs): "message": "Default legend position updated successfully.", }, 200 + @route("//reports/trigger", methods=["POST"]) + @limit_triggers() + @use_kwargs({"asset": AssetIdField(data_key="id")}, location="path") + @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: | + Queue a one-off report for a worker processing the `reporting` queue. + The caller must be able to read every input/configuration sensor and + record data on every output sensor. Each output must belong to the + asset in the URL or one of its descendants. + security: + - ApiKeyAuth: [] + parameters: + - in: path + name: id + required: true + $ref: '#/components/parameters/AssetIdPath' + requestBody: + content: + application/json: + schema: ReportTriggerSchema + responses: + 202: + description: ACCEPTED + content: + application/json: + schema: + type: object + required: + - status + - message + - job + - job-url + properties: + status: + type: string + enum: + - ACCEPTED + message: + type: string + job: + type: string + description: UUID of the queued reporting job. + job-url: + type: string + format: uri + description: URL to query the generic job status API. + example: + status: ACCEPTED + message: Request has been accepted for processing. + job: 364bfd06-c1fa-430b-8d25-8f5a547651fb + job-url: /api/v3_0/jobs/364bfd06-c1fa-430b-8d25-8f5a547651fb + 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 exc: + return unprocessable_entity(exc.messages) + + try: + reporter = get_data_generator( + source=None, # pre-defined app.data_generators + model=report_data["reporter"], + config=report_data["config"], + save_config=True, + data_generator_type=Reporter, + ) + except ValidationError as exc: + db.session.rollback() + return unprocessable_entity({"config": exc.messages}) + if reporter is None: + db.session.rollback() + return unprocessable_entity( + f"Reporter class `{report_data['reporter']}` not available." + ) + + parameters = report_data["parameters"] + try: + deserialized_parameters = reporter._parameters_schema.load(parameters) + report_sensors = resolve_data_generator_sensors( + reporter, deserialized_parameters + ) + check_sensor_access( + report_sensors["input_sensors"], report_sensors["output_sensors"] + ) + for output_sensor in report_sensors["output_sensors"]: + if not asset_is_in_subtree(asset.id, output_sensor.generic_asset_id): + raise ValueError( + f"Report output sensor {output_sensor.id} must belong to asset" + f" {asset.id} or one of its descendants." + ) + reporter.set_job_trigger("API") + result = reporter.compute(as_job=True, parameters=parameters) + except ValidationError as exc: + db.session.rollback() + return unprocessable_entity({"parameters": exc.messages}) + except ValueError as exc: + db.session.rollback() + return unprocessable_entity(str(exc)) + except Forbidden: + db.session.rollback() + raise + + return request_accepted_for_processing(result["job_id"]) + @route("//schedules/trigger", methods=["POST"]) @limit_triggers() @use_args(AssetTriggerSchemaV3(), location="args_and_json", as_kwargs=True) diff --git a/flexmeasures/api/v3_0/tests/test_rate_limiting.py b/flexmeasures/api/v3_0/tests/test_rate_limiting.py index 4c2bd6384c..3bd5b73680 100644 --- a/flexmeasures/api/v3_0/tests/test_rate_limiting.py +++ b/flexmeasures/api/v3_0/tests/test_rate_limiting.py @@ -474,6 +474,7 @@ def test_trigger_limited_views_are_registered(): from flexmeasures.api.common.rate_limiting import TRIGGER_LIMITED_VIEWS assert { + "AssetAPI.trigger_report", "AssetAPI.trigger_schedule", "SensorAPI.trigger_schedule", "SensorAPI.trigger_forecast", 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..1766c9781a --- /dev/null +++ b/flexmeasures/api/v3_0/tests/test_report_trigger_api.py @@ -0,0 +1,491 @@ +"""Tests for POST /api/v3_0/assets//reports/trigger.""" + +import logging +from datetime import datetime, timedelta, timezone + +import pytest +from flask import url_for +from rq.job import JobStatus +from sqlalchemy import func, select + +from flexmeasures.data.models.data_sources import DataSource +from flexmeasures.data.models.generic_assets import GenericAsset +from flexmeasures.data.models.time_series import Sensor, TimedBelief +from flexmeasures.utils.job_utils import work_on_rq + + +@pytest.fixture +def setup_report_sensors( + fresh_db, + add_battery_assets_fresh_db, + setup_accounts_fresh_db, + setup_generic_asset_types_fresh_db, +): + """Create report sensors inside and outside the target asset subtree.""" + battery = add_battery_assets_fresh_db["Test battery"] + input_1 = Sensor( + "report input 1", + generic_asset=battery, + event_resolution=timedelta(hours=1), + unit="kW", + ) + input_2 = Sensor( + "report input 2", + generic_asset=battery, + event_resolution=timedelta(hours=1), + unit="kW", + ) + output = Sensor( + "report output", + generic_asset=battery, + event_resolution=timedelta(hours=2), + unit="kW", + ) + cost_output = Sensor( + "cost output", + generic_asset=battery, + event_resolution=timedelta(hours=2), + unit="EUR", + ) + local_price = Sensor( + "local price", + generic_asset=battery, + event_resolution=timedelta(hours=1), + unit="EUR/kWh", + ) + sibling = GenericAsset( + name="Sibling battery", + generic_asset_type=setup_generic_asset_types_fresh_db["battery"], + owner=battery.owner, + parent_asset=battery.parent_asset, + ) + sibling_output = Sensor( + "sibling output", generic_asset=sibling, event_resolution=timedelta(hours=2) + ) + foreign = GenericAsset( + name="Foreign report asset", + generic_asset_type=setup_generic_asset_types_fresh_db["battery"], + owner=setup_accounts_fresh_db["Dummy"], + ) + foreign_input = Sensor( + "foreign input", + generic_asset=foreign, + event_resolution=timedelta(hours=1), + unit="kW", + ) + foreign_price = Sensor( + "foreign price", + generic_asset=foreign, + event_resolution=timedelta(hours=1), + unit="EUR/kWh", + ) + fresh_db.session.add_all( + [ + input_1, + input_2, + output, + cost_output, + local_price, + sibling, + sibling_output, + foreign, + foreign_input, + foreign_price, + ] + ) + fresh_db.session.flush() + return { + "asset": battery, + "input_1": input_1, + "input_2": input_2, + "output": output, + "cost_output": cost_output, + "local_price": local_price, + "sibling_output": sibling_output, + "foreign_input": foreign_input, + "foreign_price": foreign_price, + } + + +def report_message(sensor_1: Sensor, sensor_2: Sensor, output: Sensor) -> dict: + """Build a PandasReporter trigger message that adds two inputs and resamples their sum to two-hour events.""" + return { + "reporter": "PandasReporter", + "config": { + "required_input": [ + {"name": "one", "unit": "kW"}, + {"name": "two", "unit": "kW"}, + ], + "required_output": [{"name": "sum", "unit": "kW"}], + "transformations": [ + { + "df_input": "one", + "method": "add", + "args": ["@two"], + "df_output": "sum", + }, + {"method": "resample_events", "args": ["2h"]}, + ], + }, + "parameters": { + "input": [ + {"name": "one", "sensor": sensor_1.id}, + {"name": "two", "sensor": sensor_2.id}, + ], + "output": [{"name": "sum", "sensor": output.id}], + "start": "2023-04-10T00:00:00+00:00", + "end": "2023-04-10T10:00:00+00:00", + }, + } + + +def reporter_source_count(db) -> int: + return db.session.scalar( + select(func.count()) + .select_from(DataSource) + .where(DataSource.type == "reporter") + ) + + +@pytest.mark.parametrize( + "requesting_user, expected_status", + [ + (None, 401), + ("test_prosumer_user@seita.nl", 202), + ("test_dummy_user_3@seita.nl", 403), + ], + indirect=["requesting_user"], +) +def test_trigger_report_auth( + app, setup_report_sensors, clean_redis, requesting_user, expected_status +): + sensors = setup_report_sensors + with app.test_client() as client: + response = client.post( + url_for("AssetAPI:trigger_report", id=sensors["asset"].id), + json=report_message( + sensors["input_1"], sensors["input_2"], sensors["output"] + ), + ) + assert response.status_code == expected_status + + +@pytest.mark.parametrize( + "requesting_user", ["test_prosumer_user@seita.nl"], indirect=True +) +def test_trigger_report_queues_canonical_job_response( + app, setup_report_sensors, clean_redis, requesting_user +): + sensors = setup_report_sensors + with app.test_client() as client: + response = client.post( + url_for("AssetAPI:trigger_report", id=sensors["asset"].id), + json=report_message( + sensors["input_1"], sensors["input_2"], sensors["output"] + ), + ) + + assert response.status_code == 202 + assert response.json["status"] == "ACCEPTED" + job = app.queues["reporting"].jobs[0] + assert response.json["job"] == job.id + assert response.json["job-url"].endswith(f"/jobs/{job.id}") + assert job.meta["trigger"] == {"origin": "API"} + + +@pytest.mark.parametrize( + "requesting_user", ["test_prosumer_user@seita.nl"], indirect=True +) +def test_trigger_report_worker_stores_report_data( + app, fresh_db, setup_report_sensors, clean_redis, caplog, requesting_user +): + """Exercise the API, reporting queue, worker function and persisted output.""" + sensors = setup_report_sensors + source = DataSource("report input source") + fresh_db.session.add(source) + fresh_db.session.flush() + with fresh_db.session.no_autoflush: + beliefs = [ + TimedBelief( + event_start=datetime(2023, 4, 10, hour=hour, tzinfo=timezone.utc), + belief_time=datetime(2023, 4, 9, tzinfo=timezone.utc), + event_value=hour, + sensor=sensor, + source=source, + ) + for sensor in (sensors["input_1"], sensors["input_2"]) + for hour in range(10) + ] + fresh_db.session.add_all(beliefs) + fresh_db.session.commit() + + with app.test_client() as client: + response = client.post( + url_for("AssetAPI:trigger_report", id=sensors["asset"].id), + json=report_message( + sensors["input_1"], sensors["input_2"], sensors["output"] + ), + ) + assert response.status_code == 202 + + with caplog.at_level(logging.INFO): + work_on_rq(app.queues["reporting"]) + assert any( + "ran successfully" in record.message + and str(sensors["output"].id) in record.message + for record in caplog.records + ) + stored_report = sensors["output"].search_beliefs( + event_starts_after="2023-04-10T00:00:00+00:00", + event_ends_before="2023-04-10T10:00:00+00:00", + ) + # Adding the two hourly 0-9 input series yields 0, 2, ..., 18; two-hour mean resampling yields 1, 5, 9, 13, 17. + assert stored_report.values.T.tolist() == [[1, 5, 9, 13, 17]] + + +@pytest.mark.parametrize( + "requesting_user", ["test_prosumer_user@seita.nl"], indirect=True +) +@pytest.mark.parametrize( + "mutation, expected_text", + [ + (lambda message: message["parameters"].pop("start"), "start"), + (lambda message: message["parameters"].pop("output"), "output"), + (lambda message: message.update(reporter="UnknownReporter"), "UnknownReporter"), + (lambda message: message["config"].update(invalid=1), "invalid"), + ], +) +def test_trigger_report_rejects_invalid_requests( + app, + setup_report_sensors, + clean_redis, + requesting_user, + mutation, + expected_text, +): + sensors = setup_report_sensors + message = report_message(sensors["input_1"], sensors["input_2"], sensors["output"]) + mutation(message) + with app.test_client() as client: + response = client.post( + url_for("AssetAPI:trigger_report", id=sensors["asset"].id), json=message + ) + assert response.status_code == 422 + assert expected_text.casefold() in str(response.json).casefold() + assert app.queues["reporting"].jobs == [] + + +@pytest.mark.parametrize( + "requesting_user", ["test_prosumer_user@seita.nl"], indirect=True +) +@pytest.mark.parametrize("reporter", ["AggregatorReporter", "ProfitOrLossReporter"]) +def test_trigger_report_rejects_missing_specialized_dataflow_fields_without_side_effects( + app, + fresh_db, + setup_report_sensors, + clean_redis, + requesting_user, + reporter, +): + sensors = setup_report_sensors + if reporter == "AggregatorReporter": + missing_field = "output" + message = { + "reporter": reporter, + "config": {"method": "sum"}, + "parameters": { + "input": [{"sensor": sensors["input_1"].id}], + "start": "2023-04-10T00:00:00+00:00", + "end": "2023-04-10T10:00:00+00:00", + }, + } + else: + missing_field = "input" + message = { + "reporter": reporter, + "config": {"consumption_price_sensor": sensors["local_price"].id}, + "parameters": { + "output": [{"sensor": sensors["cost_output"].id}], + "start": "2023-04-10T00:00:00+00:00", + "end": "2023-04-10T10:00:00+00:00", + }, + } + + source_count = reporter_source_count(fresh_db) + with app.test_client() as client: + response = client.post( + url_for("AssetAPI:trigger_report", id=sensors["asset"].id), json=message + ) + + assert response.status_code == 422 + assert missing_field in str(response.json) + assert reporter_source_count(fresh_db) == source_count + assert app.queues["reporting"].jobs == [] + + +@pytest.mark.parametrize( + "requesting_user", ["test_prosumer_user@seita.nl"], indirect=True +) +@pytest.mark.parametrize("dependency", ["parameter", "config"]) +def test_trigger_report_rejects_inaccessible_inputs_without_side_effects( + app, + fresh_db, + setup_report_sensors, + clean_redis, + requesting_user, + dependency, +): + sensors = setup_report_sensors + if dependency == "parameter": + message = report_message( + sensors["foreign_input"], sensors["input_2"], sensors["output"] + ) + else: + message = { + "reporter": "ProfitOrLossReporter", + "config": {"consumption_price_sensor": sensors["foreign_price"].id}, + "parameters": { + "input": [{"sensor": sensors["input_1"].id}], + "output": [{"sensor": sensors["cost_output"].id}], + "start": "2023-04-10T00:00:00+00:00", + "end": "2023-04-10T10:00:00+00:00", + }, + } + source_count = reporter_source_count(fresh_db) + with app.test_client() as client: + response = client.post( + url_for("AssetAPI:trigger_report", id=sensors["asset"].id), json=message + ) + assert response.status_code == 403 + assert reporter_source_count(fresh_db) == source_count + assert app.queues["reporting"].jobs == [] + + +@pytest.mark.parametrize( + "requesting_user", ["test_prosumer_user@seita.nl"], indirect=True +) +def test_trigger_report_rejects_output_outside_asset_subtree( + app, fresh_db, setup_report_sensors, clean_redis, requesting_user +): + sensors = setup_report_sensors + message = report_message( + sensors["input_1"], sensors["input_2"], sensors["sibling_output"] + ) + source_count = reporter_source_count(fresh_db) + with app.test_client() as client: + response = client.post( + url_for("AssetAPI:trigger_report", id=sensors["asset"].id), json=message + ) + assert response.status_code == 422 + assert "must belong to asset" in str(response.json) + assert reporter_source_count(fresh_db) == source_count + assert app.queues["reporting"].jobs == [] + + +@pytest.mark.parametrize( + "requesting_user", ["test_prosumer_user@seita.nl"], indirect=True +) +def test_trigger_report_worker_reports_nothing_saved_for_misaligned_inputs( + app, fresh_db, setup_report_sensors, clean_redis, caplog, requesting_user +): + """A report whose inputs do not share a source computes only NaN, which is not saved. + + The job result should exclude rows that persistence drops. + """ + sensors = setup_report_sensors + # An hourly output sensor lets the sum be recorded without resampling, which would drop the NaN rows early. + hourly_output = Sensor( + "hourly report output", + generic_asset=sensors["asset"], + event_resolution=timedelta(hours=1), + unit="kW", + ) + # Two sources, so the inputs do not align on the source level of their belief index. + source_1 = DataSource("report input source 1") + source_2 = DataSource("report input source 2") + fresh_db.session.add_all([hourly_output, source_1, source_2]) + fresh_db.session.flush() + with fresh_db.session.no_autoflush: + beliefs = [ + TimedBelief( + event_start=datetime(2023, 4, 10, hour=hour, tzinfo=timezone.utc), + belief_time=datetime(2023, 4, 9, tzinfo=timezone.utc), + event_value=hour, + sensor=sensor, + source=source, + ) + for sensor, source in ( + (sensors["input_1"], source_1), + (sensors["input_2"], source_2), + ) + for hour in range(10) + ] + fresh_db.session.add_all(beliefs) + fresh_db.session.commit() + + # Sum the inputs without resampling, so that the NaN rows survive to be offered to the database. + message = report_message(sensors["input_1"], sensors["input_2"], hourly_output) + message["config"]["transformations"] = [ + {"df_input": "one", "method": "add", "args": ["@two"], "df_output": "sum"} + ] + with app.test_client() as client: + response = client.post( + url_for("AssetAPI:trigger_report", id=sensors["asset"].id), json=message + ) + assert response.status_code == 202 + + with caplog.at_level(logging.WARNING): + work_on_rq(app.queues["reporting"]) + + stored_report = hourly_output.search_beliefs( + event_starts_after="2023-04-10T00:00:00+00:00", + event_ends_before="2023-04-10T10:00:00+00:00", + ) + assert len(stored_report) == 0, "misaligned inputs cannot produce stored values" + + job = app.queues["reporting"].fetch_job(response.json["job"]) + assert job.get_status() == "finished" + # The count must exclude computed rows that persistence drops as NaN. + assert job.return_value() == [ + { + "sensor_id": hourly_output.id, + "n_rows": 0, + } + ] + assert any( + "produced no persistable values" in record.message for record in caplog.records + ), "an empty report should be warned about" + + +@pytest.mark.parametrize( + "requesting_user", ["test_prosumer_user@seita.nl"], indirect=True +) +def test_trigger_report_worker_fails_for_incompatible_units( + app, fresh_db, setup_report_sensors, clean_redis, requesting_user +): + """A reporter that expects power data must reject a currency input.""" + sensors = setup_report_sensors + sensors["input_1"].unit = "EUR" + fresh_db.session.commit() + + message = report_message(sensors["input_1"], sensors["input_2"], sensors["output"]) + with app.test_client() as client: + response = client.post( + url_for("AssetAPI:trigger_report", id=sensors["asset"].id), json=message + ) + assert response.status_code == 202 + + work_on_rq(app.queues["reporting"]) + + job = app.queues["reporting"].fetch_job(response.json["job"]) + assert job.get_status() == JobStatus.FAILED + assert ( + "Unit conversion from EUR to kW doesn't seem possible" + in job.latest_result().exc_string + ) + + with app.test_client() as client: + status_response = client.get(response.json["job-url"]) + assert status_response.status_code == 422 + assert status_response.json["status"] == "FAILED" + assert "Unit conversion from EUR to kW" in status_response.json["exc-info"] diff --git a/flexmeasures/app.py b/flexmeasures/app.py index c95f9e73e0..05023fd3b2 100644 --- a/flexmeasures/app.py +++ b/flexmeasures/app.py @@ -124,7 +124,11 @@ def create( # noqa C901 name="ingestion", default_timeout=get_job_timeout("ingestion", app.config, app.logger), ), - # reporting=Queue(connection=redis_conn, name="reporting"), + reporting=Queue( + connection=redis_conn, + name="reporting", + default_timeout=get_job_timeout("reporting", app.config, app.logger), + ), # labelling=Queue(connection=redis_conn, name="labelling"), # alerting=Queue(connection=redis_conn, name="alerting"), ) diff --git a/flexmeasures/cli/data_add.py b/flexmeasures/cli/data_add.py index 052b7ae473..451b70ca15 100755 --- a/flexmeasures/cli/data_add.py +++ b/flexmeasures/cli/data_add.py @@ -217,7 +217,7 @@ def new_account_role(name: str, description: str): @click.option( "--trigger-rate-limit", callback=validate_rate_limit_cli, - help="How often accounts on this plan may trigger a schedule or forecast, e.g. '60 per 5 minutes'." + help="How often accounts on this plan may trigger a schedule, forecast or report, e.g. '60 per 5 minutes'." " Defaults to the FLEXMEASURES_API_TRIGGER_RATE_LIMIT setting. Pass 'unlimited' to exempt them.", ) @click.option( @@ -2135,6 +2135,12 @@ def add_schedule( # noqa C901 is_flag=True, help="Add this flag to save the `config` in the attributes of the DataSource for future reference.", ) +@click.option( + "--as-job", + is_flag=True, + help="Whether to queue a reporting job instead of computing directly. " + "Process it with a worker on the 'reporting' queue.", +) def add_report( # noqa: C901 reporter_class: str, source: DataSource | None = None, @@ -2151,11 +2157,25 @@ def add_report( # noqa: C901 edit_parameters: bool = False, save_config: bool = False, timezone: str | None = None, + as_job: bool = False, ): """ Create a new report using the Reporter class and save the results to the database or export them as CSV or Excel file. """ + if as_job and (dry_run or output_file_pattern): + click.secho( + "The --as-job flag cannot be combined with --dry-run or --output-file:" + " the job saves the report to the database only.", + **MsgStyle.ERROR, + ) + raise click.Abort() + if as_job and not save_config: + click.secho( + "Saving the reporter config to its data source (required for --as-job).", + **MsgStyle.WARN, + ) + save_config = True config = dict() @@ -2258,6 +2278,16 @@ def add_report( # noqa: C901 if ("resolution" not in parameters) and (resolution is not None): parameters["resolution"] = pd.Timedelta(resolution).isoformat() + reporter.set_job_trigger("CLI") + + if as_job: + returns = reporter.compute(as_job=True, parameters=parameters) + click.secho( + f"Created reporting job {returns['job_id']} (the report will be saved to the database once processed).", + **MsgStyle.SUCCESS, + ) + return + click.echo("Report computation is running...") # compute the report diff --git a/flexmeasures/cli/jobs.py b/flexmeasures/cli/jobs.py index f4833fe554..804849c3ba 100644 --- a/flexmeasures/cli/jobs.py +++ b/flexmeasures/cli/jobs.py @@ -485,7 +485,7 @@ def inspect_job(job_id: str): "--queue", default=None, required=True, - help="State which queue(s) to work on (using '|' as separator), e.g. 'forecasting', 'scheduling', 'ingestion' or 'forecasting|scheduling'.", + help="State which queue(s) to work on (using '|' as separator), e.g. 'forecasting', 'scheduling', 'ingestion', 'reporting' or 'forecasting|scheduling'.", ) @click.option( "--name", @@ -504,7 +504,7 @@ def inspect_job(job_id: str): ) def run_worker(queue: str, name: str | None, with_scheduler: bool): """ - Start a worker process for forecasting, scheduling and/or ingestion jobs. + Start a worker process for forecasting, scheduling, ingestion and/or reporting jobs. We use the app context to find out which redis queues to use. """ diff --git a/flexmeasures/cli/tests/test_data_add.py b/flexmeasures/cli/tests/test_data_add.py index 7f028188da..9a7a71623b 100644 --- a/flexmeasures/cli/tests/test_data_add.py +++ b/flexmeasures/cli/tests/test_data_add.py @@ -8,6 +8,7 @@ AccountAnnotationRelationship, ) from flexmeasures.data.models.data_sources import DataSource +from flexmeasures.data.models.time_series import Sensor from flexmeasures.data.models.user import Plan, RateLimitKey from flexmeasures.cli.tests.utils import ( @@ -209,6 +210,130 @@ def test_cli_help(app): assert "Usage" in result.output +def test_add_report_as_job(app, fresh_db, setup_dummy_data, clean_redis, tmp_path): + """The report CLI can persist its reporter and queue work for a worker.""" + from flexmeasures.cli.data_add import add_report + + input_1, input_2, output, _ = setup_dummy_data + reporter_config = { + "required_input": [{"name": "one"}, {"name": "two"}], + "required_output": [{"name": "sum"}], + "transformations": [ + { + "df_input": "one", + "method": "add", + "args": ["@two"], + "df_output": "sum", + } + ], + } + config_file = tmp_path / "report-config.json" + config_file.write_text(json.dumps(reporter_config)) + parameters_file = tmp_path / "report-parameters.json" + parameters_file.write_text( + json.dumps( + { + "input": [ + {"name": "one", "sensor": input_1}, + {"name": "two", "sensor": input_2}, + ], + "output": [{"name": "sum", "sensor": output}], + } + ) + ) + + result = app.test_cli_runner().invoke( + add_report, + [ + "--config", + str(config_file), + "--parameters", + str(parameters_file), + "--start", + "2023-04-10T00:00:00+00:00", + "--end", + "2023-04-10T10:00:00+00:00", + "--as-job", + ], + ) + + check_command_ran_without_error(result) + assert "Created reporting job" in result.output + job = app.queues["reporting"].jobs[0] + assert job.timeout == app.queues["reporting"]._default_timeout + assert job.meta["trigger"] == {"origin": "CLI"} + source = fresh_db.session.get(DataSource, job.kwargs["data_source_id"]) + assert source is not None + assert source.attributes["data_generator"]["config"] == { + **reporter_config, + "droplevels": False, + } + + +def test_add_profit_report_as_job_requires_input( + app, fresh_db, setup_dummy_data, clean_redis, tmp_path +): + """The CLI must not queue a profit report without its flow sensor.""" + from flexmeasures.cli.data_add import add_report + + _, _, report_sensor_id, _ = setup_dummy_data + report_sensor = fresh_db.session.get(Sensor, report_sensor_id) + price_sensor = Sensor( + "price sensor", + generic_asset=report_sensor.generic_asset, + event_resolution=report_sensor.event_resolution, + unit="EUR/kWh", + ) + cost_sensor = Sensor( + "cost sensor", + generic_asset=report_sensor.generic_asset, + event_resolution=report_sensor.event_resolution, + unit="EUR", + ) + fresh_db.session.add_all([price_sensor, cost_sensor]) + fresh_db.session.flush() + + config_file = tmp_path / "profit-config.json" + config_file.write_text(json.dumps({"consumption_price_sensor": price_sensor.id})) + parameters_file = tmp_path / "profit-parameters.json" + parameters_file.write_text(json.dumps({"output": [{"sensor": cost_sensor.id}]})) + source_count = fresh_db.session.scalar( + select(func.count()) + .select_from(DataSource) + .where(DataSource.type == "reporter") + ) + + result = app.test_cli_runner().invoke( + add_report, + [ + "--reporter", + "ProfitOrLossReporter", + "--config", + str(config_file), + "--parameters", + str(parameters_file), + "--start", + "2023-04-10T00:00:00+00:00", + "--end", + "2023-04-10T10:00:00+00:00", + "--as-job", + ], + catch_exceptions=True, + ) + + assert result.exit_code != 0 + assert "input" in str(result.exception) + assert app.queues["reporting"].jobs == [] + assert ( + fresh_db.session.scalar( + select(func.count()) + .select_from(DataSource) + .where(DataSource.type == "reporter") + ) + == source_count + ) + + def test_add_forecast_cli_accepts_regressor_ids_and_json_reference_lists( app, fresh_db, diff --git a/flexmeasures/data/models/data_sources.py b/flexmeasures/data/models/data_sources.py index 61abcd035b..9f1a9b7aab 100644 --- a/flexmeasures/data/models/data_sources.py +++ b/flexmeasures/data/models/data_sources.py @@ -22,6 +22,7 @@ from marshmallow import Schema if TYPE_CHECKING: + from flexmeasures.data.models.time_series import Sensor from flexmeasures.data.models.user import User @@ -125,7 +126,7 @@ def output_sensors(self) -> list: return [] @staticmethod - def _resolve_sensors(*values) -> list: + def _resolve_sensors(*values) -> list[Sensor]: """Turn (lists of) sensors, sensor references or sensor IDs into a list of unique sensors. A sensor reference contributes the sensor it wraps, as the source filters only narrow down diff --git a/flexmeasures/data/models/reporting/__init__.py b/flexmeasures/data/models/reporting/__init__.py index 56ffb265fa..c116c29f2c 100644 --- a/flexmeasures/data/models/reporting/__init__.py +++ b/flexmeasures/data/models/reporting/__init__.py @@ -19,14 +19,40 @@ class Reporter(DataGenerator): _parameters_schema = ReporterParametersSchema() _config_schema = ReporterConfigSchema() - def _compute(self, check_output_resolution=True, **kwargs) -> list[dict[str, Any]]: + @property + def input_sensors(self) -> list: + """Return the sensors from which the report reads input data.""" + parameters = self._parameters or {} + return self._resolve_sensors( + [item.get("sensor") for item in parameters.get("input", [])] + ) + + @property + def output_sensors(self) -> list: + """Return the sensors on which the report records its results.""" + parameters = self._parameters or {} + return self._resolve_sensors( + [item.get("sensor") for item in parameters.get("output", [])] + ) + + def _compute( + self, check_output_resolution=True, as_job: bool = False, **kwargs + ) -> list[dict[str, Any]] | dict[str, Any]: """This method triggers the creation of a new report. The same object can generate multiple reports with different start, end, resolution and belief_time values. :param check_output_resolution: If True, checks each output for whether the event_resolution matches that of the sensor it is supposed to be recorded on. + :param as_job: If True, queue a reporting job instead of computing immediately. + :returns: A dictionary with ``job_id`` and ``n_jobs`` when queued, + otherwise the computed report results. """ + if as_job: + from flexmeasures.data.services.reporting import create_reporting_job + + job = create_reporting_job(self) + return {"job_id": job.id, "n_jobs": 1} results = self._compute_report(**kwargs) diff --git a/flexmeasures/data/models/reporting/profit.py b/flexmeasures/data/models/reporting/profit.py index 3503bdcc69..b5543f85a4 100644 --- a/flexmeasures/data/models/reporting/profit.py +++ b/flexmeasures/data/models/reporting/profit.py @@ -47,6 +47,15 @@ class ProfitOrLossReporter(Reporter): weights: dict method: str + @property + def input_sensors(self) -> list: + """Return the flow and price sensors read by this reporter.""" + return self._resolve_sensors( + super().input_sensors, + self._config.get("consumption_price_sensor"), + self._config.get("production_price_sensor"), + ) + def _compute_report( self, start: datetime, diff --git a/flexmeasures/data/schemas/reporting/__init__.py b/flexmeasures/data/schemas/reporting/__init__.py index 02e580ae42..844f411fb4 100644 --- a/flexmeasures/data/schemas/reporting/__init__.py +++ b/flexmeasures/data/schemas/reporting/__init__.py @@ -29,7 +29,9 @@ class ReporterParametersSchema(Schema): validate=validate.Length(min=1), ) - output = fields.List(fields.Nested(Output()), validate=validate.Length(min=1)) + output = fields.List( + fields.Nested(Output()), required=True, validate=validate.Length(min=1) + ) start = AwareDateTimeField(required=True) end = AwareDateTimeField(required=True) @@ -40,6 +42,18 @@ class ReporterParametersSchema(Schema): belief_horizon = DurationField(required=False) +class ReportTriggerSchema(Schema): + """Validate the request envelope for a one-off reporting job. + + The selected reporter subsequently validates ``config`` and ``parameters`` + with its concrete configuration and parameter schemas. + """ + + reporter = fields.Str(required=True) + config = fields.Dict(keys=fields.Str(), load_default=dict) + parameters = fields.Dict(keys=fields.Str(), required=True) + + class BeliefsSearchConfigSchema(Schema): """ This schema implements the required fields to perform a TimedBeliefs search diff --git a/flexmeasures/data/schemas/reporting/aggregation.py b/flexmeasures/data/schemas/reporting/aggregation.py index 165e646fb5..5ac6bc6cd1 100644 --- a/flexmeasures/data/schemas/reporting/aggregation.py +++ b/flexmeasures/data/schemas/reporting/aggregation.py @@ -56,5 +56,7 @@ class AggregatorParametersSchema(ReporterParametersSchema): # redefining output to restrict the output length to 1 output = fields.List( - fields.Nested(Output()), validate=validate.Length(min=1, max=1) + fields.Nested(Output()), + required=True, + validate=validate.Length(min=1, max=1), ) diff --git a/flexmeasures/data/schemas/reporting/profit.py b/flexmeasures/data/schemas/reporting/profit.py index 2cae5069b9..4e01ba5a34 100644 --- a/flexmeasures/data/schemas/reporting/profit.py +++ b/flexmeasures/data/schemas/reporting/profit.py @@ -94,8 +94,12 @@ class ProfitOrLossReporterParametersSchema(ReporterParametersSchema): } """ - # redefining output to restrict the input length to 1 - input = fields.List(fields.Nested(Input()), validate=validate.Length(min=1, max=1)) + # redefining input to restrict the input length to 1 + input = fields.List( + fields.Nested(Input()), + required=True, + validate=validate.Length(min=1, max=1), + ) @validates("input") def validate_input_measures_power_energy(self, value, **kwargs): diff --git a/flexmeasures/data/schemas/tests/test_reporting.py b/flexmeasures/data/schemas/tests/test_reporting.py index 337f8884d3..c669609d6d 100644 --- a/flexmeasures/data/schemas/tests/test_reporting.py +++ b/flexmeasures/data/schemas/tests/test_reporting.py @@ -6,6 +6,9 @@ ProfitOrLossReporterConfigSchema, ProfitOrLossReporterParametersSchema, ) +from flexmeasures.data.schemas.reporting.aggregation import ( + AggregatorParametersSchema, +) from flexmeasures.data.schemas.reporting import BeliefsSearchConfigSchema from marshmallow.exceptions import ValidationError @@ -231,6 +234,31 @@ def test_profit_reporter_parameters_schema( schema.load(parameters) +def test_specialized_reporter_schemas_preserve_required_dataflow_fields( + db, app, setup_dummy_sensors +): + """Overridden input/output fields remain required by the reporter contract.""" + with pytest.raises(ValidationError) as aggregator_error: + AggregatorParametersSchema().load( + { + "input": [{"sensor": 1}], + "start": start, + "end": end, + } + ) + assert "output" in aggregator_error.value.messages + + with pytest.raises(ValidationError) as profit_error: + ProfitOrLossReporterParametersSchema().load( + { + "output": [{"sensor": 3}], + "start": start, + "end": end, + } + ) + assert "input" in profit_error.value.messages + + @pytest.mark.parametrize( "make_config, is_valid", [ diff --git a/flexmeasures/data/services/data_generators.py b/flexmeasures/data/services/data_generators.py new file mode 100644 index 0000000000..a4eda1f476 --- /dev/null +++ b/flexmeasures/data/services/data_generators.py @@ -0,0 +1,42 @@ +"""Authorization helpers shared by data-generator entry points.""" + +from __future__ import annotations + +from copy import copy + +from werkzeug.exceptions import Forbidden + +from flexmeasures.auth.policy import check_access +from flexmeasures.data.models.data_sources import DataGenerator +from flexmeasures.data.models.time_series import Sensor + + +def resolve_data_generator_sensors( + data_generator: DataGenerator, deserialized_parameters: dict +) -> dict[str, list[Sensor]]: + """Return the sensors a data generator would read from and write to.""" + data_generator = copy(data_generator) + data_generator._parameters = deserialized_parameters + return { + "input_sensors": data_generator.input_sensors, + "output_sensors": data_generator.output_sensors, + } + + +def check_sensor_access( + input_sensors: list[Sensor], output_sensors: list[Sensor] +) -> None: + """Require read access to inputs and create-children access to outputs.""" + for sensors, permission, action in ( + (input_sensors, "read", "read data from"), + (output_sensors, "create-children", "record data on"), + ): + for sensor in sensors: + try: + check_access(sensor, permission) + except Forbidden as exc: + exc.api_message = ( + f"You cannot request this computation because it would {action}" + f" sensor {sensor.id}, which you cannot {action} yourself." + ) + raise diff --git a/flexmeasures/data/services/reporting.py b/flexmeasures/data/services/reporting.py new file mode 100644 index 0000000000..2e79b02d7a --- /dev/null +++ b/flexmeasures/data/services/reporting.py @@ -0,0 +1,120 @@ +"""Logic for queueing and running reporting jobs.""" + +from __future__ import annotations + +from datetime import timedelta +from typing import TYPE_CHECKING + +from flask import current_app +from rq.job import Job + +from flexmeasures.data import db +from flexmeasures.data.schemas.reporting import ReporterParametersSchema +from flexmeasures.data.utils import save_to_db + +if TYPE_CHECKING: + from flexmeasures.data.models.reporting import Reporter + + +def create_reporting_job(reporter: "Reporter", queue: str = "reporting") -> Job: + """Queue a job that computes a report and stores its results.""" + parameters = reporter._parameters_schema.dump(reporter._parameters) + ReporterParametersSchema(only=("input", "output")).load( + { + field: parameters[field] + for field in ("input", "output") + if field in parameters + } + ) + output_sensor_ids = [output["sensor"] for output in parameters["output"]] + + # The reporting worker runs in a separate process, so the data source has to be committed before the job is enqueued. + # Note that reporter.data_source may have only just created it, and that our views do not auto-commit. + reporter._data_source = db.session.merge(reporter.data_source) + db.session.flush() + data_source_id = reporter._data_source.id + db.session.commit() + + job_metadata = { + "data_source_info": {"id": data_source_id}, + "start": parameters.get("start"), + "end": parameters.get("end"), + "sensor_id": output_sensor_ids[0], + } + if reporter._job_trigger: + job_metadata["trigger"] = reporter._job_trigger + + job = Job.create( + run_report_job, + kwargs={"data_source_id": data_source_id, "parameters": parameters}, + connection=current_app.queues[queue].connection, + ttl=int( + current_app.config.get( + "FLEXMEASURES_JOB_TTL", timedelta(-1) + ).total_seconds() + ), + result_ttl=int( + current_app.config.get( + "FLEXMEASURES_PLANNING_TTL", timedelta(-1) + ).total_seconds() + ), + meta=job_metadata, + ) + current_app.queues[queue].enqueue_job(job) + for sensor_id in output_sensor_ids: + current_app.job_cache.add( + sensor_id, + job_id=job.id, + queue=queue, + asset_or_sensor_type="sensor", + ) + return job + + +def _count_persistable_values(data) -> int: + """Count computed values that will not be dropped as NaN before persistence. + + This does not account for valid values that ``save_to_db`` may skip because + they are unchanged. + """ + from timely_beliefs import BeliefsSeries + + if isinstance(data, BeliefsSeries): + return int(data.notna().sum()) + return len(data.dropna(subset=["event_value"])) + + +def run_report_job(data_source_id: int, parameters: dict) -> list[dict]: + """Compute and store a report in a reporting worker.""" + from flexmeasures.data.models.data_sources import DataSource + from flexmeasures.data.models.reporting import Reporter + + source = db.session.get(DataSource, data_source_id) + if source is None: + raise ValueError(f"Data source {data_source_id} no longer exists.") + reporter = source.data_generator + if not isinstance(reporter, Reporter): + raise ValueError(f"Data source {data_source_id} does not store a Reporter.") + reporter._parameters = None + results = reporter.compute(parameters=parameters) + saved = [] + for result in results: + n_rows = _count_persistable_values(result["data"]) + save_to_db(result["data"]) + saved.append({"sensor_id": result["sensor"].id, "n_rows": n_rows}) + db.session.commit() + + summary = ", ".join( + f"{result['n_rows']} values on sensor {result['sensor_id']}" for result in saved + ) + if any(result["n_rows"] for result in saved): + current_app.logger.info( + "Report by %s ran successfully, producing %s.", source, summary + ) + else: + current_app.logger.warning( + "Report by %s produced no persistable values (%s). This can happen when its inputs do not align on source and belief time.", + source, + summary, + ) + return saved diff --git a/flexmeasures/data/services/sensors.py b/flexmeasures/data/services/sensors.py index 07846f69a5..314bf7f66d 100644 --- a/flexmeasures/data/services/sensors.py +++ b/flexmeasures/data/services/sensors.py @@ -800,6 +800,15 @@ def build_asset_jobs_data( current_app.job_cache.get(sensor.id, "forecasting", "sensor"), ) ) + jobs.append( + ( + "reporting", + "sensor", + sensor.id, + sensor.name, + current_app.job_cache.get(sensor.id, "reporting", "sensor"), + ) + ) jobs_data = list() # Building the actual return list - we also unpack lists of jobs, each to its own entry, and we add error info @@ -814,7 +823,7 @@ def build_asset_jobs_data( ), ) job_err = ( - f"Scheduling job failed with {type(e).__name__}: {e}" + f"{queue.capitalize()} job failed with {type(e).__name__}: {e}" if job.is_failed else None ) diff --git a/flexmeasures/data/tests/test_reporting_service.py b/flexmeasures/data/tests/test_reporting_service.py new file mode 100644 index 0000000000..cbfcb9a10f --- /dev/null +++ b/flexmeasures/data/tests/test_reporting_service.py @@ -0,0 +1,28 @@ +from marshmallow import Schema, ValidationError +import pytest + +from flexmeasures.data.services.reporting import create_reporting_job + + +class EmptyParametersSchema(Schema): + pass + + +class ReporterWithoutDataflow: + _parameters_schema = EmptyParametersSchema() + _parameters = {} + + @property + def data_source(self): + raise AssertionError( + "The data source was accessed before parameter validation." + ) + + +def test_create_reporting_job_validates_dataflow_before_accessing_source(): + reporter = ReporterWithoutDataflow() + + with pytest.raises(ValidationError) as exc_info: + create_reporting_job(reporter) # type: ignore[arg-type] + + assert set(exc_info.value.messages) == {"input", "output"} diff --git a/flexmeasures/ui/static/openapi-specs.json b/flexmeasures/ui/static/openapi-specs.json index 4f156b8c06..71c358493c 100644 --- a/flexmeasures/ui/static/openapi-specs.json +++ b/flexmeasures/ui/static/openapi-specs.json @@ -4542,6 +4542,96 @@ ] } }, + "/api/v3_0/assets/{id}/reports/trigger": { + "post": { + "summary": "Trigger a one-off reporting job for this asset.", + "description": "Queue a one-off report for a worker processing the `reporting` queue.\nThe caller must be able to read every input/configuration sensor and\nrecord data on every output sensor. Each output must belong to the\nasset in the URL or one of its descendants.\n", + "security": [ + { + "ApiKeyAuth": [] + } + ], + "parameters": [ + { + "in": "path", + "name": "id", + "required": true, + "$ref": "#/components/parameters/AssetIdPath" + } + ], + "requestBody": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ReportTriggerSchema" + } + } + } + }, + "responses": { + "202": { + "description": "ACCEPTED", + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "status", + "message", + "job", + "job-url" + ], + "properties": { + "status": { + "type": "string", + "enum": [ + "ACCEPTED" + ] + }, + "message": { + "type": "string" + }, + "job": { + "type": "string", + "description": "UUID of the queued reporting job." + }, + "job-url": { + "type": "string", + "format": "uri", + "description": "URL to query the generic job status API." + } + } + }, + "example": { + "status": "ACCEPTED", + "message": "Request has been accepted for processing.", + "job": "364bfd06-c1fa-430b-8d25-8f5a547651fb", + "job-url": "/api/v3_0/jobs/364bfd06-c1fa-430b-8d25-8f5a547651fb" + } + } + } + }, + "400": { + "description": "INVALID_REQUEST, REQUIRED_INFO_MISSING, UNEXPECTED_PARAMS" + }, + "401": { + "description": "UNAUTHORIZED" + }, + "403": { + "description": "INVALID_SENDER" + }, + "422": { + "description": "UNPROCESSABLE_ENTITY" + }, + "429": { + "description": "TOO_MANY_REQUESTS - You called the API, or triggered computation, more often than your rate limits allow. Triggering endpoints share a stricter limit than the rest of the API. Wait for as long as the Retry-After header says, then try again." + } + }, + "tags": [ + "Assets" + ] + } + }, "/api/v3_0/assets/{id}/schedules/trigger": { "post": { "summary": "Trigger scheduling job for any number of devices", @@ -6296,6 +6386,27 @@ ], "additionalProperties": false }, + "ReportTriggerSchema": { + "type": "object", + "properties": { + "reporter": { + "type": "string" + }, + "config": { + "type": "object", + "additionalProperties": {} + }, + "parameters": { + "type": "object", + "additionalProperties": {} + } + }, + "required": [ + "parameters", + "reporter" + ], + "additionalProperties": false + }, "CopyAssetSchema": { "type": "object", "properties": { diff --git a/flexmeasures/utils/job_utils.py b/flexmeasures/utils/job_utils.py index 3054638a26..bc95860dcf 100644 --- a/flexmeasures/utils/job_utils.py +++ b/flexmeasures/utils/job_utils.py @@ -12,7 +12,7 @@ from rq.job import Job RQ_DEFAULT_JOB_TIMEOUT = 180 -KNOWN_JOB_QUEUES = frozenset(("forecasting", "scheduling", "ingestion")) +KNOWN_JOB_QUEUES = frozenset(("forecasting", "scheduling", "ingestion", "reporting")) def _timeout_to_seconds(timeout: timedelta | str) -> int: diff --git a/flexmeasures/utils/tests/test_job_utils.py b/flexmeasures/utils/tests/test_job_utils.py index 27ef8cb5b3..10b820f1ab 100644 --- a/flexmeasures/utils/tests/test_job_utils.py +++ b/flexmeasures/utils/tests/test_job_utils.py @@ -137,6 +137,7 @@ def test_app_queues_use_default_job_timeout(app): assert app.queues["forecasting"]._default_timeout == 180 assert app.queues["scheduling"]._default_timeout == 180 assert app.queues["ingestion"]._default_timeout == 180 + assert app.queues["reporting"]._default_timeout == 180 def test_app_queues_use_custom_global_and_queue_job_timeout(tmp_path, monkeypatch): @@ -164,3 +165,4 @@ def test_app_queues_use_custom_global_and_queue_job_timeout(tmp_path, monkeypatc assert custom_app.queues["forecasting"]._default_timeout == 3600 assert custom_app.queues["scheduling"]._default_timeout == 300 assert custom_app.queues["ingestion"]._default_timeout == 300 + assert custom_app.queues["reporting"]._default_timeout == 300