flexmeasures.data.models.planning.storage
Functions
- flexmeasures.data.models.planning.storage.add_storage_constraints(start: datetime, end: datetime, resolution: timedelta, soc_at_start: float, soc_targets: list[dict[str, datetime | float]] | pd.Series | None, soc_maxima: list[dict[str, datetime | float]] | pd.Series | None, soc_minima: list[dict[str, datetime | float]] | pd.Series | None, soc_max: float, soc_min: float) pd.DataFrame
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
- class flexmeasures.data.models.planning.storage.MetaStorageScheduler(sensor: Sensor | None = None, start: datetime | None = None, end: datetime | None = None, resolution: timedelta | None = None, belief_time: datetime | None = None, asset_or_sensor: Asset | Sensor | None = None, round_to_decimals: int | None = 6, flex_model: dict | None = None, flex_context: dict | None = None, return_multiple: bool = False)
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.
- class flexmeasures.data.models.planning.storage.StorageFallbackScheduler(sensor: Sensor | None = None, start: datetime | None = None, end: datetime | None = None, resolution: timedelta | None = None, belief_time: datetime | None = None, asset_or_sensor: Asset | Sensor | None = None, round_to_decimals: int | None = 6, flex_model: dict | None = None, flex_context: dict | None = None, return_multiple: bool = False)
- 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.
- class flexmeasures.data.models.planning.storage.StorageScheduler(sensor: Sensor | None = None, start: datetime | None = None, end: datetime | None = None, resolution: timedelta | None = None, belief_time: datetime | None = None, asset_or_sensor: Asset | Sensor | None = None, round_to_decimals: int | None = 6, flex_model: dict | None = None, flex_context: dict | None = None, return_multiple: bool = False)
- 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