From 618494043e1ab6a0fed61f8d7ddb7ec1a3468945 Mon Sep 17 00:00:00 2001 From: "F.N. Claessen" Date: Sat, 11 Jul 2026 20:37:11 +0200 Subject: [PATCH 01/15] feat: reports can run as background jobs - Activate the reporting queue (it was prepared but commented out), including worker help texts and queue cleanup. - Reporters accept as_job: a job is queued (with trigger meta data) that rebuilds the reporter from its data source, computes the report and saves the results to the database. - `flexmeasures add report --as-job` queues such a job; reporting jobs show up in the asset's jobs overview (status page and API). Part of FlexMeasures/flexmeasures#2288 Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Rbix8k1JfeUWNXEmHEZVpX --- documentation/host/queues.rst | 2 +- flexmeasures/app.py | 2 +- flexmeasures/cli/jobs.py | 12 ++- .../data/models/reporting/__init__.py | 11 +- flexmeasures/data/scripts/data_gen.py | 2 + flexmeasures/data/services/reporting.py | 101 ++++++++++++++++++ flexmeasures/data/services/sensors.py | 9 ++ 7 files changed, 131 insertions(+), 8 deletions(-) create mode 100644 flexmeasures/data/services/reporting.py diff --git a/documentation/host/queues.rst b/documentation/host/queues.rst index 5454061615..337afc75a0 100644 --- a/documentation/host/queues.rst +++ b/documentation/host/queues.rst @@ -24,7 +24,7 @@ Here is how to run one worker for each kind of job (in separate terminals): .. code-block:: bash - $ flexmeasures jobs run-worker --name our-only-worker --queue forecasting|scheduling|ingestion + $ flexmeasures jobs run-worker --name our-only-worker --queue forecasting|scheduling|ingestion|reporting Running multiple workers in parallel might be a great idea. diff --git a/flexmeasures/app.py b/flexmeasures/app.py index ee9fcab76a..4d47ea0085 100644 --- a/flexmeasures/app.py +++ b/flexmeasures/app.py @@ -110,7 +110,7 @@ def create( # noqa C901 forecasting=Queue(connection=redis_conn, name="forecasting"), scheduling=Queue(connection=redis_conn, name="scheduling"), ingestion=Queue(connection=redis_conn, name="ingestion"), - # reporting=Queue(connection=redis_conn, name="reporting"), + reporting=Queue(connection=redis_conn, name="reporting"), # labelling=Queue(connection=redis_conn, name="labelling"), # alerting=Queue(connection=redis_conn, name="alerting"), ) diff --git a/flexmeasures/cli/jobs.py b/flexmeasures/cli/jobs.py index ee31e8c620..3f119a5074 100644 --- a/flexmeasures/cli/jobs.py +++ b/flexmeasures/cli/jobs.py @@ -97,9 +97,11 @@ def run_automations(): try: returns = run_automation(automation) n_jobs = returns.get("n_jobs") if returns else 0 - queue_name = {"forecasts": "forecasting", "schedules": "scheduling"}.get( - automation.type, automation.type - ) + queue_name = { + "forecasts": "forecasting", + "schedules": "scheduling", + "reports": "reporting", + }.get(automation.type, automation.type) click.secho( f"Automation {automation.id} ('{automation.name}') queued {n_jobs} {queue_name} job(s) for asset {automation.asset_id}.", **MsgStyle.SUCCESS, @@ -420,7 +422,7 @@ def inspect_job(job_id: str): "--queue", default=None, required=True, - help="State which queue(s) to work on (using '|' as separator), e.g. 'forecasting', 'scheduling', 'ingestion' or 'forecasting|scheduling'.", + help="State which queue(s) to work on (using '|' as separator), e.g. 'forecasting', 'scheduling', 'ingestion', 'reporting' or 'forecasting|scheduling'.", ) @click.option( "--name", @@ -439,7 +441,7 @@ def inspect_job(job_id: str): ) def run_worker(queue: str, name: str | None, with_scheduler: bool): """ - Start a worker process for forecasting, scheduling and/or ingestion jobs. + Start a worker process for forecasting, scheduling, ingestion and/or reporting jobs. We use the app context to find out which redis queues to use. """ diff --git a/flexmeasures/data/models/reporting/__init__.py b/flexmeasures/data/models/reporting/__init__.py index 56ffb265fa..275a482250 100644 --- a/flexmeasures/data/models/reporting/__init__.py +++ b/flexmeasures/data/models/reporting/__init__.py @@ -19,14 +19,23 @@ class Reporter(DataGenerator): _parameters_schema = ReporterParametersSchema() _config_schema = ReporterConfigSchema() - def _compute(self, check_output_resolution=True, **kwargs) -> list[dict[str, Any]]: + def _compute( + self, check_output_resolution=True, as_job: bool = False, **kwargs + ) -> list[dict[str, Any]] | dict[str, Any]: """This method triggers the creation of a new report. The same object can generate multiple reports with different start, end, resolution and belief_time values. :param check_output_resolution: If True, checks each output for whether the event_resolution matches that of the sensor it is supposed to be recorded on. + :param as_job: If True, a job to compute (and save) the report is queued instead, + and a dict like {"job_id": , "n_jobs": 1} is returned. """ + if as_job: + from flexmeasures.data.services.reporting import create_reporting_job + + job = create_reporting_job(self) + return {"job_id": job.id, "n_jobs": 1} results = self._compute_report(**kwargs) diff --git a/flexmeasures/data/scripts/data_gen.py b/flexmeasures/data/scripts/data_gen.py index f2c1a7bde1..eb961efd33 100644 --- a/flexmeasures/data/scripts/data_gen.py +++ b/flexmeasures/data/scripts/data_gen.py @@ -233,6 +233,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)) @@ -245,6 +246,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/reporting.py b/flexmeasures/data/services/reporting.py new file mode 100644 index 0000000000..3a0fb1996d --- /dev/null +++ b/flexmeasures/data/services/reporting.py @@ -0,0 +1,101 @@ +""" +Logic for queueing and running reporting jobs. +""" + +from __future__ import annotations + +from datetime import timedelta +from typing import TYPE_CHECKING + +from flask import current_app +from rq.job import Job + +from flexmeasures.data import db +from flexmeasures.data.utils import save_to_db + +if TYPE_CHECKING: + from flexmeasures.data.models.reporting import Reporter + + +def create_reporting_job(reporter: "Reporter", queue: str = "reporting") -> Job: + """Queue a job that computes a report and saves the results to the database. + + The reporter's (loaded) parameters are re-serialized into the job, + and the reporter itself travels as its data source ID (which stores its config), + so the job is fully deserializable by the worker. + """ + # Ensure the data source ID is available in the database when the job runs. + reporter._data_source = db.session.merge(reporter.data_source) + db.session.flush() + data_source_id = reporter._data_source.id + db.session.commit() + + parameters = reporter._parameters_schema.dump(reporter._parameters) + output_sensor_ids = [ + output["sensor"] for output in parameters.get("output", []) or [] + ] + + # job metadata for tracking (datetimes as ISO strings, + # a workaround for https://github.com/Parallels/rq-dashboard/issues/510) + job_metadata = { + "data_source_info": {"id": data_source_id}, + "start": parameters.get("start"), + "end": parameters.get("end"), + "sensor_id": output_sensor_ids[0] if output_sensor_ids else None, + } + if reporter._job_trigger: + job_metadata["trigger"] = reporter._job_trigger + + job = Job.create( + run_report_job, + kwargs=dict(data_source_id=data_source_id, parameters=parameters), + connection=current_app.queues[queue].connection, + ttl=int( + current_app.config.get( + "FLEXMEASURES_JOB_TTL", timedelta(-1) + ).total_seconds() + ), + result_ttl=int( + current_app.config.get( + "FLEXMEASURES_PLANNING_TTL", timedelta(-1) + ).total_seconds() + ), # NB job.cleanup docs says a negative number of seconds means persisting forever + meta=job_metadata, + timeout=60 * 60, # 1 hour + ) + current_app.queues[queue].enqueue_job(job) + for sensor_id in output_sensor_ids: + current_app.job_cache.add( + sensor_id, + job_id=job.id, + queue=queue, + asset_or_sensor_type="sensor", + ) + return job + + +def run_report_job(data_source_id: int, parameters: dict) -> list[dict]: + """Compute a report (with the data generator stored on the given data source) + and save the results to the database. + + This function is meant to be run by a worker processing the reporting queue. + """ + from flexmeasures.data.models.data_sources import DataSource + from flexmeasures.data.models.reporting import Reporter + + source = db.session.get(DataSource, data_source_id) + if source is None: + raise ValueError(f"Data source {data_source_id} no longer exists.") + reporter = source.data_generator + if not isinstance(reporter, Reporter): + raise ValueError(f"Data source {data_source_id} does not store a Reporter.") + results = reporter.compute(parameters=parameters) + for result in results: + save_to_db(result["data"]) + db.session.commit() + + # return a light summary (the report data itself is stored in the database) + return [ + {"sensor_id": result["sensor"].id, "n_rows": len(result["data"])} + for result in results + ] diff --git a/flexmeasures/data/services/sensors.py b/flexmeasures/data/services/sensors.py index d3b2d58e3e..9b501d4f24 100644 --- a/flexmeasures/data/services/sensors.py +++ b/flexmeasures/data/services/sensors.py @@ -769,6 +769,15 @@ def build_asset_jobs_data( current_app.job_cache.get(sensor.id, "forecasting", "sensor"), ) ) + jobs.append( + ( + "reporting", + "sensor", + sensor.id, + sensor.name, + current_app.job_cache.get(sensor.id, "reporting", "sensor"), + ) + ) jobs_data = list() # Building the actual return list - we also unpack lists of jobs, each to its own entry, and we add error info From 79e1609706141ab85099e0bd35b6d25e84555def Mon Sep 17 00:00:00 2001 From: "F.N. Claessen" Date: Sat, 11 Jul 2026 20:37:48 +0200 Subject: [PATCH 02/15] feat: reports as automations Automations can now compute reports on a recurring basis: - `flexmeasures add automation --type reports --reporter ` stores the reporter config on a data source (steady across runs, so all report results attribute to the same source) and validates the report parameters. - The report window resolves freshly on each run: 'start-offset'/'end-offset' fields (comma-separated Pandas offsets, applied to the run time in the first output sensor's timezone) express a rolling window, and without any timing fields the window defaults to the last cron period (from the previous cron fire time until the run time). Absolute start/end still work, but draw a warning. - The API creation field 'forecaster' is generalized to 'generator' (also accepting reporter classes), and the UI's New automation modal gains data generator and config fields; the Reports tab is now enabled. Part of FlexMeasures/flexmeasures#2288 Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Rbix8k1JfeUWNXEmHEZVpX --- documentation/cli/change_log.rst | 3 +- documentation/cli/commands.rst | 2 +- documentation/features/forecasting.rst | 2 +- documentation/features/reporting.rst | 25 ++- flexmeasures/api/v3_0/assets.py | 2 +- flexmeasures/cli/data_add.py | 63 +++++- flexmeasures/cli/tests/test_automations.py | 161 ++++++++++++++++ flexmeasures/data/models/automations.py | 2 +- flexmeasures/data/schemas/automations.py | 13 +- flexmeasures/data/services/automations.py | 181 ++++++++++++++++-- flexmeasures/ui/static/openapi-specs.json | 16 +- .../templates/assets/asset_automations.html | 39 +++- 12 files changed, 459 insertions(+), 50 deletions(-) diff --git a/documentation/cli/change_log.rst b/documentation/cli/change_log.rst index d088c749ea..48d4b84154 100644 --- a/documentation/cli/change_log.rst +++ b/documentation/cli/change_log.rst @@ -9,7 +9,8 @@ since v1.0.0 | July XX, 2026 * 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, computing forecasts or schedules). +* Add ``flexmeasures add automation``, ``flexmeasures edit automation`` and ``flexmeasures delete automation`` to manage automations (recurring tasks on an asset, computing forecasts, schedules or reports). +* Add an ``--as-job`` flag to ``flexmeasures add report``, to queue a reporting job (processed by workers of the new ``reporting`` queue) instead of computing directly. * Add ``flexmeasures jobs run-automations`` to queue jobs for all automations that are due to run this minute (run this once per minute, e.g. via cron). since v0.33.0 | June 01, 2026 diff --git a/documentation/cli/commands.rst b/documentation/cli/commands.rst index f67e4d0643..4f658f304d 100644 --- a/documentation/cli/commands.rst +++ b/documentation/cli/commands.rst @@ -40,7 +40,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. +``flexmeasures add automation`` Add an automation: a recurring task (computing forecasts, schedules or reports) on an asset. ================================================= ======================================= diff --git a/documentation/features/forecasting.rst b/documentation/features/forecasting.rst index d492488158..6c24982b37 100644 --- a/documentation/features/forecasting.rst +++ b/documentation/features/forecasting.rst @@ -207,4 +207,4 @@ Automations defined on an asset can be viewed on the asset's *Automations* page Account admins and consultants can also create, (de)activate and delete automations right there on the page, or through the API (`[POST] /assets/(id)/automations`, `[PATCH] /assets/(id)/automations/(automation_id)` and `[DELETE] /assets/(id)/automations/(automation_id)`). -Schedules can be automated in the same way — see :ref:`automating_schedules`. +Schedules and reports can be automated in the same way — see :ref:`automating_schedules` and :ref:`automating_reports`. diff --git a/documentation/features/reporting.rst b/documentation/features/reporting.rst index 9576b035bd..2248a795c8 100644 --- a/documentation/features/reporting.rst +++ b/documentation/features/reporting.rst @@ -122,4 +122,27 @@ The input sensor stores the power/energy flow, and the output sensor will store Here, the ``ProfitOrLossReporter`` used as source (with Id 6) is the one we configured above. With the offsets, we control the timing ― we indicate that we want the new report to encompass the day of tomorrow (see Pandas offset strings). -The report sensor will now store all costs which we know will be made tomorrow by the schedule. \ No newline at end of file +The report sensor will now store all costs which we know will be made tomorrow by the schedule. + +.. _automating_reports: + +Automating reports +-------------------- + +Reports can be queued as background jobs (add ``--as-job`` to ``flexmeasures add report``, and let a worker process the ``reporting`` queue, see :ref:`redis-queue`), +and computed on a recurring basis by an *automation* defined on the asset (see :ref:`automating_forecasts` for the full introduction, including how to run automations). + +The reporter and its configuration are stored on a data source (steady across runs, so all report results attribute to the same source), +while the report parameters are stored on the automation itself and their timing is resolved freshly 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 run time, + 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 on the last cron period: from the previous cron fire time until the run time. +- 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 reports \ + --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 f76153d48e..8267248ab7 100644 --- a/flexmeasures/api/v3_0/assets.py +++ b/flexmeasures/api/v3_0/assets.py @@ -1379,7 +1379,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", ) diff --git a/flexmeasures/cli/data_add.py b/flexmeasures/cli/data_add.py index 9d4de8e340..96e7561150 100755 --- a/flexmeasures/cli/data_add.py +++ b/flexmeasures/cli/data_add.py @@ -1029,7 +1029,8 @@ def _assemble_forecaster_config_and_parameters( config = yaml.safe_load(config_file) 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 != (): config[field.data_key] = field_value if edit_config: @@ -1048,8 +1049,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 @@ -1251,19 +1252,27 @@ def add_forecast( # noqa: C901 help="Forecaster class registered in flexmeasures.data.models.forecasting or in an available flexmeasures plugin." " Use the command `flexmeasures show forecasters` to list all the available forecasters.", ) +@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 reports)." + " 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`.", + help="DataSource ID of the data generator (`Forecaster` or `Reporter`).", ) @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 data generator (forecaster or reporter).", ) @click.option( "--parameters", @@ -1271,7 +1280,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 forecasts, or a schedule trigger message for --type schedules.", + " forecast parameters for --type forecasts, a schedule trigger message for --type schedules," + " or report parameters for --type reports.", ) @make_cli_options_optional( "sensor" @@ -1285,13 +1295,14 @@ def add_automation( automation_type: str, inactive: bool = False, forecaster_class: str = "TrainPredictPipeline", + 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 @@ -1299,12 +1310,18 @@ def add_automation( --cron "0 6 * * *" --sensor 2092 --regressors 2093 flexmeasures add automation --asset 3 --name "Hourly schedules" --cron "0 * * * *" --type schedules --parameters trigger-message.yml + flexmeasures add automation --asset 3 --name "Daily self-consumption report" + --cron "0 1 * * *" --type reports --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`). """ config, parameters = _assemble_forecaster_config_and_parameters( @@ -1320,7 +1337,9 @@ def add_automation( automation_type=automation_type, active=not inactive, parameters=parameters, - forecaster_class=forecaster_class, + generator_class=( + reporter_class if automation_type == "reports" else forecaster_class + ), config=config, source=source, origin="CLI", @@ -1630,6 +1649,12 @@ def add_schedule( # noqa C901 is_flag=True, help="Add this flag to save the `config` in the attributes of the DataSource for future reference.", ) +@click.option( + "--as-job", + is_flag=True, + help="Whether to queue a reporting job instead of computing directly. " + "To process the job, run a worker (on any computer, but configured to the same databases) to process the 'reporting' queue. Defaults to False.", +) def add_report( # noqa: C901 reporter_class: str, source: DataSource | None = None, @@ -1646,11 +1671,19 @@ def add_report( # noqa: C901 edit_parameters: bool = False, save_config: bool = False, timezone: str | None = None, + as_job: bool = False, ): """ Create a new report using the Reporter class and save the results to the database or export them as CSV or Excel file. """ + if as_job and (dry_run or output_file_pattern): + click.secho( + "The --as-job flag cannot be combined with --dry-run or --output-file:" + " the job saves the report to the database only.", + **MsgStyle.ERROR, + ) + raise click.Abort() config = dict() @@ -1753,6 +1786,16 @@ def add_report( # noqa: C901 if ("resolution" not in parameters) and (resolution is not None): parameters["resolution"] = pd.Timedelta(resolution).isoformat() + reporter.set_job_trigger("CLI") + + if as_job: + returns = reporter.compute(as_job=True, parameters=parameters) + click.secho( + f"Created reporting job {returns['job_id']} (the report will be saved to the database once processed).", + **MsgStyle.SUCCESS, + ) + return + click.echo("Report computation is running...") # compute the report diff --git a/flexmeasures/cli/tests/test_automations.py b/flexmeasures/cli/tests/test_automations.py index 897bb43500..37395f044e 100644 --- a/flexmeasures/cli/tests/test_automations.py +++ b/flexmeasures/cli/tests/test_automations.py @@ -1,6 +1,7 @@ from datetime import timedelta import pytest +import yaml from sqlalchemy import select @@ -196,6 +197,166 @@ 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 + + # 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 _report_automation_cli_input( + tmp_path, sensor1_id, sensor2_id, report_sensor_id, parameters_extra=None +): + """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", "1", + "--name", "Aggregation report", + "--cron", "0 1 * * *", + "--type", "reports", + "--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 + 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"}, + ), + ) + assert "Successfully created" in result.output, result.output + automation = fresh_db.session.execute(select(Automation)).scalar_one() + assert automation.type == "reports" + 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", "reports", + ], + ) # fmt: skip + assert result.exit_code != 0 + assert "reporter is required" 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 + 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", + }, + ) + 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, + } + + # 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() + + 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. diff --git a/flexmeasures/data/models/automations.py b/flexmeasures/data/models/automations.py index e99e621165..1a318fccd1 100644 --- a/flexmeasures/data/models/automations.py +++ b/flexmeasures/data/models/automations.py @@ -22,7 +22,7 @@ class Automation(db.Model, AuthModelMixin): __tablename__ = "automation" - SUPPORTED_TYPES = ["forecasts", "schedules"] # later also "reports" + SUPPORTED_TYPES = ["forecasts", "schedules", "reports"] 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 e12a49a139..413ad85563 100644 --- a/flexmeasures/data/schemas/automations.py +++ b/flexmeasures/data/schemas/automations.py @@ -53,15 +53,20 @@ class AutomationCreationSchema(Schema): cronstr = CronField(required=True) active = fields.Bool(load_default=True) parameters = fields.Dict(keys=fields.Str(), load_default=dict) - forecaster = fields.Str( - load_default="TrainPredictPipeline", - metadata={"description": "Forecaster class (only used for type 'forecasts')."}, + generator = fields.Str( + load_default=None, + allow_none=True, + metadata={ + "description": "Data generator class, e.g. a forecaster (defaults to TrainPredictPipeline)" + " or a reporter (required for type 'reports', e.g. PandasReporter)." + " Not used for type 'schedules'." + }, ) config = fields.Dict( keys=fields.Str(), load_default=dict, metadata={ - "description": "Forecaster configuration (only used for type 'forecasts')." + "description": "Data generator configuration (only used for types 'forecasts' and 'reports')." }, ) diff --git a/flexmeasures/data/services/automations.py b/flexmeasures/data/services/automations.py index c74f20f7c9..1ddd613f88 100644 --- a/flexmeasures/data/services/automations.py +++ b/flexmeasures/data/services/automations.py @@ -11,12 +11,17 @@ from croniter import croniter import isodate import pandas as pd +import pytz from sqlalchemy import select -from flexmeasures import Forecaster +from flexmeasures import Forecaster, Reporter from flexmeasures.data import db from flexmeasures.data.models.automations import Automation -from flexmeasures.utils.time_utils import get_timezone, server_now +from flexmeasures.utils.time_utils import ( + apply_offset_chain, + get_timezone, + server_now, +) def describe_cronstr(cronstr: str) -> str: @@ -79,6 +84,81 @@ def prepare_schedule_trigger_message(parameters: dict, asset_id: int) -> dict: return message +def prepare_report_parameters( + parameters: dict, cronstr: str, now: 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 defaults to the last cron period: + from the previous cron fire time until the run time. + """ + message = dict(parameters) + if now is None: + now = server_now() + now = floor_to_minute(now) + + # 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 = now.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 last cron period: from the previous cron fire time until the run time + if start is None: + start = croniter(cronstr, now).get_prev(datetime) + 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 get_automation_job_stats(automation: Automation) -> dict[str, int]: """Count the jobs created by this automation, per job status. @@ -86,24 +166,31 @@ def get_automation_job_stats(automation: Automation) -> dict[str, int]: """ from flask import current_app - # 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 == "schedules": # Scheduling jobs are cached under the asset (multi-device wrap-up jobs) # and under individual sensors (per-device jobs). cache_refs = [(automation.asset_id, "scheduling", "asset")] + [ (sensor.id, "scheduling", "sensor") for sensor in automation.asset.sensors ] + elif automation.type == "reports": + 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] = {} @@ -126,24 +213,25 @@ def create_automation( automation_type: str = "forecasts", 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", ) -> 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. :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. :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 + from flexmeasures.data.services.data_sources import get_data_generator parameters = parameters or {} warnings: list[str] = [] @@ -152,7 +240,6 @@ def create_automation( 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") @@ -162,13 +249,13 @@ def create_automation( ) forecaster = get_data_generator( source=source, - model=forecaster_class, + 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 '{forecaster_class}'.") + raise ValueError(f"Could not set up forecaster '{generator_class}'.") generator = ( forecaster.data_source ) # looks up or creates the data source storing the forecaster config @@ -185,6 +272,40 @@ 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 == "reports": + from flexmeasures.data.schemas.reporting import ReporterParametersSchema + + if generator_class is None and source is None: + raise ValidationError( + "A reporter is required for report automations (e.g. PandasReporter)." + ) + try: + prepared_parameters = prepare_report_parameters(parameters, cronstr) + except ValueError as e: + raise ValidationError(f"Invalid time offsets: {e}") + ReporterParametersSchema().load(prepared_parameters) + 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 last cron period instead." + ) + 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}'.") + generator = ( + reporter.data_source + ) # looks up or creates the data source storing the reporter config + db.session.flush() + generator_id = generator.id else: raise ValidationError( f"Automation type '{automation_type}' is not supported (supported types: {Automation.SUPPORTED_TYPES})." @@ -260,6 +381,8 @@ def run_automation(automation: Automation) -> dict[str, Any] | None: return _run_forecast_automation(automation) elif automation.type == "schedules": return _run_schedule_automation(automation) + elif automation.type == "reports": + return _run_report_automation(automation) raise NotImplementedError( f"Automations of type '{automation.type}' cannot be run yet." ) @@ -282,6 +405,26 @@ def _run_forecast_automation(automation: Automation) -> dict[str, Any] | None: return forecaster.compute(as_job=True, parameters=dict(automation.parameters)) +def _run_report_automation(automation: Automation) -> 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." + ) + # 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) + parameters = prepare_report_parameters( + dict(automation.parameters), automation.cronstr + ) + 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/ui/static/openapi-specs.json b/flexmeasures/ui/static/openapi-specs.json index 65ce33c148..7cf3ac3669 100644 --- a/flexmeasures/ui/static/openapi-specs.json +++ b/flexmeasures/ui/static/openapi-specs.json @@ -5559,7 +5559,8 @@ "default": "forecasts", "enum": [ "forecasts", - "schedules" + "schedules", + "reports" ] }, "name": { @@ -5578,14 +5579,17 @@ "type": "object", "additionalProperties": {} }, - "forecaster": { - "type": "string", - "default": "TrainPredictPipeline", - "description": "Forecaster class (only used for type 'forecasts')." + "generator": { + "type": [ + "string", + "null" + ], + "default": null, + "description": "Data generator class, e.g. a forecaster (defaults to TrainPredictPipeline) or a reporter (required for type 'reports', e.g. PandasReporter). Not used for type 'schedules'." }, "config": { "type": "object", - "description": "Forecaster configuration (only used for type 'forecasts').", + "description": "Data generator configuration (only used for types 'forecasts' and 'reports').", "additionalProperties": {} } }, diff --git a/flexmeasures/ui/templates/assets/asset_automations.html b/flexmeasures/ui/templates/assets/asset_automations.html index af8fe21277..62ba6d73f7 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 }} @@ -48,6 +48,7 @@

New automation for {{ asset.name }}
@@ -55,11 +56,22 @@
New automation for {{ asset.name }}
In the platform timezone, e.g. "0 6 * * *" for daily at 6:00.
+
+ + +
+ Forecaster class (defaults to TrainPredictPipeline) or reporter class (required for type reports, e.g. PandasReporter). Not used for schedules. +
+
+
+ + +
- Forecast parameters (for type forecasts) or a schedule trigger message (for type schedules). + Forecast parameters (for type forecasts), a schedule trigger message (for type schedules), or report parameters (for type reports).
@@ -85,7 +97,7 @@
New automation for {{ asset.name }}
@@ -99,6 +111,11 @@
New automation for {{ asset.name }}
+
+
+
+
+
@@ -225,7 +242,7 @@
Recently created jobs
url: `/api/v3_0/assets/${assetId}/automations`, method: "GET", success: function (res) { - for (const automationType of ["forecasts", "schedules"]) { + for (const automationType of ["forecasts", "schedules", "reports"]) { makeAutomationsTable( automationType, res.automations.filter(automation => automation.type === automationType), @@ -234,7 +251,7 @@
Recently created jobs
}, error: function (xhr) { console.error("Error fetching automations:", xhr); - for (const automationType of ["forecasts", "schedules"]) { + for (const automationType of ["forecasts", "schedules", "reports"]) { makeAutomationsTable(automationType, []); } }, @@ -280,6 +297,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", @@ -289,6 +316,8 @@
Recently created jobs
type: $("#automationType").val(), cronstr: $("#automationCron").val(), active: $("#automationActive").is(":checked"), + generator: $("#automationGenerator").val().trim() || null, + config: config, parameters: parameters, }), success: () => location.reload(), From 5eac46854a204ecbfc19d5a73e35ca486f86cbf4 Mon Sep 17 00:00:00 2001 From: "F.N. Claessen" Date: Sat, 11 Jul 2026 20:39:08 +0200 Subject: [PATCH 03/15] docs: changelog entry for reports as jobs and automations Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Rbix8k1JfeUWNXEmHEZVpX --- documentation/changelog.rst | 1 + 1 file changed, 1 insertion(+) diff --git a/documentation/changelog.rst b/documentation/changelog.rst index d9810f6054..cfa087d796 100644 --- a/documentation/changelog.rst +++ b/documentation/changelog.rst @@ -15,6 +15,7 @@ New features * Automations - first roundtrip for forecasts: recurring tasks defined per asset, managed with new CLI commands (``flexmeasures add|edit|delete automation``), run by ``flexmeasures jobs run-automations``, and viewable in a new UI page and API endpoints (``[GET] /assets/(id)/automations``); jobs now also record whether they were created via the CLI, the API or an automation [see `PR #2290 `_] * Automations can also compute schedules on a recurring basis (``flexmeasures add automation --type schedules``), with the schedule start defaulting to each run's time [see `PR #2293 `_] * Automations can be created, edited and deleted in the UI and through new API endpoints (``[POST|PATCH|DELETE] /assets/(id)/automations``), by account admins and consultants [see `PR #2294 `_] +* Reports can run as background jobs (``flexmeasures add report --as-job``, processed by workers of the new ``reporting`` queue) and be computed on a recurring basis by automations, with a rolling report window expressed as Pandas offsets or defaulting to the last cron period [see `PR #2297 `_] * Breaking behaviour change: the top-level flex-context's ``relax-constraints`` field now defaults to ``True`` (matching the default already used within each ``commodities`` entry), so constraint violations are softly penalized by default instead of being hard constraints, unless explicitly set to ``False`` [see `PR #2172 `_] * In the UI, asset and sensor lists can be filtered by ID prefix through API-backed search fields [see `PR #2231 `_] * Support configurable lower and upper bounds and snapping for forecast post-processing [see `PR #2273 `_] From 791fcf43e225285f7c6b59d72ec8ec422eab5587 Mon Sep 17 00:00:00 2001 From: "F.N. Claessen" Date: Sat, 11 Jul 2026 21:31:07 +0200 Subject: [PATCH 04/15] feat: anchor default report windows to the automation's actual last run Each automation run is recorded in Redis; a report automation without timing fields then reports on the period since its actual last run, falling back to the last cron period when no last run is known (e.g. on the first run, or after a Redis flush). This gives gapless coverage even when runs are missed. Part of FlexMeasures/flexmeasures#2288 Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Rbix8k1JfeUWNXEmHEZVpX --- documentation/features/reporting.rst | 3 +- flexmeasures/cli/tests/test_automations.py | 21 +++++++ flexmeasures/data/services/automations.py | 70 ++++++++++++++++++---- 3 files changed, 81 insertions(+), 13 deletions(-) diff --git a/documentation/features/reporting.rst b/documentation/features/reporting.rst index 2248a795c8..c1fe7e6d78 100644 --- a/documentation/features/reporting.rst +++ b/documentation/features/reporting.rst @@ -137,7 +137,8 @@ while the report parameters are stored on the automation itself and their timing - Use ``start-offset`` and/or ``end-offset`` fields (comma-separated Pandas offsets, like the CLI options above) for a rolling window relative to the run time, 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 on the last cron period: from the previous cron fire time until the run time. +- Omit timing fields entirely to report on the period since the automation's actual last run + (falling back to the last cron period — from the previous cron fire time until the run time — when no last run is known, e.g. on the first run). - 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: diff --git a/flexmeasures/cli/tests/test_automations.py b/flexmeasures/cli/tests/test_automations.py index 37395f044e..83f7988890 100644 --- a/flexmeasures/cli/tests/test_automations.py +++ b/flexmeasures/cli/tests/test_automations.py @@ -213,6 +213,24 @@ def test_prepare_report_parameters(app): assert pd.Timestamp(message["start"]) == now - pd.Timedelta(hours=1) assert pd.Timestamp(message["end"]) == now + # 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 @@ -396,6 +414,9 @@ def test_run_automations(app, fresh_db, setup_dummy_data, clean_redis): 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) diff --git a/flexmeasures/data/services/automations.py b/flexmeasures/data/services/automations.py index 1ddd613f88..13eb4d6226 100644 --- a/flexmeasures/data/services/automations.py +++ b/flexmeasures/data/services/automations.py @@ -84,8 +84,44 @@ def prepare_schedule_trigger_message(parameters: dict, asset_id: int) -> dict: return message +def _last_run_redis_key(automation_id: int) -> str: + return f"automation-last-run:{automation_id}" + + +def record_automation_run(automation: Automation, now: datetime | None = None): + """Remember (in Redis) when this automation last ran. + + This is used to anchor default report windows to the actual last run. + """ + from flask import current_app + + if now is None: + now = server_now() + current_app.redis_connection.set( + _last_run_redis_key(automation.id), floor_to_minute(now).isoformat() + ) + + +def get_automation_last_run(automation_id: int) -> datetime | None: + """When this automation last ran, 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 + parameters: dict, + cronstr: str, + now: datetime | None = None, + automation_id: int | None = None, ) -> dict: """Complete stored report parameters into a message for the ReporterParametersSchema. @@ -94,8 +130,9 @@ def prepare_report_parameters( - "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 defaults to the last cron period: - from the previous cron fire time until the run time. + - Without offsets or absolutes, the window runs since the automation's actual + last run, falling back to the last cron period (from the previous cron fire + time until the run time) when no last run is known (e.g. on the first run). """ message = dict(parameters) if now is None: @@ -136,9 +173,15 @@ def prepare_report_parameters( end if end is not None else pd.Timestamp(now), end_offset ) - # Default to the last cron period: from the previous cron fire time until the run time + # Default to the window since the actual last run, falling back to the last + # cron period (from the previous cron fire time until the run time) if start is None: - start = croniter(cronstr, now).get_prev(datetime) + last_run = ( + get_automation_last_run(automation_id) + if automation_id is not None + else None + ) + start = last_run or croniter(cronstr, now).get_prev(datetime) if end is None: end = now @@ -378,14 +421,17 @@ def run_automation(automation: Automation) -> dict[str, Any] | None: :returns: a dict like {"job_id": , "n_jobs": }. """ if automation.type == "forecasts": - return _run_forecast_automation(automation) + returns = _run_forecast_automation(automation) elif automation.type == "schedules": - return _run_schedule_automation(automation) + returns = _run_schedule_automation(automation) elif automation.type == "reports": - return _run_report_automation(automation) - raise NotImplementedError( - f"Automations of type '{automation.type}' cannot be run yet." - ) + returns = _run_report_automation(automation) + else: + raise NotImplementedError( + f"Automations of type '{automation.type}' cannot be run yet." + ) + record_automation_run(automation) + return returns def _run_forecast_automation(automation: Automation) -> dict[str, Any] | None: @@ -420,7 +466,7 @@ def _run_report_automation(automation: Automation) -> dict[str, Any] | None: reporter._parameters = None reporter.set_job_trigger("automation", automation_id=automation.id) parameters = prepare_report_parameters( - dict(automation.parameters), automation.cronstr + dict(automation.parameters), automation.cronstr, automation_id=automation.id ) return reporter.compute(as_job=True, parameters=parameters) From 0dca8fcd5b4d0c9890aac9ce23c70a67ca257e12 Mon Sep 17 00:00:00 2001 From: "F.N. Claessen" Date: Sat, 11 Jul 2026 21:33:17 +0200 Subject: [PATCH 05/15] docs: add an Automations concept page Consolidates the shared automations concept (model, lifecycle, runner deployment, provenance) into documentation/features/automations.rst, with the per-feature pages linking to it and keeping only their type-specific parameter semantics. Part of FlexMeasures/flexmeasures#2288 Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Rbix8k1JfeUWNXEmHEZVpX --- documentation/features/automations.rst | 56 ++++++++++++++++++++++++++ documentation/features/forecasting.rst | 2 +- documentation/features/reporting.rst | 2 +- documentation/features/scheduling.rst | 2 +- documentation/index.rst | 1 + 5 files changed, 60 insertions(+), 3 deletions(-) create mode 100644 documentation/features/automations.rst diff --git a/documentation/features/automations.rst b/documentation/features/automations.rst new file mode 100644 index 0000000000..ad113c8bbb --- /dev/null +++ b/documentation/features/automations.rst @@ -0,0 +1,56 @@ +.. _automations: + +Automations +============ + +Hosts and users often want the three main FlexMeasures features — :ref:`forecasting`, :ref:`scheduling` and :ref:`reporting` — to run on a recurring basis, across larger numbers of sites. +*Automations* make that a first-class concept: an automation is a recurring task defined on an asset, and each time it runs, it queues jobs. + +An automation consists of: + +- a **type**: ``forecasts``, ``schedules`` or ``reports``; +- a **recurrence**: a cron string (e.g. ``"0 6 * * *"`` for daily at 6 AM), interpreted in the ``FLEXMEASURES_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. + +Managing automations +-------------------- + +Automations can be managed in three ways: + +- **CLI**: ``flexmeasures add automation``, ``flexmeasures edit automation`` (name, cron string, activation status) and ``flexmeasures delete automation``. +- **API**: list and inspect with ``[GET] /assets/(id)/automations`` and ``[GET] /assets/(id)/automations/(automation_id)``; + create, update and delete with ``[POST|PATCH|DELETE]`` on the same paths (see the `API documentation <../api/v3_0.html>`_). +- **UI**: each asset has an *Automations* page (in the breadcrumbs dropdown), with a tab per automation type. + It lists each automation's recurrence and recent job counts, and lets you create, (de)activate and delete automations. + +Creating, updating and deleting automations requires account admin or consultant rights, and is recorded in the asset's audit log. + +Running automations +-------------------- + +An automation is due whenever its cron string matches the current minute. To actually run due automations, 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 — so make sure workers are processing the relevant queues (``forecasting``, ``scheduling`` and/or ``reporting``, see :ref:`redis-queue`). +A Redis-based guard prevents queueing jobs twice if the command happens to run more than once within the same minute. + +Jobs record how they were created (via the CLI, the API or an automation), which is shown in the *Created Via* column +of the jobs table on the asset's status page, where recent jobs are listed. + +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 automation's actual last run. diff --git a/documentation/features/forecasting.rst b/documentation/features/forecasting.rst index 6c24982b37..c1111b645f 100644 --- a/documentation/features/forecasting.rst +++ b/documentation/features/forecasting.rst @@ -178,7 +178,7 @@ In `this weather forecast plugin `_ API endpoint (without the asset id). Omit the ``start`` field to schedule from the run time on each run (floored to the ``resolution`` field, if given). As usual, the flex-context and flex-model can also (partly) live on the asset itself, in which case a minimal trigger message suffices. diff --git a/documentation/index.rst b/documentation/index.rst index b2c91d487e..37c0bb0a0c 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 From 176d800cc3d17191905bd3c702d08055b67cb06b Mon Sep 17 00:00:00 2001 From: "F.N. Claessen" Date: Sat, 11 Jul 2026 22:20:29 +0200 Subject: [PATCH 06/15] fix: address stack code review findings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - API automation creation now checks that the caller may read every sensor referenced in the parameters/config and record data on the sensors the automation writes to, closing a cross-account data read/write hole. - `add report --as-job` implies --save-config (the worker rebuilds the reporter from its data source, so jobs without stored config always crashed). - Report automation parameters are validated with the chosen reporter's own parameters schema, not the base schema. - Invalid start-offset/end-offset strings are rejected at creation instead of being silently skipped at run time (which yielded empty report windows). - run_report_job wipes the shared cached reporter's parameter state, like the automation runner already did, so consecutive jobs in one worker process don't pollute each other. - Default report windows now anchor to the end of the last *successfully* covered window, recorded by the reporting job upon success — failed jobs no longer create permanent reporting gaps, and the enqueue-time minute-rollover gap is gone (the recorded anchor is the window end itself). - The cron-period fallback window is computed in the platform timezone, matching how the runner decides when automations fire. - Job stats for schedules automations also scan flex-model device sensors (which may belong to child assets), so failed per-device jobs show up. - The trigger provenance kwarg is excluded from the job cache hash, so identical schedule requests from different origins dedupe again. Part of FlexMeasures/flexmeasures#2288 Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Rbix8k1JfeUWNXEmHEZVpX --- flexmeasures/api/v3_0/assets.py | 91 +++++++ .../api/v3_0/tests/test_automations_api.py | 50 ++++ flexmeasures/cli/data_add.py | 7 + flexmeasures/cli/tests/test_automations.py | 28 ++ flexmeasures/data/services/automations.py | 253 ++++++++++++------ flexmeasures/data/services/reporting.py | 24 +- flexmeasures/data/services/utils.py | 3 + flexmeasures/data/tests/test_automations.py | 5 + 8 files changed, 371 insertions(+), 90 deletions(-) diff --git a/flexmeasures/api/v3_0/assets.py b/flexmeasures/api/v3_0/assets.py index 8267248ab7..d06b8c1784 100644 --- a/flexmeasures/api/v3_0/assets.py +++ b/flexmeasures/api/v3_0/assets.py @@ -105,6 +105,90 @@ sensors_schema = SensorSchema(many=True) +def _forecast_automation_sensor_refs( + parameters: dict, config: dict +) -> tuple[list, list]: + read_refs = [parameters.get("sensor")] + for key in ("regressors", "future-regressors", "past-regressors"): + read_refs.extend(config.get(key) or []) + write_refs = [parameters.get("sensor-to-save") or parameters.get("sensor")] + return read_refs, write_refs + + +def _report_automation_sensor_refs(parameters: dict) -> tuple[list, list]: + read_refs = [ + entry.get("sensor") + for entry in parameters.get("input") or [] + if isinstance(entry, dict) + ] + write_refs = [ + entry.get("sensor") + for entry in parameters.get("output") or [] + if isinstance(entry, dict) + ] + return read_refs, write_refs + + +def _schedule_automation_sensor_refs(parameters: dict) -> tuple[list, list]: + read_refs: list = [] + write_refs: list = [] + for entry in parameters.get("flex-model") or []: + if isinstance(entry, dict): + write_refs.append(entry.get("sensor")) + if isinstance(entry.get("state-of-charge"), dict): + read_refs.append(entry["state-of-charge"].get("sensor")) + flex_context = parameters.get("flex-context") or {} + if isinstance(flex_context, dict): + for value in flex_context.values(): + if isinstance(value, dict): + read_refs.append(value.get("sensor")) + elif isinstance(value, list): + for item in value: + read_refs.append( + item.get("sensor") if isinstance(item, dict) else item + ) + return read_refs, write_refs + + +def _check_automation_sensor_access( + automation_type: str, parameters: dict, config: dict +): + """Check that the current user may read every sensor referenced in an + automation's parameters/config, and record data on the sensors the + automation would write to. + + Sensor references that do not resolve are skipped here; schema validation + of the parameters rejects them later. + + :raises Forbidden: if any check fails (answered with 403). + """ + parameters = parameters or {} + config = config or {} + if automation_type == "forecasts": + read_refs, write_refs = _forecast_automation_sensor_refs(parameters, config) + elif automation_type == "reports": + read_refs, write_refs = _report_automation_sensor_refs(parameters) + elif automation_type == "schedules": + read_refs, write_refs = _schedule_automation_sensor_refs(parameters) + else: + read_refs, write_refs = [], [] + + def resolve(value) -> Sensor | None: + try: + return db.session.get(Sensor, int(value)) + except (TypeError, ValueError): + return None + + for value in read_refs: + sensor = resolve(value) + if sensor is not None: + check_access(sensor, "read") + for value in write_refs: + sensor = resolve(value) + if sensor is not None: + check_access(sensor, "create-children") + + def sensor_term_filter(term: str): filters = [Sensor.name.ilike(f"%{term}%")] if term.isdecimal(): @@ -1371,6 +1455,13 @@ def post_automation(self, id: int, asset: GenericAsset): automation_data = AutomationCreationSchema().load(body) except ValidationError as e: return unprocessable_entity(e.messages) + # Guard against referencing sensors outside the caller's reach + # (raises Forbidden, answered with 403) + _check_automation_sensor_access( + automation_data["type"], + automation_data["parameters"], + automation_data["config"], + ) try: automation, warnings = create_automation( asset=asset, diff --git a/flexmeasures/api/v3_0/tests/test_automations_api.py b/flexmeasures/api/v3_0/tests/test_automations_api.py index f1481c4a64..3c1ea8ce52 100644 --- a/flexmeasures/api/v3_0/tests/test_automations_api.py +++ b/flexmeasures/api/v3_0/tests/test_automations_api.py @@ -4,6 +4,7 @@ import pytest from flask import url_for +from sqlalchemy import select from flexmeasures.data.models.automations import Automation @@ -192,6 +193,55 @@ def test_post_automation( 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": "forecasts", + "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 ) diff --git a/flexmeasures/cli/data_add.py b/flexmeasures/cli/data_add.py index 96e7561150..dbd964419d 100755 --- a/flexmeasures/cli/data_add.py +++ b/flexmeasures/cli/data_add.py @@ -1684,6 +1684,13 @@ def add_report( # noqa: C901 **MsgStyle.ERROR, ) raise click.Abort() + if as_job and not save_config: + # the worker rebuilds the reporter from its data source, so the config must be stored there + click.secho( + "Saving the reporter config to its data source (required for --as-job).", + **MsgStyle.WARN, + ) + save_config = True config = dict() diff --git a/flexmeasures/cli/tests/test_automations.py b/flexmeasures/cli/tests/test_automations.py index 83f7988890..005cb8fc5e 100644 --- a/flexmeasures/cli/tests/test_automations.py +++ b/flexmeasures/cli/tests/test_automations.py @@ -327,6 +327,22 @@ def test_add_report_automation(app, fresh_db, setup_dummy_data, tmp_path): 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.""" @@ -365,6 +381,9 @@ def test_run_report_automation(app, fresh_db, setup_dummy_data, clean_redis, tmp "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) @@ -374,6 +393,15 @@ def test_run_report_automation(app, fresh_db, setup_dummy_data, clean_redis, tmp ) 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): """Active automations due this minute queue forecasting jobs (with trigger meta data); inactive ones do not. diff --git a/flexmeasures/data/services/automations.py b/flexmeasures/data/services/automations.py index 13eb4d6226..59b99e60c1 100644 --- a/flexmeasures/data/services/automations.py +++ b/flexmeasures/data/services/automations.py @@ -84,26 +84,48 @@ 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: Automation, now: datetime | None = None): - """Remember (in Redis) when this automation last ran. +def record_automation_run(automation_id: int, now: datetime | None = None): + """Remember (in Redis) until when this automation's work is covered. - This is used to anchor default report windows to the actual last run. + 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 flask import current_app if now is None: now = server_now() current_app.redis_connection.set( - _last_run_redis_key(automation.id), floor_to_minute(now).isoformat() + _last_run_redis_key(automation_id), floor_to_minute(now).isoformat() ) def get_automation_last_run(automation_id: int) -> datetime | None: - """When this automation last ran, if known (the record lives in Redis).""" + """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)) @@ -130,14 +152,16 @@ def prepare_report_parameters( - "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 automation's actual - last run, falling back to the last cron period (from the previous cron fire - time until the run time) when no last run is known (e.g. on the first run). + - 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 now is None: now = server_now() - now = floor_to_minute(now) + # Cron strings are interpreted in the platform timezone (like the runner does) + platform_now = floor_to_minute(now) # Compute the run time in the timezone local to the first output sensor # (matching `flexmeasures add report`), falling back to the platform timezone. @@ -156,7 +180,7 @@ def prepare_report_parameters( output_sensor = None if output_sensor is not None: tz = pytz.timezone(output_sensor.timezone) - now = now.astimezone(tz) + now = platform_now.astimezone(tz) start_offset = message.pop("start-offset", None) end_offset = message.pop("end-offset", None) @@ -173,15 +197,17 @@ def prepare_report_parameters( end if end is not None else pd.Timestamp(now), end_offset ) - # Default to the window since the actual last run, falling back to the last - # cron period (from the previous cron fire time until the run time) + # 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 ) - start = last_run or croniter(cronstr, now).get_prev(datetime) + # NB the cron fallback uses the platform timezone, matching how the runner + # decides when the automation fires + start = last_run or croniter(cronstr, platform_now).get_prev(datetime) if end is None: end = now @@ -215,9 +241,18 @@ def get_automation_job_stats(automation: Automation) -> dict[str, int]: parameters = automation.parameters or {} if automation.type == "schedules": # Scheduling jobs are cached under the asset (multi-device wrap-up jobs) - # and under individual sensors (per-device jobs). + # 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 sensor in automation.asset.sensors + (sensor_id, "scheduling", "sensor") for sensor_id in sensor_ids ] elif automation.type == "reports": sensor_ids = _relevant_sensor_ids( @@ -249,6 +284,103 @@ 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[int, list[str]]: + """Validate forecast automation parameters and set up the forecaster's 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}'.") + generator = ( + forecaster.data_source + ) # looks up or creates the data source storing the forecaster config + db.session.flush() + return generator.id, warnings + + +def _prepare_schedule_automation(asset, parameters: dict) -> tuple[None, list[str]]: + """Validate schedule automation parameters (the scheduler's data source is resolved at job time).""" + from flexmeasures.data.schemas.scheduling import AssetTriggerSchema + + warnings = [] + AssetTriggerSchema().load(prepare_schedule_trigger_message(parameters, asset.id)) + if "start" in parameters: + warnings.append( + "The schedule 'start' is fixed, so each run will compute the same period." + " Omit 'start' to schedule from the run time instead." + ) + return None, warnings + + +def _prepare_report_automation( + parameters: dict, + cronstr: str, + generator_class: str | None, + config: dict | None, + source, +) -> tuple[int, list[str]]: + """Validate report automation parameters and set up the reporter's 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. + 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." + ) + generator = ( + reporter.data_source + ) # looks up or creates the data source storing the reporter config + db.session.flush() + return generator.id, warnings + + def create_automation( asset, name: str, @@ -273,82 +405,18 @@ def create_automation( from marshmallow import ValidationError from flexmeasures.data.models.audit_log import AssetAuditLog - from flexmeasures.data.models.time_series import Sensor - from flexmeasures.data.services.data_sources import get_data_generator parameters = parameters or {} - warnings: list[str] = [] - generator_id = None if automation_type == "forecasts": - from flexmeasures.data.schemas.forecasting.pipeline import ( - ForecasterParametersSchema, + generator_id, warnings = _prepare_forecast_automation( + asset, parameters, generator_class, config, source ) - - 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}'.") - generator = ( - forecaster.data_source - ) # looks up or creates the data source storing the forecaster config - db.session.flush() - generator_id = generator.id elif automation_type == "schedules": - from flexmeasures.data.schemas.scheduling import AssetTriggerSchema - - AssetTriggerSchema().load( - prepare_schedule_trigger_message(parameters, asset.id) - ) - if "start" in parameters: - warnings.append( - "The schedule 'start' is fixed, so each run will compute the same period." - " Omit 'start' to schedule from the run time instead." - ) + generator_id, warnings = _prepare_schedule_automation(asset, parameters) elif automation_type == "reports": - from flexmeasures.data.schemas.reporting import ReporterParametersSchema - - if generator_class is None and source is None: - raise ValidationError( - "A reporter is required for report automations (e.g. PandasReporter)." - ) - try: - prepared_parameters = prepare_report_parameters(parameters, cronstr) - except ValueError as e: - raise ValidationError(f"Invalid time offsets: {e}") - ReporterParametersSchema().load(prepared_parameters) - 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 last cron period instead." - ) - reporter = get_data_generator( - source=source, - model=generator_class, - config=config or {}, - save_config=True, - data_generator_type=Reporter, + generator_id, warnings = _prepare_report_automation( + parameters, cronstr, generator_class, config, source ) - if reporter is None: - raise ValueError(f"Could not set up reporter '{generator_class}'.") - generator = ( - reporter.data_source - ) # looks up or creates the data source storing the reporter config - db.session.flush() - generator_id = generator.id else: raise ValidationError( f"Automation type '{automation_type}' is not supported (supported types: {Automation.SUPPORTED_TYPES})." @@ -420,17 +488,21 @@ def run_automation(automation: Automation) -> dict[str, Any] | None: :returns: a dict like {"job_id": , "n_jobs": }. """ + now = server_now() if automation.type == "forecasts": returns = _run_forecast_automation(automation) elif automation.type == "schedules": returns = _run_schedule_automation(automation) elif automation.type == "reports": - returns = _run_report_automation(automation) + # 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) else: raise NotImplementedError( f"Automations of type '{automation.type}' cannot be run yet." ) - record_automation_run(automation) + record_automation_run(automation.id, now=now) return returns @@ -451,7 +523,9 @@ def _run_forecast_automation(automation: Automation) -> dict[str, Any] | None: return forecaster.compute(as_job=True, parameters=dict(automation.parameters)) -def _run_report_automation(automation: Automation) -> dict[str, Any] | None: +def _run_report_automation( + automation: Automation, now: 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)." @@ -466,7 +540,10 @@ def _run_report_automation(automation: Automation) -> dict[str, Any] | None: reporter._parameters = None reporter.set_job_trigger("automation", automation_id=automation.id) parameters = prepare_report_parameters( - dict(automation.parameters), automation.cronstr, automation_id=automation.id + dict(automation.parameters), + automation.cronstr, + now=now, + automation_id=automation.id, ) return reporter.compute(as_job=True, parameters=parameters) diff --git a/flexmeasures/data/services/reporting.py b/flexmeasures/data/services/reporting.py index 3a0fb1996d..b784136ede 100644 --- a/flexmeasures/data/services/reporting.py +++ b/flexmeasures/data/services/reporting.py @@ -48,7 +48,11 @@ def create_reporting_job(reporter: "Reporter", queue: str = "reporting") -> Job: job = Job.create( run_report_job, - kwargs=dict(data_source_id=data_source_id, parameters=parameters), + kwargs=dict( + 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( @@ -74,11 +78,15 @@ def create_reporting_job(reporter: "Reporter", queue: str = "reporting") -> Job: return job -def run_report_job(data_source_id: int, parameters: dict) -> list[dict]: +def run_report_job( + data_source_id: int, parameters: dict, automation_id: int | None = None +) -> list[dict]: """Compute a report (with the data generator stored on the given data source) and save the results to the database. This function is meant to be run by a worker processing the reporting queue. + 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 there. """ from flexmeasures.data.models.data_sources import DataSource from flexmeasures.data.models.reporting import Reporter @@ -89,11 +97,23 @@ def run_report_job(data_source_id: int, parameters: dict) -> list[dict]: reporter = source.data_generator if not isinstance(reporter, Reporter): raise ValueError(f"Data source {data_source_id} does not store a Reporter.") + # The data generator instance is cached on the data source, which may be shared + # (e.g. within a long-lived worker process), so wipe any previous parameter state. + reporter._parameters = None results = reporter.compute(parameters=parameters) for result in results: save_to_db(result["data"]) db.session.commit() + if automation_id is not None and parameters.get("end"): + from datetime import datetime + + from flexmeasures.data.services.automations import record_automation_run + + record_automation_run( + automation_id, now=datetime.fromisoformat(parameters["end"]) + ) + # return a light summary (the report data itself is stored in the database) return [ {"sensor_id": result["sensor"].id, "n_rows": len(result["data"])} 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.py b/flexmeasures/data/tests/test_automations.py index 7e1a60a60f..f681fdabaf 100644 --- a/flexmeasures/data/tests/test_automations.py +++ b/flexmeasures/data/tests/test_automations.py @@ -44,3 +44,8 @@ def test_run_schedule_automation( "origin": "automation", "automation_id": automation.id, } + + # trigger provenance must not affect job identity: the same schedule request + # from another origin dedupes onto the same job (via the job cache) + returns_2 = run_automation(automation) + assert returns_2["job_id"] == returns["job_id"] From 292a0fa9f95bdee528043d3caf8f7798fc0caec5 Mon Sep 17 00:00:00 2001 From: Mohamed Belhsan Hmida Date: Wed, 12 Aug 2026 02:16:08 +0100 Subject: [PATCH 07/15] fix(automations): authorize reporter sensor dependencies Make reporters declare the sensors they read from and write to so automation creation, detail rendering, and sensor links use one authoritative dependency model. Profit and loss reports now include price sensors stored in reporter configuration, preventing cross-organisation report automations from bypassing access checks, and the incomplete duplicate scanner in the API is removed. Signed-off-by: Mohamed Belhsan Hmida --- flexmeasures/api/v3_0/assets.py | 91 ------------------- .../api/v3_0/tests/test_automations_api.py | 62 +++++++++++++ .../data/models/reporting/__init__.py | 22 +++++ flexmeasures/data/models/reporting/profit.py | 9 ++ flexmeasures/data/services/automations.py | 31 ++++--- 5 files changed, 111 insertions(+), 104 deletions(-) diff --git a/flexmeasures/api/v3_0/assets.py b/flexmeasures/api/v3_0/assets.py index 7dd926c559..ba36b0dfc3 100644 --- a/flexmeasures/api/v3_0/assets.py +++ b/flexmeasures/api/v3_0/assets.py @@ -117,90 +117,6 @@ sensors_schema = SensorSchema(many=True) -def _forecast_automation_sensor_refs( - parameters: dict, config: dict -) -> tuple[list, list]: - read_refs = [parameters.get("sensor")] - for key in ("regressors", "future-regressors", "past-regressors"): - read_refs.extend(config.get(key) or []) - write_refs = [parameters.get("sensor-to-save") or parameters.get("sensor")] - return read_refs, write_refs - - -def _report_automation_sensor_refs(parameters: dict) -> tuple[list, list]: - read_refs = [ - entry.get("sensor") - for entry in parameters.get("input") or [] - if isinstance(entry, dict) - ] - write_refs = [ - entry.get("sensor") - for entry in parameters.get("output") or [] - if isinstance(entry, dict) - ] - return read_refs, write_refs - - -def _schedule_automation_sensor_refs(parameters: dict) -> tuple[list, list]: - read_refs: list = [] - write_refs: list = [] - for entry in parameters.get("flex-model") or []: - if isinstance(entry, dict): - write_refs.append(entry.get("sensor")) - if isinstance(entry.get("state-of-charge"), dict): - read_refs.append(entry["state-of-charge"].get("sensor")) - flex_context = parameters.get("flex-context") or {} - if isinstance(flex_context, dict): - for value in flex_context.values(): - if isinstance(value, dict): - read_refs.append(value.get("sensor")) - elif isinstance(value, list): - for item in value: - read_refs.append( - item.get("sensor") if isinstance(item, dict) else item - ) - return read_refs, write_refs - - -def _check_automation_sensor_access( - automation_type: str, parameters: dict, config: dict -): - """Check that the current user may read every sensor referenced in an - automation's parameters/config, and record data on the sensors the - automation would write to. - - Sensor references that do not resolve are skipped here; schema validation - of the parameters rejects them later. - - :raises Forbidden: if any check fails (answered with 403). - """ - parameters = parameters or {} - config = config or {} - if automation_type == "forecasts": - read_refs, write_refs = _forecast_automation_sensor_refs(parameters, config) - elif automation_type == "reports": - read_refs, write_refs = _report_automation_sensor_refs(parameters) - elif automation_type == "schedules": - read_refs, write_refs = _schedule_automation_sensor_refs(parameters) - else: - read_refs, write_refs = [], [] - - def resolve(value) -> Sensor | None: - try: - return db.session.get(Sensor, int(value)) - except (TypeError, ValueError): - return None - - for value in read_refs: - sensor = resolve(value) - if sensor is not None: - check_access(sensor, "read") - for value in write_refs: - sensor = resolve(value) - if sensor is not None: - check_access(sensor, "create-children") - - def sensor_term_filter(term: str): filters = [Sensor.name.ilike(f"%{term}%")] if term.isdecimal(): @@ -1732,13 +1648,6 @@ def post_automation(self, id: int, asset: GenericAsset): automation_data = AutomationCreationSchema().load(body) except ValidationError as e: return unprocessable_entity(e.messages) - # Guard against referencing sensors outside the caller's reach - # (raises Forbidden, answered with 403) - _check_automation_sensor_access( - automation_data["type"], - automation_data["parameters"], - automation_data["config"], - ) try: automation, warnings = create_automation( asset=asset, diff --git a/flexmeasures/api/v3_0/tests/test_automations_api.py b/flexmeasures/api/v3_0/tests/test_automations_api.py index 366e36c57d..a5387e68d6 100644 --- a/flexmeasures/api/v3_0/tests/test_automations_api.py +++ b/flexmeasures/api/v3_0/tests/test_automations_api.py @@ -265,6 +265,68 @@ def test_post_automation_with_foreign_sensor( ) +@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": "reports", + "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 + ) + + @pytest.mark.parametrize( "requesting_user", ["test_prosumer_user_2@seita.nl"], indirect=True ) diff --git a/flexmeasures/data/models/reporting/__init__.py b/flexmeasures/data/models/reporting/__init__.py index 275a482250..d9152a54c5 100644 --- a/flexmeasures/data/models/reporting/__init__.py +++ b/flexmeasures/data/models/reporting/__init__.py @@ -19,6 +19,28 @@ class Reporter(DataGenerator): _parameters_schema = ReporterParametersSchema() _config_schema = ReporterConfigSchema() + @property + def input_sensors(self) -> list: + """The sensors from which the report reads its input data.""" + parameters = self._parameters or {} + return self._resolve_sensors( + [ + input_description.get("sensor") + for input_description in parameters.get("input", []) + ] + ) + + @property + def output_sensors(self) -> list: + """The sensors on which the report records its results.""" + parameters = self._parameters or {} + return self._resolve_sensors( + [ + output_description.get("sensor") + for output_description in parameters.get("output", []) + ] + ) + def _compute( self, check_output_resolution=True, as_job: bool = False, **kwargs ) -> list[dict[str, Any]] | dict[str, Any]: diff --git a/flexmeasures/data/models/reporting/profit.py b/flexmeasures/data/models/reporting/profit.py index 3503bdcc69..6b3f550f6a 100644 --- a/flexmeasures/data/models/reporting/profit.py +++ b/flexmeasures/data/models/reporting/profit.py @@ -47,6 +47,15 @@ class ProfitOrLossReporter(Reporter): weights: dict method: str + @property + def input_sensors(self) -> list: + """The flow input and price sensors read to compute profit or loss.""" + return self._resolve_sensors( + super().input_sensors, + self._config.get("consumption_price_sensor"), + self._config.get("production_price_sensor"), + ) + def _compute_report( self, start: datetime, diff --git a/flexmeasures/data/services/automations.py b/flexmeasures/data/services/automations.py index 29e5117037..5891d318a9 100644 --- a/flexmeasures/data/services/automations.py +++ b/flexmeasures/data/services/automations.py @@ -817,8 +817,8 @@ def _prepare_report_automation( generator_class: str | None, config: dict | None, source, -) -> tuple[int, list[str]]: - """Validate report automation parameters and set up the reporter's data 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 @@ -845,7 +845,9 @@ def _prepare_report_automation( 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. - reporter._parameters_schema.load(prepare_report_parameters(parameters, cronstr)) + 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: @@ -854,11 +856,7 @@ def _prepare_report_automation( " 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." ) - generator = ( - reporter.data_source - ) # looks up or creates the data source storing the reporter config - db.session.flush() - return generator.id, warnings + return reporter, deserialized_parameters, warnings def create_automation( @@ -897,7 +895,7 @@ def create_automation( 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 @@ -922,6 +920,7 @@ def create_automation( ) if forecaster is None: raise ValueError(f"Could not set up forecaster '{generator_class}'.") + 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). @@ -945,9 +944,15 @@ def create_automation( " Omit 'start' to schedule from the run time instead." ) elif automation_type == "reports": - generator_id, warnings = _prepare_report_automation( + reporter, deserialized_parameters, warnings = _prepare_report_automation( parameters, cronstr, generator_class, config, source ) + 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})." @@ -961,10 +966,10 @@ def create_automation( if forecast_output_sensor is not None: validate_forecast_output_scope(asset.id, forecast_output_sensor) - if forecaster is not None: - # Look up or create the data source storing the forecaster config only now that the automation is going ahead, + 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 From 8b768153219cde5a0f5ae201fd9716c119f5e87e Mon Sep 17 00:00:00 2001 From: Mohamed Belhsan Hmida Date: Wed, 12 Aug 2026 02:17:03 +0100 Subject: [PATCH 08/15] fix(reporting): require report inputs and outputs Require every report payload to declare at least one output and preserve the inherited required constraints when specialized profit and aggregation schemas narrow list lengths. Invalid automations now fail during API or CLI validation instead of being accepted and later crashing a reporting worker with missing method arguments. Signed-off-by: Mohamed Belhsan Hmida --- flexmeasures/data/schemas/reporting/__init__.py | 4 +++- .../data/schemas/reporting/aggregation.py | 2 +- flexmeasures/data/schemas/reporting/profit.py | 4 +++- .../data/schemas/tests/test_reporting.py | 16 ++++++++++++++++ 4 files changed, 23 insertions(+), 3 deletions(-) diff --git a/flexmeasures/data/schemas/reporting/__init__.py b/flexmeasures/data/schemas/reporting/__init__.py index 02e580ae42..1b72571198 100644 --- a/flexmeasures/data/schemas/reporting/__init__.py +++ b/flexmeasures/data/schemas/reporting/__init__.py @@ -29,7 +29,9 @@ class ReporterParametersSchema(Schema): validate=validate.Length(min=1), ) - output = fields.List(fields.Nested(Output()), validate=validate.Length(min=1)) + output = fields.List( + fields.Nested(Output()), required=True, validate=validate.Length(min=1) + ) start = AwareDateTimeField(required=True) end = AwareDateTimeField(required=True) diff --git a/flexmeasures/data/schemas/reporting/aggregation.py b/flexmeasures/data/schemas/reporting/aggregation.py index 165e646fb5..e3c37c55d5 100644 --- a/flexmeasures/data/schemas/reporting/aggregation.py +++ b/flexmeasures/data/schemas/reporting/aggregation.py @@ -56,5 +56,5 @@ class AggregatorParametersSchema(ReporterParametersSchema): # redefining output to restrict the output length to 1 output = fields.List( - fields.Nested(Output()), validate=validate.Length(min=1, max=1) + fields.Nested(Output()), required=True, validate=validate.Length(min=1, max=1) ) diff --git a/flexmeasures/data/schemas/reporting/profit.py b/flexmeasures/data/schemas/reporting/profit.py index 2cae5069b9..8444082ba9 100644 --- a/flexmeasures/data/schemas/reporting/profit.py +++ b/flexmeasures/data/schemas/reporting/profit.py @@ -95,7 +95,9 @@ class ProfitOrLossReporterParametersSchema(ReporterParametersSchema): """ # redefining output to restrict the input length to 1 - input = fields.List(fields.Nested(Input()), validate=validate.Length(min=1, max=1)) + input = fields.List( + fields.Nested(Input()), required=True, validate=validate.Length(min=1, max=1) + ) @validates("input") def validate_input_measures_power_energy(self, value, **kwargs): diff --git a/flexmeasures/data/schemas/tests/test_reporting.py b/flexmeasures/data/schemas/tests/test_reporting.py index 337f8884d3..f01f7c9b78 100644 --- a/flexmeasures/data/schemas/tests/test_reporting.py +++ b/flexmeasures/data/schemas/tests/test_reporting.py @@ -199,6 +199,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 From 82cb3bbe9f0571479bec7b4a48ec53e071fb9cd5 Mon Sep 17 00:00:00 2001 From: Mohamed Belhsan Hmida Date: Wed, 12 Aug 2026 02:20:09 +0100 Subject: [PATCH 09/15] fix(automations): constrain report output scope Apply the existing automation subtree invariant to report outputs during both creation and execution, after access checks have established that sensor metadata may be discussed. Reporter parameters are prepared before dependency resolution so recurring reports with relative windows expose and validate their sensors consistently on API details and worker runs. Signed-off-by: Mohamed Belhsan Hmida --- .../api/v3_0/tests/test_automations_api.py | 51 +++++++++++++++++++ flexmeasures/data/services/automations.py | 49 ++++++++++++------ 2 files changed, 84 insertions(+), 16 deletions(-) diff --git a/flexmeasures/api/v3_0/tests/test_automations_api.py b/flexmeasures/api/v3_0/tests/test_automations_api.py index a5387e68d6..51d02440bf 100644 --- a/flexmeasures/api/v3_0/tests/test_automations_api.py +++ b/flexmeasures/api/v3_0/tests/test_automations_api.py @@ -327,6 +327,57 @@ def test_post_report_automation_with_foreign_config_sensor( ) +@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": "reports", + "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 + + @pytest.mark.parametrize( "requesting_user", ["test_prosumer_user_2@seita.nl"], indirect=True ) diff --git a/flexmeasures/data/services/automations.py b/flexmeasures/data/services/automations.py index 5891d318a9..ffcb0d6bc5 100644 --- a/flexmeasures/data/services/automations.py +++ b/flexmeasures/data/services/automations.py @@ -448,8 +448,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). @@ -470,9 +470,16 @@ 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 == "reports": + parameters = prepare_report_parameters( + parameters, + automation.cronstr, + automation_id=automation.id, + ) 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( @@ -500,7 +507,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. @@ -898,7 +905,6 @@ def create_automation( data_generator = None input_sensors: list[Sensor] = [] output_sensors: list[Sensor] = [] - forecast_output_sensor: Sensor | None = None if automation_type == "forecasts": from flexmeasures.data.services.data_sources import get_data_generator from flexmeasures.data.schemas.forecasting.pipeline import ( @@ -930,7 +936,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 == "schedules": # A schedule is recorded on the sensors that the scheduler returns its results # for, and reads whatever other sensors the flex-model and flex-context refer to @@ -963,8 +968,9 @@ 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 ("forecasts", "reports"): + for output_sensor in output_sensors: + validate_automation_output_scope(asset.id, output_sensor, automation_type) 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, @@ -1073,11 +1079,13 @@ 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." ) @@ -1116,7 +1124,9 @@ 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 + ) # 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. forecaster._parameters = None @@ -1136,16 +1146,23 @@ def _run_report_automation( raise ValueError( f"Data source {automation.generator_id} of automation {automation.id} does not store a Reporter." ) - # 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) parameters = prepare_report_parameters( dict(automation.parameters), automation.cronstr, now=now, automation_id=automation.id, ) + 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) From 3e6c2ece0e480afd9d67591a82c978db29181c7a Mon Sep 17 00:00:00 2001 From: Mohamed Belhsan Hmida Date: Wed, 12 Aug 2026 02:20:30 +0100 Subject: [PATCH 10/15] fix(automations): require generators for reports Extend the database invariant that protects executable automations so report rows, like forecast rows, cannot exist without a data generator. The migration replaces the forecast-only check constraint while preserving generator-free schedule automations and provides a reversible downgrade to the previous rule. Signed-off-by: Mohamed Belhsan Hmida --- ...quire_generators_for_report_automations.py | 33 +++++++++++++++++++ flexmeasures/data/models/automations.py | 4 +-- .../data/tests/test_automations_fresh_db.py | 15 +++++++++ 3 files changed, 50 insertions(+), 2 deletions(-) create mode 100644 flexmeasures/data/migrations/versions/d2a4f6b8c901_require_generators_for_report_automations.py diff --git a/flexmeasures/data/migrations/versions/d2a4f6b8c901_require_generators_for_report_automations.py b/flexmeasures/data/migrations/versions/d2a4f6b8c901_require_generators_for_report_automations.py new file mode 100644 index 0000000000..9726544e7d --- /dev/null +++ b/flexmeasures/data/migrations/versions/d2a4f6b8c901_require_generators_for_report_automations.py @@ -0,0 +1,33 @@ +"""Require generators for forecast and report automations. + +Revision ID: d2a4f6b8c901 +Revises: c63896a97a8e +Create Date: 2026-08-12 02:20:00.000000 + +""" + +from alembic import op + +# revision identifiers, used by Alembic. +revision = "d2a4f6b8c901" +down_revision = "c63896a97a8e" +branch_labels = None +depends_on = None + + +def upgrade(): + op.drop_constraint("forecast_generator", "automation", type_="check") + op.create_check_constraint( + "automation_generator", + "automation", + "type NOT IN ('forecasts', 'reports') OR generator_id IS NOT NULL", + ) + + +def downgrade(): + op.drop_constraint("automation_generator", "automation", type_="check") + op.create_check_constraint( + "forecast_generator", + "automation", + "type != 'forecasts' OR generator_id IS NOT NULL", + ) diff --git a/flexmeasures/data/models/automations.py b/flexmeasures/data/models/automations.py index 1734c682e1..d4cee597c5 100644 --- a/flexmeasures/data/models/automations.py +++ b/flexmeasures/data/models/automations.py @@ -41,8 +41,8 @@ class Automation(db.Model, AuthModelMixin): __tablename__ = "automation" __table_args__ = ( db.CheckConstraint( - "type != 'forecasts' OR generator_id IS NOT NULL", - name="forecast_generator", + "type NOT IN ('forecasts', 'reports') OR generator_id IS NOT NULL", + name="automation_generator", ), ) diff --git a/flexmeasures/data/tests/test_automations_fresh_db.py b/flexmeasures/data/tests/test_automations_fresh_db.py index 75db9bdc1a..0fd6602672 100644 --- a/flexmeasures/data/tests/test_automations_fresh_db.py +++ b/flexmeasures/data/tests/test_automations_fresh_db.py @@ -74,6 +74,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="reports", + 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_does_not_require_generator( fresh_db, automation_with_generator ): From 601f9513d08c2d2bfd3345cb104c7368afe19e5d Mon Sep 17 00:00:00 2001 From: Mohamed Belhsan Hmida Date: Wed, 12 Aug 2026 02:23:33 +0100 Subject: [PATCH 11/15] fix(automations): anchor reports to claimed occurrences Build recurring report windows from the canonical cron occurrence claimed by the runner and interpret prior occurrences in each automation's own IANA timezone. Delayed catch-up runs now produce stable boundaries, including across daylight-saving gaps, instead of using the platform timezone and the worker's later wall-clock time. Signed-off-by: Mohamed Belhsan Hmida --- flexmeasures/cli/jobs.py | 4 +- flexmeasures/cli/tests/test_automations.py | 42 ++++++++++++++++-- flexmeasures/data/services/automations.py | 50 +++++++++++++++++----- 3 files changed, 81 insertions(+), 15 deletions(-) diff --git a/flexmeasures/cli/jobs.py b/flexmeasures/cli/jobs.py index 97544b7808..0a3561cef3 100644 --- a/flexmeasures/cli/jobs.py +++ b/flexmeasures/cli/jobs.py @@ -106,7 +106,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 queue_name = { "forecasts": "forecasting", diff --git a/flexmeasures/cli/tests/test_automations.py b/flexmeasures/cli/tests/test_automations.py index 7de17181ac..fe6a750450 100644 --- a/flexmeasures/cli/tests/test_automations.py +++ b/flexmeasures/cli/tests/test_automations.py @@ -949,6 +949,31 @@ def test_prepare_report_parameters(app): 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 occurrence 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 occurrence in Amsterdam's spring gap is canonicalized to 03:00, + # while its report starts at the prior day's real 02:30 occurrence. + spring_occurrence = datetime(2026, 3, 29, 1, 0, tzinfo=timezone.utc) + message = prepare_report_parameters( + {}, + "30 2 * * *", + cron_timezone="Europe/Amsterdam", + scheduled_at=spring_occurrence, + ) + assert pd.Timestamp(message["start"]) == pd.Timestamp("2026-03-28T01:30:00+00:00") + assert pd.Timestamp(message["end"]) == pd.Timestamp(spring_occurrence) + # 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: @@ -988,7 +1013,12 @@ def test_prepare_report_parameters(app): def _report_automation_cli_input( - tmp_path, sensor1_id, sensor2_id, report_sensor_id, parameters_extra=None + 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( @@ -1017,7 +1047,7 @@ def _report_automation_cli_input( parameters_file = tmp_path / "parameters.yml" parameters_file.write_text(yaml.dump(parameters)) return [ - "--asset", "1", + "--asset", str(asset_id), "--name", "Aggregation report", "--cron", "0 1 * * *", "--type", "reports", @@ -1032,6 +1062,9 @@ def test_add_report_automation(app, fresh_db, setup_dummy_data, tmp_path): 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, @@ -1041,6 +1074,7 @@ def test_add_report_automation(app, fresh_db, setup_dummy_data, tmp_path): 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 @@ -1088,6 +1122,7 @@ def test_run_report_automation(app, fresh_db, setup_dummy_data, clean_redis, tmp 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, @@ -1099,6 +1134,7 @@ def test_run_report_automation(app, fresh_db, setup_dummy_data, clean_redis, tmp "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) @@ -1263,7 +1299,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/services/automations.py b/flexmeasures/data/services/automations.py index ffcb0d6bc5..76f1e778c2 100644 --- a/flexmeasures/data/services/automations.py +++ b/flexmeasures/data/services/automations.py @@ -476,6 +476,7 @@ def resolve_automation_sensors(automation: Automation) -> dict[str, list[Sensor] parameters, automation.cronstr, automation_id=automation.id, + cron_timezone=automation.timezone, ) return resolve_data_generator_sensors( data_generator, @@ -619,6 +620,8 @@ def prepare_report_parameters( 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. @@ -633,10 +636,9 @@ def prepare_report_parameters( first run). """ message = dict(parameters) - if now is None: - now = server_now() - # Cron strings are interpreted in the platform timezone (like the runner does) - platform_now = floor_to_minute(now) + 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. @@ -655,7 +657,7 @@ def prepare_report_parameters( output_sensor = None if output_sensor is not None: tz = pytz.timezone(output_sensor.timezone) - now = platform_now.astimezone(tz) + now = scheduled_at.astimezone(tz) start_offset = message.pop("start-offset", None) end_offset = message.pop("end-offset", None) @@ -680,9 +682,29 @@ def prepare_report_parameters( if automation_id is not None else None ) - # NB the cron fallback uses the platform timezone, matching how the runner - # decides when the automation fires - start = last_run or croniter(cronstr, platform_now).get_prev(datetime) + 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_occurrence_time(previous_nominal, cron_tz) + # A skipped wall time can canonicalize to the first valid instant after + # the gap, which may be the current occurrence. 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_occurrence_time(previous_nominal, cron_tz) if end is None: end = now @@ -1090,7 +1112,9 @@ def validate_automation_output_scope( ) -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": }. @@ -1104,7 +1128,7 @@ def run_automation(automation: Automation) -> dict[str, Any] | None: # 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) + return _run_report_automation(automation, now=now, scheduled_at=scheduled_at) else: raise NotImplementedError( f"Automations of type '{automation.type}' cannot be run yet." @@ -1135,7 +1159,9 @@ def _run_forecast_automation(automation: Automation) -> dict[str, Any] | None: def _run_report_automation( - automation: Automation, now: datetime | None = None + automation: Automation, + now: datetime | None = None, + scheduled_at: datetime | None = None, ) -> dict[str, Any] | None: if automation.generator is None: raise ValueError( @@ -1151,6 +1177,8 @@ def _run_report_automation( 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) From f0c3fce6ec529045fa35a3951d2260da0df3855d Mon Sep 17 00:00:00 2001 From: Mohamed Belhsan Hmida Date: Wed, 12 Aug 2026 02:24:59 +0100 Subject: [PATCH 12/15] fix(automations): advance report coverage monotonically Update the Redis coverage anchor with an optimistic transaction that only accepts a later report end. Concurrent or out-of-order reporting workers can no longer let an older completion rewind the next default window and cause already reported periods to be processed again. Signed-off-by: Mohamed Belhsan Hmida --- flexmeasures/cli/tests/test_automations.py | 15 ++++++++++ flexmeasures/data/services/automations.py | 33 ++++++++++++++++++---- 2 files changed, 43 insertions(+), 5 deletions(-) diff --git a/flexmeasures/cli/tests/test_automations.py b/flexmeasures/cli/tests/test_automations.py index fe6a750450..94b20b66aa 100644 --- a/flexmeasures/cli/tests/test_automations.py +++ b/flexmeasures/cli/tests/test_automations.py @@ -1012,6 +1012,21 @@ def test_prepare_report_parameters(app): 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, diff --git a/flexmeasures/data/services/automations.py b/flexmeasures/data/services/automations.py index 76f1e778c2..9cd1d2a6e5 100644 --- a/flexmeasures/data/services/automations.py +++ b/flexmeasures/data/services/automations.py @@ -583,7 +583,7 @@ 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): +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. @@ -591,13 +591,36 @@ def record_automation_run(automation_id: int, now: datetime | None = None): instead, upon success (see run_report_job), so a failed report job does not create a permanent gap in the reported periods. """ - from flask import current_app + from redis.exceptions import WatchError if now is None: now = server_now() - current_app.redis_connection.set( - _last_run_redis_key(automation_id), floor_to_minute(now).isoformat() - ) + 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: From 2350b68cd57212e414b7a328a1c14f807b3ef235 Mon Sep 17 00:00:00 2001 From: Mohamed Belhsan Hmida Date: Wed, 12 Aug 2026 02:26:10 +0100 Subject: [PATCH 13/15] test(ui): cover report automation listings Assert that the asset automations page exposes the reports tab and its dedicated table alongside forecasts and schedules. This protects the report UI added by the stacked branch from disappearing during future template reconciliations. Signed-off-by: Mohamed Belhsan Hmida --- flexmeasures/ui/tests/test_asset_crud.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/flexmeasures/ui/tests/test_asset_crud.py b/flexmeasures/ui/tests/test_asset_crud.py index fa3f3b8003..c62bdec28b 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-forecasts"' in asset_page.data assert b'id="automationsTable-schedules"' in asset_page.data + assert b'id="automationsTable-reports"' 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 From 77c6c2d36aecd7ad9973db72d2d43c784acb18d7 Mon Sep 17 00:00:00 2001 From: Mohamed Belhsan Hmida Date: Wed, 12 Aug 2026 02:26:35 +0100 Subject: [PATCH 14/15] docs(automations): document report occurrence semantics Describe per-automation timezone selection and the successful-coverage model used by recurring reports. The documentation now distinguishes the claimed cron occurrence from delayed runner time and explains how first runs, daylight-saving transitions, and out-of-order worker completions determine report windows. Signed-off-by: Mohamed Belhsan Hmida --- documentation/features/automations.rst | 10 +++++----- documentation/features/reporting.rst | 7 ++++--- 2 files changed, 9 insertions(+), 8 deletions(-) diff --git a/documentation/features/automations.rst b/documentation/features/automations.rst index ad113c8bbb..825bd6a103 100644 --- a/documentation/features/automations.rst +++ b/documentation/features/automations.rst @@ -9,7 +9,7 @@ Hosts and users often want the three main FlexMeasures features — :ref:`foreca An automation consists of: - a **type**: ``forecasts``, ``schedules`` or ``reports``; -- a **recurrence**: a cron string (e.g. ``"0 6 * * *"`` for daily at 6 AM), interpreted in the ``FLEXMEASURES_TIMEZONE``; +- 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. @@ -22,18 +22,18 @@ Managing automations Automations can be managed in three ways: -- **CLI**: ``flexmeasures add automation``, ``flexmeasures edit automation`` (name, cron string, activation status) and ``flexmeasures delete automation``. +- **CLI**: ``flexmeasures add automation``, ``flexmeasures edit automation`` (name, cron string, timezone and activation status) and ``flexmeasures delete automation``. - **API**: list and inspect with ``[GET] /assets/(id)/automations`` and ``[GET] /assets/(id)/automations/(automation_id)``; create, update and delete with ``[POST|PATCH|DELETE]`` on the same paths (see the `API documentation <../api/v3_0.html>`_). - **UI**: each asset has an *Automations* page (in the breadcrumbs dropdown), with a tab per automation type. - It lists each automation's recurrence and recent job counts, and lets you create, (de)activate and delete automations. + It lists each automation's recurrence and recent job counts, and lets you create, edit, (de)activate and delete automations. Creating, updating and deleting automations requires account admin or consultant rights, and is recorded in the asset's audit log. Running automations -------------------- -An automation is due whenever its cron string matches the current minute. To actually run due automations, 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 @@ -53,4 +53,4 @@ The parameters stored on an automation follow the same schemas as one-off CLI/AP - :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 automation's actual last run. + or omit timing fields to report on the period since the last successfully covered report window. diff --git a/documentation/features/reporting.rst b/documentation/features/reporting.rst index ab2418ed9c..15764192d9 100644 --- a/documentation/features/reporting.rst +++ b/documentation/features/reporting.rst @@ -135,10 +135,11 @@ and computed on a recurring basis by an *automation* defined on the asset (see : The reporter and its configuration are stored on a data source (steady across runs, so all report results attribute to the same source), while the report parameters are stored on the automation itself and their timing is resolved freshly 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 run time, +- 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 on the period since the automation's actual last run - (falling back to the last cron period — from the previous cron fire time until the run time — when no last run is known, e.g. on the first run). +- 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: From 95b6187f383854263fcdebea4e05b9021afa0b43 Mon Sep 17 00:00:00 2001 From: Mohamed Belhsan Hmida Date: Wed, 12 Aug 2026 02:32:53 +0100 Subject: [PATCH 15/15] test(api): close rejected report transactions Commit the fixture-owned sensor setup after rejected report automation requests so the shared API test database does not retain an idle transaction during teardown. This keeps the complete automation API module deterministic while preserving assertions that no unauthorized automation was created. Signed-off-by: Mohamed Belhsan Hmida --- flexmeasures/api/v3_0/tests/test_automations_api.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/flexmeasures/api/v3_0/tests/test_automations_api.py b/flexmeasures/api/v3_0/tests/test_automations_api.py index 51d02440bf..d5483cbfe5 100644 --- a/flexmeasures/api/v3_0/tests/test_automations_api.py +++ b/flexmeasures/api/v3_0/tests/test_automations_api.py @@ -325,6 +325,7 @@ def test_post_report_automation_with_foreign_config_sensor( ).scalar_one_or_none() is None ) + db.session.commit() @pytest.mark.parametrize( @@ -376,6 +377,7 @@ def test_post_report_automation_rejects_output_outside_asset_subtree( assert response.status_code == 422 assert "must belong to asset" in response.text + db.session.commit() @pytest.mark.parametrize(