flexmeasures.data.models.planning.utils

Functions

flexmeasures.data.models.planning.utils.add_tiny_price_slope(orig_prices: DataFrame, col_name: str = 'event_value', d: float = 0.0001) → DataFrame

Add tiny price slope to col_name to represent e.g. inflation as a simple linear price increase. This is meant to break ties, when multiple time slots have equal prices, in favour of acting sooner. We penalise the future with at most d times the price spread (1 per thousand by default).

flexmeasures.data.models.planning.utils.ensure_prices_are_not_empty(prices: pd.DataFrame, price_variable_quantity: Sensor | list[dict] | ur.Quantity | None)

flexmeasures.data.models.planning.utils.extend_to_edges(df: pd.DataFrame | pd.Series, query_window: tuple[datetime, datetime], resolution: timedelta, kind_of_values: str = 'price', sensor: Sensor = None, allow_trimmed_query_window: bool = False)

Values are extended to the edges of the query window.

• The first available value serves as a naive backcasts.

• The last available value serves as a naive forecast.

flexmeasures.data.models.planning.utils.fallback_charging_policy(sensor: Sensor, device_constraints: DataFrame, start: datetime, end: datetime, resolution: timedelta) → Series

This fallback charging policy is to just start charging or discharging, or do neither, depending on the first target state of charge and the capabilities of the Charge Point. Note that this ignores any cause of the infeasibility and, while probably a decent policy for Charge Points, should not be considered a robust policy for other asset types.

flexmeasures.data.models.planning.utils.get_continuous_series_sensor_or_quantity(variable_quantity: Sensor | list[dict] | ur.Quantity | pd.Series | None, actuator: Sensor | Asset, unit: ur.Quantity | str, query_window: tuple[datetime, datetime], resolution: timedelta, beliefs_before: datetime | None = None, fallback_attribute: str | None = None, min_value: float | int = nan, max_value: float | int | pd.Series = nan, as_instantaneous_events: bool = False, resolve_overlaps: str = 'first', fill_sides: bool = False) → pd.Series

Creates a time series from a sensor, time series specification, or quantity within a specified window, falling back to a given fallback_attribute and making sure values stay within the domain [min_value, max_value].

Parameters:

• variable_quantity – A sensor recording the data, a time series specification or a fixed quantity.

• actuator – The actuator from which relevant defaults are retrieved.

• unit – The desired unit of the data.

• query_window – The time window (start, end) to query the data.

• resolution – The resolution or time interval for the data.

• beliefs_before – Timestamp for prior beliefs or knowledge.

• fallback_attribute – Attribute serving as a fallback default in case no quantity or sensor is given.

• min_value – Minimum value.

• max_value – Maximum value (also replacing NaN values).

• as_instantaneous_events – optionally, convert to instantaneous events, in which case the passed resolution is interpreted as the desired frequency of the data.

• resolve_overlaps – If time series segments overlap (e.g. when upsampling to instantaneous events), take the ‘max’, ‘min’ or ‘first’ value during overlapping time spans (or at instantaneous moments, such as at event boundaries).

• window (fill_sides If True, values are extended to the edges of the query) –

  • The first available value serves as a naive backcast.

  • The last available value serves as a naive forecast.

Returns:

time series data with missing values handled based on the chosen method.

flexmeasures.data.models.planning.utils.get_power_values(query_window: tuple[datetime, datetime], resolution: timedelta, beliefs_before: datetime | None, sensor: Sensor) → np.ndarray

Get measurements or forecasts of an inflexible device represented by a power or energy sensor as an array of power values in MW.

If the requested schedule lies in the future, the returned data will consist of (the most recent) forecasts (if any exist). If the requested schedule lies in the past, the returned data will consist of (the most recent) measurements (if any exist). The latter amounts to answering “What if we could have scheduled under perfect foresight?”.

Parameters:

• query_window – datetime window within which events occur (equal to the scheduling window)

• resolution – timedelta used to resample the forecasts to the resolution of the schedule

• beliefs_before – datetime used to indicate we are interested in the state of knowledge at that time

• sensor – power sensor representing an energy flow out of the device

Returns:

power measurements or forecasts (consumption is positive, production is negative)

