Schedules as automations

documentation/changelog.rst

@@ -13,6 +13,7 @@ v1.0.0 | July XX, 2026
 New features
 -------------
 * Automations - first roundtrip for forecasts: recurring tasks defined per asset, managed with new CLI commands (``flexmeasures add|edit|delete automation``), run by ``flexmeasures jobs run-automations``, and viewable in a new UI page and API endpoints (``[GET] /assets/(id)/automations``); jobs now also record whether they were created via the CLI, the API or an automation [see `PR #2290 <https://www.github.com/FlexMeasures/flexmeasures/pull/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 <https://www.github.com/FlexMeasures/flexmeasures/pull/2293>`_]
 * 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 <https://www.github.com/FlexMeasures/flexmeasures/pull/2172>`_]
 * In the UI, asset and sensor lists can be filtered by ID prefix through API-backed search fields [see `PR #2231 <https://www.github.com/FlexMeasures/flexmeasures/pull/2231>`_]
 * Support configurable lower and upper bounds and snapping for forecast post-processing [see `PR #2273 <https://www.github.com/FlexMeasures/flexmeasures/pull/2273>`_]

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

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.
 ================================================= =======================================
 
 

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`.

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

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:

flexmeasures/cli/data_add.py

@@ -37,13 +37,15 @@
     DeprecatedOption,
     DeprecatedOptionsCommand,
     add_cli_options_from_schema,
+    make_cli_options_optional,
 )
 from flexmeasures.data import db
 from flexmeasures.data.scripts.data_gen import (
     add_transmission_zone_asset,
     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,59 +1291,80 @@ 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,
         type=automation_type,
         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(

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

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.
 

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."""
 

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(