h2integrate.control.control_strategies.storage.plm_optimized_storage_controller

Contents

h2integrate.control.control_strategies.storage.plm_optimized_storage_controller#

Classes

PeakLoadManagementOptimizedControllerConfig(...)

Configuration for the Peak Load Management optimized storage controller.

PeakLoadManagementOptimizedStorageController(...)

Demand-response storage controller using a rolling-horizon MILP.
class h2integrate.control.control_strategies.storage.plm_optimized_storage_controller.PeakLoadManagementOptimizedControllerConfig(max_charge_rate, supervisory_signal, peak_window, performance_incentive=None, performance_incentive_per_event=None, charge_efficiency=1.0, discharge_efficiency=1.0, n_max_events=10, n_control_window_hours=24.0, signal_threshold_percentile=0.0, event_duration=None, min_peak_separation=None, *, max_capacity, max_soc_fraction, min_soc_fraction, init_soc_fraction, commodity, commodity_rate_units, tech_name, system_commodity_interface_limit, round_digits=4)#

Configuration for the Peak Load Management optimized storage controller.

Inherits base fields from PyomoStorageControllerBaseConfig: max_capacity, max_soc_fraction, min_soc_fraction, init_soc_fraction, n_control_window_hours, commodity, commodity_rate_units, tech_name, system_commodity_interface_limit, round_digits.

Variables:

  • max_charge_rate (float) -- Maximum charge and discharge rate (kW).

  • supervisory_signal (list[float]) -- Price, demand, or price*demand forecast time series. The rolling horizon solver uses one window of length n_control_window_hours per solve.

  • peak_window (dict) -- Hours eligible for dispatch. Keys 'start' and 'end' must be strings in HH:MM:SS format.

  • performance_incentive (float) -- Incentive revenue in $/kWh. Mutually exclusive with performance_incentive_per_event.

  • performance_incentive_per_event (float) -- Incentive revenue in $/event. Converted internally to an effective $/kWh rate using steps_per_event, dt, and P_max: incentive_kWh = incentive_event / (steps_per_event * dt_h * P_max). Mutually exclusive with performance_incentive.

  • charge_efficiency (float) -- Charge efficiency in [0, 1]. Defaults to 1.0.

  • discharge_efficiency (float) -- Discharge efficiency in [0, 1]. Defaults to 1.0.

  • n_max_events (int) -- Maximum discharge events per calendar month. Defaults to 10.

  • n_control_window_hours (float) -- Rolling window size in hours. Converted to an integer timestep count during setup() using the simulation dt, so the same value works at any resolution. Example: 10 at a 30-min dt gives a 20-timestep window. Defaults to 24.

  • signal_threshold_percentile (float) -- Percentile (0-100) used to compute the signal threshold for each rolling window. Only timesteps at or above this percentile of the window signal are eligible for dispatch. Defaults to 0.0 (all timesteps eligible).

  • event_duration (dict) -- Total dispatch-event duration expressed as a {units, val} dict, where units is any pandas timedelta unit string (e.g. 'h', 'min', 's') and val is the numeric amount. When set, the peak-signal timestep within peak_window is located and every timestep within event_duration / 2 of that peak is marked eligible (the window may extend beyond the static peak_window boundaries). When None (default) the static peak_window mask is used unchanged. Example: {units: 'h', val: 4} is +/- 2 h around the daily peak.

  • min_peak_separation (dict) -- Minimum separation between eligible peaks, expressed as a {units, val} dict like event_duration. When set, the eligible timesteps identified by signal_threshold_percentile are treated as peaks and the first peak is chosen as eligible.

Parameters:

  • max_charge_rate (float)

  • supervisory_signal (list)

  • peak_window (dict)

  • performance_incentive (float)

  • performance_incentive_per_event (float)

  • charge_efficiency (float)

  • discharge_efficiency (float)

  • n_max_events (int)

  • n_control_window_hours (float)

  • signal_threshold_percentile (float)

  • event_duration (dict)

  • min_peak_separation (dict)

  • max_capacity (float)

  • max_soc_fraction (float)

  • min_soc_fraction (float)

  • init_soc_fraction (float)

  • commodity (str)

  • commodity_rate_units (str)

  • tech_name (str)

  • system_commodity_interface_limit (float | int | str | list[float])

  • round_digits (int)

