diff --git a/documentation/api/change_log.rst b/documentation/api/change_log.rst index 66dde555d0..502a4cfe08 100644 --- a/documentation/api/change_log.rst +++ b/documentation/api/change_log.rst @@ -16,6 +16,7 @@ v3.0-32 | August 11, 2026 - 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. - Sensor references on variable-quantity flex-model and flex-context fields (such as ``soc-minima``, ``soc-maxima``, the capacity fields and the price fields) may now include a ``default`` fallback quantity, e.g. ``{"sensor": 50, "default": "0 kWh"}``. It fills the time slots for which the referenced sensor holds no value. Note that this fills *every* such slot, so a sensor recording only occasional setpoints becomes densely constrained. The field is not (yet) applied to sensor references on ``inflexible-consumption``/``inflexible-production`` or to forecaster regressors. Take particular care with a fallback of ``0`` on a ``consumption-capacity`` or ``production-capacity``: if the sensor holds no value for the whole scheduling window, the resulting all-zero capacity is read as a physical statement about the device and enforced strictly (see :ref:`the flex-model capacity fields `), rather than as a limit that may be breached at a price. +- 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 account admin or consultant rights, 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``. - Extended ``GET /api/v3_0/jobs/`` with a ``result`` field containing ``unresolved`` and ``resolved`` arrays, each keyed by asset ID. For scheduling jobs, this surfaces soft state-of-charge constraint analysis: ``soc-minima`` and ``soc-maxima`` violations (with a ``violation`` magnitude) or satisfied constraints (with a ``margin`` headroom). Both arrays are empty when no SoC constraints were defined. - **Field canonicalization** for background job tracking: * The ``job`` field is now the canonical way to identify background jobs returned by `/sensors//schedules/trigger`, `/assets//schedules/trigger`, and `/sensors//forecasts/trigger` endpoints. If applicable, the triggered response now also returns a ``results-url`` pointing to the sensor-specific results endpoint, alongside the generic ``job-url``. diff --git a/documentation/changelog.rst b/documentation/changelog.rst index b07ae36ce2..ec545fced8 100644 --- a/documentation/changelog.rst +++ b/documentation/changelog.rst @@ -56,6 +56,7 @@ New features * Automations - first roundtrip for forecasts: recurring tasks defined per asset, 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; 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 `_ and `PR #2396 `_] * 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 `_] * Automations can also compute schedules on a recurring basis (``flexmeasures add automation --type schedules``), with the schedule start defaulting to each run's time [see `PR #2293 `_] +* Automations can be created, edited and deleted in the UI and through new API endpoints (``[POST|PATCH|DELETE] /assets/(id)/automations``), by organisation admins and consultants, 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 #2294 `_] * ``flexmeasures show data-sources`` now shows which organisation a data source belongs to, and can list the sensors holding data recorded by a given source [see `PR #2401 `_] * The flex-context can now define multiple commodities, each specifying their own prices and grid capacities [see `PR #1946 `_, `PR #2172 `_, `PR #2235 `_, `PR #2271 `_, `PR #2355 `_ and `PR #2380 `_] * Support multiple feeders to a shared storage [see `PR #2001 `_, `PR #2321 `_, `PR #2322 `_, `PR #2325 `_ and `PR #2431 `_] diff --git a/flexmeasures/api/v3_0/__init__.py b/flexmeasures/api/v3_0/__init__.py index d6b73a9cf0..dfd32dfff6 100644 --- a/flexmeasures/api/v3_0/__init__.py +++ b/flexmeasures/api/v3_0/__init__.py @@ -41,6 +41,10 @@ DefaultAssetViewJSONSchema, ) 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.sensors import QuantitySchema, TimeSeriesSchema from flexmeasures.data.schemas.account import ( @@ -222,6 +226,8 @@ def create_openapi_specs(app: Flask): ("AssetAPIQuerySchema", AssetAPIQuerySchema), ("AssetSchema", AssetSchema), ("AnnotationSchema", AnnotationSchema), + ("AutomationCreationSchema", AutomationCreationSchema), + ("AutomationUpdateSchema", AutomationUpdateSchema), ("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 ad9fe3bd5a..f3be3ae146 100644 --- a/flexmeasures/api/v3_0/assets.py +++ b/flexmeasures/api/v3_0/assets.py @@ -49,12 +49,19 @@ 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, + update_automation, ) from flexmeasures.data.models.generic_assets import GenericAsset, GenericAssetType from flexmeasures.data.queries.generic_assets import ( @@ -1576,6 +1583,240 @@ 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 automations requires the same principals that may delete the asset + # (i.e. account admins and consultants), matching the Automation ACL. + @permission_required_for_context("delete", 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 account admin or consultant rights. + + 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: forecasts + 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 automations requires the same principals that may delete the asset + # (i.e. account admins and consultants), matching the Automation ACL. + @permission_required_for_context("delete", 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 account admin or consultant rights. + 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 automations requires the same principals that may delete the asset + # (i.e. account admins and consultants), matching the Automation ACL. + @permission_required_for_context("delete", 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 account admin or consultant rights. + 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("//jobs", methods=["GET"]) @use_kwargs( {"asset": AssetIdField(data_key="id")}, diff --git a/flexmeasures/api/v3_0/tests/test_automations_api.py b/flexmeasures/api/v3_0/tests/test_automations_api.py index 39652038b8..854bd88954 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,13 +164,381 @@ 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), ) assert response.status_code == 404 + + +@pytest.mark.parametrize( + "requesting_user, expected_status_code", + [ + ("test_prosumer_user@seita.nl", 403), # 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, +): + """Only account admins (and consultants) can create automations; 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": "schedules", + "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": "forecasts", + "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": "forecasts", + "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": "forecasts", + "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": "forecasts", + "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": "forecasts", + "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": "schedules", + "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", 403), # 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="forecasts", + 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 diff --git a/flexmeasures/cli/data_add.py b/flexmeasures/cli/data_add.py index 2c982b1d64..39f26cd572 100755 --- a/flexmeasures/cli/data_add.py +++ b/flexmeasures/cli/data_add.py @@ -52,12 +52,11 @@ populate_initial_structure, add_default_asset_types, ) -from flexmeasures.data.services.automations import prepare_schedule_trigger_message +from flexmeasures.data.services.automations import create_automation 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 ( @@ -1826,70 +1825,32 @@ def add_automation( " combined with --type schedules: 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 == "forecasts": - 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: # schedules - try: - AssetTriggerSchema().load( - prepare_schedule_trigger_message(parameters, asset.id) - ) - except ValidationError as e: - click.secho(f"Invalid schedule parameters: {e.messages}", **MsgStyle.ERROR) - raise click.Abort() - 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_type[:-1]} parameters: {e.messages}", + **MsgStyle.ERROR, + ) + raise click.Abort() + 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/schemas/automations.py b/flexmeasures/data/schemas/automations.py index 97ada4baf5..94481b74cf 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,57 @@ 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="forecasts", + 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 'forecasts')."}, + ) + config = fields.Dict( + keys=fields.Str(), + load_default=dict, + metadata={ + "description": "Forecaster configuration (only used for type 'forecasts')." + }, + ) + + +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 7de5d00671..e33fe047ae 100644 --- a/flexmeasures/data/services/automations.py +++ b/flexmeasures/data/services/automations.py @@ -19,9 +19,15 @@ from marshmallow import ValidationError from sqlalchemy import select, update +from werkzeug.exceptions import Forbidden + from flexmeasures import Forecaster +from flexmeasures.auth.policy import check_access 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.time_series import Sensor from flexmeasures.data.queries.generic_assets import ( asset_and_ancestor_ids, @@ -41,7 +47,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", @@ -60,7 +76,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 = {} @@ -74,6 +98,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 ( @@ -91,9 +116,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), @@ -104,6 +135,31 @@ def collect_schedule_output_sensors(message: dict) -> list[Sensor]: return list(sensors.values()) +def check_sensor_access( + input_sensors: list[Sensor], output_sensors: list[Sensor] +) -> None: + """Require access to the sensors that an automation would read from and write to. + + Reading a sensor's data requires read access to it, and recording data on a sensor + requires the same permission as recording data through the API (create-children). + """ + 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: + setattr( + exc, + "api_message", + f"You cannot set up an automation that would {action} sensor" + f" {sensor.id}, because you cannot {action} it yourself.", + ) + raise + + def describe_cronstr(cronstr: str) -> str: """Describe a cron string in natural language, e.g. "At 06:00". @@ -319,6 +375,26 @@ class AutomationSensorsUnknown(Exception): """ +def resolve_data_generator_sensors( + data_generator, deserialized_parameters: dict +) -> dict[str, list[Sensor]]: + """Ask a data generator which sensors it would read from and write to, given these parameters. + + A data generator derives this from its own config and parameters, so it also picks up a regressor that filters on sources, + which is a sensor reference rather than a plain sensor. + Work out the answer here rather than in each caller, so that displaying the sensors involved + and checking access to them can never disagree about what they are. + """ + # Work on a copy, as the data generator is cached on the data source, + # which may be shared by several automations. + data_generator = copy(data_generator) + data_generator._parameters = deserialized_parameters + return { + "input_sensors": data_generator.input_sensors, + "output_sensors": data_generator.output_sensors, + } + + def resolve_schedule_automation_sensors( parameters: dict, asset_id: int ) -> dict[str, list[Sensor]]: @@ -327,30 +403,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, @@ -380,24 +451,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 == "schedules": - 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}" @@ -537,6 +608,187 @@ 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 = "forecasts", + 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 == "forecasts": + 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 == "schedules": + # 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 + + 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 f14e1a64cd..63cfce523c 100644 --- a/flexmeasures/ui/static/openapi-specs.json +++ b/flexmeasures/ui/static/openapi-specs.json @@ -3347,6 +3347,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 account admin or consultant rights.\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", @@ -3450,6 +3502,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 account admin or consultant rights.\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": { @@ -3520,6 +3644,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 account admin or consultant rights.\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": "forecasts", + "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": { @@ -6225,6 +6414,81 @@ ], "additionalProperties": false }, + "AutomationCreationSchema": { + "type": "object", + "properties": { + "type": { + "type": "string", + "default": "forecasts", + "enum": [ + "forecasts", + "schedules" + ] + }, + "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 'forecasts')." + }, + "config": { + "type": "object", + "description": "Forecaster configuration (only used for type 'forecasts').", + "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 + }, "CopyAssetSchema": { "type": "object", "properties": { diff --git a/flexmeasures/ui/templates/assets/asset_automations.html b/flexmeasures/ui/templates/assets/asset_automations.html index 257f7cc53b..11d7c08afa 100644 --- a/flexmeasures/ui/templates/assets/asset_automations.html +++ b/flexmeasures/ui/templates/assets/asset_automations.html @@ -26,6 +26,107 @@

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 %} +