diff --git a/documentation/api/change_log.rst b/documentation/api/change_log.rst index 36846d72fa..66dde555d0 100644 --- a/documentation/api/change_log.rst +++ b/documentation/api/change_log.rst @@ -5,8 +5,13 @@ 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-33 | September 1, 2026 +""""""""""""""""""""""""""" +- Added ``GET /api/v3_0/assets//automations`` and ``GET /api/v3_0/assets//automations/`` for listing and inspecting forecast automations, including the sensors an automation reads from and writes to. Each automation shows the IANA ``timezone`` in which its cron expression is interpreted, and a ``cursor``: the offset-aware UTC time of the most recent run it committed to. The cursor advances just before queueing, so it does not indicate that queueing or the forecast itself succeeded. Asset job entries now include ``created_via`` provenance; automation identity is included only when the caller may read that automation. +- Added ``GET /api/v3_0/sources/`` to show the full record of one data source, including the attributes in which data generators store their configuration. + v3.0-32 | August 11, 2026 -"""""""""""""""""""""""""" +""""""""""""""""""""""""" - API endpoints are now rate-limited. A request which exceeds a limit is answered with a ``429 (Too Many Requests)`` status code and a ``Retry-After`` header stating how many seconds to wait. Responses also carry ``X-RateLimit-*`` headers, describing the limit that applied, how much of it is left, and when it resets. A stricter limit applies to ``POST /assets//schedules/trigger``, ``POST /sensors//schedules/trigger`` and ``POST /sensors//forecasts/trigger`` than to other endpoints; the health endpoints are exempt. Per-account overrides are set by assigning the account a plan (a ``Plan`` database row), rather than through an account attribute. - Introduced the ``inflexible-consumption`` and ``inflexible-production`` flex-context fields, which make explicit how the sign of each inflexible device's power data should be read: positive values denote consumption resp. production. Each entry is a sensor reference (``{"sensor": }``), optionally with source filters (``source-types``, ``exclude-source-types``, ``sources``, ``source-account``). Deprecated the ``inflexible-device-sensors`` field (a list of bare sensor IDs, whose sign convention is read from each sensor's ``consumption_is_positive`` attribute); it remains supported, but cannot be combined with the new fields in one flex-context. - Added a ``role`` query parameter to ``GET /api/v3_0/accounts`` for filtering accessible organisations by account role. diff --git a/documentation/changelog.rst b/documentation/changelog.rst index dd701450ee..5780401428 100644 --- a/documentation/changelog.rst +++ b/documentation/changelog.rst @@ -53,6 +53,9 @@ v1.0.0 | August 25, 2026 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 `_] +* ``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 `_] * Add support for intermediate power constraints on groups of devices, via a new ``group`` field in the storage flex-model [see `PR #2276 `_ and `issue #2092 `_] diff --git a/documentation/cli/change_log.rst b/documentation/cli/change_log.rst index 7287db1578..c3a1769089 100644 --- a/documentation/cli/change_log.rst +++ b/documentation/cli/change_log.rst @@ -12,6 +12,9 @@ since v1.0.0 | August 11, 2026 * Add ``flexmeasures add plan``, ``flexmeasures show plans`` and ``flexmeasures edit plan``, to manage the rate limits and quotas which apply to the accounts on a plan. * Add ``flexmeasures edit secret`` to store an encrypted secret on an account or asset. * Add ``flexmeasures delete secret`` to remove an encrypted secret from an account or asset. +* Add ``flexmeasures add automation``, ``flexmeasures edit automation`` and ``flexmeasures delete automation`` to manage automations (recurring tasks on an asset; for now, computing forecasts). Each automation carries its own IANA timezone (``--timezone``), in which its cron expression is interpreted. +* Add ``flexmeasures jobs run-automations`` to queue jobs for all automations that are due to run this minute from standard five-field cron expressions. Run this command once per minute. It makes at most one queueing attempt per automation per minute, including when an attempt fails after partially queueing jobs. Runs missed while the runner was down are caught up once, with several missed forecast runs coalesced into the latest useful forecast, and a run at a skipped or repeated daylight-saving-time hour happens exactly once. +* ``flexmeasures delete sensor`` now warns which automations read from or write to a sensor before it is deleted, as an automation refers to its sensors by ID and would fail on its next run. * ``flexmeasures show data-sources`` now shows the account a data source belongs to, and lists the sensors holding data recorded by a single source with ``--show-sensors``. since v0.33.0 | June 01, 2026 diff --git a/documentation/cli/commands.rst b/documentation/cli/commands.rst index 4453a213a3..88314ae233 100644 --- a/documentation/cli/commands.rst +++ b/documentation/cli/commands.rst @@ -41,6 +41,7 @@ of which some are referred to in this documentation. ``flexmeasures add annotation`` Add annotation to accounts, assets and/or sensors. ``flexmeasures add toy-account`` Create a toy account, for tutorials and trying things. ``flexmeasures add report`` Create a report. +``flexmeasures add automation`` Add an automation with its own cron timezone. ================================================= ======================================= @@ -75,6 +76,7 @@ of which some are referred to in this documentation. ``flexmeasures edit resample-data`` Assign a new event resolution to an existing sensor and resample its data accordingly. ``flexmeasures edit transfer-parenthood`` (Re)assign parent assets. ``flexmeasures edit transfer-ownership`` Transfer the ownership of an asset and its children to a different account. +``flexmeasures edit automation`` Edit an automation's name, recurrence, timezone or activation status. ================================================= ======================================= ``delete`` - Delete data @@ -93,6 +95,7 @@ of which some are referred to in this documentation. ``flexmeasures delete prognoses`` Delete forecasts and schedules (forecasts > 0). ``flexmeasures delete unchanged-beliefs`` Delete unchanged beliefs. ``flexmeasures delete nan-beliefs`` Delete NaN beliefs. +``flexmeasures delete automation`` Delete an automation. ================================================= ======================================= @@ -117,6 +120,7 @@ of which some are referred to in this documentation. ``flexmeasures jobs run-job`` Run a single job (useful for debugging it) ``flexmeasures jobs inspect-job`` Inspect a background job and print its current status, result and metadata. ``flexmeasures jobs stats`` Show estimated live statistics of the queueing system. +``flexmeasures jobs run-automations`` Handle due and missed forecast automation runs (invoke once per minute, e.g. via cron). ================================================= ======================================= diff --git a/documentation/features/automations.rst b/documentation/features/automations.rst new file mode 100644 index 0000000000..7a441226e7 --- /dev/null +++ b/documentation/features/automations.rst @@ -0,0 +1,89 @@ +.. _automations: + +Automations +============ + +An **automation** is a recurring task defined on an asset. +For now, an automation computes forecasts; automating schedules and reports is planned. + +On each run, the automation queues jobs (so make sure a worker is processing the ``forecasting`` queue, see :ref:`redis-queue`). +The parameters of the task were stored when the automation was created, and validated with the same schema that the CLI and API use. +Timing parameters are resolved on each run — for instance, the forecast start defaults to the time the automation runs, so each run produces fresh forecasts. + +Creating an automation +---------------------- + +Here is how you create an automation in the CLI, asking for daily (at 6 AM) forecasts of sensor 12: + +.. code-block:: bash + + flexmeasures add automation --asset 3 --name "Daily PV forecasts" --type forecasts \ + --cron "0 6 * * *" --timezone Europe/Amsterdam --sensor 12 + +``--type`` says what the automation computes, and defaults to ``forecasts``. +The remaining options are the ones the task itself needs: a forecast automation accepts everything `flexmeasures add forecast` accepts, such as ``--forecaster`` to pick the forecaster and ``--config`` to configure it (see :ref:`forecasting`). +The forecaster and its configuration are stored on a data source, so you can also pass ``--source`` to reuse the data source of an existing forecaster, in which case ``--forecaster`` and ``--config`` (and the individual configuration options) are not needed — the data source already determines them. +That data source is required while the automation exists, so it cannot be deleted until the automation is removed. + +The recurrence is defined by a standard five-field cron string (minute, hour, day of month, month, and day of week), which defaults to ``"0 0 * * *"`` (daily at midnight). +It is interpreted in the automation's IANA timezone. +If ``--timezone`` is omitted, the current ``FLEXMEASURES_TIMEZONE`` value is copied to the automation. +Changing that configuration later does not change existing automations. +Cron aliases and optional seconds or year fields are not supported. + +Automations are active by default (use ``--inactive`` to create them in deactivated state). +Use ``flexmeasures edit automation`` to rename, re-schedule (``--cron``), change the timezone, activate or deactivate an automation, and ``flexmeasures delete automation`` to remove one. +These changes are recorded in the asset's audit log. + +For forecast automations, the sensor on which forecasts are saved (``sensor-to-save``, falling back to ``sensor``) must belong to the automation's asset or one of its descendants. +This relationship is checked both when the automation is created and immediately before each run. + +Running automations +------------------- + +For automations to actually run, let a cron job execute the following command once per minute: + +.. code-block:: bash + + * * * * * flexmeasures jobs run-automations + +Each due automation then queues its jobs. +If the runner misses runs, because it was down or overloaded, it catches up when it resumes: it queues only the latest missed run of each automation, rather than replaying stale ones. +Timing parameters that default to the run time are resolved when that catch-up run is queued, so it produces a current forecast. + +Each scheduled run receives at most one automatic queueing attempt. +If the process crashes, or queueing fails after creating some jobs, that run is not retried automatically, because a retry could duplicate partial work. + +The jobs record how they were created, which is shown on the asset's status page (UI), where recent jobs are listed. + +Viewing automations +------------------- + +Automations defined on an asset can be viewed on the asset's *Automations* page in the UI, and listed with the API endpoint `[GET] /assets/(id)/automations <../api/v3_0.html#get--api-v3_0-assets-id-automations>`_. +An automation's details show the sensors it reads from and writes to, linking to each sensor's page. +Conversely, a sensor's page lists the automations that write data to it. + +.. _automation_cursor: + +Appendix: how the runner decides what is due +-------------------------------------------- + +This section describes the bookkeeping behind the catch-up behaviour above. +You do not need it to use automations. + +The runner is a stateless command, executed once a minute by cron, so it needs a durable record of how far each automation has got. +That record is one UTC timestamp per automation, its *cursor*: the scheduled time of the most recent run the automation has committed to. +Runs at or before the cursor are never queued again. +Before queueing any jobs, the runner advances the cursor to the run it is about to queue, and saves it. +The cursor therefore records that a run was claimed, not that queueing or the task itself succeeded. + +Keeping a single moving timestamp, rather than a record per run, is what makes the behaviour above fall out: a runner that has been down catches up by moving the cursor straight to the latest due run, and two runners started in the same minute cannot queue the same run twice, because the cursor is advanced with a conditional update that only one of them can win. + +A new automation starts from its creation minute and does not replay runs from before it existed. +Changing its cron expression or timezone, or reactivating it, restarts from the time of that change. +Deactivated automations do not accumulate catch-up work. +After upgrading an existing installation, runs scheduled before the upgrade are not replayed. + +Daylight-saving-time transitions follow wall-clock semantics. +If the clock skips a scheduled local time in spring, that run happens once at the transition boundary. +If a scheduled local time occurs twice in autumn, the first instance is the canonical run and the repeated instance is not queued again. diff --git a/documentation/features/forecasting.rst b/documentation/features/forecasting.rst index 51bc532e5f..6c0f55144c 100644 --- a/documentation/features/forecasting.rst +++ b/documentation/features/forecasting.rst @@ -213,3 +213,11 @@ Usage: Create the annotations you want to use as regressors before running the forecast. For holidays, use ``flexmeasures add holidays``, which supports both ``workalendar`` and ``holidays``. See :ref:`annotations` for details. + +.. _automating_forecasts: + +Automating forecasts +-------------------- + +Instead of asking for forecasts one at a time, you can set up an *automation*: a recurring task defined on an asset, which queues forecasting jobs on a cron schedule. +See :ref:`automations`. diff --git a/documentation/index.rst b/documentation/index.rst index 27c580e4e6..3efd5092e1 100644 --- a/documentation/index.rst +++ b/documentation/index.rst @@ -175,6 +175,7 @@ In :ref:`getting_started`, we have some helpful tips how to dive into this docum features/scheduling features/forecasting features/reporting + features/automations .. toctree:: :caption: Tutorials diff --git a/flexmeasures/api/common/utils/api_utils.py b/flexmeasures/api/common/utils/api_utils.py index b8996b1406..0ec2f8ef26 100644 --- a/flexmeasures/api/common/utils/api_utils.py +++ b/flexmeasures/api/common/utils/api_utils.py @@ -26,6 +26,7 @@ ) from flexmeasures.data.models.generic_assets import GenericAsset from flexmeasures.data.models.time_series import Sensor +from flexmeasures.data.queries.generic_assets import asset_is_in_subtree from flexmeasures.data.utils import ( SAVE_TO_DB_SUCCESS, SAVE_TO_DB_SUCCESS_BUT_NOTHING_NEW, @@ -576,23 +577,6 @@ def _determine_copy_name( return f"{source_name} (Copy {max_index + 1})" -def _asset_is_in_subtree(root_asset_id: int, candidate_asset_id: int) -> bool: - """Return True if candidate_asset_id is root or a descendant of root_asset_id.""" - current_asset_id = candidate_asset_id - visited: set[int] = set() - - while current_asset_id is not None and current_asset_id not in visited: - if current_asset_id == root_asset_id: - return True - visited.add(current_asset_id) - current_asset = db.session.get(GenericAsset, current_asset_id) - if current_asset is None: - return False - current_asset_id = current_asset.parent_asset_id - - return False - - def copy_asset( asset: GenericAsset, account=None, @@ -640,7 +624,7 @@ def copy_asset( target_account_id = int(account.id) target_parent_asset_id = int(parent_asset.id) - if target_parent_asset_id is not None and _asset_is_in_subtree( + if target_parent_asset_id is not None and asset_is_in_subtree( root_asset_id=asset.id, candidate_asset_id=target_parent_asset_id, ): diff --git a/flexmeasures/api/v3_0/assets.py b/flexmeasures/api/v3_0/assets.py index 6c420350f4..4fb54d9989 100644 --- a/flexmeasures/api/v3_0/assets.py +++ b/flexmeasures/api/v3_0/assets.py @@ -46,8 +46,16 @@ from flexmeasures.auth.decorators import permission_required_for_context from flexmeasures.data import db from flexmeasures.data.models.annotations import Annotation, get_or_create_annotation +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.services.automations import ( + AutomationSensorsUnknown, + describe_cronstr, + get_automation_job_stats, + resolve_automation_sensors, +) from flexmeasures.data.models.generic_assets import GenericAsset, GenericAssetType from flexmeasures.data.queries.generic_assets import ( filter_assets_under_root, @@ -94,6 +102,7 @@ asset_type_schema = AssetTypeSchema() asset_schema = AssetSchema() annotation_schema = AnnotationSchema() +automation_schema = AutomationSchema() # creating this once to avoid recreating it on every request default_list_assets_schema = AssetSchema(many=True, only=default_response_fields) patch_asset_schema = AssetSchema(partial=True, exclude=["account_id"]) @@ -1365,6 +1374,208 @@ def auditlog( return response, 200 + @route("//automations", methods=["GET"]) + @use_kwargs( + {"asset": AssetIdField(data_key="id")}, + location="path", + ) + @permission_required_for_context("read", ctx_arg_name="asset") + @as_json + def get_automations(self, id: int, asset: GenericAsset): + """ + .. :quickref: Assets; Get all automations defined on an asset. + + --- + get: + summary: Get all automations defined on an asset. + description: | + The response will be a list of automations: recurring tasks (for now, computing forecasts) + defined on the asset. Each entry shows the automation's ID, when it was created, + its type, name, activation status, and its recurrence, both as a cron string + and described in natural language. Each entry also shows the IANA timezone in which its cron expression is interpreted, and its cursor. + security: + - ApiKeyAuth: [] + parameters: + - in: path + name: id + required: true + description: ID of the asset to get the automations for. + schema: + type: integer + responses: + 200: + description: PROCESSED + content: + application/json: + examples: + automations: + summary: List of automations + value: + automations: + - id: 1 + created_at: "2026-07-11T00:00:00+00:00" + asset_id: 1 + type: forecasts + name: Day-ahead PV forecasts + cronstr: "0 6 * * *" + timezone: Europe/Amsterdam + cursor: "2026-07-11T04:00:00+00:00" + recurrence_description: "At 06:00" + active: true + 400: + description: INVALID_REQUEST, REQUIRED_INFO_MISSING, UNEXPECTED_PARAMS + 401: + description: UNAUTHORIZED + 403: + description: INVALID_SENDER + 422: + description: UNPROCESSABLE_ENTITY + tags: + - Assets + """ + automations_data = [] + for automation in asset.automations: + automation_data = automation_schema.dump(automation) + automation_data["recurrence_description"] = describe_cronstr( + automation.cronstr + ) + automations_data.append(automation_data) + return {"automations": automations_data}, 200 + + @route("//automations/", methods=["GET"]) + @use_kwargs( + { + "asset": AssetIdField(data_key="id"), + "automation_id": fields.Int(), + }, + location="path", + ) + @permission_required_for_context("read", ctx_arg_name="asset") + @as_json + def get_automation(self, id: int, automation_id: int, asset: GenericAsset): + """ + .. :quickref: Assets; Get details of one automation defined on an asset. + + --- + get: + summary: Get details of one automation defined on an asset. + description: | + In addition to the fields shown when listing automations, the response shows + the automation's parameters (for forecasts, these are the forecast parameters + used on each run), information about the data generator that runs it, + the sensors it reads from and writes to, + and counts of recently created jobs, per job status. + Note that jobs in Redis have a limited TTL, so not all past jobs will be counted. + The cursor is the UTC time of the most recent run the automation committed to; runs at or before it are never queued again. + It advances just before queueing, so it does not indicate that queueing or the forecast itself succeeded. + 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: + 200: + description: PROCESSED + content: + application/json: + examples: + automation: + summary: Automation details + value: + id: 1 + created_at: "2026-07-11T00:00:00+00:00" + asset_id: 1 + type: forecasts + name: Day-ahead PV forecasts + cronstr: "0 6 * * *" + timezone: Europe/Amsterdam + cursor: "2026-07-11T04:00:00+00:00" + recurrence_description: "At 06:00" + active: true + parameters: + sensor: 2092 + generator: + id: 6 + description: "forecaster 'TrainPredictPipeline' (v1)" + input_sensors: + - id: 2092 + name: power + - id: 2093 + name: irradiance + output_sensors: + - id: 2092 + name: power + job_stats: + finished: 3 + failed: 1 + redis_connection_err: null + 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 + automation_data = automation_schema.dump(automation) + automation_data["recurrence_description"] = describe_cronstr(automation.cronstr) + automation_data["parameters"] = automation.parameters + automation_data["generator"] = ( + { + "id": automation.generator.id, + "description": automation.generator.description, + } + if automation.generator is not None + else None + ) + try: + automation_sensors = resolve_automation_sensors(automation) + except AutomationSensorsUnknown as e: + # One broken automation should not keep this response from rendering, + # and there are no sensors to check access on in this case. + current_app.logger.warning(str(e)) + automation_sensors = {"input_sensors": [], "output_sensors": []} + else: + for sensor in { + sensor + for key in ("input_sensors", "output_sensors") + for sensor in automation_sensors[key] + }: + check_access(sensor, "read") + for key in ("input_sensors", "output_sensors"): + automation_data[key] = [ + {"id": sensor.id, "name": sensor.name} + for sensor in automation_sensors[key] + ] + redis_connection_err = None + try: + automation_data["job_stats"] = get_automation_job_stats(automation) + except NoRedisConfigured as e: + automation_data["job_stats"] = {} + redis_connection_err = e.args[0] + automation_data["redis_connection_err"] = redis_connection_err + return automation_data, 200 + @route("//jobs", methods=["GET"]) @use_kwargs( {"asset": AssetIdField(data_key="id")}, @@ -1407,6 +1618,7 @@ def get_jobs(self, id: int, asset: GenericAsset): status: finished err: null enqueued_at: "2023-10-01T00:00:00" + created_via: API metadata_hash: abc123 redis_connection_err: null 400: diff --git a/flexmeasures/api/v3_0/sensors.py b/flexmeasures/api/v3_0/sensors.py index 0f1c41df87..427710d43b 100644 --- a/flexmeasures/api/v3_0/sensors.py +++ b/flexmeasures/api/v3_0/sensors.py @@ -2054,6 +2054,8 @@ def trigger_forecast(self, id: int, **params): except ValidationError as err: return unprocessable_entity(err.messages) + forecaster.set_job_trigger("API") + # Queue forecasting job try: pipeline_returns = forecaster.compute(parameters=parameters, as_job=True) diff --git a/flexmeasures/api/v3_0/sources.py b/flexmeasures/api/v3_0/sources.py index 2f19e45e30..2d0efbfc57 100644 --- a/flexmeasures/api/v3_0/sources.py +++ b/flexmeasures/api/v3_0/sources.py @@ -179,21 +179,86 @@ def index(self, only_latest: bool = True): return {"types": all_types, "sources": serialized}, 200 + @route("/", methods=["GET"]) + @as_json + def get(self, id: int): + """Get one data source, including its attributes. + + .. :quickref: Sources; Get one data source. -def _serialize_source(source: DataSource) -> dict: - """Serialize a DataSource to a plain dict for the API response.""" + --- + get: + summary: Get one data source. + description: | + Returns the full record of one data source, including the attributes in which + data generators (such as forecasters, schedulers and reporters) store their + configuration. + + The access rules are the same as for listing data sources. + security: + - ApiKeyAuth: [] + parameters: + - in: path + name: id + required: true + description: ID of the data source. + schema: + type: integer + responses: + 200: + description: PROCESSED + content: + application/json: + example: + id: 6 + name: Seita + type: forecaster + model: TrainPredictPipeline + version: "1" + description: "Seita's TrainPredictPipeline model v1" + account_id: 2 + user_id: null + attributes: + data_generator: + config: + model: CustomLGBM + 401: + description: UNAUTHORIZED + 403: + description: INVALID_SENDER + 404: + description: NOT_FOUND + tags: + - Sources + """ + source = db.session.get(DataSource, id) + if source is None: + return {"message": f"No data source found with id {id}."}, 404 + accessible_account_ids = _get_accessible_account_ids() + if accessible_account_ids is not None and not ( + source.account_id in accessible_account_ids + or (source.account_id is None and source.user_id is None) + ): + return {"message": "You cannot read this data source."}, 403 + return _serialize_source(source, with_attributes=True), 200 + + +def _serialize_source(source: DataSource, with_attributes: bool = False) -> dict: + """Serialize a DataSource to a plain dict for the API response. + + With `with_attributes`, the full record is returned, including the attributes and + any fields that are not set (rather than leaving those out). + """ result = { "id": source.id, "name": source.name, "type": source.type, "description": source.description, } - if source.model is not None: - result["model"] = source.model - if source.version is not None: - result["version"] = source.version - if source.account_id is not None: - result["account_id"] = source.account_id - if source.user_id is not None: - result["user_id"] = source.user_id + for field in ("model", "version", "account_id", "user_id"): + value = getattr(source, field) + if value is not None or with_attributes: + result[field] = value + if with_attributes: + result["attributes"] = source.attributes return result diff --git a/flexmeasures/api/v3_0/tests/test_asset_jobs_api_fresh_db.py b/flexmeasures/api/v3_0/tests/test_asset_jobs_api_fresh_db.py new file mode 100644 index 0000000000..cd4bf1f156 --- /dev/null +++ b/flexmeasures/api/v3_0/tests/test_asset_jobs_api_fresh_db.py @@ -0,0 +1,96 @@ +import json +from datetime import timedelta + +import pytest +from flask import url_for + +from flexmeasures import Sensor +from flexmeasures.data.models.automations import Automation +from flexmeasures.data.models.data_sources import DataSource +from flexmeasures.data.models.generic_assets import GenericAsset + + +@pytest.mark.parametrize( + ("requesting_user", "can_read_automation"), + ( + ("test_dummy_user_3@seita.nl", False), + ("test_admin_user@seita.nl", True), + ), + indirect=["requesting_user"], +) +def test_asset_jobs_redact_inaccessible_automation_provenance( + app, + client, + fresh_db, + setup_accounts_fresh_db, + setup_roles_users_fresh_db, + setup_generic_assets_fresh_db, + clean_redis, + requesting_user, + can_read_automation, +): + source_asset = setup_generic_assets_fresh_db["test_battery"] + target_asset = GenericAsset( + name="Cross-organisation forecast target", + generic_asset_type=source_asset.generic_asset_type, + owner=setup_accounts_fresh_db["Dummy"], + ) + target_sensor = Sensor( + name="target sensor", + generic_asset=target_asset, + unit="MW", + event_resolution=timedelta(minutes=15), + ) + fresh_db.session.add_all([target_asset, target_sensor]) + fresh_db.session.flush() + generator = DataSource( + name="asset jobs automation generator", + type="forecaster", + model="TrainPredictPipeline", + ) + automation = Automation( + asset=source_asset, + generator=generator, + name="Confidential source automation", + type="forecasts", + cronstr="0 6 * * *", + parameters={"sensor": target_sensor.id}, + ) + fresh_db.session.add(automation) + fresh_db.session.flush() + + job = app.queues["forecasting"].enqueue( + sum, + [1, 2], + meta={ + "sensor_id": target_sensor.id, + "trigger": { + "origin": "automation", + "automation_id": automation.id, + }, + }, + ) + app.job_cache.add( + target_sensor.id, + job.id, + queue="forecasting", + asset_or_sensor_type="sensor", + ) + + response = client.get(url_for("AssetAPI:get_jobs", id=target_asset.id)) + + assert response.status_code == 200 + assert len(response.json["jobs"]) == 1 + job_data = response.json["jobs"][0] + metadata = json.loads(job_data["metadata"]) + assert metadata["trigger"]["origin"] == "automation" + if can_read_automation: + assert ( + job_data["created_via"] + == f"automation '{automation.name}' ({automation.id})" + ) + assert metadata["trigger"]["automation_id"] == automation.id + else: + assert job_data["created_via"] == "automation" + assert "automation_id" not in metadata["trigger"] + assert automation.name not in response.text diff --git a/flexmeasures/api/v3_0/tests/test_automations_api.py b/flexmeasures/api/v3_0/tests/test_automations_api.py new file mode 100644 index 0000000000..39652038b8 --- /dev/null +++ b/flexmeasures/api/v3_0/tests/test_automations_api.py @@ -0,0 +1,174 @@ +"""Tests for the automations endpoints (GET /api/v3_0/assets//automations[/]).""" + +from __future__ import annotations + +from datetime import datetime, timezone + +import pytest +from flask import url_for + +from flexmeasures.data.models.automations import Automation +from flexmeasures.data.models.data_sources import DataSource + + +@pytest.fixture(scope="module") +def add_automations(db, add_battery_assets): + battery = add_battery_assets["Test battery"] + generator = DataSource( + name="automations API test generator", + type="forecaster", + model="TrainPredictPipeline", + ) + automations = [ + Automation( + asset_id=battery.id, + generator=generator, + type="forecasts", + name="Day-ahead forecasts", + cronstr="0 6 * * *", + timezone="Europe/Amsterdam", + cursor=datetime(2026, 7, 11, 4, 0, tzinfo=timezone.utc), + active=True, + parameters={"sensor": battery.sensors[0].id}, + ), + Automation( + asset_id=battery.id, + generator=generator, + type="forecasts", + name="Intraday forecasts", + cronstr="0 * * * *", + timezone="UTC", + cursor=datetime(2026, 7, 11, 5, 0, tzinfo=timezone.utc), + active=False, + parameters={"sensor": battery.sensors[0].id}, + ), + ] + db.session.add_all(automations) + db.session.flush() + return automations + + +@pytest.mark.parametrize( + "requesting_user, expected_status_code", + [ + (None, 401), # not logged in + ("test_prosumer_user@seita.nl", 200), # same account + ("test_dummy_user_3@seita.nl", 403), # different account + ], + indirect=["requesting_user"], +) +def test_get_automations_auth( + app, + add_battery_assets, + add_automations, + requesting_user, + expected_status_code, +): + battery = add_battery_assets["Test battery"] + with app.test_client() as client: + response = client.get( + url_for("AssetAPI:get_automations", id=battery.id), + ) + assert response.status_code == expected_status_code + + +@pytest.mark.parametrize( + "requesting_user", ["test_prosumer_user@seita.nl"], indirect=True +) +def test_get_automations( + app, + add_battery_assets, + add_automations, + requesting_user, +): + battery = add_battery_assets["Test battery"] + with app.test_client() as client: + response = client.get( + url_for("AssetAPI:get_automations", id=battery.id), + ) + assert response.status_code == 200 + automations = response.json["automations"] + assert len(automations) == 2 + day_ahead = next(a for a in automations if a["name"] == "Day-ahead forecasts") + assert day_ahead["type"] == "forecasts" + assert day_ahead["cronstr"] == "0 6 * * *" + assert day_ahead["timezone"] == "Europe/Amsterdam" + assert day_ahead["cursor"] == "2026-07-11T04:00:00+00:00" + assert day_ahead["recurrence_description"] == "At 06:00" + assert day_ahead["active"] is True + assert day_ahead["created_at"] is not None + # generator and parameters are not listed + assert "generator_id" not in day_ahead + assert "generator" not in day_ahead + assert "parameters" not in day_ahead + + +@pytest.mark.parametrize( + "requesting_user", ["test_prosumer_user@seita.nl"], indirect=True +) +def test_get_automation_details( + app, + add_battery_assets, + add_automations, + requesting_user, +): + battery = add_battery_assets["Test battery"] + automation = add_automations[0] + with app.test_client() as client: + response = client.get( + url_for( + "AssetAPI:get_automation", + id=battery.id, + automation_id=automation.id, + ), + ) + assert response.status_code == 200 + assert response.json["name"] == "Day-ahead forecasts" + assert response.json["timezone"] == "Europe/Amsterdam" + assert response.json["cursor"] == "2026-07-11T04:00:00+00:00" + assert response.json["parameters"] == {"sensor": battery.sensors[0].id} + assert response.json["job_stats"] == {} # this automation has not queued any jobs + # the sensor to forecast is both read from (its history) and written to + sensor = {"id": battery.sensors[0].id, "name": battery.sensors[0].name} + assert response.json["input_sensors"] == [sensor] + assert response.json["output_sensors"] == [sensor] + + +@pytest.mark.parametrize( + "requesting_user", ["test_prosumer_user@seita.nl"], indirect=True +) +def test_get_automation_of_other_asset( + app, + add_battery_assets, + 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"] + automation = add_automations[0] + with app.test_client() as client: + response = client.get( + url_for( + "AssetAPI:get_automation", + id=other_asset.id, + automation_id=automation.id, + ), + ) + assert response.status_code == 404 + + +@pytest.mark.parametrize( + "requesting_user", ["test_prosumer_user@seita.nl"], indirect=True +) +def test_get_nonexistent_automation( + app, + add_battery_assets, + add_automations, + requesting_user, +): + battery = add_battery_assets["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 diff --git a/flexmeasures/api/v3_0/tests/test_automations_api_fresh_db.py b/flexmeasures/api/v3_0/tests/test_automations_api_fresh_db.py new file mode 100644 index 0000000000..cc6d033c6a --- /dev/null +++ b/flexmeasures/api/v3_0/tests/test_automations_api_fresh_db.py @@ -0,0 +1,69 @@ +"""Permission regressions for automation API details.""" + +from datetime import timedelta + +import pytest +from flask import url_for + +from flexmeasures.data.models.automations import Automation +from flexmeasures import Forecaster +from flexmeasures.data.models.time_series import Sensor +from flexmeasures.data.services.data_sources import get_data_generator + + +@pytest.mark.parametrize( + "requesting_user", ["test_prosumer_user@seita.nl"], indirect=True +) +def test_details_reject_inaccessible_sensor_metadata( + client, + fresh_db, + setup_roles_users_fresh_db, + setup_generic_assets_fresh_db, + requesting_user, +): + prosumer_asset = setup_generic_assets_fresh_db["test_battery"] + supplier_asset = setup_generic_assets_fresh_db["test_wind_turbine"] + output_sensor = Sensor( + name="prosumer output", + unit="MW", + event_resolution=timedelta(minutes=15), + generic_asset=prosumer_asset, + ) + hidden_sensor = Sensor( + name="private supplier regressor", + unit="MW", + event_resolution=timedelta(minutes=15), + generic_asset=supplier_asset, + ) + fresh_db.session.add_all([output_sensor, hidden_sensor]) + fresh_db.session.flush() + forecaster = get_data_generator( + source=None, + model="TrainPredictPipeline", + config={"regressors": [hidden_sensor.id]}, + save_config=True, + data_generator_type=Forecaster, + ) + assert forecaster is not None + generator = forecaster.data_source + automation = Automation( + asset=prosumer_asset, + generator=generator, + type="forecasts", + name="Cross-organisation details", + cronstr="0 6 * * *", + parameters={"sensor": output_sensor.id}, + ) + fresh_db.session.add(automation) + fresh_db.session.commit() + + response = client.get( + url_for( + "AssetAPI:get_automation", + id=prosumer_asset.id, + automation_id=automation.id, + ) + ) + + assert response.status_code == 403 + assert hidden_sensor.name not in response.text diff --git a/flexmeasures/api/v3_0/tests/test_sources_api.py b/flexmeasures/api/v3_0/tests/test_sources_api.py index 9cf49c7ca4..23c8a979a2 100644 --- a/flexmeasures/api/v3_0/tests/test_sources_api.py +++ b/flexmeasures/api/v3_0/tests/test_sources_api.py @@ -300,3 +300,69 @@ def test_get_sources_only_latest_tie_break_by_id( # Higher id must win the tie assert source_higher_id.id in latest_ids assert source_lower_id.id not in latest_ids + + +@pytest.mark.parametrize( + "requesting_user", + ["test_prosumer_user@seita.nl"], + indirect=True, +) +def test_get_source(client, setup_api_test_data, requesting_user, db): + """One data source can be looked up in full, including its attributes.""" + prosumer_user = find_user_by_email("test_prosumer_user@seita.nl") + source = DataSource( + name="SomeForecaster", + type="forecaster", + model="TrainPredictPipeline", + version="1", + account=prosumer_user.account, + attributes={"data_generator": {"config": {"model": "CustomLGBM"}}}, + ) + db.session.add(source) + db.session.flush() + + response = client.get(url_for("SourceAPI:get", id=source.id)) + assert response.status_code == 200 + assert response.json["id"] == source.id + assert response.json["name"] == "SomeForecaster" + assert response.json["model"] == "TrainPredictPipeline" + assert response.json["user_id"] is None # unset fields are shown, too + assert response.json["attributes"] == { + "data_generator": {"config": {"model": "CustomLGBM"}} + } + + +@pytest.mark.parametrize( + "requesting_user, expected_status_code", + [ + (None, 401), # not logged in + ("test_prosumer_user@seita.nl", 403), # different account + ("test_admin_user@seita.nl", 200), # admins see all sources + ], + indirect=["requesting_user"], +) +def test_get_source_auth( + client, setup_api_test_data, requesting_user, expected_status_code, db +): + """A data source of another account cannot be looked up.""" + supplier_user = find_user_by_email("test_supplier_user_4@seita.nl") + source = DataSource( + name="PrivateSupplierSource", + type="demo script", + account=supplier_user.account, + ) + db.session.add(source) + db.session.flush() + + response = client.get(url_for("SourceAPI:get", id=source.id)) + assert response.status_code == expected_status_code + + +@pytest.mark.parametrize( + "requesting_user", + ["test_prosumer_user@seita.nl"], + indirect=True, +) +def test_get_nonexistent_source(client, setup_api_test_data, requesting_user): + response = client.get(url_for("SourceAPI:get", id=99999)) + assert response.status_code == 404 diff --git a/flexmeasures/cli/data_add.py b/flexmeasures/cli/data_add.py index 13c45ba851..052b7ae473 100755 --- a/flexmeasures/cli/data_add.py +++ b/flexmeasures/cli/data_add.py @@ -5,7 +5,7 @@ from __future__ import annotations from contextlib import nullcontext, redirect_stdout -from datetime import datetime, timedelta +from datetime import date, datetime, timedelta from typing import Dict, Any from flexmeasures.data.schemas.forecasting.pipeline import ( TrainPredictPipelineConfigSchema, @@ -19,12 +19,16 @@ from io import StringIO from string import Template -from marshmallow import validate, ValidationError +from marshmallow import Schema, validate, ValidationError import pandas as pd import pytz from flask import current_app as app from flask.cli import with_appcontext import click + +# NB the type: ignore comments here and on ctx.get_parameter_source below are needed because types-Flask pins types-click 7.1, +# whose stubs shadow the inline types that click ships itself, and predate both of these (added in click 8.0). +from click.core import ParameterSource # type: ignore[attr-defined] import getpass from sqlalchemy.exc import IntegrityError from sqlalchemy import func, select @@ -52,6 +56,7 @@ 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 ( @@ -67,6 +72,11 @@ ) from flexmeasures.data.models.data_sources import DataSource, DEFAULT_DATASOURCE_TYPES from flexmeasures.data.models.annotations import Annotation, get_or_create_annotation +from flexmeasures.data.models.automations import ( + Automation, + get_default_automation_timezone, +) +from flexmeasures.data.schemas.automations import CronField, TimezoneField from flexmeasures.data.schemas import ( AccountIdField, AwareDateTimeField, @@ -1360,6 +1370,142 @@ def add_holidays( ) +def _normalize_yaml_value(value): + """Convert YAML-native date values to the strings expected by our schemas.""" + if isinstance(value, (date, datetime)): + return value.isoformat() + if isinstance(value, dict): + return {key: _normalize_yaml_value(item) for key, item in value.items()} + if isinstance(value, list): + return [_normalize_yaml_value(item) for item in value] + return value + + +def _load_yaml_mapping(stream: TextIOBase, option_name: str) -> dict: + """Load a YAML/JSON CLI option file whose top level must be an object.""" + value = yaml.safe_load(stream) + if value is None: + return {} + if not isinstance(value, dict): + raise click.UsageError( + f"The {option_name} file must contain a YAML or JSON object " + "at the top level." + ) + return _normalize_yaml_value(value) + + +def _normalize_yaml_mapping(value, option_name: str) -> dict: + """Validate and normalize YAML/JSON data returned by the editor.""" + if value is None: + return {} + if not isinstance(value, dict): + raise click.UsageError( + f"The {option_name} data must contain a YAML or JSON object " + "at the top level." + ) + return _normalize_yaml_value(value) + + +def _find_options_given_on_command_line( + options_by_param_name: dict[str, str], + *schemas: Schema, +) -> list[str]: + """List which of the given CLI options were actually passed on the command line. + + Options are looked up by their click parameter name, both from an explicit mapping + of parameter names to option names, and from the CLI metadata of any schema fields + (as added by `add_cli_options_from_schema`). + """ + ctx = click.get_current_context(silent=True) + if ctx is None: + return [] + options_by_param_name = dict(options_by_param_name) + for schema in schemas: + for field_name, field in schema.fields.items(): + cli = field.metadata.get("cli") + if cli: + options_by_param_name[field_name] = cli["option"] + return [ + option + for param_name, option in options_by_param_name.items() + if ctx.get_parameter_source(param_name) == ParameterSource.COMMANDLINE # type: ignore[attr-defined] + ] + + +def _assemble_forecaster_config_and_parameters( + kwargs: dict, + source: DataSource | None = None, + config_file: TextIOBase | None = None, + parameters_file: TextIOBase | None = None, + edit_config: bool = False, + edit_parameters: bool = False, +) -> tuple[dict, dict]: + """Build the forecaster config and (serialized) forecast parameters + from optional files, editors and remaining CLI options. + + CLI options matching config schema fields are popped from kwargs into the config; + all remaining options become (kebab-cased) parameters. None values are dropped. + """ + config = dict() + if config_file: + config = _load_yaml_mapping(config_file, "--config") + for field_name, field in TrainPredictPipelineConfigSchema._declared_fields.items(): + field_value = kwargs.pop(field_name, None) + if field_value is not None: + if field_name in { + "future_regressors", + "past_regressors", + "regressors", + }: + field_value = _parse_regressor_cli_values(field_value) + config[field.data_key] = field_value + + if edit_config: + config = _normalize_yaml_mapping( + launch_editor("/tmp/config.yml"), "--edit-config" + ) + + if source is not None: + # The forecaster class and its configuration are read from the data source's data + # generator attributes, so anything configured here would be silently ignored. + # Only options actually given on the command line count: the configuration options + # that were left out still show up in the config, with their schema defaults. + conflicting_options = _find_options_given_on_command_line( + { + "forecaster_class": "--forecaster", + "config_file": "--config", + "edit_config": "--edit-config", + }, + TrainPredictPipelineConfigSchema(), + ) + if conflicting_options: + raise click.UsageError( + f"{flexmeasures_inflection.join_words_into_a_list(conflicting_options)} cannot be" + " combined with --source: --source uses the forecaster configuration stored with" + " that source. Omit --source to use the supplied configuration options." + ) + + parameters = dict() + if parameters_file: + parameters = _load_yaml_mapping(parameters_file, "--parameters") + + if edit_parameters: + parameters = _normalize_yaml_mapping( + launch_editor("/tmp/parameters.yml"), "--edit-parameters" + ) + + # Move remaining kwargs to parameters, converting from snake_case to kebab-case to match schema expectation + for k, v in kwargs.items(): + kebab_key = snake_to_kebab(k) + if kebab_key not in parameters: + parameters[kebab_key] = v + + # Drop None values + parameters = {k: v for k, v in parameters.items() if v is not None} + + return config, parameters + + @fm_add_data.command("forecasts") @click.option( "--resolution", @@ -1469,42 +1615,14 @@ def add_forecast( # noqa: C901 ) del kwargs["resolution"] - config = dict() - - if config_file: - config = yaml.safe_load(config_file) - for field_name, field in TrainPredictPipelineConfigSchema._declared_fields.items(): - if field_value := kwargs.pop(field_name, None): - if field_name in { - "future_regressors", - "past_regressors", - "regressors", - }: - field_value = _parse_regressor_cli_values(field_value) - config[field.data_key] = field_value - - if edit_config: - config = launch_editor("/tmp/config.yml") - - if source is not None and config: - raise click.UsageError( - "--source uses the forecaster configuration stored with that source. " - "Omit --source to use the supplied configuration options." - ) - - parameters = dict() - - if parameters_file: - parameters = yaml.safe_load(parameters_file) - - if edit_parameters: - parameters = launch_editor("/tmp/parameters.yml") - - # Move remaining kwargs to parameters, converting from snake_case to kebab-case to match schema expectation - for k, v in kwargs.items(): - kebab_key = snake_to_kebab(k) - if kebab_key not in parameters: - parameters[kebab_key] = v + config, parameters = _assemble_forecaster_config_and_parameters( + kwargs, + source, + config_file, + parameters_file, + edit_config, + edit_parameters, + ) try: forecaster = get_data_generator( @@ -1519,9 +1637,9 @@ def add_forecast( # noqa: C901 f"Invalid forecasting configuration: {e.messages}" ) from e + forecaster.set_job_trigger("CLI") + try: - # Drop None values - parameters = {k: v for k, v in parameters.items() if v is not None} pipeline_returns = forecaster.compute(as_job=as_job, parameters=parameters) # Empty result @@ -1550,6 +1668,189 @@ def add_forecast( # noqa: C901 raise +@fm_add_data.command("automation") +@with_appcontext +@click.option( + "--asset", + "asset", + required=True, + type=AssetIdField(), + help="ID of the asset to automate a recurring task for.", +) +@click.option( + "--name", + "name", + required=True, + type=click.STRING, + help="Name of the automation.", +) +@click.option( + "--cron", + "cronstr", + default="0 0 * * *", + show_default=True, + type=CronField(), + help='Recurrence as a standard five-field cron expression, e.g. "0 6 * * *" for daily at 06:00.' + " The expression is interpreted in the automation timezone. Defaults to daily at midnight.", +) +@click.option( + "--timezone", + "timezone", + default=get_default_automation_timezone, + show_default="FLEXMEASURES_TIMEZONE", + type=TimezoneField(), + help='IANA timezone in which to interpret --cron, e.g. "UTC" or "Europe/Amsterdam". Defaults to FLEXMEASURES_TIMEZONE.', +) +@click.option( + "--type", + "automation_type", + default="forecasts", + show_default=True, + type=click.Choice(Automation.SUPPORTED_TYPES), + help="Type of task to automate.", +) +@click.option( + "--inactive", + "inactive", + is_flag=True, + help="Add this flag to create the automation in deactivated state.", +) +@click.option( + "--forecaster", + "forecaster_class", + default=None, + type=click.STRING, + help="Forecaster class registered in flexmeasures.data.models.forecasting or in an available flexmeasures plugin." + " Defaults to TrainPredictPipeline. Use the command `flexmeasures show forecasters` to list all the available forecasters." + " Cannot be combined with --source, which already determines the forecaster.", +) +@click.option( + "--source", + "source", + required=False, + type=DataSourceIdField(), + help="DataSource ID of the `Forecaster`. The forecaster class and its configuration are read from" + " the data source's data generator attributes, so --forecaster and --config are not needed (or allowed) with it.", +) +@click.option( + "--config", + "config_file", + required=False, + type=click.File("r"), + help="Path to the JSON or YAML file with the configuration of the forecaster." + " Cannot be combined with --source, which already determines the configuration.", +) +@click.option( + "--parameters", + "parameters_file", + required=False, + type=click.File("r"), + help="Path to the JSON or YAML file with the forecast parameters (passed to the compute step on each run of the automation).", +) +@add_cli_options_from_schema( + ForecasterParametersSchema(), hidden=True, force_optional=True +) +@add_cli_options_from_schema( + TrainPredictPipelineConfigSchema(), hidden=True, force_optional=True +) +def add_automation( + asset: GenericAsset, + name: str, + cronstr: str, + timezone: str, + automation_type: str, + inactive: bool = False, + forecaster_class: str | None = None, + source: DataSource | None = None, + config_file: TextIOBase | None = None, + parameters_file: TextIOBase | None = None, + **kwargs, +): + """ + Add an automation: a recurring task (for now, computing forecasts) on an asset. + + \b + Example + flexmeasures add automation --asset 3 --name "Day-ahead PV forecasts" + --cron "0 6 * * *" --timezone Europe/Amsterdam + --parameters forecast-parameters.yml + + The forecaster configuration is stored on a data source, and the forecast + parameters are validated and stored on the automation itself. + Each time the automation runs, forecasting jobs are queued + (see `flexmeasures jobs run-automations`). + + Alternatively, pass an existing data source (--source) to reuse the forecaster + and configuration stored on it. + + Every forecaster and pipeline option that `flexmeasures add forecast` accepts is accepted here, too, + but is left out of the help text above to keep it focused on the automation itself; + run `flexmeasures add forecast --help` to see them. + A configuration option given on the command line overrides the same setting from --config, + while a parameter from --parameters takes precedence over the matching command-line option. + """ + if forecaster_class is None: + forecaster_class = "TrainPredictPipeline" + + config, parameters = _assemble_forecaster_config_and_parameters( + kwargs, source, config_file, parameters_file + ) + + # Validate the parameters using the forecast parameters schema (we store them serialized) + 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, + config=config, + save_config=True, + data_generator_type=Forecaster, + ) + 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() + + 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." + ) + db.session.commit() + click.secho( + f"Successfully created {'inactive ' if inactive else ''}automation '{name}' (ID: {automation.id})" + f" to compute {automation_type} for asset {asset.id}, recurring per cron string '{cronstr}' in timezone '{timezone}'.", + **MsgStyle.SUCCESS, + ) + + @fm_add_data.command("schedule") @with_appcontext @click.option( diff --git a/flexmeasures/cli/data_delete.py b/flexmeasures/cli/data_delete.py index 6b1fbb7431..1911a6b7e3 100644 --- a/flexmeasures/cli/data_delete.py +++ b/flexmeasures/cli/data_delete.py @@ -17,7 +17,11 @@ 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.models.time_series import Sensor, TimedBelief from flexmeasures.data.schemas import ( AccountIdField, @@ -33,6 +37,7 @@ done, DeprecatedOption, DeprecatedOptionsCommand, + MsgStyle, ) from flexmeasures.utils.flexmeasures_inflection import join_words_into_a_list from flexmeasures.utils.secrets_utils import delete_secret, get_secret_paths @@ -273,6 +278,35 @@ def delete_asset_and_data(asset: GenericAsset, force: bool): db.session.commit() +@fm_delete_data.command("automation") +@with_appcontext +@click.option( + "--id", + "automation", + required=True, + type=AutomationIdField(), + help="ID of the automation to delete.", +) +@click.option("--force/--no-force", default=False, help="Skip confirmation prompt.") +def delete_automation(automation: Automation, force: bool): + """ + Delete an automation. + """ + 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) + db.session.commit() + click.secho( + f"Successfully deleted automation '{automation.name}' (ID: {automation.id}).", + **MsgStyle.SUCCESS, + ) + + @fm_delete_data.command("structure") @with_appcontext @click.option( @@ -663,6 +697,22 @@ def delete_sensor( .select_from(TimedBelief) .where(TimedBelief.sensor_id.in_([sensor.id for sensor in sensors])) ).scalar_one() + # An automation refers to its sensors by ID in its parameters, which no foreign key protects, + # so deleting one here would leave the automation to fail on its next run. Say so up front. + for sensor in sensors: + involved_automations = get_automations_involving_sensor(sensor) + if involved_automations: + click.secho( + f"Sensor {sensor.id} is used by " + + join_words_into_a_list( + [ + f"automation '{automation.name}' ({automation.id})" + for automation in involved_automations + ] + ) + + ", which will fail on the next run after this deletion.", + **MsgStyle.WARN, + ) click.confirm( f"Delete {', '.join(sensor.__repr__() for sensor in sensors)}, along with {n_beliefs} beliefs?", abort=True, diff --git a/flexmeasures/cli/data_edit.py b/flexmeasures/cli/data_edit.py index 24cd065c6f..3eaecf7426 100644 --- a/flexmeasures/cli/data_edit.py +++ b/flexmeasures/cli/data_edit.py @@ -19,7 +19,16 @@ 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.audit_log import AssetAuditLog, AuditLog +from flexmeasures.data.schemas.automations import ( + AutomationIdField, + CronField, + TimezoneField, +) from flexmeasures.data.models.time_series import TimedBelief from flexmeasures.data.utils import save_to_db from flexmeasures.cli.utils import ( @@ -54,6 +63,84 @@ def fm_edit_data(): """FlexMeasures: Edit data.""" +@fm_edit_data.command("automation") +@with_appcontext +@click.option( + "--id", + "automation", + required=True, + type=AutomationIdField(), + help="ID of the automation to edit.", +) +@click.option( + "--name", + "name", + required=False, + type=click.STRING, + help="New name of the automation.", +) +@click.option( + "--cron", + "cronstr", + required=False, + type=CronField(), + help="New recurrence as a standard five-field cron expression, interpreted in the automation timezone.", +) +@click.option( + "--timezone", + "timezone", + required=False, + type=TimezoneField(), + help='New IANA timezone in which to interpret the cron recurrence, e.g. "Europe/Amsterdam".', +) +@click.option( + "--activate/--deactivate", + "active", + default=None, + help="Activate or deactivate the automation.", +) +def edit_automation( + automation: Automation, + name: str | None = None, + cronstr: str | None = None, + timezone: str | None = None, + 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 + 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)}.", + **MsgStyle.SUCCESS, + ) + + # Plan fields which may be cleared back to NULL (meaning: server-wide behaviour applies) CLEARABLE_PLAN_FIELDS = [ "default-rate-limit", diff --git a/flexmeasures/cli/jobs.py b/flexmeasures/cli/jobs.py index 27be18f4a7..541a1e83a6 100644 --- a/flexmeasures/cli/jobs.py +++ b/flexmeasures/cli/jobs.py @@ -30,7 +30,14 @@ from tabulate import tabulate import pandas as pd +from flexmeasures.data import db from flexmeasures.data.schemas import AssetIdField, SensorIdField +from flexmeasures.data.services.automations import ( + claim_due_automation, + floor_to_minute, + get_due_automations, + run_automation, +) from flexmeasures.data.services.scheduling import handle_scheduling_exception from flexmeasures.data.services.forecasting import handle_forecasting_exception from flexmeasures.utils.job_utils import work_on_rq @@ -54,6 +61,72 @@ def fm_jobs(): """FlexMeasures: Job queueing.""" +@fm_jobs.command("run-automations") +@with_appcontext +def run_automations(): + """ + Queue jobs for all automations that are due to run this minute. + + Each cron string is interpreted in the automation's timezone. + Missed forecast runs are caught up once, with several missed runs coalesced into the latest useful forecast. + Run this command once per minute (e.g. via cron): + + \b + * * * * * flexmeasures jobs run-automations + + A Redis-based guard allows at most one queueing attempt per scheduled run. + A failed attempt is not retried automatically, because it may already have queued some jobs. + """ + now = floor_to_minute(server_now()) + due_automations = get_due_automations(now) + if not due_automations: + click.secho(f"No automations due at {now}.", **MsgStyle.SUCCESS) + return + + connection = app.queues["forecasting"].connection + n_run = 0 + n_failed = 0 + for due_automation in due_automations: + automation = due_automation.automation + # Guard the canonical run, including catch-ups and repeated wall times. + guard_key = ( + f"automation-run:{automation.id}:{due_automation.scheduled_at.isoformat()}" + ) + if not connection.set(guard_key, 1, nx=True, ex=120): + click.secho( + f"Automation {automation.id} ('{automation.name}') was already attempted for {due_automation.scheduled_at}. " + "Skipping to avoid duplicate jobs.", + **MsgStyle.WARN, + ) + continue + if not claim_due_automation(due_automation): + click.secho( + f"Automation {automation.id} ('{automation.name}') run {due_automation.scheduled_at} was already claimed. Skipping to avoid duplicate jobs.", + **MsgStyle.WARN, + ) + continue + try: + returns = run_automation(automation) + n_jobs = returns.get("n_jobs") if returns else 0 + click.secho( + f"Automation {automation.id} ('{automation.name}') queued {n_jobs} forecasting job(s) for asset {automation.asset_id}.", + **MsgStyle.SUCCESS, + ) + n_run += 1 + except Exception as e: + db.session.rollback() + # Queueing a multi-cycle forecast is not transactional. Keep the guard + # because this attempt may have queued some jobs before failing. + click.secho( + f"Automation {automation.id} ('{automation.name}') failed to queue jobs: {e}", + **MsgStyle.ERROR, + ) + n_failed += 1 + if n_failed: + click.secho(f"{n_run} automation(s) ran, {n_failed} failed.", **MsgStyle.ERROR) + raise click.exceptions.Exit(1) + + @fm_jobs.command("stats") @with_appcontext @click.option( diff --git a/flexmeasures/cli/tests/test_automations.py b/flexmeasures/cli/tests/test_automations.py new file mode 100644 index 0000000000..8373b5cf18 --- /dev/null +++ b/flexmeasures/cli/tests/test_automations.py @@ -0,0 +1,912 @@ +from datetime import datetime, timedelta, timezone +import json + +import pytest +from types import SimpleNamespace + +from sqlalchemy import select + +from flexmeasures import Sensor +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.cli.tests.utils import to_flags +from flexmeasures.utils.time_utils import get_timezone + + +@pytest.fixture(scope="function") +def clean_redis(app): + app.redis_connection.flushdb() + yield + app.redis_connection.flushdb() + + +@pytest.fixture() +def automation_scope_assets(fresh_db, setup_dummy_data): + root_sensor = fresh_db.session.get(Sensor, setup_dummy_data[0]) + root_asset = root_sensor.generic_asset + asset_type = root_asset.generic_asset_type + + ancestor = GenericAsset(name="automation ancestor", generic_asset_type=asset_type) + child = GenericAsset( + name="automation child", + generic_asset_type=asset_type, + parent_asset=root_asset, + ) + grandchild = GenericAsset( + name="automation grandchild", + generic_asset_type=asset_type, + parent_asset=child, + ) + unrelated = GenericAsset(name="automation unrelated", generic_asset_type=asset_type) + root_asset.parent_asset = ancestor + + sensors = {"root": root_sensor} + for name, asset in ( + ("ancestor", ancestor), + ("child", child), + ("grandchild", grandchild), + ("unrelated", unrelated), + ): + sensors[name] = Sensor( + f"{name} output", + generic_asset=asset, + event_resolution=root_sensor.event_resolution, + unit=root_sensor.unit, + ) + + fresh_db.session.add_all( + [ancestor, child, grandchild, unrelated, *sensors.values()] + ) + fresh_db.session.commit() + return { + "root_asset": root_asset, + "child_asset": child, + "unrelated_asset": unrelated, + "sensors": sensors, + } + + +def test_add_edit_delete_automation(app, fresh_db, setup_dummy_data): + """Roundtrip: create an automation, edit it, then delete it, checking the audit log along the way.""" + from flexmeasures.cli.data_add import add_automation + from flexmeasures.cli.data_edit import edit_automation + from flexmeasures.cli.data_delete import delete_automation + + sensor_id = setup_dummy_data[0] + runner = app.test_cli_runner() + + # add + cli_input = { + "asset": 1, + "name": "Test forecasts", + "cron": "0 6 * * *", + "timezone": "Europe/Amsterdam", + "sensor": sensor_id, + } + result = runner.invoke(add_automation, to_flags(cli_input)) + assert "Successfully created" in result.output, result.output + automation = fresh_db.session.execute( + select(Automation).filter_by(name="Test forecasts") + ).scalar_one_or_none() + assert automation is not None + assert automation.active is True + assert automation.type == "forecasts" + assert automation.cronstr == "0 6 * * *" + assert automation.timezone == "Europe/Amsterdam" + # CLI option values are stored as provided (strings); they are coerced by the schema when the automation runs + assert automation.parameters == {"sensor": str(sensor_id)} + assert automation.generator is not None + assert automation.generator.model == "TrainPredictPipeline" + assert fresh_db.session.execute( + select(AssetAuditLog).filter(AssetAuditLog.event.like("Created automation%")) + ).scalar_one_or_none() + + # edit + result = runner.invoke( + edit_automation, + [ + "--id", + automation.id, + "--name", + "Renamed", + "--timezone", + "UTC", + "--deactivate", + ], + ) + assert "Successfully updated" in result.output, result.output + assert automation.name == "Renamed" + assert automation.timezone == "UTC" + assert automation.active is False + assert fresh_db.session.execute( + select(AssetAuditLog).filter(AssetAuditLog.event.like("Updated automation%")) + ).scalar_one_or_none() + + # delete + result = runner.invoke(delete_automation, ["--id", automation.id, "--force"]) + assert "Successfully deleted" in result.output, result.output + assert fresh_db.session.execute(select(Automation)).scalar_one_or_none() is None + assert fresh_db.session.execute( + select(AssetAuditLog).filter(AssetAuditLog.event.like("Deleted automation%")) + ).scalar_one_or_none() + + +def test_add_automation_default_cron( + app, fresh_db, setup_dummy_data, freeze_server_now +): + """Without --cron, an automation recurs daily.""" + from flexmeasures.cli.data_add import add_automation + from flexmeasures.data.services.automations import ( + claim_due_automation, + get_due_automations, + ) + + # create the automation before the midnight we check, as an automation does not replay runs from before it existed + midnight = get_timezone().localize(datetime(2026, 7, 11, 0, 0)) + freeze_server_now(midnight - timedelta(hours=3)) + + sensor_id = setup_dummy_data[0] + runner = app.test_cli_runner() + result = runner.invoke( + add_automation, + to_flags({"asset": 1, "name": "Daily forecasts", "sensor": sensor_id}), + ) + assert "Successfully created" in result.output, result.output + automation = fresh_db.session.execute( + select(Automation).filter_by(name="Daily forecasts") + ).scalar_one() + assert automation.cronstr == "0 0 * * *" + + # due at midnight in the automation's timezone + due = get_due_automations(midnight) + assert [d.automation.id for d in due] == [automation.id] + + # and, once claimed, not handed out again an hour later + assert claim_due_automation(due[0]) + assert get_due_automations(midnight + timedelta(hours=1)) == [] + + +def test_add_automation_source_conflicts_with_forecaster( + app, fresh_db, setup_dummy_data +): + """--source already determines the forecaster and its config, so combining them fails.""" + from flexmeasures.cli.data_add import add_automation + + sensor_id = setup_dummy_data[0] + runner = app.test_cli_runner() + # first create an automation, so that a data source with a forecaster config exists + result = runner.invoke( + add_automation, + to_flags({"asset": 1, "name": "First", "sensor": sensor_id}), + ) + assert "Successfully created" in result.output, result.output + source_id = ( + fresh_db.session.execute(select(Automation).filter_by(name="First")) + .scalar_one() + .generator_id + ) + + result = runner.invoke( + add_automation, + to_flags( + { + "asset": 1, + "name": "Second", + "sensor": sensor_id, + "source": source_id, + "forecaster": "TrainPredictPipeline", + } + ), + ) + assert result.exit_code != 0 + assert "--forecaster cannot be combined with --source" in result.output + + # a configuration option given on the command line conflicts, too + result = runner.invoke( + add_automation, + to_flags( + { + "asset": 1, + "name": "Second", + "sensor": sensor_id, + "source": source_id, + "regressors": sensor_id, + } + ), + ) + assert result.exit_code != 0 + assert "--regressors cannot be combined with --source" in result.output + + # without the conflicting option, the same data source is simply reused + result = runner.invoke( + add_automation, + to_flags( + {"asset": 1, "name": "Second", "sensor": sensor_id, "source": source_id} + ), + ) + assert "Successfully created" in result.output, result.output + assert ( + fresh_db.session.execute(select(Automation).filter_by(name="Second")) + .scalar_one() + .generator_id + == source_id + ) + + +def test_automation_sensors(app, fresh_db, setup_dummy_data): + """An automation knows which sensors it reads from and writes to.""" + from flexmeasures.cli.data_add import add_automation + from flexmeasures.data.services.automations import get_automations_feeding_sensor + + sensor_id, regressor_id = setup_dummy_data[0], setup_dummy_data[1] + runner = app.test_cli_runner() + result = runner.invoke( + add_automation, + to_flags( + { + "asset": 1, + "name": "Test forecasts", + "sensor": sensor_id, + "regressors": regressor_id, + } + ), + ) + assert "Successfully created" in result.output, result.output + automation = fresh_db.session.execute( + select(Automation).filter_by(name="Test forecasts") + ).scalar_one() + assert [sensor.id for sensor in automation.output_sensors] == [sensor_id] + assert sorted(sensor.id for sensor in automation.input_sensors) == sorted( + [sensor_id, regressor_id] + ) + + sensor = fresh_db.session.get(Sensor, sensor_id) + assert [a.id for a in get_automations_feeding_sensor(sensor)] == [automation.id] + + +def test_delete_sensor_warns_about_automations_using_it( + app, fresh_db, setup_dummy_data +): + """Deleting a sensor an automation uses is possible, but says which automations will break.""" + from flexmeasures.cli.data_add import add_automation + from flexmeasures.cli.data_delete import delete_sensor + + sensor_id, regressor_id = setup_dummy_data[0], setup_dummy_data[1] + runner = app.test_cli_runner() + result = runner.invoke( + add_automation, + to_flags( + { + "asset": 1, + "name": "Test forecasts", + "sensor": sensor_id, + "regressors": regressor_id, + } + ), + ) + assert "Successfully created" in result.output, result.output + + # The regressor is only an input, so it is the case that get_automations_feeding_sensor misses. + result = runner.invoke(delete_sensor, to_flags({"id": regressor_id}), input="n\n") + assert "is used by automation 'Test forecasts'" in result.output, result.output + + # A sensor no automation refers to is deleted without such a warning. + unrelated = Sensor( + name="unrelated", + generic_asset=fresh_db.session.get(Sensor, sensor_id).generic_asset, + event_resolution=timedelta(minutes=15), + ) + fresh_db.session.add(unrelated) + fresh_db.session.commit() + result = runner.invoke(delete_sensor, to_flags({"id": unrelated.id}), input="n\n") + assert "is used by automation" not in result.output, result.output + + +def test_automation_sensors_with_source_filtered_regressor( + app, fresh_db, setup_dummy_data +): + """A regressor that filters on sources still counts as an input sensor. + + The source filters only narrow down which beliefs are read from that sensor, + so leaving it out would understate which sensors the automation reads from. + """ + from flexmeasures.cli.data_add import add_automation + + sensor_id, regressor_id = setup_dummy_data[0], setup_dummy_data[1] + runner = app.test_cli_runner() + result = runner.invoke( + add_automation, + to_flags( + { + "asset": 1, + "name": "Filtered regressor forecasts", + "sensor": sensor_id, + "regressors": json.dumps( + [{"sensor": regressor_id, "source-types": ["forecaster"]}] + ), + } + ), + ) + assert "Successfully created" in result.output, result.output + automation = fresh_db.session.execute( + select(Automation).filter_by(name="Filtered regressor forecasts") + ).scalar_one() + assert sorted(sensor.id for sensor in automation.input_sensors) == sorted( + [sensor_id, regressor_id] + ) + + +def test_automation_sensors_are_unknown_rather_than_empty( + app, fresh_db, setup_dummy_data +): + """When the sensors cannot be worked out, only the display helper is allowed to report none. + + Reporting no sensors to an access check would let the automation pass every check on the sensors it involves, + so the strict helper raises instead. + """ + from flexmeasures.cli.data_add import add_automation + from flexmeasures.data.services.automations import ( + AutomationSensorsUnknown, + get_automation_sensors, + resolve_automation_sensors, + ) + + sensor_id = setup_dummy_data[0] + runner = app.test_cli_runner() + result = runner.invoke( + add_automation, + to_flags({"asset": 1, "name": "Broken forecasts", "sensor": sensor_id}), + ) + assert "Successfully created" in result.output, result.output + automation = fresh_db.session.execute( + select(Automation).filter_by(name="Broken forecasts") + ).scalar_one() + + # the parameters no longer load, as happens when a sensor referred to has been deleted + automation.parameters = {"sensor": "no-such-sensor"} + fresh_db.session.commit() + + assert get_automation_sensors(automation) == { + "input_sensors": [], + "output_sensors": [], + } + with pytest.raises(AutomationSensorsUnknown): + resolve_automation_sensors(automation) + + +@pytest.mark.parametrize("cronstr", ["not a cron string", "0 0 31 2 *"]) +def test_add_automation_invalid_cron(app, fresh_db, setup_dummy_data, cronstr): + from flexmeasures.cli.data_add import add_automation + + sensor_id = setup_dummy_data[0] + runner = app.test_cli_runner() + cli_input = { + "asset": 1, + "name": "Test forecasts", + "cron": cronstr, + "sensor": sensor_id, + } + result = runner.invoke(add_automation, to_flags(cli_input)) + assert result.exit_code != 0 + # NB click reports the offending value; once it reports the validation message + # instead (see PR #2303), the cron string's own error text shows up here. + assert "Invalid value" in result.output + + +def test_add_automation_defaults_to_configured_timezone( + app, fresh_db, setup_dummy_data, monkeypatch +): + from flexmeasures.cli.data_add import add_automation + + monkeypatch.setitem(app.config, "FLEXMEASURES_TIMEZONE", "America/New_York") + result = app.test_cli_runner().invoke( + add_automation, + [ + "--asset", + "1", + "--name", + "Configured timezone", + "--cron", + "0 6 * * *", + "--sensor", + str(setup_dummy_data[0]), + ], + ) + + assert result.exit_code == 0, result.output + automation = fresh_db.session.scalars(select(Automation)).one() + assert automation.timezone == "America/New_York" + + +def test_add_and_edit_automation_reject_invalid_timezone( + app, fresh_db, setup_dummy_data +): + from flexmeasures.cli.data_add import add_automation + from flexmeasures.cli.data_edit import edit_automation + + runner = app.test_cli_runner() + invalid_add = runner.invoke( + add_automation, + [ + "--asset", + "1", + "--name", + "Invalid timezone", + "--cron", + "0 6 * * *", + "--timezone", + "Europe/NotAmsterdam", + "--sensor", + str(setup_dummy_data[0]), + ], + ) + assert invalid_add.exit_code != 0 + assert fresh_db.session.scalars(select(Automation)).all() == [] + + valid_add = runner.invoke( + add_automation, + [ + "--asset", + "1", + "--name", + "Valid timezone", + "--cron", + "0 6 * * *", + "--timezone", + "UTC", + "--sensor", + str(setup_dummy_data[0]), + ], + ) + assert valid_add.exit_code == 0, valid_add.output + automation = fresh_db.session.scalars(select(Automation)).one() + invalid_edit = runner.invoke( + edit_automation, + ["--id", str(automation.id), "--timezone", "Europe/NotAmsterdam"], + ) + assert invalid_edit.exit_code != 0 + assert automation.timezone == "UTC" + + +@pytest.mark.parametrize( + "edit_args", + ( + ["--cron", "15 10 * * *"], + ["--timezone", "Europe/Amsterdam"], + ["--activate"], + ), +) +def test_edit_automation_rebases_cursor( + app, + fresh_db, + setup_dummy_data, + freeze_server_now, + edit_args, +): + from flexmeasures.cli.data_add import add_automation + from flexmeasures.cli.data_edit import edit_automation + + freeze_server_now(datetime(2026, 1, 15, 8, 0, tzinfo=timezone.utc)) + runner = app.test_cli_runner() + add_result = runner.invoke( + add_automation, + [ + "--asset", + "1", + "--name", + "Rebased automation", + "--cron", + "0 10 * * *", + "--timezone", + "UTC", + "--inactive", + "--sensor", + str(setup_dummy_data[0]), + ], + ) + assert add_result.exit_code == 0, add_result.output + automation = fresh_db.session.scalars(select(Automation)).one() + + freeze_server_now(datetime(2026, 1, 15, 10, 0, tzinfo=timezone.utc)) + edit_result = runner.invoke( + edit_automation, ["--id", str(automation.id), *edit_args] + ) + + assert edit_result.exit_code == 0, edit_result.output + assert automation.cursor == datetime(2026, 1, 15, 9, 59, tzinfo=timezone.utc) + + +def test_add_automation_help_focuses_on_automation_options(app): + """The forecast schema options are accepted, but kept out of the help text.""" + from flexmeasures.cli.data_add import add_automation + + result = app.test_cli_runner().invoke(add_automation, ["--help"]) + + assert result.exit_code == 0, result.output + for automation_option in ( + "--asset", + "--name", + "--cron", + "--timezone", + "--config", + "--parameters", + ): + assert automation_option in result.output + for forecast_option in ("--sensor ", "--duration", "--train-start"): + assert forecast_option not in result.output + + +def test_add_automation_accepts_required_sensor_from_parameters_file( + app, fresh_db, setup_dummy_data, tmp_path +): + """The schema requires a sensor, but it may come from --parameters rather than the command line.""" + from flexmeasures.cli.data_add import add_automation + + parameters_file = tmp_path / "parameters.yml" + parameters_file.write_text(f"sensor: {setup_dummy_data[0]}\n") + + result = app.test_cli_runner().invoke( + add_automation, + to_flags( + { + "asset": 1, + "name": "YAML sensor", + "parameters": str(parameters_file), + } + ), + ) + + assert "Successfully created" in result.output, result.output + automation = fresh_db.session.execute( + select(Automation).filter_by(name="YAML sensor") + ).scalar_one() + assert automation.parameters == {"sensor": setup_dummy_data[0]} + + +@pytest.mark.parametrize( + ("output_sensor_name", "should_succeed"), + ( + ("root", True), + ("child", True), + ("grandchild", True), + ("unrelated", False), + ("ancestor", False), + ), +) +def test_add_automation_constrains_output_to_asset_subtree( + app, + fresh_db, + automation_scope_assets, + output_sensor_name, + should_succeed, +): + from flexmeasures.cli.data_add import add_automation + + root_asset = automation_scope_assets["root_asset"] + output_sensor = automation_scope_assets["sensors"][output_sensor_name] + + result = app.test_cli_runner().invoke( + add_automation, + [ + "--asset", + str(root_asset.id), + "--name", + f"{output_sensor_name} output", + "--cron", + "0 6 * * *", + "--sensor", + str(output_sensor.id), + ], + ) + + automations = fresh_db.session.scalars(select(Automation)).all() + if should_succeed: + assert result.exit_code == 0, result.output + assert len(automations) == 1 + else: + assert result.exit_code != 0 + assert "must belong to asset" in result.output + assert automations == [] + + +def test_add_automation_constrains_explicit_output_sensor( + app, fresh_db, automation_scope_assets +): + from flexmeasures.cli.data_add import add_automation + + root_asset = automation_scope_assets["root_asset"] + root_sensor = automation_scope_assets["sensors"]["root"] + unrelated_sensor = automation_scope_assets["sensors"]["unrelated"] + + result = app.test_cli_runner().invoke( + add_automation, + [ + "--asset", + str(root_asset.id), + "--name", + "unrelated explicit output", + "--cron", + "0 6 * * *", + "--sensor", + str(root_sensor.id), + "--sensor-to-save", + str(unrelated_sensor.id), + ], + ) + + assert result.exit_code != 0 + assert "must belong to asset" in result.output + assert fresh_db.session.scalars(select(Automation)).all() == [] + + +@pytest.mark.parametrize( + ("yaml_start", "expected_start"), + ( + ("2026-07-31", "2026-07-31"), + ("2026-07-31T06:00:00+01:00", "2026-07-31T06:00:00+01:00"), + ), +) +def test_add_automation_normalizes_yaml_dates( + app, + fresh_db, + setup_dummy_data, + tmp_path, + yaml_start, + expected_start, +): + from flexmeasures.cli.data_add import add_automation + + parameters_file = tmp_path / "parameters.yaml" + parameters_file.write_text(f"start: {yaml_start}\n") + result = app.test_cli_runner().invoke( + add_automation, + [ + "--asset", + "1", + "--name", + "YAML dates", + "--cron", + "0 6 * * *", + "--parameters", + str(parameters_file), + "--sensor", + str(setup_dummy_data[0]), + ], + ) + + assert result.exit_code == 0, result.output + automation = fresh_db.session.scalars(select(Automation)).one() + assert automation.parameters["start"] == expected_start + + +@pytest.mark.parametrize("option_name", ("--config", "--parameters")) +def test_add_automation_accepts_empty_yaml_file( + app, fresh_db, setup_dummy_data, tmp_path, option_name +): + from flexmeasures.cli.data_add import add_automation + + empty_file = tmp_path / "empty.yaml" + empty_file.write_text("") + result = app.test_cli_runner().invoke( + add_automation, + [ + "--asset", + "1", + "--name", + "Empty YAML", + "--cron", + "0 6 * * *", + option_name, + str(empty_file), + "--sensor", + str(setup_dummy_data[0]), + ], + ) + + assert result.exit_code == 0, result.output + + +@pytest.mark.parametrize("option_name", ("--config", "--parameters")) +def test_add_automation_rejects_non_object_yaml_file( + app, fresh_db, setup_dummy_data, tmp_path, option_name +): + from flexmeasures.cli.data_add import add_automation + + list_file = tmp_path / "list.yaml" + list_file.write_text("- not\n- an\n- object\n") + result = app.test_cli_runner().invoke( + add_automation, + [ + "--asset", + "1", + "--name", + "Invalid YAML", + "--cron", + "0 6 * * *", + option_name, + str(list_file), + "--sensor", + str(setup_dummy_data[0]), + ], + ) + + assert result.exit_code == 2, result.output + assert "must contain a YAML or JSON object at the top level" in result.output + assert "Traceback" not in result.output + + +def test_run_automations(app, fresh_db, setup_dummy_data, clean_redis): + """Active automations due this minute queue forecasting jobs (with trigger meta data); inactive ones do not. + + We use two automations with the same forecaster config (thus sharing a generator data source), + to make sure one automation's run does not pollute the other's. + """ + from flexmeasures.cli.data_add import add_automation + from flexmeasures.cli.jobs import run_automations + + sensor1_id, sensor2_id = setup_dummy_data[0], setup_dummy_data[1] + runner = app.test_cli_runner() + for name, sensor_id in [ + ("Every minute", sensor1_id), + ("Also every minute", sensor2_id), + ]: + cli_input = { + "asset": 1, + "name": name, + "cron": "* * * * *", # due every minute + "sensor": sensor_id, + } + result = runner.invoke(add_automation, to_flags(cli_input)) + assert "Successfully created" in result.output, result.output + automations = fresh_db.session.scalars(select(Automation)).all() + assert automations[0].generator_id == automations[1].generator_id + + result = runner.invoke(run_automations) + assert result.exit_code == 0, result.output + assert result.output.count("queued") == 2, result.output + + # check the queued jobs recorded how they were created + jobs = app.queues["forecasting"].jobs + assert len(jobs) > 0 + automation_ids = {automation.id for automation in automations} + assert all( + job.meta["trigger"]["origin"] == "automation" + and job.meta["trigger"]["automation_id"] in automation_ids + for job in jobs + ) + # running again within the same minute does not queue jobs twice + n_jobs = len(jobs) + result = runner.invoke(run_automations) + assert "No automations due" in result.output, result.output + assert len(app.queues["forecasting"].jobs) == n_jobs + + # inactive automations are not due + for automation in automations: + automation.active = False + fresh_db.session.commit() + app.redis_connection.flushdb() + result = runner.invoke(run_automations) + assert "No automations due" in result.output, result.output + + +def test_run_automations_catches_up_once_after_downtime( + app, + fresh_db, + setup_dummy_data, + clean_redis, + freeze_server_now, +): + from flexmeasures.cli.data_add import add_automation + from flexmeasures.cli.jobs import run_automations + + freeze_server_now(datetime(2026, 1, 15, 8, 58, 30, tzinfo=timezone.utc)) + runner = app.test_cli_runner() + add_result = runner.invoke( + add_automation, + [ + "--asset", + "1", + "--name", + "Amsterdam catch-up", + "--cron", + "0 10 * * *", + "--timezone", + "Europe/Amsterdam", + "--sensor", + str(setup_dummy_data[0]), + ], + ) + assert add_result.exit_code == 0, add_result.output + + freeze_server_now(datetime(2026, 1, 15, 9, 5, tzinfo=timezone.utc)) + first_result = runner.invoke(run_automations) + assert first_result.exit_code == 0, first_result.output + assert first_result.output.count("queued") == 1 + n_jobs = app.queues["forecasting"].count + assert n_jobs > 0 + + fresh_db.session.remove() + second_result = runner.invoke(run_automations) + assert second_result.exit_code == 0, second_result.output + assert "No automations due" in second_result.output + assert app.queues["forecasting"].count == n_jobs + + automation = fresh_db.session.scalars(select(Automation)).one() + assert automation.timezone == "Europe/Amsterdam" + assert automation.cursor == datetime(2026, 1, 15, 9, 0, tzinfo=timezone.utc) + + +def test_failed_automation_attempt_is_not_retried(app, clean_redis, mocker): + """A failure after partial queueing must not duplicate that work on retry.""" + from flexmeasures.cli.jobs import run_automations + from flexmeasures.data.services.automations import DueAutomation + + automation = SimpleNamespace(id=42, name="Partial run", asset_id=1) + due_automation = DueAutomation( + automation=automation, + scheduled_at=datetime(2026, 8, 5, 1, 0, tzinfo=timezone.utc), + expected_cursor=datetime(2026, 8, 5, 0, 0, tzinfo=timezone.utc), + expected_cronstr="0 * * * *", + expected_timezone="UTC", + ) + mocker.patch( + "flexmeasures.cli.jobs.get_due_automations", return_value=[due_automation] + ) + mocker.patch("flexmeasures.cli.jobs.claim_due_automation", return_value=True) + + def queue_then_fail(_automation): + app.queues["forecasting"].enqueue("flexmeasures.utils.time_utils.server_now") + raise RuntimeError("failed after queueing") + + mocker.patch("flexmeasures.cli.jobs.run_automation", side_effect=queue_then_fail) + runner = app.test_cli_runner() + + first_result = runner.invoke(run_automations) + assert first_result.exit_code == 1, first_result.output + assert "failed after queueing" in first_result.output + assert app.queues["forecasting"].count == 1 + + retry_result = runner.invoke(run_automations) + assert retry_result.exit_code == 0, retry_result.output + assert "already attempted" in retry_result.output + assert "Skipping to avoid duplicate jobs" in retry_result.output + assert app.queues["forecasting"].count == 1 + + +def test_run_automation_revalidates_output_scope( + app, fresh_db, automation_scope_assets, clean_redis +): + from flexmeasures.cli.data_add import add_automation + from flexmeasures.cli.jobs import run_automations + + root_asset = automation_scope_assets["root_asset"] + child_asset = automation_scope_assets["child_asset"] + unrelated_asset = automation_scope_assets["unrelated_asset"] + child_sensor = automation_scope_assets["sensors"]["child"] + runner = app.test_cli_runner() + + add_result = runner.invoke( + add_automation, + [ + "--asset", + str(root_asset.id), + "--name", + "moved output", + "--cron", + "* * * * *", + "--sensor", + str(child_sensor.id), + ], + ) + assert add_result.exit_code == 0, add_result.output + + child_asset.parent_asset = unrelated_asset + fresh_db.session.commit() + fresh_db.session.expire_all() + + run_result = runner.invoke(run_automations) + + assert run_result.exit_code == 1 + assert "must belong to asset" in run_result.output + assert app.queues["forecasting"].count == 0 diff --git a/flexmeasures/cli/tests/test_data_add.py b/flexmeasures/cli/tests/test_data_add.py index fa79824612..7f028188da 100644 --- a/flexmeasures/cli/tests/test_data_add.py +++ b/flexmeasures/cli/tests/test_data_add.py @@ -229,6 +229,9 @@ def test_add_forecast_cli_accepts_regressor_ids_and_json_reference_lists( captured_configs = [] class StubForecaster: + def set_job_trigger(self, origin): + pass + def compute(self, **kwargs): return {"n_jobs": 1} diff --git a/flexmeasures/cli/utils.py b/flexmeasures/cli/utils.py index 510e5307c9..ad02f1d680 100644 --- a/flexmeasures/cli/utils.py +++ b/flexmeasures/cli/utils.py @@ -492,8 +492,16 @@ def split_commas(ctx, param, value): return list(set([x.strip() for x in result if x.strip()])) -def add_cli_options_from_schema(schema): - """Decorator to add CLI options based on a Marshmallow schema's fields.""" +def add_cli_options_from_schema( + schema, *, hidden: bool = False, force_optional: bool = False +): + """Decorator to add CLI options based on a Marshmallow schema's fields. + + Set hidden to keep the options out of the command's help text, which is useful for a command whose help should focus on its own options, + while still accepting the schema's options. + Set force_optional to let a field that the schema requires be omitted on the command line, + so it can be supplied by another route (such as a parameters file) and be validated by the schema itself. + """ def decorator(command): for field_name, field in reversed(schema.fields.items()): @@ -518,9 +526,11 @@ def decorator(command): kwargs = { "help": help_text, - "required": field.required, + "required": field.required and not force_optional, # "default": field.load_default, } + if hidden: + kwargs["hidden"] = True if cli.get("is_flag"): kwargs["is_flag"] = True diff --git a/flexmeasures/data/config.py b/flexmeasures/data/config.py index 4a69437fdd..38f9e7e449 100644 --- a/flexmeasures/data/config.py +++ b/flexmeasures/data/config.py @@ -48,6 +48,7 @@ def configure_db_for(app: Flask): user, task_runs, forecasting, + automations, ) # noqa: F401 # This would create db structure based on models, but you should use `flask db upgrade` for that. diff --git a/flexmeasures/data/migrations/versions/4d5e6f708192_merge_automations_with_main.py b/flexmeasures/data/migrations/versions/4d5e6f708192_merge_automations_with_main.py new file mode 100644 index 0000000000..e3315c7a20 --- /dev/null +++ b/flexmeasures/data/migrations/versions/4d5e6f708192_merge_automations_with_main.py @@ -0,0 +1,21 @@ +"""merge automations with main + +Revision ID: 4d5e6f708192 +Revises: 3bc1e29ca1f4, 8ecec35b799c +Create Date: 2026-08-05 01:10:00.000000 + +""" + +# revision identifiers, used by Alembic. +revision = "4d5e6f708192" +down_revision = ("3bc1e29ca1f4", "8ecec35b799c") +branch_labels = None +depends_on = None + + +def upgrade(): + pass + + +def downgrade(): + pass diff --git a/flexmeasures/data/migrations/versions/8ecec35b799c_add_automation_table.py b/flexmeasures/data/migrations/versions/8ecec35b799c_add_automation_table.py new file mode 100644 index 0000000000..68873aaf39 --- /dev/null +++ b/flexmeasures/data/migrations/versions/8ecec35b799c_add_automation_table.py @@ -0,0 +1,52 @@ +"""add automation table + +Revision ID: 8ecec35b799c +Revises: 3c2f9e5a1d47 +Create Date: 2026-07-11 10:00:00.000000 + +""" + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + +# revision identifiers, used by Alembic. +revision = "8ecec35b799c" +down_revision = "3c2f9e5a1d47" +branch_labels = None +depends_on = None + + +def upgrade(): + op.create_table( + "automation", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("asset_id", sa.Integer(), nullable=False), + sa.Column("type", sa.String(length=80), nullable=False), + sa.Column("name", sa.String(length=80), nullable=False), + sa.Column("cronstr", sa.String(length=80), nullable=False), + sa.Column("active", sa.Boolean(), nullable=False), + sa.Column("generator_id", sa.Integer(), nullable=False), + sa.Column( + "parameters", + postgresql.JSONB(astext_type=sa.Text()), + nullable=False, + ), + sa.ForeignKeyConstraint( + ["asset_id"], + ["generic_asset.id"], + name=op.f("automation_asset_id_generic_asset_fkey"), + ondelete="CASCADE", + ), + sa.ForeignKeyConstraint( + ["generator_id"], + ["data_source.id"], + name=op.f("automation_generator_id_data_source_fkey"), + ), + sa.PrimaryKeyConstraint("id", name=op.f("automation_pkey")), + ) + + +def downgrade(): + op.drop_table("automation") diff --git a/flexmeasures/data/migrations/versions/9f2b6e1d4a73_add_automation_timezone_and_cursor.py b/flexmeasures/data/migrations/versions/9f2b6e1d4a73_add_automation_timezone_and_cursor.py new file mode 100644 index 0000000000..75b76d4903 --- /dev/null +++ b/flexmeasures/data/migrations/versions/9f2b6e1d4a73_add_automation_timezone_and_cursor.py @@ -0,0 +1,64 @@ +"""add automation timezone and cursor + +Revision ID: 9f2b6e1d4a73 +Revises: 4d5e6f708192 +Create Date: 2026-08-05 03:00:00.000000 + +""" + +from flask import current_app +from alembic import op +from pytz import all_timezones_set +import sqlalchemy as sa + +# revision identifiers, used by Alembic. +revision = "9f2b6e1d4a73" +down_revision = "4d5e6f708192" +branch_labels = None +depends_on = None + + +def upgrade(): + timezone = current_app.config.get("FLEXMEASURES_TIMEZONE", "UTC") + if timezone not in all_timezones_set: + raise ValueError( + f"Cannot migrate automations with invalid FLEXMEASURES_TIMEZONE {timezone!r}." + ) + + # Both columns are required, but existing rows have no value for them yet. + # So add them as nullable, backfill every row, and only then enforce NOT NULL. + op.add_column( + "automation", sa.Column("timezone", sa.String(length=64), nullable=True) + ) + op.add_column( + "automation", + sa.Column("cursor", sa.DateTime(timezone=True), nullable=True), + ) + automation = sa.table( + "automation", + sa.column("timezone", sa.String(length=64)), + sa.column("cursor", sa.DateTime(timezone=True)), + ) + # Existing automations predate the timezone column, and were run against the server timezone, so adopt that. + # Their cursor starts one minute before the upgrade, mirroring `get_initial_cursor` for newly created automations: + # a run scheduled in the very minute of the upgrade is still queued, while runs scheduled before that are not replayed. + op.execute( + automation.update().values( + timezone=timezone, + cursor=sa.func.date_trunc("minute", sa.func.current_timestamp()) + - sa.text("interval '1 minute'"), + ) + ) + op.alter_column("automation", "timezone", nullable=False) + op.alter_column("automation", "cursor", nullable=False) + # PostgreSQL does not index a foreign key by itself, and automations are looked up by asset + # (on an asset's automations page, and when finding the automations that feed a sensor). + op.create_index( + op.f("ix_automation_asset_id"), "automation", ["asset_id"], unique=False + ) + + +def downgrade(): + op.drop_index(op.f("ix_automation_asset_id"), table_name="automation") + op.drop_column("automation", "cursor") + op.drop_column("automation", "timezone") diff --git a/flexmeasures/data/models/automations.py b/flexmeasures/data/models/automations.py new file mode 100644 index 0000000000..db0ab6f2f5 --- /dev/null +++ b/flexmeasures/data/models/automations.py @@ -0,0 +1,133 @@ +""" +Automations: recurring tasks (for now: forecasting) defined per asset. +""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone + +from flask import current_app +from pytz import all_timezones_set +from sqlalchemy.dialects.postgresql import JSONB +from sqlalchemy.ext.mutable import MutableDict +from sqlalchemy.orm import validates + +from flexmeasures.auth.policy import AuthModelMixin +from flexmeasures.data import db +from flexmeasures.utils.time_utils import server_now + + +def get_default_automation_timezone() -> str: + """Return the timezone to snapshot when an automation is created.""" + timezone_name = current_app.config.get("FLEXMEASURES_TIMEZONE", "UTC") + if timezone_name not in all_timezones_set: + raise ValueError(f"Timezone '{timezone_name}' does not exist.") + return timezone_name + + +def get_initial_cursor() -> datetime: + """Return a cursor which keeps the automation's creation minute eligible.""" + return server_now().astimezone(timezone.utc).replace( + second=0, microsecond=0 + ) - timedelta(minutes=1) + + +class Automation(db.Model, AuthModelMixin): + """A recurring task on an asset, such as computing forecasts. + + The recurrence is defined by a cron string, and the work to be done is defined + by a data generator (e.g. a forecaster, linked through a data source) together + with the parameters to call it with. + """ + + __tablename__ = "automation" + + SUPPORTED_TYPES = ["forecasts"] # later also "schedules" and "reports" + + id = db.Column(db.Integer, autoincrement=True, primary_key=True) + created_at = db.Column( + db.DateTime(timezone=True), nullable=False, default=server_now + ) + asset_id = db.Column( + db.Integer, + db.ForeignKey("generic_asset.id", ondelete="CASCADE"), + nullable=False, + index=True, + ) + type = db.Column(db.String(80), nullable=False, default="forecasts") + name = db.Column(db.String(80), nullable=False) + cronstr = db.Column(db.String(80), nullable=False) + timezone = db.Column( + db.String(64), nullable=False, default=get_default_automation_timezone + ) + # The scheduled time of the most recent run this automation committed to. + # Runs at or before it are never queued again, which is what makes catch-up after downtime queue only the latest missed run. + # It advances just before queueing, so it records that a run was claimed, not that queueing or the forecast itself succeeded. + cursor = db.Column( + db.DateTime(timezone=True), + nullable=False, + default=get_initial_cursor, + ) + active = db.Column(db.Boolean, nullable=False, default=True) + generator_id = db.Column( + db.Integer, db.ForeignKey("data_source.id"), nullable=False + ) + parameters = db.Column(MutableDict.as_mutable(JSONB), nullable=False, default={}) + + asset = db.relationship( + "GenericAsset", + foreign_keys=[asset_id], + backref=db.backref( + "automations", lazy=True, cascade="all, delete-orphan", passive_deletes=True + ), + ) + generator = db.relationship("DataSource", foreign_keys=[generator_id]) + + @validates("timezone") + def validate_timezone(self, key: str, timezone: str) -> str: + """Require an exact timezone name from the IANA timezone database.""" + if timezone not in all_timezones_set: + raise ValueError(f"Timezone '{timezone}' does not exist.") + return timezone + + 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). + """ + if self.asset is None: + return {} + asset_acl = self.asset.__acl__() + return { + "read": asset_acl["read"], + "update": asset_acl["delete"], + "delete": asset_acl["delete"], + } + + def __repr__(self): + return "" % ( + self.id, + self.name, + self.type, + self.asset_id, + "active" if self.active else "inactive", + ) + + @property + def input_sensors(self) -> list: + """The sensors that this automation reads data from on each run, as far as they can be worked out. + + Reports no sensors if they cannot be, so do not use this to decide whether something is permitted; + see `resolve_automation_sensors` for that. + """ + from flexmeasures.data.services.automations import get_automation_sensors + + return get_automation_sensors(self)["input_sensors"] + + @property + def output_sensors(self) -> list: + """The sensors that this automation writes data to on each run. See `input_sensors`.""" + from flexmeasures.data.services.automations import get_automation_sensors + + return get_automation_sensors(self)["output_sensors"] diff --git a/flexmeasures/data/models/data_sources.py b/flexmeasures/data/models/data_sources.py index 9a25b76b18..b275127526 100644 --- a/flexmeasures/data/models/data_sources.py +++ b/flexmeasures/data/models/data_sources.py @@ -31,6 +31,7 @@ class DataGenerator: _config: dict = None _parameters: dict = None + _job_trigger: dict | None = None _parameters_schema: Schema | None = None _config_schema: Schema | None = None @@ -98,6 +99,61 @@ def __init__( elif len(kwargs) == 0: self._config = self._config_schema.load({}) + def set_job_trigger(self, origin: str, automation_id: int | None = None): + """Record how any queued jobs got created (e.g. via the CLI, the API or an automation). + + This information is stored on the jobs themselves (as job meta data). + """ + self._job_trigger = {"origin": origin} + if automation_id is not None: + self._job_trigger["automation_id"] = automation_id + + @property + def input_sensors(self) -> list: + """The sensors that this data generator reads data from. + + Overwrite in your data generator, deriving the sensors from its config and + parameters. Together with `output_sensors`, this describes the data flowing + through the data generator, which is used for linking to the sensors involved + (and, in the future, for checking access to them). + """ + return [] + + @property + def output_sensors(self) -> list: + """The sensors that this data generator writes data to. See `input_sensors`.""" + return [] + + @staticmethod + def _resolve_sensors(*values) -> list: + """Turn (lists of) sensors, sensor references or sensor IDs into a list of unique sensors. + + A sensor reference contributes the sensor it wraps, as the source filters only narrow down + which beliefs are read from that sensor, not which sensor is involved. + Sensor IDs that cannot be found, and None values, are skipped. + """ + from flexmeasures.data.models.time_series import Sensor + from flexmeasures.data.schemas.sensors import SensorReference + + sensors: dict[int, Sensor] = {} + for value in values: + if value is None: + continue + for item in value if isinstance(value, (list, tuple, set)) else [value]: + if isinstance(item, SensorReference): + sensor = item.sensor + elif isinstance(item, Sensor): + sensor = item + elif (isinstance(item, int) and not isinstance(item, bool)) or ( + isinstance(item, str) and item.isdigit() + ): + sensor = db.session.get(Sensor, int(item)) + else: + continue + if sensor is not None: + sensors[sensor.id] = sensor + return list(sensors.values()) + def _compute(self, **kwargs) -> list[dict[str, Any]]: """Overwrite with the actual computation of your data generator. diff --git a/flexmeasures/data/models/forecasting/__init__.py b/flexmeasures/data/models/forecasting/__init__.py index 7481b820fd..c449fa1621 100644 --- a/flexmeasures/data/models/forecasting/__init__.py +++ b/flexmeasures/data/models/forecasting/__init__.py @@ -30,6 +30,26 @@ class Forecaster(DataGenerator): _config_schema = ForecasterConfigSchema() + @property + def input_sensors(self) -> list: + """The regressors used to forecast, plus the history of the sensor being forecast.""" + config = self._config or {} + parameters = self._parameters or {} + return self._resolve_sensors( + config.get("past_regressors"), + config.get("future_regressors"), + config.get("regressors"), + parameters.get("sensor"), + ) + + @property + def output_sensors(self) -> list: + """The sensor that the forecast is saved to, which defaults to the sensor being forecast.""" + parameters = self._parameters or {} + return self._resolve_sensors( + parameters.get("sensor_to_save") or parameters.get("sensor") + ) + def _compute( self, check_output_resolution=True, as_job: bool = False, **kwargs ) -> list[dict[str, Any]]: diff --git a/flexmeasures/data/models/forecasting/pipelines/train_predict.py b/flexmeasures/data/models/forecasting/pipelines/train_predict.py index 7b52c25d17..e3073a7c39 100644 --- a/flexmeasures/data/models/forecasting/pipelines/train_predict.py +++ b/flexmeasures/data/models/forecasting/pipelines/train_predict.py @@ -381,7 +381,10 @@ def run( as_job: bool = False, queue: str = "forecasting", ): - logging.info( + # Only announce a pipeline run when actually running it here: with as_job, this + # method merely queues the cycles, and the workers running them log their own start. + log_start = logging.debug if as_job else logging.info + log_start( f"Starting Train-Predict Pipeline to predict for {self._parameters['predict_period_in_hours']} hours." ) connection = current_app.queues[queue].connection @@ -463,6 +466,8 @@ def run( "end": self._parameters["end_date"].isoformat(), "sensor_id": sensor_to_save_id, } + if self._job_trigger: + job_metadata["trigger"] = self._job_trigger for cycle_params in cycles_job_params: job_kwargs = { "config": job_config, diff --git a/flexmeasures/data/queries/generic_assets.py b/flexmeasures/data/queries/generic_assets.py index ab164cc012..994991f605 100644 --- a/flexmeasures/data/queries/generic_assets.py +++ b/flexmeasures/data/queries/generic_assets.py @@ -17,6 +17,27 @@ from flexmeasures.utils.flexmeasures_inflection import pluralize +def asset_and_ancestor_ids(asset_id: int | None) -> list[int]: + """List the given asset and all of its ancestors, nearest first. + + Walks up the tree one asset at a time, so keep the number of calls low on hot paths. + A cycle in the tree (which should not occur) stops the walk rather than looping forever. + """ + asset_ids: list[int] = [] + while asset_id is not None and asset_id not in asset_ids: + asset_ids.append(asset_id) + asset = db.session.get(GenericAsset, asset_id) + if asset is None: + break + asset_id = asset.parent_asset_id + return asset_ids + + +def asset_is_in_subtree(root_asset_id: int, candidate_asset_id: int) -> bool: + """Return whether an asset is the given root or one of its descendants.""" + return root_asset_id in asset_and_ancestor_ids(candidate_asset_id) + + def query_assets_by_type( type_names: list[str] | str, account_id: int | None = None, diff --git a/flexmeasures/data/schemas/automations.py b/flexmeasures/data/schemas/automations.py new file mode 100644 index 0000000000..97ada4baf5 --- /dev/null +++ b/flexmeasures/data/schemas/automations.py @@ -0,0 +1,101 @@ +from __future__ import annotations + +from datetime import datetime, timezone + +from croniter import croniter +from croniter.croniter import CroniterBadDateError +from marshmallow import fields, validates, ValidationError +from pytz import all_timezones_set + +from flexmeasures.data import ma, db +from flexmeasures.data.models.automations import Automation +from flexmeasures.data.schemas.utils import ( + FMValidationError, + MarshmallowClickMixin, + with_appcontext_if_needed, +) + + +class CronField(MarshmallowClickMixin, fields.Str): + """Field that validates a cron string (e.g. "0 6 * * *").""" + + def _deserialize(self, value, attr, obj, **kwargs) -> str: + value = super()._deserialize(value, attr, obj, **kwargs) + if len(value.split()) != 5: + raise FMValidationError( + "Automation cron expressions must contain exactly five fields " + "(minute, hour, day of month, month, and day of week)." + ) + if not croniter.is_valid(value): + raise FMValidationError(f"'{value}' is not a valid cron string.") + try: + croniter(value, datetime(2000, 1, 1, tzinfo=timezone.utc)).get_next( + datetime + ) + except CroniterBadDateError as exc: + raise FMValidationError( + f"'{value}' does not match any possible date." + ) from exc + return value + + +class TimezoneField(MarshmallowClickMixin, fields.Str): + """Field that validates an exact IANA timezone name.""" + + def _deserialize(self, value, attr, obj, **kwargs) -> str: + value = super()._deserialize(value, attr, obj, **kwargs) + if value not in all_timezones_set: + raise FMValidationError(f"Timezone '{value}' does not exist.") + return value + + +class AutomationIdField(MarshmallowClickMixin, fields.Int): + """Field that deserializes to an Automation and serializes back to an integer.""" + + @with_appcontext_if_needed() + def _deserialize(self, value, attr, obj, **kwargs) -> Automation: + """Turn an automation id into an Automation.""" + value = super()._deserialize(value, attr, obj, **kwargs) + automation = db.session.get(Automation, value) + if automation is None: + raise FMValidationError(f"No automation found with id {value}.") + return automation + + def _serialize(self, automation, attr, data, **kwargs): + """Turn an Automation into an automation id.""" + return automation.id + + +class AutomationSchema(ma.SQLAlchemySchema): + """Automation schema, with validations.""" + + class Meta: + model = Automation + + id = ma.auto_field(dump_only=True) + created_at = ma.auto_field(dump_only=True) + asset_id = ma.auto_field() + type = ma.auto_field() + name = ma.auto_field(required=True) + cronstr = CronField(required=True) + timezone = TimezoneField( + metadata={ + "description": "IANA timezone in which the cron expression is interpreted.", + "example": "Europe/Amsterdam", + } + ) + cursor = ma.auto_field( + dump_only=True, + metadata={ + "description": "UTC time of the most recent run this automation committed to. Runs at or before it are never queued again. It advances just before queueing, so it does not indicate that queueing or the forecast itself succeeded.", + "example": "2026-08-05T06:00:00+00:00", + }, + ) + active = ma.auto_field() + + @validates("type") + def validate_type(self, type: str, **kwargs): + if type not in Automation.SUPPORTED_TYPES: + raise ValidationError( + f"Automation type '{type}' is not supported (supported types: {Automation.SUPPORTED_TYPES})." + ) diff --git a/flexmeasures/data/schemas/tests/test_automations.py b/flexmeasures/data/schemas/tests/test_automations.py new file mode 100644 index 0000000000..a8961e5475 --- /dev/null +++ b/flexmeasures/data/schemas/tests/test_automations.py @@ -0,0 +1,31 @@ +import pytest +from marshmallow import ValidationError + +from flexmeasures.data.schemas.automations import CronField, TimezoneField + + +def test_cron_field_accepts_five_field_expression(): + assert CronField().deserialize("0 6 * * *") == "0 6 * * *" + + +@pytest.mark.parametrize( + "cronstr", + ( + "*/10 * * * * *", + "0 */10 * * * * 2026", + "@daily", + ), +) +def test_cron_field_rejects_non_five_field_expression(cronstr): + with pytest.raises(ValidationError, match="exactly five fields"): + CronField().deserialize(cronstr) + + +@pytest.mark.parametrize("timezone", ("UTC", "Europe/Amsterdam", "Etc/GMT+1")) +def test_timezone_field_accepts_iana_names(timezone): + assert TimezoneField().deserialize(timezone) == timezone + + +def test_timezone_field_rejects_unknown_name(): + with pytest.raises(ValidationError, match="does not exist"): + TimezoneField().deserialize("Europe/NotAmsterdam") diff --git a/flexmeasures/data/services/automations.py b/flexmeasures/data/services/automations.py new file mode 100644 index 0000000000..719d6e1227 --- /dev/null +++ b/flexmeasures/data/services/automations.py @@ -0,0 +1,426 @@ +""" +Logic for running automations (see also the CLI command `flexmeasures jobs run-automations`). +""" + +from __future__ import annotations + +from copy import copy +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from typing import Any +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError + +from cron_descriptor import get_description, Options +from croniter import croniter +from croniter.croniter import CroniterError +from flask import current_app +from marshmallow import ValidationError +from sqlalchemy import select, update + +from flexmeasures import Forecaster +from flexmeasures.data import db +from flexmeasures.data.models.automations import Automation +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.utils.time_utils import server_now + + +@dataclass(frozen=True) +class DueAutomation: + """An automation together with the canonical run it should handle.""" + + automation: Automation + scheduled_at: datetime + expected_cursor: datetime | None + expected_cronstr: str + expected_timezone: str + + +def describe_cronstr(cronstr: str) -> str: + """Describe a cron string in natural language, e.g. "At 06:00". + + Explicitly renders times in 24-hour format, as cron-descriptor otherwise + picks a format based on the system locale. + """ + options = Options() + options.use_24hour_time_format = True + try: + return get_description(cronstr, options) + except Exception: + return cronstr + + +def floor_to_minute(dt: datetime) -> datetime: + """Floor a timezone-aware datetime to a UTC minute.""" + if dt.tzinfo is None or dt.utcoffset() is None: + raise ValueError("Automation scheduling requires a timezone-aware datetime.") + return dt.astimezone(timezone.utc).replace(second=0, microsecond=0) + + +def _as_nominal_wall_time(dt: datetime) -> datetime: + """Represent local wall-clock fields on a transition-free UTC timeline.""" + return datetime(dt.year, dt.month, dt.day, dt.hour, dt.minute, tzinfo=timezone.utc) + + +def _localize_nominal_time( + nominal_time: datetime, timezone_info: ZoneInfo, fold: int +) -> datetime: + """Attach a real timezone to nominal wall-clock fields using the given fold.""" + return datetime( + nominal_time.year, + nominal_time.month, + nominal_time.day, + nominal_time.hour, + nominal_time.minute, + tzinfo=timezone_info, + fold=fold, + ) + + +def _is_valid_local_time( + localized_time: datetime, nominal_time: datetime, timezone_info: ZoneInfo +) -> bool: + """Return whether a localized time survives a UTC round trip unchanged.""" + round_tripped = localized_time.astimezone(timezone.utc).astimezone(timezone_info) + return _as_nominal_wall_time(round_tripped) == nominal_time + + +def _valid_localizations( + nominal_time: datetime, timezone_info: ZoneInfo +) -> list[datetime]: + """Return the valid physical instants for a nominal wall-clock minute.""" + localizations = [ + _localize_nominal_time(nominal_time, timezone_info, fold=0), + _localize_nominal_time(nominal_time, timezone_info, fold=1), + ] + valid_localizations = [ + localized_time + for localized_time in localizations + if _is_valid_local_time(localized_time, nominal_time, timezone_info) + ] + return list( + { + localized_time.astimezone(timezone.utc) + for localized_time in valid_localizations + } + ) + + +def _is_ambiguous_wall_time(nominal_time: datetime, timezone_info: ZoneInfo) -> bool: + """Return whether a wall-clock minute denotes two physical instants.""" + return len(_valid_localizations(nominal_time, timezone_info)) == 2 + + +def _canonical_run_time(nominal_time: datetime, timezone_info: ZoneInfo) -> datetime: + """Map one wall-clock run to its canonical effective UTC instant. + + Ambiguous times use the earlier fold. + Nonexistent times become effective at the first valid minute after the clock jump. + """ + valid_localizations = _valid_localizations(nominal_time, timezone_info) + if valid_localizations: + return min(valid_localizations) + + first_valid_nominal_time = nominal_time + for _ in range(60 * 48): + first_valid_nominal_time += timedelta(minutes=1) + valid_localizations = _valid_localizations( + first_valid_nominal_time, timezone_info + ) + if valid_localizations: + return min(valid_localizations) + raise ValueError( + f"Could not find a valid local time after {nominal_time.isoformat()} in {timezone_info.key}." + ) + + +def _cron_evaluation_time(now: datetime, timezone_info: ZoneInfo) -> datetime: + """Return the nominal wall time through which cron runs have happened. + + During the second fold of a repeated interval, the entire first fold has already happened. + Evaluate through the end of that repeated wall interval, so missed runs are coalesced instead of replayed minute by minute. + """ + localized_now = now.astimezone(timezone_info) + nominal_now = _as_nominal_wall_time(localized_now) + if localized_now.fold != 1 or not _is_ambiguous_wall_time( + nominal_now, timezone_info + ): + return nominal_now + + nominal_after_overlap = nominal_now + for _ in range(60 * 48): + nominal_after_overlap += timedelta(minutes=1) + if not _is_ambiguous_wall_time(nominal_after_overlap, timezone_info): + return nominal_after_overlap - timedelta(minutes=1) + raise ValueError( + f"Could not find the end of the repeated local-time interval in {timezone_info.key}." + ) + + +def get_latest_scheduled_run(automation: Automation, now: datetime) -> datetime: + """Return the latest canonical run for an automation through ``now``.""" + now = floor_to_minute(now) + timezone_info = ZoneInfo(automation.timezone) + evaluation_time = _cron_evaluation_time(now, timezone_info) + if croniter.match(automation.cronstr, evaluation_time): + nominal_run = evaluation_time + else: + nominal_run = croniter(automation.cronstr, evaluation_time).get_prev(datetime) + scheduled_at = _canonical_run_time(nominal_run, timezone_info) + if scheduled_at > now: + raise ValueError( + f"Cron run {nominal_run.isoformat()} in {automation.timezone} resolves after {now.isoformat()}." + ) + return scheduled_at + + +def get_due_automations(now: datetime | None = None) -> list[DueAutomation]: + """Return the newest unhandled run for each active automation.""" + if now is None: + now = server_now() + now = floor_to_minute(now) + active_automations = ( + db.session.scalars(select(Automation).filter_by(active=True)).unique().all() + ) + due_automations = [] + for automation in active_automations: + try: + scheduled_at = get_latest_scheduled_run(automation, now) + except (CroniterError, ValueError, ZoneInfoNotFoundError) as exc: + current_app.logger.error( + "Skipping automation %s (%r), because its next run could not be calculated: %s", + automation.id, + automation.name, + exc, + ) + continue + expected_cursor = automation.cursor + cursor = expected_cursor + if cursor is None: + cursor = floor_to_minute(automation.created_at) - timedelta(minutes=1) + if scheduled_at > cursor: + due_automations.append( + DueAutomation( + automation=automation, + scheduled_at=scheduled_at, + expected_cursor=expected_cursor, + expected_cronstr=automation.cronstr, + expected_timezone=automation.timezone, + ) + ) + return due_automations + + +def claim_due_automation(due_automation: DueAutomation) -> bool: + """Persist a run claim if its scheduling configuration is unchanged.""" + if due_automation.expected_cursor is None: + cursor_matches = Automation.cursor.is_(None) + else: + cursor_matches = Automation.cursor == due_automation.expected_cursor + result = db.session.execute( + update(Automation) + .where( + Automation.id == due_automation.automation.id, + Automation.active.is_(True), + Automation.cronstr == due_automation.expected_cronstr, + Automation.timezone == due_automation.expected_timezone, + cursor_matches, + ) + .values(cursor=due_automation.scheduled_at) + .execution_options(synchronize_session=False) + ) + if result.rowcount != 1: + db.session.rollback() + return False + db.session.commit() + return True + + +class AutomationSensorsUnknown(Exception): + """Raised when the sensors an automation involves cannot be worked out. + + Callers that decide whether something is allowed must let this propagate rather than treat it as "no sensors", + because an automation with no known sensors would otherwise pass every check on the sensors it involves. + """ + + +def resolve_automation_sensors(automation: Automation) -> dict[str, list[Sensor]]: + """Work out which sensors an automation reads from and writes to on each run. + + The sensors are derived from the data generator, configured with the automation's own parameters. + Raises `AutomationSensorsUnknown` if that cannot be done, e.g. because the automation has no data generator, + because its generator is not registered in this FlexMeasures instance, + or because its parameters no longer load (say, after a sensor was deleted). + Use this wherever the answer decides whether something is permitted; use `get_automation_sensors` for display. + """ + 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 {}) + ) + 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}" + ) from e + + +def get_automation_sensors(automation: Automation) -> dict[str, list[Sensor]]: + """Look up which sensors an automation reads from and writes to on each run, for display purposes. + + Automations whose sensors cannot be worked out report no sensors, so that one broken automation + does not keep a page or an API response from rendering. + Do not use this to decide whether something is permitted, as "no sensors" then reads as "nothing to check": + call `resolve_automation_sensors` instead and let its error propagate. + """ + try: + return resolve_automation_sensors(automation) + except AutomationSensorsUnknown as e: + current_app.logger.warning(str(e)) + return {"input_sensors": [], "output_sensors": []} + + +def get_automations_feeding_sensor(sensor: Sensor) -> list[Automation]: + """Find the automations that write data to the given sensor. + + Only automations on the sensor's own asset or on one of its ancestors are + considered, as an automation may only write to its asset's subtree + (see `validate_forecast_output_scope`). Working out the output sensors requires + setting up each candidate's data generator, so this keeps the work proportional + to the number of automations that could feed this sensor. + + Note that this does not filter by permission: callers showing these to a user + should check read access on each automation (e.g. with `user_can_read`). + """ + candidate_automations = db.session.scalars( + select(Automation).filter( + Automation.asset_id.in_(asset_and_ancestor_ids(sensor.generic_asset_id)) + ) + ).unique() + return [ + automation + for automation in candidate_automations + if sensor.id in [output.id for output in automation.output_sensors] + ] + + +def get_automations_involving_sensor(sensor: Sensor) -> list[Automation]: + """Find the automations that read from or write to the given sensor. + + Unlike `get_automations_feeding_sensor`, this considers every automation, because a regressor + may live anywhere in the tree, not just on the sensor's asset or one of its ancestors. + That makes this proportional to the number of automations, so keep it out of hot paths; + it is meant for rare, interactive checks, such as warning before a sensor is deleted. + """ + involved = [] + for automation in db.session.scalars(select(Automation)).unique(): + automation_sensors = get_automation_sensors(automation) + if sensor.id in { + involved_sensor.id + for key in ("input_sensors", "output_sensors") + for involved_sensor in automation_sensors[key] + }: + involved.append(automation) + return involved + + +def get_automation_job_stats(automation: Automation) -> dict[str, int]: + """Count the jobs created by this automation, per job status. + + Note that jobs in Redis have a limited TTL, so this only counts fairly recent jobs. + """ + # Jobs are cached under the forecast target sensor(s), which may belong + # to a different asset than the automation's own asset. + sensor_ids = {sensor.id for sensor in automation.asset.sensors} + for key in ("sensor", "sensor-to-save"): + value = (automation.parameters or {}).get(key) + if value is not None: + try: + sensor_ids.add(int(value)) + except (TypeError, ValueError): + pass + + counts: dict[str, int] = {} + seen_job_ids: set[str] = set() + for sensor_id in sensor_ids: + for job in current_app.job_cache.get(sensor_id, "forecasting", "sensor"): + if job.id in seen_job_ids: + continue + seen_job_ids.add(job.id) + if job.meta.get("trigger", {}).get("automation_id") == automation.id: + status = str(job.get_status().value) + counts[status] = counts.get(status, 0) + 1 + return counts + + +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") + if sensor_reference is None: + sensor_reference = parameters.get("sensor_to_save") + if sensor_reference is None: + sensor_reference = parameters.get("sensor") + + if isinstance(sensor_reference, Sensor): + return sensor_reference + try: + sensor_id = int(sensor_reference) + except (TypeError, ValueError) as exc: + raise ValueError("Forecast automation has no valid output sensor.") from exc + + sensor = db.session.get(Sensor, sensor_id) + if sensor is None: + raise ValueError( + f"Forecast automation output sensor {sensor_id} does not exist." + ) + return sensor + + +def validate_forecast_output_scope(asset_id: int, output_sensor: Sensor) -> None: + """Require forecast output on the automation asset or a descendant.""" + if not asset_is_in_subtree(asset_id, output_sensor.generic_asset_id): + raise ValueError( + f"Forecast automation output sensor {output_sensor.id} must belong to asset " + f"{asset_id} or one of its descendants." + ) + + +def run_automation(automation: Automation) -> dict[str, Any] | None: + """Queue the jobs for one run of an automation. + + :returns: the data generator's return value, e.g. {"job_id": , "n_jobs": } + for forecasting jobs. + """ + if automation.type != "forecasts": + raise NotImplementedError( + f"Automations of type '{automation.type}' cannot be run yet." + ) + if automation.generator is None: + raise ValueError( + f"Automation {automation.id} has no data generator to run (generator_id is not set)." + ) + # Work on a copy, as the data generator is cached on the data source, + # which may be shared by several automations (as in `resolve_automation_sensors`). + forecaster = copy(automation.generator.data_generator) + if not isinstance(forecaster, Forecaster): + raise ValueError( + f"Data source {automation.generator_id} of automation {automation.id} does not store a Forecaster." + ) + output_sensor = get_forecast_output_sensor(automation.parameters or {}) + validate_forecast_output_scope(automation.asset_id, output_sensor) + # Wipe any parameter state the copy inherited from a previous run. + forecaster._parameters = None + forecaster.set_job_trigger("automation", automation_id=automation.id) + return forecaster.compute(as_job=True, parameters=dict(automation.parameters)) diff --git a/flexmeasures/data/services/sensors.py b/flexmeasures/data/services/sensors.py index abb7844178..07846f69a5 100644 --- a/flexmeasures/data/services/sensors.py +++ b/flexmeasures/data/services/sensors.py @@ -7,6 +7,7 @@ from typing import Any from flask import current_app from sqlalchemy import delete +from werkzeug.exceptions import Forbidden, Unauthorized from isodate import duration_isoformat from timely_beliefs import BeliefsDataFrame @@ -21,7 +22,9 @@ from flexmeasures.data import db from flexmeasures import Sensor, Account, Asset +from flexmeasures.auth.policy import check_access from flexmeasures.data.models.audit_log import AssetAuditLog +from flexmeasures.data.models.automations import Automation from flexmeasures.data.models.data_sources import DataSource, DEFAULT_DATASOURCE_TYPES from flexmeasures.data.models.generic_assets import GenericAsset from flexmeasures.data.models.planning.devices import INFLEXIBLE_DEVICE_KEYS @@ -739,6 +742,17 @@ def serialize_sensor_status_data( return sensors +def _can_read_automation(automation: Automation | None) -> bool: + """Whether the current user may see an automation's identifying details.""" + if automation is None: + return False + try: + check_access(automation, "read") + except (Forbidden, Unauthorized): + return False + return True + + def build_asset_jobs_data( asset: Asset, ) -> list[dict]: @@ -805,7 +819,20 @@ def build_asset_jobs_data( else None ) - metadata = json.dumps({**job.meta, "job_id": job.id}, default=str, indent=4) + # Show how the job was created (e.g. via the CLI, the API or an automation) + metadata_dict = {**job.meta, "job_id": job.id} + trigger = dict(job.meta.get("trigger", {})) + created_via = trigger.get("origin", "") + if trigger.get("automation_id") is not None: + automation = db.session.get(Automation, trigger["automation_id"]) + if _can_read_automation(automation): + created_via = f"automation '{automation.name}' ({automation.id})" + else: + created_via = "automation" + trigger.pop("automation_id") + metadata_dict["trigger"] = trigger + + metadata = json.dumps(metadata_dict, default=str, indent=4) jobs_data.append( { "job_id": job.id, @@ -816,6 +843,7 @@ def build_asset_jobs_data( "status": job.get_status(), "err": job_err, "enqueued_at": job.enqueued_at, + "created_via": created_via, "metadata_hash": hashlib.sha256(metadata.encode()).hexdigest(), } ) diff --git a/flexmeasures/data/tests/test_automation_scheduling_fresh_db.py b/flexmeasures/data/tests/test_automation_scheduling_fresh_db.py new file mode 100644 index 0000000000..2e77fbc5af --- /dev/null +++ b/flexmeasures/data/tests/test_automation_scheduling_fresh_db.py @@ -0,0 +1,329 @@ +"""Regression tests for durable forecast automation run calculation.""" + +from __future__ import annotations + +from datetime import datetime, timezone + +import pytest +from flask import current_app + +from flexmeasures.data.models.automations import Automation +from flexmeasures.data.models.data_sources import DataSource +from flexmeasures.data.models.generic_assets import GenericAsset, GenericAssetType +from flexmeasures.data.services.automations import ( + claim_due_automation, + get_due_automations, +) + + +@pytest.fixture() +def automation_factory(fresh_db): + """Create persisted automations with the minimum required relationships.""" + asset_type = GenericAssetType(name="automation scheduling asset type") + asset = GenericAsset( + name="automation scheduling asset", generic_asset_type=asset_type + ) + generator = DataSource( + name="automation scheduling generator", + type="forecaster", + model="TrainPredictPipeline", + ) + fresh_db.session.add_all([asset, generator]) + fresh_db.session.flush() + + def create_automation( + *, + name: str, + cronstr: str, + timezone_name: str, + cursor: datetime, + active: bool = True, + ) -> Automation: + automation = Automation( + asset=asset, + generator=generator, + type="forecasts", + name=name, + cronstr=cronstr, + timezone=timezone_name, + cursor=cursor, + active=active, + parameters={}, + ) + fresh_db.session.add(automation) + fresh_db.session.commit() + return automation + + return create_automation + + +def test_automations_use_independent_timezones(fresh_db, automation_factory): + cursor = datetime(2026, 1, 15, 5, 59, tzinfo=timezone.utc) + amsterdam = automation_factory( + name="Amsterdam morning", + cronstr="0 7 * * *", + timezone_name="Europe/Amsterdam", + cursor=cursor, + ) + new_york = automation_factory( + name="New York morning", + cronstr="0 7 * * *", + timezone_name="America/New_York", + cursor=cursor, + ) + + due = get_due_automations(datetime(2026, 1, 15, 6, 0, tzinfo=timezone.utc)) + + assert [(item.automation.id, item.scheduled_at) for item in due] == [ + (amsterdam.id, datetime(2026, 1, 15, 6, 0, tzinfo=timezone.utc)) + ] + assert claim_due_automation(due[0]) is True + + due = get_due_automations(datetime(2026, 1, 15, 12, 0, tzinfo=timezone.utc)) + + assert [(item.automation.id, item.scheduled_at) for item in due] == [ + (new_york.id, datetime(2026, 1, 15, 12, 0, tzinfo=timezone.utc)) + ] + + +@pytest.mark.parametrize( + ("cronstr", "cursor", "now", "expected"), + ( + ( + "0 * * * *", + datetime(2026, 2, 1, 9, 0, tzinfo=timezone.utc), + datetime(2026, 2, 1, 10, 5, tzinfo=timezone.utc), + datetime(2026, 2, 1, 10, 0, tzinfo=timezone.utc), + ), + ( + "*/5 * * * *", + datetime(2026, 2, 1, 10, 0, tzinfo=timezone.utc), + datetime(2026, 2, 1, 10, 17, tzinfo=timezone.utc), + datetime(2026, 2, 1, 10, 15, tzinfo=timezone.utc), + ), + ), +) +def test_missed_runs_are_coalesced(automation_factory, cronstr, cursor, now, expected): + automation = automation_factory( + name="Catch-up", + cronstr=cronstr, + timezone_name="UTC", + cursor=cursor, + ) + + due = get_due_automations(now) + + assert [(item.automation.id, item.scheduled_at) for item in due] == [ + (automation.id, expected) + ] + + +def test_spring_forward_run_happens_at_transition_boundary( + automation_factory, +): + automation = automation_factory( + name="Skipped Amsterdam time", + cronstr="30 2 * * *", + timezone_name="Europe/Amsterdam", + cursor=datetime(2026, 3, 29, 0, 59, tzinfo=timezone.utc), + ) + + due = get_due_automations(datetime(2026, 3, 29, 1, 0, tzinfo=timezone.utc)) + + assert [(item.automation.id, item.scheduled_at) for item in due] == [ + (automation.id, datetime(2026, 3, 29, 1, 0, tzinfo=timezone.utc)) + ] + assert claim_due_automation(due[0]) is True + assert get_due_automations(datetime(2026, 3, 29, 1, 1, tzinfo=timezone.utc)) == [] + + +def test_fall_back_wall_time_runs_only_once(fresh_db, automation_factory): + automation = automation_factory( + name="Repeated Amsterdam time", + cronstr="30 2 * * *", + timezone_name="Europe/Amsterdam", + cursor=datetime(2026, 10, 25, 0, 29, tzinfo=timezone.utc), + ) + + first_fold_due = get_due_automations( + datetime(2026, 10, 25, 0, 30, tzinfo=timezone.utc) + ) + assert [(item.automation.id, item.scheduled_at) for item in first_fold_due] == [ + (automation.id, datetime(2026, 10, 25, 0, 30, tzinfo=timezone.utc)) + ] + assert claim_due_automation(first_fold_due[0]) is True + + fresh_db.session.remove() + second_fold_due = get_due_automations( + datetime(2026, 10, 25, 1, 30, tzinfo=timezone.utc) + ) + + assert second_fold_due == [] + + +def test_fall_back_resume_coalesces_completed_first_fold( + automation_factory, +): + automation = automation_factory( + name="Fall-back downtime", + cronstr="* * * * *", + timezone_name="Europe/Amsterdam", + cursor=datetime(2026, 10, 24, 23, 59, tzinfo=timezone.utc), + ) + + due = get_due_automations(datetime(2026, 10, 25, 1, 15, tzinfo=timezone.utc)) + + assert [(item.automation.id, item.scheduled_at) for item in due] == [ + (automation.id, datetime(2026, 10, 25, 0, 59, tzinfo=timezone.utc)) + ] + + +def test_persisted_cursor_survives_restart(fresh_db, automation_factory): + automation = automation_factory( + name="Persistent cursor", + cronstr="0 10 * * *", + timezone_name="UTC", + cursor=datetime(2026, 2, 1, 9, 59, tzinfo=timezone.utc), + ) + now = datetime(2026, 2, 1, 10, 0, tzinfo=timezone.utc) + due = get_due_automations(now) + assert claim_due_automation(due[0]) is True + automation_id = automation.id + + fresh_db.session.remove() + + assert get_due_automations(now) == [] + persisted = fresh_db.session.get(Automation, automation_id) + assert persisted.cursor == now + + +def test_inactive_automation_is_not_due(automation_factory): + cursor = datetime(2026, 2, 1, 9, 59, tzinfo=timezone.utc) + automation = automation_factory( + name="Inactive", + cronstr="* * * * *", + timezone_name="UTC", + cursor=cursor, + active=False, + ) + + assert get_due_automations(datetime(2026, 2, 1, 10, 0, tzinfo=timezone.utc)) == [] + assert automation.cursor == cursor + + +def test_invalid_cron_does_not_hide_other_due_automations(automation_factory, mocker): + cursor = datetime(2026, 2, 1, 9, 59, tzinfo=timezone.utc) + invalid = automation_factory( + name="Impossible date", + cronstr="0 0 31 2 *", + timezone_name="UTC", + cursor=cursor, + ) + valid = automation_factory( + name="Valid recurrence", + cronstr="* * * * *", + timezone_name="UTC", + cursor=cursor, + ) + + # Assert on the logger rather than on caplog: building an app reconfigures logging and + # replaces the root handlers, so caplog stops capturing for the rest of the test session + # once any earlier test has built one. + log_error = mocker.patch.object(current_app.logger, "error") + + due = get_due_automations(datetime(2026, 2, 1, 10, 0, tzinfo=timezone.utc)) + + assert [item.automation.id for item in due] == [valid.id] + assert log_error.call_count == 1 + assert log_error.call_args.args[0].startswith("Skipping automation") + assert log_error.call_args.args[1] == invalid.id + + +def test_claim_rejects_automation_deactivated_after_discovery( + fresh_db, automation_factory +): + cursor = datetime(2026, 2, 1, 9, 59, tzinfo=timezone.utc) + automation = automation_factory( + name="Deactivate race", + cronstr="* * * * *", + timezone_name="UTC", + cursor=cursor, + ) + due = get_due_automations(datetime(2026, 2, 1, 10, 0, tzinfo=timezone.utc))[0] + + automation.active = False + fresh_db.session.commit() + + assert claim_due_automation(due) is False + assert automation.cursor == cursor + + +@pytest.mark.parametrize( + ("field", "new_value"), + (("cronstr", "0 11 * * *"), ("timezone", "Europe/Amsterdam")), +) +def test_claim_rejects_recurrence_edited_after_discovery( + fresh_db, automation_factory, field, new_value +): + cursor = datetime(2026, 2, 1, 9, 59, tzinfo=timezone.utc) + automation = automation_factory( + name="Edit race", + cronstr="* * * * *", + timezone_name="UTC", + cursor=cursor, + ) + due = get_due_automations(datetime(2026, 2, 1, 10, 0, tzinfo=timezone.utc))[0] + + setattr(automation, field, new_value) + fresh_db.session.commit() + + assert claim_due_automation(due) is False + assert automation.cursor == cursor + + +def test_claim_rejects_cursor_changed_after_discovery(fresh_db, automation_factory): + cursor = datetime(2026, 2, 1, 9, 58, tzinfo=timezone.utc) + automation = automation_factory( + name="Cursor race", + cronstr="* * * * *", + timezone_name="UTC", + cursor=cursor, + ) + due = get_due_automations(datetime(2026, 2, 1, 10, 0, tzinfo=timezone.utc))[0] + newer_cursor = datetime(2026, 2, 1, 9, 59, tzinfo=timezone.utc) + + automation.cursor = newer_cursor + fresh_db.session.commit() + + assert claim_due_automation(due) is False + assert automation.cursor == newer_cursor + + +def test_claim_allows_name_edit_after_discovery(fresh_db, automation_factory): + automation = automation_factory( + name="Old display name", + cronstr="* * * * *", + timezone_name="UTC", + cursor=datetime(2026, 2, 1, 9, 59, tzinfo=timezone.utc), + ) + due = get_due_automations(datetime(2026, 2, 1, 10, 0, tzinfo=timezone.utc))[0] + + automation.name = "New display name" + fresh_db.session.commit() + + assert claim_due_automation(due) is True + + +def test_claim_rejects_automation_deleted_after_discovery(fresh_db, automation_factory): + automation = automation_factory( + name="Delete race", + cronstr="* * * * *", + timezone_name="UTC", + cursor=datetime(2026, 2, 1, 9, 59, tzinfo=timezone.utc), + ) + due = get_due_automations(datetime(2026, 2, 1, 10, 0, tzinfo=timezone.utc))[0] + + fresh_db.session.delete(automation) + fresh_db.session.commit() + + assert claim_due_automation(due) is False diff --git a/flexmeasures/data/tests/test_automations_fresh_db.py b/flexmeasures/data/tests/test_automations_fresh_db.py new file mode 100644 index 0000000000..47759a7e4e --- /dev/null +++ b/flexmeasures/data/tests/test_automations_fresh_db.py @@ -0,0 +1,80 @@ +from __future__ import annotations + +from datetime import timezone + +import pytest +from sqlalchemy.exc import IntegrityError + +from flexmeasures.data.models.automations import Automation +from flexmeasures.data.models.data_sources import DataSource +from flexmeasures.data.models.generic_assets import GenericAsset, GenericAssetType + + +@pytest.fixture() +def automation_with_generator(fresh_db): + asset_type = GenericAssetType(name="automation test asset type") + asset = GenericAsset(name="automation test asset", generic_asset_type=asset_type) + generator = DataSource( + name="automation test generator", + type="forecaster", + model="TrainPredictPipeline", + ) + automation = Automation( + asset=asset, + generator=generator, + type="forecasts", + name="automation generator lifecycle test", + cronstr="0 6 * * *", + parameters={}, + ) + fresh_db.session.add(automation) + fresh_db.session.commit() + return automation, generator + + +def test_referenced_automation_generator_cannot_be_deleted( + fresh_db, automation_with_generator +): + automation, generator = automation_with_generator + automation_id = automation.id + generator_id = generator.id + + fresh_db.session.delete(generator) + with pytest.raises(IntegrityError): + fresh_db.session.commit() + fresh_db.session.rollback() + + persisted_automation = fresh_db.session.get(Automation, automation_id) + assert persisted_automation is not None + assert persisted_automation.generator_id == generator_id + assert fresh_db.session.get(DataSource, generator_id) is not None + + fresh_db.session.delete(persisted_automation) + fresh_db.session.commit() + persisted_generator = fresh_db.session.get(DataSource, generator_id) + fresh_db.session.delete(persisted_generator) + fresh_db.session.commit() + assert fresh_db.session.get(DataSource, generator_id) is None + + +def test_automation_requires_generator(fresh_db, automation_with_generator): + automation, _ = automation_with_generator + automation.generator = None + + with pytest.raises(IntegrityError): + fresh_db.session.commit() + + +def test_automation_has_valid_timezone_and_aware_cursor(automation_with_generator): + automation, _ = automation_with_generator + + assert automation.timezone == "Asia/Seoul" + assert automation.cursor.tzinfo is not None + assert automation.cursor.utcoffset() == timezone.utc.utcoffset(None) + + +def test_automation_rejects_invalid_timezone(automation_with_generator): + automation, _ = automation_with_generator + + with pytest.raises(ValueError, match="does not exist"): + automation.timezone = "Europe/NotAmsterdam" diff --git a/flexmeasures/ui/static/js/flexmeasures.js b/flexmeasures/ui/static/js/flexmeasures.js index ca0d9b4fec..af6cf51281 100644 --- a/flexmeasures/ui/static/js/flexmeasures.js +++ b/flexmeasures/ui/static/js/flexmeasures.js @@ -507,6 +507,71 @@ function updateStatsTable(stats, tableBody) { }); } +function sourceIdFromKey(sourceKey) { + // Source keys are shown as " (ID: )" + const idMatch = String(sourceKey).match(/\(ID:\s*(\d+)\)$/); + return idMatch ? idMatch[1] : null; +} + +function preselectedSourceId() { + // The sensor page passes on the source query parameter, if given + const preselected = document.getElementById('sensorPageData')?.dataset.preselectedSourceId; + return preselected ? String(preselected) : null; +} + +function setUpSourceDetailsButton(sourceKey) { + // Let the button next to the source selector show the details of the selected source + const detailsButton = document.getElementById('sourceDetailsButton'); + if (!detailsButton) { return; } + const sourceId = sourceIdFromKey(sourceKey); + if (!sourceId) { + detailsButton.classList.add('d-none'); + return; + } + detailsButton.classList.remove('d-none'); + detailsButton.dataset.sourceId = sourceId; +} + +function showSourceDetails(sourceId) { + const title = document.getElementById('SourceDetailsTitle'); + const body = document.getElementById('SourceDetailsBody'); + if (!body) { return; } + title.textContent = `Data source ${sourceId}`; + body.textContent = 'Loading ...'; + fetch(`/api/v3_0/sources/${encodeURIComponent(sourceId)}`) + .then(response => { + if (!response.ok) { throw new Error(`status ${response.status}`); } + return response.json(); + }) + .then(source => { + title.textContent = `Data source ${source.id}: ${source.description}`; + const table = document.createElement('table'); + table.className = 'table table-striped'; + Object.entries(source).forEach(([field, value]) => { + const row = document.createElement('tr'); + const fieldCell = document.createElement('th'); + fieldCell.textContent = field; + const valueCell = document.createElement('td'); + if (value !== null && typeof value === 'object') { + const pre = document.createElement('pre'); + pre.className = 'mb-0'; + pre.textContent = JSON.stringify(value, null, 4); + valueCell.appendChild(pre); + } else { + valueCell.textContent = value === null ? '—' : String(value); + } + row.appendChild(fieldCell); + row.appendChild(valueCell); + table.appendChild(row); + }); + body.innerHTML = ''; + body.appendChild(table); + }) + .catch(error => { + body.textContent = `Could not load the details of this data source (${error.message}).`; + }); +} + function loadSensorStats(sensor_id, event_start_time="", event_end_time="", fresh=false) { const spinner = document.getElementById('spinner-run-simulation'); const dropdownContainer = document.getElementById('sourceKeyDropdownContainer'); @@ -569,16 +634,22 @@ function loadSensorStats(sensor_id, event_start_time="", event_end_time="", fres const selectedSourceKey = event.target.dataset.sourceKey; dropdownButton.textContent = selectedSourceKey; updateStatsTable(data[selectedSourceKey], tableBody); + setUpSourceDetailsButton(selectedSourceKey); }); dropdownItem.appendChild(dropdownLink); dropdownMenu.appendChild(dropdownItem); }); - // Update the table with the first sourceKey's data by default - const firstSourceKey = getLatestBeliefName(data); + // Show the source pre-selected via the source query parameter (e.g. when + // arriving here from an automation), or else the most recently updated one + const preselectedSourceKey = Object.keys(data).find( + sourceKey => sourceIdFromKey(sourceKey) === preselectedSourceId() + ); + const firstSourceKey = preselectedSourceKey || getLatestBeliefName(data); dropdownButton.textContent = firstSourceKey; updateStatsTable(data[firstSourceKey], tableBody); + setUpSourceDetailsButton(firstSourceKey); // Populate the "Delete data" source dropdown if it exists on the page, // re-using the stats data already fetched to avoid a duplicate API call. diff --git a/flexmeasures/ui/static/openapi-specs.json b/flexmeasures/ui/static/openapi-specs.json index 7e3ab548a7..2f7d4b32e4 100644 --- a/flexmeasures/ui/static/openapi-specs.json +++ b/flexmeasures/ui/static/openapi-specs.json @@ -3346,6 +3346,182 @@ ] } }, + "/api/v3_0/assets/{id}/automations/{automation_id}": { + "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 (for forecasts, these are the forecast parameters\nused on each run), information about the data generator that runs it,\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", + "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": { + "200": { + "description": "PROCESSED", + "content": { + "application/json": { + "examples": { + "automation": { + "summary": "Automation details", + "value": { + "id": 1, + "created_at": "2026-07-11T00:00:00+00:00", + "asset_id": 1, + "type": "forecasts", + "name": "Day-ahead PV forecasts", + "cronstr": "0 6 * * *", + "timezone": "Europe/Amsterdam", + "cursor": "2026-07-11T04:00:00+00:00", + "recurrence_description": "At 06:00", + "active": true, + "parameters": { + "sensor": 2092 + }, + "generator": { + "id": 6, + "description": "forecaster 'TrainPredictPipeline' (v1)" + }, + "input_sensors": [ + { + "id": 2092, + "name": "power" + }, + { + "id": 2093, + "name": "irradiance" + } + ], + "output_sensors": [ + { + "id": 2092, + "name": "power" + } + ], + "job_stats": { + "finished": 3, + "failed": 1 + }, + "redis_connection_err": null + } + } + } + } + } + }, + "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": { + "get": { + "summary": "Get all automations defined on an asset.", + "description": "The response will be a list of automations: recurring tasks (for now, computing forecasts)\ndefined on the asset. Each entry shows the automation's ID, when it was created,\nits type, name, activation status, and its recurrence, both as a cron string\nand described in natural language. Each entry also shows the IANA timezone in which its cron expression is interpreted, and its cursor.\n", + "security": [ + { + "ApiKeyAuth": [] + } + ], + "parameters": [ + { + "in": "path", + "name": "id", + "required": true, + "description": "ID of the asset to get the automations for.", + "schema": { + "type": "integer" + } + } + ], + "responses": { + "200": { + "description": "PROCESSED", + "content": { + "application/json": { + "examples": { + "automations": { + "summary": "List of automations", + "value": { + "automations": [ + { + "id": 1, + "created_at": "2026-07-11T00:00:00+00:00", + "asset_id": 1, + "type": "forecasts", + "name": "Day-ahead PV forecasts", + "cronstr": "0 6 * * *", + "timezone": "Europe/Amsterdam", + "cursor": "2026-07-11T04:00:00+00:00", + "recurrence_description": "At 06:00", + "active": true + } + ] + } + } + } + } + } + }, + "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": { "get": { "summary": "Download an embeddable chart with time series data", @@ -3769,6 +3945,7 @@ "status": "finished", "err": null, "enqueued_at": "2023-10-01T00:00:00", + "created_via": "API", "metadata_hash": "abc123" } ], @@ -4891,6 +5068,69 @@ }, "/api/v3_0": {}, "/api/v3_0/sensors/data": {}, + "/api/v3_0/sources/{id}": { + "get": { + "summary": "Get one data source.", + "description": "Returns the full record of one data source, including the attributes in which\ndata generators (such as forecasters, schedulers and reporters) store their\nconfiguration.\n\nThe access rules are the same as for listing data sources.\n", + "security": [ + { + "ApiKeyAuth": [] + } + ], + "parameters": [ + { + "in": "path", + "name": "id", + "required": true, + "description": "ID of the data source.", + "schema": { + "type": "integer" + } + } + ], + "responses": { + "200": { + "description": "PROCESSED", + "content": { + "application/json": { + "example": { + "id": 6, + "name": "Seita", + "type": "forecaster", + "model": "TrainPredictPipeline", + "version": "1", + "description": "Seita's TrainPredictPipeline model v1", + "account_id": 2, + "user_id": null, + "attributes": { + "data_generator": { + "config": { + "model": "CustomLGBM" + } + } + } + } + } + } + }, + "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": [ + "Sources" + ] + } + }, "/api/v3_0/sources": { "get": { "summary": "List accessible data sources and defined source types.", diff --git a/flexmeasures/ui/templates/assets/asset_automations.html b/flexmeasures/ui/templates/assets/asset_automations.html new file mode 100644 index 0000000000..2646a9bbfc --- /dev/null +++ b/flexmeasures/ui/templates/assets/asset_automations.html @@ -0,0 +1,221 @@ +{% extends "base.html" %} +{% set active_page = "assets" %} + +{% block title %} {{ asset.name }} - Automations {% endblock %} + +{% block divs %} +{% block breadcrumbs %} {{ super() }} {% endblock %} + +
+
+
+
+
+ +

+ Automations of {{ asset.name }} + +

+

+ Recent jobs created by these automations are listed on the + status page. + 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. +

+ +
    + + + +
+
+
+
+
+
+
+
+ +
+
+
+
+
+ + +{% endblock %} diff --git a/flexmeasures/ui/templates/sensors/index.html b/flexmeasures/ui/templates/sensors/index.html index 39a969e1b2..72813f5448 100644 --- a/flexmeasures/ui/templates/sensors/index.html +++ b/flexmeasures/ui/templates/sensors/index.html @@ -7,7 +7,10 @@ {% block divs %} {% block breadcrumbs %} {{ super() }} {% endblock %} - + + +
+
@@ -312,6 +315,26 @@
Attributes
{% endfor %} {% endif %} + {% if feeding_automations %} +
+ Automations + +
+ + + {% for automation in feeding_automations %} + + + + + {% endfor %} +
+ {{ automation.name }} + + {{ automation.recurrence_description }} + {% if not automation.active %}⚪{% endif %} +
+ {% endif %}
@@ -323,11 +346,30 @@
Statistics
+ +
@@ -444,6 +486,11 @@
Statistics