max_charge_rate: float#
supervisory_signal: list#
peak_window: dict#
performance_incentive: float#
performance_incentive_per_event: float#
charge_efficiency: float#
discharge_efficiency: float#
n_max_events: int#
n_control_window_hours: float#
signal_threshold_percentile: float#
event_duration: dict#
min_peak_separation: dict#
class h2integrate.control.control_strategies.storage.plm_optimized_storage_controller.PeakLoadManagementOptimizedStorageController(**kwargs)#

Demand-response storage controller using a rolling-horizon MILP.

Each call to the dispatch solver iterates over the full simulation in windows of length n_control_window_hours. For each window it builds and solves a MILP that maximizes incentive revenue, then passes the resulting dispatch commands to the performance model. The terminal SOC of each window is carried forward as the initial SOC of the next window.

dr_model: Any#
problem_state: DispatchProblemState#
_time_step_bound = (300, 3600)#
setup()#

Initialize config, register OpenMDAO inputs, and pre-compute static masks.

Raises:

ValueError -- If the length of the time series built from plant_config does not match n_timesteps.

initialize_parameters(inputs)#

Sync OpenMDAO inputs into the config.

Parameters:

inputs (dict) -- OpenMDAO inputs dict. Recognized keys are 'max_charge_rate' and 'storage_capacity'.

compute(inputs, outputs, discrete_inputs, discrete_outputs)#

Build the DR dispatch solver and write it to discrete outputs.

Parameters:

  • inputs (dict) -- OpenMDAO continuous inputs.

  • outputs (dict) -- OpenMDAO continuous outputs.

  • discrete_inputs (dict) -- OpenMDAO discrete inputs.

  • discrete_outputs (dict) -- OpenMDAO discrete outputs. The key 'pyomo_dispatch_solver' is set to the callable returned by pyomo_setup.

pyomo_setup(discrete_inputs, om_inputs)#

Return the rolling-horizon dispatch solver callable.

Parameters:

  • discrete_inputs (dict) -- OpenMDAO discrete inputs.

  • om_inputs (dict) -- OpenMDAO continuous inputs. max_charge_rate and storage_capacity are read from here so that optimizer changes to those values are reflected in each solve.

Returns:

callable -- pyomo_dispatch_solver(performance_model, performance_model_kwargs, inputs) that iterates over the simulation in windows of n_control_window_hours timesteps. For each window it:

  1. Builds a fresh MILP from the window's signal slice.

  2. Solves the MILP with a MILP solver

  3. Calls performance_model with the resulting dispatch commands.

  4. Carries the terminal SOC into the next window.

Returns (storage_out, soc_out) - two np.ndarray of length n_timesteps.

_parse_peak_window()#

Parse the peak_window config entry into datetime.time objects.

Accepts values either as HH:MM:SS strings or as plain integers (seconds since midnight). PyYAML parses unquoted HH:MM:SS values as sexagesimal integers, so both forms are equivalent in YAML.

Returns:

tuple[datetime.time, datetime.time] -- (start, end) times.

Raises:

ValueError -- If 'start' or 'end' keys are missing, or if a value is neither a valid HH:MM:SS string nor an integer.

Return type:

tuple

_compute_peak_window_mask()#

Build a boolean mask that is True for timesteps inside the peak window.

Returns:

np.ndarray -- Boolean array of shape (n_timesteps,).

Raises:

ValueError -- If peak_window end time is before start time.

Return type:

ndarray

_compute_month_ids()#

Return the calendar month index (1-12) for each timestep.

Returns:

np.ndarray -- Integer array of shape (n_timesteps,).

Return type:

ndarray

_compute_eligible_mask(signal_window, dispatch_mask=None)#

Build a boolean mask for timesteps whose signal meets the dispatch threshold.

