# -*- coding: utf-8 -*- """ verify_collisions.py 干涉/碰撞候选工具（find_collisions）运行时验证脚本。 前提：CAD 已打开并加载 cad_mcp_frame.arx（插件 DLL），127.0.0.1:8080 在监听。 用法： python verify_collisions.py # 全模型 + 每个图框各跑一次 python verify_collisions.py --top 15 # 打印前 N 条候选（默认 10） python verify_collisions.py --window # 只跑全模型，跳过图框内 覆盖： - 工具已注册（tools/list） - 全模型粗筛：entity_count / total 候选数 / ratio 分布 - min_overlap 阈值过滤（阈值升高 → 候选数不增） - 分页（limit / offset） - narrow=true 细筛（返回项带 exact 字段） - 逐图框（detect_drawing_frames 的 window）候选数 本脚本**只读**，不做任何写库操作，无需清理。 """ import socket import json import sys HOST, PORT = "127.0.0.1", 8080 # ---------------------------------------------------------------- MCP 通信 def rpc(method, params=None, timeout=60): 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(65536) if not chunk: break data += chunk if len(chunk) < 65536: 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] %-28s %s" % ("PASS" if ok else "FAIL", name, detail)) def collisions(**args): """调用 find_collisions 并返回 (ok, envelope)。""" return payload(call_tool("find_collisions", args)) def ratio_of(item): ov = item.get("overlap") or {} r = ov.get("ratio") return r if isinstance(r, (int, float)) else -1.0 def print_top(env, top): items = env.get("items") or [] if not items: print(" （无候选）") return print(" 前 %d 条候选（按重叠比降序）:" % min(top, len(items))) for it in items[:top]: a, b = it.get("a") or {}, it.get("b") or {} print(" ratio=%.3f %s/%s × %s/%s 同层=%s" % ( ratio_of(it), a.get("type"), a.get("handle"), b.get("type"), b.get("handle"), it.get("same_layer"))) def main(): top = 10 if "--top" in sys.argv: try: top = int(sys.argv[sys.argv.index("--top") + 1]) except (IndexError, ValueError): pass frames_only = "--window" in sys.argv print("=== find_collisions 运行时验证 ===") 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", [])] print("工具总数: %d" % len(names)) if "find_collisions" not in names: record("find_collisions 已注册", False, "tools/list 中不存在") return 1 record("find_collisions 已注册", True, "tool 总数 %d" % len(names)) # 1) 全模型粗筛 print("\n1) 全模型粗筛") ok, env = collisions(min_overlap=0.0, limit=200) if not ok: record("find_collisions 全模型", False, "%s" % env) return 1 total = env.get("total", 0) ent = env.get("entity_count", 0) record("find_collisions 全模型", True, "entity_count=%s total 候选=%s returned=%s" % (ent, total, env.get("returned"))) conts = env.get("excluded_containers") or [] if conts: print(" 已剔除容器 %d 个（图框块 / 分区边界）:" % len(conts)) for c in conts: print(" %-8s %-14s reason=%s deg=%s" % ( c.get("handle"), c.get("type"), c.get("reason"), c.get("degree"))) else: print(" （未剔除容器）") record("容器过滤生效", len(conts) >= 1, "剔除 %d 个容器" % len(conts)) print_top(env, top) # ratio 单调性（返回项应降序） items = env.get("items") or [] ratios = [ratio_of(i) for i in items] record("候选按重叠比降序", ratios == sorted(ratios, reverse=True), "%s" % ratios[:8]) # 2) min_overlap 阈值过滤 print("\n2) min_overlap 阈值过滤") ok2, env2 = collisions(min_overlap=0.5, limit=200) t2 = env2.get("total", -1) if ok2 else -1 record("min_overlap=0.5 过滤", ok2 and 0 <= t2 <= total, "total %s -> %s（阈值升高候选数不应增加）" % (total, t2)) ok3, env3 = collisions(min_overlap=0.99, limit=200) t3 = env3.get("total", -1) if ok3 else -1 record("min_overlap=0.99 过滤", ok3 and 0 <= t3 <= t2, "total %s -> %s" % (t2, t3)) # 3) 分页 print("\n3) 分页（limit / offset）") ok4, page1 = collisions(min_overlap=0.0, limit=3, offset=0) ok5, page2 = collisions(min_overlap=0.0, limit=3, offset=3) r1 = page1.get("returned") if ok4 else -1 r2 = page2.get("returned") if ok5 else -1 same_head = (page1.get("items") or [])[:1] == (page2.get("items") or [])[:1] record("limit=3 首页", ok4 and r1 == min(3, total), "returned=%s" % r1) record("offset=3 次页", ok5 and r2 == min(3, max(0, total - 3)), "returned=%s（应与首页不同）" % r2) record("分页不重复", not same_head or total <= 3, "首页首条 != 次页首条") # 4) narrow 细筛 print("\n4) narrow 细筛（intersectWith）") ok6, envn = collisions(min_overlap=0.0, limit=5, narrow=True) nitems = envn.get("items") or [] if ok6 else [] has_exact = all("exact" in i for i in nitems) if nitems else False exact_true = sum(1 for i in nitems if i.get("exact")) record("narrow 返回 exact 字段", ok6 and has_exact, "%d/%d 条 exact=true" % (exact_true, len(nitems))) # 5) 逐图框 if not frames_only: print("\n5) 逐图框候选") okf, fenv = payload(call_tool("detect_drawing_frames", {})) frames = (fenv.get("items") or []) if okf else [] if not frames: record("detect_drawing_frames", False, "未识别到图框") else: print(" 识别到 %d 个图框" % len(frames)) for idx, fr in enumerate(frames, 1): bbox = fr.get("bbox") if not bbox or len(bbox) != 4: record("图框%d 干涉" % idx, False, "bbox 缺失") continue win = {"min_x": bbox[0], "min_y": bbox[1], "max_x": bbox[2], "max_y": bbox[3]} okw, wev = collisions(window=win, min_overlap=0.5, limit=50) if not okw: record("图框%d 干涉" % idx, False, "%s" % wev) continue record("图框%d 干涉" % idx, True, "kind=%s entity_count=%s total(>=0.5)=%s" % ( fr.get("kind"), wev.get("entity_count"), wev.get("total"))) print_top(wev, min(top, 5)) # 6) 注入式正例：人为制造一处深度干涉，验证工具能报出（随后清理） if "--inject" in sys.argv: print("\n6) 注入式正例（人为干涉 → 应报出 → 清理）") # 放在所有图框之外（union max_x≈390012），避免与真实图元相交 ax, ay = 500000.0, 0.0 rectA = [[ax, ay], [ax + 400, ay], [ax + 400, ay + 400], [ax, ay + 400]] rectB = [[ax + 20, ay + 20], [ax + 420, ay + 20], [ax + 420, ay + 420], [ax + 20, ay + 420]] handles = [] for name, pts in (("rectA", rectA), ("rectB", rectB)): okp, penv = payload(call_tool("add_polyline", {"points": pts, "closed": True})) hs = [c.get("handle") for c in (penv.get("created") or [])] if okp else [] handles += [h for h in hs if h] record("注入 %s" % name, bool(okp and hs), "handle=%s" % hs) try: # 期望重叠比 ≈ 0.9025（380/400 的平方）；min_overlap=0.5 应能命中 oki, iev = collisions(min_overlap=0.5, limit=5) items = iev.get("items") or [] if oki else [] top_item = items[0] if items else None got = None if top_item: ha = (top_item.get("a") or {}).get("handle") hb = (top_item.get("b") or {}).get("handle") if {ha, hb} == set(handles): got = top_item if got: record("报出注入干涉", True, "ratio=%.3f %s × %s" % (ratio_of(got), (got.get("a") or {}).get("handle"), (got.get("b") or {}).get("handle"))) else: record("报出注入干涉", False, "首位候选不是注入对: %s" % (top_item and ( (top_item.get("a") or {}).get("handle"), (top_item.get("b") or {}).get("handle")))) finally: if handles: okd, denv = payload(call_tool("delete_entities", {"handles": handles})) record("清理注入实体", bool(okd and denv.get("count") == len(handles)), "删除 %s / 共 %s" % (denv.get("count"), len(handles))) else: record("清理注入实体", False, "无 handle 可删") # 汇总 print("\n=== 结果汇总 ===") passed = sum(1 for _, ok, _ in RESULTS if ok) for name, ok, detail in RESULTS: print(" %-4s %-28s %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())