diff --git a/documentation/api/change_log.rst b/documentation/api/change_log.rst index 686db20ff0..e1cdc7959a 100644 --- a/documentation/api/change_log.rst +++ b/documentation/api/change_log.rst @@ -5,6 +5,10 @@ API change log .. note:: The FlexMeasures API follows its own versioning scheme. This is also reflected in the URL (e.g. `/api/v3_0`), allowing developers to upgrade at their own pace. +v3.0-36 | September 11, 2026 +"""""""""""""""""""""""""""" +- Added ``POST /api/v3_0/assets//automations``, ``PATCH /api/v3_0/assets//automations/`` and ``DELETE /api/v3_0/assets//automations/`` for managing an asset's automations. They require the same permission as writing data under the asset, and an automation may only involve sensors that its creator can access: read access to the sensors it reads data from, and permission to record data on the sensors it writes to (a ``403`` otherwise). Both the creation and the update accept a ``timezone``, in which the automation's cron expression is interpreted; it defaults to the server's ``FLEXMEASURES_TIMEZONE``. + v3.0-35 | September 9, 2026 """"""""""""""""""""""""""" - The ``resolution`` field is now rejected with a ``422 (Unprocessable Entity)`` response unless it spans a positive amount of time. This applies wherever the API accepts one: as a query parameter on ``GET /api/v3_0/sensors//data`` and on the ``chart_data`` endpoints under ``api/dev``, and in the request body of the ``POST`` schedule trigger endpoints. Previously, a zero resolution (such as ``PT0S``) either crashed the request with a ``500`` or was silently ignored, and a negative resolution returned an empty set of values. diff --git a/documentation/changelog.rst b/documentation/changelog.rst index 813fb1f7c1..e59139f896 100644 --- a/documentation/changelog.rst +++ b/documentation/changelog.rst @@ -19,10 +19,18 @@ v1.1.0 | September XX, 2026 Select a data source to see the schedule computed under one configuration. One scheduling request still records under a single data source, including the per-device jobs of a sequential schedule. +.. warning:: A scheduler's data source now also records the flex config it computed under, where previously one data source per scheduler version recorded every schedule that scheduler made. + Schedules computed under different flex configs are therefore recorded by different data sources, and a sensor can carry schedules from several of them, as it already could for forecasts. + Values describing a single moment stay out of that config, so a ``soc-at-start``, or a ``soc-targets`` entry at a given datetime, does not make every run a new data source. + What does is a change to what the site and its devices can do, such as a device's ``power-capacity``. + After such a change, a sensor holds the schedule computed under each configuration, where the newer schedule used to supersede the older one, so a chart of that sensor draws both, and the asset's KPIs total both, as they report what the chart draws. + Select a data source to see the schedule computed under one configuration. + One scheduling request still records under a single data source, including the per-device jobs of a sequential schedule. + New features ------------- -* Automations: recurring tasks defined per asset, computing forecasts or schedules, managed with new CLI commands (``flexmeasures add|edit|delete automation``), run by ``flexmeasures jobs run-automations``, and viewable in a new UI page and API endpoints (``[GET] /assets/(id)/automations``); each automation interprets its recurrence in its own timezone, and runs missed while the runner was down are caught up once, coalesced into one current forecast; a forecast automation points at a data source holding its forecaster configuration, while a schedule automation stores what the schedule trigger endpoint accepts, and schedules from each run's own time, unless the trigger message fixes a ``start``; an automation's details link to the sensors it reads from and writes to, a sensor's page lists the automations feeding it, and deleting a sensor warns about the automations that use it; jobs now also record whether they were created via the CLI, the API or an automation [see `PR #2290 `_, `PR #2396 `_ and `PR #2293 `_] +* Automations: recurring tasks defined per asset, computing forecasts or schedules, managed with new CLI commands (``flexmeasures add|edit|delete automation``), run by ``flexmeasures jobs run-automations``, and viewable in a new UI page and API endpoints (``[GET] /assets/(id)/automations``); each automation interprets its recurrence in its own timezone, and runs missed while the runner was down are caught up once, coalesced into one current forecast; a forecast automation points at a data source holding its forecaster configuration, while a schedule automation stores what the schedule trigger endpoint accepts, and schedules from each run's own time, unless the trigger message fixes a ``start``; an automation's details link to the sensors it reads from and writes to, a sensor's page lists the automations feeding it, and deleting a sensor warns about the automations that use it; jobs now also record whether they were created via the CLI, the API or an automation; automations can also be created, edited and deleted in the UI and through new API endpoints (``[POST|PATCH|DELETE] /assets/(id)/automations``), by whoever may add data under the asset, with their recurrence expressed in a selectable IANA timezone, and only involving sensors they can access themselves (read access to the sensors an automation reads, and permission to record data on the sensors it writes to) [see `PR #2290 `_, `PR #2396 `_, `PR #2293 `_ and `PR #2294 `_] * A scheduler's data source now also records the flex config the scheduler computed under, so a schedule can be traced back to the configuration that produced it, and a schedule automation points at such a data source, the way a forecast automation points at its forecaster's [see `PR #2444 `_] * In the UI, the full record of the data source selected on a sensor page can be inspected, backed by a new API endpoint (``[GET] /sources/(id)``) [see `PR #2290 `_] * Try out a forecast without recording it, using ``flexmeasures add forecasts --dry-run``, which computes the forecast in full and reports the sensor, data source, number of beliefs and event range it would have saved [see `PR #2483 `_] diff --git a/flexmeasures/api/v3_0/__init__.py b/flexmeasures/api/v3_0/__init__.py index 835755a76b..03ed7e44f0 100644 --- a/flexmeasures/api/v3_0/__init__.py +++ b/flexmeasures/api/v3_0/__init__.py @@ -45,6 +45,10 @@ StatusPageChildJobsJSONSchema, ) from flexmeasures.data.schemas.annotations import AnnotationSchema +from flexmeasures.data.schemas.automations import ( + AutomationCreationSchema, + AutomationUpdateSchema, +) from flexmeasures.data.schemas.generic_assets import GenericAssetSchema as AssetSchema from flexmeasures.data.schemas.reporting import ReportTriggerSchema from flexmeasures.data.schemas.sensors import QuantitySchema, TimeSeriesSchema @@ -228,6 +232,8 @@ def create_openapi_specs(app: Flask): ("AssetAPIQuerySchema", AssetAPIQuerySchema), ("AssetSchema", AssetSchema), ("AnnotationSchema", AnnotationSchema), + ("AutomationCreationSchema", AutomationCreationSchema), + ("AutomationUpdateSchema", AutomationUpdateSchema), ("ReportTriggerSchema", ReportTriggerSchema), ("CopyAssetSchema", CopyAssetSchema), ("DefaultAssetViewJSONSchema", DefaultAssetViewJSONSchema), diff --git a/flexmeasures/api/v3_0/assets.py b/flexmeasures/api/v3_0/assets.py index 7fab5ec99e..b88822b5c4 100644 --- a/flexmeasures/api/v3_0/assets.py +++ b/flexmeasures/api/v3_0/assets.py @@ -48,13 +48,20 @@ from flexmeasures.data.models.automations import Automation from flexmeasures.data.models.user import Account from flexmeasures.data.models.audit_log import AssetAuditLog -from flexmeasures.data.schemas.automations import AutomationSchema +from flexmeasures.data.schemas.automations import ( + AutomationCreationSchema, + AutomationSchema, + AutomationUpdateSchema, +) from flexmeasures.data.services.automations import ( AutomationSensorsUnknown, + create_automation, + delete_automation as remove_automation, describe_cronstr, get_automation_job_stats, resolve_automation_sensors, run_automation, + update_automation, ) from flexmeasures.data.models.generic_assets import GenericAsset, GenericAssetType from flexmeasures.data.models.reporting import Reporter @@ -1476,7 +1483,7 @@ def get_automations(self, id: int, asset: GenericAsset): - id: 1 created_at: "2026-07-11T00:00:00+00:00" asset_id: 1 - type: forecasts + type: forecasting name: Day-ahead PV forecasts cronstr: "0 6 * * *" timezone: Europe/Amsterdam @@ -1556,7 +1563,7 @@ def get_automation(self, id: int, automation_id: int, asset: GenericAsset): id: 1 created_at: "2026-07-11T00:00:00+00:00" asset_id: 1 - type: forecasts + type: forecasting name: Day-ahead PV forecasts cronstr: "0 6 * * *" timezone: Europe/Amsterdam @@ -1637,6 +1644,243 @@ def get_automation(self, id: int, automation_id: int, asset: GenericAsset): automation_data["redis_connection_err"] = redis_connection_err return automation_data, 200 + @route("//automations", methods=["POST"]) + @use_kwargs( + {"asset": AssetIdField(data_key="id")}, + location="path", + ) + # Managing an automation is gated like running one: an automation exists to write data + # under the asset, so the same principals that may add data there may define it. + # The sensors it involves are checked separately, against the user's own access. + @permission_required_for_context("create-children", ctx_arg_name="asset") + @as_json + def post_automation(self, id: int, asset: GenericAsset): + """ + .. :quickref: Assets; Create an automation on an asset. + + --- + post: + summary: Create an automation on an asset. + description: | + Create a recurring task (computing forecasts or schedules) on the asset. + The parameters are validated by the schema matching the automation type: + forecast parameters for type `forecasts`, or a schedule trigger message + (without the asset id) for type `schedules`. + Requires permission to add data under the asset. + + The automation can only involve sensors that you have access to yourself: + read access to the sensors it reads data from, and permission to record data + on the sensors it writes to. + security: + - ApiKeyAuth: [] + parameters: + - in: path + name: id + required: true + description: ID of the asset to create the automation on. + schema: + type: integer + requestBody: + content: + application/json: + schema: AutomationCreationSchema + examples: + daily_forecasts: + summary: Daily forecasts of sensor 2092 + value: + name: Day-ahead PV forecasts + cronstr: "0 6 * * *" + type: forecasting + parameters: + sensor: 2092 + responses: + 201: + description: CREATED + 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: + automation_data = AutomationCreationSchema().load(body) + except ValidationError as e: + return unprocessable_entity(e.messages) + try: + automation, warnings = create_automation( + asset=asset, + name=automation_data["name"], + cronstr=automation_data["cronstr"], + timezone=automation_data["timezone"], + automation_type=automation_data["type"], + active=automation_data["active"], + parameters=automation_data["parameters"], + forecaster_class=automation_data["forecaster"], + config=automation_data["config"], + origin="API", + check_permissions=True, + ) + except ValidationError as e: + return unprocessable_entity({"parameters": e.messages}) + except ValueError as e: + return unprocessable_entity(str(e)) + db.session.commit() + response = automation_schema.dump(automation) + response["recurrence_description"] = describe_cronstr(automation.cronstr) + response["warnings"] = warnings + return response, 201 + + @route("//automations/", methods=["PATCH"]) + @use_kwargs( + { + "asset": AssetIdField(data_key="id"), + "automation_id": fields.Int(), + }, + location="path", + ) + # Managing an automation is gated like running one: an automation exists to write data + # under the asset, so the same principals that may add data there may define it. + # The sensors it involves are checked separately, against the user's own access. + @permission_required_for_context("create-children", ctx_arg_name="asset") + @as_json + def patch_automation(self, id: int, automation_id: int, asset: GenericAsset): + """ + .. :quickref: Assets; Update an automation's name, cron string or activation status. + + --- + patch: + summary: Update an automation's name, cron string or activation status. + description: | + Any subset of the fields `name`, `cronstr` and `active` can be sent. + Other automation fields cannot be updated; instead, create a new automation. + Requires permission to add data under the asset. + security: + - ApiKeyAuth: [] + parameters: + - in: path + name: id + required: true + description: ID of the asset. + schema: + type: integer + - in: path + name: automation_id + required: true + description: ID of the automation. + schema: + type: integer + requestBody: + content: + application/json: + schema: AutomationUpdateSchema + examples: + deactivate: + summary: Deactivate the automation + value: + active: false + responses: + 200: + description: PROCESSED + 400: + description: INVALID_REQUEST, REQUIRED_INFO_MISSING, UNEXPECTED_PARAMS + 401: + description: UNAUTHORIZED + 403: + description: INVALID_SENDER + 404: + description: NOT_FOUND + 422: + description: UNPROCESSABLE_ENTITY + tags: + - Assets + """ + automation = db.session.get(Automation, automation_id) + if automation is None or automation.asset_id != asset.id: + return { + "message": f"Asset {asset.id} has no automation with id {automation_id}." + }, 404 + body = request.get_json(silent=True) + if not body: + return unprocessable_entity("No JSON data provided.") + try: + automation_data = AutomationUpdateSchema().load(body) + except ValidationError as e: + return unprocessable_entity(e.messages) + update_automation(automation, origin="API", **automation_data) + db.session.commit() + response = automation_schema.dump(automation) + response["recurrence_description"] = describe_cronstr(automation.cronstr) + return response, 200 + + @route("//automations/", methods=["DELETE"]) + @use_kwargs( + { + "asset": AssetIdField(data_key="id"), + "automation_id": fields.Int(), + }, + location="path", + ) + # Managing an automation is gated like running one: an automation exists to write data + # under the asset, so the same principals that may add data there may define it. + # The sensors it involves are checked separately, against the user's own access. + @permission_required_for_context("create-children", ctx_arg_name="asset") + @as_json + def delete_automation(self, id: int, automation_id: int, asset: GenericAsset): + """ + .. :quickref: Assets; Delete an automation. + + --- + delete: + summary: Delete an automation. + description: | + Delete the automation. Any jobs it already queued are unaffected. + Requires permission to add data under the asset. + security: + - ApiKeyAuth: [] + parameters: + - in: path + name: id + required: true + description: ID of the asset. + schema: + type: integer + - in: path + name: automation_id + required: true + description: ID of the automation. + schema: + type: integer + responses: + 204: + description: DELETED + 400: + description: INVALID_REQUEST, REQUIRED_INFO_MISSING, UNEXPECTED_PARAMS + 401: + description: UNAUTHORIZED + 403: + description: INVALID_SENDER + 404: + description: NOT_FOUND + tags: + - Assets + """ + automation = db.session.get(Automation, automation_id) + if automation is None or automation.asset_id != asset.id: + return { + "message": f"Asset {asset.id} has no automation with id {automation_id}." + }, 404 + remove_automation(automation, origin="API") + db.session.commit() + return {}, 204 + @route("//automations//trigger", methods=["POST"]) @limit_triggers() @use_kwargs( diff --git a/flexmeasures/api/v3_0/tests/test_automations_api.py b/flexmeasures/api/v3_0/tests/test_automations_api.py index 6a6904acfd..9f17e819d1 100644 --- a/flexmeasures/api/v3_0/tests/test_automations_api.py +++ b/flexmeasures/api/v3_0/tests/test_automations_api.py @@ -2,18 +2,20 @@ from __future__ import annotations -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone import pytest from flask import url_for +from sqlalchemy import select from flexmeasures.data.models.automations import Automation from flexmeasures.data.models.data_sources import DataSource +from flexmeasures.data.models.time_series import Sensor -@pytest.fixture(scope="module") -def add_automations(db, add_battery_assets): - battery = add_battery_assets["Test battery"] +@pytest.fixture(scope="function") +def add_automations(fresh_db, add_battery_assets_fresh_db): + battery = add_battery_assets_fresh_db["Test battery"] generator = DataSource( name="automations API test generator", type="forecaster", @@ -43,8 +45,8 @@ def add_automations(db, add_battery_assets): parameters={"sensor": battery.sensors[0].id}, ), ] - db.session.add_all(automations) - db.session.flush() + fresh_db.session.add_all(automations) + fresh_db.session.flush() return automations @@ -59,12 +61,12 @@ def add_automations(db, add_battery_assets): ) def test_get_automations_auth( app, - add_battery_assets, + add_battery_assets_fresh_db, add_automations, requesting_user, expected_status_code, ): - battery = add_battery_assets["Test battery"] + battery = add_battery_assets_fresh_db["Test battery"] with app.test_client() as client: response = client.get( url_for("AssetAPI:get_automations", id=battery.id), @@ -77,11 +79,11 @@ def test_get_automations_auth( ) def test_get_automations( app, - add_battery_assets, + add_battery_assets_fresh_db, add_automations, requesting_user, ): - battery = add_battery_assets["Test battery"] + battery = add_battery_assets_fresh_db["Test battery"] with app.test_client() as client: response = client.get( url_for("AssetAPI:get_automations", id=battery.id), @@ -108,11 +110,11 @@ def test_get_automations( ) def test_get_automation_details( app, - add_battery_assets, + add_battery_assets_fresh_db, add_automations, requesting_user, ): - battery = add_battery_assets["Test battery"] + battery = add_battery_assets_fresh_db["Test battery"] automation = add_automations[0] with app.test_client() as client: response = client.get( @@ -139,12 +141,12 @@ def test_get_automation_details( ) def test_get_automation_of_other_asset( app, - add_battery_assets, + add_battery_assets_fresh_db, add_automations, requesting_user, ): """Requesting an automation via an asset it does not belong to should return 404.""" - other_asset = add_battery_assets["Test small battery"] + other_asset = add_battery_assets_fresh_db["Test small battery"] automation = add_automations[0] with app.test_client() as client: response = client.get( @@ -162,11 +164,11 @@ def test_get_automation_of_other_asset( ) def test_get_nonexistent_automation( app, - add_battery_assets, + add_battery_assets_fresh_db, add_automations, requesting_user, ): - battery = add_battery_assets["Test battery"] + battery = add_battery_assets_fresh_db["Test battery"] with app.test_client() as client: response = client.get( url_for("AssetAPI:get_automation", id=battery.id, automation_id=9999), @@ -174,6 +176,374 @@ def test_get_nonexistent_automation( assert response.status_code == 404 +@pytest.mark.parametrize( + "requesting_user, expected_status_code", + [ + ("test_prosumer_user@seita.nl", 201), # plain account member + ("test_prosumer_user_2@seita.nl", 201), # account admin + ("test_dummy_user_3@seita.nl", 403), # different account + ], + indirect=["requesting_user"], +) +def test_post_automation( + app, + fresh_db, + add_battery_assets_fresh_db, + requesting_user, + expected_status_code, +): + """Whoever may add data under the asset can create automations on it; parameters are validated by type.""" + battery = add_battery_assets_fresh_db["Test battery"] + with app.test_client() as client: + response = client.post( + url_for("AssetAPI:post_automation", id=battery.id), + json={ + "name": "Posted schedules", + "cronstr": "0 0 * * *", + "type": "scheduling", + "parameters": {"duration": "PT12H"}, + }, + ) + assert response.status_code == expected_status_code + if expected_status_code == 201: + assert response.json["name"] == "Posted schedules" + assert response.json["active"] is True + assert response.json["recurrence_description"] == "At 00:00" + automation = fresh_db.session.get(Automation, response.json["id"]) + assert automation.parameters == {"duration": "PT12H"} + # clean up for other tests in this module + fresh_db.session.delete(automation) + fresh_db.session.flush() + + +@pytest.mark.parametrize( + "requesting_user", ["test_prosumer_user_2@seita.nl"], indirect=True +) +def test_post_automation_with_invalid_parameters( + app, + add_battery_assets_fresh_db, + requesting_user, +): + battery = add_battery_assets_fresh_db["Test battery"] + with app.test_client() as client: + response = client.post( + url_for("AssetAPI:post_automation", id=battery.id), + json={ + "name": "Bad forecasts", + "cronstr": "0 6 * * *", + "type": "forecasting", + "parameters": {}, # missing required sensor + }, + ) + assert response.status_code == 422 + assert "sensor" in str(response.json) + + +@pytest.mark.parametrize( + "requesting_user", ["test_prosumer_user_2@seita.nl"], indirect=True +) +def test_post_and_patch_automation_timezone( + app, + fresh_db, + add_battery_assets_fresh_db, + requesting_user, +): + """An automation's timezone can be set on creation and changed afterwards, as it can from the CLI.""" + battery = add_battery_assets_fresh_db["Test battery"] + with app.test_client() as client: + response = client.post( + url_for("AssetAPI:post_automation", id=battery.id), + json={ + "name": "Seoul forecasts", + "cronstr": "0 6 * * *", + "timezone": "Asia/Seoul", + "type": "forecasting", + "parameters": {"sensor": battery.sensors[0].id}, + }, + ) + assert response.status_code == 201, response.json + assert response.json["timezone"] == "Asia/Seoul" + automation = fresh_db.session.execute( + select(Automation).filter_by(name="Seoul forecasts") + ).scalar_one() + assert automation.timezone == "Asia/Seoul" + + with app.test_client() as client: + response = client.patch( + url_for( + "AssetAPI:patch_automation", + id=battery.id, + automation_id=automation.id, + ), + json={"timezone": "Europe/Amsterdam"}, + ) + assert response.status_code == 200, response.json + assert response.json["timezone"] == "Europe/Amsterdam" + assert automation.timezone == "Europe/Amsterdam" + + with app.test_client() as client: + response = client.patch( + url_for( + "AssetAPI:patch_automation", + id=battery.id, + automation_id=automation.id, + ), + json={"timezone": "Europe/NotAmsterdam"}, + ) + assert response.status_code == 422 + + +@pytest.mark.parametrize( + "requesting_user", ["test_prosumer_user_2@seita.nl"], indirect=True +) +def test_post_automation_with_inaccessible_source_filtered_regressor( + app, + fresh_db, + add_battery_assets_fresh_db, + setup_generic_assets_fresh_db, + requesting_user, +): + """A regressor that filters on sources is a sensor reference, and still counts as a sensor read.""" + battery = add_battery_assets_fresh_db["Test battery"] + someone_elses_sensor = Sensor( + name="wind speed for a filtered regressor", + generic_asset=setup_generic_assets_fresh_db[ + "test_wind_turbine" + ], # owned by the Supplier account + event_resolution=timedelta(minutes=15), + unit="m/s", + ) + fresh_db.session.add(someone_elses_sensor) + fresh_db.session.flush() + data_sources_before = set(fresh_db.session.scalars(select(DataSource.id)).all()) + + with app.test_client() as client: + response = client.post( + url_for("AssetAPI:post_automation", id=battery.id), + json={ + "name": "Forecasts regressing on another account's sensor", + "cronstr": "0 6 * * *", + "type": "forecasting", + "parameters": {"sensor": battery.sensors[0].id}, + "config": { + "regressors": [ + { + "sensor": someone_elses_sensor.id, + "source-types": ["forecaster"], + } + ] + }, + }, + ) + assert response.status_code == 403 + assert str(someone_elses_sensor.id) in response.json["message"] + assert someone_elses_sensor.name not in response.json["message"] + assert ( + fresh_db.session.execute( + select(Automation).filter_by( + name="Forecasts regressing on another account's sensor" + ) + ).scalar_one_or_none() + is None + ) + # a refused request also leaves behind no data source for the forecaster it would have run + assert ( + set(fresh_db.session.scalars(select(DataSource.id)).all()) + == data_sources_before + ) + + +@pytest.mark.parametrize( + "requesting_user", ["test_prosumer_user_2@seita.nl"], indirect=True +) +def test_post_automation_with_inaccessible_sensor( + app, + fresh_db, + add_battery_assets_fresh_db, + setup_generic_assets_fresh_db, + requesting_user, +): + """An account admin cannot set up an automation on a sensor of another account.""" + battery = add_battery_assets_fresh_db["Test battery"] + someone_elses_sensor = Sensor( + name="wind speed", + generic_asset=setup_generic_assets_fresh_db[ + "test_wind_turbine" + ], # owned by the Supplier account + event_resolution=timedelta(minutes=15), + unit="m/s", + ) + fresh_db.session.add(someone_elses_sensor) + fresh_db.session.flush() + + with app.test_client() as client: + response = client.post( + url_for("AssetAPI:post_automation", id=battery.id), + json={ + "name": "Forecasts of another account's sensor", + "cronstr": "0 6 * * *", + "type": "forecasting", + "parameters": {"sensor": someone_elses_sensor.id}, + }, + ) + assert response.status_code == 403 + assert str(someone_elses_sensor.id) in response.json["message"] + assert someone_elses_sensor.name not in response.json["message"] + assert ( + fresh_db.session.execute( + select(Automation).filter_by(name="Forecasts of another account's sensor") + ).scalar_one_or_none() + is None + ) + + # the same automation on their own sensor is fine + own_sensor = battery.sensors[0] + with app.test_client() as client: + response = client.post( + url_for("AssetAPI:post_automation", id=battery.id), + json={ + "name": "Forecasts of their own sensor", + "cronstr": "0 6 * * *", + "type": "forecasting", + "parameters": {"sensor": own_sensor.id}, + }, + ) + assert response.status_code == 201, response.json + # clean up for other tests in this module + fresh_db.session.delete(fresh_db.session.get(Automation, response.json["id"])) + fresh_db.session.flush() + + +@pytest.mark.parametrize( + "requesting_user", ["test_prosumer_user_2@seita.nl"], indirect=True +) +def test_post_schedule_automation_with_inaccessible_output_sensor( + app, + fresh_db, + add_battery_assets_fresh_db, + setup_generic_assets_fresh_db, + requesting_user, +): + """Sensors that a schedule would be recorded on are checked, wherever they are named. + + The aggregate power schedule is recorded on the flex-context's aggregate-consumption + sensor, so that one needs to be writable, too — not just the flex-model's own sensors. + """ + battery = add_battery_assets_fresh_db["Test battery"] + someone_elses_sensor = Sensor( + name="aggregate consumption", + generic_asset=setup_generic_assets_fresh_db[ + "test_wind_turbine" + ], # owned by the Supplier account + event_resolution=timedelta(minutes=15), + unit="MW", + ) + fresh_db.session.add(someone_elses_sensor) + fresh_db.session.flush() + + with app.test_client() as client: + response = client.post( + url_for("AssetAPI:post_automation", id=battery.id), + json={ + "name": "Schedules aggregated onto another account's sensor", + "cronstr": "0 0 * * *", + "type": "scheduling", + "parameters": { + "duration": "PT12H", + "flex-context": { + "aggregate-consumption": {"sensor": someone_elses_sensor.id} + }, + }, + }, + ) + assert response.status_code == 403 + assert str(someone_elses_sensor.id) in response.json["message"] + assert someone_elses_sensor.name not in response.json["message"] + assert "record data on" in response.json["message"] + + +@pytest.mark.parametrize( + "requesting_user, expected_status_code", + [ + ("test_prosumer_user@seita.nl", 200), # plain account member + ("test_prosumer_user_2@seita.nl", 200), # account admin + ], + indirect=["requesting_user"], +) +def test_patch_automation( + app, + fresh_db, + add_battery_assets_fresh_db, + add_automations, + requesting_user, + expected_status_code, +): + battery = add_battery_assets_fresh_db["Test battery"] + automation = add_automations[0] + original_name = automation.name + with app.test_client() as client: + response = client.patch( + url_for( + "AssetAPI:patch_automation", + id=battery.id, + automation_id=automation.id, + ), + json={"name": "Renamed via API", "active": False}, + ) + assert response.status_code == expected_status_code + if expected_status_code == 200: + assert response.json["name"] == "Renamed via API" + assert response.json["active"] is False + # restore for other tests in this module + automation.name = original_name + automation.active = True + fresh_db.session.flush() + + +@pytest.mark.parametrize( + "requesting_user", ["test_prosumer_user_2@seita.nl"], indirect=True +) +def test_delete_automation( + app, + fresh_db, + add_battery_assets_fresh_db, + add_automations, + requesting_user, +): + battery = add_battery_assets_fresh_db["Test battery"] + automation = Automation( + asset_id=battery.id, + # a forecast automation is required to have a data generator holding its forecaster config + generator=add_automations[0].generator, + type="forecasting", + name="To be deleted", + cronstr="0 6 * * *", + parameters={"sensor": battery.sensors[0].id}, + ) + fresh_db.session.add(automation) + fresh_db.session.flush() + with app.test_client() as client: + response = client.delete( + url_for( + "AssetAPI:delete_automation", + id=battery.id, + automation_id=automation.id, + ), + ) + assert response.status_code == 204 + assert fresh_db.session.get(Automation, automation.id) is None + + # deleting again yields the documented 404 + response = client.delete( + url_for( + "AssetAPI:delete_automation", + id=battery.id, + automation_id=automation.id, + ), + ) + assert response.status_code == 404 + + @pytest.mark.parametrize( "requesting_user, expected_status_code", [ @@ -185,13 +555,13 @@ def test_get_nonexistent_automation( ) def test_trigger_automation_auth( app, - add_battery_assets, + add_battery_assets_fresh_db, add_automations, requesting_user, expected_status_code, mocker, ): - battery = add_battery_assets["Test battery"] + battery = add_battery_assets_fresh_db["Test battery"] automation = add_automations[0] run_automation = mocker.patch( "flexmeasures.api.v3_0.assets.run_automation", @@ -217,14 +587,14 @@ def test_trigger_automation_auth( ) def test_trigger_automation( app, - db, - add_battery_assets, + fresh_db, + add_battery_assets_fresh_db, add_automations, requesting_user, mocker, ): """Triggering a run reports the queued job, and leaves the automation's recurrence alone.""" - battery = add_battery_assets["Test battery"] + battery = add_battery_assets_fresh_db["Test battery"] automation = add_automations[1] # inactive automations can be triggered, too. cursor_before = automation.cursor mocker.patch( @@ -243,7 +613,7 @@ def test_trigger_automation( assert response.json["status"] == "ACCEPTED" assert response.json["job"] == "364bfd06-c1fa-430b-8d25-8f5a547651fb" assert response.json["n_jobs"] == 2 - db.session.expire_all() + fresh_db.session.expire_all() assert automation.cursor == cursor_before assert automation.active is False @@ -253,13 +623,13 @@ def test_trigger_automation( ) def test_trigger_automation_that_cannot_run( app, - add_battery_assets, + add_battery_assets_fresh_db, add_automations, requesting_user, mocker, ): """A run which cannot be set up is reported as such, rather than as a queued job.""" - battery = add_battery_assets["Test battery"] + battery = add_battery_assets_fresh_db["Test battery"] automation = add_automations[0] mocker.patch( "flexmeasures.api.v3_0.assets.run_automation", @@ -285,7 +655,7 @@ def test_trigger_automation_that_cannot_run( @pytest.mark.parametrize("via_other_asset", [True, False]) def test_trigger_unknown_automation( app, - add_battery_assets, + add_battery_assets_fresh_db, add_automations, requesting_user, via_other_asset, @@ -295,10 +665,10 @@ def test_trigger_unknown_automation( run_automation = mocker.patch("flexmeasures.api.v3_0.assets.run_automation") if via_other_asset: # an existing automation, requested through an asset it does not belong to. - asset = add_battery_assets["Test small battery"] + asset = add_battery_assets_fresh_db["Test small battery"] automation_id = add_automations[0].id else: - asset = add_battery_assets["Test battery"] + asset = add_battery_assets_fresh_db["Test battery"] automation_id = 9999 with app.test_client() as client: response = client.post( diff --git a/flexmeasures/cli/data_add.py b/flexmeasures/cli/data_add.py index 9dc7a76285..ae2af2ae75 100755 --- a/flexmeasures/cli/data_add.py +++ b/flexmeasures/cli/data_add.py @@ -52,16 +52,14 @@ populate_initial_structure, add_default_asset_types, ) -from flexmeasures.data.schemas.scheduling import find_momentary_flex_config_fields from flexmeasures.data.services.automations import ( - prepare_schedule_trigger_message, - resolve_schedule_generator, + create_automation, + RecurringScheduleFixesAMoment, ) from flexmeasures.data.services.data_sources import ( get_or_create_source, get_data_generator, ) -from flexmeasures.data.services.automations import validate_forecast_output_scope from flexmeasures.data.services.scheduling import make_schedule, create_scheduling_job from flexmeasures.data.services.users import create_user from flexmeasures.data.models.user import ( @@ -1722,29 +1720,6 @@ def add_forecast( # noqa: C901 raise -def _check_schedule_automation_parameters(parameters: dict, asset) -> DataSource: - """Validate a schedule automation's trigger message, and return the data generator it will run with. - - The message has to be a valid schedule trigger, and its flex config has to describe the site and its devices, - rather than one moment: the automation computes a fresh schedule on every run, - so a value tied to a fixed moment would be stale on the next one. - """ - try: - message = prepare_schedule_trigger_message(parameters, asset.id) - AssetTriggerSchema().load(message) - except ValidationError as e: - click.secho(f"Invalid schedule parameters: {e.messages}", **MsgStyle.ERROR) - raise click.Abort() - momentary_fields = find_momentary_flex_config_fields(message) - if momentary_fields: - raise click.UsageError( - f"{flexmeasures_inflection.join_words_into_a_list(momentary_fields)} fixes a moment in time," - " so it cannot configure a recurring schedule automation, which computes a fresh schedule on every run." - " Refer to a sensor instead of a fixed value, or leave the field out." - ) - return resolve_schedule_generator(asset.id, parameters) - - @fm_add_data.command("automation") @with_appcontext @click.option( @@ -1907,66 +1882,35 @@ def add_automation( " combined with --type scheduling: a schedule automation is not computed by a forecaster." ) - # Validate the parameters using the forecast parameters schema (we store them serialized) - generator_id = None - if automation_type == "forecasting": - try: - deserialized_parameters = ForecasterParametersSchema().load(parameters) - except ValidationError as e: - click.secho(f"Invalid forecast parameters: {e.messages}", **MsgStyle.ERROR) - raise click.Abort() - output_sensor = deserialized_parameters.get( - "sensor_to_save" - ) or deserialized_parameters.get("sensor") - try: - validate_forecast_output_scope(asset.id, output_sensor) - except ValueError as exc: - click.secho(str(exc), **MsgStyle.ERROR) - raise click.Abort() - - forecaster = get_data_generator( - source=source, - model=forecaster_class, + # The service validates the parameters by automation type (we store them serialized) + try: + automation, warnings = create_automation( + asset=asset, + name=name, + cronstr=cronstr, + timezone=timezone, + automation_type=automation_type, + active=not inactive, + parameters=parameters, + forecaster_class=forecaster_class, config=config, - save_config=True, - data_generator_type=Forecaster, + source=source, + origin="CLI", ) - if forecaster is None: - click.secho( - f"Could not set up forecaster '{forecaster_class}'.", **MsgStyle.ERROR - ) - raise click.Abort() - generator = ( - forecaster.data_source - ) # looks up or creates the data source storing the forecaster config - db.session.flush() - generator_id = generator.id - else: # scheduling - # The scheduler and its configuration make up the automation's data generator, - # the same way a forecaster and its configuration do for a forecast automation. - generator_id = _check_schedule_automation_parameters(parameters, asset).id - if "start" in parameters: - click.secho( - "Warning: the schedule 'start' is fixed, so each run will compute the same period." - " Omit 'start' to schedule from the run time instead.", - **MsgStyle.WARN, - ) - - automation = Automation( - asset_id=asset.id, - type=automation_type, - name=name, - cronstr=cronstr, - timezone=timezone, - active=not inactive, - generator_id=generator_id, - parameters=parameters, - ) - db.session.add(automation) - db.session.flush() - AssetAuditLog.add_record( - asset, f"Created automation '{name}' ({automation.id}) via CLI." - ) + except ValidationError as e: + click.secho( + f"Invalid {Automation.RESULT_NOUNS[automation_type]} parameters: {e.messages}", + **MsgStyle.ERROR, + ) + raise click.Abort() + except RecurringScheduleFixesAMoment as e: + # A usage error: the automation cannot be defined this way, whatever the data says. + raise click.UsageError(str(e)) + except ValueError as e: + click.secho(str(e), **MsgStyle.ERROR) + raise click.Abort() + for warning in warnings: + click.secho(f"Warning: {warning}", **MsgStyle.WARN) db.session.commit() click.secho( f"Successfully created {'inactive ' if inactive else ''}automation '{name}' (ID: {automation.id})" diff --git a/flexmeasures/cli/data_delete.py b/flexmeasures/cli/data_delete.py index 1911a6b7e3..3459f636bd 100644 --- a/flexmeasures/cli/data_delete.py +++ b/flexmeasures/cli/data_delete.py @@ -17,11 +17,13 @@ from flexmeasures import Source from flexmeasures.data import db from flexmeasures.data.models.user import Account, AccountRole, RolesAccounts, User -from flexmeasures.data.models.audit_log import AssetAuditLog from flexmeasures.data.models.automations import Automation from flexmeasures.data.models.generic_assets import GenericAsset from flexmeasures.data.schemas.automations import AutomationIdField -from flexmeasures.data.services.automations import get_automations_involving_sensor +from flexmeasures.data.services.automations import ( + delete_automation as remove_automation, + get_automations_involving_sensor, +) from flexmeasures.data.models.time_series import Sensor, TimedBelief from flexmeasures.data.schemas import ( AccountIdField, @@ -295,11 +297,7 @@ def delete_automation(automation: Automation, force: bool): if not force: prompt = f"Delete automation '{automation.name}' (ID: {automation.id}) of asset '{automation.asset.name}'?" click.confirm(prompt, abort=True) - AssetAuditLog.add_record( - automation.asset, - f"Deleted automation '{automation.name}' ({automation.id}) via CLI.", - ) - db.session.delete(automation) + remove_automation(automation, origin="CLI") db.session.commit() click.secho( f"Successfully deleted automation '{automation.name}' (ID: {automation.id}).", diff --git a/flexmeasures/cli/data_edit.py b/flexmeasures/cli/data_edit.py index 3eaecf7426..f6dc9892bc 100644 --- a/flexmeasures/cli/data_edit.py +++ b/flexmeasures/cli/data_edit.py @@ -19,16 +19,14 @@ from flexmeasures.data.schemas import AssetIdField from flexmeasures.data.schemas.sensors import SensorIdField from flexmeasures.data.models.generic_assets import GenericAsset -from flexmeasures.data.models.automations import ( - Automation, - get_initial_cursor, -) +from flexmeasures.data.models.automations import Automation from flexmeasures.data.models.audit_log import AssetAuditLog, AuditLog from flexmeasures.data.schemas.automations import ( AutomationIdField, CronField, TimezoneField, ) +from flexmeasures.data.services.automations import update_automation from flexmeasures.data.models.time_series import TimedBelief from flexmeasures.data.utils import save_to_db from flexmeasures.cli.utils import ( @@ -107,33 +105,17 @@ def edit_automation( active: bool | None = None, ): """Edit the name, recurrence, timezone or activation status of an automation.""" - changes = [] - rebase_schedule = False - if name is not None and name != automation.name: - changes.append(f"name: '{automation.name}' → '{name}'") - automation.name = name - if cronstr is not None and cronstr != automation.cronstr: - changes.append(f"cron string: '{automation.cronstr}' → '{cronstr}'") - automation.cronstr = cronstr - rebase_schedule = True - if timezone is not None and timezone != automation.timezone: - changes.append(f"timezone: '{automation.timezone}' → '{timezone}'") - automation.timezone = timezone - rebase_schedule = True - if active is not None and active != automation.active: - changes.append("activated" if active else "deactivated") - if active: - rebase_schedule = True - automation.active = active + changes = update_automation( + automation, + name=name, + cronstr=cronstr, + timezone=timezone, + active=active, + origin="CLI", + ) if not changes: click.secho("Nothing to change.", **MsgStyle.WARN) return - if rebase_schedule: - automation.cursor = get_initial_cursor() - AssetAuditLog.add_record( - automation.asset, - f"Updated automation '{automation.name}' ({automation.id}): {'; '.join(changes)}. Via CLI.", - ) db.session.commit() click.secho( f"Successfully updated automation '{automation.name}' (ID: {automation.id}): {'; '.join(changes)}.", diff --git a/flexmeasures/data/models/automations.py b/flexmeasures/data/models/automations.py index eaa7587055..902c433b0a 100644 --- a/flexmeasures/data/models/automations.py +++ b/flexmeasures/data/models/automations.py @@ -46,6 +46,10 @@ class Automation(db.Model, AuthModelMixin): SUPPORTED_TYPES = ["forecasting", "scheduling"] # later also "reporting" + # What one result of each type is called, for messages that talk about a single result, + # such as the parameters an automation of that type computes with. + RESULT_NOUNS = {"forecasting": "forecast", "scheduling": "schedule"} + id = db.Column(db.Integer, autoincrement=True, primary_key=True) created_at = db.Column( db.DateTime(timezone=True), nullable=False, default=server_now @@ -95,16 +99,16 @@ def validate_timezone(self, key: str, timezone: str) -> str: def __acl__(self): """ Whoever can read the asset can read its automations. - Updating and deleting automations is allowed for whoever can delete - the asset (i.e. account admins and consultants). + Updating and deleting automations is allowed for whoever may add data under the asset, + which is what defining an automation amounts to. """ if self.asset is None: return {} asset_acl = self.asset.__acl__() return { "read": asset_acl["read"], - "update": asset_acl["delete"], - "delete": asset_acl["delete"], + "update": asset_acl["create-children"], + "delete": asset_acl["create-children"], } def __repr__(self): diff --git a/flexmeasures/data/schemas/automations.py b/flexmeasures/data/schemas/automations.py index 97ada4baf5..5bad71f003 100644 --- a/flexmeasures/data/schemas/automations.py +++ b/flexmeasures/data/schemas/automations.py @@ -4,7 +4,7 @@ from croniter import croniter from croniter.croniter import CroniterBadDateError -from marshmallow import fields, validates, ValidationError +from marshmallow import fields, validate, validates, Schema, ValidationError from pytz import all_timezones_set from flexmeasures.data import ma, db @@ -66,6 +66,59 @@ def _serialize(self, automation, attr, data, **kwargs): return automation.id +class AutomationCreationSchema(Schema): + """Request schema for creating an automation (the asset comes from the URL path). + + The parameters are validated separately, by the schema matching the automation type. + """ + + type = fields.Str( + load_default="forecasting", + validate=validate.OneOf(Automation.SUPPORTED_TYPES), + ) + name = fields.Str(required=True, validate=validate.Length(min=1, max=80)) + cronstr = CronField(required=True) + timezone = TimezoneField( + load_default=None, + metadata={ + "description": "IANA timezone in which the cron expression is interpreted. Defaults to the server's FLEXMEASURES_TIMEZONE.", + "example": "Europe/Amsterdam", + }, + ) + active = fields.Bool(load_default=True) + parameters = fields.Dict(keys=fields.Str(), load_default=dict) + forecaster = fields.Str( + load_default="TrainPredictPipeline", + metadata={ + "description": "Forecaster class (only used for type 'forecasting')." + }, + ) + config = fields.Dict( + keys=fields.Str(), + load_default=dict, + metadata={ + "description": "Forecaster configuration (only used for type 'forecasting')." + }, + ) + + +class AutomationUpdateSchema(Schema): + """Request schema for updating an automation's name, recurrence, timezone and/or activation status. + + The parameters cannot be updated, so the sensors an automation involves stay the ones its creator was checked against. + """ + + name = fields.Str(validate=validate.Length(min=1, max=80)) + cronstr = CronField() + timezone = TimezoneField( + metadata={ + "description": "IANA timezone in which the cron expression is interpreted.", + "example": "Europe/Amsterdam", + } + ) + active = fields.Bool() + + class AutomationSchema(ma.SQLAlchemySchema): """Automation schema, with validations.""" diff --git a/flexmeasures/data/services/automations.py b/flexmeasures/data/services/automations.py index a1abaf4072..761953db6c 100644 --- a/flexmeasures/data/services/automations.py +++ b/flexmeasures/data/services/automations.py @@ -21,13 +21,20 @@ from flexmeasures import Forecaster from flexmeasures.data import db -from flexmeasures.data.models.automations import Automation +from flexmeasures.data.models.automations import ( + Automation, + get_initial_cursor, +) from flexmeasures.data.models.data_sources import DataSource from flexmeasures.data.models.time_series import Sensor from flexmeasures.data.queries.generic_assets import ( asset_and_ancestor_ids, asset_is_in_subtree, ) +from flexmeasures.data.services.data_generators import ( + check_sensor_access, + resolve_data_generator_sensors, +) from flexmeasures.utils.time_utils import server_now @@ -42,7 +49,17 @@ class DueAutomation: expected_timezone: str -# Fields naming sensors on which a scheduler records generated schedules. +# Fields naming a sensor that a scheduler records its results on, rather than reads from. +# A scheduler hands its results to `make_schedule` as (sensor, data) pairs, and these are +# the fields that decide which sensors those are: besides the power sensor of each device +# in the flex-model, its state of charge and its consumption and production sensors, plus +# the aggregates over all devices, which are defined in the flex-context. +# +# NB this list restates at set-up time what a scheduler decides at run time, so the two can drift apart. +# A scheduler that starts returning results for a sensor named by some other field would write to a sensor +# that was never checked against the creator's permissions, as this reads that sensor as an input instead. +# Extend this list whenever a flex-model or flex-context field starts naming somewhere results are recorded. +# Checking the sensors a scheduler actually returns, rather than the ones predicted here, would close the gap for good. OUTPUT_SENSOR_FIELDS = ( "consumption", "production", @@ -61,7 +78,15 @@ def collect_sensors( only_under_output_field: bool = False, _under_output_field: bool = False, ) -> list[Sensor]: - """Collect sensor objects and references from a nested scheduling structure.""" + """Collect the sensors referenced anywhere in a (possibly nested) structure. + + Both deserialized sensors and the sensor references that survive deserialization + as raw data (e.g. the flex-context and each device's flex-model, which schedulers + deserialize themselves) are picked up. + + :param only_under_output_field: only collect the sensors that are referenced under + one of the OUTPUT_SENSOR_FIELDS, at any depth. + """ if sensors is None: sensors = {} @@ -75,6 +100,7 @@ def collect(sensor: Sensor | None): for key, item in value.items(): under_output_field = _under_output_field or key in OUTPUT_SENSOR_FIELDS if key == "sensor" and isinstance(item, (int, str)): + # a sensor reference that was not deserialized, e.g. {"sensor": 12} if str(item).isdigit(): sensor = db.session.get(Sensor, int(item)) if sensor is not None and ( @@ -92,9 +118,15 @@ def collect(sensor: Sensor | None): def collect_schedule_output_sensors(message: dict) -> list[Sensor]: - """Collect sensors on which the prepared schedule trigger records results.""" + """The sensors that scheduling with this trigger message would record data on. + + That is the power sensor of each device in the flex-model, plus any sensor named by + a field that defines where generated data goes (see OUTPUT_SENSOR_FIELDS), both per + device and, for the aggregates, in the flex-context. + """ sensors: dict[int, Sensor] = {} for device in message.get("flex_model") or []: + # each device's power sensor is what its schedule is recorded on collect_sensors(device.get("sensor"), sensors) collect_sensors( device.get("sensor_flex_model", device), @@ -312,6 +344,13 @@ def claim_due_automation(due_automation: DueAutomation) -> bool: return True +class RecurringScheduleFixesAMoment(ValueError): + """Raised when a schedule automation's flex config pins a moment in time. + + Such a value would be stale on the automation's next run, so it cannot configure a recurring schedule. + """ + + class AutomationSensorsUnknown(Exception): """Raised when the sensors an automation involves cannot be worked out. @@ -328,30 +367,25 @@ def resolve_schedule_automation_sensors( from flexmeasures.data.services.scheduling import find_scheduler_class from flexmeasures.data.services.utils import get_scheduler_instance - try: - trigger_data = AssetTriggerSchema().load( - prepare_schedule_trigger_message(parameters, asset_id) - ) - start = trigger_data["start_of_schedule"] - scheduler_params = { - "start": start, - "end": start + trigger_data["duration"], - "belief_time": trigger_data.get("belief_time"), - "resolution": trigger_data.get("resolution"), - "flex_model": trigger_data["flex_model"], - "flex_context": trigger_data["flex_context"], - } - scheduler_class = find_scheduler_class(trigger_data["asset"]) - scheduler = get_scheduler_instance( - scheduler_class=scheduler_class, - asset_or_sensor=trigger_data["asset"], - scheduler_params=scheduler_params, - ) - scheduler.collect_flex_config() - except (NotImplementedError, ValidationError, ValueError) as exc: - raise AutomationSensorsUnknown( - f"Could not determine the sensors of schedule automation on asset {asset_id}: {exc}" - ) from exc + trigger_data = AssetTriggerSchema().load( + prepare_schedule_trigger_message(parameters, asset_id) + ) + start = trigger_data["start_of_schedule"] + scheduler_params = { + "start": start, + "end": start + trigger_data["duration"], + "belief_time": trigger_data.get("belief_time"), + "resolution": trigger_data.get("resolution"), + "flex_model": trigger_data["flex_model"], + "flex_context": trigger_data["flex_context"], + } + scheduler_class = find_scheduler_class(trigger_data["asset"]) + scheduler = get_scheduler_instance( + scheduler_class=scheduler_class, + asset_or_sensor=trigger_data["asset"], + scheduler_params=scheduler_params, + ) + scheduler.collect_flex_config() resolved_trigger = { "flex_model": scheduler.flex_model, @@ -381,24 +415,24 @@ def resolve_automation_sensors(automation: Automation) -> dict[str, list[Sensor] Use this wherever the answer decides whether something is permitted; use `get_automation_sensors` for display. """ if automation.type == "scheduling": - return resolve_schedule_automation_sensors( - dict(automation.parameters or {}), automation.asset_id - ) + try: + return resolve_schedule_automation_sensors( + dict(automation.parameters or {}), automation.asset_id + ) + except (NotImplementedError, ValidationError, ValueError) as exc: + raise AutomationSensorsUnknown( + f"Could not determine the sensors of schedule automation {automation.id}: {exc}" + ) from exc if automation.generator is None: raise AutomationSensorsUnknown( f"Automation {automation.id} has no data generator, so the sensors it involves are unknown." ) try: - # Work on a copy, as the data generator is cached on the data source, - # which may be shared by several automations. - data_generator = copy(automation.generator.data_generator) - data_generator._parameters = data_generator._parameters_schema.load( - dict(automation.parameters or {}) + data_generator = automation.generator.data_generator + return resolve_data_generator_sensors( + data_generator, + data_generator._parameters_schema.load(dict(automation.parameters or {})), ) - return { - "input_sensors": data_generator.input_sensors, - "output_sensors": data_generator.output_sensors, - } except (NotImplementedError, ValidationError) as e: raise AutomationSensorsUnknown( f"Could not determine the sensors of automation {automation.id}: {e}" @@ -569,6 +603,211 @@ def get_automation_job_stats(automation: Automation) -> dict[str, int]: return counts +def create_automation( + asset, + name: str, + cronstr: str, + timezone: str | None = None, + automation_type: str = "forecasting", + active: bool = True, + parameters: dict | None = None, + forecaster_class: str = "TrainPredictPipeline", + config: dict | None = None, + source=None, + origin: str = "API", + check_permissions: bool = False, +) -> tuple[Automation, list[str]]: + """Create an automation (not committed yet), validating its parameters by type. + + For forecasts, the forecaster config is stored on a data source. + An audit log record is added to the asset. + + :param check_permissions: whether to require that the current user may read the + sensors that the automation reads from, and record data + on the sensors it writes to. Set this for automations + created by a user (through the API or the UI); the CLI + runs without a user, and is trusted. + :raises marshmallow.ValidationError: if the parameters are invalid. + :raises ValueError: if the forecaster cannot be set up. + :raises werkzeug.exceptions.Forbidden: if a sensor is not accessible to the user. + :returns: the automation and a list of warnings. + """ + from marshmallow import ValidationError + + from flexmeasures.data.models.audit_log import AssetAuditLog + from flexmeasures.data.models.time_series import Sensor + + parameters = parameters or {} + warnings: list[str] = [] + generator_id = None + forecaster = None + input_sensors: list[Sensor] = [] + output_sensors: list[Sensor] = [] + forecast_output_sensor: Sensor | None = None + if automation_type == "forecasting": + from flexmeasures.data.schemas.forecasting.pipeline import ( + ForecasterParametersSchema, + ) + from flexmeasures.data.services.data_sources import get_data_generator + + deserialized_parameters = ForecasterParametersSchema().load(parameters) + sensor = deserialized_parameters.get("sensor") + if isinstance(sensor, Sensor) and sensor.generic_asset_id != asset.id: + warnings.append( + f"The sensor to forecast ({sensor.id}) does not belong to asset {asset.id}." + ) + forecaster = get_data_generator( + source=source, + model=forecaster_class, + config=config or {}, + save_config=True, + data_generator_type=Forecaster, + ) + if forecaster is None: + raise ValueError(f"Could not set up forecaster '{forecaster_class}'.") + + # A forecast reads the history of the sensor to forecast, plus its regressors, + # and records the forecast on the sensor to save to (the same sensor by default). + # The forecaster works this out from the same config and parameters it will run with. + forecast_sensors = resolve_data_generator_sensors( + forecaster, deserialized_parameters + ) + input_sensors = forecast_sensors["input_sensors"] + output_sensors = forecast_sensors["output_sensors"] + forecast_output_sensor = output_sensors[0] if output_sensors else None + elif automation_type == "scheduling": + from flexmeasures.utils import flexmeasures_inflection + from flexmeasures.data.schemas.scheduling import ( + find_momentary_flex_config_fields, + ) + + # The flex config has to describe the site and its devices, rather than one moment: + # the automation computes a fresh schedule on every run, + # so a value tied to a fixed moment would be stale on the next one. + momentary_fields = find_momentary_flex_config_fields( + prepare_schedule_trigger_message(dict(parameters), asset.id) + ) + if momentary_fields: + raise RecurringScheduleFixesAMoment( + f"{flexmeasures_inflection.join_words_into_a_list(momentary_fields)} fixes a moment in time," + " so it cannot configure a recurring schedule automation, which computes a fresh schedule on every run." + " Refer to a sensor instead of a fixed value, or leave the field out." + ) + + # A schedule is recorded on the sensors that the scheduler returns its results + # for, and reads whatever other sensors the flex-model and flex-context refer to + # (such as price sensors and the sensors of inflexible devices). + schedule_sensors = resolve_schedule_automation_sensors(parameters, asset.id) + input_sensors = schedule_sensors["input_sensors"] + output_sensors = schedule_sensors["output_sensors"] + if "start" in parameters: + warnings.append( + "The schedule 'start' is fixed, so each run will compute the same period." + " Omit 'start' to schedule from the run time instead." + ) + else: + raise ValidationError( + f"Automation type '{automation_type}' is not supported (supported types: {Automation.SUPPORTED_TYPES})." + ) + + if check_permissions: + check_sensor_access(input_sensors, output_sensors) + + # Only once the sensors are known to be the user's to involve do we say anything about them, + # so that this does not reveal where a sensor sits to someone who may not read it. + if forecast_output_sensor is not None: + validate_forecast_output_scope(asset.id, forecast_output_sensor) + + if forecaster is not None: + # Look up or create the data source storing the forecaster config only now that the automation is going ahead, + # so that a refused request leaves nothing behind, whatever the caller does with the session afterwards. + generator = forecaster.data_source + db.session.flush() + generator_id = generator.id + elif automation_type == "scheduling": + # The scheduler and its configuration make up the automation's data generator, + # the same way a forecaster and its configuration do for a forecast automation. + # It is resolved here, rather than above, for the same reason as the forecaster's. + generator_id = resolve_schedule_generator(asset.id, parameters).id + db.session.flush() + + automation_fields = dict( + asset_id=asset.id, + type=automation_type, + name=name, + cronstr=cronstr, + active=active, + generator_id=generator_id, + parameters=parameters, + ) + if timezone is not None: + automation_fields["timezone"] = timezone + automation = Automation(**automation_fields) + db.session.add(automation) + db.session.flush() + AssetAuditLog.add_record( + asset, f"Created automation '{name}' ({automation.id}) via {origin}." + ) + return automation, warnings + + +def update_automation( + automation: Automation, + name: str | None = None, + cronstr: str | None = None, + timezone: str | None = None, + active: bool | None = None, + origin: str = "API", +) -> list[str]: + """Update an automation's name, cron string, timezone and/or activation status (not committed yet). + + Anything that changes which runs are due, namely the recurrence, the timezone and reactivation, + also rebases the cursor, so that runs from before the change are not caught up on. + An audit log record is added to the asset. + + :returns: a list of (human-readable) changes; empty if nothing changed. + """ + from flexmeasures.data.models.audit_log import AssetAuditLog + + changes = [] + rebase_schedule = False + if name is not None and name != automation.name: + changes.append(f"name: '{automation.name}' → '{name}'") + automation.name = name + if cronstr is not None and cronstr != automation.cronstr: + changes.append(f"cron string: '{automation.cronstr}' → '{cronstr}'") + automation.cronstr = cronstr + rebase_schedule = True + if timezone is not None and timezone != automation.timezone: + changes.append(f"timezone: '{automation.timezone}' → '{timezone}'") + automation.timezone = timezone + rebase_schedule = True + if active is not None and active != automation.active: + changes.append("activated" if active else "deactivated") + if active: + rebase_schedule = True + automation.active = active + if rebase_schedule: + automation.cursor = get_initial_cursor() + if changes: + AssetAuditLog.add_record( + automation.asset, + f"Updated automation '{automation.name}' ({automation.id}): {'; '.join(changes)}. Via {origin}.", + ) + return changes + + +def delete_automation(automation: Automation, origin: str = "API"): + """Delete an automation (not committed yet), recording it in the asset's audit log.""" + from flexmeasures.data.models.audit_log import AssetAuditLog + + AssetAuditLog.add_record( + automation.asset, + f"Deleted automation '{automation.name}' ({automation.id}) via {origin}.", + ) + db.session.delete(automation) + + def get_forecast_output_sensor(parameters: dict[str, Any]) -> Sensor: """Resolve the sensor on which a forecast automation registers beliefs.""" sensor_reference = parameters.get("sensor-to-save") diff --git a/flexmeasures/ui/static/openapi-specs.json b/flexmeasures/ui/static/openapi-specs.json index 980b13d042..57cd8b0bb1 100644 --- a/flexmeasures/ui/static/openapi-specs.json +++ b/flexmeasures/ui/static/openapi-specs.json @@ -3368,6 +3368,58 @@ } }, "/api/v3_0/assets/{id}/automations/{automation_id}": { + "delete": { + "summary": "Delete an automation.", + "description": "Delete the automation. Any jobs it already queued are unaffected.\nRequires permission to add data under the asset.\n", + "security": [ + { + "ApiKeyAuth": [] + } + ], + "parameters": [ + { + "in": "path", + "name": "id", + "required": true, + "description": "ID of the asset.", + "schema": { + "type": "integer" + } + }, + { + "in": "path", + "name": "automation_id", + "required": true, + "description": "ID of the automation.", + "schema": { + "type": "integer" + } + } + ], + "responses": { + "204": { + "description": "DELETED" + }, + "400": { + "description": "INVALID_REQUEST, REQUIRED_INFO_MISSING, UNEXPECTED_PARAMS" + }, + "401": { + "description": "UNAUTHORIZED" + }, + "403": { + "description": "INVALID_SENDER" + }, + "404": { + "description": "NOT_FOUND" + }, + "429": { + "description": "TOO_MANY_REQUESTS - You called the API more often than your rate limit allows. Wait for as long as the Retry-After header says, then try again." + } + }, + "tags": [ + "Assets" + ] + }, "get": { "summary": "Get details of one automation defined on an asset.", "description": "In addition to the fields shown when listing automations, the response shows\nthe automation's parameters (forecast parameters or a schedule trigger message),\ninformation about its data generator (null for schedule automations),\nthe sensors it reads from and writes to,\nand counts of recently created jobs, per job status.\nNote that jobs in Redis have a limited TTL, so not all past jobs will be counted.\nThe cursor is the UTC time of the most recent run the automation committed to; runs at or before it are never queued again.\nIt advances just before queueing, so it does not indicate that queueing or the forecast itself succeeded.\n", @@ -3408,7 +3460,7 @@ "id": 1, "created_at": "2026-07-11T00:00:00+00:00", "asset_id": 1, - "type": "forecasts", + "type": "forecasting", "name": "Day-ahead PV forecasts", "cronstr": "0 6 * * *", "timezone": "Europe/Amsterdam", @@ -3471,6 +3523,78 @@ "tags": [ "Assets" ] + }, + "patch": { + "summary": "Update an automation's name, cron string or activation status.", + "description": "Any subset of the fields `name`, `cronstr` and `active` can be sent.\nOther automation fields cannot be updated; instead, create a new automation.\nRequires permission to add data under the asset.\n", + "security": [ + { + "ApiKeyAuth": [] + } + ], + "parameters": [ + { + "in": "path", + "name": "id", + "required": true, + "description": "ID of the asset.", + "schema": { + "type": "integer" + } + }, + { + "in": "path", + "name": "automation_id", + "required": true, + "description": "ID of the automation.", + "schema": { + "type": "integer" + } + } + ], + "requestBody": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/AutomationUpdateSchema" + }, + "examples": { + "deactivate": { + "summary": "Deactivate the automation", + "value": { + "active": false + } + } + } + } + } + }, + "responses": { + "200": { + "description": "PROCESSED" + }, + "400": { + "description": "INVALID_REQUEST, REQUIRED_INFO_MISSING, UNEXPECTED_PARAMS" + }, + "401": { + "description": "UNAUTHORIZED" + }, + "403": { + "description": "INVALID_SENDER" + }, + "404": { + "description": "NOT_FOUND" + }, + "422": { + "description": "UNPROCESSABLE_ENTITY" + }, + "429": { + "description": "TOO_MANY_REQUESTS - You called the API more often than your rate limit allows. Wait for as long as the Retry-After header says, then try again." + } + }, + "tags": [ + "Assets" + ] } }, "/api/v3_0/assets/{id}/automations": { @@ -3507,7 +3631,7 @@ "id": 1, "created_at": "2026-07-11T00:00:00+00:00", "asset_id": 1, - "type": "forecasts", + "type": "forecasting", "name": "Day-ahead PV forecasts", "cronstr": "0 6 * * *", "timezone": "Europe/Amsterdam", @@ -3541,6 +3665,71 @@ "tags": [ "Assets" ] + }, + "post": { + "summary": "Create an automation on an asset.", + "description": "Create a recurring task (computing forecasts or schedules) on the asset.\nThe parameters are validated by the schema matching the automation type:\nforecast parameters for type `forecasts`, or a schedule trigger message\n(without the asset id) for type `schedules`.\nRequires permission to add data under the asset.\n\nThe automation can only involve sensors that you have access to yourself:\nread access to the sensors it reads data from, and permission to record data\non the sensors it writes to.\n", + "security": [ + { + "ApiKeyAuth": [] + } + ], + "parameters": [ + { + "in": "path", + "name": "id", + "required": true, + "description": "ID of the asset to create the automation on.", + "schema": { + "type": "integer" + } + } + ], + "requestBody": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/AutomationCreationSchema" + }, + "examples": { + "daily_forecasts": { + "summary": "Daily forecasts of sensor 2092", + "value": { + "name": "Day-ahead PV forecasts", + "cronstr": "0 6 * * *", + "type": "forecasting", + "parameters": { + "sensor": 2092 + } + } + } + } + } + } + }, + "responses": { + "201": { + "description": "CREATED" + }, + "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 more often than your rate limit allows. Wait for as long as the Retry-After header says, then try again." + } + }, + "tags": [ + "Assets" + ] } }, "/api/v3_0/assets/{id}/chart": { @@ -6540,6 +6729,81 @@ ], "additionalProperties": false }, + "AutomationCreationSchema": { + "type": "object", + "properties": { + "type": { + "type": "string", + "default": "forecasting", + "enum": [ + "forecasting", + "scheduling" + ] + }, + "name": { + "type": "string", + "minLength": 1, + "maxLength": 80 + }, + "cronstr": { + "type": "string" + }, + "timezone": { + "type": [ + "string", + "null" + ], + "default": null, + "description": "IANA timezone in which the cron expression is interpreted. Defaults to the server's FLEXMEASURES_TIMEZONE.", + "example": "Europe/Amsterdam" + }, + "active": { + "type": "boolean", + "default": true + }, + "parameters": { + "type": "object", + "additionalProperties": {} + }, + "forecaster": { + "type": "string", + "default": "TrainPredictPipeline", + "description": "Forecaster class (only used for type 'forecasting')." + }, + "config": { + "type": "object", + "description": "Forecaster configuration (only used for type 'forecasting').", + "additionalProperties": {} + } + }, + "required": [ + "cronstr", + "name" + ], + "additionalProperties": false + }, + "AutomationUpdateSchema": { + "type": "object", + "properties": { + "name": { + "type": "string", + "minLength": 1, + "maxLength": 80 + }, + "cronstr": { + "type": "string" + }, + "timezone": { + "type": "string", + "description": "IANA timezone in which the cron expression is interpreted.", + "example": "Europe/Amsterdam" + }, + "active": { + "type": "boolean" + } + }, + "additionalProperties": false + }, "ReportTriggerSchema": { "type": "object", "properties": { diff --git a/flexmeasures/ui/templates/assets/asset_automations.html b/flexmeasures/ui/templates/assets/asset_automations.html index 55359d5d28..3189d175d1 100644 --- a/flexmeasures/ui/templates/assets/asset_automations.html +++ b/flexmeasures/ui/templates/assets/asset_automations.html @@ -26,24 +26,125 @@

During daylight-saving-time changes, a run at a skipped local time happens once after the clock moves forward, and a run at a repeated local time happens only once.

+ {% if user_can_manage_automations %} +
+ +
+ + + {% for timezone in available_timezones %} + + {% endfor %} + + + + + + + + {% endif %} +
-
+
-
+
@@ -58,9 +159,11 @@