Lesson 27: Advanced Capstone: Build, Playtest, Ship

If you are stuck on what to build first, build the thing the player does every few seconds, with rectangles and circles, and nothing else. Open the sample solution below and find Drone Salvage's PlayScene.update: notice that it is just "read input, move things, check collisions, publish events". Your first version can be that simple.

Your capstone is your own game, so there is no single answer. This is the lab's sample capstone, Drone Salvage, built on the same skeleton with four Advanced systems. If your instructor hands you the lab file, you will see a few extra lines marked lab runtime near the top and and frame_budget() in the loop; they let the instructor's checker run it automatically and do nothing when you run it yourself.

"""Drone Salvage: Advanced Lesson 27 sample capstone (solution).

A small but complete game built the way the capstone brief asks: one core
loop (grab salvage, dodge drones), scenes, a fixed-timestep update, and four
Advanced systems working together:
  - an event bus that isolates failing listeners (Event Systems),
  - steering drones that seek the player and keep apart (Steering & Flocking),
  - dynamic difficulty in levels per second (Difficulty & DDA),
  - telemetry saved as CSV for playtests (Playtesting & Telemetry),
plus a versioned save file written atomically (Robust Saves).

Keys: ENTER starts, WASD or arrows fly, ESC returns to the title.
Close the window to quit.
"""
import csv
import json
import os
import random
import time
from collections import defaultdict, deque
from pathlib import Path

import pygame


WIDTH, HEIGHT = 960, 600
STEP = 1 / 120                  # fixed simulation step, seconds
MAX_FRAME = 0.25                # never simulate more than this per frame (after a stall)
SAVE_PATH = Path(__file__).parent / "drone_salvage_save.json"
LOG_DIR = Path(__file__).parent / "playtest_logs"
SAVE_VERSION = 2
PLAYER_SPEED = 260              # px/s
INVULNERABLE = 1.0              # seconds after a hit
TEXT = (235, 235, 235)
ACCENT = (120, 210, 255)


def clamp(v, lo, hi):
    return max(lo, min(hi, v))


# --- events (Event Systems) -----------------------------------------------------------------
class EventBus:
    def __init__(self):
        self.handlers = defaultdict(list)
        self.errors = 0

    def subscribe(self, name, handler):
        self.handlers[name].append(handler)

    def publish(self, name, **data):
        """Call every listener; one broken listener must not stop the others (or the game)."""
        for handler in list(self.handlers[name]):
            try:
                handler(**data)
            except Exception as exc:        # isolate listener bugs, but report them
                self.errors += 1
                print(f"listener error on {name!r}: {exc!r}")


# --- saves (Robust Saves) ----------------------------------------------------------------------
DEFAULT_SAVE = {"version": SAVE_VERSION, "best_score": 0, "runs": 0}


def migrate(data):
    """Upgrade old save layouts one version at a time."""
    if data.get("version", 1) == 1:              # version 1 was {"best": n}
        data = {"version": 2, "best_score": int(data.get("best", 0)), "runs": 0}
    return data


def load_save(path):
    """Missing or damaged saves fall back to defaults instead of crashing the game."""
    try:
        with open(path, encoding="utf-8") as f:
            data = json.load(f)
        if not isinstance(data, dict):           # valid JSON, but not a save (a list, a number...)
            return dict(DEFAULT_SAVE)
        data = migrate(data)
    except (OSError, ValueError, TypeError):
        return dict(DEFAULT_SAVE)
    if not (isinstance(data.get("best_score"), int) and isinstance(data.get("runs"), int)):
        return dict(DEFAULT_SAVE)
    return data


def write_save(path, data):
    """Write to a temporary file, then atomically replace the old save."""
    tmp = path.with_suffix(".tmp")
    with open(tmp, "w", encoding="utf-8") as f:
        json.dump(data, f)
    os.replace(tmp, path)


# --- telemetry (Playtesting & Telemetry) --------------------------------------------------------
class Telemetry:
    def __init__(self, tester="T01"):
        self.session = time.strftime("%Y%m%d-%H%M%S")
        self.tester = tester
        self.clock = 0.0
        self.rows = []

    def log(self, kind, **data):
        self.rows.append({"t": round(self.clock, 3), "session": self.session, "tester": self.tester,
                          "kind": kind, "data": json.dumps(data, sort_keys=True)})

    def save(self, folder):
        folder.mkdir(parents=True, exist_ok=True)
        path = folder / f"session_{self.session}_{self.tester}.csv"
        with open(path, "w", newline="", encoding="utf-8") as f:
            writer = csv.DictWriter(f, fieldnames=["t", "session", "tester", "kind", "data"])
            writer.writeheader()
            writer.writerows(self.rows)
        return path


