diff --git a/documentation/changelog.rst b/documentation/changelog.rst index 27498b70f4..ac246dff09 100644 --- a/documentation/changelog.rst +++ b/documentation/changelog.rst @@ -43,6 +43,7 @@ New features * Add support for intermediate power constraints on groups of devices, via a new ``group`` field in the storage flex-model [see `PR #2276 `_ and `issue #2092 `_] * The ``group`` field now also accepts a ``{"asset": }`` reference (in addition to ``{"sensor": }``), allowing intermediate power constraints to be defined entirely from flex-models stored on the asset tree, with results saved via the group's ``consumption``/``production`` output sensors, without needing any flex-model in the scheduling trigger [see `issue #2092 `_] * Extended the scheduling job ``result`` field with a ``num-beliefs`` field reporting the total number of beliefs (scheduled values) saved to the database [see `PR #2280 `_] +* Flex-context commitments can be scoped to a subset of devices, via a new optional ``sensors`` list (any devices, possibly across electrical groups) or ``group`` reference (an electrical group's members); the commitment binds the net signed aggregate flow (consumption positive, production negative) of those devices — flexible and inflexible alike — as one commitment, instead of binding each device separately [see `PR #2295 `_] * Migrate the asset tree in the UI's Structure tab from Vega to ECharts, adding interactive pan/zoom navigation and refreshed node styling [see `PR #2025 `_ and `PR #2365 `_] Infrastructure / Support diff --git a/flexmeasures/data/models/planning/devices.py b/flexmeasures/data/models/planning/devices.py index e0eb1852bd..092586c2bc 100644 --- a/flexmeasures/data/models/planning/devices.py +++ b/flexmeasures/data/models/planning/devices.py @@ -624,6 +624,18 @@ 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] + def scheduled_devices_by_sensor_id(self, sensor_id: int) -> list[FlexDevice]: + """Return all devices (flexible and inflexible) whose power sensor has the given id. + + Unlike :meth:`by_sensor_id`, this includes inflexible (fixed-power) devices, + so a commitment scoped to a sensor list can bind an inflexible device's flow too. + """ + return [ + device + for device in (*self.devices, *self.inflexible_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. diff --git a/flexmeasures/data/models/planning/storage.py b/flexmeasures/data/models/planning/storage.py index acf8ab23c3..e4d6765c24 100644 --- a/flexmeasures/data/models/planning/storage.py +++ b/flexmeasures/data/models/planning/storage.py @@ -22,6 +22,7 @@ DeviceInventory, _resolve_stock_key, group_key_label, + resolve_group_reference, ) from flexmeasures.data.models.planning.linear_optimization import device_scheduler from flexmeasures.data.models.planning.utils import ( @@ -1545,6 +1546,92 @@ def device_list_series( commitments, ) + def _resolve_commitment_scope( + self, scoped_sensors, scoped_group + ) -> tuple[list[int], str]: + """Resolve a scoped commitment's device set to canonical solver indices. + + Both scopes include a device whether it is flexible or inflexible: + a ``group`` scope yields the group's (leaf) members, + and a ``sensors`` scope yields the devices recording the listed power sensors. + So listing a group's member sensors resolves to the same set as scoping by that group. + Canonical indices always come from the device inventory, + never from re-enumerating raw flex-model lists. + + The commitment then binds the *net signed* aggregate of these devices' flow (consumption positive, production negative), + so consumers add, producers subtract, and any inflexible (fixed) member contributes its fixed signed power. + + :returns: A ``(sorted device indices, human-readable scope description)`` pair. + """ + if scoped_group is not None: + group_key = resolve_group_reference(scoped_group) + scoped_devices = sorted( + self.device_inventory.group_to_devices.get(group_key, []) + ) + description = ( + f"group {group_key_label(group_key)}" + if group_key is not None + else "an unresolved group reference" + ) + return scoped_devices, description + scoped_sensor_ids = { + sensor.id if hasattr(sensor, "id") else sensor for sensor in scoped_sensors + } + scoped_devices = sorted( + device.index + for sensor_id in scoped_sensor_ids + for device in self.device_inventory.scheduled_devices_by_sensor_id( + sensor_id + ) + ) + return scoped_devices, f"sensors {sorted(scoped_sensor_ids)}" + + def _build_scoped_commitment( + self, commitment_spec, scoped_sensors, scoped_group, commitment_index + ) -> "FlowCommitment | None": + """Build one aggregate-flow FlowCommitment for a scoped commitment. + + Returns None (logged) when the scope matches no device in the flex-model, + so the commitment binds nothing rather than failing the whole schedule. + + :raises ValueError: When the scoped devices span more than one commodity. + """ + scoped_devices, scope_description = self._resolve_commitment_scope( + scoped_sensors, scoped_group + ) + if not scoped_devices: + current_app.logger.warning( + f"Commitment '{commitment_spec.get('name')}' is scoped to" + f" {scope_description}, none of which appear in the flex-model." + " This commitment will not bind any device." + ) + return None + commodities = { + self.device_inventory.by_index(d).commodity for d in scoped_devices + } + if len(commodities) > 1: + raise ValueError( + f"Commitment '{commitment_spec.get('name')}' is scoped to devices of" + f" more than one commodity ({sorted(commodities)}); a commitment binds" + " the aggregate flow of a single commodity." + ) + # A scoped commitment's commodity is defined by its scope, + # so pin it to the scoped devices' (single) commodity; + # otherwise the commitment keeps the schema's electricity default (or a mismatching explicit value), + # and its cost would be misattributed to the wrong commodity. + commitment_spec["commodity"] = next(iter(commodities)) + index = commitment_spec["index"] + # device_group maps device index -> group label; + # one shared label makes the engine bind the aggregate flow. + # The label is unique per commitment, so two scoped commitments never merge, even if they share a name. + group_label = f"scoped-commitment-{commitment_index}" + return FlowCommitment( + device=pd.Series([scoped_devices] * len(index), index=index), + device_group=pd.Series({d: group_label for d in scoped_devices}), + provenance="custom", + **commitment_spec, + ) + def convert_to_commitments( self, flex_model, @@ -1579,7 +1666,7 @@ def convert_to_commitments( commitments = [] # The specs were copied above, so converting (which pops fields) does not # mutate self.flex_context and repeated conversions see the original specs. - for commitment_spec in commitment_specs: + for commitment_index, commitment_spec in enumerate(commitment_specs): # Convert baseline, up_price and down_price to pd.Series, then create FlowCommitment if "up_price" in commitment_spec: @@ -1608,6 +1695,22 @@ def convert_to_commitments( start, end, timing_kwargs["resolution"] ) commitment_commodity = commitment_spec.get("commodity", "electricity") + + # A commitment scoped to a subset of devices binds the *aggregate* flow of those devices as one commitment, + # rather than each device separately. + # The scope is given either as a raw list of power `sensors` (a cherry-pick that may span electrical groups, + # e.g. an aFRR band on a site's e-heaters), + # or as a `group` reference (the members of an electrical group, reusing its already-resolved membership); + # the schema allows at most one of the two. + scoped_sensors = commitment_spec.pop("sensors", None) + scoped_group = commitment_spec.pop("group", None) + if scoped_sensors is not None or scoped_group is not None: + scoped = self._build_scoped_commitment( + commitment_spec, scoped_sensors, scoped_group, commitment_index + ) + if scoped is not None: + commitments.append(scoped) + continue bound_device_count = 0 for d, flex_model_d in enumerate(flex_model): device_commodity = flex_model_d.get("commodity", "electricity") diff --git a/flexmeasures/data/models/planning/tests/test_commitments.py b/flexmeasures/data/models/planning/tests/test_commitments.py index 3f9d6b6e7f..ca0fc154f3 100644 --- a/flexmeasures/data/models/planning/tests/test_commitments.py +++ b/flexmeasures/data/models/planning/tests/test_commitments.py @@ -2085,6 +2085,352 @@ def test_commitment_commodity_does_not_bind_other_commodity_devices(): assert set(electricity_commitment.device_group.unique()) == {"electricity"} +def test_sensor_scoped_commitment_binds_aggregate_of_selected_devices(app, db): + """A commitment scoped to specific sensors (here: two e-heaters) binds their aggregate flow as one commitment: + a baseline of 10 MW with a steep penalty on downward deviation keeps their combined consumption at 10 MW, + even though a cheaper allocation (0 MW) exists. + + "Band" (as in the "reserved band" commitment name) means a committed power level the aggregate is held to, + by penalising deviation from the baseline; + here only downward deviation is priced, so the band acts as a floor rather than a two-sided range. + """ + heater_type = get_or_create_model(GenericAssetType, name="e-heater") + site = GenericAsset( + name="Band site (scoped commitment test)", generic_asset_type=heater_type + ) + db.session.add(site) + db.session.flush() + + resolution = pd.Timedelta("1h") + start = pd.Timestamp("2026-02-01T00:00:00+01:00") + end = pd.Timestamp("2026-02-01T04:00:00+01:00") + + def sensor(name): + s = Sensor( + name=name, unit="MW", event_resolution=resolution, generic_asset=site + ) + db.session.add(s) + return s + + heater_1 = sensor("band heater 1") + heater_2 = sensor("band heater 2") + db.session.flush() + + flex_model = [ + { + # Heaters burn money at the consumption price; + # without the band commitment the optimum is to stay off. + "sensor": heater_1.id, + "power-capacity": "8 MW", + "consumption-capacity": "8 MW", + "production-capacity": "0 kW", + }, + { + "sensor": heater_2.id, + "power-capacity": "8 MW", + "consumption-capacity": "8 MW", + "production-capacity": "0 kW", + }, + ] + flex_context = { + "consumption-price": "50 EUR/MWh", + "production-price": "50 EUR/MWh", + "site-power-capacity": "1 GW", + "commitments": [ + { + "name": "reserved band", + "sensors": [heater_1.id, heater_2.id], + "baseline": "10 MW", + # Steep penalty for consuming less than the band (negative price penalizes downward deviation); + # consuming more is free. + "down-price": "-10000 EUR/MWh", + } + ], + } + + scheduler = StorageScheduler( + asset_or_sensor=site, + start=start, + end=end, + resolution=resolution, + belief_time=start, + flex_model=flex_model, + flex_context=flex_context, + return_multiple=True, + ) + results = scheduler.compute(skip_validation=True) + schedules = { + r["sensor"]: r["data"] for r in results if r.get("name") == "storage_schedule" + } + combined = schedules[heater_1] + schedules[heater_2] + # The band keeps the aggregate at 10 MW (cheapest way to avoid the penalty), + # even though each heater alone (8 MW max) could not carry it. + np.testing.assert_allclose(combined.iloc[:-1], 10.0, rtol=1e-4) + + +def test_commitment_scope_sensors_and_group_are_mutually_exclusive(app, db): + """A commitment's scope is either a sensor list or a group reference, not both.""" + from flexmeasures.data.schemas.scheduling import CommitmentSchema + from marshmallow import ValidationError + + heater_type = get_or_create_model(GenericAssetType, name="e-heater") + site = GenericAsset(name="scope-conflict site", generic_asset_type=heater_type) + db.session.add(site) + db.session.flush() + power = Sensor( + name="scope-conflict power", + unit="MW", + event_resolution=pd.Timedelta("1h"), + generic_asset=site, + ) + db.session.add(power) + db.session.flush() + + with pytest.raises(ValidationError, match="not both"): + CommitmentSchema().load( + { + "name": "conflicted", + "baseline": "1 MW", + "up-price": "1 EUR/MWh", + "sensors": [power.id], + "group": {"sensor": power.id}, + } + ) + + +def test_group_scoped_commitment_binds_group_aggregate(app, db): + """A commitment scoped to a ``group`` reference binds the aggregate flow of that group's members, + reusing the group's resolved membership; + the same band effect as listing the members' sensors, but pointing at the group instead. + """ + heater_type = get_or_create_model(GenericAssetType, name="e-heater") + site = GenericAsset(name="Group-scoped band site", generic_asset_type=heater_type) + db.session.add(site) + db.session.flush() + + resolution = pd.Timedelta("1h") + start = pd.Timestamp("2026-02-01T00:00:00+01:00") + end = pd.Timestamp("2026-02-01T04:00:00+01:00") + + def sensor(name): + s = Sensor( + name=name, unit="MW", event_resolution=resolution, generic_asset=site + ) + db.session.add(s) + return s + + heater_1 = sensor("group band heater 1") + heater_2 = sensor("group band heater 2") + group_sensor = sensor("group aggregate sensor") + db.session.flush() + + flex_model = [ + { + "sensor": heater_1.id, + "power-capacity": "8 MW", + "consumption-capacity": "8 MW", + "production-capacity": "0 kW", + "group": {"sensor": group_sensor.id}, + }, + { + "sensor": heater_2.id, + "power-capacity": "8 MW", + "consumption-capacity": "8 MW", + "production-capacity": "0 kW", + "group": {"sensor": group_sensor.id}, + }, + # The group entry (a loose cap so it does not itself bind the aggregate). + {"sensor": group_sensor.id, "power-capacity": "1 GW"}, + ] + flex_context = { + "consumption-price": "50 EUR/MWh", + "production-price": "50 EUR/MWh", + "site-power-capacity": "1 GW", + "commitments": [ + { + "name": "reserved band on the group", + "group": {"sensor": group_sensor.id}, + "baseline": "10 MW", + "down-price": "-10000 EUR/MWh", + } + ], + } + + scheduler = StorageScheduler( + asset_or_sensor=site, + start=start, + end=end, + resolution=resolution, + belief_time=start, + flex_model=flex_model, + flex_context=flex_context, + return_multiple=True, + ) + results = scheduler.compute(skip_validation=True) + schedules = { + r["sensor"]: r["data"] for r in results if r.get("name") == "storage_schedule" + } + combined = schedules[heater_1] + schedules[heater_2] + # The band on the group keeps its members' aggregate at 10 MW. + np.testing.assert_allclose(combined.iloc[:-1], 10.0, rtol=1e-4) + + +def test_sensor_scope_includes_inflexible_and_matches_group_scope(app): + """A sensors scope includes an inflexible device (by its power sensor), + so listing a group's member sensors binds the same device set as scoping by that group. + """ + from flexmeasures.data.models.planning.devices import DeviceInventory + + scheduler = object.__new__(StorageScheduler) + start = pd.Timestamp("2024-01-01T00:00:00+01:00") + end = pd.Timestamp("2024-01-01T03:00:00+01:00") + resolution = pd.Timedelta("1h") + + def mk(sid, name): + s = Sensor( + name=name, unit="MW", event_resolution=resolution, generic_asset_id=1 + ) + s.id = sid + return s + + battery = mk(1, "scope battery") + load = mk(12, "scope fixed load") + group_sensor = mk(10, "scope group sensor") + + flex_model = [ + {"sensor": battery, "group": {"sensor": group_sensor}}, + { + "asset": object(), + "inflexible_consumption": load, + "group": {"sensor": group_sensor}, + }, + {"sensor": group_sensor, "power_capacity_in_mw": ur.Quantity("1 GW")}, + ] + + def commitment_devices(scope): + scheduler.flex_context = { + "shared_currency_unit": "EUR", + "commitments": [ + { + "name": "band", + "baseline": ur.Quantity("1 MW"), + "up_price": ur.Quantity("1 EUR/MWh"), + **scope, + } + ], + } + scheduler.device_inventory = DeviceInventory.from_flex_config( + flex_model, scheduler.flex_context + ) + commitments = scheduler.convert_to_commitments( + flex_model, + query_window=(start, end), + resolution=resolution, + beliefs_before=start, + ) + # FlowCommitment.device is a Series of (identical) device-index lists. + return set(commitments[0].device.iloc[0]) + + by_sensors = commitment_devices({"sensors": [battery, load]}) + by_group = commitment_devices({"group": {"sensor": group_sensor}}) + + assert by_sensors == by_group # listing the members == scoping the group + assert by_sensors == {0, 1} # the flexible battery (0) and the inflexible load (1) + + +def test_scoped_commitment_pins_commodity_to_scoped_devices(app): + """A scoped commitment's commodity follows its scoped devices, + overriding the schema's electricity default, so its cost is attributed to the right commodity. + """ + from flexmeasures.data.models.planning.devices import DeviceInventory + + scheduler = object.__new__(StorageScheduler) + start = pd.Timestamp("2024-01-01T00:00:00+01:00") + end = pd.Timestamp("2024-01-01T03:00:00+01:00") + resolution = pd.Timedelta("1h") + + gas_load = Sensor( + name="scoped gas load", + unit="MW", + event_resolution=resolution, + generic_asset_id=1, + ) + gas_load.id = 12 + flex_model = [{"sensor": gas_load, "commodity": "gas"}] + scheduler.flex_context = { + "shared_currency_unit": "EUR", + "commitments": [ + { + "name": "gas band", + "commodity": "electricity", # as the schema's electricity default supplies + "sensors": [gas_load], + "baseline": ur.Quantity("1 MW"), + "up_price": ur.Quantity("1 EUR/MWh"), + } + ], + } + scheduler.device_inventory = DeviceInventory.from_flex_config( + flex_model, scheduler.flex_context + ) + + commitments = scheduler.convert_to_commitments( + flex_model, + query_window=(start, end), + resolution=resolution, + beliefs_before=start, + ) + assert len(commitments) == 1 + # Pinned to the scoped device's commodity, not the "electricity" default. + assert commitments[0].commodity == "gas" + + +def test_scoped_commitment_with_no_matching_devices_warns_and_binds_nothing( + app, caplog +): + """A scope that matches no device in the flex-model logs a warning and binds nothing, + rather than failing the whole schedule.""" + import logging + from flexmeasures.data.models.planning.devices import DeviceInventory + + scheduler = object.__new__(StorageScheduler) + start = pd.Timestamp("2024-01-01T00:00:00+01:00") + end = pd.Timestamp("2024-01-01T03:00:00+01:00") + resolution = pd.Timedelta("1h") + + present = Sensor( + name="present device", + unit="MW", + event_resolution=resolution, + generic_asset_id=1, + ) + present.id = 1 + flex_model = [{"sensor": present, "commodity": "electricity"}] + scheduler.flex_context = { + "shared_currency_unit": "EUR", + "commitments": [ + { + "name": "orphan band", + "sensors": [999], # no device in the flex-model records this sensor + "baseline": ur.Quantity("1 MW"), + "up_price": ur.Quantity("1 EUR/MWh"), + } + ], + } + scheduler.device_inventory = DeviceInventory.from_flex_config( + flex_model, scheduler.flex_context + ) + + with caplog.at_level(logging.WARNING): + commitments = scheduler.convert_to_commitments( + flex_model, + query_window=(start, end), + resolution=resolution, + beliefs_before=start, + ) + assert commitments == [] # bound nothing, did not raise + assert "will not bind any device" in caplog.text + + def test_commitments_in_commodity_contexts_are_converted(app): """Commitments saved within a commodity context (as the UI editor does per commodity tab) are picked up by the scheduler and bind that context's commodity. diff --git a/flexmeasures/data/models/planning/tests/test_device_inventory.py b/flexmeasures/data/models/planning/tests/test_device_inventory.py index 6705e1070b..67eec9e8f2 100644 --- a/flexmeasures/data/models/planning/tests/test_device_inventory.py +++ b/flexmeasures/data/models/planning/tests/test_device_inventory.py @@ -210,6 +210,24 @@ def test_by_sensor_id(): assert inventory.by_sensor_id(3) == [] +def test_scheduled_devices_by_sensor_id_includes_inflexible(): + """scheduled_devices_by_sensor_id returns flexible *and* inflexible devices, + whereas by_sensor_id returns flexible devices only.""" + battery = make_sensor(1) + load = make_sensor(12) + inventory = DeviceInventory.from_flex_config( + [ + {"sensor": battery}, + {"asset": object(), "inflexible_consumption": load}, + ] + ) + # by_sensor_id is flexible-only, so the inflexible load's sensor matches nothing. + assert inventory.by_sensor_id(12) == [] + # scheduled_devices_by_sensor_id includes the inflexible device (index 1). + assert [d.index for d in inventory.scheduled_devices_by_sensor_id(12)] == [1] + assert [d.index for d in inventory.scheduled_devices_by_sensor_id(1)] == [0] + + 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 diff --git a/flexmeasures/data/schemas/scheduling/__init__.py b/flexmeasures/data/schemas/scheduling/__init__.py index 980878efe0..ae640c2f94 100644 --- a/flexmeasures/data/schemas/scheduling/__init__.py +++ b/flexmeasures/data/schemas/scheduling/__init__.py @@ -30,6 +30,7 @@ PriceField, ) from flexmeasures.data.schemas.scheduling import metadata +from flexmeasures.data.schemas.scheduling.groups import GroupReferenceSchema from flexmeasures.data.schemas.units import UnitField from flexmeasures.utils.doc_utils import rst_to_openapi from flexmeasures.data.schemas.times import ( @@ -75,6 +76,26 @@ def forbid_time_series_specs(self, data: dict, **kwargs): class CommitmentSchema(Schema): name = fields.Str(required=True, data_key="name", validate=validate.Length(min=1)) + # Optional scoping: bind this commitment to the aggregate flow of a subset of devices, + # rather than binding each device of the commodity separately. + # Give either a list of power `sensors` (a cherry-pick that may span electrical groups, + # e.g. an aFRR band on a site's e-heaters), + # or a `group` reference (the members of an electrical group); at most one of the two. + # Either scope includes a device whether flexible or inflexible, + # so listing a group's member sensors resolves to the same set as scoping by that group. + # The commitment binds the net signed aggregate of the scoped devices (consumption positive, production negative), + # so consumers add, producers subtract, + # and any inflexible (fixed) member contributes its fixed signed power (see StorageScheduler._resolve_commitment_scope). + sensors = fields.List( + SensorIdField(), + required=False, + data_key="sensors", + ) + group = fields.Nested( + GroupReferenceSchema, + required=False, + data_key="group", + ) # Not described in UI_FLEX_CONTEXT_SCHEMA or the Sphinx docs (it does show up # in the generated OpenAPI schema, without being promoted in field descriptions). # Internal bookkeeping only: not the documented way to associate a commitment @@ -101,6 +122,15 @@ class CommitmentSchema(Schema): data_key="down-price", ) + @validates_schema + def forbid_scope_conflict(self, commitment, **kwargs): + """A commitment's scope is a sensor list or a group reference, not both.""" + if "sensors" in commitment and "group" in commitment: + raise ValidationError( + "A commitment may be scoped by 'sensors' or by 'group', not both.", + field_name="sensors", + ) + @validates_schema def require_a_price(self, commitment, **kwargs): """A commitment is worthless without at least one deviation price. diff --git a/flexmeasures/ui/static/openapi-specs.json b/flexmeasures/ui/static/openapi-specs.json index f61a713a6d..afc7ce3704 100644 --- a/flexmeasures/ui/static/openapi-specs.json +++ b/flexmeasures/ui/static/openapi-specs.json @@ -4920,6 +4920,18 @@ } } }, + "GroupReference": { + "type": "object", + "properties": { + "sensor": { + "type": "integer" + }, + "asset": { + "type": "integer" + } + }, + "description": "Reference to a group of devices whose aggregate power is constrained." + }, "Commitment": { "type": "object", "properties": { @@ -4927,6 +4939,15 @@ "type": "string", "minLength": 1 }, + "sensors": { + "type": "array", + "items": { + "type": "integer" + } + }, + "group": { + "$ref": "#/components/schemas/GroupReference" + }, "commodity": { "type": "string", "default": "electricity" @@ -6749,18 +6770,6 @@ } ] }, - "GroupReference": { - "type": "object", - "properties": { - "sensor": { - "type": "integer" - }, - "asset": { - "type": "integer" - } - }, - "description": "Reference to a group of devices whose aggregate power is constrained." - }, "StorageFlexModelSchemaOpenAPI": { "type": "object", "properties": {