Lesson 26: Playtesting & Telemetry

For to-do 2, csv.DictReader gives you strings: convert t with float() and data with json.loads(). For to-do 5, the condition is "for every task number i from 1 to k, this tester has a task_success for i": that's an all(...) around an any(...).

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 the program automatically; when you run it yourself they do nothing.

"""Instrumented Playtest: Advanced Lesson 26 practice exercise (solution).

A tiny three-task test level wired for telemetry. The tester walks to the
glowing pad and presses SPACE to activate it; one decoy pad looks almost the
same. Every press is logged as a task_attempt (with task_success or
task_fail), time spent per map cell builds a heatmap, and a path is sampled
every 0.2 s. When the window closes, the events are written with the csv
module, read back, and analyzed with empty-safe metrics.

Keys: WASD or arrows move, SPACE activates, H toggles the heatmap, P the path.
"""
import csv
import json
import random
import statistics
import time
from collections import Counter
from dataclasses import dataclass, field
from pathlib import Path

import pygame


WIDTH, HEIGHT = 800, 520
CELL = 40                      # heatmap cell size, px
SAMPLE_EVERY = 0.2             # seconds between path samples
SPEED = 220                    # px/s
LOG_DIR = Path(__file__).parent / "playtest_logs"
FIELDS = ["t", "session", "tester", "kind", "data"]
TEXT = (235, 235, 235)


@dataclass
class Event:
    t: float                   # seconds since the session started
    session: str
    tester: str
    kind: str                  # task_start, task_attempt, task_success, task_fail, session_end
    data: dict = field(default_factory=dict)


