Experimental

This namespace contains experimental features. These are under development, and their API is not necessarily stable.

Devs

Core event management functionality for Mesa’s discrete event simulation system.

This module provides the foundational data structures and classes needed for event-based simulation in Mesa. The EventList class is a priority queue implementation that maintains simulation events in chronological order while respecting event priorities. Key features:

• Priority-based event ordering

• Weak references to prevent memory leaks from canceled events

• Efficient event insertion and removal using a heap queue

• Support for event cancellation without breaking the heap structure

The module contains three main components: - Priority: An enumeration defining event priority levels (HIGH, DEFAULT, LOW) - SimulationEvent: A class representing individual events with timing and execution details - EventList: A heap-based priority queue managing the chronological ordering of events

The implementation supports both pure discrete event simulation and hybrid approaches combining agent-based modeling with event scheduling.

class Priority(*values)

Enumeration of priority levels.

class SimulationEvent(time: int | float, function: Callable, priority: Priority = Priority.DEFAULT, function_args: list[Any] | None = None, function_kwargs: dict[str, Any] | None = None)

A simulation event.

The callable is wrapped using weakref, so there is no need to explicitly cancel event if e.g., an agent is removed from the simulation.

time

The simulation time of the event

Type:

float

fn

The function to execute for this event

Type:

Callable

priority

The priority of the event

Type:

Priority

unique_id

Type:

int

function_args

Argument for the function

Type:

list[Any]

function_kwargs

Keyword arguments for the function

Type:

Dict[str, Any]

Notes

simulation events use a weak reference to the callable. Therefore, you cannot pass a lambda function in fn. A simulation event where the callable no longer exists (e.g., because the agent has been removed from the model) will fail silently.

Initialize a simulation event.

Parameters:

• time – the instant of time of the simulation event

• function – the callable to invoke

• priority – the priority of the event

• function_args – arguments for callable

• function_kwargs – keyword arguments for the callable

execute()

Execute this event.

cancel() → None

Cancel this event.

class EventList

An event list.

This is a heap queue sorted list of events. Events are always removed from the left, so heapq is a performant and appropriate data structure. Events are sorted based on their time stamp, their priority, and their unique_id as a tie-breaker, guaranteeing a complete ordering.

Initialize an event list.

add_event(event: SimulationEvent)

Add the event to the event list.

Parameters:

event (SimulationEvent) – The event to be added

peak_ahead(n: int = 1) → list[SimulationEvent]

Look at the first n non-canceled event in the event list.

Parameters:

n (int) – The number of events to look ahead

Returns:

list[SimulationEvent]

Raises:

IndexError – If the eventlist is empty

Notes

this method can return a list shorted then n if the number of non-canceled events on the event list is less than n.

pop_event() → SimulationEvent

Pop the first element from the event list.

is_empty() → bool

Return whether the event list is empty.

remove(event: SimulationEvent) → None

Remove an event from the event list.

Parameters:

event (SimulationEvent) – The event to be removed

clear()

Clear the event list.

Simulator implementations for different time advancement approaches in Mesa.

This module provides simulator classes that control how simulation time advances and how events are executed. It supports both discrete-time and continuous-time simulations through three main classes:

• Simulator: Base class defining the core simulation control interface

• ABMSimulator: A simulator for agent-based models that combines fixed time steps with event scheduling. Uses integer time units and automatically schedules model.step()

• DEVSimulator: A pure discrete event simulator using floating-point time units for continuous time simulation

Key features: - Flexible time units (integer or float) - Event scheduling using absolute or relative times - Priority-based event execution - Support for running simulations for specific durations or until specific end times

The simulators enable Mesa models to use traditional time-step based approaches, pure event-driven approaches, or hybrid combinations of both.

class Simulator(time_unit: type, start_time: int | float)

The Simulator controls the time advancement of the model.

The simulator uses next event time progression to advance the simulation time, and execute the next event

event_list

The list of events to execute

Type:

EventList

time

The current simulation time

Type:

float | int

time_unit

The unit of the simulation time

Type:

type

model

The model to simulate

Type:

Model

Initialize a Simulator instance.

Parameters:

• time_unit – type of the smulaiton time

• start_time – the starttime of the simulator

