agentlib.core package

Core module of the agentlib. The core holds all classes and functions relevant to use the agentlib. Besides some utils in the utils package, you may only check core files to understand how the agentlib works.

Submodules

agentlib.core.agent module

Module containing only the Agent class.

class agentlib.core.agent.Agent(*, config, env: Environment)[source]

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

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

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

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

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

Calls the terminate function of all modules.

pydantic model agentlib.core.agent.AgentConfig[source]

Bases: BaseModel

Class containing settings / config for an Agent.

Contains just two fields, id and modules.

Fields:
Validators:
field check_alive_interval: float = 1

Check every other check_alive_interval second if the threads of the agent are still alive.If that’s not the case, exit the main thread of the agent. Updating this value at runtime will not work as all processes have already been started.

Constraints:

  • ge = 0

field id: str | int [Required]

The ID of the Agent, should be unique in the multi-agent-system the agent is living in.

field max_queue_size: int | None = 1000

Maximal number of waiting items in data-broker queues. Set to -1 for infinity

Constraints:

  • ge = -1

field modules: List[Dict | Path] = None
Validated by:
validator check_modules  »  modules[source]

Validator to ensure all modules are in dict-format.

agentlib.core.agent.get_module_class(module_config)[source]

Return the Module-Class object for the given config.

Parameters:

module_config (dict) – Config of the module to return

Returns:

Module-Class object

Return type:

BaseModule

agentlib.core.data_broker module

The module contains the relevant classes to execute and use the DataBroker. Besides the DataBroker itself, the BrokerCallback is defined.

Internally, uses the tuple _map_tuple in the order of

(alias, source)

to match callbacks and AgentVariables.

pydantic model agentlib.core.data_broker.BrokerCallback[source]

Bases: NoCopyBrokerCallback

This broker callback always creates a deep-copy of the AgentVariable it is going to send. It is considered the safer option, as the receiving module only get’s the values and is not able to alter the AgentVariable for other modules.

Config:

  • arbitrary_types_allowed: bool = True

Fields:
Validators:
field alias: str | None = None
Validated by:
field callback: Callable [Required]
Validated by:
field kwargs: dict = {}
Validated by:
field module_id: str | None = None
Validated by:
field source: Source | None = None
Validated by:
validator auto_copy  »  callback[source]

Automatically supply the callback function with a copy

class agentlib.core.data_broker.DataBroker(logger: CustomLogger, max_queue_size: int = 1000)[source]

Bases: ABC

Handles communication and Callback triggers within an agent. Write variables to the broker with send_variable(). Variables send to the broker will trigger callbacks based on the alias and the source of the variable. Commonly, this is used to provide other modules with the variable.

Register and de-register Callbacks to the DataBroker with register_callback and deregister_callback.

static any_is_none(alias: str, source: Source) → bool[source]

Return True if any of alias or source are None.

Parameters:

  • str (alias) – The alias of the callback

  • Source (source) – The Source of the callback

deregister_callback(callback: Callable, alias: str | None = None, source: Source | None = None, **kwargs)[source]

Deregister the given callback based on given alias and source.

Parameters:

  • callable (callback) – The function of the callback

  • str (alias) – The alias of variables to trigger callback

  • Source (source) – The Source of variables to trigger callback

  • dict (kwargs) – Kwargs of the callback function

register_callback(callback: Callable, alias: str | None = None, source: Source | None = None, _unsafe_no_copy: bool = False, **kwargs) → BrokerCallback | NoCopyBrokerCallback[source]

Register a callback to the data_broker.

Parameters:

  • callable (callback) – The function of the callback

  • str (alias) – The alias of variables to trigger callback

  • Source (source) – The Source of variables to trigger callback

  • dict (kwargs) – Kwargs to be passed to the callback function

  • _unsafe_no_copy – If True, the callback will not be passed a copy, but the original AgentVariable. When using this option, the user promises to not modify the AgentVariable, as doing so could lead to wrong and difficult to debug behaviour in other modules (default False)

