flexmeasures.data.models.planning.scheduling_problem

Solver-agnostic preparation of the device scheduler’s inputs.

flexmeasures.data.models.planning.linear_optimization.device_scheduler() (Pyomo) and flexmeasures.data.models.planning.highspy_optimization.device_scheduler_highspy() (direct HiGHS) build the same mathematical model in two very different representations, so the model construction itself is necessarily written twice. Everything around it is not: normalising arguments, resolving stock groups, converting legacy commitments, deriving Big-Ms, and turning solver output back into schedules and costs is plain pandas/numpy work with no solver in it.

Keeping that work here means the two backends cannot drift apart on input handling — only on the model, which is what the equivalence tests in tests/test_highspy_equivalence.py compare. It also gives both backends a single place to grow support for a new scheduling feature’s inputs.

Module Attributes

Functions

flexmeasures.data.models.planning.scheduling_problem.aggregate_commodity_costs(commitments: list[DataFrame] | list[dict], subcommitment_costs: dict) → dict

Sum sub-commitment costs per commodity, skipping commitments without one.

Accepts either the commitment frames or the constant columns already read from them by commitment_scalars(); the latter avoids re-indexing every frame.

flexmeasures.data.models.planning.scheduling_problem.aggregate_subcommitment_costs(subcommitment_costs: dict, commitment_mapping: dict) → dict

Sum sub-commitment costs back onto the commitments they were split from.

flexmeasures.data.models.planning.scheduling_problem.commitment_scalars(commitments: list[DataFrame]) → list[dict]

Read each commitment’s constant columns once.

A commitment frame repeats its name, class, commodity and deviation prices down every row, and both model builders read them back one df[column].iloc[0] at a time — pandas builds a Series and positionally indexes it for each, costing a few microseconds where a dict lookup costs a tenth of one. With a device per charging session there are thousands of commitments per schedule, and those microseconds were a substantial part of building the problem.

Missing columns and missing values both come back as None, so callers can test for one thing rather than distinguishing “no column” from “NaN”.

flexmeasures.data.models.planning.scheduling_problem.convert_commitments_to_subcommitments(dfs: list[DataFrame]) → tuple[list[DataFrame], dict[int, int], list[dict]]

Transform commitments, each specifying a group for each time step, to sub-commitments, one per group.

‘Groups’ are a commitment concept (grouping time slots of a commitment), making it possible that deviations/breaches can be accounted for properly within this group (e.g. highest breach per calendar month defines the penalty). Here, we define sub-commitments, by separating commitments by group and by direction of deviation (up, down).

We also enumerate the time steps in a new column “j”.

For example, given contracts A and B (represented by 2 DataFrames), each with 3 groups, we return (sub)commitments A1, A2, A3, B1, B2 and B3, where A,B,C is the enumerated contract and 1,2,3 is the enumerated group.

flexmeasures.data.models.planning.scheduling_problem.loss_coefficients(efficiency: float) → tuple[float, float]

Coefficients (a, b) of one step of the stock recursion, for how=”linear”.

stock[j] = a * stock[j-1] + b * change[j]

Mirrors apply_stock_changes_and_losses(), which we cannot call here because it expects numbers, while change[j] may be a Pyomo expression.

flexmeasures.data.models.planning.scheduling_problem.planned_power_per_device(power_per_device, start, end, resolution) → list[Series]

Turn each device’s planned power values into a time series.

flexmeasures.data.models.planning.scheduling_problem.prepare_scheduling_problem(device_constraints: list[DataFrame], ems_constraints: DataFrame | list[DataFrame], commitment_quantities: list[Series] | None = None, commitment_downwards_deviation_price: list[Series] | list[float] | None = None, commitment_upwards_deviation_price: list[Series] | list[float] | None = None, commitments: list[DataFrame] | list[Commitment] | None = None, initial_stock: float | list[float] = 0, stock_groups: dict[int, list[int]] | None = None, ems_constraint_groups: list[list[int]] | None = None, device_power_bands: list[list[tuple[float, float]] | None] | None = None, coupling_groups: dict[str, list[tuple[int, float]]] | None = None, balance_groups: dict[str, list[int]] | None = None) → SchedulingProblem

Normalise and validate device_scheduler’s arguments into a SchedulingProblem.

Note

This adds a “stock delta” column to the passed device_constraints DataFrames in place, as the schedulers have always done.

flexmeasures.data.models.planning.scheduling_problem.solver_options(solver_name: str) → dict

The solver options to apply, for the given solver.

HiGHS (whether reached through Pyomo as appsi_highs or directly as highspy – both match on “highs”) gets a tight-tolerance profile, so the two backends cannot disagree on tolerances and silently produce different schedules. Operator-configured options are applied last, so they win.

