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 d715ca1624..3489628bdc 100644 --- a/documentation/api/change_log.rst +++ b/documentation/api/change_log.rst @@ -7,6 +7,7 @@ API change log v3.0-32 | August 11, 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. - API endpoints are now rate-limited. A request which exceeds a limit is answered with a ``429 (Too Many Requests)`` status code and a ``Retry-After`` header stating how many seconds to wait. Responses also carry ``X-RateLimit-*`` headers, describing the limit that applied, how much of it is left, and when it resets. A stricter limit applies to ``POST /assets//schedules/trigger``, ``POST /sensors//schedules/trigger`` and ``POST /sensors//forecasts/trigger`` than to other endpoints; the health endpoints are exempt. Per-account overrides are set by assigning the account a plan (a ``Plan`` database row), rather than through an account attribute. - Introduced the ``inflexible-consumption`` and ``inflexible-production`` flex-context fields, which make explicit how the sign of each inflexible device's power data should be read: positive values denote consumption resp. production. Each entry is a sensor reference (``{"sensor": }``), optionally with source filters (``source-types``, ``exclude-source-types``, ``sources``, ``source-account``). Deprecated the ``inflexible-device-sensors`` field (a list of bare sensor IDs, whose sign convention is read from each sensor's ``consumption_is_positive`` attribute); it remains supported, but cannot be combined with the new fields in one flex-context. - Added a ``role`` query parameter to ``GET /api/v3_0/accounts`` for filtering accessible organisations by account role. diff --git a/documentation/changelog.rst b/documentation/changelog.rst index 9e703ec80f..734fa50930 100644 --- a/documentation/changelog.rst +++ b/documentation/changelog.rst @@ -11,6 +11,8 @@ 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 `_ and `issue #2426 `_] + Infrastructure / Support ------------------------- diff --git a/documentation/cli/change_log.rst b/documentation/cli/change_log.rst index 7287db1578..62501d837c 100644 --- a/documentation/cli/change_log.rst +++ b/documentation/cli/change_log.rst @@ -7,6 +7,7 @@ FlexMeasures CLI Changelog since v1.0.0 | August 11, 2026 ================================= +* Add ``flexmeasures add report --as-job`` and the ``reporting`` worker queue for asynchronous one-off reports. * ``flexmeasures add user --roles`` now parses comma-separated role lists and accepts repeated ``--roles`` options. * ``flexmeasures db upgrade`` now runs ``VACUUM ANALYZE`` after upgrading (refreshing the query planner's statistics); opt out with ``--no-vacuum``. * Add ``flexmeasures add plan``, ``flexmeasures show plans`` and ``flexmeasures edit plan``, to manage the rate limits and quotas which apply to the accounts on a plan. diff --git a/documentation/configuration.rst b/documentation/configuration.rst index 6ea3081a67..9a327c5c57 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 31c879634a..589b6d0743 100644 --- a/flexmeasures/api/common/utils/api_utils.py +++ b/flexmeasures/api/common/utils/api_utils.py @@ -25,6 +25,7 @@ add_beliefs_to_db_and_enqueue_forecasting_jobs, ) from flexmeasures.data.models.generic_assets import GenericAsset +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, @@ -570,23 +571,6 @@ def _determine_copy_name( return f"{source_name} (Copy {max_index + 1})" -def _asset_is_in_subtree(root_asset_id: int, candidate_asset_id: int) -> bool: - """Return True if candidate_asset_id is root or a descendant of root_asset_id.""" - current_asset_id = candidate_asset_id - visited: set[int] = set() - - while current_asset_id is not None and current_asset_id not in visited: - if current_asset_id == root_asset_id: - return True - visited.add(current_asset_id) - current_asset = db.session.get(GenericAsset, current_asset_id) - if current_asset is None: - return False - current_asset_id = current_asset.parent_asset_id - - return False - - def copy_asset( asset: GenericAsset, account=None, @@ -634,7 +618,7 @@ def copy_asset( target_account_id = int(account.id) target_parent_asset_id = int(parent_asset.id) - if target_parent_asset_id is not None and _asset_is_in_subtree( + if target_parent_asset_id is not None and asset_is_in_subtree( root_asset_id=asset.id, candidate_asset_id=target_parent_asset_id, ): 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 0668cde606..8babad2046 100644 --- a/flexmeasures/api/v3_0/assets.py +++ b/flexmeasures/api/v3_0/assets.py @@ -49,13 +49,21 @@ from flexmeasures.data.models.user import Account from flexmeasures.data.models.audit_log import AssetAuditLog 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, @@ -1571,6 +1579,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, + 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") + returns = 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(returns["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..d2142c2275 --- /dev/null +++ b/flexmeasures/api/v3_0/tests/test_report_trigger_api.py @@ -0,0 +1,366 @@ +"""Tests for POST /api/v3_0/assets//reports/trigger.""" + +from datetime import datetime, timedelta, timezone + +import pytest +from flask import url_for +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) + ) + 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=setup_accounts_fresh_db["Prosumer"], + 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: + return { + "reporter": "PandasReporter", + "config": { + "required_input": [{"name": "one"}, {"name": "two"}], + "required_output": [{"name": "sum"}], + "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, 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 + + work_on_rq(app.queues["reporting"]) + 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", + ) + 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 == [] 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 deae66ee25..efe9e6f86c 100755 --- a/flexmeasures/cli/data_add.py +++ b/flexmeasures/cli/data_add.py @@ -207,7 +207,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( @@ -1834,6 +1834,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, @@ -1850,11 +1856,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() @@ -1957,6 +1977,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 27be18f4a7..15b6960927 100644 --- a/flexmeasures/cli/jobs.py +++ b/flexmeasures/cli/jobs.py @@ -355,7 +355,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", @@ -374,7 +374,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 fa79824612..d7ea272c7b 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,129 @@ 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 + config_file = tmp_path / "report-config.json" + config_file.write_text( + json.dumps( + { + "required_input": [{"name": "one"}, {"name": "two"}], + "required_output": [{"name": "sum"}], + "transformations": [ + { + "df_input": "one", + "method": "add", + "args": ["@two"], + "df_output": "sum", + } + ], + } + ) + ) + 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.attributes["data_generator"]["config"]["required_input"] + + +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 9a25b76b18..63ddcc5ca1 100644 --- a/flexmeasures/data/models/data_sources.py +++ b/flexmeasures/data/models/data_sources.py @@ -31,6 +31,7 @@ class DataGenerator: _config: dict = None _parameters: dict = None + _job_trigger: dict | None = None _parameters_schema: Schema | None = None _config_schema: Schema | None = None @@ -98,6 +99,45 @@ def __init__( elif len(kwargs) == 0: self._config = self._config_schema.load({}) + def set_job_trigger(self, origin: str): + """Record how a queued job was created, for storage in its metadata.""" + self._job_trigger = {"origin": origin} + + @property + def input_sensors(self) -> list: + """Return the sensors from which this data generator reads data.""" + return [] + + @property + def output_sensors(self) -> list: + """Return the sensors on which this data generator records data.""" + return [] + + @staticmethod + def _resolve_sensors(*values) -> list: + """Turn sensors, sensor references or sensor IDs into unique sensors.""" + from flexmeasures.data.models.time_series import Sensor + from flexmeasures.data.schemas.sensors import SensorReference + + sensors: dict[int, Sensor] = {} + for value in values: + if value is None: + continue + for item in value if isinstance(value, (list, tuple, set)) else [value]: + if isinstance(item, SensorReference): + sensor = item.sensor + elif isinstance(item, Sensor): + sensor = item + elif isinstance(item, int) or ( + isinstance(item, str) and item.isdigit() + ): + sensor = db.session.get(Sensor, int(item)) + else: + continue + if sensor is not None: + sensors[sensor.id] = sensor + return list(sensors.values()) + def _compute(self, **kwargs) -> list[dict[str, Any]]: """Overwrite with the actual computation of your data generator. diff --git a/flexmeasures/data/models/reporting/__init__.py b/flexmeasures/data/models/reporting/__init__.py index 56ffb265fa..a49a74da35 100644 --- a/flexmeasures/data/models/reporting/__init__.py +++ b/flexmeasures/data/models/reporting/__init__.py @@ -19,14 +19,38 @@ 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 and return its ID. """ + 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/queries/generic_assets.py b/flexmeasures/data/queries/generic_assets.py index ab164cc012..31e0674637 100644 --- a/flexmeasures/data/queries/generic_assets.py +++ b/flexmeasures/data/queries/generic_assets.py @@ -17,6 +17,23 @@ from flexmeasures.utils.flexmeasures_inflection import pluralize +def asset_is_in_subtree(root_asset_id: int, candidate_asset_id: int) -> bool: + """Return whether an asset is the given root or one of its descendants.""" + current_asset_id = candidate_asset_id + visited: set[int] = set() + + while current_asset_id is not None and current_asset_id not in visited: + if current_asset_id == root_asset_id: + return True + visited.add(current_asset_id) + current_asset = db.session.get(GenericAsset, current_asset_id) + if current_asset is None: + return False + current_asset_id = current_asset.parent_asset_id + + return False + + def query_assets_by_type( type_names: list[str] | str, account_id: int | None = None, 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..be8700de8c --- /dev/null +++ b/flexmeasures/data/services/reporting.py @@ -0,0 +1,91 @@ +"""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"]] + + 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 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) + for result in results: + save_to_db(result["data"]) + db.session.commit() + return [ + {"sensor_id": result["sensor"].id, "n_rows": len(result["data"])} + for result in results + ] diff --git a/flexmeasures/data/services/sensors.py b/flexmeasures/data/services/sensors.py index abb7844178..9639425fef 100644 --- a/flexmeasures/data/services/sensors.py +++ b/flexmeasures/data/services/sensors.py @@ -786,6 +786,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 @@ -800,7 +809,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 7e3ab548a7..453c19f9b7 100644 --- a/flexmeasures/ui/static/openapi-specs.json +++ b/flexmeasures/ui/static/openapi-specs.json @@ -7,7 +7,7 @@ }, "termsOfService": null, "title": "FlexMeasures", - "version": "1.0.0" + "version": "1.0.0rc5" }, "externalDocs": { "description": "FlexMeasures runs on the open source FlexMeasures technology. Read the docs here.", @@ -4294,6 +4294,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", @@ -5985,6 +6075,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": { @@ -7278,4 +7389,4 @@ } } } -} +} \ No newline at end of file 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