agentlib_mpc.data_structures package

Submodules

agentlib_mpc.data_structures.admm_datatypes module

class agentlib_mpc.data_structures.admm_datatypes.ADMMParameters(penalty_factor: float, prediction_horizon: int, time_step: float)

Bases: object

Collection of parameters which have to be shared across all agents in ADMM.

penalty_factor: float

prediction_horizon: int

time_step: float

class agentlib_mpc.data_structures.admm_datatypes.ADMMParticipation(source: Source, ready: bool = False, participating: bool = False)

Bases: object

Helper class to organize ADMM participants.

participating: bool = False

ready: bool = False

source: Source

class agentlib_mpc.data_structures.admm_datatypes.AgentDictEntry(name: str, status: ~agentlib_mpc.data_structures.coordinator_datatypes.AgentStatus = AgentStatus.pending, coup_vars: ~typing.List[str] = <factory>, exchange_vars: ~typing.List[str] = <factory>)

Bases: AgentDictEntry

Holds participating coupling variables (consensus and exchange) of a single agent in ADMM. Used in the coordinator.

coup_vars: List[str]

exchange_vars: List[str]

class agentlib_mpc.data_structures.admm_datatypes.AgentToCoordinator(local_trajectory: Dict[str, numpy.ndarray], local_exchange_trajectory: Dict[str, numpy.ndarray])

Bases: StructuredValue

local_exchange_trajectory: Dict[str, ndarray]

local_trajectory: Dict[str, ndarray]

class agentlib_mpc.data_structures.admm_datatypes.ConsensusVariable(local_trajectories: Dict[agentlib.core.datamodels.Source, list] = <factory>, mean_trajectory: list = <factory>, delta_mean: numpy.ndarray = <factory>, primal_residual: numpy.ndarray = <factory>, multipliers: Dict[agentlib.core.datamodels.Source, list] = <factory>)

Bases: CouplingVariable

flat_multipliers(sources: Iterable[Source] = None) → list[float]

multipliers: Dict[Source, list]

shift_values_by_one(horizon: int)

Shifts the trajectories

update_mean_trajectory(sources: Iterable[Source] = None)

Calculates a new mean of this variable.

Parameters:

sources – List of sources that should be included in the update. If none is given, use all variables.

update_multipliers(rho: float, sources: Iterable[Source] = None)

Performs the multiplier update.

Parameters:

• rho – penalty parameter

• sources – list of sources that should be included in the update. By default, all are included.

Returns:

class agentlib_mpc.data_structures.admm_datatypes.CoordinatorToAgent(target: str, mean_trajectory: Dict[str, list], multiplier: Dict[str, list], mean_diff_trajectory: Dict[str, list], exchange_multiplier: Dict[str, list], penalty_parameter: float)

Bases: StructuredValue

exchange_multiplier: Dict[str, list]

mean_diff_trajectory: Dict[str, list]

mean_trajectory: Dict[str, list]

multiplier: Dict[str, list]

penalty_parameter: float

target: str

class agentlib_mpc.data_structures.admm_datatypes.CouplingEntry(name: str)

Bases: object

Holds naming conventions for different optimizatin variables / parameters associated with a coupling variable in consensus ADMM.

admm_variables()

property lagged

property local

property mean

property multiplier

name: str

class agentlib_mpc.data_structures.admm_datatypes.CouplingVariable(local_trajectories: ~typing.Dict[~agentlib.core.datamodels.Source, list] = <factory>, mean_trajectory: list = <factory>, delta_mean: ~numpy.ndarray = <factory>, primal_residual: ~numpy.ndarray = <factory>)

Bases: object

Holds information about a phy

delta_mean: ndarray

flat_locals(sources: Iterable[Source] = None) → list[float]

Returns the flattened array of all local variables and their multipliers.

Parameters:

sources – list of sources that should be included in the update. By default, all are included.

Returns:

flat lists of local variables and multipliers (locals, multipliers)

get_residual(rho: float) → Tuple[ndarray, ndarray]

Returns the primal and dual residual of the last iteration as a tuple of flattened Arrays. :param rho:

Returns:

(primal residual, dual residual)

local_trajectories: Dict[Source, list]

mean_trajectory: list

property participants

Returns all agent sources that are registered to this coupling.

primal_residual: ndarray

class agentlib_mpc.data_structures.admm_datatypes.ExchangeEntry(name: str)

Bases: object

