flexmeasures.data.models.planning.storage

Module Attributes

Functions

Collect all constraints for a given storage device in a DataFrame that the device_scheduler can interpret.

Parameters:

start – Start of the schedule.
end – End of the schedule.
resolution – Timedelta used to resample the constraints to the resolution of the schedule.
soc_at_start – State of charge at the start time.
soc_targets – Exact targets for the state of charge at each time.
soc_maxima – Maximum state of charge at each time.
soc_minima – Minimum state of charge at each time.
soc_max – Maximum state of charge at all times, if configured.
soc_min – Minimum state of charge at all times, if configured.

Returns:

Constraints (StorageScheduler.COLUMNS) for a storage device, at each time step (index). See device_scheduler for possible column names.

flexmeasures.data.models.planning.storage.build_device_soc_values(soc_values: Quantity | list[dict[str, datetime | float]] | Series | None, soc_at_start: float, start_of_schedule: datetime, end_of_schedule: datetime, resolution: timedelta) → Series

Utility function to create a Pandas series from SOC values we got from the flex-model.

Should set NaN anywhere where there is no target.

SOC values should be indexed by their due date. For example, for quarter-hourly targets from 5 to 6 AM: >>> df = pd.Series(data=[1, 1.5, 2, 2.5, 3], index=pd.date_range(pd.Timestamp(“2010-01-01T05”), pd.Timestamp(“2010-01-01T06”), freq=pd.Timedelta(“PT15M”), inclusive=”both”)) >>> print(df) 2010-01-01 05:00:00 1.0 2010-01-01 05:15:00 1.5 2010-01-01 05:30:00 2.0 2010-01-01 05:45:00 2.5 2010-01-01 06:00:00 3.0 Freq: 15min, dtype: float64

TODO: this function could become the deserialization method of a new TimedEventSchema (targets, plural), which wraps TimedEventSchema.

flexmeasures.data.models.planning.storage.create_constraint_violations_message(constraint_violations: list) → str

Create a human-readable message with the constraint_violations.

Parameters:: constraint_violations – List with the constraint violations.
Returns:: Human-readable message.

flexmeasures.data.models.planning.storage.get_pattern_match_word(word: str) → str

Get a regex pattern to match a word

The conditions to delimit a word are:

start of line
whitespace
end of line
word boundary
arithmetic operations

Returns:: regex expression

flexmeasures.data.models.planning.storage.prepend_series(series: Series, value) → Series

Prepend a value to a time series

Parameters:

series – series containing the timed values
value – value to place in the first position

flexmeasures.data.models.planning.storage.sanitize_expression(expression: str, columns: list) → tuple[str, list]

Wrap column in commas to accept arbitrary column names (e.g. with spaces).

Parameters:

expression – Expression to sanitize.
columns – List with the name of the columns of the input data for the expression.

Returns:

Sanitized expression and columns (variables) used in the expression.

flexmeasures.data.models.planning.storage.validate_constraint(constraints_df: DataFrame, lhs_expression: str, inequality: str, rhs_expression: str, round_to_decimals: int | None = 6) → list[dict]

Validate the feasibility of a given set of constraints.

Parameters:

constraints_df – DataFrame with the constraints
lhs_expression – left-hand side of the inequality expression following pd.eval format. No need to use the syntax column to reference column, just use the column name.
inequality – inequality operator, one of (‘<=’, ‘<’, ‘>=’, ‘>’, ‘==’, ‘!=’).
rhs_expression – right-hand side of the inequality expression following pd.eval format. No need to use the syntax column to reference column, just use the column name.
round_to_decimals – Number of decimals to round off to before validating constraints.

Returns:

List of constraint violations, specifying their time, constraint and violation.

flexmeasures.data.models.planning.storage.validate_storage_constraints(constraints: DataFrame, soc_at_start: float, soc_min: float | None, soc_max: float | None, resolution: timedelta) → list[dict]

Check that the storage constraints are fulfilled, e.g min <= equals <= max.

