API reference

The API reference provides detailed descriptions of the mango’s classes and functions.

class mango.Agent

Base class for all agents.

add_forwarding_rule(from_addr: AgentAddress, to_addr: AgentAddress, forward_replies: bool = False) → None

Add an automatic message-forwarding rule.

After calling this, every message received from from_addr is automatically forwarded to to_addr. When forward_replies is True, replies originating from to_addr are forwarded back to from_addr.

Parameters:

• from_addr – source address to match

• to_addr – destination to forward to

• forward_replies – whether replies should be forwarded back

property addr

Return the address of the agent as AgentAddress

Returns:

_type_: AgentAddress

property aid

property category: str

Category tag for this agent.

property color: str

Visual color tag for this agent.

property current_timestamp: float

Method that returns the current unix timestamp given the clock within the container

delete_forwarding_rule(from_addr: AgentAddress, to_addr: AgentAddress | None = None) → None

Remove previously added forwarding rule(s).

Parameters:

• from_addr – source address of the rule to remove

• to_addr – if given, only remove rules that also match this destination; otherwise remove all rules matching from_addr

property description: AgentDescription

Return the agent’s AgentDescription.

handle_message(content, meta: dict[str, Any])

Has to be implemented by the user. This method is called when a message is received at the agents inbox. :param content: The deserialized message object :param meta: Meta details of the message. In case of mqtt this dict includes at least the field ‘topic’

property name: str

Human-readable name of this agent.

neighbors(state: State = State.NORMAL, *, tid: str = 'default', has_characteristic: str | None = None, include_connectors: tuple | list = (), match_func=None) → list[AgentAddress]

Return neighbor addresses from the topology.

Parameters:

• state – filter by edge state (default NORMAL)

• tid – topology identifier (default "default")

• has_characteristic – only include neighbors with this characteristic

• include_connectors – also include connector agents of these types

• match_func – optional (AgentDescription) -> bool predicate

Returns:

list of AgentAddress

property observable_tasks

on_agent_event(event: Any) → None

Called when a targeted agent event is emitted.

Override to react to events directed at this specific agent.

Parameters:

event – the event object

on_global_event(event: Any) → None

Called when a global event is emitted from the environment.

Override to react to environment-wide broadcasts.

Parameters:

event – the event object

on_ready()

Called when all container has been started using activate(…).

on_register()

Hook-in to define behavior of the agent directly after it got registered by a container

on_start()

Called when container started in which the agent is contained

on_step(env, clock, step_size_s: float) → None

Called on every simulation step (only in SimulationWorld).

Parameters:

• env – the simulation environment

• clock – the current simulation clock

• step_size_s – seconds advanced in this step

async on_stop()

Can be used as lifecycle callback when the agent is stopped

async reply_to(content: Any, received_meta: dict, **kwargs) → bool

Convenience helper to reply to a received message.

Extracts the sender address from received_meta and sends content back, preserving any tracking_id for transaction matching.

Parameters:

• content – reply content

• received_meta – the meta dict from the received message

Returns:

result of send_message()

schedule_conditional_process_task(coroutine_creator, condition_func, lookup_delay=0.1, on_stop=None, src=None)

Schedule a process task when a specified condition is met.

Parameters:

• coroutine_creator (coroutine_creator) – coroutine_creator creating coroutine to be scheduled

• condition_func (lambda () -> bool) – function for determining whether the confition is fullfiled

• lookup_delay (float) – delay between checking the condition

• src (Object) – creator of the task

schedule_conditional_task(coroutine, condition_func, lookup_delay=0.1, on_stop=None, src=None)

Schedule a task when a specified condition is met.

Parameters:

• coroutine (Coroutine) – coroutine to be scheduled

• condition_func (lambda () -> bool) – function for determining whether the confition is fullfiled

• lookup_delay (float) – delay between checking the condition

• src (Object) – creator of the task

schedule_instant_message(content, receiver_addr: AgentAddress, **kwargs)

Schedules sending a message without any delay. This is equivalent to using the schedulers ‘schedule_instant_task’ with the coroutine created by ‘container.send_message’.

Parameters:

• content – The content of the message

• receiver_addr – The address passed to the container

• kwargs – Additional parameters to provide protocol specific settings

Returns:

asyncio.Task for the scheduled coroutine

schedule_instant_process_task(coroutine_creator, on_stop=None, src=None)

Schedule an instantly executed task in another processes.

Parameters:

• coroutine_creator – coroutine_creator creating coroutine to be scheduled

• on_stop (Object) – coroutine to run on stop

• src (Object) – creator of the task

schedule_instant_task(coroutine, on_stop=None, src=None)

Schedule an instantly executed task.

Parameters:

• coroutine – coroutine to be scheduled

• on_stop (Object) – coroutine to run on stop

• src (Object) – creator of the task

schedule_periodic_process_task(coroutine_creator, delay, on_stop=None, src=None)

Schedule an open end periodically executed task in another process.

Parameters:

• coroutine_creator (Coroutine Function) – coroutine function creating coros to be scheduled

• delay (float) – delay in between the cycles

• on_stop (Object) – coroutine to run on stop

• src (Object) – creator of the task

schedule_periodic_task(coroutine_func, delay, on_stop=None, src=None)

Schedule an open end peridocally executed task.

Parameters:

• coroutine_func (Coroutine Function) – coroutine function creating coros to be scheduled

• delay (float) – delay in between the cycles

• on_stop (Object) – coroutine to run on stop

• src (Object) – creator of the task

schedule_process_task(task: ScheduledProcessTask, src=None)

Schedule a task with asyncio in another process. When the task is finished, if finite, its automatically removed afterwards. For scheduling options see the subclasses of ScheduledScheduledProcessTaskTask.

Parameters:

• task – task to be scheduled

• src – object, which represents the source of the task (for example the object in which the task got created)

schedule_recurrent_process_task(coroutine_creator, recurrency, on_stop=None, src=None)

Schedule a task using a fine-grained recurrency rule in another process.

Parameters:

• coroutine_creator (Coroutine Function) – coroutine function creating coros to be scheduled

• recurrency (dateutil.rrule.rrule) – recurrency rule to calculate next event

• on_stop (Object) – coroutine to run on stop

• src (Object) – creator of the task

schedule_recurrent_task(coroutine_func, recurrency, on_stop=None, src=None)

Schedule a task using a fine-grained recurrency rule in another process.

Parameters:

• coroutine_creator (Coroutine Function) – coroutine function creating coros to be scheduled

• recurrency (dateutil.rrule.rrule) – recurrency rule to calculate next event

• on_stop (Object) – coroutine to run on stop

• src (Object) – creator of the task

schedule_task(task: ScheduledTask, src=None)

Schedule a task with asyncio. When the task is finished, if finite, its automatically removed afterwards. For scheduling options see the subclasses of ScheduledTask.

Parameters:

• task – task to be scheduled

• src – object, which represents the source of the task (for example the object in which the task got created)

schedule_timestamp_process_task(coroutine_creator, timestamp: float, on_stop=None, src=None)

Schedule a task at specified unix timestamp dispatched to another process.

Parameters:

• coroutine_creator (coroutine_creator) – coroutine_creator creating coroutine to be scheduled

• timestamp (float) – unix timestamp defining when the task should start

• src (Object) – creator of the task

schedule_timestamp_task(coroutine, timestamp: float, on_stop=None, src=None)

Schedule a task at specified unix timestamp.

Parameters:

• coroutine (Coroutine) – coroutine to be scheduled

• timestamp (timestamp) – timestamp defining when the task should start

• src (Object) – creator of the task

async send_message(content, receiver_addr: AgentAddress, **kwargs) → bool

See container.send_message(…)

async send_messages(content, receiver_addrs: list[AgentAddress], **kwargs) → list[bool]

Send the same content to multiple recipients.

Parameters:

• content – message content (sent to every recipient)

• receiver_addrs – list of target AgentAddress instances

Returns:

list of send results (one per recipient, in order)

async send_tracked_message(content: Any, receiver_addr: AgentAddress, response_handler=None, **kwargs)

Send a message and optionally register a response handler.

A tracking_id is attached to the message so that the reply can be matched. When response_handler is provided it will be called as response_handler(reply_content, reply_meta) when the matching reply arrives.

Parameters:

• content – message content

• receiver_addr – target agent address

• response_handler – optional (content, meta) -> None callback

Returns:

the asyncio.Task for the sent message

service_of_type(type: type, default: Any = None) → Any

Return the service registered for type, creating it if absent.

If no service of type is registered yet, default is registered and returned; when default is None a new type() instance is created instead.

Parameters:

• type (type) – the type of the service

• default (Any (optional)) – the value to register if none exists; None creates a type() instance

Returns:

the service

Return type:

Any

async shutdown()

Shutdown all tasks that are running and deregister from the container

property suspendable_tasks

async tasks_complete(timeout=1)

Wait for all scheduled tasks to complete using a timeout.

Parameters:

timeout – waiting timeout. Defaults to 1.

property uid: str

Unique identifier (UUID string) of this agent.

update_description(name: str | None = None, color: str | None = None, category: str | None = None) → None

Update one or more description fields.

Parameters:

• name – new human-readable name

• color – new color tag

• category – new category tag

class mango.AgentAddress(protocol_addr: Any, aid: str)

aid: str

protocol_addr: Any

class mango.AgentDescription(name: str = '', category: str = 'agent', color: str = 'gray', uid: str = <factory>)

Metadata describing an agent (name, category, color, unique ID).

Mirrors the AgentDescription type in Mango.jl.

category: str = 'agent'

color: str = 'gray'

name: str = ''

uid: str

class mango.AgentNode(agents: list[Agent] | None = None)

A single node in a topology graph. Holds one or more agents.

add(*agents: Agent) → None

Add one or more agents to this node.

set_characteristic(agent: Agent, characteristic: str) → None

Assign a characteristic label to agent within this node.