The threshold percentile is computed from signal_window values that fall inside dispatch_mask (i.e. the current dispatch window). When dispatch_mask is None the full signal_window is used. When signal_threshold_percentile is 0.0 all timesteps are eligible. If there are multiple timesteps above the threshold within a window and min_peak_separation is set, only the first one is eligible.

Parameters:

  • signal_window (np.ndarray) -- Signal values for the current rolling window.

  • dispatch_mask (np.ndarray | None) -- Boolean mask of shape (len(signal_window),) indicating which timesteps belong to the current dispatch window. Defaults to None (use full window).

Returns:

np.ndarray --

Boolean array of shape (len(signal_window),).

True where signal_t >= threshold.

Return type:

ndarray

_compute_event_window_mask(eligible_mask)#

Expand each eligible peak timestep by +/- event_duration/2.

Every True timestep in eligible_mask is treated as a peak and all timesteps within event_duration / 2 of it are marked eligible. When event_duration is None returns eligible_mask unchanged.

Parameters:

eligible_mask (np.ndarray) -- Boolean mask of peak timesteps, shape (window_len,).

Returns:

np.ndarray -- Boolean mask of shape (window_len,).

Return type:

ndarray

_build_dr_model(window_start, window_len, init_soc, remaining_budget, P_max, storage_capacity)#

Build the DR MILP for a single rolling window.

Decision variables#

discharge[t]binary

Event indicator — 1 if a discharge event is active at timestep t. Used for event counting and window feasibility constraints only.

charge[t]binary

Event indicator — 1 if a charge event is active at timestep t.

p_discharge[t]continuous in [0, P_max]

Actual discharge power (kW). Linked to the binary via the McCormick upper-bound constraint p_discharge[t] <= P_max * discharge[t].

p_charge[t]continuous in [0, P_max]

Actual charge power (kW). Linked via p_charge[t] <= P_max * charge[t].

soc[t]continuous in [soc_min, soc_max]

State of charge (fraction).

param window_start:

Timestep index of the first timestep in this window.

type window_start:

int

param window_len:

Number of timesteps in this window.

type window_len:

int

param init_soc:

State-of-charge fraction at the start of this window.

type init_soc:

float

param remaining_budget:

Mapping of month_id (int) to remaining event slots for that month. Computed by subtracting events already dispatched in earlier windows from n_max_events, so the monthly cap is respected across windows.

type remaining_budget:

dict

param P_max:

Maximum charge/discharge rate (kW), taken from OpenMDAO inputs so optimizer changes are reflected.

type P_max:

float

param storage_capacity:

Total storage capacity (kWh), taken from OpenMDAO inputs so optimizer changes are reflected.

type storage_capacity:

float

returns:

pyomo.ConcreteModel -- Fully formed MILP ready to solve.

Parameters:

  • window_start (int)

  • window_len (int)

  • init_soc (float)

  • remaining_budget (dict)

  • P_max (float)

  • storage_capacity (float)

Return type:

ConcreteModel

solve_dispatch_model(start_time=0, n_days=0)#

Solve the DR MILP for the current window and record solver metrics.

Parameters:

  • start_time (int) -- Timestep index of the window start. Used only for error messages and metrics. Defaults to 0.

  • n_days (int) -- Total simulation days. Passed to DispatchProblemState.store_problem_metrics. Defaults to 0.

Raises:

RuntimeError -- If solver returns a not OK status or an unacceptable termination condition.

static glpk_solve_call(pyomo_model, log_name='', user_solver_options=None)#

Solve a Pyomo MILP with GLPK.

Parameters:

  • pyomo_model (pyomo.ConcreteModel) -- The model to solve.

  • log_name (str) -- Optional log file name passed to SolverOptions. Defaults to ''.

  • user_solver_options (dict | None) -- Optional overrides for GLPK solver options. Defaults to None.

Returns:

pyomo.opt.SolverResults -- Raw results object from GLPK.

_get_storage_dispatch_commands()#

Net dispatch commands for the solved window.

Returns:

list[float] -- p_discharge_t - p_charge_t (kW) for each timestep in the solved window. Positive = discharge, negative = charge.

Return type:

list

Contents