Global validation
A.1) min >= soc_min A.2) max <= soc_max
Validation in the same time frame
B.1) min <= max B.2) min <= equals B.3) equals <= max
Validation in different time frames
C.1) equals(t) - equals(t-1) <= derivative_max(t) C.2) derivative_min(t) <= equals(t) - equals(t-1) C.3) min(t) - max(t-1) <= derivative_max(t) C.4) max(t) - min(t-1) >= derivative_min(t) C.5) equals(t) - max(t-1) <= derivative_max(t) C.6) derivative_min(t) <= equals(t) - min(t-1)

Parameters:

constraints – dataframe containing the constraints of a storage device
soc_at_start – State of charge at the start time.
soc_min – Minimum state of charge at all times, if configured.
soc_max – Maximum state of charge at all times, if configured.
resolution – Constant duration between the start of each time step.

Returns:

List of constraint violations, specifying their time, constraint and violation.

Classes

This class defines the constraints of a schedule for a storage device from the flex-model, flex-context, and sensor and asset attributes

static _default_missing_directional_capacity_to_zero(flex_model_d: dict) → None: Given a missing capacity opposite a non-zero directional capacity, default the missing capacity to zero.

_get_commodity_contexts() → dict[str, dict]

Return commodity-specific flex-contexts.

Supports the new format:

“commodities”: [
{“commodity”: “electricity”, …}, {“commodity”: “gas”, …},

]

and keeps backwards compatibility with old top-level fields.

The device power capacity for each device must be known for the optimization problem to stay bounded.

We search for the power capacity in the following order: 1. Look for the power_capacity_in_mw field in the deserialized flex-model. 2. Look for the power-capacity flex-model field of the asset. 3. Look for the greatest device consumption-capacity or production-capacity. 4. Look for the site-power-capacity attribute of the asset.

_get_largest_device_capacity(flex_model_d: dict, query_window: tuple[datetime, datetime], resolution: timedelta, beliefs_before: datetime | None) → Sensor | SensorReference | list[dict] | Quantity | Series | None: Return the largest configured directional capacity, if any.

_get_soc_capacity_for_percent_conversion(flex_model: dict, sensor: Sensor | None = None) → str

Return the capacity used to convert percentage-based SoC values.

Parameters:

flex_model – Flex model containing the SoC configuration.
sensor – Optional scheduled power sensor whose asset can provide fallback capacity.

Returns:

Capacity expressed in MWh.

_get_soc_lookup_radius(sensor: Sensor | None = None, slack_steps: int = 4) → timedelta

Return the half-width of the SoC lookup interval.

We search for a nearby SoC value in the interval [self.start - slack_steps * resolution, self.start + slack_steps * resolution]. Using four resolution steps by default keeps the lookup tolerant to small timing offsets while still rejecting stale values. For example, a 15-minute resolution yields a 1-hour lookup radius.

Parameters:

sensor – Optional sensor whose resolution should be used.
slack_steps – Number of resolution steps accepted on either side of the schedule start.

Returns:

Half-width of the SoC lookup interval.

static _is_non_zero_capacity(capacity: str | int | float | Quantity | Sensor | SensorReference | list) → bool: Return whether a configured capacity should imply zero capacity in the opposite direction.

_prepare(skip_validation: bool = False) → tuple

This function prepares the required data to compute the schedule:

price data
device constraint
ems constraints

Parameters:: skip_validation – If True, skip validation of constraints specified in the data.
Returns:: Input data for the scheduler

_resolve_soc_at_start_from_sensor(state_of_charge_sensor: Sensor | SensorReference, flex_model: dict, sensor: Sensor | None = None) → float

Resolve soc-at-start from a state-of-charge sensor.

Parameters:

state_of_charge_sensor – Instantaneous SoC sensor or sensor reference (with optional source filters).
flex_model – Flex model containing the SoC configuration.
sensor – Optional scheduled power sensor.

Returns:

Starting SoC in MWh.

_resolve_soc_at_start_from_state_of_charge(flex_model: dict, sensor: Sensor | None = None) → float | None

Resolve soc-at-start from the state-of-charge field.

Parameters:

flex_model – Flex model containing the SoC configuration.
sensor – Optional scheduled power sensor.

Returns:

Starting SoC in MWh if it can be inferred.

_resolve_soc_at_start_from_time_series(soc_time_series: list[dict], sensor: Sensor | None = None) → float

