bfe-core1106-sdk/qemu/tests/qmp.py

#!/usr/bin/env python3
"""Tiny QMP client for the qemu/ device-sim tests.

    qmp.py <socket> screendump <out.ppm>
    qmp.py <socket> tap <x> <y>          # absolute 0..32767 (virtio-tablet)
    qmp.py <socket> quit
    qmp.py <socket> drive <script> <outdir> [--size N]

`drive` runs a whole interaction on ONE connection, so the VM boots once and
the scenario costs a couple of seconds per step instead of a full boot. Its
script is one command per line, '#' comments and blank lines ignored, and
coordinates are PANEL PIXELS (0..size-1, default 720) rather than the
0..32767 absolute axis the raw `tap` takes -- the point of the mode is to write
"tap the gear at 47,676" straight off a screenshot.

    shot NAME              screendump to <outdir>/NAME.ppm
    tap X Y                press, hold past several LVGL poll periods, release
    swipe X1 Y1 X2 Y2 [MS] drag, interpolated so LVGL sees real motion
    fling X1 Y1 X2 Y2      swipe fast enough to leave momentum behind
    sleep SECONDS          let animations settle or timers fire
    echo TEXT              progress marker in the driver's own output

A step that names an unknown command is a FATAL error rather than a skip: a
silently-ignored line in a scenario is a test that proves nothing.
"""
import json
import socket
import sys
import time

AXIS_MAX = 32767


def rpc(sock, sock_file, obj):
    sock.sendall((json.dumps(obj) + "\n").encode())
    while True:
        line = sock_file.readline()
        if not line:
            raise RuntimeError("QMP connection closed")
        msg = json.loads(line)
        if "return" in msg:
            return msg["return"]
        if "error" in msg:
            raise RuntimeError(f"QMP error: {msg['error']}")
        # asynchronous events are interleaved; skip them


def send_events(s, f, events):
    rpc(s, f, {"execute": "input-send-event", "arguments": {"events": events}})


def abs_ev(axis, value):
    return {"type": "abs", "data": {"axis": axis, "value": value}}


def btn_ev(down):
    return {"type": "btn", "data": {"down": down, "button": "left"}}


def to_axis(px, size):
    """Panel pixels -> the tablet's absolute axis, clamped to the panel."""
    px = max(0, min(size - 1, int(px)))
    return int(px * AXIS_MAX / (size - 1))


def do_tap(s, f, ax, ay, hold=0.2):
    send_events(s, f, [abs_ev("x", ax), abs_ev("y", ay), btn_ev(True)])
    # Hold the press across several LVGL indev poll periods (33 ms each): an
    # instantaneous press+release lands inside one poll and no click is ever
    # registered.
    time.sleep(hold)
    send_events(s, f, [btn_ev(False)])


def do_swipe(s, f, x1, y1, x2, y2, size, ms=400, steps=None):
    """Drag with interpolated motion.

    LVGL decides a gesture from the movement BETWEEN indev polls, so a press at
    the start and a release at the end (with nothing in between) reads as a
    click on whatever was under the finger, not a scroll. The interpolation
    below is the whole reason a scroll can be tested at all; the step count is
    derived from the duration so the poll period always sees a few pixels of
    travel.
    """
    ax1, ay1 = to_axis(x1, size), to_axis(y1, size)
    ax2, ay2 = to_axis(x2, size), to_axis(y2, size)
    if steps is None:
        steps = max(6, int(ms / 25))
    send_events(s, f, [abs_ev("x", ax1), abs_ev("y", ay1), btn_ev(True)])
    time.sleep(0.05)
    for i in range(1, steps + 1):
        t = i / steps
        send_events(s, f, [
            abs_ev("x", int(ax1 + (ax2 - ax1) * t)),
            abs_ev("y", int(ay1 + (ay2 - ay1) * t)),
        ])
        time.sleep(ms / 1000.0 / steps)
    send_events(s, f, [btn_ev(False)])


def drive(s, f, script_path, outdir, size):
    import os
    os.makedirs(outdir, exist_ok=True)
    with open(script_path) as fh:
        lines = fh.readlines()

    for lineno, raw in enumerate(lines, 1):
        line = raw.split("#", 1)[0].strip()
        if not line:
            continue
        parts = line.split()
        cmd, args = parts[0], parts[1:]

        if cmd == "shot":
            name = args[0]
            path = os.path.join(os.path.abspath(outdir), f"{name}.ppm")
            rpc(s, f, {"execute": "screendump", "arguments": {"filename": path}})
            print(f"  [{lineno}] shot {name}", flush=True)
        elif cmd == "tap":
            x, y = int(args[0]), int(args[1])
            do_tap(s, f, to_axis(x, size), to_axis(y, size))
            print(f"  [{lineno}] tap {x},{y}", flush=True)
        elif cmd == "swipe":
            ms = int(args[4]) if len(args) > 4 else 400
            do_swipe(s, f, int(args[0]), int(args[1]), int(args[2]), int(args[3]), size, ms)
            print(f"  [{lineno}] swipe {args[0]},{args[1]} -> {args[2]},{args[3]}", flush=True)
        elif cmd == "fling":
            do_swipe(s, f, int(args[0]), int(args[1]), int(args[2]), int(args[3]),
                     size, ms=120)
            print(f"  [{lineno}] fling {args[0]},{args[1]} -> {args[2]},{args[3]}", flush=True)
        elif cmd == "sleep":
            time.sleep(float(args[0]))
            print(f"  [{lineno}] sleep {args[0]}", flush=True)
        elif cmd == "echo":
            print(f"  [{lineno}] {' '.join(args)}", flush=True)
        else:
            sys.exit(f"FATAL: {script_path}:{lineno}: unknown command '{cmd}'")


def main():
    if len(sys.argv) < 3:
        sys.exit(__doc__)
    path, cmd = sys.argv[1], sys.argv[2]
    need = {"screendump": 4, "tap": 5, "quit": 3, "drive": 5}
    if cmd not in need:
        sys.exit(f"unknown command {cmd}\n{__doc__}")
    if len(sys.argv) < need[cmd]:
        sys.exit(f"{cmd}: missing argument(s)\n{__doc__}")

    size = 720
    if "--size" in sys.argv:
        size = int(sys.argv[sys.argv.index("--size") + 1])

    s = socket.socket(socket.AF_UNIX)
    s.connect(path)
    f = s.makefile("r")
    f.readline()  # greeting banner
    rpc(s, f, {"execute": "qmp_capabilities"})

    if cmd == "screendump":
        rpc(s, f, {"execute": "screendump", "arguments": {"filename": sys.argv[3]}})
    elif cmd == "tap":
        do_tap(s, f, int(sys.argv[3]), int(sys.argv[4]))
    elif cmd == "drive":
        drive(s, f, sys.argv[3], sys.argv[4], size)
    elif cmd == "quit":
        s.sendall(b'{"execute":"quit"}\n')


if __name__ == "__main__":
    main()