afterglow/view/effects.py¶
Part of Afterglow.
1"""Game-feel (juice) layer for Afterglow: particles, screenshake, hitstop, SFX.
2
3The game owns one :class:`Effects` instance and forwards ``room.drain_events()``
4to :meth:`Effects.play_events` each frame; everything else (one-shot particle
5bursts, camera shake, hitstop, sound) is driven from that single call. The view
6may also fire individual effects directly (e.g. a one-off ``screenshake``).
7
8Coordinate contract
9-------------------
10The sim is y-DOWN logical pixels; the diorama maps a logical point ``(x, y)`` to
11world ``(x * S, -y * S, 0)`` with ``S = world_scale = 1 / TILE_SIZE``. All public
12methods that take a *world* position expect post-mapped world coordinates;
13:meth:`play_events` does the mapping itself from each event's logical centre, so
14the game forwards raw sim events unchanged.
15
16Reduced motion
17--------------
18Construct with ``reduced_motion=True`` (photosensitive / accessibility mode) to
19damp or drop the loud effects: screenshake amplitude is scaled toward zero,
20hitstop is shortened, and particle bursts emit fewer, dimmer particles. The flag
21is honoured by every effect, so a single switch covers the whole layer.
22
23Headless safety
24---------------
25Importing and constructing :class:`Effects` never touches the GPU. Particle
26nodes are only created once :meth:`attach` wires in a live scene root, and audio
27players are only spawned when a scene tree is present. With no root attached the
28class is an inert no-op, so logic tests and ``run_headless`` stay clean.
29"""
30
31from __future__ import annotations
32
33import logging
34
35from afterglow.assets import audio
36from simvx.core import GPUParticles3D
37
38log = logging.getLogger(__name__)
39
40__all__ = ["Effects"]
41
42# Event tag -> SFX clip name (audio.get_sfx). Tags without a sound are omitted.
43_EVENT_SFX = {
44 "jump": "jump",
45 "dash": "dash",
46 "crystal": "crystal_shatter",
47 "orb": "glow_orb",
48 "shard": "shard",
49 "land": "land",
50 "spring": "spring",
51 "death": "death",
52 "win": "menu_confirm",
53}
54
55
56class Effects:
57 """Owns the juice: particle bursts, screenshake, hitstop, and SFX.
58
59 Args:
60 reduced_motion: Accessibility / photosensitive-safe mode. Scales shake
61 and hitstop toward zero and thins particle bursts.
62 screenshake_enabled: Master switch for camera shake. ``False`` disables
63 all shake regardless of ``reduced_motion``.
64 ambient_motes: Spawn the continuous floating-mote atmosphere on
65 :meth:`attach`. Off in reduced-motion unless forced.
66 """
67
68 def __init__(
69 self,
70 *,
71 reduced_motion: bool = False,
72 screenshake_enabled: bool = True,
73 ambient_motes: bool = True,
74 ) -> None:
75 self.reduced_motion = reduced_motion
76 self.screenshake_enabled = screenshake_enabled and not reduced_motion
77 self._want_motes = ambient_motes and not reduced_motion
78
79 self._root = None # scene node bursts are parented under
80 self._camera = None # Node3D shaken by screenshake
81 self._app = None # App, for time_scale (hitstop)
82
83 # Burst pool: kind -> list[GPUParticles3D]. Reused round-robin so a rapid
84 # series of events never allocates per hit.
85 self._pool: dict[str, list[GPUParticles3D]] = {}
86 self._pool_next: dict[str, int] = {}
87 self._motes: GPUParticles3D | None = None
88 self._hitstop_handle = None
89
90 # -- wiring --------------------------------------------------------------
91
92 def attach(self, scene_root, camera, app) -> None:
93 """Wire the effects layer into a live scene.
94
95 Args:
96 scene_root: Node particle bursts and audio players are parented to.
97 camera: Node3D that screenshake displaces (its ``position`` is
98 punched and restored). May be ``None`` to disable shake.
99 app: The running ``App`` (for ``time_scale`` hitstop). May be
100 ``None`` headlessly; hitstop then degrades to a no-op.
101 """
102 self._root = scene_root
103 self._camera = camera
104 self._app = app
105 if self._want_motes:
106 self._spawn_motes()
107
108 def reset(self) -> None:
109 """Drop all pooled emitters + ambient motes so the next attach rebuilds.
110
111 ``Effects`` outlives any single room (the game owns one instance for the
112 whole session), but the scene it parents nodes under is rebuilt per room
113 (``scene.clear_children()`` destroys every pooled emitter and the motes
114 node). Without this, the warm pool and ``self._motes`` keep referencing
115 now-detached nodes: ``_spawn_motes`` early-returns (no ambient motes after
116 room 1) and ``_burst`` re-arms orphaned emitters under the old scene (no
117 particle bursts after room 1). Call before re-:meth:`attach` per room.
118 """
119 self._pool.clear()
120 self._pool_next.clear()
121 self._motes = None
122 self._hitstop_handle = None
123
124 @property
125 def attached(self) -> bool:
126 return self._root is not None
127
128 # -- particle factory ----------------------------------------------------
129
130 def _make_emitter(self, **props) -> GPUParticles3D:
131 p = GPUParticles3D(**props)
132 self._root.add_child(p)
133 return p
134
135 def _scale_amount(self, amount: int) -> int:
136 """Thin bursts in reduced-motion so fewer particles flash at once."""
137 return max(8, amount // 3) if self.reduced_motion else amount
138
139 def _burst(self, kind: str, pos, builder) -> GPUParticles3D | None:
140 """Fetch (or lazily build) a pooled one-shot emitter, move it, replay.
141
142 ``builder`` is called once per pool slot to configure a fresh emitter;
143 on reuse the slot is only repositioned and restarted, so per-hit cost is
144 a transform write plus a compute re-arm.
145 """
146 if self._root is None:
147 return None
148 slots = self._pool.setdefault(kind, [])
149 idx = self._pool_next.get(kind, 0)
150 if idx < len(slots):
151 p = slots[idx]
152 else:
153 p = builder()
154 slots.append(p)
155 self._pool_next[kind] = (idx + 1) % max(1, _POOL_SIZE.get(kind, 4))
156 # Trim the round-robin index back into range once the pool is warm.
157 if len(slots) >= _POOL_SIZE.get(kind, 4):
158 self._pool_next[kind] %= len(slots)
159 p.position = (float(pos[0]), float(pos[1]), float(pos[2]))
160 p.restart()
161 return p
162
163 # -- public burst effects (world-space positions) ------------------------
164
165 def dash_trail(self, pos, direction) -> None:
166 """Streak of motion particles trailing a dash in world ``direction``."""
167 dx, dy = float(direction[0]), float(direction[1])
168 mag = (dx * dx + dy * dy) ** 0.5 or 1.0
169 # Emit opposite the travel direction so the trail lags behind the wisp.
170 d = (-dx / mag, -dy / mag, 0.0)
171
172 def build() -> GPUParticles3D:
173 return self._make_emitter(
174 amount=self._scale_amount(48),
175 lifetime=0.32,
176 one_shot=True,
177 emitting=False,
178 speed=2.2,
179 speed_variance=0.8,
180 direction=d,
181 spread=0.5,
182 gravity=(0.0, 0.0, 0.0),
183 damping=4.0,
184 emission_shape="sphere",
185 emission_radius=0.06,
186 start_colour=(0.75, 0.95, 1.0, 0.9),
187 end_colour=(0.4, 0.7, 1.0, 0.0),
188 start_scale=0.45,
189 end_scale=0.0,
190 explosiveness=0.6,
191 )
192
193 p = self._burst("dash_trail", pos, build)
194 if p is not None:
195 p.direction = d
196
197 def crystal_shatter(self, pos, colour=(0.5, 1.0, 0.6, 1.0)) -> None:
198 """Sharp radial shard burst when a crystal is shattered (its colour)."""
199 c = _rgba(colour)
200
201 def build() -> GPUParticles3D:
202 return self._make_emitter(
203 amount=self._scale_amount(96),
204 lifetime=0.55,
205 one_shot=True,
206 emitting=False,
207 speed=4.5,
208 speed_variance=2.0,
209 direction=(0.0, 1.0, 0.0),
210 spread=10.0,
211 gravity=(0.0, -3.0, 0.0),
212 damping=1.5,
213 emission_shape="point",
214 start_colour=c,
215 end_colour=(c[0], c[1], c[2], 0.0),
216 start_scale=0.4,
217 end_scale=0.05,
218 explosiveness=1.0,
219 )
220
221 p = self._burst("crystal_shatter", pos, build)
222 if p is not None:
223 p.start_colour = c
224 p.end_colour = (c[0], c[1], c[2], 0.0)
225
226 def orb_sparkle(self, pos) -> None:
227 """Gentle warm bloom of rising sparkles when a glow orb is taken."""
228
229 def build() -> GPUParticles3D:
230 return self._make_emitter(
231 amount=self._scale_amount(60),
232 lifetime=0.9,
233 one_shot=True,
234 emitting=False,
235 speed=1.2,
236 speed_variance=0.6,
237 direction=(0.0, 1.0, 0.0),
238 spread=3.0,
239 gravity=(0.0, 0.8, 0.0),
240 damping=2.0,
241 emission_shape="sphere",
242 emission_radius=0.15,
243 start_colour=(1.0, 0.95, 0.6, 0.95),
244 end_colour=(1.0, 0.7, 0.3, 0.0),
245 start_scale=0.3,
246 end_scale=0.0,
247 explosiveness=0.7,
248 )
249
250 self._burst("orb_sparkle", pos, build)
251
252 def landing_dust(self, pos) -> None:
253 """Low puff of dust kicked sideways when the player lands."""
254
255 def build() -> GPUParticles3D:
256 return self._make_emitter(
257 amount=self._scale_amount(40),
258 lifetime=0.4,
259 one_shot=True,
260 emitting=False,
261 speed=2.0,
262 speed_variance=0.8,
263 direction=(1.0, 0.0, 0.0),
264 spread=8.0,
265 gravity=(0.0, -1.0, 0.0),
266 damping=5.0,
267 emission_shape="box",
268 emission_box=(0.25, 0.02, 0.05),
269 start_colour=(0.85, 0.85, 0.8, 0.7),
270 end_colour=(0.7, 0.7, 0.65, 0.0),
271 start_scale=0.35,
272 end_scale=0.0,
273 explosiveness=0.9,
274 )
275
276 self._burst("landing_dust", pos, build)
277
278 def death_scatter(self, pos) -> None:
279 """Wisp dissolves: a full-sphere scatter of fading embers on death."""
280
281 def build() -> GPUParticles3D:
282 return self._make_emitter(
283 amount=self._scale_amount(140),
284 lifetime=0.8,
285 one_shot=True,
286 emitting=False,
287 speed=5.0,
288 speed_variance=2.5,
289 direction=(0.0, 1.0, 0.0),
290 spread=10.0,
291 gravity=(0.0, -2.0, 0.0),
292 damping=1.0,
293 emission_shape="sphere",
294 emission_radius=0.1,
295 start_colour=(0.8, 0.9, 1.0, 1.0),
296 end_colour=(0.3, 0.5, 0.9, 0.0),
297 start_scale=0.5,
298 end_scale=0.0,
299 explosiveness=1.0,
300 )
301
302 self._burst("death_scatter", pos, build)
303
304 def win_burst(self, pos) -> None:
305 """Celebratory fountain of bright motes when a room is cleared."""
306
307 def build() -> GPUParticles3D:
308 return self._make_emitter(
309 amount=self._scale_amount(180),
310 lifetime=1.4,
311 one_shot=True,
312 emitting=False,
313 speed=4.0,
314 speed_variance=2.0,
315 direction=(0.0, 1.0, 0.0),
316 spread=4.0,
317 gravity=(0.0, -2.5, 0.0),
318 damping=0.6,
319 emission_shape="sphere",
320 emission_radius=0.2,
321 start_colour=(1.0, 0.95, 0.7, 1.0),
322 end_colour=(1.0, 0.6, 0.9, 0.0),
323 start_scale=0.45,
324 end_scale=0.0,
325 explosiveness=0.85,
326 )
327
328 self._burst("win_burst", pos, build)
329
330 def spring_pop(self, pos) -> None:
331 """Small upward pop when the player hits a spring."""
332
333 def build() -> GPUParticles3D:
334 return self._make_emitter(
335 amount=self._scale_amount(36),
336 lifetime=0.5,
337 one_shot=True,
338 emitting=False,
339 speed=3.0,
340 speed_variance=1.0,
341 direction=(0.0, 1.0, 0.0),
342 spread=2.0,
343 gravity=(0.0, -2.0, 0.0),
344 damping=2.0,
345 emission_shape="point",
346 start_colour=(0.7, 1.0, 0.85, 0.9),
347 end_colour=(0.4, 0.9, 0.7, 0.0),
348 start_scale=0.35,
349 end_scale=0.0,
350 explosiveness=1.0,
351 )
352
353 self._burst("spring_pop", pos, build)
354
355 # -- ambient atmosphere --------------------------------------------------
356
357 def _spawn_motes(self) -> None:
358 """Continuous slow-drifting dust motes for room atmosphere."""
359 if self._root is None or self._motes is not None:
360 return
361 self._motes = self._make_emitter(
362 amount=self._scale_amount(120),
363 lifetime=6.0,
364 one_shot=False,
365 emitting=True,
366 speed=0.18,
367 speed_variance=0.1,
368 direction=(0.0, 1.0, 0.0),
369 spread=6.0,
370 gravity=(0.0, 0.05, 0.0),
371 damping=0.2,
372 emission_shape="box",
373 emission_box=(6.0, 4.0, 1.5),
374 start_colour=(0.9, 0.95, 1.0, 0.0),
375 end_colour=(0.9, 0.95, 1.0, 0.0),
376 start_scale=0.06,
377 end_scale=0.1,
378 )
379
380 def set_ambient_centre(self, pos) -> None:
381 """Recentre the ambient motes (call once per room on the room centre)."""
382 if self._motes is not None:
383 self._motes.position = (float(pos[0]), float(pos[1]), float(pos[2]))
384
385 def set_ambient_enabled(self, enabled: bool) -> None:
386 if self._motes is not None:
387 self._motes.emitting = bool(enabled)
388
389 # -- camera shake --------------------------------------------------------
390
391 def screenshake(self, amplitude: float, duration: float, *, frequency: float = 22.0) -> None:
392 """Damped-sine screen shake via the camera's own offset (arcade feel).
393
394 No-op when shake is disabled, in reduced-motion, or no camera is
395 attached. Amplitude is in world units. Delegates to ``GameCamera.shake``,
396 which perturbs only the framed eye's X/Y so the camera keeps its
397 pull-back Z (punching ``.position`` directly truncated the 3D eye to a
398 Vec2 and zeroed Z, blanking the whole view for the shake's duration).
399 """
400 if not self.screenshake_enabled or self._camera is None or duration <= 0.0:
401 return
402 amp = float(amplitude)
403 if amp <= 0.0:
404 return
405 shake = getattr(self._camera, "shake", None)
406 if shake is not None:
407 shake(amp, duration, frequency=frequency, decay=9.0)
408
409 # -- hitstop -------------------------------------------------------------
410
411 def hitstop(self, duration: float, *, scale: float = 0.0) -> None:
412 """Briefly slow/freeze time, then auto-restore ``App.time_scale = 1``.
413
414 Args:
415 duration: Real-time seconds the slowdown lasts.
416 scale: Time scale during the freeze (``0`` = full freeze, ``0.1`` =
417 slow-mo). Reduced-motion shortens the duration.
418
419 Safe to call repeatedly: a new hitstop replaces any in-flight one and
420 the restore is driven by a coroutine on the camera (always restoring to
421 ``1.0`` even if overlapping calls occur). A no-op without an app.
422 """
423 app = self._app
424 if app is None or duration <= 0.0:
425 return
426 dur = duration * (0.5 if self.reduced_motion else 1.0)
427 s = max(0.0, min(1.0, float(scale)))
428 # Reduced-motion never hard-freezes (avoids a jarring stutter).
429 if self.reduced_motion:
430 s = max(s, 0.35)
431
432 host = self._camera or self._root
433 if host is None:
434 return
435 if self._hitstop_handle is not None:
436 host.stop_coroutine(self._hitstop_handle)
437 self._hitstop_handle = None
438 self._hitstop_handle = host.start_coroutine(self._hitstop_co(app, s, dur))
439
440 def _hitstop_co(self, app, scale: float, duration: float):
441 # Coroutines are ticked every frame regardless of time_scale, but the
442 # yielded dt is the SCALED dt: at scale 0 (full freeze) it is 0, so the
443 # timer must run on the wall clock or the freeze would never lift.
444 import time as _time
445
446 app.time_scale = scale
447 end = _time.perf_counter() + duration
448 try:
449 while _time.perf_counter() < end:
450 yield
451 finally:
452 app.time_scale = 1.0
453 self._hitstop_handle = None
454
455 # -- event router --------------------------------------------------------
456
457 def play_events(self, events, world_scale: float) -> None:
458 """Drive every effect + SFX from a frame of sim events.
459
460 Forward ``room.drain_events()`` here each frame. Each event's logical
461 centre ``(cx, cy)`` is mapped to world ``(cx * S, -cy * S, 0)`` and the
462 matching particle burst / shake / hitstop / sound is fired. Unknown
463 tags are ignored; ``win`` has no position and bursts at the origin.
464 """
465 s = float(world_scale)
466 for ev in events:
467 tag = ev[0]
468 self._play_sfx(tag)
469 if tag == "win":
470 self.win_burst((0.0, 0.0, 0.0))
471 self.screenshake(0.05, 0.5, frequency=12.0)
472 continue
473 if len(ev) >= 3:
474 pos = (ev[1] * s, -ev[2] * s, 0.0)
475 else:
476 pos = (0.0, 0.0, 0.0)
477 self._dispatch(tag, ev, pos, s)
478
479 def _dispatch(self, tag: str, ev, pos, s: float) -> None:
480 if tag == "dash":
481 # After the usual (cx, cy) the dash event carries its unit vector in
482 # y-DOWN sim space; flip y for world. The trail streams from the
483 # player's own centre, which ``pos`` already holds.
484 ux = ev[3] if len(ev) >= 5 else 0.0
485 uy = ev[4] if len(ev) >= 5 else 0.0
486 self.dash_trail(pos, (float(ux), -float(uy)))
487 self.screenshake(0.025, 0.12)
488 elif tag == "crystal":
489 self.crystal_shatter(pos)
490 self.hitstop(0.06)
491 self.screenshake(0.06, 0.22)
492 elif tag == "orb":
493 self.orb_sparkle(pos)
494 elif tag == "shard":
495 self.orb_sparkle(pos)
496 self.screenshake(0.03, 0.18, frequency=14.0)
497 elif tag == "land":
498 self.landing_dust(pos)
499 elif tag == "spring":
500 self.spring_pop(pos)
501 self.screenshake(0.02, 0.12)
502 elif tag == "death":
503 self.death_scatter(pos)
504 self.hitstop(0.12)
505 self.screenshake(0.09, 0.35)
506
507 # -- audio ---------------------------------------------------------------
508
509 def _play_sfx(self, tag: str) -> None:
510 name = _EVENT_SFX.get(tag)
511 if name is None or self._root is None or self._root.tree is None:
512 return # no clip for tag, or root not in a live tree (skip audio)
513 from simvx.core import AudioPlayer
514
515 player = AudioPlayer(stream=audio.get_sfx(name), bus=audio.SFX_BUS)
516 player.autoplay = True
517 player.queue_free_on_end = True
518 self._root.add_child(player)
519
520
521# Round-robin pool size per burst kind: how many concurrent emitters of a kind
522# can overlap before the oldest is reused. Frequent effects get more slots.
523_POOL_SIZE = {
524 "dash_trail": 4,
525 "crystal_shatter": 4,
526 "orb_sparkle": 3,
527 "landing_dust": 4,
528 "death_scatter": 2,
529 "win_burst": 1,
530 "spring_pop": 3,
531}
532
533
534def _rgba(colour) -> tuple[float, float, float, float]:
535 """Normalise a 3- or 4-tuple colour (0..1 floats) to RGBA, alpha default 1."""
536 c = tuple(float(v) for v in colour)
537 if len(c) == 3:
538 return (c[0], c[1], c[2], 1.0)
539 return (c[0], c[1], c[2], c[3])