1 ```# -*- coding: utf-8 -*- ``` 2 3 ```""" ``` 3 ```Mesa Time Module ``` 4 ```================ ``` 5 6 ```Objects for handling the time component of a model. In particular, this module ``` 7 ```contains Schedulers, which handle agent activation. A Scheduler is an object ``` 8 ```which controls when agents are called upon to act, and when. ``` 9 10 ```The activation order can have a serious impact on model behavior, so it's ``` 11 ```important to specify it explicitly. Example simple activation regimes include ``` 12 ```activating all agents in the same order every step, shuffling the activation ``` 13 ```order every time, activating each agent *on average* once per step, and more. ``` 14 15 ```Key concepts: ``` 16 ``` Step: Many models advance in 'steps'. A step may involve the activation of ``` 17 ``` all agents, or a random (or selected) subset of them. Each agent in turn ``` 18 ``` may have their own step() method. ``` 19 20 ``` Time: Some models may simulate a continuous 'clock' instead of discrete ``` 21 ``` steps. However, by default, the Time is equal to the number of steps the ``` 22 ``` model has taken. ``` 23 ```""" ``` 24 25 3 ```from collections import OrderedDict ``` 26 27 ```# mypy ``` 28 3 ```from typing import Dict, Iterator, List, Optional, Union ``` 29 3 ```from .agent import Agent ``` 30 3 ```from .model import Model ``` 31 32 33 ```# BaseScheduler has a self.time of int, while ``` 34 ```# StagedActivation has a self.time of float ``` 35 3 ```TimeT = Union[float, int] ``` 36 37 38 3 ```class BaseScheduler: ``` 39 ``` """ Simplest scheduler; activates agents one at a time, in the order ``` 40 ``` they were added. ``` 41 42 ``` Assumes that each agent added has a *step* method which takes no arguments. ``` 43 44 ``` (This is explicitly meant to replicate the scheduler in MASON). ``` 45 46 ``` """ ``` 47 48 3 ``` def __init__(self, model: Model) -> None: ``` 49 ``` """ Create a new, empty BaseScheduler. """ ``` 50 3 ``` self.model = model ``` 51 3 ``` self.steps = 0 ``` 52 3 ``` self.time = 0 # type: TimeT ``` 53 3 ``` self._agents = OrderedDict() # type: Dict[int, Agent] ``` 54 55 3 ``` def add(self, agent: Agent) -> None: ``` 56 ``` """ Add an Agent object to the schedule. ``` 57 58 ``` Args: ``` 59 ``` agent: An Agent to be added to the schedule. NOTE: The agent must ``` 60 ``` have a step() method. ``` 61 62 ``` """ ``` 63 64 3 ``` if agent.unique_id in self._agents: ``` 65 0 ``` raise Exception("Agent with unique id {0} already added to scheduler".format(repr(agent.unique_id))) ``` 66 67 3 ``` self._agents[agent.unique_id] = agent ``` 68 69 3 ``` def remove(self, agent: Agent) -> None: ``` 70 ``` """ Remove all instances of a given agent from the schedule. ``` 71 72 ``` Args: ``` 73 ``` agent: An agent object. ``` 74 75 ``` """ ``` 76 3 ``` del self._agents[agent.unique_id] ``` 77 78 3 ``` def step(self) -> None: ``` 79 ``` """ Execute the step of all the agents, one at a time. """ ``` 80 3 ``` for agent in self.agent_buffer(shuffled=False): ``` 81 3 ``` agent.step() ``` 82 3 ``` self.steps += 1 ``` 83 3 ``` self.time += 1 ``` 84 85 3 ``` def get_agent_count(self) -> int: ``` 86 ``` """ Returns the current number of agents in the queue. """ ``` 87 3 ``` return len(self._agents.keys()) ``` 88 89 3 ``` @property ``` 90 3 ``` def agents(self) -> List[Agent]: ``` 91 3 ``` return list(self._agents.values()) ``` 92 93 3 ``` def agent_buffer(self, shuffled: bool = False) -> Iterator[Agent]: ``` 94 ``` """ Simple generator that yields the agents while letting the user ``` 95 ``` remove and/or add agents during stepping. ``` 96 97 ``` """ ``` 98 3 ``` agent_keys = list(self._agents.keys()) ``` 99 3 ``` if shuffled: ``` 100 3 ``` self.model.random.shuffle(agent_keys) ``` 101 102 3 ``` for key in agent_keys: ``` 103 3 ``` if key in self._agents: ``` 104 3 ``` yield self._agents[key] ``` 105 106 107 3 ```class RandomActivation(BaseScheduler): ``` 108 ``` """ A scheduler which activates each agent once per step, in random order, ``` 109 ``` with the order reshuffled every step. ``` 110 111 ``` This is equivalent to the NetLogo 'ask agents...' and is generally the ``` 112 ``` default behavior for an ABM. ``` 113 114 ``` Assumes that all agents have a step(model) method. ``` 115 116 ``` """ ``` 117 118 3 ``` def step(self) -> None: ``` 119 ``` """ Executes the step of all agents, one at a time, in ``` 120 ``` random order. ``` 121 122 ``` """ ``` 123 3 ``` for agent in self.agent_buffer(shuffled=True): ``` 124 3 ``` agent.step() ``` 125 3 ``` self.steps += 1 ``` 126 3 ``` self.time += 1 ``` 127 128 129 3 ```class SimultaneousActivation(BaseScheduler): ``` 130 ``` """ A scheduler to simulate the simultaneous activation of all the agents. ``` 131 132 ``` This scheduler requires that each agent have two methods: step and advance. ``` 133 ``` step() activates the agent and stages any necessary changes, but does not ``` 134 ``` apply them yet. advance() then applies the changes. ``` 135 136 ``` """ ``` 137 138 3 ``` def step(self) -> None: ``` 139 ``` """ Step all agents, then advance them. """ ``` 140 3 ``` agent_keys = list(self._agents.keys()) ``` 141 3 ``` for agent_key in agent_keys: ``` 142 3 ``` self._agents[agent_key].step() ``` 143 3 ``` for agent_key in agent_keys: ``` 144 3 ``` self._agents[agent_key].advance() ``` 145 3 ``` self.steps += 1 ``` 146 3 ``` self.time += 1 ``` 147 148 149 3 ```class StagedActivation(BaseScheduler): ``` 150 ``` """ A scheduler which allows agent activation to be divided into several ``` 151 ``` stages instead of a single `step` method. All agents execute one stage ``` 152 ``` before moving on to the next. ``` 153 154 ``` Agents must have all the stage methods implemented. Stage methods take a ``` 155 ``` model object as their only argument. ``` 156 157 ``` This schedule tracks steps and time separately. Time advances in fractional ``` 158 ``` increments of 1 / (# of stages), meaning that 1 step = 1 unit of time. ``` 159 160 ``` """ ``` 161 162 3 ``` def __init__( ``` 163 ``` self, ``` 164 ``` model: Model, ``` 165 ``` stage_list: Optional[List[str]] = None, ``` 166 ``` shuffle: bool = False, ``` 167 ``` shuffle_between_stages: bool = False, ``` 168 ``` ) -> None: ``` 169 ``` """ Create an empty Staged Activation schedule. ``` 170 171 ``` Args: ``` 172 ``` model: Model object associated with the schedule. ``` 173 ``` stage_list: List of strings of names of stages to run, in the ``` 174 ``` order to run them in. ``` 175 ``` shuffle: If True, shuffle the order of agents each step. ``` 176 ``` shuffle_between_stages: If True, shuffle the agents after each ``` 177 ``` stage; otherwise, only shuffle at the start ``` 178 ``` of each step. ``` 179 180 ``` """ ``` 181 3 ``` super().__init__(model) ``` 182 3 ``` self.stage_list = ["step"] if not stage_list else stage_list ``` 183 3 ``` self.shuffle = shuffle ``` 184 3 ``` self.shuffle_between_stages = shuffle_between_stages ``` 185 3 ``` self.stage_time = 1 / len(self.stage_list) ``` 186 187 3 ``` def step(self) -> None: ``` 188 ``` """ Executes all the stages for all agents. """ ``` 189 3 ``` agent_keys = list(self._agents.keys()) ``` 190 3 ``` if self.shuffle: ``` 191 3 ``` self.model.random.shuffle(agent_keys) ``` 192 3 ``` for stage in self.stage_list: ``` 193 3 ``` for agent_key in agent_keys: ``` 194 3 ``` getattr(self._agents[agent_key], stage)() # Run stage ``` 195 3 ``` if self.shuffle_between_stages: ``` 196 0 ``` self.model.random.shuffle(agent_keys) ``` 197 3 ``` self.time += self.stage_time ``` 198 199 3 ``` self.steps += 1 ```

Read our documentation on viewing source code .