diff --git a/documentation/api/change_log.rst b/documentation/api/change_log.rst index c75a260f70..7a8306e100 100644 --- a/documentation/api/change_log.rst +++ b/documentation/api/change_log.rst @@ -9,6 +9,8 @@ v3.0-32 | July XX, 2026 """""""""""""""""""""""" - Extended ``GET /api/v3_0/jobs/`` with a ``result`` field containing ``unresolved`` and ``resolved`` arrays, each keyed by asset ID. For scheduling jobs, this surfaces soft state-of-charge constraint analysis: ``soc-minima`` and ``soc-maxima`` violations (with a ``violation`` magnitude) or satisfied constraints (with a ``margin`` headroom). Both arrays are empty when no SoC constraints were defined. +- Added a ``group`` field to the storage flex-model, accepted by the `/assets/(id)/schedules/trigger <../api/v3_0.html#post--api-v3_0-assets-id-schedules-trigger>`_ (POST) endpoint, referencing a power sensor representing a group of devices (e.g. a shared inverter or feeder). The group's ``power-capacity`` is enforced as a hard constraint on the group's aggregate power, while its ``consumption-capacity``/``production-capacity`` are enforced as soft constraints with default breach prices; the group's scheduled aggregate power is saved to the group sensor. +- The ``group`` field also accepts a ``{"asset": }`` reference (besides ``{"sensor": }``), pointing at an asset whose own (DB-stored) flex-model defines the group's constraints. Such a group defines no power sensor of its own; its aggregate schedule is instead saved via its ``consumption``/``production`` output sensor references, following the same conventions as any other asset-only flex-model entry. This lets the entire flex-model for a device tree (including groups) live in the DB, with ``flex-model`` omitted or empty on the trigger request. v3.0-31 | 2026-06-01 """""""""""""""""""" diff --git a/documentation/changelog.rst b/documentation/changelog.rst index 0eb355d858..9c79e9ca7e 100644 --- a/documentation/changelog.rst +++ b/documentation/changelog.rst @@ -25,6 +25,8 @@ New features * 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 `_] * 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 `_] +* 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 `_] * 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 `_] diff --git a/documentation/concepts/commitments.rst b/documentation/concepts/commitments.rst index 82cb1d4f9c..d185492e6d 100644 --- a/documentation/concepts/commitments.rst +++ b/documentation/concepts/commitments.rst @@ -115,7 +115,7 @@ It is possible to define what group of schedule values is expected to not deviat The ``'device'`` attribute already lets us reduce from the values for all devices to just one. The ``' _type'``attribute offers more powerful grouping options. -For now, this extra grouping can happen across different definitions of time slots, soon also per groups of devices. +For now, this extra grouping can happen across different definitions of time slots, and, for some cases, across groups of devices (see below). - ``_type == 'each'``: penalise deviations per time slot (this is the default for time series). - ``_type == 'any'``: treat the whole commitment horizon as one group (useful @@ -124,10 +124,7 @@ For now, this extra grouping can happen across different definitions of time slo .. note:: - Near-term feature: support for **grouping over devices** is planned and - will be documented here. When enabled, grouping over devices lets you express - soft constraints that aggregate deviations across a set of devices, - for example, an intermediate capacity constraint from a feeder shared by a group of devices (via **flow commitments**), or multiple power-to-heat devices that feed a shared thermal buffer (via **stock commitments**). + Grouping over devices is now partly supported. An intermediate power constraint from a feeder or shared inverter can be modelled by adding a ``group`` field to the relevant devices' storage flex-model entries, referencing a power sensor for the group; the group's ``power-capacity`` is enforced as a hard constraint, while its ``consumption-capacity``/``production-capacity`` are enforced as soft constraints with default breach prices (via **flow commitments**). See :ref:`scheduling` for details. Multiple devices feeding a shared storage (e.g. power-to-heat devices feeding a shared thermal buffer) are also supported, via multiple feeder sensors on one storage flex-model. Fuller, more general support for grouping deviations across arbitrary sets of devices in commitments (including custom breach prices per group, and stock commitments over device groups) remains planned and will be documented here as it lands. How flex-context fields are converted into commitments diff --git a/documentation/features/scheduling.rst b/documentation/features/scheduling.rst index aec2ce3397..3873d0d30a 100644 --- a/documentation/features/scheduling.rst +++ b/documentation/features/scheduling.rst @@ -259,6 +259,9 @@ For more details on the possible formats for field values, see :ref:`variable_qu * - ``production-capacity`` - |PRODUCTION_CAPACITY.example| (only consumption) - .. include:: ../_autodoc/PRODUCTION_CAPACITY.rst + * - ``group`` + - |GROUP.example| + - .. include:: ../_autodoc/GROUP.rst .. [#quantity_field] Can only be set as a fixed quantity. @@ -268,6 +271,42 @@ For more details on the possible formats for field values, see :ref:`variable_qu For more details on the possible formats for field values, see :ref:`variable_quantities`. + +Intermediate power constraints +""""""""""""""""""""""""""""""" + +In a multi-device flex-model list, a device entry may declare a ``group`` field referencing a group of devices, for example a hybrid inverter shared by a battery and PV installation, or a feeder shared by several devices. This lets you model an intermediate power constraint that sits between the individual devices and the site as a whole. The ``group`` field accepts exactly one of two references: + +- ``{"sensor": }``: the group is identified by a power sensor, which itself gets its own flex-model entry (typically passed alongside the device entries; mainly useful for API-passed flex-models). +- ``{"asset": }``: the group is identified by the flex-model entry stored on that asset (typically a sub-EMS/asset in the asset tree, such as the inverter in the example below). Such a group entry defines no power sensor of its own; instead, like any other asset-only entry, it may define ``consumption`` and/or ``production`` output sensor references (see below) on which the group's aggregate power gets saved. + +Either way, the group reference's target (sensor or asset) gets its own flex-model entry, defining constraints on the group's aggregate (summed) power: + +- ``power-capacity`` on the group is a **hard** constraint (applied in both directions). +- ``consumption-capacity`` and ``production-capacity`` on the group are **soft** constraints, enforced with the same default breach prices used at the site level (10000 currency/kW); users cannot configure custom breach prices for groups. + +The group's scheduled aggregate power is saved as a schedule output, following the same conventions used for any device's schedule output: + +- If the group's flex-model entry has a ``sensor`` field, the aggregate power is saved directly to that sensor. +- Otherwise (an asset-only entry), the aggregate power is saved via its ``consumption`` and/or ``production`` output sensor references: with only ``consumption`` set, the full profile is saved consumption-positive; with only ``production`` set, the full profile is saved production-positive (i.e. sign-flipped before saving); with both set, the profile is split into its non-negative part (saved to ``consumption``) and its non-positive part (saved, as a positive magnitude, to ``production``). + +Groups can be nested (a group entry may itself reference a parent group), but cyclic references are rejected. Groups require a multi-device flex-model; they are rejected when scheduling a single sensor. + +Example, for a 2.5 kW hybrid inverter (sensor 5) shared by a battery (sensor 1) and PV installation (sensor 2), taken from `issue #2092 `_: + +.. code-block:: json + + [ + {"sensor": 1, "power-capacity": "2 kW", "group": {"sensor": 5}}, + {"sensor": 2, "production-capacity": "2 kW", "consumption-capacity": "0 kW", "group": {"sensor": 5}}, + {"sensor": 5, "power-capacity": "2.5 kW"} + ] + +Here, the battery and PV installation may each individually schedule up to 2 kW, but their combined power flowing through the shared inverter is hard-limited to 2.5 kW. + +The ``{"asset": }`` variant lets you define the entire flex-model on the asset tree in the DB, with no flex-model needed in the scheduling trigger at all: each device asset carries its own (partial) flex-model, including a ``group`` field pointing at the parent asset that represents the shared equipment, and that parent asset's own flex-model defines the group's constraints and output sensor(s). Triggering a schedule for the top-level site asset with an empty (or omitted) ``flex-model`` then collects the full configuration from the tree. For a hands-on walkthrough (including how to store flex-models on assets, and where the resulting schedules end up), see :ref:`tut_toy_schedule_group_constraints`. + + Usually, not the whole flexibility model is needed. FlexMeasures can infer missing values in the flex model, and even get them (as default) from the sensor's attributes. diff --git a/documentation/index.rst b/documentation/index.rst index 89ebd0bbda..15c1d622e7 100644 --- a/documentation/index.rst +++ b/documentation/index.rst @@ -184,6 +184,7 @@ In :ref:`getting_started`, we have some helpful tips how to dive into this docum tut/toy-example-from-scratch tut/toy-example-expanded tut/toy-example-multiasset-curtailment + tut/toy-example-group-constraints tut/flex-model-v2g tut/multi-feed-storage tut/multi-commodity diff --git a/documentation/tut/toy-example-group-constraints.rst b/documentation/tut/toy-example-group-constraints.rst new file mode 100644 index 0000000000..5c13a8252d --- /dev/null +++ b/documentation/tut/toy-example-group-constraints.rst @@ -0,0 +1,158 @@ +.. _tut_toy_schedule_group_constraints: + + +Toy example IV: Intermediate power constraints (groups) +================================================================ + +So far, our flexible devices (the battery and the PV inverter) have only ever been constrained directly by the building's own grid connection capacity. +But in reality, several devices are often physically wired together behind a shared piece of equipment before they reach the site's connection, and that piece of equipment has its own power limit. + +The classic example is a **hybrid inverter**: a battery and a PV installation share one inverter, and while each device could individually push a lot of power, the inverter itself caps their *combined* power flow. +This is what the ``group`` field in the storage flex-model is for (see :ref:`storage_device_scheduler` for the general explanation). This tutorial shows a fully DB-driven setup, where the entire flex-model lives on the asset tree, and you trigger a schedule for the site with an empty flex-model. + +We'll build the following little asset tree: + +.. code-block:: text + + site (building) + └── inverter (hybrid inverter, hard power-capacity 2.5 kW) + ├── battery (device, group member) + └── PV (device, group member) + +Setting up the asset tree +--------------------------------------- + +We create the site, the inverter (the group) and the two devices as assets, with the inverter and devices being children of the site. +Each device also needs output sensors to record its schedule (since these devices won't have a dedicated power sensor of their own — they are "asset-only" flex-model entries), and the inverter needs an output sensor for the group's aggregate schedule. + +.. code-block:: bash + + $ flexmeasures add asset --name "toy site" --asset-type-id 5 --account-id 1 + Successfully created asset with ID 10. + + $ flexmeasures add asset --name "hybrid inverter" --asset-type-id 5 --account-id 1 --parent-asset 10 + Successfully created asset with ID 11. + + $ flexmeasures add asset --name "toy battery" --asset-type-id 5 --account-id 1 --parent-asset 10 + Successfully created asset with ID 12. + + $ flexmeasures add asset --name "toy PV" --asset-type-id 5 --account-id 1 --parent-asset 10 + Successfully created asset with ID 13. + + $ flexmeasures add sensor --name "inverter aggregate power" --unit MW --event-resolution PT15M --asset-id 11 + Successfully created sensor with ID 21. + + $ flexmeasures add sensor --name "battery consumption" --unit MW --event-resolution PT15M --asset-id 12 + Successfully created sensor with ID 22. + $ flexmeasures add sensor --name "battery production" --unit MW --event-resolution PT15M --asset-id 12 + Successfully created sensor with ID 23. + + $ flexmeasures add sensor --name "PV production" --unit MW --event-resolution PT15M --asset-id 13 + Successfully created sensor with ID 24. + +.. note:: Asset type IDs and IDs returned above will differ in your own setup — substitute your own. + +Storing the flex-models on the assets +--------------------------------------- + +Rather than sending a flex-model with the trigger request, we store each asset's (partial) flex-model directly on the asset. FlexMeasures walks the asset tree and collects these into one combined flex-model when scheduling the site. + +You can set an asset's flex-model with ``PATCH /api/v3_0/assets/``, sending a ``flex_model`` field with the JSON below. (The FlexMeasures UI's flex-model editor on the asset's properties page supports this too, and even suggests the parent asset as a candidate for the ``group`` field.) + +The battery is a device with both a consumption and production output sensor (it can charge and discharge), belonging to the inverter's group: + +.. code-block:: json + + { + "flex_model": { + "power-capacity": "2 kW", + "consumption-capacity": "2 kW", + "production-capacity": "2 kW", + "group": {"asset": 11}, + "consumption": {"sensor": 22}, + "production": {"sensor": 23} + } + } + +Sent as ``PATCH /api/v3_0/assets/12``. + +The PV installation only produces, so it only needs a production output sensor: + +.. code-block:: json + + { + "flex_model": { + "power-capacity": "2 kW", + "consumption-capacity": "0 kW", + "production-capacity": "2 kW", + "group": {"asset": 11}, + "production": {"sensor": 24} + } + } + +Sent as ``PATCH /api/v3_0/assets/13``. + +Finally, the inverter's own flex-model defines the group's hard power-capacity and where to save the group's aggregate schedule (as it has no power sensor of its own, either): + +.. code-block:: json + + { + "flex_model": { + "power-capacity": "2.5 kW", + "consumption": {"sensor": 21} + } + } + +Sent as ``PATCH /api/v3_0/assets/11``. + +Note that neither the battery, the PV installation, nor the inverter reference a ``sensor`` field of their own for scheduling purposes — this is what makes them "asset-only" entries. Instead, results are always saved via their ``consumption``/``production`` output sensor references. + +Triggering the schedule +--------------------------------------- + +We now trigger a schedule for the site (asset 10) with an empty (or omitted) flex-model. Everything the scheduler needs is picked up from the DB-stored flex-models on the asset tree. + +.. tabs:: + + .. tab:: CLI + + .. code-block:: bash + + $ flexmeasures add schedule \ + --asset 10 \ + --start ${TOMORROW}T00:00+01:00 --duration PT4H \ + --flex-model '[]' + New schedule is stored. + + .. tab:: API + + Example call: `[POST] http://localhost:5000/api/v3_0/assets/10/schedules/trigger <../api/v3_0.html#post--api-v3_0-assets-id-schedules-trigger>`_ (update the start date to tomorrow): + + .. code-block:: json + + { + "start": "2026-07-11T00:00+01:00", + "duration": "PT4H", + "flex-model": [] + } + +Inspecting the results +--------------------------------------- + +Once the job has finished, three schedules were computed and saved: + +- The battery's schedule, split (as it can both charge and discharge) between sensor 22 (``battery consumption``, holding the non-negative, consumption-positive part) and sensor 23 (``battery production``, holding the non-positive part, stored as a positive magnitude). +- The PV installation's schedule, saved entirely to sensor 24 (``PV production``), sign-flipped to be stored as a positive magnitude (since PV only produces). +- The inverter group's aggregate schedule, saved to sensor 21 (``inverter aggregate power``), equal to the (consumption-positive) sum of the battery's and PV's schedules. + +You can inspect any of these with: + +.. code-block:: bash + + $ flexmeasures show beliefs --sensor 21 --start ${TOMORROW}T00:00:00+01:00 --duration PT4H + +The group's aggregate power never exceeds 2.5 kW in either direction — even though the battery and PV could individually reach 2 kW each (4 kW combined) — because the hybrid inverter's hard ``power-capacity`` caps their sum. Without the group, the scheduler could plan the battery to charge at full power during peak PV production, which the inverter physically cannot deliver. + +.. note:: If a device only ever consumes or only ever produces, you only need to define the corresponding single output sensor (as we did for the PV installation above). Only devices (or groups) that can go both ways need both a ``consumption`` and a ``production`` output sensor. + +This concludes our tour of intermediate power constraints. For the full field reference, see :ref:`storage_device_scheduler` and the "Intermediate power constraints" section of :ref:`scheduling`. diff --git a/flexmeasures/api/v3_0/tests/test_asset_schedules_fresh_db.py b/flexmeasures/api/v3_0/tests/test_asset_schedules_fresh_db.py index 616a83b5b6..c928c6a701 100644 --- a/flexmeasures/api/v3_0/tests/test_asset_schedules_fresh_db.py +++ b/flexmeasures/api/v3_0/tests/test_asset_schedules_fresh_db.py @@ -777,3 +777,320 @@ def test_asset_trigger_with_flex_context_commodity_not_used( assert ( len(consumption_beliefs_heat) == 0 ), "heat aggregate-consumption should be empty since no heat device was scheduled" + + +@pytest.mark.parametrize( + "requesting_user", ["test_prosumer_user@seita.nl"], indirect=True +) +def test_asset_trigger_with_group_power_constraint( + app, + fresh_db, + add_market_prices_fresh_db, + setup_roles_users_fresh_db, + add_charging_station_assets_fresh_db, + keep_scheduling_queue_empty, + requesting_user, +): + """Test that a `group` flex-model entry bounds the aggregate power of its members. + + Two power sensors (fresh storage devices, siblings under the same charging hub) + declare membership of a group (referencing a power sensor also on the hub). A + third flex-model entry, for the group sensor itself, declares a hard + `power-capacity` that is lower than the sum of the members' individual + power-capacities, so it should actually bind (each device has ample state-of-charge + room and a cheap-price incentive to charge at full power simultaneously). We verify + that: + 1. the schedule is computed successfully (200 + job success) + 2. the beliefs saved on the group sensor equal the sum of the members' beliefs + 3. the group's aggregate power never exceeds the group's power-capacity + 4. (control run) without the group constraint, the same devices do jointly exceed + the group's power-capacity, so the test setup actually exercises the constraint + """ + bidirectional_charging_station = add_charging_station_assets_fresh_db[ + "Test charging station (bidirectional)" + ] + charging_hub = bidirectional_charging_station.parent_asset + + # Two fresh storage devices (power + SoC sensor each), with plenty of SoC room + # relative to their power-capacity, so each wants to charge at full power + # whenever prices are cheap. + def make_device(name: str) -> tuple[Sensor, Sensor]: + power_sensor = Sensor( + name=f"{name} power", + generic_asset=charging_hub, + unit="MW", + event_resolution=pd.Timedelta(minutes=30), + ) + soc_sensor = Sensor( + name=f"{name} soc", + generic_asset=charging_hub, + unit="MWh", + event_resolution=pd.Timedelta(minutes=0), + ) + fresh_db.session.add(power_sensor) + fresh_db.session.add(soc_sensor) + return power_sensor, soc_sensor + + sensor_1, soc_sensor_1 = make_device("device 1") + sensor_2, soc_sensor_2 = make_device("device 2") + + # Group sensor: a power sensor on the (parent) charging hub + group_sensor = Sensor( + name="group aggregate power", + generic_asset=charging_hub, + unit="MW", + event_resolution=pd.Timedelta(minutes=30), + ) + fresh_db.session.add(group_sensor) + fresh_db.session.flush() + + price_sensor_id = add_market_prices_fresh_db["epex_da"].id + + message = message_for_trigger_schedule(resolution="PT30M") + message["flex-context"] = { + "consumption-price": {"sensor": price_sensor_id}, + "production-price": {"sensor": price_sensor_id}, + "site-power-capacity": "1 TW", # avoid infeasibilities from the site cap + } + + def make_flex_model(sensor: Sensor, soc_sensor: Sensor, group_sensor: Sensor): + return { + "sensor": sensor.id, + "state-of-charge": {"sensor": soc_sensor.id}, + "soc-at-start": 0, + "soc-min": 0, + "soc-max": 10, # MWh: plenty of room relative to the 2 MW power-capacity + "soc-unit": "MWh", + "roundtrip-efficiency": "100%", + "storage-efficiency": "100%", + "power-capacity": "2 MW", + "group": {"sensor": group_sensor.id}, + } + + CP_1_flex_model = make_flex_model(sensor_1, soc_sensor_1, group_sensor) + CP_2_flex_model = make_flex_model(sensor_2, soc_sensor_2, group_sensor) + + # The group's own entry: a hard power-capacity lower than the sum of the members' + # individual power-capacities (2 MW + 2 MW = 4 MW), so it should actually bind. + group_power_capacity = "3 MW" + group_flex_model = { + "sensor": group_sensor.id, + "power-capacity": group_power_capacity, + } + + message["flex-model"] = [CP_1_flex_model, CP_2_flex_model, group_flex_model] + + # Trigger the schedule + assert len(app.queues["scheduling"]) == 0 + with app.test_client() as client: + trigger_response = client.post( + url_for("AssetAPI:trigger_schedule", id=charging_hub.id), + json=message, + ) + if trigger_response.status_code != 200: + print(f"Error response: {trigger_response.json}") + assert trigger_response.status_code == 200 + job_id = trigger_response.json["schedule"] + + # Process the scheduling queue + scheduled_jobs = app.queues["scheduling"].jobs + scheduling_job = scheduled_jobs[0] + work_on_rq(app.queues["scheduling"], exc_handler=handle_scheduling_exception) + assert ( + Job.fetch(job_id, connection=app.queues["scheduling"].connection).is_finished + is True + ) + + # Verify scheduler data source + scheduling_job.refresh() + scheduler_source = get_data_source_for_job(scheduling_job) + assert scheduler_source is not None + + def get_beliefs_as_series(sensor: Sensor) -> pd.Series: + beliefs = ( + TimedBelief.query.filter(TimedBelief.sensor_id == sensor.id) + .filter(TimedBelief.source_id == scheduler_source.id) + .all() + ) + assert len(beliefs) > 0, f"sensor {sensor.id} should have scheduled data" + return pd.Series( + [b.event_value for b in beliefs], + index=pd.DatetimeIndex([b.event_start for b in beliefs]), + ).sort_index() + + sensor_1_schedule = get_beliefs_as_series(sensor_1) + sensor_2_schedule = get_beliefs_as_series(sensor_2) + group_schedule = get_beliefs_as_series(group_sensor) + + # The group's aggregate should equal the sum of its members (same unit: MW) + aggregate_of_members = sensor_1_schedule.add(sensor_2_schedule, fill_value=0) + pd.testing.assert_series_equal( + aggregate_of_members.sort_index(), + group_schedule.sort_index(), + check_names=False, + atol=1e-6, + ) + + # The group's aggregate power should respect the (hard) power-capacity + group_power_capacity_mw = ur.Quantity(group_power_capacity).to("MW").magnitude + assert ( + group_schedule.abs() <= group_power_capacity_mw + 1e-6 + ).all(), "the group's aggregate power should never exceed its power-capacity" + + # Sanity check (control run): without the group constraint, the same two devices + # should be able to jointly exceed the group's power-capacity (each member alone is + # allowed up to 2 MW), confirming that the group constraint actually did something + # above rather than being vacuously satisfied. + control_message = message.copy() + control_CP_1_flex_model = dict(CP_1_flex_model) + control_CP_1_flex_model.pop("group") + control_CP_2_flex_model = dict(CP_2_flex_model) + control_CP_2_flex_model.pop("group") + control_message["flex-model"] = [control_CP_1_flex_model, control_CP_2_flex_model] + control_message["force-new-job-creation"] = True + + assert len(app.queues["scheduling"]) == 0 + with app.test_client() as client: + control_response = client.post( + url_for("AssetAPI:trigger_schedule", id=charging_hub.id), + json=control_message, + ) + assert control_response.status_code == 200 + control_job_id = control_response.json["schedule"] + + work_on_rq(app.queues["scheduling"], exc_handler=handle_scheduling_exception) + control_job = Job.fetch( + control_job_id, connection=app.queues["scheduling"].connection + ) + assert control_job.is_finished is True, ( + f"control job ended as '{control_job.get_status()}': " + f"{control_job.latest_result().exc_string if control_job.latest_result() else None}" + ) + control_scheduler_source = get_data_source_for_job( + Job.fetch(control_job_id, connection=app.queues["scheduling"].connection) + ) + assert control_scheduler_source is not None + + def get_control_beliefs_as_series(sensor: Sensor) -> pd.Series: + beliefs = ( + TimedBelief.query.filter(TimedBelief.sensor_id == sensor.id) + .filter(TimedBelief.source_id == control_scheduler_source.id) + .all() + ) + assert len(beliefs) > 0, f"sensor {sensor.id} should have scheduled data" + return pd.Series( + [b.event_value for b in beliefs], + index=pd.DatetimeIndex([b.event_start for b in beliefs]), + ).sort_index() + + control_sensor_1_schedule = get_control_beliefs_as_series(sensor_1) + control_sensor_2_schedule = get_control_beliefs_as_series(sensor_2) + control_aggregate = control_sensor_1_schedule.add( + control_sensor_2_schedule, fill_value=0 + ) + assert control_aggregate.abs().max() > group_power_capacity_mw, ( + "test setup should actually exercise the group constraint " + "(members should be able to jointly exceed the cap without it)" + ) + + +@pytest.mark.parametrize( + "requesting_user", ["test_prosumer_user@seita.nl"], indirect=True +) +def test_asset_trigger_with_group_referencing_sensor_outside_asset_tree( + app, + fresh_db, + add_market_prices_fresh_db, + setup_roles_users_fresh_db, + add_charging_station_assets_fresh_db, + keep_scheduling_queue_empty, + requesting_user, +): + """A `group` entry referencing a sensor outside the triggered asset's tree should 422. + + This is enforced by `AssetTriggerSchema.check_flex_model_sensors`, which requires + every sensor named in the flex-model (including group entries, since they are + ordinary flex-model list items) to belong to the triggered asset or one of its + offspring. + """ + bidirectional_charging_station = add_charging_station_assets_fresh_db[ + "Test charging station (bidirectional)" + ] + charging_station = add_charging_station_assets_fresh_db["Test charging station"] + charging_hub = bidirectional_charging_station.parent_asset + + sensor_1 = bidirectional_charging_station.sensors[0] + sensor_2 = charging_station.sensors[0] + bi_soc_sensor = add_charging_station_assets_fresh_db["bi-soc"] + uni_soc_sensor = add_charging_station_assets_fresh_db["uni-soc"] + + # Create a standalone asset outside of the charging hub's tree to host a sensor + # that does NOT belong to the charging hub's asset tree + from flexmeasures.data.models.generic_assets import GenericAssetType + + standalone_type = GenericAssetType(name="standalone group test type") + fresh_db.session.add(standalone_type) + fresh_db.session.flush() + standalone_asset = GenericAsset( + name="Standalone asset outside hub", + owner=charging_hub.owner, + generic_asset_type=standalone_type, + latitude=10, + longitude=100, + ) + fresh_db.session.add(standalone_asset) + fresh_db.session.flush() + outside_group_sensor = Sensor( + name="outside group sensor", + generic_asset=standalone_asset, + unit="MW", + event_resolution=pd.Timedelta(minutes=15), + ) + fresh_db.session.add(outside_group_sensor) + fresh_db.session.flush() + + price_sensor_id = add_market_prices_fresh_db["epex_da"].id + + message = message_for_trigger_schedule(resolution="PT30M") + message["flex-context"] = { + "consumption-price": {"sensor": price_sensor_id}, + "production-price": {"sensor": price_sensor_id}, + "site-power-capacity": "1 TW", + } + + CP_1_flex_model = message["flex-model"].copy() + CP_1_flex_model["state-of-charge"] = {"sensor": bi_soc_sensor.id} + CP_1_flex_model["sensor"] = sensor_1.id + CP_1_flex_model["group"] = {"sensor": outside_group_sensor.id} + + CP_2_flex_model = message["flex-model"].copy() + CP_2_flex_model["state-of-charge"] = {"sensor": uni_soc_sensor.id} + CP_2_flex_model["sensor"] = sensor_2.id + CP_2_flex_model["group"] = {"sensor": outside_group_sensor.id} + + group_flex_model = { + "sensor": outside_group_sensor.id, + "power-capacity": "3 MW", + } + + message["flex-model"] = [CP_1_flex_model, CP_2_flex_model, group_flex_model] + + assert len(app.queues["scheduling"]) == 0 + with app.test_client() as client: + trigger_response = client.post( + url_for("AssetAPI:trigger_schedule", id=charging_hub.id), + json=message, + ) + assert trigger_response.status_code == 422 + error_json = trigger_response.json + print(f"Error response: {error_json}") + errors = error_json.get("errors", error_json) + error_message = str(errors) + assert str(outside_group_sensor.id) in error_message + assert ( + "does not belong to asset" in error_message + or "does not belong to" in error_message + ) + + # No scheduling job should have been queued + assert len(app.queues["scheduling"]) == 0 diff --git a/flexmeasures/conftest.py b/flexmeasures/conftest.py index 8be3b72480..7cbaf607c9 100644 --- a/flexmeasures/conftest.py +++ b/flexmeasures/conftest.py @@ -2066,6 +2066,27 @@ def add_test_sensor_with_anomalous_beliefs( return {"anomaly-sensor": sensor} +def _patch_server_now_in_module(module, module_name: str, value, originals: dict): + """Patch server_now in a single module, remembering the original only the first + time we patch it, so repeated freeze calls still restore the true original.""" + try: + originals.setdefault(module_name, module.server_now) + setattr(module, "server_now", lambda: value) + except Exception: + # skip modules that cannot be inspected or modified + pass + + +def _patch_server_now_in_loaded_modules(value, originals: dict): + """Patch server_now in all currently loaded FlexMeasures modules.""" + for module in list(sys.modules.values()): # copy to avoid RuntimeError + if not isinstance(module, type(sys)): # skip placeholders + continue + name = getattr(module, "__name__", "") + if name.startswith("flexmeasures") and hasattr(module, "server_now"): + _patch_server_now_in_module(module, name, value, originals) + + @pytest.fixture def freeze_server_now(): """ @@ -2075,40 +2096,24 @@ def freeze_server_now(): def test_x(freeze_server_now): freeze_server_now(pd.Timestamp("2025-01-15T12:23:58+01")) """ - patched_modules = set() + original_server_nows: dict = {} # module name -> original server_now function + original_import = builtins.__import__ def _freeze(value: datetime | pd.Timestamp): if isinstance(value, pd.Timestamp): value = value.to_pydatetime() # Patch currently loaded FlexMeasures modules - for module in list(sys.modules.values()): # copy to avoid RuntimeError - try: - if not isinstance(module, type(sys)): # skip placeholders - continue - name = getattr(module, "__name__", "") - if not name.startswith("flexmeasures"): - continue - if hasattr(module, "server_now"): - setattr(module, "server_now", lambda: value) - patched_modules.add(module.__name__) - except Exception: - # skip modules that cannot be inspected or modified - pass + _patch_server_now_in_loaded_modules(value, original_server_nows) # Optionally, warn if new modules are imported later - original_import = builtins.__import__ - def import_hook(name, *args, **kwargs): mod = original_import(name, *args, **kwargs) - if hasattr(mod, "server_now") and mod not in patched_modules: + mod_name = getattr(mod, "__name__", name) + if hasattr(mod, "server_now") and mod_name not in original_server_nows: warnings.warn( f"Module {name} imported after server_now was frozen; patching it now." ) - try: - setattr(mod, "server_now", lambda: value) - patched_modules.add(name) - except Exception: - pass + _patch_server_now_in_module(mod, mod_name, value, original_server_nows) return mod builtins.__import__ = import_hook @@ -2117,5 +2122,15 @@ def import_hook(name, *args, **kwargs): yield _freeze - # cleanup: restore the original import function - builtins.__import__ = builtins.__import__ + # Cleanup: restore the original import function and unfreeze server_now in all + # patched modules. Without this, the frozen clock leaks into every test that runs + # afterwards in the same process (e.g. scheduling jobs then reuse the exact same + # belief_time, causing unique-key violations on saving beliefs). + builtins.__import__ = original_import + for module_name, original_server_now in original_server_nows.items(): + module = sys.modules.get(module_name) + if module is not None: + try: + setattr(module, "server_now", original_server_now) + except Exception: + pass diff --git a/flexmeasures/data/models/planning/devices.py b/flexmeasures/data/models/planning/devices.py index b7eb4a98c2..ebe86fdccb 100644 --- a/flexmeasures/data/models/planning/devices.py +++ b/flexmeasures/data/models/planning/devices.py @@ -1,8 +1,9 @@ """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 +stock-only entries carrying SoC parameters for a shared stock, +group entries carrying constraints on the aggregate power of a set of member devices, +and — in the future — 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. @@ -11,6 +12,7 @@ - which entries are schedulable devices, and their canonical solver indices, - which entries only carry SoC parameters for a shared stock (stock-only entries), +- which entries are group entries, and which devices belong to each group, - the inflexible devices from the flex-context, in canonical solver order, and - the stock groups (devices sharing a state-of-charge sensor). @@ -34,8 +36,7 @@ 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 + Extension point (not yet implemented): CONVERTER_PORT (a commodity port of a multi-commodity converter). """ @@ -43,6 +44,8 @@ class DeviceRole(Enum): DEVICE = "device" #: An entry carrying SoC parameters for a shared stock; not itself scheduled. STOCK_ONLY = "stock-only" + #: An entry constraining the aggregate power of a set of member devices; not itself scheduled. + GROUP = "group" #: An inflexible device from the flex-context (scheduled with a fixed profile). INFLEXIBLE = "inflexible" @@ -91,6 +94,11 @@ def consumption_sensor(self) -> Sensor | None: def production_sensor(self) -> Sensor | None: return _resolve_output_sensor(self.flex_model, "production") + @property + def group_key(self) -> tuple[str, int] | None: + """The key of the group this entry belongs to (via its "group" field), if any.""" + return resolve_group_key(self.flex_model) + def _resolve_output_sensor(flex_model: dict | None, output_field: str) -> Sensor | None: """Resolve the sensor of a nested consumption/production output reference.""" @@ -147,6 +155,115 @@ def _resolve_stock_key(state_of_charge: Any) -> int | None: return key +def _ref_id(value: Any) -> int | None: + """Return the id of a sensor/asset reference, which may be a model object or a raw id.""" + if value is None: + return None + return value.id if hasattr(value, "id") else value + + +def resolve_group_key(flex_model: dict | None) -> tuple[str, int] | None: + """Return a normalized ("sensor", id) or ("asset", id) key for the group a flex-model entry's "group" field references, or None if it has none.""" + if flex_model is None: + return None + group = flex_model.get("group") + if not group: + return None + if isinstance(group, dict): + group_sensor_id = _ref_id(group.get("sensor")) + group_asset_id = _ref_id(group.get("asset")) + else: + # Backwards compatibility: a raw sensor id/object. + group_sensor_id = _ref_id(group) + group_asset_id = None + if group_sensor_id is not None: + return ("sensor", group_sensor_id) + if group_asset_id is not None: + return ("asset", group_asset_id) + return None + + +def group_key_label(group_key: tuple[str, int]) -> str: + """Return a human-readable label for a group key, for use in error messages.""" + kind, gid = group_key + return f"{kind} {gid}" + + +def _match_own_group_key( + flex_model: dict, referenced_group_keys: set[tuple[str, int]] +) -> tuple[str, int] | None: + """Return the group key under which this entry is referenced as a group, if any. + + An entry is a group entry when its own "sensor" matches the sensor referenced by + another entry's "group" field, or when its own "asset" matches the asset referenced + by another entry's "group" field. + + :raises ValueError: When an asset-referenced group entry also carries a "sensor" field. + """ + own_sensor_id = _ref_id(flex_model.get("sensor")) + if own_sensor_id is not None and ("sensor", own_sensor_id) in referenced_group_keys: + return ("sensor", own_sensor_id) + own_asset_id = _ref_id(flex_model.get("asset")) + if own_asset_id is not None and ("asset", own_asset_id) in referenced_group_keys: + if own_sensor_id is not None: + raise ValueError( + f"Group entry for asset {own_asset_id} is referenced by asset," + " but also carries a 'sensor' field;" + " an asset-referenced group entry must not define its own power sensor." + ) + return ("asset", own_asset_id) + return None + + +def _collect_referenced_group_keys( + flex_model_list: list[dict], is_single_sensor_mode: bool +) -> set[tuple[str, int]]: + """Collect the group keys referenced by entries' "group" fields. + + :raises ValueError: When a single-sensor flex-model carries a "group" field + (groups are only supported in multi-device mode). + """ + if is_single_sensor_mode: + if any(isinstance(fm, dict) and fm.get("group") for fm in flex_model_list): + raise ValueError( + "The 'group' field is only supported in multi-device flex-models." + ) + return set() + referenced_group_keys: set[tuple[str, int]] = set() + for fm in flex_model_list: + group_key = resolve_group_key(fm) + if group_key is not None: + referenced_group_keys.add(group_key) + return referenced_group_keys + + +def _classify_group_entry(inventory: DeviceInventory, fm: dict) -> bool: + """Classify a flex-model entry as a group entry, if its own sensor/asset is referenced as a group. + + Group entries are not schedulable devices; + they carry constraints on the summed power of their member devices, + so they must be classified before their power sensor or output sensors could make them pass for devices. + + :returns: True if the entry was classified (and registered) as a group entry. + """ + if inventory.is_single_sensor_mode: + return False + own_group_key = _match_own_group_key(fm, inventory.referenced_group_keys) + if own_group_key is None: + return False + own_sensor = fm.get("sensor") + entry = FlexDevice( + role=DeviceRole.GROUP, + index=None, + flex_model=fm, + power_sensor=own_sensor if isinstance(own_sensor, Sensor) else None, + asset=(own_sensor.asset if isinstance(own_sensor, Sensor) else fm.get("asset")), + ) + inventory.entries.append(entry) + inventory.group_entries[own_group_key] = fm + return True + + #: 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") @@ -175,6 +292,10 @@ class DeviceInventory: 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) + #: Group entries (raw flex-model dicts) per group key: ("sensor", id) or ("asset", id). + group_entries: dict[tuple[str, int], dict] = field(default_factory=dict) + #: The group keys referenced by entries' "group" fields (used to detect dangling references). + referenced_group_keys: set[tuple[str, int]] = field(default_factory=set) is_single_sensor_mode: bool = False @classmethod @@ -199,6 +320,12 @@ def from_flex_config( inventory = cls(is_single_sensor_mode=is_single_sensor_mode) + # Collect the group keys referenced by entries' "group" fields; + # the entries whose own sensor/asset matches a referenced key are classified as group entries below. + inventory.referenced_group_keys = _collect_referenced_group_keys( + flex_model_list, 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) @@ -214,6 +341,11 @@ def register_stock_params(stock_key: int, fm: dict) -> None: inventory.stock_entries[stock_key] = fm for fm in flex_model_list: + # Group entry (multi-device mode only): this entry's own sensor/asset is + # the aggregate sensor/asset referenced by another entry's "group" field. + if _classify_group_entry(inventory, fm): + continue + if is_single_sensor_mode: power_sensor = sensor else: @@ -344,6 +476,50 @@ 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 group_to_devices(self) -> dict[tuple[str, int], list[int]]: + """Map each group key to the indices of the (leaf) member devices of that group. + + Membership is resolved transitively: + a group entry may itself belong to another group (via its own "group" field), + in which case its member devices count as members of the outer group, too. + + :raises ValueError: When group entries reference each other cyclically. + """ + resolved: dict[tuple[str, int], list[int]] = {} + + def resolve_leaf_devices( + group_key: tuple[str, int], path: tuple[tuple[str, int], ...] = () + ) -> list[int]: + if group_key in path: + raise ValueError( + f"Cyclic 'group' reference detected involving group " + f"{group_key_label(group_key)}." + ) + if group_key in resolved: + return resolved[group_key] + leaves: list[int] = [] + seen: set[int] = set() + for device in self.devices: + if device.group_key == group_key and device.index not in seen: + leaves.append(device.index) + seen.add(device.index) + # Also resolve members that are themselves groups pointing at this group. + for other_key, other_entry in self.group_entries.items(): + if other_key == group_key: + continue + if resolve_group_key(other_entry) == group_key: + for leaf in resolve_leaf_devices(other_key, path + (group_key,)): + if leaf not in seen: + leaves.append(leaf) + seen.add(leaf) + resolved[group_key] = leaves + return leaves + + for group_key in self.group_entries: + resolve_leaf_devices(group_key) + return resolved + @cached_property def commodity_to_devices(self) -> dict[str, list[int]]: """Map each commodity to its device indices, in canonical solver order.""" diff --git a/flexmeasures/data/models/planning/storage.py b/flexmeasures/data/models/planning/storage.py index a6686748b8..53f2b501dc 100644 --- a/flexmeasures/data/models/planning/storage.py +++ b/flexmeasures/data/models/planning/storage.py @@ -18,7 +18,7 @@ SchedulerOutputType, StockCommitment, ) -from flexmeasures.data.models.planning.devices import DeviceInventory +from flexmeasures.data.models.planning.devices import DeviceInventory, group_key_label from flexmeasures.data.models.planning.linear_optimization import device_scheduler from flexmeasures.data.models.planning.utils import ( add_tiny_price_slope, @@ -148,6 +148,77 @@ def _prepare(self, skip_validation: bool = False) -> tuple: # noqa: C901 # The stock groups' device indices align with the device models self.stock_groups = inventory.stock_groups + # Group entries (intermediate power constraints on groups of devices, e.g. a + # sub-EMS) come classified from the inventory, together with the resolved + # (leaf) group membership. Accessing `group_to_devices` also detects cyclic + # group references (raising a ValueError). + self._group_models = inventory.group_entries + self._group_to_devices = inventory.group_to_devices + + if not skip_validation: + dangling = inventory.referenced_group_keys - set( + inventory.group_entries.keys() + ) + if dangling: + raise ValueError( + "The 'group' field references " + f"{sorted(group_key_label(g) for g in dangling)}, but no " + "flex-model entry was found for it. Add a flex-model entry for " + "the group, carrying the group's power-capacity, " + "consumption-capacity and/or production-capacity." + ) + device_only_fields = ( + "soc_at_start", + "soc_min", + "soc_max", + "soc_minima", + "soc_maxima", + "soc_targets", + "soc_gain", + "soc_usage", + "state_of_charge", + "storage_efficiency", + "charging_efficiency", + "discharging_efficiency", + "roundtrip_efficiency", + ) + for gkey, group_entry in inventory.group_entries.items(): + offending = [ + field for field in device_only_fields if group_entry.get(field) + ] + if offending: + raise ValueError( + f"Group entry for {group_key_label(gkey)} carries " + f"device-only field(s) {offending}, which is not allowed: " + "group entries only describe constraints on the group's " + "aggregate power." + ) + if not any( + group_entry.get(field) is not None + for field in ( + "power_capacity_in_mw", + "consumption_capacity", + "production_capacity", + ) + ): + raise ValueError( + f"Group entry for {group_key_label(gkey)} defines none of " + "'power-capacity', 'consumption-capacity' or " + "'production-capacity'; such an entry has no effect." + ) + for gkey, leaves in inventory.group_to_devices.items(): + if not leaves: + raise ValueError( + f"The 'group' field references {group_key_label(gkey)}, " + "but no device in the flex-model belongs to that group." + ) + commodities = {inventory.devices[d].commodity for d in leaves} + if len(commodities) > 1: + raise ValueError( + f"All member devices of group {group_key_label(gkey)} must " + f"share the same commodity; found {sorted(commodities)}." + ) + # List the asset(s) and sensor(s) being scheduled sensors: list[Sensor | None] = inventory.power_sensors assets: list[Asset | None] = inventory.assets @@ -606,6 +677,153 @@ def device_list_series( ems_constraints.append(commodity_ems_constraints) ems_constraint_groups.append(list(devices)) + # Intermediate power constraints on groups of devices (sub-EMS's), declared via + # the `group` field on flex-model entries. See MetaStorageScheduler._prepare + # docstring notes on groups above. + default_group_breach_price = ur.Quantity( + f"10000 {self.flex_context['shared_currency_unit']}/kW" + ) + for group_key, leaf_members in self._group_to_devices.items(): + if not leaf_members: + continue + group_entry = self._group_models[group_key] + group_label = f"{group_key[0]}:{group_key[1]}" + group_commodity = inventory.devices[leaf_members[0]].commodity + group_devices = device_list_series(leaf_members, index) + + group_power_capacity = get_continuous_series_sensor_or_quantity( + variable_quantity=group_entry.get("power_capacity_in_mw"), + unit="MW", + query_window=(start, end), + resolution=resolution, + beliefs_before=belief_time, + resolve_overlaps="min", + ) + group_consumption_capacity = get_continuous_series_sensor_or_quantity( + variable_quantity=group_entry.get("consumption_capacity"), + unit="MW", + query_window=(start, end), + resolution=resolution, + beliefs_before=belief_time, + max_value=group_power_capacity, + resolve_overlaps="min", + ) + group_production_capacity = -1 * get_continuous_series_sensor_or_quantity( + variable_quantity=group_entry.get("production_capacity"), + unit="MW", + query_window=(start, end), + resolution=resolution, + beliefs_before=belief_time, + max_value=group_power_capacity, + resolve_overlaps="min", + ) + + # Hard bound: the group's summed power may never exceed its (physical) + # power capacity, in either direction. + group_ems_constraints = initialize_df( + StorageScheduler.COLUMNS, start, end, resolution + ) + group_ems_constraints["derivative max"] = group_power_capacity + group_ems_constraints["derivative min"] = -group_power_capacity + ems_constraints.append(group_ems_constraints) + ems_constraint_groups.append(list(leaf_members)) + + currency_unit = self.flex_context["shared_currency_unit"] + + # Soft bound: directional (consumption/production) capacities on the group + # are enforced via breach commitments with default breach prices (no + # user-set prices are supported for groups), mirroring the site-level + # ems_consumption_breach_price / ems_production_breach_price pattern. + if group_entry.get("consumption_capacity") is not None: + any_group_consumption_breach_price = ( + get_continuous_series_sensor_or_quantity( + variable_quantity=default_group_breach_price, + unit=currency_unit + "/MW", + query_window=(start, end), + resolution=resolution, + beliefs_before=belief_time, + fill_sides=True, + ) + ) + all_group_consumption_breach_price = ( + get_continuous_series_sensor_or_quantity( + variable_quantity=default_group_breach_price, + unit=currency_unit + "/MW*h", + query_window=(start, end), + resolution=resolution, + beliefs_before=belief_time, + fill_sides=True, + ) + ) + commitments.append( + FlowCommitment( + name=f"group {group_label} any consumption breach", + quantity=group_consumption_capacity, + upwards_deviation_price=any_group_consumption_breach_price, + _type="any", + index=index, + device=group_devices, + device_group=f"group:{group_label}", + commodity=group_commodity, + ) + ) + commitments.append( + FlowCommitment( + name=f"group {group_label} all consumption breaches", + quantity=group_consumption_capacity, + upwards_deviation_price=all_group_consumption_breach_price, + index=index, + device=group_devices, + device_group=f"group:{group_label}", + commodity=group_commodity, + ) + ) + + if group_entry.get("production_capacity") is not None: + any_group_production_breach_price = ( + get_continuous_series_sensor_or_quantity( + variable_quantity=default_group_breach_price, + unit=currency_unit + "/MW", + query_window=(start, end), + resolution=resolution, + beliefs_before=belief_time, + fill_sides=True, + ) + ) + all_group_production_breach_price = ( + get_continuous_series_sensor_or_quantity( + variable_quantity=default_group_breach_price, + unit=currency_unit + "/MW*h", + query_window=(start, end), + resolution=resolution, + beliefs_before=belief_time, + fill_sides=True, + ) + ) + commitments.append( + FlowCommitment( + name=f"group {group_label} any production breach", + quantity=group_production_capacity, + downwards_deviation_price=-any_group_production_breach_price, + _type="any", + index=index, + device=group_devices, + device_group=f"group:{group_label}", + commodity=group_commodity, + ) + ) + commitments.append( + FlowCommitment( + name=f"group {group_label} all production breaches", + quantity=group_production_capacity, + downwards_deviation_price=-all_group_production_breach_price, + index=index, + device=group_devices, + device_group=f"group:{group_label}", + commodity=group_commodity, + ) + ) + # Keep one price frame for later preference logic. # The existing "prefer charging sooner" code uses `up_deviation_prices`. # Prefer electricity prices if available, otherwise use the first commodity price. @@ -2329,55 +2547,86 @@ def _build_consumption_production_schedules( """ schedules: dict = {} for d, flex_model_d in enumerate(flex_model): - consumption_field = flex_model_d.get("consumption") - production_field = flex_model_d.get("production") - consumption_sensor = ( - consumption_field["sensor"] - if isinstance(consumption_field, dict) and "sensor" in consumption_field - else None - ) - production_sensor = ( - production_field["sensor"] - if isinstance(production_field, dict) and "sensor" in production_field - else None - ) - if consumption_sensor is None and production_sensor is None: - continue power_series = ems_schedule[d] # in MW; consumption is positive - if consumption_sensor is not None and production_sensor is None: - # Full power profile on the consumption sensor (consumption positive, production negative). - schedules[consumption_sensor] = convert_units( - power_series, - "MW", - consumption_sensor.unit, - event_resolution=consumption_sensor.event_resolution, - ) - elif production_sensor is not None and consumption_sensor is None: - # Full power profile on the production sensor in native scheduler convention. - # make_schedule inverts the sign via consumption_is_positive=False on the sensor. - schedules[production_sensor] = convert_units( - power_series, - "MW", - production_sensor.unit, - event_resolution=production_sensor.event_resolution, - ) - else: - # Both sensors defined: clip to non-negative (consumption) and non-positive (production) parts. - # make_schedule inverts the sign for the production sensor via consumption_is_positive=False. - schedules[consumption_sensor] = convert_units( - power_series.clip(lower=0), - "MW", - consumption_sensor.unit, - event_resolution=consumption_sensor.event_resolution, - ) - schedules[production_sensor] = convert_units( - power_series.clip(upper=0), - "MW", - production_sensor.unit, - event_resolution=production_sensor.event_resolution, - ) + StorageScheduler._split_schedule_over_output_sensors( + flex_model_d, power_series, schedules + ) return schedules + @staticmethod + def _split_schedule_over_output_sensors( + flex_model_d: dict, + power_series: pd.Series, + schedules: dict, + ) -> None: + """Save a power schedule (in MW, consumption positive) to the output sensor(s) + (``consumption``/``production``) defined on a single flex-model entry, in-place + on ``schedules`` (mapping output sensor -> power schedule). + + Follows the same conventions as :func:`_build_consumption_production_schedules`: + + - **Only** ``consumption`` **sensor defined**: the full power schedule is written to that + sensor using the scheduler's native sign convention (consumption positive, production + negative). ``make_schedule`` applies no further sign change because the sensor already + has ``consumption_is_positive=True``. + - **Only** ``production`` **sensor defined**: the full power schedule is written to that + sensor in the scheduler's native sign convention (consumption positive, production + negative). ``make_schedule`` inverts the sign based on the sensor's + ``consumption_is_positive=False`` attribute so that production is stored as positive values. + - **Both** ``consumption`` **and** ``production`` **sensors defined**: only the non-negative + part of the schedule (charging / consuming) is written to the consumption sensor, and only + the non-positive part (discharging / producing, still as negative values) is written to + the production sensor. ``make_schedule`` inverts the sign for the production sensor. + + Unit conversion from MW to each sensor's unit is applied. + """ + consumption_field = flex_model_d.get("consumption") + production_field = flex_model_d.get("production") + consumption_sensor = ( + consumption_field["sensor"] + if isinstance(consumption_field, dict) and "sensor" in consumption_field + else None + ) + production_sensor = ( + production_field["sensor"] + if isinstance(production_field, dict) and "sensor" in production_field + else None + ) + if consumption_sensor is None and production_sensor is None: + return + if consumption_sensor is not None and production_sensor is None: + # Full power profile on the consumption sensor (consumption positive, production negative). + schedules[consumption_sensor] = convert_units( + power_series, + "MW", + consumption_sensor.unit, + event_resolution=consumption_sensor.event_resolution, + ) + elif production_sensor is not None and consumption_sensor is None: + # Full power profile on the production sensor in native scheduler convention. + # make_schedule inverts the sign via consumption_is_positive=False on the sensor. + schedules[production_sensor] = convert_units( + power_series, + "MW", + production_sensor.unit, + event_resolution=production_sensor.event_resolution, + ) + else: + # Both sensors defined: clip to non-negative (consumption) and non-positive (production) parts. + # make_schedule inverts the sign for the production sensor via consumption_is_positive=False. + schedules[consumption_sensor] = convert_units( + power_series.clip(lower=0), + "MW", + consumption_sensor.unit, + event_resolution=consumption_sensor.event_resolution, + ) + schedules[production_sensor] = convert_units( + power_series.clip(upper=0), + "MW", + production_sensor.unit, + event_resolution=production_sensor.event_resolution, + ) + def _reconstruct_commodity_to_devices(self) -> dict[str, list[int]]: """Return the mapping of commodity -> device indices, as enumerated by the device inventory. @@ -2505,6 +2754,70 @@ def _compute_commodity_aggregate_schedules( commodity_aggregate.clip(upper=0) ) + @staticmethod + def _merge_group_output_schedules( + consumption_production_schedule: dict, group_output_schedules: dict + ) -> None: + """Merge group output schedules (already unit-converted) into + ``consumption_production_schedule`` in-place, avoiding overwrite of a device's + own output sensor with a group's.""" + for out_sensor, out_schedule in group_output_schedules.items(): + if out_sensor in consumption_production_schedule: + raise ValueError( + f"Sensor {out_sensor.id} is used as an output sensor both by a " + "device and by a group; a sensor cannot be both." + ) + consumption_production_schedule[out_sensor] = out_schedule + + def _add_group_schedules( + self, storage_schedule: dict, ems_schedule: list[pd.Series] + ) -> dict: + """Save each group's aggregate power schedule. + + - Sensor-referenced groups: the aggregate is saved under the group's own power + sensor (as before), added in-place to ``storage_schedule`` (still in MW, + native scheduler convention; unit conversion happens later, alongside other + device sensors). + - Asset-referenced groups: the group entry defines no power sensor of its own; + instead, the aggregate is saved via the group entry's own ``consumption``/ + ``production`` output sensors, if defined. + - Either kind of group may additionally define ``consumption``/``production`` + output sensors, in which case the aggregate is also split/saved there. + + The ``consumption``/``production`` output schedules are returned separately + (already unit-converted, like ``_build_consumption_production_schedules``) + rather than added to ``storage_schedule``, to avoid double unit conversion. + + :returns: Dict mapping each group output sensor to its power schedule. + """ + group_models = getattr(self, "_group_models", None) or {} + group_to_devices = getattr(self, "_group_to_devices", None) or {} + group_output_schedules: dict = {} + for group_key, leaf_members in group_to_devices.items(): + group_entry = group_models.get(group_key) + if group_entry is None or not leaf_members: + continue + group_aggregate = pd.concat( + [ems_schedule[d] for d in leaf_members], + axis=1, + ).sum(axis=1) + + group_sensor = group_entry.get("sensor") + if group_sensor is not None: + if group_sensor in storage_schedule: + raise ValueError( + f"Sensor {group_sensor.id} is used both as a device sensor " + "and as a group sensor; a sensor cannot be both." + ) + storage_schedule[group_sensor] = group_aggregate + + # For both sensor-ref and asset-ref groups, also honor any consumption/ + # production output sensors defined on the group entry itself. + self._split_schedule_over_output_sensors( + group_entry, group_aggregate, group_output_schedules + ) + return group_output_schedules + def compute(self, skip_validation: bool = False) -> SchedulerOutputType: """Schedule a battery or Charge Point based directly on the latest beliefs regarding market prices within the specified time window. For the resulting consumption schedule, consumption is defined as positive values. @@ -2555,6 +2868,10 @@ def compute(self, skip_validation: bool = False) -> SchedulerOutputType: elif sensor is not None and sensor in storage_schedule: storage_schedule[sensor] += ems_schedule[d] + group_output_schedules = self._add_group_schedules( + storage_schedule, ems_schedule + ) + # 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) @@ -2605,6 +2922,9 @@ def compute(self, skip_validation: bool = False) -> SchedulerOutputType: consumption_production_schedule = self._build_consumption_production_schedules( flex_model_for_soc, ems_schedule ) + self._merge_group_output_schedules( + consumption_production_schedule, group_output_schedules + ) # Resample each device schedule to the resolution of the device's power sensor if self.resolution is None: @@ -2682,9 +3002,11 @@ def compute(self, skip_validation: bool = False) -> SchedulerOutputType: for sensor, soc in soc_schedule.items() ] # Determine which sensors are consumption vs. production output sensors + group_models_for_output = getattr(self, "_group_models", None) or {} consumption_output_sensors = { flex_model_d["consumption"]["sensor"] - for flex_model_d in flex_model_for_soc + for flex_model_d in list(flex_model_for_soc) + + list(group_models_for_output.values()) if isinstance(flex_model_d.get("consumption"), dict) and "sensor" in flex_model_d["consumption"] } diff --git a/flexmeasures/data/models/planning/tests/test_device_inventory.py b/flexmeasures/data/models/planning/tests/test_device_inventory.py index 9fb0d2b996..e49984ba90 100644 --- a/flexmeasures/data/models/planning/tests/test_device_inventory.py +++ b/flexmeasures/data/models/planning/tests/test_device_inventory.py @@ -248,3 +248,85 @@ def test_conflicting_stock_params_raise(): [stock_only_entry, device_without_params] ) assert inventory.stock_params(soc.id) is stock_only_entry + + +def test_group_entry_classification(): + """An entry whose own sensor is referenced by another entry's "group" field is + classified as a group entry (not as a device), and the group's member devices + resolve to their canonical device indices.""" + power_a = make_sensor(1) + power_b = make_sensor(2) + group_sensor = make_sensor(10) + + group_entry = {"sensor": group_sensor, "power_capacity_in_mw": 0.001} + inventory = DeviceInventory.from_flex_config( + [ + {"sensor": power_a, "group": {"sensor": group_sensor}}, + {"sensor": power_b, "group": {"sensor": group_sensor}}, + group_entry, + ] + ) + + assert [entry.role for entry in inventory.entries] == [ + DeviceRole.DEVICE, + DeviceRole.DEVICE, + DeviceRole.GROUP, + ] + # The group entry is not a schedulable device, so it gets no device index. + assert inventory.num_flexible == 2 + assert inventory.power_sensors == [power_a, power_b] + assert inventory.group_entries == {("sensor", group_sensor.id): group_entry} + assert inventory.group_to_devices == {("sensor", group_sensor.id): [0, 1]} + + +def test_nested_group_resolves_to_leaf_devices(): + """Group membership is resolved transitively: a group entry belonging to another + group contributes its member devices to the outer group.""" + power_a = make_sensor(1) + inner_group_sensor = make_sensor(10) + outer_group_sensor = make_sensor(20) + + inventory = DeviceInventory.from_flex_config( + [ + {"sensor": power_a, "group": {"sensor": inner_group_sensor}}, + { + "sensor": inner_group_sensor, + "power_capacity_in_mw": 0.001, + "group": {"sensor": outer_group_sensor}, + }, + {"sensor": outer_group_sensor, "power_capacity_in_mw": 0.002}, + ] + ) + + assert inventory.num_flexible == 1 + assert inventory.group_to_devices == { + ("sensor", inner_group_sensor.id): [0], + ("sensor", outer_group_sensor.id): [0], + } + + +def test_cyclic_group_reference_raises(): + group_a_sensor = make_sensor(10) + group_b_sensor = make_sensor(20) + + inventory = DeviceInventory.from_flex_config( + [ + {"sensor": make_sensor(1), "group": {"sensor": group_a_sensor}}, + {"sensor": group_a_sensor, "group": {"sensor": group_b_sensor}}, + {"sensor": group_b_sensor, "group": {"sensor": group_a_sensor}}, + ] + ) + + with pytest.raises(ValueError, match="Cyclic"): + inventory.group_to_devices + + +def test_group_field_rejected_in_single_sensor_mode(): + power = make_sensor(1) + group_sensor = make_sensor(10) + + with pytest.raises(ValueError, match="multi-device"): + DeviceInventory.from_flex_config( + {"soc_at_start": 0.0, "group": {"sensor": group_sensor}}, + sensor=power, + ) diff --git a/flexmeasures/data/models/planning/tests/test_group_constraints.py b/flexmeasures/data/models/planning/tests/test_group_constraints.py new file mode 100644 index 0000000000..4a08ee0342 --- /dev/null +++ b/flexmeasures/data/models/planning/tests/test_group_constraints.py @@ -0,0 +1,789 @@ +import uuid +from datetime import timedelta + +import pytest +import pandas as pd + +from flexmeasures.data.models.generic_assets import GenericAsset, GenericAssetType +from flexmeasures.data.models.time_series import Sensor +from flexmeasures.data.models.planning.storage import StorageScheduler +from flexmeasures.utils.unit_utils import ur + + +def _unique_name(prefix: str) -> str: + return f"{prefix} {uuid.uuid4().hex[:8]}" + + +def make_sensors(db, building, n=2, unit="MW"): + sensors = [] + for i in range(n): + s = Sensor( + name=_unique_name(f"group test power sensor {i}"), + generic_asset=building, + event_resolution=timedelta(hours=1), + unit=unit, + ) + db.session.add(s) + sensors.append(s) + db.session.commit() + return sensors + + +def make_group_sensor(db, building, unit="MW"): + s = Sensor( + name=_unique_name("group aggregate sensor"), + generic_asset=building, + event_resolution=timedelta(hours=1), + unit=unit, + ) + db.session.add(s) + db.session.commit() + return s + + +def make_sub_asset(db, building): + """Create a child asset (e.g. an inverter/sub-EMS) under `building`.""" + asset_type = GenericAssetType(name=_unique_name("group test asset type")) + db.session.add(asset_type) + asset = GenericAsset( + name=_unique_name("group test sub-asset"), + generic_asset_type=asset_type, + parent_asset=building, + owner=building.owner, + ) + db.session.add(asset) + db.session.commit() + return asset + + +def make_output_sensor(db, asset, unit="MW"): + s = Sensor( + name=_unique_name("output sensor"), + generic_asset=asset, + event_resolution=timedelta(hours=1), + unit=unit, + ) + db.session.add(s) + db.session.commit() + return s + + +def base_flex_context(): + return { + "consumption_price": ur.Quantity("100 EUR/MWh"), + "production_price": ur.Quantity("100 EUR/MWh"), + "shared_currency_unit": "EUR", + } + + +def run_scheduler(building, flex_model, flex_context, **kwargs): + start = pd.Timestamp("2023-01-01T00:00:00", tz="Europe/Amsterdam") + end = start + timedelta(hours=4) + resolution = timedelta(hours=1) + scheduler = StorageScheduler( + asset_or_sensor=building, + start=start, + end=end, + resolution=resolution, + flex_model=flex_model, + flex_context=flex_context, + **kwargs, + ) + scheduler.config_deserialized = True + return scheduler + + +def test_group_hard_power_capacity_caps_aggregate(db, building): + """Inverter-like case (#2092): battery + PV producer under a group with a tight + power-capacity should cap the *sum* of their schedules, even though each device + individually would want to go to its own max.""" + battery, pv = make_sensors(db, building, n=2) + group_sensor = make_group_sensor(db, building) + + flex_model = [ + { + "sensor": battery, + "soc_at_start": 1.0, + "soc_min": 0.0, + "soc_max": 2.0, + "power_capacity_in_mw": ur.Quantity("2 MW"), + "consumption_capacity": ur.Quantity("2 MW"), + "production_capacity": ur.Quantity("2 MW"), + "group": {"sensor": group_sensor}, + }, + { + "sensor": pv, + "power_capacity_in_mw": ur.Quantity("2 MW"), + "consumption_capacity": ur.Quantity("0 MW"), + "production_capacity": ur.Quantity("2 MW"), + "group": {"sensor": group_sensor}, + }, + { + "sensor": group_sensor, + "power_capacity_in_mw": ur.Quantity("2.5 MW"), + }, + ] + scheduler = run_scheduler( + building, flex_model, base_flex_context(), return_multiple=True + ) + results = scheduler.compute() + + schedules = { + r["sensor"]: r["data"] for r in results if r.get("name") == "storage_schedule" + } + assert battery in schedules + assert pv in schedules + assert group_sensor in schedules + + aggregate = schedules[group_sensor] + assert (aggregate.abs() <= 2.5 + 1e-6).all() + # the group's aggregate should equal the sum of its members + assert ((schedules[battery] + schedules[pv] - aggregate).abs() < 1e-6).all() + + +def test_group_absent_allows_larger_aggregate(db, building): + """Control run: without the group constraint, the same devices are not capped.""" + battery, pv = make_sensors(db, building, n=2) + + flex_model = [ + { + "sensor": battery, + "soc_at_start": 1.0, + "soc_min": 0.0, + "soc_max": 2.0, + "power_capacity_in_mw": ur.Quantity("2 MW"), + "consumption_capacity": ur.Quantity("2 MW"), + "production_capacity": ur.Quantity("2 MW"), + }, + { + "sensor": pv, + "power_capacity_in_mw": ur.Quantity("2 MW"), + "consumption_capacity": ur.Quantity("0 MW"), + "production_capacity": ur.Quantity("2 MW"), + }, + ] + scheduler = run_scheduler( + building, flex_model, base_flex_context(), return_multiple=True + ) + results = scheduler.compute() + schedules = { + r["sensor"]: r["data"] for r in results if r.get("name") == "storage_schedule" + } + combined_abs_max = (schedules[battery] + schedules[pv]).abs().max() + assert combined_abs_max > 2.5 + + +def test_group_soft_directional_capacity(db, building): + """A tight consumption-capacity on the group is soft (breach commitments), but the + hard power-capacity remains a true cap.""" + battery, pv = make_sensors(db, building, n=2) + group_sensor = make_group_sensor(db, building) + + flex_model = [ + { + "sensor": battery, + "soc_at_start": 1.0, + "soc_min": 0.0, + "soc_max": 2.0, + "power_capacity_in_mw": ur.Quantity("2 MW"), + "consumption_capacity": ur.Quantity("2 MW"), + "production_capacity": ur.Quantity("2 MW"), + "group": {"sensor": group_sensor}, + }, + { + "sensor": pv, + "power_capacity_in_mw": ur.Quantity("2 MW"), + "consumption_capacity": ur.Quantity("0 MW"), + "production_capacity": ur.Quantity("2 MW"), + "group": {"sensor": group_sensor}, + }, + { + "sensor": group_sensor, + "power_capacity_in_mw": ur.Quantity("2.5 MW"), + "consumption_capacity": ur.Quantity("1 MW"), + }, + ] + scheduler = run_scheduler( + building, flex_model, base_flex_context(), return_multiple=True + ) + results = scheduler.compute() + + schedules = { + r["sensor"]: r["data"] for r in results if r.get("name") == "storage_schedule" + } + aggregate = schedules[group_sensor] + # With default (very high) breach prices and moderate test prices, breaching + # should never be worthwhile: the soft directional bound behaves like a cap. + assert (aggregate <= 1 + 1e-6).all() + # Hard power-capacity remains a hard cap regardless. + assert (aggregate.abs() <= 2.5 + 1e-6).all() + + commitment_costs = next( + r["data"] for r in results if r.get("name") == "commitment_costs" + ) + assert any( + "group" in name and "consumption breach" in name for name in commitment_costs + ) + + +def test_group_dangling_reference_raises(db, building): + battery, pv = make_sensors(db, building, n=2) + other_sensor = make_group_sensor(db, building) + + flex_model = [ + { + "sensor": battery, + "soc_at_start": 1.0, + "soc_min": 0.0, + "soc_max": 2.0, + "power_capacity_in_mw": ur.Quantity("2 MW"), + "group": {"sensor": other_sensor}, + }, + { + "sensor": pv, + "power_capacity_in_mw": ur.Quantity("2 MW"), + }, + ] + scheduler = run_scheduler(building, flex_model, base_flex_context()) + with pytest.raises(ValueError, match="group"): + scheduler.compute() + + +def test_group_entry_with_soc_at_start_raises(db, building): + battery, pv = make_sensors(db, building, n=2) + group_sensor = make_group_sensor(db, building) + + flex_model = [ + { + "sensor": battery, + "soc_at_start": 1.0, + "soc_min": 0.0, + "soc_max": 2.0, + "power_capacity_in_mw": ur.Quantity("2 MW"), + "group": {"sensor": group_sensor}, + }, + { + "sensor": pv, + "power_capacity_in_mw": ur.Quantity("2 MW"), + "group": {"sensor": group_sensor}, + }, + { + "sensor": group_sensor, + "power_capacity_in_mw": ur.Quantity("2.5 MW"), + "soc_at_start": 0.5, + }, + ] + scheduler = run_scheduler(building, flex_model, base_flex_context()) + with pytest.raises(ValueError, match="device-only"): + scheduler.compute() + + +def test_group_mixed_commodities_raises(db, building): + battery, pv = make_sensors(db, building, n=2) + group_sensor = make_group_sensor(db, building) + + flex_model = [ + { + "sensor": battery, + "soc_at_start": 1.0, + "soc_min": 0.0, + "soc_max": 2.0, + "power_capacity_in_mw": ur.Quantity("2 MW"), + "commodity": "electricity", + "group": {"sensor": group_sensor}, + }, + { + "sensor": pv, + "power_capacity_in_mw": ur.Quantity("2 MW"), + "commodity": "gas", + "group": {"sensor": group_sensor}, + }, + { + "sensor": group_sensor, + "power_capacity_in_mw": ur.Quantity("2.5 MW"), + }, + ] + scheduler = run_scheduler(building, flex_model, base_flex_context()) + with pytest.raises(ValueError, match="commodity"): + scheduler.compute() + + +def test_group_entry_without_capacity_fields_raises(db, building): + battery, pv = make_sensors(db, building, n=2) + group_sensor = make_group_sensor(db, building) + + flex_model = [ + { + "sensor": battery, + "soc_at_start": 1.0, + "soc_min": 0.0, + "soc_max": 2.0, + "power_capacity_in_mw": ur.Quantity("2 MW"), + "group": {"sensor": group_sensor}, + }, + { + "sensor": pv, + "power_capacity_in_mw": ur.Quantity("2 MW"), + "group": {"sensor": group_sensor}, + }, + { + "sensor": group_sensor, + }, + ] + scheduler = run_scheduler(building, flex_model, base_flex_context()) + with pytest.raises(ValueError, match="none of"): + scheduler.compute() + + +def test_group_in_single_sensor_mode_raises(db, building): + battery = make_sensors(db, building, n=1)[0] + group_sensor = make_group_sensor(db, building) + + scheduler = StorageScheduler( + asset_or_sensor=battery, + start=pd.Timestamp("2023-01-01T00:00:00", tz="Europe/Amsterdam"), + end=pd.Timestamp("2023-01-01T04:00:00", tz="Europe/Amsterdam"), + resolution=timedelta(hours=1), + flex_model={ + "soc_at_start": 1.0, + "soc_min": 0.0, + "soc_max": 2.0, + "power_capacity_in_mw": ur.Quantity("2 MW"), + "group": {"sensor": group_sensor}, + }, + flex_context=base_flex_context(), + ) + scheduler.config_deserialized = True + with pytest.raises(ValueError, match="multi-device"): + scheduler.compute() + + +def test_nested_group_leaf_resolution(db, building): + """Two-level nesting: device -> inner group -> outer group. Verify the outer + group's aggregate power equals the sum of the leaf device's schedule (the only + leaf here), i.e. leaf resolution skips the intermediate group.""" + battery, pv = make_sensors(db, building, n=2) + inner_group = make_group_sensor(db, building) + outer_group = make_group_sensor(db, building) + + flex_model = [ + { + "sensor": battery, + "soc_at_start": 1.0, + "soc_min": 0.0, + "soc_max": 2.0, + "power_capacity_in_mw": ur.Quantity("2 MW"), + "group": {"sensor": inner_group}, + }, + { + "sensor": pv, + "power_capacity_in_mw": ur.Quantity("2 MW"), + "consumption_capacity": ur.Quantity("0 MW"), + "group": {"sensor": inner_group}, + }, + { + "sensor": inner_group, + "power_capacity_in_mw": ur.Quantity("3 MW"), + "group": {"sensor": outer_group}, + }, + { + "sensor": outer_group, + "power_capacity_in_mw": ur.Quantity("3.5 MW"), + }, + ] + scheduler = run_scheduler( + building, flex_model, base_flex_context(), return_multiple=True + ) + results = scheduler.compute() + schedules = { + r["sensor"]: r["data"] for r in results if r.get("name") == "storage_schedule" + } + assert inner_group in schedules + assert outer_group in schedules + expected = schedules[battery] + schedules[pv] + assert ((schedules[inner_group] - expected).abs() < 1e-6).all() + assert ((schedules[outer_group] - expected).abs() < 1e-6).all() + + +def test_group_cycle_raises(db, building): + battery = make_sensors(db, building, n=1)[0] + group_a = make_group_sensor(db, building) + group_b = make_group_sensor(db, building) + + flex_model = [ + { + "sensor": battery, + "soc_at_start": 1.0, + "soc_min": 0.0, + "soc_max": 2.0, + "power_capacity_in_mw": ur.Quantity("2 MW"), + "group": {"sensor": group_a}, + }, + { + "sensor": group_a, + "power_capacity_in_mw": ur.Quantity("2 MW"), + "group": {"sensor": group_b}, + }, + { + "sensor": group_b, + "power_capacity_in_mw": ur.Quantity("2 MW"), + "group": {"sensor": group_a}, + }, + ] + scheduler = run_scheduler(building, flex_model, base_flex_context()) + with pytest.raises(ValueError, match="Cyclic"): + scheduler.compute() + + +def test_group_asset_ref_hard_cap(db, building): + """An asset-referenced group entry (no power sensor of its own) still caps the + aggregate power of its members, and saves the aggregate on its consumption output + sensor (consumption-only case: full profile, consumption positive).""" + battery, pv = make_sensors(db, building, n=2) + inverter = make_sub_asset(db, building) + consumption_sensor = make_output_sensor(db, inverter) + + flex_model = [ + { + "sensor": battery, + "soc_at_start": 1.0, + "soc_min": 0.0, + "soc_max": 2.0, + "power_capacity_in_mw": ur.Quantity("2 MW"), + "consumption_capacity": ur.Quantity("2 MW"), + "production_capacity": ur.Quantity("2 MW"), + "group": {"asset": inverter}, + }, + { + "sensor": pv, + "power_capacity_in_mw": ur.Quantity("2 MW"), + "consumption_capacity": ur.Quantity("0 MW"), + "production_capacity": ur.Quantity("2 MW"), + "group": {"asset": inverter}, + }, + { + "asset": inverter, + "power_capacity_in_mw": ur.Quantity("2.5 MW"), + "consumption": {"sensor": consumption_sensor}, + }, + ] + scheduler = run_scheduler( + building, flex_model, base_flex_context(), return_multiple=True + ) + results = scheduler.compute() + + storage_schedules = { + r["sensor"]: r["data"] for r in results if r.get("name") == "storage_schedule" + } + assert battery in storage_schedules + assert pv in storage_schedules + # The group entry has no power sensor of its own, so it isn't in storage_schedules. + assert not any( + getattr(sensor, "asset", None) == inverter for sensor in storage_schedules + ) + + consumption_schedules = { + r["sensor"]: r["data"] + for r in results + if r.get("name") == "consumption_schedule" + } + assert consumption_sensor in consumption_schedules + aggregate = consumption_schedules[consumption_sensor] + assert (aggregate.abs() <= 2.5 + 1e-6).all() + assert ( + (storage_schedules[battery] + storage_schedules[pv] - aggregate).abs() < 1e-6 + ).all() + + +def test_group_asset_ref_production_only_output(db, building): + """An asset-referenced group entry with only a production output sensor gets the + full aggregate profile in native (consumption-positive) convention; sign inversion + to production-positive happens downstream in make_schedule.""" + battery, pv = make_sensors(db, building, n=2) + inverter = make_sub_asset(db, building) + production_sensor = make_output_sensor(db, inverter) + + flex_model = [ + { + "sensor": battery, + "soc_at_start": 1.0, + "soc_min": 0.0, + "soc_max": 2.0, + "power_capacity_in_mw": ur.Quantity("2 MW"), + "group": {"asset": inverter}, + }, + { + "sensor": pv, + "power_capacity_in_mw": ur.Quantity("2 MW"), + "consumption_capacity": ur.Quantity("0 MW"), + "group": {"asset": inverter}, + }, + { + "asset": inverter, + "power_capacity_in_mw": ur.Quantity("2.5 MW"), + "production": {"sensor": production_sensor}, + }, + ] + scheduler = run_scheduler( + building, flex_model, base_flex_context(), return_multiple=True + ) + results = scheduler.compute() + + storage_schedules = { + r["sensor"]: r["data"] for r in results if r.get("name") == "storage_schedule" + } + production_schedules = { + r["sensor"]: r["data"] + for r in results + if r.get("name") == "production_schedule" + } + assert production_sensor in production_schedules + expected = storage_schedules[battery] + storage_schedules[pv] + assert ((production_schedules[production_sensor] - expected).abs() < 1e-6).all() + + +def test_group_asset_ref_both_outputs_split(db, building): + """An asset-referenced group entry with both consumption and production output + sensors gets the clip-split of the aggregate: non-negative to consumption, + non-positive to production.""" + battery, pv = make_sensors(db, building, n=2) + inverter = make_sub_asset(db, building) + consumption_sensor = make_output_sensor(db, inverter) + production_sensor = make_output_sensor(db, inverter) + + flex_model = [ + { + "sensor": battery, + "soc_at_start": 1.0, + "soc_min": 0.0, + "soc_max": 2.0, + "power_capacity_in_mw": ur.Quantity("2 MW"), + "group": {"asset": inverter}, + }, + { + "sensor": pv, + "power_capacity_in_mw": ur.Quantity("2 MW"), + "consumption_capacity": ur.Quantity("0 MW"), + "group": {"asset": inverter}, + }, + { + "asset": inverter, + "power_capacity_in_mw": ur.Quantity("2.5 MW"), + "consumption": {"sensor": consumption_sensor}, + "production": {"sensor": production_sensor}, + }, + ] + scheduler = run_scheduler( + building, flex_model, base_flex_context(), return_multiple=True + ) + results = scheduler.compute() + + storage_schedules = { + r["sensor"]: r["data"] for r in results if r.get("name") == "storage_schedule" + } + consumption_schedules = { + r["sensor"]: r["data"] + for r in results + if r.get("name") == "consumption_schedule" + } + production_schedules = { + r["sensor"]: r["data"] + for r in results + if r.get("name") == "production_schedule" + } + aggregate = storage_schedules[battery] + storage_schedules[pv] + assert ( + (consumption_schedules[consumption_sensor] - aggregate.clip(lower=0)).abs() + < 1e-6 + ).all() + assert ( + (production_schedules[production_sensor] - aggregate.clip(upper=0)).abs() < 1e-6 + ).all() + # Consistency: consumption plus production reconstructs the full aggregate. + assert ( + ( + consumption_schedules[consumption_sensor] + + production_schedules[production_sensor] + - aggregate + ).abs() + < 1e-6 + ).all() + + +def test_group_asset_ref_dangling_raises(db, building): + battery = make_sensors(db, building, n=1)[0] + other_asset = make_sub_asset(db, building) + + flex_model = [ + { + "sensor": battery, + "power_capacity_in_mw": ur.Quantity("2 MW"), + "group": {"asset": other_asset}, + }, + ] + scheduler = run_scheduler(building, flex_model, base_flex_context()) + with pytest.raises(ValueError, match="group"): + scheduler.compute() + + +def test_group_asset_ref_with_sensor_raises(db, building): + """An asset-referenced group entry must not also carry a `sensor` field.""" + battery, pv = make_sensors(db, building, n=2) + inverter = make_sub_asset(db, building) + bogus_sensor = make_group_sensor(db, building) + + flex_model = [ + { + "sensor": battery, + "power_capacity_in_mw": ur.Quantity("2 MW"), + "group": {"asset": inverter}, + }, + { + "sensor": pv, + "power_capacity_in_mw": ur.Quantity("2 MW"), + "group": {"asset": inverter}, + }, + { + "asset": inverter, + "sensor": bogus_sensor, + "power_capacity_in_mw": ur.Quantity("2.5 MW"), + }, + ] + scheduler = run_scheduler(building, flex_model, base_flex_context()) + with pytest.raises(ValueError, match="asset-"): + scheduler.compute() + + +def test_nested_group_mixed_ref_kinds(db, building): + """Inner group referenced by asset, outer group referenced by sensor: leaf + resolution must work transitively across both kinds.""" + battery = make_sensors(db, building, n=1)[0] + inverter = make_sub_asset(db, building) + outer_group_sensor = make_group_sensor(db, building) + + flex_model = [ + { + "sensor": battery, + "soc_at_start": 1.0, + "soc_min": 0.0, + "soc_max": 2.0, + "power_capacity_in_mw": ur.Quantity("2 MW"), + "group": {"asset": inverter}, + }, + { + "asset": inverter, + "power_capacity_in_mw": ur.Quantity("3 MW"), + "group": {"sensor": outer_group_sensor}, + }, + { + "sensor": outer_group_sensor, + "power_capacity_in_mw": ur.Quantity("3.5 MW"), + }, + ] + scheduler = run_scheduler( + building, flex_model, base_flex_context(), return_multiple=True + ) + results = scheduler.compute() + storage_schedules = { + r["sensor"]: r["data"] for r in results if r.get("name") == "storage_schedule" + } + assert outer_group_sensor in storage_schedules + assert ( + (storage_schedules[outer_group_sensor] - storage_schedules[battery]).abs() + < 1e-6 + ).all() + + +def test_pure_db_tree_group_constraint(db, building): + """End-to-end (planning-level): the entire flex-model lives on the asset tree in + the DB (no flex-model entries are passed to the scheduler at all). A site asset + (``building``) has two child assets (battery-like and PV-like) that are both + asset-only device entries (no power sensor of their own; results are saved via + consumption/production output sensors) belonging to a group referenced by a third + child asset (an "inverter"), which itself defines the group's hard power-capacity + and saves the group's aggregate via its own consumption output sensor. + + Triggering the site asset with an empty flex-model list should still produce a + correctly constrained schedule, entirely from `GenericAsset.flex_model` attributes, + via `Scheduler.collect_flex_config`. + """ + inverter = make_sub_asset(db, building) + battery_asset = make_sub_asset(db, building) + pv_asset = make_sub_asset(db, building) + + inverter_consumption_sensor = make_output_sensor(db, inverter) + battery_consumption_sensor = make_output_sensor(db, battery_asset) + battery_production_sensor = make_output_sensor(db, battery_asset) + pv_production_sensor = make_output_sensor(db, pv_asset) + + # Store the flex-model entirely on the assets in the DB (asset-only entries: no + # "sensor" key, so results are saved via consumption/production output sensors). + battery_asset.flex_model = { + "power-capacity": "2 MW", + "consumption-capacity": "2 MW", + "production-capacity": "2 MW", + "group": {"asset": inverter.id}, + "consumption": {"sensor": battery_consumption_sensor.id}, + "production": {"sensor": battery_production_sensor.id}, + } + pv_asset.flex_model = { + "power-capacity": "2 MW", + "consumption-capacity": "0 MW", + "production-capacity": "2 MW", + "group": {"asset": inverter.id}, + "production": {"sensor": pv_production_sensor.id}, + } + inverter.flex_model = { + "power-capacity": "2.5 MW", + "consumption": {"sensor": inverter_consumption_sensor.id}, + } + db.session.add_all([battery_asset, pv_asset, inverter]) + db.session.commit() + + scheduler = StorageScheduler( + asset_or_sensor=building, + start=pd.Timestamp("2023-01-01T00:00:00", tz="Europe/Amsterdam"), + end=pd.Timestamp("2023-01-01T04:00:00", tz="Europe/Amsterdam"), + resolution=timedelta(hours=1), + flex_model=[], # entirely DB-driven + # `building`'s own flex-context (in the DB) already sets a large + # site-power-capacity; override the (unpopulated) sensor-based + # consumption-price with fixed quantities here. Real deserialization + # (collect_flex_config + schema loading) is exercised, unlike in the other + # tests in this file (which bypass it). + flex_context={ + "consumption-price": "100 EUR/MWh", + "production-price": "100 EUR/MWh", + }, + return_multiple=True, + ) + results = scheduler.compute() + + consumption_schedules = { + r["sensor"]: r["data"] + for r in results + if r.get("name") == "consumption_schedule" + } + production_schedules = { + r["sensor"]: r["data"] + for r in results + if r.get("name") == "production_schedule" + } + + assert inverter_consumption_sensor in consumption_schedules + aggregate = consumption_schedules[inverter_consumption_sensor] + # Hard cap on the group's aggregate power respected. + assert (aggregate.abs() <= 2.5 + 1e-6).all() + + # The aggregate equals the sum of the (signed, consumption-positive) member + # device schedules, reconstructed from their consumption/production outputs. + battery_signed = consumption_schedules.get( + battery_consumption_sensor, + pd.Series(0.0, index=aggregate.index), + ) + production_schedules.get( + battery_production_sensor, pd.Series(0.0, index=aggregate.index) + ) + pv_signed = production_schedules.get( + pv_production_sensor, pd.Series(0.0, index=aggregate.index) + ) + assert ((battery_signed + pv_signed - aggregate).abs() < 1e-6).all() diff --git a/flexmeasures/data/schemas/scheduling/__init__.py b/flexmeasures/data/schemas/scheduling/__init__.py index 93c86ac555..da91718107 100644 --- a/flexmeasures/data/schemas/scheduling/__init__.py +++ b/flexmeasures/data/schemas/scheduling/__init__.py @@ -943,6 +943,15 @@ def check_prices(self, data: dict, original_data: dict, **kwargs): }, "example-units": EXAMPLE_UNIT_TYPES["power"], }, + "group": { + "default": None, + "description": rst_to_openapi(metadata.GROUP.description), + "types": { + "backend": "typeTwo", + "ui": "A power sensor or an asset representing a group of devices; a sensor-referenced group also records the group's scheduled aggregate power, while an asset-referenced group records it via its own consumption/production output sensors.", + }, + "example-units": EXAMPLE_UNIT_TYPES["power"], + }, "commodity": { "default": "electricity", "description": rst_to_openapi(metadata.COMMODITY_FLEX_MODEL.description), @@ -1256,7 +1265,11 @@ def check_flex_model_sensors(self, data, **kwargs): asset = data["asset"] sensors = [] for sensor_flex_model in data.get("flex_model", []): - sensor = sensor_flex_model["sensor"] + sensor = sensor_flex_model.get("sensor") + if sensor is None: + # Asset-only entries (e.g. an asset-referenced `group` entry) carry no + # sensor of their own; nothing to check here. + continue if sensor in sensors: raise FMValidationError( f"Sensor {sensor_flex_model['sensor'].id} should not occur more than once in the flex-model" diff --git a/flexmeasures/data/schemas/scheduling/metadata.py b/flexmeasures/data/schemas/scheduling/metadata.py index 4f6f9d4298..c6340194a0 100644 --- a/flexmeasures/data/schemas/scheduling/metadata.py +++ b/flexmeasures/data/schemas/scheduling/metadata.py @@ -392,3 +392,11 @@ def to_dict(self): """, example="0 kW", ) +GROUP = MetaData( + description="""Reference to a group of devices whose aggregate power is constrained, given as either a power sensor (``{"sensor": }``) or an asset (``{"asset": }``) - exactly one of the two. +The referenced sensor or asset should itself get its own flex-model entry defining the group's ``power-capacity`` (hard constraint) and/or ``consumption-capacity``/``production-capacity`` (soft constraints with default breach prices). +When the group is referenced by ``sensor``, the group's scheduled aggregate power is saved to that group sensor. +When the group is referenced by ``asset`` (e.g. a sub-EMS asset in the tree), the group entry defines no power sensor of its own; the group's aggregate power is instead saved via that entry's own ``consumption`` and/or ``production`` output sensors, following the usual output-sensor conventions. +""", + example={"sensor": 5}, +) diff --git a/flexmeasures/data/schemas/scheduling/storage.py b/flexmeasures/data/schemas/scheduling/storage.py index 60d9da3448..0c5df11492 100644 --- a/flexmeasures/data/schemas/scheduling/storage.py +++ b/flexmeasures/data/schemas/scheduling/storage.py @@ -19,7 +19,9 @@ from flexmeasures.data.schemas.units import QuantityField from flexmeasures.data.schemas.scheduling import metadata from flexmeasures.data.schemas.sensors import ( + SensorIdField, SensorReference, + SensorReferenceSchema, OutputSensorReferenceSchema, VariableQuantityField, ) @@ -31,6 +33,44 @@ ALLOWED_COMMODITIES = {"electricity", "gas"} + +def _validate_group_sensor_is_power_sensor(group: dict): + """Check that the sensor referenced by the `group` field measures power.""" + sensor = group.get("sensor") + if isinstance(sensor, (Sensor, SensorReference)) and not is_power_unit(sensor.unit): + raise ValidationError( + "The `group` field must reference a sensor with a power unit.", + field_name="sensor", + ) + + +class GroupReferenceSchema(SensorReferenceSchema): + """Reference to a group of devices whose aggregate power is constrained. + + Accepts exactly one of: + - ``{"sensor": }``: the group's aggregate power is stored on this power sensor + (the sensor must itself carry a flex-model entry defining the group's + constraints). + - ``{"asset": }``: the group is identified by the flex-model entry on this + asset (typically a sub-EMS/asset in the tree). Such a group entry defines no + power sensor of its own; instead it may define ``consumption`` and/or + ``production`` output sensors on which the group's aggregate power gets saved, + following the usual output-sensor conventions. + """ + + sensor = SensorIdField(required=False) + asset = GenericAssetIdField(required=False) + + @validates_schema + def validate_exactly_one_reference(self, data: dict, **kwargs): + has_sensor = "sensor" in data + has_asset = "asset" in data + if has_sensor == has_asset: # both or neither + raise ValidationError( + "The `group` field must reference exactly one of 'sensor' or 'asset'." + ) + + # Telling type hints what to expect after schema parsing SoCTarget = TypedDict( "SoCTarget", @@ -148,6 +188,13 @@ class StorageFlexModelSchema(Schema): metadata=metadata.PRODUCTION_CAPACITY.to_dict(), ) + group = fields.Nested( + GroupReferenceSchema, + data_key="group", + required=False, + metadata=metadata.GROUP.to_dict(), + ) + # Activation prices prefer_curtailing_later = fields.Bool( data_key="prefer-curtailing-later", @@ -341,6 +388,10 @@ def validate_state_of_charge( "The `state-of-charge` field can only be a Sensor or a time series." ) + @validates("group") + def validate_group(self, group: dict, **kwargs): + _validate_group_sensor_is_power_sensor(group) + @validates("asset") def validate_asset(self, asset: Asset, **kwargs): if self.sensor is not None and self.sensor.asset != asset: @@ -428,6 +479,13 @@ class DBStorageFlexModelSchema(Schema): consumption = fields.Nested(OutputSensorReferenceSchema) production = fields.Nested(OutputSensorReferenceSchema) + group = fields.Nested( + GroupReferenceSchema, + data_key="group", + required=False, + metadata=metadata.GROUP.to_dict(), + ) + soc_min = VariableQuantityField( to_unit="MWh", data_key="soc-min", @@ -568,6 +626,10 @@ def __init__(self, *args, **kwargs): for field in self.declared_fields } + @validates("group") + def validate_group(self, group: dict, **kwargs): + _validate_group_sensor_is_power_sensor(group) + @validates_schema def forbid_time_series_specs(self, data: dict, **kwargs): """Do not allow time series specs for the flex-model fields saved in the db.""" diff --git a/flexmeasures/data/schemas/tests/test_scheduling.py b/flexmeasures/data/schemas/tests/test_scheduling.py index 90842d2cf9..88bf86ce9c 100644 --- a/flexmeasures/data/schemas/tests/test_scheduling.py +++ b/flexmeasures/data/schemas/tests/test_scheduling.py @@ -856,6 +856,15 @@ def test_flex_context_schema_rejects_filtered_aggregate_power( }, False, ), + # group must reference a power sensor + ( + {"group": {"sensor": "power-sensor"}}, + False, + ), + ( + {"group": {"sensor": "energy-sensor"}}, + {"group": "The `group` field must reference a sensor with a power unit."}, + ), ], ) def test_flex_model_schemas( @@ -911,6 +920,42 @@ def test_flex_model_schemas( schema.load(flex_model) +def test_storage_flex_model_group_field(db, app, setup_dummy_sensors, dummy_asset): + """The `group` field should load a `{"sensor": }` reference to a power Sensor, + reject non-power sensors, and reject unknown sensor IDs.""" + energy_sensor, _, _, power_sensor = setup_dummy_sensors + + for schema in ( + StorageFlexModelSchema( + start=datetime(2026, 6, 1, tzinfo=pytz.UTC), sensor=None + ), + DBStorageFlexModelSchema(), + ): + # Valid group reference loads to a Sensor + flex_model = schema.load({"group": {"sensor": power_sensor.id}}) + assert flex_model["group"]["sensor"] == power_sensor + + # A non-power sensor is rejected + with pytest.raises(ValidationError, match="power unit"): + schema.load({"group": {"sensor": energy_sensor.id}}) + + # An unknown sensor ID is rejected (by SensorIdField) + with pytest.raises(ValidationError, match="No sensor found"): + schema.load({"group": {"sensor": -1}}) + + # A valid asset reference loads to a GenericAsset + flex_model = schema.load({"group": {"asset": dummy_asset.id}}) + assert flex_model["group"]["asset"] == dummy_asset + + # Both sensor and asset given: rejected + with pytest.raises(ValidationError, match="exactly one"): + schema.load({"group": {"sensor": power_sensor.id, "asset": dummy_asset.id}}) + + # Neither sensor nor asset given: rejected + with pytest.raises(ValidationError, match="exactly one"): + schema.load({"group": {}}) + + @pytest.mark.parametrize( ["flex_context", "fails"], [ diff --git a/flexmeasures/ui/static/openapi-specs.json b/flexmeasures/ui/static/openapi-specs.json index 0eab864eef..79a2700370 100644 --- a/flexmeasures/ui/static/openapi-specs.json +++ b/flexmeasures/ui/static/openapi-specs.json @@ -7,7 +7,7 @@ }, "termsOfService": null, "title": "FlexMeasures", - "version": "1.0.0" + "version": "0.33.2" }, "externalDocs": { "description": "FlexMeasures runs on the open source FlexMeasures technology. Read the docs here.", @@ -6200,6 +6200,62 @@ ], "additionalProperties": false }, + "GroupReference": { + "type": "object", + "properties": { + "sensor": { + "type": "integer" + }, + "source-types": { + "type": [ + "array", + "null" + ], + "default": null, + "description": "Only use beliefs from sources with these source types (e.g. 'user', 'script', 'forecaster', 'scheduler').", + "items": { + "type": "string" + } + }, + "exclude-source-types": { + "type": [ + "array", + "null" + ], + "default": null, + "description": "Exclude beliefs from sources with these source types.", + "items": { + "type": "string" + } + }, + "sources": { + "type": [ + "array", + "null" + ], + "default": null, + "description": "Only use beliefs from these data source IDs.", + "items": { + "type": "integer" + } + }, + "source-account": { + "type": [ + "array", + "null" + ], + "default": null, + "description": "Only use beliefs from data sources linked to these account IDs.", + "items": { + "type": "integer" + } + }, + "asset": { + "type": "integer" + } + }, + "description": "Sensor reference from which to look up a variable quantity." + }, "StorageFlexModelSchemaOpenAPI": { "type": "object", "properties": { @@ -6259,6 +6315,13 @@ "example": "0 kW", "$ref": "#/components/schemas/VariableQuantityOpenAPI" }, + "group": { + "description": "Reference to a group of devices whose aggregate power is constrained, given as either a power sensor ({\"sensor\": }) or an asset ({\"asset\": }) - exactly one of the two.\nThe referenced sensor or asset should itself get its own flex-model entry defining the group's power-capacity (hard constraint) and/or consumption-capacity/production-capacity (soft constraints with default breach prices).\nWhen the group is referenced by sensor, the group's scheduled aggregate power is saved to that group sensor.\nWhen the group is referenced by asset (e.g. a sub-EMS asset in the tree), the group entry defines no power sensor of its own; the group's aggregate power is instead saved via that entry's own consumption and/or production output sensors, following the usual output-sensor conventions.\n", + "example": { + "sensor": 5 + }, + "$ref": "#/components/schemas/GroupReference" + }, "prefer-curtailing-later": { "type": "boolean", "default": true, diff --git a/flexmeasures/ui/templates/assets/asset_properties.html b/flexmeasures/ui/templates/assets/asset_properties.html index b29dc02847..f0ec2e5c3e 100644 --- a/flexmeasures/ui/templates/assets/asset_properties.html +++ b/flexmeasures/ui/templates/assets/asset_properties.html @@ -391,6 +391,22 @@
Edit {{ asset.name }}'s flex-model
+ {% if parent_has_power_capacity %} + + {% elif own_has_power_capacity and has_child_assets %} + + {% endif %} +
@@ -446,6 +462,15 @@
Edit {{ asset.name }}'s flex-model
// This variable is used to prevent reRenderForm from running on initial load let hasInitialized = false; + // Info used to hint the user towards using the parent asset as the `group` + // for the flex-model's `group` field (and vice versa, hinting the parent + // that its children can join its power-capacity group). + const parentAssetId = {{ parent_asset_id | tojson | safe }}; + const parentAssetName = {{ parent_asset_name | tojson | safe }}; + const parentHasPowerCapacity = {{ parent_has_power_capacity | tojson | safe }}; + const ownHasPowerCapacity = {{ own_has_power_capacity | tojson | safe }}; + const hasChildAssets = {{ has_child_assets | tojson | safe }}; + const schemaSpecs = {{ flex_model_schema | tojson | safe }}; const assetFlexModelSchema = {}; const FlexModelFieldValidTypes = {}; @@ -596,6 +621,27 @@
Edit {{ asset.name }}'s flex-model
senSearchResEle.style.display = 'none'; } + function setCardGroupAsset(assetId) { + if (!assetId) { + return; + } + + const flexModel = getFlexModel(); + flexModel["group"] = { "asset": assetId }; + setFlexModel(flexModel); + + setTimeout(() => { + const card = document.getElementById('group-control'); + if (!card) { + return; + } + setActiveCard(card); + card.classList.add('border-on-click'); + renderSelectInfoCards('group'); + renderFlexInputOptions(); + }, 500); + } + async function searchSensors() { const searchValue = sensorSearchBar.value.toLowerCase(); spinnerElement.style.display = 'flex'; @@ -694,6 +740,37 @@
Edit {{ asset.name }}'s flex-model
` : ""; + // Hint towards using the parent asset as the value for `group`, + // especially when the parent already limits the group's power. + let groupParentHint = ""; + if (modelKey === "group" && parentAssetId) { + const powerCapacityNote = parentHasPowerCapacity + ? `The parent asset ${parentAssetName} defines a power-capacity. Consider joining that group so this device counts toward the parent's limit.` + : `Consider using the parent asset ${parentAssetName} as the group.`; + groupParentHint = ` +
+ ${powerCapacityNote} +
+ +
+
+ `; + } + + // Hint towards the fact that children can reference this asset + // as their `group`, when this asset itself has a power-capacity. + let powerCapacityChildrenHint = ""; + if (modelKey === "power-capacity" && ownHasPowerCapacity && hasChildAssets) { + powerCapacityChildrenHint = ` +
+ + This asset has child assets. They can set group: {"asset": {{ asset.id }} } in their own flex-model's group field, to share this power-capacity limit. +
+ `; + } + flexInfoContainer.innerHTML = `
${storageEfficiencyWarning} + ${groupParentHint} + ${powerCapacityChildrenHint} -
+
Possible types: ${schemaSpecs[modelKey]["types"]["ui"] || "No types available."}
Example units: ${schemaSpecs[modelKey]["example-units"].join(", ")}
`; + + const useParentAsGroupBtn = document.getElementById('useParentAsGroupBtn'); + if (useParentAsGroupBtn) { + useParentAsGroupBtn.onclick = function () { + setCardGroupAsset(parentAssetId); + }; + } } else { flexInfoContainer.innerHTML = ""; } diff --git a/flexmeasures/ui/tests/test_asset_crud.py b/flexmeasures/ui/tests/test_asset_crud.py index 2341d631d8..2f327a830d 100644 --- a/flexmeasures/ui/tests/test_asset_crud.py +++ b/flexmeasures/ui/tests/test_asset_crud.py @@ -496,3 +496,75 @@ def test_admin_only_buttons_on_properties_page( assert b"Create asset" in page.data else: assert b"Create asset" not in page.data + + +def test_group_field_hints_on_properties_page( + db, client, as_admin, setup_accounts, setup_generic_asset_types +): + """The properties page should hint at using the parent asset as `group`, + depending on whether the (child's) parent or the asset itself defines a + power-capacity in its flex-model.""" + + # Parent with a power-capacity in its flex-model, and a child without a group set. + parent_with_capacity = GenericAsset( + name="parent-with-power-capacity", + generic_asset_type=setup_generic_asset_types["battery"], + owner=setup_accounts["Prosumer"], + latitude=10, + longitude=100, + flex_model={"power-capacity": "400kW"}, + ) + db.session.add(parent_with_capacity) + db.session.flush() + + child = GenericAsset( + name="child-of-parent-with-power-capacity", + generic_asset_type=setup_generic_asset_types["battery"], + owner=setup_accounts["Prosumer"], + latitude=10, + longitude=100, + parent_asset_id=parent_with_capacity.id, + ) + db.session.add(child) + + # Parent without children with power-capacity and without children (no hints). + lone_asset = GenericAsset( + name="lone-asset-no-hints", + generic_asset_type=setup_generic_asset_types["battery"], + owner=setup_accounts["Prosumer"], + latitude=10, + longitude=100, + ) + db.session.add(lone_asset) + db.session.commit() + + # (a) child whose parent has power-capacity: expect the "join parent group" hint. + child_page = client.get( + url_for("AssetCrudUI:properties", id=child.id), + follow_redirects=True, + ) + assert child_page.status_code == 200 + assert b"parent-with-power-capacity" in child_page.data + assert b"power-capacity" in child_page.data + assert f'group: {{"asset": {parent_with_capacity.id} }}'.encode() in child_page.data + + # (b) parent with power-capacity and children: expect the "children can join" hint. + parent_page = client.get( + url_for("AssetCrudUI:properties", id=parent_with_capacity.id), + follow_redirects=True, + ) + assert parent_page.status_code == 200 + assert b"Child assets can" in parent_page.data + assert ( + f'group: {{"asset": {parent_with_capacity.id} }}'.encode() in parent_page.data + ) + + # (c) an asset with neither a parent with power-capacity, nor children with + # power-capacity of its own: expect no hints. + lone_page = client.get( + url_for("AssetCrudUI:properties", id=lone_asset.id), + follow_redirects=True, + ) + assert lone_page.status_code == 200 + assert b"Consider setting" not in lone_page.data + assert b"Child assets can" not in lone_page.data diff --git a/flexmeasures/ui/views/assets/views.py b/flexmeasures/ui/views/assets/views.py index 30084e7ad9..f795318d00 100644 --- a/flexmeasures/ui/views/assets/views.py +++ b/flexmeasures/ui/views/assets/views.py @@ -435,6 +435,21 @@ def properties(self, id: str): .all() ) + # Info to help the user set the `group` field of the flex-model: + # - if this asset has a parent, suggest referencing the parent as the group + # - if the parent already defines a power-capacity, hint that this asset + # should join that group to count toward the parent's limit + # - if this asset itself defines a power-capacity and has children, hint + # that those children can reference this asset as their group + parent_asset_id = asset.parent_asset.id if asset.parent_asset else None + parent_asset_name = asset.parent_asset.name if asset.parent_asset else None + parent_has_power_capacity = bool( + asset.parent_asset + and "power-capacity" in (asset.parent_asset.flex_model or {}) + ) + own_has_power_capacity = "power-capacity" in (asset.flex_model or {}) + has_child_assets = bool(asset.child_assets) + # Can the user create a sibling of this asset? # - Has a parent → check create-children on that parent asset. # - No parent, owned account → check create-children on that account. @@ -485,4 +500,9 @@ def properties(self, id: str): attributes_label=ATTRIBUTES_FIELD_LABEL, attributes_description=ATTRIBUTES_FIELD_DESCRIPTION, stored_secrets=get_secret_overview(asset.secrets), + parent_asset_id=parent_asset_id, + parent_asset_name=parent_asset_name, + parent_has_power_capacity=parent_has_power_capacity, + own_has_power_capacity=own_has_power_capacity, + has_child_assets=has_child_assets, )