afterglow/progress.py

Part of Afterglow.

  1"""Afterglow save / progression layer (plain JSON).
  2
  3Why JSON and not ``simvx.core.save_manager.SaveManager``:
  4
  5``SaveManager`` snapshots ``Property(persist=True)`` values off a live ``Node``
  6tree and pickles them. Afterglow's progress is free-form, key-addressed data:
  7per-room best times keyed by ``(world_id, room_index)``, completion/shard flags,
  8a deaths counter, an unlocked-worlds set, and an options dict (volumes, toggles,
  9key rebinds). None of that lives on a node, and pickle's "unpickling executes
 10arbitrary code" trust boundary is a poor fit for a file the player may copy
 11between machines. SimVX only mandates ``.py`` for *scenes*; save files may use
 12any serialization, so a single human-diffable JSON document is the cleanest fit:
 13no tree to walk, no Property descriptors to mirror, and a forward-compatible
 14schema (unknown keys are tolerated on load).
 15
 16Writes are atomic (temp + ``fsync`` + ``os.replace``) so a crash mid-save never
 17corrupts the live file. A single ``.bak`` is kept for manual recovery.
 18
 19Coordinate / data contract this layer assumes (and nothing more):
 20  * a "room" is addressed by ``(world_id: str, room_index: int)``;
 21  * a "world" is addressed by ``world_id: str`` and has a known room count so we
 22    can tell when its last room was beaten (passed to ``record_room_result`` as
 23    ``world_room_count`` so this layer needs no import of ``rooms_data``);
 24  * worlds unlock in the order given by ``WORLD_ORDER`` (first world always
 25    unlocked); beating the last room of a world unlocks the next one.
 26"""
 27
 28from __future__ import annotations
 29
 30import json
 31import logging
 32import os
 33from pathlib import Path
 34from typing import Any
 35
 36log = logging.getLogger(__name__)
 37
 38#: Bump when the on-disk envelope shape changes incompatibly.
 39SAVE_VERSION = 1
 40
 41#: Canonical world unlock order. Mirrors ``rooms_data.WORLDS`` ids but is kept
 42#: here as a literal so the save layer carries zero sim/content imports. Keep in
 43#: sync with ``rooms_data`` if world ids ever change.
 44WORLD_ORDER: tuple[str, ...] = ("glade", "caverns", "spire")
 45
 46#: Defaults for every option key. ``set_option`` rejects unknown keys, so this
 47#: doubles as the option schema. ``key_rebinds`` maps an action name to a key
 48#: name (empty = engine default bindings).
 49DEFAULT_OPTIONS: dict[str, Any] = {
 50    # Ship a comfortable, quiet default mix: at 0 dB the layered music drone plus
 51    # overlapping SFX summed through Master was harsh and fatiguing. These are the
 52    # default slider positions; the player can raise any of them in Options.
 53    "master_volume": 0.4,
 54    "sfx_volume": 0.4,
 55    "music_volume": 0.22,
 56    "screenshake": True,
 57    "photosensitive_safe": False,
 58    "assist_mode": False,
 59    "assist_invincible": False,
 60    "assist_slow_time": False,
 61    "key_rebinds": {},
 62}
 63
 64
 65def default_save_path() -> Path:
 66    """Per-user save location, overridable via ``$AFTERGLOW_SAVE_DIR``.
 67
 68    Honours ``$XDG_DATA_HOME`` (falling back to ``~/.local/share``) on every
 69    platform; the demo has no platform-specific path needs beyond that.
 70    """
 71    override = os.environ.get("AFTERGLOW_SAVE_DIR")
 72    base = Path(override) if override else Path(os.environ.get("XDG_DATA_HOME", Path.home() / ".local" / "share"))
 73    return base / "afterglow" / "progress.json"
 74
 75
 76def _room_key(world_id: str, room_index: int) -> str:
 77    """Flat JSON-safe key for a ``(world, room)`` pair (JSON keys are strings)."""
 78    return f"{world_id}:{room_index}"
 79
 80
 81class Progress:
 82    """Persistent player progression and options for Afterglow.
 83
 84    All mutators update in-memory state only; call :meth:`save` to persist.
 85    Construct with :meth:`load` to read an existing file (or get defaults).
 86    """
 87
 88    __slots__ = ("path", "rooms", "total_deaths", "unlocked_worlds", "options")
 89
 90    def __init__(self, path: Path | str | None = None) -> None:
 91        self.path = Path(path) if path is not None else default_save_path()
 92        # rooms[key] = {"best_time": float|None, "shard": bool, "completed": bool}
 93        self.rooms: dict[str, dict[str, Any]] = {}
 94        self.total_deaths: int = 0
 95        # First world is always reachable.
 96        self.unlocked_worlds: set[str] = {WORLD_ORDER[0]} if WORLD_ORDER else set()
 97        self.options: dict[str, Any] = _fresh_options()
 98
 99    # -- persistence ------------------------------------------------------
