From 0b345b2ec7cc283a6a617f9657156242ee13d882 Mon Sep 17 00:00:00 2001 From: "F.N. Claessen" Date: Sat, 11 Jul 2026 18:29:55 +0200 Subject: [PATCH 1/3] feat: record on scheduling jobs how they were created The scheduling job creators accept an optional trigger dict (stored as job meta data), like the forecasting pipeline already does. The API trigger endpoint records origin API; the CLI and automations follow in the next commit. The status page's 'Created Via' column picks this up automatically. 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 | 1 + flexmeasures/data/services/scheduling.py | 11 +++++++++++ 2 files changed, 12 insertions(+) diff --git a/flexmeasures/api/v3_0/assets.py b/flexmeasures/api/v3_0/assets.py index b6af05ff32..9f1ef8656c 100644 --- a/flexmeasures/api/v3_0/assets.py +++ b/flexmeasures/api/v3_0/assets.py @@ -1712,6 +1712,7 @@ def trigger_schedule( asset=asset, enqueue=True, force_new_job_creation=force_new_job_creation, + trigger={"origin": "API"}, **scheduler_kwargs, ) except ValidationError as err: diff --git a/flexmeasures/data/services/scheduling.py b/flexmeasures/data/services/scheduling.py index 01b3d6b922..e9fdc86ce7 100644 --- a/flexmeasures/data/services/scheduling.py +++ b/flexmeasures/data/services/scheduling.py @@ -212,6 +212,7 @@ def create_scheduling_job( scheduler_specs: dict | None = None, depends_on: Job | list[Job] | None = None, success_callback: Callable | None = None, + trigger: dict | None = None, **scheduler_kwargs, ) -> Job: """ @@ -236,6 +237,8 @@ def create_scheduling_job( :param force_new_job_creation: If True, this attribute forces a new job to be created (skipping cache). :param success_callback: Callback function that runs on success (this argument is used by the @job_cache decorator). + :param trigger: Optionally, info about how the job got created (e.g. via the CLI, + the API or an automation), stored as job meta data. :returns: The job. """ @@ -288,6 +291,8 @@ def create_scheduling_job( ) job.meta["asset_or_sensor"] = asset_or_sensor + if trigger: + job.meta["trigger"] = trigger job.meta["scheduler_kwargs"] = scheduler_kwargs # Serialize start, end, resolution and belief_time @@ -347,6 +352,7 @@ def create_sequential_scheduling_job( scheduler_specs: dict | None = None, depends_on: list[Job] | None = None, success_callback: Callable | None = None, + trigger: dict | None = None, **scheduler_kwargs, ) -> Job: """Create a chain of underlying jobs, one for each device, with one additional job to wrap up. @@ -410,6 +416,7 @@ def create_sequential_scheduling_job( enqueue=enqueue, depends_on=previous_job, force_new_job_creation=force_new_job_creation, + trigger=trigger, ) jobs.append(job) previous_sensors.append(sensor.id) @@ -434,6 +441,8 @@ def create_sequential_scheduling_job( connection=current_app.queues["scheduling"].connection, ) job.meta["asset_or_sensor"] = get_asset_or_sensor_ref(asset) + if trigger: + job.meta["trigger"] = trigger job.save_meta() try: @@ -463,6 +472,7 @@ def create_simultaneous_scheduling_job( scheduler_specs: dict | None = None, depends_on: list[Job] | None = None, success_callback: Callable | None = None, + trigger: dict | None = None, **scheduler_kwargs, ) -> Job: """Create a single job to schedule all devices at once. @@ -510,6 +520,7 @@ def create_simultaneous_scheduling_job( depends_on=depends_on, success_callback=success_callback, force_new_job_creation=force_new_job_creation, + trigger=trigger, ) try: From 4c0c5dbac26d64492686ae44b6bfcf7d1bf8a4d0 Mon Sep 17 00:00:00 2001 From: "F.N. Claessen" Date: Sat, 11 Jul 2026 18:31:36 +0200 Subject: [PATCH 2/3] feat: schedules as automations MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Automations now also support the 'schedules' type: - `flexmeasures add automation --type schedules` validates the parameters as a schedule trigger message (per the AssetTriggerSchema, as accepted by the API trigger endpoint, without the asset id). The schedule 'start' may be omitted, in which case each run schedules from the run time (floored to the message's resolution, if given) — a fixed start draws a warning. - The runner dispatches schedules automations to the same job creators as the API trigger endpoint (sequential or simultaneous), recording trigger meta data (origin automation) on the queued jobs; `flexmeasures add schedule --as-job` now records origin CLI. - Job stats for schedules automations are counted from the scheduling job cache (asset-level wrap-up jobs and per-sensor device jobs). - The UI automations page's Schedules tab is now enabled, with automations filtered by type per tab. Part of FlexMeasures/flexmeasures#2288 Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Rbix8k1JfeUWNXEmHEZVpX --- documentation/cli/change_log.rst | 2 +- documentation/cli/commands.rst | 2 +- documentation/features/forecasting.rst | 2 + documentation/features/scheduling.rst | 18 +++ flexmeasures/cli/data_add.py | 101 ++++++++++------ flexmeasures/cli/jobs.py | 5 +- flexmeasures/cli/tests/test_automations.py | 112 ++++++++++++++++++ flexmeasures/cli/utils.py | 17 +++ flexmeasures/data/models/automations.py | 2 +- flexmeasures/data/services/automations.py | 109 ++++++++++++++--- flexmeasures/data/tests/test_automations.py | 46 +++++++ .../templates/assets/asset_automations.html | 56 +++++---- 12 files changed, 392 insertions(+), 80 deletions(-) create mode 100644 flexmeasures/data/tests/test_automations.py diff --git a/documentation/cli/change_log.rst b/documentation/cli/change_log.rst index 3128ee4420..d088c749ea 100644 --- a/documentation/cli/change_log.rst +++ b/documentation/cli/change_log.rst @@ -9,7 +9,7 @@ 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; for now, computing forecasts). +* 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 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 ad884bafac..f67e4d0643 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 (for now, computing forecasts) on an asset. +``flexmeasures add automation`` Add an automation: a recurring task (computing forecasts or schedules) on an asset. ================================================= ======================================= diff --git a/documentation/features/forecasting.rst b/documentation/features/forecasting.rst index 9c12ea0bca..f6c5d57113 100644 --- a/documentation/features/forecasting.rst +++ b/documentation/features/forecasting.rst @@ -204,3 +204,5 @@ Each due automation then queues its forecasting jobs. The jobs record how they were created, which is shown on the asset's status page (UI), where recent jobs are listed. Automations defined on an asset can be viewed on the asset's *Automations* page in the UI, and listed with the API endpoint `[GET] /assets/(id)/automations <../api/v3_0.html>`_. + +Schedules can be automated in the same way — see :ref:`automating_schedules`. diff --git a/documentation/features/scheduling.rst b/documentation/features/scheduling.rst index aec2ce3397..5e73401ddf 100644 --- a/documentation/features/scheduling.rst +++ b/documentation/features/scheduling.rst @@ -578,3 +578,21 @@ Here are some thoughts on further innovation: - Aggregating flexibility of a group of assets (e.g. a neighborhood) and optimizing its aggregated usage (e.g. for grid congestion support) is also an exciting direction for expansion. + + +.. _automating_schedules: + +Automating schedules +-------------------- + +Like forecasts, schedules can be 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 automation's parameters form a schedule trigger message, as accepted by the `[POST] /assets/(id)/schedules/trigger <../api/v3_0.html>`_ 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. + +For example, this automation queues a scheduling job every hour, each time scheduling the next 12 hours: + +.. code-block:: bash + + echo 'duration: "PT12H"' > trigger-message.yml + flexmeasures add automation --asset 3 --name "Hourly schedules" --cron "0 * * * *" --type schedules --parameters trigger-message.yml diff --git a/flexmeasures/cli/data_add.py b/flexmeasures/cli/data_add.py index 5951800186..1de8452f7f 100755 --- a/flexmeasures/cli/data_add.py +++ b/flexmeasures/cli/data_add.py @@ -37,6 +37,7 @@ DeprecatedOption, DeprecatedOptionsCommand, add_cli_options_from_schema, + make_cli_options_optional, ) from flexmeasures.data import db from flexmeasures.data.scripts.data_gen import ( @@ -44,6 +45,7 @@ populate_initial_structure, add_default_asset_types, ) +from flexmeasures.data.services.automations import prepare_schedule_trigger_message from flexmeasures.data.services.data_sources import ( get_or_create_source, get_data_generator, @@ -1268,8 +1270,12 @@ def add_forecast( # noqa: C901 "parameters_file", required=False, type=click.File("r"), - help="Path to the JSON or YAML file with the forecast parameters (passed to the compute step on each run of the automation).", + 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.", ) +@make_cli_options_optional( + "sensor" +) # only required for --type forecasts, which the forecast parameter schema enforces @add_cli_options_from_schema(ForecasterParametersSchema()) @add_cli_options_from_schema(TrainPredictPipelineConfigSchema()) def add_automation( @@ -1285,51 +1291,72 @@ def add_automation( **kwargs, ): """ - Add an automation: a recurring task (for now, computing forecasts) on an asset. + Add an automation: a recurring task (computing forecasts or schedules) on an asset. \b - Example + Examples flexmeasures add automation --asset 3 --name "Day-ahead PV forecasts" --cron "0 6 * * *" --sensor 2092 --regressors 2093 - - The forecaster configuration is stored on a data source, and the forecast - parameters are validated and stored on the automation itself. - Each time the automation runs, forecasting jobs are queued - (see `flexmeasures jobs run-automations`). + flexmeasures add automation --asset 3 --name "Hourly schedules" + --cron "0 * * * *" --type schedules --parameters trigger-message.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 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. + Each time the automation runs, jobs are queued (see `flexmeasures jobs run-automations`). """ config, parameters = _assemble_forecaster_config_and_parameters( kwargs, config_file, parameters_file ) # Validate the parameters using the forecast parameters schema (we store them serialized) - try: - deserialized_parameters = ForecasterParametersSchema().load(parameters) - except ValidationError as e: - click.secho(f"Invalid forecast parameters: {e.messages}", **MsgStyle.ERROR) - raise click.Abort() - sensor = deserialized_parameters.get("sensor") - if isinstance(sensor, Sensor) and sensor.generic_asset_id != asset.id: - click.secho( - f"Warning: the sensor to forecast ({sensor.id}) does not belong to asset {asset.id}.", - **MsgStyle.WARN, - ) + generator_id = None + if automation_type == "forecasts": + try: + deserialized_parameters = ForecasterParametersSchema().load(parameters) + except ValidationError as e: + click.secho(f"Invalid forecast parameters: {e.messages}", **MsgStyle.ERROR) + raise click.Abort() + sensor = deserialized_parameters.get("sensor") + if isinstance(sensor, Sensor) and sensor.generic_asset_id != asset.id: + click.secho( + f"Warning: the sensor to forecast ({sensor.id}) does not belong to asset {asset.id}.", + **MsgStyle.WARN, + ) - forecaster = get_data_generator( - source=source, - model=forecaster_class, - config=config, - save_config=True, - data_generator_type=Forecaster, - ) - if forecaster is None: - click.secho( - f"Could not set up forecaster '{forecaster_class}'.", **MsgStyle.ERROR + forecaster = get_data_generator( + source=source, + model=forecaster_class, + config=config, + save_config=True, + data_generator_type=Forecaster, ) - raise click.Abort() - generator = ( - forecaster.data_source - ) # looks up or creates the data source storing the forecaster config - db.session.flush() + if forecaster is None: + click.secho( + f"Could not set up forecaster '{forecaster_class}'.", **MsgStyle.ERROR + ) + raise click.Abort() + generator = ( + forecaster.data_source + ) # looks up or creates the data source storing the forecaster config + db.session.flush() + generator_id = generator.id + else: # schedules + try: + AssetTriggerSchema().load( + prepare_schedule_trigger_message(parameters, asset.id) + ) + except ValidationError as e: + click.secho(f"Invalid schedule parameters: {e.messages}", **MsgStyle.ERROR) + raise click.Abort() + if "start" in parameters: + click.secho( + "Warning: the schedule 'start' is fixed, so each run will compute the same period." + " Omit 'start' to schedule from the run time instead.", + **MsgStyle.WARN, + ) automation = Automation( asset_id=asset.id, @@ -1337,7 +1364,7 @@ def add_automation( name=name, cronstr=cronstr, active=not inactive, - generator_id=generator.id, + generator_id=generator_id, parameters=parameters, ) db.session.add(automation) @@ -1513,7 +1540,9 @@ def add_schedule( # noqa C901 if as_job: job = create_scheduling_job( - asset_or_sensor=asset_or_sensor, **scheduling_kwargs + asset_or_sensor=asset_or_sensor, + trigger={"origin": "CLI"}, + **scheduling_kwargs, ) if job: click.secho( diff --git a/flexmeasures/cli/jobs.py b/flexmeasures/cli/jobs.py index 918d76b216..ee31e8c620 100644 --- a/flexmeasures/cli/jobs.py +++ b/flexmeasures/cli/jobs.py @@ -97,8 +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 + ) click.secho( - f"Automation {automation.id} ('{automation.name}') queued {n_jobs} forecasting job(s) for asset {automation.asset_id}.", + f"Automation {automation.id} ('{automation.name}') queued {n_jobs} {queue_name} job(s) for asset {automation.asset_id}.", **MsgStyle.SUCCESS, ) n_run += 1 diff --git a/flexmeasures/cli/tests/test_automations.py b/flexmeasures/cli/tests/test_automations.py index d4e0895876..897bb43500 100644 --- a/flexmeasures/cli/tests/test_automations.py +++ b/flexmeasures/cli/tests/test_automations.py @@ -1,3 +1,5 @@ +from datetime import timedelta + import pytest from sqlalchemy import select @@ -84,6 +86,116 @@ def test_add_automation_invalid_cron(app, fresh_db, setup_dummy_data): assert "Invalid value for '--cron'" in result.output +def test_add_schedule_automation(app, fresh_db, setup_dummy_data, tmp_path): + """Create a schedules automation; parameters are validated as a schedule trigger message.""" + from flexmeasures.cli.data_add import add_automation + + runner = app.test_cli_runner() + + # invalid parameters (unknown field) are rejected + parameters_file = tmp_path / "parameters.yml" + parameters_file.write_text("not-a-trigger-field: 1\n") + result = runner.invoke( + add_automation, + [ + "--asset", "1", + "--name", "Bad schedules", + "--cron", "0 * * * *", + "--type", "schedules", + "--parameters", str(parameters_file), + ], + ) # fmt: skip + assert result.exit_code != 0 + assert "Invalid schedule parameters" in result.output + + # minimal valid parameters (flex config can live on the asset) + parameters_file.write_text('duration: "PT12H"\n') + result = runner.invoke( + add_automation, + [ + "--asset", "1", + "--name", "Half-day schedules", + "--cron", "0 * * * *", + "--type", "schedules", + "--parameters", str(parameters_file), + ], + ) # fmt: skip + assert "Successfully created" in result.output, result.output + automation = fresh_db.session.execute( + select(Automation).filter_by(name="Half-day schedules") + ).scalar_one() + assert automation.type == "schedules" + assert automation.generator_id is None + assert automation.parameters == {"duration": "PT12H"} + + # a fixed start draws a warning + parameters_file.write_text('start: "2026-01-01T00:00:00+01:00"\n') + result = runner.invoke( + add_automation, + [ + "--asset", "1", + "--name", "Fixed-start schedules", + "--cron", "0 * * * *", + "--type", "schedules", + "--parameters", str(parameters_file), + ], + ) # fmt: skip + assert "Successfully created" in result.output, result.output + assert "each run will compute the same period" in result.output + + +def test_run_schedule_automation_dispatch(app, fresh_db, setup_dummy_data, monkeypatch): + """Running a schedules automation queues a scheduling job with trigger meta data. + + We monkeypatch the job creator to avoid needing a fully schedulable asset here. + """ + from flexmeasures.data.models.generic_assets import GenericAsset + from flexmeasures.data.services import scheduling + from flexmeasures.data.services.automations import run_automation + from flexmeasures.utils.time_utils import server_now + + asset = fresh_db.session.get(GenericAsset, 1) + automation = Automation( + asset_id=asset.id, + type="schedules", + name="Test schedules", + cronstr="0 * * * *", + parameters={"duration": "PT12H", "resolution": "PT15M"}, + ) + fresh_db.session.add(automation) + fresh_db.session.flush() + + calls = {} + + def fake_create_simultaneous_scheduling_job(asset, **kwargs): + calls["asset"] = asset + calls["kwargs"] = kwargs + + class FakeJob: + id = "fake-job-id" + + return FakeJob() + + monkeypatch.setattr( + scheduling, + "create_simultaneous_scheduling_job", + fake_create_simultaneous_scheduling_job, + ) + + returns = run_automation(automation) + assert returns == {"job_id": "fake-job-id", "n_jobs": 1} + assert calls["asset"].id == asset.id + assert calls["kwargs"]["trigger"] == { + "origin": "automation", + "automation_id": automation.id, + } + # start defaulted to (roughly) now, floored to the 15-minute resolution + start = calls["kwargs"]["start"] + assert start.minute % 15 == 0 + assert abs((server_now() - start).total_seconds()) < 16 * 60 + assert calls["kwargs"]["end"] - start == timedelta(hours=12) + + 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/cli/utils.py b/flexmeasures/cli/utils.py index f2a2ba10bb..4e7b2333d5 100644 --- a/flexmeasures/cli/utils.py +++ b/flexmeasures/cli/utils.py @@ -447,6 +447,23 @@ def split_commas(ctx, param, value): return list(set([x.strip() for x in result if x.strip()])) +def make_cli_options_optional(*option_names): + """Decorator to relax the required flag on the named CLI options. + + Useful when schema-derived options (see add_cli_options_from_schema) are + only conditionally required; apply above the decorator that adds them, + and enforce requiredness in the command body (e.g. by schema validation). + """ + + def decorator(command): + for param in getattr(command, "__click_params__", []): + if param.name in option_names: + param.required = False + return command + + return decorator + + def add_cli_options_from_schema(schema): """Decorator to add CLI options based on a Marshmallow schema's fields.""" diff --git a/flexmeasures/data/models/automations.py b/flexmeasures/data/models/automations.py index dd9fc723b9..e99e621165 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"] # later also "schedules" and "reports" + SUPPORTED_TYPES = ["forecasts", "schedules"] # later also "reports" id = db.Column(db.Integer, autoincrement=True, primary_key=True) created_at = db.Column( diff --git a/flexmeasures/data/services/automations.py b/flexmeasures/data/services/automations.py index e7aaf5115a..af9ea6db55 100644 --- a/flexmeasures/data/services/automations.py +++ b/flexmeasures/data/services/automations.py @@ -9,6 +9,8 @@ from cron_descriptor import get_description, Options from croniter import croniter +import isodate +import pandas as pd from sqlalchemy import select from flexmeasures import Forecaster @@ -54,6 +56,29 @@ def get_due_automations(now: datetime | None = None) -> list[Automation]: ] +def prepare_schedule_trigger_message(parameters: dict, asset_id: int) -> dict: + """Complete stored schedule parameters into a message for the AssetTriggerSchema. + + The asset id is injected, and the (required) schedule start defaults to now, + floored to the message's resolution (if given, otherwise to the minute), + so recurring automations produce fresh schedules on each run. + """ + message = dict(parameters) + message["id"] = asset_id + if "start" not in message: + start = floor_to_minute(server_now()) + if message.get("resolution") is not None: + try: + resolution = isodate.parse_duration(message["resolution"]) + start = ( + pd.Timestamp(start).floor(pd.Timedelta(resolution)).to_pydatetime() + ) + except Exception: + pass # leave start floored to the minute + message["start"] = start.isoformat() + return message + + def get_automation_job_stats(automation: Automation) -> dict[str, int]: """Count the jobs created by this automation, per job status. @@ -61,21 +86,30 @@ def get_automation_job_stats(automation: Automation) -> dict[str, int]: """ from flask import current_app - # 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 + # Determine the job cache entries to scan. + 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 + ] + 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 + cache_refs = [(sensor_id, "forecasting", "sensor") for sensor_id in sensor_ids] counts: dict[str, int] = {} seen_job_ids: set[str] = set() - for sensor_id in sensor_ids: - for job in current_app.job_cache.get(sensor_id, "forecasting", "sensor"): + for entity_id, queue, asset_or_sensor_type in cache_refs: + for job in current_app.job_cache.get(entity_id, queue, asset_or_sensor_type): if job.id in seen_job_ids: continue seen_job_ids.add(job.id) @@ -88,13 +122,18 @@ def get_automation_job_stats(automation: Automation) -> dict[str, int]: def run_automation(automation: Automation) -> dict[str, Any] | None: """Queue the jobs for one run of an automation. - :returns: the data generator's return value, e.g. {"job_id": , "n_jobs": } - for forecasting jobs. + :returns: a dict like {"job_id": , "n_jobs": }. """ - if automation.type != "forecasts": - raise NotImplementedError( - f"Automations of type '{automation.type}' cannot be run yet." - ) + if automation.type == "forecasts": + return _run_forecast_automation(automation) + elif automation.type == "schedules": + return _run_schedule_automation(automation) + raise NotImplementedError( + f"Automations of type '{automation.type}' cannot be run yet." + ) + + +def _run_forecast_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)." @@ -109,3 +148,37 @@ def run_automation(automation: Automation) -> dict[str, Any] | None: forecaster._parameters = None forecaster.set_job_trigger("automation", automation_id=automation.id) return forecaster.compute(as_job=True, parameters=dict(automation.parameters)) + + +def _run_schedule_automation(automation: Automation) -> dict[str, Any]: + from flexmeasures.data.schemas.scheduling import AssetTriggerSchema + from flexmeasures.data.services.scheduling import ( + create_sequential_scheduling_job, + create_simultaneous_scheduling_job, + ) + + message = prepare_schedule_trigger_message( + dict(automation.parameters), automation.asset_id + ) + trigger_data = AssetTriggerSchema().load(message) + start = trigger_data["start_of_schedule"] + scheduler_kwargs = dict( + start=start, + end=start + trigger_data["duration"], + belief_time=trigger_data.get("belief_time"), # server time if not set + resolution=trigger_data.get("resolution"), + flex_model=trigger_data["flex_model"], + flex_context=trigger_data["flex_context"], + ) + if trigger_data["sequential"]: + f = create_sequential_scheduling_job + else: + f = create_simultaneous_scheduling_job + job = f( + asset=trigger_data["asset"], + enqueue=True, + force_new_job_creation=trigger_data.get("force_new_job_creation", False), + trigger={"origin": "automation", "automation_id": automation.id}, + **scheduler_kwargs, + ) + return {"job_id": job.id, "n_jobs": 1} diff --git a/flexmeasures/data/tests/test_automations.py b/flexmeasures/data/tests/test_automations.py new file mode 100644 index 0000000000..7e1a60a60f --- /dev/null +++ b/flexmeasures/data/tests/test_automations.py @@ -0,0 +1,46 @@ +"""Integration tests for running automations (service level).""" + +from __future__ import annotations + +import pytest +from rq.job import Job + +from flexmeasures.api.v3_0.tests.utils import message_for_trigger_schedule +from flexmeasures.data.models.automations import Automation +from flexmeasures.data.services.automations import run_automation + + +@pytest.fixture(scope="function") +def keep_scheduling_queue_empty(app): + app.queues["scheduling"].empty() + yield + app.queues["scheduling"].empty() + + +def test_run_schedule_automation( + db, app, add_battery_assets, add_market_prices, keep_scheduling_queue_empty +): + """A schedules automation queues a scheduling job carrying trigger meta data.""" + battery = add_battery_assets["Test battery"] + message = message_for_trigger_schedule() + flex_model = message.pop("flex-model") + flex_model["sensor"] = battery.sensors[0].id + + automation = Automation( + asset_id=battery.id, + type="schedules", + name="Nightly schedules", + cronstr="0 0 * * *", + parameters={**message, "flex-model": [flex_model]}, + ) + db.session.add(automation) + db.session.flush() + + returns = run_automation(automation) + assert returns["n_jobs"] == 1 + + job = Job.fetch(returns["job_id"], connection=app.queues["scheduling"].connection) + assert job.meta["trigger"] == { + "origin": "automation", + "automation_id": automation.id, + } diff --git a/flexmeasures/ui/templates/assets/asset_automations.html b/flexmeasures/ui/templates/assets/asset_automations.html index 90b7f95f69..523ae3e2b6 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 }} @@ -30,7 +30,7 @@