agentlib.utils package

Module containing all util functions for the agentlib.

Most notably, the custom injection enabling dynamic loading of custom models and modules.

agentlib.utils.custom_injection(config: dict, module_name: str | None = None)

Function to dynamically load new python files into the agentlib. Using this, users may use custom modules oder custom models together with the existing agentlib objects.

Parameters:

• config (dict) – Config dict containing the following items: file (str): Filepath to a python file (.py) class_name (str): Name of the class to be imported

• module_name (str, optional) – Name of the imported module in the sys.modules list. Carefully check if duplicate module keys raise unexpected behaviour. If so, use randomly generated strings or similar in classes calling this function. Default is None. In that case, the path is converted to a matching string.

Returns:

The class object specified by class_name

Return type:

class (object)

Subpackages

• agentlib.utils.plotting package
  • Submodules
  • agentlib.utils.plotting.simulator_dashboard module
    • create_layout()
    • create_plot()
    • format_time_axis()
    • get_port()
    • load_new_data()
    • simulator_dashboard()
    • update_data()

Submodules

agentlib.utils.broker module

Module containing a basic Broker which may be inherited by specialized classes.

class agentlib.utils.broker.Broker(*args, **kwargs)

Bases: object

Base Broker class

delete_all_clients()

Delete all clients from the list of clients

delete_client(client)

Delete the given client from the list of clients

register_client(client)

Append the given client to the list of clients.

class agentlib.utils.broker.Singleton

Bases: type

Global singleton to ensure only one broker exists

agentlib.utils.fuzzy_matching module

agentlib.utils.fuzzy_matching.fuzzy_match(target: str, choices: Iterable[str]) → None | List[str]

agentlib.utils.load_config module

agentlib.utils.load_config.load_config(config: ConfigT | Path | str | dict, config_type: Type[ConfigT]) → ConfigT

Generic config loader, either accepting a path to a json file, a json string, a dict or passing through a valid config object.

agentlib.utils.local_broadcast_broker module

Module containing a LocalBroadcastBroker that enables local broadcast communication.

class agentlib.utils.local_broadcast_broker.LocalBroadcastBroker(*args, **kwargs)

Bases: Broker

Local broadcast broker class which broadcasts messages

broadcast(agent_id: str, message: bytes | AgentVariable)

Broadcast message object to all agents but itself

agentlib.utils.local_broker module

Module containing a LocalBroker that enables local communication with subscriptions.

class agentlib.utils.local_broker.LocalBroker(*args, **kwargs)

Bases: Broker

Local broker class which sends messages

send(agent_id: str, message: bytes | AgentVariable)

Send the given message to all clients if the source matches. :param agent_id: Source to match :param message: The message to send

Returns:

agentlib.utils.multi_agent_system module

Module containing a local agency to test any LocalMASAgency system without the need of cloneMAP.

pydantic model agentlib.utils.multi_agent_system.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.utils.multi_agent_system.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.

pydantic model agentlib.utils.multi_agent_system.MAS

Bases: BaseModel

Parent class for all MAS

Config:

• arbitrary_types_allowed: bool = True

Fields:

• agent_configs (List[dict | pathlib.Path | str])

• env (agentlib.core.environment.Environment | dict | pathlib.Path)

• variable_logging (bool)

Validators:

• setup_agents » agent_configs

field agent_configs: List[dict | Path | str] [Required]

Validated by:

• setup_agents

field env: Environment | dict | Path [Optional]

The environment for the agents.

field variable_logging: bool = False

Enable variable logging in all agents with sampling rate of environment.

add_agent(config: AgentConfig)

Add an agent to the local agency with the given agent config.

Parameters:

Dict (config) – agent config

static add_agent_logger(config: AgentConfig, sampling=60) → AgentConfig

Adds the AgentLogger to the list of configs.

Parameters:

• dict (config) – The config to be updated

• sampling= –

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.

abstract run(until)

Run the MAS. :param until: The time until which the simulation should run.

Returns:

validator setup_agents  »  agent_configs

Load agent configs and add them.

pydantic model agentlib.utils.multi_agent_system.MultiProcessingMAS

Bases: MAS

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

Config:

• arbitrary_types_allowed: bool = True

Fields:

• agent_configs (List[Union[dict, FilePath, str]])

• cleanup (bool)

• env (dict | pathlib.Path)

• log_level (int)

• variable_logging (bool)

Validators:

• setup_agents » agent_configs

• setup_env » env

field agent_configs: List[dict | FilePath | str] [Required]

Validated by:

• setup_agents

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.

field variable_logging: bool = False

