From 1dbe4537316e6985641cf2279871df191aa41ae4 Mon Sep 17 00:00:00 2001 From: Ryan Malloy Date: Mon, 17 Aug 2026 10:39:23 -0600 Subject: [PATCH] Add vm_mouse_move: chunked relative PS/2 motion with corner homing and clicks --- README.md | 2 +- src/mcqemu/tools/display.py | 88 ++++++++++++++++++++++++++++++++++++- tests/test_display.py | 77 ++++++++++++++++++++++++++++++++ 3 files changed, 165 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 0c4586a..5bb589d 100644 --- a/README.md +++ b/README.md @@ -29,7 +29,7 @@ claude mcp add mcqemu -- uv run --directory /path/to/mcqemu mcqemu | Sandboxes | `sandbox_vm` (overlay + launch + wait-for-agent in one call), `sandbox_destroy` | | Inspect | `list_vms`, `vm_info` | | Live snapshots | `vm_snapshot_create` / `restore` / `delete` / `list` | -| See & drive | `vm_screenshot` (PNG), `vm_send_keys`, `vm_type_text`, `vm_click`, `vm_serial_read` | +| See & drive | `vm_screenshot` (PNG), `vm_send_keys`, `vm_type_text`, `vm_click`, `vm_mouse_move` (relative PS/2, for guests without tablet drivers), `vm_serial_read` | | Disk images | `image_create`, `image_info`, `image_convert`, `image_resize`, `image_snapshot_*` | | Guest agent | `guest_ping`, `guest_info`, `guest_exec`, `guest_file_read`, `guest_file_write` | diff --git a/src/mcqemu/tools/display.py b/src/mcqemu/tools/display.py index 2abc739..dfe36c0 100644 --- a/src/mcqemu/tools/display.py +++ b/src/mcqemu/tools/display.py @@ -1,8 +1,10 @@ """See & drive: screenshots, keyboard/mouse injection, serial console.""" import asyncio +import math import struct from pathlib import Path +from typing import Literal from fastmcp import Context, FastMCP from fastmcp.exceptions import ToolError @@ -15,6 +17,9 @@ from ._common import require_vm _ABS_MAX = 32767 # QEMU absolute-pointer coordinate space +# PS/2-style relative buttons for HMP mouse_button (bitmask). +_MOUSE_BUTTONS = {"left": 1, "right": 2, "middle": 4} + def _png_dimensions(data: bytes) -> tuple[int, int]: if len(data) < 24 or data[:8] != b"\x89PNG\r\n\x1a\n": @@ -145,6 +150,87 @@ async def vm_click( return {"name": name, "clicked": f"{button} at ({x}, {y})", "double": double} +def _steps(delta: int, step: int) -> list[int]: + """Split a delta into chunks no larger than `step` (preserving sign).""" + if delta == 0: + return [] + sign = 1 if delta > 0 else -1 + full, rest = divmod(abs(delta), step) + return [sign * step] * full + ([sign * rest] if rest else []) + + +async def vm_mouse_move( + name: str, + dx: int = 0, + dy: int = 0, + home: Literal["top-left", "top-right", "bottom-left", "bottom-right"] | None = None, + click: Literal["left", "right", "middle"] | None = None, + double: bool = False, + step: int = 32, + ctx: Context = None, +) -> dict: + """Relative mouse control for guests WITHOUT absolute-pointer (tablet) + drivers — most pre-2010 OSes. Use this when vm_click has no visible + effect. Motion goes to the emulated PS/2 mouse in small steps (<= `step` + px per packet) because guests often desync or apply acceleration on large + deltas. + + Recommended pattern: pass home="bottom-right" (or another corner) to pin + the cursor to a known position first, then dx/dy toward the target, then + vm_screenshot to verify where the cursor actually landed (guest + acceleration may scale motion), correct with further small moves, and + finally click. `click` presses that button after moving; double=True + double-clicks.""" + if click is not None and click not in _MOUSE_BUTTONS: + raise ToolError(f"Unsupported button {click!r}: use left, right, or middle.") + if step < 1 or step > 120: + raise ToolError("step must be between 1 and 120 (PS/2 deltas are small signed bytes).") + + _, record = require_vm(ctx, name) + + homed_steps = 0 + if home is not None: + # Screen size bounds how far the corner can be; overshoot a little so + # the cursor pins against the edge regardless of starting position. + png = await _screendump(ctx, name) + width, height = _png_dimensions(png) + sx = -1 if "left" in home else 1 + sy = -1 if "top" in home else 1 + homed_steps = math.ceil(max(width, height) / step) + 2 + + async with qmp_session(record) as client: + + async def hmp(cmd: str): + return await execute(client, "human-monitor-command", {"command-line": cmd}) + + for _ in range(homed_steps): + await hmp(f"mouse_move {sx * step} {sy * step}") + await asyncio.sleep(0.02) + for chunk_x in _steps(dx, step): + await hmp(f"mouse_move {chunk_x} 0") + await asyncio.sleep(0.02) + for chunk_y in _steps(dy, step): + await hmp(f"mouse_move 0 {chunk_y}") + await asyncio.sleep(0.02) + if click is not None: + for _ in range(2 if double else 1): + await hmp(f"mouse_button {_MOUSE_BUTTONS[click]}") + await asyncio.sleep(0.08) + await hmp("mouse_button 0") + await asyncio.sleep(0.12) + + return { + "name": name, + "homed": home, + "moved": [dx, dy], + "clicked": f"{click}{' double' if double and click else ''}" if click else None, + "note": ( + "Motion is relative and the guest may scale it (acceleration). " + "Take a vm_screenshot to verify the cursor position before clicking." + ), + } + + async def vm_serial_read(name: str, tail_lines: int = 50, ctx: Context = None) -> dict: """Read the last lines of the VM's serial console log. Only useful when the guest writes to its serial port (kernel console=ttyS0, or text-mode @@ -167,5 +253,5 @@ async def vm_serial_read(name: str, tail_lines: int = 50, ctx: Context = None) - def register(mcp: FastMCP) -> None: - for fn in (vm_screenshot, vm_send_keys, vm_type_text, vm_click, vm_serial_read): + for fn in (vm_screenshot, vm_send_keys, vm_type_text, vm_click, vm_mouse_move, vm_serial_read): mcp.tool(fn) diff --git a/tests/test_display.py b/tests/test_display.py index c41db93..4dacacc 100644 --- a/tests/test_display.py +++ b/tests/test_display.py @@ -119,3 +119,80 @@ async def test_serial_read_tail(dirs, fake_qmp): ) assert data["total_lines"] == 100 assert data["tail"] == "line97\nline98\nline99" + + +def hmp_calls(): + return [a["command-line"] for c, a in FakeQMPClient.calls if c == "human-monitor-command"] + + +async def test_mouse_move_chunks_deltas(dirs, fake_qmp): + FakeQMPClient.responses["human-monitor-command"] = "" + write_registry(dirs, seeded_record(dirs, "vm1")) + async with Client(mcp) as client: + data = result_data( + await client.call_tool( + "vm_mouse_move", {"name": "vm1", "dx": 70, "dy": -75, "step": 32} + ) + ) + assert data["moved"] == [70, -75] + assert hmp_calls() == [ + "mouse_move 32 0", + "mouse_move 32 0", + "mouse_move 6 0", + "mouse_move 0 -32", + "mouse_move 0 -32", + "mouse_move 0 -11", + ] + + +async def test_mouse_move_home_pins_to_corner(dirs, fake_qmp): + arm_screendump(width=320, height=200) + FakeQMPClient.responses["human-monitor-command"] = "" + write_registry(dirs, seeded_record(dirs, "vm1")) + async with Client(mcp) as client: + data = result_data( + await client.call_tool( + "vm_mouse_move", {"name": "vm1", "home": "bottom-right", "step": 100} + ) + ) + assert data["homed"] == "bottom-right" + moves = hmp_calls() + # ceil(320/100) + 2 = 6 homing steps, all toward +x/+y + assert moves == ["mouse_move 100 100"] * 6 + + +async def test_mouse_move_home_top_left_direction(dirs, fake_qmp): + arm_screendump(width=100, height=100) + FakeQMPClient.responses["human-monitor-command"] = "" + write_registry(dirs, seeded_record(dirs, "vm1")) + async with Client(mcp) as client: + await client.call_tool("vm_mouse_move", {"name": "vm1", "home": "top-left", "step": 50}) + assert set(hmp_calls()) == {"mouse_move -50 -50"} + + +async def test_mouse_move_click_sequence(dirs, fake_qmp): + FakeQMPClient.responses["human-monitor-command"] = "" + write_registry(dirs, seeded_record(dirs, "vm1")) + async with Client(mcp) as client: + data = result_data( + await client.call_tool( + "vm_mouse_move", {"name": "vm1", "click": "left", "double": True} + ) + ) + assert data["clicked"] == "left double" + assert hmp_calls() == ["mouse_button 1", "mouse_button 0"] * 2 + + +async def test_mouse_move_right_click(dirs, fake_qmp): + FakeQMPClient.responses["human-monitor-command"] = "" + write_registry(dirs, seeded_record(dirs, "vm1")) + async with Client(mcp) as client: + await client.call_tool("vm_mouse_move", {"name": "vm1", "click": "right"}) + assert hmp_calls() == ["mouse_button 2", "mouse_button 0"] + + +async def test_mouse_move_rejects_bad_step(dirs, fake_qmp): + write_registry(dirs, seeded_record(dirs, "vm1")) + async with Client(mcp) as client: + with pytest.raises(ToolError, match="step"): + await client.call_tool("vm_mouse_move", {"name": "vm1", "dx": 10, "step": 500})