agentlib package

Subpackages

• agentlib.core package
  • Submodules
  • agentlib.core.agent module
  • agentlib.core.data_broker module
  • agentlib.core.datamodels module
  • agentlib.core.environment module
  • agentlib.core.errors module
  • agentlib.core.model module
  • agentlib.core.module module
  • Module contents
• agentlib.models package
  • Submodules
  • agentlib.models.fmu_model module
  • agentlib.models.scipy_model module
  • Module contents
• agentlib.modules package
  • Subpackages
  • Submodules
  • agentlib.modules.simulator module
  • Module contents
• agentlib.utils package
  • Submodules
  • agentlib.utils.broker module
  • agentlib.utils.fuzzy_matching module
  • agentlib.utils.load_config module
  • agentlib.utils.local_broadcast_broker module
  • agentlib.utils.local_broker module
  • agentlib.utils.multi_agent_system module
  • agentlib.utils.multi_processing_broker module
  • agentlib.utils.plugin_import module
  • agentlib.utils.validators module
  • Module contents

Module contents

Top-level module of the agentlib. Besides import of submodules, nothing happens here.

class agentlib.Agent(*, config, env: Environment)

Bases: object

The base class for all reactive agent implementations.

Parameters:

• config (Union[AgentConfig, FilePath, str, dict]) – A config object to initialize the agents config

• env (Environment) – The environment the agent is running in

clean_results()

Calls the cleanup_results function of all modules, removing files that were created by them.

property config: AgentConfig

Get the config (AgentConfig) of the agent

Returns:

An instance of AgentConfig

Return type:

AgentConfig

property data_broker: DataBroker

Get the data_broker of the agent

Returns:

An instance of the DataBroker class

Return type:

DataBroker

property env: CustomSimpyEnvironment

Get the environment the agent is in

Returns:

The environment instance

Return type:

Environment

get_module(module_id: str) → BaseModuleClass

Get the module by given module_id. If no such module exists, None is returned :param module_id: Id of the module to return :type module_id: str

Returns:

Module with the given name

Return type:

BaseModule

get_results(cleanup=True)

Gets the results of this agent. :param cleanup: If true, created files are deleted.

property id: str

Getter for current agent’s id

Returns:

current id of agent

Return type:

str

property modules: List[BaseModuleClass]

Get all modules of agent

Returns:

List of all modules

Return type:

List[BaseModule]

register_thread(thread: Thread)

Registers the given thread to the dictionary of threads which need to run in order for the agent to work.

Parameters:

threading.Thread (thread) – The thread object

terminate()

Calls the terminate function of all modules.

class agentlib.AgentVariable(*, 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 = None, alias: str | None = None, source: dict | Source | str = Source(agent_id=None, module_id=None), shared: bool | None = None, rdf_class: str | None = None)

Bases: BaseVariable

The basic AgentVariable. The AgentVariable is the central messaging piece in the AgentLib. It can hold arbitrary (but json-serializable!) values as Agent States, Configuration objects or

messages.

In addition to fields defined in BaseVariable, any AgentVariable holds the - alias: The publicly known name of the variable - source: Indicating which agent and or module the variable belong to - shared: Whether the variable is going to be shared to other Agents - rdf_class: Class in the resource description framework

Check the description of each field for further information.

alias: str

copy(update: dict | None = None, deep: bool = False)

Creates a copy of the Variable.

dict(exclude: Container[str] | None = None) → dict

Generates a dict from the Variable.

classmethod from_json(s: str | bytes, validate=False)

Instantiates a new AgentVariable from json.

rdf_class: str | None

run_validation()

Performs all validations.

shared: bool | None

source: Source | str

class agentlib.BaseModule(*, config: dict, agent: Agent)

Bases: ABC

Basic module used by any agent. Besides a common configuration, where ids and variables are defined, this class manages the setting and getting of variables and relevant attributes.

property agent: Agent

Get the agent this module is located in.

cleanup_results()

Deletes all files this module created.

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

property config: BaseModuleConfigClass

The module config.

Returns:

Config of type self.config_type

Return type:

BaseModuleConfigClass

property env: CustomSimpyEnvironment

Get the environment of the agent.

get(name: str) → AgentVariable

Get any variable matching the given name:

Parameters:

name (str) – The item to get by name of Variable. Hence, item=AgentVariable.name

Returns:

The matching variable

Return type:

var (AgentVariable)

Raises:

KeyError – If the item was not found in the variables of the module.

classmethod get_config_type() → Type[BaseModuleConfigClass]

get_results()

Returns results of this modules run.

Override this method, if your module creates data that you would like to obtain

after the run.

Returns:

Some form of results data, often in the form of a pandas DataFrame.

get_value(name: str) → Any

Get the value of the variable matching the given name:

Parameters:

name (str) – The item to get by name of Variable. Hence, item=AgentVariable.name

Returns:

The matching value

Return type:

var (Any)

Raises:

KeyError – If the item was not found in the variables of the module.

property id: str

Get the module’s id

abstract process()

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