flexmeasures.data.models.planning.scheduling_problem.validate_highs_options(options: dict) → None

Raise if HiGHS would refuse any of these options.

Pyomo’s appsi_highs interface applies solver options without checking HiGHS’ return status, so an unknown name, an invalid value, or a feature missing from the installed HiGHS build is otherwise ignored without a word. That silently turns a mis-typed option into a no-op, and a benchmark of it into a false negative. Probing a throwaway Highs instance surfaces the rejection instead.

Classes

class flexmeasures.data.models.planning.scheduling_problem.SchedulingProblem(start: object, end: object, resolution: object, device_constraints: list[~pandas.core.frame.DataFrame], ems_constraints_list: list[~pandas.core.frame.DataFrame], ems_constraint_device_groups: list[list[int]], device_to_group: dict[int, str], group_to_devices: dict[str, list[int]], commitments: list[~pandas.core.frame.DataFrame], commitment_mapping: dict[int, int], commitment_scalars: list[dict], device_group_lookup: dict[int, dict], convex_cost_curve: bool, Md: float, Mc: float, band_lookup: dict[int, list[tuple[float, float]]], coupling_device_specs: list[tuple[int, int, float]], balance_group_specs: list[list[int]], initial_stock: float | list[float], original_commitments: list[~pandas.core.frame.DataFrame] = <factory>)

Everything both scheduler backends need before building their model.

Produced by prepare_scheduling_problem(); see device_scheduler’s docstring for what the underlying arguments mean.

Md: float

Big-Ms bounding the search space for device power (Md) and commitment deviations (Mc)

__init__(start: object, end: object, resolution: object, device_constraints: list[~pandas.core.frame.DataFrame], ems_constraints_list: list[~pandas.core.frame.DataFrame], ems_constraint_device_groups: list[list[int]], device_to_group: dict[int, str], group_to_devices: dict[str, list[int]], commitments: list[~pandas.core.frame.DataFrame], commitment_mapping: dict[int, int], commitment_scalars: list[dict], device_group_lookup: dict[int, dict], convex_cost_curve: bool, Md: float, Mc: float, band_lookup: dict[int, list[tuple[float, float]]], coupling_device_specs: list[tuple[int, int, float]], balance_group_specs: list[list[int]], initial_stock: float | list[float], original_commitments: list[~pandas.core.frame.DataFrame] = <factory>) → None

balance_group_specs: list[list[int]]

device lists of the balance groups (internal commodity nodes), empty groups dropped

band_lookup: dict[int, list[tuple[float, float]]]

device index -> its signed power bands (S2 operation modes)

commitment_scalars: list[dict]

Each sub-commitment’s constant columns (name, class, commodity, deviation prices), gathered while splitting. See commitment_scalars().

commitments: list[DataFrame]

Sub-commitments (one per commitment group and deviation direction), and the mapping from each sub-commitment index back to its original commitment index

property commodity_devices: dict

commodity -> set(device indices).

Computed on demand: only the EMS-level flow commitment constraints need it, and the per-row scan is not cheap enough to pay for unconditionally.

convex_cost_curve: bool

Whether the summed deviation prices describe a convex cost curve (a non-convex curve needs binary commitment-sign variables).

coupling_device_specs: list[tuple[int, int, float]]

(group index, device index, coefficient) triples for hard flow-coupling constraints

device_constraints: list[DataFrame]

Device constraints, with a “stock delta” column guaranteed to be present

device_group_lookup: dict[int, dict]

sub-commitment index -> {device group label -> member device indices}

device_to_group: dict[int, str]

device -> its primary stock group key, and stock group key -> member devices

ems_constraints_list: list[DataFrame]

EMS constraints, normalised to a list, plus the device indices each applies to

initial_stock_of(d) → float

The initial stock of device d, defaulting to 0.

Device indices reaching this from a commitment’s “device” column may be numpy floats, hence the cast.

original_commitments: list[DataFrame]

The commitments as passed in, before the sub-commitment split. Only kept to derive commodity_devices lazily.

start: object

Timing, taken from the first device

class flexmeasures.data.models.planning.scheduling_problem.SubCommitmentFrames(specs: list[tuple])

The sub-commitment frames, each built the first time it is asked for.

Splitting a commitment produces one sub-commitment per group, and a group is usually a single time step, so a schedule can hold thousands of one-row frames. The HiGHS backend works entirely from the arrays gathered during the split and never looks at them; the Pyomo backend uses them throughout. Building them all up front therefore cost more than the rest of the split put together, and for the default backend it bought nothing.

Each entry knows the frame it came from, the row positions of its group, and the price column its half of a two-sided group drops.

__init__(specs: list[tuple])