Holds naming conventions for different optimizatin variables / parameters associated with a coupling variable in exchange ADMM.

admm_variables()

property lagged

property local

property mean_diff

property multiplier

name: str

class agentlib_mpc.data_structures.admm_datatypes.ExchangeVariable(local_trajectories: Dict[agentlib.core.datamodels.Source, list] = <factory>, mean_trajectory: list = <factory>, delta_mean: numpy.ndarray = <factory>, primal_residual: numpy.ndarray = <factory>, diff_trajectories: Dict[agentlib.core.datamodels.Source, list[float]] = <factory>, multiplier: list[float] = <factory>)

Bases: CouplingVariable

diff_trajectories: Dict[Source, list[float]]

multiplier: list[float]

shift_values_by_one(horizon: int)

Shifts the trajectories

update_diff_trajectories(sources: Iterable[Source] = None)

Calculates a new mean of this variable.

Parameters:

sources – List of sources that should be included in the update. If none is given, use all variables.

update_multiplier(rho: float)

Performs the multiplier update.

Parameters:

rho – penalty parameter

Returns:

class agentlib_mpc.data_structures.admm_datatypes.StructuredValue

Bases: object

Base Class to specify the structure of an AgentVariable Value. It will be efficiently sent and deserialized.

classmethod from_json(data: str)

to_json() → str

Serialize self to json bytes, can be used by the communicator.

class agentlib_mpc.data_structures.admm_datatypes.VariableReference(states: ~typing.List[str] = <factory>, controls: ~typing.List[str] = <factory>, inputs: ~typing.List[str] = <factory>, parameters: ~typing.List[str] = <factory>, outputs: ~typing.List[str] = <factory>, couplings: list[agentlib_mpc.data_structures.admm_datatypes.CouplingEntry] = <factory>, exchange: list[agentlib_mpc.data_structures.admm_datatypes.ExchangeEntry] = <factory>)

Bases: FullVariableReference

Holds info about all variables of an MPC and their role in the optimization problem.

all_variables() → List[str]

Returns a list of all variables registered in the var_ref

couplings: list[agentlib_mpc.data_structures.admm_datatypes.CouplingEntry]

exchange: list[agentlib_mpc.data_structures.admm_datatypes.ExchangeEntry]

classmethod from_config(config: BaseModuleConfigClass)

Creates an instance from a pydantic values dict which includes lists of AgentVariables with the keys corresponding to ‘states’, ‘inputs’, etc..

agentlib_mpc.data_structures.admm_datatypes.coupling_alias(alias: str) → str

Naming convention for local variables to send and receive.

agentlib_mpc.data_structures.admm_datatypes.exchange_alias(alias: str) → str

Naming convention for local exchange variables to send and receive.

agentlib_mpc.data_structures.casadi_utils module

Stores all sorts of Dataclasses, Enums or Factories to help with the CasadiBackend.

pydantic model agentlib_mpc.data_structures.casadi_utils.CasadiDiscretizationOptions

Bases: DiscretizationOptions

Config:

• extra: str = forbid

Fields:

• collocation_method (agentlib_mpc.data_structures.casadi_utils.CollocationMethod)

• collocation_order (int)

• integrator (agentlib_mpc.data_structures.casadi_utils.Integrators)

• method (agentlib_mpc.data_structures.casadi_utils.DiscretizationMethod)

field collocation_method: CollocationMethod = CollocationMethod.legendre

field collocation_order: int = 3

Constraints:

• ge = 1

• le = 9

field integrator: Integrators = Integrators.cvodes

field method: DiscretizationMethod = DiscretizationMethod.collocation

