diff --git a/pqcli/pqw.py b/pqcli/pqw.py new file mode 100644 index 0000000..5eede19 --- /dev/null +++ b/pqcli/pqw.py @@ -0,0 +1,353 @@ +"""Import and export Progress Quest web saves (.pqw files). + +A .pqw file is base64 of one character's JSON, as the web version's roster +page downloads it. Characters keep their generator state both ways, so they +carry on exactly as they would have where they came from. +""" + +import base64 +import binascii +import copy +import datetime +import json +from typing import Any, cast + +from pqcli.config import Class, EquipmentType, Monster, Race, StatType +from pqcli.game_data import GameData +from pqcli.lingo import act_name, to_arabic, to_roman +from pqcli.mechanic import ( + BaseTask, + BuyTask, + HeadingToKillingFieldsTask, + HeadingToMarketTask, + InventoryItem, + KillTask, + LoadingTask, + Player, + PlotTask, + RegularTask, + SellTask, + Simulation, + Spell, + Stats, +) +from pqcli.random import Alea, RngState + +# Spells upstream renamed; its loader merges old saves the same way. +SPELL_RENAMES = [ + ("Innoculate", "Inoculate"), + ("Tonsilectomy", "Tonsillectomy"), +] + +# Fields of the web save that pqcli has no use for. +ONLINE_FIELDS = ("online", "guild", "motto") + +LOADING_CAPTION = "Loading...." + + +class PqwError(Exception): + pass + + +def decode(text: str) -> dict[str, Any]: + """Turn the contents of a .pqw file into the web save's game object.""" + try: + raw = base64.b64decode("".join(text.split()), validate=True) + except (binascii.Error, ValueError) as ex: + raise PqwError(f"not a .pqw file: {ex}") from ex + try: + decoded = raw.decode("utf-8") + except UnicodeDecodeError: + # saves from before 2022 were base64 of the raw JS string + decoded = raw.decode("latin-1") + try: + game = json.loads(decoded) + except json.JSONDecodeError as ex: + raise PqwError(f"not a .pqw file: {ex}") from ex + if not isinstance(game, dict) or "Traits" not in game: + raise PqwError("not a .pqw file: no character in it") + return game + + +def encode(game: dict[str, Any]) -> str: + """Turn a web save's game object into the contents of a .pqw file.""" + text = json.dumps(game, ensure_ascii=False, separators=(",", ":")) + return base64.b64encode(text.encode("utf-8")).decode("ascii") + + +def _field(game: dict[str, Any], *path: str | int) -> Any: + value: Any = game + for key in path: + try: + value = value[key] + except (KeyError, IndexError, TypeError) as ex: + name = ".".join(map(str, path)) + raise PqwError(f"missing field {name}") from ex + return value + + +def _int(game: dict[str, Any], *path: str | int) -> int: + value = _field(game, *path) + try: + return int(value) + except (TypeError, ValueError) as ex: + name = ".".join(map(str, path)) + raise PqwError(f"field {name} is not a number: {value!r}") from ex + + +def _bar(game: dict[str, Any], name: str) -> tuple[int, float]: + position = _field(game, name, "position") + if not isinstance(position, int | float): + raise PqwError(f"field {name}.position is not a number") + return _int(game, name, "max"), position + + +def _parse_monster(text: str) -> Monster | None: + if not text: + return None + parts = text.split("|") + if len(parts) != 3: + raise PqwError(f"cannot read monster {text!r}") + name, level, item = parts + try: + return Monster(name, int(level), None if item == "*" else item) + except ValueError as ex: + raise PqwError(f"cannot read monster {text!r}") from ex + + +def _encode_monster(monster: Monster) -> str: + return f"{monster.name}|{monster.level}|{monster.item or '*'}" + + +def _parse_task(game: dict[str, Any]) -> BaseTask: + kind = str(_field(game, "task")) + caption = str(_field(game, "kill")) + duration = _bar(game, "TaskBar")[0] + if kind.startswith("kill|"): + monster = _parse_monster(kind.removeprefix("kill|")) + return KillTask(caption.removesuffix("..."), duration, monster) + if caption == LOADING_CAPTION: + return LoadingTask("Loading", duration) + task_class: type[BaseTask] = { + "buying": BuyTask, + "market": HeadingToMarketTask, + "sell": SellTask, + "heading": HeadingToKillingFieldsTask, + "": RegularTask, + }.get(kind, RegularTask) + return task_class(caption.removesuffix("..."), duration) + + +def _encode_task(task: BaseTask) -> tuple[str, str]: + if isinstance(task, LoadingTask): + return "", LOADING_CAPTION + caption = task.description + "..." + if isinstance(task, KillTask): + monster = task.monster or Monster("", 0, None) + return "kill|" + _encode_monster(monster), caption + for task_class, kind in [ + (BuyTask, "buying"), + (HeadingToMarketTask, "market"), + (SellTask, "sell"), + (HeadingToKillingFieldsTask, "heading"), + ]: + if isinstance(task, task_class): + return kind, caption + return "", caption + + +def _parse_queue_entry(entry: Any) -> RegularTask: + parts = str(entry).split("|", 2) + if len(parts) != 3 or parts[0] not in {"task", "plot"}: + raise PqwError(f"cannot read queued task {entry!r}") + try: + duration = int(parts[1]) * 1000 + except ValueError as ex: + raise PqwError(f"cannot read queued task {entry!r}") from ex + if parts[0] == "plot": + # renamed after the act it starts once it comes up + return PlotTask("Loading", duration) + return RegularTask(parts[2], duration) + + +def _encode_queue_entry(task: BaseTask) -> str: + seconds = task.duration // 1000 + if isinstance(task, PlotTask): + return f"plot|{seconds}|Loading" + return f"task|{seconds}|{task.description}" + + +def _parse_spells(game: dict[str, Any]) -> list[Spell]: + rows = [(str(row[0]), str(row[1])) for row in _field(game, "Spells")] + levels = {name: to_arabic(roman) for name, roman in rows} + spells = [] + for name, _roman in rows: + renamed = dict(SPELL_RENAMES).get(name) + if renamed is not None: + levels[renamed] = levels.get(renamed, 0) + levels[name] + name = renamed + elif any(new == name and old in levels for old, new in SPELL_RENAMES): + continue # merged into the old spell's place + spells.append(Spell(name, levels[name])) + return spells + + +def _find_by_name( + items: tuple[Race, ...] | tuple[Class, ...], name: str +) -> Race | Class | None: + return next((item for item in items if item.name == name), None) + + +def import_game( + game: dict[str, Any], data: GameData +) -> tuple[Player, list[str]]: + """Build a pqcli character from a web save, plus any warnings.""" + warnings = [] + for field in ONLINE_FIELDS: + if game.get(field): + warnings.append(f"dropped {field!r}: pqcli plays offline only") + + name = str(_field(game, "Traits", "Name")) + race_name = str(_field(game, "Traits", "Race")) + class_name = str(_field(game, "Traits", "Class")) + race = _find_by_name(data.races, race_name) or Race(race_name, []) + class_ = _find_by_name(data.classes, class_name) or Class(class_name, []) + + stats = Stats({stat: _int(game, "Stats", stat.value) for stat in StatType}) + seed = _field(game, "seed") + if not (isinstance(seed, list) and len(seed) == 4): + raise PqwError("field seed is not a generator state") + stamp = game.get("birthstamp") + birthday = ( + datetime.datetime.fromtimestamp(stamp / 1000) + if isinstance(stamp, int | float) + else datetime.datetime.now() + ) + + player = Player( + name=name, + birthday=birthday, + race=cast(Race, race), + class_=cast(Class, class_), + stats=stats, + rng=Alea.from_state(cast(RngState, tuple(seed))), + ) + player.level = _int(game, "Traits", "Level") + player.elapsed = float(game.get("elapsed", 0)) * 1000 + + player.exp_bar.reset(*_bar(game, "ExpBar")) + player.task_bar.reset(*_bar(game, "TaskBar")) + player.quest_book.plot_bar.reset(*_bar(game, "PlotBar")) + player.quest_book.quest_bar.reset(*_bar(game, "QuestBar")) + + player.quest_book._act = _int(game, "act") + player.quest_book._quests = [str(q) for q in _field(game, "Quests")] + player.quest_book.monster = _parse_monster( + str(game.get("questmonster", "")) + ) + + for kind in EquipmentType: + item = str(_field(game, "Equips").get(kind.value, "")) + if item: + player.equipment.put(kind, item) + player.equipment.best = str(game.get("bestequip", player.equipment.best)) + + rows = _field(game, "Inventory") + if not rows or rows[0][0] != "Gold": + raise PqwError("field Inventory does not start with Gold") + player.inventory._gold = _int(game, "Inventory", 0, 1) + player.inventory._items = [ + InventoryItem(str(row[0]), int(row[1])) for row in rows[1:] + ] + player.inventory.encum_bar.reset(*_bar(game, "EncumBar")) + + player.spell_book._spells = _parse_spells(game) + + player.task = _parse_task(game) + player.queue = [_parse_queue_entry(e) for e in _field(game, "queue")] + return player, warnings + + +def export_game(player: Player, data: GameData) -> dict[str, Any]: + """Build a web save from a pqcli character.""" + if player.task is None: + # never played: set up the Prologue first, on a copy + player = copy.deepcopy(player) + Simulation(player, data).tick(0) + assert player.task is not None + + seed = list(player.rng.state()) + birthstamp = int(player.birthday.timestamp() * 1000) + now = datetime.datetime.now() + kind, caption = _encode_task(player.task) + best_stat = player.stats.best_prime + best_spell = player.spell_book.best + + quest_monster = player.quest_book.monster + # upstream picks the quest monster by its index in the monster list + quest_monster_index = ( + data.monsters.index(quest_monster) + if quest_monster in data.monsters + else 0 + ) + + def bar(position: float, max_: int) -> dict[str, float]: + return {"position": position, "max": max_} + + return { + "Traits": { + "Name": player.name, + "Race": player.race.name, + "Class": player.class_.name, + "Level": player.level, + }, + "dna": seed, + "seed": seed, + "birthday": player.birthday.strftime("%a %b %d %Y %H:%M:%S"), + "birthstamp": birthstamp, + "date": now.strftime("%a %b %d %Y %H:%M:%S"), + "stamp": int(now.timestamp() * 1000), + "Stats": {stat.value: player.stats[stat] for stat in StatType}, + "beststat": f"{best_stat.value} {player.stats[best_stat]}", + "task": kind, + "tasks": 0, + "elapsed": int(player.elapsed // 1000), + "bestequip": player.equipment.best, + "Equips": { + kind.value: player.equipment[kind] or "" for kind in EquipmentType + }, + "Inventory": [["Gold", player.inventory.gold]] + + [[item.name, item.quantity] for item in player.inventory], + "Spells": [ + [spell.name, to_roman(spell.level)] for spell in player.spell_book + ], + "bestspell": ( + f"{best_spell.name} {to_roman(best_spell.level)}" + if best_spell + else "" + ), + "act": player.quest_book.act, + "bestplot": act_name(player.quest_book.act), + "Quests": list(player.quest_book.quests), + "bestquest": player.quest_book.current_quest or "", + "questmonster": ( + _encode_monster(quest_monster) if quest_monster else "" + ), + "questmonsterindex": quest_monster_index, + "kill": caption, + "ExpBar": bar(player.exp_bar.position, player.exp_bar.max_), + "EncumBar": bar( + player.inventory.encum_bar.position, + player.inventory.encum_bar.max_, + ), + "PlotBar": bar( + player.quest_book.plot_bar.position, + player.quest_book.plot_bar.max_, + ), + "QuestBar": bar( + player.quest_book.quest_bar.position, + player.quest_book.quest_bar.max_, + ), + "TaskBar": bar(player.task_bar.position, player.task_bar.max_), + "queue": [_encode_queue_entry(task) for task in player.queue], + } diff --git a/tests/test_pqw.py b/tests/test_pqw.py new file mode 100644 index 0000000..992c58f --- /dev/null +++ b/tests/test_pqw.py @@ -0,0 +1,133 @@ +import base64 +import datetime +import json + +import pytest +from upstream import bridge + +from pqcli.config import StatType +from pqcli.game_data import load_game_data +from pqcli.mechanic import Player, Simulation, Stats +from pqcli.pqw import PqwError, decode, encode, export_game, import_game +from pqcli.random import Alea + +DATA = load_game_data() + + +def played(ticks: int, seed: str = "pqw") -> Simulation: + player = Player( + name="Exportée", + birthday=datetime.datetime(2020, 2, 2, 12, 0, 0), + race=DATA.races[5], + class_=DATA.classes[7], + stats=Stats({stat: 9 for stat in StatType}), + rng=Alea(seed), + ) + simulation = Simulation(player, DATA) + simulation.tick(0) + for _ in range(ticks): + simulation.tick(100) + return simulation + + +def test_encode_decode_round_trip() -> None: + game = {"Traits": {"Name": "Zoë 🐉"}, "act": 3} + assert decode(encode(game)) == game + + +def test_decode_ignores_whitespace() -> None: + text = encode({"Traits": {"Name": "x"}}) + assert decode("\n".join(text[i : i + 10] for i in range(0, 99, 10))) + + +def test_decode_reads_old_latin1_saves() -> None: + # before 2022 the web version stored base64 of the raw JS string + raw = json.dumps({"Traits": {"Name": "Jürgen"}}, ensure_ascii=False) + text = base64.b64encode(raw.encode("latin-1")).decode() + assert decode(text)["Traits"]["Name"] == "Jürgen" + + +@pytest.mark.parametrize( + "text", + [ + "not base64 at all!", + base64.b64encode(b"not json").decode(), + base64.b64encode(b"[1, 2]").decode(), + base64.b64encode(b'{"act": 1}').decode(), + ], +) +def test_decode_rejects_garbage(text: str) -> None: + with pytest.raises(PqwError, match="not a .pqw file"): + decode(text) + + +def test_import_reports_missing_fields() -> None: + game = export_game(played(500).player, DATA) + del game["ExpBar"] + with pytest.raises(PqwError, match="ExpBar"): + import_game(game, DATA) + + +def test_import_rejects_bad_numbers() -> None: + game = export_game(played(500).player, DATA) + game["Traits"]["Level"] = "lots" + with pytest.raises(PqwError, match="Traits.Level"): + import_game(game, DATA) + + +def test_import_warns_about_online_play() -> None: + game = export_game(played(500).player, DATA) + game["online"] = {"realm": "Alpaquil", "passkey": 1} + game["motto"] = "hi" + _player, warnings = import_game(game, DATA) + assert len(warnings) == 2 + assert "online" in warnings[0] + + +def test_import_merges_renamed_spells() -> None: + game = export_game(played(500).player, DATA) + game["Spells"] = [ + ["Inoculate", "II"], + ["Slime Finger", "I"], + ["Innoculate", "III"], + ] + player, _warnings = import_game(game, DATA) + assert [(s.name, s.level) for s in player.spell_book] == [ + ("Slime Finger", 1), + ("Inoculate", 5), + ] + + +def test_import_keeps_unknown_race_and_class() -> None: + game = export_game(played(500).player, DATA) + game["Traits"]["Race"] = "Moon Wombat" + player, _warnings = import_game(game, DATA) + assert player.race.name == "Moon Wombat" + assert player.race.attr == [] + + +def test_export_of_unplayed_character() -> None: + player = Player( + name="Fresh", + birthday=datetime.datetime(2020, 2, 2), + race=DATA.races[0], + class_=DATA.classes[0], + stats=Stats({stat: 9 for stat in StatType}), + ) + game = export_game(player, DATA) + assert game["kill"] == "Loading...." + assert len(game["queue"]) == 5 + assert player.task is None # exported a copy + + +@pytest.mark.parametrize("ticks", [0, 1, 25, 3000, 40000, 123457]) +def test_round_trip_keeps_playing_identically(ticks: int) -> None: + original = played(ticks) + copy, _warnings = import_game( + decode(encode(export_game(original.player, DATA))), DATA + ) + assert bridge.snapshot(copy) == bridge.snapshot(original.player) + resumed = Simulation(copy, DATA) + assert bridge.run(resumed, 10**8, 2000) == bridge.run( + original, 10**8, 2000 + ) diff --git a/tests/test_upstream.py b/tests/test_upstream.py index 04d27f3..a0ccfc0 100644 --- a/tests/test_upstream.py +++ b/tests/test_upstream.py @@ -13,11 +13,13 @@ from pqcli import random from pqcli.game_data import DATA_FILES, load_game_data from pqcli.lingo import generate_name from pqcli.mechanic import ( + Simulation, impressive_guy, monster_task, named_monster, special_item, ) +from pqcli.pqw import decode, encode, export_game, import_game from pqcli.random import Alea, RngState pytestmark = pytest.mark.skipif( @@ -133,7 +135,14 @@ def test_game_matches_upstream(seed: str, race: str, class_: str) -> None: [{"cmd": "play", "spec": spec, "ticks": ticks, "tasks": tasks}] ) ours = bridge.play(DATA, spec, ticks, tasks) - assert len(ours) == len(theirs) == tasks + assert len(ours) == tasks + assert_same_games(ours, theirs) + + +def assert_same_games( + ours: list[dict[str, Any]], theirs: list[dict[str, Any]] +) -> None: + assert len(ours) == len(theirs) for i, (mine, upstream) in enumerate(zip(ours, theirs)): if mine != upstream: previous = [snap["caption"] for snap in ours[max(0, i - 5) : i]] @@ -150,3 +159,43 @@ def test_game_matches_upstream(seed: str, race: str, class_: str) -> None: for key, (ours_value, theirs_value) in fields.items() ) ) + + +# (tasks, ticks): stop right after a task, or in the middle of one +SAVE_POINTS = [(3000, 10**9), (10**9, 123457), (10**9, 1)] + + +@pytest.mark.parametrize("tasks,ticks", SAVE_POINTS) +def test_import_from_upstream_plays_on_identically( + tasks: int, ticks: int +) -> None: + spec = bridge.make_spec("import", "Eel Man", "Inner Mason") + (game,) = bridge.run_harness( + [{"cmd": "export", "spec": spec, "ticks": ticks, "tasks": tasks}] + ) + (text, theirs) = bridge.run_harness( + [ + {"cmd": "encode", "game": game}, + {"cmd": "resume", "game": game, "ticks": 10**9, "tasks": 3000}, + ] + ) + player, warnings = import_game(decode(text), DATA) + assert warnings == [] + ours = bridge.run(Simulation(player, DATA), 10**9, 3000) + assert_same_games(ours, theirs) + + +@pytest.mark.parametrize("tasks,ticks", SAVE_POINTS) +def test_export_to_upstream_plays_on_identically( + tasks: int, ticks: int +) -> None: + spec = bridge.make_spec("export", "Panda Man", "Robot Monk") + simulation = bridge.new_simulation(DATA, spec) + bridge.run(simulation, ticks, tasks) + text = encode(export_game(simulation.player, DATA)) + (game,) = bridge.run_harness([{"cmd": "decode", "text": text}]) + (theirs,) = bridge.run_harness( + [{"cmd": "resume", "game": game, "ticks": 10**9, "tasks": 3000}] + ) + ours = bridge.run(simulation, 10**9, 3000) + assert_same_games(ours, theirs) diff --git a/tests/upstream/bridge.py b/tests/upstream/bridge.py index 2b0e46a..87ca098 100644 --- a/tests/upstream/bridge.py +++ b/tests/upstream/bridge.py @@ -146,6 +146,10 @@ def play( data: GameData, spec: dict[str, Any], ticks: int, tasks: int ) -> list[dict[str, Any]]: """Play pqcli the way the harness plays upstream: 100 ms ticks.""" + return run(new_simulation(data, spec), ticks, tasks) + + +def new_simulation(data: GameData, spec: dict[str, Any]) -> Simulation: race = next(race for race in data.races if race.name == spec["race"]) class_ = next(c for c in data.classes if c.name == spec["class"]) seed = cast(RngState, tuple(spec["seed"])) @@ -159,6 +163,14 @@ def play( ) simulation = Simulation(player, data) simulation.tick(0) # sets up the Prologue, as upstream's sold() does + return simulation + + +def run( + simulation: Simulation, ticks: int, tasks: int +) -> list[dict[str, Any]]: + """Keep playing in 100 ms ticks, snapshotting after each task.""" + player = simulation.player snapshots = [] for _ in range(ticks): completing = player.task_bar.done diff --git a/tests/upstream/harness.js b/tests/upstream/harness.js index 6c4e379..93d53b0 100644 --- a/tests/upstream/harness.js +++ b/tests/upstream/harness.js @@ -114,8 +114,26 @@ function snapshot() { function play(request) { newGame(request.spec); + return run(request.ticks, request.tasks); +} + +// Play a game to some point, then return its save, as .pqw files hold it. +function exportGame(request) { + newGame(request.spec); + run(request.ticks, request.tasks); + SaveGame(); + return JSON.parse(store.roster)[request.spec.name]; +} + +// Load a save and keep playing it. +function resume(request) { + LoadGame(JSON.parse(JSON.stringify(request.game))); + return run(request.ticks, request.tasks); +} + +function run(ticks, tasks) { const snapshots = []; - for (let i = 0; i < request.ticks; i++) { + for (let i = 0; i < ticks; i++) { const completing = TaskBar.done(); now += 100; clock.lasttick = now - 100; @@ -123,7 +141,7 @@ function play(request) { if (completing) { // copy, since game's lists keep changing after this tick snapshots.push(JSON.parse(JSON.stringify(snapshot()))); - if (snapshots.length >= request.tasks) break; + if (snapshots.length >= tasks) break; } } return snapshots; @@ -148,7 +166,24 @@ function lists() { return result; } -const handlers = { play: play, call: call, lists: lists }; +// The roster page's .pqw encoding, from roster.js. +function encode(request) { + return btoa(unescape(encodeURIComponent(JSON.stringify(request.game)))); +} + +function decode(request) { + return JSON.parse(decodeURIComponent(escape(atob(request.text)))); +} + +const handlers = { + encode: encode, + decode: decode, + play: play, + export: exportGame, + resume: resume, + call: call, + lists: lists, +}; const requests = JSON.parse(fs.readFileSync(0, "utf8")); const responses = requests.map((request) => handlers[request.cmd](request)); process.stdout.write(JSON.stringify(responses));