Control and Optimisation¶

Controllers are simulators that read measurements, optionally look ahead with a Forecast and write set-points (MPCController solves an economic LP/MILP over library Blocks, OPFController wraps pandapower OPF, PyPSARollingHorizonController wraps pypsa.Network.optimize). Solver-based controllers need scipy (pip install energysim[control]). The status of each feature is tracked in features.yaml; see Validation for the cross-checks against independent LPs.

Controllers¶

`Controller`¶

Module: energysim.control.base

Controller(
    name: 'str' = 'controller',
    measurement_inputs: 'Sequence[str]' = (),
    setpoint_outputs: 'Sequence[str]' = (),
    forecasts: 'Optional[Mapping[str, Any]]' = None,
    initial_setpoints: 'Optional[Mapping[str, float]]' = None,
    initial_measurements: 'Optional[Mapping[str, float]]' = None,
    units: 'Optional[Mapping[str, str]]' = None,
    decide_at_init: 'Optional[bool]' = None,
    actuation_delay: 'bool' = True,
    step_size: 'float' = 900.0,
)

Base class of controllers (optimisers, MPC, OPF, rule-based ...).

Parameters¶

Parameter	Type	Description
`name`	str	Controller name (used in messages).
`measurement_inputs`	list[str]	Names of the inputs (measurements / external signals).
`setpoint_outputs`	list[str]	Names of the set-point outputs. The diagnostic outputs `solver_status` (0 = optimal, 1 = limit, 2 = infeasible, 3 = unbounded, 4 = error, -1 = not solved yet), `solve_time_s`, `objective` and `fallback` (0 none, 1 previous plan, 2 rule) are always added.
`forecasts`	dict[str, Forecast], optional	Forecast providers evaluated at every decision over `horizon_times` and handed to `decide`.
`initial_setpoints`	dict, optional	Output values before the first decision (default 0.0).
`initial_measurements`	dict, optional	Measurement values assumed before the first exchange (default 0.0).
`units`	dict, optional	`{variable: unit}` for inputs / outputs (used by world unit conversion).
`decide_at_init`	bool, optional	Take a decision in `init` (default: True when `initial_measurements` were given).
`actuation_delay`	bool, default True	See the module docstring.
`step_size`	float, default 900	Decision interval used by `step` (`advance` decides once per call).

Methods¶

Method	Summary
`advance(start_time, stop_time)`	One decision per world macro step (delay = macro step).
`advance_with_recording(start_time, stop_time, outputs)`	Advance while recording output values at every micro-step.
`configure(start_time, stop_time, macro_step)`	Receive the run's time frame before `init` is called.
`decide(t, measurements, forecasts)`	Return `{setpoint: value}` for the interval starting at `t`.
`evaluate_forecasts(t_act, measurements)`
`get_available_variables()`	Return `{'inputs': [...], 'outputs': [...]}` of known variable names.
`get_value(parameters, time=None)`	Return a list of values corresponding to `parameters`.
`horizon_times(t_act)`	Start times of the forecast horizon intervals (default: just `t_act`).
`init()`	Start a fresh run: outputs, measurements, history and forecasts are reset.
`reset_controller()`	Hook for subclasses: clear internal state (plans, estimates) at `init()`.
`set_value(parameters, values)`	Set a list of `parameters` to the given `values`.
`step(time)`	Decide once at `time` (stand-alone use; delay = `step_size`).

`RuleController`¶

Module: energysim.control.base

RuleController(rule, measurement_inputs=(), setpoint_outputs=(), **kwargs)

Controller defined by a plain function rule(t, measurements, forecasts) -> dict.

Handy as a baseline or as the fallback of an optimising controller.

Methods¶

Method	Summary
`decide(t, measurements, forecasts)`	Return `{setpoint: value}` for the interval starting at `t`.

`MPCController`¶

Module: energysim.control.mpc

MPCController(
    blocks: 'Sequence[Block]',
    forecasts: 'Optional[Mapping[str, Any]]' = None,
    horizon: 'int' = 24,
    dt: 'float' = 900.0,
    co2_price: 'float' = 0.0,
    unserved_penalty: 'Optional[float]' = 10000.0,
    spill_penalty: 'Optional[float]' = None,
    extra_inputs: 'Sequence[str]' = (),
    fallback: 'Union[str, Callable]' = 'hold',
    horizon_end: 'Optional[float]' = None,
    delay_compensation: 'bool' = True,
    time_limit: 'Optional[float]' = None,
    mip_rel_gap: 'Optional[float]' = 1e-06,
    name: 'str' = 'mpc',
    decide_at_init: 'bool' = True,
    **kwargs,
)

