diff --git a/documentation/changelog.rst b/documentation/changelog.rst index d010a74639..05c3f4bf6f 100644 --- a/documentation/changelog.rst +++ b/documentation/changelog.rst @@ -12,12 +12,12 @@ v1.0.0 | July XX, 2026 New features ------------- +* Floor off-clock API datetimes to a non-instantaneous sensor's resolution by default when ingesting sensor data, uploading sensor data, and handling scheduler flex-model timed events; configurable with the ``floor_datetimes_to_resolution`` sensor attribute [see `PR #2146 `_ and `PR #2194 `_] * In the UI, asset and sensor charts now render with Apache ECharts (canvas) by default, for much faster drawing and interaction on dense time series, while staying visually and functionally equivalent to the previous Vega-Lite charts, which remain available as a fallback via a toggle [see `PR #2234 `_] * Breaking behaviour change: the top-level flex-context's ``relax-constraints`` field now defaults to ``True`` (matching the default already used within each ``commodities`` entry), so constraint violations are softly penalized by default instead of being hard constraints, unless explicitly set to ``False`` [see `PR #2172 `_] * Support for creating new assets by using another asset as a template from the UI. [see `PR #2195 `_ and `PR #2268 `_ * In the UI, asset and sensor lists can be filtered by ID prefix through API-backed search fields [see `PR #2231 `_] * Support configurable lower and upper bounds and snapping for forecast post-processing [see `PR #2273 `_] -* Floor off-clock API datetimes to a non-instantaneous sensor's resolution by default when ingesting sensor data, uploading sensor data, and handling scheduler flex-model timed events; configurable with the ``floor_datetimes_to_resolution`` sensor attribute [see `PR #2146 `_] * Sensor references in flex-model and flex-context support various ways of filtering by source [see `PR #2209 `_] * Let storage scheduling infer missing ``power-capacity`` from directional device capacities before falling back to site capacity, and default the missing opposite capacity to zero when only a non-zero ``consumption-capacity`` or ``production-capacity`` is configured [see `PR #2222 `_] * Support multiple feeders to a shared storage [see `PR #2001 `_, `PR #2321 `_, `PR #2322 `_ and `PR #2325 `_] diff --git a/documentation/concepts/data-model.rst b/documentation/concepts/data-model.rst index 6bb300a0c0..ac6a3fa20c 100644 --- a/documentation/concepts/data-model.rst +++ b/documentation/concepts/data-model.rst @@ -100,6 +100,32 @@ Each belief links to a sensor and a data source. Here are two examples: See also :ref:`one_or_multiple_sensors` for guidance on when such beliefs are best recorded on one shared sensor and when separate sensors are preferable. +.. _projecting_scheduling_constraints: + +Projecting scheduling constraints to a fixed resolution +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + +Schedulers operate on a fixed scheduling resolution, such as 15 minutes. +This means the optimization problem can enforce state-of-charge constraints only on the scheduling ticks implied by that resolution. +When a storage flex-model contains point-like ``soc-targets``, ``soc-minima`` or ``soc-maxima`` between two scheduling ticks, FlexMeasures projects those constraints onto the surrounding ticks instead of simply flooring them. + +For an off-tick ``soc-target``, the target value is moved to the next scheduling tick as an exact target. +The previous tick receives lower and upper bounds that reflect how much the asset could still charge or discharge between the previous tick and the original target time. +For off-tick ``soc-minima``, both surrounding ticks receive lower bounds that preserve whether the requested minimum can still be reached. +For off-tick ``soc-maxima``, both surrounding ticks receive upper bounds with the same reachability logic. + +The projection uses the ``consumption-capacity`` and ``production-capacity`` active at the relevant ticks. +If multiple projected lower bounds land on the same tick, the highest lower bound is kept. +If multiple projected upper bounds land on the same tick, the lowest upper bound is kept. +Because projection can introduce additional bounds and more complex combinations can become infeasible, FlexMeasures enables ``relax-soc-constraints`` automatically when off-tick SoC constraints are submitted (unless it was explicitly set to ``False``, which is respected with a logged warning). +When relaxation is enabled purely because of off-tick projection (rather than by the flex-context settings), the softening is scoped to the devices that actually use off-tick SoC constraints; other devices keep their hard SoC constraints. + +The starting state of charge is projected as well: when ``soc-at-start`` is resolved from the ``state-of-charge`` field and the underlying measurement was taken at an off-tick time within the first scheduling interval, the SoC is assumed to hold until that time, and the next scheduling tick receives bounds reflecting how much the device can still (dis)charge after it. + +Projection can be disabled per sensor by setting the ``floor_datetimes_to_resolution`` sensor attribute to ``False``. +In that case, off-tick point-like SoC constraints cannot be enforced on the scheduling ticks and are disregarded (with a logged warning). + + .. _signs_of_power_beliefs: About signs of power & energy values diff --git a/documentation/features/scheduling.rst b/documentation/features/scheduling.rst index 0fa9735df1..f728d422a6 100644 --- a/documentation/features/scheduling.rst +++ b/documentation/features/scheduling.rst @@ -293,6 +293,8 @@ For more details on the possible formats for field values, see :ref:`variable_qu .. [#minimum_overlap] In case this field defines partially overlapping time periods, the minimum value is selected. See :ref:`variable_quantities`. +.. [#projecting_scheduling_constraints] Off-tick ``soc-targets``, ``soc-minima`` and ``soc-maxima`` are projected to the surrounding scheduling ticks. See :ref:`projecting_scheduling_constraints`. + For more details on the possible formats for field values, see :ref:`variable_quantities`. Usually, not the whole flexibility model is needed. diff --git a/flexmeasures/data/models/planning/soc_projection.py b/flexmeasures/data/models/planning/soc_projection.py new file mode 100644 index 0000000000..266a7b0099 --- /dev/null +++ b/flexmeasures/data/models/planning/soc_projection.py @@ -0,0 +1,489 @@ +"""Projection of off-tick point-like SoC constraints onto scheduling ticks.""" + +from __future__ import annotations + +import copy +import logging +from dataclasses import dataclass, field +from datetime import datetime, timedelta +from typing import Literal + +import pandas as pd + +from flexmeasures.data.models.time_series import Sensor +from flexmeasures.data.schemas.scheduling.utils import is_on_schedule_tick +from flexmeasures.utils.unit_utils import ur + +logger = logging.getLogger(__name__) + +SocBoundType = Literal["min", "max"] +SocProjectionTick = Literal["previous", "next"] + +TimedEventList = list[dict[str, datetime | float]] +SocSpecification = TimedEventList | pd.Series | Sensor | ur.Quantity | None + + +@dataclass(frozen=True) +class SocProjectionRule: + """One projected bound for an off-tick point-like SoC event. + + A rule only needs to state *which* bound lands on *which* surrounding + scheduling tick; everything else follows from preserving reachability: + + - The capacity period runs from the tick to the event time (``previous``) + or from the event time to the tick (``next``). + - A bound is loosened by the energy the device can move through that period + towards satisfying the original event: lower bounds by charging up to it + (on the previous tick) or by discharging away from it (on the next tick), + and upper bounds vice versa. + - Lower bounds are loosened downwards, upper bounds upwards. + """ + + bound_type: SocBoundType + tick: SocProjectionTick + + @property + def uses_charging(self) -> bool: + """Whether the bound is loosened by chargeable (rather than dischargeable) energy.""" + return (self.bound_type == "min") == (self.tick == "previous") + + @property + def sign(self) -> int: + """Lower bounds are loosened downwards (-1), upper bounds upwards (+1).""" + return -1 if self.bound_type == "min" else +1 + + +#: How each type of off-tick point-like SoC event is projected onto the +#: surrounding scheduling ticks (see :class:`SocProjectionRule`). +#: Off-tick ``soc-targets`` additionally become an exact target on the next +#: tick, and ``soc-at-start`` denotes a starting SoC known at an off-tick time +#: within the first scheduling interval; both are handled by their respective +#: entry points below. +SOC_PROJECTION_POLICIES: dict[str, tuple[SocProjectionRule, ...]] = { + "soc-targets": ( + SocProjectionRule("min", "previous"), + SocProjectionRule("max", "previous"), + ), + "soc-minima": ( + SocProjectionRule("min", "previous"), + SocProjectionRule("min", "next"), + ), + "soc-maxima": ( + SocProjectionRule("max", "previous"), + SocProjectionRule("max", "next"), + ), + "soc-at-start": ( + SocProjectionRule("min", "next"), + SocProjectionRule("max", "next"), + ), +} + + +def _soc_value_in_mwh(value: ur.Quantity | float | int) -> float: + """Return the SoC value as a plain float in MWh.""" + if isinstance(value, ur.Quantity): + return value.to("MWh").magnitude + return float(value) + + +def _optional_soc_value_in_mwh(value: ur.Quantity | float | int | None) -> float | None: + """Return the SoC value as a plain float in MWh, passing through None.""" + if value is None: + return None + return _soc_value_in_mwh(value) + + +def _soc_event_at( + soc_event: dict[str, datetime | float], + dt: pd.Timestamp, + value: float, +) -> dict[str, datetime | float]: + """Return a copy of a point-like SoC event, moved to the given tick with the given value. + + All timing fields are set to the tick (any 'duration' is dropped), so the result + remains a point-like (instantaneous) event. + """ + shifted_event = copy.copy(soc_event) + shifted_event["value"] = value + shifted_event["start"] = dt.to_pydatetime() + shifted_event["end"] = dt.to_pydatetime() + if "datetime" in shifted_event: + shifted_event["datetime"] = dt.to_pydatetime() + shifted_event.pop("duration", None) + return shifted_event + + +def _efficiency_at( + efficiency: pd.Series | ur.Quantity | float | None, + tick: pd.Timestamp, +) -> float: + """Return the (dis)charging efficiency at the given tick as a plain float. + + Efficiencies may be given as a series (indexed by tick start), a fixed + dimensionless quantity, or a plain number. Missing values default to 1. + """ + if efficiency is None: + return 1 + if isinstance(efficiency, pd.Series): + if efficiency.index.tz is not None: + tick = tick.tz_convert(efficiency.index.tz) + value = efficiency.get(tick) + if value is None or pd.isna(value): + return 1 + return float(value) + if isinstance(efficiency, ur.Quantity): + return float(efficiency.to("dimensionless").magnitude) + return float(efficiency) + + +def _reachable_energy( + capacity: pd.Series, + start: pd.Timestamp, + end: pd.Timestamp, + resolution: timedelta, + efficiency: pd.Series | ur.Quantity | float | None = None, + efficiency_affects_stock: str = "multiply", +) -> float: + """Compute by how much energy (in MWh) the device can move its stock between two times. + + The period from ``start`` to ``end`` always lies within a single scheduling + tick (from the tick just before an off-tick event to the event, or from the + event to the tick just after it), so the given power capacity series (in MW, + indexed by tick start at the given resolution) applies at a single constant + value. A missing capacity value counts as zero (conservative: the projected + bound then sticks close to the original event value). + + The capacity limits the power exchanged with the grid, while the projected + bounds concern the stock (SoC). Charging at power P raises the stock at rate + P * charging_efficiency (``efficiency_affects_stock="multiply"``; note that a + charging efficiency can exceed 1, e.g. a heat pump's COP), while discharging + at power P lowers the stock at rate P / discharging_efficiency + (``efficiency_affects_stock="divide"``). + """ + if end <= start: + return 0 + tick = start.floor(resolution) + if capacity.index.tz is not None: + tick = tick.tz_convert(capacity.index.tz) + capacity_in_mw = capacity.get(tick) + if capacity_in_mw is None or pd.isna(capacity_in_mw): + return 0 + efficiency_value = _efficiency_at(efficiency, tick) + stock_rate_in_mw = float(capacity_in_mw) + if efficiency_affects_stock == "multiply": + stock_rate_in_mw *= efficiency_value + elif efficiency_value != 0: + stock_rate_in_mw /= efficiency_value + else: + # A zero discharging efficiency means the stock can drop arbitrarily + # fast without producing any power, so the bound becomes unbounded. + stock_rate_in_mw = float("inf") + return stock_rate_in_mw * ((end - start) / pd.Timedelta(hours=1)) + + +def _add_soc_bound( + soc_events: list[dict[str, datetime | float]], + soc_event: dict[str, datetime | float], + bound_type: str, +) -> None: + """Add a SoC bound to a list of timed events, merging bounds on the same period. + + If an event with the same start and end already exists, the stricter bound wins: + the maximum of two lower bounds, or the minimum of two upper bounds. + """ + for existing_event in soc_events: + if existing_event.get("start") == soc_event.get("start") and existing_event.get( + "end" + ) == soc_event.get("end"): + existing_value = _soc_value_in_mwh(existing_event["value"]) + soc_value = _soc_value_in_mwh(soc_event["value"]) + existing_event["value"] = ( + max(existing_value, soc_value) + if bound_type == "min" + else min(existing_value, soc_value) + ) + return + soc_events.append(soc_event) + + +def _projected_soc_events_or_original( + original_soc_events: SocSpecification, + projected_soc_events: TimedEventList, + field_name: str, +) -> SocSpecification: + """Choose between the projected event list and the original specification. + + Only list-based (or missing) specifications can absorb projected bounds; + sensors, series and fixed quantities are returned unchanged. If projected + bounds had to be dropped as a result, a warning is logged. + """ + if isinstance(original_soc_events, list): + return projected_soc_events + if original_soc_events is None and projected_soc_events: + return projected_soc_events + if projected_soc_events: + logger.warning( + f"Dropping {len(projected_soc_events)} projected SoC bound(s): " + f"the '{field_name}' field is not given as a list of timed events, " + f"so projected bounds cannot be merged into it." + ) + return original_soc_events + + +@dataclass +class _SocProjection: + """Working state for projecting the off-tick SoC events of one device. + + Bundles the device's capacities, efficiencies and global SoC limits, and + accumulates the projected lower and upper bounds while rules are applied. + """ + + consumption_capacity: pd.Series + production_capacity: pd.Series + resolution: timedelta + soc_min: float | None + soc_max: float | None + charging_efficiency: pd.Series | ur.Quantity | float | None = None + discharging_efficiency: pd.Series | ur.Quantity | float | None = None + minima: TimedEventList = field(default_factory=list) + maxima: TimedEventList = field(default_factory=list) + + def reachable_energy( + self, charging: bool, start: pd.Timestamp, end: pd.Timestamp + ) -> float: + """The energy (in MWh) the stock can move up (charging) or down (discharging).""" + if charging: + return _reachable_energy( + self.consumption_capacity, + start, + end, + self.resolution, + efficiency=self.charging_efficiency, + efficiency_affects_stock="multiply", + ) + return _reachable_energy( + self.production_capacity, + start, + end, + self.resolution, + efficiency=self.discharging_efficiency, + efficiency_affects_stock="divide", + ) + + def apply_rule( + self, + rule: SocProjectionRule, + soc_event: dict[str, datetime | float], + event_time: pd.Timestamp, + ) -> None: + """Apply one projection rule to one off-tick SoC event. + + Computes the reachability-adjusted bound value, clamps it to the global + SoC limits, and merges it into the projected minima or maxima (keeping + the stricter bound if one already exists on the same tick). + """ + previous_tick = event_time.floor(self.resolution) + next_tick = event_time.ceil(self.resolution) + if rule.tick == "previous": + tick, period = previous_tick, (previous_tick, event_time) + else: + tick, period = next_tick, (event_time, next_tick) + value = _soc_value_in_mwh( + soc_event["value"] + ) + rule.sign * self.reachable_energy(rule.uses_charging, *period) + if rule.bound_type == "min": + if self.soc_min is not None: + value = max(self.soc_min, value) + _add_soc_bound(self.minima, _soc_event_at(soc_event, tick, value), "min") + else: + if self.soc_max is not None: + value = min(self.soc_max, value) + _add_soc_bound(self.maxima, _soc_event_at(soc_event, tick, value), "max") + + def apply_policy( + self, + field_name: str, + soc_event: dict[str, datetime | float], + event_time: pd.Timestamp, + ) -> None: + """Apply all projection rules of the given policy to one off-tick SoC event.""" + for rule in SOC_PROJECTION_POLICIES[field_name]: + self.apply_rule(rule, soc_event, event_time) + + +def _is_projectable( + soc_event: dict[str, datetime | float], resolution: timedelta +) -> bool: + """Whether the SoC event is point-like and falls between two scheduling ticks.""" + return soc_event["start"] == soc_event["end"] and not is_on_schedule_tick( + soc_event["end"], resolution + ) + + +def project_off_tick_soc_constraints( + soc_targets: SocSpecification, + soc_maxima: SocSpecification, + soc_minima: SocSpecification, + consumption_capacity: pd.Series, + production_capacity: pd.Series, + resolution: timedelta, + soc_min: ur.Quantity | float | None, + soc_max: ur.Quantity | float | None, + charging_efficiency: pd.Series | ur.Quantity | float | None = None, + discharging_efficiency: pd.Series | ur.Quantity | float | None = None, +) -> tuple[SocSpecification, SocSpecification, SocSpecification]: + """Project off-tick point-like SoC constraints onto scheduling ticks. + + The scheduler can only enforce constraints at its fixed scheduling resolution. + Point-like ``soc-targets``, ``soc-minima`` and ``soc-maxima`` that fall between + two scheduling ticks are therefore replaced by constraints on the previous and + next tick that preserve reachability using the available charge and discharge + capacity between the original event time and those ticks + (see :data:`SOC_PROJECTION_POLICIES`). + + For an off-tick event with value ``v`` at time ``t``, between previous tick ``p`` + and next tick ``n``: + + - ``soc-targets`` become an exact target ``v`` on ``n``, plus bounds on ``p`` that + keep the target reachable at ``t``: a lower bound of ``v`` minus the energy that + can still be charged between ``p`` and ``t``, and an upper bound of ``v`` plus + the energy that can still be discharged between ``p`` and ``t``. + - ``soc-minima`` become lower bounds on both surrounding ticks: on ``p``, ``v`` + minus the energy that can be charged between ``p`` and ``t``; on ``n``, ``v`` + minus the energy that can be discharged between ``t`` and ``n``. + - ``soc-maxima`` become upper bounds on both surrounding ticks: on ``p``, ``v`` + plus the energy that can be discharged between ``p`` and ``t``; on ``n``, ``v`` + plus the energy that can be charged between ``t`` and ``n``. + + The reachable energy accounts for the (dis)charging efficiencies: charging at + grid power P moves the stock at rate P * charging_efficiency (which can exceed + 1, e.g. a heat pump's COP), and discharging at grid power P moves the stock at + rate P / discharging_efficiency. + + If multiple projected bounds land on the same tick, the stricter lower or upper + bound is kept. Projected bounds are clamped to the global ``soc-min``/``soc-max``. + + Returns ``(soc_targets, soc_maxima, soc_minima)`` with projected list-based + timed events. Non-list specifications such as sensors, series, fixed + quantities, or ``None`` are returned unchanged unless projected bounds need to + be added to a missing list. + """ + + if not any( + isinstance(soc_events, list) and soc_events + for soc_events in (soc_targets, soc_maxima, soc_minima) + ): + return soc_targets, soc_maxima, soc_minima + + projection = _SocProjection( + consumption_capacity=consumption_capacity, + production_capacity=production_capacity, + resolution=resolution, + soc_min=_optional_soc_value_in_mwh(soc_min), + soc_max=_optional_soc_value_in_mwh(soc_max), + charging_efficiency=charging_efficiency, + discharging_efficiency=discharging_efficiency, + minima=copy.deepcopy(soc_minima) if isinstance(soc_minima, list) else [], + maxima=copy.deepcopy(soc_maxima) if isinstance(soc_maxima, list) else [], + ) + + if isinstance(soc_targets, list): + projected_targets = [] + for soc_target in soc_targets: + if not _is_projectable(soc_target, resolution): + projected_targets.append(copy.copy(soc_target)) + continue + target_time = pd.Timestamp(soc_target["end"]) + # Exact target on the next tick, plus previous-tick bounds that keep + # the target reachable at the original (off-tick) target time. + projected_targets.append( + _soc_event_at( + soc_target, + target_time.ceil(resolution), + _soc_value_in_mwh(soc_target["value"]), + ) + ) + projection.apply_policy("soc-targets", soc_target, target_time) + else: + projected_targets = soc_targets + + for field_name, soc_events in ( + ("soc-minima", soc_minima), + ("soc-maxima", soc_maxima), + ): + if not isinstance(soc_events, list): + continue + for soc_event in copy.deepcopy(soc_events): + if _is_projectable(soc_event, resolution): + projection.apply_policy( + field_name, soc_event, pd.Timestamp(soc_event["end"]) + ) + + return ( + projected_targets, + _projected_soc_events_or_original(soc_maxima, projection.maxima, "soc-maxima"), + _projected_soc_events_or_original(soc_minima, projection.minima, "soc-minima"), + ) + + +def project_off_tick_soc_at_start( + soc_at_start_time: datetime, + soc_at_start: ur.Quantity | float, + soc_maxima: SocSpecification, + soc_minima: SocSpecification, + schedule_start: datetime, + consumption_capacity: pd.Series, + production_capacity: pd.Series, + resolution: timedelta, + soc_min: ur.Quantity | float | None, + soc_max: ur.Quantity | float | None, + charging_efficiency: pd.Series | ur.Quantity | float | None = None, + discharging_efficiency: pd.Series | ur.Quantity | float | None = None, +) -> tuple[SocSpecification, SocSpecification]: + """Project an off-tick starting state of charge onto the next scheduling tick. + + When the starting SoC is known at a time ``t`` between the schedule start and + the next scheduling tick ``n`` (e.g. because the ``state-of-charge`` field + resolved to a measurement taken at ``t``), the SoC is assumed to hold from the + schedule start until ``t`` (the device is not moving its stock before then), and + the SoC at ``n`` is bounded by how much the device can (dis)charge between ``t`` + and ``n``: + + - an upper bound of ``soc_at_start`` plus the energy chargeable between ``t`` and ``n``, + - a lower bound of ``soc_at_start`` minus the energy dischargeable between ``t`` and ``n``. + + Both bounds are clamped to the global ``soc-min``/``soc-max`` and merged into + the given ``soc-maxima``/``soc-minima`` (the stricter bound wins on collisions). + Known SoC times on a scheduling tick, or outside the first scheduling interval, + leave the bounds unchanged. + + Returns ``(soc_maxima, soc_minima)``. + """ + event_time = pd.Timestamp(soc_at_start_time) + start = pd.Timestamp(schedule_start) + if is_on_schedule_tick(event_time, resolution) or not ( + start < event_time < start + resolution + ): + return soc_maxima, soc_minima + + projection = _SocProjection( + consumption_capacity=consumption_capacity, + production_capacity=production_capacity, + resolution=resolution, + soc_min=_optional_soc_value_in_mwh(soc_min), + soc_max=_optional_soc_value_in_mwh(soc_max), + charging_efficiency=charging_efficiency, + discharging_efficiency=discharging_efficiency, + minima=copy.deepcopy(soc_minima) if isinstance(soc_minima, list) else [], + maxima=copy.deepcopy(soc_maxima) if isinstance(soc_maxima, list) else [], + ) + soc_event = { + "start": event_time.to_pydatetime(), + "end": event_time.to_pydatetime(), + "value": _soc_value_in_mwh(soc_at_start), + } + projection.apply_policy("soc-at-start", soc_event, event_time) + return ( + _projected_soc_events_or_original(soc_maxima, projection.maxima, "soc-maxima"), + _projected_soc_events_or_original(soc_minima, projection.minima, "soc-minima"), + ) diff --git a/flexmeasures/data/models/planning/storage.py b/flexmeasures/data/models/planning/storage.py index e6c936fe97..cbe73d0485 100644 --- a/flexmeasures/data/models/planning/storage.py +++ b/flexmeasures/data/models/planning/storage.py @@ -18,7 +18,10 @@ SchedulerOutputType, StockCommitment, ) -from flexmeasures.data.models.planning.devices import DeviceInventory +from flexmeasures.data.models.planning.devices import ( + DeviceInventory, + _resolve_stock_key, +) from flexmeasures.data.models.planning.linear_optimization import device_scheduler from flexmeasures.data.models.planning.utils import ( add_tiny_price_slope, @@ -38,6 +41,15 @@ MultiSensorFlexModelSchema, SharedSchema, ) +from flexmeasures.data.models.planning.soc_projection import ( + project_off_tick_soc_at_start, + project_off_tick_soc_constraints, +) +from flexmeasures.data.schemas.scheduling.utils import ( + flex_model_has_off_tick_soc_constraints, + get_soc_constraint_resolution, + should_project_off_tick_soc_constraints, +) from flexmeasures.data.schemas.sensors import SensorReference, VariableQuantityField from flexmeasures.data.services.scheduling_result import SchedulingJobResult from flexmeasures.utils.calculations import ( @@ -50,6 +62,7 @@ storage_asset_types = ["one-way_evse", "two-way_evse", "battery", "heat-storage"] + #: Key used to store and retrieve the ``SchedulingJobResult`` in RQ job metadata #: and in the multi-result list returned by ``StorageScheduler.compute()``. SCHEDULING_RESULT_KEY = "scheduling_result" @@ -150,6 +163,8 @@ def _prepare(self, skip_validation: bool = False) -> tuple: # noqa: C901 self.stock_groups = inventory.stock_groups # Soft SoC constraints are attached to their stock (see StockCommitment.stock), # so the solver couples them to the stock group rather than to a device index. + # Off-tick SoC relaxation scoping and starting-SoC projection also track + # stocks (not devices), so look up each device's stock key. device_stock_key = { d: stock_key for stock_key, group_devices in self.stock_groups.items() @@ -207,7 +222,7 @@ def _prepare(self, skip_validation: bool = False) -> tuple: # noqa: C901 # so the device could discharge more energy than its store holds. if soc_at_start[d0] is None: resolved_soc_at_start = self._resolve_stock_soc_at_start( - stock_model, sensor=sensors[d0] + stock_model, sensor=sensors[d0], stock_key=stock_id ) if resolved_soc_at_start is not None: soc_at_start[d0] = resolved_soc_at_start @@ -732,72 +747,6 @@ def device_list_series( as_instantaneous_events=True, resolve_overlaps="max", ) - if ( - self.flex_context.get("soc_minima_breach_price") is not None - and soc_minima[d] is not None - ): - soc_minima_breach_price = self.flex_context["soc_minima_breach_price"] - any_soc_minima_breach_price = get_continuous_series_sensor_or_quantity( - variable_quantity=soc_minima_breach_price, - unit=self.flex_context["shared_currency_unit"] + "/MWh", - query_window=(start + resolution, end + resolution), - resolution=resolution, - beliefs_before=belief_time, - fill_sides=True, - ).shift(-1, freq=resolution) - all_soc_minima_breach_price = get_continuous_series_sensor_or_quantity( - variable_quantity=soc_minima_breach_price, - unit=self.flex_context["shared_currency_unit"] - + "/MWh*h", # from EUR/MWh² to EUR/MWh/resolution - query_window=(start + resolution, end + resolution), - resolution=resolution, - beliefs_before=belief_time, - fill_sides=True, - ).shift(-1, freq=resolution) - # Set up commitments DataFrame - # soc_minima_d is a temp variable because add_storage_constraints can't deal with Series yet - soc_minima_d = get_continuous_series_sensor_or_quantity( - variable_quantity=soc_minima[d], - unit="MWh", - query_window=(start + resolution, end + resolution), - resolution=resolution, - beliefs_before=belief_time, - as_instantaneous_events=True, - resolve_overlaps="max", - ) - # shift soc minima by one resolution (they define a state at a certain time, - # while the commitment defines what the total stock should be at the end of a time slot, - # where the time slot is indexed by its starting time) - soc_minima_d = soc_minima_d.shift(-1, freq=resolution) * ( - timedelta(hours=1) / resolution - ) - soc_at_start[d] * (timedelta(hours=1) / resolution) - - commitment = StockCommitment( - name="any soc minima", - quantity=soc_minima_d, - # negative price because breaching in the downwards (shortage) direction is penalized - downwards_deviation_price=-any_soc_minima_breach_price, - index=index, - _type="any", - device=d, - stock=device_stock_key.get(d), - ) - commitments.append(commitment) - - commitment = StockCommitment( - name="all soc minima", - quantity=soc_minima_d, - # negative price because breaching in the downwards (shortage) direction is penalized - downwards_deviation_price=-all_soc_minima_breach_price, - index=index, - device=d, - stock=device_stock_key.get(d), - ) - commitments.append(commitment) - - # soc-minima will become a soft constraint (modelled as stock commitments), so remove hard constraint - soc_minima[d] = None - if isinstance(soc_maxima[d], (Sensor, SensorReference)): soc_maxima[d] = get_continuous_series_sensor_or_quantity( variable_quantity=soc_maxima[d], @@ -808,95 +757,6 @@ def device_list_series( as_instantaneous_events=True, resolve_overlaps="min", ) - if ( - self.flex_context.get("soc_maxima_breach_price") is not None - and soc_maxima[d] is not None - ): - soc_maxima_breach_price = self.flex_context["soc_maxima_breach_price"] - any_soc_maxima_breach_price = get_continuous_series_sensor_or_quantity( - variable_quantity=soc_maxima_breach_price, - unit=self.flex_context["shared_currency_unit"] + "/MWh", - query_window=(start + resolution, end + resolution), - resolution=resolution, - beliefs_before=belief_time, - fill_sides=True, - ).shift(-1, freq=resolution) - all_soc_maxima_breach_price = get_continuous_series_sensor_or_quantity( - variable_quantity=soc_maxima_breach_price, - unit=self.flex_context["shared_currency_unit"] - + "/MWh*h", # from EUR/MWh² to EUR/MWh/resolution - query_window=(start + resolution, end + resolution), - resolution=resolution, - beliefs_before=belief_time, - fill_sides=True, - ).shift(-1, freq=resolution) - # Set up commitments DataFrame - # soc_maxima_d is a temp variable because add_storage_constraints can't deal with Series yet - soc_maxima_d = get_continuous_series_sensor_or_quantity( - variable_quantity=soc_maxima[d], - unit="MWh", - query_window=(start + resolution, end + resolution), - resolution=resolution, - beliefs_before=belief_time, - as_instantaneous_events=True, - resolve_overlaps="min", - ) - # shift soc maxima by one resolution (they define a state at a certain time, - # while the commitment defines what the total stock should be at the end of a time slot, - # where the time slot is indexed by its starting time) - soc_maxima_d = soc_maxima_d.shift(-1, freq=resolution) * ( - timedelta(hours=1) / resolution - ) - soc_at_start[d] * (timedelta(hours=1) / resolution) - - commitment = StockCommitment( - name="any soc maxima", - quantity=soc_maxima_d, - # positive price because breaching in the upwards (surplus) direction is penalized - upwards_deviation_price=any_soc_maxima_breach_price, - index=index, - _type="any", - device=d, - stock=device_stock_key.get(d), - ) - commitments.append(commitment) - - commitment = StockCommitment( - name="all soc maxima", - quantity=soc_maxima_d, - # positive price because breaching in the upwards (surplus) direction is penalized - upwards_deviation_price=all_soc_maxima_breach_price, - index=index, - device=d, - stock=device_stock_key.get(d), - ) - commitments.append(commitment) - - # soc-maxima will become a soft constraint (modelled as stock commitments), so remove hard constraint - soc_maxima[d] = None - - # only apply SOC constraints to the first device of a shared stock - apply_soc_constraints = True - - for stock_id, devices in self.stock_groups.items(): - if d in devices and d != devices[0]: - apply_soc_constraints = False - break - - if soc_at_start[d] is not None and apply_soc_constraints: - device_constraints[d] = add_storage_constraints( - start, - end, - resolution, - soc_at_start[d], - soc_targets[d], - soc_maxima[d], - soc_minima[d], - soc_max[d], - soc_min[d], - ) - else: - # No need to validate non-existing storage constraints - skip_validation = True power_capacity_in_mw[d] = get_continuous_series_sensor_or_quantity( variable_quantity=power_capacity_in_mw[d], @@ -913,6 +773,9 @@ def device_list_series( if sensor_d is not None and sensor_d.get_attribute( "is_strictly_non_positive" ): + production_capacity_d = pd.Series( + 0, index=power_capacity_in_mw[d].index + ) device_constraints[d]["derivative min"] = 0 else: production_capacity_d = get_continuous_series_sensor_or_quantity( @@ -981,6 +844,9 @@ def device_list_series( if sensor_d is not None and sensor_d.get_attribute( "is_strictly_non_negative" ): + consumption_capacity_d = pd.Series( + 0, index=power_capacity_in_mw[d].index + ) device_constraints[d]["derivative max"] = 0 else: consumption_capacity_d = get_continuous_series_sensor_or_quantity( @@ -1045,38 +911,6 @@ def device_list_series( # consumption-capacity will become a hard constraint device_constraints[d]["derivative max"] = consumption_capacity_d - all_stock_delta = [] - - for is_usage, soc_delta in zip([False, True], [soc_gain[d], soc_usage[d]]): - if soc_delta is None: - # Try to get fallback - soc_delta = [None] - - for component in soc_delta: - stock_delta_series = get_continuous_series_sensor_or_quantity( - variable_quantity=component, - unit="MW", - query_window=(start, end), - resolution=resolution, - beliefs_before=belief_time, - ) - - # example: 4 MW sustained over 15 minutes gives 1 MWh - stock_delta_series *= resolution / timedelta( - hours=1 - ) # MW -> MWh / resolution - - if is_usage: - stock_delta_series *= -1 - - all_stock_delta.append(stock_delta_series) - - if len(all_stock_delta) > 0: - all_stock_delta = pd.concat(all_stock_delta, axis=1) - - device_constraints[d]["stock delta"] = all_stock_delta.sum(1) - device_constraints[d]["stock delta"] *= timedelta(hours=1) / resolution - # Apply round-trip efficiency evenly to charging and discharging charging_efficiency[d] = ( get_continuous_series_sensor_or_quantity( @@ -1114,6 +948,241 @@ def device_list_series( charging_efficiency[d] = roundtrip_efficiency**0.5 discharging_efficiency[d] = roundtrip_efficiency**0.5 + # Project off-tick point-like SoC constraints onto the scheduling ticks + # before they are turned into soft commitments or hard constraints, + # so that both paths consume on-tick events. + if should_project_off_tick_soc_constraints(sensor_d): + # A starting SoC known at an off-tick time within the first + # scheduling interval bounds the SoC on the next tick. The timing + # is tracked per stock; a None key covers the single-sensor case + # where the stock key cannot be resolved at record time. + soc_at_start_datetimes = getattr(self, "soc_at_start_datetimes", {}) + soc_at_start_time = soc_at_start_datetimes.get(device_stock_key.get(d)) + if soc_at_start_time is None: + soc_at_start_time = soc_at_start_datetimes.get(None) + if soc_at_start_time is not None and soc_at_start[d] is not None: + soc_maxima[d], soc_minima[d] = project_off_tick_soc_at_start( + soc_at_start_time, + soc_at_start[d], + soc_maxima[d], + soc_minima[d], + start, + consumption_capacity_d, + production_capacity_d, + resolution, + soc_min[d], + soc_max[d], + charging_efficiency=charging_efficiency[d], + discharging_efficiency=discharging_efficiency[d], + ) + ( + soc_targets[d], + soc_maxima[d], + soc_minima[d], + ) = project_off_tick_soc_constraints( + soc_targets[d], + soc_maxima[d], + soc_minima[d], + consumption_capacity_d, + production_capacity_d, + resolution, + soc_min[d], + soc_max[d], + charging_efficiency=charging_efficiency[d], + discharging_efficiency=discharging_efficiency[d], + ) + + if ( + self.flex_context.get("soc_minima_breach_price") is not None + and soc_minima[d] is not None + and self._soc_relaxation_applies_to(device_stock_key.get(d), sensor_d) + ): + soc_minima_breach_price = self.flex_context["soc_minima_breach_price"] + any_soc_minima_breach_price = get_continuous_series_sensor_or_quantity( + variable_quantity=soc_minima_breach_price, + unit=self.flex_context["shared_currency_unit"] + "/MWh", + query_window=(start + resolution, end + resolution), + resolution=resolution, + beliefs_before=belief_time, + fill_sides=True, + ).shift(-1, freq=resolution) + all_soc_minima_breach_price = get_continuous_series_sensor_or_quantity( + variable_quantity=soc_minima_breach_price, + unit=self.flex_context["shared_currency_unit"] + + "/MWh*h", # from EUR/MWh² to EUR/MWh/resolution + query_window=(start + resolution, end + resolution), + resolution=resolution, + beliefs_before=belief_time, + fill_sides=True, + ).shift(-1, freq=resolution) + # Set up commitments DataFrame + # soc_minima_d is a temp variable because add_storage_constraints can't deal with Series yet + soc_minima_d = get_continuous_series_sensor_or_quantity( + variable_quantity=soc_minima[d], + unit="MWh", + query_window=(start + resolution, end + resolution), + resolution=resolution, + beliefs_before=belief_time, + as_instantaneous_events=True, + resolve_overlaps="max", + ) + # shift soc minima by one resolution (they define a state at a certain time, + # while the commitment defines what the total stock should be at the end of a time slot, + # where the time slot is indexed by its starting time) + soc_minima_d = soc_minima_d.shift(-1, freq=resolution) * ( + timedelta(hours=1) / resolution + ) - soc_at_start[d] * (timedelta(hours=1) / resolution) + + commitment = StockCommitment( + name="any soc minima", + quantity=soc_minima_d, + # negative price because breaching in the downwards (shortage) direction is penalized + downwards_deviation_price=-any_soc_minima_breach_price, + index=index, + _type="any", + device=d, + stock=device_stock_key.get(d), + ) + commitments.append(commitment) + + commitment = StockCommitment( + name="all soc minima", + quantity=soc_minima_d, + # negative price because breaching in the downwards (shortage) direction is penalized + downwards_deviation_price=-all_soc_minima_breach_price, + index=index, + device=d, + stock=device_stock_key.get(d), + ) + commitments.append(commitment) + + # soc-minima will become a soft constraint (modelled as stock commitments), so remove hard constraint + soc_minima[d] = None + + if ( + self.flex_context.get("soc_maxima_breach_price") is not None + and soc_maxima[d] is not None + and self._soc_relaxation_applies_to(device_stock_key.get(d), sensor_d) + ): + soc_maxima_breach_price = self.flex_context["soc_maxima_breach_price"] + any_soc_maxima_breach_price = get_continuous_series_sensor_or_quantity( + variable_quantity=soc_maxima_breach_price, + unit=self.flex_context["shared_currency_unit"] + "/MWh", + query_window=(start + resolution, end + resolution), + resolution=resolution, + beliefs_before=belief_time, + fill_sides=True, + ).shift(-1, freq=resolution) + all_soc_maxima_breach_price = get_continuous_series_sensor_or_quantity( + variable_quantity=soc_maxima_breach_price, + unit=self.flex_context["shared_currency_unit"] + + "/MWh*h", # from EUR/MWh² to EUR/MWh/resolution + query_window=(start + resolution, end + resolution), + resolution=resolution, + beliefs_before=belief_time, + fill_sides=True, + ).shift(-1, freq=resolution) + # Set up commitments DataFrame + # soc_maxima_d is a temp variable because add_storage_constraints can't deal with Series yet + soc_maxima_d = get_continuous_series_sensor_or_quantity( + variable_quantity=soc_maxima[d], + unit="MWh", + query_window=(start + resolution, end + resolution), + resolution=resolution, + beliefs_before=belief_time, + as_instantaneous_events=True, + resolve_overlaps="min", + ) + # shift soc maxima by one resolution (they define a state at a certain time, + # while the commitment defines what the total stock should be at the end of a time slot, + # where the time slot is indexed by its starting time) + soc_maxima_d = soc_maxima_d.shift(-1, freq=resolution) * ( + timedelta(hours=1) / resolution + ) - soc_at_start[d] * (timedelta(hours=1) / resolution) + + commitment = StockCommitment( + name="any soc maxima", + quantity=soc_maxima_d, + # positive price because breaching in the upwards (surplus) direction is penalized + upwards_deviation_price=any_soc_maxima_breach_price, + index=index, + _type="any", + device=d, + stock=device_stock_key.get(d), + ) + commitments.append(commitment) + + commitment = StockCommitment( + name="all soc maxima", + quantity=soc_maxima_d, + # positive price because breaching in the upwards (surplus) direction is penalized + upwards_deviation_price=all_soc_maxima_breach_price, + index=index, + device=d, + stock=device_stock_key.get(d), + ) + commitments.append(commitment) + + # soc-maxima will become a soft constraint (modelled as stock commitments), so remove hard constraint + soc_maxima[d] = None + + # only apply SOC constraints to the first device of a shared stock + apply_soc_constraints = True + for stock_id, devices in self.stock_groups.items(): + if d in devices and d != devices[0]: + apply_soc_constraints = False + break + + if soc_at_start[d] is not None and apply_soc_constraints: + storage_constraints = add_storage_constraints( + start, + end, + resolution, + soc_at_start[d], + soc_targets[d], + soc_maxima[d], + soc_minima[d], + soc_max[d], + soc_min[d], + ) + for column in ("equals", "min", "max"): + device_constraints[d][column] = storage_constraints[column] + else: + # No need to validate non-existing storage constraints + skip_validation = True + + all_stock_delta = [] + + for is_usage, soc_delta in zip([False, True], [soc_gain[d], soc_usage[d]]): + if soc_delta is None: + # Try to get fallback + soc_delta = [None] + + for component in soc_delta: + stock_delta_series = get_continuous_series_sensor_or_quantity( + variable_quantity=component, + unit="MW", + query_window=(start, end), + resolution=resolution, + beliefs_before=belief_time, + ) + + # example: 4 MW sustained over 15 minutes gives 1 MWh + stock_delta_series *= resolution / timedelta( + hours=1 + ) # MW -> MWh / resolution + + if is_usage: + stock_delta_series *= -1 + + all_stock_delta.append(stock_delta_series) + + if len(all_stock_delta) > 0: + all_stock_delta = pd.concat(all_stock_delta, axis=1) + + device_constraints[d]["stock delta"] = all_stock_delta.sum(1) + device_constraints[d]["stock delta"] *= timedelta(hours=1) / resolution + device_constraints[d]["derivative down efficiency"] = ( discharging_efficiency[d] ) @@ -1336,8 +1405,19 @@ def deserialize_flex_config(self): self.flex_model = {} self.collect_flex_config() - self._deserialize_flex_context() + if self.flex_context is None: + self.flex_context = {} + #: Stock keys of stocks whose off-tick SoC constraints triggered automatic + #: relaxation (None marks an entry whose stock key cannot be resolved from + #: its serialized form, e.g. a state-of-charge time series). + self.off_tick_stock_keys: set = set() + #: Whether SoC constraint softening should apply only to those devices + #: (True when relaxation was enabled purely because of off-tick projection). + self.scope_soc_relaxation_to_off_tick_devices: bool = False + # The flex-model is deserialized first, because off-tick SoC constraints + # may enable relax-soc-constraints on the still-serialized flex-context. self._deserialize_flex_model() + self._deserialize_flex_context() # Classify all flex-model entries (and the flex-context's inflexible devices) # once; scheduling and result mapping rely on this inventory for device @@ -1407,12 +1487,17 @@ def _deserialize_flex_model(self): if self.sensor.generic_asset.asset_type.name in storage_asset_types: self.ensure_soc_at_start() + self._possibly_relax_off_tick_soc_constraints( + self.flex_model, sensor=self.sensor, power_sensor=self.sensor + ) + # Now it's time to check if our flex configuration holds up to schemas - self.flex_model = StorageFlexModelSchema( + schema = StorageFlexModelSchema( start=self.start, sensor=self.sensor, default_soc_unit=self.flex_model.get("soc-unit"), - ).load(self.flex_model) + ) + self.flex_model = schema.load(self.flex_model) # Extend schedule period in case a target exceeds its end self.possibly_extend_end(soc_targets=self.flex_model.get("soc_targets")) @@ -1429,17 +1514,24 @@ def _deserialize_flex_model(self): soc_sensor = None if soc_sensor_id is not None: soc_sensor = Sensor.query.filter_by(id=soc_sensor_id).first() - self.flex_model[d] = StorageFlexModelSchema( + sensor_d = ( + sensor_flex_model.get("sensor") + if sensor_flex_model.get("sensor") is not None + else soc_sensor + ) + self._possibly_relax_off_tick_soc_constraints( + sensor_flex_model["sensor_flex_model"], + sensor=sensor_d, + power_sensor=sensor_flex_model.get("sensor"), + ) + schema = StorageFlexModelSchema( start=self.start, - sensor=( - sensor_flex_model.get("sensor") - if sensor_flex_model.get("sensor") is not None - else soc_sensor - ), + sensor=sensor_d, default_soc_unit=sensor_flex_model["sensor_flex_model"].get( "soc-unit" ), - ).load(sensor_flex_model["sensor_flex_model"]) + ) + self.flex_model[d] = schema.load(sensor_flex_model["sensor_flex_model"]) self.flex_model[d]["sensor"] = sensor_flex_model.get("sensor") self.flex_model[d]["asset"] = sensor_flex_model.get("asset") @@ -1455,6 +1547,127 @@ def _deserialize_flex_model(self): return self.flex_model + def _possibly_relax_off_tick_soc_constraints( + self, + flex_model: dict, + sensor: Sensor | None, + power_sensor: Sensor | None = None, + ) -> None: + """Enable SoC constraint relaxation if the (serialized) flex-model contains off-tick SoC events. + + The detection uses the scheduler's actual resolution (falling back to the + sensor's event resolution), matching the resolution later used to project + off-tick SoC constraints onto the scheduling ticks. + + When relaxation is enabled purely because of off-tick projection (rather + than by the user's own flex-context settings), softening is scoped to the + stocks that actually use off-tick SoC constraints (tracked here by their + stock key, so all devices sharing the stock are covered). An entry without + a resolvable stock key is tracked by its power sensor instead (its stock + gets a synthetic key only later, when the device inventory is built). + """ + if not should_project_off_tick_soc_constraints(sensor): + return + resolution = get_soc_constraint_resolution( + self.resolution, sensor, self.default_resolution + ) + if flex_model_has_off_tick_soc_constraints(flex_model, resolution=resolution): + stock_key = _resolve_stock_key(flex_model.get("state-of-charge")) + if stock_key is None and power_sensor is not None: + stock_key = ("sensor", power_sensor.id) + self.off_tick_stock_keys.add(stock_key) + self.scope_soc_relaxation_to_off_tick_devices = ( + not self._soc_relaxation_user_enabled() + ) + self.enable_relax_soc_constraints() + + def _soc_relaxation_user_enabled(self) -> bool: + """Whether the user's own (serialized) flex-context already softens SoC constraints. + + That is the case when any context defines a SoC breach price explicitly, + sets ``relax-soc-constraints`` to ``True``, or leaves relaxation to the + general ``relax-constraints`` flag (which defaults to ``True``). + """ + if isinstance(self.flex_context, dict): + contexts = [self.flex_context] + list( + self.flex_context.get("commodities", []) + ) + elif isinstance(self.flex_context, list): + contexts = self.flex_context + else: + return True + for context in contexts: + if ( + context.get("soc-minima-breach-price") is not None + or context.get("soc-maxima-breach-price") is not None + ): + return True + if context.get("relax-soc-constraints") is True: + return True + if context.get("relax-soc-constraints") is None and context.get( + "relax-constraints", True + ): + return True + return False + + def _soc_relaxation_applies_to( + self, stock_key, sensor_d: Sensor | None = None + ) -> bool: + """Whether SoC constraint softening applies to the stock with this key. + + Softening applies to all stocks, unless relaxation was auto-enabled purely + for off-tick SoC constraint projection, in which case it is scoped to the + stocks that use off-tick SoC constraints (covering all devices sharing them). + A stock tracked by power sensor (for lack of a resolvable stock key) is + matched via the device's power sensor. + """ + if not getattr(self, "scope_soc_relaxation_to_off_tick_devices", False): + return True + off_tick_stock_keys = getattr(self, "off_tick_stock_keys", set()) + if None in off_tick_stock_keys: + # An entry with neither a resolvable stock key nor a power sensor + # cannot be matched to a stock. + return True + if stock_key is not None and stock_key in off_tick_stock_keys: + return True + return sensor_d is not None and ("sensor", sensor_d.id) in off_tick_stock_keys + + def enable_relax_soc_constraints(self) -> None: + """Relax SoC constraints when off-tick SoC events require scheduling-tick projection. + + Projection can add bounds (and stricter combinations of bounds), which could + render the problem infeasible if they remain hard constraints. Therefore, + ``relax-soc-constraints`` is enabled unless the user explicitly disabled it, + in which case we respect that choice and only log a warning. + """ + + def _enable(context: dict) -> None: + if context.get("relax-soc-constraints") is False: + current_app.logger.warning( + "Off-tick SoC constraints are projected onto the scheduling ticks, " + "which can add bounds that render the scheduling problem infeasible, " + "but 'relax-soc-constraints' is explicitly disabled. " + "Keeping SoC constraints hard." + ) + return + if context.get("relax-soc-constraints") is not True: + current_app.logger.info( + "Enabling 'relax-soc-constraints' because off-tick SoC constraints " + "are projected onto the scheduling ticks." + ) + context["relax-soc-constraints"] = True + + if self.flex_context is None: + self.flex_context = {} + if isinstance(self.flex_context, dict): + _enable(self.flex_context) + for commodity_context in self.flex_context.get("commodities", []): + _enable(commodity_context) + return + if isinstance(self.flex_context, list): + for commodity_context in self.flex_context: + _enable(commodity_context) + def has_soc_at_start(self) -> bool: return ( "soc-at-start" in self.flex_model @@ -1465,6 +1678,22 @@ def has_soc_at_start(self) -> bool: def has_soc_at_start_in(flex_model: dict) -> bool: return "soc-at-start" in flex_model and flex_model["soc-at-start"] is not None + def _record_soc_at_start_datetime( + self, stock_key, soc_datetime: datetime | None + ) -> None: + """Remember at which time a stock's starting state of charge is actually known. + + Keyed by the stock key (a None key covers the single-sensor case where the + stock key cannot be resolved, e.g. a state-of-charge time series). Used to + project an off-tick starting SoC onto the next scheduling tick (see + :func:`flexmeasures.data.models.planning.soc_projection.project_off_tick_soc_at_start`). + """ + if soc_datetime is None: + return + if not hasattr(self, "soc_at_start_datetimes"): + self.soc_at_start_datetimes: dict = {} + self.soc_at_start_datetimes[stock_key] = soc_datetime + def _get_soc_lookup_radius( self, sensor: Sensor | None = None, slack_steps: int = 4 ) -> timedelta: @@ -1590,6 +1819,7 @@ def _resolve_soc_at_start_from_sensor( beliefs_df["time_distance"] == beliefs_df["time_distance"].min() ] nearest_belief = nearest_beliefs.loc[nearest_beliefs["event_start"].idxmax()] + self._record_soc_at_start_datetime(soc_sensor.id, nearest_belief["event_start"]) return self._convert_soc_value_to_mwh( value=nearest_belief["event_value"], @@ -1599,12 +1829,14 @@ def _resolve_soc_at_start_from_sensor( ) def _resolve_soc_at_start_from_time_series( - self, soc_time_series: list[dict], sensor: Sensor | None = None + self, soc_time_series: list[dict], sensor: Sensor | None = None, stock_key=None ) -> float: """Resolve ``soc-at-start`` from a ``state-of-charge`` time series. :param soc_time_series: SoC time series specification. :param sensor: Optional scheduled power sensor. + :param stock_key: Key of the stock whose SoC is being resolved, if known + (a time series does not resolve to a stock key by itself). :returns: Starting SoC in MWh. """ lookup_radius = self._get_soc_lookup_radius(sensor) @@ -1642,6 +1874,7 @@ def _resolve_soc_at_start_from_time_series( ) _, nearest_segment = min(candidate_segments, key=lambda item: item[0]) + self._record_soc_at_start_datetime(stock_key, nearest_segment["start"]) return (nearest_segment["value"] / ur.Quantity("MWh")).magnitude def _resolve_soc_at_start_from_state_of_charge( @@ -1693,7 +1926,7 @@ def _resolve_soc_at_start_from_state_of_charge( return None def _resolve_stock_soc_at_start( - self, stock_model: dict, sensor: Sensor | None = None + self, stock_model: dict, sensor: Sensor | None = None, stock_key=None ) -> float | None: """Resolve a stock's soc-at-start (in MWh) from its (deserialized) state-of-charge. @@ -1721,7 +1954,9 @@ def _resolve_stock_soc_at_start( state_of_charge, percent_conversion_model, sensor ) if isinstance(state_of_charge, list): - return self._resolve_soc_at_start_from_time_series(state_of_charge, sensor) + return self._resolve_soc_at_start_from_time_series( + state_of_charge, sensor, stock_key=stock_key + ) return None def possibly_extend_end(self, soc_targets, sensor: Sensor = None): @@ -1738,6 +1973,15 @@ def possibly_extend_end(self, soc_targets, sensor: Sensor = None): if soc_targets and not isinstance(soc_targets, (Sensor, SensorReference)): max_target_datetime = max([soc_target["end"] for soc_target in soc_targets]) + # Off-tick target times are preserved during deserialization, and their + # projection moves the target to the next scheduling tick. Ceil to the + # scheduling resolution, so the projected target still falls within the + # schedule (instead of being disregarded as beyond its end). + resolution = self.resolution or sensor.event_resolution + if resolution not in (None, timedelta(0)): + max_target_datetime = ( + pd.Timestamp(max_target_datetime).ceil(resolution).to_pydatetime() + ) if max_target_datetime > self.end: max_server_horizon = get_max_planning_horizon(sensor.event_resolution) if max_server_horizon: @@ -2880,6 +3124,20 @@ def build_device_soc_values( device_values.loc[soc_constraint_start:end_of_schedule] = soc continue + if ( + soc_constraint_start == soc_constraint_end + and soc_constraint_start not in device_values.index + ): + # Point-like events between scheduling ticks match no index entry. + # This can happen when off-tick projection is disabled through the + # sensor's floor_datetimes_to_resolution attribute. + current_app.logger.warning( + f"Disregarding off-tick SoC constraint at {soc_constraint_start} " + f"(value: {soc}): it does not fall on the scheduling ticks and " + f"off-tick projection is disabled for this sensor." + ) + continue + device_values.loc[soc_constraint_start:soc_constraint_end] = soc if not disregarded_periods: diff --git a/flexmeasures/data/models/planning/tests/test_storage.py b/flexmeasures/data/models/planning/tests/test_storage.py index a03d09e05d..9435e2c3db 100644 --- a/flexmeasures/data/models/planning/tests/test_storage.py +++ b/flexmeasures/data/models/planning/tests/test_storage.py @@ -8,6 +8,11 @@ import pandas as pd from flexmeasures.data.models.planning import Scheduler +from flexmeasures.data.models.planning.soc_projection import ( + project_off_tick_soc_at_start, + project_off_tick_soc_constraints, +) +from flexmeasures.data.models.generic_assets import GenericAsset from flexmeasures.data.models.planning.storage import StorageScheduler from flexmeasures.data.models.planning.utils import initialize_index from flexmeasures.data.models.time_series import Sensor, TimedBelief @@ -934,6 +939,741 @@ def spy( assert entry_most_relevant_only["soc-minima"][0] == expected_tightest +def test_off_tick_soc_target_is_projected_to_scheduling_ticks(add_battery_assets, db): + """Off-tick targets become a next-tick target and a reachable previous-tick bound.""" + _, battery = get_sensors_from_db( + db, add_battery_assets, battery_name="Test battery" + ) + tz = pytz.timezone("Europe/Amsterdam") + start = tz.localize(datetime(2015, 1, 1, 16, 45)) + end = tz.localize(datetime(2015, 1, 1, 17, 15)) + resolution = timedelta(minutes=15) + + scheduler = StorageScheduler( + battery, + start, + end, + resolution, + flex_model={ + "soc-at-start": "0 MWh", + "soc-min": "0 MWh", + "soc-max": "1 MWh", + "power-capacity": "0.04 MW", + "consumption-capacity": "0.04 MW", + "production-capacity": "0 MW", + "roundtrip-efficiency": 1, + "storage-efficiency": 1, + "soc-targets": [ + { + "datetime": "2015-01-01T17:12:00+01:00", + "value": "1 MWh", + } + ], + }, + flex_context={ + "consumption-price": "0 EUR/MWh", + "production-price": "0 EUR/MWh", + "site-power-capacity": "1 MW", + # keep SoC constraints hard, so we can assert the projected bounds directly + "relax-constraints": False, + "relax-soc-constraints": False, + }, + ) + + _, _, _, _, _, device_constraints, _, _ = scheduler._prepare(skip_validation=True) + storage_constraints = device_constraints[0].tz_convert(tz) + + assert pd.isna( + storage_constraints.loc[start, "equals"] + ), "off-tick targets should not become exact constraints on the previous tick" + assert storage_constraints.loc[start, "min"] == pytest.approx( + 0.992 * 4 + ), "previous tick should allow charging the missing 0.008 MWh before the target time" + assert storage_constraints.loc[start + resolution, "equals"] == pytest.approx( + 4 + ), "next tick should carry the projected exact target" + + +def test_off_tick_soc_target_is_projected_for_instantaneous_sensor( + add_battery_assets, db +): + """Off-tick projection also applies when the scheduled sensor is instantaneous.""" + _, battery = get_sensors_from_db( + db, add_battery_assets, battery_name="Test battery" + ) + instantaneous_power_sensor = Sensor( + name="instantaneous-power", + generic_asset=battery.generic_asset, + event_resolution=timedelta(0), + unit="MW", + ) + db.session.add(instantaneous_power_sensor) + db.session.flush() + tz = pytz.timezone("Europe/Amsterdam") + start = tz.localize(datetime(2015, 1, 1, 16, 45)) + end = tz.localize(datetime(2015, 1, 1, 17, 15)) + resolution = timedelta(minutes=15) + + scheduler = StorageScheduler( + instantaneous_power_sensor, + start, + end, + resolution, + flex_model={ + "soc-at-start": "0 MWh", + "soc-min": "0 MWh", + "soc-max": "1 MWh", + "power-capacity": "0.04 MW", + "consumption-capacity": "0.04 MW", + "production-capacity": "0 MW", + "roundtrip-efficiency": 1, + "storage-efficiency": 1, + "soc-targets": [ + { + "datetime": "2015-01-01T17:12:00+01:00", + "value": "1 MWh", + } + ], + }, + flex_context={ + "consumption-price": "0 EUR/MWh", + "production-price": "0 EUR/MWh", + "site-power-capacity": "1 MW", + # keep SoC constraints hard, so we can assert the projected bounds directly + "relax-constraints": False, + "relax-soc-constraints": False, + }, + ) + + _, _, _, _, _, device_constraints, _, _ = scheduler._prepare(skip_validation=True) + storage_constraints = device_constraints[0].tz_convert(tz) + + assert storage_constraints.loc[start, "min"] == pytest.approx( + 0.992 * 4 + ), "instantaneous sensors should still project the previous-tick minimum" + assert storage_constraints.loc[start + resolution, "equals"] == pytest.approx( + 4 + ), "instantaneous sensors should still project the exact target to the next tick" + + +@pytest.mark.parametrize( + "soc_minima, soc_maxima, expected_previous_value, expected_next_value", + [ + ( + [ + { + "datetime": "2015-01-01T17:12:00+01:00", + "start": "2015-01-01T17:12:00+01:00", + "end": "2015-01-01T17:12:00+01:00", + "value": 1, + } + ], + None, + 0.992, + 1, + ), + ( + None, + [ + { + "datetime": "2015-01-01T17:12:00+01:00", + "start": "2015-01-01T17:12:00+01:00", + "end": "2015-01-01T17:12:00+01:00", + "value": 0.5, + } + ], + 0.5, + 0.502, + ), + ], +) +def test_off_tick_soc_bounds_are_projected_to_scheduling_ticks( + soc_minima, + soc_maxima, + expected_previous_value, + expected_next_value, +): + """Off-tick minima and maxima are projected as reachable bounds on surrounding ticks.""" + tz = pytz.timezone("Europe/Amsterdam") + resolution = timedelta(minutes=15) + previous_tick = pd.Timestamp(tz.localize(datetime(2015, 1, 1, 17))) + next_tick = previous_tick + resolution + capacity = pd.Series( + 0.04, index=pd.date_range(previous_tick, next_tick, freq=resolution) + ) + + _, projected_maxima, projected_minima = project_off_tick_soc_constraints( + soc_targets=None, + soc_maxima=soc_maxima, + soc_minima=soc_minima, + consumption_capacity=capacity, + production_capacity=pd.Series(0, index=capacity.index), + resolution=resolution, + soc_min=0, + soc_max=1, + ) + + projected_events = projected_minima or projected_maxima + + assert _soc_event_value_at(projected_events, previous_tick) == pytest.approx( + expected_previous_value + ), "previous tick should use the capacity-adjusted projected SoC bound" + assert _soc_event_value_at(projected_events, next_tick) == pytest.approx( + expected_next_value + ), "next tick should use the projected SoC bound implied by reachability" + + +def test_off_tick_soc_projection_accepts_missing_global_bounds(): + """Missing global SoC bounds leave projected off-tick bounds unclamped.""" + tz = pytz.timezone("Europe/Amsterdam") + resolution = timedelta(minutes=15) + previous_tick = pd.Timestamp(tz.localize(datetime(2015, 1, 1, 17))) + next_tick = previous_tick + resolution + capacity = pd.Series( + 0.04, index=pd.date_range(previous_tick, next_tick, freq=resolution) + ) + + _, projected_maxima, projected_minima = project_off_tick_soc_constraints( + soc_targets=[ + { + "datetime": "2015-01-01T17:12:00+01:00", + "start": "2015-01-01T17:12:00+01:00", + "end": "2015-01-01T17:12:00+01:00", + "value": 0.5, + } + ], + soc_maxima=None, + soc_minima=None, + consumption_capacity=capacity, + production_capacity=capacity, + resolution=resolution, + soc_min=None, + soc_max=None, + ) + + assert _soc_event_value_at(projected_minima, previous_tick) == pytest.approx( + 0.492 + ), "missing global soc-min should not clamp the projected previous-tick minimum" + assert _soc_event_value_at(projected_maxima, previous_tick) == pytest.approx( + 0.508 + ), "missing global soc-max should not clamp the projected previous-tick maximum" + + +def _soc_event_value_at(events, dt): + matches = [ + event + for event in events + if pd.Timestamp(event["start"]) == dt and pd.Timestamp(event["end"]) == dt + ] + assert len(matches) == 1, "projection should create exactly one event per tick" + return matches[0]["value"] + + +def test_off_tick_soc_bounds_are_merged_on_the_same_scheduling_tick(): + """Projected bounds sharing a tick keep the stricter minimum or maximum.""" + tz = pytz.timezone("Europe/Amsterdam") + resolution = timedelta(minutes=15) + previous_tick = pd.Timestamp(tz.localize(datetime(2015, 1, 1, 17))) + next_tick = previous_tick + resolution + capacity = pd.Series( + 0, index=pd.date_range(previous_tick, next_tick, freq=resolution) + ) + + _, projected_maxima, projected_minima = project_off_tick_soc_constraints( + soc_targets=None, + soc_maxima=[ + { + "datetime": tz.localize(datetime(2015, 1, 1, 17, 4)), + "start": tz.localize(datetime(2015, 1, 1, 17, 4)), + "end": tz.localize(datetime(2015, 1, 1, 17, 4)), + "value": 0.8, + }, + { + "datetime": tz.localize(datetime(2015, 1, 1, 17, 8)), + "start": tz.localize(datetime(2015, 1, 1, 17, 8)), + "end": tz.localize(datetime(2015, 1, 1, 17, 8)), + "value": 0.6, + }, + ], + soc_minima=[ + { + "datetime": tz.localize(datetime(2015, 1, 1, 17, 4)), + "start": tz.localize(datetime(2015, 1, 1, 17, 4)), + "end": tz.localize(datetime(2015, 1, 1, 17, 4)), + "value": 0.4, + }, + { + "datetime": tz.localize(datetime(2015, 1, 1, 17, 8)), + "start": tz.localize(datetime(2015, 1, 1, 17, 8)), + "end": tz.localize(datetime(2015, 1, 1, 17, 8)), + "value": 0.7, + }, + ], + consumption_capacity=capacity, + production_capacity=capacity, + resolution=resolution, + soc_min=0, + soc_max=1, + ) + + assert _soc_event_value_at(projected_minima, previous_tick) == pytest.approx( + 0.7 + ), "merged minima should keep the stricter lower bound on the previous tick" + assert _soc_event_value_at(projected_minima, next_tick) == pytest.approx( + 0.7 + ), "merged minima should keep the stricter lower bound on the next tick" + assert _soc_event_value_at(projected_maxima, previous_tick) == pytest.approx( + 0.6 + ), "merged maxima should keep the stricter upper bound on the previous tick" + assert _soc_event_value_at(projected_maxima, next_tick) == pytest.approx( + 0.6 + ), "merged maxima should keep the stricter upper bound on the next tick" + + +@pytest.mark.parametrize("explicit_relax_setting", [None, False]) +def test_off_tick_soc_constraints_enable_relax_soc_constraints( + add_battery_assets, db, explicit_relax_setting +): + """Off-tick SoC constraints enable relaxation because projection can add bounds. + + An explicit ``relax-soc-constraints: False`` is respected, though. + """ + _, battery = get_sensors_from_db( + db, add_battery_assets, battery_name="Test battery" + ) + tz = pytz.timezone("Europe/Amsterdam") + start = tz.localize(datetime(2015, 1, 1, 16, 45)) + end = tz.localize(datetime(2015, 1, 1, 17, 15)) + resolution = timedelta(minutes=15) + + scheduler = StorageScheduler( + battery, + start, + end, + resolution, + flex_model={ + "soc-at-start": "0 MWh", + "soc-min": "0 MWh", + "soc-max": "1 MWh", + "power-capacity": "0.04 MW", + "soc-targets": [ + { + "datetime": "2015-01-01T17:12:00+01:00", + "value": "1 MWh", + } + ], + }, + flex_context={ + "consumption-price": "0 EUR/MWh", + "production-price": "0 EUR/MWh", + **( + {} + if explicit_relax_setting is None + else {"relax-soc-constraints": explicit_relax_setting} + ), + }, + ) + + scheduler.deserialize_config() + + if explicit_relax_setting is False: + assert ( + scheduler.flex_context["relax_soc_constraints"] is False + ), "an explicit relax-soc-constraints: False should be respected" + else: + assert ( + scheduler.flex_context["relax_soc_constraints"] is True + ), "off-tick SoC constraints should automatically enable SoC relaxation" + assert ( + scheduler.flex_context["soc_minima_breach_price"] is not None + ), "auto-enabled SoC relaxation should include a minima breach price" + assert ( + scheduler.flex_context["soc_maxima_breach_price"] is not None + ), "auto-enabled SoC relaxation should include a maxima breach price" + + +def test_off_tick_soc_projection_accounts_for_efficiencies(): + """Reachable energy converts grid power to stock change using the (dis)charging efficiencies.""" + tz = pytz.timezone("Europe/Amsterdam") + resolution = timedelta(minutes=15) + previous_tick = pd.Timestamp(tz.localize(datetime(2015, 1, 1, 17))) + next_tick = previous_tick + resolution + capacity = pd.Series( + 0.04, index=pd.date_range(previous_tick, next_tick, freq=resolution) + ) + + _, _, projected_minima = project_off_tick_soc_constraints( + soc_targets=None, + soc_maxima=None, + soc_minima=[ + { + "datetime": tz.localize(datetime(2015, 1, 1, 17, 12)), + "start": tz.localize(datetime(2015, 1, 1, 17, 12)), + "end": tz.localize(datetime(2015, 1, 1, 17, 12)), + "value": 1, + } + ], + consumption_capacity=capacity, + production_capacity=capacity, + resolution=resolution, + soc_min=0, + soc_max=None, + charging_efficiency=4, # e.g. a heat pump's COP + discharging_efficiency=0.8, + ) + + # Charging between 17:00 and 17:12 moves the stock by 0.04 MW * 4 * 0.2 h. + assert _soc_event_value_at(projected_minima, previous_tick) == pytest.approx( + 1 - 0.04 * 4 * 0.2 + ), "the previous-tick minimum should account for the charging efficiency" + # Discharging between 17:12 and 17:15 moves the stock by 0.04 MW / 0.8 * 0.05 h. + assert _soc_event_value_at(projected_minima, next_tick) == pytest.approx( + 1 - 0.04 / 0.8 * 0.05 + ), "the next-tick minimum should account for the discharging efficiency" + + +def test_off_tick_soc_target_extends_schedule_end_to_next_tick(add_battery_assets, db): + """A target beyond the schedule end extends it to a scheduling tick covering the projection.""" + _, battery = get_sensors_from_db( + db, add_battery_assets, battery_name="Test battery" + ) + tz = pytz.timezone("Europe/Amsterdam") + start = tz.localize(datetime(2015, 1, 1, 16, 45)) + end = tz.localize(datetime(2015, 1, 1, 17, 0)) # before the off-tick target + resolution = timedelta(minutes=15) + + scheduler = StorageScheduler( + battery, + start, + end, + resolution, + flex_model={ + "soc-at-start": "0 MWh", + "soc-min": "0 MWh", + "soc-max": "1 MWh", + "power-capacity": "0.04 MW", + "consumption-capacity": "0.04 MW", + "production-capacity": "0 MW", + "roundtrip-efficiency": 1, + "storage-efficiency": 1, + "soc-targets": [ + { + "datetime": "2015-01-01T17:12:00+01:00", + "value": "1 MWh", + } + ], + }, + flex_context={ + "consumption-price": "0 EUR/MWh", + "production-price": "0 EUR/MWh", + "site-power-capacity": "1 MW", + # keep SoC constraints hard, so we can assert the projected target directly + "relax-soc-constraints": False, + }, + ) + + _, _, schedule_end, _, _, device_constraints, _, _ = scheduler._prepare( + skip_validation=True + ) + + assert schedule_end == tz.localize( + datetime(2015, 1, 1, 17, 15) + ), "the schedule end should be ceiled to the tick carrying the projected target" + storage_constraints = device_constraints[0].tz_convert(tz) + assert storage_constraints.loc[start + resolution, "equals"] == pytest.approx( + 4 + ), "the projected target should fall within the (extended) schedule" + + +def test_off_tick_soc_minima_are_projected_into_soft_commitments( + add_battery_assets, db +): + """With a breach price, projected off-tick minima feed the soft commitments, not hard bounds.""" + _, battery = get_sensors_from_db( + db, add_battery_assets, battery_name="Test battery" + ) + tz = pytz.timezone("Europe/Amsterdam") + start = tz.localize(datetime(2015, 1, 1, 16, 45)) + end = tz.localize(datetime(2015, 1, 1, 17, 15)) + resolution = timedelta(minutes=15) + + scheduler = StorageScheduler( + battery, + start, + end, + resolution, + flex_model={ + "soc-at-start": "0 MWh", + "soc-min": "0 MWh", + "soc-max": "1 MWh", + "power-capacity": "0.04 MW", + "consumption-capacity": "0.04 MW", + "production-capacity": "0.04 MW", + "roundtrip-efficiency": 1, + "storage-efficiency": 1, + "soc-minima": [ + { + "datetime": "2015-01-01T17:12:00+01:00", + "value": "1 MWh", + } + ], + }, + flex_context={ + "consumption-price": "0 EUR/MWh", + "production-price": "0 EUR/MWh", + "site-power-capacity": "1 MW", + "soc-minima-breach-price": "1000 EUR/MWh", + }, + ) + + _, _, _, _, _, device_constraints, _, commitments = scheduler._prepare( + skip_validation=True + ) + + storage_constraints = device_constraints[0].tz_convert(tz) + assert ( + storage_constraints["min"] == 0 + ).all(), ( + "with a breach price, only the global soc-min should remain a hard constraint" + ) + + soc_minima_commitments = [ + c for c in commitments if getattr(c, "name", "") == "any soc minima" + ] + assert len(soc_minima_commitments) == 1 + quantity = soc_minima_commitments[0].quantity.tz_convert(tz) + # The projected previous-tick minimum (1 MWh - 0.04 MW * 0.2 h = 0.992 MWh) + # constrains the stock at the end of the slot starting at 16:45. + assert quantity.loc[start] == pytest.approx( + 0.992 * 4 + ), "the soft commitment should use the projected previous-tick minimum" + # The projected next-tick minimum (1 MWh - 0.04 MW * 0.05 h = 0.998 MWh) + # constrains the stock at the end of the slot starting at 17:00. + assert quantity.loc[start + resolution] == pytest.approx( + 0.998 * 4 + ), "the soft commitment should use the projected next-tick minimum" + + +def test_off_tick_soc_relaxation_is_scoped_to_the_off_tick_device( + add_battery_assets, db +): + """In a multi-device flex-model, auto-relaxation softens only the off-tick device. + + Device 0 uses an off-tick soc-minima (triggering automatic relaxation and + projection), while device 1 uses an on-tick soc-minima. With relaxation + otherwise disabled, device 0's minima should become soft commitments and + device 1's minima should remain hard constraints. + """ + template = add_battery_assets["Test battery"] + asset = GenericAsset( + name="Test multi-device battery site", + generic_asset_type=template.generic_asset_type, + owner=template.owner, + ) + sensor_0 = Sensor( + name="multi-device power 0", + generic_asset=asset, + event_resolution=timedelta(minutes=15), + unit="MW", + ) + sensor_1 = Sensor( + name="multi-device power 1", + generic_asset=asset, + event_resolution=timedelta(minutes=15), + unit="MW", + ) + db.session.add_all([asset, sensor_0, sensor_1]) + db.session.flush() + tz = pytz.timezone("Europe/Amsterdam") + start = tz.localize(datetime(2015, 1, 1, 16, 45)) + end = tz.localize(datetime(2015, 1, 1, 17, 15)) + resolution = timedelta(minutes=15) + + common_flex_model = { + "soc-at-start": "0 MWh", + "soc-min": "0 MWh", + "soc-max": "1 MWh", + "power-capacity": "0.04 MW", + "consumption-capacity": "0.04 MW", + "production-capacity": "0.04 MW", + "roundtrip-efficiency": 1, + "storage-efficiency": 1, + } + scheduler = StorageScheduler( + asset, + start, + end, + resolution, + flex_model=[ + { + "sensor": sensor_0.id, + **common_flex_model, + "soc-minima": [ + { + "datetime": "2015-01-01T17:12:00+01:00", + "value": "1 MWh", + } + ], + }, + { + "sensor": sensor_1.id, + **common_flex_model, + "soc-minima": [ + { + "datetime": "2015-01-01T17:00:00+01:00", + "value": "1 MWh", + } + ], + }, + ], + flex_context={ + "consumption-price": "0 EUR/MWh", + "production-price": "0 EUR/MWh", + "site-power-capacity": "1 MW", + # relaxation is otherwise off, so any softening is due to off-tick projection + "relax-constraints": False, + }, + ) + + _, _, _, _, _, device_constraints, _, commitments = scheduler._prepare( + skip_validation=True + ) + + assert ( + scheduler.flex_context["relax_soc_constraints"] is True + ), "off-tick SoC constraints should automatically enable SoC relaxation" + + soc_minima_commitments = [ + c for c in commitments if getattr(c, "name", "") == "any soc minima" + ] + assert ( + len(soc_minima_commitments) == 1 + and (soc_minima_commitments[0].device == 0).all() + ), "only the off-tick device should have its soc-minima softened into commitments" + quantity = soc_minima_commitments[0].quantity.tz_convert(tz) + assert quantity.loc[start] == pytest.approx( + 0.992 * 4 + ), "the soft commitment should use the projected previous-tick minimum" + assert quantity.loc[start + resolution] == pytest.approx( + 0.998 * 4 + ), "the soft commitment should use the projected next-tick minimum" + + constraints_0 = device_constraints[0].tz_convert(tz) + assert ( + constraints_0["min"] == 0 + ).all(), "the off-tick device should keep only the global soc-min as a hard bound" + + constraints_1 = device_constraints[1].tz_convert(tz) + assert constraints_1.loc[start, "min"] == pytest.approx( + 4 + ), "the on-tick device's soc-minima should remain a hard constraint" + + +def test_project_off_tick_soc_at_start_bounds_the_next_tick(): + """An off-tick starting SoC bounds the next tick by reachable (dis)charge energy.""" + tz = pytz.timezone("Europe/Amsterdam") + resolution = timedelta(minutes=15) + start = pd.Timestamp(tz.localize(datetime(2015, 1, 1, 16, 45))) + next_tick = start + resolution + capacity = pd.Series(0.04, index=pd.date_range(start, next_tick, freq=resolution)) + + soc_maxima, soc_minima = project_off_tick_soc_at_start( + soc_at_start_time=tz.localize(datetime(2015, 1, 1, 16, 47)), + soc_at_start=0.5, + soc_maxima=None, + soc_minima=None, + schedule_start=start, + consumption_capacity=capacity, + production_capacity=capacity, + resolution=resolution, + soc_min=0, + soc_max=1, + charging_efficiency=0.9, + discharging_efficiency=0.8, + ) + + # 13 minutes remain between the known SoC (16:47) and the next tick (17:00). + assert _soc_event_value_at(soc_maxima, next_tick) == pytest.approx( + 0.5 + 0.04 * 0.9 * (13 / 60) + ), "the next tick's upper bound should reflect the chargeable energy since 16:47" + assert _soc_event_value_at(soc_minima, next_tick) == pytest.approx( + 0.5 - 0.04 / 0.8 * (13 / 60) + ), "the next tick's lower bound should reflect the dischargeable energy since 16:47" + + # A known SoC time on a scheduling tick (or outside the first interval) is a no-op. + assert project_off_tick_soc_at_start( + soc_at_start_time=start.to_pydatetime(), + soc_at_start=0.5, + soc_maxima=None, + soc_minima=None, + schedule_start=start, + consumption_capacity=capacity, + production_capacity=capacity, + resolution=resolution, + soc_min=0, + soc_max=1, + ) == (None, None) + + +def test_off_tick_state_of_charge_bounds_first_scheduling_interval( + add_battery_assets, db +): + """A state-of-charge measurement at an off-tick time caps the SoC at the next tick.""" + _, battery = get_sensors_from_db( + db, add_battery_assets, battery_name="Test battery" + ) + tz = pytz.timezone("Europe/Amsterdam") + start = tz.localize(datetime(2015, 1, 1, 16, 45)) + end = tz.localize(datetime(2015, 1, 1, 17, 15)) + resolution = timedelta(minutes=15) + + scheduler = StorageScheduler( + battery, + start, + end, + resolution, + flex_model={ + "soc-min": "0 MWh", + "soc-max": "1 MWh", + "power-capacity": "0.04 MW", + "consumption-capacity": "0.04 MW", + "production-capacity": "0.04 MW", + "roundtrip-efficiency": 1, + "storage-efficiency": 1, + # the starting SoC is known at 16:47, not at the schedule start + "state-of-charge": [ + { + "start": "2015-01-01T16:47:00+01:00", + "end": "2015-01-01T16:47:00+01:00", + "value": "0 MWh", + } + ], + }, + flex_context={ + "consumption-price": "0 EUR/MWh", + "production-price": "0 EUR/MWh", + "site-power-capacity": "1 MW", + # keep SoC constraints hard, so we can assert the projected bounds directly + "relax-constraints": False, + "relax-soc-constraints": False, + }, + ) + + _, _, _, _, soc_at_start, device_constraints, _, _ = scheduler._prepare( + skip_validation=True + ) + + assert soc_at_start[0] == pytest.approx(0), "the starting SoC should be resolved" + storage_constraints = device_constraints[0].tz_convert(tz) + # Charging can only start at 16:47, so by 17:00 at most 13 minutes of charging fit. + assert storage_constraints.loc[start, "max"] == pytest.approx( + 0.04 * (13 / 60) * 4 + ), "the first interval should be capped by the chargeable energy since 16:47" + assert storage_constraints.loc[start, "min"] == pytest.approx( + 0 + ), "the lower bound should be clamped to soc-min" + + def test_deserialize_storage_soc_at_start_from_state_of_charge_sensor( add_charging_station_assets, setup_markets, setup_sources, db ): @@ -1223,3 +1963,116 @@ def test_resolve_soc_at_start_from_percent_sensor_uses_device_sensor_fallback( ) == 2.5 ) + + +def test_off_tick_soc_relaxation_covers_all_devices_of_a_shared_stock( + add_battery_assets, db +): + """Auto-relaxation scoped by stock covers the whole stock group. + + Devices 0 and 1 share a stock whose SoC parameters - including an off-tick + soc-minima - live on a stock-only entry, while device 2 uses an on-tick + soc-minima of its own. The shared stock's minima should be softened into + commitments (landing on the group's first device), while device 2's minima + should remain hard constraints. + """ + template = add_battery_assets["Test battery"] + asset = GenericAsset( + name="Test shared-stock battery site", + generic_asset_type=template.generic_asset_type, + owner=template.owner, + ) + power_sensors = [ + Sensor( + name=f"shared-stock power {i}", + generic_asset=asset, + event_resolution=timedelta(minutes=15), + unit="MW", + ) + for i in range(3) + ] + soc_sensor = Sensor( + name="shared-stock state of charge", + generic_asset=asset, + event_resolution=timedelta(0), + unit="MWh", + ) + db.session.add_all([asset, soc_sensor, *power_sensors]) + db.session.flush() + tz = pytz.timezone("Europe/Amsterdam") + start = tz.localize(datetime(2015, 1, 1, 16, 45)) + end = tz.localize(datetime(2015, 1, 1, 17, 15)) + resolution = timedelta(minutes=15) + + device_properties = { + "power-capacity": "0.04 MW", + "consumption-capacity": "0.04 MW", + "production-capacity": "0.04 MW", + "roundtrip-efficiency": 1, + } + soc_parameters = { + "soc-at-start": "0 MWh", + "soc-min": "0 MWh", + "soc-max": "1 MWh", + } + off_tick_minima = [{"datetime": "2015-01-01T17:12:00+01:00", "value": "1 MWh"}] + on_tick_minima = [{"datetime": "2015-01-01T17:00:00+01:00", "value": "1 MWh"}] + + scheduler = StorageScheduler( + asset, + start, + end, + resolution, + flex_model=[ + { + "sensor": power_sensors[0].id, + "state-of-charge": {"sensor": soc_sensor.id}, + **device_properties, + }, + { + "sensor": power_sensors[1].id, + "state-of-charge": {"sensor": soc_sensor.id}, + **device_properties, + }, + { + # Stock-only entry holding the shared stock's SoC parameters + "state-of-charge": {"sensor": soc_sensor.id}, + **soc_parameters, + "soc-minima": off_tick_minima, + }, + { + "sensor": power_sensors[2].id, + **device_properties, + **soc_parameters, + "soc-minima": on_tick_minima, + }, + ], + flex_context={ + "consumption-price": "0 EUR/MWh", + "production-price": "0 EUR/MWh", + "site-power-capacity": "1 MW", + # relaxation is otherwise off, so any softening is due to off-tick projection + "relax-constraints": False, + }, + ) + + _, _, _, _, _, device_constraints, _, commitments = scheduler._prepare( + skip_validation=True + ) + + assert ( + scheduler.flex_context["relax_soc_constraints"] is True + ), "off-tick SoC constraints should automatically enable SoC relaxation" + + soc_minima_commitments = [ + c for c in commitments if getattr(c, "name", "") == "any soc minima" + ] + assert ( + len(soc_minima_commitments) == 1 + and (soc_minima_commitments[0].device == 0).all() + ), "the shared stock's soc-minima should be softened once, on the group's first device" + + constraints_2 = device_constraints[2].tz_convert(tz) + assert constraints_2.loc[start, "min"] == pytest.approx( + 4 + ), "the on-tick device's soc-minima should remain a hard constraint" diff --git a/flexmeasures/data/schemas/scheduling/metadata.py b/flexmeasures/data/schemas/scheduling/metadata.py index 4f6f9d4298..cfcbcc56a8 100644 --- a/flexmeasures/data/schemas/scheduling/metadata.py +++ b/flexmeasures/data/schemas/scheduling/metadata.py @@ -285,7 +285,7 @@ def to_dict(self): SOC_MINIMA = MetaData( description="""Set points that form lower boundaries, e.g. to target a full car battery in the morning. If a ``soc-minima-breach-price`` is defined, the ``soc-minima`` become soft constraints in the optimization problem. -Otherwise, they become hard constraints. [#maximum_overlap]_. Both single points in time and ranges are possible, see example.""", +Otherwise, they become hard constraints. [#maximum_overlap]_. Both single points in time and ranges are possible, see example. [#projecting_scheduling_constraints]_""", example=[ {"datetime": "2024-02-05T08:00:00+01:00", "value": "8.2 kWh"}, { @@ -298,7 +298,7 @@ def to_dict(self): SOC_MAXIMA = MetaData( description="""Set points that form upper boundaries at certain times, e.g. to target an empty heat buffer before a maintenance window. If a ``soc-maxima-breach-price`` is defined, the ``soc-maxima`` become soft constraints in the optimization problem. -Otherwise, they become hard constraints. [#minimum_overlap]_""", +Otherwise, they become hard constraints. [#minimum_overlap]_ [#projecting_scheduling_constraints]_""", example=[ { "value": "51 kWh", @@ -310,7 +310,7 @@ def to_dict(self): SOC_TARGETS = MetaData( description=""" Exact set point(s) of the storage's state of charge that the scheduler needs to realize. -These are hard constraints, which means that any infeasible state-of-charge targets would prevent a complete schedule from being computed. +These are hard constraints, which means that any infeasible state-of-charge targets would prevent a complete schedule from being computed. [#projecting_scheduling_constraints]_ """, example=[{"datetime": "2024-02-05T08:00:00+01:00", "value": "3.2 kWh"}], ) diff --git a/flexmeasures/data/schemas/scheduling/storage.py b/flexmeasures/data/schemas/scheduling/storage.py index 60d9da3448..e39de7164b 100644 --- a/flexmeasures/data/schemas/scheduling/storage.py +++ b/flexmeasures/data/schemas/scheduling/storage.py @@ -258,14 +258,6 @@ def __init__( self.start = start self.sensor = sensor self.timezone = sensor.timezone if sensor is not None else None - self.flooring_resolution = ( - sensor.event_resolution - if sensor is not None - and sensor.event_resolution != timedelta(0) - and sensor.get_attribute("floor_datetimes_to_resolution", True) - else None - ) - # guess default soc-unit if default_soc_unit is None: if self.sensor is not None and self.sensor.unit in ("MWh", "kWh"): @@ -275,34 +267,17 @@ def __init__( else: default_soc_unit = "MWh" - self.soc_maxima = VariableQuantityField( - to_unit="MWh", - default_src_unit=default_soc_unit, - timezone=self.timezone, - event_resolution=self.flooring_resolution, - data_key="soc-maxima", - ) - - self.soc_minima = VariableQuantityField( - to_unit="MWh", - default_src_unit=default_soc_unit, - timezone=self.timezone, - event_resolution=self.flooring_resolution, - data_key="soc-minima", - value_validator=validate.Range(min=0), - ) - self.soc_targets = VariableQuantityField( - to_unit="MWh", - default_src_unit=default_soc_unit, - timezone=self.timezone, - event_resolution=self.flooring_resolution, - data_key="soc-targets", - ) - super().__init__(*args, **kwargs) - if default_soc_unit is not None: - for field in self.fields.keys(): - if field.startswith("soc_"): + for field in self.fields.keys(): + if field.startswith("soc_"): + # Override the class-level placeholders. Note that assigning new + # instance-level fields would be inert (marshmallow resolves fields + # from the class-level declared fields), so we set attributes on + # the bound fields instead. SoC event datetimes are deliberately + # not floored (no event_resolution is set): off-tick events are + # preserved and later projected onto the scheduling ticks. + setattr(self.fields[field], "timezone", self.timezone) + if default_soc_unit is not None: setattr(self.fields[field], "default_src_unit", default_soc_unit) @validates_schema diff --git a/flexmeasures/data/schemas/scheduling/utils.py b/flexmeasures/data/schemas/scheduling/utils.py new file mode 100644 index 0000000000..07d15a07ca --- /dev/null +++ b/flexmeasures/data/schemas/scheduling/utils.py @@ -0,0 +1,61 @@ +from __future__ import annotations + +from datetime import datetime, timedelta + +import pandas as pd + +from flexmeasures import Sensor + + +SOC_TIMED_EVENT_FIELDS = ("soc-targets", "soc-minima", "soc-maxima") + + +def is_on_schedule_tick(dt: datetime, resolution: timedelta) -> bool: + timestamp = pd.Timestamp(dt) + return timestamp == timestamp.floor(resolution) + + +def get_soc_constraint_resolution( + schedule_resolution: timedelta | None, + sensor: Sensor | None, + default_resolution: timedelta, +) -> timedelta: + if schedule_resolution not in (None, timedelta(0)): + return schedule_resolution + if sensor is not None and sensor.event_resolution != timedelta(0): + return sensor.event_resolution + return default_resolution + + +def should_project_off_tick_soc_constraints(sensor: Sensor | None) -> bool: + return sensor is None or sensor.get_attribute("floor_datetimes_to_resolution", True) + + +def flex_model_has_off_tick_soc_constraints( + flex_model: dict, + resolution: timedelta | None, +) -> bool: + if resolution in (None, timedelta(0)): + return False + + for field_name in SOC_TIMED_EVENT_FIELDS: + field_value = flex_model.get( + field_name, flex_model.get(field_name.replace("-", "_")) + ) + if not isinstance(field_value, list): + continue + for soc_event in field_value: + if not isinstance(soc_event, dict): + continue + for timing_field in ("datetime", "start", "end"): + if soc_event.get(timing_field) is None: + continue + try: + is_on_tick = is_on_schedule_tick( + soc_event[timing_field], resolution + ) + except (TypeError, ValueError): + continue + if not is_on_tick: + return True + return False diff --git a/flexmeasures/data/schemas/tests/test_scheduling.py b/flexmeasures/data/schemas/tests/test_scheduling.py index ed94d711a6..e4dcea7b18 100644 --- a/flexmeasures/data/schemas/tests/test_scheduling.py +++ b/flexmeasures/data/schemas/tests/test_scheduling.py @@ -1,4 +1,4 @@ -from datetime import datetime, timedelta +from datetime import datetime import pytz import pytest @@ -14,7 +14,6 @@ StorageFlexModelSchema, DBStorageFlexModelSchema, ) -from flexmeasures.data.models.time_series import Sensor from flexmeasures.data.schemas.sensors import TimedEventSchema, VariableQuantityField from flexmeasures.utils.unit_utils import ur @@ -164,24 +163,28 @@ def test_process_scheduler_flex_model_process_type(db, app, setup_dummy_sensors) assert process_scheduler_flex_model["process_type"] == ProcessType.SHIFTABLE -def test_storage_flex_model_schema_does_not_floor_instantaneous_sensor( - db, app, dummy_asset +def test_storage_flex_model_schema_preserves_off_tick_soc_datetimes( + db, app, setup_dummy_sensors ): - sensor = Sensor( - "instantaneous power sensor", - generic_asset=dummy_asset, - event_resolution=timedelta(0), - unit="MW", - ) - db.session.add(sensor) - db.session.flush() + sensor1, _, _, _ = setup_dummy_sensors schema = StorageFlexModelSchema( - sensor=sensor, + sensor=sensor1, start=datetime(2023, 1, 1, tzinfo=pytz.UTC), ) - assert schema.flooring_resolution is None + flex_model = schema.load( + { + "soc-at-start": "0 MWh", + "soc-targets": [ + {"datetime": "2023-01-01T00:04:40+00:00", "value": "1 MWh"} + ], + } + ) + + assert flex_model["soc_targets"][0]["datetime"] == pd.Timestamp( + "2023-01-01T00:04:40+00:00" + ) @pytest.mark.parametrize( diff --git a/flexmeasures/tests/test_schemas.py b/flexmeasures/tests/test_schemas.py index 66e76a3ef9..2ad823d992 100644 --- a/flexmeasures/tests/test_schemas.py +++ b/flexmeasures/tests/test_schemas.py @@ -18,12 +18,18 @@ def iter_marshmallow_field_subclasses(): """Yield (class, module) for Marshmallow Field subclasses.""" for module in iter_flexmeasures_modules(): for obj in vars(module).values(): - if ( - inspect.isclass(obj) - and issubclass(obj, ma_fields.Field) - and obj is not ma_fields.Field - and obj.__module__ == module.__name__ - ): + try: + is_field_subclass = ( + inspect.isclass(obj) + and issubclass(obj, ma_fields.Field) + and obj is not ma_fields.Field + and obj.__module__ == module.__name__ + ) + except TypeError: + # On Python 3.10, generic aliases like list[dict] pass + # inspect.isclass but make issubclass raise (bpo-46080). + continue + if is_field_subclass: yield obj, module