class mango.AgentsRecording(timeseries: dict[str, list[~typing.Any]]=<factory>, agent_time: dict[str, list[float]]=<factory>, time: list[float] = <factory>)

Per-agent time-series recording.

timeseries maps each agent AID to a list of recorded values. agent_time maps each agent AID to the elapsed simulation seconds at which each of that agent’s values was recorded; it stays aligned with timeseries even when agents are recorded sparsely (e.g. registered mid-simulation or gated by a filter). time holds every step’s timestamp as a shared axis for agents recorded on every step.

agent_time: dict[str, list[float]]

time: list[float]

timeseries: dict[str, list[Any]]

class mango.Area2D(width: float = 10.0, height: float = 10.0)

A rectangular 2-D space.

Agents without a predefined position receive a random location within [0, width] × [0, height] during initialize().

Example:

space = Area2D(width=100.0, height=100.0)

agents_within(center, radius: float, agents: list) → list

Return all agents from agents within radius of center.

center itself is excluded from the result. Only agents that have a registered position are considered.

Parameters:

• center – the reference agent

• radius – search radius in space units

• agents – candidate agents to search among

Returns:

list of agents within radius of center

Example:

nearby = space.agents_within(my_agent, 5.0, world._agents.values())

distance(agent_a, agent_b) → float

Return the Euclidean distance between two agents.

Both agents must have a registered position.

Parameters:

• agent_a – first agent

• agent_b – second agent

Returns:

Euclidean distance in space units

Example:

d = space.distance(agent1, agent2)

has_position(agent) → bool

Return True if agent has a registered position.

initialize(agents: list, clock: Clock) → None

Called once before the first simulation step.

install(agent, **kwargs) → None

Install agent in the space (called when agent is added to env).

location(agent) → Position2D

Return the current position of agent.

move(agent, position: Position2D) → None

Move agent to position.

move_toward(agent, target: Position2D | object, max_step: float) → None

Move agent toward target by at most max_step units.

target may be either another agent (with a registered position) or a Position2D directly. If the agent is already within max_step of the target, it is moved exactly to the target position.

Parameters:

• agent – the agent to move

• target – destination agent or Position2D

• max_step – maximum distance to travel in one call

Example:

space.move_toward(rover, base_station, max_step=1.0)

class mango.AsyncioClock

The AsyncioClock

sleep(t) → Future

Sleeping via asyncio sleep

property time: float

Current time using the time module

class mango.Behavior

Abstract environment behavior.

Override on_step() to model environment dynamics and initialize() for one-time setup.

initialize(environment: Environment, clock: Clock) → None

Called once before the first simulation step.

install(agent, **kwargs) → None

Called when an agent is installed in the environment.

on_step(environment: Environment, clock: Clock, step_size_s: float) → None

Called on every simulation step.

class mango.CommunicationSimulation

Abstract base class for communication simulations.

Implement this to define custom message delay and loss behaviour. The same MessagePackage may be passed multiple times within a simulation step; implementations must return identical results for the same package (determinism requirement).

abstractmethod calculate_communication(current_time: float, messages: list[MessagePackage]) → CommunicationSimulationResult

Calculate delivery results for the given messages.

Parameters:

• current_time – current simulation time in seconds

• messages – messages to evaluate

Returns:

one PackageResult per message, in the same order

class mango.CommunicationSimulationResult(package_results: list[PackageResult])

Aggregated result for a set of message packages.

package_results: list[PackageResult]

class mango.DefaultEnvironment(space: Space | None = None, behavior: Behavior | None = None)

Full environment implementation with pluggable space and behavior.

Example:

env = DefaultEnvironment(space=Area2D(100, 100))
world = create_world(start_time=0.0, environment=env)

add_observer(observer: WorldObserver) → None

Register an observer to receive global events.

property behavior: Behavior

emit_agent_event(event: Any, agent_id: Any) → None

Deliver event to the agent registered under agent_id.

emit_global_event(event: Any) → None

Broadcast event to all registered observers.

initialize(agents: list, clock: Clock) → None

Initialize the environment with the given agents.

initialized() → bool

Return whether the environment has been initialized.

install(agent, agent_id: Any = None, **kwargs) → None

Register agent in the space and behavior.

Parameters:

• agent – the agent to install

• agent_id – identifier used to retrieve the agent later

property space: Space

step(clock: Clock, step_size_s: float) → None

Step the environment forward by step_size_s seconds.

class mango.DelayProviderCommunicationSimulation(default_delay_s_provider: ~collections.abc.Callable[[], float] = <function DelayProviderCommunicationSimulation.<lambda>>, delay_s_directed_edge_dict: dict[tuple[str | None, str], ~collections.abc.Callable[[], float]] | None = None)

Communication simulation where delays come from callable providers.

Use this when delays should be drawn from a distribution. Each provider is a zero-argument callable that returns a delay in seconds.

Example:

import random
sim = DelayProviderCommunicationSimulation(
    default_delay_s_provider=lambda: random.gauss(0.1, 0.01),
)

calculate_communication(current_time: float, messages: list[MessagePackage]) → CommunicationSimulationResult

Calculate delivery results for the given messages.

Parameters:

• current_time – current simulation time in seconds

• messages – messages to evaluate

Returns:

one PackageResult per message, in the same order

delay_s_directed_edge_dict: dict[tuple[str | None, str], Callable[[], float]]

class mango.Environment

Abstract environment interface.

abstractmethod emit_agent_event(event: Any, agent_id: Any) → None

Deliver event to a specific agent identified by agent_id.

abstractmethod emit_global_event(event: Any) → None

Broadcast event to all registered observers (and thus agents).

abstractmethod initialize(agents: list, clock: Clock) → None

Initialize the environment with the given agents.

abstractmethod initialized() → bool

Return whether the environment has been initialized.

abstractmethod step(clock: Clock, step_size_s: float) → None

Step the environment forward by step_size_s seconds.

class mango.ExternalClock(start_time: float = 0)

An external clock that proceeds only when set_time is called

get_next_activity() → float

set_time(t: float)

New time is set

sleep(t: float) → Future

Sleeps for t based on the external clock

property time: float

Current time of the external clock

class mango.ForwardingRule(from_addr: AgentAddress, to_addr: AgentAddress, forward_replies: bool = False)

Rule for automatic message forwarding.

When the agent receives a message from from_addr it is automatically forwarded to to_addr. If forward_replies is True, replies from to_addr are forwarded back to from_addr.

forward_replies: bool = False

from_addr: AgentAddress

to_addr: AgentAddress

class mango.JSON

A Codec that uses JSON to encode and decode messages.

add_serializer(otype, serialize, deserialize, type_id=None)

Add methods to serialize and deserialize objects typed otype.

This can be used to de-/encode objects that the codec otherwise couldn’t encode.

serialize will receive the unencoded object and needs to return an encodable serialization of it.

deserialize will receive an objects representation and should return an instance of the original object.

decode(data)

Decode data from bytes to the original data structure.

deserialize_obj(obj_repr)

Deserialize the original object from obj_repr.

encode(data)

Encode the given data and return a bytes object.

make_type_id(otype)

Create a type id for otype using: - type name - function names in the class - signature of the class and return a 32 bit integer type id.

serialize_obj(obj)

Serialize obj to something that the codec can encode.

class mango.MessagePackage(sender_id: str | None, receiver_id: str, sent_time: float, content: Any)

Describes a message between two agents in the simulation.

content: Any

receiver_id: str

sender_id: str | None

sent_time: float

class mango.MessagePreprocessor

Abstract base for message preprocessors in the role system.

A preprocessor intercepts messages before they reach a role’s handler, allowing transformation or rate-limiting. Pass an instance to RoleContext.subscribe_message() via the preprocessor keyword.

Subclasses must implement handle(). Override process() to transform the message content/meta before delivery.

Example:

class LoggingPreprocessor(MessagePreprocessor):
    def handle(self, role, handler, content, meta):
        print(f"[{role}] received: {content}")
        handler(content, meta)

class MyRole(Role):
    def setup(self):
        self.context.subscribe_message(
            self, self.on_msg, lambda c, m: True,
            preprocessor=LoggingPreprocessor(),
        )

    def on_msg(self, content, meta):
        ...

abstractmethod handle(role_or_agent: Any, handler: Callable, content: Any, meta: dict) → None

Intercept a message. Must call handler(content, meta) to deliver.

Parameters:

• role_or_agent – the subscribing role or agent

• handler – the original message handler

• content – message content

• meta – message metadata

init(role_or_agent: Any) → None

Called once when the preprocessor is registered.

Parameters:

role_or_agent – the role (or agent) that owns the subscription

process(content: Any, meta: dict) → tuple[Any, dict]

Transform message before delivery. Default: identity.

Parameters:

• content – message content

• meta – message metadata

Returns:

transformed (content, meta) tuple

class mango.MessageTransaction(sender_id: str | None, receiver_id: str, sent_time: float, arriving_time: float, content: Any)

Records a message that was delivered during the simulation.

arriving_time: float

content: Any

receiver_id: str

sender_id: str | None

sent_time: float

class mango.NoSpace

A space without spatial positioning.

Use this when agents do not need positions. Every agent reports no position, and attempts to move() or location() raise a RuntimeError. This is the default space for DefaultEnvironment.

Example:

env = DefaultEnvironment()          # uses NoSpace by default
env = DefaultEnvironment(space=NoSpace())

has_position(agent) → bool

Return True if agent has a registered position.

initialize(agents: list, clock: Clock) → None

Called once before the first simulation step.

install(agent, **kwargs) → None

Install agent in the space (called when agent is added to env).

location(agent) → Position

Return the current position of agent.

move(agent, position: Position) → None

Move agent to position.

class mango.PROTOBUF

add_serializer(otype, serialize, deserialize, type_id=None)

Add methods to serialize and deserialize objects typed otype.

This can be used to de-/encode objects that the codec otherwise couldn’t encode.

