# -*- coding: utf-8 -*- """ verify_annotate.py 批注工具集运行时验证脚本。 前提：CAD 已打开并加载 cad_mcp_frame.arx（插件 DLL），127.0.0.1:8080 在监听。 用法： python verify_annotate.py # 绘制 -> 校验 -> 清理（默认清理） python verify_annotate.py --keep # 保留批注，便于肉眼观察 覆盖工具： draw_revision_cloud / add_text / add_mtext / add_polyline / add_line / add_leader_annotation / set_entity_properties / delete_entities """ import socket import json import sys HOST, PORT = "127.0.0.1", 8080 ANNOT_LAYER = "AI_REVIEW" # ---------------------------------------------------------------- MCP 通信 def rpc(method, params=None, timeout=20): s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.settimeout(timeout) s.connect((HOST, PORT)) req = {"jsonrpc": "2.0", "id": 1, "method": method} if params is not None: req["params"] = params s.sendall(json.dumps(req).encode("utf-8")) data = b"" while True: chunk = s.recv(8192) if not chunk: break data += chunk if len(chunk) < 8192: break s.close() return json.loads(data.decode("utf-8")) def call_tool(name, arguments): return rpc("tools/call", {"name": name, "arguments": arguments}) def payload(resp): """从 tools/call 响应里取出数据类工具返回的内层 JSON。""" if resp is None: return False, "no response" if "error" in resp: return False, resp["error"] try: for item in resp["result"]["content"]: if item.get("type") == "text": return True, json.loads(item["text"]) except Exception as e: # noqa: BLE001 return False, "parse error: %s" % e return False, "no text content" # ---------------------------------------------------------------- 测试驱动 RESULTS = [] def record(name, ok, detail): RESULTS.append((name, ok, detail)) print(" [%s] %-24s %s" % ("PASS" if ok else "FAIL", name, detail)) def expect_created(name, resp, want_type): """校验工具返回 ok 且 created[0].type == want_type，返回 handle 列表。""" ok, obj = payload(resp) if not ok: record(name, False, "调用失败: %s" % obj) return [] if not obj.get("ok"): record(name, False, "返回非 ok: %s" % obj) return [] created = obj.get("created", []) handles = [c.get("handle", "") for c in created] types = [c.get("type", "") for c in created] layers = [c.get("layer", "") for c in created] if want_type not in types: record(name, False, "类型不符: 期望 %s 实际 %s" % (want_type, types)) return handles bad_layer = [l for l in layers if l != ANNOT_LAYER] if bad_layer: record(name, False, "图层不符: %s" % layers) return handles record(name, True, "created=%s layer=%s" % (types, layers)) return handles def main(): keep = "--keep" in sys.argv print("=== 批注工具集运行时验证 ===") print("目标: %s:%d\n" % (HOST, PORT)) # 0) 连接 & 工具清单 try: tools_resp = rpc("tools/list") except Exception as e: # noqa: BLE001 print("[错误] 无法连接 MCP 服务: %s" % e) print("请确认 CAD 已打开、插件已加载、8080 端口在监听。") return 2 names = [t["name"] for t in tools_resp.get("result", {}).get("tools", [])] want = ["draw_revision_cloud", "add_text", "add_mtext", "add_polyline", "add_line", "add_leader_annotation", "delete_entities", "set_entity_properties"] missing = [w for w in want if w not in names] print("工具总数: %d" % len(names)) if missing: print("[FAIL] 未注册的工具: %s" % missing) return 1 print("[PASS] 8 个批注工具均已注册\n") created_handles = [] # 1) 云线（矩形） print("1) 绘制工具") h = expect_created("draw_revision_cloud", call_tool("draw_revision_cloud", { "shape": "rect", "min_x": 0, "min_y": 0, "max_x": 100, "max_y": 60, }), "LWPOLYLINE") created_handles += h # 2) 单行文字 h = expect_created("add_text", call_tool("add_text", { "text": "AI批注-单行", "position": [0, 80], "height": 5, }), "TEXT") created_handles += h # 3) 多行文字 h = expect_created("add_mtext", call_tool("add_mtext", { "text": "AI批注-多行\n第二行", "position": [0, 100], "height": 5, }), "MTEXT") created_handles += h # 4) 多段线 h = expect_created("add_polyline", call_tool("add_polyline", { "points": [[0, 130], [60, 130], [60, 170]], "closed": False, }), "LWPOLYLINE") created_handles += h # 5) 直线 h = expect_created("add_line", call_tool("add_line", { "start": [0, 190], "end": [100, 190], }), "LINE") created_handles += h # 6) 引线批注（返回 MTEXT + LEADER 两个） h = expect_created("add_leader_annotation", call_tool("add_leader_annotation", { "text": "引线批注", "points": [[300, 200], [200, 150]], "height": 5, }), "LEADER") created_handles += h # 7) 下钻校验：取第一个实体详情，确认图层 print("\n2) 校验") if created_handles: ok, obj = payload(call_tool("get_entity_detail", {"handle": created_handles[0]})) if ok and obj.get("layer") == ANNOT_LAYER: record("get_entity_detail", True, "handle=%s layer=%s" % (created_handles[0], obj.get("layer"))) else: record("get_entity_detail", False, "%s" % obj) else: record("get_entity_detail", False, "无可用 handle") # 8) 修改实体属性（把第一个实体改红） if created_handles: ok, obj = payload(call_tool("set_entity_properties", { "handle": created_handles[0], "color": 1, })) record("set_entity_properties", bool(ok and obj.get("count") == 1), "%s" % obj) # 9) 清理 print("\n3) 清理") if keep: record("delete_entities", True, "已跳过（--keep）；保留 handle: %s" % created_handles) elif created_handles: ok, obj = payload(call_tool("delete_entities", {"handles": created_handles})) good = ok and obj.get("count") == len(created_handles) record("delete_entities", good, "删除 %s / 共 %s" % (obj.get("count"), len(created_handles))) else: record("delete_entities", False, "无 handle 可删") # 汇总 print("\n=== 结果汇总 ===") passed = sum(1 for _, ok, _ in RESULTS if ok) for name, ok, detail in RESULTS: print(" %-4s %-24s %s" % ("OK" if ok else "NG", name, detail)) print("\n通过 %d / %d" % (passed, len(RESULTS))) return 0 if passed == len(RESULTS) else 1 if __name__ == "__main__": sys.exit(main())