267 lines
11 KiB
Python
267 lines
11 KiB
Python
|
# -*- 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())
|