serialize will receive the unencoded object and needs to return an encodable serialization of it.

deserialize will receive an objects representation and should return an instance of the original object.

decode(data)

Decode data from bytes to the original data structure.

deserialize_obj(obj_repr)

Deserialize the original object from obj_repr.

encode(data)

Encode the given data and return a bytes object.

make_type_id(otype)

Create a type id for otype using: - type name - function names in the class - signature of the class and return a 32 bit integer type id.

register_proto_type(proto_class)

serialize_obj(obj)

Serialize obj to something that the codec can encode.

class mango.PackageResult(reached: bool, delay_s: float)

Result for a single message package.

delay_s: float

reached: bool

class mango.Performatives(*values)

member values (must be unique) could be used as priority values if not replaced by enum.auto. See http://www.fipa.org/specs/fipa00037/SC00037J.html for a description of performatives.

accept_proposal = 1

agree = 2

call_for_proposal = 5

cancel = 3

cfp = 4

confirm = 6

disconfirm = 7

failure = 8

inform = 9

inform_if = 20

not_understood = 10

propagate = 22

propose = 11

proxy = 21

query_if = 12

query_ref = 13

refuse = 14

reject_proposal = 15

request = 16

request_when = 17

request_whenever = 18

subscribe = 19

class mango.Position

Marker base class for position types.

class mango.Position2D(x: float, y: float)

A 2-D Cartesian position.

x: float

y: float

class mango.Role

General role class, defining the API every role can use. A role implements one responsibility of an agent.

Every role must be added to a RoleAgent and is defined by some lifecycle methods:

• Role.setup() is called when the Role is added to the agent, so its the perfect place for

  initialization and scheduling of tasks

• Role.on_stop() is called when the container the agent lives in, is shut down

To interact with the environment you have to use the context, accessible via :func:Role.context.

property context: RoleContext

Return the context of the role. This context can be send as bridge to the agent.

Returns:

the context of the role

handle_message(content: Any, meta: dict)

on_agent_event(event: Any) → None

Called when a targeted agent event is emitted.

Parameters:

event – the event object

on_change_model(model) → None

Will be invoked when a subscribed model changes via RoleContext.update().

Parameters:

model – the model

on_deactivation(src) → None

Hook in, which will be called when another role deactivates this instance (temporarily)

on_global_event(event: Any) → None

Called when a global event is emitted from the environment.

Parameters:

event – the event object

on_ready()

Called after the start of all container using activate

on_start() → None

Called when container started in which the agent is contained

on_step(env, clock, step_size_s: float) → None

Called on every simulation step (only in SimulationWorld).

Parameters:

• env – the simulation environment

• clock – the current simulation clock

• step_size_s – seconds advanced in this step

async on_stop() → None

Lifecycle hook in, which will be called when the container is shut down or if the role got removed.

setup() → None

Lifecycle hook in, which will be called on adding the role to agent. The role context is known from hereon.

class mango.RoleAgent

Agent, which support the role API-system. When you want to use the role-api you always need a RoleAgent as base for your agents. A role can be added with RoleAgent.add_role().

add_forwarding_rule(from_addr: AgentAddress, to_addr: AgentAddress, forward_replies: bool = False) → None

Add an automatic message-forwarding rule.

After calling this, every message received from from_addr is automatically forwarded to to_addr. When forward_replies is True, replies originating from to_addr are forwarded back to from_addr.

Parameters:

• from_addr – source address to match

• to_addr – destination to forward to

• forward_replies – whether replies should be forwarded back

add_role(role: Role)

Add a role to the agent. This will lead to the call of Role.setup().

Parameters:

role – the role to add

property addr

Return the address of the agent as AgentAddress

Returns:

_type_: AgentAddress

property aid

property category: str

Category tag for this agent.

property color: str

Visual color tag for this agent.

context: AgentContext

property current_timestamp: float

Method that returns the current unix timestamp given the clock within the container

delete_forwarding_rule(from_addr: AgentAddress, to_addr: AgentAddress | None = None) → None

Remove previously added forwarding rule(s).

Parameters:

• from_addr – source address of the rule to remove

• to_addr – if given, only remove rules that also match this destination; otherwise remove all rules matching from_addr

property description: AgentDescription

Return the agent’s AgentDescription.

handle_message(content, meta: dict[str, Any])

Has to be implemented by the user. This method is called when a message is received at the agents inbox. :param content: The deserialized message object :param meta: Meta details of the message. In case of mqtt this dict includes at least the field ‘topic’

property name: str

Human-readable name of this agent.

neighbors(state: State = State.NORMAL, *, tid: str = 'default', has_characteristic: str | None = None, include_connectors: tuple | list = (), match_func=None) → list[AgentAddress]

Return neighbor addresses from the topology.

Parameters:

• state – filter by edge state (default NORMAL)

• tid – topology identifier (default "default")

• has_characteristic – only include neighbors with this characteristic

• include_connectors – also include connector agents of these types

• match_func – optional (AgentDescription) -> bool predicate

Returns:

list of AgentAddress

property observable_tasks

on_agent_event(event: Any) → None

Called when a targeted agent event is emitted.

Override to react to events directed at this specific agent.

Parameters:

event – the event object

on_global_event(event: Any) → None

Called when a global event is emitted from the environment.

Override to react to environment-wide broadcasts.

Parameters:

event – the event object

on_ready()

Called when all container has been started using activate(…).

on_register()

Hook-in to define behavior of the agent directly after it got registered by a container

on_start()

Called when container started in which the agent is contained

on_step(env, clock, step_size_s: float) → None

Called on every simulation step (only in SimulationWorld).

Parameters:

• env – the simulation environment

• clock – the current simulation clock

• step_size_s – seconds advanced in this step

async on_stop()

Can be used as lifecycle callback when the agent is stopped

remove_role(role: Role)

Remove a role permanently from the agent.

Parameters:

role (Role) – [description]

async reply_to(content: Any, received_meta: dict, **kwargs) → bool

Convenience helper to reply to a received message.

Extracts the sender address from received_meta and sends content back, preserving any tracking_id for transaction matching.

Parameters:

• content – reply content

• received_meta – the meta dict from the received message

Returns:

result of send_message()

property roles: list[Role]

Returns list of roles

Returns:

list of roles

schedule_conditional_process_task(coroutine_creator, condition_func, lookup_delay=0.1, on_stop=None, src=None)

Schedule a process task when a specified condition is met.

Parameters:

• coroutine_creator (coroutine_creator) – coroutine_creator creating coroutine to be scheduled

• condition_func (lambda () -> bool) – function for determining whether the confition is fullfiled

• lookup_delay (float) – delay between checking the condition

• src (Object) – creator of the task

schedule_conditional_task(coroutine, condition_func, lookup_delay=0.1, on_stop=None, src=None)

Schedule a task when a specified condition is met.

Parameters:

• coroutine (Coroutine) – coroutine to be scheduled

• condition_func (lambda () -> bool) – function for determining whether the confition is fullfiled

• lookup_delay (float) – delay between checking the condition

• src (Object) – creator of the task

schedule_instant_message(content, receiver_addr: AgentAddress, **kwargs)

Schedules sending a message without any delay. This is equivalent to using the schedulers ‘schedule_instant_task’ with the coroutine created by ‘container.send_message’.

Parameters:

• content – The content of the message

• receiver_addr – The address passed to the container

• kwargs – Additional parameters to provide protocol specific settings

Returns:

asyncio.Task for the scheduled coroutine

schedule_instant_process_task(coroutine_creator, on_stop=None, src=None)

Schedule an instantly executed task in another processes.

Parameters:

• coroutine_creator – coroutine_creator creating coroutine to be scheduled

• on_stop (Object) – coroutine to run on stop

• src (Object) – creator of the task

schedule_instant_task(coroutine, on_stop=None, src=None)

Schedule an instantly executed task.

Parameters:

• coroutine – coroutine to be scheduled

• on_stop (Object) – coroutine to run on stop

• src (Object) – creator of the task

schedule_periodic_process_task(coroutine_creator, delay, on_stop=None, src=None)

Schedule an open end periodically executed task in another process.

Parameters:

• coroutine_creator (Coroutine Function) – coroutine function creating coros to be scheduled

• delay (float) – delay in between the cycles

• on_stop (Object) – coroutine to run on stop

• src (Object) – creator of the task

schedule_periodic_task(coroutine_func, delay, on_stop=None, src=None)

Schedule an open end peridocally executed task.

Parameters:

• coroutine_func (Coroutine Function) – coroutine function creating coros to be scheduled

• delay (float) – delay in between the cycles

• on_stop (Object) – coroutine to run on stop

• src (Object) – creator of the task

schedule_process_task(task: ScheduledProcessTask, src=None)

Schedule a task with asyncio in another process. When the task is finished, if finite, its automatically removed afterwards. For scheduling options see the subclasses of ScheduledScheduledProcessTaskTask.

Parameters:

• task – task to be scheduled

• src – object, which represents the source of the task (for example the object in which the task got created)

schedule_recurrent_process_task(coroutine_creator, recurrency, on_stop=None, src=None)

Schedule a task using a fine-grained recurrency rule in another process.

Parameters:

• coroutine_creator (Coroutine Function) – coroutine function creating coros to be scheduled

• recurrency (dateutil.rrule.rrule) – recurrency rule to calculate next event

• on_stop (Object) – coroutine to run on stop

• src (Object) – creator of the task

schedule_recurrent_task(coroutine_func, recurrency, on_stop=None, src=None)

Schedule a task using a fine-grained recurrency rule in another process.

Parameters:

• coroutine_creator (Coroutine Function) – coroutine function creating coros to be scheduled

• recurrency (dateutil.rrule.rrule) – recurrency rule to calculate next event

• on_stop (Object) – coroutine to run on stop

• src (Object) – creator of the task

schedule_task(task: ScheduledTask, src=None)