# --- difficulty (Difficulty & DDA) -------------------------------------------------------------------
class Director:
    def __init__(self, level=3.0, target=0.6, gain=1.5, max_rate=0.4):
        self.level, self.target, self.gain, self.max_rate = level, target, gain, max_rate
        self.recent = deque()                   # (time, +1 salvage / -1 hit)
        self.clock = 0.0

    def record(self, value):
        self.recent.append((self.clock, value))

    def performance(self):
        while self.recent and self.clock - self.recent[0][0] > 20.0:
            self.recent.popleft()
        good = sum(1 for _, v in self.recent if v > 0)
        bad = sum(1 for _, v in self.recent if v < 0)
        return clamp(0.5 + 0.05 * good - 0.2 * bad, 0.0, 1.0)

    def update(self, dt):
        self.clock += dt
        error = self.performance() - self.target
        rate = clamp(self.gain * error, -self.max_rate, self.max_rate) if abs(error) > 0.05 else 0.0
        self.level = clamp(self.level + rate * dt, 1.0, 10.0)

    def params(self):
        mult = self.level / 5
        return {"drone_speed": 120 * mult ** 0.8, "drone_count": 2 + round(self.level),
                "salvage_value": max(1, round(10 * mult ** -0.3))}


# --- steering (Steering & Flocking) --------------------------------------------------------------------
def seek(pos, vel, target, max_speed, max_force):
    """Steering force toward target, limited to max_force (px/s^2)."""
    desired = target - pos
    if desired.length_squared() == 0:
        return pygame.Vector2()
    force = desired.normalize() * max_speed - vel
    if force.length() > max_force:
        force.scale_to_length(max_force)
    return force


def separation(pos, others, radius):
    """Push away from neighbors closer than radius; stronger when closer."""
    push = pygame.Vector2()
    for other in others:
        offset = pos - other
        d = offset.length()
        if 0 < d < radius:
            push += offset / d * (radius - d) / radius
    return push


class Drone:
    def __init__(self, pos):
        self.pos = pygame.Vector2(pos)
        self.vel = pygame.Vector2()

    def update(self, target, neighbors, speed, dt):
        force = seek(self.pos, self.vel, target, speed, 400) + separation(self.pos, neighbors, 60) * 600
        self.vel += force * dt
        if self.vel.length() > speed:
            self.vel.scale_to_length(speed)
        self.pos += self.vel * dt


# --- scenes (Game States & Scenes) --------------------------------------------------------------------------
class TitleScene:
    def __init__(self, game):
        self.game = game

    def handle(self, event):
        if event.type == pygame.KEYDOWN and event.key == pygame.K_RETURN:
            self.game.change(PlayScene(self.game))

    def update(self, dt):
        pass

    def draw(self, screen):
        screen.fill((16, 20, 30))
        g = self.game
        title = g.big.render("DRONE SALVAGE", True, ACCENT)
        screen.blit(title, title.get_rect(center=(WIDTH / 2, HEIGHT / 2 - 40)))
        info = g.font.render(f"ENTER to start   best {g.save['best_score']}   runs {g.save['runs']}", True, TEXT)
        screen.blit(info, info.get_rect(center=(WIDTH / 2, HEIGHT / 2 + 20)))


