diff --git a/documentation/changelog.rst b/documentation/changelog.rst index 89a909a897..4fc2f13886 100644 --- a/documentation/changelog.rst +++ b/documentation/changelog.rst @@ -12,12 +12,13 @@ v1.0.0 | July XX, 2026 New features ------------- +* Breaking behaviour change: the top-level flex-context's ``relax-constraints`` field now defaults to ``True`` (matching the default already used within each ``commodities`` entry), so constraint violations are softly penalized by default instead of being hard constraints, unless explicitly set to ``False`` [see `PR #2172 `_] * In the UI, asset and sensor lists can be filtered by ID prefix through API-backed search fields [see `PR #2231 `_] * 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 `_ ] -* The flex-context can now define multiple commodities, each specifying their own prices and grid capacities [see `PR #1946 `_, `PR #2172 `_ and `PR #2235 `_] +* 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 `_] * Extended ``GET /api/v3_0/jobs/`` with a ``result`` field containing ``unresolved`` and ``resolved`` soft state-of-charge constraint analysis (``soc-minima``/``soc-maxima`` violations or satisfied constraints, keyed by asset ID) for scheduling jobs; both arrays are empty when no SoC constraints were defined [see `PR #2072 `_] diff --git a/flexmeasures/data/models/planning/storage.py b/flexmeasures/data/models/planning/storage.py index cc1f9786bb..dc0c02b7f7 100644 --- a/flexmeasures/data/models/planning/storage.py +++ b/flexmeasures/data/models/planning/storage.py @@ -43,7 +43,7 @@ ) from flexmeasures.utils.time_utils import get_max_planning_horizon from flexmeasures.utils.time_utils import determine_minimum_resampling_resolution -from flexmeasures.utils.unit_utils import ur, convert_units +from flexmeasures.utils.unit_utils import ur, convert_units, units_are_convertible storage_asset_types = ["one-way_evse", "two-way_evse", "battery", "heat-storage"] @@ -300,8 +300,9 @@ def _prepare(self, skip_validation: bool = False) -> tuple: # noqa: C901 ] # Get info from flex-context - inflexible_device_sensors = self.flex_context.get( - "inflexible_device_sensors", [] + # (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) @@ -359,17 +360,40 @@ def device_list_series( number_inflexible_devices = len( self.flex_context.get("inflexible_device_sensors", []) ) - commodity_to_devices["electricity"] += list( + 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) + commodity_contexts = self._get_commodity_contexts() price_frames_by_commodity = {} for commodity, devices in commodity_to_devices.items(): + # Skip commodities without any devices (e.g. no electricity devices in an + # all-gas flex-model, or a commodity context that no device refers to): + # they need no prices, commitments or EMS constraints, and empty device + # groups would make the optimization problem unbounded. + if not devices: + continue commodity_devices = device_list_series(devices, index) commodity_context = commodity_contexts.get(commodity, {}) @@ -730,12 +754,20 @@ def device_list_series( ) commitments.append(commitment) - # Set up device constraints: scheduled flexible devices for this EMS (from index 0 to D-1), plus the forecasted inflexible devices (at indices D to n). + # 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. device_constraints = [ initialize_df(StorageScheduler.COLUMNS, start, end, resolution) - for i in range(num_flexible_devices + len(inflexible_device_sensors)) + for i in range( + num_flexible_devices + + len(inflexible_device_sensors) + + len(commodity_context_inflexible_sensors) + ) ] - for i, inflexible_sensor in enumerate(inflexible_device_sensors): + for i, inflexible_sensor in enumerate( + inflexible_device_sensors + commodity_context_inflexible_sensors + ): device_constraints[i + num_flexible_devices]["derivative equals"] = ( get_power_values( query_window=(start, end), @@ -1318,12 +1350,14 @@ def convert_to_commitments( commitment_spec["index"] = initialize_index( start, end, timing_kwargs["resolution"] ) + commitment_commodity = commitment_spec.get("commodity", "electricity") for d, flex_model_d in enumerate(flex_model): + device_commodity = flex_model_d.get("commodity", "electricity") + if device_commodity != commitment_commodity: + continue commitment = FlowCommitment( device=d, - # todo: is flex_model_d guaranteed to have "commodity? Consider defaulting the device commodity to "electricity" - # todo: should there not be something matching the "commodity" from the commitment_spec (default to "electricity") to the device commodity? - device_group=flex_model_d["commodity"], + device_group=device_commodity, **commitment_spec, ) commitments.append(commitment) @@ -1375,17 +1409,13 @@ def _deserialize_flex_context(self): ) # Ensure all flex-contexts share the same currency unit - # todo: move this into a validator for FlexContextSchema.commodity_contexts? shared_currency_unit = None for commodity_flex_context in self.flex_context: - shared_currency_unit = commodity_flex_context["shared_currency_unit"] + context_currency_unit = commodity_flex_context["shared_currency_unit"] if shared_currency_unit is None: - shared_currency_unit = commodity_flex_context[ - "shared_currency_unit" - ] - elif ( - commodity_flex_context["shared_currency_unit"] - != shared_currency_unit + shared_currency_unit = context_currency_unit + elif not units_are_convertible( + context_currency_unit, shared_currency_unit ): raise ValidationError( f"All prices in the flex-context must share the same currency unit (in this case: '{shared_currency_unit}')." @@ -2419,30 +2449,17 @@ def _build_consumption_production_schedules( ) return schedules - def _compute_commodity_aggregate_schedules( - self, - storage_schedule: dict, - ems_schedule: pd.DataFrame, - # sensors: list[Sensor | None], - ) -> None: - """Compute per-commodity aggregate power flows for aggregate-consumption and aggregate-production sensors. - - This method populates the storage_schedule dict with aggregate schedules for each commodity - that defines aggregate-consumption and/or aggregate-production sensors in its commodity context. + def _reconstruct_commodity_to_devices(self) -> dict[str, list[int]]: + """Reconstruct the mapping of commodity -> device indices as enumerated by `_prepare()`. - The sign convention and split logic follows the same pattern as _build_consumption_production_schedules: - - Only aggregate-consumption defined: full aggregate schedule (consumption +, production -) - - Only aggregate-production defined: full aggregate schedule (consumption +, production -) - (sign will be flipped by make_schedule based on consumption_is_positive=False) - - Both defined: consumption sensor gets non-negative part, production sensor gets non-positive part - (sign will be flipped for production by make_schedule) + Device enumeration order: + 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. - For backwards compatibility, when no commodity_contexts are defined, all devices are treated - as electricity devices and use the top-level flex-context fields. - - :param storage_schedule: Dict to populate with aggregate schedules (will be modified in-place) - :param ems_schedule: DataFrame of per-device power schedules in MW (consumption positive) - :param sensors: List of sensors corresponding to device indices + This mirrors `_prepare()`'s device enumeration exactly, so 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) @@ -2457,7 +2474,7 @@ def _compute_commodity_aggregate_schedules( flex_model = [flex_model] # Reconstruct commodity_to_devices mapping - commodity_to_devices = {} + 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) @@ -2467,7 +2484,7 @@ def _compute_commodity_aggregate_schedules( # - 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 step 2 below, a commodity's inflexible demand (e.g. a heat load) is left + # 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( @@ -2496,6 +2513,37 @@ def _compute_commodity_aggregate_schedules( ) num_devices += len(commodity_inflexible_device_sensors) + return commodity_to_devices + + def _electricity_device_indices(self) -> list[int]: + """Return the device indices (flexible and inflexible) belonging to the electricity commodity.""" + return self._reconstruct_commodity_to_devices().get("electricity", []) + + def _compute_commodity_aggregate_schedules( + self, + storage_schedule: dict, + ems_schedule: pd.DataFrame, + ) -> None: + """Compute per-commodity aggregate power flows for aggregate-consumption and aggregate-production sensors. + + This method populates the storage_schedule dict with aggregate schedules for each commodity + that defines aggregate-consumption and/or aggregate-production sensors in its commodity context. + + The sign convention and split logic follows the same pattern as _build_consumption_production_schedules: + - Only aggregate-consumption defined: full aggregate schedule (consumption +, production -) + - Only aggregate-production defined: full aggregate schedule (consumption +, production -) + (sign will be flipped by make_schedule based on consumption_is_positive=False) + - Both defined: consumption sensor gets non-negative part, production sensor gets non-positive part + (sign will be flipped for production by make_schedule) + + For backwards compatibility, when no commodity_contexts are defined, all devices are treated + as electricity devices and use the top-level flex-context fields. + + :param storage_schedule: Dict to populate with aggregate schedules (will be modified in-place) + :param ems_schedule: DataFrame of per-device power schedules in MW (consumption positive) + """ + commodity_to_devices = self._reconstruct_commodity_to_devices() + # Get commodity contexts (handles backwards compatibility) commodity_contexts = self._get_commodity_contexts() @@ -2617,16 +2665,16 @@ def compute(self, skip_validation: bool = False) -> SchedulerOutputType: storage_schedule[sensor] += ems_schedule[d] # Obtain the aggregate power schedule, too, if the flex-context states the associated sensor. Fill with the sum of schedules made here. + # Restricted to electricity devices (flexible and inflexible), per decision. aggregate_power_sensor = self.flex_context.get("aggregate_power", None) if isinstance(aggregate_power_sensor, Sensor): + electricity_devices = self._electricity_device_indices() storage_schedule[aggregate_power_sensor] = pd.concat( - ems_schedule, - axis=1, # todo: select only electric devices (flexible and inflexible) + [ems_schedule[d] for d in electricity_devices if d < len(ems_schedule)], + axis=1, ).sum(axis=1) # Compute per-commodity aggregate power flows for aggregate-consumption and aggregate-production sensors - self._compute_commodity_aggregate_schedules( - storage_schedule, ems_schedule # , sensors - ) + self._compute_commodity_aggregate_schedules(storage_schedule, ems_schedule) # Convert each device schedule to the unit of the device's power sensor storage_schedule = { diff --git a/flexmeasures/data/models/planning/tests/test_commitments.py b/flexmeasures/data/models/planning/tests/test_commitments.py index d8b8f54ff1..4275892c53 100644 --- a/flexmeasures/data/models/planning/tests/test_commitments.py +++ b/flexmeasures/data/models/planning/tests/test_commitments.py @@ -12,7 +12,8 @@ initialize_index, ) from flexmeasures.data.models.planning.storage import StorageScheduler -from flexmeasures.data.models.time_series import Sensor +from flexmeasures.data.models.time_series import Sensor, TimedBelief +from flexmeasures.data.models.data_sources import DataSource from flexmeasures.data.models.planning.linear_optimization import device_scheduler from flexmeasures.data.models.generic_assets import GenericAsset, GenericAssetType from flexmeasures.data.utils import save_to_db @@ -1553,3 +1554,231 @@ def test_simulation_with_dynamic_consumption_capacity(app, db): assert heater_schedule.loc["2026-04-07T08:00:00+00:00"] == pytest.approx( 80.0 ), "Electric heater should have one expected partial 80 kW dispatch step before the first cheap-electricity window." + + +def test_all_gas_flex_model_without_electricity_device(app, db): + """test_all_gas_flex_model_without_electricity_device: a flex-model with only gas + devices (no electricity device at all) should not raise a KeyError, now that + commodity_to_devices["electricity"] is built with setdefault(). + """ + boiler_type = get_or_create_model(GenericAssetType, name="gas-boiler") + + start = pd.Timestamp("2024-01-01T00:00:00+01:00") + end = pd.Timestamp("2024-01-02T00:00:00+01:00") + resolution = pd.Timedelta("1h") + + gas_boiler = GenericAsset( + name="Gas Boiler (all-gas flex-model test)", + generic_asset_type=boiler_type, + ) + db.session.add(gas_boiler) + db.session.commit() + + boiler_power = Sensor( + name="boiler power", + unit="kW", + event_resolution=resolution, + generic_asset=gas_boiler, + ) + db.session.add(boiler_power) + db.session.commit() + + flex_model = [ + { + "sensor": boiler_power.id, + "commodity": "gas", + "power-capacity": "30 kW", + "consumption-capacity": "30 kW", + "production-capacity": "0 kW", + "soc-usage": ["1 kW"], + "soc-min": 0.0, + "soc-max": 0.0, + "soc-at-start": 0.0, + }, + ] + + flex_context = [ + { + "commodity": "gas", + "consumption-price": "50 EUR/MWh", + "production-price": "50 EUR/MWh", + }, + ] + + scheduler = StorageScheduler( + asset_or_sensor=gas_boiler, + start=start, + end=end, + resolution=resolution, + belief_time=start, + flex_model=flex_model, + flex_context=flex_context, + return_multiple=True, + ) + + # This used to raise KeyError("electricity") in _prepare(). + schedules = scheduler.compute(skip_validation=True) + + storage_schedules = [ + entry for entry in schedules if entry.get("name") == "storage_schedule" + ] + assert len(storage_schedules) == 1 + boiler_schedule = storage_schedules[0]["data"] + assert (boiler_schedule == 1.0).all() + + +def test_per_commodity_inflexible_device_sensors(app, db): + """test_per_commodity_inflexible_device_sensors: an inflexible-device-sensor declared + inside a (non-electricity) commodity context should constrain that commodity's site + capacity and be reflected in the flexible device's schedule (since its consumption + capacity leaves no room for the inflexible load plus more). + """ + boiler_type = get_or_create_model(GenericAssetType, name="gas-boiler") + + start = pd.Timestamp("2024-01-01T00:00:00+01:00") + end = pd.Timestamp("2024-01-01T04:00:00+01:00") + resolution = pd.Timedelta("1h") + + gas_site = GenericAsset( + name="Gas Site (per-commodity inflexible sensor test)", + generic_asset_type=boiler_type, + ) + db.session.add(gas_site) + db.session.commit() + + flexible_boiler_power = Sensor( + name="flexible boiler power", + unit="kW", + event_resolution=resolution, + generic_asset=gas_site, + ) + inflexible_gas_load = Sensor( + name="inflexible gas load", + unit="kW", + event_resolution=resolution, + generic_asset=gas_site, + ) + gas_aggregate_consumption = Sensor( + name="gas aggregate consumption", + unit="kW", + event_resolution=resolution, + generic_asset=gas_site, + ) + db.session.add_all( + [flexible_boiler_power, inflexible_gas_load, gas_aggregate_consumption] + ) + db.session.commit() + + # A constant 8 kW inflexible gas load, recorded as beliefs. + # By default, power sensors store consumption as negative values + # (get_power_values flips the sign to the scheduler's consumption-positive convention). + index = initialize_index(start, end, resolution) + + source = get_or_create_model(DataSource, name="test source", type="forecaster") + beliefs = [ + TimedBelief( + sensor=inflexible_gas_load, + source=source, + event_start=dt, + belief_time=start, + event_value=-8.0, + ) + for dt in index + ] + db.session.add_all(beliefs) + db.session.commit() + + flex_model = [ + { + "sensor": flexible_boiler_power.id, + "commodity": "gas", + "power-capacity": "30 kW", + "consumption-capacity": "30 kW", + "production-capacity": "0 kW", + "soc-usage": ["1 kW"], + "soc-min": 0.0, + "soc-max": 0.0, + "soc-at-start": 0.0, + }, + ] + + flex_context = [ + { + "commodity": "gas", + "consumption-price": "50 EUR/MWh", + "production-price": "50 EUR/MWh", + "site-consumption-capacity": "10 kW", + "inflexible-device-sensors": [inflexible_gas_load.id], + "aggregate-consumption": {"sensor": gas_aggregate_consumption.id}, + }, + ] + + scheduler = StorageScheduler( + asset_or_sensor=gas_site, + start=start, + end=end, + resolution=resolution, + belief_time=start, + flex_model=flex_model, + flex_context=flex_context, + return_multiple=True, + ) + + schedules = scheduler.compute(skip_validation=True) + + storage_schedules = [ + entry for entry in schedules if entry.get("name") == "storage_schedule" + ] + boiler_schedule = next( + entry for entry in storage_schedules if entry["sensor"] == flexible_boiler_power + )["data"] + + # With an 8 kW inflexible gas load counted against the 10 kW site-consumption-capacity, + # the flexible boiler (which otherwise consumes a constant 1 kW) is left at most 2 kW of + # headroom. + assert (boiler_schedule <= 2.0 + 1e-6).all() + + # The aggregate-consumption schedule for the gas commodity must include the inflexible + # gas load. Before the fix, the per-commodity inflexible sensor got no device constraints + # (its device index was silently dropped), so the aggregate would only reflect the + # flexible boiler's ~1 kW instead of 8 + 1 = 9 kW. + aggregate_schedule = next( + entry + for entry in storage_schedules + if entry["sensor"] == gas_aggregate_consumption + )["data"] + expected_aggregate = boiler_schedule + 8.0 + assert aggregate_schedule.values == pytest.approx( + expected_aggregate.values, abs=1e-6 + ), ( + "Aggregate gas consumption should include the 8 kW inflexible gas load " + "on top of the flexible boiler's schedule." + ) + + +def test_electricity_device_indices_exclude_other_commodities(): + """test_electricity_device_indices_exclude_other_commodities: the device indices used + for the aggregate-power sum should cover only electricity devices (flexible and + inflexible), not gas devices, nor per-commodity inflexible devices of other commodities. + """ + scheduler = object.__new__(StorageScheduler) + # Flexible devices: 0 = electricity, 1 = gas, 2 = electricity (implicit default) + scheduler._device_models = [ + {"commodity": "electricity"}, + {"commodity": "gas"}, + {}, # defaults to electricity + ] + scheduler.flex_model = scheduler._device_models + # Top-level inflexible sensors (electricity): indices 3 and 4. + # Gas commodity context with one inflexible sensor: index 5. + scheduler.flex_context = { + "inflexible_device_sensors": ["el_sensor_a", "el_sensor_b"], + "commodity_contexts": [ + {"commodity": "gas", "inflexible_device_sensors": ["gas_sensor"]}, + ], + } + + mapping = scheduler._reconstruct_commodity_to_devices() + assert mapping["electricity"] == [0, 2, 3, 4] + assert mapping["gas"] == [1, 5] + assert scheduler._electricity_device_indices() == [0, 2, 3, 4] diff --git a/flexmeasures/data/models/planning/tests/test_solver.py b/flexmeasures/data/models/planning/tests/test_solver.py index edb0ad45a8..182f2cc400 100644 --- a/flexmeasures/data/models/planning/tests/test_solver.py +++ b/flexmeasures/data/models/planning/tests/test_solver.py @@ -1523,6 +1523,10 @@ def test_capacity( flex_context = { "production-price": {"sensor": add_market_prices["epex_da_production"].id}, "consumption-price": {"sensor": add_market_prices["epex_da"].id}, + # This test asserts hard-capacity semantics (constraint DataFrames), so opt out + # of the default constraint relaxation (which would use the physical capacity + # as the hard bound and penalize the contracted capacity only softly). + "relax-constraints": False, } def set_if_not_none(dictionary, key, value): @@ -1621,7 +1625,8 @@ def test_device_power_capacity_uses_directional_capacity_before_site_fallback( "soc-max": 5, **configured_capacities, }, - flex_context={"consumption-price": "1 EUR/MWh"}, + # relax-constraints False: this test asserts hard-capacity semantics. + flex_context={"consumption-price": "1 EUR/MWh", "relax-constraints": False}, ) scheduler.deserialize_config() @@ -1879,7 +1884,11 @@ def test_battery_power_capacity_as_sensor( resolution = timedelta(minutes=15) soc_at_start = 10 - flex_context = {"site-power-capacity": "1100 kVA"} # 1.1 MW + flex_context = { + "site-power-capacity": "1100 kVA", # 1.1 MW + # relax-constraints False: this test asserts hard-capacity semantics. + "relax-constraints": False, + } if site_consumption_capacity_sensor: flex_context["site-consumption-capacity"] = { "sensor": capacity_sensors["site_power_capacity"].id diff --git a/flexmeasures/data/schemas/scheduling/__init__.py b/flexmeasures/data/schemas/scheduling/__init__.py index 5bd8021e6a..93c86ac555 100644 --- a/flexmeasures/data/schemas/scheduling/__init__.py +++ b/flexmeasures/data/schemas/scheduling/__init__.py @@ -41,6 +41,7 @@ is_capacity_price_unit, is_energy_price_unit, is_power_unit, + is_currency_unit, is_energy_unit, ) from flexmeasures.utils.validation_utils import validate_variable_quantity @@ -269,7 +270,7 @@ class SharedSchema(Schema): # Relaxation fields relax_constraints = fields.Bool( data_key="relax-constraints", - load_default=False, + load_default=True, metadata=metadata.RELAX_CONSTRAINTS.to_dict(), ) relax_soc_constraints = fields.Bool( @@ -393,13 +394,6 @@ class CommodityFlexContextSchema(SharedSchema): metadata=metadata.COMMODITY_FLEX_CONTEXT.to_dict(), ) - # For flex-context listings (per commodity), default relax_constraints to True - relax_constraints = fields.Bool( - data_key="relax-constraints", - load_default=True, - metadata=metadata.RELAX_CONSTRAINTS.to_dict(), - ) - def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) @@ -412,6 +406,23 @@ def __init__(self, *args, **kwargs): class FlexContextSchema(SharedSchema): """This schema defines fields that provide context to the portfolio to be optimized.""" + # The single-dict flex-context form only supports the electricity commodity. + # Other commodities must be defined via the `commodities` list. + # Not part of the documented UI/OpenAPI fields. + commodity = fields.Str( + required=False, + load_default="electricity", + data_key="commodity", + validate=validate.OneOf( + ["electricity"], + error="The top-level flex-context dict only supports the 'electricity' " + "commodity. Use the `commodities` list to define other commodities.", + ), + metadata=dict( + description="Commodity of the single-dict flex-context form; only 'electricity' is supported here. Use the `commodities` list to define other commodities.", + ), + ) + commodity_contexts = fields.Nested( CommodityFlexContextSchema, data_key="commodities", @@ -448,41 +459,62 @@ def validate_aggregate_power_is_sensor( if not isinstance(aggregate_power, Sensor): raise ValidationError("The `aggregate-power` field can only be a Sensor.") + @validates("commodity_contexts") + def validate_commodity_contexts_unique( + self, commodity_contexts: list[dict], **kwargs + ): + """Validate that each commodity is listed at most once. + + `_get_commodity_contexts` (storage.py) builds a dict keyed by commodity, so + duplicate entries would otherwise silently overwrite each other. + """ + commodities = [context["commodity"] for context in commodity_contexts] + seen = set() + duplicates = set() + for commodity in commodities: + if commodity in seen: + duplicates.add(commodity) + seen.add(commodity) + if duplicates: + raise ValidationError( + f"Each commodity may only be listed once in `commodities`. Duplicate(s): {sorted(duplicates)}." + ) + @validates("commodity_contexts") def validate_commodity_contexts_shared_currency( self, commodity_contexts: list[dict], **kwargs ): - """Validate that all prices across commodity contexts share the same currency.""" + """Validate that all prices across commodity contexts share the same currency. + + Each commodity context already computed its own normalized ``shared_currency_unit`` + (a base-unit currency string, e.g. "EUR") via the inherited + ``_try_to_convert_price_units`` schema-level validator. We simply compare those. + """ if not commodity_contexts: return shared_currency_unit = None for context in commodity_contexts: - # Check all price fields in this context - for field_name, field_value in context.items(): - if field_name.endswith("_price") and field_value is not None: - # Get the price unit - if hasattr(field_value, "units"): - price_unit = str(field_value.units) - elif isinstance(field_value, ur.Quantity): - price_unit = str(field_value.units) - else: - continue - - # Extract currency from the price unit - # Price units are typically like "EUR/MWh" or "USD/MW" - # Split by "/" and take first part as currency - currency_unit = price_unit.split("/")[0].strip() - - if shared_currency_unit is None: - shared_currency_unit = str( - ur.Quantity(currency_unit).to_base_units().units - ) - elif not units_are_convertible(currency_unit, shared_currency_unit): - raise ValidationError( - "all prices in the flex-context must share the same currency unit" - ) + context_currency_unit = context.get("shared_currency_unit") + if context_currency_unit is None: + continue + if shared_currency_unit is None: + shared_currency_unit = context_currency_unit + elif not units_are_convertible(context_currency_unit, shared_currency_unit): + raise ValidationError( + "all prices in the flex-context must share the same currency unit" + f" (found both '{shared_currency_unit}' and '{context_currency_unit}')" + ) + + # Note: we deliberately tolerate a `commodities` list combined with top-level + # commodity-specific (SharedSchema) fields. In the API path, a multi-commodity + # list is normalized to {"commodities": [...]} and collect_flex_config then + # dict-merges the asset's db-stored flex-context (e.g. "site-power-capacity", + # "consumption-price") at the top level, so rejecting this mix would 422 any + # asset with stored electricity flex-context fields. Semantics: top-level fields + # serve as the electricity context only when the commodities list has no + # electricity entry (see _get_commodity_contexts in storage.py). @validates_schema(pass_original=True) def check_prices(self, data: dict, original_data: dict, **kwargs): @@ -507,8 +539,10 @@ def check_prices(self, data: dict, original_data: dict, **kwargs): field.data_key: field_var for field_var, field in self.declared_fields.items() } + # Only count fields that were explicitly passed (not filled in by a load_default, + # such as relax-constraints, which defaults to True). if any( - field_map[field] in data and data[field_map[field]] + field in original_data and data.get(field_map[field]) for field in ( "soc-minima-breach-price", "soc-maxima-breach-price", @@ -539,6 +573,35 @@ def check_prices(self, data: dict, original_data: dict, **kwargs): data = self._try_to_convert_price_units(data, original_data) shared_currency = ur.Quantity(data["shared_currency_unit"]) + # Also check that top-level prices share their currency with any per-commodity contexts + for context in data.get("commodity_contexts", []) or []: + context_currency_unit = context.get("shared_currency_unit") + if context_currency_unit is None: + continue + if not units_are_convertible( + context_currency_unit, data["shared_currency_unit"] + ): + raise ValidationError( + "all prices in the flex-context must share the same currency unit" + f" (found both '{data['shared_currency_unit']}' at the top level and" + f" '{context_currency_unit}' in a commodity context)", + field_name="commodities", + ) + + # Skip filling default breach prices when: + # - the deprecated price sensor fields are used (those predate relaxation + # support; filling defaults would silently change legacy behaviour), or + # - the shared currency is not an actual currency (e.g. a mis-united price + # field slipped through _try_to_convert_price_units); filling defaults in a + # nonsense currency would misattribute unit errors to the breach price + # fields in downstream validation (e.g. DBFlexContextSchema). + if ( + "consumption_price_sensor" in data + or "production_price_sensor" in data + or not is_currency_unit(data["shared_currency_unit"]) + ): + return data + # Fill in default soc breach prices when asked to relax SoC constraints, unless already set explicitly. if ( data["relax_soc_constraints"] @@ -1176,6 +1239,15 @@ def normalize_flex_context_format(self, data, **kwargs): # Regular list case data["flex-context"] = {"commodities": raw_flex_context} # else: already a dict, leave as-is + + # By now, flex-context should always be normalized to a dict. If it isn't + # (e.g. a bare string or number was passed), raise a 422 here instead of + # letting downstream code fail with a TypeError. + if not isinstance(data["flex-context"], dict): + raise ValidationError( + "`flex-context` must be an object, or a list of objects.", + field_name="flex-context", + ) return data @validates_schema diff --git a/flexmeasures/data/schemas/tests/test_scheduling.py b/flexmeasures/data/schemas/tests/test_scheduling.py index aef5fbef9d..90842d2cf9 100644 --- a/flexmeasures/data/schemas/tests/test_scheduling.py +++ b/flexmeasures/data/schemas/tests/test_scheduling.py @@ -932,7 +932,7 @@ def test_flex_model_schemas( }, False, ), - # Test that relax_constraints defaults to False in FlexContextSchema + # Test that relax_constraints defaults to True in FlexContextSchema ( {"site-power-capacity": "1 MVA"}, False, @@ -1135,3 +1135,65 @@ def test_flex_context_listing_shared_currency( schema = FlexContextSchema() check_schema_loads_data(schema=schema, data=flex_context_listing, fails=fails) + + +def test_flex_context_listing_rejects_duplicate_commodities(db, app): + """test_flex_context_listing_rejects_duplicate_commodities: a commodity listed twice must be rejected.""" + schema = FlexContextSchema() + flex_context = { + "commodities": [ + {"commodity": "electricity", "consumption-price": "1 EUR/MWh"}, + {"commodity": "electricity", "production-price": "1 EUR/MWh"}, + ] + } + check_schema_loads_data( + schema=schema, + data=flex_context, + fails={"commodities": "may only be listed once"}, + ) + + +def test_flex_context_single_dict_rejects_non_electricity_commodity(db, app): + """test_flex_context_single_dict_rejects_non_electricity_commodity: the single-dict form only supports electricity.""" + schema = FlexContextSchema() + flex_context = {"commodity": "gas", "consumption-price": "1 EUR/MWh"} + check_schema_loads_data( + schema=schema, + data=flex_context, + fails={"commodity": "only supports the 'electricity' commodity"}, + ) + + +def test_flex_context_single_dict_allows_explicit_electricity_commodity(db, app): + """test_flex_context_single_dict_allows_explicit_electricity_commodity: explicit electricity is fine.""" + schema = FlexContextSchema() + flex_context = {"commodity": "electricity", "consumption-price": "1 EUR/MWh"} + check_schema_loads_data(schema=schema, data=flex_context, fails=False) + + +def test_flex_context_tolerates_commodities_with_top_level_shared_fields(db, app): + """test_flex_context_tolerates_commodities_with_top_level_shared_fields: mixing must be tolerated. + + The API path dict-merges an asset's db-stored (electricity) flex-context fields at the + top level after normalizing a multi-commodity list to {"commodities": [...]}, so this + mix must load fine. Top-level fields serve as the electricity context only when the + commodities list has no electricity entry (see _get_commodity_contexts in storage.py). + """ + schema = FlexContextSchema() + flex_context = { + "consumption-price": "1 EUR/MWh", + "commodities": [ + {"commodity": "gas", "consumption-price": "1 EUR/MWh"}, + ], + } + check_schema_loads_data(schema=schema, data=flex_context, fails=False) + + +def test_asset_trigger_schema_rejects_malformed_flex_context(app): + """test_asset_trigger_schema_rejects_malformed_flex_context: a non-dict/list flex-context must raise a ValidationError, not a TypeError.""" + from flexmeasures.data.schemas.scheduling import AssetTriggerSchema + + schema = AssetTriggerSchema() + with pytest.raises(ValidationError) as e_info: + schema.normalize_flex_context_format({"flex-context": "not-a-dict-or-list"}) + assert "flex-context" in str(e_info.value) diff --git a/flexmeasures/ui/tests/test_utils.py b/flexmeasures/ui/tests/test_utils.py index eff2114fc6..b9925823b7 100644 --- a/flexmeasures/ui/tests/test_utils.py +++ b/flexmeasures/ui/tests/test_utils.py @@ -80,6 +80,7 @@ def test_ui_flexcontext_schema(): "consumption-price-sensor", "production-price-sensor", "commodities", # todo: https://github.com/FlexMeasures/flexmeasures/issues/2230 + "commodity", # single-dict form is electricity-only; not exposed in the UI ] schema_keys = []