Schedule a task with asyncio. When the task is finished, if finite, its automatically removed afterwards. For scheduling options see the subclasses of ScheduledTask.

Parameters:

• task – task to be scheduled

• src – object, which represents the source of the task (for example the object in which the task got created)

schedule_timestamp_process_task(coroutine_creator, timestamp: float, on_stop=None, src=None)

Schedule a task at specified unix timestamp dispatched to another process.

Parameters:

• coroutine_creator (coroutine_creator) – coroutine_creator creating coroutine to be scheduled

• timestamp (float) – unix timestamp defining when the task should start

• src (Object) – creator of the task

schedule_timestamp_task(coroutine, timestamp: float, on_stop=None, src=None)

Schedule a task at specified unix timestamp.

Parameters:

• coroutine (Coroutine) – coroutine to be scheduled

• timestamp (timestamp) – timestamp defining when the task should start

• src (Object) – creator of the task

scheduler: Scheduler

async send_message(content, receiver_addr: AgentAddress, **kwargs) → bool

See container.send_message(…)

async send_messages(content, receiver_addrs: list[AgentAddress], **kwargs) → list[bool]

Send the same content to multiple recipients.

Parameters:

• content – message content (sent to every recipient)

• receiver_addrs – list of target AgentAddress instances

Returns:

list of send results (one per recipient, in order)

async send_tracked_message(content: Any, receiver_addr: AgentAddress, response_handler=None, **kwargs)

Send a message and optionally register a response handler.

A tracking_id is attached to the message so that the reply can be matched. When response_handler is provided it will be called as response_handler(reply_content, reply_meta) when the matching reply arrives.

Parameters:

• content – message content

• receiver_addr – target agent address

• response_handler – optional (content, meta) -> None callback

Returns:

the asyncio.Task for the sent message

service_of_type(type: type, default: Any = None) → Any

Return the service registered for type, creating it if absent.

If no service of type is registered yet, default is registered and returned; when default is None a new type() instance is created instead.

Parameters:

• type (type) – the type of the service

• default (Any (optional)) – the value to register if none exists; None creates a type() instance

Returns:

the service

Return type:

Any

async shutdown()

Shutdown all tasks that are running and deregister from the container

property suspendable_tasks

async tasks_complete(timeout=1)

Wait for all scheduled tasks to complete using a timeout.

Parameters:

timeout – waiting timeout. Defaults to 1.

property uid: str

Unique identifier (UUID string) of this agent.

update_description(name: str | None = None, color: str | None = None, category: str | None = None) → None

Update one or more description fields.

Parameters:

• name – new human-readable name

• color – new color tag

• category – new category tag

class mango.RoleContext(role_handler: RoleHandler, aid: str, inbox)

Implementation of the RoleContext.

activate(role) → None

add_forwarding_rule(from_addr: AgentAddress, to_addr: AgentAddress, forward_replies: bool = False) → None

Add an automatic message-forwarding rule.

After calling this, every message received from from_addr is automatically forwarded to to_addr. When forward_replies is True, replies originating from to_addr are forwarded back to from_addr.

Parameters:

• from_addr – source address to match

• to_addr – destination to forward to

• forward_replies – whether replies should be forwarded back

add_role(role: Role)

Add a role to the context.

Parameters:

role – the Role

property addr

Return the address of the agent as AgentAddress

Returns:

_type_: AgentAddress

property aid

property category: str

Category tag for this agent.

property color: str

Visual color tag for this agent.

context: AgentContext

property current_timestamp: float

Method that returns the current unix timestamp given the clock within the container

property data

Return data container of the agent

Returns:

the data container

Return type:

DataContainer

deactivate(role) → None

delete_forwarding_rule(from_addr: AgentAddress, to_addr: AgentAddress | None = None) → None

Remove previously added forwarding rule(s).

Parameters:

• from_addr – source address of the rule to remove

• to_addr – if given, only remove rules that also match this destination; otherwise remove all rules matching from_addr

property description: AgentDescription

Return the agent’s AgentDescription.

emit_event(event: Any, event_source: Any = None)

Emit an custom event to other roles.

Parameters:

• event (Any) – the event

• event_source (Any, optional) – emitter of the event (mostly the emitting role), defaults to None

get_or_create_model(cls)

get_role(cls: type) → Role | None

returns the first role of a given class returns None if no role of this type exists in the current context

handle_message(content, meta: dict[str, Any])

Handle an incoming message, delegating it to all applicable subscribers

for role, message_condition, method, _ in self._message_subs:
    if self._is_role_active(role) and message_condition(content, meta):
        method(content, meta)

Parameters:

• content – content

• meta – meta

inbox_length()

property name: str

Human-readable name of this agent.

neighbors(state: State = State.NORMAL, *, tid: str = 'default', has_characteristic: str | None = None, include_connectors: tuple | list = (), match_func=None) → list[AgentAddress]

Return neighbor addresses from the topology.

Parameters:

• state – filter by edge state (default NORMAL)

• tid – topology identifier (default "default")

• has_characteristic – only include neighbors with this characteristic

• include_connectors – also include connector agents of these types

• match_func – optional (AgentDescription) -> bool predicate

Returns:

list of AgentAddress

on_agent_event(event: Any) → None

Called when a targeted agent event is emitted.

Override to react to events directed at this specific agent.

Parameters:

event – the event object

on_global_event(event: Any) → None

Called when a global event is emitted from the environment.

Override to react to environment-wide broadcasts.

Parameters:

event – the event object

on_ready()

Called when all container has been started using activate(…).

on_start()

Called when container started in which the agent is contained

on_step(env, clock, step_size_s: float) → None

Called on every simulation step (only in SimulationWorld).

Parameters:

• env – the simulation environment

• clock – the current simulation clock

• step_size_s – seconds advanced in this step

remove_role(role: Role)

Remove a role and call on_stop for clean up

Parameters:

role (Role) – the role to remove

async reply_to(content: Any, received_meta: dict, **kwargs) → bool

Convenience helper to reply to a received message.

Extracts the sender address from received_meta and sends content back, preserving any tracking_id for transaction matching.

Parameters:

• content – reply content

• received_meta – the meta dict from the received message

Returns:

result of send_message()

schedule_conditional_process_task(coroutine_creator, condition_func, lookup_delay=0.1, on_stop=None, src=None)

Schedule a process task when a specified condition is met.

Parameters:

• coroutine_creator (coroutine_creator) – coroutine_creator creating coroutine to be scheduled

• condition_func (lambda () -> bool) – function for determining whether the confition is fullfiled

• lookup_delay (float) – delay between checking the condition

• src (Object) – creator of the task

schedule_conditional_task(coroutine, condition_func, lookup_delay=0.1, on_stop=None, src=None)

Schedule a task when a specified condition is met.

Parameters:

• coroutine (Coroutine) – coroutine to be scheduled

• condition_func (lambda () -> bool) – function for determining whether the confition is fullfiled

• lookup_delay (float) – delay between checking the condition

• src (Object) – creator of the task

schedule_instant_message(content, receiver_addr: AgentAddress, **kwargs)

Schedules sending a message without any delay. This is equivalent to using the schedulers ‘schedule_instant_task’ with the coroutine created by ‘container.send_message’.

Parameters:

• content – The content of the message

• receiver_addr – The address passed to the container

• kwargs – Additional parameters to provide protocol specific settings

Returns:

asyncio.Task for the scheduled coroutine

schedule_instant_process_task(coroutine_creator, on_stop=None, src=None)

Schedule an instantly executed task in another processes.

Parameters:

• coroutine_creator – coroutine_creator creating coroutine to be scheduled

• on_stop (Object) – coroutine to run on stop

• src (Object) – creator of the task

schedule_instant_task(coroutine, on_stop=None, src=None)

Schedule an instantly executed task.

Parameters:

• coroutine – coroutine to be scheduled

• on_stop (Object) – coroutine to run on stop

• src (Object) – creator of the task

schedule_periodic_process_task(coroutine_creator, delay, on_stop=None, src=None)

Schedule an open end periodically executed task in another process.

Parameters:

• coroutine_creator (Coroutine Function) – coroutine function creating coros to be scheduled

• delay (float) – delay in between the cycles

• on_stop (Object) – coroutine to run on stop

• src (Object) – creator of the task

schedule_periodic_task(coroutine_func, delay, on_stop=None, src=None)

Schedule an open end peridocally executed task.

Parameters:

• coroutine_func (Coroutine Function) – coroutine function creating coros to be scheduled

• delay (float) – delay in between the cycles

• on_stop (Object) – coroutine to run on stop

• src (Object) – creator of the task

schedule_process_task(task: ScheduledProcessTask, src=None)

Schedule a task with asyncio in another process. When the task is finished, if finite, its automatically removed afterwards. For scheduling options see the subclasses of ScheduledScheduledProcessTaskTask.

Parameters:

• task – task to be scheduled

• src – object, which represents the source of the task (for example the object in which the task got created)

schedule_recurrent_process_task(coroutine_creator, recurrency, on_stop=None, src=None)

Schedule a task using a fine-grained recurrency rule in another process.

Parameters:

• coroutine_creator (Coroutine Function) – coroutine function creating coros to be scheduled

• recurrency (dateutil.rrule.rrule) – recurrency rule to calculate next event

• on_stop (Object) – coroutine to run on stop

• src (Object) – creator of the task

schedule_recurrent_task(coroutine_func, recurrency, on_stop=None, src=None)

Schedule a task using a fine-grained recurrency rule in another process.

Parameters:

• coroutine_creator (Coroutine Function) – coroutine function creating coros to be scheduled

• recurrency (dateutil.rrule.rrule) – recurrency rule to calculate next event

• on_stop (Object) – coroutine to run on stop

• src (Object) – creator of the task

schedule_task(task: ScheduledTask, src=None)

Schedule a task with asyncio. When the task is finished, if finite, its automatically removed afterwards. For scheduling options see the subclasses of ScheduledTask.

