Monolith to Composed¶
Structure a game as small, single-purpose nodes.
▶ Run in browserTags: tutorial architecture signals
Monolith to Composed¶
You can build games now. This last tutorial is about building them well: how to structure a game so it stays easy to change. The game itself is tiny, a dodge game where you slide a paddle along the bottom (arrow keys, A/D, or hold the pointer) to avoid falling blocks. The point is the shape of the code.
main.py is the composed version. The full before/after refactor, starting from
one giant node that does everything, is the walkthrough in
From Monolithic to Composed.
The idea: one node, one job¶
The tempting first draft is a single DodgeGame node that tracks the player’s x,
runs a spawn timer, holds the score, draws everything, and checks collisions. It
works, but every change touches the same 80-line method and nothing can be tested
or reused in isolation.
The fix is to give each concern its own node:
node |
its one job |
|---|---|
|
read input, move, clamp, draw itself |
|
fall, and announce when it escapes off the bottom |
|
a timer that emits each new enemy |
|
hold the score and draw it |
|
a bare |
|
own the graph and the one thing only it can see: collision |
Signals connect them¶
Each node declares its signals at class scope, exactly like it declares its properties, and stays ignorant of who is listening. The root does the wiring:
class Spawner(Node2D):
spawned = Signal(Enemy) # declared once; each instance gets its own
...
self.spawner.spawned.connect(self._on_spawned) # new enemy -> root places it
def _on_spawned(self, enemy):
enemy.escaped.connect(lambda: self.score.add(1)) # dodged -> +1
self.world.add_child(enemy)
The signal-pairing rule: every signal has an emitter and a listener. Here
Spawner.spawned, Enemy.escaped, and ScoreLabel.changed are all paired. A
signal with no listener (or a global reached for instead of a signal) is a smell
that the structure is wrong.
Composition makes “stop the game” one line¶
The monolith froze on death for free: one node, one if alive guard. Split into
five nodes, that guard only stops the root, and the spawner keeps spawning behind
the Game Over text. The composed answer is not five guards, it is one grouping node:
every gameplay node is a child of a World node, so tree.paused = True freezes the
whole subtree at once. The root sets update_mode = UpdateMode.ALWAYS so it keeps
ticking through the pause and can hear the restart, and World sets
UpdateMode.PAUSABLE so it does not inherit that exemption.
The payoff¶
The root’s on_update shrinks to pure game flow (alive check, restart, collision
sweep). Per-entity behaviour lives in the entity. Each node is small enough to test
on its own with a SceneRunner, and reusable in the next game.
Run it¶
# In your own copy of this directory
python main.py
# From the root of a repository checkout
uv run python examples/tutorials/monolith_to_composed/main.py
Source¶
1"""Monolith to Composed: Structure a game as small, single-purpose nodes.
2
3The capstone of the basics track. This is the *composed* version of a tiny dodge
4game: move the paddle along the bottom with the arrow keys (or A/D, or by holding
5the pointer) and avoid the falling blocks; each block you dodge scores a point.
6The lesson is the structure, not the game: instead of one giant node that does
7input, spawning, scoring, and collision, every concern is its own node, and they
8talk through signals.
9
10Read the README for the before/after refactor; this file is the after.
11
12# /// simvx
13# tags = ["tutorial", "architecture", "signals"]
14# web = { root = "DodgeGame", width = 800, height = 600, responsive = true }
15# ///
16
17## What you will learn
18
19- **One node, one job** -- `Player`, `Enemy`, `Spawner`, and `ScoreLabel` each own a
20 single concern, so each is short and testable in isolation.
21- **Paired signals** -- signals are declared at class scope (`escaped = Signal()`),
22 and every one here is emitted by one node and connected by another:
23 `Spawner.spawned`, `Enemy.escaped`, `ScoreLabel.changed`. A signal with no
24 listener (or a listener with no emitter) is a smell.
25- **The root owns the graph, not the state** -- `DodgeGame` only decides which nodes
26 exist and which signals wire to which slots, plus the one thing only it can see:
27 player-versus-enemy collision.
28- **One flag freezes the whole game** -- death sets `tree.paused`, which stops the
29 whole gameplay subtree at once, while the root keeps ticking (`UpdateMode.ALWAYS`)
30 so it can still hear the restart.
31
32Run: uv run python examples/tutorials/monolith_to_composed/main.py
33Headless self-check: uv run python examples/tutorials/monolith_to_composed/main.py --test
34"""
35
36import random
37
38from simvx.core import Input, Key, MouseButton, Node2D, Property, Signal, UpdateMode, Vec2
39from simvx.graphics import App
40
41WIDTH, HEIGHT = 800, 600
42PLAYER_W, PLAYER_H = 80, 16
43ENEMY_W, ENEMY_H = 40, 40
44
45
46class Player(Node2D):
47 """Owns its position and its one input concern: move left/right, clamped."""
48
49 speed = Property(360.0, range=(50, 800))
50
51 def on_update(self, dt: float):
52 dx = Input.get_strength("right") - Input.get_strength("left")
53 if Input.is_mouse_button_pressed(MouseButton.LEFT):
54 # Hold (or drag) the pointer to steer; a touch reads as the left button.
55 dx = max(-1.0, min(1.0, (Input.mouse_position.x - self.position.x) / 32.0))
56 new_x = self.position.x + dx * self.speed * dt
57 self.position = Vec2(max(PLAYER_W / 2, min(WIDTH - PLAYER_W / 2, new_x)), self.position.y)
58
59 def on_draw(self, renderer):
60 renderer.draw_rect(
61 (self.position.x - PLAYER_W / 2, self.position.y - PLAYER_H / 2),
62 (PLAYER_W, PLAYER_H),
63 colour=(0.5, 0.9, 1.0, 1),
64 filled=True,
65 )
66
67
68class Enemy(Node2D):
69 """Falls straight down and announces when it escapes off the bottom."""
70
71 speed = Property(220.0, range=(50, 600))
72
73 escaped = Signal() # the player dodged it
74
75 def on_update(self, dt: float):
76 self.position = Vec2(self.position.x, self.position.y + self.speed * dt)
77 if self.position.y > HEIGHT + ENEMY_H:
78 self.escaped.emit()
79 self.destroy()
80
81 def on_draw(self, renderer):
82 renderer.draw_rect(
83 (self.position.x - ENEMY_W / 2, self.position.y - ENEMY_H / 2),
84 (ENEMY_W, ENEMY_H),
85 colour=(1.0, 0.4, 0.3, 1),
86 filled=True,
87 )
88
89
90class Spawner(Node2D):
91 """No drawing: just a timer that emits each new enemy for the root to place."""
92
93 min_delay = Property(0.4)
94 max_delay = Property(0.9)
95
96 spawned = Signal(Enemy) # carries the new Enemy node
97
98 def on_ready(self):
99 self._timer = 0.0
100
101 def on_update(self, dt: float):
102 self._timer -= dt
103 if self._timer <= 0:
104 self._timer = random.uniform(self.min_delay, self.max_delay)
105 self.spawned.emit(
106 Enemy(speed=random.uniform(150.0, 300.0), position=Vec2(random.uniform(20, WIDTH - 20), -ENEMY_H))
107 )
108
109
110class ScoreLabel(Node2D):
111 """Owns the score and its rendering; announces when it changes."""
112
113 score = Property(0)
114
115 changed = Signal(int)
116
117 def add(self, n: int = 1):
118 self.score += n
119 self.changed.emit(self.score)
120
121 def on_draw(self, renderer):
122 renderer.draw_text(f"Score: {self.score}", (20, 20), scale=2, colour=(1, 1, 1, 1))
123
124
125class DodgeGame(Node2D):
126 """Root: owns the graph (which nodes exist, which signals wire to which slots)
127 and the one thing only it can see, player-versus-enemy collision."""
128
129 input_actions = {
130 "left": [Key.A, Key.LEFT],
131 "right": [Key.D, Key.RIGHT],
132 "restart": [Key.ENTER, MouseButton.LEFT],
133 }
134
135 def on_ready(self):
136 # Death pauses the tree. The root has to keep ticking through that to
137 # notice the restart, so it opts out of the pause and the gameplay
138 # subtree opts back in (see ``_start``).
139 self.update_mode = UpdateMode.ALWAYS
140 self._start()
141
142 def _start(self):
143 self.alive = True
144 self.tree.paused = False
145 self.queue_redraw() # alive flipped (restart) -> clear the "Game Over" text from on_draw
146 # Everything that plays lives under one node, so one flag freezes it all
147 # and one destroy() tears the round down.
148 self.world = self.add_child(Node2D(name="World"))
149 self.world.update_mode = UpdateMode.PAUSABLE
150 self.player = self.world.add_child(Player(position=Vec2(WIDTH / 2, HEIGHT - 60)))
151 self.spawner = self.world.add_child(Spawner())
152 self.score = self.world.add_child(ScoreLabel())
153 self.spawner.spawned.connect(self._on_spawned)
154 self.score.changed.connect(self._on_score_changed)
155
156 def _on_spawned(self, enemy: Enemy):
157 # The root decides where the enemy lives, and pairs its escape to a point.
158 enemy.escaped.connect(lambda: self.score.add(1))
159 self.world.add_child(enemy)
160
161 def _on_score_changed(self, score: int):
162 # A rising score ramps the difficulty: the root tightens the spawn delay.
163 self.spawner.min_delay = max(0.12, 0.4 - score * 0.01)
164 self.spawner.max_delay = max(0.3, 0.9 - score * 0.02)
165
166 def on_update(self, dt: float):
167 if not self.alive:
168 if Input.is_action_just_pressed("restart"):
169 self.world.destroy()
170 self._start()
171 return
172 px, py = self.player.position
173 for child in list(self.world.children):
174 if (
175 isinstance(child, Enemy)
176 and abs(child.position.x - px) < (PLAYER_W + ENEMY_W) / 2
177 and abs(child.position.y - py) < (PLAYER_H + ENEMY_H) / 2
178 ):
179 self.alive = False
180 # One flag stops the spawner, the falling enemies, and the scoring:
181 # nothing keeps simulating behind the "Game Over" text.
182 self.tree.paused = True
183 self.queue_redraw() # alive flipped (death) -> the retained renderer redraws "Game Over"
184 return
185
186 def on_draw(self, renderer):
187 renderer.draw_rect((0, 0), (WIDTH, HEIGHT), colour=(0.05, 0.05, 0.08, 1), filled=True)
188 if not self.alive:
189 renderer.draw_text(
190 "Game Over: Enter or tap to restart", (WIDTH / 2 - 190, HEIGHT / 2), scale=2, colour=(1, 1, 0.3, 1)
191 )
192
193
194def _selftest() -> bool:
195 """Headless: play a round -- dodge, die, restart -- and check the structure holds.
196
197 The paddle is only ever steered by holding the pointer, which is one of the two
198 routes the Player reads, so the round is played rather than simulated. What is
199 checked is what the lesson teaches: that each signal has an emitter and a
200 listener at the other end, that one paused flag freezes the whole gameplay
201 subtree while the root keeps ticking, and that a restart rebuilds it.
202 """
203 from simvx.core.testing import InputSimulator
204 from simvx.graphics.testing import assert_not_blank, save_png
205
206 DODGE_UNTIL = 4 # points to score by dodging before deliberately colliding
207 FROZEN_FOR = 30 # frames to watch after death, to see that nothing moves
208 TOTAL = 1800
209
210 random.seed(5)
211 app = App(title="Monolith to Composed", width=WIDTH, height=HEIGHT, visible=False)
212 scene = DodgeGame(name="DodgeGame")
213 sim = InputSimulator()
214 seen: dict[str, object] = {}
215 pointing = False
216
217 def steer(x: float) -> None:
218 nonlocal pointing
219 sim.move_mouse(max(0.0, min(WIDTH, x)), HEIGHT - 60.0)
220 if not pointing:
221 sim.press_mouse(MouseButton.LEFT)
222 pointing = True
223
224 def let_go() -> None:
225 nonlocal pointing
226 if pointing:
227 sim.release_mouse(MouseButton.LEFT)
228 pointing = False
229
230 def enemies() -> list[Enemy]:
231 return [c for c in scene.world.children if isinstance(c, Enemy)]
232
233 def on_frame(idx: int, _t: float) -> bool:
234 if idx == 1:
235 seen["world_children"] = sorted(
236 {type(c).__name__ for c in scene.world.children if not isinstance(c, Enemy)}
237 )
238 seen["first_world"] = id(scene.world)
239 seen["delays_at_zero"] = (scene.spawner.min_delay, scene.spawner.max_delay)
240 seen["enemies_in_world"] = True
241
242 if idx > 0 and not all(e.parent is scene.world for e in enemies()):
243 seen["enemies_in_world"] = False
244
245 if scene.alive and "died_at" not in seen:
246 threats = [e for e in enemies() if e.position.y > HEIGHT / 2]
247 if scene.score.score < DODGE_UNTIL:
248 # Dodging: stand as far from the nearest threat as the court allows.
249 if threats:
250 nearest = min(threats, key=lambda e: abs(e.position.x - scene.player.position.x))
251 steer(0.0 if nearest.position.x > WIDTH / 2 else WIDTH)
252 else:
253 steer(scene.player.position.x)
254 else:
255 if "delays_after_dodging" not in seen:
256 seen["delays_after_dodging"] = (scene.spawner.min_delay, scene.spawner.max_delay)
257 seen["score_from_dodging"] = scene.score.score
258 # Now stand under one, on purpose.
259 if threats:
260 steer(float(min(threats, key=lambda e: -e.position.y).position.x))
261 elif "died_at" not in seen:
262 let_go()
263 seen["died_at"] = idx
264 seen["paused"] = scene.tree.paused
265 seen["score_at_death"] = scene.score.score
266 seen["frozen_from"] = [(id(e), float(e.position.y)) for e in enemies()]
267 elif idx == seen["died_at"] + FROZEN_FOR:
268 seen["frozen_to"] = [(id(e), float(e.position.y)) for e in enemies()]
269 seen["score_after_freeze"] = scene.score.score
270 sim.press_key(Key.ENTER)
271 elif idx == seen["died_at"] + FROZEN_FOR + 1:
272 sim.release_key(Key.ENTER)
273 seen["restarted"] = (scene.alive, scene.tree.paused, id(scene.world), scene.score.score)
274 return False
275 return True
276
277 frames = app.run_headless(scene, frames=TOTAL, on_frame=on_frame, capture_frames=[0])
278 assert_not_blank(frames[0])
279 save_png(frames[0], "/tmp/monolith_to_composed_test.png")
280
281 ok = True
282
283 def check(label: str, passed: bool, detail: str) -> None:
284 nonlocal ok
285 ok = ok and passed
286 print(f"{'ok ' if passed else 'FAIL'} {label}: {detail}")
287
288 check(
289 "the round is built from one node per job, all under a single World",
290 seen["world_children"] == ["Player", "ScoreLabel", "Spawner"] and seen["enemies_in_world"],
291 ", ".join(seen["world_children"]) + ", and the root files every spawned Enemy there too",
292 )
293 check(
294 "dodging an enemy scores a point, so escaped reaches the score through the root",
295 seen["score_from_dodging"] >= DODGE_UNTIL,
296 f"{seen['score_from_dodging']} points scored by letting enemies fall past the paddle",
297 )
298 before, after = seen["delays_at_zero"], seen["delays_after_dodging"]
299 check(
300 "and the rising score tightens the spawner, which is the other end of changed",
301 after[0] < before[0] and after[1] < before[1],
302 f"spawn delay {before[0]:.2f}..{before[1]:.2f}s at 0 points, {after[0]:.2f}..{after[1]:.2f}s "
303 f"at {seen['score_from_dodging']}",
304 )
305 check(
306 "being hit ends the round and pauses the tree",
307 seen["paused"],
308 f"tree.paused = {seen['paused']} on the frame the collision was seen, at frame {seen['died_at']}",
309 )
310 still = [
311 (y_from, y_to)
312 for (id_from, y_from), (id_to, y_to) in zip(seen["frozen_from"], seen["frozen_to"], strict=True)
313 if id_from == id_to
314 ]
315 check(
316 "one flag freezes the whole gameplay subtree: nothing falls and nothing scores",
317 bool(still) and all(a == b for a, b in still) and seen["score_after_freeze"] == seen["score_at_death"],
318 f"{len(still)} enemies held their y for {FROZEN_FOR} frames, score held at {seen['score_at_death']}",
319 )
320 alive, paused, world, score = seen["restarted"]
321 check(
322 "the root keeps ticking through the pause, so Enter restarts into a fresh round",
323 alive and not paused and world != seen["first_world"] and score == 0,
324 f"alive again, unpaused, a new World, score back to {score}",
325 )
326
327 print("screenshot: /tmp/monolith_to_composed_test.png")
328 print("SELFTEST:", "PASS" if ok else "FAIL")
329 return ok
330
331
332if __name__ == "__main__":
333 import sys
334
335 if "--test" in sys.argv:
336 sys.exit(0 if _selftest() else 1)
337 App(title="Monolith to Composed", width=WIDTH, height=HEIGHT).run(DodgeGame())