path: root/gbctl

#!/usr/bin/env python3
"""gbctl — drive the gbc emulator's socket control/debug channel from the CLI.

Model mirrors gdb-driver: spawn a tmux pane running the emulator with a control
socket, then drive it with terse verbs, then stop. Socket auto-resolves to the
single live instance (else pass --sock / set GBCTL_SOCK).

  gbctl spawn <rom> [emu-args...]     # tmux pane + emulator --sock; waits ready
  gbctl cpu | state | stopinfo        # introspection (one-line replies)
  gbctl read 0xC000 16                # hex dump
  gbctl write 0xC000 de ad be ef      # hex bytes (round-trips with read)
  gbctl reg pc 0x150                  # set a register/flag
  gbctl step [n]                      # single-step (synchronous)
  gbctl break 0x150 | break | delete all
  gbctl watch 0xFF44 | unwatch all
  gbctl continue [secs]              # resume; if secs given, block for a stop event
  gbctl pause
  gbctl press a b   |  gbctl hold left  |  gbctl release   # inject input
  gbctl send <raw protocol line...>   # escape hatch: send any line verbatim
  gbctl monitor [secs]                # stream async events (default until Ctrl-C)
  gbctl screen [color]                # tmux capture of the emulator pane
  gbctl list [--prune]                # inventory of live instances
  gbctl stop                          # quit emulator + kill pane + rm socket
"""
import json, os, re, socket, subprocess, sys, time, glob

HERE = os.path.dirname(os.path.abspath(__file__))
BIN  = os.path.join(HERE, "build", "gbc")
REGDIR = os.path.join(os.environ.get("XDG_RUNTIME_DIR", "/tmp"), "gbctl")

# ---------------------------------------------------------------------------
# registry
# ---------------------------------------------------------------------------
def _ensure_regdir():
    os.makedirs(REGDIR, exist_ok=True)

def _reg_path(inst_id):
    return os.path.join(REGDIR, inst_id + ".json")

def _load_all():
    out = []
    for p in glob.glob(os.path.join(REGDIR, "*.json")):
        try:
            with open(p) as f:
                d = json.load(f)
            d["_path"] = p
            out.append(d)
        except Exception:
            pass
    return out

def _pane_alive(pane):
    if not pane:
        return False
    try:
        r = subprocess.run(["tmux", "list-panes", "-a", "-F", "#{pane_id}"],
                           capture_output=True, text=True)
        return pane in r.stdout.split()
    except Exception:
        return False

def _live_instances(prune=False):
    live = []
    for d in _load_all():
        ok = os.path.exists(d.get("sock", "")) and _pane_alive(d.get("pane"))
        if ok:
            live.append(d)
        elif prune:
            try:
                os.unlink(d["_path"])
                if d.get("sock") and os.path.exists(d["sock"]):
                    os.unlink(d["sock"])
            except Exception:
                pass
    return live

# ---------------------------------------------------------------------------
# socket resolution + I/O
# ---------------------------------------------------------------------------
def _resolve_sock(argv):
    # explicit --sock anywhere
    if "--sock" in argv:
        i = argv.index("--sock")
        s = argv[i + 1]
        del argv[i:i + 2]
        return s
    if os.environ.get("GBCTL_SOCK"):
        return os.environ["GBCTL_SOCK"]
    live = _live_instances()
    if len(live) == 1:
        return live[0]["sock"]
    if not live:
        die("no live emulator; run: gbctl spawn <rom>")
    die("multiple instances; pass --sock <path>:\n" +
        "\n".join("  %s  %s" % (d["sock"], d.get("rom", "")) for d in live))

class Conn:
    """Line-buffered wrapper around a connected unix socket."""
    def __init__(self, sock_path, timeout=5.0):
        self.s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        self.s.settimeout(timeout)
        self.s.connect(sock_path)
        self.buf = b""

    def read_line(self, timeout=None):
        if timeout is not None:
            self.s.settimeout(timeout)
        while b"\n" not in self.buf:
            try:
                d = self.s.recv(65536)
            except socket.timeout:
                return None
            if not d:
                if self.buf:
                    line, self.buf = self.buf, b""
                    return line.decode(errors="replace")
                return None
            self.buf += d
        line, _, self.buf = self.buf.partition(b"\n")
        return line.decode(errors="replace")

    def send(self, line):
        self.s.sendall((line + "\n").encode())

    def close(self):
        try:
            self.s.close()
        except Exception:
            pass

def _connect(sock_path, timeout=5.0):
    c = Conn(sock_path, timeout)
    c.read_line()          # consume greeting
    return c