Parameters:

• task – task to be scheduled

• src – object, which represents the source of the task (for example the object in which the task got created)

schedule_timestamp_process_task(coroutine_creator, timestamp: float, on_stop=None, src=None)

Schedule a task at specified unix timestamp dispatched to another process.

Parameters:

• coroutine_creator (coroutine_creator) – coroutine_creator creating coroutine to be scheduled

• timestamp (float) – unix timestamp defining when the task should start

• src (Object) – creator of the task

schedule_timestamp_task(coroutine, timestamp: float, on_stop=None, src=None)

Schedule a task at specified unix timestamp.

Parameters:

• coroutine (Coroutine) – coroutine to be scheduled

• timestamp (timestamp) – timestamp defining when the task should start

• src (Object) – creator of the task

scheduler: Scheduler

async send_message(content, receiver_addr: AgentAddress, **kwargs) → bool

See container.send_message(…)

async send_messages(content, receiver_addrs: list[AgentAddress], **kwargs) → list[bool]

Send the same content to multiple recipients.

Parameters:

• content – message content (sent to every recipient)

• receiver_addrs – list of target AgentAddress instances

Returns:

list of send results (one per recipient, in order)

async send_tracked_message(content: Any, receiver_addr: AgentAddress, response_handler=None, **kwargs)

Send a message and optionally register a response handler.

A tracking_id is attached to the message so that the reply can be matched. When response_handler is provided it will be called as response_handler(reply_content, reply_meta) when the matching reply arrives.

Parameters:

• content – message content

• receiver_addr – target agent address

• response_handler – optional (content, meta) -> None callback

Returns:

the asyncio.Task for the sent message

service_of_type(type: type, default: Any = None) → Any

Return the service registered for type, creating it if absent.

If no service of type is registered yet, default is registered and returned; when default is None a new type() instance is created instead.

Parameters:

• type (type) – the type of the service

• default (Any (optional)) – the value to register if none exists; None creates a type() instance

Returns:

the service

Return type:

Any

subscribe_event(role: Role, event_type: Any, handler_method: Callable)

Subscribe to specific event types. The listener will be evaluated based on their order of subscription

Parameters:

• role (Role) – the role in which you want to handle the event

• event_type (Any) – the event type you want to handle

subscribe_message(role, method, message_condition, priority=0, preprocessor: MessagePreprocessor | None = None)

subscribe_model(role, role_model_type)

subscribe_send(role, method)

async tasks_complete(timeout=1)

Wait for all scheduled tasks to complete using a timeout.

Parameters:

timeout – waiting timeout. Defaults to 1.

property uid: str

Unique identifier (UUID string) of this agent.

update(role_model)

update_description(name: str | None = None, color: str | None = None, category: str | None = None) → None

Update one or more description fields.

Parameters:

• name – new human-readable name

• color – new color tag

• category – new category tag

exception mango.SerializationError

Raised when an object cannot be serialized.

add_note()

Exception.add_note(note) – add a note to the exception

args

with_traceback()

Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

class mango.SimpleCommunicationSimulation(loss_percent: float = 0.0, default_delay_s: float = 0.0, delay_s_directed_edge_dict: dict[tuple[str | None, str], float] | None = None)

Default communication simulation with configurable loss and delay.

Per-link delays can be set via delay_s_directed_edge_dict using (sender_id, receiver_id) tuples as keys. Message loss is drawn independently per package from loss_percent; each package is evaluated exactly once by the world, so no result cache is kept.

Example:

sim = SimpleCommunicationSimulation(
    default_delay_s=0.1,
    loss_percent=0.05,
    delay_s_directed_edge_dict={("a1", "a2"): 0.5},
)

calculate_communication(current_time: float, messages: list[MessagePackage]) → CommunicationSimulationResult

Calculate delivery results for the given messages.

Parameters:

• current_time – current simulation time in seconds

• messages – messages to evaluate

Returns:

one PackageResult per message, in the same order

delay_s_directed_edge_dict: dict[tuple[str | None, str], float]

class mango.SimulationResult(time_elapsed_s: float, step_size_s: float, messages_delivered: int)

Return value of step_simulation().

messages_delivered: int

step_size_s: float

time_elapsed_s: float

class mango.SimulationWorld(clock: ExternalClock, communication_sim: CommunicationSimulation, environment: Environment | None = None)

A local, clock-driven simulation container.

Do not instantiate directly; use create_world() instead.

The world acts as the container that agents register against. It satisfies the minimal container interface expected by mango’s tasks_complete_or_sleeping() helper (inbox, _agents).

addr: str

clock: ExternalClock

communication_sim: CommunicationSimulation

data_agent_collections: dict[str, AgentsRecording]

data_collections: dict[str, WorldRecording]

deregister(aid: str) → None

environment: Environment

inbox: Queue | None

is_aid_available(aid: str) → bool

property name: str

ready: bool

recorded_messages: list[MessageTransaction]

register(agent: Agent, suggested_aid: str | None = None) → Agent

Register agent with the world and return it.

Parameters:

• agent – agent instance to register

• suggested_aid – optional preferred agent ID

Returns:

the registered agent (same object)

running: bool

async send_message(content: Any, receiver_addr: AgentAddress, sender_id: str | None = None, **kwargs) → bool

Send a message, applying communication simulation.

Messages are queued with a delivery time determined by the communication simulation. They are delivered during the next call to step_simulation().

async shutdown() → None

Shut down all agents.

class mango.Space

Abstract space in which agents can be placed and moved.

abstractmethod has_position(agent) → bool

Return True if agent has a registered position.

initialize(agents: list, clock: Clock) → None

Called once before the first simulation step.

install(agent, **kwargs) → None

Install agent in the space (called when agent is added to env).

abstractmethod location(agent) → Position

Return the current position of agent.

abstractmethod move(agent, position: Position) → None

Move agent to position.

class mango.State(*values)

BROKEN = 2

EXT_CONNECTION = 4

INACTIVE = 1

NORMAL = 0

UNKNOWN = 3

class mango.Topology(graph: Graph, *, tid: str = 'default')

A graph-based agent neighborhood topology.

Manages a NetworkX graph where each node holds one or more agents. After populating nodes and edges, call inject() (or use create_topology() / per_node()) to push neighbor information into each agent’s TopologyService.

Parameters:

• graph – an undirected or directed NetworkX graph

• tid – topology identifier; agents in multiple topologies use this to distinguish neighbor sets (default "default")

add_edge(node_from: int, node_to: int, state: State = State.NORMAL) → None

Add an undirected edge between node_from and node_to.

Parameters:

• node_from – source node ID

• node_to – destination node ID

• state – initial edge state (default NORMAL)

add_node(*agents: Agent) → int

Add a new node containing agents and return the node ID.

Parameters:

agents – zero or more agents to place at this node

Returns:

integer node ID

property agents: list[Agent]

All agents in the topology across all nodes.

property graph: Graph

The underlying NetworkX graph.

inject() → None

Push neighborhood data into every agent’s TopologyService.

Called automatically by create_topology(), per_node(), modify_topology(), and connect_topologies().

remove_edge(node_from: int, node_to: int) → None

Remove the edge between node_from and node_to.

Parameters:

• node_from – source node ID

• node_to – destination node ID

remove_node(node_id: int) → None

Remove the node with node_id and all its incident edges.

Parameters:

node_id – ID of the node to remove

set_as_connector(*agents: Agent, connector_type: str = 'default') → None

Mark agents as connectors for inter-topology links.

Connectors are agents that bridge two topologies when connect_topologies() is called.

Parameters:

• agents – agents to mark as connectors

• connector_type – connection type label (default "default")

set_characteristic(node_id: int, agent: Agent, characteristic: str) → None

Assign a characteristic label to agent in node node_id.

Characteristics are short string labels (e.g. "leader") that allow neighbors to filter by the agent’s role in the graph.

Parameters:

• node_id – ID of the node the agent belongs to

• agent – the agent to label

• characteristic – label string

set_edge_state(node_from: int, node_to: int, state: State, *, both_directions: bool = True) → None

Set the state of the edge between node_from and node_to.

Parameters:

• node_from – source node ID

• node_to – destination node ID

• state – new edge state

• both_directions – also update the reverse edge if present (default True)

property tid: str

The topology identifier.

class mango.TopologyNeighbor(agent: Any, description: AgentDescription, characteristic: str = '')

A neighbor in the topology graph.

Addresses are resolved lazily so that topologies can be built before agents are registered in a container.

Parameters:

• agent – the neighbor agent (address resolved at access time)

• description – the agent’s AgentDescription

• characteristic – an optional label for the agent’s role in its node (e.g. "leader"); empty string means no characteristic

property address: AgentAddress

The neighbor’s current AgentAddress (resolved lazily).

class mango.TopologyService

Stores neighborhood data injected by the topology system.

An instance is attached to each agent via service_of_type().

characteristic(tid: str = 'default') → str

Return this agent’s characteristic label in topology tid.

connection_types(tid: str = 'default') → list[str]

Return the connection type labels for connectors in topology tid.

connectors(tid: str = 'default', *, include_connectors: tuple[str, ...] | list[str] = (), match_func: Any = None) → list[AgentAddress]

Return addresses of connector agents for topology tid.

neighbors(state: State = State.NORMAL, *, tid: str = 'default', has_characteristic: str | None = None, include_connectors: tuple[str, ...] | list[str] = (), match_func: Any = None) → list[AgentAddress]

Return addresses of neighbors in topology tid with edge state.

Parameters:

• state – only return neighbors reachable via edges in this state

• tid – topology identifier

• has_characteristic – if given, only neighbors with this characteristic

• include_connectors – also include connectors of these connection types

• match_func – optional predicate (AgentDescription) -> bool

node_id(tid: str = 'default') → int

Return the node ID this agent occupies in topology tid.

