From fafaf45450704053759ed3e1ffa3531066d387bc Mon Sep 17 00:00:00 2001 From: "F.N. Claessen" Date: Wed, 2 Sep 2026 17:59:43 +0200 Subject: [PATCH 1/8] Make the Scheduler a DataGenerator, and give every automation a generator A scheduler's data source recorded only its class, version and author, so one source described every schedule that scheduler ever made, whatever it computed. It now also records the flex config the scheduler computed under, the way a reporter's and a forecaster's source records theirs, so a schedule can be traced back to the configuration that produced it. `Scheduler` therefore subclasses `DataGenerator`, with a config of the asset and its serialized flex-model and flex-context. Timing stays out of it: start, end and resolution differ from run to run, which is what `DataGenerator._clean_parameters` already says about parameters. The config is snapshotted while still serialized, because a deserialized flex config holds sensors, quantities and time series which do not survive a round trip. `resolve_flex_config` returns the config as passed, and `StorageScheduler` overrides it to merge in what the asset tree stores, so a scheduler which does not read the asset tree keeps describing exactly what it was given. One scheduling request stays one data source: `create_sequential_scheduling_job` resolves the request's source once and hands it to each device job, so a schedule can still be retrieved per device from the request's job, rather than each device job resolving a source from its own slice of the flex-model. A schedule automation now points at such a source, so `generator_id` is required for every automation and the constraint requiring it only for forecasts is gone. That generator is derived rather than chosen: the scheduler follows from the asset and the config from the asset tree, so the runner resolves it again on every run and moves the automation when either has changed. For the same reason, a schedule automation's flex config may only describe the site and its devices: a field fixing a moment, such as `soc-at-start` or a `soc-targets` entry with a `datetime`, is refused when the automation is created, naming the field. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Lp1bUhWjEQtyDbnvRZQgQs Signed-off-by: F.N. Claessen --- documentation/changelog.rst | 6 + documentation/cli/change_log.rst | 1 + documentation/features/automations.rst | 11 +- .../tests/test_automations_api_fresh_db.py | 5 +- flexmeasures/cli/data_add.py | 39 +++++-- flexmeasures/cli/tests/test_automations.py | 81 +++++++++++++- ...require_a_generator_on_every_automation.py | 48 ++++++++ flexmeasures/data/models/automations.py | 18 ++- flexmeasures/data/models/planning/__init__.py | 83 +++++++++++++- flexmeasures/data/models/planning/storage.py | 9 ++ flexmeasures/data/schemas/scheduler_config.py | 49 ++++++++ .../data/schemas/scheduling/__init__.py | 3 + flexmeasures/data/services/automations.py | 80 +++++++++++++ flexmeasures/data/services/scheduling.py | 33 ++++-- .../data/tests/test_automations_fresh_db.py | 105 +++++++++++++++--- .../data/tests/test_scheduling_jobs.py | 11 +- .../tests/test_scheduling_simultaneous.py | 27 +++-- 17 files changed, 548 insertions(+), 61 deletions(-) create mode 100644 flexmeasures/data/migrations/versions/b8f4d2617ac9_require_a_generator_on_every_automation.py create mode 100644 flexmeasures/data/schemas/scheduler_config.py diff --git a/documentation/changelog.rst b/documentation/changelog.rst index 2b153ab8f6..50a33f8fd9 100644 --- a/documentation/changelog.rst +++ b/documentation/changelog.rst @@ -14,10 +14,16 @@ 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. + Charts then show one series per configuration used, and a query which aggregates a sensor's schedules should filter by source. + 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..f0ed68f4e6 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 it is derived rather than chosen: it is the data source describing the scheduler the asset resolves to, and the flex config that scheduler computes under. +That config is the trigger message merged with what the asset tree stores, so it changes when the asset does. +The runner therefore resolves it again on every run, and moves the automation to another data source when either the scheduler's version or the flex config has changed. +Editing an asset's flex-model is a configuration change, and shows up as such: 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..60a4a8ba5c 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.services.automations import ( + find_momentary_flex_config_fields, + 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/b8f4d2617ac9_require_a_generator_on_every_automation.py b/flexmeasures/data/migrations/versions/b8f4d2617ac9_require_a_generator_on_every_automation.py new file mode 100644 index 0000000000..60808123fa --- /dev/null +++ b/flexmeasures/data/migrations/versions/b8f4d2617ac9_require_a_generator_on_every_automation.py @@ -0,0 +1,48 @@ +"""Require a data generator on every automation. + +A schedule automation now points at the data source describing its scheduler and the flex config it computes under, +just as a forecast automation points at the one describing its forecaster and configuration, +so the column no longer has to be nullable and the constraint requiring it only for forecasts can go. + +Revision ID: b8f4d2617ac9 +Revises: a71d6f2c9b04 +Create Date: 2026-09-02 16:00:00.000000 + +""" + +from alembic import op +import sqlalchemy as sa + +# revision identifiers, used by Alembic. +revision = "b8f4d2617ac9" +down_revision = "a71d6f2c9b04" +branch_labels = None +depends_on = None + + +def upgrade(): + connection = op.get_bind() + without_generator = connection.execute( + sa.text("SELECT id, name FROM automation WHERE generator_id IS NULL") + ).fetchall() + if without_generator: + listing = ", ".join(f"{row.id} ('{row.name}')" for row in without_generator) + raise RuntimeError( + "These automations have no data generator, which this revision makes mandatory: " + f"{listing}." + " They are schedule automations created before a schedule automation resolved its scheduler's data source." + " Resolving one takes the scheduler and the asset's flex config, which this migration cannot do," + " so recreate them with `flexmeasures add automation` (`flexmeasures delete automation --id ` removes one)," + " and run this migration again." + ) + op.drop_constraint("forecast_generator", "automation", type_="check") + op.alter_column("automation", "generator_id", nullable=False) + + +def downgrade(): + op.alter_column("automation", "generator_id", nullable=True) + op.create_check_constraint( + "forecast_generator", + "automation", + "type != 'forecasting' OR generator_id IS NOT NULL", + ) diff --git a/flexmeasures/data/models/automations.py b/flexmeasures/data/models/automations.py index 7fd7ec11d1..41e4911928 100644 --- a/flexmeasures/data/models/automations.py +++ b/flexmeasures/data/models/automations.py @@ -33,18 +33,14 @@ 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, while a schedule automation's + follows from its asset, so the runner resolves that one 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 +69,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..ee5f5599c1 100644 --- a/flexmeasures/data/models/planning/__init__.py +++ b/flexmeasures/data/models/planning/__init__.py @@ -1,6 +1,8 @@ from __future__ import annotations from collections import defaultdict +from copy import deepcopy +import json from collections.abc import Iterable from dataclasses import dataclass, field from datetime import datetime, timedelta @@ -12,8 +14,10 @@ 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.scheduler_config import SchedulerConfigSchema from flexmeasures.utils.coding_utils import deprecated, merge_or_append from .devices import INFLEXIBLE_DEVICE_KEYS from .exceptions import WrongEntityException @@ -21,6 +25,16 @@ SchedulerOutputType = pd.Series | list[dict[str, Any]] | None +def _json_safe(value: Any) -> Any: + """Return `value` with anything JSON cannot hold replaced by its string form. + + 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 quantities and sensors instead, + and those are recorded by how they print, so that the configuration is still described rather than lost. + """ + return json.loads(json.dumps(value, default=str)) + + def _shadow_inflexible_device_keys( db_flex_context: dict, passed_flex_context: dict | list | None ) -> None: @@ -50,7 +64,7 @@ def _shadow_inflexible_device_keys( db_flex_context.pop(key, None) -class Scheduler: +class Scheduler(DataGenerator): """ Superclass for all FlexMeasures Schedulers. @@ -70,6 +84,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 +209,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 +264,56 @@ 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, + "flex-model": deepcopy(self.flex_model), + "flex-context": deepcopy(self.flex_context), + } + + @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 +329,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 +411,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/schemas/scheduler_config.py b/flexmeasures/data/schemas/scheduler_config.py new file mode 100644 index 0000000000..8ff3f47d91 --- /dev/null +++ b/flexmeasures/data/schemas/scheduler_config.py @@ -0,0 +1,49 @@ +"""The schema describing a scheduler's configuration. + +This lives apart from the rest of the scheduling schemas because ``flexmeasures.data.models.planning`` imports it, +and that module is imported while the ``flexmeasures`` package itself is still initialising. +""" + +from __future__ import annotations + +from marshmallow import Schema, fields + + +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/__init__.py b/flexmeasures/data/schemas/scheduling/__init__.py index 99b0f9e4fe..181684790b 100644 --- a/flexmeasures/data/schemas/scheduling/__init__.py +++ b/flexmeasures/data/schemas/scheduling/__init__.py @@ -21,6 +21,9 @@ from flexmeasures import Sensor from flexmeasures.data.schemas.generic_assets import GenericAssetIdField +from flexmeasures.data.schemas.scheduler_config import ( # noqa: F401 + SchedulerConfigSchema, +) # noqa: F401 from flexmeasures.data.schemas.sensors import ( VariableQuantityField, SensorIdField, diff --git a/flexmeasures/data/services/automations.py b/flexmeasures/data/services/automations.py index 44263dd001..197d964d6b 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,78 @@ def prepare_schedule_trigger_message(parameters: dict, asset_id: int) -> dict: return message +#: 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. +_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 _find_momentary_flex_fields(value, path: str) -> list[str]: + """Find the fields under `value` which fix a moment in time, reporting each by its path in the flex config.""" + if isinstance(value, dict): + if any(key in value for key in _MOMENT_KEYS): + return [path] + found = [] + for key, item in value.items(): + if key in _MOMENTARY_FLEX_FIELDS: + found.append(f"{path}.{key}") + continue + found += _find_momentary_flex_fields(item, f"{path}.{key}") + return found + if isinstance(value, list): + found = [] + for index, item in enumerate(value): + found += _find_momentary_flex_fields(item, f"{path}[{index}]") + return found + return [] + + +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, and its data source records the config it computes under, + so a value tied to a fixed moment is both stale on the next run and misleading as a description of the automation. + 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"): + found += _find_momentary_flex_fields(message.get(key), key) + return sorted(set(found)) + + +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 +683,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..16d9f23bf9 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_id: int | 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_id=data_source_id, **scheduler_kwargs, ), id=job_id, @@ -470,6 +472,20 @@ 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. + 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_id = request_scheduler.data_source.id + jobs = [] previous_sensors = [] previous_job = depends_on @@ -488,6 +504,7 @@ def create_sequential_scheduling_job( job = create_scheduling_job( **current_scheduler_kwargs, + data_source_id=data_source_id, scheduler_specs=scheduler_specs, requeue=requeue, job_id=job_id, @@ -793,6 +810,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_id: int | None = None, dry_run: bool = False, **scheduler_kwargs: dict, ) -> dict: @@ -881,12 +899,13 @@ 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 source of the request it belongs to, + # so that one request records one schedule per sensor, rather than one per device's own config. + if data_source_id is not None: + data_source = db.session.get(DataSource, data_source_id) + else: + 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_simultaneous.py b/flexmeasures/data/tests/test_scheduling_simultaneous.py index ebf4c4cd8e..58e2150e04 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]) From 414f4303fe00cdfbc9ae391b5a43d0c1e4a8b727 Mon Sep 17 00:00:00 2001 From: "F.N. Claessen" Date: Thu, 3 Sep 2026 13:09:01 +0200 Subject: [PATCH 2/8] Address review of the scheduler data generator Keep values that describe one moment out of a scheduler's data source. A `soc-at-start`, or a `soc-targets` entry at a given datetime, differs on every trigger, so recording it would have made every schedule the work of a brand new data source. Only what the site and its devices can do now tells one scheduler source from another. The rule that already refused such fields on a schedule automation is the same one, so both now live next to the config schema they are about. That schema moves back in with the other scheduling schemas, into a submodule of its own. The circular import which had pushed it out is fixed at its root instead: three scheduling schema modules imported `Sensor` and `Asset` from the `flexmeasures` package root, which is still initialising while they load, so they now import from the modules that define them. The migration relaxing the generator constraint and the one restoring it cancel out, and both were unreleased, so they are gone, along with the merge revision that only existed to rejoin the relaxing one. `generator_id` keeps the NOT NULL it was created with, the type rename no longer has a constraint to recreate, and the automation feature arrives in one migration rather than three. Also, fail with the source's id when a job is told to record under a data source that no longer exists, rather than an AttributeError one frame later, and reflow the comments and docstrings which broke mid-phrase. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Lp1bUhWjEQtyDbnvRZQgQs Signed-off-by: F.N. Claessen --- documentation/changelog.rst | 4 +- documentation/features/automations.rst | 8 +- flexmeasures/cli/data_add.py | 2 +- ...58_merge_schedule_automations_with_main.py | 6 +- ..._schedule_automations_without_generator.py | 29 ----- ...04_name_automation_types_after_the_task.py | 13 -- ...require_a_generator_on_every_automation.py | 48 -------- ...7a8e_merge_the_automation_timezone_and_.py | 25 ---- flexmeasures/data/models/automations.py | 10 +- flexmeasures/data/models/planning/__init__.py | 9 +- .../planning/tests/test_utils_fresh_db.py | 40 ++++++ flexmeasures/data/schemas/scheduler_config.py | 49 -------- .../data/schemas/scheduling/__init__.py | 8 +- .../data/schemas/scheduling/config.py | 114 ++++++++++++++++++ .../data/schemas/scheduling/groups.py | 2 +- .../data/schemas/scheduling/storage.py | 3 +- flexmeasures/data/schemas/scheduling/utils.py | 2 +- flexmeasures/data/services/automations.py | 41 ------- flexmeasures/data/services/scheduling.py | 12 +- .../tests/test_scheduling_simultaneous.py | 4 +- 20 files changed, 196 insertions(+), 233 deletions(-) delete mode 100644 flexmeasures/data/migrations/versions/5a9c0e3b7d21_allow_schedule_automations_without_generator.py delete mode 100644 flexmeasures/data/migrations/versions/b8f4d2617ac9_require_a_generator_on_every_automation.py delete mode 100644 flexmeasures/data/migrations/versions/c63896a97a8e_merge_the_automation_timezone_and_.py delete mode 100644 flexmeasures/data/schemas/scheduler_config.py create mode 100644 flexmeasures/data/schemas/scheduling/config.py diff --git a/documentation/changelog.rst b/documentation/changelog.rst index 50a33f8fd9..f3a650f674 100644 --- a/documentation/changelog.rst +++ b/documentation/changelog.rst @@ -16,7 +16,9 @@ v1.1.0 | September XX, 2026 .. 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. - Charts then show one series per configuration used, and a query which aggregates a sensor's schedules should filter by source. + 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 a schedule for the same period from each configuration, so a chart draws both, and a query which sums a sensor's scheduled power over that period adds both up unless it selects a source. One scheduling request still records under a single data source, including the per-device jobs of a sequential schedule. New features diff --git a/documentation/features/automations.rst b/documentation/features/automations.rst index f0ed68f4e6..fa5318dd41 100644 --- a/documentation/features/automations.rst +++ b/documentation/features/automations.rst @@ -46,10 +46,10 @@ Use the canonical API field names, including ``flex-model``, ``flex-context`` an 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 it is derived rather than chosen: it is the data source describing the scheduler the asset resolves to, and the flex config that scheduler computes under. -That config is the trigger message merged with what the asset tree stores, so it changes when the asset does. -The runner therefore resolves it again on every run, and moves the automation to another data source when either the scheduler's version or the flex config has changed. -Editing an asset's flex-model is a configuration change, and shows up as such: the schedules computed before and after it carry different data sources. +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. diff --git a/flexmeasures/cli/data_add.py b/flexmeasures/cli/data_add.py index 60a4a8ba5c..a58d6e2bfc 100755 --- a/flexmeasures/cli/data_add.py +++ b/flexmeasures/cli/data_add.py @@ -52,8 +52,8 @@ populate_initial_structure, add_default_asset_types, ) +from flexmeasures.data.schemas.scheduling import find_momentary_flex_config_fields from flexmeasures.data.services.automations import ( - find_momentary_flex_config_fields, prepare_schedule_trigger_message, resolve_schedule_generator, ) 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/b8f4d2617ac9_require_a_generator_on_every_automation.py b/flexmeasures/data/migrations/versions/b8f4d2617ac9_require_a_generator_on_every_automation.py deleted file mode 100644 index 60808123fa..0000000000 --- a/flexmeasures/data/migrations/versions/b8f4d2617ac9_require_a_generator_on_every_automation.py +++ /dev/null @@ -1,48 +0,0 @@ -"""Require a data generator on every automation. - -A schedule automation now points at the data source describing its scheduler and the flex config it computes under, -just as a forecast automation points at the one describing its forecaster and configuration, -so the column no longer has to be nullable and the constraint requiring it only for forecasts can go. - -Revision ID: b8f4d2617ac9 -Revises: a71d6f2c9b04 -Create Date: 2026-09-02 16:00:00.000000 - -""" - -from alembic import op -import sqlalchemy as sa - -# revision identifiers, used by Alembic. -revision = "b8f4d2617ac9" -down_revision = "a71d6f2c9b04" -branch_labels = None -depends_on = None - - -def upgrade(): - connection = op.get_bind() - without_generator = connection.execute( - sa.text("SELECT id, name FROM automation WHERE generator_id IS NULL") - ).fetchall() - if without_generator: - listing = ", ".join(f"{row.id} ('{row.name}')" for row in without_generator) - raise RuntimeError( - "These automations have no data generator, which this revision makes mandatory: " - f"{listing}." - " They are schedule automations created before a schedule automation resolved its scheduler's data source." - " Resolving one takes the scheduler and the asset's flex config, which this migration cannot do," - " so recreate them with `flexmeasures add automation` (`flexmeasures delete automation --id ` removes one)," - " and run this migration again." - ) - op.drop_constraint("forecast_generator", "automation", type_="check") - op.alter_column("automation", "generator_id", nullable=False) - - -def downgrade(): - op.alter_column("automation", "generator_id", nullable=True) - op.create_check_constraint( - "forecast_generator", - "automation", - "type != 'forecasting' 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 41e4911928..eaa7587055 100644 --- a/flexmeasures/data/models/automations.py +++ b/flexmeasures/data/models/automations.py @@ -33,11 +33,13 @@ 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. Every automation has a data generator, - linked through a data source: a forecaster and its configuration for a forecast automation, + 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, while a schedule automation's - follows from its asset, so the runner resolves that one afresh on every run. + 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" diff --git a/flexmeasures/data/models/planning/__init__.py b/flexmeasures/data/models/planning/__init__.py index ee5f5599c1..6a45c17b80 100644 --- a/flexmeasures/data/models/planning/__init__.py +++ b/flexmeasures/data/models/planning/__init__.py @@ -17,7 +17,10 @@ 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.scheduler_config import SchedulerConfigSchema +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 @@ -282,8 +285,8 @@ def resolve_flex_config(self) -> dict: asset_id = None return { "asset": asset_id, - "flex-model": deepcopy(self.flex_model), - "flex-context": deepcopy(self.flex_context), + "flex-model": strip_momentary_flex_fields(deepcopy(self.flex_model)), + "flex-context": strip_momentary_flex_fields(deepcopy(self.flex_context)), } @property 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..50359d2955 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,46 @@ 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_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/scheduler_config.py b/flexmeasures/data/schemas/scheduler_config.py deleted file mode 100644 index 8ff3f47d91..0000000000 --- a/flexmeasures/data/schemas/scheduler_config.py +++ /dev/null @@ -1,49 +0,0 @@ -"""The schema describing a scheduler's configuration. - -This lives apart from the rest of the scheduling schemas because ``flexmeasures.data.models.planning`` imports it, -and that module is imported while the ``flexmeasures`` package itself is still initialising. -""" - -from __future__ import annotations - -from marshmallow import Schema, fields - - -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/__init__.py b/flexmeasures/data/schemas/scheduling/__init__.py index 181684790b..b40b35e32c 100644 --- a/flexmeasures/data/schemas/scheduling/__init__.py +++ b/flexmeasures/data/schemas/scheduling/__init__.py @@ -18,12 +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.scheduler_config import ( # noqa: F401 +from flexmeasures.data.schemas.scheduling.config import ( # noqa: F401 SchedulerConfigSchema, -) # noqa: F401 + 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..0d7f0bea03 --- /dev/null +++ b/flexmeasures/data/schemas/scheduling/config.py @@ -0,0 +1,114 @@ +"""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 _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 + item, item_found = _walk_momentary_fields(item, f"{path}.{key}", drop) + found += item_found + kept[key] = item + 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/services/automations.py b/flexmeasures/data/services/automations.py index 197d964d6b..a1abaf4072 100644 --- a/flexmeasures/data/services/automations.py +++ b/flexmeasures/data/services/automations.py @@ -477,47 +477,6 @@ def prepare_schedule_trigger_message(parameters: dict, asset_id: int) -> dict: return message -#: 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. -_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 _find_momentary_flex_fields(value, path: str) -> list[str]: - """Find the fields under `value` which fix a moment in time, reporting each by its path in the flex config.""" - if isinstance(value, dict): - if any(key in value for key in _MOMENT_KEYS): - return [path] - found = [] - for key, item in value.items(): - if key in _MOMENTARY_FLEX_FIELDS: - found.append(f"{path}.{key}") - continue - found += _find_momentary_flex_fields(item, f"{path}.{key}") - return found - if isinstance(value, list): - found = [] - for index, item in enumerate(value): - found += _find_momentary_flex_fields(item, f"{path}[{index}]") - return found - return [] - - -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, and its data source records the config it computes under, - so a value tied to a fixed moment is both stale on the next run and misleading as a description of the automation. - 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"): - found += _find_momentary_flex_fields(message.get(key), key) - return sorted(set(found)) - - 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. diff --git a/flexmeasures/data/services/scheduling.py b/flexmeasures/data/services/scheduling.py index 16d9f23bf9..b71ae8c47d 100644 --- a/flexmeasures/data/services/scheduling.py +++ b/flexmeasures/data/services/scheduling.py @@ -472,10 +472,10 @@ 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. + # 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. request_scheduler = get_scheduler_instance( scheduler_class=scheduler_class, asset_or_sensor=asset, @@ -904,6 +904,10 @@ def make_schedule( # noqa: C901 # so that one request records one schedule per sensor, rather than one per device's own config. if data_source_id is not None: data_source = db.session.get(DataSource, data_source_id) + if data_source is None: + raise ValueError( + f"Data source {data_source_id}, which this job was told to record its schedule under, no longer exists." + ) else: data_source = scheduler.data_source diff --git a/flexmeasures/data/tests/test_scheduling_simultaneous.py b/flexmeasures/data/tests/test_scheduling_simultaneous.py index 58e2150e04..5fefade6c4 100644 --- a/flexmeasures/data/tests/test_scheduling_simultaneous.py +++ b/flexmeasures/data/tests/test_scheduling_simultaneous.py @@ -81,8 +81,8 @@ def test_create_simultaneous_jobs( assert job.get_status() == "finished" # 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. + # 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 From 3fa582438e447ce6db3b17cdf959b989e5d6f33c Mon Sep 17 00:00:00 2001 From: "F.N. Claessen" Date: Thu, 3 Sep 2026 14:06:50 +0200 Subject: [PATCH 3/8] docs/changelog: say what a second schedule source means for reading a sensor The warning read as if summing two schedules of one period were a thing anyone would want to do, which made it sound like a defect rather than a change in provenance. It now says what actually changes: the newer schedule used to supersede the older one, and now both are kept, so a chart draws both and the asset's KPIs total both, because they deliberately report what the chart draws. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Lp1bUhWjEQtyDbnvRZQgQs Signed-off-by: F.N. Claessen --- documentation/changelog.rst | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/documentation/changelog.rst b/documentation/changelog.rst index f3a650f674..2f778ab043 100644 --- a/documentation/changelog.rst +++ b/documentation/changelog.rst @@ -18,7 +18,8 @@ v1.1.0 | September XX, 2026 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 a schedule for the same period from each configuration, so a chart draws both, and a query which sums a sensor's scheduled power over that period adds both up unless it selects a source. + 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 From e2dd6ee8ab29fba43c6255a752188cd2d40fce9e Mon Sep 17 00:00:00 2001 From: "F.N. Claessen" Date: Thu, 3 Sep 2026 14:27:54 +0200 Subject: [PATCH 4/8] Keep a scheduler's data source identity stable Two ways the recorded config could differ while describing the same configuration, both from the Copilot review. A caller which deserialized the flex config first hands over sensors and assets, which were recorded by how they print. A sensor prints as its name, so renaming one described a different configuration, and two sensors sharing a name described the same one. They are now recorded by their id, which is what the serialized flex config names them by. A single moment may be written as one mapping or as a list of them, and stripping the momentary values left a null behind in the first case and an empty list in the second, so the same configuration looked like three different ones depending on how it was written. A field that stripping empties is now left out altogether, which is what leaving it out of the trigger message does too. A field that arrived empty stays as it was given, and a list with static entries beside momentary ones keeps them. The snapshot no longer deep-copies. It goes straight through the JSON-safe encoder, which yields plain structures, where the copy used to carry sensors along and be read after a commit had expired them. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Lp1bUhWjEQtyDbnvRZQgQs Signed-off-by: F.N. Claessen --- flexmeasures/data/models/planning/__init__.py | 26 +++++++--- .../planning/tests/test_utils_fresh_db.py | 44 +++++++++++++++++ .../data/schemas/scheduling/config.py | 18 ++++++- .../schemas/tests/test_scheduler_config.py | 49 +++++++++++++++++++ 4 files changed, 128 insertions(+), 9 deletions(-) create mode 100644 flexmeasures/data/schemas/tests/test_scheduler_config.py diff --git a/flexmeasures/data/models/planning/__init__.py b/flexmeasures/data/models/planning/__init__.py index 6a45c17b80..0c0c90806b 100644 --- a/flexmeasures/data/models/planning/__init__.py +++ b/flexmeasures/data/models/planning/__init__.py @@ -1,7 +1,6 @@ from __future__ import annotations from collections import defaultdict -from copy import deepcopy import json from collections.abc import Iterable from dataclasses import dataclass, field @@ -29,13 +28,23 @@ def _json_safe(value: Any) -> Any: - """Return `value` with anything JSON cannot hold replaced by its string form. + """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 quantities and sensors instead, - and those are recorded by how they print, so that the configuration is still described rather than lost. + 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. """ - return json.loads(json.dumps(value, default=str)) + + 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( @@ -285,8 +294,11 @@ def resolve_flex_config(self) -> dict: asset_id = None return { "asset": asset_id, - "flex-model": strip_momentary_flex_fields(deepcopy(self.flex_model)), - "flex-context": strip_momentary_flex_fields(deepcopy(self.flex_context)), + # 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)), } @property 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 50359d2955..e2ce1c0fad 100644 --- a/flexmeasures/data/models/planning/tests/test_utils_fresh_db.py +++ b/flexmeasures/data/models/planning/tests/test_utils_fresh_db.py @@ -352,6 +352,50 @@ def source_for(soc_at_start: str, power_capacity: str = "2 MW"): 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/config.py b/flexmeasures/data/schemas/scheduling/config.py index 0d7f0bea03..7a8e147edb 100644 --- a/flexmeasures/data/schemas/scheduling/config.py +++ b/flexmeasures/data/schemas/scheduling/config.py @@ -18,6 +18,15 @@ def _describes_a_moment(value) -> bool: 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. @@ -35,9 +44,14 @@ def _walk_momentary_fields(value, path: str, drop: bool): continue kept[key] = item continue - item, item_found = _walk_momentary_fields(item, f"{path}.{key}", drop) + stripped, item_found = _walk_momentary_fields(item, f"{path}.{key}", drop) found += item_found - kept[key] = item + 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 = [] 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]"] From 5d6540dc37250e979b1237942a8e03b0d49ab8cf Mon Sep 17 00:00:00 2001 From: "F.N. Claessen" Date: Thu, 3 Sep 2026 15:18:12 +0200 Subject: [PATCH 5/8] Send the request's configuration to the device jobs, not a data source id FlexMeasures does not auto-commit the session of the request that enqueues a scheduling job (see `flexmeasures.data.transactional`), and the schedule trigger endpoint does not commit either. A data source resolved while enqueueing therefore lives in a transaction the workers never see, so handing its id to the device jobs of a sequential schedule would have failed to find it. The jobs now carry the request's configuration instead, and each worker resolves the data source from it and commits, as `make_schedule` already does for everything else it writes. The device jobs of one request still describe one configuration, so they still share one source. A test pins that enqueueing writes no data source of its own, and that the device jobs carry the same configuration. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Lp1bUhWjEQtyDbnvRZQgQs Signed-off-by: F.N. Claessen --- flexmeasures/data/models/planning/__init__.py | 18 ++++-- flexmeasures/data/services/scheduling.py | 28 +++++----- .../data/tests/test_scheduling_sequential.py | 56 +++++++++++++++++++ 3 files changed, 84 insertions(+), 18 deletions(-) diff --git a/flexmeasures/data/models/planning/__init__.py b/flexmeasures/data/models/planning/__init__.py index 0c0c90806b..aff867f8d2 100644 --- a/flexmeasures/data/models/planning/__init__.py +++ b/flexmeasures/data/models/planning/__init__.py @@ -84,11 +84,12 @@ class Scheduler(DataGenerator): 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 + (under what constraints can the schedule be optimized?), + and the system's flexibility context (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. @@ -301,6 +302,15 @@ def resolve_flex_config(self) -> dict: "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. diff --git a/flexmeasures/data/services/scheduling.py b/flexmeasures/data/services/scheduling.py index b71ae8c47d..946481229e 100644 --- a/flexmeasures/data/services/scheduling.py +++ b/flexmeasures/data/services/scheduling.py @@ -223,7 +223,7 @@ def create_scheduling_job( depends_on: Job | list[Job] | None = None, success_callback: Callable | None = None, trigger: dict | None = None, - data_source_id: int | None = None, + data_source_config: dict | None = None, **scheduler_kwargs, ) -> Job: """ @@ -282,7 +282,7 @@ def create_scheduling_job( kwargs=dict( asset_or_sensor=asset_or_sensor, scheduler_specs=scheduler_specs, - data_source_id=data_source_id, + data_source_config=data_source_config, **scheduler_kwargs, ), id=job_id, @@ -476,6 +476,9 @@ def create_sequential_scheduling_job( # 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, @@ -484,7 +487,7 @@ def create_sequential_scheduling_job( "flex_model": MultiSensorFlexModelSchema(many=True).dump(flex_model), }, ) - data_source_id = request_scheduler.data_source.id + data_source_config = request_scheduler.resolve_flex_config() jobs = [] previous_sensors = [] @@ -504,7 +507,7 @@ def create_sequential_scheduling_job( job = create_scheduling_job( **current_scheduler_kwargs, - data_source_id=data_source_id, + data_source_config=data_source_config, scheduler_specs=scheduler_specs, requeue=requeue, job_id=job_id, @@ -810,7 +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_id: int | None = None, + data_source_config: dict | None = None, dry_run: bool = False, **scheduler_kwargs: dict, ) -> dict: @@ -900,16 +903,13 @@ def make_schedule( # noqa: C901 rq_job.meta["scheduler_info"] = scheduler.info # 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 source of the request it belongs to, + # 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. - if data_source_id is not None: - data_source = db.session.get(DataSource, data_source_id) - if data_source is None: - raise ValueError( - f"Data source {data_source_id}, which this job was told to record its schedule under, no longer exists." - ) - else: - data_source = scheduler.data_source + # It is handed the configuration rather than a data source id, because the 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_scheduling_sequential.py b/flexmeasures/data/tests/test_scheduling_sequential.py index f219c62b60..b7a1251ae4 100644 --- a/flexmeasures/data/tests/test_scheduling_sequential.py +++ b/flexmeasures/data/tests/test_scheduling_sequential.py @@ -3,12 +3,68 @@ 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 + (see `flexmeasures.data.transactional`), so a data source created while enqueueing would never reach the workers. + 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, + ) + + device_jobs = [ + Job.fetch(job_id, connection=queue.connection) + for job_id in queue.job_ids + if Job.fetch(job_id, connection=queue.connection).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 + ) + + def test_create_sequential_jobs(db, app, flex_description_sequential, smart_building): """Test sequential scheduling capabilities. From 9039cde2796991af4a7c2306d24648a923b931a4 Mon Sep 17 00:00:00 2001 From: "F.N. Claessen" Date: Thu, 3 Sep 2026 20:07:42 +0200 Subject: [PATCH 6/8] tests: clear Redis after the test that only queues jobs The new sequential test never runs what it queues, so it left the jobs, the queue's deferred registry and the job cache behind. The job cache was the one that bit: a scheduling job's id is derived from what it schedules, so the next test's identical request was skipped as already made, and it saw no jobs queued. Also reflow a docstring line that ended mid-phrase. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01LtMZ49GH6LaiY5MdgtfNe2 Signed-off-by: F.N. Claessen --- flexmeasures/data/models/planning/__init__.py | 6 +++--- flexmeasures/data/tests/test_scheduling_sequential.py | 5 +++++ 2 files changed, 8 insertions(+), 3 deletions(-) diff --git a/flexmeasures/data/models/planning/__init__.py b/flexmeasures/data/models/planning/__init__.py index aff867f8d2..fdc0a808bc 100644 --- a/flexmeasures/data/models/planning/__init__.py +++ b/flexmeasures/data/models/planning/__init__.py @@ -87,9 +87,9 @@ class Scheduler(DataGenerator): 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. diff --git a/flexmeasures/data/tests/test_scheduling_sequential.py b/flexmeasures/data/tests/test_scheduling_sequential.py index b7a1251ae4..74522bca6d 100644 --- a/flexmeasures/data/tests/test_scheduling_sequential.py +++ b/flexmeasures/data/tests/test_scheduling_sequential.py @@ -64,6 +64,11 @@ def test_sequential_jobs_carry_the_request_config_not_a_data_source_id( == 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. From bcf75707e9a3bd0c4d9f959179f52d0cebcf6517 Mon Sep 17 00:00:00 2001 From: "F.N. Claessen" Date: Thu, 3 Sep 2026 21:06:36 +0200 Subject: [PATCH 7/8] docs: reflow two lines that broke mid-phrase Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01LtMZ49GH6LaiY5MdgtfNe2 Signed-off-by: F.N. Claessen --- flexmeasures/data/services/scheduling.py | 5 +++-- flexmeasures/data/tests/test_scheduling_sequential.py | 4 ++-- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/flexmeasures/data/services/scheduling.py b/flexmeasures/data/services/scheduling.py index 946481229e..b4802c0348 100644 --- a/flexmeasures/data/services/scheduling.py +++ b/flexmeasures/data/services/scheduling.py @@ -905,8 +905,9 @@ def make_schedule( # noqa: C901 # 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 the id would come from a row - # created while the request that enqueued this job was still open, and this session never sees it. + # 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 diff --git a/flexmeasures/data/tests/test_scheduling_sequential.py b/flexmeasures/data/tests/test_scheduling_sequential.py index 74522bca6d..2423d48901 100644 --- a/flexmeasures/data/tests/test_scheduling_sequential.py +++ b/flexmeasures/data/tests/test_scheduling_sequential.py @@ -17,8 +17,8 @@ def test_sequential_jobs_carry_the_request_config_not_a_data_source_id( ): """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 - (see `flexmeasures.data.transactional`), so a data source created while enqueueing would never reach the workers. + 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 From 4e348cc3519c324f5e2d77dae4438bcd0e0ba6a5 Mon Sep 17 00:00:00 2001 From: "F.N. Claessen" Date: Thu, 3 Sep 2026 21:36:30 +0200 Subject: [PATCH 8/8] tests: fetch each queued job once Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01LtMZ49GH6LaiY5MdgtfNe2 Signed-off-by: F.N. Claessen --- flexmeasures/data/tests/test_scheduling_sequential.py | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/flexmeasures/data/tests/test_scheduling_sequential.py b/flexmeasures/data/tests/test_scheduling_sequential.py index 2423d48901..ba3051eccd 100644 --- a/flexmeasures/data/tests/test_scheduling_sequential.py +++ b/flexmeasures/data/tests/test_scheduling_sequential.py @@ -43,11 +43,10 @@ def test_sequential_jobs_carry_the_request_config_not_a_data_source_id( **flex_description_sequential, ) - device_jobs = [ - Job.fetch(job_id, connection=queue.connection) - for job_id in queue.job_ids - if Job.fetch(job_id, connection=queue.connection).kwargs.get("asset_or_sensor") + 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(