"""MCP 工具层：13 个高层工具，统一输出契约（与 CLI --json 同构）。 成功 {"ok": true, "tool": name, "data": {...}} 失败 {"ok": false, "tool": name, "error": {"code", "message", "hint"?}} """ from __future__ import annotations import asyncio import base64 import binascii import datetime import io import os from PIL import Image from bleak.exc import BleakDeviceNotFoundError from .client import PPClient from .device_manager import TRANSPORT_ERRORS _TIMEOUT_HINT = "设备长睡眠、广播窗口 30-60s；调大 --connect-timeout 或稍后重试" _NOTFOUND_HINT = "未发现 NRF- 前缀设备；确认设备在位且未处于深睡" def decode_image_source(source: str) -> Image.Image: """source = 服务器本地路径，或 base64（可带 data URI 前缀）。""" if os.path.exists(source): try: return Image.open(source) except OSError: # 文件存在但不是可识别图片（含截断） raise ValueError("source 不是可识别的图片") from None data = source if data.startswith("data:"): if "," not in data: raise ValueError("data URI 缺少 ',' 分隔") data = data.split(",", 1)[1] try: raw = base64.b64decode(data, validate=True) except (binascii.Error, ValueError): raise ValueError("source 既不是存在的文件路径，也不是合法 base64") from None try: return Image.open(io.BytesIO(raw)) except OSError: # 合法 base64 但不是可识别图片（PIL.UnidentifiedImageError 属 OSError） raise ValueError("source 不是可识别的图片") from None def _err(tool: str, code: str, message: str, hint: str | None = None) -> dict: error = {"code": code, "message": message} if hint: error["hint"] = hint return {"ok": False, "tool": tool, "error": error} def _map_transport_error(tool: str, exc: Exception) -> dict: """传输层异常全家（TRANSPORT_ERRORS）→ 统一错误码。 真实硬件路径的异常来源：SDK 语义 TransportError（base/local 两个同名类）、 bleak 裸抛 BleakError、平台层 OSError / asyncio.TimeoutError。 类型判定优先于消息匹配（BleakDeviceNotFoundError 的英文消息含 "not found"， 但语义是"已知 MAC 守候窗口内未出现"，归 CONNECT_TIMEOUT 而非 DEVICE_NOT_FOUND）。 """ msg = str(exc) low = msg.lower() if isinstance(exc, (asyncio.TimeoutError, BleakDeviceNotFoundError)): return _err(tool, "CONNECT_TIMEOUT", msg, _TIMEOUT_HINT) if "超时" in msg or "timeout" in low or "timed out" in low: return _err(tool, "CONNECT_TIMEOUT", msg, _TIMEOUT_HINT) if "未找到" in msg: return _err(tool, "DEVICE_NOT_FOUND", msg, _NOTFOUND_HINT) if "连接失败" in msg: # 已知 MAC 守候未连上（bleak not found 的包装形态） return _err(tool, "CONNECT_TIMEOUT", msg, _TIMEOUT_HINT) if "not found" in low or "no device" in low: return _err(tool, "DEVICE_NOT_FOUND", msg, _NOTFOUND_HINT) return _err(tool, "BLE_ERROR", msg) def register_tools(mcp, manager) -> None: """把 13 个工具注册到 FastMCP 实例；manager 为 DeviceManager（测试可注入 fake）。""" async def _call(tool: str, op) -> dict: try: data = await manager.run(op) return {"ok": True, "tool": tool, "data": data if data is not None else {}} except TRANSPORT_ERRORS as exc: return _map_transport_error(tool, exc) except ValueError as exc: return _err(tool, "INVALID_PARAM", str(exc)) except Exception as exc: # noqa: BLE001 - 工具层兜底，错误须回 agent 而非炸会话 return _err(tool, "INTERNAL", f"{type(exc).__name__}: {exc}") @mcp.tool() async def scan_devices(timeout: float = 5.0) -> dict: """扫描附近 NRF- 前缀墨水屏设备。返回形状：data.devices = [{name,address,rssi}]。""" try: devices = await PPClient.scan_devices("local", timeout=timeout) return {"ok": True, "tool": "scan_devices", "data": {"devices": devices}} except TRANSPORT_ERRORS as exc: return _map_transport_error("scan_devices", exc) except Exception as exc: # noqa: BLE001 return _err("scan_devices", "INTERNAL", f"{type(exc).__name__}: {exc}") @mcp.tool() async def device_status() -> dict: """查询连接状态/绑定 MAC/设备 ID/空闲秒数。 注意：会触发连接守候（最长 --connect-timeout 秒）并刷新空闲计时—— 周期轮询本工具会阻止 idle 自动断开。 """ try: return {"ok": True, "tool": "device_status", "data": await manager.status()} except TRANSPORT_ERRORS as exc: return _map_transport_error("device_status", exc) except Exception as exc: # noqa: BLE001 return _err("device_status", "INTERNAL", f"{type(exc).__name__}: {exc}") @mcp.tool() async def set_time(tz: float = 8.0) -> dict: """对时；tz 为时区偏移（默认 8=北京时间）。""" return await _call("set_time", lambda dev: dev.set_time(tz=tz)) @mcp.tool() async def set_mode(mode: str) -> dict: """切换显示模式：clock1-3/calendar1-3/image0-3/tricolor/mono。""" return await _call("set_mode", lambda dev: dev.set_mode(mode)) @mcp.tool() async def toggle(name: str) -> dict: """单字节切换：invert/font/rotate180/hour_format/clock_color。""" return await _call("toggle", lambda dev: dev.toggle(name)) @mcp.tool() async def upload_image(source: str, slot: int = 0, algo: str = "atkinson", mono: bool = False, threshold: float | None = None, diffusion: float | None = None, brightness: float | None = None, contrast: float | None = None, saturation: float | None = None, rotate: float | None = None) -> dict: """传图到指定槽位。source=服务器路径或 base64；algo∈ none/floydsteinberg/atkinson/bayer/stucki/jarvis。""" try: adjust = {k: v for k, v in { "threshold": threshold, "diffusion": diffusion, "brightness": brightness, "contrast": contrast, "saturation": saturation, "rotate": rotate}.items() if v is not None} img = decode_image_source(source) except ValueError as exc: return _err("upload_image", "INVALID_PARAM", str(exc)) return await _call("upload_image", lambda dev: dev.upload_image( img, slot=slot, algo=algo, mono=mono, **adjust)) @mcp.tool() async def render_template(name: str, payload: dict | None = None, slot: int = 0, algo: str = "atkinson") -> dict: """本地渲染模板并上传：schedule/businesscard/memo/course/qrcode/custom。""" return await _call("render_template", lambda dev: dev.render_template( name, payload, slot=slot, algo=algo)) @mcp.tool() async def countdown(date: str, mode: str = "clock", prefix: str | None = None) -> dict: """倒计时；date 为 ISO YYYY-MM-DD；mode=clock/calendar。""" try: d = datetime.date.fromisoformat(date) except ValueError as exc: return _err("countdown", "INVALID_PARAM", str(exc)) return await _call("countdown", lambda dev: dev.countdown( d, mode=mode, prefix=prefix)) @mcp.tool() async def countdown_off() -> dict: """关闭倒计时。""" return await _call("countdown_off", lambda dev: dev.countdown_off()) @mcp.tool() async def calendar_text(text: str) -> dict: """设置日历中文文字并切到日历模式一。""" return await _call("calendar_text", lambda dev: dev.calendar_text(text)) @mcp.tool() async def set_sleep(on: bool, start_h: int, end_h: int) -> dict: """设置休眠时段（0-23 时）。""" return await _call("set_sleep", lambda dev: dev.sleep(on, start_h, end_h)) @mcp.tool() async def clear_screen() -> dict: """刷屏（EPD 清屏）。""" return await _call("clear_screen", lambda dev: dev.clear()) @mcp.tool() async def raw_send(data_hex: str, channel: str = "alt") -> dict: """逃生舱：直发十六进制字节串；channel=rxtx/epd/alt。""" try: data = bytes.fromhex(data_hex) except ValueError as exc: return _err("raw_send", "INVALID_PARAM", str(exc)) return await _call("raw_send", lambda dev: dev.raw(data, channel))