diff --git a/documentation/changelog.rst b/documentation/changelog.rst index e59139f896..ac46ed1633 100644 --- a/documentation/changelog.rst +++ b/documentation/changelog.rst @@ -117,6 +117,7 @@ v1.0.0 | August 25, 2026 New features ------------- +* Reports can be computed on a recurring basis by automations (``flexmeasures add automation --type reporting``), with a rolling report window expressed as Pandas offsets, or defaulting to the period since the automation's last covered window [see `PR #2297 `_] * ``flexmeasures show data-sources`` now shows which organisation a data source belongs to, and can list the sensors holding data recorded by a given source [see `PR #2401 `_] * The flex-context can now define multiple commodities, each specifying their own prices and grid capacities [see `PR #1946 `_, `PR #2172 `_, `PR #2235 `_, `PR #2271 `_, `PR #2355 `_ and `PR #2380 `_] * Support multiple feeders to a shared storage [see `PR #2001 `_, `PR #2321 `_, `PR #2322 `_, `PR #2325 `_ and `PR #2431 `_] diff --git a/documentation/cli/change_log.rst b/documentation/cli/change_log.rst index 26bfbdb900..5028bf2885 100644 --- a/documentation/cli/change_log.rst +++ b/documentation/cli/change_log.rst @@ -24,7 +24,7 @@ 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, with ``--type forecasting`` or ``--type scheduling`` saying which task to automate). Each automation carries its own IANA timezone (``--timezone``), in which its cron expression is interpreted. +* Add ``flexmeasures add automation``, ``flexmeasures edit automation`` and ``flexmeasures delete automation`` to manage automations (recurring tasks on an asset, with ``--type forecasting``, ``--type scheduling`` or ``--type reporting`` saying which task to automate). Each automation carries its own IANA timezone (``--timezone``), in which its cron expression is interpreted. * ``flexmeasures add automation --type scheduling`` refuses a flex config field which fixes a moment in time, such as ``soc-at-start`` or a ``soc-targets`` entry with a ``datetime``, naming the field: a recurring schedule automation computes a fresh schedule on every run, so such a value would be stale on the next one. * 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. * Add ``flexmeasures jobs run-automation --automation `` to queue the jobs for a single run of one automation, now, on top of its recurring runs. This leaves the automation's cursor alone, so its next recurring run still happens as scheduled, and inactive automations can be run this way, too. diff --git a/documentation/cli/commands.rst b/documentation/cli/commands.rst index 7d81a1e309..205b9c41d6 100644 --- a/documentation/cli/commands.rst +++ b/documentation/cli/commands.rst @@ -41,7 +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: a recurring task (computing forecasts or schedules) on an asset, with its own cron timezone. +``flexmeasures add automation`` Add an automation: a recurring task (computing forecasts, schedules or reports) on an asset, with its own cron timezone. ================================================= ======================================= diff --git a/documentation/features/automations.rst b/documentation/features/automations.rst index f797568018..f6e6292b81 100644 --- a/documentation/features/automations.rst +++ b/documentation/features/automations.rst @@ -4,23 +4,30 @@ Automations ============ An **automation** is a recurring task defined on an asset. -For now, an automation computes forecasts or schedules; automating reports is planned. +An automation computes forecasts, schedules or reports. -On each run, the automation queues jobs (so make sure a worker is processing the ``forecasting`` or ``scheduling`` queue, whichever the automation needs, see :ref:`redis-queue`). +On each run, the automation queues jobs (so make sure a worker is processing the ``forecasting``, ``scheduling`` or ``reporting`` queue, whichever the automation needs, 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 or schedule start defaults to the time the automation runs, so each run produces fresh results. -Creating an automation ----------------------- +- a **type**: ``forecasts``, ``schedules`` or ``reports``; +- a **recurrence**: a cron string (e.g. ``"0 6 * * *"`` for daily at 6 AM), interpreted in the automation's own IANA timezone; +- a **data generator** (for forecasts and reports): the forecaster or reporter class and its configuration, stored on a data source. + The data source stays the same across runs, so all results the automation produces attribute to one steady source; +- **parameters**: what to compute on each run, validated by the same schema the CLI and API use for one-off runs. + Timing parameters are resolved freshly on each run, so a recurring automation always computes fresh periods + (see the type-specific sections below for the exact rules); +- an **activation status**: only active automations run. -Here is how you create an automation in the CLI, asking for daily (at 6 AM) forecasts of sensor 12: +Managing automations +-------------------- -.. code-block:: bash +Automations can be managed in three ways: flexmeasures add automation --asset 3 --name "Daily PV forecasts" --type forecasting \ --cron "0 6 * * *" --timezone Europe/Amsterdam --sensor 12 -``--type`` says which task to automate (``forecasting`` or ``scheduling``, matching the queue the jobs go to), and defaults to ``forecasting``. +``--type`` says which task to automate (``forecasting``, ``scheduling`` or ``reporting``, matching the queue the jobs go to), and defaults to ``forecasting``. 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. @@ -68,10 +75,32 @@ For example, this automation queues a scheduling job every hour, each time sched echo 'duration: "PT12H"' > trigger-message.yml flexmeasures add automation --asset 3 --name "Hourly schedules" --cron "0 * * * *" --type scheduling --parameters trigger-message.yml +Automating reports +------------------ + +A report automation's parameters are report parameters, as ``flexmeasures add report`` accepts them, and its reporter is named with ``--reporter`` and configured with ``--config`` (see :ref:`reporting`). +As for a forecast automation, the reporter and its configuration are stored on a data source, so ``--source`` can reuse the data source of an existing reporter instead. + +The report window is resolved on every run, so that each run reports on a fresh period. +Give ``start-offset`` and ``end-offset`` in the parameters for a rolling window: both take comma-separated Pandas offsets, plus ``DB`` (day begin) and ``HB`` (hour begin), applied to the run time. +For instance, ``start-offset: "-1D,DB"`` with ``end-offset: "DB"`` reports on the whole of the previous day. +Offsets are resolved in the timezone of the first output sensor, falling back to the platform timezone. + +Leave the timing fields out to report on the period since the automation last covered one, falling back to the last cron period on the first run. +That coverage is recorded by the reporting job itself, once it has succeeded, so a failed report leaves no permanent gap: the next run starts where the last successful one ended. +An absolute ``start`` or ``end`` is passed through untouched, which means every run then reports on the same period. + +For example, this automation queues a reporting job every night, reporting on the previous day: + +.. code-block:: bash + + flexmeasures add automation --asset 3 --name "Daily self-consumption report" --cron "0 1 * * *" --type reporting \ + --reporter PandasReporter --config reporter-config.yml --parameters report-parameters.yml + Running automations -------------------- +-------------------- -For automations to actually run, let a cron job execute the following command once per minute: +An automation is due whenever its cron string matches the current minute in its configured timezone. To actually run due automations, let a cron job execute the following command once per minute: .. code-block:: bash @@ -84,7 +113,8 @@ Timing parameters that default to the run time are resolved when that catch-up r 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. +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 result. Running one automation on demand -------------------------------- @@ -111,6 +141,16 @@ Automations defined on an asset can be viewed on the asset's *Automations* page 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. +Automating each feature +----------------------- + +The parameters stored on an automation follow the same schemas as one-off CLI/API calls, with type-specific rules for resolving timing on each run: + +- :ref:`automating_forecasts` — forecast parameters; the forecast start defaults to the run time. +- :ref:`automating_schedules` — a schedule trigger message; omit ``start`` to schedule from the run time. +- :ref:`automating_reports` — report parameters; use ``start-offset``/``end-offset`` (Pandas offsets) for a rolling window, + or omit timing fields to report on the period since the last successfully covered report window. + .. _automation_cursor: Appendix: how the runner decides what is due diff --git a/documentation/features/forecasting.rst b/documentation/features/forecasting.rst index 32991bcd58..32e80475c5 100644 --- a/documentation/features/forecasting.rst +++ b/documentation/features/forecasting.rst @@ -219,6 +219,20 @@ Usage: 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`. -Schedules can be automated in the same way — see :ref:`automating_schedules`. +Instead of asking for forecasts one at a time, you can set up an *automation*: a recurring task defined on an asset (see :ref:`automations` for the full concept, including how to manage and run automations). +On each run, the automation queues forecasting jobs (so make sure a worker is processing the ``forecasting`` queue, see :ref:`redis-queue`). +When the automation was created, its forecast parameters (see above) were stored, 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. +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. + +Here is how you create a forecast 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 forecasting \ + --cron "0 6 * * *" --timezone Europe/Amsterdam --sensor 12 + +A forecast automation accepts everything ``flexmeasures add forecast`` accepts, such as ``--forecaster`` to pick the forecaster and ``--config`` to configure it. +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. diff --git a/documentation/features/reporting.rst b/documentation/features/reporting.rst index 5262af925c..c6b749ce1b 100644 --- a/documentation/features/reporting.rst +++ b/documentation/features/reporting.rst @@ -129,3 +129,28 @@ Here, the ``ProfitOrLossReporter`` used as source (with Id 6) is the one we conf With the offsets, we control the timing ― we indicate that we want the new report to encompass the day of tomorrow (see Pandas offset strings). The report sensor will now store all costs which we know will be made tomorrow by the schedule. + +.. _automating_reports: + +Automating reports +-------------------- + +Besides running a report once, a report can be computed on a recurring basis by an *automation* defined on the asset. +See :ref:`automations` for the full concept, including how to manage and run automations. + +The reporter and its configuration are stored on a data source, which stays the same across runs, so all of the automation's report results attribute to one source. +The report parameters are stored on the automation itself, and their timing is resolved afresh on each run: + +- Use ``start-offset`` and/or ``end-offset`` fields (comma-separated Pandas offsets, like the CLI options above) for a rolling window relative to the claimed cron occurrence, in the timezone of the first output sensor. + For instance, ``"start-offset": "-1D,DB"`` with ``"end-offset": "DB"`` reports on the whole previous day. +- Omit timing fields entirely to report from the end of the latest successfully completed report window through the claimed cron occurrence. + When no completed window is known, such as on the first run, the start falls back to the previous cron occurrence in the automation's timezone. + The completion marker only moves forward, so concurrent reporting workers that finish out of order cannot reopen an already covered period. +- Absolute ``start``/``end`` fields are also accepted, but draw a warning, as each run would then compute the same period. + +For example, this automation computes a report over each past day, every morning at 1 AM: + +.. code-block:: bash + + flexmeasures add automation --asset 3 --name "Daily aggregation report" --cron "0 1 * * *" --type reporting \ + --reporter PandasReporter --config reporter-config.yml --parameters report-parameters.yml diff --git a/flexmeasures/api/v3_0/assets.py b/flexmeasures/api/v3_0/assets.py index b88822b5c4..b4794227a6 100644 --- a/flexmeasures/api/v3_0/assets.py +++ b/flexmeasures/api/v3_0/assets.py @@ -1723,7 +1723,7 @@ def post_automation(self, id: int, asset: GenericAsset): automation_type=automation_data["type"], active=automation_data["active"], parameters=automation_data["parameters"], - forecaster_class=automation_data["forecaster"], + generator_class=automation_data["generator"], config=automation_data["config"], origin="API", check_permissions=True, diff --git a/flexmeasures/api/v3_0/tests/test_automations_api.py b/flexmeasures/api/v3_0/tests/test_automations_api.py index 9f17e819d1..2c1d11f1d6 100644 --- a/flexmeasures/api/v3_0/tests/test_automations_api.py +++ b/flexmeasures/api/v3_0/tests/test_automations_api.py @@ -216,6 +216,170 @@ def test_post_automation( fresh_db.session.flush() +@pytest.mark.parametrize( + "requesting_user", ["test_prosumer_user_2@seita.nl"], indirect=True +) +def test_post_automation_with_foreign_sensor( + app, + db, + setup_accounts, + add_battery_assets, + requesting_user, +): + """Referencing a sensor outside the caller's reach is forbidden.""" + from datetime import timedelta + + from flexmeasures.data.models.generic_assets import GenericAsset + from flexmeasures.data.models.time_series import Sensor + + battery = add_battery_assets["Test battery"] + foreign_asset = GenericAsset( + name="Foreign asset", + generic_asset_type=battery.generic_asset_type, + owner=setup_accounts["Dummy"], + ) + foreign_sensor = Sensor( + "foreign power", + generic_asset=foreign_asset, + event_resolution=timedelta(minutes=15), + unit="MW", + ) + db.session.add(foreign_sensor) + db.session.flush() + with app.test_client() as client: + response = client.post( + url_for("AssetAPI:post_automation", id=battery.id), + json={ + "name": "Sneaky forecasts", + "cronstr": "0 6 * * *", + "type": "forecasting", + "parameters": {"sensor": foreign_sensor.id}, + }, + ) + assert response.status_code == 403 + assert ( + db.session.execute( + select(Automation).filter_by(name="Sneaky forecasts") + ).scalar_one_or_none() + is None + ) + + +@pytest.mark.parametrize( + "requesting_user", ["test_prosumer_user_2@seita.nl"], indirect=True +) +def test_post_report_automation_with_foreign_config_sensor( + app, + db, + setup_accounts, + add_battery_assets, + requesting_user, +): + """Reporter configuration may not read a sensor outside the caller's reach.""" + from flexmeasures.data.models.generic_assets import GenericAsset + + battery = add_battery_assets["Test battery"] + foreign_asset = GenericAsset( + name="Foreign price asset", + generic_asset_type=battery.generic_asset_type, + owner=setup_accounts["Dummy"], + ) + foreign_price_sensor = Sensor( + "private foreign price", + generic_asset=foreign_asset, + event_resolution=timedelta(hours=1), + unit="EUR/MWh", + ) + report_sensor = Sensor( + "profit report", + generic_asset=battery, + event_resolution=timedelta(hours=1), + unit="EUR", + ) + db.session.add_all([foreign_price_sensor, report_sensor]) + db.session.flush() + + with app.test_client() as client: + response = client.post( + url_for("AssetAPI:post_automation", id=battery.id), + json={ + "name": "Cross-organisation profit report", + "cronstr": "0 1 * * *", + "type": "reporting", + "generator": "ProfitOrLossReporter", + "config": { + "consumption_price_sensor": foreign_price_sensor.id, + }, + "parameters": { + "input": [{"sensor": battery.sensors[0].id}], + "output": [{"sensor": report_sensor.id}], + }, + }, + ) + + assert response.status_code == 403 + assert foreign_price_sensor.name not in response.text + assert ( + db.session.execute( + select(Automation).filter_by(name="Cross-organisation profit report") + ).scalar_one_or_none() + is None + ) + db.session.commit() + + +@pytest.mark.parametrize( + "requesting_user", ["test_prosumer_user_2@seita.nl"], indirect=True +) +def test_post_report_automation_rejects_output_outside_asset_subtree( + app, + db, + add_battery_assets, + requesting_user, +): + """Report output must stay on the automation asset or a descendant.""" + battery = add_battery_assets["Test battery"] + sibling_battery = add_battery_assets["Test small battery"] + report_sensor = Sensor( + "sibling report output", + generic_asset=sibling_battery, + event_resolution=timedelta(hours=1), + unit="MW", + ) + db.session.add(report_sensor) + db.session.flush() + + with app.test_client() as client: + response = client.post( + url_for("AssetAPI:post_automation", id=battery.id), + json={ + "name": "Misplaced report output", + "cronstr": "0 1 * * *", + "type": "reporting", + "generator": "PandasReporter", + "config": { + "required_input": [{"name": "flow"}], + "required_output": [{"name": "copied_flow"}], + "transformations": [ + { + "df_input": "flow", + "df_output": "copied_flow", + "method": "copy", + } + ], + }, + "parameters": { + "input": [{"name": "flow", "sensor": battery.sensors[0].id}], + "output": [{"name": "copied_flow", "sensor": report_sensor.id}], + }, + }, + ) + + assert response.status_code == 422 + assert "must belong to asset" in response.text + db.session.commit() + + @pytest.mark.parametrize( "requesting_user", ["test_prosumer_user_2@seita.nl"], indirect=True ) diff --git a/flexmeasures/cli/data_add.py b/flexmeasures/cli/data_add.py index ae2af2ae75..cf512c46da 100755 --- a/flexmeasures/cli/data_add.py +++ b/flexmeasures/cli/data_add.py @@ -1460,7 +1460,8 @@ def _assemble_forecaster_config_and_parameters( 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: + # skip unset options (click passes None, or an empty tuple for multiple-value options) + if field_value is not None and field_value != (): if field_name in { "future_regressors", "past_regressors", @@ -1509,8 +1510,8 @@ def _assemble_forecaster_config_and_parameters( 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} + # Drop unset values + parameters = {k: v for k, v in parameters.items() if v is not None and v != ()} return config, parameters @@ -1776,20 +1777,28 @@ def add_forecast( # noqa: C901 " 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( + "--reporter", + "reporter_class", + required=False, + type=click.STRING, + help="Reporter class registered in flexmeasures.data.models.reporting or in an available flexmeasures plugin (only used for --type reporting)." + " Use the command `flexmeasures show reporters` to list all the available reporters.", +) @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.", + help="DataSource ID of the data generator (`Forecaster` or `Reporter`). The generator class and its configuration are read from" + " the data source's attributes, so --forecaster/--reporter 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." + help="Path to the JSON or YAML file with the configuration of the forecaster or reporter." " Cannot be combined with --source, which already determines the configuration.", ) @click.option( @@ -1798,7 +1807,8 @@ def add_forecast( # noqa: C901 required=False, type=click.File("r"), help="Path to the JSON or YAML file with the parameters used on each run of the automation:" - " forecast parameters for --type forecasting, or a schedule trigger message for --type scheduling.", + " forecast parameters for --type forecasting, a schedule trigger message for --type scheduling," + " or report parameters for --type reporting.", ) @add_cli_options_from_schema( ForecasterParametersSchema(), hidden=True, force_optional=True @@ -1814,13 +1824,14 @@ def add_automation( automation_type: str, inactive: bool = False, forecaster_class: str | None = None, + reporter_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 (computing forecasts or schedules) on an asset. + Add an automation: a recurring task (computing forecasts, schedules or reports) on an asset. \b Examples @@ -1829,12 +1840,19 @@ def add_automation( --parameters forecast-parameters.yml flexmeasures add automation --asset 3 --name "Hourly schedules" --cron "0 * * * *" --type scheduling --parameters trigger-message.yml + flexmeasures add automation --asset 3 --name "Daily self-consumption report" + --cron "0 1 * * *" --type reporting --reporter PandasReporter + --config reporter-config.yml --parameters report-parameters.yml + - For forecasts, the forecaster configuration is stored on a data source, and - the forecast parameters are validated and stored on the automation itself. + For forecasts and reports, the data generator configuration is stored on a + data source, and the parameters are validated and stored on the automation itself. For schedules, the parameters form a schedule trigger message (as accepted by the [POST] /assets/(id)/schedules/trigger API endpoint, without the asset id); omit its "start" field to schedule from the run time on each run. + For reports, use "start-offset"/"end-offset" (comma-separated Pandas offsets, + applied to the run time) for a rolling report window, or omit timing fields + entirely to report on the last cron period. Each time the automation runs, jobs are queued (see `flexmeasures jobs run-automations`). Alternatively, pass an existing data source (--source) to reuse the forecaster @@ -1892,7 +1910,9 @@ def add_automation( automation_type=automation_type, active=not inactive, parameters=parameters, - forecaster_class=forecaster_class, + generator_class=( + reporter_class if automation_type == "reporting" else forecaster_class + ), config=config, source=source, origin="CLI", @@ -2256,6 +2276,7 @@ def add_report( # noqa: C901 ) raise click.Abort() if as_job and not save_config: + click.secho( "Saving the reporter config to its data source (required for --as-job).", **MsgStyle.WARN, diff --git a/flexmeasures/cli/jobs.py b/flexmeasures/cli/jobs.py index c674267914..c45a111e48 100644 --- a/flexmeasures/cli/jobs.py +++ b/flexmeasures/cli/jobs.py @@ -116,7 +116,9 @@ def run_automations(): ) continue try: - returns = run_automation(automation) + returns = run_automation( + automation, scheduled_at=due_automation.scheduled_at + ) n_jobs = returns.get("n_jobs") if returns else 0 click.secho( f"Automation {automation.id} ('{automation.name}') queued {n_jobs} {automation.type} job(s) for asset {automation.asset_id}.", diff --git a/flexmeasures/cli/tests/test_automations.py b/flexmeasures/cli/tests/test_automations.py index 0731ed3302..e3a9ccb3c0 100644 --- a/flexmeasures/cli/tests/test_automations.py +++ b/flexmeasures/cli/tests/test_automations.py @@ -2,6 +2,7 @@ import json import pytest +import yaml import pytz from types import SimpleNamespace @@ -1113,6 +1114,263 @@ class FakeJob: assert calls["kwargs"]["end"] - start == timedelta(hours=12) +def test_prepare_report_parameters(app): + """Report start/end resolve per run: from Pandas offsets, or defaulting to the last cron period.""" + import pandas as pd + + from flexmeasures.data.services.automations import prepare_report_parameters + from flexmeasures.utils.time_utils import get_timezone + + now = pd.Timestamp("2026-07-11T14:00:00+02:00") + # without an output sensor, offsets resolve in the platform timezone + local_now = now.tz_convert(get_timezone()) + + # default: the last cron period (hourly cron -> the previous hour) + message = prepare_report_parameters({}, "0 * * * *", now=now) + assert pd.Timestamp(message["start"]) == now - pd.Timedelta(hours=1) + assert pd.Timestamp(message["end"]) == now + + # The fallback cron period is interpreted in the automation timezone and + # ends at the claimed run rather than at a delayed runner's wall time. + scheduled_at = datetime(2026, 1, 1, 16, 0, tzinfo=timezone.utc) + message = prepare_report_parameters( + {}, + "0 1 * * *", + now=datetime(2026, 1, 2, 0, 30, tzinfo=timezone.utc), + cron_timezone="Asia/Seoul", + scheduled_at=scheduled_at, + ) + assert pd.Timestamp(message["start"]) == pd.Timestamp("2025-12-31T16:00:00+00:00") + assert pd.Timestamp(message["end"]) == pd.Timestamp(scheduled_at) + + # A cron run in Amsterdam's spring gap is canonicalized to 03:00, + # while its report starts at the prior day's real 02:30 run. + spring_run = datetime(2026, 3, 29, 1, 0, tzinfo=timezone.utc) + message = prepare_report_parameters( + {}, + "30 2 * * *", + cron_timezone="Europe/Amsterdam", + scheduled_at=spring_run, + ) + assert pd.Timestamp(message["start"]) == pd.Timestamp("2026-03-28T01:30:00+00:00") + assert pd.Timestamp(message["end"]) == pd.Timestamp(spring_run) + + # with a known actual last run, the window starts there instead + app.redis_connection.set("automation-last-run:1234", "2026-07-11T09:30:00+02:00") + try: + message = prepare_report_parameters( + {}, "0 * * * *", now=now, automation_id=1234 + ) + assert pd.Timestamp(message["start"]) == pd.Timestamp( + "2026-07-11T09:30:00+02:00" + ) + assert pd.Timestamp(message["end"]) == now + # an unknown automation id still falls back to the last cron period + message = prepare_report_parameters( + {}, "0 * * * *", now=now, automation_id=5678 + ) + assert pd.Timestamp(message["start"]) == now - pd.Timedelta(hours=1) + finally: + app.redis_connection.delete("automation-last-run:1234") + + # offsets applied to the run time; "DB" floors to the day begin + message = prepare_report_parameters( + {"start-offset": "-1D,DB", "end-offset": "DB"}, "0 1 * * *", now=now + ) + assert ( + pd.Timestamp(message["start"]) == (local_now - pd.Timedelta(days=1)).normalize() + ) + assert pd.Timestamp(message["end"]) == local_now.normalize() + assert "start-offset" not in message and "end-offset" not in message + + # absolute datetimes pass through untouched + message = prepare_report_parameters( + {"start": "2026-01-01T00:00:00+01:00", "end": "2026-01-02T00:00:00+01:00"}, + "0 1 * * *", + now=now, + ) + assert pd.Timestamp(message["start"]) == pd.Timestamp("2026-01-01T00:00:00+01:00") + assert pd.Timestamp(message["end"]) == pd.Timestamp("2026-01-02T00:00:00+01:00") + + +def test_report_coverage_cannot_move_backwards(app, clean_redis): + """An older report finishing later may not reopen already covered periods.""" + from flexmeasures.data.services.automations import ( + get_automation_last_run, + record_automation_run, + ) + + later_end = datetime(2026, 1, 3, tzinfo=timezone.utc) + older_end = datetime(2026, 1, 2, tzinfo=timezone.utc) + + assert record_automation_run(42, later_end) is True + assert record_automation_run(42, older_end) is False + assert get_automation_last_run(42) == later_end + + +def _report_automation_cli_input( + tmp_path, + sensor1_id, + sensor2_id, + report_sensor_id, + parameters_extra=None, + asset_id=1, +): + """CLI input for a report automation using a simple PandasReporter aggregation.""" + reporter_config = dict( + required_input=[{"name": "sensor_1"}, {"name": "sensor_2"}], + required_output=[{"name": "df_agg"}], + transformations=[ + dict( + df_input="sensor_1", + method="add", + args=["@sensor_2"], + df_output="df_agg", + ), + dict(method="resample_events", args=["2h"]), + ], + ) + parameters = dict( + input=[ + dict(name="sensor_1", sensor=sensor1_id), + dict(name="sensor_2", sensor=sensor2_id), + ], + output=[dict(name="df_agg", sensor=report_sensor_id)], + **(parameters_extra or {}), + ) + config_file = tmp_path / "reporter_config.yml" + config_file.write_text(yaml.dump(reporter_config)) + parameters_file = tmp_path / "parameters.yml" + parameters_file.write_text(yaml.dump(parameters)) + return [ + "--asset", str(asset_id), + "--name", "Aggregation report", + "--cron", "0 1 * * *", + "--type", "reporting", + "--reporter", "PandasReporter", + "--config", str(config_file), + "--parameters", str(parameters_file), + ] # fmt: skip + + +def test_add_report_automation(app, fresh_db, setup_dummy_data, tmp_path): + """Create a reports automation; the reporter config lands on a data source.""" + from flexmeasures.cli.data_add import add_automation + + sensor1_id, sensor2_id, report_sensor_id, _ = setup_dummy_data + from flexmeasures.data.models.time_series import Sensor + + report_sensor = fresh_db.session.get(Sensor, report_sensor_id) + runner = app.test_cli_runner() + result = runner.invoke( + add_automation, + _report_automation_cli_input( + tmp_path, + sensor1_id, + sensor2_id, + report_sensor_id, + parameters_extra={"start-offset": "-1D,DB", "end-offset": "DB"}, + asset_id=report_sensor.generic_asset_id, + ), + ) + assert "Successfully created" in result.output, result.output + automation = fresh_db.session.execute(select(Automation)).scalar_one() + assert automation.type == "reporting" + assert automation.generator is not None + assert automation.generator.model == "PandasReporter" + assert automation.parameters["start-offset"] == "-1D,DB" + + # a reports automation without a reporter is rejected + result = runner.invoke( + add_automation, + [ + "--asset", "1", + "--name", "No reporter", + "--cron", "0 1 * * *", + "--type", "reporting", + ], + ) # fmt: skip + assert result.exit_code != 0 + assert "reporter is required" in result.output + + # invalid time offsets are rejected (they would otherwise be silently skipped at run time) + result = runner.invoke( + add_automation, + _report_automation_cli_input( + tmp_path, + sensor1_id, + sensor2_id, + report_sensor_id, + parameters_extra={ + "start-offset": "P1D,DB" + }, # ISO duration, not a Pandas offset + ), + ) + assert result.exit_code != 0 + assert "Invalid start-offset" in result.output + + +def test_run_report_automation(app, fresh_db, setup_dummy_data, clean_redis, tmp_path): + """A due reports automation queues a reporting job; a worker computes and saves the report.""" + from flexmeasures.cli.data_add import add_automation + from flexmeasures.cli.jobs import run_automations + from flexmeasures.data.models.time_series import Sensor + from flexmeasures.utils.job_utils import work_on_rq + + sensor1_id, sensor2_id, report_sensor_id, _ = setup_dummy_data + report_sensor = fresh_db.session.get(Sensor, report_sensor_id) + runner = app.test_cli_runner() + cli_input = _report_automation_cli_input( + tmp_path, + sensor1_id, + sensor2_id, + report_sensor_id, + # the dummy data lives in April 2023, so use an absolute reporting window + parameters_extra={ + "start": "2023-04-10T00:00:00+00:00", + "end": "2023-04-10T10:00:00+00:00", + }, + asset_id=report_sensor.generic_asset_id, + ) + cli_input[cli_input.index("0 1 * * *")] = "* * * * *" # due every minute + result = runner.invoke(add_automation, cli_input) + assert "Successfully created" in result.output, result.output + automation = fresh_db.session.execute(select(Automation)).scalar_one() + + result = runner.invoke(run_automations) + assert result.exit_code == 0, result.output + assert "queued 1 reporting job(s)" in result.output, result.output + + # the queued job recorded how it was created + jobs = app.queues["reporting"].jobs + assert len(jobs) == 1 + assert jobs[0].meta["trigger"] == { + "origin": "automation", + "automation_id": automation.id, + } + + # the covered-until anchor is only recorded once the job succeeds + assert not app.redis_connection.get(f"automation-last-run:{automation.id}") + + # process the job and check the report got saved + work_on_rq(app.queues["reporting"]) + report_sensor = fresh_db.session.get(Sensor, report_sensor_id) + stored_report = report_sensor.search_beliefs( + event_starts_after="2023-04-10T00:00:00+00:00", + event_ends_before="2023-04-10T10:00:00+00:00", + ) + assert (stored_report.values.T == [1, 2 + 3, 4 + 5, 6 + 7, 8 + 9]).all() + + # the successful job recorded the end of the report window as covered + import pandas as pd + + covered_until = app.redis_connection.get(f"automation-last-run:{automation.id}") + assert covered_until is not None + assert pd.Timestamp(covered_until.decode()) == pd.Timestamp( + "2023-04-10T10:00:00+00:00" + ) + + def test_run_automations( app, fresh_db, setup_dummy_data, clean_redis, freeze_server_now ): @@ -1158,6 +1416,9 @@ def test_run_automations( and job.meta["trigger"]["automation_id"] in automation_ids for job in jobs ) + # the run got recorded (used e.g. to anchor default report windows) + for automation in automations: + assert app.redis_connection.get(f"automation-last-run:{automation.id}") # running again within the same minute does not queue jobs twice n_jobs = len(jobs) result = runner.invoke(run_automations) @@ -1238,7 +1499,7 @@ def test_failed_automation_attempt_is_not_retried(app, clean_redis, mocker): ) mocker.patch("flexmeasures.cli.jobs.claim_due_automation", return_value=True) - def queue_then_fail(_automation): + def queue_then_fail(_automation, **_kwargs): app.queues["forecasting"].enqueue("flexmeasures.utils.time_utils.server_now") raise RuntimeError("failed after queueing") diff --git a/flexmeasures/data/models/automations.py b/flexmeasures/data/models/automations.py index 902c433b0a..0191a94b72 100644 --- a/flexmeasures/data/models/automations.py +++ b/flexmeasures/data/models/automations.py @@ -36,19 +36,24 @@ class Automation(db.Model, AuthModelMixin): The recurrence is defined by a cron string. Every automation has a data generator, linked through a data source: a forecaster and its configuration for a forecast automation, - and a scheduler and the flex config it computes under for a schedule automation. - A forecast automation's generator is chosen when it is created. + a scheduler and the flex config it computes under for a schedule automation, + and a reporter and its configuration for a report automation. + A forecast automation's generator is chosen when it is created, and so is a report automation's. A schedule automation's is assembled from the trigger message and what its asset stores, so the runner puts it together afresh on every run. """ __tablename__ = "automation" - SUPPORTED_TYPES = ["forecasting", "scheduling"] # later also "reporting" + SUPPORTED_TYPES = ["forecasting", "scheduling", "reporting"] # What one result of each type is called, for messages that talk about a single result, # such as the parameters an automation of that type computes with. - RESULT_NOUNS = {"forecasting": "forecast", "scheduling": "schedule"} + RESULT_NOUNS = { + "forecasting": "forecast", + "scheduling": "schedule", + "reporting": "report", + } id = db.Column(db.Integer, autoincrement=True, primary_key=True) created_at = db.Column( diff --git a/flexmeasures/data/schemas/automations.py b/flexmeasures/data/schemas/automations.py index 5bad71f003..93314f1860 100644 --- a/flexmeasures/data/schemas/automations.py +++ b/flexmeasures/data/schemas/automations.py @@ -87,17 +87,20 @@ class AutomationCreationSchema(Schema): ) active = fields.Bool(load_default=True) parameters = fields.Dict(keys=fields.Str(), load_default=dict) - forecaster = fields.Str( - load_default="TrainPredictPipeline", + generator = fields.Str( + load_default=None, + allow_none=True, metadata={ - "description": "Forecaster class (only used for type 'forecasting')." + "description": "Data generator class, e.g. a forecaster (defaults to TrainPredictPipeline)" + " or a reporter (required for type 'reporting', e.g. PandasReporter)." + " Not used for type 'scheduling'." }, ) config = fields.Dict( keys=fields.Str(), load_default=dict, metadata={ - "description": "Forecaster configuration (only used for type 'forecasting')." + "description": "Data generator configuration (only used for types 'forecasting' and 'reporting')." }, ) diff --git a/flexmeasures/data/schemas/tests/test_reporting.py b/flexmeasures/data/schemas/tests/test_reporting.py index c669609d6d..25dde8983e 100644 --- a/flexmeasures/data/schemas/tests/test_reporting.py +++ b/flexmeasures/data/schemas/tests/test_reporting.py @@ -202,6 +202,22 @@ def test_profit_reporter_config_schema(config, is_valid, db, app, setup_dummy_se }, True, ), + ( # missing required input + { + "output": [{"sensor": 3}], + "start": start, + "end": end, + }, + False, + ), + ( # missing required output + { + "input": [{"sensor": 4}], + "start": start, + "end": end, + }, + False, + ), ( # wrong output unit { "input": [{"sensor": 4}], # unit: MW diff --git a/flexmeasures/data/scripts/data_gen.py b/flexmeasures/data/scripts/data_gen.py index 047e95039d..73efdedcd9 100644 --- a/flexmeasures/data/scripts/data_gen.py +++ b/flexmeasures/data/scripts/data_gen.py @@ -468,6 +468,7 @@ def depopulate_prognoses( if not sensor: num_forecasting_jobs_deleted = app.queues["forecasting"].empty() num_scheduling_jobs_deleted = app.queues["scheduling"].empty() + num_reporting_jobs_deleted = app.queues["reporting"].empty() # Clear all forecasts (data with positive horizon) query = delete(TimedBelief).filter(TimedBelief.belief_horizon > timedelta(hours=0)) @@ -480,6 +481,7 @@ def depopulate_prognoses( if not sensor: click.echo("Deleted %d Forecast Jobs" % num_forecasting_jobs_deleted) click.echo("Deleted %d Schedule Jobs" % num_scheduling_jobs_deleted) + click.echo("Deleted %d Report Jobs" % num_reporting_jobs_deleted) click.echo("Deleted %d forecasts (ex-ante beliefs)" % num_forecasts_deleted) diff --git a/flexmeasures/data/services/automations.py b/flexmeasures/data/services/automations.py index 761953db6c..a28c875841 100644 --- a/flexmeasures/data/services/automations.py +++ b/flexmeasures/data/services/automations.py @@ -14,12 +14,14 @@ from croniter import croniter from croniter.croniter import CroniterError import isodate +import pandas as pd +import pytz from isodate.isoerror import ISO8601Error from flask import current_app from marshmallow import ValidationError from sqlalchemy import select, update -from flexmeasures import Forecaster +from flexmeasures import Forecaster, Reporter from flexmeasures.data import db from flexmeasures.data.models.automations import ( Automation, @@ -35,7 +37,7 @@ check_sensor_access, resolve_data_generator_sensors, ) -from flexmeasures.utils.time_utils import server_now +from flexmeasures.utils.time_utils import apply_offset_chain, get_timezone, server_now @dataclass(frozen=True) @@ -407,8 +409,8 @@ def resolve_schedule_automation_sensors( def resolve_automation_sensors(automation: Automation) -> dict[str, list[Sensor]]: """Work out which sensors an automation reads from and writes to on each run. - Forecast sensors are derived from the data generator, while schedule sensors are - derived from the same prepared trigger message used to queue the scheduling job. + Forecast and report sensors are derived from the data generator, while schedule sensors + are derived from the same prepared trigger message used to queue the scheduling job. Raises `AutomationSensorsUnknown` if that cannot be done, e.g. because a forecast 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). @@ -429,9 +431,17 @@ def resolve_automation_sensors(automation: Automation) -> dict[str, list[Sensor] ) try: data_generator = automation.generator.data_generator + parameters = dict(automation.parameters or {}) + if automation.type == "reporting": + parameters = prepare_report_parameters( + parameters, + automation.cronstr, + automation_id=automation.id, + cron_timezone=automation.timezone, + ) return resolve_data_generator_sensors( data_generator, - data_generator._parameters_schema.load(dict(automation.parameters or {})), + data_generator._parameters_schema.load(parameters), ) except (NotImplementedError, ValidationError) as e: raise AutomationSensorsUnknown( @@ -459,7 +469,7 @@ def get_automations_feeding_sensor(sensor: Sensor) -> list[Automation]: 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 + (see `validate_automation_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. @@ -511,6 +521,194 @@ def prepare_schedule_trigger_message(parameters: dict, asset_id: int) -> dict: return message +def validate_offset_chain(offset_chain: str): + """Raise a ValueError on any offset that apply_offset_chain would silently skip. + + Valid offsets are Pandas offset strings, plus "DB" (day begin) and "HB" (hour begin). + """ + from pandas.tseries.frequencies import to_offset + + for offset in str(offset_chain).split(","): + offset = offset.strip() + if offset.lower() in ("db", "hb"): + continue + try: + to_offset(offset) + except ValueError: + raise ValueError( + f"'{offset}' is not a valid Pandas offset string (nor 'DB'/'HB')." + ) + + +def _last_run_redis_key(automation_id: int) -> str: + return f"automation-last-run:{automation_id}" + + +def record_automation_run(automation_id: int, now: datetime | None = None) -> bool: + """Remember (in Redis) until when this automation's work is covered. + + For forecasts and schedules automations, this is the (enqueue) run time. + For reports automations, the reporting job records the end of the report window + instead, upon success (see run_report_job), so a failed report job does not + create a permanent gap in the reported periods. + """ + from redis.exceptions import WatchError + + if now is None: + now = server_now() + candidate = floor_to_minute(now) + key = _last_run_redis_key(automation_id) + connection = current_app.redis_connection + while True: + with connection.pipeline() as pipeline: + try: + pipeline.watch(key) + value = pipeline.get(key) + if value: + if isinstance(value, bytes): + value = value.decode() + try: + current = floor_to_minute(datetime.fromisoformat(value)) + except ValueError: + current = None + if current is not None and current >= candidate: + pipeline.unwatch() + return False + pipeline.multi() + pipeline.set(key, candidate.isoformat()) + pipeline.execute() + return True + except WatchError: + # Another worker updated the coverage after our read. Re-read it + # and only advance from the new value. + continue + + +def get_automation_last_run(automation_id: int) -> datetime | None: + """Until when this automation's work is covered, if known (the record lives in Redis).""" + from flask import current_app + + value = current_app.redis_connection.get(_last_run_redis_key(automation_id)) + if not value: + return None + if isinstance(value, bytes): + value = value.decode() + try: + return datetime.fromisoformat(value) + except ValueError: + return None + + +def prepare_report_parameters( + parameters: dict, + cronstr: str, + now: datetime | None = None, + automation_id: int | None = None, + cron_timezone: str | None = None, + scheduled_at: datetime | None = None, +) -> dict: + """Complete stored report parameters into a message for the ReporterParametersSchema. + + The (required) start and end of the report are resolved on each run: + + - "start-offset" and "end-offset" fields hold comma-separated Pandas offsets + (e.g. "-1D,DB" for the start of the previous day), applied to the run time + (or to the given absolute start/end), in the timezone of the first output sensor. + - Without offsets or absolutes, the window runs since the end of the automation's + last (successfully) covered window, falling back to the last cron period (from + the previous cron fire time until the run time) when none is known (e.g. on the + first run). + """ + message = dict(parameters) + if scheduled_at is None: + scheduled_at = now if now is not None else server_now() + scheduled_at = floor_to_minute(scheduled_at) + + # Compute the run time in the timezone local to the first output sensor + # (matching `flexmeasures add report`), falling back to the platform timezone. + tz = get_timezone() + outputs = message.get("output") or [] + if ( + outputs + and isinstance(outputs[0], dict) + and outputs[0].get("sensor") is not None + ): + from flexmeasures.data.models.time_series import Sensor + + try: + output_sensor = db.session.get(Sensor, int(outputs[0]["sensor"])) + except (TypeError, ValueError): + output_sensor = None + if output_sensor is not None: + tz = pytz.timezone(output_sensor.timezone) + now = scheduled_at.astimezone(tz) + + start_offset = message.pop("start-offset", None) + end_offset = message.pop("end-offset", None) + start = pd.Timestamp(message["start"]) if "start" in message else None + end = pd.Timestamp(message["end"]) if "end" in message else None + + # Apply offsets to the given absolute datetime, or to the run time + if start_offset is not None: + start = apply_offset_chain( + start if start is not None else pd.Timestamp(now), start_offset + ) + if end_offset is not None: + end = apply_offset_chain( + end if end is not None else pd.Timestamp(now), end_offset + ) + + # Default to the window since the last covered window's end, falling back to + # the last cron period (from the previous cron fire time until the run time) + if start is None: + last_run = ( + get_automation_last_run(automation_id) + if automation_id is not None + else None + ) + if last_run is not None: + start = last_run + else: + cron_tz = ( + ZoneInfo(cron_timezone) + if cron_timezone is not None + else ZoneInfo(str(get_timezone())) + ) + nominal_scheduled_at = _as_nominal_wall_time( + scheduled_at.astimezone(cron_tz) + ) + previous_nominal = croniter(cronstr, nominal_scheduled_at).get_prev( + datetime + ) + start = _canonical_run_time(previous_nominal, cron_tz) + # A skipped wall time can canonicalize to the first valid instant after + # the gap, which may be the current run. Step back once more so + # the first report still covers a non-empty cron period. + if start >= scheduled_at: + previous_nominal = croniter(cronstr, previous_nominal).get_prev( + datetime + ) + start = _canonical_run_time(previous_nominal, cron_tz) + if end is None: + end = now + + message["start"] = pd.Timestamp(start).isoformat() + message["end"] = pd.Timestamp(end).isoformat() + return message + + +def _relevant_sensor_ids(automation: Automation, parameter_values: list) -> set[int]: + """The asset's sensor ids, plus any (castable) sensor ids among the given parameter values.""" + sensor_ids = {sensor.id for sensor in automation.asset.sensors} + for value in parameter_values: + if value is not None: + try: + sensor_ids.add(int(value)) + except (TypeError, ValueError): + pass + return sensor_ids + + def resolve_schedule_generator(asset_id: int, parameters: dict) -> DataSource: """The data source describing the scheduler a schedule automation runs, and the flex config it runs with. @@ -567,27 +765,40 @@ def get_automation_job_stats(automation: Automation) -> dict[str, int]: Note that jobs in Redis have a limited TTL, so this only counts fairly recent jobs. """ - # Determine the job cache entries to scan. + # Determine the job cache entries to scan. Forecasting and reporting jobs are cached under their target/output sensor(s), + # which may belong to a different asset than the automation's own asset. + parameters = automation.parameters or {} if automation.type == "scheduling": + # Scheduling jobs are cached under the asset (multi-device wrap-up jobs) - # and under individual sensors (per-device jobs). - assets = [automation.asset, *automation.asset.offspring] + # and under individual device sensors (per-device jobs), which may belong + # to child assets rather than the automation's own (site) asset. + sensor_ids = _relevant_sensor_ids( + automation, + [ + entry.get("sensor") + for entry in parameters.get("flex-model", []) or [] + if isinstance(entry, dict) + ], + ) cache_refs = [(automation.asset_id, "scheduling", "asset")] + [ - (sensor.id, "scheduling", "sensor") - for asset in assets - for sensor in asset.sensors + (sensor_id, "scheduling", "sensor") for sensor_id in sensor_ids ] + elif automation.type == "reporting": + sensor_ids = _relevant_sensor_ids( + automation, + [ + output.get("sensor") + for output in parameters.get("output", []) or [] + if isinstance(output, dict) + ], + ) + cache_refs = [(sensor_id, "reporting", "sensor") for sensor_id in sensor_ids] else: - # Forecasting 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 + sensor_ids = _relevant_sensor_ids( + automation, + [parameters.get("sensor"), parameters.get("sensor-to-save")], + ) cache_refs = [(sensor_id, "forecasting", "sensor") for sensor_id in sensor_ids] counts: dict[str, int] = {} @@ -603,6 +814,83 @@ def get_automation_job_stats(automation: Automation) -> dict[str, int]: return counts +def _prepare_forecast_automation( + asset, parameters: dict, generator_class: str | None, config: dict | None, source +) -> tuple[Forecaster, dict, list[str]]: + """Validate forecast automation parameters and set up the forecaster, without creating a data source.""" + from flexmeasures.data.models.time_series import Sensor + from flexmeasures.data.schemas.forecasting.pipeline import ( + ForecasterParametersSchema, + ) + from flexmeasures.data.services.data_sources import get_data_generator + + warnings = [] + deserialized_parameters = ForecasterParametersSchema().load(parameters) + sensor = deserialized_parameters.get("sensor") + if isinstance(sensor, Sensor) and sensor.generic_asset_id != asset.id: + warnings.append( + f"The sensor to forecast ({sensor.id}) does not belong to asset {asset.id}." + ) + forecaster = get_data_generator( + source=source, + model=generator_class or "TrainPredictPipeline", + config=config or {}, + save_config=True, + data_generator_type=Forecaster, + ) + if forecaster is None: + raise ValueError(f"Could not set up forecaster '{generator_class}'.") + return forecaster, deserialized_parameters, warnings + + +def _prepare_report_automation( + parameters: dict, + cronstr: str, + generator_class: str | None, + config: dict | None, + source, +) -> tuple[Reporter, dict, list[str]]: + """Validate report automation parameters without creating a data source.""" + from marshmallow import ValidationError + + from flexmeasures.data.services.data_sources import get_data_generator + + warnings = [] + if generator_class is None and source is None: + raise ValidationError( + "A reporter is required for report automations (e.g. PandasReporter)." + ) + for offset_field in ("start-offset", "end-offset"): + if offset_field in parameters: + try: + validate_offset_chain(parameters[offset_field]) + except ValueError as e: + raise ValidationError(f"Invalid {offset_field}: {e}") + reporter = get_data_generator( + source=source, + model=generator_class, + config=config or {}, + save_config=True, + data_generator_type=Reporter, + ) + if reporter is None: + raise ValueError(f"Could not set up reporter '{generator_class}'.") + # Validate with the chosen reporter's own parameters schema, + # which may extend the base ReporterParametersSchema. + deserialized_parameters = reporter._parameters_schema.load( + prepare_report_parameters(parameters, cronstr) + ) + if ( + "start" in parameters or "end" in parameters + ) and "start-offset" not in parameters: + warnings.append( + "The report period is (partly) fixed, so each run may compute the same period." + " Use 'start-offset'/'end-offset' (Pandas offsets applied to the run time)," + " or omit timing fields to report on the period since the last run instead." + ) + return reporter, deserialized_parameters, warnings + + def create_automation( asset, name: str, @@ -611,7 +899,7 @@ def create_automation( automation_type: str = "forecasting", active: bool = True, parameters: dict | None = None, - forecaster_class: str = "TrainPredictPipeline", + generator_class: str | None = None, config: dict | None = None, source=None, origin: str = "API", @@ -619,7 +907,7 @@ def create_automation( ) -> tuple[Automation, list[str]]: """Create an automation (not committed yet), validating its parameters by type. - For forecasts, the forecaster config is stored on a data source. + For forecasts and reports, the data generator config is stored on a data source. An audit log record is added to the asset. :param check_permissions: whether to require that the current user may read the @@ -628,43 +916,28 @@ def create_automation( created by a user (through the API or the UI); the CLI runs without a user, and is trusted. :raises marshmallow.ValidationError: if the parameters are invalid. - :raises ValueError: if the forecaster cannot be set up. + :raises ValueError: if the data generator cannot be set up. :raises werkzeug.exceptions.Forbidden: if a sensor is not accessible to the user. :returns: the automation and a list of warnings. """ from marshmallow import ValidationError from flexmeasures.data.models.audit_log import AssetAuditLog - from flexmeasures.data.models.time_series import Sensor parameters = parameters or {} warnings: list[str] = [] generator_id = None - forecaster = None + data_generator = None input_sensors: list[Sensor] = [] output_sensors: list[Sensor] = [] - forecast_output_sensor: Sensor | None = None if automation_type == "forecasting": - from flexmeasures.data.schemas.forecasting.pipeline import ( - ForecasterParametersSchema, - ) - from flexmeasures.data.services.data_sources import get_data_generator - - deserialized_parameters = ForecasterParametersSchema().load(parameters) - sensor = deserialized_parameters.get("sensor") - if isinstance(sensor, Sensor) and sensor.generic_asset_id != asset.id: - warnings.append( - f"The sensor to forecast ({sensor.id}) does not belong to asset {asset.id}." + forecaster, deserialized_parameters, forecast_warnings = ( + _prepare_forecast_automation( + asset, parameters, generator_class, config, source ) - forecaster = get_data_generator( - source=source, - model=forecaster_class, - config=config or {}, - save_config=True, - data_generator_type=Forecaster, ) - if forecaster is None: - raise ValueError(f"Could not set up forecaster '{forecaster_class}'.") + warnings.extend(forecast_warnings) + data_generator = forecaster # A forecast reads the history of the sensor to forecast, plus its regressors, # and records the forecast on the sensor to save to (the same sensor by default). @@ -674,7 +947,6 @@ def create_automation( ) input_sensors = forecast_sensors["input_sensors"] output_sensors = forecast_sensors["output_sensors"] - forecast_output_sensor = output_sensors[0] if output_sensors else None elif automation_type == "scheduling": from flexmeasures.utils import flexmeasures_inflection from flexmeasures.data.schemas.scheduling import ( @@ -705,6 +977,17 @@ def create_automation( "The schedule 'start' is fixed, so each run will compute the same period." " Omit 'start' to schedule from the run time instead." ) + elif automation_type == "reporting": + reporter, deserialized_parameters, report_warnings = _prepare_report_automation( + parameters, cronstr, generator_class, config, source + ) + warnings.extend(report_warnings) + data_generator = reporter + report_sensors = resolve_data_generator_sensors( + reporter, deserialized_parameters + ) + input_sensors = report_sensors["input_sensors"] + output_sensors = report_sensors["output_sensors"] else: raise ValidationError( f"Automation type '{automation_type}' is not supported (supported types: {Automation.SUPPORTED_TYPES})." @@ -715,13 +998,14 @@ def create_automation( # Only once the sensors are known to be the user's to involve do we say anything about them, # so that this does not reveal where a sensor sits to someone who may not read it. - if forecast_output_sensor is not None: - validate_forecast_output_scope(asset.id, forecast_output_sensor) + if automation_type in ("forecasting", "reporting"): + for output_sensor in output_sensors: + validate_automation_output_scope(asset.id, output_sensor, automation_type) - if forecaster is not None: - # Look up or create the data source storing the forecaster config only now that the automation is going ahead, + if data_generator is not None: + # Look up or create the data source storing the generator config only now that the automation is going ahead, # so that a refused request leaves nothing behind, whatever the caller does with the session afterwards. - generator = forecaster.data_source + generator = data_generator.data_source db.session.flush() generator_id = generator.id elif automation_type == "scheduling": @@ -831,27 +1115,40 @@ def get_forecast_output_sensor(parameters: dict[str, Any]) -> Sensor: return sensor -def validate_forecast_output_scope(asset_id: int, output_sensor: Sensor) -> None: - """Require forecast output on the automation asset or a descendant.""" +def validate_automation_output_scope( + asset_id: int, output_sensor: Sensor, automation_type: str +) -> None: + """Require generated 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"{automation_type.capitalize()} 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: +def run_automation( + automation: Automation, scheduled_at: datetime | None = None +) -> dict[str, Any] | None: """Queue the jobs for one run of an automation. :returns: a dict like {"job_id": , "n_jobs": }. """ + now = server_now() if automation.type == "forecasting": - return _run_forecast_automation(automation) + returns = _run_forecast_automation(automation) elif automation.type == "scheduling": - return _run_schedule_automation(automation) - raise NotImplementedError( - f"Automations of type '{automation.type}' cannot be run yet." - ) + returns = _run_schedule_automation(automation) + elif automation.type == "reporting": + # NB the reporting job itself records the end of the report window upon + # success (see run_report_job), so failed jobs do not create gaps in the + # reported periods. + return _run_report_automation(automation, now=now, scheduled_at=scheduled_at) + else: + raise NotImplementedError( + f"Automations of type '{automation.type}' cannot be run yet." + ) + record_automation_run(automation.id, now=now) + return returns def _run_forecast_automation(automation: Automation) -> dict[str, Any] | None: @@ -867,13 +1164,51 @@ def _run_forecast_automation(automation: Automation) -> dict[str, Any] | None: 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) + validate_automation_output_scope( + automation.asset_id, output_sensor, automation.type + ) # 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)) +def _run_report_automation( + automation: Automation, + now: datetime | None = None, + scheduled_at: datetime | None = None, +) -> dict[str, Any] | None: + if automation.generator is None: + raise ValueError( + f"Automation {automation.id} has no data generator to run (generator_id is not set)." + ) + reporter = automation.generator.data_generator + if not isinstance(reporter, Reporter): + raise ValueError( + f"Data source {automation.generator_id} of automation {automation.id} does not store a Reporter." + ) + parameters = prepare_report_parameters( + dict(automation.parameters), + automation.cronstr, + now=now, + automation_id=automation.id, + cron_timezone=automation.timezone, + scheduled_at=scheduled_at, + ) + report_sensors = resolve_data_generator_sensors( + reporter, reporter._parameters_schema.load(parameters) + ) + for output_sensor in report_sensors["output_sensors"]: + validate_automation_output_scope( + automation.asset_id, output_sensor, automation.type + ) + # The data generator instance is cached on the data source, which may be shared + # by several automations, so wipe any parameter state from a previous run. + reporter._parameters = None + reporter.set_job_trigger("automation", automation_id=automation.id) + return reporter.compute(as_job=True, parameters=parameters) + + def _run_schedule_automation(automation: Automation) -> dict[str, Any]: from flexmeasures.data.schemas.scheduling import AssetTriggerSchema from flexmeasures.data.services.scheduling import ( diff --git a/flexmeasures/data/services/reporting.py b/flexmeasures/data/services/reporting.py index e2b4350c32..93521925d9 100644 --- a/flexmeasures/data/services/reporting.py +++ b/flexmeasures/data/services/reporting.py @@ -2,7 +2,7 @@ from __future__ import annotations -from datetime import timedelta +from datetime import datetime, timedelta from typing import TYPE_CHECKING from flask import current_app @@ -45,7 +45,11 @@ def create_reporting_job(reporter: "Reporter", queue: str = "reporting") -> Job: job = Job.create( run_report_job, - kwargs={"data_source_id": data_source_id, "parameters": parameters}, + kwargs={ + "data_source_id": data_source_id, + "parameters": parameters, + "automation_id": (reporter._job_trigger or {}).get("automation_id"), + }, connection=current_app.queues[queue].connection, ttl=int( current_app.config.get( @@ -83,8 +87,15 @@ def _count_persistable_values(data) -> int: return len(data.dropna(subset=["event_value"])) -def run_report_job(data_source_id: int, parameters: dict) -> list[dict]: - """Compute and store a report in a reporting worker.""" +def run_report_job( + data_source_id: int, parameters: dict, automation_id: int | None = None +) -> list[dict]: + """Compute and store a report in a reporting worker. + + If the report was triggered by an automation, the end of the report window is recorded upon success, + so the automation's next default window starts where this one ended. + A failed report job therefore leaves no permanent gap in the reported periods. + """ from flexmeasures.data.models.data_sources import DataSource from flexmeasures.data.models.reporting import Reporter @@ -116,4 +127,12 @@ def run_report_job(data_source_id: int, parameters: dict) -> list[dict]: source, summary, ) + + if automation_id is not None and parameters.get("end"): + from flexmeasures.data.services.automations import record_automation_run + + record_automation_run( + automation_id, now=datetime.fromisoformat(parameters["end"]) + ) + return saved diff --git a/flexmeasures/data/services/utils.py b/flexmeasures/data/services/utils.py index 62dc2066c9..e6b15ec435 100644 --- a/flexmeasures/data/services/utils.py +++ b/flexmeasures/data/services/utils.py @@ -286,6 +286,9 @@ def wrapper(*args, **kwargs): "force_new_job_creation", False ) + # provenance meta data (how the job got created) must not affect job identity + kwargs_for_hash.pop("trigger", None) + # creating a hash from args and kwargs_for_hash args_hash = f"{queue}:{func.__name__}:{hash_function_arguments(args, kwargs_for_hash)}" diff --git a/flexmeasures/data/tests/test_automations_fresh_db.py b/flexmeasures/data/tests/test_automations_fresh_db.py index 061a06f4ee..6038d25df9 100644 --- a/flexmeasures/data/tests/test_automations_fresh_db.py +++ b/flexmeasures/data/tests/test_automations_fresh_db.py @@ -88,6 +88,21 @@ def test_automation_requires_generator(fresh_db, automation_with_generator): fresh_db.session.commit() +def test_report_automation_requires_generator(fresh_db, automation_with_generator): + forecast_automation, _ = automation_with_generator + report_automation = Automation( + asset=forecast_automation.asset, + type="reporting", + name="generator-free report", + cronstr="0 1 * * *", + parameters={}, + ) + fresh_db.session.add(report_automation) + + with pytest.raises(IntegrityError): + fresh_db.session.commit() + + def test_schedule_automation_generator_describes_its_scheduler_and_config( fresh_db, automation_with_generator ): @@ -150,6 +165,11 @@ def test_run_schedule_automation( "automation_id": automation.id, } + # Trigger provenance must not affect job identity: the same schedule request + # from another origin deduplicates onto the same job through the job cache. + returns_2 = run_automation(automation) + assert returns_2["job_id"] == returns["job_id"] + @pytest.mark.parametrize("sequential", (False, True)) def test_run_minimal_schedule_automation_with_stored_flex_config( diff --git a/flexmeasures/ui/static/openapi-specs.json b/flexmeasures/ui/static/openapi-specs.json index 57cd8b0bb1..cf05c138fe 100644 --- a/flexmeasures/ui/static/openapi-specs.json +++ b/flexmeasures/ui/static/openapi-specs.json @@ -6737,7 +6737,8 @@ "default": "forecasting", "enum": [ "forecasting", - "scheduling" + "scheduling", + "reporting" ] }, "name": { @@ -6765,14 +6766,17 @@ "type": "object", "additionalProperties": {} }, - "forecaster": { - "type": "string", - "default": "TrainPredictPipeline", - "description": "Forecaster class (only used for type 'forecasting')." + "generator": { + "type": [ + "string", + "null" + ], + "default": null, + "description": "Data generator class, e.g. a forecaster (defaults to TrainPredictPipeline) or a reporter (required for type 'reporting', e.g. PandasReporter). Not used for type 'scheduling'." }, "config": { "type": "object", - "description": "Forecaster configuration (only used for type 'forecasting').", + "description": "Data generator configuration (only used for types 'forecasting' and 'reporting').", "additionalProperties": {} } }, diff --git a/flexmeasures/ui/templates/assets/asset_automations.html b/flexmeasures/ui/templates/assets/asset_automations.html index 3189d175d1..71f589d19c 100644 --- a/flexmeasures/ui/templates/assets/asset_automations.html +++ b/flexmeasures/ui/templates/assets/asset_automations.html @@ -15,7 +15,7 @@

Automations of {{ asset.name }} @@ -55,6 +55,7 @@

New automation for {{ asset.name }}
@@ -67,11 +68,22 @@
New automation for {{ asset.name }}
Choose the IANA timezone in which this recurrence should follow the local clock. The asset timezone is selected by default.
+
+ + +
+ Forecaster class (defaults to TrainPredictPipeline) or reporter class (required for type reporting, e.g. PandasReporter). Not used for scheduling. +
+
+
+ + +
- Forecast parameters (for type forecasting) or a schedule trigger message (for type scheduling). + Forecast parameters (for type forecasting), a schedule trigger message (for type scheduling), or report parameters (for type reporting).
@@ -135,7 +147,7 @@
Edit automation
@@ -149,6 +161,11 @@
Edit automation
+
+
+
+
+
@@ -356,7 +373,7 @@
Recently created jobs
method: "GET", success: function (res) { res.automations.forEach(automation => automationsById.set(automation.id, automation)); - for (const automationType of ["forecasting", "scheduling"]) { + for (const automationType of ["forecasting", "scheduling", "reporting"]) { makeAutomationsTable( automationType, res.automations.filter(automation => automation.type === automationType), @@ -364,7 +381,8 @@
Recently created jobs
} }, error: function (xhr) { - for (const automationType of ["forecasting", "scheduling"]) { + console.error("Error fetching automations:", xhr); + for (const automationType of ["forecasting", "scheduling", "reporting"]) { makeAutomationsTable(automationType, []); $(`#automationsTable-${automationType}`).hide(); } @@ -450,6 +468,16 @@
Recently created jobs
return; } } + let config = {}; + const configText = $("#automationConfig").val().trim(); + if (configText) { + try { + config = JSON.parse(configText); + } catch (e) { + $("#newAutomationErr").removeClass("d-none").text("The data generator config is not valid JSON."); + return; + } + } $.ajax({ url: `/api/v3_0/assets/${assetId}/automations`, method: "POST", @@ -460,6 +488,8 @@
Recently created jobs
cronstr: $("#automationCron").val(), timezone: $("#automationTimezone").val(), active: $("#automationActive").is(":checked"), + generator: $("#automationGenerator").val().trim() || null, + config: config, parameters: parameters, }), success: () => location.reload(), diff --git a/flexmeasures/ui/tests/test_asset_crud.py b/flexmeasures/ui/tests/test_asset_crud.py index a6e790964c..c8126ef25e 100644 --- a/flexmeasures/ui/tests/test_asset_crud.py +++ b/flexmeasures/ui/tests/test_asset_crud.py @@ -72,8 +72,10 @@ def test_asset_page(db, client, setup_assets, as_prosumer_user1, view): assert "Automations of".encode() in asset_page.data assert "Forecasts".encode() in asset_page.data assert "Schedules".encode() in asset_page.data + assert "Reports".encode() in asset_page.data assert b'id="automationsTable-forecasting"' in asset_page.data assert b'id="automationsTable-scheduling"' in asset_page.data + assert b'id="automationsTable-reporting"' in asset_page.data assert b"automation.type === automationType" in asset_page.data assert b"No ${automationType} automations" in asset_page.data assert b'id="automations_err"' in asset_page.data