vi
A smol simulator framework built on top of PyGame.
- Automatic agent wandering behaviour
- Fully deterministic simulations with PRNG seeds
- Install Violet with a simple
pip install😎 - Matrix-powered multi-threaded configuration testing
- Polars-powered simulation analytics
- Replay-able simulations with a ✨ time machine ✨
- Type-safe configuration system
Want to get started right away? Check out the Violet Starter Kit!
A Tour of Violet
Violet is all about creating and researching collective intelligence. And what's better than programming your own little video game to do so?
Under the hood, Violet is using PyGame to render all those fancy pixels to your very screen. However, you don't need to have any knowledge of PyGame whatsoever to get started!
Instead, you only need to familiarise yourself with Violet's vi.agent and vi.simulation modules.
You see, all that's needed to create a video game is a new instance of the vi.simulation.Simulation class.
from vi import Config, Simulation
Simulation(Config())
Yep, that's all it takes to start your simulation. However, it closes right away...
To actually run your simulation, you need to add one more thing.
from vi import Config, Simulation
Simulation(Config()).run()
There, now you have a nice black window appear in the middle of your screen!
Obviously, creating a black window doesn't really count as having created a simulation just yet. So let's add some agents!
from vi import Agent, Config, Simulation
(
Simulation(Config())
.batch_spawn_agents(100, Agent, ["examples/images/white.png"])
.run()
)
We now have 100 agents wiggling around our screen. They just don't particularly do anything just yet. Sure, they're moving, but we want them to interact!
That's where you come in! Customising the behaviour of the agents is central to creating simulations. And it couldn't be any easier to customise what these agents do!
Violet is built around the concept of inheritance.
In a nutshell, inheritance allows you to well, inherit,
all the functions and properties that the Agent class exposes.
But that's not all!
Inheritance also allows you to build on top of the Agent class,
allowing you to implement your agent's custom behaviour simply by implementing an update method.
Let's see some inheritance in action.
class MyAgent(Agent): ...
Here we create a new class called MyAgent, which inherits the Agent class.
Now, the three dots simply tells Python that we still have to add things to it.
But before we do that, we can already start a simulation with our new MyAgent.
(
Simulation(Config())
.batch_spawn_agents(100, MyAgent, ["examples/images/white.png"])
.run()
)
If you look very closely, you can see that there's absolutely no difference! The agent is still aimlessly wandering about. And that's the key takeaway of inheritance. Without adding or changing things, our agent's behaviour will be exactly the same!
To actually customise our agent's behaviour,
we have to implement the vi.agent.Agent.update method.
The update method will run on every tick of the simulation,
after the positions of all the agents have been updated.
Now, you might wonder why the agent's position isn't updated in the update method.
Long story short, in most collective intelligence simulations,
your agent needs to interact with other agents.
Therefore, it needs to have a sense of which other agents are in its proximity.
To make sure that whenever agent A sees agent B, agent B also sees agent A,
we need to separate the position changing code from the behavioural code.
This way, either both agents see each other, or they don't see each other at all.
On the topic of agents seeing each other,
let's implement our own update method in which we want our agent
to change its colour whenever it sees another agent.
To be able to change colours, we need to supply two different images to our Agent. I'll go with a white and a red circle.
(
Simulation(Config())
.batch_spawn_agents(100, MyAgent, [
"examples/images/white.png",
"examples/images/red.png",
])
.run()
)
If we run the simulation again, we see that the white image is picked automatically, as it is the first image in the list of images.
Now, to change the currently selected image,
we can call the vi.agent.Agent.change_image method.
class MyAgent(Agent):
def update(self) -> None:
self.change_image(1)
Remember that the first index of a Python list is 0, so changing the image to index 1 actually selects our second image. Cool! Our agents are now red instead! 😎
However, we don't want them to always be red. Instead, we only want our agents to turn red when they see at least one other agent.
Fortunately, Violet already keeps track of who sees who for us, so we don't have to implement any code ourselves!
Instead, we can utilise the vi.agent.Agent.in_proximity_accuracy method.
This will return an iterator. Combined with vi.util.count, we can count the number of agents in proximity.
Let's say that if we count at least one other agent, we turn red. Otherwise, we change the image back to white!
from vi.util import count
class MyAgent(Agent):
def update(self) -> None:
if count(self.in_proximity_accuracy()) >= 1:
self.change_image(1)
else:
self.change_image(0)
And there we have it! A disco of a simulation with agents swapping colours whenever they get close to someone.
Now, there's way more to explore. But this should give you an impression on how to change the behaviour of your agents with Violet. Explore some of the modules on the left and experiment away!
1"""A smol simulator framework built on top of [PyGame](https://www.pygame.org/docs/). 2 3- Automatic agent wandering behaviour 4- Fully deterministic simulations with PRNG seeds 5- Install Violet with a simple `pip install` 😎 6- Matrix-powered multi-threaded configuration testing 7- [Polars](https://docs.pola.rs)-powered simulation analytics 8- Replay-able simulations with a ✨ time machine ✨ 9- Type-safe configuration system 10 11Want to get started right away? 12Check out the [Violet Starter Kit](https://github.com/m-rots/violet-starter-kit)! 13 14# A Tour of Violet 15 16Violet is all about creating and researching collective intelligence. 17And what's better than programming your own little video game to do so? 18 19Under the hood, Violet is using [PyGame](https://www.pygame.org/docs/) to render all those fancy pixels to your very screen. 20However, you don't need to have any knowledge of PyGame whatsoever to get started! 21 22Instead, you only need to familiarise yourself with Violet's `vi.agent` and `vi.simulation` modules. 23You see, all that's needed to create a *video game* is a new instance of the `vi.simulation.Simulation` class. 24 25```python 26from vi import Config, Simulation 27 28 29Simulation(Config()) 30``` 31 32Yep, that's all it takes to start your simulation. 33However, it closes right away... 34 35To actually `run` your simulation, you need to add one more thing. 36 37```python 38from vi import Config, Simulation 39 40 41Simulation(Config()).run() 42``` 43 44There, now you have a nice black window appear in the middle of your screen! 45 46Obviously, creating a black window doesn't really count as having created a simulation just yet. 47So let's add some agents! 48 49```python 50from vi import Agent, Config, Simulation 51 52 53( 54 Simulation(Config()) 55 .batch_spawn_agents(100, Agent, ["examples/images/white.png"]) 56 .run() 57) 58``` 59 60We now have 100 agents wiggling around our screen. 61They just don't particularly do anything just yet. 62Sure, they're *moving*, but we want them to interact! 63 64That's where you come in! 65Customising the behaviour of the agents is central to creating simulations. 66And it couldn't be any easier to customise what these agents do! 67 68Violet is built around the concept of *inheritance*. 69In a nutshell, inheritance allows you to well, *inherit*, 70all the functions and properties that the `Agent` class exposes. 71But that's not all! 72Inheritance also allows you to build on top of the `Agent` class, 73allowing you to implement your agent's custom behaviour simply by implementing an `update` method. 74 75Let's see some inheritance in action. 76 77```python 78class MyAgent(Agent): ... 79``` 80 81Here we create a new class called `MyAgent`, which inherits the `Agent` class. 82Now, the three dots simply tells Python that we still have to add things to it. 83But before we do that, we can already start a simulation with our new `MyAgent`. 84 85```python 86( 87 Simulation(Config()) 88 .batch_spawn_agents(100, MyAgent, ["examples/images/white.png"]) 89 .run() 90) 91``` 92 93If you look very closely, you can see that there's absolutely no difference! 94The agent is still aimlessly wandering about. 95And that's the key takeaway of inheritance. 96Without adding or changing things, 97our agent's behaviour will be exactly the same! 98 99To actually customise our agent's behaviour, 100we have to implement the `vi.agent.Agent.update` method. 101 102The `update` method will run on every tick of the simulation, 103after the positions of all the agents have been updated. 104Now, you might wonder why the agent's position isn't updated in the `update` method. 105Long story short, in most collective intelligence simulations, 106your agent needs to interact with other agents. 107Therefore, it needs to have a sense of which other agents are in its proximity. 108To make sure that whenever agent A sees agent B, agent B also sees agent A, 109we need to separate the position changing code from the behavioural code. 110This way, either both agents see each other, or they don't see each other at all. 111 112On the topic of agents seeing each other, 113let's implement our own `update` method in which we want our agent 114to change its colour whenever it sees another agent. 115 116To be able to change colours, 117we need to supply two different images to our Agent. 118I'll go with a white and a red circle. 119 120```python 121( 122 Simulation(Config()) 123 .batch_spawn_agents(100, MyAgent, [ 124 "examples/images/white.png", 125 "examples/images/red.png", 126 ]) 127 .run() 128) 129``` 130 131If we run the simulation again, 132we see that the white image is picked automatically, 133as it is the first image in the list of images. 134 135Now, to change the currently selected image, 136we can call the `vi.agent.Agent.change_image` method. 137 138```python 139class MyAgent(Agent): 140 def update(self) -> None: 141 self.change_image(1) 142``` 143 144Remember that the first index of a Python list is 0, 145so changing the image to index 1 actually selects our second image. 146Cool! Our agents are now red instead! 😎 147 148However, we don't want them to always be red. 149Instead, we only want our agents to turn red when they see at least one other agent. 150 151Fortunately, Violet already keeps track of who sees who for us, so we don't have to implement any code ourselves! 152Instead, we can utilise the `vi.agent.Agent.in_proximity_accuracy` method. 153This will return an iterator. Combined with `vi.util.count`, we can count the number of agents in proximity. 154 155Let's say that if we count at least one other agent, we turn red. Otherwise, we change the image back to white! 156 157```python 158from vi.util import count 159 160 161class MyAgent(Agent): 162 def update(self) -> None: 163 if count(self.in_proximity_accuracy()) >= 1: 164 self.change_image(1) 165 else: 166 self.change_image(0) 167``` 168 169And there we have it! 170A disco of a simulation with agents swapping colours whenever they get close to someone. 171 172Now, there's way more to explore. 173But this should give you an impression on how to change the behaviour of your agents with Violet. 174Explore some of the modules on the left and experiment away! 175""" 176 177from .agent import Agent 178from .config import Config, Matrix, Window 179from .simulation import HeadlessSimulation, Simulation 180 181 182__all__ = [ 183 "Agent", 184 "Config", 185 "HeadlessSimulation", 186 "Matrix", 187 "Simulation", 188 "Window", 189 "agent", 190 "config", 191 "metrics", # type: ignore[reportUnsupportedDunderAll] 192 "replay", # # type: ignore[reportUnsupportedDunderAll] 193 "simulation", 194 "util", # type: ignore[reportUnsupportedDunderAll] 195]
class
Agent(pygame.sprite.Sprite, typing.Generic[ConfigClass]):
38class Agent[ConfigClass: Config = Config](Sprite): 39 """The `Agent` class is home to Violet's various additions and is built on top of [PyGame's Sprite](https://www.pygame.org/docs/ref/sprite.html) class. 40 41 While you can simply add this `Agent` class to your simulations by calling `batch_spawn_agents`, 42 the agents won't actually do anything interesting. 43 Sure, they'll move around the screen very energetically, but they don't *interact* with each other. 44 45 That's where you come in! 46 By inheriting this `Agent` class you can make use of all its utilities, 47 while also programming the behaviour of your custom agent. 48 """ 49 50 id: int 51 """The unique identifier of the agent.""" 52 53 config: ConfigClass 54 """The config of the simulation that's shared with all agents. 55 56 The config can be overriden when inheriting the Agent class. 57 However, the config must always: 58 59 1. Inherit `Config` 60 2. Be decorated by `@dataclass` 61 """ 62 63 shared: Shared 64 """Attributes that are shared between the simulation and all agents.""" 65 66 _images: list[Surface] 67 """A list of images which you can use to change the current image of the agent.""" 68 69 _image_index: int 70 """The currently selected image.""" 71 72 _image_cache: tuple[int, Surface] | None = None 73 74 _area: Rect 75 """The area in which the agent is free to move.""" 76 77 move: Vector2 78 """A two-dimensional vector representing the delta between the agent's current and next position. 79 80 Note that `move` isn't added to the agent's `pos` automatically. 81 Instead, you should manually add the move delta to `pos`, like so: 82 83 ```python 84 self.pos += self.move 85 ``` 86 87 The vector `Vector2(2, 1)` indicates that the agent will be moving 2 pixels along the x axis and 1 pixel along the y axis. 88 You can use the `Vector2` class to calculate the vector's magnitude. 89 90 This property is also used to automatically rotate the agent's image 91 when `vi.config.Schema.image_rotation` is enabled. 92 """ 93 94 pos: Vector2 95 """The current (centre) position of the agent.""" 96 97 _obstacles: Group[Any] 98 """The group of obstacles the agent can collide with.""" 99 100 _sites: Group[Any] 101 """The group of sites on which the agent can appear.""" 102 103 __simulation: HeadlessSimulation[ConfigClass] 104 105 _moving: bool = True 106 """The agent's movement will freeze when moving is set to False.""" 107 108 def __init__( 109 self, 110 images: list[Surface], 111 simulation: HeadlessSimulation[ConfigClass], 112 pos: Vector2 | None = None, 113 move: Vector2 | None = None, 114 ) -> None: 115 Sprite.__init__(self, simulation._all, simulation._agents) 116 117 self.__simulation = simulation 118 119 self.id = simulation._agent_id() 120 self.config = simulation.config 121 self.shared = simulation.shared 122 123 # Default to first image in case no image is given 124 self._image_index = 0 125 self._images = images 126 127 self._obstacles = simulation._obstacles 128 self._sites = simulation._sites 129 130 self._area = simulation._area 131 self.move = ( 132 move 133 if move is not None 134 else random_angle(self.config.movement_speed, prng=self.shared.prng_move) 135 ) 136 137 # On spawn acts like the __init__ for non-pygame facing state. 138 # It could be used to override the initial image if necessary. 139 self.on_spawn() 140 141 if pos is not None: 142 self.pos = pos 143 144 if not hasattr(self, "pos"): 145 # Keep changing the position until the position no longer collides with any obstacle. 146 while True: 147 self.pos = random_pos(self._area, prng=self.shared.prng_move) 148 149 obstacle_hit = pg.sprite.spritecollideany( 150 self, # type: ignore[reportArgumentType] 151 self._obstacles, 152 pg.sprite.collide_mask, 153 ) 154 155 if not bool(obstacle_hit) and self._area.contains(self.rect): 156 break 157 158 def _get_image(self) -> Surface: 159 image = self._images[self._image_index] 160 161 if self.config.image_rotation: 162 angle = self.move.angle_to(Vector2((0, -1))) 163 164 return pg.transform.rotate(image, angle) 165 166 return image 167 168 @property 169 def image(self) -> Surface: 170 """The read-only image that's used for PyGame's rendering.""" 171 if self._image_cache is not None: 172 frame, image = self._image_cache 173 if frame == self.shared.counter: 174 return image 175 176 new_image = self._get_image() 177 self._image_cache = (self.shared.counter, new_image) 178 179 return new_image 180 181 @property 182 def center(self) -> tuple[int, int]: 183 """The read-only centre position of the agent.""" 184 return round_pos(self.pos) 185 186 @property 187 def rect(self) -> Rect: 188 """The read-only bounding-box that's used for PyGame's rendering.""" 189 rect = self.image.get_rect() 190 rect.center = self.center 191 192 return rect 193 194 @property 195 def mask(self) -> Mask: 196 """A read-only bit-mask of the image used for collision detection with obstacles and sites.""" 197 return pg.mask.from_surface(self.image) 198 199 def update(self) -> None: 200 """Run your own agent logic at every tick of the simulation. 201 202 Every frame of the simulation, this method is called automatically for every agent of the simulation. 203 To add your own logic, inherit the `Agent` class and override this method with your own. 204 """ 205 206 def obstacle_intersections( 207 self, 208 scale: float = 1, 209 ) -> Generator[Vector2]: 210 """Retrieve the centre coordinates of all obstacle intersections. 211 212 If you not only want to check for obstacle collision, 213 but also want to retrieve the coordinates of pixel groups 214 that are colliding with your agent, then `obstacle_intersections` is for you! 215 216 If you only have one obstacle in your environment, 217 but it doesn't have a fill, only a stroke, 218 then your agent could possibly be colliding with different areas of the stroke. 219 Therefore, this method checks the bitmasks of both the agent and the obstacle 220 to calculate the overlapping bitmask. 221 From this overlapping bitmask, 222 the centre coordinates of all groups of connected pixels are returned. 223 224 To emulate a bigger (or smaller) radius, 225 you can pass along the `scale` option. 226 A scale of 2 makes your agent twice as big, 227 but only for calculating the intersecting bitmasks. 228 """ 229 mask = pg.mask.from_surface(self.image) 230 231 # Scale the mask to the desired size 232 width, height = mask.get_size() 233 mask = mask.scale((width * scale, height * scale)) 234 235 # Align the mask to the centre position of the agent 236 rect = mask.get_rect() 237 rect.center = self.center 238 239 for sprite in self._obstacles.sprites(): 240 obstacle: _StaticSprite = sprite 241 242 # Calculate the mask offset 243 x = obstacle.rect.x - rect.x 244 y = obstacle.rect.y - rect.y 245 246 overlap = mask.overlap_mask(obstacle.mask, offset=(x, y)) 247 248 # For some reason PyGame has the wrong type hint here (single instead of list) 249 overlap_rects: list[pg.rect.Rect] = overlap.get_bounding_rects() # type: ignore[reportAssignmentType] 250 251 for overlap_rect in overlap_rects: 252 # Undo the offset 253 overlap_rect.x += rect.x 254 overlap_rect.y += rect.y 255 256 # Return the centre coordinates of the connected pixels 257 yield Vector2(overlap_rect.center) 258 259 def on_spawn(self) -> None: 260 """Run any code when the agent is spawned into the simulation. 261 262 This method is a replacement for `__init__`, which you should not overwrite directly. 263 Instead, you can make alterations to your Agent within this function instead. 264 265 You should override this method when inheriting Agent to add your own logic. 266 267 Some examples include: 268 - Changing the image or state of your Agent depending on its assigned identifier. 269 """ 270 271 def there_is_no_escape(self) -> bool: 272 """Pac-Man-style teleport the agent to the other side of the screen when it is outside of the playable area. 273 274 Examples 275 -------- 276 An agent that will always move to the right, until snapped back to reality. 277 278 ```python 279 class MyAgent(Agent): 280 def on_spawn(self): 281 self.move = Vector2((5, 0)) 282 283 def change_position(self): 284 self.there_is_no_escape() 285 self.pos += self.move 286 ``` 287 288 """ 289 changed = False 290 291 if self.pos.x < self._area.left: 292 changed = True 293 self.pos.x = self._area.right 294 295 if self.pos.x > self._area.right: 296 changed = True 297 self.pos.x = self._area.left 298 299 if self.pos.y < self._area.top: 300 changed = True 301 self.pos.y = self._area.bottom 302 303 if self.pos.y > self._area.bottom: 304 changed = True 305 self.pos.y = self._area.top 306 307 return changed 308 309 def change_position(self) -> None: 310 """Change the position of the agent. 311 312 The agent's new position is calculated as follows: 313 1. The agent checks whether it's outside of the visible screen area. 314 If this is the case, then the agent will be teleported to the other edge of the screen. 315 2. If the agent collides with any obstacles, then the agent will turn around 180 degrees. 316 3. If the agent has not collided with any obstacles, it will have the opportunity to slightly change its angle. 317 """ 318 if not self._moving: 319 return 320 321 changed = self.there_is_no_escape() 322 323 prng = self.shared.prng_move 324 325 # Always calculate the random angle so a seed could be used. 326 deg = prng.uniform(-30, 30) 327 328 # Only update angle if the agent was teleported to a different area of the simulation. 329 if changed: 330 self.move.rotate_ip(deg) 331 332 # Obstacle Avoidance 333 obstacle_hit = pg.sprite.spritecollideany( 334 self, # type: ignore[reportArgumentType] 335 self._obstacles, 336 pg.sprite.collide_mask, 337 ) 338 339 collision = bool(obstacle_hit) 340 341 # Reverse direction when colliding with an obstacle. 342 if collision and not self._still_stuck: 343 self.move.rotate_ip(180) 344 self._still_stuck = True 345 346 if not collision: 347 self._still_stuck = False 348 349 # Random opportunity to slightly change angle. 350 # Probabilities are pre-computed so a seed could be used. 351 should_change_angle = prng.random() 352 deg = prng.uniform(-10, 10) 353 354 # Only allow the angle opportunity to take place when no collisions have occured. 355 # This is done so an agent always turns 180 degrees. Any small change in the number of degrees 356 # allows the agent to possibly escape the obstacle. 357 if not collision and not self._still_stuck and should_change_angle < 0.25: # noqa: PLR2004 358 self.move.rotate_ip(deg) 359 360 # Actually update the position at last. 361 self.pos += self.move 362 363 def in_proximity_accuracy( 364 self, 365 ) -> Generator[tuple[Agent[ConfigClass], float]]: 366 """Retrieve other agents that are in the `vi.config.Schema.radius` of the current agent. 367 368 This proximity method calculates the distances between agents to determine whether 369 an agent is in the radius of the current agent. 370 371 To calculate the distances between agents, up to four chunks have to be retrieved from the Proximity Engine. 372 These extra lookups, in combination with the vector distance calculation, have a noticable impact on performance. 373 Note however that this performance impact is only noticable with >1000 agents. 374 375 If you want to speed up your simulation at the cost of some accuracy, 376 consider using the `in_proximity_performance` method instead. 377 378 This function doesn't return the agents as a `list` or as a `set`. 379 Instead, you are given a generator. 380 381 Examples 382 -------- 383 Count the number of agents that are in proximity 384 and change to image `1` if there is at least one agent nearby. 385 386 ```python 387 from vi.util import count 388 389 390 class MyAgent(Agent): 391 def update(self) -> None: 392 if count(self.in_proximity_accuracy()) >= 1: 393 self.change_image(1) 394 else: 395 self.change_image(0) 396 ``` 397 398 Kill the first `Human` agent that's in proximity. 399 400 ```python 401 class Zombie(Agent): 402 def update(self) -> None: 403 for agent, _ in self.in_proximity_accuracy(): 404 # Don't want to kill other zombies 405 if isinstance(Agent, Human): 406 agent.kill() 407 break 408 ``` 409 410 Calculate the average distance of agents that are in proximity. 411 412 ```python 413 from statistics import fmean 414 415 class Heimerdinger(Agent): 416 def update(self) -> None: 417 distances = [dist for _, dist in self.in_proximity_accuracy()] 418 dist_mean = fmean(distances) if len(distances) > 0 else 0 419 ``` 420 421 """ 422 return self.__simulation._proximity.in_proximity_accuracy(self) 423 424 def in_proximity_performance( 425 self, 426 ) -> Generator[Agent[ConfigClass]]: 427 """Retrieve other agents that are in the `vi.config.Schema.radius` of the current agent. 428 429 Unlike `in_proximity_accuracy`, this proximity method does not calculate the distances between agents. 430 Instead, it retrieves agents that are in the same chunk as the current agent, 431 irrespective of their position within the chunk. 432 433 If you find yourself limited by the performance of `in_proximity_accuracy`, 434 you can swap the function call for this one instead. 435 This performance method roughly doubles the frame rates of the simulation. 436 """ 437 return self.__simulation._proximity.in_proximity_performance(self) 438 439 def on_site(self) -> bool: 440 """Check whether the agent is currently on a site. 441 442 Examples 443 -------- 444 Stop the agent's movement when it reaches a site (think of a nice beach). 445 446 ```python 447 class TravellingAgent(Agent): 448 def update(self) -> None: 449 if self.on_site(): 450 # crave that star damage 451 self.freeze_movement() 452 ``` 453 454 """ 455 return self.on_site_id() is not None 456 457 def on_site_id(self) -> int | None: 458 """Get the identifier of the site the agent is currently on. 459 460 Examples 461 -------- 462 Stop the agent's movement when it reaches a site to inspect. 463 In addition, the current site identifier is saved to the DataFrame. 464 465 ```python 466 class SiteInspector(Agent): 467 def update(self) -> None: 468 site_id = self.on_site_id() 469 470 # Save the site id to the DataFrame 471 self.save_data("site", site_id) 472 473 # bool(site_id) would be inaccurate 474 # as a site_id of 0 will return False. 475 # Therefore, we check whether it is not None instead. 476 if site_id is not None: 477 # Inspect the site 478 self.freeze_movement() 479 ``` 480 481 """ 482 site: _StaticSprite | None = pg.sprite.spritecollideany( 483 self, # type: ignore[reportArgumentType] 484 self._sites, 485 pg.sprite.collide_mask, 486 ) 487 488 if site is not None: 489 return site.id 490 return None 491 492 def freeze_movement(self) -> None: 493 """Freeze the movement of the agent. The movement can be continued by calling `continue_movement`.""" 494 self._moving = False 495 496 def continue_movement(self) -> None: 497 """Continue the movement of the agent from before its movement was frozen.""" 498 self._moving = True 499 500 def change_image(self, index: int) -> None: 501 """Change the image of the agent. 502 503 If you want to change the agent's image to the second image in the images list, 504 then you can change the image to index 1: 505 506 ```python 507 self.change_image(1) 508 ``` 509 """ 510 self._image_index = index 511 512 def save_data(self, column: str, value: Any) -> None: # noqa: ANN401 513 """Add extra data to the simulation's metrics. 514 515 The following data is collected automatically: 516 - agent identifier 517 - current frame 518 - x and y coordinates 519 520 Examples 521 -------- 522 Saving the number of agents that are currently in proximity: 523 524 ```python 525 from vi.util import count 526 527 528 class MyAgent(Agent): 529 def update(self) -> None: 530 in_proximity = count(self.in_proximity_accuracy()) 531 532 self.save_data("in_proximity", in_proximity) 533 ``` 534 535 """ 536 self.__simulation._metrics._temporary_snapshots[column].append(value) 537 538 def _collect_replay_data(self) -> None: 539 """Add the minimum data needed for the replay simulation to the dataframe.""" 540 x, y = self.center 541 snapshots = self.__simulation._metrics._temporary_snapshots 542 543 snapshots["frame"].append(self.shared.counter) 544 snapshots["id"].append(self.id) 545 546 snapshots["x"].append(x) 547 snapshots["y"].append(y) 548 549 snapshots["image_index"].append(self._image_index) 550 551 if self.config.image_rotation: 552 angle = self.move.angle_to(Vector2((0, -1))) 553 snapshots["angle"].append(round(angle)) 554 555 def __copy__(self) -> Self: 556 """Create a copy of this agent and spawn it into the simulation. 557 558 Note that this only copies the `pos` and `move` vectors. 559 Any other attributes will be set to their defaults. 560 """ 561 cls = self.__class__ 562 agent = cls(self._images, self.__simulation) 563 564 # We want to make sure to copy the position and movement vectors. 565 # Otherwise, the original as well as the clone will continue sharing these vectors. 566 agent.pos = self.pos.copy() 567 agent.move = self.move.copy() 568 569 return agent 570 571 def reproduce(self) -> Self: 572 """Create a new agent and spawn it into the simulation. 573 574 All values will be reset to their defaults, 575 except for the agent's position and movement vector. 576 These will be cloned from the original agent. 577 """ 578 return copy(self) 579 580 def kill(self) -> None: 581 """Kill the agent. 582 583 While violence usually isn't the right option, 584 sometimes you just want to murder some innocent agents inside your little computer. 585 586 But fear not! 587 By *killing* the agent, all you're really doing is removing it from the simulation. 588 """ 589 super().kill() 590 591 def is_dead(self) -> bool: 592 """Is the agent dead? 593 594 Death occurs when `kill` is called. 595 """ 596 return not self.is_alive() 597 598 def is_alive(self) -> bool: 599 """Is the agent still alive? 600 601 Death occurs when `kill` is called. 602 """ 603 return super().alive()
The Agent class is home to Violet's various additions and is built on top of PyGame's Sprite class.
While you can simply add this Agent class to your simulations by calling batch_spawn_agents,
the agents won't actually do anything interesting.
Sure, they'll move around the screen very energetically, but they don't interact with each other.
That's where you come in!
By inheriting this Agent class you can make use of all its utilities,
while also programming the behaviour of your custom agent.
config: 'ConfigClass'
The config of the simulation that's shared with all agents.
The config can be overriden when inheriting the Agent class. However, the config must always:
- Inherit
Config - Be decorated by
@dataclass
move: pygame.math.Vector2
A two-dimensional vector representing the delta between the agent's current and next position.
Note that move isn't added to the agent's pos automatically.
Instead, you should manually add the move delta to pos, like so:
self.pos += self.move
The vector Vector2(2, 1) indicates that the agent will be moving 2 pixels along the x axis and 1 pixel along the y axis.
You can use the Vector2 class to calculate the vector's magnitude.
This property is also used to automatically rotate the agent's image
when vi.config.Schema.image_rotation is enabled.
image: pygame.surface.Surface
168 @property 169 def image(self) -> Surface: 170 """The read-only image that's used for PyGame's rendering.""" 171 if self._image_cache is not None: 172 frame, image = self._image_cache 173 if frame == self.shared.counter: 174 return image 175 176 new_image = self._get_image() 177 self._image_cache = (self.shared.counter, new_image) 178 179 return new_image
The read-only image that's used for PyGame's rendering.
center: tuple[int, int]
181 @property 182 def center(self) -> tuple[int, int]: 183 """The read-only centre position of the agent.""" 184 return round_pos(self.pos)
The read-only centre position of the agent.
rect: pygame.rect.Rect
186 @property 187 def rect(self) -> Rect: 188 """The read-only bounding-box that's used for PyGame's rendering.""" 189 rect = self.image.get_rect() 190 rect.center = self.center 191 192 return rect
The read-only bounding-box that's used for PyGame's rendering.
mask: pygame.mask.Mask
194 @property 195 def mask(self) -> Mask: 196 """A read-only bit-mask of the image used for collision detection with obstacles and sites.""" 197 return pg.mask.from_surface(self.image)
A read-only bit-mask of the image used for collision detection with obstacles and sites.
def
obstacle_intersections(self, scale: float = 1) -> Generator[pygame.math.Vector2]:
206 def obstacle_intersections( 207 self, 208 scale: float = 1, 209 ) -> Generator[Vector2]: 210 """Retrieve the centre coordinates of all obstacle intersections. 211 212 If you not only want to check for obstacle collision, 213 but also want to retrieve the coordinates of pixel groups 214 that are colliding with your agent, then `obstacle_intersections` is for you! 215 216 If you only have one obstacle in your environment, 217 but it doesn't have a fill, only a stroke, 218 then your agent could possibly be colliding with different areas of the stroke. 219 Therefore, this method checks the bitmasks of both the agent and the obstacle 220 to calculate the overlapping bitmask. 221 From this overlapping bitmask, 222 the centre coordinates of all groups of connected pixels are returned. 223 224 To emulate a bigger (or smaller) radius, 225 you can pass along the `scale` option. 226 A scale of 2 makes your agent twice as big, 227 but only for calculating the intersecting bitmasks. 228 """ 229 mask = pg.mask.from_surface(self.image) 230 231 # Scale the mask to the desired size 232 width, height = mask.get_size() 233 mask = mask.scale((width * scale, height * scale)) 234 235 # Align the mask to the centre position of the agent 236 rect = mask.get_rect() 237 rect.center = self.center 238 239 for sprite in self._obstacles.sprites(): 240 obstacle: _StaticSprite = sprite 241 242 # Calculate the mask offset 243 x = obstacle.rect.x - rect.x 244 y = obstacle.rect.y - rect.y 245 246 overlap = mask.overlap_mask(obstacle.mask, offset=(x, y)) 247 248 # For some reason PyGame has the wrong type hint here (single instead of list) 249 overlap_rects: list[pg.rect.Rect] = overlap.get_bounding_rects() # type: ignore[reportAssignmentType] 250 251 for overlap_rect in overlap_rects: 252 # Undo the offset 253 overlap_rect.x += rect.x 254 overlap_rect.y += rect.y 255 256 # Return the centre coordinates of the connected pixels 257 yield Vector2(overlap_rect.center)
Retrieve the centre coordinates of all obstacle intersections.
If you not only want to check for obstacle collision,
but also want to retrieve the coordinates of pixel groups
that are colliding with your agent, then obstacle_intersections is for you!
If you only have one obstacle in your environment, but it doesn't have a fill, only a stroke, then your agent could possibly be colliding with different areas of the stroke. Therefore, this method checks the bitmasks of both the agent and the obstacle to calculate the overlapping bitmask. From this overlapping bitmask, the centre coordinates of all groups of connected pixels are returned.
To emulate a bigger (or smaller) radius,
you can pass along the scale option.
A scale of 2 makes your agent twice as big,
but only for calculating the intersecting bitmasks.
def
on_spawn(self) -> None:
259 def on_spawn(self) -> None: 260 """Run any code when the agent is spawned into the simulation. 261 262 This method is a replacement for `__init__`, which you should not overwrite directly. 263 Instead, you can make alterations to your Agent within this function instead. 264 265 You should override this method when inheriting Agent to add your own logic. 266 267 Some examples include: 268 - Changing the image or state of your Agent depending on its assigned identifier. 269 """
Run any code when the agent is spawned into the simulation.
This method is a replacement for __init__, which you should not overwrite directly.
Instead, you can make alterations to your Agent within this function instead.
You should override this method when inheriting Agent to add your own logic.
Some examples include:
- Changing the image or state of your Agent depending on its assigned identifier.
def
there_is_no_escape(self) -> bool:
271 def there_is_no_escape(self) -> bool: 272 """Pac-Man-style teleport the agent to the other side of the screen when it is outside of the playable area. 273 274 Examples 275 -------- 276 An agent that will always move to the right, until snapped back to reality. 277 278 ```python 279 class MyAgent(Agent): 280 def on_spawn(self): 281 self.move = Vector2((5, 0)) 282 283 def change_position(self): 284 self.there_is_no_escape() 285 self.pos += self.move 286 ``` 287 288 """ 289 changed = False 290 291 if self.pos.x < self._area.left: 292 changed = True 293 self.pos.x = self._area.right 294 295 if self.pos.x > self._area.right: 296 changed = True 297 self.pos.x = self._area.left 298 299 if self.pos.y < self._area.top: 300 changed = True 301 self.pos.y = self._area.bottom 302 303 if self.pos.y > self._area.bottom: 304 changed = True 305 self.pos.y = self._area.top 306 307 return changed
Pac-Man-style teleport the agent to the other side of the screen when it is outside of the playable area.
Examples
An agent that will always move to the right, until snapped back to reality.
class MyAgent(Agent):
def on_spawn(self):
self.move = Vector2((5, 0))
def change_position(self):
self.there_is_no_escape()
self.pos += self.move
def
change_position(self) -> None:
309 def change_position(self) -> None: 310 """Change the position of the agent. 311 312 The agent's new position is calculated as follows: 313 1. The agent checks whether it's outside of the visible screen area. 314 If this is the case, then the agent will be teleported to the other edge of the screen. 315 2. If the agent collides with any obstacles, then the agent will turn around 180 degrees. 316 3. If the agent has not collided with any obstacles, it will have the opportunity to slightly change its angle. 317 """ 318 if not self._moving: 319 return 320 321 changed = self.there_is_no_escape() 322 323 prng = self.shared.prng_move 324 325 # Always calculate the random angle so a seed could be used. 326 deg = prng.uniform(-30, 30) 327 328 # Only update angle if the agent was teleported to a different area of the simulation. 329 if changed: 330 self.move.rotate_ip(deg) 331 332 # Obstacle Avoidance 333 obstacle_hit = pg.sprite.spritecollideany( 334 self, # type: ignore[reportArgumentType] 335 self._obstacles, 336 pg.sprite.collide_mask, 337 ) 338 339 collision = bool(obstacle_hit) 340 341 # Reverse direction when colliding with an obstacle. 342 if collision and not self._still_stuck: 343 self.move.rotate_ip(180) 344 self._still_stuck = True 345 346 if not collision: 347 self._still_stuck = False 348 349 # Random opportunity to slightly change angle. 350 # Probabilities are pre-computed so a seed could be used. 351 should_change_angle = prng.random() 352 deg = prng.uniform(-10, 10) 353 354 # Only allow the angle opportunity to take place when no collisions have occured. 355 # This is done so an agent always turns 180 degrees. Any small change in the number of degrees 356 # allows the agent to possibly escape the obstacle. 357 if not collision and not self._still_stuck and should_change_angle < 0.25: # noqa: PLR2004 358 self.move.rotate_ip(deg) 359 360 # Actually update the position at last. 361 self.pos += self.move
Change the position of the agent.
The agent's new position is calculated as follows:
- The agent checks whether it's outside of the visible screen area. If this is the case, then the agent will be teleported to the other edge of the screen.
- If the agent collides with any obstacles, then the agent will turn around 180 degrees.
- If the agent has not collided with any obstacles, it will have the opportunity to slightly change its angle.
def
in_proximity_accuracy(self) -> 'Generator[tuple[Agent[ConfigClass], float]]':
363 def in_proximity_accuracy( 364 self, 365 ) -> Generator[tuple[Agent[ConfigClass], float]]: 366 """Retrieve other agents that are in the `vi.config.Schema.radius` of the current agent. 367 368 This proximity method calculates the distances between agents to determine whether 369 an agent is in the radius of the current agent. 370 371 To calculate the distances between agents, up to four chunks have to be retrieved from the Proximity Engine. 372 These extra lookups, in combination with the vector distance calculation, have a noticable impact on performance. 373 Note however that this performance impact is only noticable with >1000 agents. 374 375 If you want to speed up your simulation at the cost of some accuracy, 376 consider using the `in_proximity_performance` method instead. 377 378 This function doesn't return the agents as a `list` or as a `set`. 379 Instead, you are given a generator. 380 381 Examples 382 -------- 383 Count the number of agents that are in proximity 384 and change to image `1` if there is at least one agent nearby. 385 386 ```python 387 from vi.util import count 388 389 390 class MyAgent(Agent): 391 def update(self) -> None: 392 if count(self.in_proximity_accuracy()) >= 1: 393 self.change_image(1) 394 else: 395 self.change_image(0) 396 ``` 397 398 Kill the first `Human` agent that's in proximity. 399 400 ```python 401 class Zombie(Agent): 402 def update(self) -> None: 403 for agent, _ in self.in_proximity_accuracy(): 404 # Don't want to kill other zombies 405 if isinstance(Agent, Human): 406 agent.kill() 407 break 408 ``` 409 410 Calculate the average distance of agents that are in proximity. 411 412 ```python 413 from statistics import fmean 414 415 class Heimerdinger(Agent): 416 def update(self) -> None: 417 distances = [dist for _, dist in self.in_proximity_accuracy()] 418 dist_mean = fmean(distances) if len(distances) > 0 else 0 419 ``` 420 421 """ 422 return self.__simulation._proximity.in_proximity_accuracy(self)
Retrieve other agents that are in the vi.config.Schema.radius of the current agent.
This proximity method calculates the distances between agents to determine whether an agent is in the radius of the current agent.
To calculate the distances between agents, up to four chunks have to be retrieved from the Proximity Engine. These extra lookups, in combination with the vector distance calculation, have a noticable impact on performance. Note however that this performance impact is only noticable with >1000 agents.
If you want to speed up your simulation at the cost of some accuracy,
consider using the in_proximity_performance method instead.
This function doesn't return the agents as a list or as a set.
Instead, you are given a generator.
Examples
Count the number of agents that are in proximity
and change to image 1 if there is at least one agent nearby.
from vi.util import count
class MyAgent(Agent):
def update(self) -> None:
if count(self.in_proximity_accuracy()) >= 1:
self.change_image(1)
else:
self.change_image(0)
Kill the first Human agent that's in proximity.
class Zombie(Agent):
def update(self) -> None:
for agent, _ in self.in_proximity_accuracy():
# Don't want to kill other zombies
if isinstance(Agent, Human):
agent.kill()
break
Calculate the average distance of agents that are in proximity.
from statistics import fmean
class Heimerdinger(Agent):
def update(self) -> None:
distances = [dist for _, dist in self.in_proximity_accuracy()]
dist_mean = fmean(distances) if len(distances) > 0 else 0
def
in_proximity_performance(self) -> 'Generator[Agent[ConfigClass]]':
424 def in_proximity_performance( 425 self, 426 ) -> Generator[Agent[ConfigClass]]: 427 """Retrieve other agents that are in the `vi.config.Schema.radius` of the current agent. 428 429 Unlike `in_proximity_accuracy`, this proximity method does not calculate the distances between agents. 430 Instead, it retrieves agents that are in the same chunk as the current agent, 431 irrespective of their position within the chunk. 432 433 If you find yourself limited by the performance of `in_proximity_accuracy`, 434 you can swap the function call for this one instead. 435 This performance method roughly doubles the frame rates of the simulation. 436 """ 437 return self.__simulation._proximity.in_proximity_performance(self)
Retrieve other agents that are in the vi.config.Schema.radius of the current agent.
Unlike in_proximity_accuracy, this proximity method does not calculate the distances between agents.
Instead, it retrieves agents that are in the same chunk as the current agent,
irrespective of their position within the chunk.
If you find yourself limited by the performance of in_proximity_accuracy,
you can swap the function call for this one instead.
This performance method roughly doubles the frame rates of the simulation.
def
on_site(self) -> bool:
439 def on_site(self) -> bool: 440 """Check whether the agent is currently on a site. 441 442 Examples 443 -------- 444 Stop the agent's movement when it reaches a site (think of a nice beach). 445 446 ```python 447 class TravellingAgent(Agent): 448 def update(self) -> None: 449 if self.on_site(): 450 # crave that star damage 451 self.freeze_movement() 452 ``` 453 454 """ 455 return self.on_site_id() is not None
Check whether the agent is currently on a site.
Examples
Stop the agent's movement when it reaches a site (think of a nice beach).
class TravellingAgent(Agent):
def update(self) -> None:
if self.on_site():
# crave that star damage
self.freeze_movement()
def
on_site_id(self) -> int | None:
457 def on_site_id(self) -> int | None: 458 """Get the identifier of the site the agent is currently on. 459 460 Examples 461 -------- 462 Stop the agent's movement when it reaches a site to inspect. 463 In addition, the current site identifier is saved to the DataFrame. 464 465 ```python 466 class SiteInspector(Agent): 467 def update(self) -> None: 468 site_id = self.on_site_id() 469 470 # Save the site id to the DataFrame 471 self.save_data("site", site_id) 472 473 # bool(site_id) would be inaccurate 474 # as a site_id of 0 will return False. 475 # Therefore, we check whether it is not None instead. 476 if site_id is not None: 477 # Inspect the site 478 self.freeze_movement() 479 ``` 480 481 """ 482 site: _StaticSprite | None = pg.sprite.spritecollideany( 483 self, # type: ignore[reportArgumentType] 484 self._sites, 485 pg.sprite.collide_mask, 486 ) 487 488 if site is not None: 489 return site.id 490 return None
Get the identifier of the site the agent is currently on.
Examples
Stop the agent's movement when it reaches a site to inspect. In addition, the current site identifier is saved to the DataFrame.
class SiteInspector(Agent):
def update(self) -> None:
site_id = self.on_site_id()
# Save the site id to the DataFrame
self.save_data("site", site_id)
# bool(site_id) would be inaccurate
# as a site_id of 0 will return False.
# Therefore, we check whether it is not None instead.
if site_id is not None:
# Inspect the site
self.freeze_movement()
def
freeze_movement(self) -> None:
492 def freeze_movement(self) -> None: 493 """Freeze the movement of the agent. The movement can be continued by calling `continue_movement`.""" 494 self._moving = False
Freeze the movement of the agent. The movement can be continued by calling continue_movement.
def
continue_movement(self) -> None:
496 def continue_movement(self) -> None: 497 """Continue the movement of the agent from before its movement was frozen.""" 498 self._moving = True
Continue the movement of the agent from before its movement was frozen.
def
change_image(self, index: int) -> None:
500 def change_image(self, index: int) -> None: 501 """Change the image of the agent. 502 503 If you want to change the agent's image to the second image in the images list, 504 then you can change the image to index 1: 505 506 ```python 507 self.change_image(1) 508 ``` 509 """ 510 self._image_index = index
Change the image of the agent.
If you want to change the agent's image to the second image in the images list, then you can change the image to index 1:
self.change_image(1)
def
save_data(self, column: str, value: Any) -> None:
512 def save_data(self, column: str, value: Any) -> None: # noqa: ANN401 513 """Add extra data to the simulation's metrics. 514 515 The following data is collected automatically: 516 - agent identifier 517 - current frame 518 - x and y coordinates 519 520 Examples 521 -------- 522 Saving the number of agents that are currently in proximity: 523 524 ```python 525 from vi.util import count 526 527 528 class MyAgent(Agent): 529 def update(self) -> None: 530 in_proximity = count(self.in_proximity_accuracy()) 531 532 self.save_data("in_proximity", in_proximity) 533 ``` 534 535 """ 536 self.__simulation._metrics._temporary_snapshots[column].append(value)
Add extra data to the simulation's metrics.
The following data is collected automatically:
- agent identifier
- current frame
- x and y coordinates
Examples
Saving the number of agents that are currently in proximity:
from vi.util import count
class MyAgent(Agent):
def update(self) -> None:
in_proximity = count(self.in_proximity_accuracy())
self.save_data("in_proximity", in_proximity)
def
reproduce(self) -> Self:
571 def reproduce(self) -> Self: 572 """Create a new agent and spawn it into the simulation. 573 574 All values will be reset to their defaults, 575 except for the agent's position and movement vector. 576 These will be cloned from the original agent. 577 """ 578 return copy(self)
Create a new agent and spawn it into the simulation.
All values will be reset to their defaults, except for the agent's position and movement vector. These will be cloned from the original agent.
Config =
vi.config.Schema[Mono[int], Mono[float]]
class
HeadlessSimulation(typing.Generic[ConfigClass]):
89class HeadlessSimulation[ConfigClass: Config]: 90 """The Headless Mode equivalent of `Simulation`. 91 92 Headless Mode removes all the rendering logic from the simulation 93 to not only remove the annoying simulation window from popping up every time, 94 but to also speed up your simulation when it's GPU bound. 95 96 Combining Headless Mode with `vi.config.Matrix` and Python's [multiprocessing](https://docs.python.org/3/library/multiprocessing.html) opens a realm of new possibilities. 97 Vi's Matrix is `vi.config.Config` on steroids. 98 It allows you to pass lists of values instead of single values on supported parameters, 99 to then effortlessly combine each unique combination of values into its own `vi.config.Config`. 100 When combined with [multiprocessing](https://docs.python.org/3/library/multiprocessing.html), 101 we can run multiple configs in parallel. 102 103 ```python 104 from multiprocessing import Pool 105 106 import polars as pl 107 108 from vi import Agent, Config, HeadlessSimulation, Matrix 109 110 111 def run_simulation(config: Config) -> pl.DataFrame: 112 return ( 113 HeadlessSimulation(config) 114 .batch_spawn_agents(100, Agent, ["examples/images/white.png"]) 115 .run() 116 .snapshots 117 ) 118 119 120 if __name__ == "__main__": 121 # We create a threadpool to run our simulations in parallel 122 with Pool() as p: 123 # The matrix will create four unique configs 124 matrix = Matrix(radius=[25, 50], seed=[1, 2]) 125 126 # Create unique combinations of matrix values 127 configs = matrix.to_configs(Config) 128 129 # Combine our individual DataFrames into one big DataFrame 130 df = pl.concat(p.map(run_simulation, configs)) 131 132 print(df) 133 ``` 134 """ 135 136 shared: Shared 137 """Attributes that are shared between the simulation and all agents.""" 138 139 _running: bool = False 140 """The simulation keeps running as long as running is True.""" 141 142 _area: pg.rect.Rect 143 144 # Sprite Groups 145 _all: pg.sprite.Group[Any] 146 _agents: pg.sprite.Group[Any] 147 _obstacles: pg.sprite.Group[Any] 148 _sites: pg.sprite.Group[Any] 149 150 _next_agent_id: int = 0 151 """The agent identifier to be given next.""" 152 153 _next_obstacle_id: int = 0 154 """The obstacle identifier to be given next.""" 155 156 _next_site_id: int = 0 157 """The site identifier to be given next.""" 158 159 # Proximity 160 _proximity: ProximityEngine[ConfigClass] 161 162 # Config that's passed on to agents as well 163 config: ConfigClass 164 """The config of the simulation that's shared with all agents. 165 166 The config can be overriden when inheriting the Simulation class. 167 However, the config must always: 168 169 1. Inherit `Config` 170 2. Be decorated by `@serde` 171 """ 172 173 _metrics: Metrics 174 """A collection of all the Snapshots that have been created in the simulation. 175 176 Each agent produces a Snapshot at every frame in the simulation. 177 """ 178 179 def __init__(self, config: ConfigClass) -> None: 180 self.config = config 181 self._metrics = Metrics() 182 183 # Initiate the seed as early as possible. 184 random.seed(self.config.seed) 185 186 # Using a custom generator for agent movement 187 prng_move = random.Random() # noqa: S311 188 prng_move.seed(self.config.seed) 189 190 self.shared = Shared(prng_move=prng_move) 191 192 width, height = self.config.window.as_tuple() 193 self._area = pg.rect.Rect(0, 0, width, height) 194 195 # Create sprite groups 196 self._all = pg.sprite.Group() 197 self._agents = pg.sprite.Group() 198 self._obstacles = pg.sprite.Group() 199 self._sites = pg.sprite.Group() 200 201 # Proximity! 202 self._proximity = ProximityEngine[ConfigClass](self._agents, self.config.radius) 203 204 def batch_spawn_agents( 205 self, 206 count: int, 207 agent_class: type[Agent[ConfigClass]], 208 images: list[str], 209 ) -> Self: 210 """Spawn multiple agents into the simulation. 211 212 Examples 213 -------- 214 Spawn 100 `vi.agent.Agent`'s into the simulation with `examples/images/white.png` as image. 215 216 ```python 217 ( 218 Simulation(Config()) 219 .batch_spawn_agents(100, Agent, ["examples/images/white.png"]) 220 .run() 221 ) 222 ``` 223 224 """ 225 # Load images once so the files don't have to be read multiple times. 226 loaded_images = self._load_images(images) 227 228 for _ in range(count): 229 agent_class(images=loaded_images, simulation=self) 230 231 return self 232 233 def spawn_agent( 234 self, 235 agent_class: type[Agent[ConfigClass]], 236 images: list[str], 237 ) -> Self: 238 """Spawn one agent into the simulation. 239 240 While you can run `spawn_agent` in a for-loop, 241 you probably want to call `batch_spawn_agents` instead 242 as `batch_spawn_agents` optimises the image loading process. 243 244 Examples 245 -------- 246 Spawn a single `vi.agent.Agent` into the simulation with `examples/images/white.png` as image: 247 248 ```python 249 ( 250 Simulation(Config()) 251 .spawn_agent(Agent, ["examples/images/white.png"]) 252 .run() 253 ) 254 ``` 255 256 """ 257 agent_class(images=self._load_images(images), simulation=self) 258 259 return self 260 261 def spawn_obstacle(self, image_path: str, x: int, y: int) -> Self: 262 """Spawn one obstacle into the simulation. The given coordinates will be the centre of the obstacle. 263 264 When agents collide with an obstacle, they will make a 180 degree turn. 265 266 Examples 267 -------- 268 Spawn a single obstacle into the simulation with `examples/images/bubble-full.png` as image. 269 In addition, we place the obstacle in the centre of our window. 270 271 ```python 272 config = Config() 273 x, y = config.window.as_tuple() 274 275 ( 276 Simulation(config) 277 .spawn_obstacle("examples/images/bubble-full.png", x // 2, y // 2) 278 .run() 279 ) 280 ``` 281 282 """ 283 _StaticSprite( 284 containers=[self._all, self._obstacles], 285 id=self._obstacle_id(), 286 image=self._load_image(image_path), 287 pos=Vector2((x, y)), 288 ) 289 290 return self 291 292 def spawn_site(self, image_path: str, x: int, y: int) -> Self: 293 """Spawn one site into the simulation. The given coordinates will be the centre of the site. 294 295 Examples 296 -------- 297 Spawn a single site into the simulation with `examples/images/site.png` as image. 298 In addition, we give specific coordinates where the site should be placed. 299 300 ```python 301 ( 302 Simulation(Config()) 303 .spawn_site("examples/images/site.png", x=375, y=375) 304 .run() 305 ) 306 ``` 307 308 """ 309 _StaticSprite( 310 containers=[self._all, self._sites], 311 id=self._site_id(), 312 image=self._load_image(image_path), 313 pos=Vector2((x, y)), 314 ) 315 316 return self 317 318 def run(self) -> Metrics: 319 """Run the simulation until it's ended by closing the window or when the `vi.config.Schema.duration` has elapsed.""" 320 self._running = True 321 322 while self._running: 323 self.tick() 324 325 return self._metrics 326 327 def before_update(self) -> None: 328 """Run any code before the agents are updated in every tick. 329 330 You should override this method when inheriting Simulation to add your own logic. 331 332 Some examples include: 333 - Processing events from PyGame's event queue. 334 """ 335 336 def after_update(self) -> None: ... 337 338 def tick(self) -> None: 339 """Advance the simulation with one tick.""" 340 self.before_update() 341 342 # Update the position of all agents 343 self.__update_positions() 344 345 # If the radius was changed by an event, 346 # also update the radius in the proximity engine 347 self._proximity._set_radius(self.config.radius) 348 349 # Calculate proximity chunks 350 self._proximity.update() 351 352 # Save the replay data of all agents 353 self.__collect_replay_data() 354 355 # Update all agents 356 self._all.update() 357 358 # Merge the collected snapshots into the dataframe. 359 self._metrics._merge() 360 361 self.after_update() 362 363 # If we've reached the duration of the simulation, then stop the simulation. 364 if self.config.duration > 0 and self.shared.counter == self.config.duration: 365 self.stop() 366 return 367 368 self.shared.counter += 1 369 370 def stop(self) -> None: 371 """Stop the simulation. 372 373 The simulation isn't stopped directly. 374 Instead, the current tick is completed, after which the simulation will end. 375 """ 376 self._running = False 377 378 def __collect_replay_data(self) -> None: 379 """Collect the replay data for all agents.""" 380 for sprite in self._agents: 381 agent: Agent = sprite 382 agent._collect_replay_data() 383 384 def __update_positions(self) -> None: 385 """Update the position of all agents.""" 386 for sprite in self._agents.sprites(): 387 agent: Agent = sprite 388 agent.change_position() 389 390 def _load_image(self, path: str) -> pg.surface.Surface: 391 return pg.image.load(path) 392 393 def _load_images(self, images: list[str]) -> list[pg.surface.Surface]: 394 return [self._load_image(path) for path in images] 395 396 def _agent_id(self) -> int: 397 agent_id = self._next_agent_id 398 self._next_agent_id += 1 399 400 return agent_id 401 402 def _obstacle_id(self) -> int: 403 obstacle_id = self._next_obstacle_id 404 self._next_obstacle_id += 1 405 406 return obstacle_id 407 408 def _site_id(self) -> int: 409 site_id = self._next_site_id 410 self._next_site_id += 1 411 412 return site_id
The Headless Mode equivalent of Simulation.
Headless Mode removes all the rendering logic from the simulation to not only remove the annoying simulation window from popping up every time, but to also speed up your simulation when it's GPU bound.
Combining Headless Mode with vi.config.Matrix and Python's multiprocessing opens a realm of new possibilities.
Vi's Matrix is vi.config.Config on steroids.
It allows you to pass lists of values instead of single values on supported parameters,
to then effortlessly combine each unique combination of values into its own vi.config.Config.
When combined with multiprocessing,
we can run multiple configs in parallel.
from multiprocessing import Pool
import polars as pl
from vi import Agent, Config, HeadlessSimulation, Matrix
def run_simulation(config: Config) -> pl.DataFrame:
return (
HeadlessSimulation(config)
.batch_spawn_agents(100, Agent, ["examples/images/white.png"])
.run()
.snapshots
)
if __name__ == "__main__":
# We create a threadpool to run our simulations in parallel
with Pool() as p:
# The matrix will create four unique configs
matrix = Matrix(radius=[25, 50], seed=[1, 2])
# Create unique combinations of matrix values
configs = matrix.to_configs(Config)
# Combine our individual DataFrames into one big DataFrame
df = pl.concat(p.map(run_simulation, configs))
print(df)
HeadlessSimulation(config: 'ConfigClass')
179 def __init__(self, config: ConfigClass) -> None: 180 self.config = config 181 self._metrics = Metrics() 182 183 # Initiate the seed as early as possible. 184 random.seed(self.config.seed) 185 186 # Using a custom generator for agent movement 187 prng_move = random.Random() # noqa: S311 188 prng_move.seed(self.config.seed) 189 190 self.shared = Shared(prng_move=prng_move) 191 192 width, height = self.config.window.as_tuple() 193 self._area = pg.rect.Rect(0, 0, width, height) 194 195 # Create sprite groups 196 self._all = pg.sprite.Group() 197 self._agents = pg.sprite.Group() 198 self._obstacles = pg.sprite.Group() 199 self._sites = pg.sprite.Group() 200 201 # Proximity! 202 self._proximity = ProximityEngine[ConfigClass](self._agents, self.config.radius)
config: 'ConfigClass'
The config of the simulation that's shared with all agents.
The config can be overriden when inheriting the Simulation class. However, the config must always:
- Inherit
Config - Be decorated by
@serde
def
batch_spawn_agents( self, count: int, agent_class: 'type[Agent[ConfigClass]]', images: list[str]) -> Self:
204 def batch_spawn_agents( 205 self, 206 count: int, 207 agent_class: type[Agent[ConfigClass]], 208 images: list[str], 209 ) -> Self: 210 """Spawn multiple agents into the simulation. 211 212 Examples 213 -------- 214 Spawn 100 `vi.agent.Agent`'s into the simulation with `examples/images/white.png` as image. 215 216 ```python 217 ( 218 Simulation(Config()) 219 .batch_spawn_agents(100, Agent, ["examples/images/white.png"]) 220 .run() 221 ) 222 ``` 223 224 """ 225 # Load images once so the files don't have to be read multiple times. 226 loaded_images = self._load_images(images) 227 228 for _ in range(count): 229 agent_class(images=loaded_images, simulation=self) 230 231 return self
Spawn multiple agents into the simulation.
Examples
Spawn 100 vi.agent.Agent's into the simulation with examples/images/white.png as image.
(
Simulation(Config())
.batch_spawn_agents(100, Agent, ["examples/images/white.png"])
.run()
)
def
spawn_agent(self, agent_class: 'type[Agent[ConfigClass]]', images: list[str]) -> Self:
233 def spawn_agent( 234 self, 235 agent_class: type[Agent[ConfigClass]], 236 images: list[str], 237 ) -> Self: 238 """Spawn one agent into the simulation. 239 240 While you can run `spawn_agent` in a for-loop, 241 you probably want to call `batch_spawn_agents` instead 242 as `batch_spawn_agents` optimises the image loading process. 243 244 Examples 245 -------- 246 Spawn a single `vi.agent.Agent` into the simulation with `examples/images/white.png` as image: 247 248 ```python 249 ( 250 Simulation(Config()) 251 .spawn_agent(Agent, ["examples/images/white.png"]) 252 .run() 253 ) 254 ``` 255 256 """ 257 agent_class(images=self._load_images(images), simulation=self) 258 259 return self
Spawn one agent into the simulation.
While you can run spawn_agent in a for-loop,
you probably want to call batch_spawn_agents instead
as batch_spawn_agents optimises the image loading process.
Examples
Spawn a single vi.agent.Agent into the simulation with examples/images/white.png as image:
(
Simulation(Config())
.spawn_agent(Agent, ["examples/images/white.png"])
.run()
)
def
spawn_obstacle(self, image_path: str, x: int, y: int) -> Self:
261 def spawn_obstacle(self, image_path: str, x: int, y: int) -> Self: 262 """Spawn one obstacle into the simulation. The given coordinates will be the centre of the obstacle. 263 264 When agents collide with an obstacle, they will make a 180 degree turn. 265 266 Examples 267 -------- 268 Spawn a single obstacle into the simulation with `examples/images/bubble-full.png` as image. 269 In addition, we place the obstacle in the centre of our window. 270 271 ```python 272 config = Config() 273 x, y = config.window.as_tuple() 274 275 ( 276 Simulation(config) 277 .spawn_obstacle("examples/images/bubble-full.png", x // 2, y // 2) 278 .run() 279 ) 280 ``` 281 282 """ 283 _StaticSprite( 284 containers=[self._all, self._obstacles], 285 id=self._obstacle_id(), 286 image=self._load_image(image_path), 287 pos=Vector2((x, y)), 288 ) 289 290 return self
Spawn one obstacle into the simulation. The given coordinates will be the centre of the obstacle.
When agents collide with an obstacle, they will make a 180 degree turn.
Examples
Spawn a single obstacle into the simulation with examples/images/bubble-full.png as image.
In addition, we place the obstacle in the centre of our window.
config = Config()
x, y = config.window.as_tuple()
(
Simulation(config)
.spawn_obstacle("examples/images/bubble-full.png", x // 2, y // 2)
.run()
)
def
spawn_site(self, image_path: str, x: int, y: int) -> Self:
292 def spawn_site(self, image_path: str, x: int, y: int) -> Self: 293 """Spawn one site into the simulation. The given coordinates will be the centre of the site. 294 295 Examples 296 -------- 297 Spawn a single site into the simulation with `examples/images/site.png` as image. 298 In addition, we give specific coordinates where the site should be placed. 299 300 ```python 301 ( 302 Simulation(Config()) 303 .spawn_site("examples/images/site.png", x=375, y=375) 304 .run() 305 ) 306 ``` 307 308 """ 309 _StaticSprite( 310 containers=[self._all, self._sites], 311 id=self._site_id(), 312 image=self._load_image(image_path), 313 pos=Vector2((x, y)), 314 ) 315 316 return self
Spawn one site into the simulation. The given coordinates will be the centre of the site.
Examples
Spawn a single site into the simulation with examples/images/site.png as image.
In addition, we give specific coordinates where the site should be placed.
(
Simulation(Config())
.spawn_site("examples/images/site.png", x=375, y=375)
.run()
)
318 def run(self) -> Metrics: 319 """Run the simulation until it's ended by closing the window or when the `vi.config.Schema.duration` has elapsed.""" 320 self._running = True 321 322 while self._running: 323 self.tick() 324 325 return self._metrics
Run the simulation until it's ended by closing the window or when the vi.config.Schema.duration has elapsed.
def
before_update(self) -> None:
327 def before_update(self) -> None: 328 """Run any code before the agents are updated in every tick. 329 330 You should override this method when inheriting Simulation to add your own logic. 331 332 Some examples include: 333 - Processing events from PyGame's event queue. 334 """
Run any code before the agents are updated in every tick.
You should override this method when inheriting Simulation to add your own logic.
Some examples include:
- Processing events from PyGame's event queue.
def
tick(self) -> None:
338 def tick(self) -> None: 339 """Advance the simulation with one tick.""" 340 self.before_update() 341 342 # Update the position of all agents 343 self.__update_positions() 344 345 # If the radius was changed by an event, 346 # also update the radius in the proximity engine 347 self._proximity._set_radius(self.config.radius) 348 349 # Calculate proximity chunks 350 self._proximity.update() 351 352 # Save the replay data of all agents 353 self.__collect_replay_data() 354 355 # Update all agents 356 self._all.update() 357 358 # Merge the collected snapshots into the dataframe. 359 self._metrics._merge() 360 361 self.after_update() 362 363 # If we've reached the duration of the simulation, then stop the simulation. 364 if self.config.duration > 0 and self.shared.counter == self.config.duration: 365 self.stop() 366 return 367 368 self.shared.counter += 1
Advance the simulation with one tick.
def
stop(self) -> None:
370 def stop(self) -> None: 371 """Stop the simulation. 372 373 The simulation isn't stopped directly. 374 Instead, the current tick is completed, after which the simulation will end. 375 """ 376 self._running = False
Stop the simulation.
The simulation isn't stopped directly. Instead, the current tick is completed, after which the simulation will end.
Matrix =
vi.config.Schema[Poly[int], Poly[float]]
415class Simulation[ConfigClass: Config](HeadlessSimulation[ConfigClass]): 416 """Offers the same functionality as `HeadlessSimulation`, but adds logic to automatically draw all agents, obstacles and sites to your screen. 417 418 If a custom config isn't provided when creating the simulation, the default values of `Config` will be used instead. 419 """ 420 421 _background: pg.surface.Surface 422 _clock: pg.time.Clock 423 _screen: pg.surface.Surface 424 425 def __init__(self, config: ConfigClass) -> None: 426 super().__init__(config) 427 428 pg.display.init() 429 pg.display.set_caption("Violet") 430 431 size = self.config.window.as_tuple() 432 self._screen = pg.display.set_mode(size) 433 434 # Initialise background 435 self._background = pg.surface.Surface(size).convert() 436 self._background.fill((0, 0, 0)) 437 438 # Show background immediately (before spawning agents) 439 self._screen.blit(self._background, (0, 0)) 440 pg.display.flip() 441 442 # Initialise the clock. Used to cap FPS. 443 self._clock = pg.time.Clock() 444 445 def before_update(self) -> None: 446 rebound: list[Event] = [] 447 for event in pg.event.get(eventtype=[pg.QUIT, pg.KEYDOWN]): 448 if event.type == pg.QUIT: 449 self.stop() 450 elif event.type == pg.KEYDOWN: 451 if event.key == pg.K_HOME: 452 self.config.radius += 1 453 elif event.key == pg.K_END: 454 self.config.radius -= 1 455 else: 456 # If a different key was pressed, then we want to re-emit the vent 457 # so other code can handle it. 458 rebound.append(event) 459 460 for event in rebound: 461 pg.event.post(event) 462 463 # Clear the screen before the update so agents can draw stuff themselves too! 464 self._all.clear(self._screen, self._background) 465 self._screen.blit(self._background, (0, 0)) 466 467 def after_update(self) -> None: 468 # Draw everything to the screen 469 self._all.draw(self._screen) 470 471 if self.config.visualise_chunks: 472 self.__visualise_chunks() 473 474 # Update the screen with the new image 475 pg.display.flip() 476 477 self._clock.tick(self.config.fps_limit) 478 479 current_fps = self._clock.get_fps() 480 if current_fps > 0: 481 self._metrics.fps._push(current_fps) 482 483 if self.config.print_fps: 484 print(f"FPS: {current_fps:.1f}") # noqa: T201 485 486 def __visualise_chunks(self) -> None: 487 """Visualise the proximity chunks by drawing their borders.""" 488 colour = pg.Color(255, 255, 255, 122) 489 chunk_size = self._proximity.chunk_size 490 491 width, height = self.config.window.as_tuple() 492 493 for x in range(chunk_size, width, chunk_size): 494 vline(self._screen, x, 0, height, colour) 495 496 for y in range(chunk_size, height, chunk_size): 497 hline(self._screen, 0, width, y, colour) 498 499 def _load_image(self, path: str) -> pg.surface.Surface: 500 return super()._load_image(path).convert_alpha()
Offers the same functionality as HeadlessSimulation, but adds logic to automatically draw all agents, obstacles and sites to your screen.
If a custom config isn't provided when creating the simulation, the default values of Config will be used instead.
Simulation(config: 'ConfigClass')
425 def __init__(self, config: ConfigClass) -> None: 426 super().__init__(config) 427 428 pg.display.init() 429 pg.display.set_caption("Violet") 430 431 size = self.config.window.as_tuple() 432 self._screen = pg.display.set_mode(size) 433 434 # Initialise background 435 self._background = pg.surface.Surface(size).convert() 436 self._background.fill((0, 0, 0)) 437 438 # Show background immediately (before spawning agents) 439 self._screen.blit(self._background, (0, 0)) 440 pg.display.flip() 441 442 # Initialise the clock. Used to cap FPS. 443 self._clock = pg.time.Clock()
def
before_update(self) -> None:
445 def before_update(self) -> None: 446 rebound: list[Event] = [] 447 for event in pg.event.get(eventtype=[pg.QUIT, pg.KEYDOWN]): 448 if event.type == pg.QUIT: 449 self.stop() 450 elif event.type == pg.KEYDOWN: 451 if event.key == pg.K_HOME: 452 self.config.radius += 1 453 elif event.key == pg.K_END: 454 self.config.radius -= 1 455 else: 456 # If a different key was pressed, then we want to re-emit the vent 457 # so other code can handle it. 458 rebound.append(event) 459 460 for event in rebound: 461 pg.event.post(event) 462 463 # Clear the screen before the update so agents can draw stuff themselves too! 464 self._all.clear(self._screen, self._background) 465 self._screen.blit(self._background, (0, 0))
Run any code before the agents are updated in every tick.
You should override this method when inheriting Simulation to add your own logic.
Some examples include:
- Processing events from PyGame's event queue.
def
after_update(self) -> None:
467 def after_update(self) -> None: 468 # Draw everything to the screen 469 self._all.draw(self._screen) 470 471 if self.config.visualise_chunks: 472 self.__visualise_chunks() 473 474 # Update the screen with the new image 475 pg.display.flip() 476 477 self._clock.tick(self.config.fps_limit) 478 479 current_fps = self._clock.get_fps() 480 if current_fps > 0: 481 self._metrics.fps._push(current_fps) 482 483 if self.config.print_fps: 484 print(f"FPS: {current_fps:.1f}") # noqa: T201
Inherited Members
@dataclass
class
Window:
76@dataclass 77class Window: 78 """Settings related to the simulation window.""" 79 80 width: int = 750 81 """The width of the simulation window in pixels.""" 82 83 height: int = 750 84 """The height of the simulation window in pixels.""" 85 86 @classmethod 87 def square(cls, size: int) -> Self: 88 return cls(width=size, height=size) 89 90 def as_tuple(self) -> tuple[int, int]: 91 return (self.width, self.height)
Settings related to the simulation window.