class agentlib_mpc.data_structures.casadi_utils.CollocationMethod(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: str, Enum

legendre = 'legendre'

radau = 'radau'

class agentlib_mpc.data_structures.casadi_utils.Constraint(function: casadi.casadi.MX, lb: casadi.casadi.MX, ub: casadi.casadi.MX)

Bases: object

function: MX

lb: MX

ub: MX

class agentlib_mpc.data_structures.casadi_utils.DiscretizationMethod(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: str, Enum

collocation = 'collocation'

multiple_shooting = 'multiple_shooting'

class agentlib_mpc.data_structures.casadi_utils.Integrators(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: str, Enum

cvodes = 'cvodes'

euler = 'euler'

rk = 'rk'

class agentlib_mpc.data_structures.casadi_utils.ModelConstraint(lb, function, ub)

Bases: NamedTuple

function: MX

Alias for field number 1

lb: MX | SX | Sparsity | DM | float | int

Alias for field number 0

ub: MX | SX | Sparsity | DM | float | int

Alias for field number 2

class agentlib_mpc.data_structures.casadi_utils.OptParMXContainer(var: ~typing.List[~casadi.casadi.MX] = <factory>, grid: ~typing.List[float] = <factory>)

Bases: object

Stores the necessary MX variables created during discretization for OptimizationParameters.

grid: List[float]

var: List[MX]

class agentlib_mpc.data_structures.casadi_utils.OptVarMXContainer(var: ~typing.List[~casadi.casadi.MX] = <factory>, grid: ~typing.List[float] = <factory>, lb: ~typing.List[~casadi.casadi.MX] = <factory>, ub: ~typing.List[~casadi.casadi.MX] = <factory>, guess: ~typing.List[~casadi.casadi.MX] = <factory>, opt: ~casadi.casadi.DM = None)

Bases: OptParMXContainer

Stores the necessary MX variables created during discretization for OptimizationVariables.

guess: List[MX]

lb: List[MX]

opt: DM = None

ub: List[MX]

class agentlib_mpc.data_structures.casadi_utils.SolverFactory(do_jit: bool, bat_file: ~pathlib.Path = None, name: str = None, options: ~agentlib_mpc.data_structures.casadi_utils.SolverOptions = <factory>, logger: ~logging.Logger = None)

Bases: object

Creates a solver given an NLP and an options construct.

bat_file: Path = None

create_solver(nlp: dict | str, discrete: List[bool] = None, equalities: list[bool] = None) → Function

do_jit: bool

logger: Logger = None

make_casadi_nlp(problem: dict | str, solver: str, opts: dict, problem_type: Literal['nlp', 'qp'] = 'nlp')

name: str = None

options: SolverOptions

pydantic model agentlib_mpc.data_structures.casadi_utils.SolverOptions

Bases: BaseModel

Config:

• extra: str = forbid

Fields:

• name (agentlib_mpc.data_structures.casadi_utils.Solvers)

• options (dict)

field name: Solvers = 'ipopt'

field options: dict = {}

class agentlib_mpc.data_structures.casadi_utils.Solvers(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: str, Enum

bonmin = 'bonmin'

fatrop = 'fatrop'

gurobi = 'gurobi'

ipopt = 'ipopt'

osqp = 'osqp'

proxqp = 'proxqp'

qpoases = 'qpoases'

sqpmethod = 'sqpmethod'

agentlib_mpc.data_structures.casadi_utils.compile_solver(bat_file: Path, name: str, optimizer: Function) → str

Code-generates an ipopt solver and compiles it. Currently, only works on Windows! Requires a batch file that knows how to setup Visual Studio command line and compiles the source code.

Returns:

The Path to the .dll file for the compiled solver.

Raises:

• TypeError –

• FileNotFoundError –

• RuntimeError –

agentlib_mpc.data_structures.casadi_utils.temporary_directory(path)

agentlib_mpc.data_structures.coordinator_datatypes module

class agentlib_mpc.data_structures.coordinator_datatypes.AgentDictEntry(name: str, status: AgentStatus = AgentStatus.pending)

Bases: object

Dataclass holding the status of a participating agent in DMPC.

name: str

optimization_data = OptimizationData(x=array([], dtype=float64), u=array([], dtype=float64))

status: AgentStatus = 'pending'

class agentlib_mpc.data_structures.coordinator_datatypes.AgentStatus(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: str, Enum

Enum used within a DMPC-Coordinator to keep track of the statuses of its participating agents.

busy = 'busy'

pending = 'pending'

ready = 'ready'

standby = 'standby'

class agentlib_mpc.data_structures.coordinator_datatypes.CoordinatorStatus(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: str, Enum

Enum used to keep track of the status of a DMPC-Coordinator.

init_iterations = 'init_iterations'

optimization = 'optimization'

sleeping = 'sleeping'

updating = 'updating'

class agentlib_mpc.data_structures.coordinator_datatypes.OptimizationData(x: ~numpy.array = <factory>, u: ~numpy.array = <factory>)

Bases: object

Contains specific variables (or time series) of the agent

classmethod from_dict(data: dict)

to_dict() → dict

u: array

x: array

class agentlib_mpc.data_structures.coordinator_datatypes.RegistrationMessage(status: AgentStatus = None, opts: dict = None, agent_id: str = None, coupling: list = None)

Bases: object

Dataclass structuring the communication during registration between a participating agent and the coordinator in DMPC.

agent_id: str = None

coupling: list = None

opts: dict = None

status: AgentStatus = None

agentlib_mpc.data_structures.mpc_datamodels module

class agentlib_mpc.data_structures.mpc_datamodels.BaseVariableReference

Bases: object

all_variables() → List[str]

Returns a list of all variables registered in the var_ref

classmethod from_config(config: BaseModuleConfigClass)

Creates an instance from a pydantic values dict which includes lists of AgentVariables with the keys corresponding to ‘states’, ‘inputs’, etc..

pydantic model agentlib_mpc.data_structures.mpc_datamodels.DiscretizationOptions

Bases: BaseModel

Class defining the options to discretize an MPC. Can be extended for different optimization implementations.

Config:

• extra: str = allow

Fields:

• prediction_horizon (int)

• time_step (float)

field prediction_horizon: int = 5

Prediction horizon of the MPC.

Constraints:

• ge = 0

field time_step: float = 60

Time step of the MPC.

Constraints:

• ge = 0

class agentlib_mpc.data_structures.mpc_datamodels.FullVariableReference(states: List[str] = <factory>, controls: List[str] = <factory>, inputs: List[str] = <factory>, parameters: List[str] = <factory>, outputs: List[str] = <factory>)

Bases: VariableReference

property r_del_u: List[str]

class agentlib_mpc.data_structures.mpc_datamodels.InitStatus(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: str, Enum

Keep track of the readyness status of the MPC.

during_update = '2'

pre_module_init = '1'

ready = '3'

class agentlib_mpc.data_structures.mpc_datamodels.MHEVariableReference(states: List[str] = <factory>, measured_states: List[str] = <factory>, weights_states: List[str] = <factory>, estimated_inputs: List[str] = <factory>, estimated_parameters: List[str] = <factory>, known_inputs: List[str] = <factory>, known_parameters: List[str] = <factory>, outputs: List[str] = <factory>)

Bases: BaseVariableReference

all_variables() → Iterable[str]

Returns a list of all variables registered in the var_ref which the MHE can get from the config with get()

estimated_inputs: List[str]

estimated_parameters: List[str]

known_inputs: List[str]

known_parameters: List[str]

measured_states: List[str]

outputs: List[str]

states: List[str]

weights_states: List[str]

class agentlib_mpc.data_structures.mpc_datamodels.MINLPVariableReference(states: List[str] = <factory>, controls: List[str] = <factory>, inputs: List[str] = <factory>, parameters: List[str] = <factory>, outputs: List[str] = <factory>, binary_controls: List[str] = <factory>)

Bases: VariableReference

binary_controls: List[str]

class agentlib_mpc.data_structures.mpc_datamodels.MPCVariable(*, name: str, type: str | None = None, timestamp: float | None = None, unit: str = 'Not defined', description: str = 'Not defined', ub=inf, lb=-inf, clip: bool = False, allowed_values: List[Any] = [], value: Any = None, alias: str = None, source: dict | Source | str = Source(agent_id=None, module_id=None), shared: bool | None = None, rdf_class: str | None = None, interpolation_method: InterpolationMethods = InterpolationMethods.linear)

Bases: AgentVariable

AgentVariable used to define input variables of MPC.

interpolation_method: InterpolationMethods

class agentlib_mpc.data_structures.mpc_datamodels.Results(*args, **kwargs)

Bases: Protocol

df: DataFrame

class agentlib_mpc.data_structures.mpc_datamodels.VariableReference(states: List[str] = <factory>, controls: List[str] = <factory>, inputs: List[str] = <factory>, parameters: List[str] = <factory>, outputs: List[str] = <factory>)

Bases: BaseVariableReference

controls: List[str]

inputs: List[str]

outputs: List[str]

parameters: List[str]

states: List[str]

agentlib_mpc.data_structures.mpc_datamodels.cia_relaxed_results_path(path: Path | str) → Path

agentlib_mpc.data_structures.mpc_datamodels.r_del_u_convention(name: str) → str

Turns the name of a control variable into its weight via convention

agentlib_mpc.data_structures.mpc_datamodels.stats_path(path: Path | str) → Path

Module contents