diff --git a/documentation/changelog.rst b/documentation/changelog.rst index 72921d0e39..6d502ae3d0 100644 --- a/documentation/changelog.rst +++ b/documentation/changelog.rst @@ -14,9 +14,20 @@ v1.1.0 | September XX, 2026 If you maintain indexes of your own on this table, note that the reordered primary key leads with ``(sensor_id, source_id, event_start, belief_horizon)``, so any index you keep on a prefix of that is now redundant and can be dropped once the migration has run. The migration names the ones it finds and leaves them in place, as it cannot know which ones you meant to keep. +.. warning:: A scheduler's data source now also records the flex config it computed under, where previously one data source per scheduler version recorded every schedule that scheduler made. + Schedules computed under different flex configs are therefore recorded by different data sources, and a sensor can carry schedules from several of them, as it already could for forecasts. + Values describing a single moment stay out of that config, so a ``soc-at-start``, or a ``soc-targets`` entry at a given datetime, does not make every run a new data source. + What does is a change to what the site and its devices can do, such as a device's ``power-capacity``. + After such a change, a sensor holds the schedule computed under each configuration, where the newer schedule used to supersede the older one, so a chart of that sensor draws both, and the asset's KPIs total both, as they report what the chart draws. + Select a data source to see the schedule computed under one configuration. + One scheduling request still records under a single data source, including the per-device jobs of a sequential schedule. + New features ------------- +* Automations: recurring tasks defined per asset, computing forecasts or schedules, 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``); each automation interprets its recurrence in its own timezone, and runs missed while the runner was down are caught up once, coalesced into one current forecast; a forecast automation points at a data source holding its forecaster configuration, while a schedule automation stores what the schedule trigger endpoint accepts, and schedules from each run's own time, unless the trigger message fixes a ``start``; an automation's details link to the sensors it reads from and writes to, a sensor's page lists the automations feeding it, and deleting a sensor warns about the automations that use it; jobs now also record whether they were created via the CLI, the API or an automation [see `PR #2290 `_, `PR #2396 `_ and `PR #2293 `_] +* A scheduler's data source now also records the flex config the scheduler computed under, so a schedule can be traced back to the configuration that produced it, and a schedule automation points at such a data source, the way a forecast automation points at its forecaster's [see `PR #2444 `_] +* 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 `_] * Try out a forecast without recording it, using ``flexmeasures add forecasts --dry-run``, which computes the forecast in full and reports the sensor, data source, number of beliefs and event range it would have saved [see `PR #2483 `_] * Run one-off reports as background jobs from the CLI or the asset API, with sensor-level authorization and a dedicated reporting worker queue [see `PR #2298 `_] * A single automation can now be run on demand, from the CLI (``flexmeasures jobs run-automation``), the API (``POST /assets//automations//trigger``) and the asset's *Automations* page (a *Run now* button), which is useful to try out a new automation, to re-run one after fixing what made it fail, or to refresh its results after late input data arrived [see `PR #2460 `_] @@ -80,8 +91,6 @@ v1.0.0 | August 25, 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``); each automation interprets its recurrence in its own timezone, and runs missed while the runner was down are caught up once, coalesced into one current forecast; an automation's details link to the sensors it reads from and writes to, a sensor's page lists the automations feeding it, and deleting a sensor warns about the automations that use it; jobs now also record whether they were created via the CLI, the API or an automation [see `PR #2290 `_ and `PR #2396 `_] -* 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 `_] * ``flexmeasures show data-sources`` now shows which organisation a data source belongs to, and can list the sensors holding data recorded by a given source [see `PR #2401 `_] * The flex-context can now define multiple commodities, each specifying their own prices and grid capacities [see `PR #1946 `_, `PR #2172 `_, `PR #2235 `_, `PR #2271 `_, `PR #2355 `_ and `PR #2380 `_] * Support multiple feeders to a shared storage [see `PR #2001 `_, `PR #2321 `_, `PR #2322 `_, `PR #2325 `_ and `PR #2431 `_] diff --git a/documentation/cli/change_log.rst b/documentation/cli/change_log.rst index 7e12ea497a..e1dad6c05e 100644 --- a/documentation/cli/change_log.rst +++ b/documentation/cli/change_log.rst @@ -20,7 +20,8 @@ since v1.0.0 | August 11, 2026 * Add ``flexmeasures add plan``, ``flexmeasures show plans`` and ``flexmeasures edit plan``, to manage the rate limits and quotas which apply to the accounts on a plan. * Add ``flexmeasures edit secret`` to store an encrypted secret on an account or asset. * Add ``flexmeasures delete secret`` to remove an encrypted secret from an account or asset. -* Add ``flexmeasures add automation``, ``flexmeasures edit automation`` and ``flexmeasures delete automation`` to manage automations (recurring tasks on an asset; for now, computing forecasts). Each automation carries its own IANA timezone (``--timezone``), in which its cron expression is interpreted. +* Add ``flexmeasures add automation``, ``flexmeasures edit automation`` and ``flexmeasures delete automation`` to manage automations (recurring tasks on an asset, with ``--type forecasting`` or ``--type scheduling`` saying which task to automate). Each automation carries its own IANA timezone (``--timezone``), in which its cron expression is interpreted. +* ``flexmeasures add automation --type scheduling`` refuses a flex config field which fixes a moment in time, such as ``soc-at-start`` or a ``soc-targets`` entry with a ``datetime``, naming the field: a recurring schedule automation computes a fresh schedule on every run, so such a value would be stale on the next one. * Add ``flexmeasures jobs run-automations`` to queue jobs for all automations that are due to run this minute from standard five-field cron expressions. Run this command once per minute. It makes at most one queueing attempt per automation per minute, including when an attempt fails after partially queueing jobs. Runs missed while the runner was down are caught up once, with several missed forecast runs coalesced into the latest useful forecast, and a run at a skipped or repeated daylight-saving-time hour happens exactly once. * Add ``flexmeasures jobs run-automation --automation `` to queue the jobs for a single run of one automation, now, on top of its recurring runs. This leaves the automation's cursor alone, so its next recurring run still happens as scheduled, and inactive automations can be run this way, too. * ``flexmeasures delete sensor`` now warns which automations read from or write to a sensor before it is deleted, as an automation refers to its sensors by ID and would fail on its next run. diff --git a/documentation/cli/commands.rst b/documentation/cli/commands.rst index 88314ae233..7d81a1e309 100644 --- a/documentation/cli/commands.rst +++ b/documentation/cli/commands.rst @@ -41,7 +41,7 @@ of which some are referred to in this documentation. ``flexmeasures add annotation`` Add annotation to accounts, assets and/or sensors. ``flexmeasures add toy-account`` Create a toy account, for tutorials and trying things. ``flexmeasures add report`` Create a report. -``flexmeasures add automation`` Add an automation 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/automations.rst b/documentation/features/automations.rst index ae8d1a74cb..f797568018 100644 --- a/documentation/features/automations.rst +++ b/documentation/features/automations.rst @@ -4,11 +4,11 @@ Automations ============ An **automation** is a recurring task defined on an asset. -For now, an automation computes forecasts; automating schedules and reports is planned. +For now, an automation computes forecasts or schedules; automating reports is planned. -On each run, the automation queues jobs (so make sure a worker is processing the ``forecasting`` queue, see :ref:`redis-queue`). +On each run, the automation queues jobs (so make sure a worker is processing the ``forecasting`` or ``scheduling`` queue, whichever the automation needs, see :ref:`redis-queue`). The parameters of the task were stored when the automation was created, and validated with the same schema that the CLI and API use. -Timing parameters are resolved on each run — for instance, the forecast start defaults to the time the automation runs, so each run produces fresh forecasts. +Timing parameters are resolved on each run — for instance, the forecast or schedule start defaults to the time the automation runs, so each run produces fresh results. Creating an automation ---------------------- @@ -17,10 +17,10 @@ Here is how you create an automation in the CLI, asking for daily (at 6 AM) fore .. code-block:: bash - flexmeasures add automation --asset 3 --name "Daily PV forecasts" --type forecasts \ + flexmeasures add automation --asset 3 --name "Daily PV forecasts" --type forecasting \ --cron "0 6 * * *" --timezone Europe/Amsterdam --sensor 12 -``--type`` says what the automation computes, and defaults to ``forecasts``. +``--type`` says which task to automate (``forecasting`` or ``scheduling``, matching the queue the jobs go to), and defaults to ``forecasting``. The remaining options are the ones the task itself needs: a forecast automation accepts everything `flexmeasures add forecast` accepts, such as ``--forecaster`` to pick the forecaster and ``--config`` to configure it (see :ref:`forecasting`). The forecaster and its configuration are stored on a data source, so you can also pass ``--source`` to reuse the data source of an existing forecaster, in which case ``--forecaster`` and ``--config`` (and the individual configuration options) are not needed — the data source already determines them. That data source is required while the automation exists, so it cannot be deleted until the automation is removed. @@ -38,6 +38,36 @@ These changes are recorded in the asset's audit log. For forecast automations, the sensor on which forecasts are saved (``sensor-to-save``, falling back to ``sensor``) must belong to the automation's asset or one of its descendants. This relationship is checked both when the automation is created and immediately before each run. +Automating schedules +-------------------- + +A schedule 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``. +The message is passed in a file, through ``--parameters``, and validated when the automation is created. +The forecaster options above configure a forecaster, so they do not apply here, and are refused when combined with ``--type scheduling``. + +A schedule automation has a data generator too, but you do not name it separately. +It is put together from choices you have already made: the flex config in the trigger message, the flex config saved on the asset tree, and the scheduler that the asset resolves to. +Because those live in two places, and the asset can be edited without touching the automation, the runner puts the generator together again on every run, and moves the automation to another data source when the combination has changed. +Editing an asset's flex-model is therefore a configuration change, and shows up as one: the schedules computed before and after it carry different data sources. + +Because the schedule is recomputed on every run, the flex config may only describe the site and its devices, not one moment. +A field with a fixed moment in it, such as ``soc-at-start`` or a ``soc-targets`` entry with a ``datetime``, is refused when the automation is created, and the error names the field. +Refer to a sensor instead, which says where to look rather than what was true once. + +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 scheduling --parameters trigger-message.yml + Running automations ------------------- @@ -49,7 +79,7 @@ For automations to actually run, let a cron job execute the following command on Each due automation then queues its jobs. If the runner misses runs, because it was down or overloaded, it catches up when it resumes: it queues only the latest missed run of each automation, rather than replaying stale ones. -Timing parameters that default to the run time are resolved when that catch-up run is queued, so it produces a current forecast. +Timing parameters that default to the run time are resolved when that catch-up run is queued, so it produces a current forecast or schedule. Each scheduled run receives at most one automatic queueing attempt. If the process crashes, or queueing fails after creating some jobs, that run is not retried automatically, because a retry could duplicate partial work. diff --git a/documentation/features/forecasting.rst b/documentation/features/forecasting.rst index 6c0f55144c..32991bcd58 100644 --- a/documentation/features/forecasting.rst +++ b/documentation/features/forecasting.rst @@ -221,3 +221,4 @@ Automating forecasts Instead of asking for forecasts one at a time, you can set up an *automation*: a recurring task defined on an asset, which queues forecasting jobs on a cron schedule. See :ref:`automations`. +Schedules can be automated in the same way — see :ref:`automating_schedules`. diff --git a/documentation/features/scheduling.rst b/documentation/features/scheduling.rst index f05cd5b766..3b765ac42d 100644 --- a/documentation/features/scheduling.rst +++ b/documentation/features/scheduling.rst @@ -368,3 +368,12 @@ 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 +-------------------- + +Instead of triggering schedules one at a time, you can set up an *automation*: a recurring task defined on an asset, which queues scheduling jobs on a cron schedule. +See :ref:`automations`. diff --git a/flexmeasures/api/v3_0/assets.py b/flexmeasures/api/v3_0/assets.py index 28c84dfd79..3a66755057 100644 --- a/flexmeasures/api/v3_0/assets.py +++ b/flexmeasures/api/v3_0/assets.py @@ -1397,7 +1397,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 cursor. @@ -1469,8 +1469,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. @@ -2222,10 +2222,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: @@ -2235,6 +2236,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 4ce2fe26e3..4d1117e2bf 100644 --- a/flexmeasures/api/v3_0/sensors.py +++ b/flexmeasures/api/v3_0/sensors.py @@ -1078,6 +1078,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_jobs_api_fresh_db.py b/flexmeasures/api/v3_0/tests/test_asset_jobs_api_fresh_db.py index cd4bf1f156..eab4d8a9b6 100644 --- a/flexmeasures/api/v3_0/tests/test_asset_jobs_api_fresh_db.py +++ b/flexmeasures/api/v3_0/tests/test_asset_jobs_api_fresh_db.py @@ -52,7 +52,7 @@ def test_asset_jobs_redact_inaccessible_automation_provenance( asset=source_asset, generator=generator, name="Confidential source automation", - type="forecasts", + type="forecasting", cronstr="0 6 * * *", parameters={"sensor": target_sensor.id}, ) 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.py b/flexmeasures/api/v3_0/tests/test_automations_api.py index e7ef018575..6a6904acfd 100644 --- a/flexmeasures/api/v3_0/tests/test_automations_api.py +++ b/flexmeasures/api/v3_0/tests/test_automations_api.py @@ -23,7 +23,7 @@ def add_automations(db, add_battery_assets): Automation( asset_id=battery.id, generator=generator, - type="forecasts", + type="forecasting", name="Day-ahead forecasts", cronstr="0 6 * * *", timezone="Europe/Amsterdam", @@ -34,7 +34,7 @@ def add_automations(db, add_battery_assets): Automation( asset_id=battery.id, generator=generator, - type="forecasts", + type="forecasting", name="Intraday forecasts", cronstr="0 * * * *", timezone="UTC", @@ -90,7 +90,7 @@ def test_get_automations( automations = response.json["automations"] assert len(automations) == 2 day_ahead = next(a for a in automations if a["name"] == "Day-ahead forecasts") - assert day_ahead["type"] == "forecasts" + assert day_ahead["type"] == "forecasting" assert day_ahead["cronstr"] == "0 6 * * *" assert day_ahead["timezone"] == "Europe/Amsterdam" assert day_ahead["cursor"] == "2026-07-11T04:00:00+00:00" 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..0beeef0aab 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 @@ -6,6 +6,7 @@ from flask import url_for from flexmeasures.data.models.automations import Automation +from flexmeasures.data.services.automations import resolve_schedule_generator from flexmeasures import Forecaster from flexmeasures.data.models.time_series import Sensor from flexmeasures.data.services.data_sources import get_data_generator @@ -49,7 +50,7 @@ def test_details_reject_inaccessible_sensor_metadata( automation = Automation( asset=prosumer_asset, generator=generator, - type="forecasts", + type="forecasting", name="Cross-organisation details", cronstr="0 6 * * *", parameters={"sensor": output_sensor.id}, @@ -67,3 +68,68 @@ 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}, + } + parameters = {"duration": "PT1H"} + automation = Automation( + asset=asset, + type="scheduling", + name="Minimal schedule details", + cronstr="0 6 * * *", + parameters=parameters, + generator_id=resolve_schedule_generator(asset.id, parameters).id, + ) + 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 bd832d70e6..9dc7a76285 100755 --- a/flexmeasures/cli/data_add.py +++ b/flexmeasures/cli/data_add.py @@ -52,6 +52,11 @@ populate_initial_structure, add_default_asset_types, ) +from flexmeasures.data.schemas.scheduling import find_momentary_flex_config_fields +from flexmeasures.data.services.automations import ( + prepare_schedule_trigger_message, + resolve_schedule_generator, +) from flexmeasures.data.services.data_sources import ( get_or_create_source, get_data_generator, @@ -1384,7 +1389,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): @@ -1712,6 +1722,29 @@ def add_forecast( # noqa: C901 raise +def _check_schedule_automation_parameters(parameters: dict, asset) -> DataSource: + """Validate a schedule automation's trigger message, and return the data generator it will run with. + + The message has to be a valid schedule trigger, and its flex config has to describe the site and its devices, + rather than one moment: the automation computes a fresh schedule on every run, + so a value tied to a fixed moment would be stale on the next one. + """ + try: + message = prepare_schedule_trigger_message(parameters, asset.id) + AssetTriggerSchema().load(message) + except ValidationError as e: + click.secho(f"Invalid schedule parameters: {e.messages}", **MsgStyle.ERROR) + raise click.Abort() + momentary_fields = find_momentary_flex_config_fields(message) + if momentary_fields: + raise click.UsageError( + f"{flexmeasures_inflection.join_words_into_a_list(momentary_fields)} fixes a moment in time," + " so it cannot configure a recurring schedule automation, which computes a fresh schedule on every run." + " Refer to a sensor instead of a fixed value, or leave the field out." + ) + return resolve_schedule_generator(asset.id, parameters) + + @fm_add_data.command("automation") @with_appcontext @click.option( @@ -1748,7 +1781,7 @@ def add_forecast( # noqa: C901 @click.option( "--type", "automation_type", - default="forecasts", + default="forecasting", show_default=True, type=click.Choice(Automation.SUPPORTED_TYPES), help="Type of task to automate.", @@ -1789,7 +1822,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 forecasting, or a schedule trigger message for --type scheduling.", ) @add_cli_options_from_schema( ForecasterParametersSchema(), hidden=True, force_optional=True @@ -1811,18 +1845,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 scheduling --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. @@ -1830,6 +1868,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. """ @@ -1840,46 +1879,78 @@ def add_automation( kwargs, source, config_file, parameters_file ) - # An automation exists to record forecasts, so a dry run would render it pointless. - # Popping the parameter also keeps it out of the parameters stored on the automation. + # An automation exists to record what it computes, so a dry run would render it pointless. + # Popping the parameter also keeps it out of the parameters stored on the automation, + # where a schedule trigger message would reject it as an unknown field. if parameters.pop("dry-run", False): click.secho( - "The dry-run option is not supported for automations, which exist to record the forecasts they compute.", + "The dry-run option is not supported for automations, which exist to record what they compute.", **MsgStyle.ERROR, ) raise click.Abort() + if automation_type == "scheduling": + # 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 scheduling: 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 == "forecasting": + 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: # scheduling + # The scheduler and its configuration make up the automation's data generator, + # the same way a forecaster and its configuration do for a forecast automation. + generator_id = _check_schedule_automation_parameters(parameters, asset).id + 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, @@ -1888,7 +1959,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) @@ -1899,7 +1970,7 @@ def add_automation( db.session.commit() click.secho( f"Successfully created {'inactive ' if inactive else ''}automation '{name}' (ID: {automation.id})" - f" to compute {automation_type} for asset {asset.id}, recurring per cron string '{cronstr}' in timezone '{timezone}'.", + f" for {automation_type} on asset {asset.id}, recurring per cron string '{cronstr}' in timezone '{timezone}'.", **MsgStyle.SUCCESS, ) @@ -2074,7 +2145,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 1e71717d7c..c674267914 100644 --- a/flexmeasures/cli/jobs.py +++ b/flexmeasures/cli/jobs.py @@ -119,7 +119,7 @@ def run_automations(): returns = run_automation(automation) n_jobs = returns.get("n_jobs") if returns else 0 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} {automation.type} 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 511a570990..0731ed3302 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 @@ -91,7 +92,7 @@ def test_add_edit_delete_automation(app, fresh_db, setup_dummy_data): ).scalar_one_or_none() assert automation is not None assert automation.active is True - assert automation.type == "forecasts" + assert automation.type == "forecasting" assert automation.cronstr == "0 6 * * *" assert automation.timezone == "Europe/Amsterdam" # CLI option values are stored as provided (strings); they are coerced by the schema when the automation runs @@ -736,6 +737,382 @@ 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 + + +@pytest.mark.parametrize("automation_type", ("forecasting", "scheduling")) +def test_add_automation_rejects_dry_run( + app, fresh_db, setup_dummy_data, tmp_path, automation_type +): + """An automation records what it computes, so asking it not to record is refused. + + The option reaches every automation type, whatever it computes, + so it is answered before the parameters are read as a forecast or as a schedule trigger. + """ + from flexmeasures.cli.data_add import add_automation + + parameters_file = tmp_path / "parameters.yml" + parameters_file.write_text('duration: "PT12H"\n') + result = app.test_cli_runner().invoke( + add_automation, + [ + "--asset", "1", + "--name", "Dry run", + "--cron", "0 * * * *", + "--type", automation_type, + "--parameters", str(parameters_file), + "--dry-run", + ], + ) # fmt: skip + assert result.exit_code != 0, result.output + assert "dry-run option is not supported for automations" in result.output + assert ( + fresh_db.session.execute( + select(Automation).filter_by(name="Dry run") + ).scalar_one_or_none() + is None + ) + + +def test_add_schedule_automation_rejects_momentary_flex_fields( + app, fresh_db, setup_dummy_data, tmp_path +): + """A flex config field describing one moment cannot configure a recurring schedule automation. + + Such a value is stale on the next run, and it would misdescribe the automation on its data source, + which records the configuration the scheduler computes under. + """ + from flexmeasures.cli.data_add import add_automation + + runner = app.test_cli_runner() + parameters_file = tmp_path / "parameters.yml" + + # a state of charge that held at one moment + parameters_file.write_text( + 'duration: "PT12H"\n' + "flex-model:\n" + " - sensor: 1\n" + ' soc-at-start: "5 kWh"\n' + ) + result = runner.invoke( + add_automation, + [ + "--asset", "1", + "--name", "Momentary state of charge", + "--cron", "0 * * * *", + "--type", "scheduling", + "--parameters", str(parameters_file), + ], + ) # fmt: skip + assert result.exit_code == 2, result.output + assert "flex-model[0].soc-at-start fixes a moment in time" in result.output + assert "Traceback" not in result.output + + # a target tied to a datetime + parameters_file.write_text( + 'duration: "PT12H"\n' + "flex-model:\n" + " - sensor: 1\n" + " soc-targets:\n" + ' - datetime: "2026-01-15T10:00+01:00"\n' + ' value: "5 kWh"\n' + ) + result = runner.invoke( + add_automation, + [ + "--asset", "1", + "--name", "Momentary target", + "--cron", "0 * * * *", + "--type", "scheduling", + "--parameters", str(parameters_file), + ], + ) # fmt: skip + assert result.exit_code == 2, result.output + assert "flex-model[0].soc-targets[0] fixes a moment in time" in result.output + + assert ( + fresh_db.session.execute( + select(Automation).filter_by(name="Momentary state of charge") + ).scalar_one_or_none() + is None + ) + + +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", "scheduling", + "--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", "scheduling", + "--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 == "scheduling" + # The scheduler and the flex config it computes under are the automation's data generator. + assert automation.generator is not None + assert automation.generator.type == "scheduler" + assert ( + automation.generator.attributes["data_generator"]["config"]["asset"] + == automation.asset_id + ) + 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", "scheduling", + "--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", + "scheduling", + "--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", + "scheduling", + "--regressors", + str(setup_dummy_data[0]), + ], + ) + + assert result.exit_code == 2 + assert "--regressors cannot be combined with --type scheduling" 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", + "scheduling", + "--forecaster", + "TrainPredictPipeline", + ], + ) + + assert result.exit_code == 2 + assert "--forecaster cannot be combined with --type scheduling" 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 ( + resolve_schedule_generator, + run_automation, + ) + from flexmeasures.utils.time_utils import server_now + + asset = fresh_db.session.get(GenericAsset, 1) + parameters = {"duration": "PT12H", "resolution": "PT15M"} + automation = Automation( + asset_id=asset.id, + type="scheduling", + name="Test schedules", + cronstr="0 * * * *", + parameters=parameters, + generator_id=resolve_schedule_generator(asset.id, parameters).id, + ) + 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, freeze_server_now ): diff --git a/flexmeasures/data/migrations/versions/3e91c47b0a58_merge_schedule_automations_with_main.py b/flexmeasures/data/migrations/versions/3e91c47b0a58_merge_schedule_automations_with_main.py new file mode 100644 index 0000000000..01afc0e051 --- /dev/null +++ b/flexmeasures/data/migrations/versions/3e91c47b0a58_merge_schedule_automations_with_main.py @@ -0,0 +1,25 @@ +"""merge the schedule automation migrations with main + +Two migrations branched off the same revision: the one adding an automation's timezone and cursor, +and those reordering the timed belief primary key and adding the sensor data source association. +They touch different tables, so this merge only rejoins them and has nothing of its own to do. + +Revision ID: 3e91c47b0a58 +Revises: 9f2b6e1d4a73, 84f268f5153c +Create Date: 2026-09-02 10:30:00.000000 + +""" + +# revision identifiers, used by Alembic. +revision = "3e91c47b0a58" +down_revision = ("9f2b6e1d4a73", "84f268f5153c") +branch_labels = None +depends_on = None + + +def upgrade(): + pass + + +def downgrade(): + pass diff --git a/flexmeasures/data/migrations/versions/a71d6f2c9b04_name_automation_types_after_the_task.py b/flexmeasures/data/migrations/versions/a71d6f2c9b04_name_automation_types_after_the_task.py new file mode 100644 index 0000000000..68f20645d2 --- /dev/null +++ b/flexmeasures/data/migrations/versions/a71d6f2c9b04_name_automation_types_after_the_task.py @@ -0,0 +1,28 @@ +"""Name automation types after the task, not its results. + +The rest of the codebase calls these tasks "forecasting" and "scheduling" (queue names, job types), +so the automation types follow suit: 'forecasts' becomes 'forecasting' and 'schedules' becomes 'scheduling'. + +Revision ID: a71d6f2c9b04 +Revises: 3e91c47b0a58 +Create Date: 2026-09-02 11:00:00.000000 + +""" + +from alembic import op + +# revision identifiers, used by Alembic. +revision = "a71d6f2c9b04" +down_revision = "3e91c47b0a58" +branch_labels = None +depends_on = None + + +def upgrade(): + op.execute("UPDATE automation SET type = 'forecasting' WHERE type = 'forecasts'") + op.execute("UPDATE automation SET type = 'scheduling' WHERE type = 'schedules'") + + +def downgrade(): + op.execute("UPDATE automation SET type = 'forecasts' WHERE type = 'forecasting'") + op.execute("UPDATE automation SET type = 'schedules' WHERE type = 'scheduling'") diff --git a/flexmeasures/data/migrations/versions/c7a2f13b9e04_merge_schedule_automations_with_main.py b/flexmeasures/data/migrations/versions/c7a2f13b9e04_merge_schedule_automations_with_main.py new file mode 100644 index 0000000000..21aad8aa2f --- /dev/null +++ b/flexmeasures/data/migrations/versions/c7a2f13b9e04_merge_schedule_automations_with_main.py @@ -0,0 +1,25 @@ +"""merge the schedule automation migrations with main + +Two migrations branched off the same revision: the one naming the automation types after the task, +and the one dropping the tables that earlier cleanups missed. +They touch different tables, so this merge only rejoins them and has nothing of its own to do. + +Revision ID: c7a2f13b9e04 +Revises: a71d6f2c9b04, 8f4a1d0c2e77 +Create Date: 2026-09-08 10:00:00.000000 + +""" + +# revision identifiers, used by Alembic. +revision = "c7a2f13b9e04" +down_revision = ("a71d6f2c9b04", "8f4a1d0c2e77") +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 db0ab6f2f5..eaa7587055 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,18 @@ def get_initial_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. + Every automation has a data generator, linked through a data source: + a forecaster and its configuration for a forecast automation, + and a scheduler and the flex config it computes under for a schedule automation. + A forecast automation's generator is chosen when it is created. + A schedule automation's is assembled from the trigger message and what its asset stores, + so the runner puts it together afresh on every run. """ __tablename__ = "automation" - SUPPORTED_TYPES = ["forecasts"] # later also "schedules" and "reports" + SUPPORTED_TYPES = ["forecasting", "scheduling"] # later also "reporting" id = db.Column(db.Integer, autoincrement=True, primary_key=True) created_at = db.Column( @@ -54,7 +56,7 @@ class Automation(db.Model, AuthModelMixin): nullable=False, index=True, ) - type = db.Column(db.String(80), nullable=False, default="forecasts") + type = db.Column(db.String(80), nullable=False, default="forecasting") name = db.Column(db.String(80), nullable=False) cronstr = db.Column(db.String(80), nullable=False) timezone = db.Column( diff --git a/flexmeasures/data/models/planning/__init__.py b/flexmeasures/data/models/planning/__init__.py index 6aebfd65d0..fdc0a808bc 100644 --- a/flexmeasures/data/models/planning/__init__.py +++ b/flexmeasures/data/models/planning/__init__.py @@ -1,6 +1,7 @@ from __future__ import annotations from collections import defaultdict +import json from collections.abc import Iterable from dataclasses import dataclass, field from datetime import datetime, timedelta @@ -12,8 +13,13 @@ from flask import current_app from flexmeasures.data import db +from flexmeasures.data.models.data_sources import DataGenerator, DataSource from flexmeasures.data.models.time_series import Sensor from flexmeasures.data.models.generic_assets import GenericAsset as Asset +from flexmeasures.data.schemas.scheduling.config import ( + SchedulerConfigSchema, + strip_momentary_flex_fields, +) from flexmeasures.utils.coding_utils import deprecated, merge_or_append from .devices import INFLEXIBLE_DEVICE_KEYS from .exceptions import WrongEntityException @@ -21,6 +27,26 @@ SchedulerOutputType = pd.Series | list[dict[str, Any]] | None +def _json_safe(value: Any) -> Any: + """Return `value` with anything JSON cannot hold replaced by a stable, serializable stand-in. + + A flex config usually reaches a scheduler serialized, so this changes nothing. + A caller which deserialized it first, as ``flexmeasures add schedule`` does, hands over sensors, assets and quantities instead. + + A sensor or an asset is recorded by its id, which is what the serialized flex config names it by. + Recording how it prints would tie the configuration to its name, so renaming a sensor would describe a different configuration, + and two sensors sharing a name would describe the same one. + Anything else is recorded by how it prints, so that the configuration is still described rather than lost. + """ + + def encode(obj: Any) -> Any: + if isinstance(obj, (Sensor, Asset)): + return obj.id + return str(obj) + + return json.loads(json.dumps(value, default=encode)) + + def _shadow_inflexible_device_keys( db_flex_context: dict, passed_flex_context: dict | list | None ) -> None: @@ -50,7 +76,7 @@ def _shadow_inflexible_device_keys( db_flex_context.pop(key, None) -class Scheduler: +class Scheduler(DataGenerator): """ Superclass for all FlexMeasures Schedulers. @@ -58,11 +84,12 @@ class Scheduler: TODO: extend to multiple flexible assets. The scheduler knows the power sensor of the flexible asset. - It also knows the basic timing parameter of the schedule (start, end, resolution), including the point in time when - knowledge can be assumed to be available (belief_time). + It also knows the basic timing parameter of the schedule (start, end, resolution), + including the point in time when knowledge can be assumed to be available (belief_time). - Furthermore, the scheduler needs to have knowledge about the asset's flexibility model (under what constraints - can the schedule be optimized?) and the system's flexibility context (which other sensors are relevant, e.g. prices). + Furthermore, the scheduler needs to have knowledge about the asset's flexibility model, + which says under what constraints the schedule may be optimized, + and about the system's flexibility context, which says which other sensors are relevant, e.g. prices. These two flexibility configurations are usually fed in from outside, so the scheduler should check them. The deserialize_flex_config function can be used for that. @@ -70,6 +97,10 @@ class Scheduler: __version__ = None __author__ = None + __data_generator_base__ = "scheduler" + + _config_schema = SchedulerConfigSchema() + _save_config = True sensor: Sensor | None = None asset: Asset | None = None @@ -191,6 +222,10 @@ def __init__( self.resolution = resolution self.belief_time = belief_time self.round_to_decimals = round_to_decimals + # The data generator state, kept per instance rather than per class. + self._config = None + self._data_source = None + self._flex_config_collected = False if flex_model is None: flex_model = {} self.flex_model = flex_model @@ -242,6 +277,68 @@ def get_data_source_info(cls: type) -> dict: ) return source_info + def resolve_flex_config(self) -> dict: + """The serialized flex config this scheduler computes with, which is what its data source records. + + By default, that is the flex config as it was passed in. + A scheduler which also reads flex config from the asset tree should override this, + so that its data source describes the configuration the scheduler actually used. + + The config is kept serialized, as the trigger message and the asset tree spell it, + because a deserialized flex config holds sensors, quantities and time series, which do not survive a round trip. + """ + if self.asset is not None: + asset_id = self.asset.id + elif self.sensor is not None: + asset_id = self.sensor.generic_asset.id + else: + asset_id = None + return { + "asset": asset_id, + # Take the snapshot through `_json_safe`, which yields plain JSON structures. + # Deep-copying instead would carry sensors and assets along, detached from the session, + # and the copy would be read long after the objects it copied had been expired by a commit. + "flex-model": strip_momentary_flex_fields(_json_safe(self.flex_model)), + "flex-context": strip_momentary_flex_fields(_json_safe(self.flex_context)), + } + + def record_config(self, config: dict) -> None: + """Record `config` on this scheduler's data source, rather than the config it resolves itself. + + A device job of a sequential schedule uses the configuration of the request it belongs to, + so that one request records one schedule per sensor, rather than one per device's own slice of the flex-model. + """ + self._config = config + self._data_source = None + + @property + def data_source(self) -> DataSource: + """The data source describing this scheduler, its version and its configuration. + + Unlike reporters and forecasters, a scheduler names its source after the scheduler's author, + and versions it by the scheduler's ``__version__``, so this does not defer to `DataGenerator.data_source`. + What it does share is that the configuration is part of the source's identity: + two schedules computed under different flex configs are recorded by different sources. + """ + from flexmeasures.data.services.data_sources import get_or_create_source + + if self._data_source is None: + if self._config is None: + self._config = self.resolve_flex_config() + source_info = self.get_data_source_info() + self._data_source = get_or_create_source( + source=source_info["name"], + source_type=self.__data_generator_base__, + model=source_info["model"], + version=source_info["version"], + attributes={ + "data_generator": { + "config": _json_safe(self._config_schema.dump(self._config)) + } + }, + ) + return self._data_source + def persist_flex_model(self): """ If useful, (parts of) the flex model can be persisted here, @@ -257,11 +354,16 @@ def _get_sensor_or_raise(sensor_id: int) -> Sensor: raise ValueError(f"No sensor found with ID {sensor_id}.") return sensor - def collect_flex_config(self): + def collect_flex_config(self): # noqa: C901 """Merge the flex-config from the db (from the asset and its ancestors) with the initialization flex-config. Note that self.flex_context overrides db_flex_context (from the asset and its ancestors). + Merging twice would be wrong rather than merely wasteful, so this returns early when it already ran. """ + # Tolerate a subclass which does not call this class's __init__, as plugins may not. + if getattr(self, "_flex_config_collected", False): + return + self._flex_config_collected = True if self.asset is not None: asset = self.asset else: @@ -318,7 +420,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: @@ -330,6 +436,8 @@ def deserialize_config(self): Check all configurations we have, throwing either ValidationErrors or ValueErrors. Other code can decide if/how to handle those. """ + # Record the configuration while it is still serialized, as the data source stores it. + self._config = self.resolve_flex_config() self.deserialize_timing_config() self.deserialize_flex_config() self.config_deserialized = True diff --git a/flexmeasures/data/models/planning/storage.py b/flexmeasures/data/models/planning/storage.py index 43e1c11d69..283c64fb42 100644 --- a/flexmeasures/data/models/planning/storage.py +++ b/flexmeasures/data/models/planning/storage.py @@ -1924,6 +1924,15 @@ def persist_flex_model(self): "soc_in_mwh", self.flex_model.get("soc_at_start") ) + def resolve_flex_config(self) -> dict: + """A storage scheduler also reads flex config from the asset tree, so merge that in before recording it. + + Without this, the data source would describe only what the trigger message said, + which for a minimal trigger message is next to nothing. + """ + self.collect_flex_config() + return super().resolve_flex_config() + def deserialize_flex_config(self): """ Deserialize storage flex model and the flex context against schemas. diff --git a/flexmeasures/data/models/planning/tests/test_utils_fresh_db.py b/flexmeasures/data/models/planning/tests/test_utils_fresh_db.py index a4473409cb..e2ce1c0fad 100644 --- a/flexmeasures/data/models/planning/tests/test_utils_fresh_db.py +++ b/flexmeasures/data/models/planning/tests/test_utils_fresh_db.py @@ -312,6 +312,90 @@ def test_collect_flex_config_missing_sensor_raises(fresh_db): scheduler_soc.collect_flex_config() +def test_momentary_flex_fields_do_not_make_a_new_data_source(fresh_db): + """A value describing one moment stays out of a scheduler's data source identity. + + A state of charge measured at the start of a schedule differs on every trigger, + so recording it would make every schedule the work of a brand new data source. + What the site and its devices can do is what tells one scheduler source from another. + """ + asset_type = GenericAssetType(name="test-asset-type-momentary-config") + fresh_db.session.add(asset_type) + asset = GenericAsset( + name="test-asset-momentary-config", generic_asset_type=asset_type + ) + fresh_db.session.add(asset) + fresh_db.session.commit() + + start = datetime(2023, 1, 1, tzinfo=ZoneInfo("UTC")) + + def source_for(soc_at_start: str, power_capacity: str = "2 MW"): + scheduler = StorageScheduler( + asset_or_sensor=asset, + start=start, + end=start + timedelta(hours=1), + resolution=timedelta(hours=1), + flex_model=[ + { + "soc-at-start": soc_at_start, + "soc-min": "0 kWh", + "power-capacity": power_capacity, + } + ], + flex_context={}, + ) + return scheduler.data_source + + # Two schedules of the same device, from different states of charge + assert source_for("4 kWh") == source_for("7 kWh") + # ... but a device that can draw less power is a different configuration + assert source_for("4 kWh") != source_for("4 kWh", power_capacity="1 MW") + + +def test_renaming_a_sensor_does_not_make_a_new_data_source(fresh_db): + """A scheduler's data source records the sensors its config names by id, not by name. + + A caller which deserialized the flex config first, as ``flexmeasures add schedule`` does, + hands the scheduler sensor objects rather than their ids. + Recording how those print would tie the configuration to the sensor's name, + so renaming a sensor would describe a different configuration, and two sensors sharing a name the same one. + """ + asset_type = GenericAssetType(name="test-asset-type-renamed-sensor") + fresh_db.session.add(asset_type) + asset = GenericAsset( + name="test-asset-renamed-sensor", generic_asset_type=asset_type + ) + fresh_db.session.add(asset) + price_sensor = Sensor( + name="day-ahead prices", + generic_asset=asset, + event_resolution=timedelta(hours=1), + unit="EUR/MWh", + ) + fresh_db.session.add(price_sensor) + fresh_db.session.commit() + + start = datetime(2023, 1, 1, tzinfo=ZoneInfo("UTC")) + + def source_now(): + scheduler = StorageScheduler( + asset_or_sensor=asset, + start=start, + end=start + timedelta(hours=1), + resolution=timedelta(hours=1), + flex_model=[{"soc-min": "0 kWh", "power-capacity": "2 MW"}], + # As a caller hands it over once deserialized: the sensor itself, not its id. + flex_context={"consumption-price": price_sensor}, + ) + return scheduler.data_source + + before = source_now() + price_sensor.name = "day-ahead prices (renamed)" + fresh_db.session.commit() + + assert source_now() == before + + def test_get_power_values_sign_conventions_and_source_filters(fresh_db): """The explicit sign convention wins; None defers to the sensor attribute; source filters on a SensorReference are honored. diff --git a/flexmeasures/data/schemas/scheduling/__init__.py b/flexmeasures/data/schemas/scheduling/__init__.py index cd0a4e942d..b40b35e32c 100644 --- a/flexmeasures/data/schemas/scheduling/__init__.py +++ b/flexmeasures/data/schemas/scheduling/__init__.py @@ -18,9 +18,14 @@ post_load, ) -from flexmeasures import Sensor +from flexmeasures.data.models.time_series import Sensor from flexmeasures.data.schemas.generic_assets import GenericAssetIdField +from flexmeasures.data.schemas.scheduling.config import ( # noqa: F401 + SchedulerConfigSchema, + find_momentary_flex_config_fields, + strip_momentary_flex_fields, +) from flexmeasures.data.schemas.sensors import ( VariableQuantityField, SensorIdField, @@ -1831,6 +1836,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/schemas/scheduling/config.py b/flexmeasures/data/schemas/scheduling/config.py new file mode 100644 index 0000000000..7a8e147edb --- /dev/null +++ b/flexmeasures/data/schemas/scheduling/config.py @@ -0,0 +1,128 @@ +"""What makes up a scheduler's configuration, as recorded on its data source.""" + +from __future__ import annotations + +from marshmallow import Schema, fields + +#: Flex-config fields which state what was true at one moment, rather than where to look it up. +#: A recurring automation would carry such a value into every later run, long after the moment it described, +#: and a data source recording one would be a new source on every run. +MOMENTARY_FLEX_FIELDS = ("soc-at-start",) + +#: Keys which mark a value as describing one moment or period, as a time series segment does. +_MOMENT_KEYS = ("datetime", "start", "end") + + +def _describes_a_moment(value) -> bool: + """Whether this value pins itself to a moment, as a time series segment does.""" + return isinstance(value, dict) and any(key in value for key in _MOMENT_KEYS) + + +def _was_emptied(original, stripped) -> bool: + """Whether stripping left nothing of a value which did hold something.""" + if stripped is None: + return original is not None + if isinstance(stripped, (list, dict)) and not stripped: + return bool(original) + return False + + +def _walk_momentary_fields(value, path: str, drop: bool): + """Find, and optionally drop, the parts of a flex config which describe one moment. + + Returns the value (with those parts removed when `drop`) and the paths at which they were found. + """ + found: list[str] = [] + if isinstance(value, dict): + if _describes_a_moment(value): + return (None if drop else value), [path] + kept = {} + for key, item in value.items(): + if key in MOMENTARY_FLEX_FIELDS: + found.append(f"{path}.{key}") + if drop: + continue + kept[key] = item + continue + stripped, item_found = _walk_momentary_fields(item, f"{path}.{key}", drop) + found += item_found + if drop and _was_emptied(item, stripped): + # Leave the field out altogether, rather than keeping it as a null or an empty list. + # Otherwise the same configuration would look different depending on how it was written, + # since a single moment may be given as one mapping or as a list of them. + continue + kept[key] = stripped + return kept, found + if isinstance(value, list): + kept = [] + for index, item in enumerate(value): + item, item_found = _walk_momentary_fields(item, f"{path}[{index}]", drop) + found += item_found + if drop and item is None: + continue + kept.append(item) + return kept, found + return value, found + + +def find_momentary_flex_config_fields(message: dict) -> list[str]: + """Find the flex config fields which describe one moment, rather than the site and its devices. + + A schedule automation recomputes its schedule on every run, so a value tied to a fixed moment is stale on the next one. + Sensor references and plain quantities are fine: they say where to look, or what always holds, rather than what was true once. + """ + found: list[str] = [] + for key in ("flex-model", "flex-context"): + _, key_found = _walk_momentary_fields(message.get(key), key, drop=False) + found += key_found + return sorted(set(found)) + + +def strip_momentary_flex_fields(value): + """Return the flex config without the parts which describe one moment. + + A scheduler's data source is identified by its configuration, so anything that changes from run to run has to stay out of it. + A state of charge measured at the start of one schedule, or a target at one datetime, would otherwise make every run a new data source. + """ + stripped, _ = _walk_momentary_fields(value, "", drop=True) + return stripped + + +class SchedulerConfigSchema(Schema): + """The configuration of a scheduler: which asset it schedules, and the flex config it uses. + + Together with the scheduler's class and version, this is what tells one scheduler data source from another, + so that a schedule can be traced back to the configuration it was computed under. + Timing fields are deliberately absent: start, end, resolution and belief time differ from run to run, + and are the scheduler's parameters rather than its configuration. + + The flex config is kept in its serialized form, as the trigger message and the asset tree spell it, + because that is the form every scheduler shares. + Deserialized flex configs hold sensors, quantities and time series, which each scheduler resolves in its own way. + """ + + asset = fields.Integer( + required=False, + allow_none=True, + metadata=dict( + description="ID of the asset (or of the sensor's asset) that this scheduler schedules.", + ), + ) + flex_model = fields.Raw( + attribute="flex-model", + data_key="flex-model", + required=False, + allow_none=True, + metadata=dict( + description="The flex-model the scheduler uses, after merging the trigger message with what the asset tree stores.", + ), + ) + flex_context = fields.Raw( + attribute="flex-context", + data_key="flex-context", + required=False, + allow_none=True, + metadata=dict( + description="The flex-context the scheduler uses, after merging the trigger message with what the asset tree stores.", + ), + ) diff --git a/flexmeasures/data/schemas/scheduling/groups.py b/flexmeasures/data/schemas/scheduling/groups.py index fc69e6d6ca..d3f81a0f7c 100644 --- a/flexmeasures/data/schemas/scheduling/groups.py +++ b/flexmeasures/data/schemas/scheduling/groups.py @@ -10,7 +10,7 @@ from marshmallow import validates_schema, ValidationError -from flexmeasures import Sensor +from flexmeasures.data.models.time_series import Sensor from flexmeasures.data.schemas.generic_assets import GenericAssetIdField from flexmeasures.data.schemas.sensors import ( SensorIdField, diff --git a/flexmeasures/data/schemas/scheduling/storage.py b/flexmeasures/data/schemas/scheduling/storage.py index 770d13b7ad..d453b34c08 100644 --- a/flexmeasures/data/schemas/scheduling/storage.py +++ b/flexmeasures/data/schemas/scheduling/storage.py @@ -14,7 +14,8 @@ ) from marshmallow.validate import OneOf, ValidationError -from flexmeasures import Asset, Sensor +from flexmeasures.data.models.generic_assets import GenericAsset as Asset +from flexmeasures.data.models.time_series import Sensor from flexmeasures.data.schemas.generic_assets import GenericAssetIdField from flexmeasures.data.schemas.units import QuantityField from flexmeasures.data.schemas.scheduling import metadata diff --git a/flexmeasures/data/schemas/scheduling/utils.py b/flexmeasures/data/schemas/scheduling/utils.py index 92d1c2ee98..e25f2fb2f5 100644 --- a/flexmeasures/data/schemas/scheduling/utils.py +++ b/flexmeasures/data/schemas/scheduling/utils.py @@ -4,7 +4,7 @@ import pandas as pd -from flexmeasures import Sensor +from flexmeasures.data.models.time_series import Sensor SOC_TIMED_EVENT_FIELDS = ("soc-targets", "soc-minima", "soc-maxima") diff --git a/flexmeasures/data/schemas/tests/test_scheduler_config.py b/flexmeasures/data/schemas/tests/test_scheduler_config.py new file mode 100644 index 0000000000..10741538e3 --- /dev/null +++ b/flexmeasures/data/schemas/tests/test_scheduler_config.py @@ -0,0 +1,49 @@ +from flexmeasures.data.schemas.scheduling.config import ( + find_momentary_flex_config_fields, + strip_momentary_flex_fields, +) + + +def test_a_moment_leaves_no_trace_whichever_shape_it_came_in(): + """However a momentary value was written, the configuration left behind is the same. + + A single moment may be given as one mapping or as a list of them, + so keeping the field as a null or an empty list would make one spelling a different configuration than the other, + and both different from leaving the field out. + """ + as_a_list = [{"sensor": 1, "soc-targets": [{"datetime": "x", "value": 1}]}] + as_a_mapping = [{"sensor": 1, "soc-targets": {"datetime": "x", "value": 1}}] + left_out = [{"sensor": 1}] + + assert strip_momentary_flex_fields(as_a_list) == left_out + assert strip_momentary_flex_fields(as_a_mapping) == left_out + assert strip_momentary_flex_fields(left_out) == left_out + + +def test_stripping_keeps_what_holds_beyond_one_moment(): + """Sensor references and plain quantities survive, and so do the static entries of a mixed list.""" + flex_model = [ + { + "sensor": 1, + "power-capacity": "2 MW", + "soc-at-start": "5 kWh", + "soc-targets": [{"datetime": "x", "value": 1}, {"sensor": 9}], + } + ] + assert strip_momentary_flex_fields(flex_model) == [ + {"sensor": 1, "power-capacity": "2 MW", "soc-targets": [{"sensor": 9}]} + ] + # An empty list was not emptied by stripping, so it stays as it was given. + assert strip_momentary_flex_fields([{"soc-targets": []}]) == [{"soc-targets": []}] + + +def test_momentary_fields_are_reported_by_their_path(): + """The paths name the field at fault, which is what the CLI tells the user.""" + assert find_momentary_flex_config_fields( + { + "flex-model": [ + {"soc-at-start": "5 kWh", "soc-targets": [{"datetime": "x"}]}, + ], + "flex-context": {"consumption-price": {"sensor": 3}}, + } + ) == ["flex-model[0].soc-at-start", "flex-model[0].soc-targets[0]"] diff --git a/flexmeasures/data/services/automations.py b/flexmeasures/data/services/automations.py index 719d6e1227..a1abaf4072 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 @@ -20,6 +22,7 @@ from flexmeasures import Forecaster from flexmeasures.data import db from flexmeasures.data.models.automations import Automation +from flexmeasures.data.models.data_sources import DataSource from flexmeasures.data.models.time_series import Sensor from flexmeasures.data.queries.generic_assets import ( asset_and_ancestor_ids, @@ -39,6 +42,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". @@ -60,6 +126,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) @@ -247,15 +320,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 == "scheduling": + 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." @@ -316,6 +444,70 @@ 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 resolve_schedule_generator(asset_id: int, parameters: dict) -> DataSource: + """The data source describing the scheduler a schedule automation runs, and the flex config it runs with. + + The scheduler class follows from the asset, and the config is the trigger message merged with what the asset tree stores, + so both can change without the automation changing. + That is why this is resolved afresh on every run, rather than only when the automation is created. + """ + from flexmeasures.data.schemas.scheduling import AssetTriggerSchema + from flexmeasures.data.services.scheduling import ( + find_scheduler_class, + get_scheduler_instance, + ) + + message = prepare_schedule_trigger_message(dict(parameters or {}), asset_id) + trigger_data = AssetTriggerSchema().load(message) + asset = trigger_data["asset"] + scheduler = get_scheduler_instance( + scheduler_class=find_scheduler_class(asset), + asset_or_sensor=asset, + scheduler_params=dict( + start=trigger_data["start_of_schedule"], + end=trigger_data["start_of_schedule"] + trigger_data["duration"], + # The flex config goes in as the message spells it, which is the form the data source records. + flex_model=message.get("flex-model"), + flex_context=message.get("flex-context"), + return_multiple=True, + ), + ) + return scheduler.data_source + + def get_automations_involving_sensor(sensor: Sensor) -> list[Automation]: """Find the automations that read from or write to the given sensor. @@ -341,21 +533,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 == "scheduling": + # Scheduling jobs are cached under the asset (multi-device wrap-up jobs) + # and under individual sensors (per-device jobs). + assets = [automation.asset, *automation.asset.offspring] + 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) @@ -400,13 +604,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 == "forecasting": + return _run_forecast_automation(automation) + elif automation.type == "scheduling": + return _run_schedule_automation(automation) + raise NotImplementedError( + f"Automations of type '{automation.type}' cannot be run yet." + ) + + +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)." @@ -424,3 +633,46 @@ 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, + ) + + # The scheduler and the flex config it merges in can both change between runs, + # so record which data source this run actually computes under. + generator = resolve_schedule_generator(automation.asset_id, automation.parameters) + if automation.generator_id != generator.id: + automation.generator_id = generator.id + db.session.commit() + + 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 cad1f32963..ac37dbf26f 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 @@ -39,7 +40,7 @@ from flexmeasures.data.models.generic_assets import GenericAsset as Asset from flexmeasures.data.models.data_sources import DataSource from flexmeasures.data.schemas.scheduling import MultiSensorFlexModelSchema -from flexmeasures.data.utils import get_data_source, save_to_db +from flexmeasures.data.utils import save_to_db from flexmeasures.utils.time_utils import server_now from flexmeasures.data.services.utils import ( job_cache, @@ -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,8 @@ 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, + data_source_config: dict | None = None, **scheduler_kwargs, ) -> Job: """ @@ -237,6 +248,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. """ @@ -269,6 +282,7 @@ def create_scheduling_job( kwargs=dict( asset_or_sensor=asset_or_sensor, scheduler_specs=scheduler_specs, + data_source_config=data_source_config, **scheduler_kwargs, ), id=job_id, @@ -289,6 +303,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 +397,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 +410,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 +435,60 @@ 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()) + + # A scheduling request is one run of one generator, + # so all of its device jobs record their schedules under one data source, describing the request's own configuration. + # Without this, each device job would resolve a source of its own, from its own slice of the flex-model, + # and a schedule could no longer be retrieved per device from the request's job. + # The configuration travels with the jobs, rather than the data source it belongs to: + # a source created here would live in the transaction of the request that enqueued the jobs, + # which FlexMeasures does not commit (see `flexmeasures.data.transactional`), so the workers would never see it. + request_scheduler = get_scheduler_instance( + scheduler_class=scheduler_class, + asset_or_sensor=asset, + scheduler_params={ + **scheduler_kwargs, + "flex_model": MultiSensorFlexModelSchema(many=True).dump(flex_model), + }, + ) + data_source_config = request_scheduler.resolve_flex_config() + jobs = [] previous_sensors = [] previous_job = depends_on @@ -436,12 +507,14 @@ def create_sequential_scheduling_job( job = create_scheduling_job( **current_scheduler_kwargs, + data_source_config=data_source_config, scheduler_specs=scheduler_specs, requeue=requeue, job_id=job_id, 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 +539,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 +570,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 +583,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 +619,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: @@ -735,6 +813,7 @@ def make_schedule( # noqa: C901 flex_context: dict | None = None, flex_config_has_been_deserialized: bool = False, scheduler_specs: dict | None = None, + data_source_config: dict | None = None, dry_run: bool = False, **scheduler_kwargs: dict, ) -> dict: @@ -823,12 +902,15 @@ def make_schedule( # noqa: C901 click.echo("Job %s made schedule." % rq_job.id) rq_job.meta["scheduler_info"] = scheduler.info - data_source = get_data_source( - data_source_name=data_source_info["name"], - data_source_model=data_source_info["model"], - data_source_version=data_source_info["version"], - data_source_type="scheduler", - ) + # The scheduler's own data source, which also records the flex config it computed under. + # A device job of a sequential schedule is handed the configuration of the request it belongs to, + # so that one request records one schedule per sensor, rather than one per device's own config. + # It is handed the configuration rather than a data source id, + # because that id would come from a row created while the request that enqueued this job was still open, + # and this session never sees it. + if data_source_config is not None: + scheduler.record_config(data_source_config) + data_source = scheduler.data_source # saving info on the job, so the API for a job can look the data up if rq_job: diff --git a/flexmeasures/data/tests/test_automation_scheduling_fresh_db.py b/flexmeasures/data/tests/test_automation_scheduling_fresh_db.py index 2e77fbc5af..c5c78392c7 100644 --- a/flexmeasures/data/tests/test_automation_scheduling_fresh_db.py +++ b/flexmeasures/data/tests/test_automation_scheduling_fresh_db.py @@ -42,7 +42,7 @@ def create_automation( automation = Automation( asset=asset, generator=generator, - type="forecasts", + type="forecasting", name=name, cronstr=cronstr, timezone=timezone_name, diff --git a/flexmeasures/data/tests/test_automations_fresh_db.py b/flexmeasures/data/tests/test_automations_fresh_db.py index 47759a7e4e..061a06f4ee 100644 --- a/flexmeasures/data/tests/test_automations_fresh_db.py +++ b/flexmeasures/data/tests/test_automations_fresh_db.py @@ -1,13 +1,36 @@ 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.services.automations import resolve_schedule_generator 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, +) + + +def build_schedule_automation(asset, **kwargs) -> Automation: + """Build a schedule automation, with the data generator that its creation and its runs resolve. + + A schedule automation's generator describes the scheduler the asset resolves to, and the flex config it computes under, + so it is derived from the asset and the parameters rather than chosen. + """ + automation = Automation(asset=asset, type="scheduling", **kwargs) + automation.generator_id = resolve_schedule_generator( + asset.id, automation.parameters + ).id + return automation @pytest.fixture() @@ -22,7 +45,7 @@ def automation_with_generator(fresh_db): automation = Automation( asset=asset, generator=generator, - type="forecasts", + type="forecasting", name="automation generator lifecycle test", cronstr="0 6 * * *", parameters={}, @@ -65,6 +88,253 @@ def test_automation_requires_generator(fresh_db, automation_with_generator): fresh_db.session.commit() +def test_schedule_automation_generator_describes_its_scheduler_and_config( + fresh_db, automation_with_generator +): + """A schedule automation's generator names the scheduler, and records the flex config it computes under.""" + forecast_automation, _ = automation_with_generator + schedule_automation = build_schedule_automation( + forecast_automation.asset, + name="scheduling the asset", + cronstr="0 * * * *", + parameters={"duration": "PT1H"}, + ) + fresh_db.session.add(schedule_automation) + fresh_db.session.commit() + + generator = schedule_automation.generator + assert generator is not None + assert generator.type == "scheduler" + assert generator.model == "StorageScheduler" + config = generator.attributes["data_generator"]["config"] + assert config["asset"] == forecast_automation.asset.id + # The flex config is recorded as the asset tree and the trigger message spell it, not as timing. + assert set(config) == {"asset", "flex-model", "flex-context"} + + +@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 = build_schedule_automation( + battery, + 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 = build_schedule_automation( + building, + 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_schedule_automation_follows_its_asset_flex_config( + fresh_db, + app, + add_battery_assets_fresh_db, + add_market_prices_fresh_db, + clean_scheduling_redis, +): + """Editing the asset's flex config moves the automation to another data generator. + + A schedule automation's generator describes the flex config the scheduler computes under, + and that config is the trigger message merged with what the asset tree stores, + so a change to the asset shows up as a different generator on the automation's next run. + """ + 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-min": "0 MWh", + "soc-max": "5 MWh", + "power-capacity": "2 MW", + } + automation = build_schedule_automation( + building, + name="Schedule following the asset", + cronstr="0 * * * *", + parameters={"duration": "PT1H"}, + ) + fresh_db.session.add(automation) + fresh_db.session.commit() + + run_automation(automation) + generator_before = automation.generator + assert "2 MW" in str( + generator_before.attributes["data_generator"]["config"]["flex-model"] + ) + + # The site can now draw less power, which is a different configuration to schedule under. + battery.flex_model = {**battery.flex_model, "power-capacity": "1 MW"} + fresh_db.session.commit() + + run_automation(automation) + generator_after = automation.generator + assert generator_after.id != generator_before.id + assert "1 MW" in str( + generator_after.attributes["data_generator"]["config"]["flex-model"] + ) + # Both describe the same scheduler, so only the configuration tells them apart. + assert generator_after.model == generator_before.model + assert generator_after.version == generator_before.version + + +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 = build_schedule_automation( + building, + 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 = build_schedule_automation( + root, + 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/data/tests/test_scheduling_jobs.py b/flexmeasures/data/tests/test_scheduling_jobs.py index 2842a066ae..e1285813bd 100644 --- a/flexmeasures/data/tests/test_scheduling_jobs.py +++ b/flexmeasures/data/tests/test_scheduling_jobs.py @@ -18,6 +18,7 @@ from flexmeasures.data.tests.utils import exception_reporter from flexmeasures.utils.job_utils import work_on_rq from flexmeasures.data.services.scheduling import ( + get_data_source_for_job, create_scheduling_job, load_custom_scheduler, handle_scheduling_exception, @@ -77,12 +78,16 @@ def test_scheduling_a_battery( work_on_rq(app.queues["scheduling"], exc_handler=exception_reporter) - scheduler_source = fresh_db.session.execute( - select(DataSource).filter_by(name="Seita", type="scheduler") - ).scalar_one_or_none() + # Ask the job which source it wrote with, rather than looking one up by name: + # a scheduler's source also records the flex config it computed under, + # so several sources can share the scheduler's name, model and version. + scheduler_source = get_data_source_for_job( + Job.fetch(job.id, connection=app.queues["scheduling"].connection) + ) assert ( scheduler_source is not None ) # Make sure the scheduler data source is now there + assert scheduler_source.name == "Seita" and scheduler_source.type == "scheduler" power_values = fresh_db.session.scalars( select(TimedBelief) diff --git a/flexmeasures/data/tests/test_scheduling_sequential.py b/flexmeasures/data/tests/test_scheduling_sequential.py index f219c62b60..ba3051eccd 100644 --- a/flexmeasures/data/tests/test_scheduling_sequential.py +++ b/flexmeasures/data/tests/test_scheduling_sequential.py @@ -3,12 +3,72 @@ import pandas as pd from rq.job import Job +from sqlalchemy import select + +from flexmeasures.data.models.data_sources import DataSource from flexmeasures.data.services.scheduling import create_sequential_scheduling_job from flexmeasures.utils.job_utils import work_on_rq from flexmeasures.data.services.scheduling import handle_scheduling_exception from flexmeasures.data.models.time_series import Sensor +def test_sequential_jobs_carry_the_request_config_not_a_data_source_id( + db, app, flex_description_sequential, smart_building +): + """The device jobs of one request share a data source, without depending on an uncommitted row. + + FlexMeasures does not auto-commit the session of the request that enqueues the jobs, + so a data source created while enqueueing would never reach the workers (see `flexmeasures.data.transactional`). + The jobs therefore carry the request's configuration, and each worker resolves the source from it and commits. + """ + assets, sensors, soc_sensors = smart_building + queue = app.queues["scheduling"] + start = pd.Timestamp("2015-01-03").tz_localize("Europe/Amsterdam") + flex_description_sequential["start"] = start + flex_description_sequential["end"] = pd.Timestamp("2015-01-04").tz_localize( + "Europe/Amsterdam" + ) + + scheduler_sources_before = db.session.scalars( + select(DataSource).filter_by(type="scheduler") + ).all() + + create_sequential_scheduling_job( + asset=assets["Test Site"], + scheduler_specs={ + "module": "flexmeasures.data.models.planning.storage", + "class": "StorageScheduler", + }, + enqueue=True, + **flex_description_sequential, + ) + + queued_jobs = [ + Job.fetch(job_id, connection=queue.connection) for job_id in queue.job_ids + ] + device_jobs = [job for job in queued_jobs if job.kwargs.get("asset_or_sensor")] + assert device_jobs, "the request should have queued a job per device" + configs = [job.kwargs.get("data_source_config") for job in device_jobs] + assert all( + config is not None for config in configs + ), "every device job should carry the request's configuration" + assert all( + config == configs[0] for config in configs + ), "the device jobs of one request describe one configuration, so they share one data source" + assert "data_source_id" not in device_jobs[0].kwargs + + # Enqueueing wrote no data source of its own, which is what it must not rely on. + assert ( + db.session.scalars(select(DataSource).filter_by(type="scheduler")).all() + == scheduler_sources_before + ) + + # This test never runs the jobs it queued, so clear Redis rather than leaking them into the next test: + # the queue and its deferred registry, the jobs themselves, whose ids are derived from what they schedule, + # and the job cache, which would otherwise skip the next test's identical request as already made. + app.redis_connection.flushdb() + + def test_create_sequential_jobs(db, app, flex_description_sequential, smart_building): """Test sequential scheduling capabilities. diff --git a/flexmeasures/data/tests/test_scheduling_simultaneous.py b/flexmeasures/data/tests/test_scheduling_simultaneous.py index ebf4c4cd8e..5fefade6c4 100644 --- a/flexmeasures/data/tests/test_scheduling_simultaneous.py +++ b/flexmeasures/data/tests/test_scheduling_simultaneous.py @@ -2,7 +2,10 @@ import numpy as np import pandas as pd -from flexmeasures.data.services.scheduling import create_simultaneous_scheduling_job +from flexmeasures.data.services.scheduling import ( + create_simultaneous_scheduling_job, + get_data_source_for_job, +) from flexmeasures.utils.job_utils import work_on_rq from flexmeasures.data.models.time_series import Sensor @@ -77,21 +80,29 @@ def test_create_simultaneous_jobs( job.perform() assert job.get_status() == "finished" - # Get power and SoC values - ev_power = sensors["Test EV"].search_beliefs() + # Get power and SoC values, from the source this job wrote with. + # A scheduler's data source records the flex config it computed under, + # so these sensors can also carry schedules computed under another config, by another job, from another source. + scheduler_source = get_data_source_for_job(job) + assert scheduler_source is not None + + def schedule_of(sensor): + return sensor.search_beliefs(source=scheduler_source) + + ev_power = schedule_of(sensors["Test EV"]) assert ev_power.sources.unique()[0].model == "StorageScheduler" - ev_soc = soc_sensors["Test EV"].search_beliefs() + ev_soc = schedule_of(soc_sensors["Test EV"]) assert ev_soc.sources.unique()[0].model == "StorageScheduler" if use_heterogeneous_resolutions: - battery_power = sensors["Test Battery 1h"].search_beliefs() + battery_power = schedule_of(sensors["Test Battery 1h"]) assert len(battery_power) == 24 - battery_soc = soc_sensors["Test Battery 1h"].search_beliefs() + battery_soc = schedule_of(soc_sensors["Test Battery 1h"]) assert len(battery_soc) == 97 else: - battery_power = sensors["Test Battery"].search_beliefs() + battery_power = schedule_of(sensors["Test Battery"]) assert len(battery_power) == 96 - battery_soc = soc_sensors["Test Battery"].search_beliefs() + battery_soc = schedule_of(soc_sensors["Test Battery"]) assert len(battery_soc) == 97 ev_power = ev_power.droplevel([1, 2, 3]) diff --git a/flexmeasures/ui/static/openapi-specs.json b/flexmeasures/ui/static/openapi-specs.json index aef8d19a55..6302e0da96 100644 --- a/flexmeasures/ui/static/openapi-specs.json +++ b/flexmeasures/ui/static/openapi-specs.json @@ -3369,7 +3369,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 cursor is the UTC time of the most recent run the automation committed to; runs at or before it are never queued again.\nIt advances just before queueing, so it does not indicate that queueing or the forecast itself succeeded.\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 cursor is the UTC time of the most recent run the automation committed to; runs at or before it are never queued again.\nIt advances just before queueing, so it does not indicate that queueing or the forecast itself succeeded.\n", "security": [ { "ApiKeyAuth": [] @@ -3475,7 +3475,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 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 cursor.\n", "security": [ { "ApiKeyAuth": [] diff --git a/flexmeasures/ui/templates/assets/asset_automations.html b/flexmeasures/ui/templates/assets/asset_automations.html index 455b09760e..55359d5d28 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 @@