Time#

Underlying modules for event scheduling and time advancement.

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

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

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 list[Any]: Argument for the function

function_kwargs dict[str, Any]: Keyword arguments for the function

Notes

Simulation events use a weak reference to the callable. If the callback no longer exists at execution time (e.g., because an agent has been removed), execution will fail silently. Lambda callbacks are rejected at Event creation.

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() → None[source]#: Execute this event.

cancel() → None[source]#: Cancel this event.

class EventGenerator(model: Model, function: Callable[..., None], schedule: Schedule, priority: Priority = Priority.DEFAULT)[source]#

A generator that creates recurring events based on a Schedule.

Unlike a single Event, an EventGenerator is persistent and can be stopped or configured with stop conditions.

model#: The model this generator belongs to

function#: The callable to execute for each generated event

schedule#: The Schedule defining when events occur

priority#: Priority level for generated events

Notes

Event generators 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. If you want to use functools.partial, please assign the partial function to a variable prior to creating the generator.

Initialize an EventGenerator.

Parameters:

model – The model this generator belongs to
function – The callable to execute for each generated event. Use functools.partial to bind arguments.
schedule – The Schedule defining timing
priority – Priority level for generated events

property is_active: bool#: Return whether the generator is currently active.

property execution_count: int#: Return the number of times this generator has executed.

property next_scheduled_time: float | None#: Return the time of the next scheduled execution, or None if not scheduled.

start() → EventGenerator[source]#

Start the event generator.

Returns:: Self for method chaining

stop() → None[source]#: Stop the event generator immediately.

pause() → None[source]#

Pause the event generator temporarily.

This cancels the currently scheduled event but keeps the generator active in the model. Execution can be resumed later using resume().

resume() → None[source]#: Resume a paused event generator.

class EventList[source]#

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

Add the event to the event list.

Parameters:: event (Event) – The event to be added

peek_ahead(n: int = 1) → list[Event][source]#

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

Parameters:: n (int) – The number of events to look ahead
Returns:: list[Event]
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() → Event[source]#: Pop the first element from the event list.

compact() → None[source]#

Remove all canceled events from the heap and rebuild it.

If there are many canceled events, compaction can speed up performance substantially.

is_empty() → bool[source]#: Return whether the event list is empty.

remove(event: Event) → None[source]#

Remove an event from the event list.

Parameters:: event (Event) – The event to be removed

clear() → None[source]#: Clear the event list.

class Priority(*values)[source]#

Enumeration of priority levels.

conjugate()#: Returns self, the complex conjugate of any int.

bit_length()#

Number of bits necessary to represent self in binary.

>>> bin(37)
'0b100101'
>>> (37).bit_length()
6

bit_count()#

Number of ones in the binary representation of the absolute value of self.

Also known as the population count.

>>> bin(13)
'0b1101'
>>> (13).bit_count()
3

to_bytes(length=1, byteorder='big', *, signed=False)#

Return an array of bytes representing an integer.

length: Length of bytes object to use. An OverflowError is raised if the integer is not representable with the given number of bytes. Default is length 1.
byteorder: The byte order used to represent the integer. If byteorder is ‘big’, the most significant byte is at the beginning of the byte array. If byteorder is ‘little’, the most significant byte is at the end of the byte array. To request the native byte order of the host system, use sys.byteorder as the byte order value. Default is to use ‘big’.
signed: Determines whether two’s complement is used to represent the integer. If signed is False and a negative integer is given, an OverflowError is raised.

classmethod from_bytes(bytes, byteorder='big', *, signed=False)#

Return the integer represented by the given array of bytes.

bytes: Holds the array of bytes to convert. The argument must either support the buffer protocol or be an iterable object producing bytes. Bytes and bytearray are examples of built-in objects that support the buffer protocol.
byteorder: The byte order used to represent the integer. If byteorder is ‘big’, the most significant byte is at the beginning of the byte array. If byteorder is ‘little’, the most significant byte is at the end of the byte array. To request the native byte order of the host system, use sys.byteorder as the byte order value. Default is to use ‘big’.
signed: Indicates whether two’s complement is used to represent the integer.

as_integer_ratio()#

Return a pair of integers, whose ratio is equal to the original int.

The ratio is in lowest terms and has a positive denominator.

>>> (10).as_integer_ratio()
(10, 1)
>>> (-10).as_integer_ratio()
(-10, 1)
>>> (0).as_integer_ratio()
(0, 1)

is_integer()#: Returns True. Exists for duck type compatibility with float.is_integer.

real#: the real part of a complex number

imag#: the imaginary part of a complex number

numerator#: the numerator of a rational number in lowest terms

denominator#: the denominator of a rational number in lowest terms

Defines when something should happen repeatedly.

interval#

Time between executions (fixed value or callable returning value)

Type:: float | int | Callable[[Model], float | int]

start#

Absolute time to begin (None = use current model time + interval)

Type:: float | None

end#

Absolute time to stop (None = no end)

Type:: float | None

count#

Maximum executions (None = unlimited)

Type:: int | None