send_variable(variable: AgentVariable, copy: bool = True)[source]

Send variable to data_broker. Evokes callbacks associated with this variable.

Parameters:

  • AgentVariable (variable) – The variable to set.

  • boolean (copy) – Whether to copy the variable before sending. Default is True.

class agentlib.core.data_broker.LocalDataBroker(env: Environment, logger: CustomLogger, max_queue_size: int = 1000)[source]

Bases: DataBroker

Local variation of the DataBroker written for fast-as-possible simulation within a single non-realtime Environment.

pydantic model agentlib.core.data_broker.NoCopyBrokerCallback[source]

Bases: BaseModel

Basic broker callback. This object does not copy the AgentVariable before calling the callback, which can be unsafe.

This class checks if the given callback function adheres to the signature it needs to be correctly called. The first argument will be an AgentVariable. If a type-hint is specified, it must be AgentVariable or “AgentVariable”. Any further arguments must match the kwargs specified in the class and will also be the ones you pass to this class.

Example: >>> def my_callback(variable: “AgentVariable”, some_static_info: str): >>> print(variable, some_other_info) >>> NoCopyBrokerCallback( >>> callback=my_callback, >>> kwargs={“some_static_info”: “Hello World”} >>> )

Config:

  • arbitrary_types_allowed: bool = True

Fields:
Validators:
field alias: str | None = None
Validated by:
field callback: Callable [Required]
Validated by:
field kwargs: dict = {}
Validated by:
field module_id: str | None = None
Validated by:
field source: Source | None = None
Validated by:
validator check_valid_callback_function  »  all fields[source]

Ensures the callback function signature is valid.

class agentlib.core.data_broker.RTDataBroker(env: Environment, logger: CustomLogger, max_queue_size: int = 1000)[source]

Bases: DataBroker

DataBroker written for Realtime operation regardless of Environment.

register_callback(callback: Callable, alias: str | None = None, source: Source | None = None, _unsafe_no_copy: bool = False, **kwargs) → NoCopyBrokerCallback | BrokerCallback[source]

Register a callback to the data_broker.

Parameters:

  • callable (callback) – The function of the callback

  • str (alias) – The alias of variables to trigger callback

  • Source (source) – The Source of variables to trigger callback

  • dict (kwargs) – Kwargs to be passed to the callback function

  • _unsafe_no_copy – If True, the callback will not be passed a copy, but the original AgentVariable. When using this option, the user promises to not modify the AgentVariable, as doing so could lead to wrong and difficult to debug behaviour in other modules (default False)

agentlib.core.data_broker.log_queue_status(logger: Logger, queue_object: Queue, max_queue_size: int, queue_name: str)[source]

Log the current load of the given queue in percent.

Parameters:

  • logger (logging.Logger) – A logger instance

  • queue_object (queue.Queue) – The queue object

  • max_queue_size (int) – Maximal queue size

  • queue_name (str) – Name associated with the queue

agentlib.core.datamodels module

The datamodels module contains all classes defining basic models to handle data.

class agentlib.core.datamodels.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)[source]

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

Creates a copy of the Variable.

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

Generates a dict from the Variable.

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

Instantiates a new AgentVariable from json.

rdf_class: str | None
run_validation()[source]

Performs all validations.

shared: bool | None
source: Source | str
class agentlib.core.datamodels.BaseVariable(*, 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)[source]

Bases: AttrsToPydanticAdaptor

BaseVariable for all Variables inside the agentlib. This includes Model Variables and AgentVariables. A Variable can have an arbitrary value with several forms of validation for type, boundaries and allowed values.

allowed_values: List[Any]
clip: bool
classmethod create(data: dict | BaseVariable) → BaseVariable[source]

Constructor for pydantic.

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

Generates a dict from the Variable.

json() → str[source]

Serializes the Variable in json format and returns a string

lb: float | int
name: str
run_validation()[source]

Performs all validations.