Economic MPC over a set of Block objects.

Parameters¶

Parameter	Type	Description
`blocks`	list[Block]	Assets of the problem. Outputs are named `"<block>.<var>"` (e.g. `bess.P_cmd`), measured states likewise (`bess.SoC`).
`forecasts`	dict[str, Forecast]	Providers referenced by name in block parameters (`'price'`, `'load'`...).
`horizon`	int, default 24	Number of intervals.
`dt`	float, default 900	Interval length in seconds.
`co2_price`	float, default 0	EUR per tonne CO2.
`unserved_penalty`	float or None, default 1e4	EUR per MWh of unserved demand per carrier; `None` makes the balances hard (the problem can then be infeasible).
`spill_penalty`	float or None	If set, surplus may be spilled at this cost (EUR/MWh).
`extra_inputs`	list[str]	Additional inputs (e.g. a connected `price` signal usable as a persistence forecast by name).
`fallback`	'hold' \| 'zero' \| callable	Rule used when the solve fails and no previous plan covers the current time: keep the previous set-points, use zeros, or call `fallback(t, measurements, forecasts) -> dict`.
`horizon_end`	float, optional	Absolute time beyond which the horizon is not extended (shrinking horizon).
`delay_compensation`	bool, default True	Predict the state at the actuation time (see module docstring).
`time_limit`	float, optional	HiGHS options.
`mip_rel_gap`	float, optional	HiGHS options.

Methods¶

Method	Summary
`block(name)`
`decide(t, measurements, forecasts)`	Return `{setpoint: value}` for the interval starting at `t`.
`horizon_times(t_act)`	Start times of the forecast horizon intervals (default: just `t_act`).
`plan_open_loop(t0, n_steps=None, measurements=None)`	Solve one problem from `t0` over `n_steps` (default `horizon`) from the initial
`reset_controller()`	Hook for subclasses: clear internal state (plans, estimates) at `init()`.
`solve_horizon(times, states, forecasts=None, measurements=None)`	Build and solve the problem over `times` from the given block states.
`status_text()`

`OPFController`¶

Module: energysim.control.opf

OPFController(
    network: 'Any',
    inputs: 'Sequence[str]' = (),
    outputs: 'Sequence[str]' = (),
    controllable: 'Optional[Iterable[str]]' = None,
    costs: 'Optional[Mapping[str, Union[float, Mapping[str, float]]]]' = None,
    mode: 'str' = 'ac',
    copy: 'bool' = True,
    opf_kwargs: 'Optional[Mapping[str, Any]]' = None,
    name: 'str' = 'opf',
    **kwargs,
)

pandapower OPF wrapped as an energysim controller.

Parameters¶

Parameter	Type	Description
`network`	pandapowerNet, pp_adapter, or path to a pickle / json network	The network model (deep-copied unless `copy=False`).
`inputs`	list[str]	`"element.column"` names set on the network before every OPF.
`outputs`	list[str]	`"element.column"` result names (and the special outputs `cost`, `max_line_loading_percent`, `max_trafo_loading_percent`).
`controllable`	iterable of element names, optional	Elements whose `controllable` flag is set to True.
`costs`	dict, optional	`{element: cp1_eur_per_mw}` or `{element: {cp0_eur:..., cp1_eur_per_mw:..., ...}}`; missing `poly_cost` rows are created with `pp.create_poly_cost`.
`mode`	'ac' \| 'dc'	`runopp` or `rundcopp`.
`opf_kwargs`	dict, optional	Extra keyword arguments for the pandapower OPF call.

Validity and notes¶

On non-convergence the previous set-points are kept (fallback = 1) and solver_status is 2 (infeasible / not converged) or 4 (other error).

Methods¶

Method	Summary
`decide(t, measurements, forecasts)`	Return `{setpoint: value}` for the interval starting at `t`.
`reset_controller()`	Hook for subclasses: clear internal state (plans, estimates) at `init()`.
`run_opf()`	Run the pandapower OPF on the controller's network (raises on failure).

`PyPSARollingHorizonController`¶

Module: energysim.control.pypsa_rolling

PyPSARollingHorizonController(
    network: 'Any',
    window: 'int',
    setpoints: 'Mapping[str, Sequence]',
    measurements: 'Optional[Mapping[str, Sequence]]' = None,
    snapshot_seconds: 'Optional[float]' = None,
    start_snapshot: 'int' = 0,
    solver_name: 'str' = 'highs',
    solver_options: 'Optional[Mapping]' = None,
    copy: 'bool' = True,
    name: 'str' = 'pypsa_mpc',
    **kwargs,
)

