diff --git a/documentation/changelog.rst b/documentation/changelog.rst index 3f3af3aa18..c00ca7c53f 100644 --- a/documentation/changelog.rst +++ b/documentation/changelog.rst @@ -33,6 +33,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``); an automation's details link to the sensors it reads from and writes to, and a sensor's page lists the automations feeding it; jobs now also record whether they were created via the CLI, the API or an automation [see `PR #2290 `_] * In the UI, the full record of the data source selected on a sensor page can be inspected, backed by a new API endpoint (``[GET] /sources/(id)``) [see `PR #2290 `_] * Let forecast automations use their own timezone and catch up only the latest missed occurrence after downtime, with skipped daylight-saving times handled once and repeated wall-clock times not duplicated [see `PR #2396 `_] +* 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 `_] * ``flexmeasures show data-sources`` now shows which organisation a data source belongs to, and can list the sensors holding data recorded by a given source [see `PR #2401 `_] * New ``inflexible-consumption`` and ``inflexible-production`` flex-context fields make explicit how the sign of each inflexible device's power data should be read (positive values denote consumption resp. production), accepting sensor references with optional source filters; they replace the now-deprecated ``inflexible-device-sensors`` field (bare sensor IDs, sign read from each sensor's ``consumption_is_positive`` attribute), which remains supported [see `PR #2358 `_] * An inflexible (unschedulable) device can be modelled as its own asset by giving its flex-model entry a single ``inflexible-consumption`` or ``inflexible-production`` sensor reference; such a device joins a ``group`` like any other member, so its fixed (measured) load counts towards the group's intermediate power constraint [see `PR #2374 `_] diff --git a/documentation/cli/change_log.rst b/documentation/cli/change_log.rst index b3ac7245a3..4cb14ec6c6 100644 --- a/documentation/cli/change_log.rst +++ b/documentation/cli/change_log.rst @@ -12,7 +12,7 @@ since v1.0.0 | July XX, 2026 * Add ``flexmeasures add plan``, ``flexmeasures show plans`` and ``flexmeasures edit plan``, to manage the rate limits and quotas which apply to the accounts on a plan. * Add ``flexmeasures edit secret`` to store an encrypted secret on an account or asset. * Add ``flexmeasures delete secret`` to remove an encrypted secret from an account or asset. -* Add ``flexmeasures add automation``, ``flexmeasures edit automation`` and ``flexmeasures delete automation`` to manage automations (recurring tasks on an asset; for now, computing forecasts). +* 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 from standard five-field cron expressions. Run this command once per minute. It makes at most one queueing attempt per automation per minute, including when an attempt fails after partially queueing jobs. * Add ``--timezone`` to ``flexmeasures add automation`` and ``flexmeasures edit automation``. ``flexmeasures jobs run-automations`` now persists scheduling progress, catches up only the latest missed forecast occurrence, and handles skipped or repeated daylight-saving-time occurrences once. Failed or partially completed queueing attempts are still not retried automatically. * ``flexmeasures show data-sources`` now shows the account a data source belongs to, and lists the sensors holding data recorded by a single source with ``--show-sensors``. diff --git a/documentation/cli/commands.rst b/documentation/cli/commands.rst index 35160b64a1..bb4b0c525c 100644 --- a/documentation/cli/commands.rst +++ b/documentation/cli/commands.rst @@ -41,7 +41,7 @@ of which some are referred to in this documentation. ``flexmeasures add annotation`` Add annotation to accounts, assets and/or sensors. ``flexmeasures add toy-account`` Create a toy account, for tutorials and trying things. ``flexmeasures add report`` Create a report. -``flexmeasures add automation`` Add a forecast automation with its own cron timezone. +``flexmeasures add automation`` Add an automation: a recurring task (computing forecasts or schedules) on an asset, with its own cron timezone. ================================================= ======================================= diff --git a/documentation/features/forecasting.rst b/documentation/features/forecasting.rst index b977bf8ed2..9be296a695 100644 --- a/documentation/features/forecasting.rst +++ b/documentation/features/forecasting.rst @@ -277,3 +277,5 @@ The jobs record how they were created, which is shown on the asset's status page Automations defined on an asset can be viewed on the asset's *Automations* page in the UI, and listed with the API endpoint `[GET] /assets/(id)/automations <../api/v3_0.html#get--api-v3_0-assets-id-automations>`_. An automation's details show the sensors it reads from and writes to, linking to each sensor's page. Conversely, a sensor's page lists the automations that write data to it. + +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 f05cd5b766..6d599df855 100644 --- a/documentation/features/scheduling.rst +++ b/documentation/features/scheduling.rst @@ -368,3 +368,26 @@ Here are some thoughts on further innovation: This is ongoing architecture design work, and therefore happens in development settings, until we are happy with the outcomes. Thoughts welcome :) - 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#post--api-v3_0-assets-id-schedules-trigger>`_ API endpoint (without the asset id). +Use the canonical API field names, including ``flex-model``, ``flex-context`` and ``force-new-job-creation``. + +Omit the ``start`` field to calculate it afresh from the server time on each run. +It is floored to the fixed, positive ``resolution`` when given, or otherwise to the minute. +A fixed ``start`` is accepted, but every run then schedules the same period and the CLI warns about this when creating the automation. +The ``duration`` must be positive; ``resolution`` does not accept nominal durations such as a month. +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/api/v3_0/assets.py b/flexmeasures/api/v3_0/assets.py index cf8fd0466a..273729d80d 100644 --- a/flexmeasures/api/v3_0/assets.py +++ b/flexmeasures/api/v3_0/assets.py @@ -1389,7 +1389,7 @@ def get_automations(self, id: int, asset: GenericAsset): get: summary: Get all automations defined on an asset. description: | - The response will be a list of automations: recurring tasks (for now, computing forecasts) + The response will be a list of automations: recurring forecasting or scheduling tasks defined on the asset. Each entry shows the automation's ID, when it was created, its type, name, activation status, and its recurrence, both as a cron string and described in natural language. Each entry also shows the IANA timezone in which its cron expression is interpreted and its persistent scheduling cursor. @@ -1461,8 +1461,8 @@ def get_automation(self, id: int, automation_id: int, asset: GenericAsset): summary: Get details of one automation defined on an asset. description: | In addition to the fields shown when listing automations, the response shows - the automation's parameters (for forecasts, these are the forecast parameters - used on each run), information about the data generator that runs it, + the automation's parameters (forecast parameters or a schedule trigger message), + information about its data generator (null for schedule automations), the sensors it reads from and writes to, and counts of recently created jobs, per job status. Note that jobs in Redis have a limited TTL, so not all past jobs will be counted. @@ -1979,10 +1979,11 @@ def trigger_schedule( start=start_of_schedule, end=end_of_schedule, belief_time=belief_time, # server time if no prior time was sent - resolution=resolution, flex_model=flex_model, flex_context=flex_context, ) + if resolution is not None: + scheduler_kwargs["resolution"] = resolution if sequential: f = create_sequential_scheduling_job else: @@ -1992,6 +1993,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/api/v3_0/sensors.py b/flexmeasures/api/v3_0/sensors.py index f9badc6003..06a403c4a6 100644 --- a/flexmeasures/api/v3_0/sensors.py +++ b/flexmeasures/api/v3_0/sensors.py @@ -1075,6 +1075,7 @@ def trigger_schedule( **scheduler_kwargs, enqueue=True, force_new_job_creation=force_new_job_creation, + trigger={"origin": "API"}, ) except ValidationError as err: return unprocessable_entity(err.messages) diff --git a/flexmeasures/api/v3_0/tests/test_asset_schedules_fresh_db.py b/flexmeasures/api/v3_0/tests/test_asset_schedules_fresh_db.py index 8100b4af8a..8603fbc5fe 100644 --- a/flexmeasures/api/v3_0/tests/test_asset_schedules_fresh_db.py +++ b/flexmeasures/api/v3_0/tests/test_asset_schedules_fresh_db.py @@ -121,6 +121,10 @@ def test_asset_trigger_and_get_schedule( ), "the whole scheduling job is handled as a single job (simultaneous scheduling)" done_job_id = scheduled_jobs[0].id scheduling_job = scheduled_jobs[0] + assert all( + job.meta["trigger"] == {"origin": "API"} + for job in [*scheduled_jobs, *deferred_jobs] + ) print(scheduling_job.kwargs) if sequential: diff --git a/flexmeasures/api/v3_0/tests/test_automations_api_fresh_db.py b/flexmeasures/api/v3_0/tests/test_automations_api_fresh_db.py index cc6d033c6a..8f12a989fb 100644 --- a/flexmeasures/api/v3_0/tests/test_automations_api_fresh_db.py +++ b/flexmeasures/api/v3_0/tests/test_automations_api_fresh_db.py @@ -67,3 +67,66 @@ def test_details_reject_inaccessible_sensor_metadata( assert response.status_code == 403 assert hidden_sensor.name not in response.text + + +@pytest.mark.parametrize( + "requesting_user", ["test_prosumer_user@seita.nl"], indirect=True +) +def test_schedule_details_include_stored_flex_sensors( + client, + fresh_db, + setup_roles_users_fresh_db, + setup_generic_assets_fresh_db, + requesting_user, +): + asset = setup_generic_assets_fresh_db["test_battery"] + power_sensor = Sensor( + name="scheduled power", + unit="MW", + event_resolution=timedelta(minutes=15), + generic_asset=asset, + ) + price_sensor = Sensor( + name="schedule price", + unit="EUR/MWh", + event_resolution=timedelta(hours=1), + generic_asset=asset, + ) + fresh_db.session.add_all([power_sensor, price_sensor]) + fresh_db.session.flush() + asset.flex_model = { + "consumption": {"sensor": power_sensor.id}, + "soc-at-start": "2.5 MWh", + "soc-min": "0 MWh", + "soc-max": "5 MWh", + "power-capacity": "2 MW", + } + asset.flex_context = { + "site-power-capacity": "2 MVA", + "consumption-price": {"sensor": price_sensor.id}, + } + automation = Automation( + asset=asset, + type="schedules", + name="Minimal schedule details", + cronstr="0 6 * * *", + parameters={"duration": "PT1H"}, + ) + fresh_db.session.add(automation) + fresh_db.session.commit() + + response = client.get( + url_for( + "AssetAPI:get_automation", + id=asset.id, + automation_id=automation.id, + ) + ) + + assert response.status_code == 200 + assert response.json["input_sensors"] == [ + {"id": price_sensor.id, "name": price_sensor.name} + ] + assert response.json["output_sensors"] == [ + {"id": power_sensor.id, "name": power_sensor.name} + ] diff --git a/flexmeasures/api/v3_0/tests/test_sensor_schedules_fresh_db.py b/flexmeasures/api/v3_0/tests/test_sensor_schedules_fresh_db.py index 8204d1c74f..903854c1d6 100644 --- a/flexmeasures/api/v3_0/tests/test_sensor_schedules_fresh_db.py +++ b/flexmeasures/api/v3_0/tests/test_sensor_schedules_fresh_db.py @@ -90,6 +90,7 @@ def test_trigger_and_get_schedule( len(app.queues["scheduling"]) == 1 ) # only 1 schedule should be made for 1 asset job = app.queues["scheduling"].jobs[0] + assert job.meta["trigger"] == {"origin": "API"} print(job.kwargs) assert job.kwargs["asset_or_sensor"]["id"] == sensor.id assert job.kwargs["start"] == parse_datetime(message["start"]) diff --git a/flexmeasures/cli/data_add.py b/flexmeasures/cli/data_add.py index 40dfdb5bcf..d166a14fbd 100755 --- a/flexmeasures/cli/data_add.py +++ b/flexmeasures/cli/data_add.py @@ -52,6 +52,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, @@ -1383,7 +1384,12 @@ def _normalize_yaml_value(value): def _load_yaml_mapping(stream: TextIOBase, option_name: str) -> dict: """Load a YAML/JSON CLI option file whose top level must be an object.""" - value = yaml.safe_load(stream) + try: + value = yaml.safe_load(stream) + except yaml.YAMLError as exc: + raise click.UsageError( + f"The {option_name} file is not valid YAML or JSON." + ) from exc if value is None: return {} if not isinstance(value, dict): @@ -1745,7 +1751,8 @@ 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.", ) @add_cli_options_from_schema( ForecasterParametersSchema(), hidden=True, force_optional=True @@ -1767,18 +1774,22 @@ 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 * * *" --timezone Europe/Amsterdam --parameters forecast-parameters.yml + flexmeasures add automation --asset 3 --name "Hourly schedules" + --cron "0 * * * *" --type schedules --parameters trigger-message.yml - The forecaster configuration is stored on a data source, and the forecast - parameters are validated and stored on the automation itself. - Each time the automation runs, forecasting jobs are queued - (see `flexmeasures jobs run-automations`). + 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`). Alternatively, pass an existing data source (--source) to reuse the forecaster and configuration stored on it. @@ -1786,6 +1797,7 @@ def add_automation( Every forecaster and pipeline option that `flexmeasures add forecast` accepts is accepted here, too, but is left out of the help text above to keep it focused on the automation itself; run `flexmeasures add forecast --help` to see them. + They only apply to forecast automations. A configuration option given on the command line overrides the same setting from --config, while a parameter from --parameters takes precedence over the matching command-line option. """ @@ -1796,37 +1808,72 @@ def add_automation( kwargs, source, config_file, parameters_file ) + if automation_type == "schedules": + # Only options actually given on the command line count: the forecaster and the + # configuration options that were left out still show up here, with their defaults. + forecast_options = _find_options_given_on_command_line( + { + "forecaster_class": "--forecaster", + "source": "--source", + "config_file": "--config", + "edit_config": "--edit-config", + }, + TrainPredictPipelineConfigSchema(), + ) + if forecast_options: + raise click.UsageError( + f"{flexmeasures_inflection.join_words_into_a_list(forecast_options)} cannot be" + " combined with --type schedules: a schedule automation is not computed by a forecaster." + ) + # Validate the parameters using the forecast parameters schema (we store them serialized) - try: - deserialized_parameters = ForecasterParametersSchema().load(parameters) - except ValidationError as e: - click.secho(f"Invalid forecast parameters: {e.messages}", **MsgStyle.ERROR) - raise click.Abort() - output_sensor = deserialized_parameters.get( - "sensor_to_save" - ) or deserialized_parameters.get("sensor") - try: - validate_forecast_output_scope(asset.id, output_sensor) - except ValueError as exc: - click.secho(str(exc), **MsgStyle.ERROR) - raise click.Abort() + 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() + output_sensor = deserialized_parameters.get( + "sensor_to_save" + ) or deserialized_parameters.get("sensor") + try: + validate_forecast_output_scope(asset.id, output_sensor) + except ValueError as exc: + click.secho(str(exc), **MsgStyle.ERROR) + raise click.Abort() - forecaster = get_data_generator( - source=source, - model=forecaster_class, - config=config, - save_config=True, - data_generator_type=Forecaster, - ) - if forecaster is None: - click.secho( - f"Could not set up forecaster '{forecaster_class}'.", **MsgStyle.ERROR + 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, @@ -1835,7 +1882,7 @@ def add_automation( cronstr=cronstr, timezone=timezone, active=not inactive, - generator_id=generator.id, + generator_id=generator_id, parameters=parameters, ) db.session.add(automation) @@ -2011,7 +2058,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 ceb3116135..63a2b06d68 100644 --- a/flexmeasures/cli/jobs.py +++ b/flexmeasures/cli/jobs.py @@ -108,8 +108,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 2b647a6a50..f75bcfc7c3 100644 --- a/flexmeasures/cli/tests/test_automations.py +++ b/flexmeasures/cli/tests/test_automations.py @@ -2,6 +2,7 @@ import json import pytest +import pytz from types import SimpleNamespace from sqlalchemy import select @@ -700,6 +701,273 @@ def test_add_automation_rejects_non_object_yaml_file( assert "Traceback" not in result.output +@pytest.mark.parametrize("option_name", ("--config", "--parameters")) +def test_add_automation_rejects_malformed_yaml_file( + app, fresh_db, setup_dummy_data, tmp_path, option_name +): + from flexmeasures.cli.data_add import add_automation + + malformed_file = tmp_path / "malformed.yaml" + malformed_file.write_text("field: [\n") + result = app.test_cli_runner().invoke( + add_automation, + [ + "--asset", + "1", + "--name", + "Malformed YAML", + "--cron", + "0 6 * * *", + option_name, + str(malformed_file), + "--sensor", + str(setup_dummy_data[0]), + ], + ) + + assert result.exit_code == 2, result.output + assert f"The {option_name} file is not valid YAML or JSON" in result.output + assert "Traceback" not 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 + + +@pytest.mark.parametrize( + "parameters_yaml", + ( + 'resolution: "P1M"\n', + 'resolution: "PT0S"\n', + 'resolution: "-PT15M"\n', + 'duration: "PT0S"\n', + 'duration: "-PT1H"\n', + ), +) +def test_add_schedule_automation_rejects_unsupported_durations( + app, fresh_db, setup_dummy_data, tmp_path, parameters_yaml +): + from flexmeasures.cli.data_add import add_automation + + parameters_file = tmp_path / "parameters.yml" + parameters_file.write_text(parameters_yaml) + result = app.test_cli_runner().invoke( + add_automation, + [ + "--asset", + "1", + "--name", + "Invalid schedule durations", + "--cron", + "0 * * * *", + "--type", + "schedules", + "--parameters", + str(parameters_file), + ], + ) + + assert result.exit_code != 0 + assert "Invalid schedule parameters" in result.output + + +def test_add_schedule_automation_rejects_forecast_config( + app, fresh_db, setup_dummy_data +): + from flexmeasures.cli.data_add import add_automation + + result = app.test_cli_runner().invoke( + add_automation, + [ + "--asset", + "1", + "--name", + "Schedule with ignored forecast config", + "--cron", + "0 * * * *", + "--type", + "schedules", + "--regressors", + str(setup_dummy_data[0]), + ], + ) + + assert result.exit_code == 2 + assert "--regressors cannot be combined with --type schedules" in result.output + assert "Traceback" not in result.output + + +def test_add_schedule_automation_rejects_the_default_forecaster_when_given( + app, fresh_db, setup_dummy_data +): + """Naming the default forecaster is still naming a forecaster, so it is refused. + + The check asks whether the option was given, rather than comparing its value against the default, + which would let the default pass silently and leave the user thinking it applied. + """ + from flexmeasures.cli.data_add import add_automation + + result = app.test_cli_runner().invoke( + add_automation, + [ + "--asset", + "1", + "--name", + "Schedule naming the default forecaster", + "--cron", + "0 * * * *", + "--type", + "schedules", + "--forecaster", + "TrainPredictPipeline", + ], + ) + + assert result.exit_code == 2 + assert "--forecaster cannot be combined with --type schedules" in result.output + assert ( + fresh_db.session.execute( + select(Automation).filter_by(name="Schedule naming the default forecaster") + ).scalar_one_or_none() + is None + ) + + +def test_add_forecast_automation_still_requires_sensor(app, fresh_db, setup_dummy_data): + from flexmeasures.cli.data_add import add_automation + + result = app.test_cli_runner().invoke( + add_automation, + ["--asset", "1", "--name", "No sensor", "--cron", "0 * * * *"], + ) + + assert result.exit_code != 0 + assert "Invalid forecast parameters" in result.output + + +@pytest.mark.parametrize("is_dst", (True, False)) +def test_prepare_schedule_start_floors_both_dst_folds(app, monkeypatch, is_dst): + from flexmeasures.data.services import automations + + timezone = pytz.timezone("Europe/Amsterdam") + now = timezone.localize(datetime(2026, 10, 25, 2, 7, 30), is_dst=is_dst) + monkeypatch.setattr(automations, "server_now", lambda: now) + parameters = {"duration": "PT1H", "resolution": "PT15M"} + + message = automations.prepare_schedule_trigger_message(parameters, asset_id=1) + + assert datetime.fromisoformat(message["start"]) == now.replace( + minute=0, second=0, microsecond=0 + ) + assert parameters == {"duration": "PT1H", "resolution": "PT15M"} + + +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/data/migrations/versions/5a9c0e3b7d21_allow_schedule_automations_without_generator.py b/flexmeasures/data/migrations/versions/5a9c0e3b7d21_allow_schedule_automations_without_generator.py new file mode 100644 index 0000000000..1b728361ce --- /dev/null +++ b/flexmeasures/data/migrations/versions/5a9c0e3b7d21_allow_schedule_automations_without_generator.py @@ -0,0 +1,29 @@ +"""Allow schedule automations without a generator. + +Revision ID: 5a9c0e3b7d21 +Revises: 4d5e6f708192 +Create Date: 2026-08-05 12:15:00.000000 + +""" + +from alembic import op + +# revision identifiers, used by Alembic. +revision = "5a9c0e3b7d21" +down_revision = "4d5e6f708192" +branch_labels = None +depends_on = None + + +def upgrade(): + op.alter_column("automation", "generator_id", nullable=True) + op.create_check_constraint( + "forecast_generator", + "automation", + "type != 'forecasts' OR generator_id IS NOT NULL", + ) + + +def downgrade(): + op.drop_constraint("forecast_generator", "automation", type_="check") + op.alter_column("automation", "generator_id", nullable=False) diff --git a/flexmeasures/data/migrations/versions/c63896a97a8e_merge_the_automation_timezone_and_.py b/flexmeasures/data/migrations/versions/c63896a97a8e_merge_the_automation_timezone_and_.py new file mode 100644 index 0000000000..9afd0fd28e --- /dev/null +++ b/flexmeasures/data/migrations/versions/c63896a97a8e_merge_the_automation_timezone_and_.py @@ -0,0 +1,25 @@ +"""merge the automation timezone and schedule generator migrations + +Two migrations branched off the same revision: one adding an automation's timezone and scheduling cursor, +the other allowing a schedule automation to exist without a data generator. +They touch different columns, so this merge only rejoins them and has nothing of its own to do. + +Revision ID: c63896a97a8e +Revises: 5a9c0e3b7d21, 9f2b6e1d4a73 +Create Date: 2026-08-11 01:06:28.121631 + +""" + +# revision identifiers, used by Alembic. +revision = "c63896a97a8e" +down_revision = ("5a9c0e3b7d21", "9f2b6e1d4a73") +branch_labels = None +depends_on = None + + +def upgrade(): + pass + + +def downgrade(): + pass diff --git a/flexmeasures/data/models/automations.py b/flexmeasures/data/models/automations.py index 87c699461c..fd13bc24dc 100644 --- a/flexmeasures/data/models/automations.py +++ b/flexmeasures/data/models/automations.py @@ -1,6 +1,4 @@ -""" -Automations: recurring tasks (for now: forecasting) defined per asset. -""" +"""Automations: recurring forecasting or scheduling tasks defined per asset.""" from __future__ import annotations @@ -35,14 +33,20 @@ def get_initial_scheduling_cursor() -> datetime: class Automation(db.Model, AuthModelMixin): """A recurring task on an asset, such as computing forecasts. - The recurrence is defined by a cron string, and the work to be done is defined - by a data generator (e.g. a forecaster, linked through a data source) together - with the parameters to call it with. + The recurrence is defined by a cron string. Forecast automations use a data + generator (e.g. a forecaster linked through a data source), while schedule + automations use only their stored parameters. """ __tablename__ = "automation" + __table_args__ = ( + db.CheckConstraint( + "type != 'forecasts' OR generator_id IS NOT NULL", + name="forecast_generator", + ), + ) - 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( @@ -65,9 +69,7 @@ class Automation(db.Model, AuthModelMixin): default=get_initial_scheduling_cursor, ) active = db.Column(db.Boolean, nullable=False, default=True) - generator_id = db.Column( - db.Integer, db.ForeignKey("data_source.id"), nullable=False - ) + generator_id = db.Column(db.Integer, db.ForeignKey("data_source.id"), nullable=True) parameters = db.Column(MutableDict.as_mutable(JSONB), nullable=False, default={}) asset = db.relationship( diff --git a/flexmeasures/data/models/planning/__init__.py b/flexmeasures/data/models/planning/__init__.py index 6aebfd65d0..e32f75564f 100644 --- a/flexmeasures/data/models/planning/__init__.py +++ b/flexmeasures/data/models/planning/__init__.py @@ -318,7 +318,11 @@ def collect_flex_config(self): ] combined_flex_model = amended_db_flex_model + amended_flex_model # For the single-asset case, revert the flex-model listification - if len(combined_flex_model) == 1 and "sensor" not in combined_flex_model[0]: + if ( + len(combined_flex_model) == 1 + and "sensor" not in combined_flex_model[0] + and self.asset is None + ): # Single-asset case self.flex_model = combined_flex_model[0] else: diff --git a/flexmeasures/data/schemas/scheduling/__init__.py b/flexmeasures/data/schemas/scheduling/__init__.py index cd0a4e942d..99b0f9e4fe 100644 --- a/flexmeasures/data/schemas/scheduling/__init__.py +++ b/flexmeasures/data/schemas/scheduling/__init__.py @@ -1831,6 +1831,26 @@ class AssetTriggerSchema(Schema): ), ) + @validates_schema + def validate_schedule_durations(self, data, **kwargs): + """Require a positive horizon and a positive fixed resolution.""" + start = data["start_of_schedule"] + duration = DurationField.ground_from(data["duration"], start) + if duration <= timedelta(0): + raise ValidationError( + "Schedule duration must be positive.", field_name="duration" + ) + data["duration"] = duration + + resolution = data.get("resolution") + if resolution is not None and ( + not isinstance(resolution, timedelta) or resolution <= timedelta(0) + ): + raise ValidationError( + "Schedule resolution must be a positive, fixed duration.", + field_name="resolution", + ) + @pre_load def normalize_flex_context_format(self, data, **kwargs): """Normalize flex_context to always be a dict. diff --git a/flexmeasures/data/services/automations.py b/flexmeasures/data/services/automations.py index bd4dc4d943..84ba9a9f9b 100644 --- a/flexmeasures/data/services/automations.py +++ b/flexmeasures/data/services/automations.py @@ -13,6 +13,8 @@ from cron_descriptor import get_description, Options from croniter import croniter from croniter.croniter import CroniterError +import isodate +from isodate.isoerror import ISO8601Error from flask import current_app from marshmallow import ValidationError from sqlalchemy import select, update @@ -36,6 +38,69 @@ class DueAutomation: expected_timezone: str +# Fields naming sensors on which a scheduler records generated schedules. +OUTPUT_SENSOR_FIELDS = ( + "consumption", + "production", + "state-of-charge", + "state_of_charge", + "aggregate-consumption", + "aggregate_consumption", + "aggregate-production", + "aggregate_production", +) + + +def collect_sensors( + value: Any, + sensors: dict[int, Sensor] | None = None, + only_under_output_field: bool = False, + _under_output_field: bool = False, +) -> list[Sensor]: + """Collect sensor objects and references from a nested scheduling structure.""" + if sensors is None: + sensors = {} + + def collect(sensor: Sensor | None): + if sensor is not None and (_under_output_field or not only_under_output_field): + sensors[sensor.id] = sensor + + if isinstance(value, Sensor): + collect(value) + elif isinstance(value, dict): + for key, item in value.items(): + under_output_field = _under_output_field or key in OUTPUT_SENSOR_FIELDS + if key == "sensor" and isinstance(item, (int, str)): + if str(item).isdigit(): + sensor = db.session.get(Sensor, int(item)) + if sensor is not None and ( + under_output_field or not only_under_output_field + ): + sensors[sensor.id] = sensor + else: + collect_sensors( + item, sensors, only_under_output_field, under_output_field + ) + elif isinstance(value, (list, tuple, set)): + for item in value: + collect_sensors(item, sensors, only_under_output_field, _under_output_field) + return list(sensors.values()) + + +def collect_schedule_output_sensors(message: dict) -> list[Sensor]: + """Collect sensors on which the prepared schedule trigger records results.""" + sensors: dict[int, Sensor] = {} + for device in message.get("flex_model") or []: + collect_sensors(device.get("sensor"), sensors) + collect_sensors( + device.get("sensor_flex_model", device), + sensors, + only_under_output_field=True, + ) + collect_sensors(message.get("flex_context"), sensors, only_under_output_field=True) + return list(sensors.values()) + + def describe_cronstr(cronstr: str) -> str: """Describe a cron string in natural language, e.g. "At 06:00". @@ -57,6 +122,13 @@ def floor_to_minute(dt: datetime) -> datetime: return dt.astimezone(timezone.utc).replace(second=0, microsecond=0) +def floor_to_resolution(dt: datetime, resolution: timedelta) -> datetime: + """Floor an aware datetime to a fixed resolution without losing its DST fold.""" + delta_seconds = resolution.total_seconds() + floored = dt.timestamp() - (dt.timestamp() % delta_seconds) + return datetime.fromtimestamp(floored, tz=dt.tzinfo) + + def _as_nominal_wall_time(dt: datetime) -> datetime: """Represent local wall-clock fields on a transition-free UTC timeline.""" return datetime(dt.year, dt.month, dt.day, dt.hour, dt.minute, tzinfo=timezone.utc) @@ -250,15 +322,70 @@ class AutomationSensorsUnknown(Exception): """ +def resolve_schedule_automation_sensors( + parameters: dict, asset_id: int +) -> dict[str, list[Sensor]]: + """Resolve the sensors declared by a prepared schedule trigger.""" + from flexmeasures.data.schemas.scheduling import AssetTriggerSchema + from flexmeasures.data.services.scheduling import find_scheduler_class + from flexmeasures.data.services.utils import get_scheduler_instance + + try: + trigger_data = AssetTriggerSchema().load( + prepare_schedule_trigger_message(parameters, asset_id) + ) + start = trigger_data["start_of_schedule"] + scheduler_params = { + "start": start, + "end": start + trigger_data["duration"], + "belief_time": trigger_data.get("belief_time"), + "resolution": trigger_data.get("resolution"), + "flex_model": trigger_data["flex_model"], + "flex_context": trigger_data["flex_context"], + } + scheduler_class = find_scheduler_class(trigger_data["asset"]) + scheduler = get_scheduler_instance( + scheduler_class=scheduler_class, + asset_or_sensor=trigger_data["asset"], + scheduler_params=scheduler_params, + ) + scheduler.collect_flex_config() + except (NotImplementedError, ValidationError, ValueError) as exc: + raise AutomationSensorsUnknown( + f"Could not determine the sensors of schedule automation on asset {asset_id}: {exc}" + ) from exc + + resolved_trigger = { + "flex_model": scheduler.flex_model, + "flex_context": scheduler.flex_context, + } + output_sensors = collect_schedule_output_sensors(resolved_trigger) + output_sensor_ids = {sensor.id for sensor in output_sensors} + input_sensors = [ + sensor + for sensor in collect_sensors(resolved_trigger) + if sensor.id not in output_sensor_ids + ] + return { + "input_sensors": input_sensors, + "output_sensors": output_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. - The sensors are derived from the data generator, configured with the automation's own parameters. - Raises `AutomationSensorsUnknown` if that cannot be done, e.g. because the automation has no data generator, + 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. + 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). Use this wherever the answer decides whether something is permitted; use `get_automation_sensors` for display. """ + if automation.type == "schedules": + return resolve_schedule_automation_sensors( + dict(automation.parameters or {}), automation.asset_id + ) if automation.generator is None: raise AutomationSensorsUnknown( f"Automation {automation.id} has no data generator, so the sensors it involves are unknown." @@ -319,6 +446,39 @@ def get_automations_feeding_sensor(sensor: Sensor) -> 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 = server_now() + if message.get("resolution") is not None: + try: + resolution = isodate.parse_duration(message["resolution"]) + except (ISO8601Error, TypeError) as exc: + raise ValidationError( + {"resolution": ["Not a valid ISO 8601 duration."]} + ) from exc + if not isinstance(resolution, timedelta) or resolution <= timedelta(0): + raise ValidationError( + { + "resolution": [ + "Schedule resolution must be a positive, fixed duration." + ] + } + ) + start = floor_to_resolution(start, resolution) + else: + start = floor_to_minute(start) + message["start"] = start.isoformat() + return message + + def _asset_and_ancestor_ids(asset_id: int | None) -> list[int]: """List the given asset and all of its ancestors, nearest first.""" from flexmeasures.data.models.generic_assets import GenericAsset @@ -338,21 +498,33 @@ def get_automation_job_stats(automation: Automation) -> dict[str, int]: Note that jobs in Redis have a limited TTL, so this only counts fairly recent jobs. """ - # 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). + assets = [automation.asset, *automation.asset.offspring] + cache_refs = [(automation.asset_id, "scheduling", "asset")] + [ + (sensor.id, "scheduling", "sensor") + for asset in assets + for sensor in 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) @@ -397,13 +569,18 @@ def validate_forecast_output_scope(asset_id: int, output_sensor: Sensor) -> None 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)." @@ -420,3 +597,39 @@ 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 + flex_model=trigger_data["flex_model"], + flex_context=trigger_data["flex_context"], + ) + if trigger_data.get("resolution") is not None: + scheduler_kwargs["resolution"] = trigger_data["resolution"] + 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, + ) + n_jobs = len(job.args[0]) + 1 if trigger_data["sequential"] else 1 + return {"job_id": job.id, "n_jobs": n_jobs} diff --git a/flexmeasures/data/services/scheduling.py b/flexmeasures/data/services/scheduling.py index d169924255..61dfbc0e1c 100644 --- a/flexmeasures/data/services/scheduling.py +++ b/flexmeasures/data/services/scheduling.py @@ -18,6 +18,7 @@ import click from flask import current_app from isodate import duration_isoformat +from marshmallow import ValidationError from rq import get_current_job, Callback from rq.exceptions import InvalidJobOperation from rq.job import Job @@ -189,6 +190,7 @@ def trigger_optional_fallback(job, connection, type, value, traceback): enqueue=False, scheduler_specs=scheduler_specs, success_callback=Callback(success_callback), + trigger=job.meta.get("trigger"), **scheduler_kwargs, ) @@ -201,6 +203,13 @@ def trigger_optional_fallback(job, connection, type, value, traceback): job.meta["fallback_job_id"] = fallback_job.id job.save_meta() current_app.queues["scheduling"].enqueue_job(fallback_job) + asset_or_sensor_ref = get_asset_or_sensor_ref(asset_or_sensor) + current_app.job_cache.add( + asset_or_sensor_ref["id"], + fallback_job.id, + queue="scheduling", + asset_or_sensor_type=asset_or_sensor_ref["class"].lower(), + ) @job_cache("scheduling") @@ -213,6 +222,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: """ @@ -237,6 +247,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. """ @@ -289,6 +301,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 @@ -381,6 +395,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. @@ -393,6 +408,7 @@ def create_sequential_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: Optional provenance metadata stored on every device job and the wrap-up job. :param scheduler_kwargs: Dict containing start and end (both deserialized) the flex-context (serialized), and the flex-model (partially deserialized, see example below). :returns: The wrap-up job. @@ -417,7 +433,43 @@ def create_sequential_scheduling_job( raise NotImplementedError( "See why: https://github.com/FlexMeasures/flexmeasures/pull/1313/files#r1971479492" ) + if scheduler_specs: + scheduler_class: Type[Scheduler] = load_custom_scheduler(scheduler_specs) + else: + scheduler_class = find_scheduler_class(asset) + if not scheduler_kwargs["flex_model"]: + scheduler = get_scheduler_instance( + scheduler_class=scheduler_class, + asset_or_sensor=asset, + scheduler_params=scheduler_kwargs, + ) + scheduler.collect_flex_config() + collected_flex_model = deepcopy(scheduler.flex_model) + scheduler_kwargs["flex_context"] = scheduler.flex_context + scheduler.deserialize_config() + scheduler_kwargs["flex_model"] = MultiSensorFlexModelSchema(many=True).load( + collected_flex_model + ) + flex_model = scheduler_kwargs["flex_model"] + for child_flex_model in flex_model: + if child_flex_model.get("sensor") is not None: + continue + sensor_ids = { + sensor_reference["sensor"] + for field in ("consumption", "production") + if (sensor_reference := child_flex_model["sensor_flex_model"].get(field)) + is not None + } + if len(sensor_ids) != 1: + asset = child_flex_model.get("asset") + raise ValidationError( + "Sequential scheduling requires each stored device flex-model to " + "reference exactly one output sensor through 'consumption' or " + f"'production' (asset {asset.id if asset else 'unknown'})." + ) + child_flex_model["sensor"] = db.session.get(Sensor, sensor_ids.pop()) + jobs = [] previous_sensors = [] previous_job = depends_on @@ -442,6 +494,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) @@ -466,6 +519,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: @@ -495,6 +550,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. @@ -507,9 +563,10 @@ def create_simultaneous_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: Optional provenance metadata stored on the scheduling job. :param scheduler_kwargs: Dict containing start and end (both deserialized) the flex-context (serialized), and the flex-model (partially deserialized, see example below). - :returns: The wrap-up job. + :returns: The scheduling job. Example of a partially deserialized flex-model per sensor: @@ -542,6 +599,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: diff --git a/flexmeasures/data/tests/test_automations_fresh_db.py b/flexmeasures/data/tests/test_automations_fresh_db.py index 40021f31ca..e0f2d85536 100644 --- a/flexmeasures/data/tests/test_automations_fresh_db.py +++ b/flexmeasures/data/tests/test_automations_fresh_db.py @@ -1,13 +1,22 @@ from __future__ import annotations -from datetime import timezone +from datetime import timedelta, timezone import pytest +from rq.job import Job from sqlalchemy.exc import IntegrityError +from flexmeasures.api.v3_0.tests.utils import message_for_trigger_schedule from flexmeasures.data.models.automations import Automation from flexmeasures.data.models.data_sources import DataSource from flexmeasures.data.models.generic_assets import GenericAsset, GenericAssetType +from flexmeasures.data.models.time_series import Sensor +from flexmeasures.data.services.automations import ( + get_automations_feeding_sensor, + get_automation_job_stats, + resolve_automation_sensors, + run_automation, +) @pytest.fixture() @@ -65,6 +74,198 @@ def test_automation_requires_generator(fresh_db, automation_with_generator): fresh_db.session.commit() +def test_schedule_automation_does_not_require_generator( + fresh_db, automation_with_generator +): + forecast_automation, _ = automation_with_generator + schedule_automation = Automation( + asset=forecast_automation.asset, + type="schedules", + name="generator-free schedule", + cronstr="0 * * * *", + parameters={"duration": "PT1H"}, + ) + fresh_db.session.add(schedule_automation) + fresh_db.session.commit() + + assert schedule_automation.generator_id is None + + +@pytest.fixture() +def clean_scheduling_redis(app): + app.redis_connection.flushdb() + yield + app.redis_connection.flushdb() + + +def test_run_schedule_automation( + fresh_db, + app, + add_battery_assets_fresh_db, + add_market_prices_fresh_db, + clean_scheduling_redis, +): + """A schedules automation queues a scheduling job carrying trigger meta data.""" + battery = add_battery_assets_fresh_db["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]}, + ) + fresh_db.session.add(automation) + fresh_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, + } + + +@pytest.mark.parametrize("sequential", (False, True)) +def test_run_minimal_schedule_automation_with_stored_flex_config( + fresh_db, + app, + add_battery_assets_fresh_db, + add_market_prices_fresh_db, + clean_scheduling_redis, + sequential, +): + """A minimal trigger inherits a single device's flex config from the asset tree.""" + battery = add_battery_assets_fresh_db["Test battery"] + building = battery.parent_asset + power_sensor = next(sensor for sensor in battery.sensors if sensor.name == "power") + battery.flex_model = { + "consumption": {"sensor": power_sensor.id}, + "soc-at-start": "2.5 MWh", + "soc-min": "0 MWh", + "soc-max": "5 MWh", + "power-capacity": "2 MW", + } + automation = Automation( + asset=building, + type="schedules", + name="Minimal stored-flex schedule", + cronstr="0 * * * *", + parameters={"duration": "PT1H", "sequential": sequential}, + ) + fresh_db.session.add(automation) + fresh_db.session.commit() + + returns = run_automation(automation) + job = Job.fetch(returns["job_id"], connection=app.redis_connection) + + if sequential: + assert returns["n_jobs"] == 2 + device_job = Job.fetch(job.args[0][0], connection=app.redis_connection) + assert device_job.meta["asset_or_sensor"] == { + "id": power_sensor.id, + "class": "Sensor", + } + else: + assert returns["n_jobs"] == 1 + assert job.meta["asset_or_sensor"] == {"id": building.id, "class": "Asset"} + + +def test_minimal_schedule_automation_reports_stored_flex_sensors( + fresh_db, add_battery_assets_fresh_db +): + battery = add_battery_assets_fresh_db["Test battery"] + building = battery.parent_asset + power_sensor = next(sensor for sensor in battery.sensors if sensor.name == "power") + price_sensor = fresh_db.session.get( + Sensor, battery.flex_context["consumption-price"]["sensor"] + ) + building.flex_context = { + **building.flex_context, + "consumption-price": {"sensor": price_sensor.id}, + } + battery.flex_model = { + "consumption": {"sensor": power_sensor.id}, + "soc-at-start": "2.5 MWh", + "soc-min": "0 MWh", + "soc-max": "5 MWh", + "power-capacity": "2 MW", + } + automation = Automation( + asset=building, + type="schedules", + name="Minimal stored-flex sensor details", + cronstr="0 * * * *", + parameters={"duration": "PT1H"}, + ) + fresh_db.session.add(automation) + fresh_db.session.commit() + + sensors = resolve_automation_sensors(automation) + + assert sensors["output_sensors"] == [power_sensor] + assert price_sensor in sensors["input_sensors"] + assert get_automations_feeding_sensor(power_sensor) == [automation] + + +def test_schedule_automation_stats_include_descendant_jobs_once( + fresh_db, app, automation_with_generator, clean_scheduling_redis +): + forecast_automation, _ = automation_with_generator + root = forecast_automation.asset + child = GenericAsset( + name="automation child", + generic_asset_type=root.generic_asset_type, + parent_asset=root, + ) + child_sensor = Sensor( + name="child power", + generic_asset=child, + event_resolution=timedelta(minutes=15), + unit="MW", + ) + schedule_automation = Automation( + asset=root, + type="schedules", + name="descendant schedules", + cronstr="0 * * * *", + parameters={"duration": "PT1H"}, + ) + fresh_db.session.add_all([child_sensor, schedule_automation]) + fresh_db.session.flush() + + queue = app.queues["scheduling"] + job = Job.create( + "flexmeasures.utils.time_utils.server_now", connection=queue.connection + ) + job.meta["trigger"] = { + "origin": "automation", + "automation_id": schedule_automation.id, + } + job.save_meta() + queue.enqueue_job(job) + app.job_cache.add(root.id, job.id, "scheduling", "asset") + app.job_cache.add(child_sensor.id, job.id, "scheduling", "sensor") + + other_job = Job.create( + "flexmeasures.utils.time_utils.server_now", connection=queue.connection + ) + other_job.meta["trigger"] = { + "origin": "automation", + "automation_id": schedule_automation.id + 1, + } + other_job.save_meta() + queue.enqueue_job(other_job) + app.job_cache.add(child_sensor.id, other_job.id, "scheduling", "sensor") + + assert get_automation_job_stats(schedule_automation) == {"queued": 1} + + def test_automation_has_valid_timezone_and_aware_cursor(automation_with_generator): automation, _ = automation_with_generator diff --git a/flexmeasures/ui/static/openapi-specs.json b/flexmeasures/ui/static/openapi-specs.json index 9677fb6607..07b8ed4b7f 100644 --- a/flexmeasures/ui/static/openapi-specs.json +++ b/flexmeasures/ui/static/openapi-specs.json @@ -3349,7 +3349,7 @@ "/api/v3_0/assets/{id}/automations/{automation_id}": { "get": { "summary": "Get details of one automation defined on an asset.", - "description": "In addition to the fields shown when listing automations, the response shows\nthe automation's parameters (for forecasts, these are the forecast parameters\nused on each run), information about the data generator that runs it,\nthe sensors it reads from and writes to,\nand counts of recently created jobs, per job status.\nNote that jobs in Redis have a limited TTL, so not all past jobs will be counted.\nThe scheduling cursor is a UTC watermark: occurrences at or before it are ineligible for another automatic queueing attempt. It is not a successful-run timestamp.\n", + "description": "In addition to the fields shown when listing automations, the response shows\nthe automation's parameters (forecast parameters or a schedule trigger message),\ninformation about its data generator (null for schedule automations),\nthe sensors it reads from and writes to,\nand counts of recently created jobs, per job status.\nNote that jobs in Redis have a limited TTL, so not all past jobs will be counted.\nThe scheduling cursor is a UTC watermark: occurrences at or before it are ineligible for another automatic queueing attempt. It is not a successful-run timestamp.\n", "security": [ { "ApiKeyAuth": [] @@ -3455,7 +3455,7 @@ "/api/v3_0/assets/{id}/automations": { "get": { "summary": "Get all automations defined on an asset.", - "description": "The response will be a list of automations: recurring tasks (for now, computing forecasts)\ndefined on the asset. Each entry shows the automation's ID, when it was created,\nits type, name, activation status, and its recurrence, both as a cron string\nand described in natural language. Each entry also shows the IANA timezone in which its cron expression is interpreted and its persistent scheduling cursor.\n", + "description": "The response will be a list of automations: recurring forecasting or scheduling tasks\ndefined on the asset. Each entry shows the automation's ID, when it was created,\nits type, name, activation status, and its recurrence, both as a cron string\nand described in natural language. Each entry also shows the IANA timezone in which its cron expression is interpreted and its persistent scheduling cursor.\n", "security": [ { "ApiKeyAuth": [] diff --git a/flexmeasures/ui/templates/assets/asset_automations.html b/flexmeasures/ui/templates/assets/asset_automations.html index f69774e498..d09a069bfb 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 }} @@ -28,10 +28,10 @@