#!/usr/bin/env python3 """ migrate.py -- turn the data we already have into an open, trainable dataset. Run: python3 migrate.py [--out dataset] [--verify] WHAT THIS MIGRATES Game run history (37 files) -> runs.jsonl + decisions.jsonl Live captures (600+ files) -> states/ (content-addressed, gzipped) WHAT IT CANNOT MIGRATE, AND WHY Per-step combat state/action pairs. The session logs record the action, the source and the reason, but NOT the observation, and the captures are only written for combat. So a combat step has a state with no action, or an action with no state -- never both. Those pairs are NOT reconstructed here. Inventing them would poison the corpus, so the collector (collect.py) records them going forward instead. DESIGN RULES 1. Lossless raw, derived views. Nothing is dropped, nothing is edited. 2. Content-addressed states, so the 44% duplicate rate costs nothing. 3. Every inferred field is marked. `deck_reconstructed` is not the same thing as an observed deck and the corpus must say which it is. 4. Provenance on every row: game build, mod commit, source file. 5. No secrets, no absolute paths, no personal data in the output. """ from __future__ import annotations import argparse import gzip import hashlib import json import os import pathlib import shutil import sys from collections import Counter SCHEMA_RUN = "sts2.run/1" SCHEMA_DECISION = "sts2.decision/1" SCHEMA_MANIFEST = "sts2.manifest/1" GAME_BUILD = "0.107.1" GAME_COMMIT = "59260271" MOD_COMMIT = "55e0648" HOME = pathlib.Path.home() RUN_HISTORY = ( HOME / "Library/Application Support/SlayTheSpire2/steam" / "76561198141226155/modded/profile1/saves/history" ) CAPTURE_DIR = pathlib.Path("capture") # -------------------------------------------------------------------------- # helpers # -------------------------------------------------------------------------- def short_id(text: str, n: int = 12) -> str: return hashlib.sha256(text.encode()).hexdigest()[:n] def strip_prefix(value, prefix: str) -> str: """'CARD.STRIKE_IRONCLAD' -> 'STRIKE_IRONCLAD'. Never mangles non-matching.""" text = str(value or "") return text[len(prefix):] if text.startswith(prefix) else text def write_jsonl(path: pathlib.Path, rows) -> int: n = 0 with path.open("w") as fh: for row in rows: fh.write(json.dumps(row, sort_keys=True, separators=(",", ":")) + "\n") n += 1 return n def verify(out: pathlib.Path) -> int: """Check an existing dataset. Returns 0 when clean, else a count of failures. Separate from the migrate path on purpose: verifying a dataset that was just regenerated proves nothing, because it is correct by construction. This is the entry point that can actually catch corruption. """ print() print("verifying round-trip and invariants...") bad = 0 def fail(msg): nonlocal bad bad += 1 print(f" FAIL {msg}") for name in ("runs.jsonl", "decisions.jsonl", "states_index.jsonl"): if not (out / name).exists(): fail(f"missing {name}") return bad run_rows = [json.loads(l) for l in (out / "runs.jsonl").read_text().splitlines()] dec_rows = [json.loads(l) for l in (out / "decisions.jsonl").read_text().splitlines()] st_rows = [json.loads(l) for l in (out / "states_index.jsonl").read_text().splitlines()] # state blobs must exist and hash back to their digest for rec in st_rows: blob = out / rec["path"] if not blob.exists(): fail(f"missing state {rec['path']}") continue if hashlib.sha256(gzip.decompress(blob.read_bytes())).hexdigest() \ != rec["digest"].split(":")[1]: fail(f"hash mismatch {rec['path']}") for row in dec_rows: # a reward decision that picked something must pick exactly one option if row["kind"].endswith("reward") and row.get("chosen"): n_picked = sum(1 for o in row["options"] if o.get("picked")) if n_picked != 1: fail(f"{row['decision_id']}: {n_picked} picked options") # `chosen` must be one of `options` -- guards the mapping, not the parse if row.get("chosen") and row["chosen"] not in \ [o.get("id") for o in row["options"]]: fail(f"{row['decision_id']}: chosen not in options") # no secrets and no absolute paths in a corpus meant to be published blob = (out / "decisions.jsonl").read_text() + (out / "runs.jsonl").read_text() if "/Users/" in blob or "/home/" in blob: fail("absolute path leaked into the corpus") for needle in ("sk-", "API_KEY", "Bearer "): if needle in blob: fail(f"possible secret leaked: {needle!r}") # every row must carry its own outcome, and reference a real run ids = {r["run_id"] for r in run_rows} for row in dec_rows: if "outcome" not in row: fail(f"{row['decision_id']}: no outcome attached") if row["run_id"] not in ids: fail(f"{row['decision_id']}: unknown run_id") print(f" {'OK' if not bad else str(bad) + ' PROBLEMS'}: " f"{len(run_rows)} runs, {len(dec_rows)} decisions, {len(st_rows)} states") return bad # -------------------------------------------------------------------------- # run history -> runs.jsonl + decisions.jsonl # -------------------------------------------------------------------------- def deck_as_counts(cards) -> list[dict]: """Collapse a card list to [{'id':..,'count':..}], sorted for stability.""" counts = Counter(strip_prefix((c or {}).get("id"), "CARD.") for c in (cards or [])) return [{"id": k, "count": v} for k, v in sorted(counts.items()) if k] def migrate_runs(out: pathlib.Path, verbose: bool = True): runs, decisions = [], [] files = sorted(RUN_HISTORY.glob("*.run")) if not files: print(f" no run files under {RUN_HISTORY}") return runs, decisions for path in files: try: d = json.loads(path.read_text()) except Exception as exc: # noqa: BLE001 print(f" SKIP {path.name}: {exc}") continue run_id = path.stem player = (d.get("players") or [{}])[0] history = d.get("map_point_history") or [] # `map_point_history` is a list of acts, each a list of map points. acts = [a for a in history if isinstance(a, list)] total_points = sum(len(a) for a in acts) killed_by = strip_prefix(d.get("killed_by_encounter"), "ENCOUNTER.") killed_by_event = strip_prefix(d.get("killed_by_event"), "EVENT.") runs.append({ "schema": SCHEMA_RUN, "run_id": run_id, "seed": d.get("seed"), "character": strip_prefix(player.get("character"), "CHARACTER."), "ascension": d.get("ascension"), "game_mode": d.get("game_mode"), "build_id": d.get("build_id"), "platform": d.get("platform_type"), "modifiers": d.get("modifiers"), "start_time": d.get("start_time"), "run_time_seconds": d.get("run_time"), "was_abandoned": d.get("was_abandoned"), "win": bool(d.get("win")), "killed_by": killed_by or None, "killed_by_event": killed_by_event or None, "progress": {"acts_entered": len(acts), "map_points": total_points}, "final": { "deck": deck_as_counts(player.get("deck")), "deck_size": len(player.get("deck") or []), "relics": [strip_prefix((r or {}).get("id"), "RELIC.") for r in (player.get("relics") or [])], "potions": [strip_prefix((p or {}).get("id"), "POTION.") for p in (player.get("potions") or [])], "max_potion_slots": player.get("max_potion_slot_count"), }, "source": {"kind": "game_run_history", "file": path.name}, }) # ---- per-map-point decisions ------------------------------------ outcome = { "run_win": bool(d.get("win")), "run_map_points": total_points, "run_killed_by": killed_by or None, "run_acts_entered": len(acts), } for act_i, act in enumerate(acts): for pt_i, point in enumerate(act): if not isinstance(point, dict): continue stats = (point.get("player_stats") or [{}])[0] rooms = point.get("rooms") or [] room = rooms[0] if rooms else {} where = f"{run_id}:act{act_i}:pt{pt_i}" # Context at this map point. NOTE: the game does not record the # deck as it stood here, so `deck` is NOT included. What is # recorded is the observed delta, which the consumer can fold # forward from the known starting deck. Marked as observed. context = { "act": act_i, "map_point": pt_i, "map_point_type": strip_prefix(point.get("map_point_type"), "MAP_POINT."), "hp": stats.get("current_hp"), "max_hp": stats.get("max_hp"), "gold": stats.get("current_gold"), "encounter": strip_prefix(room.get("model_id"), "ENCOUNTER.") or None, "room_type": strip_prefix(room.get("room_type"), "ROOM.") or None, "monsters": [strip_prefix(m, "MONSTER.") for m in (room.get("monster_ids") or [])], } # Per-fight reward signal. This is the dense signal the corpus # exists for: damage and turns are what a policy actually pays. fight = { "damage_taken": stats.get("damage_taken"), "hp_healed": stats.get("hp_healed"), "turns_taken": room.get("turns_taken"), "gold_gained": stats.get("gold_gained"), } def emit(kind, options, chosen, extra=None): row = { "schema": SCHEMA_DECISION, "decision_id": f"{where}:{kind}", "run_id": run_id, "kind": kind, "context": context, "options": options, "chosen": chosen, "fight": fight, "outcome": outcome, "source": {"kind": "game_run_history", "file": path.name}, } if extra: row.update(extra) decisions.append(row) # card reward: every offered card, with the pick flag ccs = stats.get("card_choices") or [] if ccs: options = [{ "id": strip_prefix((cc.get("card") or {}).get("id"), "CARD."), "picked": bool(cc.get("was_picked")), "floor_added": (cc.get("card") or {}).get("floor_added_to_deck"), } for cc in ccs] picked = [o["id"] for o in options if o["picked"]] emit("card_reward", options, picked[0] if picked else None, {"skipped": not picked}) if stats.get("relic_choices"): options = [{ "id": strip_prefix((rc.get("relic") or {}).get("id"), "RELIC."), "picked": bool(rc.get("was_picked")), } for rc in stats["relic_choices"]] picked = [o["id"] for o in options if o["picked"]] emit("relic_reward", options, picked[0] if picked else None, {"skipped": not picked}) if stats.get("potion_choices"): options = [{ "id": strip_prefix((pc.get("potion") or {}).get("id"), "POTION."), "picked": bool(pc.get("was_picked")), } for pc in stats["potion_choices"]] picked = [o["id"] for o in options if o["picked"]] emit("potion_reward", options, picked[0] if picked else None, {"skipped": not picked}) if stats.get("event_choices"): # Shape varies by event; keep it raw rather than guessing. emit("event", stats["event_choices"], None, {"raw": True, "note": "event option shape varies; stored unparsed"}) if stats.get("rest_site_choices"): emit("rest_site", [{"id": str(c)} for c in stats["rest_site_choices"]], str(stats["rest_site_choices"][0])) if stats.get("upgraded_cards"): emit("upgrade", [{"id": strip_prefix(c, "CARD.")} for c in stats["upgraded_cards"]], strip_prefix(stats["upgraded_cards"][0], "CARD.")) if stats.get("cards_removed"): emit("remove", [{"id": strip_prefix(c, "CARD.")} for c in stats["cards_removed"]], strip_prefix(stats["cards_removed"][0], "CARD.")) if stats.get("cards_transformed"): emit("transform", [{"id": strip_prefix(c, "CARD.")} for c in stats["cards_transformed"]], strip_prefix(stats["cards_transformed"][0], "CARD.")) if stats.get("cards_enchanted"): emit("enchant", [{"id": strip_prefix(c, "CARD.")} for c in stats["cards_enchanted"]], strip_prefix(stats["cards_enchanted"][0], "CARD.")) if stats.get("ancient_choice"): emit("ancient", stats["ancient_choice"], None, {"raw": True}) bought = [] for key in ("bought_relics", "bought_colorless", "bought_potions"): for item in (stats.get(key) or []): bought.append({"item": item, "kind": key}) if bought: emit("shop_purchase", bought, None, {"raw": True}) if verbose: print(f" {run_id} pts={total_points:<3} win={bool(d.get('win'))!s:<5} " f"killed_by={killed_by or '-'}") return runs, decisions # -------------------------------------------------------------------------- # captures -> content-addressed states/ # -------------------------------------------------------------------------- def migrate_states(out: pathlib.Path, verbose: bool = True) -> dict: store = out / "states" store.mkdir(parents=True, exist_ok=True) index, seen, total_raw, total_gz = [], {}, 0, 0 files = sorted(CAPTURE_DIR.glob("*.json")) for path in files: raw = path.read_bytes() digest = hashlib.sha256(raw).hexdigest() total_raw += len(raw) if digest in seen: seen[digest]["seen_in"].append(path.name) continue try: obs = json.loads(raw) except Exception: # noqa: BLE001 continue blob = gzip.compress(raw, 6) total_gz += len(blob) sub = store / digest[:2] sub.mkdir(exist_ok=True) (sub / f"{digest}.json.gz").write_bytes(blob) state_type = obs.get("state_type") seen[digest] = { "digest": f"sha256:{digest}", "path": f"states/{digest[:2]}/{digest}.json.gz", "bytes_raw": len(raw), "bytes_gz": len(blob), "state_type": state_type, "menu_screen": obs.get("menu_screen"), "seen_in": [path.name], } index.append(seen[digest]) write_jsonl(out / "states_index.jsonl", index) if verbose: print(f" {len(files)} capture files -> {len(index)} unique states " f"({100*len(index)/max(1,len(files)):.0f}% unique)") print(f" {total_raw/1e6:.2f} MB raw -> {total_gz/1e6:.2f} MB stored") return {"files": len(files), "unique": len(index), "bytes_raw": total_raw, "bytes_gz": total_gz} # -------------------------------------------------------------------------- # main # -------------------------------------------------------------------------- def main() -> int: ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--out", default="dataset", help="output directory") ap.add_argument("--verify", action="store_true", help="re-read the output and check it round-trips") ap.add_argument("--check-only", action="store_true", help="verify an EXISTING dataset without regenerating it. " "--verify alone rebuilds first, so it can only ever " "see data that is correct by construction.") args = ap.parse_args() out = pathlib.Path(args.out) if args.check_only: if not out.exists(): print(f"no dataset at {out}/; run without --check-only first") return 2 return verify(out) if out.exists(): shutil.rmtree(out) out.mkdir(parents=True) print("migrating run history...") runs, decisions = migrate_runs(out) n_runs = write_jsonl(out / "runs.jsonl", runs) n_dec = write_jsonl(out / "decisions.jsonl", decisions) print() print("migrating captures...") state_stats = migrate_states(out) kinds = Counter(d["kind"] for d in decisions) print() print("=== dataset ===") print(f" runs.jsonl {n_runs:>6} rows") print(f" decisions.jsonl {n_dec:>6} rows") print(f" states_index.jsonl{state_stats['unique']:>6} rows") print() print(" decisions by kind:") for kind, n in kinds.most_common(): print(f" {kind:<18}{n:>6}") manifest = { "schema": SCHEMA_MANIFEST, "generated_by": "migrate.py", "provenance": { "game_build": GAME_BUILD, "game_commit": GAME_COMMIT, "mod_commit": MOD_COMMIT, }, "counts": {"runs": n_runs, "decisions": n_dec, "unique_states": state_stats["unique"], "capture_files": state_stats["files"]}, "decisions_by_kind": dict(kinds), "limitations": [ "No per-step combat state/action pairs: the logs record the action " "but not the observation, and captures exist only for combat. " "collect.py records these going forward.", "Per-map-point deck composition is not recorded by the game. The " "corpus stores observed deltas (cards_gained, cards_removed, " "upgraded_cards) and the final deck; folding them forward is left " "to the consumer.", "0 wins in this corpus. The reward signal has no positive class.", ], "license_note": "Slay the Spire 2 is (c) Mega Crit. Game-derived " "identifiers are included for research and education.", } (out / "manifest.json").write_text(json.dumps(manifest, indent=2, sort_keys=True)) if args.verify: rc = verify(out) if rc: return rc print() print(f"written to {out}/") return 0 if __name__ == "__main__": sys.exit(main())