abstract register_callbacks()

set(name: str, value: Any, timestamp: float | None = None)

Set any variable by using the name:

Parameters:

• name (str) – The item to get by name of Variable. Hence, item=AgentVariable.name

• value (Any) – Any value to set to the Variable

• timestamp (float) – The timestamp associated with the variable. If None, current environment time is used.

Raises:

AttributeError – If the item was not found in the variables of the module.

property source: Source

Get the source of the module, containing the agent and module id

terminate()

Terminate all relevant processes of the module. This is necessary to correctly terminate an agent at runtime. Not all modules may need this, hence it is not an abstract method.

update_variables(variables: List[AgentVariable], timestamp: float | None = None)

Updates the given list of variables in the current data_broker. If a given Variable is not in the config of the module, an error is raised. TODO: check if this is needed, we currently don’t use it anywhere

Parameters:

• variables – List with agent_variables.

• timestamp – The timestamp associated with the variable. If None, current environment time is used.

property variables: List[AgentVariable]

Return all values as a list.

pydantic model agentlib.BaseModuleConfig

Bases: BaseModel

Pydantic data model for basic module configuration

Config:

• arbitrary_types_allowed: bool = True

• validate_assignment: bool = True

• extra: str = forbid

• frozen: bool = True

Fields:

• log_level (str | None)

• module_id (str)

• shared_variable_fields (List[str])

• type (str | Dict[str, str])

• validate_incoming_values (bool | None)

Validators:

• check_valid_fields » shared_variable_fields

• check_valid_level » log_level

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 shared_variable_fields: List[str] = []

A list of strings with each string being a field of the Modules configs. The field must be or contain an AgentVariable. If the field is added to this list, all shared attributes of the AgentVariables will be set to True.

Validated by:

• check_valid_fields

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.

classmethod check_if_variables_are_unique(names)

Check if a given iterable of AgentVariables have a unique name.

validator check_valid_fields  »  shared_variable_fields

Check if the shared_variables_fields are valid fields.

validator check_valid_level  »  log_level

Check if the given log_level is valid

classmethod default(field: str)

get_variables()

Return the private attribute with all AgentVariables

classmethod merge_variables(pre_validated_instance: BaseModuleConfig, user_config: dict, agent_id: str, shared_variable_fields: List[str])

Merge, rigorously check and validate the input of all AgentVariables into the module. This function:

• Collects all variables

