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, order: Literal['asc', 'desc'] = 'asc') → 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), divided over the number of planning steps.

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

flexmeasures.data.models.planning.utils.extend_to_edges(df: DataFrame | 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.get_continuous_series_sensor_or_quantity(variable_quantity: Sensor | list[dict] | Quantity | Series | None, unit: Quantity | str, query_window: tuple[datetime, datetime], resolution: timedelta, beliefs_before: datetime | None = None, min_value: float | int = nan, max_value: float | int | Series = nan, as_instantaneous_events: bool = False, resolve_overlaps: str = 'first', fill_sides: bool = False) → Series

Creates a time series from a sensor, time series specification, or quantity within a specified window, 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.

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

• 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).

• fill_sides –

  If True, values are extended to the edges of the query window:

  • 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 | SensorReference, consumption_is_positive: bool | None = None) → 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, or a SensorReference wrapping such a sensor with source filters

• consumption_is_positive – sign convention of the sensor’s data, as determined by the flex-context field the device was listed under (inflexible-consumption → True, inflexible-production → False); None means the sign convention is read from the sensor’s consumption_is_positive attribute (the deprecated inflexible-device-sensors field’s behavior)

Returns:

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

flexmeasures.data.models.planning.utils.get_series_from_quantity_or_sensor(variable_quantity: Sensor | SensorReference | list[dict] | Quantity, unit: 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) → 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 SensorReference wrapping a Sensor with optional source filters

  • 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).

• fill_sides –

  If True, values are extended to the edges of the query window:

  • 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.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') → DatetimeIndex

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

flexmeasures.data.models.planning.utils.nanmin_of_series_and_value(s: Series, value: float | Series) → 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.

flexmeasures.data.models.planning.utils.solver_output_kept_off_our_log_handlers(solver) → Iterator[None]

Show an appsi solver’s console output once, without passing it through the application’s log handlers.

Pyomo’s appsi solvers (e.g. appsi_highs) solve inside capture_output(capture_fd=True), which points file descriptors 1 and 2 at a pipe that a reader thread drains into config.solver_output_logger. By default, that is a Pyomo logger that propagates to the root, so a console handler writes each line back into the pipe it came from. Lines then pile up with one more prefix per pass, and once a line no longer fits in the pipe, the reader thread blocks on its own write while holding the handler’s lock, which hangs the next thread that logs (see issue #2742).

So we hand the solver a logger of its own, whose one handler writes to a copy of stdout taken before the capture starts. Whatever reaches the captured descriptors then appears on the console exactly once: the solver’s log, when output_flag lets HiGHS print one (see solver_options()), and anything our own handlers wrote there during the solve, such as Pyomo’s warnings, already formatted by them.