Simulation

World

SimulationWorld – a self-contained simulation container for mango.

Mirrors the World type from Mango.jl. Agents registered in a SimulationWorld share an ExternalClock and can be stepped forward in discrete or fixed-size time increments.

Typical usage:

async def run():
    world = create_world(start_time=0.0)
    agent = world.register(MyAgent())

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

asyncio.run(run())

Or for a fully automated discrete-event run:

async def run():
    world = create_world(start_time=0.0)
    agent = world.register(MyAgent())
    async with world:
        await discrete_step_until(world, max_advance_time_s=60.0)

asyncio.run(run())

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

Bases: object

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.simulation.world.MessageTransaction(sender_id: str | None, receiver_id: str, sent_time: float, arriving_time: float, content: Any)

Bases: object

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.simulation.world.SimulationResult(time_elapsed_s: float, step_size_s: float, messages_delivered: int)

Bases: object

Return value of step_simulation().

messages_delivered: int

step_size_s: float

time_elapsed_s: float

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

Bases: object

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.simulation.world.WorldRecording(timeseries: list[Any] = <factory>, time: list[float] = <factory>)

Bases: object

Time-series recording of world-level data.

time: list[float]

timeseries: list[Any]

mango.simulation.world.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.simulation.world.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.simulation.world.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),
)

async mango.simulation.world.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.simulation.world.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.simulation.world.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.simulation.world.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.simulation.world.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.simulation.world.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))

async mango.simulation.world.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

Communication

Communication simulation for mango’s SimulationWorld.

Provides deterministic simulation of message delays and packet loss between agents.

class mango.simulation.communication.CommunicationSimulation

Bases: ABC

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.simulation.communication.CommunicationSimulationResult(package_results: list[PackageResult])

Bases: object

Aggregated result for a set of message packages.

package_results: list[PackageResult]

class mango.simulation.communication.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)

Bases: CommunicationSimulation

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.simulation.communication.MessagePackage(sender_id: str | None, receiver_id: str, sent_time: float, content: Any)

Bases: object

Describes a message between two agents in the simulation.

content: Any

receiver_id: str

sender_id: str | None

sent_time: float

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

Bases: object

Result for a single message package.

delay_s: float

reached: bool

class mango.simulation.communication.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)

Bases: CommunicationSimulation

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]

mango.simulation.communication.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
    ),
)

Environment

Environment system for mango’s SimulationWorld.

Provides a spatial environment with pluggable space and behavior models.

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

Bases: Space

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.simulation.environment.Behavior

Bases: ABC

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.simulation.environment.DefaultEnvironment(space: Space | None = None, behavior: Behavior | None = None)

Bases: Environment

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.simulation.environment.Environment

Bases: ABC

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.simulation.environment.NoSpace

Bases: Space

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.

location(agent) → Position

Return the current position of agent.

move(agent, position: Position) → None

Move agent to position.

class mango.simulation.environment.Position

Bases: object

Marker base class for position types.

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

Bases: Position

A 2-D Cartesian position.

x: float

y: float

class mango.simulation.environment.Space

Bases: ABC

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.simulation.environment.WorldObserver

Bases: ABC

Observer that can receive global events from the environment.

abstractmethod dispatch_global_event(clock: Clock, event: Any) → None

Dispatch a global event to observers.

mango.simulation.environment.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

Visualization

Visualization utilities for SimulationWorld recordings.

Requires matplotlib (pip install matplotlib).

Functions

• plot_world() – line chart of a world-level recording

• plot_agents() – multi-line chart of a per-agent recording

• plot_recordings() – grid of all recordings in a world

• show_communication_data() – message-flow timeline diagram

mango.simulation.visualization.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.simulation.visualization.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.simulation.visualization.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.simulation.visualization.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",
)