diff --git a/documentation/changelog.rst b/documentation/changelog.rst index 2b153ab8f6..2f778ab043 100644 --- a/documentation/changelog.rst +++ b/documentation/changelog.rst @@ -14,10 +14,19 @@ 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 `_] * Changing the selected time range on an asset or sensor chart now only loads the data that is actually new, instead of reloading the whole range, which makes stepping through or extending a long period much faster; reloading the page, or leaving it open for five minutes, still fetches everything afresh [see `PR #2433 `_] diff --git a/documentation/cli/change_log.rst b/documentation/cli/change_log.rst index ca09478a84..50cddf05fb 100644 --- a/documentation/cli/change_log.rst +++ b/documentation/cli/change_log.rst @@ -13,6 +13,7 @@ since v1.0.0 | August 11, 2026 * Add ``flexmeasures edit secret`` to store an encrypted secret on an account or asset. * Add ``flexmeasures delete secret`` to remove an encrypted secret from an account or asset. * Add ``flexmeasures add automation``, ``flexmeasures edit automation`` and ``flexmeasures delete automation`` to manage automations (recurring tasks on an asset, 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. * ``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. * ``flexmeasures show data-sources`` now shows the account a data source belongs to, and lists the sensors holding data recorded by a single source with ``--show-sensors``. diff --git a/documentation/features/automations.rst b/documentation/features/automations.rst index 29e80a1770..fa5318dd41 100644 --- a/documentation/features/automations.rst +++ b/documentation/features/automations.rst @@ -44,7 +44,16 @@ 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. -No forecaster or data source is involved, so the forecaster options above do not apply to a schedule automation, and are refused when combined with ``--type scheduling``. +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. 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 656df4d289..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 @@ -105,12 +106,14 @@ def test_schedule_details_include_stored_flex_sensors( "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={"duration": "PT1H"}, + parameters=parameters, + generator_id=resolve_schedule_generator(asset.id, parameters).id, ) fresh_db.session.add(automation) fresh_db.session.commit() diff --git a/flexmeasures/cli/data_add.py b/flexmeasures/cli/data_add.py index 3f148104b8..a58d6e2bfc 100755 --- a/flexmeasures/cli/data_add.py +++ b/flexmeasures/cli/data_add.py @@ -52,7 +52,11 @@ populate_initial_structure, add_default_asset_types, ) -from flexmeasures.data.services.automations import prepare_schedule_trigger_message +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, @@ -1674,6 +1678,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( @@ -1861,13 +1888,9 @@ def add_automation( db.session.flush() generator_id = generator.id else: # scheduling - try: - AssetTriggerSchema().load( - prepare_schedule_trigger_message(parameters, asset.id) - ) - except ValidationError as e: - click.secho(f"Invalid schedule parameters: {e.messages}", **MsgStyle.ERROR) - raise click.Abort() + # 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." diff --git a/flexmeasures/cli/tests/test_automations.py b/flexmeasures/cli/tests/test_automations.py index 421cd15dcb..737f66d464 100644 --- a/flexmeasures/cli/tests/test_automations.py +++ b/flexmeasures/cli/tests/test_automations.py @@ -766,6 +766,70 @@ def test_add_automation_rejects_malformed_yaml_file( assert "Traceback" not in result.output +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 @@ -805,7 +869,13 @@ def test_add_schedule_automation(app, fresh_db, setup_dummy_data, tmp_path): select(Automation).filter_by(name="Half-day schedules") ).scalar_one() assert automation.type == "scheduling" - assert automation.generator_id is None + # 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 @@ -959,16 +1029,21 @@ def test_run_schedule_automation_dispatch(app, fresh_db, setup_dummy_data, monke """ from flexmeasures.data.models.generic_assets import GenericAsset from flexmeasures.data.services import scheduling - from flexmeasures.data.services.automations import run_automation + 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={"duration": "PT12H", "resolution": "PT15M"}, + parameters=parameters, + generator_id=resolve_schedule_generator(asset.id, parameters).id, ) fresh_db.session.add(automation) fresh_db.session.flush() 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 index defc8ab4e5..01afc0e051 100644 --- a/flexmeasures/data/migrations/versions/3e91c47b0a58_merge_schedule_automations_with_main.py +++ b/flexmeasures/data/migrations/versions/3e91c47b0a58_merge_schedule_automations_with_main.py @@ -1,18 +1,18 @@ """merge the schedule automation migrations with main -Two migrations branched off the same revision: those adding schedule automations, +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: c63896a97a8e, 84f268f5153c +Revises: 9f2b6e1d4a73, 84f268f5153c Create Date: 2026-09-02 10:30:00.000000 """ # revision identifiers, used by Alembic. revision = "3e91c47b0a58" -down_revision = ("c63896a97a8e", "84f268f5153c") +down_revision = ("9f2b6e1d4a73", "84f268f5153c") branch_labels = None depends_on = None diff --git a/flexmeasures/data/migrations/versions/5a9c0e3b7d21_allow_schedule_automations_without_generator.py b/flexmeasures/data/migrations/versions/5a9c0e3b7d21_allow_schedule_automations_without_generator.py deleted file mode 100644 index 1b728361ce..0000000000 --- a/flexmeasures/data/migrations/versions/5a9c0e3b7d21_allow_schedule_automations_without_generator.py +++ /dev/null @@ -1,29 +0,0 @@ -"""Allow schedule automations without a generator. - -Revision ID: 5a9c0e3b7d21 -Revises: 4d5e6f708192 -Create Date: 2026-08-05 12:15:00.000000 - -""" - -from alembic import op - -# revision identifiers, used by Alembic. -revision = "5a9c0e3b7d21" -down_revision = "4d5e6f708192" -branch_labels = None -depends_on = None - - -def upgrade(): - op.alter_column("automation", "generator_id", nullable=True) - op.create_check_constraint( - "forecast_generator", - "automation", - "type != 'forecasts' OR generator_id IS NOT NULL", - ) - - -def downgrade(): - op.drop_constraint("forecast_generator", "automation", type_="check") - op.alter_column("automation", "generator_id", nullable=False) diff --git a/flexmeasures/data/migrations/versions/a71d6f2c9b04_name_automation_types_after_the_task.py b/flexmeasures/data/migrations/versions/a71d6f2c9b04_name_automation_types_after_the_task.py index 3eaf225d43..68f20645d2 100644 --- 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 @@ -2,7 +2,6 @@ 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'. -The check constraint requiring a data generator for forecast automations is recreated with the new value. Revision ID: a71d6f2c9b04 Revises: 3e91c47b0a58 @@ -20,22 +19,10 @@ def upgrade(): - op.drop_constraint("forecast_generator", "automation", type_="check") op.execute("UPDATE automation SET type = 'forecasting' WHERE type = 'forecasts'") op.execute("UPDATE automation SET type = 'scheduling' WHERE type = 'schedules'") - op.create_check_constraint( - "forecast_generator", - "automation", - "type != 'forecasting' OR generator_id IS NOT NULL", - ) def downgrade(): - op.drop_constraint("forecast_generator", "automation", type_="check") op.execute("UPDATE automation SET type = 'forecasts' WHERE type = 'forecasting'") op.execute("UPDATE automation SET type = 'schedules' WHERE type = 'scheduling'") - op.create_check_constraint( - "forecast_generator", - "automation", - "type != 'forecasts' OR generator_id IS NOT NULL", - ) diff --git a/flexmeasures/data/migrations/versions/c63896a97a8e_merge_the_automation_timezone_and_.py b/flexmeasures/data/migrations/versions/c63896a97a8e_merge_the_automation_timezone_and_.py deleted file mode 100644 index 9afd0fd28e..0000000000 --- a/flexmeasures/data/migrations/versions/c63896a97a8e_merge_the_automation_timezone_and_.py +++ /dev/null @@ -1,25 +0,0 @@ -"""merge the automation timezone and schedule generator migrations - -Two migrations branched off the same revision: one adding an automation's timezone and scheduling cursor, -the other allowing a schedule automation to exist without a data generator. -They touch different columns, so this merge only rejoins them and has nothing of its own to do. - -Revision ID: c63896a97a8e -Revises: 5a9c0e3b7d21, 9f2b6e1d4a73 -Create Date: 2026-08-11 01:06:28.121631 - -""" - -# revision identifiers, used by Alembic. -revision = "c63896a97a8e" -down_revision = ("5a9c0e3b7d21", "9f2b6e1d4a73") -branch_labels = None -depends_on = None - - -def upgrade(): - pass - - -def downgrade(): - pass diff --git a/flexmeasures/data/models/automations.py b/flexmeasures/data/models/automations.py index 7fd7ec11d1..eaa7587055 100644 --- a/flexmeasures/data/models/automations.py +++ b/flexmeasures/data/models/automations.py @@ -33,18 +33,16 @@ 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. Forecast automations use a data - generator (e.g. a forecaster linked through a data source), while schedule - automations use only their stored parameters. + 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" - __table_args__ = ( - db.CheckConstraint( - "type != 'forecasting' OR generator_id IS NOT NULL", - name="forecast_generator", - ), - ) SUPPORTED_TYPES = ["forecasting", "scheduling"] # later also "reporting" @@ -73,7 +71,9 @@ class Automation(db.Model, AuthModelMixin): default=get_initial_cursor, ) active = db.Column(db.Boolean, nullable=False, default=True) - generator_id = db.Column(db.Integer, db.ForeignKey("data_source.id"), nullable=True) + generator_id = db.Column( + db.Integer, db.ForeignKey("data_source.id"), nullable=False + ) parameters = db.Column(MutableDict.as_mutable(JSONB), nullable=False, default={}) asset = db.relationship( diff --git a/flexmeasures/data/models/planning/__init__.py b/flexmeasures/data/models/planning/__init__.py index e32f75564f..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: @@ -334,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 99b0f9e4fe..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, 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 44263dd001..a1abaf4072 100644 --- a/flexmeasures/data/services/automations.py +++ b/flexmeasures/data/services/automations.py @@ -22,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, @@ -476,6 +477,37 @@ def prepare_schedule_trigger_message(parameters: dict, asset_id: int) -> dict: 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. @@ -610,6 +642,13 @@ def _run_schedule_automation(automation: Automation) -> dict[str, Any]: 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 ) diff --git a/flexmeasures/data/services/scheduling.py b/flexmeasures/data/services/scheduling.py index 61dfbc0e1c..b4802c0348 100644 --- a/flexmeasures/data/services/scheduling.py +++ b/flexmeasures/data/services/scheduling.py @@ -40,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, @@ -223,6 +223,7 @@ def create_scheduling_job( 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: """ @@ -281,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, @@ -470,6 +472,23 @@ def create_sequential_scheduling_job( ) 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 @@ -488,6 +507,7 @@ 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, @@ -793,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: @@ -881,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_automations_fresh_db.py b/flexmeasures/data/tests/test_automations_fresh_db.py index 6505f5a185..061a06f4ee 100644 --- a/flexmeasures/data/tests/test_automations_fresh_db.py +++ b/flexmeasures/data/tests/test_automations_fresh_db.py @@ -8,6 +8,7 @@ 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 @@ -19,6 +20,19 @@ ) +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() def automation_with_generator(fresh_db): asset_type = GenericAssetType(name="automation test asset type") @@ -74,21 +88,28 @@ def test_automation_requires_generator(fresh_db, automation_with_generator): fresh_db.session.commit() -def test_schedule_automation_does_not_require_generator( +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 = Automation( - asset=forecast_automation.asset, - type="scheduling", - name="generator-free schedule", + 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() - assert schedule_automation.generator_id is None + 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() @@ -111,9 +132,8 @@ def test_run_schedule_automation( flex_model = message.pop("flex-model") flex_model["sensor"] = battery.sensors[0].id - automation = Automation( - asset_id=battery.id, - type="scheduling", + automation = build_schedule_automation( + battery, name="Nightly schedules", cronstr="0 0 * * *", parameters={**message, "flex-model": [flex_model]}, @@ -151,9 +171,8 @@ def test_run_minimal_schedule_automation_with_stored_flex_config( "soc-max": "5 MWh", "power-capacity": "2 MW", } - automation = Automation( - asset=building, - type="scheduling", + automation = build_schedule_automation( + building, name="Minimal stored-flex schedule", cronstr="0 * * * *", parameters={"duration": "PT1H", "sequential": sequential}, @@ -176,6 +195,58 @@ def test_run_minimal_schedule_automation_with_stored_flex_config( 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 ): @@ -196,9 +267,8 @@ def test_minimal_schedule_automation_reports_stored_flex_sensors( "soc-max": "5 MWh", "power-capacity": "2 MW", } - automation = Automation( - asset=building, - type="scheduling", + automation = build_schedule_automation( + building, name="Minimal stored-flex sensor details", cronstr="0 * * * *", parameters={"duration": "PT1H"}, @@ -229,9 +299,8 @@ def test_schedule_automation_stats_include_descendant_jobs_once( event_resolution=timedelta(minutes=15), unit="MW", ) - schedule_automation = Automation( - asset=root, - type="scheduling", + schedule_automation = build_schedule_automation( + root, name="descendant schedules", cronstr="0 * * * *", parameters={"duration": "PT1H"}, 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])