diff --git a/documentation/changelog.rst b/documentation/changelog.rst index 4025664a6f..a15e209916 100644 --- a/documentation/changelog.rst +++ b/documentation/changelog.rst @@ -20,7 +20,7 @@ New features * Floor off-clock API datetimes to a non-instantaneous sensor's resolution by default when ingesting sensor data, uploading sensor data, and handling scheduler flex-model timed events; configurable with the ``floor_datetimes_to_resolution`` sensor attribute [see `PR #2146 `_] * Sensor references in flex-model and flex-context support various ways of filtering by source [see `PR #2209 `_] * Let storage scheduling infer missing ``power-capacity`` from directional device capacities before falling back to site capacity, and default the missing opposite capacity to zero when only a non-zero ``consumption-capacity`` or ``production-capacity`` is configured [see `PR #2222 `_] -* Support multiple feeders to a shared storage [see `PR #2001 `_ ] +* Support multiple feeders to a shared storage [see `PR #2001 `_ and `PR #2321 `_] * The flex-context can now define multiple commodities, each specifying their own prices and grid capacities [see `PR #1946 `_, `PR #2172 `_, `PR #2235 `_ and `PR #2271 `_] * CLI support for adding/editing account attributes [see `PR #2242 `_] * Improve chart axis domain for event values not around zero, with a per-sub-chart ``y-axis`` option in ``sensors_to_show`` (default ``zero``, which pads the axis out to include zero) that can be set to ``data`` to fit a sub-chart's y-axis to the values shown, to an explicit ``[min, max]`` domain that the axis will cover at least (expanding to fit data beyond it), or to a strict ``{"min": min, "max": max}`` domain that the axis will never exceed (clamping data beyond it, with a warning when that happens), editable from the graph editor [see `PR #2244 `_] @@ -43,6 +43,7 @@ Infrastructure / Support * Warn hosts when the database schema is not at the latest migration, and skip startup template provisioning until migrations are applied [see `PR #2309 `_] * Stop manual runs of the Docker publishing workflow from overwriting the ``latest`` image tag, and let them opt in to it explicitly [see `PR #2316 `_] * Add a pre-commit hook that blocks image files (png, jpg, gif, bmp, tiff, webp, ico, psd) from being committed outside of ``flexmeasures/ui/static/`` and ``documentation/``, to protect the git history from binary bloat; screenshots belong in the ``FlexMeasures/screenshots`` repo instead [see `PR #2315 `_] +* Schedulers track devices via a typed device inventory, which classifies every flex-model entry once and serves as the single source of truth for device roles and canonical device indices [see `PR #2321 `_] Bugfixes ----------- diff --git a/flexmeasures/conftest.py b/flexmeasures/conftest.py index 8ef85d5dc3..8be3b72480 100644 --- a/flexmeasures/conftest.py +++ b/flexmeasures/conftest.py @@ -1371,6 +1371,7 @@ def create_charging_station_assets( ) db.session.add(bi_soc) db.session.add(uni_soc) + db.session.flush() # assign IDs, so tests don't depend on earlier tests flushing return { "Test charging station": charging_station, "Test charging station (bidirectional)": bidirectional_charging_station, diff --git a/flexmeasures/data/models/generic_assets.py b/flexmeasures/data/models/generic_assets.py index 7b5e57c447..0e0ceef93c 100644 --- a/flexmeasures/data/models/generic_assets.py +++ b/flexmeasures/data/models/generic_assets.py @@ -755,7 +755,9 @@ def get_flex_context(self) -> dict: flex_context = {} parent_asset = self.parent_asset while set(flex_context.keys()) != flex_context_field_names and parent_asset: - flex_context = {**parent_asset.flex_context, **flex_context} + # An ancestor's flex_context may still be None (e.g. a pending asset + # created without one, before its column default is applied on flush). + flex_context = {**(parent_asset.flex_context or {}), **flex_context} parent_asset = parent_asset.parent_asset return flex_context diff --git a/flexmeasures/data/models/planning/__init__.py b/flexmeasures/data/models/planning/__init__.py index 0f8c606d2f..5b8b20d134 100644 --- a/flexmeasures/data/models/planning/__init__.py +++ b/flexmeasures/data/models/planning/__init__.py @@ -55,6 +55,9 @@ class Scheduler: flex_model: list[dict] | dict | None = None flex_context: dict | None = None stock_groups: dict | None = None + #: Typed classification of the flex config (see planning.devices.DeviceInventory); + #: derived state, (re)built when the flex config is deserialized. + device_inventory = None fallback_scheduler_class: "Type[Scheduler] | None" = None info: dict | None = None @@ -71,6 +74,12 @@ class Scheduler: def _build_stock_groups(flex_model: list[dict]) -> dict: """ Build stock groups where devices sharing the same state-of-charge sensor are grouped together. + + Deprecated: use ``DeviceInventory.stock_groups`` (see ``planning.devices``), + which classifies flex-model entries once and keeps stock-group keys in sync + with the stock parameters. Note that this function's synthetic keys (for + devices without a state-of-charge sensor) depend on the length of the passed + list, so they only match ``stock_models`` keys built from the same list. """ groups = defaultdict(list) soc_usage = defaultdict(list) diff --git a/flexmeasures/data/models/planning/devices.py b/flexmeasures/data/models/planning/devices.py new file mode 100644 index 0000000000..b7eb4a98c2 --- /dev/null +++ b/flexmeasures/data/models/planning/devices.py @@ -0,0 +1,381 @@ +"""Typed device tracking for schedulers. + +Multi-device flex-models describe several kinds of entries (schedulable devices, +stock-only entries carrying SoC parameters for a shared stock, and — in the future — +group entries and converter ports). Historically, each scheduling feature re-derived +an entry's kind from the raw dicts and kept parallel lists aligned by integer position, +which is where several alignment bugs crept in. + +This module classifies every entry exactly once, right after flex-config +deserialization, into a :class:`DeviceInventory`: the single source of truth for + +- which entries are schedulable devices, and their canonical solver indices, +- which entries only carry SoC parameters for a shared stock (stock-only entries), +- the inflexible devices from the flex-context, in canonical solver order, and +- the stock groups (devices sharing a state-of-charge sensor). + +The raw (deserialized) flex-model dicts are kept as-is on each :class:`FlexDevice`, +so downstream code and new flex-model fields need no dataclass changes. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import Enum +from functools import cached_property +from typing import Any + +from marshmallow import ValidationError + +from flexmeasures.data.models.time_series import Sensor +from flexmeasures.data.models.generic_assets import GenericAsset as Asset + + +class DeviceRole(Enum): + """The role a flex-model (or flex-context) entry plays in the scheduling problem. + + Extension points (not yet implemented): GROUP (an entry constraining the aggregate + power of a set of member devices) and CONVERTER_PORT (a commodity port of a + multi-commodity converter). + """ + + #: A schedulable flexible device (usually with a power sensor). + DEVICE = "device" + #: An entry carrying SoC parameters for a shared stock; not itself scheduled. + STOCK_ONLY = "stock-only" + #: An inflexible device from the flex-context (scheduled with a fixed profile). + INFLEXIBLE = "inflexible" + + +@dataclass +class FlexDevice: + """A single classified flex-model (or flex-context) entry. + + Note that ``flex_model`` references the original deserialized dict (it is not a + copy); code that mutates per-device parameters should work on copies. + """ + + role: DeviceRole + #: Canonical solver device index; None for stock-only entries. + index: int | None + #: The deserialized flex-model entry (with underscore keys); None for inflexible devices. + flex_model: dict | None + #: The device's power sensor, resolved from the entry's top-level "sensor" key, else from a nested consumption/production output reference. + #: None for entries that reference no power sensor at all (e.g. asset-only entries). + power_sensor: Sensor | None + #: The device's asset, resolved from the power sensor or the entry's "asset" key. + asset: Asset | None + commodity: str = "electricity" + #: Key of the stock this device draws from: the id of its state-of-charge sensor, or a unique negative synthetic key for devices without one. + #: None for inflexible devices. + stock_key: int | None = None + + @property + def sensor_id(self) -> int | None: + if self.power_sensor is not None: + return getattr(self.power_sensor, "id", None) + return None + + @property + def state_of_charge(self) -> Any: + # A flex-model entry always has a dict; this None check is a precaution for inflexible devices, which have no flex-model entry. + if self.flex_model is None: + return None + return self.flex_model.get("state_of_charge") + + @property + def consumption_sensor(self) -> Sensor | None: + return _resolve_output_sensor(self.flex_model, "consumption") + + @property + def production_sensor(self) -> Sensor | None: + return _resolve_output_sensor(self.flex_model, "production") + + +def _resolve_output_sensor(flex_model: dict | None, output_field: str) -> Sensor | None: + """Resolve the sensor of a nested consumption/production output reference.""" + if flex_model is None: + return None + output_ref = flex_model.get(output_field) + if isinstance(output_ref, dict): + output_sensor = output_ref.get("sensor") + if isinstance(output_sensor, Sensor): + return output_sensor + # Tolerate SensorReference-like objects without importing schema modules. + if isinstance(getattr(output_sensor, "sensor", None), Sensor): + return output_sensor.sensor + return None + + +def _resolve_power_sensor(flex_model: dict) -> Sensor | None: + """Resolve a flex-model entry's power sensor. + + The top-level "sensor" key takes precedence. Devices that reference their power + sensor only via a nested output reference (e.g. ``{"consumption": {"sensor": N}}``) + resolve to that output sensor, so they are recognized as schedulable devices + (rather than being misclassified as stock-only entries and silently dropped). + """ + sensor = flex_model.get("sensor") + if sensor is not None: + return sensor + for output_field in ("consumption", "production"): + output_sensor = _resolve_output_sensor(flex_model, output_field) + if output_sensor is not None: + return output_sensor + return None + + +def _resolve_stock_key(state_of_charge: Any) -> int | None: + """Resolve a state-of-charge reference to a stock key (the SoC sensor id). + + Only a sensor reference can link devices into a shared stock. A + state-of-charge given as a value or time series (e.g. a list of timed values) + resolves to None: the device keeps a stock of its own. + """ + if state_of_charge is None: + return None + if hasattr(state_of_charge, "id"): + return state_of_charge.id + key = state_of_charge + if isinstance(state_of_charge, dict) and "sensor" in state_of_charge: + sensor = state_of_charge["sensor"] + key = sensor.id if hasattr(sensor, "id") else sensor + try: + hash(key) + except TypeError: + return None + return key + + +#: Flex-model fields that make a device entry (with a state-of-charge sensor) +#: also carry the SoC parameters of its stock. +SOC_PARAM_FIELDS = ("soc_at_start", "soc_min", "soc_max", "soc_targets") + + +@dataclass +class DeviceInventory: + """All devices of a scheduling problem, classified once, in canonical solver order. + + The canonical device enumeration is: + + 1. flexible devices (flex-model entries with role DEVICE), in flex-model order, + 2. top-level (electricity) inflexible-device-sensors from the flex-context, in order, + 3. each commodity context's own inflexible-device-sensors, in the order the + commodity contexts are given. + + This is the one enumeration both `_prepare()` and the result mapping rely on, + so they cannot drift apart. + """ + + #: All flex-model entries, in their original order (including stock-only entries). + entries: list[FlexDevice] = field(default_factory=list) + #: The schedulable devices; ``devices[d].index == d``. + devices: list[FlexDevice] = field(default_factory=list) + #: The inflexible devices from the flex-context, with indices following the devices. + inflexible_devices: list[FlexDevice] = field(default_factory=list) + #: SoC parameters per stock key. Keys are shared with :attr:`stock_groups`. + stock_entries: dict[int, dict] = field(default_factory=dict) + is_single_sensor_mode: bool = False + + @classmethod + def from_flex_config( + cls, + flex_model: list[dict] | dict, + flex_context: dict | None = None, + sensor: Sensor | None = None, + ) -> "DeviceInventory": + """Classify a deserialized flex-model (and flex-context) into an inventory. + + :param flex_model: The deserialized flex-model: a dict (single-sensor mode, + in which case ``sensor`` is the device's power sensor) + or a list of entry dicts (multi-device mode). + :param flex_context: The deserialized flex-context, used for the inflexible + devices (top-level and per commodity context). + :param sensor: The scheduler's target sensor (single-sensor mode only). + """ + flex_context = flex_context or {} + is_single_sensor_mode = not isinstance(flex_model, list) + flex_model_list = [flex_model] if is_single_sensor_mode else flex_model + + inventory = cls(is_single_sensor_mode=is_single_sensor_mode) + + # One counter yields the synthetic stock keys for devices without a + # state-of-charge sensor, so stock_entries and stock_groups always share keys. + synthetic_stock_key = -len(flex_model_list) + + def register_stock_params(stock_key: int, fm: dict) -> None: + """Register the flex-model entry holding a stock's SoC parameters, failing fast on conflicts.""" + existing = inventory.stock_entries.get(stock_key) + if existing is not None and existing is not fm: + raise ValidationError( + f"Multiple flex-model entries define state-of-charge parameters for the same stock" + f" (state-of-charge sensor {stock_key}). Please define them on a single entry." + ) + inventory.stock_entries[stock_key] = fm + + for fm in flex_model_list: + if is_single_sensor_mode: + power_sensor = sensor + else: + power_sensor = _resolve_power_sensor(fm) + state_of_charge = fm.get("state_of_charge") + stock_key = _resolve_stock_key(state_of_charge) + + # Stock-only entry: SoC parameters for a shared stock, but no power sensor + # (only in multi-device mode; in single-sensor mode the power sensor is the + # scheduler's target sensor rather than a flex-model field). + if ( + not is_single_sensor_mode + and power_sensor is None + and state_of_charge is not None + ): + if stock_key is None: + stock_key = synthetic_stock_key + synthetic_stock_key += 1 + entry = FlexDevice( + role=DeviceRole.STOCK_ONLY, + index=None, + flex_model=fm, + power_sensor=None, + asset=fm.get("asset"), + stock_key=stock_key, + ) + inventory.entries.append(entry) + register_stock_params(stock_key, fm) + continue + + # Device entry. + if stock_key is None: + stock_key = synthetic_stock_key + synthetic_stock_key += 1 + # A device without a state-of-charge *sensor* (it may still define a + # state of charge as a value or time series) keeps its own SoC + # parameters, under its synthetic stock key. + register_stock_params(stock_key, fm) + elif any(param in fm for param in SOC_PARAM_FIELDS): + # A device entry may also carry the SoC parameters of its stock itself, + # as long as only one entry per stock does. + register_stock_params(stock_key, fm) + + device = FlexDevice( + role=DeviceRole.DEVICE, + index=len(inventory.devices), + flex_model=fm, + power_sensor=power_sensor, + asset=( + power_sensor.asset if power_sensor is not None else fm.get("asset") + ), + commodity=fm.get("commodity", "electricity"), + stock_key=stock_key, + ) + inventory.entries.append(device) + inventory.devices.append(device) + + # Inflexible devices from the flex-context: top-level (electricity) sensors + # first, then each commodity context's own sensors, in context order. + index = len(inventory.devices) + for inflexible_sensor in flex_context.get("inflexible_device_sensors", []): + inventory.inflexible_devices.append( + FlexDevice( + role=DeviceRole.INFLEXIBLE, + index=index, + flex_model=None, + power_sensor=inflexible_sensor, + asset=getattr(inflexible_sensor, "asset", None), + commodity="electricity", + ) + ) + index += 1 + for commodity_context in flex_context.get("commodity_contexts", []): + commodity = commodity_context["commodity"] + for inflexible_sensor in commodity_context.get( + "inflexible_device_sensors", [] + ): + inventory.inflexible_devices.append( + FlexDevice( + role=DeviceRole.INFLEXIBLE, + index=index, + flex_model=None, + power_sensor=inflexible_sensor, + asset=getattr(inflexible_sensor, "asset", None), + commodity=commodity, + ) + ) + index += 1 + + assert all( + device.index == d for d, device in enumerate(inventory.devices) + ), "Device indices must match their position among the schedulable devices." + return inventory + + @property + def num_flexible(self) -> int: + """The number of schedulable (flexible) devices.""" + return len(self.devices) + + @property + def num_scheduled(self) -> int: + """The number of devices in the optimization problem (flexible + inflexible).""" + return len(self.devices) + len(self.inflexible_devices) + + def by_index(self, d: int) -> FlexDevice: + """Return the device with canonical solver index ``d`` (flexible or inflexible).""" + if d < len(self.devices): + return self.devices[d] + return self.inflexible_devices[d - len(self.devices)] + + def by_sensor_id(self, sensor_id: int) -> list[FlexDevice]: + """Return the flexible devices whose power sensor has the given id.""" + return [device for device in self.devices if device.sensor_id == sensor_id] + + @cached_property + def stock_groups(self) -> dict[int, list[int]]: + """Map each stock key to the indices of the devices drawing from that stock. + + Devices sharing a state-of-charge sensor are grouped together; devices without + one form singleton groups under their synthetic (negative) stock key. + """ + groups: dict[int, list[int]] = {} + for device in self.devices: + groups.setdefault(device.stock_key, []).append(device.index) + return groups + + def stock_params(self, stock_key: int) -> dict | None: + """Return the flex-model entry holding the SoC parameters of the given stock.""" + return self.stock_entries.get(stock_key) + + @cached_property + def commodity_to_devices(self) -> dict[str, list[int]]: + """Map each commodity to its device indices, in canonical solver order.""" + mapping: dict[str, list[int]] = {} + for device in self.devices: + mapping.setdefault(device.commodity, []).append(device.index) + # Inflexible devices are electricity by default, so the electricity group + # exists even when empty. + mapping.setdefault("electricity", []) + for device in self.inflexible_devices: + mapping.setdefault(device.commodity, []).append(device.index) + return mapping + + @property + def inflexible_sensors(self) -> list[Sensor]: + """The inflexible devices' power sensors, in canonical solver order. + + Inflexible devices are constructed from the flex-context's sensors, so their power sensor is always set. + """ + return [device.power_sensor for device in self.inflexible_devices] + + @property + def power_sensors(self) -> list[Sensor | None]: + """The flexible devices' power sensors, by device index.""" + return [device.power_sensor for device in self.devices] + + @property + def assets(self) -> list[Asset | None]: + """The flexible devices' assets, by device index.""" + return [device.asset for device in self.devices] + + @property + def device_flex_models(self) -> list[dict]: + """The flexible devices' flex-model entries, by device index.""" + return [device.flex_model for device in self.devices] diff --git a/flexmeasures/data/models/planning/storage.py b/flexmeasures/data/models/planning/storage.py index dc0c02b7f7..9a1f18a627 100644 --- a/flexmeasures/data/models/planning/storage.py +++ b/flexmeasures/data/models/planning/storage.py @@ -18,6 +18,7 @@ SchedulerOutputType, StockCommitment, ) +from flexmeasures.data.models.planning.devices import DeviceInventory from flexmeasures.data.models.planning.linear_optimization import device_scheduler from flexmeasures.data.models.planning.utils import ( add_tiny_price_slope, @@ -127,123 +128,43 @@ def _prepare(self, skip_validation: bool = False) -> tuple: # noqa: C901 resolution = self.resolution belief_time = self.belief_time - # For backwards compatibility with the single asset scheduler - # Track whether we started with a single dict (single-sensor mode) or a list - is_single_sensor_mode = not isinstance(self.flex_model, list) - flex_model = self.flex_model.copy() - if not isinstance(flex_model, list): - flex_model = [flex_model] - - # Identify stock models: entries not defining a power sensor, but only a (state-of-charge) sensor - self.stock_models = {} - - device_models = [] # everything except stock models - stock_models = {} # stock models only - - missing_soc_sensor_i = -len(flex_model) - for fm in flex_model: - - # stock model: entry in the flex-model list where the sensor key is the state-of-charge sensor of the device (e.g. a stock) - # Only apply this detection in multi-device mode; in single-sensor mode the power sensor is self.sensor (not in the fm dict) - if ( - not is_single_sensor_mode - and fm.get("sensor") is None - and (soc_sensor := fm.get("state_of_charge")) - ): - stock_models[ - soc_sensor.id if isinstance(soc_sensor, Sensor) else soc_sensor - ] = fm - continue - - """ - [ - { - "sensor": 1, - "charging-efficiency": 0.9, - "state-of-charge": {"sensor": 2}, - }, - { - "sensor": 3, - "charging-efficiency": 0.9, - "state-of-charge": {"sensor": 2}, - }, - { - "state-of-charge": {"sensor": 2}, - "storage-efficiency": 0.99, - }, - ] - """ - - # Check if this is a stock-only model (no power sensor) - # Stock-only entries have SOC parameters but no power sensor - # Only apply in multi-device mode; single-sensor mode devices have no "sensor" key by design - soc_sensor = fm.get("state_of_charge") - if ( - not is_single_sensor_mode - and fm.get("sensor") is None - and soc_sensor is not None - ): - # This is a stock-only entry, add to stock_models only - soc_id = soc_sensor.id if isinstance(soc_sensor, Sensor) else soc_sensor - stock_models[soc_id] = fm - continue + # Look up the device inventory: every flex-model entry (and the flex-context's + # inflexible devices) classified once, as the single source of truth for + # device roles and canonical device indices. Tests may bypass deserialization + # (setting config_deserialized) with an already deserialized flex config, + # in which case we classify it here. + inventory = self.device_inventory + if inventory is None: + inventory = DeviceInventory.from_flex_config( + self.flex_model, self.flex_context, sensor=self.sensor + ) + self.device_inventory = inventory - # device model: entry in the flex-model list where the sensor key is the power sensor of the device (e.g. a feeder) - device_models.append(fm) - - # If this device has state-of-charge parameters (soc-at-start, soc-min, etc.), - # also create a stock model entry so those parameters are properly captured - if soc_sensor is not None: - soc_id = soc_sensor.id if isinstance(soc_sensor, Sensor) else soc_sensor - # Check if there are SOC parameters in this device entry - has_soc_params = any( - param in fm - for param in ["soc_at_start", "soc_min", "soc_max", "soc_targets"] - ) - if has_soc_params: - stock_models[soc_id] = fm - elif fm.get("state_of_charge") is None: - stock_models[missing_soc_sensor_i] = fm - missing_soc_sensor_i += 1 - - flex_model = device_models - self.stock_models = stock_models + device_models = inventory.device_flex_models self._device_models = ( device_models # Store filtered model for later use in _build_soc_schedule ) - - # Rebuild stock_groups using only device_models (which have sensors) - # This ensures the mapping aligns with the device indices - self.stock_groups = self._build_stock_groups(device_models) + self.stock_models = inventory.stock_entries + # The stock groups' device indices align with the device models + self.stock_groups = inventory.stock_groups # List the asset(s) and sensor(s) being scheduled + sensors: list[Sensor | None] = inventory.power_sensors + assets: list[Asset | None] = inventory.assets if self.asset is not None: if not isinstance(self.flex_model, list): self.flex_model = [self.flex_model] - sensors: list[Sensor | None] = [fm.get("sensor") for fm in device_models] - assets: list[Asset | None] = [ # noqa: F841 - s.asset if s is not None else flex_model_d.get("asset") - for s, flex_model_d in zip(sensors, self.flex_model) - ] if resolution is None: # in case of no sensors with a non-instantaneous resolution, schedule with a 15-minute resolution resolution = determine_minimum_resampling_resolution( [s.event_resolution for s in sensors if s is not None], fallback_resolution=self.default_resolution, ) - asset = self.asset - else: - # For backwards compatibility with the single asset scheduler - sensors = [self.sensor] - asset = self.sensor.generic_asset - assets = [asset] # noqa: F841 - - # For backwards compatibility with the single asset scheduler - flex_model = self.flex_model.copy() - if not isinstance(flex_model, list): - flex_model = [flex_model] - else: - flex_model = [flex_model_d.copy() for flex_model_d in flex_model] + + # Work on copies of the device flex-models (aligned with the device indices, + # unlike the unfiltered self.flex_model), so the defaults applied here don't + # leak back into the inventory's raw entries. + flex_model = [flex_model_d.copy() for flex_model_d in device_models] for flex_model_d in flex_model: self._default_missing_directional_capacity_to_zero(flex_model_d) num_flexible_devices = len(device_models) @@ -299,12 +220,6 @@ def _prepare(self, skip_validation: bool = False) -> tuple: # noqa: C901 flex_model_d.get("discharging_efficiency") for flex_model_d in flex_model ] - # Get info from flex-context - # (normalize to a list; tests may pass e.g. dict_values when bypassing the schema) - inflexible_device_sensors = list( - self.flex_context.get("inflexible_device_sensors", []) - ) - # Fetch the device's power capacity (required to keep the optimization problem bounded) power_capacity_in_mw = self._get_device_power_capacity( flex_model, @@ -348,41 +263,12 @@ def device_list_series( ) -> pd.Series: return pd.Series([tuple(devices)] * len(index), index=index, name="device") - # Enumerate only device models (not stock entries), so device indices line up - # with the sensors and device_constraints lists. - commodity_to_devices = {} - for d, flex_model_d in enumerate(device_models): - commodity = flex_model_d.get("commodity", "electricity") - commodity_to_devices.setdefault(commodity, []).append(d) - - # inflexible devices are electricity by default - number_flexible_devices = len(device_models) - number_inflexible_devices = len( - self.flex_context.get("inflexible_device_sensors", []) - ) - commodity_to_devices.setdefault("electricity", []).extend( - range( - number_flexible_devices, - number_flexible_devices + number_inflexible_devices, - ) - ) - - # Per-commodity inflexible-device-sensors, enumerated after the top-level - # (electricity) inflexible devices, in the order the commodity contexts are - # given. This mirrors the enumeration that - # `_compute_commodity_aggregate_schedules` already assumes. - commodity_context_inflexible_sensors: list[Sensor] = [] - num_devices = number_flexible_devices + number_inflexible_devices - for commodity_context in self.flex_context.get("commodity_contexts", []): - commodity = commodity_context["commodity"] - commodity_inflexible_sensors = commodity_context.get( - "inflexible_device_sensors", [] - ) - commodity_to_devices.setdefault(commodity, []).extend( - range(num_devices, num_devices + len(commodity_inflexible_sensors)) - ) - commodity_context_inflexible_sensors.extend(commodity_inflexible_sensors) - num_devices += len(commodity_inflexible_sensors) + # The canonical device enumeration comes from the inventory: flexible devices + # (indices lining up with the sensors and device_constraints lists), then + # top-level (electricity) inflexible devices, then each commodity context's + # own inflexible devices. This is the same enumeration that + # `_compute_commodity_aggregate_schedules` relies on. + commodity_to_devices = inventory.commodity_to_devices commodity_contexts = self._get_commodity_contexts() price_frames_by_commodity = {} @@ -756,18 +642,12 @@ def device_list_series( # Set up device constraints: scheduled flexible devices for this EMS (from index 0 to D-1), # plus the forecasted top-level (electricity) inflexible devices, plus each commodity - # context's own inflexible devices, in that order. + # context's own inflexible devices, in that order (the inventory's canonical order). device_constraints = [ initialize_df(StorageScheduler.COLUMNS, start, end, resolution) - for i in range( - num_flexible_devices - + len(inflexible_device_sensors) - + len(commodity_context_inflexible_sensors) - ) + for i in range(inventory.num_scheduled) ] - for i, inflexible_sensor in enumerate( - inflexible_device_sensors + commodity_context_inflexible_sensors - ): + for i, inflexible_sensor in enumerate(inventory.inflexible_sensors): device_constraints[i + num_flexible_devices]["derivative equals"] = ( get_power_values( query_window=(start, end), @@ -1397,6 +1277,13 @@ def deserialize_flex_config(self): self._deserialize_flex_context() self._deserialize_flex_model() + # Classify all flex-model entries (and the flex-context's inflexible devices) + # once; scheduling and result mapping rely on this inventory for device + # identity and canonical device indices. + self.device_inventory = DeviceInventory.from_flex_config( + self.flex_model, self.flex_context, sensor=self.sensor + ) + def _deserialize_flex_context(self): if isinstance(self.flex_context, dict): # Load the one flex-context for electricity @@ -1477,8 +1364,6 @@ def _deserialize_flex_model(self): soc_targets=self.flex_model[d].get("soc_targets"), sensor=self.flex_model[d]["sensor"], ) - self.stock_groups = self._build_stock_groups(self.flex_model) - else: raise TypeError( f"Unsupported type of flex-model: '{type(self.flex_model)}'" @@ -2450,70 +2335,32 @@ def _build_consumption_production_schedules( return schedules def _reconstruct_commodity_to_devices(self) -> dict[str, list[int]]: - """Reconstruct the mapping of commodity -> device indices as enumerated by `_prepare()`. + """Return the mapping of commodity -> device indices, as enumerated by the device inventory. - Device enumeration order: + Device enumeration order (the inventory's canonical order, also used by `_prepare()`): 1. flexible devices (from the flex-model), in order, 2. top-level (electricity) inflexible-device-sensors, in order, 3. each commodity context's own inflexible-device-sensors, in the order the commodity contexts are given. - This mirrors `_prepare()`'s device enumeration exactly, so the returned device - indices line up with entries of `ems_schedule` / `device_constraints`. + The returned device indices line up with entries of `ems_schedule` / + `device_constraints`. """ - # Get the device models to reconstruct commodity_to_devices mapping - flex_model = getattr(self, "_device_models", None) - if flex_model is None: - # Fallback: reconstruct if not available (shouldn't happen in normal flow) - flex_model = ( - self.flex_model.copy() - if isinstance(self.flex_model, dict) - else [fm for fm in self.flex_model if fm.get("sensor") is not None] - ) - if not isinstance(flex_model, list): - flex_model = [flex_model] - - # Reconstruct commodity_to_devices mapping - commodity_to_devices: dict[str, list[int]] = {} - for d, flex_model_d in enumerate(flex_model): - commodity = flex_model_d.get("commodity", "electricity") - commodity_to_devices.setdefault(commodity, []).append(d) - - # Add inflexible devices to commodities, mirroring _prepare()'s device - # enumeration so the device indices line up with ems_schedule: - # - top-level inflexible-device-sensors go to electricity (backwards compat), - # - then each commodity context's own inflexible-device-sensors are appended to - # that commodity, in the order the commodity contexts are given. - # Without this, a commodity's inflexible demand (e.g. a heat load) is left - # out of its aggregate schedule, so an aggregate-consumption sensor only reflects - # the flexible devices of that commodity. - inflexible_device_sensors = self.flex_context.get( - "inflexible_device_sensors", [] - ) - number_flexible_devices = len(flex_model) - commodity_to_devices.setdefault("electricity", []).extend( - range( - number_flexible_devices, - number_flexible_devices + len(inflexible_device_sensors), - ) - ) - - # Per-commodity inflexible devices, enumerated after the top-level ones. - num_devices = number_flexible_devices + len(inflexible_device_sensors) - for commodity_context in self.flex_context.get("commodity_contexts", []): - commodity = commodity_context["commodity"] - commodity_inflexible_device_sensors = commodity_context.get( - "inflexible_device_sensors", [] - ) - commodity_to_devices.setdefault(commodity, []).extend( - range( - num_devices, - num_devices + len(commodity_inflexible_device_sensors), - ) - ) - num_devices += len(commodity_inflexible_device_sensors) - - return commodity_to_devices + inventory = self.device_inventory + if inventory is None: + # Fallback (e.g. bare schedulers in tests): classify the stored device + # models, or failing that, the flex-model itself. + flex_model = getattr(self, "_device_models", None) + if flex_model is None: + flex_model = ( + self.flex_model.copy() + if isinstance(self.flex_model, dict) + else [fm for fm in self.flex_model if fm.get("sensor") is not None] + ) + inventory = DeviceInventory.from_flex_config( + flex_model, self.flex_context, sensor=getattr(self, "sensor", None) + ) + return inventory.commodity_to_devices def _electricity_device_indices(self) -> list[int]: """Return the device indices (flexible and inflexible) belonging to the electricity commodity.""" diff --git a/flexmeasures/data/models/planning/tests/test_device_inventory.py b/flexmeasures/data/models/planning/tests/test_device_inventory.py new file mode 100644 index 0000000000..9fb0d2b996 --- /dev/null +++ b/flexmeasures/data/models/planning/tests/test_device_inventory.py @@ -0,0 +1,250 @@ +"""Unit tests for the DeviceInventory: classification of flex-model entries and +canonical device enumeration. These tests run without a database; sensors are +unpersisted model instances with manually assigned ids. +""" + +from datetime import timedelta + +import pytest +from marshmallow import ValidationError + +from flexmeasures.data.models.time_series import Sensor +from flexmeasures.data.models.planning.devices import ( + DeviceInventory, + DeviceRole, +) + + +def make_sensor(sensor_id: int, unit: str = "kW") -> Sensor: + sensor = Sensor( + name=f"sensor {sensor_id}", + generic_asset_id=1, + unit=unit, + event_resolution=timedelta(hours=1) if unit == "kW" else timedelta(0), + ) + sensor.id = sensor_id + return sensor + + +@pytest.mark.parametrize("stock_only_position", [0, 1, 2]) +def test_stock_only_entry_position_does_not_shift_devices(stock_only_position): + """Devices keep their properties and order regardless of where a stock-only entry sits.""" + soc = make_sensor(100, unit="kWh") + power_a = make_sensor(1) + power_b = make_sensor(2) + + stock_only_entry = { + "state_of_charge": soc, + "soc_at_start": 0.0, + "soc_max": 0.1, + } + device_entries = [ + {"sensor": power_a, "state_of_charge": soc, "power_capacity_in_mw": 0.001}, + {"sensor": power_b, "power_capacity_in_mw": 0.009}, + ] + flex_model = device_entries.copy() + flex_model.insert(stock_only_position, stock_only_entry) + + inventory = DeviceInventory.from_flex_config(flex_model) + + assert [entry.role for entry in inventory.entries] == [ + DeviceRole.STOCK_ONLY if fm is stock_only_entry else DeviceRole.DEVICE + for fm in flex_model + ] + # Device indices and properties are independent of the stock-only entry's position. + assert inventory.num_flexible == 2 + assert inventory.power_sensors == [power_a, power_b] + assert inventory.device_flex_models == device_entries + assert [device.index for device in inventory.devices] == [0, 1] + # Device A shares the stock described by the stock-only entry. + assert inventory.stock_groups[soc.id] == [0] + assert inventory.stock_params(soc.id) is stock_only_entry + + +def test_stock_keys_shared_between_params_and_groups(): + """Every stock group must be able to look up its SoC parameters, also for devices + without a state-of-charge sensor and in the presence of stock-only entries.""" + soc = make_sensor(100, unit="kWh") + power_a = make_sensor(1) + power_c = make_sensor(3) + + device_c_entry = { + "sensor": power_c, + "soc_at_start": 0.0, + "soc_max": 0.1, + } + inventory = DeviceInventory.from_flex_config( + [ + {"state_of_charge": soc, "soc_at_start": 0.0}, + {"sensor": power_a, "state_of_charge": soc}, + device_c_entry, # no state-of-charge sensor, own SoC parameters + ] + ) + + for stock_key, devices in inventory.stock_groups.items(): + assert inventory.stock_params(stock_key) is not None, ( + f"Stock group {stock_key} (devices {devices}) cannot find its SoC " + "parameters; its key is missing from stock_entries." + ) + device_c = inventory.devices[1] + assert device_c.power_sensor is power_c + assert inventory.stock_params(device_c.stock_key) is device_c_entry + + +def test_single_sensor_mode(): + """A dict flex-model describes one device, whose power sensor is the scheduler's target.""" + power = make_sensor(1) + flex_model = {"soc_at_start": 0.0, "soc_max": 0.1} + + inventory = DeviceInventory.from_flex_config(flex_model, sensor=power) + + assert inventory.is_single_sensor_mode + assert inventory.num_flexible == 1 + device = inventory.devices[0] + assert device.role == DeviceRole.DEVICE + assert device.power_sensor is power + assert device.index == 0 + # The single device has no state-of-charge sensor, so it keeps its own SoC + # parameters under its synthetic stock key. + assert inventory.stock_groups == {device.stock_key: [0]} + assert inventory.stock_params(device.stock_key) is flex_model + + +def test_single_sensor_mode_with_state_of_charge_sensor(): + power = make_sensor(1) + soc = make_sensor(100, unit="kWh") + flex_model = {"state_of_charge": soc, "soc_at_start": 0.0} + + inventory = DeviceInventory.from_flex_config(flex_model, sensor=power) + + device = inventory.devices[0] + assert device.role == DeviceRole.DEVICE + assert device.stock_key == soc.id + assert inventory.stock_params(soc.id) is flex_model + + +def test_nested_output_reference_resolves_power_sensor(): + """An entry referencing its power sensor only via a nested consumption/production + output reference is a schedulable device, not a stock-only entry.""" + soc = make_sensor(100, unit="kWh") + output = make_sensor(5) + + inventory = DeviceInventory.from_flex_config( + [ + { + "state_of_charge": soc, + "consumption": {"sensor": output}, + "soc_at_start": 0.0, + }, + ] + ) + + assert inventory.num_flexible == 1 + device = inventory.devices[0] + assert device.role == DeviceRole.DEVICE + assert device.power_sensor is output + assert device.consumption_sensor is output + assert device.stock_key == soc.id + + +def test_asset_only_entry_is_a_device_without_power_sensor(): + """An entry with neither a sensor nor a state-of-charge reference is still a device.""" + fake_asset = object() + + inventory = DeviceInventory.from_flex_config( + [{"asset": fake_asset, "power_capacity_in_mw": 0.001}] + ) + + assert inventory.num_flexible == 1 + device = inventory.devices[0] + assert device.role == DeviceRole.DEVICE + assert device.power_sensor is None + assert device.asset is fake_asset + + +def test_commodity_enumeration_includes_inflexible_tail(): + """Inflexible devices follow the flexible devices: top-level (electricity) sensors + first, then each commodity context's own sensors, in context order.""" + inventory = DeviceInventory.from_flex_config( + [ + {"sensor": make_sensor(1), "commodity": "electricity"}, + {"sensor": make_sensor(2), "commodity": "gas"}, + {"sensor": make_sensor(3)}, # defaults to electricity + ], + flex_context={ + "inflexible_device_sensors": [make_sensor(11), make_sensor(12)], + "commodity_contexts": [ + {"commodity": "gas", "inflexible_device_sensors": [make_sensor(13)]}, + ], + }, + ) + + assert inventory.commodity_to_devices["electricity"] == [0, 2, 3, 4] + assert inventory.commodity_to_devices["gas"] == [1, 5] + assert inventory.num_scheduled == 6 + assert [device.sensor_id for device in inventory.inflexible_devices] == [11, 12, 13] + assert inventory.by_index(5).commodity == "gas" + + +def test_electricity_group_exists_even_without_electricity_devices(): + """The electricity commodity group is always present (inflexible devices are + electricity by default), even when empty.""" + inventory = DeviceInventory.from_flex_config( + [{"sensor": make_sensor(1), "commodity": "gas"}] + ) + assert inventory.commodity_to_devices == {"gas": [0], "electricity": []} + + +def test_by_sensor_id(): + power_a = make_sensor(1) + power_b = make_sensor(2) + inventory = DeviceInventory.from_flex_config( + [ + {"sensor": power_a}, + {"sensor": power_b}, + {"sensor": power_b}, # two devices may share a power sensor + ] + ) + assert [device.index for device in inventory.by_sensor_id(1)] == [0] + assert [device.index for device in inventory.by_sensor_id(2)] == [1, 2] + assert inventory.by_sensor_id(3) == [] + + +def test_state_of_charge_as_time_series_forms_own_stock(): + """A state of charge given as a value or time series (rather than a sensor + reference) cannot link devices into a shared stock: the device keeps its own + (synthetic) stock, with its own SoC parameters.""" + power = make_sensor(1) + flex_model = { + "state_of_charge": [{"start": "2015-01-01T00:00+01", "value": "3.1 MWh"}], + "soc_at_start": 3.1, + } + + inventory = DeviceInventory.from_flex_config(flex_model, sensor=power) + + device = inventory.devices[0] + assert device.role == DeviceRole.DEVICE + assert device.stock_key < 0 # synthetic + assert inventory.stock_params(device.stock_key) is flex_model + + +def test_conflicting_stock_params_raise(): + """When multiple entries carry SoC parameters for the same stock, we fail fast + rather than letting one entry silently win.""" + soc = make_sensor(100, unit="kWh") + stock_only_entry = {"state_of_charge": soc, "soc_at_start": 0.0} + device_with_params = { + "sensor": make_sensor(1), + "state_of_charge": soc, + "soc_at_start": 1.0, + } + + with pytest.raises(ValidationError, match="single entry"): + DeviceInventory.from_flex_config([stock_only_entry, device_with_params]) + + # A device entry without SoC parameters does not conflict. + device_without_params = {"sensor": make_sensor(2), "state_of_charge": soc} + inventory = DeviceInventory.from_flex_config( + [stock_only_entry, device_without_params] + ) + assert inventory.stock_params(soc.id) is stock_only_entry diff --git a/flexmeasures/data/models/planning/tests/test_solver.py b/flexmeasures/data/models/planning/tests/test_solver.py index 182f2cc400..3af8026f6a 100644 --- a/flexmeasures/data/models/planning/tests/test_solver.py +++ b/flexmeasures/data/models/planning/tests/test_solver.py @@ -11,6 +11,7 @@ from sqlalchemy import select from flexmeasures.data.models.data_sources import DataSource +from flexmeasures.data.models.generic_assets import GenericAsset from flexmeasures.data.models.time_series import TimedBelief from flexmeasures.data.models.time_series import Sensor from flexmeasures.data.models.planning import Scheduler @@ -25,11 +26,13 @@ from flexmeasures.data.models.planning.tests.utils import ( check_constraints, get_sensors_from_db, + series_to_ts_specs, ) from flexmeasures.data.models.planning.utils import ( initialize_device_commitment, initialize_df, initialize_energy_commitment, + initialize_index, initialize_series, ) from flexmeasures.data.schemas.sensors import TimedEventSchema @@ -3244,6 +3247,439 @@ def test_prefer_full_storage_skips_non_storage_devices(db, building): ) +def _add_parent_site(db, building, name: str) -> GenericAsset: + """Add a fresh parent site for an alignment test. + + Note that we deliberately do not schedule on the shared (module-scoped) ``building`` + fixture asset directly: earlier tests move legacy sensor attributes into its + flex-model, which ``collect_flex_config`` would then inject into our flex-model + as an extra asset-only device entry. + """ + site = GenericAsset( + name=name, + generic_asset_type=building.generic_asset_type, + owner=building.owner, + flex_context={}, + ) + db.session.add(site) + db.session.flush() + return site + + +def _add_battery_device( + db, parent, battery_type, name: str, with_soc_sensor: bool = True +) -> tuple[Sensor, Sensor | None]: + """Add a child battery asset with a power sensor and (optionally) a SoC sensor.""" + asset = GenericAsset( + name=name, + generic_asset_type=battery_type, + owner=parent.owner, + parent_asset_id=parent.id, + ) + db.session.add(asset) + db.session.flush() + power_sensor = Sensor( + name=f"{name} power", + generic_asset=asset, + event_resolution=timedelta(hours=1), + unit="kW", + ) + sensors = [power_sensor] + soc_sensor = None + if with_soc_sensor: + soc_sensor = Sensor( + name=f"{name} soc", + generic_asset=asset, + event_resolution=timedelta(0), + unit="kWh", + ) + sensors.append(soc_sensor) + db.session.add_all(sensors) + db.session.flush() + return power_sensor, soc_sensor + + +def test_stock_only_entry_first_keeps_device_alignment( + db, building, setup_generic_asset_types +): + """A stock-only entry listed before the device entries must not shift device properties. + + The flex-model lists a stock-only entry first (carrying decoy power-capacity and + charging-efficiency values), followed by two devices feeding the shared stock: + + - Device A: 1 kW power capacity, perfect charging efficiency (by default). + - Device B: 9 kW power capacity, poor (10%) charging efficiency. + + Reaching the 2 kWh soc-minimum through device A is 10x cheaper, so a correctly + aligned scheduler uses device A only. On misaligned code, each device inherits the + previous entry's properties (A gets the decoys, B gets A's), flipping the optimum. + """ + battery_type = setup_generic_asset_types["battery"] + site = _add_parent_site(db, building, "alignment test site") + a_power, a_soc = _add_battery_device(db, site, battery_type, "alignment test A") + b_power, _ = _add_battery_device( + db, site, battery_type, "alignment test B", with_soc_sensor=False + ) + db.session.commit() + + resolution = timedelta(hours=1) + start = pd.Timestamp("2020-01-01T00:00:00", tz="Europe/Amsterdam") + end = start + 4 * resolution + + flex_model = [ + { + # Stock-only entry (no power sensor): SoC parameters for the shared stock, + # plus decoy device properties that no device should inherit. + "state-of-charge": {"sensor": a_soc.id}, + "soc-at-start": "0 kWh", + "soc-min": "0 kWh", + "soc-max": "100 kWh", + "soc-minima": [{"datetime": end.isoformat(), "value": "2 kWh"}], + "power-capacity": "9 kW", # decoy + "charging-efficiency": "50%", # decoy + }, + { + "sensor": a_power.id, + "state-of-charge": {"sensor": a_soc.id}, + "power-capacity": "1 kW", + }, + { + "sensor": b_power.id, + "state-of-charge": {"sensor": a_soc.id}, + "power-capacity": "9 kW", + "charging-efficiency": "10%", + }, + ] + + scheduler: Scheduler = StorageScheduler( + asset_or_sensor=site, + start=start, + end=end, + resolution=resolution, + flex_model=flex_model, + flex_context={ + "consumption-price": "100 EUR/MWh", + "production-price": "0 EUR/MWh", + "site-power-capacity": "100 kW", + }, + return_multiple=True, + ) + results = scheduler.compute() + + a_schedule = next( + r["data"] + for r in results + if r.get("name") == "storage_schedule" and r.get("sensor") is a_power + ) + b_schedule = next( + r["data"] + for r in results + if r.get("name") == "storage_schedule" and r.get("sensor") is b_power + ) + + # Device A does all the charging (2 kWh), respecting its own 1 kW capacity. + np.testing.assert_allclose(a_schedule.sum(), 2, atol=1e-3) + assert (a_schedule <= 1 + 1e-3).all(), ( + "Device A charged above its own 1 kW power capacity, " + "indicating it inherited the stock-only entry's decoy capacity." + ) + # Device B (10% charging efficiency) is 10x more expensive per stored kWh, so it stays idle. + np.testing.assert_allclose(b_schedule.abs().sum(), 0, atol=1e-3) + + +def test_device_without_soc_sensor_keeps_soc_params_when_stock_only_entry_present( + db, building, setup_generic_asset_types +): + """A device without a state-of-charge sensor must keep its own SoC parameters, + also when a stock-only entry is present in the flex-model. + + Device C has no state-of-charge sensor, but carries its own soc-at-start and a + 1 kWh soc-target. On buggy code, the presence of the stock-only entry offsets the + synthetic stock keys, so device C's SoC parameters are silently dropped and its + target is never met. + """ + battery_type = setup_generic_asset_types["battery"] + site = _add_parent_site(db, building, "soc params test site") + a_power, a_soc = _add_battery_device(db, site, battery_type, "soc params test A") + c_power, _ = _add_battery_device( + db, site, battery_type, "soc params test C", with_soc_sensor=False + ) + db.session.commit() + + resolution = timedelta(hours=1) + start = pd.Timestamp("2020-01-01T00:00:00", tz="Europe/Amsterdam") + end = start + 4 * resolution + + flex_model = [ + { + # Stock-only entry for device A's stock. + "state-of-charge": {"sensor": a_soc.id}, + "soc-at-start": "0 kWh", + "soc-min": "0 kWh", + "soc-max": "100 kWh", + }, + { + "sensor": a_power.id, + "state-of-charge": {"sensor": a_soc.id}, + "power-capacity": "1 kW", + }, + { + # Device C: no state-of-charge sensor, but its own SoC parameters. + "sensor": c_power.id, + "power-capacity": "2 kW", + "soc-at-start": "0 kWh", + "soc-min": "0 kWh", + "soc-max": "100 kWh", + "soc-targets": [{"datetime": end.isoformat(), "value": "1 kWh"}], + }, + ] + + scheduler: Scheduler = StorageScheduler( + asset_or_sensor=site, + start=start, + end=end, + resolution=resolution, + flex_model=flex_model, + flex_context={ + "consumption-price": "100 EUR/MWh", + "production-price": "0 EUR/MWh", + "site-power-capacity": "100 kW", + }, + return_multiple=True, + ) + results = scheduler.compute() + + c_schedule = next( + r["data"] + for r in results + if r.get("name") == "storage_schedule" and r.get("sensor") is c_power + ) + # Device C charges 1 kWh in total to meet its own soc-target. + np.testing.assert_allclose(c_schedule.sum(), 1, atol=1e-3) + + +def test_flex_context_commitments_target_devices_not_stock_only_entries( + db, building, setup_generic_asset_types +): + """Flex-context commitments must bind the scheduled devices, not stock-only entries. + + With a stock-only entry listed first, a flex-context commitment should still yield + one commitment per scheduled device (indices 0 and 1), rather than one per + flex-model entry (indices 0, 1 and 2, of which index 2 does not exist as a + flexible device). + """ + battery_type = setup_generic_asset_types["battery"] + site = _add_parent_site(db, building, "commitment test site") + a_power, a_soc = _add_battery_device(db, site, battery_type, "commitment test A") + b_power, _ = _add_battery_device( + db, site, battery_type, "commitment test B", with_soc_sensor=False + ) + db.session.commit() + + resolution = timedelta(hours=1) + start = pd.Timestamp("2020-01-01T00:00:00", tz="Europe/Amsterdam") + end = start + 4 * resolution + + flex_model = [ + { + "state-of-charge": {"sensor": a_soc.id}, + "soc-at-start": "0 kWh", + "soc-min": "0 kWh", + "soc-max": "100 kWh", + }, + { + "sensor": a_power.id, + "state-of-charge": {"sensor": a_soc.id}, + "power-capacity": "1 kW", + }, + { + "sensor": b_power.id, + "power-capacity": "2 kW", + }, + ] + + scheduler: Scheduler = StorageScheduler( + asset_or_sensor=site, + start=start, + end=end, + resolution=resolution, + flex_model=flex_model, + flex_context={ + "consumption-price": "100 EUR/MWh", + "production-price": "0 EUR/MWh", + "site-power-capacity": "100 kW", + "commitments": [ + { + "name": "test commitment", + "up-price": "1 EUR/MWh", + "down-price": "-1 EUR/MWh", + } + ], + }, + return_multiple=True, + ) + *_, commitments = scheduler._prepare(skip_validation=True) + + test_commitments = [c for c in commitments if c.name == "test commitment"] + num_devices = 2 + assert len(test_commitments) == num_devices, ( + f"Expected one commitment per scheduled device ({num_devices}), " + f"got {len(test_commitments)} (one per flex-model entry, including the " + "stock-only entry)." + ) + commitment_devices = {int(d) for c in test_commitments for d in c.device.unique()} + assert commitment_devices == set(range(num_devices)), ( + f"Commitments target device indices {sorted(commitment_devices)}, " + f"expected {sorted(range(num_devices))}." + ) + + +def test_multi_device_flex_model_alignment(db, building, setup_generic_asset_types): + """Regression test for two bugs in multi-device storage scheduling. + + The flex-model lists two devices under a common parent asset: + + - Device A: a normal storage device with a top-level ``sensor`` and no explicit + ``power-capacity`` (so its capacity is looked up from its own asset's flex-model, + a small 1 kW). + - Device B: a storage device whose power sensor is referenced only via a nested + ``consumption`` output reference (no top-level ``sensor``), with its own + state-of-charge sensor and a much larger 9 kW ``power-capacity``. + + Device B is listed first so that, on unfixed code, filtering it out shifts the + device/asset alignment for device A. + + Two bugs manifest without the fixes: + + 1. Device B is silently dropped (misclassified as a stock-only entry because it has + no top-level ``sensor``), so its consumption output sensor never receives a + schedule. + 2. Because B is dropped, the remaining device A picks up B's (larger) power capacity + through the misaligned device/asset zip, letting A charge above its own 1 kW + capacity. + """ + parent = _add_parent_site(db, building, "multi device test site") + battery_type = setup_generic_asset_types["battery"] + owner = parent.owner + + resolution = timedelta(hours=1) + tz = pytz.timezone("Europe/Amsterdam") + start = tz.localize(datetime(2020, 1, 1)) + end = start + 4 * resolution + end_iso = end.isoformat() + + def make_device(name: str, power_capacity: str) -> tuple[Sensor, Sensor]: + asset = GenericAsset( + name=name, + generic_asset_type=battery_type, + owner=owner, + parent_asset_id=parent.id, + flex_model={"power-capacity": power_capacity}, + ) + db.session.add(asset) + db.session.flush() + power_sensor = Sensor( + name=f"{name} power", + generic_asset=asset, + event_resolution=resolution, + unit="kW", + ) + soc_sensor = Sensor( + name=f"{name} soc", + generic_asset=asset, + event_resolution=timedelta(0), + unit="kWh", + ) + db.session.add_all([power_sensor, soc_sensor]) + db.session.flush() + return power_sensor, soc_sensor + + # Device A: small (1 kW) capacity, provided only via its asset's flex-model. + a_power, a_soc = make_device("device A", "1 kW") + # Device B: larger (9 kW) capacity, power sensor referenced only via "consumption". + b_power, b_soc = make_device("device B", "9 kW") + b_output = Sensor( + name="device B consumption output", + generic_asset=b_power.generic_asset, + event_resolution=resolution, + unit="kW", + ) + db.session.add(b_output) + db.session.commit() + + # Make the first slot the cheapest so that a device allowed to exceed 1 kW would + # front-load its charging into slot 0 (revealing the wrong, larger capacity). + index = initialize_index(start=start, end=end, resolution=resolution) + consumption_prices = pd.Series([1, 100, 100, 100], index=index) + + # Device B is listed first, so dropping it (bug 2) shifts the alignment for device A. + flex_model = [ + { + "state-of-charge": {"sensor": b_soc.id}, + "consumption": {"sensor": b_output.id}, + "power-capacity": "9 kW", + "soc-at-start": "0 kWh", + "soc-min": "0 kWh", + "soc-max": "100 kWh", + "soc-minima": [{"datetime": end_iso, "value": "2 kWh"}], + }, + { + "sensor": a_power.id, + "state-of-charge": {"sensor": a_soc.id}, + "soc-at-start": "0 kWh", + "soc-min": "0 kWh", + "soc-max": "100 kWh", + "soc-minima": [{"datetime": end_iso, "value": "4 kWh"}], + }, + ] + + scheduler: Scheduler = StorageScheduler( + asset_or_sensor=parent, + start=start, + end=end, + resolution=resolution, + flex_model=flex_model, + flex_context={ + "consumption-price": series_to_ts_specs(consumption_prices, unit="EUR/MWh"), + "production-price": "0 EUR/MWh", + "site-power-capacity": "100 kW", + "soc-minima-breach-price": "6000 EUR/kWh", + }, + return_multiple=True, + ) + results = scheduler.compute() + + # Bug 1: device A must respect its own 1 kW capacity, not device B's 9 kW. + a_schedule = next( + r["data"] + for r in results + if r.get("name") == "storage_schedule" and r.get("sensor") is a_power + ) + # Charge is in kW (sensor unit); allow a small numerical tolerance. + assert (a_schedule <= 1 + 1e-3).all(), ( + "Device A charged above its own 1 kW power capacity, " + "indicating it inherited device B's larger capacity through a misaligned zip." + ) + # Sanity: device A still meets its 4 kWh soc-minimum (1 kW across all four hours). + np.testing.assert_allclose(a_schedule.values, 1.0, atol=1e-3) + + # Bug 2: device B must be scheduled and its consumption output sensor must receive data. + consumption_result = next( + ( + r + for r in results + if r.get("name") == "consumption_schedule" and r.get("sensor") is b_output + ), + None, + ) + assert ( + consumption_result is not None + ), "Device B was dropped: its consumption output sensor received no schedule." + # Device B charges 2 kWh in total (consumption is positive). + b_consumption = consumption_result["data"] + np.testing.assert_allclose(b_consumption.sum(), 2.0, atol=1e-3) + + def test_multiple_devices_sequential_scheduler(): start = pd.Timestamp("2023-01-01T00:00:00") end = pd.Timestamp("2023-01-02T00:00:00")