agentlib_mpc.modules package

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)[source]

Bases: object

import_class()[source]

Subpackages

Submodules

agentlib_mpc.modules.data_source module

class agentlib_mpc.modules.data_source.DataSource(config: dict, agent: Agent)[source]

Bases: BaseModule

process()[source]

Write the current data values into data_broker every t_sample

register_callbacks()[source]

Don’t do anything as this module is not event-triggered

transform_index(data: DataFrame) → DataFrame[source]

Handles the index and ensures it is numeric, with correct offset

pydantic model agentlib_mpc.modules.data_source.DataSourceConfig[source]

Bases: BaseModuleConfig

Config:

  • arbitrary_types_allowed: bool = True

  • validate_assignment: bool = True

  • extra: str = forbid

  • frozen: bool = True

Fields:
Validators:
field columns: List[str] | None = None

Optional list of columns of data frame that should be sent.If ommited, all datapoint in frame are sent.

field data: DataFrame | Path = Empty DataFrame Columns: [] Index: []

Data that should be communicated during execution.Index should be either numeric or Datetime, numeric values are interpreted as seconds.

Validated by:
field data_offset: Timedelta | float | None = 0

Offset will be subtracted from index.

field interpolation_method: InterpolationMethods | None = InterpolationMethods.previous

Interpolation method used for resampling of data.Only ‘linear’ and ‘previous’ are allowed.

Validated by:
field shared: bool | None = True

Indicates if the variable is going to be shared with other agents.

field t_sample: float | int = 1

Sample time of data source. Default is 1 s.

validator check_data  »  data[source]

Makes sure data is a data frame, and loads it if required.

validator check_interpolation_method  »  interpolation_method[source]
model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

  • self – The BaseModel instance.

  • context – The context.

agentlib_mpc.modules.minlp_mpc module

agentlib_mpc.modules.ml_model_simulator module

agentlib_mpc.modules.mpc module

Holds the base class for MPCs.

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

Bases: BaseModule

A model predictive controller. More info to follow.

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

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)[source]

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()[source]

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][source]

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

do_step()[source]

Performs an MPC step.

get_results() → DataFrame | None[source]

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()[source]

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()[source]

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)[source]

Re-initializes the optimization backend with new parameters.

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

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()[source]

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

set_actuation(solution: Results)[source]

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

set_output(solution: Results)[source]

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

warn_for_missed_solves(stats: DataFrame | None)[source]

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

pydantic model agentlib_mpc.modules.mpc.BaseMPCConfig[source]

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:
Validators:
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 log_level: str | None = None

The log level for this Module. Default uses the root-loggers level.Options: DEBUG; INFO; WARNING; ERROR; CRITICAL

Validated by:

  • check_valid_level

field module_id: str [Required]

The unqiue id of the module within an agent, used only to communicate withing the agent.

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:
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

field type: str | Dict[str, str] [Required]

The type of the Module. Used to find the Python-Object from all agentlib-core and plugin Module options. If a dict is given,it must contain the keys ‘file’ and ‘class_name’. ‘file’ is the filepath of a python file containing the Module.’class_name’ is the name of the Module class within this file.

field validate_incoming_values: bool | None = True

If true, the validator of the AgentVariable value is called when receiving a new value from the DataBroker.

validator default_sampling_time  »  sampling_time[source]
model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

  • self – The BaseModel instance.

  • context – The context.

agentlib_mpc.modules.mpc.create_optimization_backend(optimization_backend, agent_id)[source]

Set up the optimization_backend

agentlib_mpc.modules.mpc_full module

Holds the class for full featured MPCs.

class agentlib_mpc.modules.mpc_full.MPC(config: dict, agent: Agent)[source]

Bases: BaseMPC

A model predictive controller. More info to follow.

collect_variables_for_optimization(var_ref: VariableReference = None) → dict[str, agentlib.core.datamodels.AgentVariable][source]

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

do_step()[source]

Performs an MPC step.

register_callbacks_for_lagged_variables()[source]

Registers callbacks which listen to the variables which have to be saved as time series. These callbacks save the values in the history for use in the optimization.

pydantic model agentlib_mpc.modules.mpc_full.MPCConfig[source]

Bases: BaseMPCConfig

Pydantic data model for MPC configuration parser

Config:

  • arbitrary_types_allowed: bool = True

  • validate_assignment: bool = True

  • extra: str = forbid

  • frozen: bool = True

Fields:
Validators:
field r_del_u: dict[str, float] = {}

Weights that are applied to the change in control variables.

Validated by:
validator check_r_del_u_in_controls  »  r_del_u[source]

Ensures r_del_u is only set for control variables.

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

  • self – The BaseModel instance.

  • context – The context.