# -*- coding: utf-8 -*-
"""MCP-сервер: каталог ДКС + проект RamCube (XML).
Инструменты:
dkc_search — поиск аппаратов в каталоге ДКС (библиотека DKC_data.mdf)
dkc_device — полные параметры позиции + габарит
ramcube_project_read — сводка проекта RamCube из XML
ramcube_apparatus_replace — замена аппаратов в схеме XML (оба набора свойств)
ramcube_set_project — параметры проекта (IP, секционирование, ввод кабелей, Iном, Iкз)
Запуск: python server.py (stdio)
"""
from __future__ import annotations
import os
import re
import subprocess
import xml.etree.ElementTree as ET
from typing import Any, Literal
import pyodbc
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("dkc-ramcube")
def _find_localdb() -> str:
"""SqlLocalDB.exe: берём из переменной окружения или ищем старшую версию."""
env = os.environ.get("DKC_LOCALDB")
if env:
return env
base = r"C:\Program Files\Microsoft SQL Server"
found = []
for name in os.listdir(base) if os.path.isdir(base) else []:
if not name.isdigit():
continue
p = os.path.join(base, name, "Tools", "Binn", "SqlLocalDB.exe")
if os.path.exists(p):
found.append((int(name), p))
if not found:
raise RuntimeError("SqlLocalDB.exe не найден: задайте DKC_LOCALDB")
return max(found)[1]
LOCALDB = _find_localdb()
INSTANCE = os.environ.get("DKC_LOCALDB_INSTANCE", "MSSQLLocalDB")
DB = os.environ.get("DKC_DB", "DKC_cat_ro_v44")
CONN_STR = os.environ.get("DKC_CONN", "") # готовые строки подключения: перебивает LocalDB
# id параметров каталога (проверено по базе 4.4)
P_IN, P_IN_R, P_ICU, P_CODE, P_POLES = 2, 3, 8, 9, 10
P_VENDOR, P_CAT_TYPE, P_DESIGN = 11, 12, 43
# модели 3D по семействам (пути внутри <RAMcube>\dwg)
MODELS = {
"NW08-NW25": r".\Schneider Electric\Drawings\Models\3D\ACB\NW08-NW25 MTZ2 (630-2000) (W) 3p_3D.dwg",
"NW32-NW40": r".\Schneider Electric\Drawings\Models\3D\ACB\NW32-NW40 MTZ2 (3200-4000А) (W) 3p_3D.dwg",
}
# ---------------------------------------------------------------- БД каталога
_conn: pyodbc.Connection | None = None
def _db() -> pyodbc.Connection:
"""Подключение к базе каталога через pipe LocalDB (драйвер {SQL Server})."""
global _conn
if _conn is not None:
return _conn
# ODBC-драйвер {SQL Server} раскрывает %WINDIR%: в урезанном окружении stdio-процесса
# переменных может не быть — иначе ошибка IM003 «не найден SQLSRV32.dll».
os.environ.setdefault("SystemRoot", r"C:\Windows")
os.environ.setdefault("WINDIR", r"C:\Windows")
if CONN_STR:
_conn = pyodbc.connect(CONN_STR, timeout=30)
return _conn
subprocess.run([LOCALDB, "start", INSTANCE], capture_output=True, timeout=90)
info = subprocess.run([LOCALDB, "info", INSTANCE], capture_output=True, text=True, timeout=60).stdout
m = re.search(r"(np:\\.*?tsql\\query)", info)
if not m:
raise RuntimeError("LocalDB не запущен: нет instance pipe name")
_conn = pyodbc.connect(
"DRIVER={SQL Server};SERVER=%s;DATABASE=%s;Trusted_Connection=yes;" % (m.group(1), DB),
timeout=30,
)
return _conn
def _params(device_id: int) -> dict[int, str]:
cur = _db().cursor()
q = """SELECT deviceValueParameter, CAST(deviceValueContent AS nvarchar(400))
FROM tblDeviceValues WHERE deviceValueDevice = ?"""
# deviceValueContent — тип sql_variant, pyodbc его не читает (ODBC type -150): только через CAST
return {int(pid): ("" if val is None else str(val)) for pid, val in cur.execute(q, device_id)}
def _bounds(device_id: int) -> dict[str, float] | None:
cur = _db().cursor()
q = """SELECT unitBoundMinX, unitBoundMaxX, unitBoundMinY, unitBoundMaxY,
unitBoundMinZ, unitBoundMaxZ
FROM tblDrawingUnits WHERE unitDevice = ?"""
row = cur.execute(q, device_id).fetchone()
if not row:
return None
return {"width": round(row[1] - row[0], 1), "depth": round(row[3] - row[2], 1), "height": round(row[5] - row[4], 1)}
def _series(code: str) -> str | None:
"""Семейство ACB по коду: ACB NW08..NW25 / ACB NW32..NW40."""
m = re.search(r"NW(\d+)", code or "")
if not m:
return None
n = int(m.group(1))
if 8 <= n <= 25:
return "NW08-NW25"
if 32 <= n <= 40:
return "NW32-NW40"
return None
# ---------------------------------------------------------------- инструменты
@mcp.tool()
def dkc_search(query: str = "", in_min: int | None = None, in_max: int | None = None,
with_geometry_only: bool = False, limit: int = 20) -> list[dict[str, Any]]:
"""Поиск аппаратов в каталоге ДКС.
query — подстрока артикула/кода (например 'NW40', 'MNX250', 'MD63N');
in_min/in_max — фильтр по номинальному току In, А;
with_geometry_only — только позиции с 3D-геометрией (те, что встанут в сборку).
"""
cur = _db().cursor()
# без alias и ORDER BY: legacy-драйвер {SQL Server} + pyodbc ломается на описании колонок
rows = cur.execute("SELECT deviceId, Code FROM tblDevices").fetchall()
if query:
q = query.lower()
rows = [r for r in rows if q in (r[1] or "").lower()]
rows.sort(key=lambda r: r[1] or "")
out: list[dict[str, Any]] = []
for device_id, code in rows:
pr = _params(device_id)
try:
in_a = int(float(pr.get(P_IN, "") or 0))
except ValueError:
in_a = None
if in_min is not None and (in_a is None or in_a < in_min):
continue
if in_max is not None and (in_a is None or in_a > in_max):
continue
b = _bounds(device_id)
if with_geometry_only and b is None:
continue
out.append({
"device_id": device_id,
"code": pr.get(P_CODE, code),
"catalog_type": pr.get(P_CAT_TYPE, ""),
"in_a": in_a,
"icu_ka": pr.get(P_ICU, ""),
"poles": pr.get(P_POLES, ""),
"design": pr.get(P_DESIGN, ""),
"vendor": pr.get(P_VENDOR, ""),
"bounds_mm": b,
"series": _series(pr.get(P_CODE, code)),
})
if len(out) >= limit:
break
return out
@mcp.tool()
def dkc_device(device_id: int) -> dict[str, Any]:
"""Полные параметры позиции каталога по deviceId (deviceId = elementDevice в XML RamCube)."""
pr = _params(device_id)
cur = _db().cursor()
row = cur.execute("SELECT Code FROM tblDevices WHERE deviceId = ?", device_id).fetchone()
if not row:
raise ValueError("deviceId %s не найден в каталоге" % device_id)
return {
"device_id": device_id,
"code": pr.get(P_CODE, row[0]),
"params": {str(k): v for k, v in sorted(pr.items())},
"bounds_mm": _bounds(device_id),
"series": _series(pr.get(P_CODE, row[0])),
"model_3d": MODELS.get(_series(pr.get(P_CODE, row[0])) or "", None),
}
# ---------------------------------------------------------------- XML проекта
def _load(path: str) -> ET.ElementTree:
if not os.path.exists(path):
raise FileNotFoundError(path)
return ET.parse(path)
def _cab_widths(root: ET.Element) -> list[dict[str, Any]]:
ass = {}
for e in root.find("elementsAssembly") or []:
a = e.attrib
try:
w = float(a.get("elementBoundMaxX", 0)) - float(a.get("elementBoundMinX", 0))
except (TypeError, ValueError):
w = 0.0
ass[a["elementId"]] = (a.get("elementDevice"), w)
inst = {i.attrib["elementInstanceId"]: i.attrib for i in root.find("elementInstancesAssembly") or []}
rows = []
for c in root.find("cabinets") or []:
sig = c.attrib.get("cabinetSignature", "?")
i = inst.get(c.attrib.get("cabinetElementInstance"), {})
dev, w = ass.get(i.get("elementInstanceElement"), ("?", 0.0))
kind = ("cable" if "Кабельн" in sig else "bus" if "Шинн" in sig else
"input" if "Вводн" in sig else "section" if "Секцион" in sig else
"distribution" if "Распредел" in sig else "other")
rows.append({"signature": sig, "device": dev, "width_mm": round(w, 1), "kind": kind})
return rows
@mcp.tool()
def ramcube_project_read(xml_path: str) -> dict[str, Any]:
"""Сводка проекта RamCube из XML: параметры, аппараты схемы, панели, габарит."""
root = _load(xml_path).getroot()
a = root.attrib
dict_props = {r.attrib["propertyProperty"]: r.attrib["propertyName"]
for r in root.find("properties") if "propertyProperty" in r.attrib}
vals: dict[str, dict[str, str]] = {}
for r in root.find("properties"):
x = r.attrib
if "propertyValueInstance" in x:
vals.setdefault(x["propertyValueInstance"], {})[x["propertyValueProperty"]] = x["propertyValueContent"]
elems = {e.attrib["elementId"]: e.attrib for e in root.find("elements")}
apparatus = []
for ins in root.find("elementInstances"):
i = ins.attrib
e = elems.get(i["elementInstanceElement"], {})
pr = vals.get(i["elementInstanceInstance"], {})
apparatus.append({
"instance": i["elementInstanceId"],
"element": i["elementInstanceElement"],
"device": e.get("elementDevice"),
"code": pr.get("9"),
"in_a": pr.get("2"),
"x": i.get("elementInstancePositionX"),
"signature": i.get("elementInstanceSignature", ""),
})
cabs = _cab_widths(root)
total = round(sum(c["width_mm"] for c in cabs), 1)
useful = round(sum(c["width_mm"] for c in cabs if c["kind"] in ("input", "distribution", "section")), 1)
return {
"file": xml_path,
"project": {k: a.get(k) for k in (
"projectName", "projectIP", "projectAmperageNominal", "projectAmperageShortCircuit",
"projectSectioning", "projectCableEntrance", "projectDimensionsHeight", "projectExistsAVR")},
"counts": {t: len(list(root.find(t))) for t in (
"elements", "elementsAssembly", "elementInstances", "elementInstancesAssembly",
"cabinets", "schemes", "buses", "wires")},
"apparatus": apparatus,
"cabinets": cabs,
"length_mm": {"total": total, "useful": useful,
"service": round(total - useful, 1)},
"notes": {"cable_entrance": "1 = сверху, 0 = снизу"},
}
@mcp.tool()
def ramcube_apparatus_replace(xml_in: str, xml_out: str,
replacements: list[dict[str, Any]]) -> dict[str, Any]:
"""Замена аппаратов в схеме XML.
replacements — список: {"element_id": "-174426", "new_device_id": 11919}
(или вместо element_id — "element_device": "11765", тогда меняются все элементы с этим deviceId).
Правятся оба набора свойств (эталон элемента и каждый экземпляр) — иначе RamCube
на загрузке выдаёт «код экземпляра не соответствует элементу».
"""
tree = _load(xml_in)
root = tree.getroot()
elems = {e.attrib["elementId"]: e for e in root.find("elements")}
targets: dict[str, dict[str, str]] = {} # elementId -> props нового аппарата
report = []
for rep in replacements:
new_id = int(rep["new_device_id"])
pr = _params(new_id)
code = pr.get(P_CODE, "")
series = _series(code)
props = {
"9": code,
"2": pr.get(P_IN, ""),
"3": pr.get(P_IN_R, ""),
"8": pr.get(P_ICU, ""),
"12": pr.get(P_CAT_TYPE, ""),
"43": pr.get(P_DESIGN, ""),
}
model = rep.get("model") or MODELS.get(series or "")
matched = []
for eid, el in elems.items():
if rep.get("element_id"):
hit = eid == str(rep["element_id"])
else:
hit = el.attrib.get("elementDevice") == str(rep.get("element_device"))
if not hit:
continue
old = el.attrib.get("elementDevice")
el.attrib["elementDevice"] = str(new_id)
if model:
el.attrib["elementContentFileName"] = model
targets[eid] = props
matched.append({"element_id": eid, "old_device": old, "new_device": str(new_id),
"code": code, "series": series})
report.extend(matched)
if not report:
raise ValueError("ни один элемент не найден по заданным criteria")
# эталон элемента + наборы экземпляров
inst_props: dict[str, dict[str, str]] = {}
for eid, props in targets.items():
ei = elems[eid].attrib.get("elementInstance")
if ei:
inst_props[ei] = props
for ins in root.find("elementInstances"):
eid = ins.attrib.get("elementInstanceElement")
if eid in targets:
inst_props[ins.attrib["elementInstanceInstance"]] = targets[eid]
changed = 0
for rec in root.find("properties"):
x = rec.attrib
inst = x.get("propertyValueInstance")
if inst in inst_props and x.get("propertyValueProperty") in inst_props[inst]:
x["propertyValueContent"] = inst_props[inst][x["propertyValueProperty"]]
changed += 1
tree.write(xml_out, encoding="utf-8", xml_declaration=True)
# валидация: перечитать и сверить коды
check = ET.parse(xml_out).getroot()
cvals: dict[str, dict[str, str]] = {}
for r in check.find("properties"):
x = r.attrib
if "propertyValueInstance" in x:
cvals.setdefault(x["propertyValueInstance"], {})[x["propertyValueProperty"]] = x["propertyValueContent"]
verified = []
for ins in check.find("elementInstances"):
i = ins.attrib
if i.get("elementInstanceElement") in targets:
pr = cvals.get(i["elementInstanceInstance"], {})
verified.append({"instance": i["elementInstanceId"], "code": pr.get("9"), "in_a": pr.get("2"),
"catalog_type": pr.get("12"), "design": pr.get("43")})
return {"out": xml_out, "elements": report, "property_sets": len(inst_props),
"values_changed": changed, "verified_instances": verified}
@mcp.tool()
def ramcube_set_project(xml_path: str, name: str | None = None, ip: str | None = None,
sectioning: Literal["3b", "4b"] | None = None,
cable_entrance: Literal["сверху", "снизу"] | None = None,
in_nominal: int | None = None, ikz_ka: float | None = None,
exists_avr: bool | None = None) -> dict[str, Any]:
"""Параметры проекта RamCube. cable_entrance: 'сверху' = 1, 'снизу' = 0."""
tree = _load(xml_path)
a = tree.getroot().attrib
if name:
a["projectName"] = name
if ip:
a["projectIP"] = str(ip)
if sectioning:
a["projectSectioning"] = sectioning
if cable_entrance:
a["projectCableEntrance"] = "1" if cable_entrance == "сверху" else "0"
if in_nominal:
a["projectAmperageNominal"] = str(in_nominal)
if ikz_ka is not None:
a["projectAmperageShortCircuit"] = str(ikz_ka)
if exists_avr is not None:
a["projectExistsAVR"] = "true" if exists_avr else "false"
tree.write(xml_path, encoding="utf-8", xml_declaration=True)
return {k: tree.getroot().attrib.get(k) for k in (
"projectName", "projectIP", "projectSectioning", "projectCableEntrance",
"projectAmperageNominal", "projectAmperageShortCircuit", "projectExistsAVR")}
if __name__ == "__main__":
mcp.run()