Receding-horizon controller around pypsa.Network.optimize.

Parameters¶

Parameter	Type	Description
`network`	pypsa.Network	Multi-period network with time series over (at least) the simulated period.
`window`	int	Number of snapshots optimised at each decision.
`setpoints`	dict[str, tuple]	`{output_name: (component, attribute, element[, scale])}` read from the first window snapshot (e.g. `{'bess.P_cmd': ('StorageUnit', 'p', 'bess', -1.0)}`).
`measurements`	dict[str, tuple], optional	`{input_name: (component, attribute, element[, scale])}` written before each solve (e.g. `{'bess.SoC': ('StorageUnit', 'state_of_charge_initial', 'bess', 4.0)}`).
`snapshot_seconds`	float, optional	Snapshot spacing in seconds (default: inferred from a DatetimeIndex, else 3600).

Methods¶

Method	Summary
`decide(t, measurements, forecasts)`	Return `{setpoint: value}` for the interval starting at `t`.
`reset_controller()`	Hook for subclasses: clear internal state (plans, estimates) at `init()`.
`snapshot_index(t)`

Forecasts¶

`Forecast`¶

Module: energysim.control.forecast

Forecast()

Base class of forecast providers.

`PerfectForecast`¶

Module: energysim.control.forecast

PerfectForecast(
    data,
    column: 'Optional[str]' = None,
    time_column: 'Optional[str]' = None,
    periodic: 'Optional[float]' = None,
    scale: 'float' = 1.0,
)

Read ahead in a known time series (an oracle forecast).

Parameters¶

Parameter	Type	Description
`data`	DataFrame, Series, str (CSV path), dict or `(times, values)` tuple	The time series. For a DataFrame / CSV the time axis is the `time_column` (default `'time'` if present, else the index); for a Series its index. Times are seconds.
`column`	str, optional	Value column (required for DataFrames with more than one candidate).
`periodic`	float, optional	Period in seconds; times are wrapped modulo the period (e.g. 86400 to repeat a daily profile). Without it values beyond the series hold the first / last sample.
`scale`	float, default 1.0	Multiplier applied to the values (unit conversion, e.g. 1e-3 kW->MW).

`PersistenceForecast`¶

Module: energysim.control.forecast

PersistenceForecast(input: 'Optional[str]' = None, default: 'float' = 0.0)

Repeat the latest measured value over the whole horizon.

Parameters¶

Parameter	Type	Description
`input`	str, optional	Name of the controller input holding the measured value.
`default`	float, default 0.0	Value used when no measurement is available.

`CallableForecast`¶

Module: energysim.control.forecast

CallableForecast(fn: 'Callable[[float, np.ndarray], Any]')

Wrap fn(t, horizon_times) -> array | scalar.

`NoisyForecast`¶

Module: energysim.control.forecast

NoisyForecast(
    base: 'Forecast',
    sigma: 'float',
    seed: 'Optional[int]' = None,
    relative: 'bool' = False,
    growth: 'float' = 0.0,
    exact_first: 'bool' = False,
)

Add Gaussian noise to another forecast (for robustness / ensemble studies).

sigma is absolute (same unit as the base forecast) unless relative=True (fraction of the base value). The noise standard deviation grows linearly with lead time: sigma_k = sigma * (1 + growth * k). The first horizon value is left noise-free when exact_first=True. The generator is re-seeded at every reset() (i.e. every run), so equal seeds give identical runs.

MPC building blocks¶

`Block`¶

Module: energysim.control.blocks

Block(name: 'str')

Base class: one asset in the optimisation problem.

Subclasses implement build and extract; stateful blocks also initial_state, state_from_measurements and predict. Variable names exposed by the controller are prefixed with "<block name>.".

`StorageBlock`¶

Module: energysim.control.blocks

StorageBlock(
    name: 'str',
    capacity: 'float',
    p_charge_max: 'float',
    p_discharge_max: 'Optional[float]' = None,
    eta_c: 'float' = 1.0,
    eta_d: 'float' = 1.0,
    soc_min: 'float' = 0.0,
    soc_max: 'float' = 1.0,
    soc0: 'float' = 0.5,
    self_discharge: 'float' = 0.0,
    carrier: 'str' = 'electricity',
    setpoint: 'str' = 'P_cmd',
    charge_positive: 'bool' = True,
    soc_measurement: 'Optional[str]' = 'SoC',
    soc_terminal: 'Optional[float]' = None,
    terminal_value: 'Optional[float]' = None,
    cycle_cost: 'float' = 0.0,
    use_binary: 'bool' = False,
)