Resolve soc-at-start from a state-of-charge time series.

Parameters:

soc_time_series – SoC time series specification.
sensor – Optional scheduled power sensor.

Returns:

Starting SoC in MWh.

compute_schedule() → Series | None

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.

Deprecated method in v0.14. As an alternative, use MetaStorageScheduler.compute().

convert_to_commitments(flex_model, **timing_kwargs) → list[FlowCommitment | StockCommitment]: Convert list of commitment specifications (dicts) to a list of FlowCommitments.

deserialize_flex_config()

Deserialize storage flex model and the flex context against schemas. Before that, we fill in values from wider context, if possible. Mostly, we allow several fields to come from sensor attributes. TODO: this work could maybe go to the schema as a pre-load hook (if we pass in the sensor to schema initialization)

Note: Before we apply the flex config schemas, we need to use the flex config identifiers with hyphens,: (this is how they are represented to outside, e.g. by the API), after deserialization we use internal schema names (with underscores).

ensure_soc_at_start(flex_model: dict | None = None, sensor: Sensor | None = None) → dict: Ensure we have a starting state of charge - if needed. Preferably, a starting soc is given. Otherwise, we try to retrieve the current state of charge from the configured state-of-charge field. If that doesn’t work, we try the (old-style) asset attribute. Finally, we default the starting soc to 0 (only if there are soc limits, though, as some assets don’t use the concept of a state of charge, and without soc targets and limits the starting soc doesn’t matter).

get_min_max_targets() → tuple[float | None, float | None]: This happens before deserializing the flex-model.

persist_flex_model()

Store new soc info as GenericAsset attributes

This method should become obsolete when all SoC information is recorded on a sensor, instead.

Deprecated: get rid of this when moving to v1.0 (requiring to also remove attributes from test data assets)

possibly_extend_end(soc_targets, sensor: Sensor = None)

Extend schedule period in case a target exceeds its end.

The schedule’s duration is possibly limited by the server config setting ‘FLEXMEASURES_MAX_PLANNING_HORIZON’.

todo: when deserialize_flex_config becomes a single schema for the whole scheduler,: this function would become a class method with a @post_load decorator.

compute(skip_validation: bool = False) → Series | list[dict[str, Any]] | None

Schedule a battery or Charge Point by just starting to charge, discharge, or do neither,

depending on the first target state of charge and the capabilities of the Charge Point. For the resulting consumption schedule, consumption is defined as positive values.

Note that this ignores any cause of the infeasibility.

Parameters:: skip_validation – If True, skip validation of constraints specified in the data.
Returns:: The computed schedule.

static _build_consumption_production_schedules(flex_model: list[dict], ems_schedule: DataFrame) → dict

Build consumption and/or production power schedules for devices that define output sensors.

Each device’s flex model may define a consumption sensor, a production sensor, or both. The schedule stored on each sensor depends on which sensors are defined:

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.

The consumption_is_positive attribute is set on each output sensor when the scheduling job is created (see create_scheduling_job), not here. This method only clips the series; sign handling is left entirely to make_schedule.

Unit conversion from MW to each sensor’s unit is applied.

Parameters:

flex_model – List of per-device flex models (after deserialization).
ems_schedule – DataFrame of per-device power schedules in MW (consumption positive).

Returns:

Dict mapping each output sensor to its power schedule.

static _build_soc_schedule(flex_model: list[dict], ems_schedule: list[Series], soc_at_start: list[float], device_constraints: list, resolution: timedelta, stock_groups: dict[int, list[int]]) → tuple[dict, dict]

Build the state-of-charge schedule for each stock group.

Supports both: - original logic: one device per stock group - local/shared-stock logic: multiple devices contribute to one shared stock

For shared stock groups, each device contribution is integrated separately with its own efficiencies and stock delta, then summed on top of the shared initial stock.

Also computes the MWh SoC for devices that have soc-minima or soc-maxima constraints (even without a state-of-charge sensor) so that unresolved targets can be checked later. For a shared stock group, every device in the group is tracked against the group’s combined MWh SoC series, since they share the same physical stock.

Converts the integrated stock schedule from MWh to the state-of-charge sensor unit. For ‘%’ sensors, the soc-max flex-model field is used as capacity. If soc-max is missing or zero for a ‘%’ sensor, the schedule is skipped with a warning.