class mango.WaitingMessagePreprocessor

Prevents concurrent message handling for a role.

Messages are queued and dispatched one at a time. The next message is only delivered after the handler for the current one has returned (or its coroutine completed). This avoids race conditions when a role’s handler is async or triggers further messages.

Example:

class MyRole(Role):
    def setup(self):
        self.context.subscribe_message(
            self, self.on_data, lambda c, m: True,
            preprocessor=WaitingMessagePreprocessor(),
        )

    async def on_data(self, content, meta):
        await asyncio.sleep(0.1)   # safe – next msg waits
        ...

handle(role_or_agent: Any, handler: Callable, content: Any, meta: dict) → None

Intercept a message. Must call handler(content, meta) to deliver.

Parameters:

• role_or_agent – the subscribing role or agent

• handler – the original message handler

• content – message content

• meta – message metadata

init(role_or_agent: Any) → None

Called once when the preprocessor is registered.

Parameters:

role_or_agent – the role (or agent) that owns the subscription

process(content: Any, meta: dict) → tuple[Any, dict]

Transform message before delivery. Default: identity.

Parameters:

• content – message content

• meta – message metadata

Returns:

transformed (content, meta) tuple

class mango.WorldRecording(timeseries: list[Any] = <factory>, time: list[float] = <factory>)

Time-series recording of world-level data.

time: list[float]

timeseries: list[Any]

mango.activate(*containers: Container) → ContainerActivationManager

Create and return an async activation context manager. This can be used with the async with syntax to run code while the container(s) are active. The containers are started first, after your code under async with will run, and at the end the container will shut down (even when an error occurs).

Example:

# Single container
async with activate(container) as container:
    # do your stuff

# Multiple container
async with activate(container_list) as container_list:
    # do your stuff

Returns:

The context manager to be used as described

Return type:

ContainerActivationManager

mango.addr(protocol_addr: Any, aid: str) → AgentAddress

Create an Address from the topic.

Args:

protocol_addr (Any): the container part of the addr, e.g. topic for mqtt, or host/port for tcp, … aid (str): the agent id

Returns:

AgentAddress: the address

mango.agent_composed_of(*roles: Role, register_in: None | Container = None, suggested_aid: None | str = None) → ComposedAgent

Create an agent composed of the given roles. If a container is provided, the created agent is automatically registered with the container register_in.

Parameters:

• register_in (None | Container) – container in which the created agent is registered, if provided

• suggested_aid (str) – the suggested aid for registration

Returns:

the composed agent

Return type:

ComposedAgent

mango.assign_agents(condition: Callable[[Agent, AgentNode], bool], topology: Topology, agents: list[Agent]) → None

Assign agents to nodes based on a predicate.

condition receives each (agent, node) pair and should return True when that agent should be placed in that node.

Parameters:

• condition – (agent, node) -> bool

• topology – the target Topology

• agents – pool of agents to assign

Example:

assign_agents(
    lambda agent, node: isinstance(agent, SensorAgent),
    topology,
    all_agents,
)

mango.auto_assign(topology: Topology, agents: list[Agent]) → None

Distribute agents across nodes in round-robin order.

Agents are assigned to nodes in the order the graph iterates them. If there are more agents than nodes the assignment wraps around.

Parameters:

• topology – the target Topology

• agents – list of agents to distribute

Example:

topology = complete_topology(3)
auto_assign(topology, my_agents)

mango.behavior_in(world, func, *, on_message=None, on_global_event=None, on_agent_event=None, agent_types=(), role_types=(), has_roles=(), match_names=(), match_colors=(), preprocessor=None)

Attach message handlers and event subscriptions to a matched set of agents.

This is a simulation-only helper that lets you inject behavior into agents without modifying their class definitions. All matching criteria are optional; if none are given, every agent in world is matched.

The handler func is called with the matched agent (or role, when role_types is used) as the first argument:

• on_message: func(agent, content, meta)

• on_global_event: func(agent, event)

• on_agent_event: func(agent, event)

When role_types is provided the first argument is the matched role instead of the agent.

Parameters:

• world – the SimulationWorld

• func – handler callable

• on_message – message type to match (isinstance check), or None to skip message subscription

• on_global_event – event type to match for global events

• on_agent_event – event type to match for targeted agent events

• agent_types – restrict to agents that are instances of these types

• role_types – attach handler to matching roles (first arg is role)

• has_roles – restrict to agents that have at least one of these roles

• match_names – restrict to agents whose name is in this collection

• match_colors – restrict to agents whose color is in this collection

• preprocessor – optional MessagePreprocessor for message handling

Example:

from mango import behavior_in

behavior_in(
    world,
    lambda agent, content, meta: print(agent.aid, content),
    on_message=str,
    agent_types=MyAgent,
)

async mango.broadcast_to_neighbors(agent_or_role: Any, content: Any, *, state: State = State.NORMAL, tid: str = 'default', has_characteristic: str | None = None, include_connectors: tuple[str, ...] | list[str] = (), match_func: Callable[[AgentDescription], bool] | None = None, **kwargs: Any) → list[AgentAddress]

Send content to all topology neighbors of agent_or_role.

Accepts all keyword arguments of topology_neighbors() for filtering, plus any extra kwargs that are forwarded to send_message.

Parameters:

• agent_or_role – sender agent or role

• content – message content

• state – edge state filter (default NORMAL)

• tid – topology identifier (default "default")

• has_characteristic – neighbor characteristic filter

• include_connectors – also broadcast to these connector types

• match_func – optional (AgentDescription) -> bool filter

Returns:

list of addresses that received the message

Example:

async def on_ready(self):
    await broadcast_to_neighbors(self, "ping")

mango.collect_agent_data(world: SimulationWorld, key: str, collector: Callable[[SimulationWorld, Agent, AgentsRecording], None]) → None

Register an agent-level data collector.

collector is called for every agent after every simulation step with the world, the agent, and the AgentsRecording for key. A shared time entry is appended once per step.

Example:

collect_agent_data(world, "state", lambda w, a, rec: (
    rec.timeseries.setdefault(a.aid, []).append(a.some_state),
))

mango.collect_data(world: SimulationWorld, key: str, collector: Callable[[SimulationWorld, WorldRecording], None]) → None

Register a world-level data collector.

collector is called after every simulation step with the world and the WorldRecording identified by key.

Example:

collect_data(world, "total_msgs", lambda w, rec: (
    rec.timeseries.append(len(w.recorded_messages)),
    rec.time.append(w.clock.time),
))

mango.complete_topology(number_of_nodes: int, *, tid: str = 'default') → Topology

Create a fully-connected (complete) topology.

Parameters:

• number_of_nodes – number of nodes in the topology

• tid – topology identifier (default "default")

Returns:

Topology

Example:

topology = complete_topology(4)
for node in per_node(topology):
    node.add(MyAgent())

mango.connect_topologies(topology_one: Topology, topology_two: Topology, connection_type: str = 'default', *, directed: bool = False) → None

Link two topologies so their connector agents can reach each other.

Connectors are agents registered via Topology.set_as_connector() or mark_as_connector(). After connecting, connector agents on each side will have the opposing side’s connectors in their connectors() list.

Parameters:

• topology_one – first topology

• topology_two – second topology

• connection_type – the link type label (default "default")

• directed – if True only topology_one → topology_two (default False)

mango.create_acl(content, receiver_addr: AgentAddress, sender_addr: AgentAddress, acl_metadata: None | dict[str, Any] = None, is_anonymous_acl=False)

mango.create_distribution_based_com_sim(aid_graph: nx.Graph, default_delay_per_edge_ms: float = 10.0, base_delay_per_message_ms: float = 0.0, max_edge_delay_ms: float | None = None, distribution_provider: Callable[[float], Callable[[], float]] | None = None) → DelayProviderCommunicationSimulation

Create a topology-aware communication simulation.

Delays are derived from shortest-path distances in aid_graph. Each node in the graph must be an agent AID (string). The delay for a (sender, receiver) pair is:

mean_delay_ms = base_delay_per_message_ms
                + shortest_path_length * default_delay_per_edge_ms

capped at max_edge_delay_ms when provided. A distribution_provider maps a mean delay (in ms) to a zero-argument callable that samples the actual delay (in seconds). The default provider returns the mean value deterministically.

Parameters:

• aid_graph – a networkx.Graph (or DiGraph) whose nodes are agent AID strings

• default_delay_per_edge_ms – additional delay per hop (ms)

• base_delay_per_message_ms – fixed base delay for every message (ms)

• max_edge_delay_ms – optional cap on the total mean delay (ms)

• distribution_provider – (mean_ms: float) -> () -> float factory; each call returns a provider that samples a delay in seconds. Defaults to a constant provider (no randomness).

Returns:

a configured DelayProviderCommunicationSimulation

Example:

import networkx as nx
import random

g = nx.path_graph(["a0", "a1", "a2"])
# Gaussian delays, mean grows with graph distance
sim = create_distribution_based_com_sim(
    g,
    default_delay_per_edge_ms=20.0,
    distribution_provider=lambda mean_ms: (
        lambda: max(0.0, random.gauss(mean_ms, mean_ms * 0.1)) / 1000.0
    ),
)

mango.create_ec_container(codec: Codec = None, clock: Clock = None, addr: None | str | tuple[str, int] = None, **kwargs: dict[str, Any])

mango.create_mqtt_container(broker_addr: tuple | dict | str, client_id: str, codec: Codec = None, clock: Clock = None, inbox_topic: str | None = None, copy_internal_messages: bool = False, **kwargs)

This method is called to instantiate an MQTT container

Parameters:

• broker_addr – The address of the broker this container will connect to. it has to be a tuple of (host, port).

• client_id – The id of the MQTT Client

• codec – Defines the codec to use. Defaults to JSON

• clock – The clock that the scheduler of the agent should be based on. Defaults to the AsyncioClock

• inbox_topic – Default subscription to the a specific MQTT topic