Generic storage (battery, thermal tank, hydrogen tank ...).

Dynamics (e stored energy, dt_h hours):

e[k+1] = (1 - self_discharge * dt_h) * e[k] + eta_c * pc[k] * dt_h - pd[k] / eta_d * dt_h

Parameters¶

Parameter	Type	Description
`capacity`	float	Usable capacity in MWh (kg for hydrogen).
`p_charge_max`	float	Power limits (MW; kg/h for hydrogen).
`p_discharge_max`	float	Power limits (MW; kg/h for hydrogen).
`eta_c`	float	Charge / discharge efficiencies.
`eta_d`	float	Charge / discharge efficiencies.
`soc_min`	float	SoC window and initial SoC (fractions).
`soc_max`	float	SoC window and initial SoC (fractions).
`soc0`	float	SoC window and initial SoC (fractions).
`self_discharge`	float	Fraction of the stored energy lost per hour.
`setpoint`	str, default 'P_cmd'	Name of the set-point output; `charge_positive` gives its sign.
`soc_measurement`	str or None, default 'SoC'	Name of the measured SoC input (fraction); `None` = open loop.
`soc_terminal`	float, optional	Minimum SoC at the end of the horizon.
`terminal_value`	float, optional	Value (EUR/MWh) of energy left at the end of the horizon.
`cycle_cost`	float, default 0	Throughput cost (EUR/MWh charged + discharged).
`use_binary`	bool, default False	Forbid simultaneous charge and discharge with a binary (MILP).

`ElectrolyzerBlock`¶

Module: energysim.control.blocks

ElectrolyzerBlock(
    name: 'str',
    p_max: 'float',
    p_min: 'float' = 0.0,
    curve: 'Optional[Sequence[Tuple[float, float]]]' = None,
    spec_consumption: 'float' = 55.0,
    use_binary: 'bool' = False,
    setpoint: 'str' = 'P_cmd',
    carrier_in: 'str' = 'electricity',
    carrier_out: 'str' = 'hydrogen',
    standby_cost: 'float' = 0.0,
)

Electrolyser: electricity -> hydrogen with a piecewise-linear production curve.

P = p_min * on + sum(seg_j), H2 = h(p_min) * on + sum(slope_j * seg_j) with 0 <= seg_j <= width_j * on. The curve is replaced by its concave envelope (exact for concave curves, slightly optimistic otherwise). With use_binary=True on is binary and the minimum turndown p_min is enforced exactly (MILP); otherwise on is relaxed to [0, 1] (LP).

Parameters¶

Parameter	Type	Description
`p_max`	float	Rated electrical power (MW).
`p_min`	float	Minimum turndown (MW).
`curve`	sequence of (P_MW, H2_kg_per_h), optional	Production curve between `p_min` and `p_max`. Default: constant `spec_consumption` (kWh/kg).

`HeatPumpBlock`¶

Module: energysim.control.blocks

HeatPumpBlock(
    name: 'str',
    q_max: 'float',
    cop: 'Any' = 3.0,
    q_min: 'float' = 0.0,
    setpoint: 'str' = 'Q_heat_cmd_MW',
    carrier_in: 'str' = 'electricity',
    carrier_out: 'str' = 'heat',
)

Heat pump: P_elec = Q_heat / COP; COP fixed or time-varying (forecast).

`BoilerBlock`¶

Module: energysim.control.blocks

BoilerBlock(
    name: 'str',
    p_max: 'float',
    efficiency: 'float' = 1.0,
    fuel: 'Optional[str]' = 'electricity',
    fuel_price: 'Any' = 0.0,
    co2_intensity: 'float' = 0.0,
    ramp: 'Optional[float]' = None,
    setpoint: 'str' = 'P_cmd',
    carrier_out: 'str' = 'heat',
)

Boiler: heat = efficiency * fuel.

fuel is a carrier name whose balance supplies the input (e.g. 'electricity' for an electric boiler) or None for fuel bought at fuel_price (EUR/MWh fuel). co2_intensity is t CO2 per MWh fuel. The set-point is the fuel / electric input power.

`GeneratorBlock`¶

Module: energysim.control.blocks

GeneratorBlock(
    name: 'str',
    p_max: 'float',
    p_min: 'float' = 0.0,
    cost: 'Any' = 0.0,
    no_load_cost: 'float' = 0.0,
    ramp: 'Optional[float]' = None,
    co2_intensity: 'float' = 0.0,
    no_load_co2: 'float' = 0.0,
    heat_ratio: 'float' = 0.0,
    carrier: 'str' = 'electricity',
    heat_carrier: 'str' = 'heat',
    use_binary: 'bool' = False,
    setpoint: 'str' = 'P_cmd',
)

