mirror of
https://github.com/kuhyx/utils.git
synced 2026-11-10 21:27:14 +01:00
Krzysztof kuhy Rudnicki
694c013b6b
Some checks failed
dependency freshness / dependencies on newest stable (push) Has been cancelled
file length / file-length (push) Has been cancelled
md-naming / markdown naming convention (push) Has been cancelled
palette drift / palette (push) Has been cancelled
shared gate tests / gate unit tests (push) Has been cancelled
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
95 lines
3.1 KiB
Python
Executable File
95 lines
3.1 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""One-shot QMP client for vmbox: issue a command and print the reply.
|
|
|
|
Used for actions that need a round trip rather than passive event recording --
|
|
notably `screendump`, which is how locker/X11 tests get a screenshot without
|
|
any VNC or SPICE viewer installed on the host.
|
|
|
|
Usage:
|
|
qmp.py <qmp.sock> screendump <out.png>
|
|
qmp.py <qmp.sock> status
|
|
qmp.py <qmp.sock> pause-on-shutdown # a guest poweroff/hibernate leaves
|
|
# QEMU alive in 'shutdown' (vm wake)
|
|
qmp.py <qmp.sock> quit
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import socket
|
|
import sys
|
|
from typing import Any, TextIO
|
|
|
|
|
|
def open_session(path: str) -> tuple[socket.socket, TextIO]:
|
|
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
sock.settimeout(20)
|
|
sock.connect(path)
|
|
stream = sock.makefile("rw", encoding="utf-8", newline="\n")
|
|
stream.readline() # greeting
|
|
command(stream, "qmp_capabilities")
|
|
return sock, stream
|
|
|
|
|
|
def command(stream: TextIO, name: str, **args: Any) -> dict[str, Any]:
|
|
"""Send one command and return its reply, skipping any events in between."""
|
|
payload: dict[str, Any] = {"execute": name}
|
|
if args:
|
|
payload["arguments"] = args
|
|
stream.write(json.dumps(payload) + "\n")
|
|
stream.flush()
|
|
while True:
|
|
line = stream.readline()
|
|
if not line:
|
|
raise SystemExit(f"qmp: connection closed during '{name}'")
|
|
msg = json.loads(line)
|
|
if "return" in msg or "error" in msg:
|
|
return msg
|
|
|
|
|
|
def main() -> int:
|
|
if len(sys.argv) < 3:
|
|
print(__doc__, file=sys.stderr)
|
|
return 2
|
|
path, action = sys.argv[1], sys.argv[2]
|
|
|
|
if not os.path.exists(path):
|
|
print(f"qmp: no such socket: {path} (is the sandbox running?)", file=sys.stderr)
|
|
return 1
|
|
|
|
sock, stream = open_session(path)
|
|
try:
|
|
if action == "screendump":
|
|
if len(sys.argv) < 4:
|
|
print("qmp: screendump needs an output path", file=sys.stderr)
|
|
return 2
|
|
out = os.path.abspath(sys.argv[3])
|
|
# PNG is native to qemu's screendump; no ImageMagick needed.
|
|
reply = command(stream, "screendump", filename=out, format="png")
|
|
elif action == "status":
|
|
reply = command(stream, "query-status")
|
|
elif action == "pause-on-shutdown":
|
|
# Runtime equivalent of -no-shutdown, so launch defaults stay as
|
|
# they are: the machine stops, but its CMOS (and the RTC alarm in
|
|
# it) stays readable instead of vanishing with the process.
|
|
reply = command(stream, "set-action", shutdown="pause")
|
|
elif action == "quit":
|
|
reply = command(stream, "quit")
|
|
else:
|
|
print(f"qmp: unknown action '{action}'", file=sys.stderr)
|
|
return 2
|
|
|
|
if "error" in reply:
|
|
print(f"qmp: {reply['error'].get('desc', reply['error'])}", file=sys.stderr)
|
|
return 1
|
|
print(json.dumps(reply.get("return", {})))
|
|
return 0
|
|
finally:
|
|
stream.close()
|
|
sock.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|