A simulation models a set of objects that act on their own and interact through shared state. This chapter builds three, each giving its agents less to work with than the last. A pack of rats coordinates through a shared blackboard, a single robot walks a maze where each object it meets decides what happens, and a plate of vibrating sand runs on grains that know nothing. The first two confirm a design you can predict from the code. The third produces a pattern nobody wrote down.
The chapter works the first example, the pack of rats, from
end to end. It puts asyncio tasks, a shared coordination object,
and structural typing together in one small program. Concurrency
introduces the asyncio mechanics
(async def, await,
gather, run).
The problem has three types.
A maze knows its own layout. Given a coordinate, it reports whether each neighboring cell is a wall or an opening, and it hands out an entry point. The maze never decides anything. It only answers questions.
A blackboard is the shared surface on which every
rat writes. Blackboard is a classic coordination pattern.
Independent agents read from and write to one common data
structure instead of talking to each other directly. Here the
blackboard owns the maze, records which cells the rats have
explored, hands out rat numbers, and launches new rats. The rats
run as cooperative asyncio tasks. They take turns
instead of running at the same instant, so the blackboard needs
no lock. Nothing interrupts a rat partway through an update.
A rat explores. Each rat runs as its own task. From its current cell it looks at the four neighbors and tries to claim the open ones. By claiming a cell, a rat both marks it visited and reserves it. This way, no two rats cover the same ground. When a rat finds more than one open neighbor, it keeps the first for itself and spawns a new rat down each of the others, then yields so its siblings can run. When it can claim nothing, it has reached a dead end and its task ends. When the last rat dies, the pack has mapped every cell reachable from the entry.
The rat does not import the blackboard. It needs only an
object with matching methods, so a Protocol
describes what it expects. This is structural typing from Static
Typing. The rat works with anything that can claim a cell,
spawn a rat, record a message, and hand out a number.
# rats_and_mazes/rat.py
import asyncio
from dataclasses import dataclass, field
from typing import Final, Protocol
# South, north, west, east
DIRECTIONS: Final[list[tuple[int, int]]] = [
(0, 1), (0, -1), (-1, 0), (1, 0)]
class Recorder(Protocol):
def claim(self, x: int, y: int) -> bool: ...
def spawn(self, x: int, y: int) -> None: ...
def log(self, message: str) -> None: ...
def next_number(self) -> int: ...
@dataclass
class Rat:
blackboard: Recorder
x: int
y: int
number: int = field(init=False)
def __post_init__(self) -> None:
self.number = self.blackboard.next_number()
self.blackboard.log(
f"Rat {self.number} starts at {(self.x, self.y)}.")
async def run(self) -> None:
while True:
neighbors = [
(self.x + dx, self.y + dy) for dx, dy in DIRECTIONS]
moves = [pos for pos in neighbors
if self.blackboard.claim(*pos)]
if not moves:
self.blackboard.log(
f"Rat {self.number} dead-ends "
f"at {(self.x, self.y)}.")
return
for branch in moves[1:]:
self.blackboard.spawn(*branch)
self.x, self.y = moves[0]
await asyncio.sleep(0) # Yield so sibling rats can runInitializing number requires calling
blackboard.next_number(), a side-effecting method,
not a static default. Marking it field(init=False)
leaves it out of the generated __init__, and
__post_init__ runs immediately after that
__init__ finishes, so it fills in
number and logs the rat’s start, once
blackboard, x, and y hold
their values.
The maze is a grid of characters. A * is a wall
and a space is an opening. Out-of-bounds coordinates count as
walls, so the rats stay inside.
# rats_and_mazes/maze.py
from enum import StrEnum
from pathlib import Path
from typing import Self
type Coord = tuple[int, int] # (column, row)
class Maze:
class Cell(StrEnum):
WALL = "*"
OPEN = " "
def __init__(self, rows: list[str]) -> None:
self.height = len(rows)
self.width = max((len(r) for r in rows), default=0)
self.rows = [
r.ljust(self.width, self.Cell.WALL) for r in rows]
@classmethod
def from_text(cls, text: str) -> Self:
rows = [line for line in text.splitlines()
if line and not line.lstrip().startswith("#")]
return cls(rows)
@classmethod
def from_file(cls, filename: str) -> Self:
return cls.from_text(
Path(filename).read_text(encoding="utf-8"))
def is_open(self, x: int, y: int) -> bool:
return (0 <= y < self.height and 0 <= x < self.width
and self.rows[y][x] == self.Cell.OPEN)
def entry(self) -> Coord:
for y in range(self.height):
for x in range(self.width):
if self.is_open(x, y):
return x, y
raise ValueError("the maze has no open cell")Cell nests inside Maze because it
names concepts only Maze uses, and it is a
StrEnum rather than an Enum so its
members keep acting like real strings. WALL still
works as the fill character for ljust(), and
comparing self.rows[y][x] against
Cell.OPEN still works, because a
StrEnum member is its string value.
The blackboard holds everything the rats share.
claim() is the heart of the program. It tests and
marks a cell in one step with no await in between,
so a single rat gets each cell even when several reach it. It
sidesteps the read-modify-write race from Concurrency.
That race needs a suspension point inside the update, and
claim() contains none, so the atomicity comes from
the shape of the code rather than from a lock (exercise 3
inserts a suspension point and looks at what breaks).
next_number() hands out rat numbers from
itertools.count(), the endless counter from Iterators.
explore() claims the entry and releases the first
rat inside an asyncio.TaskGroup:
# rats_and_mazes/blackboard.py
import asyncio
import itertools
from collections.abc import Iterator
from dataclasses import dataclass, field
from maze import Coord, Maze
from rat import Rat
@dataclass
class Blackboard:
maze: Maze
visited: set[Coord] = field(init=False, default_factory=set)
tasks: list[asyncio.Task[None]] = field(
init=False, default_factory=list)
messages: list[str] = field(init=False, default_factory=list)
_numbers: Iterator[int] = field(
init=False, default_factory=lambda: itertools.count(1))
group: asyncio.TaskGroup = field(init=False)
def claim(self, x: int, y: int) -> bool:
# No await between the test and the add, so this is atomic
if self.maze.is_open(x, y) and (x, y) not in self.visited:
self.visited.add((x, y))
return True
return False
def spawn(self, x: int, y: int) -> None:
rat = Rat(self, x, y)
self.tasks.append(self.group.create_task(rat.run()))
def next_number(self) -> int:
return next(self._numbers)
def log(self, message: str) -> None:
self.messages.append(message)
async def explore(self) -> None:
start = self.maze.entry()
self.claim(*start)
async with asyncio.TaskGroup() as group:
self.group = group
self.spawn(*start)
def render(self) -> str:
lines = []
for y in range(self.maze.height):
row = []
for x in range(self.maze.width):
if not self.maze.is_open(x, y):
row.append("#")
elif (x, y) in self.visited:
row.append(".")
else:
row.append(" ")
lines.append("".join(row))
return "\n".join(lines)A TaskGroup does not close until every task
inside it has finished, including tasks created after the block
began, which is the shape this problem has: each rat can create
more rats. A single asyncio.gather(*self.tasks)
would not do, because gather() fixes its argument
list at the moment of the call, and half the rats do not exist
yet.
group’s declaration is
field(init=False), and only explore()
assigns it, the same declaration-without-assignment the robot
example later in this chapter uses for Robot.room.
The other four fields are internal bookkeeping rather than
constructor arguments: init=False keeps them out of
the generated signature, and each default_factory
builds a fresh object per blackboard.
The maze layout lives in a text file. The loader drops blank
lines and any line beginning with #, so the first
line, naming the file’s path, drops out and the rest is the
maze.
# rats_and_mazes/amaze.txt
*********************
* * * *
* * * ******* *** * *
* * * * * *
* ***** *** ******* *
* * * * * *
***** *** ***** *** *
* * * * *
* * ******* *** * ***
* * * * *
* ***** * ********* *
* * * * *
***** * *** *********
* * *
*********************
Running it turns the rats loose, then prints the first eight log messages and the mapped maze. The log shows what the map cannot: rat 1 spawns rat 2 and then dies before it, and numbers arrive in spawn order rather than completion order. The full log runs to eighteen messages, two per rat.
# rats_and_mazes/rats_and_mazes.py
import asyncio
from blackboard import Blackboard
from maze import Maze
async def main() -> None:
maze = Maze.from_file("amaze.txt")
blackboard = Blackboard(maze)
await blackboard.explore()
for message in blackboard.messages[:8]:
print(message)
print("Mapped maze (# wall, . visited):")
print(blackboard.render())
print(f"{len(blackboard.tasks)} rats mapped "
f"{len(blackboard.visited)} cells.")
asyncio.run(main())
#: Rat 1 starts at (1, 1).
#: Rat 2 starts at (6, 3).
#: Rat 1 dead-ends at (7, 5).
#: Rat 3 starts at (6, 1).
#: Rat 2 dead-ends at (3, 3).
#: Rat 4 starts at (18, 1).
#: Rat 3 dead-ends at (15, 1).
#: Rat 5 starts at (12, 13).
#: Mapped maze (# wall, . visited):
#: #####################
#: #.#...........#.....#
#: #.#.#.#######.###.#.#
#: #.#.#.......#.....#.#
#: #.#####.###.#######.#
#: #.....#.#...#.....#.#
#: #####.###.#####.###.#
#: #...#.....#.....#...#
#: #.#.#######.###.#.###
#: #.#.........#...#...#
#: #.#####.#.#########.#
#: #.....#.#.#.........#
#: #####.#.###.#########
#: #.....#.............#
#: #####################
#: 9 rats mapped 139 cells.Because claiming is atomic, the rats always cover every cell reachable from the entry, no matter how the tasks interleave. The test verifies this by comparing the cells the rats visited against a flood fill of the same maze.
# rats_and_mazes/test_rats_and_mazes.py
import asyncio
from typing import Final
from blackboard import Blackboard
from maze import Coord, Maze
LAYOUT: Final[str] = """\
*********
* *
*** *** *
* * *
* ***** *
* *
*********
"""
def flood(maze: Maze, start: Coord) -> set[Coord]:
seen: set[Coord] = set()
stack = [start]
while stack:
x, y = stack.pop()
if (x, y) in seen or not maze.is_open(x, y):
continue
seen.add((x, y))
stack += [(x + 1, y), (x - 1, y), (x, y + 1), (x, y - 1)]
return seen
def test_rats_map_every_reachable_cell() -> None:
maze = Maze.from_text(LAYOUT)
blackboard = Blackboard(maze)
asyncio.run(blackboard.explore())
assert blackboard.visited == flood(maze, maze.entry())The same model drives a GUI demonstration.
rats_view.py lets the rats finish exploring,
records the order in which they claimed cells, and replays that
order on a tkinter canvas: walls in gray, then each
claimed cell turning green one after another, so you watch the
pack move through the maze from the entry outward. Each of this
chapter’s three views is a separate file holding all the display
code, the model-view split of Observer.
The missing piece is the subscription: no model in this chapter
notifies anybody, so each view drives or replays its model
instead of waiting for a notification. The harness skips it,
like every windowed view in this book
(tools/data/norun.txt lists all three of this
chapter’s views):
# rats_and_mazes/rats_view.py
import asyncio
import tkinter as tk
from typing import Final, override
from blackboard import Blackboard
from maze import Coord, Maze
CELL: Final[int] = 26
class RecordingBlackboard(Blackboard):
def __init__(self, maze: Maze) -> None:
super().__init__(maze)
self.order: list[Coord] = []
@override
def claim(self, x: int, y: int) -> bool:
claimed = super().claim(x, y)
if claimed:
self.order.append((x, y))
return claimed
def show(layout: str = "amaze.txt", step_ms: int = 60) -> None:
maze = Maze.from_file(layout)
board = RecordingBlackboard(maze)
asyncio.run(board.explore())
root = tk.Tk()
root.title("Rats and Mazes")
canvas = tk.Canvas(root, highlightthickness=0,
width=maze.width * CELL,
height=maze.height * CELL)
canvas.pack()
def box(x: int, y: int, color: str) -> None:
canvas.create_rectangle(
x * CELL, y * CELL, (x + 1) * CELL, (y + 1) * CELL,
fill=color, outline="gray")
for y in range(maze.height):
for x in range(maze.width):
box(x, y, "white" if maze.is_open(x, y) else "dimgray")
cells = iter(board.order)
def step() -> None:
cell = next(cells, None)
if cell is not None:
box(cell[0], cell[1], "palegreen")
root.after(step_ms, step)
step()
root.mainloop()
if __name__ == "__main__":
show()Jeremy Meyer wrote the original Java version of this example.
Concurrency is one way to build a simulation. Object-oriented
design is another. This second example, adapted from my
Atomic Kotlin book, walks a single robot through a
maze. It shows how polymorphism removes conditionals. A
Room asks its occupant what to do, and each type of
occupant answers for itself.
The occupants are Items.
Room.enter() calls
occupant.interact(), and returns the room in which
the robot ends up. A wall keeps the robot where it is, food
feeds the robot and lets it in, a teleport returns a distant
room. No if or elif on the type of
occupant appears in the movement code:
# robot_explorer/items.py
from enum import Enum, auto
from typing import TYPE_CHECKING, ClassVar, override
if TYPE_CHECKING:
from world import Room
class Urge(Enum):
NORTH = auto()
SOUTH = auto()
EAST = auto()
WEST = auto()
class Item:
symbol: ClassVar[str] = ""
def interact(self, robot: Robot, room: Room) -> Room:
return room # Default: the robot enters the room
def __str__(self) -> str:
return self.symbol
class Robot(Item):
symbol = "R"
room: Room # Set by the builder when the robot is placed
def __init__(self) -> None:
self.finished = False # Set when the robot reaches the end
def move(self, urge: Urge) -> None:
self.room = self.room.doors.open(urge).enter(self)
class Wall(Item):
symbol = "#"
@override
def interact(self, robot: Robot, room: Room) -> Room:
return robot.room # Cannot pass: stay put
class Food(Item):
symbol = "."
@override
def interact(self, robot: Robot, room: Room) -> Room:
room.occupant = Empty() # Eaten
return room
class Teleport(Item):
symbol = "" # Set per target letter
target_room: Room # Paired up by the builder
def __init__(self, target: str) -> None:
self.target = target
@override
def interact(self, robot: Robot, room: Room) -> Room:
return self.target_room
@override
def __str__(self) -> str:
return self.target
class Empty(Item):
symbol = "_"
@override
def interact(self, robot: Robot, room: Room) -> Room:
return room
class Edge(Item):
symbol = "/"
@override
def interact(self, robot: Robot, room: Room) -> Room:
return robot.room # The void outside the maze: stay put
class EndGame(Item):
symbol = "!"
@override
def interact(self, robot: Robot, room: Room) -> Room:
robot.finished = True # Recorded, not printed
return room
def item_factory(symbol: str) -> Item:
for item_type in Item.__subclasses__():
if symbol == item_type.symbol:
return item_type()
return Teleport(symbol) # Anything else is a teleport targetworld.py imports Item,
Robot, and Urge from
items.py, so from world import Room
here is circular. if TYPE_CHECKING: is
False at runtime, so that import never runs, and no
cycle forms. It is True only for a type checker
reading the file. Every use of Room below is an
annotation (room: Room, -> Room),
never a runtime lookup.
Robot holds its two pieces of state in different
ways. __init__ assigns finished, so
each robot owns its own flag from the start. The code only
declares room, writing room: Room with
no value. That line stores nothing, not even None.
It is a declaration: it tells the type checker that a
Room belongs there, which GameBuilder
guarantees when it places the robot and sets
robot.room. The attribute does not exist until
then, so reading it earlier raises an
AttributeError, and the builder runs first, so
nothing reads it earlier. Declaring it this way keeps the type
Room instead of Room | None, so no
code that reads room needs a None
check.
item_factory() turns a maze character into an
Item. It searches
Item.__subclasses__() for a matching
symbol, so adding a new kind of item needs no
change here. If you define the subclass with its symbol, the
factory finds it. This is the registry idea from Factory,
using the class hierarchy as the registry.
__subclasses__() reports only direct subclasses
(that chapter’s Simple Factory
Method shows the recursion for deeper hierarchies), so a new
item must inherit from Item itself. Deriving from
Food to borrow its behavior hides the class from
the factory, which falls through to the last line and builds a
Teleport instead.
A Room holds one item and connects to its
neighbors through a Doors object. Doors that lead
nowhere point at one shared EDGE room, the void
outside the maze, so the robot can try any direction without a
special case. EDGE is a Null Object:
it answers like any other room and sends the robot back where it
started:
# robot_explorer/world.py
from typing import Final
from items import Edge, Item, Robot, Urge
type Coord = tuple[int, int] # (row, col)
type RoomMap = dict[Coord, Room]
class Room:
def __init__(self, occupant: Item) -> None:
self.occupant = occupant
self.doors = Doors()
def enter(self, robot: Robot) -> Room:
return self.occupant.interact(robot, self)
def __repr__(self) -> str:
return f"Room({self.occupant})"
class Doors:
def __init__(self) -> None:
self.neighbors: dict[Urge, Room] = {}
def connect(self, row: int, col: int,
rooms: RoomMap) -> None:
for urge, coord in {
Urge.NORTH: (row - 1, col),
Urge.SOUTH: (row + 1, col),
Urge.EAST: (row, col + 1),
Urge.WEST: (row, col - 1),
}.items():
if coord in rooms:
self.neighbors[urge] = rooms[coord]
def open(self, urge: Urge) -> Room:
return self.neighbors.get(urge, EDGE)
# Created once both classes exist; its own doors stay unset
EDGE: Final[Room] = Room(Edge())The Coord here counts (row, col),
the opposite order from the rats example’s
(column, row), because GameBuilder
walks the maze text line by line.
One move is one chain. Robot.move() asks
doors.open(urge) for the neighboring room,
EDGE when no door leads that way.
enter() hands the robot to that room, and whatever
room its occupant’s interact() returns becomes
robot.room. Every rule of the game lives in some
interact().
GameBuilder assembles the maze in three stages:
a room for every character, then the connections between rooms,
then the teleport pairs. Each stage depends on the one before
it, so splitting them into labeled passes keeps the construction
readable instead of tangling it into one loop. Factory counts this as one of
the cases where Builder survives in Python, because construction
here is genuinely a process rather than a single call.
run() walks a string of moves, and
show_maze() renders the current state:
# robot_explorer/game.py
# Build the maze in three stages, then run it.
from items import Empty, Robot, Teleport, Urge, item_factory
from world import Room, RoomMap
class GameBuilder:
def __init__(self, maze: str) -> None:
self.rooms: RoomMap = {}
teleports: list[Room] = []
# Stage 1: a Room for every character
for row, line in enumerate(maze.splitlines()):
for col, char in enumerate(line):
occupant = item_factory(char)
if isinstance(occupant, Robot):
room = Room(Empty())
self.robot = occupant
self.robot.room = room
else:
room = Room(occupant)
self.rooms[row, col] = room
if isinstance(occupant, Teleport):
teleports.append(room)
# Stage 2: connect each room to its neighbors
for (row, col), room in self.rooms.items():
room.doors.connect(row, col, self.rooms)
# Stage 3: pair the teleports that share a target letter
def target(room: Room) -> str:
assert isinstance(room.occupant, Teleport)
return room.occupant.target
teleports.sort(key=target)
pairs = iter(teleports)
for room1, room2 in zip(pairs, pairs):
assert isinstance(room1.occupant, Teleport)
assert isinstance(room2.occupant, Teleport)
room1.occupant.target_room = room2
room2.occupant.target_room = room1
def show_maze(self) -> str:
rows: list[str] = []
current = -1
for (row, _), room in self.rooms.items():
if row != current:
rows.append("")
current = row
if room is self.robot.room:
rows[-1] += str(self.robot)
else:
rows[-1] += str(room.occupant)
return "\n".join(rows)
def run(self, solution: str) -> None:
moves = {"n": Urge.NORTH, "s": Urge.SOUTH,
"e": Urge.EAST, "w": Urge.WEST}
for char in "".join(solution.split()):
self.robot.move(moves[char])
string_maze = """
###############################
#R#.____#____.#_______#_______#
#_###_#_###_#_#_#_#####_#####_#
#___#_#___#_#_#_#.#__b__#___#_#
###_#_###_#_#_###_#_#####_#_#_#
#.#_#_#.__#_#__.#_#__b__#_#___#
#_#_#_#_###_###_#_#####_#_#####
#_#_#_#__.#_#_#_____#___#_____#
#_#_#_###_#_#_#_#####_#######_#
#.#___#___#_#___#____.#_____#_#
#_#####_###_#_###_#####_#_###_#
#___#a__#.__#.__#__.#___#_#___#
#_#_#_###_#####_###_###_###_#_#
#_#.#_#___#!______#_____#___#_#
#_#_#_###_#############_#_###_#
#_#_#__a#_______________#___#_#
#_#####_###_###########_###_#_#
#_____#.__#_#___#_____#_#___#_#
#_#_#####_###_#_#_###_###_###_#
#.#___________#___#____.__#___#
###############################
""".strip()
solution = (
"sseesssssseennnnnnnneesseesswwsseesswwsswwsseesseeeenneessee"
"nneeeesseeeenneennwwnneenneennnnwwwwnnnneesseennnnwwwwwwssww"
"eesswwsswwwwsseesseeeesswwwwwwwwwwwwwwnnnneennnnnnnnnneessss"
"eesssswwsseesswwww"
)Running the demo prints the maze before and after the walk:
# robot_explorer/robot_demo.py
from game import GameBuilder, solution, string_maze
game = GameBuilder(string_maze)
print("start:")
print(game.show_maze())
#: start:
#: ###############################
#: #R#.____#____.#_______#_______#
#: #_###_#_###_#_#_#_#####_#####_#
#: #___#_#___#_#_#_#.#__b__#___#_#
#: ###_#_###_#_#_###_#_#####_#_#_#
#: #.#_#_#.__#_#__.#_#__b__#_#___#
#: #_#_#_#_###_###_#_#####_#_#####
#: #_#_#_#__.#_#_#_____#___#_____#
#: #_#_#_###_#_#_#_#####_#######_#
#: #.#___#___#_#___#____.#_____#_#
#: #_#####_###_#_###_#####_#_###_#
#: #___#a__#.__#.__#__.#___#_#___#
#: #_#_#_###_#####_###_###_###_#_#
#: #_#.#_#___#!______#_____#___#_#
#: #_#_#_###_#############_#_###_#
#: #_#_#__a#_______________#___#_#
#: #_#####_###_###########_###_#_#
#: #_____#.__#_#___#_____#_#___#_#
#: #_#_#####_###_#_#_###_###_###_#
#: #.#___________#___#____.__#___#
#: ###############################
game.run(solution)
if game.robot.finished:
print("Game over!")
#: Game over!
print("\nfinal:")
print(game.show_maze())
#:
#: final:
#: ###############################
#: #_#.____#_____#_______#_______#
#: #_###_#_###_#_#_#_#####_#####_#
#: #___#_#___#_#_#_#.#__b__#___#_#
#: ###_#_###_#_#_###_#_#####_#_#_#
#: #.#_#_#___#_#___#_#__b__#_#___#
#: #_#_#_#_###_###_#_#####_#_#####
#: #_#_#_#___#_#_#_____#___#_____#
#: #_#_#_###_#_#_#_#####_#######_#
#: #.#___#___#_#___#_____#_____#_#
#: #_#####_###_#_###_#####_#_###_#
#: #___#a__#___#___#___#___#_#___#
#: #_#_#_###_#####_###_###_###_#_#
#: #_#.#_#___#R______#_____#___#_#
#: #_#_#_###_#############_#_###_#
#: #_#_#__a#_______________#___#_#
#: #_#####_###_###########_###_#_#
#: #_____#___#_#___#_____#_#___#_#
#: #_#_#####_###_#_#_###_###_###_#
#: #.#___________#___#_______#___#
#: ###############################The robot eats the food along its path, jumps through both
teleports (a, then b), and reaches the
! that ends the game.
Stage 3 pairs the teleports with a small idiom.
pairs = iter(teleports) makes one iterator, and
zip(pairs, pairs) pulls from that same iterator
twice per loop, so each pass consumes two rooms: the first and
second a, then the two bs. The sort by
target letter lines those partners up beforehand. Avoid
zip(teleports, teleports), which walks two
independent passes over the list and pairs every room with
itself. The assert isinstance lines that follow are
for the type checker as much as for safety: each proves to the
checker that the occupant really is a Teleport
before the code touches target_room.
Stage 1 does test types, with
isinstance(occupant, Robot) and
isinstance(occupant, Teleport). That is not the
type switch polymorphism removes. GameBuilder still
must tell the kinds of item apart, once, and the movement code
that runs afterward never asks again. The Robot
branch also explains Room(Empty()): the robot is
the one item that does not become an occupant. Its cell gets an
Empty occupant instead, so when the robot moves
away the room behaves like any other empty room, and
show_maze() draws the R by checking
which room the robot holds rather than reading an occupant.
show_maze() renders the maze into a string, so a
test can check the model without opening a window. Build the
maze, run the solution, and check that the robot finished on the
! square and that the final rendering matches, food
eaten and all:
# robot_explorer/test_robot.py
from typing import Final
from game import GameBuilder, solution, string_maze
from items import EndGame
FINISHED: Final[str] = """
###############################
#_#.____#_____#_______#_______#
#_###_#_###_#_#_#_#####_#####_#
#___#_#___#_#_#_#.#__b__#___#_#
###_#_###_#_#_###_#_#####_#_#_#
#.#_#_#___#_#___#_#__b__#_#___#
#_#_#_#_###_###_#_#####_#_#####
#_#_#_#___#_#_#_____#___#_____#
#_#_#_###_#_#_#_#####_#######_#
#.#___#___#_#___#_____#_____#_#
#_#####_###_#_###_#####_#_###_#
#___#a__#___#___#___#___#_#___#
#_#_#_###_#####_###_###_###_#_#
#_#.#_#___#R______#_____#___#_#
#_#_#_###_#############_#_###_#
#_#_#__a#_______________#___#_#
#_#####_###_###########_###_#_#
#_____#___#_#___#_____#_#___#_#
#_#_#####_###_#_#_###_###_###_#
#.#___________#___#_______#___#
###############################
""".strip()
def test_solution_walks_the_robot_to_the_end() -> None:
game = GameBuilder(string_maze)
game.run(solution)
room = game.robot.room
assert isinstance(room.occupant, EndGame) # Finished on the "!"
assert game.robot.finished # And the model recorded it
assert game.show_maze() == FINISHED # Food eaten, robot moved
def test_walls_block_and_food_is_eaten() -> None:
game = GameBuilder("R.#") # Robot, food, wall in one row
start = game.robot.room
game.run("e") # East: eat the food and move in
assert "." not in game.show_maze() # Food gone
assert game.robot.room is not start
blocked = game.robot.room
game.run("e") # East again: a wall, so stay put
assert game.robot.room is blockedThat same model drives a graphical view.
maze_view.py imports the maze and the moves, draws
each room as a colored cell, and steps the robot along the
solution on a timer. The view is the only part that touches the
screen.
# robot_explorer/maze_view.py
import tkinter as tk
from typing import Final
from game import GameBuilder, solution, string_maze
from items import Urge
CELL: Final[int] = 20
FILL: Final[dict[str, str]] = {
"#": "dimgray", "!": "tomato", ".": "khaki",
"_": "white", "R": "royalblue"}
MOVES: Final[dict[str, Urge]] = {
"n": Urge.NORTH, "s": Urge.SOUTH,
"e": Urge.EAST, "w": Urge.WEST}
def show(maze: str = string_maze, moves: str = solution,
step_ms: int = 80) -> None:
game = GameBuilder(maze)
rows = maze.splitlines()
width = max(len(row) for row in rows)
root = tk.Tk()
root.title("Robot in a Maze")
canvas = tk.Canvas(root, highlightthickness=0,
width=width * CELL, height=len(rows) * CELL)
canvas.pack()
def draw() -> None:
canvas.delete("all")
for (row, col), room in game.rooms.items():
symbol = ("R" if room is game.robot.room
else str(room.occupant))
canvas.create_rectangle(
col * CELL, row * CELL,
(col + 1) * CELL, (row + 1) * CELL,
fill=FILL.get(symbol, "palegreen"), outline="gray")
queue = list("".join(moves.split()))
def step() -> None:
draw()
if queue:
game.robot.move(MOVES[queue.pop(0)])
root.after(step_ms, step)
step()
root.mainloop()
if __name__ == "__main__":
show()Three ideas from earlier chapters carry the design. Polymorphism replaces a type switch, a factory builds objects from data, and a Null Object removes the check for a missing door. None of them needs concurrency.
Two further resources on mazes: a survey of algorithms to create mazes, and Craig Reynolds on steering behavior for autonomous moving objects, which is where a robot that decided its own route would start.
The two simulations so far confirm designs. The rats cover
every reachable cell because claim() is atomic. The
robot reaches the goal because polymorphism handles every
encounter. In each case you knew the outcome in advance and ran
the program to confirm it. This final example is different. Its
result appears in no line of its code. That is simulation’s
other purpose, to discover behavior instead of confirming
it.
In 1787 Ernst Chladni sprinkled sand across a metal plate and drew a violin bow along its edge. The bow made the plate ring. A ringing plate does not move evenly. Standing waves divide it into regions that swing up and down, and the nodal lines between them stay still. The vibration bounces sand out of the moving regions. When a grain comes to rest on a still line, nothing kicks it away again. Within seconds the random motion sweeps the sand into sharp, symmetric curves. Bowing a different spot rings the plate in a different mode and draws a different figure.
The model needs almost nothing. amplitude() is
the standing-wave field of a square plate ringing in mode
(m, n). Physics supplies the formula, an
approximation for a plate with free edges. Treat it as given.
All that matters here is its shape. It is zero along curves, and
those curves are the nodal lines. A Grain is a
position. step() is the entire simulation. Every
grain takes one random step, and the plate’s vibration at that
grain’s location scales it. Grains never look at each other and
remember nothing. Nothing in the code knows the pattern
exists.
# chladni_plate/chladni.py
import math
import random
from dataclasses import dataclass
type Mode = tuple[int, int] # Vibration pattern (m, n)
def amplitude(x: float, y: float, mode: Mode) -> float:
m, n = mode
return abs(
math.cos(m * math.pi * x) * math.cos(n * math.pi * y)
- math.cos(n * math.pi * x) * math.cos(m * math.pi * y))
def bounce(v: float) -> float:
if v < 0.0:
return -v
if v > 1.0:
return 2.0 - v
return v
@dataclass
class Grain:
x: float
y: float
class Plate:
def __init__(self, grains: int, mode: Mode,
seed: int | None = None) -> None:
self.rng = random.Random(seed)
self.mode = mode
self.grains = [
Grain(self.rng.random(), self.rng.random())
for _ in range(grains)]
def step(self, kick: float = 0.05) -> None:
for g in self.grains:
a = amplitude(g.x, g.y, self.mode)
g.x = bounce(
g.x + self.rng.uniform(-kick, kick) * a)
g.y = bounce(
g.y + self.rng.uniform(-kick, kick) * a)
def agitation(self) -> float:
return sum(
amplitude(g.x, g.y, self.mode)
for g in self.grains) / len(self.grains)
def render(self, width: int = 60, height: int = 30) -> str:
counts: list[list[int]] = [
[0] * width for _ in range(height)]
for g in self.grains:
col = min(int(g.x * width), width - 1)
row = min(int(g.y * height), height - 1)
counts[row][col] += 1
shades = " .:*#"
return "\n".join(
"".join(shades[min(c, len(shades) - 1)]
for c in row).rstrip()
for row in counts)bounce() reflects a kicked grain off the edge
instead of letting it leave the plate. agitation()
measures the mean vibration strength directly under the grains.
Grains scattered at random feel the field’s average, so
agitation starts high. A grain resting on a nodal line feels
zero. One number summarizes how settled the sand is.
render() draws grain density as characters, in the
same spirit as Blackboard.render(), so the model
can show its state without a window. The demo shakes the plate
1200 times, printing agitation at four checkpoints along the
way:
# chladni_plate/chladni_demo.py
from chladni import Plate
plate = Plate(grains=2000, mode=(2, 3), seed=42)
steps = 0
for target in (0, 100, 400, 1200):
for _ in range(target - steps):
plate.step()
steps = target
print(f"steps {target:4}: "
f"agitation {plate.agitation():.3f}")
#: steps 0: agitation 0.585
#: steps 100: agitation 0.073
#: steps 400: agitation 0.005
#: steps 1200: agitation 0.000
print(plate.render())
#: *. # #
#: :## # #
#: ## ## #
#: ## ##. #
#: ## ### .#
#: ## ###### ##########
#: ## ########
#: ## ###
#: ## ##
#: ## #*
#: ## #
#: ## #
#: ####### ## #
#: *## ## #:
#: ## ## #
#: # ## #:
#: ## ## ##*
#: # ## #######
#: # ##
#: # ##
#: :# ##
#: ## ##
#: ##* ##
#: ######## ##
#: ######*### ###### ##
#: ## ### ##
#: # .## ##
#: # ## ##.
#: # # ##.
#: # # .:##Agitation collapses toward zero, and the picture shows why.
The grains have gathered on the nodal lines of mode
(2, 3). Nothing steered them there. A loud region
flings its grains around until a random wander crosses a quiet
line, where the kicks shrink toward nothing. Noise can carry a
grain into a quiet place. It cannot carry the grain back out.
The randomness is not fighting the order but producing it.
A test cannot guess where a particular grain ends up after a
million random kicks. It pins down the aggregate instead.
Shaking must collapse agitation, and no kick may throw a grain
off the plate. Seeding random.Random makes any
failure reproducible.
# chladni_plate/test_chladni.py
from chladni import Plate
def test_noise_settles_grains_onto_quiet_lines() -> None:
plate = Plate(grains=500, mode=(2, 3), seed=1)
before = plate.agitation()
for _ in range(400):
plate.step()
assert plate.agitation() < before / 10
def test_kicks_never_knock_grains_off_the_plate() -> None:
plate = Plate(grains=200, mode=(3, 5), seed=2)
for _ in range(300):
plate.step(kick=0.2)
assert all(0.0 <= g.x <= 1.0 and 0.0 <= g.y <= 1.0
for g in plate.grains)The tkinter view shows what the text version cannot: the collapse as it unfolds, and the pattern surviving a change of rules. Each grain keeps one color from a small palette, so you can watch individual grains mix while the collective figure forms. Every 200 frames the view switches the plate to a new mode. The old figure suddenly sits on loud regions of the new field. It bursts back into chaos, mixes, and condenses into a different figure. The order is not a property of the grains. It belongs to the field on which they sit.
# chladni_plate/chladni_view.py
import itertools
import tkinter as tk
from typing import Final
from chladni import Mode, Plate
SIZE: Final[int] = 560
DOT: Final[int] = 3
COLORS: Final[list[str]] = [
"gold", "coral", "palegreen", "skyblue", "plum"]
MODES: Final[list[Mode]] = [(1, 2), (2, 3), (3, 4), (3, 5)]
def show(grains: int = 1200, step_ms: int = 30,
frames_per_mode: int = 200) -> None:
plate = Plate(grains, MODES[0])
root = tk.Tk()
root.title(f"Chladni Plate {plate.mode}")
canvas = tk.Canvas(root, width=SIZE, height=SIZE,
background="black", highlightthickness=0)
canvas.pack()
palette = itertools.cycle(COLORS)
dots = [
canvas.create_oval(0, 0, DOT, DOT, outline="",
fill=next(palette))
for _ in plate.grains]
modes = itertools.cycle(MODES[1:] + MODES[:1])
frames = itertools.count(1)
def frame() -> None:
if next(frames) % frames_per_mode == 0:
plate.mode = next(modes)
root.title(f"Chladni Plate {plate.mode}")
for _ in range(3):
plate.step()
for dot, g in zip(dots, plate.grains):
canvas.moveto(dot, g.x * SIZE - DOT / 2,
g.y * SIZE - DOT / 2)
root.after(step_ms, frame)
frame()
root.mainloop()
if __name__ == "__main__":
show()itertools.cycle() constructs an infinite
iterator from any finite iterable: it yields the source’s
elements in sequence and starts over when it reaches the end.
itertools.count(1) numbers the frames, the same
endless counter that numbered the rats.
The chapter began by defining a simulation as objects that act on their own and interact through shared state. The grains push that definition to its limit. The shared state is the plate, and the grains only read it. They never sense each other. Even so, structure that no agent encodes appears in the aggregate. This is emergence: global order arising from local rules that never mention it. The less the agents understand, the more the run can tell you, because the outcome lives in the interactions rather than the instructions. When behavior emerges, reading the code is not enough. Run it.
Rat with a fake blackboard. Because
Rat depends only on the Recorder
Protocol, you can drive it with a stand-in. Write a
fake whose claim() returns a scripted sequence of
results and whose spawn() only records the
coordinates it receives, run one rat with
asyncio.run(rat.run()), and assert which cell the
rat kept for itself and which cells it spawned. You need no real
Blackboard, Maze, or task
scheduling.explore() finishes, compare
blackboard.visited against every open cell of the
Maze and print the open cells that no rat claimed.
Build a maze for which that set is not empty, and explain what
makes a cell unreachable.claim(). Make
claim() an async def, which pulls the
Recorder protocol, Rat.run()’s
comprehension, and explore() along with it, and put
await asyncio.sleep(0) between the membership test
and self.visited.add(...). Then count how many
calls return True and compare that count with
len(blackboard.visited).
test_rats_and_mazes.py still passes, because
visited is a set: the guarantee that broke is “one
rat per cell”, not “every cell visited”. What does the extra
success cost the rats, and why does the original
claim(), with no await inside it, need
no lock?Item to the robot maze.
Define a Coin subclass of Item with
the symbol $ whose interact() removes
itself the way Food does and adds one to a coin
count carried by the Robot. Place a few
$ characters in the maze and report how many the
robot collects. You shouldn’t need to touch
item_factory(), Room, or
GameBuilder. Explain why the factory finds your new
item on its own, and what it does if you derive
Coin from Food instead.GameBuilder and searches the
rooms for a path from the robot’s room to the
EndGame room, the way flood() searches
maze cells in test_rats_and_mazes.py. The room
graph has no is_open(), so the occupant decides
whether a room is passable: refuse any room holding a
Wall or an Edge. Turning the room path
back into
n/s/e/w
means tracking which Urge produced each step. Your
path comes out shorter than the hard-coded solution
and need not use the teleports, so assert only that the robot
finishes on the ! square, as
test_robot.py does.MODES starting at (2, 2). Work out
what amplitude() returns whenever
m == n, and explain why the result is neither chaos
nor a figure. Then explain why the main diagonal shows up in
every figure this plate makes. Swapping x and
y in the two terms of amplitude() is
the clue.amplitude() with
abs(math.sin(m * math.pi * x) * math.sin(n * math.pi * y)),
the standing waves of a membrane fixed at its edges, like a
drumhead. Predict the figures before you run the view. Why are
the nodal lines now straight?chladni_demo.py passing
kick=0.005 and then kick=0.5 to
plate.step(), printing agitation at the same
checkpoints. One setting produces order too slowly and the other
never sharpens. Explain both failures, and why an intermediate
kick avoids them.