• copy_internal_messages – Explicitly copy internal messages. Defaults to False

Returns:

The instance of a MQTTContainer

mango.create_tcp_container(addr: str | tuple[str, int], codec: Codec = None, clock: Clock = None, copy_internal_messages: bool = False, auto_port=False, **kwargs: dict[str, Any]) → Container

This method is called to instantiate a tcp container

Parameters:

• addr – The address to use. it has to be a tuple of (host, port).

• codec – Defines the codec to use. Defaults to JSON

• clock – The clock that the scheduler of the agent should be based on. Defaults to the AsyncioClock

• copy_internal_messages – Explicitly copy internal messages. Defaults to False

• auto_port – Whether you want to let the operating system pick the port. Defaults to False

Returns:

The instance of a TCPContainer

mango.create_topology(*, directed: bool = False, tid: str = 'default') → Iterator[Topology]

Context manager that builds a topology and injects neighborhoods on exit.

Parameters:

• directed – use a directed graph (default False)

• tid – topology identifier (default "default")

Yield:

an empty Topology to populate

Example:

agents = [MyAgent(), MyAgent(), MyAgent()]
with create_topology() as topology:
    n1 = topology.add_node(agents[0])
    n2 = topology.add_node(agents[1])
    n3 = topology.add_node(agents[2])
    topology.add_edge(n1, n2)
    topology.add_edge(n1, n3)

mango.create_world(start_time: float = 0.0, communication_sim: CommunicationSimulation | None = None, environment: Environment | None = None) → SimulationWorld

Create a SimulationWorld.

Parameters:

• start_time – initial simulation time in seconds (default 0)

• communication_sim – communication simulation to use; defaults to SimpleCommunicationSimulation with zero delay and no loss

• environment – environment to use; defaults to DefaultEnvironment

Returns:

a ready-to-use SimulationWorld

Example:

world = create_world(
    start_time=0.0,
    communication_sim=SimpleCommunicationSimulation(default_delay_s=0.1),
)

mango.custom_topology(graph: Graph) → Topology

Create a topology from an existing NetworkX graph (alias for graph_topology()).

mango.cycle_topology(number_of_nodes: int, *, tid: str = 'default') → Topology

Create a ring (cycle) topology.

Parameters:

• number_of_nodes – number of nodes in the ring

• tid – topology identifier (default "default")

Returns:

Topology

Example:

topology = cycle_topology(6)
for node in per_node(topology):
    node.add(MyAgent())

async mango.discrete_step_until(world: SimulationWorld, max_advance_time_s: float) → list[SimulationResult]

Run a discrete-event simulation until max_advance_time_s has elapsed.

The simulation stops when the world clock has advanced by max_advance_time_s from its current position, or when there are no more events to process.

Parameters:

• world – the simulation world

• max_advance_time_s – maximum total time to simulate in seconds

Returns:

list of SimulationResult from each step

Example:

results = await discrete_step_until(world, max_advance_time_s=3600.0)

mango.distance(pa: Position2D, pb: Position2D) → float

Return the Euclidean distance between two Position2D points.

Parameters:

• pa – first position

• pb – second position

Returns:

Euclidean distance

Example:

d = distance(Position2D(0, 0), Position2D(3, 4))  # → 5.0

mango.graph_topology(graph: Graph, *, tid: str = 'default') → Topology

Create a topology from an existing NetworkX graph.

Parameters:

• graph – the graph to use as the topology structure

• tid – topology identifier (default "default")

Returns:

Topology

mango.json_serializable(cls=None, repr=True)

This is a direct copy from aiomas: https://gitlab.com/sscherfke/aiomas/-/blob/master/src/aiomas/codecs.py

Class decorator that makes the decorated class serializable by the json codec (or any codec that can handle python dictionaries).

The decorator tries to extract all arguments to the class’ __init__(). That means, the arguments must be available as attributes with the same name.

The decorator adds the following methods to the decorated class:

• __asdict__(): Returns a dict with all __init__ parameters

• __fromdict__(dict): Creates a new class instance from dict

• __serializer__(): Returns a tuple with args for Codec.add_serializer()

• __repr__(): Returns a generic instance representation. Adding this method can be deactivated by passing repr=False to the decorator.

mango.mark_as_connector(agent: Agent, connector_type: str = 'default') → None

Mark agent as a connector for later connect_topologies() calls.

Unlike Topology.set_as_connector(), this can be called before the topology is built. The mark is picked up during Topology.inject().

Parameters:

• agent – the agent to mark

• connector_type – connection type label (default "default")

mango.modify_topology(topology: Topology) → Iterator[Topology]

Context manager for mutating an existing topology.

Re-injects neighborhoods after the with block exits.

Parameters:

topology – the Topology to modify

Yield:

the same topology

Example:

with modify_topology(my_topology) as t:
    t.remove_edge(0, 1)
    t.set_edge_state(0, 2, State.BROKEN)

mango.per_node(topology: Topology) → Iterator[AgentNode]

Iterate over topology nodes, yielding each AgentNode.

Neighborhoods are injected after the last iteration.

Parameters:

topology – the Topology to iterate

Yield:

each AgentNode in graph order

Example:

topology = complete_topology(3)
for node in per_node(topology):
    node.add(MyAgent())

mango.plot_agents(world: SimulationWorld, recording_key: str, *, title: str | None = None, xlabel: str = 'Time (s)', ylabel: str = 'Value', color: str | list[str] | None = None, write_to: str | None = None) → Any

Plot a per-agent recording with one line per agent.

Parameters:

• world – the simulation world

• recording_key – key of the AgentsRecording to plot

• title – plot title

• xlabel – x-axis label

• ylabel – y-axis label

• color – single colour or list of colours (one per agent)

• write_to – if given, save the figure to this file path

Returns:

the matplotlib Figure

Example:

record_agent(world, "soc", lambda a: a.soc)
async with world:
    await discrete_step_until(world, 3600.0)
plot_agents(world, "soc", ylabel="State of charge (kWh)")

mango.plot_recordings(world: SimulationWorld, *, figsize: tuple[float, float] | None = None, colormap: str | None = None, write_to: str | None = None) → Any

Plot all recordings in a grid layout.

World-level recordings appear as single-line charts. Per-agent recordings show one line per agent.

Parameters:

• world – the simulation world

• figsize – optional (width, height) override

• colormap – optional matplotlib colormap name for agent colours

• write_to – if given, save the figure to this file path

Returns:

the matplotlib Figure

Example:

record_world(world, "msgs", lambda: len(world.recorded_messages))
record_agent(world, "state", lambda a: a.state)
async with world:
    await discrete_step_until(world, 60.0)
plot_recordings(world, write_to="results.png")

mango.plot_world(world: SimulationWorld, recording_key: str, *, title: str | None = None, xlabel: str = 'Time (s)', ylabel: str = 'Value', color: str | None = None, write_to: str | None = None) → Any

Plot a world-level recording as a line chart.

Parameters:

• world – the simulation world

• recording_key – key of the WorldRecording to plot

• title – plot title (defaults to the recording key)

• xlabel – x-axis label

• ylabel – y-axis label

• color – line colour (any matplotlib colour string)

• write_to – if given, save the figure to this file path instead of returning it

Returns:

the matplotlib Figure

Example:

record_world(world, "msg_count", lambda: len(world.recorded_messages))
async with world:
    await discrete_step_until(world, 60.0)
fig = plot_world(world, "msg_count", ylabel="# messages")
fig.show()

mango.position_history(world: SimulationWorld, key: str = 'positions') → AgentsRecording

Return the AgentsRecording populated by record_position().

Parameters:

• world – the simulation world

• key – recording key (default "positions")

Returns:

the recording

mango.record_agent(world: SimulationWorld, key: str, recorder: Callable[[Agent], Any], filter_fn: Callable[[Agent], bool] | None = None) → None

Record a per-agent scalar after every step.

recorder receives each agent and returns the value to store. An optional filter_fn restricts recording to a subset of agents — pass an isinstance-based predicate to record only agents of a particular type:

record_agent(world, "soc", lambda a: a.soc_kwh,
             filter_fn=lambda a: isinstance(a, EVAgent))

Parameters:

• world – the simulation world

• key – recording key

• recorder – (agent) -> value callable

• filter_fn – optional (agent) -> bool predicate; None records all registered agents

mango.record_agent_having(world: SimulationWorld, key: str, role_type: type, recorder: Callable[[Agent], Any]) → None

Record a per-agent scalar for agents that carry a specific role type.

Only agents that have at least one role that is an instance of role_type are included in the recording. recorder receives the agent and returns the value to store.

Parameters:

• world – the simulation world

• key – recording key

• role_type – only record agents that have a role of this type

• recorder – (agent) -> value callable

Example:

record_agent_having(world, "energy", EnergyRole, lambda a: a.roles[0].energy)

mango.record_position(world: SimulationWorld, key: str = 'positions', filter_fn: Callable[[Agent], bool] | None = None) → None

Record the spatial position of every agent after each step.

Only agents that have a position in the world’s space are recorded. An optional filter_fn restricts recording to a subset of agents.

Parameters:

• world – the simulation world

• key – recording key (default "positions")

• filter_fn – (agent) -> bool predicate; None means all agents

Example:

record_position(world)
history = position_history(world)
# history.timeseries["agent0"]  -> list of Position2D

mango.record_world(world: SimulationWorld, key: str, recorder: Callable[[], Any]) → None

Record a world-level scalar after every step.

recorder is a zero-argument callable whose return value is appended to the recording’s timeseries.

Example:

record_world(world, "agent_count", lambda: len(world._agents))

mango.run_with_mqtt(num: int, *agents: tuple[Agent, dict], broker_addr: tuple[str, int] = ('127.0.0.1', 1883), codec: None | Codec = None) → RunWithMQTTManager

Create and return an async context manager, which can be used to run the given agents in num automatically created mqtt container. The agents are distributed according to the topic

