agentlib_mpc.modules.mpc package

Submodules

agentlib_mpc.modules.mpc.mpc module

Holds the base class for MPCs.

class agentlib_mpc.modules.mpc.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.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 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

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.mpc.create_optimization_backend(optimization_backend, agent_id)[source]

Set up the optimization_backend

agentlib_mpc.modules.mpc.mpc_full module

Holds the class for full featured MPCs.

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

Bases: BaseMPC, SkippableMixin

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.mpc_full.MPCConfig[source]

Bases: BaseMPCConfig, SkippableMixinConfig

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.

agentlib_mpc.modules.mpc.skippable_mixin module

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

Bases: BaseModule

check_if_should_be_skipped()[source]

Checks if mpc steps should be skipped based on external activation flag.

pydantic model agentlib_mpc.modules.mpc.skippable_mixin.SkippableMixinConfig[source]

Bases: BaseModuleConfig

Config:

  • arbitrary_types_allowed: bool = True

  • validate_assignment: bool = True

  • extra: str = forbid

  • frozen: bool = True

Fields:
Validators:
field active: AgentVariable = AgentVariable(name='MPC_FLAG_ACTIVE', type='bool', timestamp=None, unit='Not defined', description='MPC is active', ub=inf, lb=-inf, clip=False, allowed_values=[], value=True, alias='MPC_FLAG_ACTIVE', source=Source(agent_id=None, module_id=None), shared=False, rdf_class=None)

Variable used to activate or deactivate the MPC operation

Validated by:
field deactivation_source: Source | None = None

Source for the deactivation signal.

field enable_deactivation: bool = False

If true, the MPC module uses an AgentVariable active whichother modules may change to disable the MPC operation temporarily

validator add_deactivation_source  »  active[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.