setup(model: Model) → None

Set up the simulator with the model to simulate.

Parameters:

model (Model) – The model to simulate

Raises:

• Exception if simulator.time is not equal to simulator.starttime –

• Exception if event list is not empty –

reset()

Reset the simulator by clearing the event list and removing the model to simulate.

run_until(end_time: int | float) → None

Run the simulator until the end time.

Parameters:

end_time (int | float) – The end time for stopping the simulator

Raises:

Exception if simulator.setup() has not yet been called –

run_next_event()

Execute the next event.

Raises:

Exception if simulator.setup() has not yet been called –

run_for(time_delta: int | float)

Run the simulator for the specified time delta.

Parameters:

time_delta (float| int) – The time delta. The simulator is run from the current time to the current time plus the time delta

schedule_event_now(function: Callable, priority: Priority = Priority.DEFAULT, function_args: list[Any] | None = None, function_kwargs: dict[str, Any] | None = None) → SimulationEvent

Schedule event for the current time instant.

Parameters:

• function (Callable) – The callable to execute for this event

• priority (Priority) – the priority of the event, optional

• function_args (List[Any]) – list of arguments for function

• function_kwargs (Dict[str, Any]) – dict of keyword arguments for function

Returns:

the simulation event that is scheduled

Return type:

SimulationEvent

schedule_event_absolute(function: Callable, time: int | float, priority: Priority = Priority.DEFAULT, function_args: list[Any] | None = None, function_kwargs: dict[str, Any] | None = None) → SimulationEvent

Schedule event for the specified time instant.

Parameters:

• function (Callable) – The callable to execute for this event

• time (int | float) – the time for which to schedule the event

• priority (Priority) – the priority of the event, optional

• function_args (List[Any]) – list of arguments for function

• function_kwargs (Dict[str, Any]) – dict of keyword arguments for function

Returns:

the simulation event that is scheduled

Return type:

SimulationEvent

schedule_event_relative(function: Callable, time_delta: int | float, priority: Priority = Priority.DEFAULT, function_args: list[Any] | None = None, function_kwargs: dict[str, Any] | None = None) → SimulationEvent

Schedule event for the current time plus the time delta.

Parameters:

• function (Callable) – The callable to execute for this event

• time_delta (int | float) – the time delta

• priority (Priority) – the priority of the event, optional

• function_args (List[Any]) – list of arguments for function

• function_kwargs (Dict[str, Any]) – dict of keyword arguments for function

Returns:

the simulation event that is scheduled

Return type:

SimulationEvent

cancel_event(event: SimulationEvent) → None

Remove the event from the event list.

Parameters:

event (SimulationEvent) – The simulation event to remove

class ABMSimulator

This simulator uses incremental time progression, while allowing for additional event scheduling.

The basic time unit of this simulator is an integer. It schedules model.step for each tick with the highest priority. This implies that by default, model.step is the first event executed at a specific tick. In addition, discrete event scheduling, using integer as the time unit is fully supported, paving the way for hybrid ABM-DEVS simulations.

Initialize a ABM simulator.

setup(model)

Set up the simulator with the model to simulate.

Parameters:

model (Model) – The model to simulate

check_time_unit(time) → bool

Check whether the time is of the correct unit.

Parameters:

time (int | float) – the time

Returns:

whether the time is of the correct unit

Return type:

bool

schedule_event_next_tick(function: Callable, priority: Priority = Priority.DEFAULT, function_args: list[Any] | None = None, function_kwargs: dict[str, Any] | None = None) → SimulationEvent

Schedule a SimulationEvent for the next tick.

Parameters:

• function (Callable) – the callable to execute

• priority (Priority) – the priority of the event

• function_args (List[Any]) – List of arguments to pass to the callable

• function_kwargs (Dict[str, Any]) – List of keyword arguments to pass to the callable

run_until(end_time: int) → None

Run the simulator up to and included the specified end time.

Parameters:

end_time (float| int) – The end_time delta. The simulator is until the specified end time

Raises:

Exception if simulator.setup() has not yet been called –

class DEVSimulator

A simulator where the unit of time is a float.

Can be used for full-blown discrete event simulating using event scheduling.

Initialize a DEVS simulator.

check_time_unit(time) → bool