Note: soc-max is a QuantityField (not a VariableQuantityField), so it is always a float after deserialization and cannot be a sensor reference. The isinstance guard below is therefore a defensive check for forward-compatibility.

Returns:: Tuple of (soc_schedule keyed by SoC sensor in sensor unit, soc_schedule_mwh keyed by device index in MWh).

_compute_commodity_aggregate_schedules(storage_schedule: dict, ems_schedule: DataFrame) → None

Compute per-commodity aggregate power flows for aggregate-consumption and aggregate-production sensors.

This method populates the storage_schedule dict with aggregate schedules for each commodity that defines aggregate-consumption and/or aggregate-production sensors in its commodity context.

The sign convention and split logic follows the same pattern as _build_consumption_production_schedules: - Only aggregate-consumption defined: full aggregate schedule (consumption +, production -) - Only aggregate-production defined: full aggregate schedule (consumption +, production -)

(sign will be flipped by make_schedule based on consumption_is_positive=False)

Both defined: consumption sensor gets non-negative part, production sensor gets non-positive part (sign will be flipped for production by make_schedule)

For backwards compatibility, when no commodity_contexts are defined, all devices are treated as electricity devices and use the top-level flex-context fields.

Parameters:

storage_schedule – Dict to populate with aggregate schedules (will be modified in-place)
ems_schedule – DataFrame of per-device power schedules in MW (consumption positive)

_compute_unresolved_targets(flex_model: list[dict], soc_schedule_mwh: dict, start: datetime, end: datetime, resolution: timedelta, most_relevant_only: bool = False) → tuple[list, list]

Compute unmet and met SoC minima/maxima targets per device.

For each device that has soc-minima or soc-maxima constraints in the flex model, compares the computed MWh SoC schedule against those constraints. Devices without a state_of_charge Sensor are included as long as a device key can be determined from the power sensor.

The result includes asset ID for each constraint. Devices for which an asset ID cannot be determined are skipped.

Constraints are evaluated over the window (start + resolution, end) (i.e. the first scheduled slot through the end of the schedule). The start slot itself is the initial condition (soc_at_start), not a scheduled value, so it is excluded.

Note: soc-targets are modelled as hard constraints and are not checked here, as by definition the scheduler will not allow any deviation from them.

Parameters:

flex_model – The deserialized flex model (list of per-device dicts).
soc_schedule_mwh – MWh SoC schedule keyed by device index d.
start – Start of the schedule.
end – End of the schedule.
resolution – Schedule resolution.
most_relevant_only – If False (the default), report every violated/met slot. If True, report only the single most relevant slot (the first violation, or the tightest margin). Either way, the result holds a list.

Returns:

A tuple (unresolved, resolved).

unresolved is a list of dicts, each with "asset" field and constraint info. Each constraint entry is a list of dicts {"datetime": <ISO 8601 UTC string>, "violation": "<value> kWh"} (one per violated slot, or just the first if most_relevant_only is True), where violation is always positive.

resolved is also a list of dicts with "asset" field and constraint info. Each constraint entry is a list of dicts {"datetime": <ISO 8601 UTC string>, "margin": "<value> kWh"} (one per met slot, or just the slot with the tightest/smallest positive margin if most_relevant_only is True).

_electricity_device_indices() → list[int]: Return the device indices (flexible and inflexible) belonging to the electricity commodity.

_reconstruct_commodity_to_devices() → dict[str, list[int]]

Reconstruct the mapping of commodity -> device indices as enumerated by _prepare().

Device enumeration order:

flexible devices (from the flex-model), in order,
top-level (electricity) inflexible-device-sensors, in order,
each commodity context’s own inflexible-device-sensors, in the order the commodity contexts are given.

This mirrors _prepare()’s device enumeration exactly, so the returned device indices line up with entries of ems_schedule / device_constraints.

compute(skip_validation: bool = False) → Series | list[dict[str, Any]] | None

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.

Parameters:: skip_validation – If True, skip validation of constraints specified in the data.
Returns:: The computed schedule.

fallback_scheduler_class: alias of StorageFallbackScheduler