diff --git a/mesa/experimental/devs/__init__.py b/mesa/experimental/devs/__init__.py index a3f19f7f2ee..618ff16d52a 100644 --- a/mesa/experimental/devs/__init__.py +++ b/mesa/experimental/devs/__init__.py @@ -18,7 +18,14 @@ combining agent-based modeling with event scheduling. """ -from .eventlist import Priority, SimulationEvent +from .eventlist import EventList, Priority, SimulationEvent from .simulator import ABMSimulator, DEVSimulator, Simulator -__all__ = ["ABMSimulator", "DEVSimulator", "Priority", "SimulationEvent", "Simulator"] +__all__ = [ + "ABMSimulator", + "DEVSimulator", + "EventList", + "Priority", + "SimulationEvent", + "Simulator", +] diff --git a/mesa/experimental/devs/scheduler.py b/mesa/experimental/devs/scheduler.py new file mode 100644 index 00000000000..bcac84860ef --- /dev/null +++ b/mesa/experimental/devs/scheduler.py @@ -0,0 +1,289 @@ +"""Event scheduling and model execution. + +This module provides the Scheduler class which handles event scheduling +and model execution, and the @scheduled decorator for marking recurring methods. +""" + +from __future__ import annotations + +from collections.abc import Callable +from typing import TYPE_CHECKING, Any + +from .eventlist import EventList, Priority, SimulationEvent + +if TYPE_CHECKING: + from mesa import Model + + +# Attribute name used to mark scheduled methods +_SCHEDULED_ATTR = "_mesa_scheduled_interval" + + +def scheduled(interval: float = 1.0): + """Decorator to mark a method as scheduled for recurring execution. + + Args: + interval: Time interval between executions (default: 1.0). + + Returns: + Decorated method with scheduling metadata. + + Examples: + class MyModel(Model): + @scheduled() # Called every 1.0 time units + def step(self): + self.agents.shuffle_do("step") + + @scheduled(interval=7.0) # Called every 7.0 time units + def weekly_update(self): + self.collect_stats() + + @scheduled(interval=0.1) # Called every 0.1 time units + def fast_process(self): + self.update_physics() + """ + + def decorator(method: Callable) -> Callable: + setattr(method, _SCHEDULED_ATTR, interval) + return method + + return decorator + + +def get_scheduled_methods(obj: object) -> dict[str, float]: + """Find all methods decorated with @scheduled on an object. + + Args: + obj: Object to inspect for scheduled methods. + + Returns: + Dictionary mapping method names to their intervals. + """ + scheduled_methods = {} + for name in dir(obj): + if name.startswith("_"): + continue + try: + method = getattr(obj, name) + if callable(method) and hasattr(method, _SCHEDULED_ATTR): + interval = getattr(method, _SCHEDULED_ATTR) + scheduled_methods[name] = interval + except AttributeError: + continue + return scheduled_methods + + +class Scheduler: + """Handles event scheduling and model execution. + + This class manages the event list and provides methods for scheduling + events and running the model. It is used internally by Model. + + Attributes: + model: The model this scheduler is attached to. + """ + + def __init__(self, model: Model) -> None: + """Initialize the scheduler. + + Args: + model: The model instance to schedule events for. + """ + self._model = model + self._event_list = EventList() + self._recurring_events: dict[str, SimulationEvent] = {} + + def start_recurring( + self, scheduled_methods: dict[str, float] | None = None + ) -> None: + """Start all recurring methods. + + Args: + scheduled_methods: Dictionary mapping method names to intervals. + If None, scans the model for @scheduled decorated methods. + """ + if scheduled_methods is None: + scheduled_methods = get_scheduled_methods(self._model) + + for method_name, interval in scheduled_methods.items(): + method = getattr(self._model, method_name) + self._schedule_recurring(method_name, method, interval) + + def _schedule_recurring(self, name: str, method: Callable, interval: float) -> None: + """Schedule a recurring method execution. + + Args: + name: Name of the method (for tracking). + method: The method to call. + interval: Time interval between calls. + """ + + # Create a wrapper that reschedules after execution + def recurring_wrapper(): + method() + # Reschedule for next interval + next_time = self._model.time + interval + event = SimulationEvent( + next_time, recurring_wrapper, priority=Priority.DEFAULT + ) + self._event_list.add_event(event) + self._recurring_events[name] = event + + # Schedule first execution + first_time = self._model.time + interval + event = SimulationEvent( + first_time, recurring_wrapper, priority=Priority.DEFAULT + ) + self._event_list.add_event(event) + self._recurring_events[name] = event + + # ------------------------------------------------------------------------- + # Event Scheduling + # ------------------------------------------------------------------------- + + def schedule( + self, + callback: Callable, + *, + at: float | None = None, + after: float | None = None, + priority: Priority = Priority.DEFAULT, + args: list[Any] | None = None, + kwargs: dict[str, Any] | None = None, + ) -> SimulationEvent: + """Schedule an event to be executed at a specific time. + + Args: + callback: The callable to execute for this event. + at: Absolute time at which to execute the event. + after: Time delta from now at which to execute the event. + priority: Priority level for simultaneous events. + args: Positional arguments for the callback. + kwargs: Keyword arguments for the callback. + + Returns: + SimulationEvent: The scheduled event (can be used to cancel). + + Raises: + ValueError: If neither `at` nor `after` is specified, or both are. + """ + if (at is None) == (after is None): + raise ValueError("Specify exactly one of 'at' or 'after'") + + if at is not None: + event_time = at + if event_time < self._model.time: + raise ValueError( + f"Cannot schedule event in the past " + f"(at={at}, current time={self._model.time})" + ) + else: + event_time = self._model.time + after + + event = SimulationEvent( + event_time, + callback, + priority=priority, + function_args=args, + function_kwargs=kwargs, + ) + self._event_list.add_event(event) + return event + + def cancel(self, event: SimulationEvent) -> None: + """Cancel a scheduled event. + + Args: + event: The event to cancel. + """ + self._event_list.remove(event) + + # ------------------------------------------------------------------------- + # Model Execution + # ------------------------------------------------------------------------- + + def run( + self, + *, + until: float | None = None, + duration: float | None = None, + steps: int | None = None, + condition: Callable[[Model], bool] | None = None, + ) -> None: + """Run the model. + + Args: + until: Run until simulation time reaches this value. + duration: Run for this many time units from current time. + steps: Run for this many steps (calls to step method, if exists). + condition: Run while this condition returns True. + + Raises: + ValueError: If no termination criterion is specified. + """ + end_time = self._determine_end_time(until, duration, steps, condition) + + # Main simulation loop + while self._model.running: + if condition is not None and not condition(self._model): + break + + if self._event_list.is_empty(): + self._model.time = end_time + break + + # Peek at next event + try: + next_events = self._event_list.peak_ahead(1) + if not next_events: + break + next_event = next_events[0] + except IndexError: + break + + # Check if next event is within our time horizon + if next_event.time > end_time: + self._model.time = end_time + break + + # Execute the event + event = self._event_list.pop_event() + self._model.time = event.time + event.execute() + + def _determine_end_time( + self, + until: float | None, + duration: float | None, + steps: int | None, + condition: Callable[[Model], bool] | None, + ) -> float: + """Determine the end time based on provided arguments.""" + if until is not None: + return until + elif duration is not None: + return self._model.time + duration + elif steps is not None: + # For backward compat: steps means number of step() calls + # Each step is at interval 1.0 by default + return self._model.time + steps + elif condition is not None: + return float("inf") + else: + raise ValueError( + "Specify at least one of: 'until', 'duration', 'steps', or 'condition'" + ) + + def clear(self) -> None: + """Clear all scheduled events.""" + self._event_list.clear() + self._recurring_events.clear() + + @property + def is_empty(self) -> bool: + """Check if there are no scheduled events.""" + return self._event_list.is_empty() + + def peek(self, n: int = 1) -> list[SimulationEvent]: + """Look at the next n events without removing them.""" + return self._event_list.peak_ahead(n) diff --git a/mesa/experimental/devs/simulator.py b/mesa/experimental/devs/simulator.py index f4c43626031..5097e1566d0 100644 --- a/mesa/experimental/devs/simulator.py +++ b/mesa/experimental/devs/simulator.py @@ -1,28 +1,16 @@ -"""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. +"""Backward-compatible simulator wrappers. + +These classes provide the legacy Simulator API while delegating to Model's +integrated event scheduling. New code should use Model.run() and Model.schedule() +directly. + +Deprecated: + ABMSimulator and DEVSimulator are deprecated. Use Model.run() and + Model.schedule() directly instead. """ from __future__ import annotations -import numbers import warnings from collections.abc import Callable from typing import TYPE_CHECKING, Any @@ -34,140 +22,115 @@ class Simulator: - """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 - - Attributes: - event_list (EventList): The list of events to execute - time (float | int): The current simulation time - time_unit (type) : The unit of the simulation time - model (Model): The model to simulate - + """Legacy simulator base class. + Deprecated: + Use Model.run() and Model.schedule() directly instead. """ - # TODO: add replication support - # TODO: add experimentation support - def __init__(self, time_unit: type, start_time: int | float): """Initialize a Simulator instance. Args: - time_unit: type of the smulaiton time - start_time: the starttime of the simulator + time_unit: Type of the simulation time. + start_time: The start time of the simulator. """ - # should model run in a separate thread, - # and we can then interact with start, stop, run_until, and step? - self.event_list = EventList() + warnings.warn( + "Simulator classes are deprecated. Use Model.run() and Model.schedule() " + "directly instead.", + FutureWarning, + stacklevel=2, + ) self.start_time = start_time self.time_unit = time_unit self.model: Model | None = None + self._local_event_list = EventList() @property def time(self) -> float: - """Simulator time (deprecated).""" + """Simulator time (deprecated, use model.time).""" warnings.warn( "simulator.time is deprecated, use model.time instead", FutureWarning, stacklevel=2, ) - return self.model.time + return self.model.time if self.model else self.start_time + + @property + def event_list(self) -> EventList: + """Return the model's event list or local one if no model.""" + if self.model: + return self.model._scheduler._event_list + return self._local_event_list - def check_time_unit(self, time: int | float) -> bool: ... # noqa: D102 + def check_time_unit(self, time: int | float) -> bool: + """Check whether the time is of the correct unit.""" + return True def setup(self, model: Model) -> None: - """Set up the simulator with the model to simulate. + """Set up the simulator with the model. Args: - model (Model): The model to simulate - - Raises: - Exception if simulator.time is not equal to simulator.starttime - Exception if event list is not empty - + model: The model to simulate. """ if model.time != self.start_time: raise ValueError( - f"Model time ({model.time}) does not match simulator start_time ({self.start_time}). " - "Has the model already been run?" + f"Model time ({model.time}) does not match simulator start_time ({self.start_time})." ) - if not self.event_list.is_empty(): + + if not self._local_event_list.is_empty(): raise ValueError("Events already scheduled. Call setup before scheduling.") self.model = model - model._simulator = self # Register simulator with model + model._simulator = self # Set reference for backward compatibility - def reset(self): + def reset(self) -> None: """Reset the simulator.""" - self.event_list.clear() - if self.model is not None: - self.model._simulator = None + if self.model: + self.model._scheduler.clear() self.model.time = self.start_time self.model = None + self._local_event_list.clear() def run_until(self, end_time: int | float) -> None: """Run the simulator until the end time. Args: - end_time (int | float): The end time for stopping the simulator - - Raises: - Exception if simulator.setup() has not yet been called - + end_time: The end time for stopping the simulator. """ if self.model is None: raise RuntimeError( "Simulator not set up. Call simulator.setup(model) first." ) + self.model.run(until=end_time) - while True: - try: - event = self.event_list.pop_event() - except IndexError: - self.model.time = end_time - break - - if event.time <= end_time: - self.model.time = event.time - event.execute() - else: - self.model.time = end_time - self._schedule_event(event) - break - - def run_next_event(self): - """Execute the next event. - - Raises: - Exception if simulator.setup() has not yet been called + def run_for(self, time_delta: int | float) -> None: + """Run the simulator for the specified time delta. + Args: + time_delta: The time delta to run for. """ + if self.model is None: + raise RuntimeError( + "Simulator not set up. Call simulator.setup(model) first." + ) + self.model.run(duration=time_delta) + + def run_next_event(self) -> None: + """Execute only the next event.""" if self.model is None: raise RuntimeError( "Simulator not set up. Call simulator.setup(model) first." ) - try: - event = self.event_list.pop_event() - except IndexError: + event_list = self.model._scheduler._event_list + if event_list.is_empty(): return + event = event_list.pop_event() self.model.time = event.time event.execute() - def run_for(self, time_delta: int | float): - """Run the simulator for the specified time delta. - - Args: - time_delta (float| int): The time delta. The simulator is run from the current time to the current time - plus the time delta - - """ - # fixme, raise initialization error or something like it if model.setup has not been called - end_time = self.model.time + time_delta - self.run_until(end_time) - def schedule_event_now( self, function: Callable, @@ -175,18 +138,7 @@ def schedule_event_now( function_args: list[Any] | None = None, function_kwargs: dict[str, Any] | None = None, ) -> SimulationEvent: - """Schedule event for the current time instant. - - Args: - 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: - SimulationEvent: the simulation event that is scheduled - - """ + """Schedule event for the current time instant.""" return self.schedule_event_relative( function, 0.0, @@ -203,22 +155,16 @@ def schedule_event_absolute( function_args: list[Any] | None = None, function_kwargs: dict[str, Any] | None = None, ) -> SimulationEvent: - """Schedule event for the specified time instant. - - Args: - 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: - SimulationEvent: the simulation event that is scheduled - - """ - if self.model.time > time: - raise ValueError("trying to schedule an event in the past") - + """Schedule event for the specified time instant.""" + if self.model: + return self.model.schedule( + function, + at=time, + priority=priority, + args=function_args, + kwargs=function_kwargs, + ) + # No model yet, store locally event = SimulationEvent( time, function, @@ -226,7 +172,7 @@ def schedule_event_absolute( function_args=function_args, function_kwargs=function_kwargs, ) - self._schedule_event(event) + self._local_event_list.add_event(event) return event def schedule_event_relative( @@ -237,88 +183,53 @@ def schedule_event_relative( function_args: list[Any] | None = None, function_kwargs: dict[str, Any] | None = None, ) -> SimulationEvent: - """Schedule event for the current time plus the time delta. - - Args: - 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: - SimulationEvent: the simulation event that is scheduled - - """ + """Schedule event relative to current time.""" + if self.model: + return self.model.schedule( + function, + after=time_delta, + priority=priority, + args=function_args, + kwargs=function_kwargs, + ) + # No model yet + current = self.start_time event = SimulationEvent( - self.model.time + time_delta, + current + time_delta, function, priority=priority, function_args=function_args, function_kwargs=function_kwargs, ) - self._schedule_event(event) + self._local_event_list.add_event(event) return event def cancel_event(self, event: SimulationEvent) -> None: - """Remove the event from the event list. - - Args: - event (SimulationEvent): The simulation event to remove - - """ - self.event_list.remove(event) - - def _schedule_event(self, event: SimulationEvent): - if not self.check_time_unit(event.time): - raise ValueError( - f"time unit mismatch {event.time} is not of time unit {self.time_unit}" - ) - - # check timeunit of events - self.event_list.add_event(event) + """Cancel a scheduled event.""" + if self.model: + self.model.cancel(event) + else: + self._local_event_list.remove(event) class ABMSimulator(Simulator): - """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. + """Legacy ABM simulator with integer time steps. + Deprecated: + Use Model.run() and Model.schedule() directly instead. """ def __init__(self): - """Initialize a ABM simulator.""" + """Initialize an ABM simulator.""" super().__init__(int, 0) - def setup(self, model): - """Set up the simulator with the model to simulate. - - Args: - model (Model): The model to simulate - - """ - super().setup(model) - self.schedule_event_next_tick(self.model.step, priority=Priority.HIGH) - - def check_time_unit(self, time) -> bool: - """Check whether the time is of the correct unit. - - Args: - time (int | float): the time - - Returns: - bool: whether the time is of the correct unit - - """ + def check_time_unit(self, time: int | float) -> bool: + """Check whether the time is an integer.""" if isinstance(time, int): return True if isinstance(time, float): return time.is_integer() - else: - return False + return False def schedule_event_next_tick( self, @@ -327,15 +238,7 @@ def schedule_event_next_tick( function_args: list[Any] | None = None, function_kwargs: dict[str, Any] | None = None, ) -> SimulationEvent: - """Schedule a SimulationEvent for the next tick. - - Args: - 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 - - """ + """Schedule event for the next tick.""" return self.schedule_event_relative( function, 1, @@ -344,63 +247,18 @@ def schedule_event_next_tick( function_kwargs=function_kwargs, ) - def run_until(self, end_time: int) -> None: - """Run the simulator up to and included the specified end time. - - Args: - 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 - - """ - if self.model is None: - raise RuntimeError( - "Simulator not set up. Call simulator.setup(model) first." - ) - - while True: - try: - event = self.event_list.pop_event() - except IndexError: - self.model.time = float(end_time) - break - - if event.time <= end_time: - self.model.time = float(event.time) - - # Reschedule model.step for next tick if this is a step event - if event.fn() == self.model.step: - self.schedule_event_next_tick( - self.model.step, priority=Priority.HIGH - ) - - event.execute() - else: - self.model.time = float(end_time) - self._schedule_event(event) - break - class DEVSimulator(Simulator): - """A simulator where the unit of time is a float. - - Can be used for full-blown discrete event simulating using event scheduling. + """Legacy DEVS simulator with floating-point time. + Deprecated: + Use Model.run() and Model.schedule() directly instead. """ def __init__(self): """Initialize a DEVS simulator.""" super().__init__(float, 0.0) - def check_time_unit(self, time) -> bool: - """Check whether the time is of the correct unit. - - Args: - time (float): the time - - Returns: - bool: whether the time is of the correct unit - - """ - return isinstance(time, numbers.Number) + def check_time_unit(self, time: int | float) -> bool: + """Check whether the time is numeric.""" + return isinstance(time, (int, float)) diff --git a/mesa/model.py b/mesa/model.py index 43c3677fc63..fd972c20bbc 100644 --- a/mesa/model.py +++ b/mesa/model.py @@ -3,30 +3,23 @@ Core Objects: Model """ -# Mypy; for the `|` operator purpose -# Remove this __future__ import once the oldest supported Python is 3.10 from __future__ import annotations import random import sys -from collections.abc import Sequence - -# mypy +from collections.abc import Callable, Sequence from typing import Any import numpy as np from mesa.agent import Agent, AgentSet -from mesa.experimental.devs import Simulator -from mesa.mesa_logging import create_module_logger, method_logger +from mesa.experimental.devs.eventlist import Priority, SimulationEvent +from mesa.experimental.devs.scheduler import Scheduler SeedLike = int | np.integer | Sequence[int] | np.random.SeedSequence RNGLike = np.random.Generator | np.random.BitGenerator -_mesa_logger = create_module_logger() - - class Model: """Base class for models in the Mesa ABM library. @@ -36,116 +29,166 @@ class Model: Attributes: running: A boolean indicating if the model should continue running. - steps: the number of times `model.step()` has been called. - random: a seeded python.random number generator. - rng : a seeded numpy.random.Generator + steps: The number of times `model.step()` has been called. + time: The current simulation time. + random: A seeded python.random number generator. + rng: A seeded numpy.random.Generator. Notes: - Model.agents returns the AgentSet containing all agents registered with the model. Changing - the content of the AgentSet directly can result in strange behavior. If you want change the - composition of this AgentSet, ensure you operate on a copy. - + Model.agents returns the AgentSet containing all agents registered with + the model. Changing the content of the AgentSet directly can result in + strange behavior. If you want to change the composition of this AgentSet, + ensure you operate on a copy. """ - @method_logger(__name__) def __init__( self, *args: Any, seed: float | None = None, rng: RNGLike | SeedLike | None = None, - step_duration: float = 1.0, **kwargs: Any, ) -> None: """Create a new model. - Overload this method with the actual code to initialize the model. Always start with super().__init__() - to initialize the model object properly. - Args: - args: arguments to pass onto super - seed: the seed for the random number generator - rng : Pseudorandom number generator state. When `rng` is None, a new `numpy.random.Generator` is created - using entropy from the operating system. Types other than `numpy.random.Generator` are passed to - `numpy.random.default_rng` to instantiate a `Generator`. - step_duration: How much time advances each step (default 1.0) - kwargs: keyword arguments to pass onto super + args: Arguments to pass onto super. + seed: The seed for the random number generator. + rng: Pseudorandom number generator state. When `rng` is None, a new + `numpy.random.Generator` is created using entropy from the OS. + kwargs: Keyword arguments to pass onto super. Notes: - you have to pass either seed or rng, but not both. - + You have to pass either seed or rng, but not both. """ super().__init__(*args, **kwargs) self.running: bool = True self.steps: int = 0 self.time: float = 0.0 - self._step_duration: float = step_duration - # Track if a simulator is controlling time - self._simulator: Simulator | None = None + # Initialize scheduler for event management + self._scheduler = Scheduler(self) + + # Random number generator setup + self._init_random(seed, rng) + # Agent registration data structures + self._agents = {} + self._agents_by_type: dict[type[Agent], AgentSet] = {} + self._all_agents = AgentSet([], random=self.random) + + def _init_random(self, seed: float | None, rng: RNGLike | SeedLike | None) -> None: + """Initialize random number generators.""" if (seed is not None) and (rng is not None): raise ValueError("you have to pass either rng or seed, not both") elif seed is None: self.rng: np.random.Generator = np.random.default_rng(rng) - self._rng = ( - self.rng.bit_generator.state - ) # this allows for reproducing the rng - + self._rng = self.rng.bit_generator.state try: self.random = random.Random(rng) except TypeError: seed = int(self.rng.integers(np.iinfo(np.int32).max)) self.random = random.Random(seed) - self._seed = seed # this allows for reproducing stdlib.random - elif rng is None: + self._seed = seed + else: self.random = random.Random(seed) - self._seed = seed # this allows for reproducing stdlib.random - + self._seed = seed try: - self.rng: np.random.Generator = np.random.default_rng(seed) + self.rng = np.random.default_rng(seed) except TypeError: rng = self.random.randint(0, sys.maxsize) - self.rng: np.random.Generator = np.random.default_rng(rng) + self.rng = np.random.default_rng(rng) self._rng = self.rng.bit_generator.state - # Wrap the user-defined step method - self._user_step = self.step - self.step = self._wrapped_step - - # setup agent registration data structures - self._agents = {} # the hard references to all agents in the model - self._agents_by_type: dict[ - type[Agent], AgentSet - ] = {} # a dict with an agentset for each class of agents - self._all_agents = AgentSet( - [], random=self.random - ) # an agenset with all agents - - def _wrapped_step(self, *args: Any, **kwargs: Any) -> None: - """Automatically increments time and steps after calling the user's step method.""" - # Automatically increment time and step counters - self.steps += 1 - # Only auto-increment time if no simulator is controlling it - if self._simulator is None: - self.time += self._step_duration - - _mesa_logger.info( - f"calling model.step for step {self.steps} at time {self.time}" + # Event Scheduling API (delegates to Scheduler) + def schedule( + self, + callback: Callable, + *, + at: float | None = None, + after: float | None = None, + priority: Priority = Priority.DEFAULT, + args: list[Any] | None = None, + kwargs: dict[str, Any] | None = None, + ) -> SimulationEvent: + """Schedule an event to be executed at a specific time. + + Args: + callback: The callable to execute for this event. + at: Absolute time at which to execute the event. + after: Time delta from now at which to execute the event. + priority: Priority level for simultaneous events. + args: Positional arguments for the callback. + kwargs: Keyword arguments for the callback. + + Returns: + SimulationEvent: The scheduled event (can be used to cancel). + + Examples: + model.schedule(callback, at=50.0) + model.schedule(callback, after=10.0) + model.schedule(callback, at=50.0, priority=Priority.HIGH) + """ + return self._scheduler.schedule( + callback, at=at, after=after, priority=priority, args=args, kwargs=kwargs + ) + + def cancel(self, event: SimulationEvent) -> None: + """Cancel a scheduled event. + + Args: + event: The event to cancel. + """ + self._scheduler.cancel(event) + + # Running API (delegates to Scheduler) + def run( + self, + *, + until: float | None = None, + duration: float | None = None, + steps: int | None = None, + condition: Callable[[Model], bool] | None = None, + ) -> None: + """Run the model. + + Args: + until: Run until simulation time reaches this value. + duration: Run for this many time units from current time. + steps: Run for this many steps (each step = 1.0 time units). + condition: Run while this condition returns True. + + Examples: + model.run(until=100) + model.run(duration=50) + model.run(steps=100) + model.run(condition=lambda m: m.running) + """ + self._scheduler.run( + until=until, duration=duration, steps=steps, condition=condition ) - # Call the original user-defined step method - self._user_step(*args, **kwargs) + def step(self) -> None: + """A single step of the model. Override this in subclasses.""" + + def run_model(self) -> None: + """Run the model until running is False. + + Deprecated: Use run(condition=lambda m: m.running) instead. + """ + self.run(condition=lambda m: m.running) + + # Agent Management @property def agents(self) -> AgentSet: - """Provides an AgentSet of all agents in the model, combining agents from all types.""" + """Provides an AgentSet of all agents in the model.""" return self._all_agents @agents.setter def agents(self, agents: Any) -> None: raise AttributeError( - "You are trying to set model.agents. In Mesa 3.0 and higher, this attribute is " - "used by Mesa itself, so you cannot use it directly anymore." - "Please adjust your code to use a different attribute name for custom agent storage." + "You are trying to set model.agents. In Mesa 3.0 and higher, " + "this attribute is used by Mesa itself, so you cannot use it " + "directly anymore. Please use a different attribute name." ) @property @@ -155,70 +198,49 @@ def agent_types(self) -> list[type]: @property def agents_by_type(self) -> dict[type[Agent], AgentSet]: - """A dictionary where the keys are agent types and the values are the corresponding AgentSets.""" + """A dictionary where keys are agent types and values are AgentSets.""" return self._agents_by_type - def register_agent(self, agent): + def register_agent(self, agent: Agent) -> None: """Register the agent with the model. Args: agent: The agent to register. Notes: - This method is called automatically by ``Agent.__init__``, so there - is no need to use this if you are subclassing Agent and calling its - super in the ``__init__`` method. + This method is called automatically by Agent.__init__. """ self._agents[agent] = None - - # because AgentSet requires model, we cannot use defaultdict - # tricks with a function won't work because model then cannot be pickled try: self._agents_by_type[type(agent)].add(agent) except KeyError: - self._agents_by_type[type(agent)] = AgentSet( - [ - agent, - ], - random=self.random, - ) - + self._agents_by_type[type(agent)] = AgentSet([agent], random=self.random) self._all_agents.add(agent) - _mesa_logger.debug( - f"registered {agent.__class__.__name__} with agent_id {agent.unique_id}" - ) - def deregister_agent(self, agent): + def deregister_agent(self, agent: Agent) -> None: """Deregister the agent with the model. Args: agent: The agent to deregister. Notes: - This method is called automatically by ``Agent.remove`` - + This method is called automatically by Agent.remove(). """ del self._agents[agent] self._agents_by_type[type(agent)].remove(agent) self._all_agents.remove(agent) - _mesa_logger.debug(f"deregistered agent with agent_id {agent.unique_id}") - def run_model(self) -> None: - """Run the model until the end condition is reached. - - Overload as needed. - """ - while self.running: - self.step() - - def step(self) -> None: - """A single step. Fill in here.""" + def remove_all_agents(self) -> None: + """Remove all agents from the model.""" + for agent in list(self._agents.keys()): + agent.remove() + # Random Number Generator Management def reset_randomizer(self, seed: int | None = None) -> None: """Reset the model random number generator. Args: - seed: A new seed for the RNG; if None, reset using the current seed + seed: A new seed for the RNG; if None, reset using the current seed. """ if seed is None: seed = self._seed @@ -226,23 +248,10 @@ def reset_randomizer(self, seed: int | None = None) -> None: self._seed = seed def reset_rng(self, rng: RNGLike | SeedLike | None = None) -> None: - """Reset the model random number generator. + """Reset the numpy random number generator. Args: - rng: A new seed for the RNG; if None, reset using the current seed + rng: A new seed for the RNG; if None, reset using the current seed. """ self.rng = np.random.default_rng(rng) self._rng = self.rng.bit_generator.state - - def remove_all_agents(self): - """Remove all agents from the model. - - Notes: - This method calls agent.remove for all agents in the model. If you need to remove agents from - e.g., a SingleGrid, you can either explicitly implement your own agent.remove method or clean this up - near where you are calling this method. - - """ - # we need to wrap keys in a list to avoid a RunTimeError: dictionary changed size during iteration - for agent in list(self._agents.keys()): - agent.remove()