Check whether the time is of the correct unit.

Parameters:

time (float) – the time

Returns: bool: whether the time is of the correct unit

Continuous Space

A Continuous Space class.

class ContinuousSpace(dimensions: Buffer | _SupportsArray[dtype[Any]] | _NestedSequence[_SupportsArray[dtype[Any]]] | bool | int | float | complex | str | bytes | _NestedSequence[bool | int | float | complex | str | bytes], torus: bool = False, random: Random | None = None, n_agents: int = 100)

Continuous space where each agent can have an arbitrary position.

Create a new continuous space.

Parameters:

• dimensions – a numpy array like object where each row specifies the minimum and maximum value of that dimension.

• torus – boolean for whether the space wraps around or not

• random – a seeded stdlib random.Random instance

• n_agents – the expected number of agents in the space

Internally, a numpy array is used to store the positions of all agents. This is resized if needed, but you can control the initial size explicitly by passing n_agents.

property agents: AgentSet

Return an AgentSet with the agents in the space.

calculate_difference_vector(point: ndarray, agents=None) → ndarray

Calculate the difference vector between the point and all agenents.

Parameters:

• point – the point to calculate the difference vector for

• agents – the agents to calculate the difference vector of point with. By default, all agents are considered.

calculate_distances(point: Buffer | _SupportsArray[dtype[Any]] | _NestedSequence[_SupportsArray[dtype[Any]]] | bool | int | float | complex | str | bytes | _NestedSequence[bool | int | float | complex | str | bytes], agents: Iterable[Agent] | None = None, **kwargs) → tuple[ndarray, list]

Calculate the distance between the point and all agents.

Parameters:

• point – the point to calculate the difference vector for

• agents – the agents to calculate the difference vector of point with. By default, all agents are considered.

• kwargs – any additional keyword arguments are passed to scipy’s cdist, which is used only if torus is False. This allows for non-Euclidian distance measures.

get_agents_in_radius(point: Buffer | _SupportsArray[dtype[Any]] | _NestedSequence[_SupportsArray[dtype[Any]]] | bool | int | float | complex | str | bytes | _NestedSequence[bool | int | float | complex | str | bytes], radius: float | int = 1) → tuple[list, ndarray]

Return the agents and their distances within a radius for the point.

get_k_nearest_agents(point: Buffer | _SupportsArray[dtype[Any]] | _NestedSequence[_SupportsArray[dtype[Any]]] | bool | int | float | complex | str | bytes | _NestedSequence[bool | int | float | complex | str | bytes], k: int = 1) → tuple[list, ndarray]

Return the k nearest agents and their distances to the point.

Notes

This method returns exactly k agents, ignoring ties. In case of ties, the earlier an agent is inserted the higher it will rank.

in_bounds(point: Buffer | _SupportsArray[dtype[Any]] | _NestedSequence[_SupportsArray[dtype[Any]]] | bool | int | float | complex | str | bytes | _NestedSequence[bool | int | float | complex | str | bytes]) → bool

Check if point is inside the bounds of the space.

torus_correct(point: Buffer | _SupportsArray[dtype[Any]] | _NestedSequence[_SupportsArray[dtype[Any]]] | bool | int | float | complex | str | bytes | _NestedSequence[bool | int | float | complex | str | bytes]) → ndarray

Apply a torus correction to the point.

Continuous space agents.

class HasPositionProtocol(*args, **kwargs)

Protocol for continuous space position holders.

class ContinuousSpaceAgent(space: ContinuousSpace, model)

A continuous space agent.

space

the continuous space in which the agent is located

Type:

ContinuousSpace

position

the position of the agent

Type:

np.ndarray

Initialize a continuous space agent.

Parameters:

• space – the continuous space in which the agent is located

• model – the model to which the agent belongs

property position: ndarray

Position of the agent.

remove() → None

Remove and delete the agent from the model and continuous space.

get_neighbors_in_radius(radius: float | int = 1) → tuple[list, ndarray]

Get neighbors within radius.

Parameters:

radius – radius within which to look for neighbors

get_nearest_neighbors(k: int = 1) → tuple[list, ndarray]

Get neighbors within radius.

Parameters:

k – the number of nearest neighbors to return