agentlib_mpc.modules package

Submodules

agentlib_mpc.modules.mpc module

Holds the base class for MPCs.

class agentlib_mpc.modules.mpc.BaseMPC(config: dict, agent: Agent)

Bases: BaseModule

A model predictive controller. More info to follow.

assert_mpc_variables_are_in_model() → dict[str, set[str]]

Checks whether all variables of var_ref are contained in the model. Returns names of model variables not contained in the var_ref, sorted by keys: ‘states’, ‘inputs’, ‘outputs’, ‘parameters’.

assert_subset(mpc_names, model_names, message_head)

Helper function for assert assert_mpc_variables_are_in_model. Asserts the variables of the var_ref corresponding to ref_key are a subset of a list of names provided (usually obtained from the model) and prints out an error if false. Returns the portion of model_names that are not in the given var_ref.

cleanup_results()

Deletes all files this module created.

Override this method, if your module creates e.g. results files etc.

collect_variables_for_optimization(var_ref: VariableReference = None) → Dict[str, AgentVariable]

Gets all variables noted in the var ref and puts them in a flat dictionary.

do_step()

Performs an MPC step.

get_results() → DataFrame | None

Read the results that were saved from the optimization backend and returns them as Dataframe.

Returns:

(results, stats) tuple of Dataframes.

property model: Model

Getter for current simulation model

Returns:

Current simulation model

Return type:

agentlib.model

pre_computation_hook()

This method is called in every computation step before the optimization starts. Overwrite this method in a derived subclass if you want to take some actions each time before the optimal control problem is solved.

process()

This abstract method must be implemented in order to sync the module with the other processes of the agent and the whole MAS.

re_init_optimization(parameter: AgentVariable)

Re-initializes the optimization backend with new parameters.

static read_results_file(results_file: str) → Tuple[DataFrame, DataFrame]

Read the provided csv-file as an MPC results file. :param results_file: File path

Returns:

results, stats results is the Dataframe with all inputs and outputs of the MPC optimizations. stats is the Dataframe with matching solver stats

register_callbacks()

Registers the init_optimization callback to all parameters which cannot be changed without recreating the optimization problem.

set_actuation(solution: Results)

Takes the solution from optimization backend and sends the first step to AgentVariables.

set_output(solution: Results)

Takes the solution from optimization backend and sends it to AgentVariables.

warn_for_missed_solves(stats: DataFrame | None)

Read the solver information from the optimization :returns: Warning if solver fails

pydantic model agentlib_mpc.modules.mpc.BaseMPCConfig

Bases: BaseModuleConfig

Pydantic data model for MPC configuration parser

Config:

• arbitrary_types_allowed: bool = True

• validate_assignment: bool = True

• extra: str = forbid

• frozen: bool = True

Fields:

• controls (List[agentlib_mpc.data_structures.mpc_datamodels.MPCVariable])

• inputs (List[agentlib_mpc.data_structures.mpc_datamodels.MPCVariable])

• optimization_backend (dict)

• outputs (List[agentlib_mpc.data_structures.mpc_datamodels.MPCVariable])

• parameters (List[agentlib_mpc.data_structures.mpc_datamodels.MPCVariable])

• prediction_horizon (int)

• sampling_time (float | None)

• set_outputs (bool)

• shared_variable_fields (list[str])

• states (List[agentlib_mpc.data_structures.mpc_datamodels.MPCVariable])

• time_step (float)

Validators:

• check_valid_fields » shared_variable_fields

• default_sampling_time » sampling_time

field controls: List[MPCVariable] = []

List of all control variables of the MPC.

field inputs: List[MPCVariable] = []

List of all input variables of the MPC. Includes predictions for disturbances, set_points, dynamic constraint boundaries etc.

field optimization_backend: dict [Required]

field outputs: List[MPCVariable] = []

List of all shared outputs of the MPC.

field parameters: List[MPCVariable] = []

List of model parameters of the MPC. They are constant over the horizon. Parameters not listed here will have their default from the model file.

field prediction_horizon: int = 5

Prediction horizon of the MPC.

Constraints:

• ge = 0

field sampling_time: float | None = None

Sampling interval for control steps. If None, will be the same as time step. Does not affect the discretization of the MPC, only the interval with which there will be optimization steps.

Validated by:

• default_sampling_time

field set_outputs: bool = False

Sets the full output time series to the data broker.

field shared_variable_fields: list[str] = ['outputs', 'controls']

Validated by:

• check_valid_fields

field states: List[MPCVariable] = []

List of all differential states of the MPC. The entries can define the boundaries and the source for the measurements

field time_step: float = 60

Time step of the MPC.

Constraints:

• ge = 0

validator default_sampling_time  »  sampling_time

model_post_init(context: Any, /) → None

We need to both initialize private attributes and call the user-defined model_post_init method.

agentlib_mpc.modules.mpc.create_optimization_backend(optimization_backend, agent_id)

Set up the optimization_backend

Module contents

This package contains all modules for the distributed model predictive control using multi agent systems.

It contains classes for local optimization and global coordination.

class agentlib_mpc.modules.ModuleImport(module_path: str, class_name: str)

Bases: object

import_class()