def send_cmd(sock_path, line, wait_event=0.0):
    """Send one protocol line; print the reply. If wait_event>0, keep reading
    for an async 'event ...' line up to that many seconds and print it too."""
    c = _connect(sock_path)
    c.send(line)
    reply = c.read_line(timeout=5.0)
    if reply is not None:
        print(reply)
    if wait_event > 0:
        deadline = time.time() + wait_event
        while time.time() < deadline:
            ev = c.read_line(timeout=max(0.01, deadline - time.time()))
            if ev is None:
                break
            if ev.startswith("event"):
                print(ev)
                break
    c.close()

# ---------------------------------------------------------------------------
# helpers
# ---------------------------------------------------------------------------
def die(msg, code=1):
    print(msg, file=sys.stderr)
    sys.exit(code)

def _wait_ready(sock_path, secs=10.0):
    deadline = time.time() + secs
    while time.time() < deadline:
        if os.path.exists(sock_path):
            try:
                c = _connect(sock_path, timeout=1.0)
                c.send("ping")
                ok = (c.read_line(1.0) or "").strip() == "pong"
                c.close()
                if ok:
                    return True
            except Exception:
                pass
        time.sleep(0.1)
    return False

# ---------------------------------------------------------------------------
# subcommands
# ---------------------------------------------------------------------------
def cmd_spawn(argv):
    if not os.path.exists(BIN):
        die("emulator not built: run `make` in %s" % HERE)
    split = "-v"                              # stacked keeps full width for render
    while argv and argv[0].startswith("--split"):
        val = argv.pop(0).split("=", 1)
        mode = (val[1] if len(val) == 2 else argv.pop(0)) if val[0] == "--split" else "v"
        split = {"h": "-h", "v": "-v", "window": "window"}.get(mode, "-v")
    if not argv:
        die("usage: gbctl spawn [--split h|v|window] <rom> [emu-args...]")
    rom = argv[0]
    emu_args = argv[1:]
    if not os.path.exists(rom):
        die("rom not found: %s" % rom)

    _ensure_regdir()
    inst_id = "%d-%s" % (os.getpid(), os.urandom(3).hex())
    sock_path = os.path.join(REGDIR, inst_id + ".sock")

    if "TMUX" not in os.environ:
        die("not inside tmux; gbctl spawn needs a tmux session")

    cmd = [BIN, "--sock", sock_path] + emu_args + [rom]   # ROM is the final arg
    # exec via a shell wrapper so the pane stays around & shows a title
    shell_cmd = "printf '\\033]2;gbc %s\\007'; exec %s" % (
        os.path.basename(rom), " ".join(_shquote(c) for c in cmd))
    if split == "window":
        r = subprocess.run(["tmux", "new-window", "-P", "-F", "#{pane_id}",
                            shell_cmd], capture_output=True, text=True)
    else:
        r = subprocess.run(["tmux", "split-window", split, "-P", "-F",
                            "#{pane_id}", shell_cmd], capture_output=True, text=True)
    if r.returncode != 0:
        die("tmux spawn failed: %s" % r.stderr.strip())
    pane = r.stdout.strip()

    with open(_reg_path(inst_id), "w") as f:
        json.dump({"id": inst_id, "sock": sock_path, "pane": pane,
                   "rom": rom, "created": time.time()}, f)

    if _wait_ready(sock_path):
        print("ready pane=%s sock=%s rom=%s" % (pane, sock_path, rom))
    else:
        die("spawned pane=%s but control socket never became ready\n"
            "  check the pane for errors (gbctl screen)" % pane)

def _shquote(s):
    if re.fullmatch(r"[A-Za-z0-9_./=:+-]+", s or ""):
        return s
    return "'" + s.replace("'", "'\\''") + "'"

def cmd_list(argv):
    prune = "--prune" in argv
    live = _live_instances(prune=prune)
    if not live:
        print("(no live instances)")
        return
    for d in live:
        print("pane=%s sock=%s rom=%s" % (d.get("pane"), d["sock"],
                                          d.get("rom", "")))

def cmd_stop(argv):
    sock_path = _resolve_sock(argv)
    # find matching registry entry (for pane id)
    inst = None
    for d in _load_all():
        if d.get("sock") == sock_path:
            inst = d
            break
    # ask the emulator to quit
    try:
        c = _connect(sock_path, timeout=2.0)
        c.send("quit")
        c.read_line(1.0)
        c.close()
    except Exception:
        pass
    time.sleep(0.2)
    if inst:
        if _pane_alive(inst.get("pane")):
            subprocess.run(["tmux", "kill-pane", "-t", inst["pane"]],
                          capture_output=True)
        try:
            os.unlink(inst["_path"])
        except Exception:
            pass
    if os.path.exists(sock_path):
        try:
            os.unlink(sock_path)
        except Exception:
            pass
    print("stopped %s" % sock_path)

