path: root/tools/gbdemo

#!/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"]
    # bare --bios: resolve the emulator's default boot ROM to an absolute path
    # (the tmux pane's cwd is wherever the user happens to be)
    if "--bios" in args:
        i = args.index("--bios")
        if i + 1 >= len(args) or args[i + 1].startswith("-"):
            args.insert(i + 1, os.path.join(GBC, "bios/gbc_bios.bin"))
    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()