cad-agent/verify_annotate.py

# -*- 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())