set_value(value, validate: bool = False)[source]

Sets the value of the variable. If validate is True (default False), also do the following:

  • Convert to the given type

  • Check if inside the list of allowed_values

  • If bounds can be applied, check if inside the given bounds and maybe even clip accordingly

timestamp: float | None
type: str | None
ub: float | int
unit: str
classmethod validate_data(data: dict) → BaseVariable[source]

Constructor that performs all validation.

value: Any
class agentlib.core.datamodels.Causality(value)[source]

Bases: str, Enum

Enumeration that defines the causality of a variable.

The default causality is “local”.

Allowed params of his enumeration:

calculatedParameter = 'calculatedParameter'
independent = 'independent'
input = 'input'
local = 'local'
output = 'output'
parameter = 'parameter'
class agentlib.core.datamodels.ModelInput(*, name: str, 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, causality: Causality | None = None, variability: Variability | None = None, type: str | None = 'float', sim_time: float = 0.0)[source]

Bases: ModelVariable

The ModelInput variable. Inherits

  • BaseInput: The causality and variability associated with an input

  • ModelVariable: The fields unique to a ModelVariable.

  • BaseModelVariable: All fields associated with any model variable.

class agentlib.core.datamodels.ModelOutput(*, name: str, 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, causality: Causality | None = None, variability: Variability | None = None, type: str | None = 'float', sim_time: float = 0.0)[source]

Bases: ModelVariable

The ModelOutput variable. Inherits

  • BaseOutput: The causality and variability associated with an output

  • ModelVariable: The fields unique to a ModelVariable.

  • BaseModelVariable: All fields associated with any model variable.

class agentlib.core.datamodels.ModelParameter(*, name: str, 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, causality: Causality | None = None, variability: Variability | None = None, type: str | None = 'float', sim_time: float = 0.0)[source]

Bases: ModelVariable

The ModelParameter variable. Inherits

  • BaseParameter: The causality and variability associated with a parameter

  • ModelVariable: The fields unique to a ModelVariable.

class agentlib.core.datamodels.ModelState(*, name: str, 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, causality: Causality | None = None, variability: Variability | None = None, type: str | None = 'float', sim_time: float = 0.0)[source]

Bases: ModelVariable

The ModelState variable. Inherits

  • BaseLocal: The causality and variability associated with a local

  • ModelVariable: The fields unique to a ModelVariable.

  • BaseModelVariable: All fields associated with any model variable.

class agentlib.core.datamodels.ModelVariable(*, name: str, 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, causality: Causality | None = None, variability: Variability | None = None, type: str | None = 'float', sim_time: float = 0.0)[source]

Bases: BaseModelVariable

The basic ModelVariable. Aside from only allowing number for values, this class enables calculation with the object itself.

sim_time: float
class agentlib.core.datamodels.Source(agent_id: str | None = None, module_id: str | None = None)[source]

Bases: AttrsToPydanticAdaptor

Object to define the source of a variable or possible other object. As objects are passed both module-internally by agents or across multiple agents, both the agent_id and module_id build up the source object.

However, with methods like ‘matches’, one can indicate setting any id to None that the id is irrelevant.

agent_id: str | None
classmethod create(data: dict | Source | str)[source]

Constructor for this class, used by pydantic.

dict()[source]

Overwrite pydantic method to be faster.

json() → str[source]

Returns json serialization of the Source

matches(other) → bool[source]

Define if the current source matches another source:

First, convert other object to a dict. The dict must contain the variables agent_id and module_id. If one of the values is None, it is excluded from the comparison. :param other Union[Source: Another source to compare :param Dict]: Another source to compare

Returns:

Boolean if they match

module_id: str | None
class agentlib.core.datamodels.Variability(value)[source]

Bases: str, Enum

Enumeration that defines the time dependency of the variable, in other words,it defines the time instants when a variable can change its value. [The purpose of this attribute is to define when a result value needs to be inquired and to be stored. For example, discrete variables change their params only at event instants (ModelExchange) or at a communication point (CoSimulation) and it is therefore only necessary to inquire and store them at event times].

