qemu-xteink/scripts/xteink/cli.py

"""Command line front end for the xteink emulator."""

import argparse
import sys
from pathlib import Path

from . import emu
from .web import serve


def state_dir(args):
    return Path(args.state_dir).expanduser().resolve()


def run(args, interactive):
    state = state_dir(args)
    process = emu.launch(
        state,
        args.firmware,
        args.sdcard,
        variant=args.variant,
        power_hold_ms=args.power_hold_ms,
        interactive=interactive,
    )
    print(f"started PID {process.pid}")
    print(f"state: {state}")
    if not interactive:
        print(f"serial log: {state / 'serial.log'}")
        return

    print("keys: Left/Right, 1-4 bottom buttons, P power; close the window to stop")
    try:
        returncode = process.wait()
    except KeyboardInterrupt:
        if process.poll() is None:
            try:
                emu.qmp_command(state, "quit")
            except OSError:
                process.terminate()
        returncode = process.wait()
    finally:
        (state / "pid").unlink(missing_ok=True)
        (state / "qmp.sock").unlink(missing_ok=True)
    if returncode:
        raise RuntimeError(f"QEMU exited with status {returncode}")


def start(args):
    run(args, interactive=False)


def interactive(args):
    run(args, interactive=True)


def stop(args):
    print(f"stopped PID {emu.shutdown(state_dir(args))}")


def wait(args):
    line = emu.wait_for_log(state_dir(args), args.match, args.timeout)
    print(line, end="" if line.endswith("\n") else "\n")


def button(args):
    state = state_dir(args)
    emu.require_running(state)
    if (args.down or args.up) and len(args.name) != 1:
        raise RuntimeError("--down/--up accept exactly one button")

    for name in args.name:
        if args.down or args.up:
            emu.set_button(state, name, down=args.down)
            return
        emu.press(state, name, args.samples, args.timeout, args.hold_ms)


def screenshot(args):
    state = state_dir(args)
    emu.require_running(state)
    print(emu.screendump(state, args.output, args.settle_ms))


def memory(args):
    state = state_dir(args)
    emu.require_running(state)
    if args.dump:
        print(emu.dump_memory(state, args.region, args.dump))
        return
    for row in emu.memory_usage(state):
        percent = 100 * row["touched"] / row["size"]
        print(
            f"{row['region']:<7}{row['base']:#010x}  {row['size'] // 1024:>4} KiB total"
            f"  {row['touched'] // 1024:>4} KiB touched ({percent:.0f}%)"
        )


def web(args):
    serve(state_dir(args), args.host, args.port, args.log_bytes)


def parser():
    root = argparse.ArgumentParser(prog="xteink-emu")
    commands = root.add_subparsers(dest="command", required=True)

    def add_state(command):
        command.add_argument("--state-dir", default=emu.default_state_dir())

    def add_run_arguments(command):
        add_state(command)
        command.add_argument("--firmware", required=True)
        command.add_argument("--sdcard", required=True)
        command.add_argument("--variant", choices=emu.VARIANTS, default="x3")
        command.add_argument("--power-hold-ms", type=int, default=2000)

    command = commands.add_parser("start")
    add_run_arguments(command)
    command.set_defaults(handler=start)

    command = commands.add_parser("interactive")
    add_run_arguments(command)
    command.set_defaults(handler=interactive)

    command = commands.add_parser("stop")
    add_state(command)
    command.set_defaults(handler=stop)

    command = commands.add_parser("wait")
    add_state(command)
    command.add_argument("--timeout", type=float, required=True, help="seconds")
    command.add_argument("--match", required=True, help="regular expression")
    command.set_defaults(handler=wait)

    command = commands.add_parser("button")
    add_state(command)
    command.add_argument("name", nargs="+", choices=emu.BUTTONS, help="one or more buttons pressed in order")
    command.add_argument("--hold-ms", type=int, default=500, help="hold time for the power button")
    command.add_argument("--samples", type=int, default=4, help="guest ADC reads required per press and release")
    command.add_argument("--timeout", type=float, default=90, help="seconds to wait for guest ADC sampling")
    button_mode = command.add_mutually_exclusive_group()
    button_mode.add_argument("--down", action="store_true", help="press and hold")
    button_mode.add_argument("--up", action="store_true", help="release")
    command.set_defaults(handler=button)

    command = commands.add_parser("screenshot")
    add_state(command)
    command.add_argument("--output", required=True)
    command.add_argument("--settle-ms", type=int, default=1500, help="wait for the panel to refresh before capturing")
    command.set_defaults(handler=screenshot)

    command = commands.add_parser("memory", help="RAM high-water marks, or dump a region with --dump")
    add_state(command)
    command.add_argument("--region", choices=emu.MEMORY_REGIONS, default="dram")
    command.add_argument("--dump", help="write the raw region to this file")
    command.set_defaults(handler=memory)

    command = commands.add_parser("web")
    add_state(command)
    command.add_argument("--host", default="127.0.0.1")
    command.add_argument("--port", type=int, default=8080)
    command.add_argument("--log-bytes", type=int, default=200_000, help="tail size of the serial log")
    command.set_defaults(handler=web)
    return root


def main():
    args = parser().parse_args()
    if getattr(args, "timeout", 0) < 0:
        raise RuntimeError("--timeout must not be negative")
    if getattr(args, "hold_ms", 0) < 0:
        raise RuntimeError("--hold-ms must not be negative")
    if getattr(args, "samples", 1) < 1:
        raise RuntimeError("--samples must be positive")
    if getattr(args, "settle_ms", 0) < 0:
        raise RuntimeError("--settle-ms must not be negative")
    if getattr(args, "power_hold_ms", 0) < 0:
        raise RuntimeError("--power-hold-ms must not be negative")
    args.handler(args)


def run_cli():
    try:
        main()
    except (RuntimeError, OSError) as error:
        print(f"xteink-emu: {error}", file=sys.stderr)
        sys.exit(1)