The function takes a list of agents which shall run, it is possible to provide a tuple (Agent, dict), the dict supports “aid” for the suggested_aid and “topics” as list of topics the agent wants to subscribe to.

Parameters:

• num (int) – _description_

• broker_addr (tuple[str, int], optional) – Address of the broker the container shall connect to, defaults to (“127.0.0.1”, 1883)

• codec (None | Codec, optional, The codec of the container) – _description_, defaults to None

Returns:

the async context manager

Return type:

RunWithMQTTManager

mango.run_with_simulation(*agents: Agent | tuple[Agent, dict], start_time: float = 0.0, communication_sim=None, environment=None) → RunWithSimulationManager

Create and return an async context manager backed by a SimulationWorld.

Agents are registered in a single simulation world. Pass a custom communication_sim or environment to override the defaults.

from mango import run_with_simulation, step_simulation

async with run_with_simulation(MyAgent(), MyAgent()) as world:
    await step_simulation(world, step_size_s=1.0)

Parameters:

• agents – agent instances or (agent, {"aid": "preferred_id"}) tuples

• start_time – initial simulation clock time in seconds (default 0.0)

• communication_sim – custom communication simulation; defaults to SimpleCommunicationSimulation with zero delay and no loss

• environment – custom environment; defaults to DefaultEnvironment

Returns:

async context manager that yields the SimulationWorld

Return type:

RunWithSimulationManager

mango.run_with_tcp(num: int, *agents: Agent | tuple[Agent, dict], addr: tuple[str, int] = ('127.0.0.1', 5555), codec: None | Codec = None, auto_port: bool = False) → RunWithTCPManager

Create and return an async context manager, which can be used to run the given agents in num automatically created tcp container. The agents are distributed evenly.

async with run_with_tcp(2, Agent(), Agent(), (Agent(), dict(aid="my_agent_id"))) as c:
    # do your stuff

Parameters:

• num (int) – number of tcp container

• addr (tuple[str, int], optional) – the starting addr of the containers, defaults to (“127.0.0.1”, 5555)

• codec (None | Codec, optional) – the codec for the containers, defaults to None

• auto_port (bool) – set if the port should be chosen automatically

Returns:

the async context manager to run the agents with

Return type:

RunWithTCPManager

mango.sender_addr(meta: dict) → AgentAddress

Extract the sender_addr from the meta dict.

Args:

meta (dict): the meta you received

Returns:

AgentAddress: Extracted agent address to be used for replying to messages

mango.show_communication_data(world: SimulationWorld, *, aid_to_name: dict[str, str] | None = None, aid_to_color: dict[str, str] | None = None, write_to: str | None = None) → Any

Plot a message-flow timeline for all recorded messages.

Each agent appears as a labelled horizontal lane. Every MessageTransaction is drawn as an arrow from the sender lane at sent_time to the receiver lane at arriving_time, with a short content label at the midpoint.

Parameters:

• world – the simulation world (reads world.recorded_messages)

• aid_to_name – optional mapping of AID → display label

• aid_to_color – optional mapping of AID → matplotlib colour string

• write_to – if given, save the figure to this file path

Returns:

the matplotlib Figure

Example:

async with world:
    await discrete_step_until(world, 10.0)
show_communication_data(
    world,
    aid_to_name={"agent0": "Producer", "agent1": "Consumer"},
    write_to="comms.png",
)

mango.star_topology(number_of_nodes: int, *, tid: str = 'default') → Topology

Create a star topology (one hub connected to all leaves).

Parameters:

• number_of_nodes – total number of nodes including the hub

• tid – topology identifier (default "default")

Returns:

Topology

Example:

topology = star_topology(5)  # node 0 is the hub
for node in per_node(topology):
    node.add(MyAgent())

async mango.step_simulation(world: SimulationWorld, step_size_s: float = -1.0, max_advance_time_s: float = -1.0) → SimulationResult | None

Advance the simulation by step_size_s seconds.

When step_size_s is DISCRETE_EVENT (the default), the step size is determined automatically as the time until the next scheduled event (message arrival or agent task wakeup).

Parameters:

• world – the simulation world to step

• step_size_s – step size in seconds, or DISCRETE_EVENT

• max_advance_time_s – abort if the determined discrete step would exceed this value; -1 means no limit

Returns:

SimulationResult, or None if there is nothing to do

Example:

result = await step_simulation(world, step_size_s=1.0)
result = await step_simulation(world)  # discrete-event

mango.topology_characteristic(agent_or_role: Any, *, tid: str = 'default') → str

Return the characteristic label of agent_or_role in topology tid.

Parameters:

• agent_or_role – the agent or role to query

• tid – topology identifier (default "default")

Returns:

characteristic string, or "" if none was set

mango.topology_connection_types(agent_or_role: Any, *, tid: str = 'default') → list[str]

Return connection type labels for connectors in topology tid.

Parameters:

• agent_or_role – the agent or role to query

• tid – topology identifier (default "default")

Returns:

list of connection type strings

mango.topology_connectors(agent_or_role: Any, *, tid: str = 'default', include_connectors: tuple[str, ...] | list[str] = (), match_func: Callable[[AgentDescription], bool] | None = None) → list[AgentAddress]

Return connector addresses visible to agent_or_role in topology tid.

Parameters:

• agent_or_role – the agent or role to query

• tid – topology identifier (default "default")

• include_connectors – filter to these connection type labels

• match_func – optional (AgentDescription) -> bool predicate

Returns:

list of AgentAddress

mango.topology_neighbors(agent_or_role: Any, *, state: State = State.NORMAL, tid: str = 'default', has_characteristic: str | None = None, include_connectors: tuple[str, ...] | list[str] = (), match_func: Callable[[AgentDescription], bool] | None = None) → list[AgentAddress]

Return the neighbor addresses of agent_or_role in topology tid.

Accepts both Agent instances and Role instances.

Parameters:

• agent_or_role – the agent or role whose neighbors to retrieve

• state – filter by edge state (default NORMAL)

• tid – topology identifier (default "default")

• has_characteristic – only include neighbors with this characteristic

• include_connectors – also include connector agents of these types

• match_func – optional (AgentDescription) -> bool predicate

Returns:

list of AgentAddress

Example:

addrs = topology_neighbors(agent)
addrs = topology_neighbors(role, state=State.INACTIVE, tid="overlay")

mango.topology_node_id(agent_or_role: Any, *, tid: str = 'default') → int

Return the node ID agent_or_role occupies in topology tid.

Parameters:

• agent_or_role – the agent or role to query

• tid – topology identifier (default "default")

Returns:

integer node ID

Raises:

KeyError – if the agent has not been injected into tid

mango.topology_to_aid_graph(topology: Topology) → Graph

Convert a Topology to an agent-level NetworkX graph.

Each agent becomes an individual node (labelled by AID). Nodes in the same topology node are connected with NORMAL edges; nodes in adjacent topology nodes inherit the topology edge state.

Parameters:

topology – the source Topology

Returns:

a networkx.Graph with AID-labelled nodes and State-labelled edges

Example:

aid_graph = topology_to_aid_graph(topology)
# Use with create_distribution_based_com_sim for topology-aware delays

class mango.PrintingAgent

handle_message(content, meta: dict[str, Any])

Has to be implemented by the user. This method is called when a message is received at the agents inbox. :param content: The deserialized message object :param meta: Meta details of the message. In case of mqtt this dict includes at least the field ‘topic’

class mango.DistributedClockManager(receiver_clock_addresses: list)

async broadcast(message, add_futures=True)

Broadcast the given message to all receiver clock addresses. If add_futures is set, a future is added which is finished when an answer by the receiving clock agent was received.

Args:

message (object): the given message add_futures (bool, optional): Adds futures which can be awaited until a response to a message is given. Defaults to True.

async distribute_time(time=None)

Waits until the current container is done. Brodcasts the new time to all the other clock agents. Thn awaits until the work in the other agents is done and their next event is received.

Args:

time (number, optional): The new time which is set. Defaults to None.

Returns:

number or None: The time at which the next event happens

async get_next_event()

Get the next event from the scheduler by requesting all known clock agents

handle_message(content: float, meta)

Has to be implemented by the user. This method is called when a message is received at the agents inbox. :param content: The deserialized message object :param meta: Meta details of the message. In case of mqtt this dict includes at least the field ‘topic’

on_ready()

Called when all container has been started using activate(…).

async send_current_time(time=None)

Broadcasts the current time to all receiver clock addresses. Does not add futures to wait for responses, as no response is expected here.

Args:

time (number, optional): The current time which is set. Defaults to None.

async shutdown()

Shutdown all tasks that are running and deregister from the container

async wait_all_online()

sends a broadcast to ask for the next event to all expected addresses. Waits one second and repeats this behavior until a response by all addresses is receivd. This effectively waits until all agents are up and running and the manager can start the simulation.

This is needed, as there is no way in paho mqtt to check whether a message was retrieved, except for by sending ping pong messages.

async wait_for_futures()

Waits for all futures in self.futures

Gives debug log output to see which agent is waited for.

class mango.DistributedClockAgent

handle_message(content: float, meta)

Has to be implemented by the user. This method is called when a message is received at the agents inbox. :param content: The deserialized message object :param meta: Meta details of the message. In case of mqtt this dict includes at least the field ‘topic’

Note

Note that, most classes and functions described in the API reference should be imported using from mango import …, as the stable and public API generally will be available by using mango and the internal module structure might change, even in minor releases.

By subpackages

• Express API
  • API
• Agents API
  • Agent core
  • Roles
• Container API
  • Container core
  • Container external coupling
  • Container creation
  • MQTT Container
  • TCP Container
• Codecs
  • Codec implementations
• Utilities
  • Clock
  • Distributed clock
  • Scheduling
  • Termination detection
• Simulation
  • World
  • Communication
  • Environment
  • Visualization