martin/musicmouse
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add backend support for the typing game: curriculum, progress, and reward unlocks

Browse Source 
Moves the typing app's lesson plan and progress from client-side YAML/localStorage
into the backend, and adds a reward system that ties passing a lesson to unlocking
part of the music library. Lock state is always recomputed live from curriculum x
progress x the live library, never persisted separately.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Martin Bauer
2026-09-12 21:03:45 +02:00
parent a5210fead2
commit f7a5d24d8d
18 changed files with 2794 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

330
python-backend/musicmouse/tippen/progress.py Normal file
View File

@@ -0,0 +1,330 @@
"""Server-side typing progress: a JSON sidecar, atomic-written, defensively loaded.
Same convention as everywhere else this codebase persists something outside
``config.yml`` - see ``musicmouse.library.podcast_feeds``'s failed-downloads sidecar. A
missing or corrupt file is just a fresh start, never a crash: nothing here is precious
enough to raise over.
``record_run`` is a near-verbatim port of tippen's own (client-side, until now)
``recordRun`` in ``tippen/src/lib/progress.ts`` - grading a keystroke-by-keystroke run
into stars/points/an animal stays entirely client-side (see ``TippenRunIn`` in
``musicmouse.services.web.schemas``); this only owns what happens to progress once a
graded result arrives.
"""
from __future__ import annotations
import json
import logging
import os
from dataclasses import dataclass
from datetime import date, timedelta
from pathlib import Path
from typing import Literal
from pydantic import BaseModel, Field, ValidationError
from musicmouse.tippen.curriculum import (
CreatureId,
Curriculum,
first_lesson_id,
lesson_by_id,
next_lesson,
world_reward,
)
_log = logging.getLogger(__name__)
__all__ = [
"DILIGENCE_ATTEMPTS",
"AnimalId",
"KeyStat",
"LessonProgress",
"RecordOutcome",
"RunResult",
"Streak",
"Stroke",
"TippenSettings",
"TypingProgress",
"fresh_progress",
"load_progress",
"record_run",
"save_progress",
"today",
]
#: How many attempts at one lesson unlock the next regardless of score - the safety
#: valve against getting stuck on a single stubborn key.
DILIGENCE_ATTEMPTS = 5
AnimalId = Literal[
"snail",
"crab",
"turtle",
"jellyfish",
"fish",
"penguin",
"seal",
"dolphin",
"shark",
"orca",
]
# ------------------------------------------------------------------------- domain
@dataclass(frozen=True, slots=True)
class Stroke:
key: str
expected: str
correct: bool
#: ms timestamp, from the run's own clock.
at: float
@dataclass(frozen=True, slots=True)
class RunResult:
"""What the client already graded a finished run as - see ``grading.ts``."""
stars: Literal[0, 1, 2, 3]
animal: AnimalId
points: float
passed: bool
pearls: int
strokes: tuple[Stroke, ...]
# --------------------------------------------------------------------------- store
class GhostStroke(BaseModel):
key: str
at: float
class LessonProgress(BaseModel):
unlocked: bool = False
runs: int = 0
best_stars: Literal[0, 1, 2, 3] = 0
best_animal: AnimalId | None = None
best_points: float = 0
#: Best-run keystrokes, replayed as the opponent in race mode.
ghost: list[GhostStroke] | None = None
@property
def earned(self) -> bool:
"""Passed on its own merits, or given up on gracefully after enough tries.
Always derived from ``best_stars``/``runs`` rather than stored, so it can never
drift from the rule that computed it. Two stars is the same "passed" gate the
frontend's ``isPassed`` uses (accuracy >= 93 %); the fifth attempt is the
diligence fallback.
"""
return self.best_stars >= 2 or self.runs >= DILIGENCE_ATTEMPTS
class KeyStat(BaseModel):
#: Smoothed reaction time in ms - keybr's exponential moving average.
ema: float = 0
attempts: int = 0
errors: int = 0
class Streak(BaseModel):
days: int = 0
last_played: str | None = None
class TippenSettings(BaseModel):
sound: bool = True
keyboard_hint: Literal["auto", "on", "off"] = "auto"
class TypingProgress(BaseModel):
version: Literal[1] = 1
lessons: dict[str, LessonProgress] = Field(default_factory=dict)
key_stats: dict[str, KeyStat] = Field(default_factory=dict)
pearls: int = 0
#: Pets that have moved into the aquarium, in the order they arrived.
aquarium: list[CreatureId] = Field(default_factory=list)
streak: Streak = Field(default_factory=Streak)
settings: TippenSettings = Field(default_factory=TippenSettings)
def fresh_progress(curriculum: Curriculum) -> TypingProgress:
first = first_lesson_id(curriculum)
lessons = {
lesson.id: LessonProgress(unlocked=lesson.id == first) for lesson in curriculum.lessons
}
return TypingProgress(lessons=lessons)
def load_progress(path: Path, curriculum: Curriculum) -> TypingProgress:
"""A missing or corrupt file is just a fresh start - never fatal."""
try:
raw = path.read_text(encoding="utf-8")
except FileNotFoundError:
return fresh_progress(curriculum)
except OSError as exc:
_log.warning("Could not read %s; starting fresh: %s", path, exc)
return fresh_progress(curriculum)
try:
progress = TypingProgress.model_validate(json.loads(raw))
except (ValueError, ValidationError) as exc:
_log.warning("Could not parse %s; starting fresh: %s", path, exc)
return fresh_progress(curriculum)
# A lesson added to the curriculum since the last save needs an entry too, and the
# first lesson is unlocked by definition - a save that says otherwise is wrong.
fresh = fresh_progress(curriculum)
lessons = dict(progress.lessons)
for lesson_id, blank in fresh.lessons.items():
lessons.setdefault(lesson_id, blank)
first = first_lesson_id(curriculum)
if first is not None and first in lessons:
lessons[first] = lessons[first].model_copy(update={"unlocked": True})
return progress.model_copy(update={"lessons": lessons})
def save_progress(path: Path, progress: TypingProgress) -> None:
"""Write through a sibling temp file so an interrupted save cannot truncate the real
one - the same atomic-write idiom used throughout this codebase."""
temp = path.with_name(f"{path.name}.tmp{os.getpid()}")
try:
temp.write_text(progress.model_dump_json(), encoding="utf-8")
temp.replace(path)
except BaseException:
temp.unlink(missing_ok=True)
raise
# --------------------------------------------------------------------------- record
def today(day: date | None = None) -> str:
"""Today (or `day`) as YYYY-MM-DD."""
return (day or date.today()).isoformat()
def _is_better(candidate: RunResult, best_stars: int, best_points: float) -> bool:
"""Stars come first, points break the tie - a careful run is never displaced by a
sloppy fast one."""
if candidate.stars != best_stars:
return candidate.stars > best_stars
return candidate.points > best_points
def _fold_key_stats(stats: dict[str, KeyStat], strokes: tuple[Stroke, ...]) -> dict[str, KeyStat]:
"""Fold the per-key reaction times of a run into the stored averages. The EMA weight
of 0.3 is slow enough that one distracted run does not rewrite what is known."""
next_stats = dict(stats)
previous_at: float | None = None
for stroke in strokes:
key = stroke.expected.lower()
current = next_stats.get(key, KeyStat())
gap = None if previous_at is None else stroke.at - previous_at
previous_at = stroke.at
# Reaction times over two seconds are a pause for thought, not a measure of the
# key, so they are ignored rather than averaged in.
ema = current.ema
if gap is not None and gap < 2000:
ema = gap if current.ema == 0 else current.ema * 0.7 + gap * 0.3
next_stats[key] = KeyStat(
ema=ema,
attempts=current.attempts + 1,
errors=current.errors + (0 if stroke.correct else 1),
)
return next_stats
def _bump_streak(streak: Streak, day: str) -> Streak:
if streak.last_played == day:
return streak
yesterday = (date.fromisoformat(day) - timedelta(days=1)).isoformat()
consecutive = streak.last_played == yesterday
# A missed day restarts at 1, never at 0 - playing today always counts for something.
return Streak(days=streak.days + 1 if consecutive else 1, last_played=day)
@dataclass(frozen=True, slots=True)
class RecordOutcome:
progress: TypingProgress
#: Set when this run unlocked the following lesson, for the lesson-map celebration.
unlocked_lesson_id: str | None
#: Whether *this* lesson's pass threshold was crossed by this run - the signal a
#: reward attached to it (``Lesson.unlocks``) should now be checked.
newly_earned: bool
#: Set when this run finished a world, for the creature that moved in.
new_creature: CreatureId | None
is_new_best: bool
def record_run(
progress: TypingProgress,
lesson_id: str,
result: RunResult,
curriculum: Curriculum,
day: str | None = None,
) -> RecordOutcome:
"""Record a finished run: stars, animal, pearls, key stats, streak, and the unlock.
The unlock rule, in one place: two stars unlocks the next lesson, and so does the
fifth attempt whatever the score. Speed is nowhere in it.
"""
day = day or today()
before = progress.lessons.get(lesson_id) or LessonProgress(unlocked=True)
was_earned = before.earned
runs = before.runs + 1
improved = _is_better(result, before.best_stars, before.best_points)
lessons = dict(progress.lessons)
updated = before.model_copy(
update={
"runs": runs,
"best_stars": result.stars if improved else before.best_stars,
"best_animal": result.animal if improved else before.best_animal,
"best_points": result.points if improved else before.best_points,
"ghost": (
[GhostStroke(key=s.key, at=s.at) for s in result.strokes if s.correct]
if improved
else before.ghost
),
}
)
lessons[lesson_id] = updated
newly_earned = updated.earned and not was_earned
nxt = next_lesson(curriculum, lesson_id)
unlocked_lesson_id: str | None = None
if newly_earned and nxt is not None and not lessons.get(nxt.id, LessonProgress()).unlocked:
next_before = lessons.get(nxt.id, LessonProgress())
lessons[nxt.id] = next_before.model_copy(update={"unlocked": True})
unlocked_lesson_id = nxt.id
# Finishing the last lesson of a world releases that world's creature. Checked
# against the aquarium so it is only ever awarded once.
aquarium = list(progress.aquarium)
new_creature: CreatureId | None = None
if unlocked_lesson_id and nxt is not None:
finished = lesson_by_id(curriculum, lesson_id)
if finished is not None and nxt.world != finished.world:
reward = world_reward(curriculum, finished.world)
if reward and reward not in aquarium:
aquarium.append(reward)
new_creature = reward
updated_progress = progress.model_copy(
update={
"lessons": lessons,
"aquarium": aquarium,
"key_stats": _fold_key_stats(progress.key_stats, result.strokes),
"pearls": progress.pearls + result.pearls,
"streak": _bump_streak(progress.streak, day),
}
)
return RecordOutcome(
progress=updated_progress,
unlocked_lesson_id=unlocked_lesson_id,
newly_earned=newly_earned,
new_creature=new_creature,
is_new_best=improved,
)