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 forecasts 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_targets(targets: list[dict[str, datetime | float]] | pd.Series, soc_at_start: float, start_of_schedule: datetime, end_of_schedule: datetime, resolution: timedelta) → pd.Series

flexmeasures.data.models.planning.storage.build_device_soc_values(soc_values: list[dict[str, datetime | float]] | pd.Series, soc_at_start: float, start_of_schedule: datetime, end_of_schedule: datetime, resolution: timedelta) → pd.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: 15T, 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_serie(serie: Series, value) → Series

Prepend a value to a time series series

Parameters:

serie – serie 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: pd.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

_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

compute_schedule() → pd.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().

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_min_max(): Make sure we have min and max SOC. If not passed directly, then get default from sensor or targets.

persist_flex_model(): Store new soc info as GenericAsset attributes

possibly_extend_end()

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.

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.

compute_schedule() → pd.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().

fallback_scheduler_class: alias of StorageFallbackScheduler