100
101    @classmethod
102    def load(cls, path: Path | str | None = None) -> Progress:
103        """Return a ``Progress`` loaded from ``path`` (defaults if absent/corrupt).
104
105        A missing file yields fresh defaults. A corrupt file is logged and also
106        yields defaults rather than crashing the game on launch.
107        """
108        prog = cls(path)
109        if not prog.path.exists():
110            return prog
111        try:
112            data = json.loads(prog.path.read_text(encoding="utf-8"))
113        except (OSError, ValueError) as exc:
114            log.error("Afterglow save at %s is unreadable (%s); using defaults", prog.path, exc)
115            return prog
116        prog._apply(data)
117        return prog
118
119    def _apply(self, data: dict[str, Any]) -> None:
120        if not isinstance(data, dict):
121            log.error("Afterglow save root is %s, expected dict; ignoring", type(data).__name__)
122            return
123        version = data.get("version", SAVE_VERSION)
124        if isinstance(version, int) and version > SAVE_VERSION:
125            log.warning("Afterglow save version %s is newer than %s; loading best-effort", version, SAVE_VERSION)
126        rooms = data.get("rooms")
127        if isinstance(rooms, dict):
128            for key, entry in rooms.items():
129                if isinstance(entry, dict):
130                    self.rooms[key] = {
131                        "best_time": entry.get("best_time"),
132                        "shard": bool(entry.get("shard", False)),
133                        "completed": bool(entry.get("completed", False)),
134                    }
135        self.total_deaths = int(data.get("total_deaths", 0))
136        unlocked = data.get("unlocked_worlds")
137        if isinstance(unlocked, list):
138            self.unlocked_worlds |= {str(w) for w in unlocked}
139        opts = data.get("options")
140        if isinstance(opts, dict):
141            for key, value in opts.items():
142                if key in DEFAULT_OPTIONS:
143                    self.options[key] = value
144
145    def to_dict(self) -> dict[str, Any]:
146        """Serialisable snapshot of the whole progress document."""
147        return {
148            "version": SAVE_VERSION,
149            "rooms": self.rooms,
150            "total_deaths": self.total_deaths,
151            "unlocked_worlds": sorted(self.unlocked_worlds),
152            "options": self.options,
153        }
154
155    def save(self) -> Path:
156        """Write the document to ``self.path`` atomically; return the path.
157
158        Rotates one ``.bak`` backup. Raises ``OSError`` on write failure (the
159        temp file is best-effort cleaned and the live file is untouched).
160        """
161        self.path.parent.mkdir(parents=True, exist_ok=True)
162        tmp = self.path.with_name(self.path.name + ".tmp")
163        payload = json.dumps(self.to_dict(), indent=2, sort_keys=True)
164        try:
165            with open(tmp, "w", encoding="utf-8") as fh:
166                fh.write(payload)
167                fh.flush()
168                os.fsync(fh.fileno())
169        except OSError:
170            tmp.unlink(missing_ok=True)
171            raise
172        if self.path.exists():
173            os.replace(self.path, self.path.with_name(self.path.name + ".bak"))
174        os.replace(tmp, self.path)
175        return self.path
176
177    # -- room results -----------------------------------------------------
178
179    def record_room_result(
180        self,
181        world_id: str,
182        room_index: int,
183        time: float,
184        shard: bool,
185        died_count: int = 0,
186        *,
187        world_room_count: int | None = None,
188    ) -> None:
189        """Record a room clear.
190
191        Best time only improves (``min``). ``shard`` and ``completed`` are
192        sticky (once true, stay true). ``died_count`` adds to the lifetime
193        deaths counter. If ``world_room_count`` is given and this is the last
194        room of the world (``room_index == world_room_count - 1``), the next
195        world in :data:`WORLD_ORDER` is unlocked.
196        """
197        key = _room_key(world_id, room_index)
198        entry = self.rooms.setdefault(key, {"best_time": None, "shard": False, "completed": False})
199        prev = entry["best_time"]
200        entry["best_time"] = time if prev is None else min(prev, time)
201        entry["shard"] = entry["shard"] or bool(shard)
202        entry["completed"] = True
203        if died_count:
204            self.total_deaths += int(died_count)
205        if world_room_count is not None and room_index >= world_room_count - 1:
206            self._unlock_next(world_id)
207
208    def _unlock_next(self, world_id: str) -> None:
209        if world_id not in WORLD_ORDER:
210            return
211        idx = WORLD_ORDER.index(world_id)
212        if idx + 1 < len(WORLD_ORDER):
213            self.unlocked_worlds.add(WORLD_ORDER[idx + 1])
214
215    # -- queries ----------------------------------------------------------
216
217    def best_time(self, world_id: str, room_index: int) -> float | None:
218        """Best recorded time for a room, or ``None`` if never cleared."""
219        entry = self.rooms.get(_room_key(world_id, room_index))
220        return entry["best_time"] if entry else None
221
222    def has_shard(self, world_id: str, room_index: int) -> bool:
223        entry = self.rooms.get(_room_key(world_id, room_index))
224        return bool(entry and entry["shard"])
225
226    def is_completed(self, world_id: str, room_index: int) -> bool:
227        entry = self.rooms.get(_room_key(world_id, room_index))
228        return bool(entry and entry["completed"])
229
230    def is_world_unlocked(self, world_id: str) -> bool:
231        return world_id in self.unlocked_worlds
232
233    # -- options ----------------------------------------------------------
234
235    def get_option(self, key: str) -> Any:
236        """Return an option value, falling back to its default."""
237        if key not in DEFAULT_OPTIONS:
238            raise KeyError(f"Unknown option {key!r}")
239        return self.options.get(key, DEFAULT_OPTIONS[key])
240
241    def set_option(self, key: str, value: Any) -> None:
242        """Set an option (rejecting unknown keys)."""
243        if key not in DEFAULT_OPTIONS:
244            raise KeyError(f"Unknown option {key!r}")
245        self.options[key] = value
246
247
248def _fresh_options() -> dict[str, Any]:
249    """Deep-ish copy of the option defaults (so the dict value isn't shared)."""
250    opts = dict(DEFAULT_OPTIONS)
251    opts["key_rebinds"] = dict(DEFAULT_OPTIONS["key_rebinds"])
252    return opts