musicmouse/python-backend/musicmouse/tippen/progress.py

"""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
    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)
    #: 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, 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),
            "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,
    )