Enable variable logging in all agents with sampling rate of environment.

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.

agentlib.utils.multi_agent_system.agent_process(agent_config: dict | Path, until: float, env: dict | Path, results_dict: dict, cleanup=True, log_level=40)

Function to initialize and start an agent in its own process. Collects results from the agent and stores them in the passed results_dict. :param cleanup: :param agent_config: Config for an agent. :param until: Simulation runtime :param env: config for an environment :param results_dict: dict from process manager :param log_level: the log level for this process

Returns:

agentlib.utils.multi_processing_broker module

Module containing a MultiProcessingBroker that enables communication across different processes.

class agentlib.utils.multi_processing_broker.BrokerManager(address=None, authkey=None, serializer='pickle', ctx=None)

Bases: SyncManager

get_queue(*args, **kwds)

class agentlib.utils.multi_processing_broker.MPClient(agent_id, read, write)

Bases: tuple

agent_id

Alias for field number 0

read

Alias for field number 1

write

Alias for field number 2

class agentlib.utils.multi_processing_broker.Message(agent_id, payload)

Bases: tuple

agent_id

Alias for field number 0

payload

Alias for field number 1

class agentlib.utils.multi_processing_broker.MultiProcessingBroker(*args, **kwargs)

Bases: Broker

Singleton which acts as a broker for distributed simulations among multiple local processes. Establishes a connection to a multiprocessing.Manager object, which defines a queue. This queue is used to receive connection requests from local clients. The clients send a Conn object (from multiprocessing.Pipe()) object through which the connection is established. For each connected client, a thread waits for incoming objects.

property config: MultiProcessingBrokerConfig

Return the config of the environment

send(source, message)

Send the given message to all clients if the source matches. :param source: Source to match :param message: The message to send

Returns:

pydantic model agentlib.utils.multi_processing_broker.MultiProcessingBrokerConfig

Bases: BaseModel

Class describing the configuration options for the MultiProcessingBroker.

Fields:

• authkey (bytes)

• ipv4 (ipaddress.IPv4Address)

• port (int)

field authkey: bytes = b'useTheAgentlib'

Authorization key for the connection with the broker.

field ipv4: IPv4Address = '127.0.0.1'

IP Address for the communication server. Defaults to localhost.

field port: int = 50000

Port for setting up the connection with the server.

agentlib.utils.plugin_import module

Module containing all function to import new plugins

pydantic model agentlib.utils.plugin_import.ModuleImport

Bases: BaseModel

Data-Class to import a given python file from import_path and load the given class_name

Fields:

• class_name (str)

• import_path (str)

field class_name: str [Required]

field import_path: str [Required]

import_class()

Import the Module with class_name from the import path

class agentlib.utils.plugin_import.SaveUpdateDict

Bases: dict

Custom object to safely update the dictionary. Duplicate entries will raise an error.

update(new_dict, **kwargs) → None

Check if all modules have distinct identifier strings

Parameters:

• new_dict –

• **kwargs –

agentlib.utils.plugin_import.load_plugin(name: str, loaded_classes: SaveUpdateDict, plugin_types_name: str)

Loads the plugin based on the given name.

Parameters:

• str (plugin_types_name) – Name of the plugin

• SaveUpdateDict (loaded_classes) – SaveUpdateDict instance with already loaded classes (modules or models)

• str – Name of the dictionary in the plugin. Typical values are “MODULE_TYPES” or “MODEL_TYPES”.

agentlib.utils.validators module

Module with validator function used in multiple parts of the agentlib

agentlib.utils.validators.convert_to_list(obj)

Function to convert an object to a list. Either is is already a list. Or it is None, then [] is returned, or it is a scalar object and thus converted to a list.

agentlib.utils.validators.include_defaults_in_root(variables: ~typing.List[~typing.Dict], field: ~pydantic.fields.FieldInfo, make_shared: bool, agent_id: str, field_name: str, type_: ~typing.Type[~agentlib.core.datamodels.AgentVariable] = <class 'agentlib.core.datamodels.AgentVariable'>) → List[AgentVariable]

Validator building block to merge default variables with config variables in the root 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.

agentlib.utils.validators.is_list_of_agent_variables(ls: Any)

agentlib.utils.validators.is_valid_agent_var_config(data: dict, field_name: str, type_: ~agentlib.core.datamodels.AgentVariable = <class 'agentlib.core.datamodels.AgentVariable'>)

agentlib.utils.validators.update_default_agent_variable(default_var: AgentVariable, user_data: dict, make_shared: bool, agent_id: str, field_name: str)

Update a variable based on it’s default