feat(M1): MCP 截图坐标空间统一——de_screenshot 附 native_size,de_tap/de_swipe 接受截图坐标由 server 按比例换算原生(模型只感知截图坐标系)
This commit is contained in:
@@ -18,6 +18,11 @@ logging.basicConfig(level=logging.INFO,
|
|||||||
_log = logging.getLogger("mcp")
|
_log = logging.getLogger("mcp")
|
||||||
|
|
||||||
S = config.Settings()
|
S = config.Settings()
|
||||||
|
|
||||||
|
# 坐标空间缓存:serial -> (display_w, display_h, native_w, native_h)
|
||||||
|
# de_screenshot 返回的图与 de_tap/de_swipe 的坐标同一空间(display),
|
||||||
|
# server 按比例换算为设备原生坐标(模型只感知截图坐标系,不感知原生分辨率)。
|
||||||
|
_coord = {}
|
||||||
audit.init(S.audit_file)
|
audit.init(S.audit_file)
|
||||||
|
|
||||||
mcp = FastMCP("mobile-control")
|
mcp = FastMCP("mobile-control")
|
||||||
@@ -47,6 +52,16 @@ def _check_write():
|
|||||||
raise PlatformError("write_disabled", "写操作未启用(MCP_ALLOW_WRITE=1 开启)")
|
raise PlatformError("write_disabled", "写操作未启用(MCP_ALLOW_WRITE=1 开启)")
|
||||||
|
|
||||||
|
|
||||||
|
def _to_native(serial, x, y):
|
||||||
|
"""截图坐标 → 设备原生坐标(按最近一次截图的比例换算)。"""
|
||||||
|
c = _coord.get(serial)
|
||||||
|
if not c:
|
||||||
|
raise PlatformError("invalid_param",
|
||||||
|
"请先对该设备执行 de_screenshot(需要建立坐标空间)")
|
||||||
|
dw, dh, nw, nh = c
|
||||||
|
return (round(x * nw / dw), round(y * nh / dh))
|
||||||
|
|
||||||
|
|
||||||
def _err(e: PlatformError):
|
def _err(e: PlatformError):
|
||||||
return {"ok": False, "error": {"code": e.code, "message": e.message}}
|
return {"ok": False, "error": {"code": e.code, "message": e.message}}
|
||||||
|
|
||||||
@@ -96,40 +111,52 @@ def de_screenshot(serial: str) -> dict:
|
|||||||
buf = io.BytesIO()
|
buf = io.BytesIO()
|
||||||
img.convert("RGB").save(buf, "JPEG", quality=S.jpeg_quality)
|
img.convert("RGB").save(buf, "JPEG", quality=S.jpeg_quality)
|
||||||
data = base64.b64encode(buf.getvalue()).decode()
|
data = base64.b64encode(buf.getvalue()).decode()
|
||||||
|
# 记录坐标空间(display=返回图尺寸,native=设备原生),供 tap/swipe 换算
|
||||||
|
nw, nh = platform().screen_size(serial)
|
||||||
|
_coord[serial] = (img.size[0], img.size[1], nw, nh)
|
||||||
except PlatformError as e:
|
except PlatformError as e:
|
||||||
return _err(e)
|
return _err(e)
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
return _err(PlatformError("platform_unavailable", f"截图处理失败: {e}"))
|
return _err(PlatformError("platform_unavailable", f"截图处理失败: {e}"))
|
||||||
audit.audit("de_screenshot", serial, f"{w}x{h}", "ok")
|
audit.audit("de_screenshot", serial, f"{w}x{h}", "ok")
|
||||||
|
dw, dh, nw, nh = _coord[serial]
|
||||||
return _ok({"image": {"type": "image", "data": data,
|
return _ok({"image": {"type": "image", "data": data,
|
||||||
"mimeType": "image/jpeg"},
|
"mimeType": "image/jpeg"},
|
||||||
"width": w, "height": h,
|
"width": dw, "height": dh,
|
||||||
|
"native_size": {"width": nw, "height": nh},
|
||||||
"screen_state": screen_state})
|
"screen_state": screen_state})
|
||||||
|
|
||||||
|
|
||||||
@mcp.tool()
|
@mcp.tool()
|
||||||
def de_tap(serial: str, x: int, y: int) -> dict:
|
def de_tap(serial: str, x: int, y: int) -> dict:
|
||||||
"""点击设备屏幕指定坐标(设备原生分辨率像素,与截图 1:1 换算)。"""
|
"""点击设备屏幕指定坐标。
|
||||||
|
|
||||||
|
坐标空间 = de_screenshot 返回的图像坐标(display 空间)——先截图拿到
|
||||||
|
native_size 后再点击,server 自动换算为设备原生坐标。
|
||||||
|
"""
|
||||||
try:
|
try:
|
||||||
_check_write()
|
_check_write()
|
||||||
serial = _check_serial(serial)
|
serial = _check_serial(serial)
|
||||||
if x < 0 or y < 0:
|
if x < 0 or y < 0:
|
||||||
raise PlatformError("invalid_param", "坐标不能为负")
|
raise PlatformError("invalid_param", "坐标不能为负")
|
||||||
platform().tap(serial, x, y)
|
nx, ny = _to_native(serial, x, y)
|
||||||
|
platform().tap(serial, nx, ny)
|
||||||
except PlatformError as e:
|
except PlatformError as e:
|
||||||
return _err(e)
|
return _err(e)
|
||||||
audit.audit("de_tap", serial, f"({x},{y})", "ok")
|
audit.audit("de_tap", serial, f"({x},{y})->native({nx},{ny})", "ok")
|
||||||
return _ok({"action": "tap", "serial": serial, "x": x, "y": y})
|
return _ok({"action": "tap", "serial": serial, "x": x, "y": y})
|
||||||
|
|
||||||
|
|
||||||
@mcp.tool()
|
@mcp.tool()
|
||||||
def de_swipe(serial: str, x1: int, y1: int, x2: int, y2: int,
|
def de_swipe(serial: str, x1: int, y1: int, x2: int, y2: int,
|
||||||
duration: float = 0.2) -> dict:
|
duration: float = 0.2) -> dict:
|
||||||
"""在设备屏幕上滑动(起终点为原生分辨率像素)。"""
|
"""在设备屏幕上滑动(坐标空间同 de_tap:截图坐标,server 换算原生)。"""
|
||||||
try:
|
try:
|
||||||
_check_write()
|
_check_write()
|
||||||
serial = _check_serial(serial)
|
serial = _check_serial(serial)
|
||||||
platform().swipe(serial, x1, y1, x2, y2, duration)
|
nx1, ny1 = _to_native(serial, x1, y1)
|
||||||
|
nx2, ny2 = _to_native(serial, x2, y2)
|
||||||
|
platform().swipe(serial, nx1, ny1, nx2, ny2, duration)
|
||||||
except PlatformError as e:
|
except PlatformError as e:
|
||||||
return _err(e)
|
return _err(e)
|
||||||
audit.audit("de_swipe", serial, f"({x1},{y1})->({x2},{y2})", "ok")
|
audit.audit("de_swipe", serial, f"({x1},{y1})->({x2},{y2})", "ok")
|
||||||
|
|||||||
@@ -80,6 +80,19 @@ class PlatformClient:
|
|||||||
f"截图 HTTP {r.status_code}")
|
f"截图 HTTP {r.status_code}")
|
||||||
return r.content, r.headers.get("X-Screen-State", "unknown")
|
return r.content, r.headers.get("X-Screen-State", "unknown")
|
||||||
|
|
||||||
|
def screen_size(self, serial):
|
||||||
|
"""屏幕原生分辨率(GET /api/screen/size),返回 (w, h)。"""
|
||||||
|
r = self._get("/api/screen/size", serial=serial)
|
||||||
|
if r.status_code == 503:
|
||||||
|
raise PlatformError("device_offline", r.text[:120])
|
||||||
|
if r.status_code != 200:
|
||||||
|
raise PlatformError("platform_unavailable",
|
||||||
|
f"分辨率 HTTP {r.status_code}")
|
||||||
|
j = r.json() or {}
|
||||||
|
if not j.get("ok"):
|
||||||
|
raise PlatformError("device_offline", str(j.get("error", "取分辨率失败"))[:120])
|
||||||
|
return int(j["width"]), int(j["height"])
|
||||||
|
|
||||||
def tap(self, serial, x, y):
|
def tap(self, serial, x, y):
|
||||||
"""点击(POST /api/screen/tap)。"""
|
"""点击(POST /api/screen/tap)。"""
|
||||||
r = self._post("/api/screen/tap", json={"serial": serial,
|
r = self._post("/api/screen/tap", json={"serial": serial,
|
||||||
|
|||||||
Reference in New Issue
Block a user