diff options
Diffstat (limited to '')
| -rwxr-xr-x | tools/gbdemo | 309 |
1 files changed, 309 insertions, 0 deletions
diff --git a/tools/gbdemo b/tools/gbdemo new file mode 100755 index 0000000..e2b8e13 --- /dev/null +++ b/tools/gbdemo @@ -0,0 +1,309 @@ +#!/usr/bin/env python3 +"""gbdemo - scripted, screen-synced demo driver for gbos promo captures. + +Runs a demo "script": spawns a sl0pboy Game Boy on the gbhub network, records +the LCD to a .gbv, injects console text (via the hub, like gbtype) and raw +button events (via the emulator control socket, like gbctl) at the right +moments, and turns the capture into a GIF. + +The key trick is `waitfor`: gbos's terminal keeps an ASCII shadow of the screen +in WRAM (TERM_BUF $D600, 18 slots x 64 B, row order via wAssign[18] $D590), so +the driver READS THE SCREEN over the emulator's debug socket and syncs each +step on real output - no blind frame counting, robust against network timing. + +Usage: + sudo tools/gbhub --daemon # first, in another terminal + tools/gbdemo demo/readme.gbd # then run the script (inside tmux) + +Script verbs (shlex lines, '#' comments): + spawn [emu-args...] spawn GB on the hub (gbctl tmux pane + --sock) + record <out.gbv> [everyN] start LCD capture (default every 2nd frame) + recordstop stop the capture + type <text...> inject a console line at once (+ Enter) + ('#' only comments at line start, so IRC + channel names pass through unquoted) + slowtype <text...> inject char-by-char, human-paced (+ Enter) + key <gbctl-input-args...> button events, e.g. `key press select`, + `key hold a`, `key release` + movie <file.gbmv> play a TAS movie (tools in ~/dev/gbc) + wait <seconds> fixed delay + waitfor <text> [timeout=30] block until <text> appears on the GB screen + waitgone <text> [timeout=30] block until <text> is NOT on the screen + screen print the decoded 40x18 screen (script tuning) + shot <out.bmp> VRAM screenshot via gbshot.py + gif <out.gif> [gbgif-args] convert the last recording (gbgif.py) + echo <text...> progress note to stderr + stop quit emulator + pane (also runs on error) +""" +import os, re, shlex, socket, subprocess, sys, time + +HOME = os.path.expanduser("~") +GBC = os.environ.get("GBDEMO_GBC", os.path.join(HOME, "dev/gbc")) +GBCTL = os.path.join(GBC, "gbctl") +ROM = os.path.join(HOME, "dev/gbos/gbos.gb") +HUB = "/tmp/gbhub.sock" +CTL = "/tmp/gbhub.ctl" + +TERM_BUF, ASSIGN, ROWS, COLS, SLOT = 0xD600, 0xD590, 18, 40, 64 + +state = {"sock": None, "gbv": None, "spawned": False} + + +def die(msg): + sys.stderr.write("gbdemo: %s\n" % msg) + cleanup() + sys.exit(1) + + +def note(msg): + sys.stderr.write("[gbdemo] %s\n" % msg) + + +# ---- emulator control socket (same line protocol gbctl speaks) -------------- +def sock_cmd(line, timeout=5.0): + s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + s.settimeout(timeout) + s.connect(state["sock"]) + s.sendall(line.encode() + b"\n") + # skip the connect greeting and async "event ..." pushes; keep reading + # until the actual reply line for our command arrives. + buf = b"" + while True: + nl = buf.find(b"\n") + if nl >= 0: + ln, buf = buf[:nl].decode(errors="replace").strip(), buf[nl + 1:] + if ln.startswith("event ") or "control channel" in ln: + continue + s.close() + return ln + d = s.recv(65536) + if not d: + s.close() + return "" + buf += d + + +def read_mem(addr, n): + r = sock_cmd("read 0x%04X %d" % (addr, n)) + parts = r.split() # ok read 0xD600 1152 <hex> + if len(parts) < 5 or parts[0] != "ok": + die("read 0x%04X failed: %s" % (addr, r)) + return bytes.fromhex(parts[4]) + + +def screen_text(): + """Decode the gbos terminal: 18 rows x 40 cols of ASCII, top to bottom.""" + assign = read_mem(ASSIGN, ROWS) + buf = read_mem(TERM_BUF, ROWS * SLOT) + rows = [] + for r in range(ROWS): + line = buf[assign[r] * SLOT: assign[r] * SLOT + COLS] + rows.append("".join(chr(c) if 32 <= c < 127 else " " for c in line)) + return rows + + +# ---- hub console injection (same wire protocol as gbtype) ------------------- +def hub_type(payload: bytes): + s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + s.settimeout(5) + try: + s.connect(CTL) + except OSError as e: + die("cannot connect hub ctl %s (%s) - is gbhub running?" % (CTL, e)) + s.sendall(b"type\n" + payload) + s.shutdown(socket.SHUT_WR) + out = b"" + while True: + try: + d = s.recv(4096) + except OSError: + break + if not d: + break + out += d + s.close() + if out.startswith(b"err"): + die("hub: %s" % out.decode(errors="replace").strip()) + + +# ---- verbs ------------------------------------------------------------------ +def v_spawn(args): + nohub = "--nohub" in args # offline: no link (script smoke-tests) + args = [a for a in args if a != "--nohub"] + if not nohub and not os.path.exists(HUB): + die("no hub at %s - start it first: sudo tools/gbhub --daemon" % HUB) + rom = "/tmp/gbdemo_%d.gb" % os.getpid() # private copy: no .sav clashes + subprocess.run(["cp", ROM, rom], check=True) + cmd = [GBCTL, "spawn", "--split", "v", rom] + if not nohub: + cmd += ["--serial-sock", HUB] + if "--shrink" not in args and "--sixel" not in args: + cmd += ["--shrink", "2"] # keep the monitor pane small; capture + # and recording stay full-res + cmd += args + r = subprocess.run(cmd, cwd=GBC, capture_output=True, text=True) + m = re.search(r"sock=(\S+)", r.stdout + r.stderr) + if r.returncode != 0 or not m: + die("spawn failed:\n%s%s" % (r.stdout, r.stderr)) + state["sock"], state["spawned"] = m.group(1), True + note("spawned, sock=%s" % state["sock"]) + + +def v_record(args): + if not args: + die("record needs an output path") + path = os.path.abspath(args[0]) + every = args[1] if len(args) > 1 else "2" + r = sock_cmd("record start %s %s" % (path, every)) + if not r.startswith("ok"): + die("record: %s" % r) + state["gbv"] = path + note("recording -> %s (every %s frames)" % (path, every)) + + +def v_recordstop(args): + note(sock_cmd("record stop")) + + +def v_type(args): + text = " ".join(args) + note("type: %s" % text) + hub_type(text.encode() + b"\n") + + +def v_slowtype(args): + text = " ".join(args) + note("slowtype: %s" % text) + for ch in text: + hub_type(ch.encode()) + time.sleep(0.06) + time.sleep(0.25) + hub_type(b"\n") + + +def v_key(args): + if not args: + die("key needs gbctl input args (press/hold/release ...)") + r = subprocess.run([GBCTL, "--sock", state["sock"]] + args, + cwd=GBC, capture_output=True, text=True) + note("key %s: %s" % (" ".join(args), (r.stdout + r.stderr).strip())) + + +def v_movie(args): + if not args: + die("movie needs a .gbmv path") + note(sock_cmd("input play %s" % os.path.abspath(args[0]))) + + +def v_wait(args): + time.sleep(float(args[0])) + + +def _wait_screen(args, want_present): + if not args: + die("waitfor/waitgone need <text> [timeout]") + text = args[0] + timeout = float(args[1]) if len(args) > 1 else 30.0 + deadline = time.time() + timeout + while time.time() < deadline: + present = any(text in row for row in screen_text()) + if present == want_present: + note("%s %r" % ("saw" if want_present else "gone:", text)) + return + time.sleep(0.25) + for row in screen_text(): + sys.stderr.write(" |%s|\n" % row) + die("timeout (%gs) waiting for %r %s" % + (timeout, text, "to appear" if want_present else "to go away")) + + +def v_waitfor(args): + _wait_screen(args, True) + + +def v_waitgone(args): + _wait_screen(args, False) + + +def v_screen(args): + for row in screen_text(): + print("|%s|" % row) + + +def v_shot(args): + if not args: + die("shot needs an output path") + subprocess.run(["python3", os.path.join(GBC, "tools/gbshot.py"), + state["sock"], os.path.abspath(args[0])], check=True) + note("shot -> %s" % args[0]) + + +def v_gif(args): + if not state["gbv"]: + die("gif: nothing recorded yet") + if not args: + die("gif needs an output path") + out = os.path.abspath(args[0]) + cmd = ["python3", os.path.join(GBC, "tools/gbgif.py"), + state["gbv"], out] + args[1:] + subprocess.run(cmd, check=True) + note("gif -> %s" % out) + + +def v_echo(args): + note(" ".join(args)) + + +def v_stop(args): + cleanup() + + +def cleanup(): + if state["spawned"]: + state["spawned"] = False + subprocess.run([GBCTL, "stop"], cwd=GBC, + capture_output=True, text=True) + note("emulator stopped") + + +VERBS = {"spawn": v_spawn, "record": v_record, "recordstop": v_recordstop, + "type": v_type, "slowtype": v_slowtype, "key": v_key, + "movie": v_movie, "wait": v_wait, "waitfor": v_waitfor, + "waitgone": v_waitgone, "screen": v_screen, "shot": v_shot, + "gif": v_gif, "echo": v_echo, "stop": v_stop} + + +def main(): + if len(sys.argv) != 2 or sys.argv[1] in ("-h", "--help"): + sys.stdout.write(__doc__ + "\n") + sys.exit(0 if len(sys.argv) == 2 else 1) + try: + lines = open(sys.argv[1]).read().splitlines() + except OSError as e: + sys.exit("gbdemo: %s" % e) + try: + for n, raw in enumerate(lines, 1): + if raw.lstrip().startswith("#"): + continue # full-line comment + try: + toks = shlex.split(raw) # no inline comments: '#gbos' etc. + except ValueError as e: + die("line %d: %s" % (n, e)) + if not toks: + continue + verb, args = toks[0].lower(), toks[1:] + if verb not in VERBS: + die("line %d: unknown verb %r" % (n, verb)) + if verb not in ("spawn", "echo", "wait") and not state["sock"] \ + and verb not in ("stop",): + die("line %d: %r before spawn" % (n, verb)) + VERBS[verb](args) + except KeyboardInterrupt: + die("interrupted") + finally: + cleanup() + note("done") + + +if __name__ == "__main__": + main() |