The default is “continuous”

Allowed params of this enumeration:

constant = 'constant'
continuous = 'continuous'
discrete = 'discrete'
fixed = 'fixed'
tunable = 'tunable'

agentlib.core.environment module

This module contains the Environment class, used by all Agents and Modules.

This module contains modified code of simpy (https://gitlab.com/team-simpy/simpy). Simpy is distributed under the MIT License

The MIT License (MIT)

Copyright (c) 2013 Ontje Lünsdorf and Stefan Scherfke (also see AUTHORS.txt)

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the “Software”), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

class agentlib.core.environment.CustomSimpyEnvironment(initial_time: int | float = 0)[source]

Bases: Environment

A customized version of the simpy environment. Handles execution of modules processes and manages time for instant execution mode.

clock()[source]

Define a clock loop to increase the now-timer every other second (Or whatever t_sample is)

property config: EnvironmentConfig

Return the config of the environment

property offset: EnvironmentConfig

Start time offset of the environment.

pretty_time()[source]
property time: float

Get the current time of the environment.

class agentlib.core.environment.Environment(*args, **kwargs)[source]

Bases: object

Simpy Environment Distributor. Handles synchronous processes.

pydantic model agentlib.core.environment.EnvironmentConfig[source]

Bases: BaseModel

Config for the Environment

Config:

  • validate_assignment: bool = True

  • arbitrary_types_allowed: bool = True

  • extra: str = forbid

Fields:
field clock: bool = False
field factor: float = 1.0
Constraints:

  • gt = 0

field offset: float = 0

Used to offset the now-time ofthe environment using the clock function.

field rt: bool = False
field strict: bool = False
field t_sample: float = 60

Used to increase the now-time ofthe environment using the clock function.

Constraints:

  • gt = 0

class agentlib.core.environment.InstantEnvironment(*, config: EnvironmentConfig)[source]

Bases: CustomSimpyEnvironment

A customized version of the simpy environment. Handles execution of modules processes and manages time for instant execution mode.

pretty_time() → str[source]

Returns the time in seconds.

class agentlib.core.environment.RealtimeEnvironment(*, config: EnvironmentConfig)[source]

Bases: RealtimeEnvironment, CustomSimpyEnvironment

A customized version of the simpy environment. Handles execution of modules processes and manages time for real time execution mode.

pretty_time() → str[source]

Returns the time in a datetime format.

run(until: int | float | Event | None = None) → Any | None[source]

Executes step() until the given criterion until is met.

  • If it is None (which is the default), this method will return when there are no further events to be processed.

  • If it is an Event, the method will continue stepping until this event has been triggered and will return its value. Raises a RuntimeError if there are no further events to be processed and the until event was not triggered.

  • If it is a number, the method will continue stepping until the environment’s time reaches until.

silent_clock()[source]

A silent clock, which does not log anything.

property time: float

Get the current system time as unix timestamp, with the enivronement offset.

class agentlib.core.environment.ScaledRealtimeEnvironment(*, config: EnvironmentConfig)[source]

Bases: RealtimeEnvironment, CustomSimpyEnvironment

A customized version of the simpy environment. Handles execution of modules processes and manages time for scaled real time execution mode.

pretty_time() → str[source]

Returns the time in seconds.

run(until: int | float | Event | None = None) → Any | None[source]

Executes step() until the given criterion until is met.

  • If it is None (which is the default), this method will return when there are no further events to be processed.

  • If it is an Event, the method will continue stepping until this event has been triggered and will return its value. Raises a RuntimeError if there are no further events to be processed and the until event was not triggered.

  • If it is a number, the method will continue stepping until the environment’s time reaches until.

silent_clock()[source]

A silent clock, which does not log anything.

property time: float

Get the current time of the environment.

