flexmeasures.data.models.planning.storage

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.
soc_min – Minimum state of charge at all times.

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: list[dict[str, datetime | float]] | Series, 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, soc_max: float, 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.
soc_max – Maximum state of charge at all times.
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

_get_device_power_capacity(flex_model: list[dict], assets: list[GenericAsset]) → list[Quantity]

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 site-power-capacity attribute of the asset.

_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.

_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, 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.
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(**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).

ensure_soc_min_max(): Make sure we have min and max SOC. If not passed directly, then get default from asset or targets. This happens before deserializing the flex-model.

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

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_soc_schedule(flex_model: list[dict], ems_schedule: DataFrame, soc_at_start: list[float], device_constraints: list, resolution: timedelta) → dict

Build the state-of-charge schedule for each device that has a state-of-charge sensor.

Converts the integrated power schedule from MWh to the sensor’s unit. For sensors with a ‘%’ unit, 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.

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