#!/usr/bin/env python3 # -*- coding: utf-8 -*- """Сборка условных обозначений по электрике через MCP-сервер AutoCAD. Повторяет то, что было сделано 01.10.2026: каждый символ собирается из простых примитивов в ИМЕНОВАННЫЙ блок с атрибутами (ATTDEF), затем вставляется на чертёж и сохраняется в DXF. Запуск: python3 build_symbols_mcp.py --port 30041 --out ./out (токен берётся из /opt/data/.autocad_mcp_token, либо из переменной ACAD_MCP_TOKEN) Проверка после сборки — командой inspect_dxf.py или block_list/block_get_attributes. """ from __future__ import annotations import argparse import base64 import json import os import urllib.request # --- Описание символов: примитивы и атрибуты (единицы чертежа = мм плана) ------- SYMBOLS = { "СВ_ВСТР": { "layer": "СО_СВЕТИЛЬНИКИ", "primitives": [ ("entity_create_circle", {"cx": 0, "cy": 0, "radius": 150}), ("entity_create_line", {"x1": -106, "y1": -106, "x2": 106, "y2": 106}), ("entity_create_line", {"x1": -106, "y1": 106, "x2": 106, "y2": -106}), ], "attdefs": [ ("МАРКА", "Марка светильника", -60, 200, 120, "Т1"), ("ТИП", "Тип по спецификации", -300, -320, 90, "встраиваемый"), ("ВЫСОТА", "Высота от пола, мм", -120, -450, 90, "0"), ("ГРУППА", "Группа включения", -60, 350, 90, "1.1"), ], }, "СВ_БРА": { "layer": "СО_СВЕТИЛЬНИКИ", "primitives": [ ("entity_create_arc", {"cx": 0, "cy": 0, "radius": 150, "start_angle": 0, "end_angle": 180}), ("entity_create_line", {"x1": -150, "y1": 0, "x2": 150, "y2": 0}), ("entity_create_line", {"x1": 0, "y1": 0, "x2": 0, "y2": -120}), ], "attdefs": [ ("МАРКА", "Марка светильника", -60, 220, 120, "Т10"), ("ТИП", "Тип по спецификации", -300, -350, 90, "бра"), ("ВЫСОТА", "Высота от пола, мм", -120, -480, 90, "1800"), ("ГРУППА", "Группа включения", -60, 360, 90, "1.1"), ], }, "РОЗ_2": { "layer": "ЭО_РОЗЕТКИ", "primitives": [ ("entity_create_arc", {"cx": 0, "cy": 0, "radius": 150, "start_angle": 0, "end_angle": 180}), ("entity_create_line", {"x1": -150, "y1": 0, "x2": 150, "y2": 0}), ("entity_create_line", {"x1": 0, "y1": 0, "x2": 0, "y2": 200}), ("entity_create_line", {"x1": -60, "y1": 200, "x2": 60, "y2": 200}), ], "attdefs": [ ("МАРКА", "Марка розетки", -60, -250, 120, "S1"), ("ТИП", "Тип", -300, -380, 90, "розетка двойная с з/к"), ("ВЫСОТА", "Высота от пола, мм", -120, -510, 90, "300"), ("НАЗНАЧ", "Назначение", -150, -640, 90, "общее"), ("ГРУППА", "Группа", -60, 260, 90, "1.1"), ], }, "ВЫКЛ_3": { "layer": "СО_ВЫКЛЮЧАТЕЛИ", "primitives": [ ("entity_create_circle", {"cx": 0, "cy": 0, "radius": 150}), ("entity_create_line", {"x1": 0, "y1": 0, "x2": 0, "y2": 300}), ("entity_create_line", {"x1": -100, "y1": 300, "x2": 100, "y2": 300}), ], "attdefs": [ ("МАРКА", "Марка выключателя", -60, -250, 120, "V1"), ("ТИП", "Тип", -300, -380, 90, "выключатель 3-клавишный"), ("ВЫСОТА", "Высота от пола, мм", -120, -510, 90, "900"), ("ГРУППЫ", "Группы включения", -200, -640, 90, "1.1;1.2;1.3"), ], }, } INSERTS = [("СВ_ВСТР", 1200, 900, "Т1"), ("СВ_ВСТР", 3000, 900, "Т2"), ("ВЫКЛ_3", 1800, 0, "V1"), ("РОЗ_2", 3600, 0, "S1")] class Mcp: def __init__(self, port: str, token: str): self.url = f"http://10.26.5.108:{port}/mcp" self.token = token self.sid = None self._id = 0 def _rpc(self, payload): headers = {"Content-Type": "application/json", "Accept": "application/json, text/event-stream", "Authorization": f"Bearer {self.token}"} if self.sid: headers["Mcp-Session-Id"] = self.sid req = urllib.request.Request(self.url, data=json.dumps(payload).encode(), headers=headers, method="POST") with urllib.request.urlopen(req, timeout=300) as r: self.sid = r.headers.get("Mcp-Session-Id") or self.sid raw = r.read().decode("utf-8", "replace") msgs = [] for line in raw.splitlines(): if line.strip().startswith("data:"): try: msgs.append(json.loads(line[5:].strip())) except Exception: pass if not msgs: try: msgs = [json.loads(raw)] except Exception: msgs = [] for m in msgs: if m.get("id") == payload.get("id"): return m return msgs[-1] if msgs else {} def init(self): m = self._rpc({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": {"name": "build_symbols", "version": "1"}}}) self._rpc({"jsonrpc": "2.0", "method": "notifications/initialized"}) return ((m.get("result") or {}).get("serverInfo") or {}) def call(self, name, args=None): self._id += 1 res = self._rpc({"jsonrpc": "2.0", "id": 1000 + self._id, "method": "tools/call", "params": {"name": name, "arguments": args or {}}}) out = res.get("result") or {} if res.get("error") or out.get("isError"): raise RuntimeError(f"{name}: {str(res.get('error') or out)[:300]}") return out.get("structuredContent") or out def build(mcp: Mcp, out_dir: str): os.makedirs(out_dir, exist_ok=True) mcp.call("drawing_new") for layer, color in [("СО_СВЕТИЛЬНИКИ", 3), ("СО_ВЫКЛЮЧАТЕЛИ", 4), ("ЭО_РОЗЕТКИ", 5)]: try: mcp.call("layer_create", {"name": layer, "color": color}) except RuntimeError as e: print(f" слой {layer}: {e}") for name, spec in SYMBOLS.items(): handles = [] for tool, args in spec["primitives"]: r = mcp.call(tool, {**args, "layer": spec["layer"]}) h = r.get("handle") if h: handles.append(h) mcp.call("block_create_from_entities", {"name": name, "handles": handles, "base_x": 0, "base_y": 0}) for tag, prompt, x, y, height, default in spec["attdefs"]: mcp.call("entity_create_attdef", {"block_name": name, "tag": tag, "prompt": prompt, "x": x, "y": y, "height": height, "default": default}) print(f" блок {name}: примитивов {len(handles)}, атрибутов {len(spec['attdefs'])}") for name, x, y, mark in INSERTS: attrs = {"МАРКА": mark, "ТИП": SYMBOLS[name]["attdefs"][1][5], "ВЫСОТА": SYMBOLS[name]["attdefs"][2][5], "ГРУППА": "1.1"} mcp.call("block_insert", {"name": name, "x": x, "y": y, "layer": SYMBOLS[name]["layer"], "attributes": attrs}) print(f" вставок на чертёж: {len(INSERTS)}") dxf_in_container = "/opt/data/dxf/электрика_блоки.dxf" mcp.call("drawing_save_as", {"path": dxf_in_container}) got = mcp.call("file_get", {"path": dxf_in_container}) data = got.get("content_base64") or "" dst = os.path.join(out_dir, "электрика_блоки.dxf") if data: open(dst, "wb").write(base64.b64decode(data)) print(f" сохранено: {dst} ({os.path.getsize(dst)} байт)") blocks = mcp.call("block_list", {"compact": True, "fields": ["name", "attribute_count"]}) print(" проверка block_list:", json.dumps(blocks.get("result", {}) or blocks, ensure_ascii=False)[:300]) anon = [r for r in (blocks.get("result", {}).get("rows") or []) if str(r[0]).startswith("*U")] print(" анонимных блоков *U:", len(anon)) if __name__ == "__main__": ap = argparse.ArgumentParser() ap.add_argument("--port", default="30041", help="порт MCP-сервера AutoCAD (30015 — прод, 30041 — 1.6.0-ours)") ap.add_argument("--out", default="./out") a = ap.parse_args() token = os.environ.get("ACAD_MCP_TOKEN") or open("/opt/data/.autocad_mcp_token", encoding="utf-8").read().strip() m = Mcp(a.port, token) print("сервер:", m.init().get("version")) build(m, a.out)