• Checks if duplicate names (will cause errors in the get() function.

classmethod model_json_schema(*args, **kwargs) → dict

Custom schema method to - Add JSON Schema for custom attrs types Source and AgentVariable - put log_level last, as it is the only optional field of the module config. Used to better display relevant options of children classes in GUIs.

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.

class agentlib.Environment(*args, **kwargs)

Bases: object

Simpy Environment Distributor. Handles synchronous processes.

pydantic model agentlib.LocalCloneMAPAgency

Bases: LocalMASAgency

Local LocalMASAgency agency class which tries to mimic cloneMAP behaviour for the local execution.

Config:

• arbitrary_types_allowed: bool = True

Fields:

Validators:

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.

run(until=None)

Execute the LocalMASAgency and terminate it after run is finished

setup_agent(id: str)

Setup the agent matching the given id

terminate_agents()

Terminate all agents modules.

pydantic model agentlib.LocalMASAgency

Bases: MAS

Local LocalMASAgency agency class which holds the agents in a common environment, executes and terminates them.

Config:

• arbitrary_types_allowed: bool = True

Fields:

Validators:

• setup_env » env

add_agent(config: AgentConfig)

Also setup the agent directly

get_agent(id: str) → Agent

Get the agent matching the given id

get_results(cleanup: bool = True) → Dict[str, DataFrame]

Get all results of the agentLogger :param cleanup: If true, read files are deleted.

Returns:

key is the agent_id, value the dataframe

Return type:

Dict[str, pd.DataFrame]

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.

run(until)

Execute the LocalMASAgency and terminate it after run is finished

setup_agent(id: str) → Agent

Setup the agent matching the given id

validator setup_env  »  env

Setup the env if a config is given.

stop_agency()

Stop all threads

terminate_agents()

Terminate all agents modules.

class agentlib.Model(**kwargs)

Bases: ABC

Base class for simulation models. To implement your own model, inherit from this class.

property config: ModelConfig

Get the current config, which is a ModelConfig object.

property description

Get model description

abstract do_step(*, t_start: float, t_sample: float)

Performing one simulation step :param t_start: start time for integration :param t_sample: increment of solver integration

Returns:

property dt

Get time increment of simulation

generate_variables_config(filename: str | None = None, **kwargs) → str

Generate a config file (.json) to enable an user friendly configuration of the model.

Parameters:

• filename (str) – Optional path where to store the config. If None, current model name and workdir are used.

• kwargs – Kwargs directly passed to the json.dump method.

Returns:

Filepath where the json is stored

Return type:

filepath (str)

get(name: str) → ModelVariable

Get any variable from using name:

Parameters:

name (str) – The item to get from config by name of Variable. Hence, item=ModelVariable.name

Returns:

The matching variable

Return type:

var (ModelVariable)

Raises:

AttributeError – If the item was not found in the variables of the module.

classmethod get_config_type() → Type[ModelConfig]

get_input(name: str)

Get model input based on given name.

get_input_names()

Returns:

A list containing all input names

Return type:

names (list)

get_inputs(names: List[str])

Get model inputs based on given names.

get_output(name: str)

Get model output based on given name.

get_output_names()

Returns:

A list containing all output names

Return type:

names (list)

get_outputs(names: List[str])

Get model outputs based on given names.

get_parameter(name: str)

Get model parameter based on given name.

get_parameter_names()

Returns:

A list containing all state names

Return type:

names (list)

get_parameters(names: List[str])

Get model parameters based on given names.

get_state(name: str)

Get model state based on given name.

get_state_names()

Returns:

A list containing all state names

Return type:

names (list)

get_states(names: List[str])

Get model states based on given names.

abstract initialize(**kwargs)

Abstract method to define what to do in order to initialize the model in use.

property inputs: List[ModelInput]

Get all model inputs as a list

property name

Get model name

property outputs: List[ModelOutput]

Get all model outputs as a list

property parameters: List[ModelParameter]

Get all model parameters as a list

set(name: str, value: Any)

Set any variable from using name:

Parameters:

• name (str) – The item to get from data_broker by name of Variable. Hence, item=AgentVariable.name

• value (Any) – Any value to set to the Variable

Raises:

AttributeError – If the item was not found in the variables of the module.

set_input_value(name: str, value: float | int | bool)

Just used from external modules like simulator to set new input values

set_input_values(names: List[str], values: List[float | int | bool])

Just used from external modules like simulator to set new input values

set_parameter_value(name: str, value: float | int | bool)

Used externally to write new parameter values from e.g. a calibration process

set_parameter_values(names: List[str], values: List[float | int | bool])

Used externally to write new parameter values from e.g. a calibration process

property sim_time

Get the current simulation time

property states: List[ModelState]

Get all model states as a list

terminate()

Terminate the model if applicable by subclass.

property variables

Get all model variables as a list

pydantic model agentlib.ModelConfig

Bases: BaseModel

Pydantic data model for controller configuration parser

Config:

• validate_assignment: bool = True

• arbitrary_types_allowed: bool = True

• extra: str = forbid

Fields:

• description (str)

• dt (float | int)

• inputs (List[agentlib.core.datamodels.ModelInput])

• name (str | None)

• outputs (List[agentlib.core.datamodels.ModelOutput])

• parameters (List[agentlib.core.datamodels.ModelParameter])

• sim_time (float)

• states (List[agentlib.core.datamodels.ModelState])

• user_config (dict)

• validate_variables (bool)

Validators:

• check_name » name

• include_default_model_variables » inputs

• include_default_model_variables » outputs

• include_default_model_variables » parameters

• include_default_model_variables » states

field description: str = 'You forgot to document your model!'

field dt: float | int = 1

field inputs: List[ModelInput] = []

Validated by:

• include_default_model_variables

field name: str | None = None

Validated by:

• check_name

field outputs: List[ModelOutput] = []

Validated by:

• include_default_model_variables

field parameters: List[ModelParameter] = []

Validated by:

• include_default_model_variables

field sim_time: float = 0

field states: List[ModelState] = []

Validated by:

• include_default_model_variables

field user_config: dict = None

The config given by the user to instantiate this class.Will be stored to enable a valid overwriting of the default config and to better restart modules.Is also useful to debug validators and the general BaseModuleConfig.

field validate_variables: bool = True

If true, the validator of a variables value is called whenever a new value is set. Disabled by default for performance reasons.

validator check_name  »  name

Check if name of model is given. If not, use the name of the model class.

get_variable_names()

Returns the names of every variable as list

validator include_default_model_variables  »  inputs, parameters, outputs, states

Validator building block to merge default variables with config variables in a standard validator. Updates default variables when a variable with the same name is present in the config. Then returns the union of the default variables and the external config variables.

This validator ensures default variables are kept when the config provides new 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.

pydantic model agentlib.MultiProcessingMAS

Bases: MAS

Helper class to conveniently run multi-agent-systems in separate processes.

Config:

• arbitrary_types_allowed: bool = True

Fields:

• cleanup (bool)

• env (dict | pathlib.Path)

• log_level (int)

Validators:

• setup_env » env

field cleanup: bool = False

Whether agents should clean the results files after running.

field env: dict | Path [Optional]

The environment for the agents.

Validated by:

• setup_env

field log_level: int = 40

Loglevel to set for the processes.

get_results() → Dict[str, DataFrame]

Get all results of the agentLogger :returns: key is the agent_id, value the dataframe :rtype: Dict[str, pd.DataFrame]

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.

run(until)

Execute the multi-agent-system in parallel and terminate it after run is finished

validator setup_env  »  env

Setup the env if a config is given.