def cmd_screen(argv):
    color = "color" in argv
    argv = [a for a in argv if a != "color"]
    sock_path = _resolve_sock(argv)
    inst = next((d for d in _load_all() if d.get("sock") == sock_path), None)
    if not inst or not inst.get("pane"):
        die("no pane for %s" % sock_path)
    args = ["tmux", "capture-pane", "-p", "-t", inst["pane"]]
    if color:
        args.insert(2, "-e")
    r = subprocess.run(args, capture_output=True, text=True)
    sys.stdout.write(r.stdout)

def _kill_hud_panes():
    # kill any pane whose start command references the HUD, so there's never
    # more than one HUD running (avoids the double-spawn tmux flicker).
    n = 0
    try:
        r = subprocess.run(
            ["tmux", "list-panes", "-a", "-F", "#{pane_id}\t#{pane_start_command}"],
            capture_output=True, text=True)
        for line in r.stdout.splitlines():
            pid, _, startcmd = line.partition("\t")
            if "gbhud.py" in startcmd:
                subprocess.run(["tmux", "kill-pane", "-t", pid], capture_output=True)
                n += 1
    except Exception:
        pass
    return n

def cmd_hud(argv):
    # spawn a tmux pane running the live debug HUD, bound to the resolved socket
    killonly = "--kill" in argv
    if killonly:
        argv = [a for a in argv if a != "--kill"]
        k = _kill_hud_panes()
        print("killed %d hud pane(s)" % k)
        return
    sock_path = _resolve_sock(argv)
    if "TMUX" not in os.environ:
        die("not inside tmux; gbctl hud needs a tmux session")
    killed = _kill_hud_panes()          # never leave a second HUD around
    hud = os.path.join(HERE, "tools", "gbhud.py")
    split = "-h"
    if "--split" in argv:
        i = argv.index("--split"); split = {"h":"-h","v":"-v"}.get(argv[i+1],"-h")
    cmd = "exec python3 %s --sock %s" % (_shquote(hud), _shquote(sock_path))
    r = subprocess.run(["tmux", "split-window", split, "-P", "-F", "#{pane_id}", cmd],
                       capture_output=True, text=True)
    if r.returncode != 0:
        die("tmux spawn failed: %s" % r.stderr.strip())
    note = (" (replaced %d)" % killed) if killed else ""
    print("hud pane=%s sock=%s%s" % (r.stdout.strip(), sock_path, note))

def cmd_monitor(argv):
    secs = float(argv[0]) if argv and _isnum(argv[0]) else None
    if secs is not None:
        argv = argv[1:]
    sock_path = _resolve_sock(argv)
    c = _connect(sock_path, timeout=None)
    print("# monitoring %s (Ctrl-C to stop)" % sock_path, file=sys.stderr)
    deadline = (time.time() + secs) if secs else None
    try:
        while True:
            if deadline and time.time() >= deadline:
                break
            to = None if deadline is None else max(0.01, deadline - time.time())
            line = c.read_line(timeout=to)
            if line is None:
                if deadline:
                    break
                continue
            print(line, flush=True)
    except KeyboardInterrupt:
        pass
    finally:
        c.close()

def _isnum(x):
    try:
        float(x)
        return True
    except Exception:
        return False

# ---------------------------------------------------------------------------
# main dispatch
# ---------------------------------------------------------------------------
LIFECYCLE = {"spawn": cmd_spawn, "list": cmd_list, "stop": cmd_stop,
             "screen": cmd_screen, "monitor": cmd_monitor, "hud": cmd_hud}

# convenience aliases -> raw protocol verbs
ALIASES = {"press": None, "tap": None, "hold": None, "buttons": None}

def main():
    argv = sys.argv[1:]
    if not argv or argv[0] in ("-h", "--help", "help") and len(argv) == 1:
        print(__doc__)
        return
    verb = argv[0]

    if verb in LIFECYCLE:
        return LIFECYCLE[verb](argv[1:])

    # everything else is a protocol line sent to the resolved socket
    rest = argv[1:]
    sock_path = _resolve_sock(rest)   # strips --sock if present

    # input convenience: press/tap/hold map to button tokens
    if verb in ("press", "tap"):
        line = " ".join(rest)                       # e.g. "a b"
    elif verb == "hold":
        line = " ".join("+" + t for t in rest)      # +left +a
    elif verb == "send":
        line = " ".join(rest)                       # raw passthrough
    else:
        line = " ".join([verb] + rest)              # verb + args

    # continue with an optional wait-for-event timeout (positional seconds)
    wait = 0.0
    if verb in ("continue", "cont", "c", "run"):
        # a trailing numeric arg means "block this long for a stop event"
        if rest and _isnum(rest[-1]):
            wait = float(rest[-1])
            line = " ".join([verb] + rest[:-1])

    send_cmd(sock_path, line, wait_event=wait)

if __name__ == "__main__":
    main()