class Telemetry:
    """Three channels: events (what happened), heat (where time was spent), path (in what order)."""
    def __init__(self, session, tester):
        self.session = session
        self.tester = tester
        self.clock = 0.0
        self.events = []
        self.heat = Counter()          # (cx, cy) -> seconds spent in that cell
        self.path = []                 # (t, x, y) every SAMPLE_EVERY seconds
        self._since_sample = SAMPLE_EVERY

    def tick(self, pos, dt):
        self.clock += dt
        self.heat[(int(pos.x // CELL), int(pos.y // CELL))] += dt
        self._since_sample += dt
        if self._since_sample >= SAMPLE_EVERY:
            self._since_sample = 0.0
            self.path.append((round(self.clock, 2), round(pos.x), round(pos.y)))

    def log(self, kind, **data):
        self.events.append(Event(round(self.clock, 3), self.session, self.tester, kind, data))


def write_events(path, events):
    """Write events with csv.DictWriter: commas, quotes and newlines in the data are escaped for us."""
    path.parent.mkdir(parents=True, exist_ok=True)
    with open(path, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=FIELDS)
        writer.writeheader()
        for e in events:
            writer.writerow({"t": e.t, "session": e.session, "tester": e.tester, "kind": e.kind,
                             "data": json.dumps(e.data, sort_keys=True)})


def read_events(path):
    with open(path, newline="", encoding="utf-8") as f:
        return [Event(float(r["t"]), r["session"], r["tester"], r["kind"], json.loads(r["data"]))
                for r in csv.DictReader(f)]


# --- analysis: every function is safe with no data --------------------------------------
def by_tester(events):
    groups = {}
    for e in events:
        groups.setdefault(e.tester, []).append(e)
    return groups


def completion_rate(events, n_tasks):
    """Share of testers who finished every task, or None when there are no testers."""
    groups = by_tester(events)
    if not groups:
        return None
    done = sum(1 for evs in groups.values()
               if len({e.data.get("task") for e in evs if e.kind == "task_success"}) >= n_tasks)
    return done / len(groups)


def task_stats(events):
    """task -> (successes, attempts, success rate or None)."""
    stats = {}
    for e in events:
        if e.kind in ("task_attempt", "task_success"):
            s, a = stats.get(e.data["task"], (0, 0))
            stats[e.data["task"]] = (s + (e.kind == "task_success"), a + (e.kind == "task_attempt"))
    return {task: (s, a, s / a if a else None) for task, (s, a) in stats.items()}


def funnel(events, n_tasks):
    """How many testers completed task 1, then 1 and 2, ... Never above 100% and never rising."""
    groups = by_tester(events)
    rows = []
    for k in range(1, n_tasks + 1):
        count = sum(1 for evs in groups.values()
                    if all(any(e.kind == "task_success" and e.data.get("task") == i for e in evs)
                           for i in range(1, k + 1)))
        rows.append((k, count, 100 * count / len(groups) if groups else None))
    return rows


def hotspots(heat, top=3):
    return heat.most_common(top)


def sus_score(responses):
    """System Usability Scale: 10 answers from 1 to 5; odd items positive, even items negative."""
    if len(responses) != 10 or any(r not in (1, 2, 3, 4, 5) for r in responses):
        raise ValueError("SUS needs exactly 10 answers, each from 1 to 5")
    total = sum((r - 1) if i % 2 == 0 else (5 - r) for i, r in enumerate(responses))
    return total * 2.5


def cohens_d(a, b):
    """Difference in means divided by the pooled standard deviation."""
    if len(a) < 2 or len(b) < 2:
        return None
    pooled = (((len(a) - 1) * statistics.variance(a) + (len(b) - 1) * statistics.variance(b))
              / (len(a) + len(b) - 2)) ** 0.5
    return (statistics.mean(a) - statistics.mean(b)) / pooled if pooled else None


def permutation_p(a, b, rounds=5000, seed=1):
    """Two-sided permutation test on the difference in means (standard library only).

    Shuffle the group labels many times and count how often a difference at least
    as large as the real one appears by chance."""
    if not a or not b:
        return None
    rng = random.Random(seed)
    observed = abs(statistics.mean(a) - statistics.mean(b))
    pooled = list(a) + list(b)
    hits = 0
    for _ in range(rounds):
        rng.shuffle(pooled)
        diff = abs(statistics.mean(pooled[:len(a)]) - statistics.mean(pooled[len(a):]))
        hits += diff >= observed - 1e-12
    return (hits + 1) / (rounds + 1)


def compare(name_a, a, name_b, b, alpha=0.05):
    """An honest one-line verdict: a difference is only claimed when the test supports it."""
    if len(a) < 2 or len(b) < 2:
        return "Not enough data to compare."
    p = permutation_p(a, b)
    d = cohens_d(a, b)
    ma, mb = statistics.mean(a), statistics.mean(b)
    summary = f"{name_a} mean {ma:.1f} vs {name_b} mean {mb:.1f}, d = {d:.2f}, p = {p:.3f}"
    if p < alpha:
        faster = name_a if ma < mb else name_b
        return f"{summary}: {faster} is faster in this sample."
    return f"{summary}: no clear difference; test more players before deciding."


# --- the test level ------------------------------------------------------------------------
PADS = {1: pygame.FRect(660, 60, 70, 70), 2: pygame.FRect(80, 380, 70, 70), 3: pygame.FRect(370, 220, 70, 70)}
DECOY = pygame.FRect(170, 380, 70, 70)       # looks like pad 2: a deliberate usability trap


class PlaytestSession:
    def __init__(self, tester="T01"):
        self.telemetry = Telemetry(session=time.strftime("%Y%m%d-%H%M%S"), tester=tester)
        self.pos = pygame.Vector2(100, 100)
        self.task = 1
        self.telemetry.log("task_start", task=1)

    @property
    def finished(self):
        return self.task > len(PADS)

    def update(self, move, dt):
        if move.length_squared() > 0:
            self.pos += move.normalize() * SPEED * dt
            self.pos.x = max(10.0, min(WIDTH - 10.0, self.pos.x))
            self.pos.y = max(10.0, min(HEIGHT - 10.0, self.pos.y))
        self.telemetry.tick(self.pos, dt)

    def action(self):
        """SPACE: every press is an attempt; it either succeeds or fails with a reason."""
        if self.finished:
            return
        t = self.telemetry
        t.log("task_attempt", task=self.task, x=round(self.pos.x), y=round(self.pos.y))
        if PADS[self.task].collidepoint(self.pos):
            t.log("task_success", task=self.task)
            self.task += 1
            if not self.finished:
                t.log("task_start", task=self.task)
        else:
            on_other = DECOY.collidepoint(self.pos) or any(r.collidepoint(self.pos) for r in PADS.values())
            t.log("task_fail", task=self.task, reason="wrong pad" if on_other else "not on a pad")

    def draw(self, screen, font, show_heat, show_path):
        screen.fill((24, 26, 34))
        if show_heat and self.telemetry.heat:
            most = max(self.telemetry.heat.values())
            overlay = pygame.Surface((WIDTH, HEIGHT), pygame.SRCALPHA)
            for (cx, cy), secs in self.telemetry.heat.items():
                alpha = max(0, min(255, int(40 + 180 * secs / most)))
                overlay.fill((230, 70, 60, alpha), (cx * CELL, cy * CELL, CELL, CELL))
            screen.blit(overlay, (0, 0))
        if show_path and len(self.telemetry.path) > 1:
            pygame.draw.lines(screen, (90, 190, 240), False, [(x, y) for _, x, y in self.telemetry.path], 2)
        pygame.draw.rect(screen, (70, 120, 90), DECOY, border_radius=8)
        for n, rect in PADS.items():
            color = (120, 230, 140) if n == self.task else (60, 80, 70) if n > self.task else (40, 60, 50)
            pygame.draw.rect(screen, color, rect, border_radius=8)
            screen.blit(font.render(str(n), True, TEXT), rect.move(28, 24).topleft)
        pygame.draw.circle(screen, (250, 250, 250), self.pos, 10)
        status = "All tasks done - close the window" if self.finished else f"Task {self.task}: stand on the green pad, press SPACE"
        screen.blit(font.render(status, True, TEXT), (10, 8))
        screen.blit(font.render(f"events {len(self.telemetry.events)}   H heatmap   P path", True, TEXT), (10, 30))


def report(events, heat):
    lines = []
    rate = completion_rate(events, len(PADS))
    lines.append("Completion: " + ("no testers" if rate is None else f"{rate:.0%}"))
    for task, (s, a, r) in sorted(task_stats(events).items()):
        lines.append(f"Task {task}: {s}/{a} attempts succeeded" + ("" if r is None else f" ({r:.0%})"))
    lines.append("Hotspots (cell, seconds): " + ", ".join(f"{c}:{s:.1f}" for c, s in hotspots(heat)))
    return lines


def main():
    pygame.init()
    screen = pygame.display.set_mode((WIDTH, HEIGHT))
    pygame.display.set_caption("Instrumented Playtest")
    clock = pygame.time.Clock()
    font = pygame.font.Font(None, 24)               # created once
    session = PlaytestSession()
    held = set()
    show_heat, show_path = True, True

    running = True
    while running:
        dt = min(clock.tick(60) / 1000, 0.05)
        for event in pygame.event.get():
            if event.type == pygame.QUIT:
                running = False
            elif event.type == pygame.KEYDOWN:
                held.add(event.key)
                if event.key == pygame.K_SPACE:
                    session.action()
                elif event.key == pygame.K_h:
                    show_heat = not show_heat
                elif event.key == pygame.K_p:
                    show_path = not show_path
            elif event.type == pygame.KEYUP:
                held.discard(event.key)

        move = pygame.Vector2(
            (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))
        session.update(move, dt)
        session.draw(screen, font, show_heat, show_path)
        pygame.display.flip()

    pygame.quit()
    t = session.telemetry
    t.log("session_end", finished=session.finished)
    path = LOG_DIR / f"session_{t.session}_{t.tester}.csv"
    write_events(path, t.events)
    events = read_events(path)                      # analyze what was saved, not what is in memory
    print(f"Saved {len(events)} events to {path.name}")
    for line in report(events, t.heat):
        print(line)
    # Example A/B data (made up for practice): seconds to finish the tutorial, two layouts.
    print(compare("Layout A", [48, 55, 61, 52, 70, 58], "Layout B", [51, 49, 57, 60, 54, 62]))
    print("Playtest closed cleanly.")


if __name__ == "__main__":
    main()