class PlayScene:
    def __init__(self, game, seed=None):
        self.game = game
        self.rng = random.Random(seed)
        self.player = pygame.Vector2(WIDTH / 2, HEIGHT / 2)
        self.move = pygame.Vector2()
        self.shield = 3
        self.invulnerable = 0.0
        self.score = 0
        self.drones = []
        self.salvage = [self.random_point() for _ in range(4)]
        self.director = Director()
        game.bus.publish("run_started")

    def random_point(self):
        return pygame.Vector2(self.rng.uniform(40, WIDTH - 40), self.rng.uniform(60, HEIGHT - 40))

    def handle(self, event):
        if event.type == pygame.KEYDOWN and event.key == pygame.K_ESCAPE:
            self.game.change(TitleScene(self.game))

    def update(self, dt):
        g = self.game
        self.director.update(dt)
        p = self.director.params()
        held = g.held
        self.move.update((pygame.K_RIGHT in held or pygame.K_d in held) - (pygame.K_LEFT in held or pygame.K_a in held),
                         (pygame.K_DOWN in held or pygame.K_s in held) - (pygame.K_UP in held or pygame.K_w in held))
        if self.move.length_squared() > 0:
            self.player += self.move.normalize() * PLAYER_SPEED * dt
        self.player.x = clamp(self.player.x, 12, WIDTH - 12)
        self.player.y = clamp(self.player.y, 12, HEIGHT - 12)

        while len(self.drones) < p["drone_count"]:
            edge = self.rng.choice([(0, self.rng.uniform(0, HEIGHT)), (WIDTH, self.rng.uniform(0, HEIGHT))])
            self.drones.append(Drone(edge))
        positions = [d.pos for d in self.drones]
        for d in self.drones:
            d.update(self.player, positions, p["drone_speed"], dt)

        for s in list(self.salvage):
            if s.distance_to(self.player) < 22:
                self.salvage.remove(s)
                self.salvage.append(self.random_point())
                self.score += p["salvage_value"]
                self.director.record(+1)
                g.bus.publish("salvage", value=p["salvage_value"], x=round(s.x), y=round(s.y))

        self.invulnerable = max(0.0, self.invulnerable - dt)
        if self.invulnerable == 0 and any(d.pos.distance_to(self.player) < 24 for d in self.drones):
            self.shield -= 1
            self.invulnerable = INVULNERABLE
            self.director.record(-1)
            g.bus.publish("hit", shield=self.shield, level=round(self.director.level, 2))
            if self.shield <= 0:
                g.bus.publish("game_over", score=self.score)
                g.change(GameOverScene(g, self.score))

    def draw(self, screen):
        g = self.game
        screen.fill((12, 16, 26))
        for s in self.salvage:
            pygame.draw.rect(screen, (240, 200, 80), (s.x - 7, s.y - 7, 14, 14), border_radius=3)
        for d in self.drones:
            pygame.draw.circle(screen, (230, 80, 90), d.pos, 11)
        blink = self.invulnerable > 0 and int(self.invulnerable * 10) % 2 == 0
        if not blink:
            pygame.draw.circle(screen, ACCENT, self.player, 12)
        hud = g.font.render(f"score {self.score}   shield {self.shield}   level {self.director.level:.1f}", True, TEXT)
        screen.blit(hud, (12, 10))


class GameOverScene:
    def __init__(self, game, score):
        self.game = game
        self.score = score
        game.save["runs"] += 1
        self.new_best = score > game.save["best_score"]
        if self.new_best:
            game.save["best_score"] = score
        write_save(game.save_path, game.save)

    def handle(self, event):
        if event.type == pygame.KEYDOWN and event.key == pygame.K_RETURN:
            self.game.change(PlayScene(self.game))

    def update(self, dt):
        pass

    def draw(self, screen):
        g = self.game
        screen.fill((26, 14, 20))
        lines = ["GAME OVER", f"score {self.score}" + ("   NEW BEST!" if self.new_best else ""), "ENTER to fly again"]
        for i, text in enumerate(lines):
            img = (g.big if i == 0 else g.font).render(text, True, TEXT)
            screen.blit(img, img.get_rect(center=(WIDTH / 2, HEIGHT / 2 - 50 + i * 50)))


class Game:
    def __init__(self, save_path=SAVE_PATH):
        self.bus = EventBus()
        self.telemetry = Telemetry()
        self.save_path = save_path
        self.save = load_save(save_path)
        self.held = set()
        self.font = pygame.font.Font(None, 28)      # fonts created once, shared by every scene
        self.big = pygame.font.Font(None, 72)
        self.accumulator = 0.0
        for name in ("run_started", "salvage", "hit", "game_over"):
            self.bus.subscribe(name, lambda _n=name, **data: self.telemetry.log(_n, **data))
        self.scene = TitleScene(self)

    def change(self, scene):
        self.scene = scene

    def advance(self, dt):
        """Fixed timestep: run whole STEPs of simulation for the real time that passed."""
        self.accumulator = min(self.accumulator + dt, MAX_FRAME)
        steps = 0
        while self.accumulator >= STEP:
            self.telemetry.clock += STEP
            self.scene.update(STEP)
            self.accumulator -= STEP
            steps += 1
        return steps


def main():
    pygame.init()
    screen = pygame.display.set_mode((WIDTH, HEIGHT))
    pygame.display.set_caption("Drone Salvage")
    clock = pygame.time.Clock()
    game = Game()

    running = True
    while running:
        dt = clock.tick(60) / 1000
        for event in pygame.event.get():
            if event.type == pygame.QUIT:
                running = False
            elif event.type == pygame.KEYDOWN:
                game.held.add(event.key)
            elif event.type == pygame.KEYUP:
                game.held.discard(event.key)
            game.scene.handle(event)
        game.advance(dt)
        game.scene.draw(screen)
        pygame.display.flip()

    pygame.quit()
    log = game.telemetry.save(LOG_DIR)
    print(f"Telemetry: {len(game.telemetry.rows)} events in {log.name}")
    print(f"Best score: {game.save['best_score']}  runs: {game.save['runs']}")
    print("Studio closed cleanly.")


if __name__ == "__main__":
    main()