flexmeasures.data.models.planning.utils.get_quantity_from_attribute(entity: Asset | Sensor, attribute: str | None, unit: str | ur.Quantity) → ur.Quantity

Get the value (in the given unit) of a quantity stored as an entity attribute.

Parameters:

• entity – The entity (sensor or asset) containing the attribute to retrieve the value from.

• attribute – The attribute name to extract the value from.

• unit – The unit in which the value should be returned.

Returns:

The retrieved quantity or the provided default.

flexmeasures.data.models.planning.utils.get_series_from_quantity_or_sensor(variable_quantity: Sensor | list[dict] | ur.Quantity, unit: ur.Quantity | str, query_window: tuple[datetime, datetime], resolution: timedelta, beliefs_before: datetime | None = None, as_instantaneous_events: bool = True, resolve_overlaps: str = 'first', fill_sides: bool = False) → pd.Series

Get a time series given a quantity or sensor defined on a time window.

Parameters:

• variable_quantity – Variable quantity measuring e.g. power capacity or efficiency. One of the following types: - a timely-beliefs Sensor recording the data - a list of dictionaries representing a time series specification - a pint Quantity representing a fixed quantity

• unit – Unit of the output data.

• query_window – Tuple representing the start and end of the requested data.

• resolution – Time resolution of the requested data.

• beliefs_before – Optional datetime used to indicate we are interested in the state of knowledge at that time.

• as_instantaneous_events – Optionally, convert to instantaneous events, in which case the passed resolution is interpreted as the desired frequency of the data.

• resolve_overlaps – If time series segments overlap (e.g. when upsampling to instantaneous events), take the ‘max’, ‘min’ or ‘first’ value during overlapping time spans (or at instantaneous moments, such as at event boundaries).

• window (fill_sides If True, values are extended to the edges of the query) –

  • The first available value serves as a naive backcast.

  • The last available value serves as a naive forecast.

Returns:

Pandas Series with the requested time series data.

flexmeasures.data.models.planning.utils.idle_after_reaching_target(schedule: Series, target: Series, initial_state: float = 0) → Series

Stop planned (dis)charging after target is reached (or constraint is met).

flexmeasures.data.models.planning.utils.initialize_device_commitment(start: Timestamp, end: Timestamp, resolution: timedelta, device: int, target_datetime: str, target_value: float, soc_at_start: float, soc_target_penalty: float) → DataFrame

Model penalties for demand unmet per device.

flexmeasures.data.models.planning.utils.initialize_df(columns: list[str], start: datetime, end: datetime, resolution: timedelta, inclusive: str = 'left') → DataFrame

flexmeasures.data.models.planning.utils.initialize_energy_commitment(start: Timestamp, end: Timestamp, resolution: timedelta, market_prices: list[float]) → DataFrame

Model energy contract for the site.

flexmeasures.data.models.planning.utils.initialize_index(start: date | datetime | str, end: date | datetime | str, resolution: timedelta | str, inclusive: str = 'left') → pd.DatetimeIndex

flexmeasures.data.models.planning.utils.initialize_series(data: pd.Series | list[float] | np.ndarray | float | None, start: datetime, end: datetime, resolution: timedelta, inclusive: str = 'left') → pd.Series

flexmeasures.data.models.planning.utils.nanmin_of_series_and_value(s: pd.Series, value: float | pd.Series) → pd.Series

Perform a nanmin between a Series and a float.

flexmeasures.data.models.planning.utils.process_time_series_segments(index: DatetimeIndex, variable_quantity: list[dict], unit: str, resolution: timedelta, resolve_overlaps: str, fill_sides: bool = False) → Series

Process a time series defined by a list of dicts, while resolving overlapping segments.

Parameters:

index: The index for the time series DataFrame. variable_quantity: List of events, where each event is a dictionary containing:

• ‘value’: The value of the event (can be a Quantity or scalar).

• ‘start’: The start datetime of the event.

• ‘end’: The end datetime of the event.

unit: The unit to convert the value into if it’s a Quantity. resolution: The resolution to subtract from the ‘end’ to avoid overlap. resolve_overlaps: How to handle overlaps (e.g., ‘first’, ‘last’, ‘mean’, etc.). fill_sides: Whether to extend values to cover the whole index.

Returns: A time series with resolved event values.