Dispatchable generator (optionally CHP) with unit-commitment relaxation.

p_min * on <= p <= p_max * on; cost cost * p + no_load_cost * on (EUR/MWh, EUR/h); CO2 co2_intensity * p + no_load_co2 * on (t/MWh, t/h); ramp limit in MW/h; recoverable heat <= heat_ratio * p. on is binary with use_binary=True (MILP) else relaxed to [0, 1].

`GridBlock`¶

Module: energysim.control.blocks

GridBlock(
    name: 'str' = 'grid',
    import_limit: 'Any' = inf,
    export_limit: 'Any' = inf,
    price_import: 'Any' = 0.0,
    price_export: 'Any' = 0.0,
    peak_cost: 'float' = 0.0,
    co2_intensity: 'Any' = 0.0,
    carrier: 'str' = 'electricity',
)

Market / grid connection: import and export at (time-varying) prices.

peak_cost (EUR/MW) prices the maximum import over the horizon (demand charge). co2_intensity (t/MWh) applies to imports. Make sure price_export <= price_import or limit the exchange, otherwise the problem is unbounded. Works for any carrier (e.g. a hydrogen offtaker with carrier='hydrogen', import_limit=0, price_export=4.0).

`LoadBlock`¶

Module: energysim.control.blocks

LoadBlock(
    name: 'str',
    demand: 'Any',
    carrier: 'str' = 'electricity',
    shiftable_fraction: 'float' = 0.0,
    shift_max: 'Optional[float]' = None,
)

Demand (forecast) with an optional shiftable share.

The fixed part (1 - f) * d must be served when it occurs; the shiftable part f * d may be served at any interval of the horizon (0 <= s <= shift_max, total energy conserved within the horizon). The output P_set is the planned load (a set-point for flexible loads).

`RenewableBlock`¶

Module: energysim.control.blocks

RenewableBlock(
    name: 'str',
    available: 'Any',
    curtailable: 'bool' = True,
    carrier: 'str' = 'electricity',
    cost: 'Any' = 0.0,
)

Variable renewable source: 0 <= p <= available (== available if not curtailable).

`PlanContext`¶

Module: energysim.control.blocks

PlanContext(
    lp: 'LinearProblem',
    times: 'np.ndarray',
    dt: 'float',
    forecasts: 'Optional[Mapping[str, np.ndarray]]' = None,
    measurements: 'Optional[Mapping[str, float]]' = None,
    co2_price: 'float' = 0.0,
)

Shared state while the blocks build one horizon problem.

Linear programs¶

`LinearProblem`¶

Module: energysim.control.lp

LinearProblem() -> 'None'

Incrementally built min c'x s.t. lo <= A x <= hi, lb <= x <= ub.

`LPResult`¶

Module: energysim.control.lp

LPResult(
    status: 'int',
    x: 'Optional[np.ndarray]',
    objective: 'float',
    message: 'str' = '',
    solve_time_s: 'float' = 0.0,
)

Outcome of LinearProblem.solve.

Control and Optimisation¶

Controllers¶

Controller¶

Parameters¶

Methods¶

RuleController¶

Methods¶

MPCController¶

Parameters¶

Methods¶

OPFController¶

Parameters¶

Validity and notes¶

Methods¶

PyPSARollingHorizonController¶

Parameters¶

Methods¶

Forecasts¶

Forecast¶

PerfectForecast¶

Parameters¶

PersistenceForecast¶

Parameters¶

CallableForecast¶

NoisyForecast¶

MPC building blocks¶

Block¶

StorageBlock¶

Parameters¶

ElectrolyzerBlock¶

Parameters¶

HeatPumpBlock¶

BoilerBlock¶

GeneratorBlock¶

GridBlock¶

LoadBlock¶

RenewableBlock¶

PlanContext¶

Linear programs¶

LinearProblem¶

LPResult¶

`Controller`¶

`RuleController`¶

`MPCController`¶

`OPFController`¶

`PyPSARollingHorizonController`¶

`Forecast`¶

`PerfectForecast`¶

`PersistenceForecast`¶

`CallableForecast`¶

`NoisyForecast`¶

`Block`¶

`StorageBlock`¶

`ElectrolyzerBlock`¶

`HeatPumpBlock`¶

`BoilerBlock`¶

`GeneratorBlock`¶

`GridBlock`¶

`LoadBlock`¶

`RenewableBlock`¶

`PlanContext`¶

`LinearProblem`¶

`LPResult`¶