agentlib.core.environment.make_env_config(config: dict | EnvironmentConfig | str | None) → EnvironmentConfig[source]

Creates the environment config from different sources.

agentlib.core.environment.monkey_patch_simpy_process()[source]

Removes the exception catching in simpy processes. This removes some of simpys features that we do not need. In return, it improves debugging and makes error messages more concise.

agentlib.core.errors module

This module implements custom errors to ensure a better understand of errors in the agentlib

exception agentlib.core.errors.ConfigurationError[source]

Bases: Exception

Exception raised for errors due to wrong configuration of modules.

exception agentlib.core.errors.InitializationError[source]

Bases: Exception

Exception raised for errors due to wrong initialization of modules.

exception agentlib.core.errors.OptionalDependencyError(used_object: str, dependency_install: str, dependency_name: str | None = None)[source]

Bases: Exception

Exception to indicate that an optional dependency is missing which can always be fixed by installing the missing dependency.

agentlib.core.logging_ module

class agentlib.core.logging_.CustomLogger(name, env: Environment, level=0)[source]

Bases: Logger

Subclass of Logger that adds the env_time to the record, allowing it to print the current time.

makeRecord(name, level, fn, lno, msg, args, exc_info, func=None, extra=None, sinfo=None)[source]

A factory method which can be overridden in subclasses to create specialized LogRecords.

agentlib.core.logging_.create_logger(env: Environment, name: str) → CustomLogger[source]

Creates a logger that displays the environment time when logging.

agentlib.core.model module

This module contains just the basic Model.

class agentlib.core.model.Model(**kwargs)[source]

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

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

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

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][source]
get_input(name: str)[source]

Get model input based on given name.

get_input_names()[source]
Returns:

A list containing all input names

Return type:

names (list)

get_inputs(names: List[str])[source]

Get model inputs based on given names.

get_output(name: str)[source]

Get model output based on given name.

get_output_names()[source]
Returns:

A list containing all output names

Return type:

names (list)

get_outputs(names: List[str])[source]

Get model outputs based on given names.

get_parameter(name: str)[source]

Get model parameter based on given name.

get_parameter_names()[source]
Returns:

A list containing all state names

Return type:

names (list)

get_parameters(names: List[str])[source]

Get model parameters based on given names.

get_state(name: str)[source]

Get model state based on given name.

get_state_names()[source]
Returns:

A list containing all state names

Return type:

names (list)

get_states(names: List[str])[source]

Get model states based on given names.

abstract initialize(**kwargs)[source]

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

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

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

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

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

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

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

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

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

Terminate the model if applicable by subclass.

property variables

Get all model variables as a list

pydantic model agentlib.core.model.ModelConfig[source]

Bases: BaseModel

Pydantic data model for controller configuration parser

Config:

  • validate_assignment: bool = True

  • arbitrary_types_allowed: bool = True

  • extra: str = forbid

Fields:
Validators:
field description: str = 'You forgot to document your model!'
field dt: float | int = 1
field inputs: List[ModelInput] = []
Validated by:
field name: str | None = None
Validated by:
field outputs: List[ModelOutput] = []
Validated by:
field parameters: List[ModelParameter] = []
Validated by:
field sim_time: float = 0
field states: List[ModelState] = []
Validated by:
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[source]

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

get_variable_names()[source]

Returns the names of every variable as list

validator include_default_model_variables  »  inputs, parameters, outputs, states[source]

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.

agentlib.core.module module

This module contains the base AgentModule.

class agentlib.core.module.BaseModule(*, config: dict, agent: Agent)[source]

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

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

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

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

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

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()[source]
set(name: str, value: Any, timestamp: float | None = None)[source]

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

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

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.core.module.BaseModuleConfig[source]

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

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

validator check_valid_fields  »  shared_variable_fields[source]

Check if the shared_variables_fields are valid fields.

validator check_valid_level  »  log_level[source]

Check if the given log_level is valid

classmethod default(field: str)[source]
get_variables()[source]

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

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

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.