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

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

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

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]

name: str

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

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

Bases: CouplingVariable

delta_mean: ndarray

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

local_trajectories: Dict[Source, list]

mean_trajectory: list

multipliers: Dict[Source, list]

primal_residual: ndarray

shift_values_by_one(horizon: int)[source]: Shifts the trajectories

update_mean_trajectory(sources: Iterable[Source] = None)[source]

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

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

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

Bases: object

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

admm_variables()[source]

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

Bases: object

Holds information about a phy

delta_mean: ndarray

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

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

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

Bases: object

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

admm_variables()[source]

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

Bases: CouplingVariable

delta_mean: ndarray

diff_trajectories: Dict[Source, list[float]]

local_trajectories: Dict[Source, list]

mean_trajectory: list

multiplier: list[float]

primal_residual: ndarray

shift_values_by_one(horizon: int)[source]: Shifts the trajectories

update_diff_trajectories(sources: Iterable[Source] = None)[source]

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

Performs the multiplier update.

Parameters:: rho – penalty parameter

Returns:

class agentlib_mpc.data_structures.admm_datatypes.StructuredValue[source]

Bases: object

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

classmethod from_json(data: str)[source]

to_json() → str[source]: 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>)[source]

Bases: FullVariableReference

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

all_variables() → List[str][source]: Returns a list of all variables registered in the var_ref

controls: List[str]

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

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

classmethod from_config(config: BaseModuleConfigClass)[source]: Creates an instance from a pydantic values dict which includes lists of AgentVariables with the keys corresponding to ‘states’, ‘inputs’, etc..

inputs: List[str]

outputs: List[str]

parameters: List[str]

states: List[str]

agentlib_mpc.data_structures.admm_datatypes.coupling_alias(alias: str) → str[source]: Naming convention for local variables to send and receive.

agentlib_mpc.data_structures.admm_datatypes.exchange_alias(alias: str) → str[source]: 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[source]

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)
prediction_horizon (int)
time_step (float)

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

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.casadi_utils.CollocationMethod(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

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

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

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

Bases: str, Enum

cvodes = 'cvodes'

euler = 'euler'

rk = 'rk'

class agentlib_mpc.data_structures.casadi_utils.ModelConstraint(lb, function, ub)[source]

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

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

Bases: OptParMXContainer

Stores the necessary MX variables created during discretization for OptimizationVariables.

grid: List[float]

guess: List[MX]

lb: List[MX]

opt: DM = None

ub: List[MX]

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

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

do_jit: bool

logger: Logger = None

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

name: str = None

options: SolverOptions

pydantic model agentlib_mpc.data_structures.casadi_utils.SolverOptions[source]

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

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

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

agentlib_mpc.data_structures.coordinator_datatypes module

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

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

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

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

Bases: object

Contains specific variables (or time series) of the agent

classmethod from_dict(data: dict)[source]

to_dict() → dict[source]

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

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.interpolation module

class agentlib_mpc.data_structures.interpolation.InterpolationMethods(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: str, Enum

linear = 'linear'

mean_over_interval = 'mean_over_interval'

no_interpolation = 'no_interpolation'

previous = 'previous'

spline3 = 'spline3'

agentlib_mpc.data_structures.ml_model_datatypes module

pydantic model agentlib_mpc.data_structures.ml_model_datatypes.EarlyStoppingCallback[source]

Bases: BaseModel

Fields:

activate (bool)
patience (int)
restore_best_weights (bool)
verbose (Literal[0, 1])

field activate: bool = False

field patience: int = (1000,)

field restore_best_weights: bool = True

field verbose: Literal[0, 1] = 0

callback()[source]

pydantic model agentlib_mpc.data_structures.ml_model_datatypes.Feature[source]

Bases: BaseModel

Fields:

lag (int)
name (str)

field lag: int = 1

field name: str [Required]

pydantic model agentlib_mpc.data_structures.ml_model_datatypes.OutputFeature[source]

Bases: Feature

Fields:

output_type (agentlib_mpc.data_structures.ml_model_datatypes.OutputType)
recursive (bool)

Validators:

non_recursive_features_have_to_be_absolute » recursive

field output_type: OutputType [Required]: What kind of output this is. If ‘absolute’ an forward pass of the MLModel will yield the absolute value of the featuer at the next time step. If it is ‘difference’, the difference to the last time step will be generated, so it has to be added again.

field recursive: bool = True

If the output feature is recursive, it will also be used as an input for the model. This is useful for mpc, where we want to model the evolution of states based on their previous value. If false, can be used to model algebraic relationships. Default is true.

Validated by:

non_recursive_features_have_to_be_absolute

validator non_recursive_features_have_to_be_absolute » recursive[source]

class agentlib_mpc.data_structures.ml_model_datatypes.OutputType(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: str, Enum

absolute = 'absolute'

difference = 'difference'

class agentlib_mpc.data_structures.ml_model_datatypes.TrainingData(training_inputs: DataFrame, training_outputs: DataFrame, validation_inputs: DataFrame, validation_outputs: DataFrame, test_inputs: DataFrame, test_outputs: DataFrame)[source]

Bases: object

Stores the data which is used to train a model.

classmethod load(path: Path)[source]

save(path: Path)[source]: Saves three csv files in the path location. The csv files contain the test, training and validation data

test_inputs: DataFrame

test_outputs: DataFrame

training_inputs: DataFrame

training_outputs: DataFrame

validation_inputs: DataFrame

validation_outputs: DataFrame

agentlib_mpc.data_structures.ml_model_datatypes.column_order(inputs: dict[str, agentlib_mpc.data_structures.ml_model_datatypes.Feature], outputs: dict[str, agentlib_mpc.data_structures.ml_model_datatypes.OutputFeature]) → list[str][source]: Defines the order of the columns in which Training data should be passed to keras, and saved for the Serialization.

agentlib_mpc.data_structures.ml_model_datatypes.name_with_lag(name: str, lag: int) → str[source]

agentlib_mpc.data_structures.mpc_datamodels module

class agentlib_mpc.data_structures.mpc_datamodels.BaseVariableReference[source]

Bases: object

all_variables() → List[str][source]: Returns a list of all variables registered in the var_ref

classmethod from_config(config: BaseModuleConfigClass)[source]: 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[source]

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

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

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

Bases: BaseVariableReference

all_variables() → Iterable[str][source]: 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>)[source]

Bases: VariableReference

binary_controls: List[str]

controls: List[str]

inputs: List[str]

outputs: List[str]

parameters: List[str]

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

Bases: AgentVariable

AgentVariable used to define input variables of MPC.

interpolation_method: InterpolationMethods

class agentlib_mpc.data_structures.mpc_datamodels.Results(*args, **kwargs)[source]

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

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

agentlib_mpc.data_structures.mpc_datamodels.r_del_u_convention(name: str) → str[source]: Turns the name of a control variable into its weight via convention

agentlib_mpc.data_structures.mpc_datamodels.stats_path(path: Path | str) → Path[source]