#!/usr/bin/env python3
# Copyright 2006-2026 The QElectroTech Team
# This file is part of QElectroTech.
#
# QElectroTech is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 2 of the License, or
# (at your option) any later version.
#
# QElectroTech is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with QElectroTech. If not, see .
"""
qet-mcp — a Model Context Protocol server over QElectroTech projects.
WHY THIS EXISTS
Verifying a QET change by screenshot is unreliable. Twice in one review
session a screenshot was read as showing a defect that the saved file
proved had not happened: once "dragging a multi-selection leaves the
symbols behind and detaches their labels" (the XML showed all four
elements moved and no label moved), and once "Apply does nothing" (Apply
was disabled because a required field was empty). Both times the pixels
misled and the model told the truth.
So the primary tools here read the *model*, not the screen, and the
primary tool is qet_diff: do the thing, then ask what actually changed.
DESIGN
Most tools parse the .qet XML directly and never launch QElectroTech.
That is deliberate: it is fast, deterministic, needs no display, and
cannot be confused by a dialog. Only qet_export shells out to the
binary, and it carries the launch traps with it (see _run_qet).
The project database would be a better query surface than XML, but it is
not reachable from outside the application: projectDataBase::newQuery()
and isReadOnlySelect() are C++-internal and the JavaScript scripting API
exposes no SQL binding. Until it does, structure lives here.
PROTOCOL
Line-delimited JSON-RPC 2.0 on stdin/stdout, per MCP's stdio transport.
Nothing but protocol goes to stdout; diagnostics go to stderr.
No third-party dependencies — the MCP SDK is not assumed to be present.
`--call [arguments]` runs one tool without an MCP client, for an
assistant that can execute Python but cannot launch a server (a web chat
with code execution). It goes through the same dispatcher, so the
workspace policy applies exactly as it does over stdio.
"""
from __future__ import annotations
import datetime
import json
import re
import os
import shutil
import subprocess
import sys
import tempfile
import xml.etree.ElementTree as ET
from pathlib import Path, PurePath, PurePosixPath, PureWindowsPath
SERVER_NAME = "qet-mcp"
SERVER_VERSION = "0.1.0"
DEFAULT_PROTOCOL = "2025-06-18"
EXPORT_FORMATS = {
"pdf": "--export-pdf",
"png": "--export-png",
"svg": "--export-svg",
"dxf": "--export-dxf",
"bom": "--export-bom",
"cables": "--export-cables",
"wires": "--export-wires",
"wiring": "--export-wiring",
"nets": "--export-nets",
"links": "--export-links",
"info": "--info",
}
# --------------------------------------------------------------------------
# model reading
# --------------------------------------------------------------------------
def _root(path: str) -> ET.Element:
p = Path(path).expanduser()
if not p.is_file():
raise ValueError(f"no such file: {p}")
try:
return ET.parse(p).getroot()
except ET.ParseError as exc:
raise ValueError(f"{p.name} is not parseable XML: {exc}") from exc
def _element_info(el: ET.Element) -> dict:
"""The bag, as a plain dict."""
out = {}
bag = el.find("elementInformations")
if bag is not None:
for info in bag.findall("elementInformation"):
name = info.get("name")
if name:
out[name] = (info.text or "").strip()
return out
def _folios(root: ET.Element):
"""Yield (index, diagram) for each folio, 1-based as the UI numbers them."""
for i, d in enumerate(root.iter("diagram"), start=1):
yield i, d
def _elements(root: ET.Element):
for i, d in _folios(root):
for el in d.iter("element"):
yield i, el
def _wires(diagram: ET.Element):
"""The folio's conductors: children of only. A folio's wire
numbering rule is also saved as a tag, under , and
is not a wire."""
return diagram.findall("conductors/conductor")
def _conductors(root: ET.Element):
definitions = _definition_terminals(root)
for i, d in _folios(root):
index = _terminal_index(d, definitions)
for c in _wires(d):
yield i, c, index
def _uuid_key(value: str) -> str:
return (value or "").strip().strip("{}").lower()
# Orientation as a placed symbol's record writes it (an int,
# Qet::Orientation) or as a definition does (n/e/s/w).
_ORIENTATIONS = {"n": 0, "e": 1, "s": 2, "w": 3, "0": 0, "1": 1, "2": 2, "3": 3}
# Where QElectroTech docks a wire, relative to the terminal's position in
# its definition (Terminal's constructor, Terminal::terminalSize = 4). A
# placed symbol's record is written at that point.
_DOCK_OFFSET = {0: (0.0, 4.0), 1: (-4.0, 0.0), 2: (0.0, -4.0), 3: (4.0, 0.0)}
def _definition_terminals(root: ET.Element) -> dict:
"""Map each symbol stored in the project ("embed://" + its path in the
) to its terminals: {terminal uuid: (x, y, orientation)},
the position being the one in the definition."""
out = {}
def walk(node, path):
for child in node:
if child.tag == "category":
walk(child, path + [child.get("name", "")])
elif child.tag == "element":
terminals = {}
for t in child.findall("definition/description/terminal"):
try:
terminals[_uuid_key(t.get("uuid"))] = (
float(t.get("x")), float(t.get("y")),
_ORIENTATIONS.get((t.get("orientation") or "n")[:1], 0))
except (TypeError, ValueError):
continue
terminals.pop("", None)
if terminals:
out["embed://" + "/".join(path + [child.get("name", "")])] = terminals
collection = root.find("collection")
if collection is not None:
walk(collection, [])
return out
def _element_row(folio: int, el: ET.Element) -> dict:
info = _element_info(el)
etype = el.get("type", "")
return {
"folio": folio,
"uuid": el.get("uuid", ""),
"type": etype,
"name": etype.rsplit("/", 1)[-1].removesuffix(".elmt"),
"x": el.get("x"),
"y": el.get("y"),
"label": info.get("label", ""),
"info": info,
}
def _terminal_index(diagram: ET.Element, definitions: dict | None = None) -> dict:
"""Map a folio's terminal ids to an identity that survives a save.
A conductor names its ends with terminal1/terminal2, which are plain
integers scoped to the folio -- and QElectroTech reassigns them on every
write, in whatever order it happens to serialise the elements. The same
untouched conductor comes back as terminal1="1" terminal2="16" before a
save and terminal1="34" terminal2="15" after one. Keying a conductor on
that pair, which this tool used to do, made every conductor in the file
read as removed-and-re-added whenever the "after" side had been through
QElectroTech -- which is exactly the case qet_edit produces, so the
conductor half of the diff was noise precisely when it was needed.
So resolve each id to (owning element uuid, terminal position and
orientation inside that element). Element uuids are persisted and
stable; the terminal's local geometry comes from the element definition
and does not move when the element moves. That pair is the same basis
QET's own Terminal::stableUuid() uses for terminals with no uuid of
their own, and it is stable for the same reasons.
Conductors in the corpus carry no element1/element2 attribute -- 0 of
47 in ArduinoLCD.qet, 0 of 67 in 741.qet -- so this mapping has to be
built from the elements rather than read off the conductor.
A conductor can also name its ends by terminal uuid (element1 +
terminal1), and QElectroTech writes that form as soon as the terminal
has a uuid -- which, since a project gives every terminal one on
opening, is the first save of any older file. So the same untouched
conductor is written in the numbered form before a save and the uuid
form after it. With @p definitions (from _definition_terminals()), a
uuid end is resolved too, keyed (element uuid, terminal uuid), to the
very same identity as the numbered end: the terminal's definition
position, moved to where the wire docks, is where the placed symbol's
record is.
"""
index = {}
for el in diagram.iter("element"):
uuid = el.get("uuid", "")
if not uuid:
# Old enough to predate persisted element uuids. Leaving these
# ids unresolved is deliberate: keyed on terminal geometry
# alone, every element of the same type collapses together --
# in schema_indus.qet that merged nine distinct conductors onto
# one key, which is worse than the instability it was meant to
# fix. An unresolved end keeps them apart and stays visibly
# marked with a "#" so the caller can see the diff is on the
# unstable footing that file forces.
continue
records = []
for t in el.iter("terminal"):
tid = t.get("id")
if tid is None:
continue
key = (f"{uuid}@{t.get('x','?')},{t.get('y','?')}"
f",{t.get('orientation','?')}")
index[tid] = key
records.append((t, key))
for tuuid, (x, y, o) in (definitions or {}).get(el.get("type", ""), {}).items():
dx, dy = _DOCK_OFFSET[o]
for t, key in records:
try:
if (abs(float(t.get("x")) - (x + dx)) < 1e-6
and abs(float(t.get("y")) - (y + dy)) < 1e-6
and _ORIENTATIONS.get((t.get("orientation") or "")[:1]) == o):
index[(_uuid_key(uuid), tuuid)] = key
break
except (TypeError, ValueError):
continue
return index
def _conductor_key(folio: int, c: ET.Element, index: dict) -> str:
"""Identify a conductor by its two ends, in whichever scheme it uses.
The project format has two, and a file can hold both at once -- the
same folio, after an edit, carries legacy conductors and new ones:
- legacy: terminal1/terminal2 are the folio-scoped integer ids, and
there is no element1/element2. Resolve them through index.
- current: terminal1/terminal2 are terminal uuids from the element
*definition*, with element1/element2 naming the placed instances.
The terminal uuid alone is not an identity -- two coils of the same
type have the same one on both ends, so a conductor between them
would key as a self-loop -- so it is the (instance, terminal) pair
that identifies an end.
"""
ends = []
for elem_attr, term_attr, name_attr in (("element1", "terminal1", "terminalname1"),
("element2", "terminal2", "terminalname2")):
tid = c.get(term_attr, "?")
owner = c.get(elem_attr)
if owner:
ends.append(index.get((_uuid_key(owner), _uuid_key(tid)))
or f"{owner}/{tid or c.get(name_attr, '?')}")
else:
# An id with no element behind it stays visible as itself
# rather than silently collapsing conductors onto one key.
ends.append(index.get(tid, f"#{tid}"))
# A conductor is undirected: whichever end QET happens to write first,
# it is the same connection.
return f"{folio}:" + "--".join(sorted(ends))
def _conductor_row(folio: int, c: ET.Element, index: dict | None = None) -> dict:
return {
"folio": folio,
"uuid": c.get("uuid", ""),
"key": _conductor_key(folio, c, index or {}),
"num": c.get("num", ""),
"formula": c.get("formula", ""),
"cable": c.get("cable", ""),
"bus": c.get("bus", ""),
"function": c.get("function", ""),
"color": c.get("conductor_color", ""),
"section": c.get("conductor_section", ""),
"type": c.get("type", ""),
}
# --------------------------------------------------------------------------
# tools
# --------------------------------------------------------------------------
def tool_project_info(path: str) -> dict:
root = _root(path)
folios = []
for i, d in _folios(root):
folios.append({
"index": i,
"uuid": d.get("uuid", ""),
"title": d.get("title", ""),
"elements": sum(1 for _ in d.iter("element")),
"conductors": len(_wires(d)),
})
return {
"file": str(Path(path).expanduser()),
"title": root.get("title", ""),
"version": root.get("version", ""),
"folio_count": len(folios),
"element_count": sum(f["elements"] for f in folios),
"conductor_count": sum(f["conductors"] for f in folios),
"folios": folios,
}
def tool_elements(path: str, folio: int | None = None,
name_contains: str | None = None, limit: int = 200) -> dict:
rows = []
for i, el in _elements(_root(path)):
if folio is not None and i != folio:
continue
row = _element_row(i, el)
if name_contains and name_contains.lower() not in row["name"].lower():
continue
rows.append(row)
return {"count": len(rows), "truncated": len(rows) > limit,
"elements": rows[:limit]}
ITEM_KINDS = ["text", "shape", "image", "table", "element_text"]
def tool_items(path: str, folio: int | None = None, kind: str | None = None,
limit: int = 500) -> dict:
"""Every drawn item that is not a symbol or a wire, with its uuid.
Free texts, shapes, pictures, tables and the text fields of symbols --
the items a qet_edit op or a qet_diff entry names by uuid. Folios are
numbered from 1, as in qet_elements. An item saved before these items
carried a uuid has "" here; QElectroTech gives it one on the next save.
"""
if kind is not None and kind not in ITEM_KINDS:
raise ValueError(f"kind must be one of {ITEM_KINDS}, not {kind!r}")
ex = _extras(_root(path))
rows = []
for name, records in (("text", ex["texts"]), ("shape", ex["shapes"]),
("image", ex["images"]), ("table", ex["tables"])):
for r in records:
rows.append({"kind": name, "uuid": r["uuid"], **r["label"], **r["value"]})
for k, v in ex["element_texts"].items():
rows.append({"kind": "element_text", "uuid": ex["element_text_uuids"][k],
"folio": ex["element_text_folios"][k], "element": k[0],
"source": k[1], "bound_to": k[2], "n": k[3], **v})
rows = [r for r in rows if (folio is None or r["folio"] == folio)
and (kind is None or r["kind"] == kind)]
rows.sort(key=lambda r: (r["folio"], ITEM_KINDS.index(r["kind"])))
return {"count": len(rows), "truncated": len(rows) > limit, "items": rows[:limit]}
def tool_conductors(path: str, folio: int | None = None,
attribute: str | None = None,
non_empty: bool = False, limit: int = 200) -> dict:
rows = []
for i, c, ix in _conductors(_root(path)):
if folio is not None and i != folio:
continue
row = _conductor_row(i, c, ix)
if attribute is not None:
value = c.get(attribute, "")
if non_empty and not value.strip():
continue
row["value"] = value
rows.append(row)
return {"count": len(rows), "truncated": len(rows) > limit,
"conductors": rows[:limit]}
# No "version": that attribute is the file-format stamp QElectroTech rewrites
# on every save, so diffing it made every folio of any re-saved project look
# edited, and it is not something a script can set (see setFolioProperty).
_FOLIO_FIELDS = ("title", "author", "plant", "locmach", "indexrev",
"folio", "filename",
# the frame: attribute names as the file writes them
"cols", "colsize", "rows", "rowsize", "displaycols", "displayrows",
"titleblocktemplate")
def _plain_text(html: str) -> str:
"""The visible text of an independent text's HTML, which is what a
person means by "the text". The file stores a whole HTML document."""
import re
body = re.search(r"]*>(.*)", html or "", re.S)
inner = body.group(1) if body else (html or "")
inner = re.sub(r"<[^>]+>", "", inner)
for a, b in (("<", "<"), (">", ">"), ("&", "&"), (""", '"'), ("'", "'")):
inner = inner.replace(a, b)
return " ".join(inner.split())
def _angle(value: str) -> str:
"""A rotation in degrees, reduced to [0, 360) so equal angles compare equal.
QElectroTech writes the same angle in more than one way: rotating a symbol
and undoing it leaves its text fields at "-270" where they were "90", or
"-90" where they were "270". Compared as written, that read as a change.
Anything that is not a number is returned unchanged.
"""
try:
deg = float(value) % 360
except (TypeError, ValueError):
return value
return f"{deg:g}"
def _extras(root: ET.Element) -> dict:
"""Everything a folio holds besides elements and conductors.
Independent texts, shapes and images are records, not a keyed dict:
files written since they carry a uuid are compared by it (see
_diff_items), so an edited or moved text reads as that text, changed.
Older files have only position to go on, and there a change reads as
the old one removed and a new one added, with both shown.
"""
folios, folio_uuids, texts, shapes, images, tables = {}, {}, [], [], [], []
def record(el, label, value):
return {"uuid": el.get("uuid", ""), "key": tuple(label.values()),
"label": label, "value": value}
for n, d in _folios(root):
folios[n] = {f: d.get(f, "") for f in _FOLIO_FIELDS}
folio_uuids[n] = d.get("uuid", "")
for tb in d.findall("tables/graphics_table"):
tables.append(record(
tb, {"folio": n, "name": tb.get("name", "")},
{"x": tb.get("x", ""), "y": tb.get("y", ""), "width": tb.get("width", ""),
"height": tb.get("height", ""), "rows_shown": tb.get("display_n_row", "")}))
# The folio's own items only: direct children of its ,
# and . Symbols in older files carry their own
# texts, which iter() would count as free texts
# (122 extra in schema_indus.qet).
for t in d.findall("inputs/input"):
texts.append(record(
t, {"folio": n, "x": t.get("x", ""), "y": t.get("y", ""),
"text": _plain_text(t.get("text", ""))},
{"rotation": _angle(t.get("rotation", "0")),
"font": t.get("font", ""), "color": t.get("color", "")}))
for sh in d.findall("shapes/shape"):
pen, brush = sh.find("pen"), sh.find("brush")
shapes.append(record(
sh, {"folio": n, "type": sh.get("type", ""),
"from": [sh.get("x1", ""), sh.get("y1", "")],
"to": [sh.get("x2", ""), sh.get("y2", "")]},
{"line_color": pen.get("color", "") if pen is not None else "",
"line_style": pen.get("style", "") if pen is not None else "",
"line_width": pen.get("widthF", "") if pen is not None else "",
"fill": (brush.get("color", "") if brush is not None and
brush.get("style", "") != "NoBrush" else "none"),
"rotation": _angle(sh.get("rotation", "0"))}))
for im in d.findall("images/image"):
images.append(record(
im, {"folio": n, "x": im.get("x", ""), "y": im.get("y", "")},
{"scale": im.get("size", ""), "rotation": _angle(im.get("rotation", ""))}))
element_texts, element_text_uuids, element_text_folios = {}, {}, {}
for n, d in _folios(root):
for el in d.iter("element"):
uuid = el.get("uuid", "")
seen = {}
for t in el.iter("dynamic_elmt_text"):
src = t.get("text_from", "")
what = (t.findtext("info_name") if src == "ElementInfo"
else t.findtext("composite_text") if src == "CompositeText"
else t.findtext("text")) or ""
base = (uuid, src, what)
seen[base] = seen.get(base, 0) + 1
fs = (t.get("font", "").split(",") + ["", ""])[1]
element_text_uuids[base + (seen[base],)] = t.get("uuid", "")
element_text_folios[base + (seen[base],)] = n
element_texts[base + (seen[base],)] = {
"x": t.get("x", ""), "y": t.get("y", ""), "size": fs,
"frame": t.get("frame", ""), "rotation": _angle(t.get("rotation", "")),
"width": t.get("text_width", ""),
"shows": t.findtext("text") or ""}
strips = {}
for st in root.iter("terminal_strip"):
data = st.find("terminal_strip_data")
if data is None:
continue
info = {i.get("name"): (i.text or "") for i in data.iter("information")}
strips[data.get("uuid", "")] = {
"installation": info.get("installation", ""),
"location": info.get("location", ""),
"name": info.get("name", ""),
"terminals": sum(1 for _ in st.iter("real_terminal"))}
return {"folios": folios, "folio_uuids": folio_uuids, "texts": texts, "shapes": shapes,
"images": images, "tables": tables, "strips": strips,
"element_texts": element_texts, "element_text_uuids": element_text_uuids,
"element_text_folios": element_text_folios}
def _usable_ids(*sides) -> bool:
"""Whether uuids can identify items: present on every item, unique on each
side. Copying a symbol keeps its text fields' uuids, so a project can
hold the same field uuid twenty times; keying on it would merge them."""
return (any(sides) and all(all(s) for s in sides)
and all(len(set(s)) == len(s) for s in sides))
def _diff_keyed(a: dict, b: dict, label) -> dict:
"""added / removed / changed for two dicts keyed by identity."""
changed = []
for k in sorted(set(a) & set(b), key=str):
delta = {f: [a[k][f], b[k][f]] for f in a[k] if a[k][f] != b[k].get(f)}
if delta:
changed.append({"item": label(k), "changed": delta})
return {"before": len(a), "after": len(b),
"added": [label(k) for k in sorted(set(b) - set(a), key=str)][:50],
"removed": [label(k) for k in sorted(set(a) - set(b), key=str)][:50],
"changed": changed[:50]}
def _diff_items(a: list, b: list) -> dict:
"""_diff_keyed() over _extras() records of one kind.
Keyed on uuid only when every item on both sides has one. A file saved
before these items carried a uuid has none, and the first save by a
current QElectroTech gives them one, so a mixed pair falls back to
position for the whole kind rather than reading as everything removed
and re-added. On uuid, position is part of what is compared, so a move
is a change to that item.
"""
by_uuid = _usable_ids([r["uuid"] for r in a], [r["uuid"] for r in b])
def key(r):
return r["uuid"] if by_uuid else str(r["key"])
def keyed(rs):
return {key(r): ({**r["label"], **r["value"]} if by_uuid else r["value"])
for r in rs}
labels = {key(r): ({**r["label"], "uuid": r["uuid"]} if by_uuid else r["label"])
for r in a + b}
out = _diff_keyed(keyed(a), keyed(b), lambda k: labels[k])
out["keyed_by"] = "uuid" if by_uuid else "position"
return out
def _diff_folios(a: dict, b: dict) -> dict:
"""Folio fields, keyed by the folio's uuid when every folio has one.
By uuid, a folio moved to another position is reported once, under
"reordered", instead of as every folio after it changing its fields.
Without uuids (older files) folios are keyed by position, and a
removal or reorder in the middle shifts every later index -- the note
says so when the count changed.
"""
ua, ub = a["folio_uuids"], b["folio_uuids"]
by_uuid = _usable_ids(list(ua.values()), list(ub.values()))
changed, reordered, added, removed = [], [], [], []
if by_uuid:
pos_a = {u: n for n, u in ua.items()}
pos_b = {u: n for n, u in ub.items()}
for u in sorted(set(pos_a) & set(pos_b), key=lambda u: pos_b[u]):
fa, fb = a["folios"][pos_a[u]], b["folios"][pos_b[u]]
delta = {f: [fa[f], fb[f]] for f in _FOLIO_FIELDS if fa[f] != fb[f]}
if delta:
changed.append({"folio": pos_b[u], "uuid": u, "changed": delta})
if pos_a[u] != pos_b[u]:
reordered.append({"uuid": u, "title": fb["title"],
"from": pos_a[u], "to": pos_b[u]})
added = [{"folio": pos_b[u], "uuid": u, "title": b["folios"][pos_b[u]]["title"]}
for u in sorted(set(pos_b) - set(pos_a), key=lambda u: pos_b[u])]
removed = [{"folio": pos_a[u], "uuid": u, "title": a["folios"][pos_a[u]]["title"]}
for u in sorted(set(pos_a) - set(pos_b), key=lambda u: pos_a[u])]
else:
for n in sorted(set(a["folios"]) & set(b["folios"])):
delta = {f: [a["folios"][n][f], b["folios"][n][f]] for f in _FOLIO_FIELDS
if a["folios"][n][f] != b["folios"][n][f]}
if delta:
changed.append({"folio": n, "changed": delta})
out = {"before": len(a["folios"]), "after": len(b["folios"]),
"keyed_by": "uuid" if by_uuid else "position", "changed": changed[:50]}
if by_uuid:
out.update(added=added[:50], removed=removed[:50], reordered=reordered[:50])
elif len(a["folios"]) != len(b["folios"]) and changed:
out["note"] = ("the folio count changed, so changes listed here may be "
"later folios shifting position rather than edits")
return out
def _diff_element_texts(a: dict, b: dict) -> dict:
"""Element text fields, keyed by their own uuid when every field has one.
Otherwise by element, what the field is bound to, and the nth such field
-- which cannot tell a field that was removed from one that moved down
the list. A field's own text is also compared ("shows"), so relabelling
an element shows up here as well as in the element's information.
"""
# A field's uuid is unique only within its symbol (copies keep them), so
# a field is identified by its symbol's uuid and its own.
ka = {k: (k[0], u) if k[0] and u else "" for k, u in a["element_text_uuids"].items()}
kb = {k: (k[0], u) if k[0] and u else "" for k, u in b["element_text_uuids"].items()}
by_uuid = _usable_ids(list(ka.values()), list(kb.values()))
def label(k):
return {"element": k[0], "source": k[1], "bound_to": k[2], "n": k[3]}
if not by_uuid:
out = _diff_keyed(a["element_texts"], b["element_texts"], label)
out["keyed_by"] = "position"
return out
labels = {u: {**label(k), "uuid": u[1]} for side in (ka, kb) for k, u in side.items()}
out = _diff_keyed({ka[k]: v for k, v in a["element_texts"].items()},
{kb[k]: v for k, v in b["element_texts"].items()},
lambda u: labels[u])
out["keyed_by"] = "uuid"
return out
def _diff_extras(before: ET.Element, after: ET.Element) -> dict:
a, b = _extras(before), _extras(after)
out = {}
ta, tb = before.get("title", ""), after.get("title", "")
out["project"] = {"changed": {"title": [ta, tb]} if ta != tb else {}}
out["folios"] = _diff_folios(a, b)
out["texts"] = _diff_items(a["texts"], b["texts"])
out["shapes"] = _diff_items(a["shapes"], b["shapes"])
out["images"] = _diff_items(a["images"], b["images"])
out["tables"] = _diff_items(a["tables"], b["tables"])
out["element_texts"] = _diff_element_texts(a, b)
out["terminal_strips"] = _diff_keyed(
a["strips"], b["strips"],
lambda k: (lambda v: f"{v['installation']} {v['location']} {v['name']}".strip())(
(b["strips"].get(k) or a["strips"].get(k))))
return out
def tool_diff(before: str, after: str) -> dict:
"""Structural diff of two .qet files.
This is the tool that answers "what did that edit actually change",
which is the question a screenshot answers badly.
"""
# Key on uuid where there is one. Files written before conductors and
# elements carried persisted uuids fall back to a positional key, which
# is why a move in such a file reads as remove+add rather than a move.
a_el, b_el = {}, {}
for i, e in _elements(_root(before)):
r = _element_row(i, e)
# Rotation is saved as "orientation", in quarter turns (0-3); it is
# the only thing a rotation changes, so without it a rotated symbol
# reads as untouched.
r["orientation"] = e.get("orientation", "0")
a_el[r["uuid"] or f"{i}:{r['x']},{r['y']}:{r['name']}"] = r
for i, e in _elements(_root(after)):
r = _element_row(i, e)
r["orientation"] = e.get("orientation", "0")
b_el[r["uuid"] or f"{i}:{r['x']},{r['y']}:{r['name']}"] = r
moved, rotated, relabelled, changed_info = [], [], [], []
for k, a in a_el.items():
b = b_el.get(k)
if b is None:
continue
if a["orientation"] != b["orientation"]:
rotated.append({"uuid": k, "name": a["name"], "folio": a["folio"],
"orientation": [a["orientation"], b["orientation"]]})
if (a["x"], a["y"]) != (b["x"], b["y"]):
moved.append({
"uuid": k, "name": a["name"], "folio": a["folio"],
"from": [a["x"], a["y"]], "to": [b["x"], b["y"]],
"delta": [_num(b["x"]) - _num(a["x"]),
_num(b["y"]) - _num(a["y"])],
})
if a["label"] != b["label"]:
relabelled.append({"uuid": k, "name": a["name"],
"from": a["label"], "to": b["label"]})
# An empty field and a missing one mean the same thing, and
# QElectroTech drops empty ones when it saves, so compare only the
# fields that hold a value.
a_info = {n: v for n, v in a["info"].items() if v}
b_info = {n: v for n, v in b["info"].items() if v}
if a_info != b_info:
changed_info.append({"uuid": k, "name": a["name"],
"from": a_info, "to": b_info})
a_rows = [_conductor_row(i, c, ix) for i, c, ix in _conductors(_root(before))]
b_rows = [_conductor_row(i, c, ix) for i, c, ix in _conductors(_root(after))]
# Keyed by the conductor's own uuid when every conductor on both sides
# has one, so a rewired conductor is that conductor, changed ("ends").
# QElectroTech keeps a uuid only on conductors that were loaded with one
# or created since, so an older file keys on its two ends instead.
co_by_uuid = _usable_ids([r["uuid"] for r in a_rows], [r["uuid"] for r in b_rows])
co_id = (lambda r: r["uuid"]) if co_by_uuid else (lambda r: r["key"])
a_co = {co_id(r): r for r in a_rows}
b_co = {co_id(r): r for r in b_rows}
# An end that could not be resolved to an element is keyed on the
# folio-scoped integer id, which QElectroTech reassigns on every write.
# Say so rather than presenting the result as if it were comparable:
# in such a file an untouched conductor can read as removed and re-added.
shaky = 0 if co_by_uuid else sum(1 for k in set(a_co) | set(b_co) if "#" in k)
unstable = {} if not shaky else {
"unstable_keys": shaky,
"warning": "some conductors sit on elements with no persisted uuid, so "
"they are keyed on folio-scoped terminal ids that "
"QElectroTech renumbers on save; added/removed entries "
"marked with # may be the same conductor, not a change",
}
conductor_changes = []
for k, a in a_co.items():
b = b_co.get(k)
if b is None:
continue
fields = {f: [a[f], b[f]] for f in
("num", "formula", "cable", "bus", "color", "section",
"function", "type")
if a[f] != b[f]}
if a["key"] != b["key"]:
fields["ends"] = [a["key"], b["key"]]
if fields:
conductor_changes.append({"key": b["key"], **({"uuid": k} if co_by_uuid else {}),
"changed": fields})
deltas = sorted({tuple(m["delta"]) for m in moved})
return {
"elements": {
"before": len(a_el), "after": len(b_el),
"added": sorted(set(b_el) - set(a_el))[:50],
"removed": sorted(set(a_el) - set(b_el))[:50],
"moved": moved[:100],
"moved_count": len(moved),
"distinct_move_deltas": [list(d) for d in deltas],
"relabelled": relabelled[:50],
"info_changed": changed_info[:50],
"rotated": rotated[:50],
},
"conductors": {
"before": len(a_co), "after": len(b_co),
"keyed_by": "uuid" if co_by_uuid else "ends",
"added": sorted(b_co[k]["key"] for k in set(b_co) - set(a_co))[:50],
"removed": sorted(a_co[k]["key"] for k in set(a_co) - set(b_co))[:50],
"changed": conductor_changes[:100],
"changed_count": len(conductor_changes),
**unstable,
},
**_diff_extras(_root(before), _root(after)),
}
def _num(v) -> float:
try:
return float(v)
except (TypeError, ValueError):
return 0.0
def tool_scan(directory: str, tag: str = "conductor",
attribute: str = "cable", recursive: bool = True) -> dict:
"""Sweep every .qet in a directory, counting how many carry a
non-empty `attribute`.
This is the corpus question: "3190 conductors, 0 cable values across
25 projects" is exactly one call to this tool.
"""
d = Path(directory).expanduser()
if not d.is_dir():
raise ValueError(f"not a directory: {d}")
files = sorted(d.rglob("*.qet") if recursive else d.glob("*.qet"))
total = non_empty = 0
per_file, values, unreadable = [], {}, []
for f in files:
try:
root = ET.parse(f).getroot()
except ET.ParseError as exc:
unreadable.append({"file": f.name, "error": str(exc)})
continue
n = ne = 0
for node in root.iter(tag):
n += 1
v = (node.get(attribute) or "").strip()
if v:
ne += 1
values[v] = values.get(v, 0) + 1
total += n
non_empty += ne
per_file.append({"file": f.name, tag: n, "non_empty": ne})
return {
"directory": str(d), "files": len(files), "unreadable": unreadable,
"tag": tag, "attribute": attribute,
"total": total, "non_empty": non_empty,
"distinct_values": sorted(values.items(), key=lambda kv: -kv[1])[:25],
"per_file": per_file,
}
def _terminals_in_index_order(terminal_nodes) -> tuple:
"""Order an element's terminals the way QElectroTech indexes them.
Not file order. Element::parseTerminal() re-sorts the terminals every
time it adds one, top to bottom and then left to right, on each
terminal's local (y, x) -- Terminal::dockConductor() is mapToScene() of
its position, evaluated while the element still sits unrotated at the
origin. So the terminal a script reaches as index 0 is the topmost one,
whatever order the .elmt lists them in: bobine_ka_a_remanence.elmt
writes A2 (y=20) before A1 (y=-20), and add_conductor's index 0 is A1.
Getting this wrong wires the wrong end of a coil and nothing complains.
Returns (nodes in index order, ambiguous). Two terminals at the same
point tie, and the C++ sort is not stable, so which one is index 0 is
not defined; ambiguous says so instead of pretending.
"""
def key(t):
try:
return (float(t.get("y", 0)), float(t.get("x", 0)))
except ValueError:
return (0.0, 0.0)
ordered = sorted(terminal_nodes, key=key)
keys = [key(t) for t in ordered]
return ordered, len(keys) != len(set(keys))
def tool_element_info(path: str) -> dict:
"""Introspect a .elmt: names, terminals, and which info fields it carries."""
root = _root(path)
names = {n.get("lang"): (n.text or "") for n in root.iter("name")}
ordered, ambiguous = _terminals_in_index_order(list(root.iter("terminal")))
terminals = [{"index": i, "x": t.get("x"), "y": t.get("y"),
"orientation": t.get("orientation"),
"name": t.get("name", ""), "type": t.get("type", ""),
"uuid": t.get("uuid", "")}
for i, t in enumerate(ordered)]
info_fields = sorted({(i.text or "").strip()
for i in root.iter("info_name") if (i.text or "").strip()})
parts = {}
part_list = []
desc = root.find("description")
for child in (desc if desc is not None else []):
parts[child.tag] = parts.get(child.tag, 0) + 1
if child.tag != "terminal":
# uuid is empty for a part saved before parts carried one; the
# element editor gives it one on the next save.
part_list.append({"type": child.tag, "uuid": child.get("uuid", "")})
return {
"file": str(Path(path).expanduser()),
"type": root.get("type", ""), "link_type": root.get("link_type", ""),
"width": root.get("width"), "height": root.get("height"),
"names": names,
"terminal_count": len(terminals), "terminals": terminals,
"terminal_order": "index order: top to bottom then left to right, "
"not the file's order" + (
"; two terminals share a point, so their relative "
"index is undefined" if ambiguous else ""),
"info_fields": info_fields,
"parts": parts,
"part_list": part_list,
}
def _launch_executable(src: Path, sandbox: Path, windows: bool) -> Path:
"""The executable _run_qet() starts: a private copy, except on Windows.
The copy gives each run its own SingleApplication key, which is derived
from the executable's path. On Windows a program loads its DLLs from its
own folder, so a copy on its own dies before main() (0xC0000135, DLL not
found) and nothing could ever be exported or edited there. Run the
original instead: every flag this server passes is a CLI export flag or
--run, and QElectroTech handles both and returns before it constructs
SingleApplication (main.cpp), so there is no instance to be handed to.
The copy stays elsewhere for builds from before that early return.
"""
if windows:
return src
exe = sandbox / f"qet-mcp-{os.getpid()}"
shutil.copy2(src, exe)
return exe
def _launch_env(env: dict, home: Path, windows: bool) -> dict:
"""The environment _run_qet() starts QElectroTech in.
A private HOME and XDG directories, and QET_SETTINGS_DIR pointing into
them. HOME and XDG move QElectroTech's settings only on Linux: Qt keeps
them in the registry on Windows and in the user's preferences on
macOS, so without QET_SETTINGS_DIR every run there read the user's own
settings and never saw a collection path written for it
(qelectrotech-source-mirror#1178). A QElectroTech that knows the
variable keeps its settings in an INI file in that folder instead.
Except on Windows, also Qt's offscreen platform so no display is
needed. The Windows packages ship only the qwindows platform plugin:
asked for "offscreen", Qt finds no plugin and stops at a message box
nobody can close, so every call hung until its timeout. The export
flags and --run open no window there, so the default platform is what
they need.
"""
env = dict(env,
HOME=str(home),
XDG_CONFIG_HOME=str(home / ".config"),
XDG_DATA_HOME=str(home / ".local" / "share"),
QET_SETTINGS_DIR=str(home / ".config"))
if not windows:
env["QT_QPA_PLATFORM"] = "offscreen"
return env
def _collection_setting(collection: PurePath) -> str:
"""The settings file that points QElectroTech at @p collection.
Forward slashes: Qt reads a backslash in these files as an escape, so a
Windows path written as-is arrives mangled."""
return ("[elements-collections]\n"
f"common-collection-path={collection.as_posix()}\n")
def _run_qet(binary: str, args: list[str], timeout: int = 180,
elements_dir: str | None = None,
script: str | None = None, tail: int = 4000,
extra_env: dict | None = None) -> dict:
"""Launch QElectroTech headlessly, carrying the known launch traps.
SingleApplication keys its socket on applicationFilePath(), so a second
launch of the same path forwards its request to an already-running
instance and returns THAT process's answer with no error. Copying the
binary to a unique path gives this run its own socket. A symlink will
not do: applicationFilePath() resolves it back.
The sandbox HOME that isolation buys also costs something, and it is
not obvious: with no settings file, QETApp::commonElementsDir() falls
back to the compiled-in QET_COMMON_COLLECTION_PATH, which on a machine
that has never run `make install` does not exist. Every "common://..."
path then fails to resolve and the only symptom is addElement()
reporting "does not resolve to an element" for a file that is plainly
there. elements_dir writes the one setting that fixes it. The file name
is not free-choice: QSettings derives it from the organisation and
application names main.cpp sets before this branch runs. It is written
twice: QElectroTech/QElectroTech.ini is what a QElectroTech that knows
QET_SETTINGS_DIR reads, on every system (see _launch_env()), and
QElectroTech/QElectroTech.conf is what an older one reads on Linux.
script, when given, is written into the sandbox and passed to --run.
It lives inside the temporary directory so it cannot collide with a
concurrent call, and it is returned to the caller on failure, because a
generated script nobody can see is not debuggable.
The environment is inherited, not rebuilt, so QET_ENABLE_SCRIPTING
reaches QElectroTech from wherever this server was started -- normally
the "env" block of the MCP client's own configuration. That is the
consent: whoever configured this server and pointed it at a
QElectroTech binary made the choice, and their interactive
QElectroTech keeps whatever its own setting says. This server does not
set the variable itself, because a switch a program turns on for
itself is not a switch.
"""
src = Path(binary).expanduser()
if not src.is_file() or not os.access(src, os.X_OK):
raise ValueError(f"not an executable: {src}")
with tempfile.TemporaryDirectory(prefix="qet-mcp-") as tmp:
sandbox = Path(tmp)
exe = _launch_executable(src, sandbox, os.name == "nt")
home = sandbox / "home"
(home / ".config").mkdir(parents=True)
(home / ".local" / "share").mkdir(parents=True)
if elements_dir:
coll = Path(elements_dir).expanduser()
if not coll.is_dir():
raise ValueError(f"no such elements directory: {coll}")
cfg = home / ".config" / "QElectroTech"
cfg.mkdir(parents=True, exist_ok=True)
for name in ("QElectroTech.ini", "QElectroTech.conf"):
(cfg / name).write_text(_collection_setting(coll), encoding="utf-8")
if script is not None:
script_path = sandbox / "qet-mcp-edit.js"
script_path.write_text(script, encoding="utf-8")
args = ["--run", str(script_path), *args]
env = _launch_env(dict(os.environ), home, os.name == "nt")
env.update(extra_env or {})
try:
p = subprocess.run([str(exe), *args], env=env, timeout=timeout,
capture_output=True, text=True)
except subprocess.TimeoutExpired:
return {"ok": False, "timed_out": True, "timeout_s": timeout,
"hint": "a modal dialog during load will hang a headless "
"run; check the project's format version"}
result = {"ok": p.returncode == 0, "exit_code": p.returncode,
"stdout": p.stdout[-tail:], "stderr": p.stderr[-tail:]}
# QElectroTech refuses --run when scripting is switched off, which
# it is by default. Its own message is clear but French, and it
# names a settings dialog that nobody driving this server is
# looking at -- so say the thing that actually applies here. Keyed
# on QElectroTech naming the variable, with exit 3 as a fallback
# for a future build that words the refusal differently.
if script is not None and not result["ok"] and (
"QET_ENABLE_SCRIPTING" in p.stderr or p.returncode == 3):
result["hint"] = (
"this tool drives QElectroTech through a script, and this "
"QElectroTech has scripting switched off. Add "
"QET_ENABLE_SCRIPTING=1 to the environment this server is "
"started in -- in an MCP client that is the \"env\" block of "
"its entry in the client configuration. Only qet_query, "
"qet_continuity, qet_check, qet_layout_check, qet_project_new, "
"qet_edit, "
"qet_script_api, qet_script_test, qet_script_install, "
"qet_script_remove and qet_recording_check need it; every "
"other tool either reads "
"the file directly "
"or uses a plain CLI flag.")
return result
def _source_date_epoch(value) -> int:
"""The SOURCE_DATE_EPOCH a reproducible export uses: @p value when the
caller gives one, else the one this server was started with, else 0.
0 (1 January 1970) rather than a date read from the project: the file's
modification time changes with every copy and checkout, and a date the
drawing carries is a title block field, not when the PDF was made. A
caller who wants a meaningful date passes it.
"""
if value is None:
value = os.environ.get("SOURCE_DATE_EPOCH", "0")
try:
seconds = None if isinstance(value, (bool, float)) else int(value)
except (TypeError, ValueError):
seconds = None
if seconds is None or seconds < 0:
raise ValueError(f"source_date_epoch must be a whole number of seconds "
f"since 1970, got {value!r}")
return seconds
def tool_export(binary: str, project: str, format: str, output: str,
timeout: int = 180, reproducible: bool = False,
source_date_epoch: int | None = None) -> dict:
if format not in EXPORT_FORMATS:
raise ValueError(f"unknown format {format!r}; "
f"expected one of {', '.join(sorted(EXPORT_FORMATS))}")
proj = Path(project).expanduser()
if not proj.is_file():
raise ValueError(f"no such project: {proj}")
# The CLI matches its flags by exact string (cli_export.cpp:828) and takes
# the project and output as the two positional arguments after the flag
# (:862, :882). A "--export-bom=out.csv" form is NOT recognised: it fails
# the flag test, so the run is not treated as an export at all and the
# application starts its GUI instead, which then hangs on an offscreen
# platform. Order matters here.
flag = EXPORT_FORMATS[format]
# SOURCE_DATE_EPOCH (reproducible-builds.org) makes --export-pdf write
# the same bytes for the same project: the dates it names instead of
# now, a document id from the project file, fonts in a fixed order.
extra_env = None
if reproducible or source_date_epoch is not None:
epoch = _source_date_epoch(source_date_epoch)
extra_env = {"SOURCE_DATE_EPOCH": str(epoch)}
result = _run_qet(binary, [flag, str(proj), output], timeout,
extra_env=extra_env)
out = Path(output).expanduser()
result["output"] = str(out)
result["output_exists"] = out.exists()
if out.exists() and out.is_file():
result["output_bytes"] = out.stat().st_size
if extra_env and format == "pdf":
# A QElectroTech older than this option ignores the variable and
# writes the time of the export, so say whether this one did.
stamp = datetime.datetime.fromtimestamp(epoch, datetime.timezone.utc)
honoured = (b"/CreationDate (D:" + stamp.strftime("%Y%m%d%H%M%S").encode()
+ b"Z)") in out.read_bytes()
result["reproducible"] = honoured
if not honoured:
result["hint"] = ("this QElectroTech ignores SOURCE_DATE_EPOCH, so the "
"PDF carries the time of the export; it needs a build "
"with repeatable PDF export")
return result
# --------------------------------------------------------------------------
# qet_element_build: author a .elmt definition
# --------------------------------------------------------------------------
# Derived from the 6,918 shipped elements rather than from documentation:
# these are the attributes each part tag actually carries. Everything not
# listed here is refused, so a typo becomes an error instead of an
# attribute QElectroTech silently ignores.
PART_SCHEMA = {
"line": {"required": ("x1", "y1", "x2", "y2"),
"optional": ("end1", "end2", "length1", "length2")},
"rect": {"required": ("x", "y", "width", "height"), "optional": ("rx", "ry")},
"ellipse": {"required": ("x", "y", "width", "height"), "optional": ()},
"circle": {"required": ("x", "y", "diameter"), "optional": ()},
"arc": {"required": ("x", "y", "width", "height", "start", "angle"),
"optional": ()},
"polygon": {"required": ("points",), "optional": ("closed",)},
"text": {"required": ("x", "y", "text"), "optional": ("size", "rotation", "color")},
}
DEFAULT_STYLE = "line-style:normal;line-weight:normal;filling:none;color:black"
TERMINAL_ORIENTATIONS = ("n", "s", "e", "w")
# As used in the collection. "thumbnail" is included because it is the
# second most common value, not because this tool can build a good one.
LINK_TYPES = ("simple", "thumbnail", "master", "slave", "terminal",
"next_report", "previous_report")
def _f(value, where: str) -> float:
try:
return float(value)
except (TypeError, ValueError):
raise ValueError(f"{where}: expected a number, got {value!r}")
def _part_extent(kind: str, part: dict) -> list:
"""The x,y points a part reaches, for the bounding box."""
g = lambda k: _f(part[k], f"{kind}.{k}")
if kind == "line":
return [(g("x1"), g("y1")), (g("x2"), g("y2"))]
if kind in ("rect", "ellipse", "arc"):
x, y, w, h = g("x"), g("y"), g("width"), g("height")
return [(x, y), (x + w, y + h)]
if kind == "circle":
x, y, d = g("x"), g("y"), g("diameter")
return [(x, y), (x + d, y + d)]
if kind == "polygon":
return [(_f(px, "polygon point"), _f(py, "polygon point"))
for px, py in part["points"]]
if kind == "text":
return [(g("x"), g("y"))]
return []
def _element_geometry(parts: list, terminals: list) -> dict:
"""Bounding box, then a declared box that contains it.
The .elmt header carries width/height/hotspot_x/hotspot_y, and the
relationship to the drawing is a containment constraint rather than a
formula: the declared box runs from (-hotspot_x, -hotspot_y) to
(width - hotspot_x, height - hotspot_y) in the element's own
coordinates, and the drawing has to fit inside it. Checked against the
shipped collection, where authors chose their own margins -- one
element pads 2 units on the left and 3 on the right, another 8 and 2 --
so there is nothing to copy, only an invariant to satisfy.
Sizes are rounded out to multiples of 10, which is what every element
sampled from the collection uses and what keeps terminals on the grid.
"""
points = []
for part in parts:
points += _part_extent(part["type"], part)
for t in terminals:
points.append((_f(t["x"], "terminal.x"), _f(t["y"], "terminal.y")))
if not points:
raise ValueError("an element needs at least one part or terminal")
min_x = min(x for x, _ in points)
max_x = max(x for x, _ in points)
min_y = min(y for _, y in points)
max_y = max(y for _, y in points)
import math
pad = 5.0
hotspot_x = int(math.ceil((-min_x + pad) / 10.0) * 10)
hotspot_y = int(math.ceil((-min_y + pad) / 10.0) * 10)
width = int(math.ceil((max_x + hotspot_x + pad) / 10.0) * 10)
height = int(math.ceil((max_y + hotspot_y + pad) / 10.0) * 10)
# The invariant, asserted rather than trusted: an element whose drawing
# escapes its declared box is the classic way a hand-written .elmt
# renders clipped in the collection panel while looking fine in XML.
if not (-hotspot_x <= min_x and max_x <= width - hotspot_x
and -hotspot_y <= min_y and max_y <= height - hotspot_y):
raise ValueError(
f"internal error: declared box ({-hotspot_x}, {-hotspot_y}) to "
f"({width - hotspot_x}, {height - hotspot_y}) does not contain the "
f"drawing ({min_x}, {min_y}) to ({max_x}, {max_y})")
return {"width": width, "height": height,
"hotspot_x": hotspot_x, "hotspot_y": hotspot_y,
"bbox": [min_x, min_y, max_x, max_y]}
def _validate_part(index: int, part) -> str:
if not isinstance(part, dict):
raise ValueError(f"part {index} is not an object: {part!r}")
kind = part.get("type")
if kind not in PART_SCHEMA:
raise ValueError(f"part {index}: unknown type {kind!r}; expected one of "
f"{', '.join(sorted(PART_SCHEMA))}")
spec = PART_SCHEMA[kind]
for key in spec["required"]:
if key not in part:
raise ValueError(f"part {index} ({kind}) is missing {key!r}")
allowed = set(spec["required"]) | set(spec["optional"]) | {"type", "style", "antialias", "uuid"}
for key in part:
if key not in allowed:
raise ValueError(f"part {index} ({kind}): unexpected {key!r}; "
f"allowed: {', '.join(sorted(allowed))}")
if "uuid" in part and not (isinstance(part["uuid"], str)
and _UUID_RE.fullmatch(part["uuid"])):
raise ValueError(f"part {index} ({kind}): uuid must look like "
f"{{xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx}}, got {part['uuid']!r}")
if kind == "polygon":
pts = part["points"]
if not isinstance(pts, list) or len(pts) < 2:
raise ValueError(f"part {index} (polygon) needs at least two points")
for pt in pts:
if not (isinstance(pt, (list, tuple)) and len(pt) == 2):
raise ValueError(f"part {index} (polygon): each point is [x, y], got {pt!r}")
return kind
def _part_uuid(part: dict) -> str:
"""The caller's uuid for this part, braced as QElectroTech writes it, or
a new one. Every part carries one, as the element editor saves them."""
given = part.get("uuid")
if given:
return given if given.startswith("{") else "{" + given + "}"
return "{" + str(__import__("uuid").uuid4()) + "}"
def _part_element(part: dict, uuid: str) -> ET.Element:
kind = part["type"]
node = ET.Element(kind)
node.set("uuid", uuid)
if kind == "polygon":
for n, (px, py) in enumerate(part["points"], start=1):
node.set(f"x{n}", _fmt(px))
node.set(f"y{n}", _fmt(py))
node.set("closed", "true" if part.get("closed", True) else "false")
elif kind == "text":
node.set("x", _fmt(part["x"]))
node.set("y", _fmt(part["y"]))
node.set("text", str(part["text"]))
node.set("rotation", _fmt(part.get("rotation", 0)))
node.set("font", f"Sans Serif,{int(part.get('size', 9))},-1,5,50,0,0,0,0,0")
node.set("color", str(part.get("color", "#000000")))
return node
else:
for key in PART_SCHEMA[kind]["required"] + PART_SCHEMA[kind]["optional"]:
if key in part:
node.set(key, _fmt(part[key]))
node.set("antialias", "true" if part.get("antialias", True) else "false")
node.set("style", part.get("style", DEFAULT_STYLE))
return node
def _fmt(v) -> str:
"""Numbers the way QElectroTech writes them: no trailing .0."""
if isinstance(v, bool):
return "true" if v else "false"
if isinstance(v, (int, float)):
f = float(v)
return str(int(f)) if f == int(f) else repr(f)
return str(v)
def tool_element_build(output: str, names: dict, parts: list,
terminals: list | None = None,
link_type: str = "simple",
informations: dict | None = None,
uuid: str | None = None) -> dict:
"""Write a .elmt element definition.
Unlike a project, an element definition is not rewritten by
QElectroTech on a round trip, so generating one here is safe in a way
that generating a .qet would not be: there is no toXml() that will
drop what this writer did not know to emit.
What it will not do is invent geometry. The caller supplies the parts;
this validates them against the schema the shipped collection actually
uses, computes the width/height/hotspot header so the declared box
contains the drawing, and refuses anything it cannot place.
"""
terminals = terminals or []
if not isinstance(names, dict) or not names:
raise ValueError('names must be a non-empty object, e.g. {"en": "Coil", "fr": "Bobine"}')
if link_type not in LINK_TYPES:
raise ValueError(f"unknown link_type {link_type!r}; expected one of "
f"{', '.join(LINK_TYPES)}")
if not isinstance(parts, list):
raise ValueError("parts must be a list")
for i, part in enumerate(parts):
_validate_part(i, part)
for i, t in enumerate(terminals):
if not isinstance(t, dict):
raise ValueError(f"terminal {i} is not an object: {t!r}")
for key in ("x", "y", "orientation"):
if key not in t:
raise ValueError(f"terminal {i} is missing {key!r}")
if t["orientation"] not in TERMINAL_ORIENTATIONS:
raise ValueError(f"terminal {i}: orientation is one of "
f"{', '.join(TERMINAL_ORIENTATIONS)}, got {t['orientation']!r}")
# A master with no terminal cannot be wired, and a slave with none
# cannot be placed on a rail -- both are silent failures at use time.
if link_type in ("master", "slave", "simple") and not terminals:
raise ValueError(f"a {link_type} element with no terminals cannot be connected; "
"add terminals, or use link_type 'thumbnail' for a drawing-only element")
geometry = _element_geometry(parts, terminals)
root = ET.Element("definition", {
"version": "0.100.0", "type": "element", "link_type": link_type,
"width": str(geometry["width"]), "height": str(geometry["height"]),
"hotspot_x": str(geometry["hotspot_x"]), "hotspot_y": str(geometry["hotspot_y"]),
})
ET.SubElement(root, "uuid", {"uuid": uuid or "{" + str(__import__("uuid").uuid4()) + "}"})
names_node = ET.SubElement(root, "names")
for lang in sorted(names):
ET.SubElement(names_node, "name", {"lang": lang}).text = str(names[lang])
if informations:
kind = ET.SubElement(root, "kindInformations")
for key in sorted(informations):
ET.SubElement(kind, "kindInformation", {"name": key}).text = str(informations[key])
ET.SubElement(root, "informations")
description = ET.SubElement(root, "description")
part_uuids = [_part_uuid(part) for part in parts]
if len(set(part_uuids)) != len(part_uuids):
raise ValueError("two parts were given the same uuid")
for part, part_uuid in zip(parts, part_uuids):
description.append(_part_element(part, part_uuid))
for t in terminals:
attrs = {"x": _fmt(t["x"]), "y": _fmt(t["y"]),
"orientation": t["orientation"],
"type": t.get("type", "Generic"),
"uuid": "{" + str(__import__("uuid").uuid4()) + "}"}
if t.get("name"):
attrs["name"] = str(t["name"])
ET.SubElement(description, "terminal", attrs)
out = Path(output).expanduser()
out.parent.mkdir(parents=True, exist_ok=True)
ET.indent(root, space=" ")
out.write_bytes(ET.tostring(root, encoding="utf-8", xml_declaration=True))
# Read it back with the same reader every other tool here uses, rather
# than reporting what was intended.
check = tool_element_info(str(out))
return {"ok": True, "output": str(out), "bytes": out.stat().st_size,
**geometry,
# The order add_conductor will use, which is not the order the
# caller listed them in.
"terminal_index_order": [t["name"] or f"({t['x']},{t['y']})"
for t in check["terminals"]],
# In the order the parts were given.
"part_uuids": part_uuids,
"verified": check}
# --------------------------------------------------------------------------
# qet_edit: drive the scripting API, then prove what it did
# --------------------------------------------------------------------------
# op name -> (qet method, argument spec). A spec entry is (json key, kind),
# where kind says how the value is turned into JavaScript and, for "folio"
# and "elmt", that it may be a "$name" reference to an earlier op's result.
OPS = {
"add_folio": (None, []),
"set_folio_title": ("setFolioTitle", [("folio", "folio"), ("title", "str")]),
"add_element": ("addElement", [("folio", "folio"), ("path", "str"),
("x", "num"), ("y", "num")]),
"set_position": ("setElementPosition", [("folio", "folio"), ("element", "elmt"),
("x", "num"), ("y", "num")]),
"move_element": ("moveElement", [("folio", "folio"), ("element", "elmt"),
("dx", "num"), ("dy", "num")]),
"rotate_element": ("rotateElement", [("folio", "folio"), ("element", "elmt"),
("angle", "num")]),
"set_label": ("setElementLabel", [("folio", "folio"), ("element", "elmt"),
("label", "str")]),
"set_info": ("setElementInfo", [("folio", "folio"), ("element", "elmt"),
("key", "str"), ("value", "str")]),
"add_conductor": ("addConductor", [("folio", "folio"),
("from", "elmt"), ("from_terminal", "term"),
("to", "elmt"), ("to_terminal", "term")]),
"delete_element": ("deleteElement", [("folio", "folio"), ("element", "elmt")]),
"set_conductor": ("setConductorProperty", [("folio", "folio"), ("element", "elmt"),
("terminal", "term"), ("property", "str"),
("value", "str")]),
"move_conductor_segment": ("moveConductorSegment", [("folio", "folio"), ("element", "elmt"),
("terminal", "term"), ("segment", "num"),
("dx", "num"), ("dy", "num")]),
# Redraws one conductor around the symbols in its way; the result is
# "routed", or "no-route" when there is none, the path then unchanged.
"route_conductor": ("routeConductor", [("folio", "folio"), ("element", "elmt"),
("terminal", "term")]),
"link_elements": ("linkElements", [("folio", "folio"), ("element", "elmt"),
("to_folio", "folio"), ("to", "elmt")]),
"link_plc_io": ("linkElements", [("folio", "folio"), ("element", "elmt"),
("to_folio", "folio"), ("to", "elmt"),
("io_index", "folio")]),
"unlink_element": ("unlinkElement", [("folio", "folio"), ("element", "elmt")]),
"add_plc_io": ("addPlcIO", [("folio", "folio"), ("element", "elmt"),
("type", "str"), ("address", "str"),
("function", "str"), ("comment", "str")]),
"set_plc_io": ("setPlcIO", [("folio", "folio"), ("element", "elmt"),
("index", "folio"), ("property", "str"),
("value", "str")]),
"remove_plc_io": ("removePlcIO", [("folio", "folio"), ("element", "elmt"),
("index", "folio")]),
# Texts, shapes and images are addressed by "index": their index in a
# position-sorted listing, which add_text/add_shape return, or their
# uuid, which does not shift when another one is added. So
# it can be named as "$id". Indexes shift when one is added or deleted.
"delete_conductor": ("deleteConductor", [("folio", "folio"), ("element", "elmt"),
("terminal", "term")]),
"remove_folio": ("removeFolio", [("folio", "folio")]),
"set_folio": ("setFolioProperty", [("folio", "folio"), ("property", "str"),
("value", "str")]),
"add_terminal_strip": ("addTerminalStrip", [("installation", "str"), ("location", "str"),
("name", "str")]),
"remove_terminal_strip": ("removeTerminalStrip", [("strip", "folio")]),
"add_to_strip": ("addTerminalToStrip", [("strip", "folio"), ("folio", "folio"),
("element", "elmt")]),
"group_terminals": ("groupTerminals", [("strip", "folio"), ("indices", "indices")]),
"bridge_terminals": ("bridgeTerminals", [("strip", "folio"), ("indices", "indices")]),
"sort_terminal_strip": ("sortTerminalStrip", [("strip", "folio")]),
"add_autonum": ("addAutoNum", [("kind", "str"), ("name", "str"),
("parts", "list")]),
"remove_autonum": ("removeAutoNum", [("kind", "str"), ("name", "str")]),
"rename_autonum": ("renameAutoNum", [("kind", "str"), ("name", "str"),
("new_name", "str")]),
"use_conductor_autonum": ("useConductorAutoNum", [("folio", "folio"), ("name", "str")]),
"use_element_autonum": ("useElementAutoNum", [("name", "str")]),
"number_element": ("numberElement", [("folio", "folio"), ("element", "elmt")]),
"renumber_element_autonum": ("renumberElementAutoNum", [("name", "str")]),
"free_element_numbers": ("freeElementNumbers", [("folio", "folio"), ("element", "elmt")]),
"assign_element_number": ("assignElementNumber", [("folio", "folio"), ("element", "elmt"),
("number", "num")]),
"assign_element_autonum": ("assignElementAutoNum", [("name", "str"), ("folio", "folio"),
("element", "elmt"), ("overwrite", "bool")]),
# The text fields drawn on a symbol. Indexed within the element's own
# list, which follows its definition and shifts on delete (and undo of a
# delete puts the field back at the end).
"add_element_text": ("addElementText", [("folio", "folio"), ("element", "elmt"),
("source", "str"), ("value", "str"),
("x", "num"), ("y", "num")]),
"set_element_text": ("setElementTextProperty", [("folio", "folio"), ("element", "elmt"),
("index", "element_text"), ("property", "str"),
("value", "str")]),
"delete_element_text": ("deleteElementText", [("folio", "folio"), ("element", "elmt"),
("index", "element_text")]),
# Returns the uuids of the copies IN THE ORDER the elements were named,
# so "$copies[0]" is the copy of the first one. Conductors between the
# copied elements are copied with them; copies arrive without labels or
# wire numbers, as they do on a paste in the application.
"insert_folio": ("insertFolio", [("position", "folio")]),
# Reads, not edits: the result is reported in the operations list, so a
# follow-up call can lay something out relative to it.
"element_geometry": ("elementGeometry", [("folio", "folio"), ("element", "elmt")]),
# Undo/redo act on QElectroTech's undo stack for this run, one command at
# a time. Consecutive edits to the same property or information key merge
# into one command, so one undo can revert several of them.
"undo": ("undo", []),
"redo": ("redo", []),
"search_and_replace": ("searchAndReplace", [("kind", "str"), ("field", "str"),
("pattern", "str"), ("replacement", "str"),
("regex", "bool"), ("case_sensitive", "bool")]),
"set_project_title": ("setProjectTitle", [("title", "str")]),
"set_folio_border": ("setFolioBorder", [("folio", "folio"), ("property", "str"),
("value", "str")]),
# A folio's conductor defaults, or with "folio": -1 the project's ones
# that each new folio copies.
"set_conductor_default": ("setConductorDefault", [("folio", "folio"), ("property", "str"),
("value", "str")]),
"embed_title_block_template": ("embedTitleBlockTemplate", [("name", "str")]),
"duplicate_elements": ("duplicateElements", [("folio", "folio"), ("elements", "elmts"),
("to_folio", "folio"), ("x", "num"), ("y", "num")]),
"add_text": ("addText", [("folio", "folio"), ("text", "str"),
("x", "num"), ("y", "num")]),
"set_text": ("setTextContent", [("folio", "folio"), ("index", "text"),
("text", "str")]),
"set_text_color": ("setTextColor", [("folio", "folio"), ("index", "text"),
("color", "str")]),
"rotate_text": ("setTextRotation", [("folio", "folio"), ("index", "text"),
("angle", "num")]),
"delete_text": ("deleteText", [("folio", "folio"), ("index", "text")]),
"add_shape": ("addShape", [("folio", "folio"), ("shape", "str"),
("x1", "num"), ("y1", "num"),
("x2", "num"), ("y2", "num")]),
"set_shape": ("setShapeProperty", [("folio", "folio"), ("index", "shape"),
("property", "str"), ("value", "str")]),
"add_image": ("addImage", [("folio", "folio"), ("file", "str"),
("x", "num"), ("y", "num")]),
"add_pdf_page": ("addPdfPage", [("folio", "folio"), ("file", "str"),
("page", "num"), ("dpi", "num"),
("x", "num"), ("y", "num")]),
"scale_image": ("setImageScale", [("folio", "folio"), ("index", "image"),
("factor", "num")]),
"rotate_image": ("setImageRotation", [("folio", "folio"), ("index", "image"),
("angle", "num")]),
"delete_image": ("deleteImage", [("folio", "folio"), ("index", "image")]),
"delete_shape": ("deleteShape", [("folio", "folio"), ("index", "shape")]),
"add_polygon": ("addPolygon", [("folio", "folio"), ("points", "points"),
("closed", "bool")]),
"set_shape_polygon": ("setShapePolygon", [("folio", "folio"), ("index", "shape"),
("points", "points")]),
"add_path": ("addPath", [("folio", "folio"), ("nodes", "nodes"),
("closed", "bool")]),
"set_shape_path_nodes": ("setShapePathNodes", [("folio", "folio"), ("index", "shape"),
("nodes", "nodes")]),
"set_shape_closed": ("setShapeClosed", [("folio", "folio"), ("index", "shape"),
("closed", "bool")]),
"add_table": ("addTable", [("folio", "folio"), ("kind", "str"),
("name", "str"), ("query", "str")]),
"set_table_position": ("setTablePosition", [("folio", "folio"), ("table", "table"),
("x", "num"), ("y", "num")]),
"delete_table": ("deleteTable", [("folio", "folio"), ("table", "table")]),
# Lining symbols up. These run several calls each, in helpers the
# generated script defines (qetMcp*), not one scripting call.
"align_elements": ("qetMcpAlign", [("folio", "folio"), ("elements", "elmts"),
("edge", "str"), ("to", "elmt")]),
"distribute_elements": ("qetMcpDistribute", [("folio", "folio"), ("elements", "elmts"),
("axis", "str"), ("pitch", "num")]),
# Place a symbol so one of its terminals is exactly in line with
# another symbol's: the wire between the two is then straight. Returns
# the new element's uuid, as add_element does.
"place_element": ("qetMcpPlace", [("folio", "folio"), ("path", "str"),
("terminal", "anyterm"), ("next_to", "elmt"),
("next_to_terminal", "anyterm"),
("side", "str"), ("gap", "num"),
("angle", "num")]),
"align_terminal": ("qetMcpAlignTerminal", [("folio", "folio"), ("element", "elmt"),
("terminal", "anyterm"), ("to", "elmt"),
("to_terminal", "anyterm")]),
}
# Arguments those ops may leave out.
OP_DEFAULTS = {
"align_elements": {"to": ""},
"distribute_elements": {"pitch": 0},
"place_element": {"side": "", "gap": 40, "angle": 0},
}
ALIGN_EDGES = ["left", "center", "right", "top", "middle", "bottom"]
DISTRIBUTE_AXES = ["horizontal", "vertical"]
PLACE_SIDES = ["", "below", "above", "right", "left"]
# What the helpers call, beyond what every edit needs.
_HELPER_NEEDS = {
"qetMcpAlign": {"elementGeometry", "moveElement"},
"qetMcpDistribute": {"elementGeometry", "moveElement"},
"qetMcpPlace": {"elementGeometry", "addElement", "deleteElement", "moveElement",
"rotateElement", "terminalPosition", "terminalIndex"},
"qetMcpAlignTerminal": {"moveElement", "terminalPosition", "terminalIndex"},
}
SHAPES = ["line", "rectangle", "ellipse", "polygon"]
FOLIO_BORDER_PROPERTIES = ["columns", "column-width", "display-columns",
"rows", "row-height", "display-rows", "preset"]
# The sheets set_folio_border's "preset" sizes a folio for, as the scripting
# API names them (QetScriptApi::folioPresets()). Tabloid and ledger are the
# same 11 x 17 in sheet.
FOLIO_PRESETS = [f"{paper}-{orientation}"
for paper in ("a0", "a1", "a2", "a3", "a4", "a5",
"letter", "legal", "tabloid", "ledger")
for orientation in ("portrait", "landscape")]
ELEMENT_TEXT_SOURCES = ["text", "info", "composite"]
ELEMENT_TEXT_PROPERTIES = ["text", "source", "info", "composite", "frame", "size",
"x", "y", "rotation", "width"]
SHAPE_PROPERTIES = ["color", "fill", "width", "line-style", "rotation"]
AUTONUM_KINDS = ["conductor", "element", "folio"]
SEARCH_REPLACE_KINDS = ["element_info", "conductor", "text"]
FOLIO_PROPERTIES = ["title", "author", "filename", "plant", "locmach",
"indexrev", "folio", "template"]
# The ops that make a folio: their "$id" is an index the next insert_folio
# or remove_folio can shift, so the script also keeps the folio's uuid.
FOLIO_MAKING_OPS = ("add_folio", "insert_folio")
# The ops that address one conductor by element + terminal, and so also
# take "conductor": "{uuid}" (qet_conductors reports each one's uuid).
CONDUCTOR_UUID_OPS = ("set_conductor", "move_conductor_segment", "delete_conductor",
"route_conductor")
# add_conductor's "route": "default" is the application's own two or three
# segments; "avoid" then redraws the new conductor around the symbols.
ROUTE_MODES = ["default", "avoid"]
# Routing arrived after the other drawing verbs, so it is required only by
# an edit that routes: every other edit still runs on a build without it.
_ROUTE_METHODS = {"routeConductor", "routeConductorBetween"}
# Accepted by set_conductor. The names are the project file's own, so what
# a script sets is what qet_conductors reports back.
CONDUCTOR_PROPERTIES = ["num", "formula", "function", "bus", "cable",
"tension_protocol", "conductor_color",
"conductor_section", "color", "text_color",
# the conductor's look, under the file's own names
"color2", "bicolor", "style", "dash-size",
"condsize", "numsize", "displaytext"]
# A folio's conductor defaults take the same names, plus the folio-wide
# "one text per potential" switch, which no single conductor carries.
CONDUCTOR_DEFAULT_PROPERTIES = ["onetextperfolio"] + CONDUCTOR_PROPERTIES
# Methods this tool needs that only exist in a build carrying the drawing
# verbs. Probed in the script rather than assumed, because the failure mode
# otherwise is a TypeError on line N of a generated file the caller never
# sees, reported as "the edit failed".
_REQUIRED_METHODS = sorted(({m for m, _ in OPS.values()
if m and not m.startswith("qetMcp")} - _ROUTE_METHODS) |
{"save", "folioCount", "conductorCount", "elementCount"})
_MARKER = "QETEDIT "
_UUID_RE = re.compile(r"\{?[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-"
r"[0-9a-fA-F]{4}-[0-9a-fA-F]{12}\}?")
def _js(value) -> str:
"""A JSON literal is a JavaScript literal for every type used here."""
return json.dumps(value)
def _build_script(operations: list, output: str) -> str:
"""Turn the operation list into a script, or raise on a bad operation.
Every op is validated here, before QElectroTech is launched at all: a
typo in an op name should cost nothing, not a process start and a
JavaScript exception.
"""
refs: set[str] = set()
# "$name"s made by an op that creates a folio: held as the folio's uuid
# too, since a later insert_folio or remove_folio shifts its index.
folio_refs: set[str] = set()
# Resolvers a uuid reference needs; required only when one is used, so
# an index-only edit still runs on a build that predates them.
uuid_methods: set[str] = set()
folio_js = "0"
element_js = None # the op's element, for lookups scoped to it
owner_js = None # the element a terminal argument belongs to
lines = [
"// generated by qet-mcp; do not edit",
"var R = {};", # $name -> value from an earlier op
"var F = {};", # $name -> uuid of a folio an op made
"var missing = [];",
"var need = @NEED@;",
"for (var i = 0; i < need.length; i++) {",
" if (typeof qet[need[i]] !== 'function') missing.push(need[i]);",
"}",
f"qet.log({_js(_MARKER)} + JSON.stringify("
"{kind: 'capabilities', missing: missing}));",
"var stop = false;",
# A conductor named by uuid is turned into one of its ends, as the
# conductor calls take it: an end whose terminal carries no other
# conductor, so the call cannot pick the wrong one. null, with the
# reason logged, if it is not on the folio or both ends are shared.
"function qetMcpConductorEnd(index, folio, uuid) {",
" var ends = qet.conductorEnds(folio, uuid);",
" var why = 'no conductor ' + uuid + ' on folio ' + folio;",
" if (ends && ends.length === 2) {",
" var lines = qet.conductors(folio);",
" for (var k = 0; k < 2; k++) {",
" if (ends[k] === '?') continue;",
" var n = 0;",
" for (var j = 0; j < lines.length; j++) {",
" var p = lines[j].split(' : ')[0].split(' -- ');",
" if (p[0] === ends[k] || p[1] === ends[k]) n++;",
" }",
" if (n === 1) {",
" var m = ends[k].split(' terminal ');",
" return {element: m[0], terminal: parseInt(m[1], 10)};",
" }",
" }",
" why = 'conductor ' + uuid + ' shares both of its terminals with other '",
" + 'conductors; the conductor calls address one by a terminal carrying '",
" + 'only it';",
" }",
f" qet.log({_js(_MARKER)} + JSON.stringify("
"{kind: 'op_note', index: index, note: why}));",
" return null;",
"}",
# A terminal named by uuid is turned into the index the calls take,
# on its own element; -1, with the reason logged, if that element
# has no terminal with it.
"function qetMcpTerminal(index, key, folio, element, uuid) {",
" var t = qet.terminalIndex(folio, element, uuid);",
" if (t < 0) qet.log(" + _js(_MARKER) + " + JSON.stringify({kind: 'op_note', "
"index: index, note: key + ': no terminal ' + uuid + ' on element ' + element + "
"' (or two of its terminals carry it)'}));",
" return t;",
"}",
# The layout ops' helpers. qetMcpOp is the index of the op running,
# for the notes they log.
"var qetMcpOp = -1;",
"function qetMcpNote(note) {",
f" qet.log({_js(_MARKER)} + JSON.stringify("
"{kind: 'op_note', index: qetMcpOp, note: note}));",
"}",
"function qetMcpTerm(folio, element, t) {",
" return typeof t === 'string' ? qet.terminalIndex(folio, element, t) : t;",
"}",
"function qetMcpAlign(folio, els, edge, to) {",
" var ref = qet.elementGeometry(folio, to || els[0]);",
" if (ref.left === undefined) { qetMcpNote('no element ' + (to || els[0])); return false; }",
" var across = edge === 'left' || edge === 'center' || edge === 'right';",
" function at(g) {",
" if (edge === 'center') return (g.left + g.right) / 2;",
" if (edge === 'middle') return (g.top + g.bottom) / 2;",
" return g[edge];",
" }",
" var moved = 0;",
" for (var i = 0; i < els.length; i++) {",
" if (els[i] === to) continue;",
" var g = qet.elementGeometry(folio, els[i]);",
" if (g.left === undefined) { qetMcpNote('no element ' + els[i]); return false; }",
" var d = at(ref) - at(g);",
" if (Math.abs(d) < 1e-9) continue;",
" if (!qet.moveElement(folio, els[i], across ? d : 0, across ? 0 : d)) return false;",
" moved++;",
" }",
" return moved;",
"}",
"function qetMcpDistribute(folio, els, axis, pitch) {",
" var k = axis === 'horizontal' ? 'x' : 'y', gs = [];",
" for (var i = 0; i < els.length; i++) {",
" var g = qet.elementGeometry(folio, els[i]);",
" if (g.left === undefined) { qetMcpNote('no element ' + els[i]); return false; }",
" gs.push({e: els[i], v: g[k]});",
" }",
" gs.sort(function (a, b) { return a.v - b.v; });",
" var n = gs.length, first = gs[0].v;",
" var step = pitch > 0 ? pitch : (gs[n - 1].v - first) / (n - 1);",
" var moved = 0;",
" for (var j = 1; j < n; j++) {",
" var t = first + j * step;",
# an even split of a gap that is not a whole number of grid steps
# rounds each place to the grid, so symbols on it stay on it
" if (pitch <= 0 && j < n - 1) t = first + Math.round((t - first) / 10) * 10;",
" var d = t - gs[j].v;",
" if (Math.abs(d) < 1e-9) continue;",
" if (!qet.moveElement(folio, gs[j].e, k === 'x' ? d : 0, k === 'y' ? d : 0)) return false;",
" moved++;",
" }",
" return moved;",
"}",
"var qetMcpStep = {below: [0, 1], above: [0, -1], right: [1, 0], left: [-1, 0]};",
"var qetMcpFacingSide = {s: 'below', n: 'above', e: 'right', w: 'left'};",
"var qetMcpBack = {below: 'n', above: 's', right: 'w', left: 'e'};",
"function qetMcpPlace(folio, path, term, near, nearTerm, side, gap, angle) {",
" var g = qet.elementGeometry(folio, near);",
" if (g.left === undefined) { qetMcpNote('no element ' + near); return ''; }",
" var to = qet.terminalPosition(folio, near, qetMcpTerm(folio, near, nearTerm));",
" if (to.x === undefined) { qetMcpNote('next_to_terminal ' + nearTerm + ' is not a terminal of ' + near); return ''; }",
" var el = qet.addElement(folio, path, g.x, g.y);",
" if (!el) return '';",
# turned first, so the terminal is measured where it ends up
" if (angle && !qet.rotateElement(folio, el, angle)) { qet.deleteElement(folio, el); return ''; }",
" var mine = qet.terminalPosition(folio, el, qetMcpTerm(folio, el, term));",
" if (mine.x === undefined) {",
" qet.deleteElement(folio, el);",
" qetMcpNote('terminal ' + term + ' is not a terminal of ' + path);",
" return '';",
" }",
" side = side || qetMcpFacingSide[to.facing];",
" var s = qetMcpStep[side];",
" if (!qet.moveElement(folio, el, to.x + s[0] * gap - mine.x, to.y + s[1] * gap - mine.y)) return '';",
" if (mine.facing !== qetMcpBack[side])",
" qetMcpNote('terminal ' + term + ' of the new symbol faces ' + mine.facing + ', not '",
" + qetMcpBack[side] + ': the wire to it will bend. Rotate the symbol (rotate_element) '",
" + 'or name another terminal');",
" return el;",
"}",
"function qetMcpAlignTerminal(folio, el, term, to, toTerm) {",
" var a = qet.terminalPosition(folio, to, qetMcpTerm(folio, to, toTerm));",
" var b = qet.terminalPosition(folio, el, qetMcpTerm(folio, el, term));",
" if (a.x === undefined || b.x === undefined) {",
" qetMcpNote('terminal not found: ' + (a.x === undefined ? to + ' ' + toTerm : el + ' ' + term));",
" return false;",
" }",
" var along = a.facing === 'n' || a.facing === 's';",
" return qet.moveElement(folio, el, along ? a.x - b.x : 0, along ? 0 : a.y - b.y);",
"}",
"if (missing.length === 0) {",
]
def ref_or(value, kind: str, op_index: int, key: str) -> str:
if kind == "elmts":
if not isinstance(value, list) or not value or not all(isinstance(v, str) for v in value):
raise ValueError(f"operation {op_index}: {key!r} must be a non-empty list of "
f"elements (uuids or \"$id\" references), got {value!r}")
return "[" + ", ".join(ref_or(v, "elmt", op_index, key) for v in value) + "]"
if isinstance(value, str) and value.startswith("$"):
indexed = re.fullmatch(r"\$([A-Za-z0-9_]+)\[(\d+)\]", value)
if indexed:
# one item of a list result, e.g. the second copy from
# duplicate_elements
name, n = indexed.group(1), int(indexed.group(2))
if name not in refs:
raise ValueError(
f"operation {op_index} refers to {value!r}, which no earlier "
f"operation defined (set \"id\": {name!r} on the op that creates it)")
return f"R[{_js(name)}][{n}]"
name = value[1:]
if name not in refs:
raise ValueError(
f"operation {op_index} refers to {value!r}, which no earlier "
f"operation defined (set \"id\": {name!r} on the op that creates it)")
if kind == "folio" and name in folio_refs:
# the folio's index now, not when it was made; the stored
# index on a build that cannot report folio uuids
ref = _js(name)
return f"(F[{ref}] ? qet.folioIndex(F[{ref}]) : R[{ref}])"
return f"R[{_js(name)}]"
if kind == "num":
if not isinstance(value, (int, float)) or isinstance(value, bool):
raise ValueError(f"operation {op_index}: {key!r} must be a number, "
f"got {value!r}")
return _js(value)
if kind == "list":
if not isinstance(value, list) or not all(isinstance(x, str) for x in value):
raise ValueError(f"operation {op_index}: {key!r} must be a list of strings, "
f"got {value!r}")
return _js(value)
if kind == "indices":
if (not isinstance(value, list) or not value
or not all(isinstance(x, int) and not isinstance(x, bool) for x in value)):
raise ValueError(f"operation {op_index}: {key!r} must be a non-empty list of "
f"integer indices, got {value!r}")
return _js(value)
if kind == "table":
# As for texts below: a uuid is resolved to the current index at
# run time, since deleting an earlier table shifts every index.
if isinstance(value, str) and _UUID_RE.fullmatch(value):
uuid_methods.add("tableIndex")
return f"qet.tableIndex({folio_js}, {_js(value)})"
if not isinstance(value, int) or isinstance(value, bool):
raise ValueError(f"operation {op_index}: {key!r} must be a table index "
f"or its uuid, got {value!r}")
return _js(value)
if kind == "term":
# A terminal's uuid comes from its symbol's definition, so it
# names a terminal only on its element: the op's element, or for
# add_conductor the end's own. Turned into the index the call
# takes at run time; unlike the index, it is defined between two
# terminals at the same point.
if isinstance(value, str) and _UUID_RE.fullmatch(value):
uuid_methods.add("terminalIndex")
return (f"qetMcpTerminal({op_index}, {_js(key)}, {folio_js}, {owner_js}, "
f"{_js(value)})")
if not isinstance(value, int) or isinstance(value, bool):
raise ValueError(f"operation {op_index}: {key!r} must be a terminal index "
f"or its uuid, got {value!r}")
return _js(value)
if kind == "anyterm":
# A terminal of an element the helper resolves itself (for
# place_element, one that does not exist yet): its index, or its
# uuid, which the helper turns into the index.
if isinstance(value, str) and _UUID_RE.fullmatch(value):
return _js(value)
if not isinstance(value, int) or isinstance(value, bool) or value < 0:
raise ValueError(f"operation {op_index}: {key!r} must be a terminal index "
f"or its uuid, got {value!r}")
return _js(value)
if kind == "element_text":
# A field's uuid is unique only within its element (copies keep
# them), so the lookup takes the op's element too.
if isinstance(value, str) and _UUID_RE.fullmatch(value):
uuid_methods.add("elementTextIndex")
return f"qet.elementTextIndex({folio_js}, {element_js}, {_js(value)})"
if not isinstance(value, int) or isinstance(value, bool):
raise ValueError(f"operation {op_index}: {key!r} must be a text field index "
f"or its uuid, got {value!r}")
return _js(value)
if kind in ("text", "shape", "image"):
# A uuid names the item for good; it is turned into the index
# the call takes at run time, by the item's own folio.
if isinstance(value, str) and _UUID_RE.fullmatch(value):
method = {"text": "textIndex", "shape": "shapeIndex",
"image": "imageIndex"}[kind]
uuid_methods.add(method)
return f"qet.{method}({folio_js}, {_js(value)})"
if not isinstance(value, int) or isinstance(value, bool):
raise ValueError(f"operation {op_index}: {key!r} must be a {kind} index, "
f"its uuid, or a \"$name\" reference, got {value!r}")
return _js(value)
if kind == "folio":
# A uuid names the folio for good; it is turned into the index
# the call takes at run time, since adding or removing a folio
# shifts every index after it.
if isinstance(value, str) and _UUID_RE.fullmatch(value):
uuid_methods.add("folioIndex")
return f"qet.folioIndex({_js(value)})"
if value == "":
raise ValueError(f"operation {op_index}: {key!r} is empty -- a folio saved "
f"without a uuid has none in the file until the project is "
f"saved once; give its index instead")
if not isinstance(value, int) or isinstance(value, bool):
raise ValueError(f"operation {op_index}: {key!r} must be a folio index, "
f"its uuid, or a \"$name\" reference, got {value!r}")
return _js(value)
if kind == "bool":
if not isinstance(value, bool):
raise ValueError(f"operation {op_index}: {key!r} must be true or false, "
f"got {value!r}")
return _js(value)
if kind in ("points", "nodes"):
def _num(v):
return isinstance(v, (int, float)) and not isinstance(v, bool)
if not isinstance(value, list) or len(value) < 2:
raise ValueError(f"operation {op_index}: {key!r} must be a list of at "
f"least 2 {'points' if kind == 'points' else 'nodes'}, "
f"got {value!r}")
for item in value:
if not isinstance(item, dict) or not _num(item.get("x")) or not _num(item.get("y")):
raise ValueError(f"operation {op_index}: {key!r} entries must be "
f"{{\"x\": num, \"y\": num, ...}}, got {item!r}")
if kind == "nodes":
if "kind" in item and item["kind"] not in ("corner", "smooth", "symmetric"):
raise ValueError(f"operation {op_index}: {key!r} entry kind "
f"{item['kind']!r} must be corner, smooth or symmetric")
for hkey in ("inHandle", "outHandle"):
if hkey in item and (not isinstance(item[hkey], dict)
or not _num(item[hkey].get("x"))
or not _num(item[hkey].get("y"))):
raise ValueError(f"operation {op_index}: {key!r} entry "
f"{hkey!r} must be {{\"x\": num, \"y\": num}}")
return _js(value)
return _js("" if value is None else str(value))
for i, op in enumerate(operations):
if not isinstance(op, dict):
raise ValueError(f"operation {i} is not an object: {op!r}")
name = op.get("op")
if name not in OPS:
raise ValueError(f"operation {i}: unknown op {name!r}; "
f"expected one of {', '.join(sorted(OPS))}")
method, spec = OPS[name]
if name == "set_conductor" and op.get("property") not in CONDUCTOR_PROPERTIES:
raise ValueError(f"operation {i}: unknown conductor property "
f"{op.get('property')!r}; expected one of "
f"{', '.join(CONDUCTOR_PROPERTIES)}")
route = op.get("route", "default") if name == "add_conductor" else "default"
if route not in ROUTE_MODES:
raise ValueError(f"operation {i}: unknown route {route!r}; "
f"expected one of {', '.join(ROUTE_MODES)}")
if name == "route_conductor":
uuid_methods.add("routeConductor")
if route == "avoid":
uuid_methods.add("routeConductorBetween")
if name == "set_folio" and op.get("property") not in FOLIO_PROPERTIES:
raise ValueError(f"operation {i}: unknown folio property "
f"{op.get('property')!r}; expected one of "
f"{', '.join(FOLIO_PROPERTIES)}")
if name in ("add_autonum", "remove_autonum", "rename_autonum") and op.get("kind") not in AUTONUM_KINDS:
raise ValueError(f"operation {i}: unknown kind {op.get('kind')!r}; "
f"expected one of {', '.join(AUTONUM_KINDS)}")
if name == "search_and_replace":
if op.get("kind") not in SEARCH_REPLACE_KINDS:
raise ValueError(f"operation {i}: unknown kind {op.get('kind')!r}; "
f"expected one of {', '.join(SEARCH_REPLACE_KINDS)}")
if op.get("kind") == "conductor" and op.get("field") not in CONDUCTOR_PROPERTIES:
raise ValueError(f"operation {i}: unknown conductor field "
f"{op.get('field')!r}; expected one of "
f"{', '.join(CONDUCTOR_PROPERTIES)}")
if op.get("kind") == "element_info" and not op.get("field"):
raise ValueError(f"operation {i}: element_info needs a non-empty "
f"\"field\" (information key)")
if (name == "set_conductor_default"
and op.get("property") not in CONDUCTOR_DEFAULT_PROPERTIES):
raise ValueError(f"operation {i}: unknown conductor default "
f"{op.get('property')!r}; expected one of "
f"{', '.join(CONDUCTOR_DEFAULT_PROPERTIES)}")
if name == "set_folio_border" and op.get("property") not in FOLIO_BORDER_PROPERTIES:
raise ValueError(f"operation {i}: unknown folio border property "
f"{op.get('property')!r}; expected one of "
f"{', '.join(FOLIO_BORDER_PROPERTIES)}")
preset = name == "set_folio_border" and op.get("property") == "preset"
if preset:
if str(op.get("value", "")).lower() not in FOLIO_PRESETS:
raise ValueError(f"operation {i}: unknown folio preset {op.get('value')!r}; "
f"expected one of {', '.join(FOLIO_PRESETS)}")
# Only a build that has it can say why it refused one.
uuid_methods.add("folioPresets")
if name == "add_element_text" and op.get("source") not in ELEMENT_TEXT_SOURCES:
raise ValueError(f"operation {i}: unknown source {op.get('source')!r}; "
f"expected one of {', '.join(ELEMENT_TEXT_SOURCES)}")
if name == "set_element_text" and op.get("property") not in ELEMENT_TEXT_PROPERTIES:
raise ValueError(f"operation {i}: unknown element-text property "
f"{op.get('property')!r}; expected one of "
f"{', '.join(ELEMENT_TEXT_PROPERTIES)}")
if name == "set_shape" and op.get("property") not in SHAPE_PROPERTIES:
raise ValueError(f"operation {i}: unknown shape property "
f"{op.get('property')!r}; expected one of "
f"{', '.join(SHAPE_PROPERTIES)}")
if name == "add_shape" and op.get("shape") not in SHAPES:
raise ValueError(f"operation {i}: unknown shape {op.get('shape')!r}; "
f"expected one of {', '.join(SHAPES)}")
# The conductor ops take "conductor": "{uuid}" in place of element +
# terminal: a uuid names one conductor for good, where a terminal
# can carry several.
if name in OP_DEFAULTS:
op = {**OP_DEFAULTS[name], **op}
if name == "align_elements":
if op.get("edge") not in ALIGN_EDGES:
raise ValueError(f"operation {i}: unknown edge {op.get('edge')!r}; "
f"expected one of {', '.join(ALIGN_EDGES)}")
if not isinstance(op.get("elements"), list) or len(op["elements"]) < 2:
raise ValueError(f"operation {i}: align_elements needs at least 2 elements")
if name == "distribute_elements":
if op.get("axis") not in DISTRIBUTE_AXES:
raise ValueError(f"operation {i}: unknown axis {op.get('axis')!r}; "
f"expected one of {', '.join(DISTRIBUTE_AXES)}")
pitch = op.get("pitch")
if isinstance(pitch, (int, float)) and not isinstance(pitch, bool) and pitch < 0:
raise ValueError(f"operation {i}: pitch must be >= 0 (0: spread evenly)")
least = 2 if isinstance(pitch, (int, float)) and pitch > 0 else 3
if not isinstance(op.get("elements"), list) or len(op["elements"]) < least:
raise ValueError(f"operation {i}: distribute_elements needs at least "
f"{least} elements" + ("" if least == 2 else
" (or 2 with a \"pitch\")"))
if name == "place_element":
if op.get("side") not in PLACE_SIDES:
raise ValueError(f"operation {i}: unknown side {op.get('side')!r}; expected "
f"one of {', '.join(s for s in PLACE_SIDES if s)}, or none "
"for the way next_to_terminal faces")
gap = op.get("gap")
if isinstance(gap, (int, float)) and not isinstance(gap, bool) and gap <= 0:
raise ValueError(f"operation {i}: gap must be > 0")
if method and method.startswith("qetMcp"):
uuid_methods.update(_HELPER_NEEDS[method])
conductor_js = None
if name in CONDUCTOR_UUID_OPS and "conductor" in op:
if "element" in op or "terminal" in op:
raise ValueError(f"operation {i} ({name}): give either \"conductor\" "
f"(its uuid) or \"element\" + \"terminal\", not both")
if op["conductor"] == "":
raise ValueError(f"operation {i}: \"conductor\" is empty -- a conductor "
f"from a project saved before conductors carried a uuid "
f"has none in the file; name it by \"element\" + "
f"\"terminal\" instead")
if not isinstance(op["conductor"], str) or not _UUID_RE.fullmatch(op["conductor"]):
raise ValueError(f"operation {i}: \"conductor\" must be a conductor uuid, "
f"got {op['conductor']!r}")
conductor_js = _js(op["conductor"])
uuid_methods.add("conductorEnds")
op = {**op, "element": "{00000000-0000-0000-0000-000000000000}", "terminal": 0}
args = []
for key, kind in spec:
if key not in op:
raise ValueError(f"operation {i} ({name}) is missing {key!r}")
folio_js = args[0] if args else "0"
element_js = args[1] if len(args) > 1 else None
owner_js = args[-1] if args else None # a terminal's element precedes it
args.append(ref_or(op[key], kind, i, key))
if conductor_js is not None:
args[1], args[2] = f"e{i}.element", f"e{i}.terminal"
ident = op.get("id")
if ident is not None:
if not isinstance(ident, str) or not ident or ident.startswith("$"):
raise ValueError(f"operation {i}: \"id\" must be a non-empty name "
f"without a leading $, got {ident!r}")
if ident in refs:
raise ValueError(f"operation {i}: \"id\" {ident!r} is already used")
call = "qet.addFolio()" if method is None else f"qet.{method}({', '.join(args)})"
if method and method.startswith("qetMcp"):
call = f"{method}({', '.join(args)})"
lines.append(" if (!stop) {")
lines.append(f" qetMcpOp = {i};")
if conductor_js is not None:
lines.append(f" var e{i} = qetMcpConductorEnd({i}, {args[0]}, {conductor_js});")
call = f"(e{i} ? {call} : false)"
lines.append(f" var v{i} = {call};")
if preset:
# Say what the preset chose, and the size of the frame and title
# block as a PDF export measures it: plus its one-pixel line, at
# 96 pixels an inch, in points. The export writes that on the
# standard sheet it is within 3 pt of.
f = args[0]
lines.append(
f" if (v{i}) qet.log({_js(_MARKER)} + JSON.stringify({{kind: 'op_note', "
f"index: {i}, note: qet.folioBorder({f}, 'columns') + ' columns of ' + "
f"qet.folioBorder({f}, 'column-width') + ', ' + qet.folioBorder({f}, 'rows') + "
f"' rows of ' + qet.folioBorder({f}, 'row-height') + '; frame ' + "
f"Math.ceil(Number(qet.folioBorder({f}, 'width')) + 1) * 0.75 + ' x ' + "
f"Math.ceil(Number(qet.folioBorder({f}, 'height')) + 1) * 0.75 + ' pt'}}));")
if route == "avoid":
# The wire exists either way; a route not found leaves it on the
# default path, which is a note on the op, not a failure of it.
lines.append(f" if (v{i} === true) {{ var r{i} = qet.routeConductorBetween("
f"{', '.join(args)}); if (r{i} !== 'routed') qet.log({_js(_MARKER)} + "
f"JSON.stringify({{kind: 'op_note', index: {i}, note: 'no route "
f"around the symbols was found; the conductor keeps the default "
f"path'}})); }}")
if name == "route_conductor":
lines.append(f" if (v{i} === 'no-route') qet.log({_js(_MARKER)} + "
f"JSON.stringify({{kind: 'op_note', index: {i}, note: 'no route "
f"around the symbols was found; the conductor keeps its path'}}));")
if ident is not None:
lines.append(f" R[{_js(ident)}] = v{i};")
refs.add(ident)
if name in FOLIO_MAKING_OPS:
lines.append(f" F[{_js(ident)}] = (typeof qet.folioUuid === 'function' "
f"&& v{i} >= 0) ? qet.folioUuid(v{i}) : '';")
folio_refs.add(ident)
lines.append(
f" qet.log({_js(_MARKER)} + JSON.stringify("
f"{{kind: 'op', index: {i}, op: {_js(name)}, "
f"id: {_js(ident)}, result: v{i}}}));")
# An op that failed usually invalidates the ones after it -- a
# conductor to an element that was never placed is not a second,
# independent finding, it is noise on top of the first one. The
# three falsey returns are the three the API uses: false for a
# refused edit, "" for an addElement that placed nothing, -1 for an
# addFolio that added none.
# An empty list is a failure too (duplicateElements returns [] when it
# refuses). Duck-typed on .length, not Array.isArray: QJSEngine hands
# an empty QStringList back as an array-like wrapper for which
# Array.isArray is false, so that test never fired and the run went
# on past the failed operation.
lines.append(f" if (v{i} === false || v{i} === '' || v{i} === -1 || "
f"(typeof v{i} === 'object' && v{i} !== null && "
f"(v{i}.length === 0 || (v{i}.length === undefined && "
f"Object.keys(v{i}).length === 0)))) "
"stop = true;")
lines.append(" }")
lines.append(" // ---- end of operations ----")
# Save even after a failed op: a partial result that can be inspected
# beats no result at all, and the diff is what says how far it got.
lines.append(f" var saved = qet.save({_js(output)});")
lines.append(f" qet.log({_js(_MARKER)} + JSON.stringify("
"{kind: 'save', result: saved, stopped_early: stop}));")
lines.append("}")
need = _js(sorted(set(_REQUIRED_METHODS) | uuid_methods))
return "\n".join(line.replace("@NEED@", need) for line in lines) + "\n"
def _parse_script_output(text: str) -> dict:
"""Read the marker lines the generated script emits.
They arrive on stderr, not stdout: QetScriptApi::log() is a
QTextStream(stderr). Read both anyway rather than depending on that --
the cost is nothing and the failure it prevents is silent (an edit that
worked, reported as having run no operations at all, which is what the
first version of this tool did)."""
caps, ops, saved, stopped, notes = None, [], None, False, {}
for line in text.splitlines():
idx = line.find(_MARKER)
if idx < 0:
continue
try:
rec = json.loads(line[idx + len(_MARKER):])
except json.JSONDecodeError:
continue
if rec.get("kind") == "capabilities":
caps = rec.get("missing") or []
elif rec.get("kind") == "op":
rec.pop("kind", None)
# Not "in (False, ...)": 0 == False in Python, and 0 is a valid
# index/folio result. The script side already uses strict ===.
r = rec.get("result")
rec["succeeded"] = not (r is None or r is False or r == "" or r == [] or r == {} or
(isinstance(r, int) and not isinstance(r, bool) and r == -1))
ops.append(rec)
elif rec.get("kind") == "op_note":
# An op can log more than one (add_conductor, one per end):
# keep them all rather than only the last.
idx = rec.get("index")
notes[idx] = (notes[idx] + "; " if idx in notes else "") + str(rec.get("note"))
elif rec.get("kind") == "save":
saved = bool(rec.get("result"))
stopped = bool(rec.get("stopped_early"))
for rec in ops:
if rec.get("index") in notes:
rec["note"] = notes[rec["index"]]
return {"missing_methods": caps, "operations": ops, "saved": saved,
"stopped_early": stopped}
def tool_query(binary: str, project: str, sql: str,
elements_dir: str | None = None, timeout: int = 180) -> dict:
"""Run a read-only SELECT against the project's SQLite database.
This is the surface the rest of this server has done without. Every
other structural tool here re-derives its answer from the XML, because
the database was unreachable from outside the application; it is
reachable now, through the same guarded path QElectroTech's own
"Requête SQL personnalisée" box uses, so a structural question can be
asked of the database that already knows it.
The three *_view names are the surface to depend on --
element_nomenclature_view, project_summary_view, wiring_list_view.
They exist to be queried. The underlying tables are how the cache is
arranged today and a column may move; qet_query with sql omitted
lists both.
"""
proj = Path(project).expanduser()
if not proj.is_file():
raise ValueError(f"no such project: {proj}")
if sql:
# Same first-word rule projectDataBase::isReadOnlySelect() applies,
# checked here too so an obvious write is refused without paying
# for a process launch. QET still enforces it; this is not the
# guard, only an early one.
head = sql.strip().lstrip("(").split(None, 1)[0].upper() if sql.strip() else ""
if head not in ("SELECT", "WITH"):
raise ValueError("only read-only queries are allowed: a statement must "
f"begin with SELECT or WITH, not {head or '(nothing)'}")
script = ("var out = %s ? qet.query(%s) : qet.tables();\n"
"qet.log(%s + JSON.stringify({kind: 'query', rows: out, "
"error: qet.queryError ? qet.queryError() : ''}));\n"
% (json.dumps(bool(sql)), json.dumps(sql or ""), json.dumps(_MARKER)))
result = _run_qet(binary, [str(proj)], timeout=timeout,
elements_dir=elements_dir, script=script, tail=400_000)
streams = result.get("stdout", "") + "\n" + result.get("stderr", "")
rows, error = None, ""
for line in streams.splitlines():
idx = line.find(_MARKER)
if idx < 0:
continue
try:
rec = json.loads(line[idx + len(_MARKER):])
except json.JSONDecodeError:
continue
if rec.get("kind") == "query":
rows, error = rec.get("rows"), rec.get("error") or ""
for key in ("stdout", "stderr"):
kept = [ln for ln in result.get(key, "").splitlines() if _MARKER not in ln]
result[key] = "\n".join(kept)[-4000:]
if rows is None:
result["ok"] = False
result.setdefault("hint", "the query returned nothing at all -- this build's "
"scripting API may predate qet.query()")
return result
if error:
result["ok"] = False
result["error"] = error
result["rows"] = rows
result["row_count"] = len(rows)
result["listing"] = not sql
return result
# --------------------------------------------------------------------------
# qet_continuity: electrical continuity / ERC-style structural checks
# --------------------------------------------------------------------------
def tool_continuity(binary: str, project: str, folio: int | None = None,
elements_dir: str | None = None, timeout: int = 180) -> dict:
"""Run qet.checkContinuity() and get its findings back.
Three checks, against the live Terminal/Conductor object graph rather
than a heuristic read of the XML (that is qet_check's job, and the two
are complementary, not redundant -- qet_check looks at labels and
numbering conventions, this looks at the electrical graph itself):
unconnected_terminal (info -- routine, not necessarily a mistake),
potential_mismatch (error -- two conductors QElectroTech's own
setConductorProperty() would always keep identical, found disagreeing,
which only happens from hand-edited XML, a legacy file, or an external
tool), and report_link_mismatch (warning -- a next_report/
previous_report folio-jump pair whose conductors disagree on colour,
style, num, etc.; unlike potential_mismatch this one CAN happen through
ordinary use, since LinkElementCommand::isLinkable() never checks
conductor properties, only type and freedom -- see
qelectrotech/qelectrotech-source-mirror#974, which this check
reproduces exactly: one folio-link conductor drawn in two different
colours on either side of the link). See qet.checkContinuity()'s own
doc comment (qetscriptapi.cpp) for what this deliberately does not
check: pin electrical direction/power conflicts and No/Nc/Common
contact shorts, since QElectroTech's terminal data model does not
carry the information either would need.
"""
proj = Path(project).expanduser()
if not proj.is_file():
raise ValueError(f"no such project: {proj}")
if folio is not None:
# qet.checkContinuity() answers an index it has no folio for with an
# empty list, which reads exactly like a clean folio. The index counts
# from 0 while qet_elements numbers folios from 1, so the likely
# mistake -- passing the last folio's number -- would pass silently.
count = len(list(_folios(_root(str(proj)))))
if not 0 <= folio < count:
raise ValueError(
f"folio {folio} does not exist: the project has {count} folio(s), "
f"indexed 0 to {count - 1} here. qet_continuity counts folios from 0; "
"the folio qet_elements calls N is N - 1.")
folio_arg = -1 if folio is None else folio
script = ("var out = qet.checkContinuity(%s);\n"
"qet.log(%s + JSON.stringify({kind: 'continuity', findings: out}));\n"
% (json.dumps(folio_arg), json.dumps(_MARKER)))
result = _run_qet(binary, [str(proj)], timeout=timeout,
elements_dir=elements_dir, script=script, tail=400_000)
streams = result.get("stdout", "") + "\n" + result.get("stderr", "")
findings = None
for line in streams.splitlines():
idx = line.find(_MARKER)
if idx < 0:
continue
try:
rec = json.loads(line[idx + len(_MARKER):])
except json.JSONDecodeError:
continue
if rec.get("kind") == "continuity":
findings = rec.get("findings")
for key in ("stdout", "stderr"):
kept = [ln for ln in result.get(key, "").splitlines() if _MARKER not in ln]
result[key] = "\n".join(kept)[-4000:]
if findings is None:
result["ok"] = False
result.setdefault("hint", "no findings came back at all -- this build's "
"scripting API may predate qet.checkContinuity()")
return result
for f in findings:
# "folio" is the 0-based index qet.checkContinuity() uses; add the
# number qet_elements and the application show, so the two can be
# matched without arithmetic.
if isinstance(f, dict) and isinstance(f.get("folio"), int):
f["folio_number"] = f["folio"] + 1
result["findings"] = findings
result["finding_count"] = len(findings)
result["errors"] = sum(1 for f in findings if f.get("severity") == "error")
result["warnings"] = sum(1 for f in findings if f.get("severity") == "warning")
result["info"] = sum(1 for f in findings if f.get("severity") == "info")
return result
# --------------------------------------------------------------------------
# qet_element_search: find a symbol in the collection
# --------------------------------------------------------------------------
_ELEMENT_INDEX: dict = {}
def _fold(text: str) -> str:
"""Case- and accent-insensitive form, so 'resistance' finds 'Résistance'."""
import unicodedata
return "".join(c for c in unicodedata.normalize("NFKD", text.lower())
if not unicodedata.combining(c))
# Spellings of "normally open" and "normally closed" in the collection's names
# and file names, folded to one word each: N/O, NF (French "normalement
# fermé"), the words written out.
_CONTACT_SYNONYMS = (
(re.compile(r"\bnormal(?:ly|ement)\s+(?:closed|fermee?s?)\b"), " nc "),
(re.compile(r"\bnormal(?:ly|ement)\s+(?:open|ouverte?s?)\b"), " no "),
(re.compile(r"\bn\s*/\s*([ocf])\b"), r" n\1 "),
)
def _search_words(text: str) -> list:
"""The words of text as a search sees them: folded, split on anything
that is not a letter or digit (so a path's '_' and '/' separate words),
with every spelling of NO and NC reduced to 'no' and 'nc'."""
t = _fold(text)
for pattern, repl in _CONTACT_SYNONYMS:
t = pattern.sub(repl, t)
return ["nc" if w == "nf" else w for w in re.findall(r"[^\W_]+", t)]
def _word_matches(word: str, words: frozenset) -> bool:
"""A query of one or two letters must be a whole word, so 'nc' does
not find 'remanence' and 'no' does not find 'normal'. Anything longer,
or with a digit, may be part of a word: 'schutz' finds
'Leitungsschutzschalter', '3p' finds '3pn'."""
if word in words:
return True
if len(word) <= 2 and word.isalpha():
return False
return any(word in w for w in words)
# Folders of drawings that are not schematic symbols, or of one maker's
# parts: offered after everything else, so "emergency stop" gives the
# push button before an assembly-plan drawing of one.
_SECONDARY_FOLDER = re.compile(r"(?:^|/)\d+_(?:graphics|manufacturers_articles|miscellaneous_unsorted)/")
def _collection_signature(root: Path):
"""Cheap change detector: file count and newest mtime, no parsing."""
count, newest = 0, 0.0
for f in root.rglob("*.elmt"):
count += 1
try:
newest = max(newest, f.stat().st_mtime)
except OSError:
pass
return count, newest
def _index_collection(root: Path) -> list:
"""Parse every .elmt under root once and keep what a search needs.
Cached for the life of the process and rebuilt when the file count or
the newest modification time changes -- which is what makes a symbol
written by qet_element_build findable straight away, without the caller
knowing there is an index at all.
"""
key = str(root.resolve())
sig = _collection_signature(root)
cached = _ELEMENT_INDEX.get(key)
if cached and cached["sig"] == sig:
return cached["items"]
items = []
for f in sorted(root.rglob("*.elmt")):
try:
d = ET.parse(f).getroot()
except (ET.ParseError, OSError):
continue
if d.tag != "definition":
continue
names = {n.get("lang", ""): (n.text or "").strip() for n in d.iter("name")}
kind = ""
for ki in d.iter("kindInformation"):
if ki.get("name") == "type":
kind = (ki.text or "").strip()
ordered, ambiguous = _terminals_in_index_order(list(d.iter("terminal")))
terminals = [t.get("name") or "" for t in ordered]
rel = f.relative_to(root).as_posix()
items.append({
"path": "common://" + rel,
"file": str(f),
"name": names.get("en") or names.get("fr") or next(iter(names.values()), ""),
"names": names,
"link_type": d.get("link_type", "simple"),
"kind": kind,
"terminals": len(terminals),
"terminal_names": terminals, # index order, not file order
"terminal_order_ambiguous": ambiguous,
"width": d.get("width"), "height": d.get("height"),
"haystack": frozenset(_search_words(" ".join([*names.values(), rel, kind]))),
})
_ELEMENT_INDEX[key] = {"sig": sig, "items": items}
return items
def tool_element_search(directory: str, query: str = "", link_type: str | None = None,
min_terminals: int | None = None, max_terminals: int | None = None,
kind: str | None = None, limit: int = 25) -> dict:
"""Search an element collection by name, type and terminal count.
Matches every word of query against all the translated names, the
element's path and its kind, ignoring case and accents -- so a French
or German search finds the same symbol an English one does. Results
carry a common:// path that qet_edit's add_element takes directly, and
the terminal names in the order add_conductor indexes them -- which is
top to bottom then left to right, not the order the file lists them.
"""
root = Path(directory).expanduser()
if not root.is_dir():
raise ValueError(f"no such directory: {root}")
if link_type is not None and link_type not in LINK_TYPES:
raise ValueError(f"unknown link_type {link_type!r}; expected one of "
f"{', '.join(LINK_TYPES)}")
if limit < 1:
raise ValueError("limit must be >= 1")
words = _fold(query).split()
search = _search_words(query)
matches = []
for it in _index_collection(root):
if link_type and it["link_type"] != link_type:
continue
if kind and _fold(kind) not in _fold(it["kind"]):
continue
if min_terminals is not None and it["terminals"] < min_terminals:
continue
if max_terminals is not None and it["terminals"] > max_terminals:
continue
if not all(_word_matches(w, it["haystack"]) for w in search):
continue
matches.append(it)
# Schematic symbols before drawings and makers' parts; then whole-name
# hits before substring hits, then shorter names first: a search for
# "coil" should offer "Coil" before "Remanence coil, latching".
def rank(it):
name = _fold(it["name"])
secondary = 1 if _SECONDARY_FOLDER.search(it["path"]) else 0
exact = 0 if (words and name == " ".join(words)) else 1
starts = 0 if (words and name.startswith(words[0])) else 1
return (secondary, exact, starts, len(it["name"]), it["path"])
matches.sort(key=rank)
shown = [{k: v for k, v in it.items() if k not in ("haystack", "names", "file")}
| {"languages": sorted(it["names"])} for it in matches[:limit]]
return {"query": query, "total_matches": len(matches), "returned": len(shown),
"indexed": len(_ELEMENT_INDEX[str(root.resolve())]["items"]),
"results": shown}
# Design-rule checks, each one a read-only query over the project database.
#
# Every check is a SELECT that returns the offending rows, so "no rows" is a
# pass and the same query is what a human would write by hand. The severity
# and the note say how much to trust a hit, because these are heuristics
# tuned against the 24 shipped examples, not standards:
#
# - Every text comparison is COALESCE'd. A value that was never set is NULL
# in the database when the element was placed in this session and an empty
# string when it was loaded from a file, and `col = ''` matches only the
# second -- which made the first version of these checks silently pass on
# exactly the freshly-edited projects qet_edit produces. The JavaScript
# side renders both as "", so the difference is invisible until a check
# fails to fire. Likewise exclude_from_bom is text, not a number.
# - An unnumbered conductor is '' in some files and '_' in others: '_' is the
# placeholder QElectroTech assigns when no numbering is configured, so
# testing for '' alone passed on industrial.qet's 36 placeholder conductors
# and on every project qet_edit builds without a numbering context.
# - Slaves and terminals are excluded from the duplicate-label check on
# purpose. A slave contact carries its master coil's label by design, and
# terminals repeat their numbers from one strip to the next; counting
# either would bury the real findings.
# - "simple" elements are checked for duplicates too but only as a warning:
# industrial.qet reuses V1..V6 across folios on purpose.
CHECKS = {
"duplicate_master_labels": {
"severity": "error",
"note": "Two master elements with the same label are ambiguous in every "
"report that keys on it (BOM, cross-references, wiring list).",
"sql": "SELECT label, COUNT(*) AS n FROM element_nomenclature_view "
"WHERE COALESCE(label,'') <> '' AND element_type = 'master' "
"GROUP BY label HAVING n > 1 ORDER BY n DESC, label",
},
"duplicate_simple_labels": {
"severity": "warning",
"note": "Legitimate when a label is reused on purpose across folios "
"(industrial.qet does); worth a look otherwise.",
"sql": "SELECT label, COUNT(*) AS n FROM element_nomenclature_view "
"WHERE COALESCE(label,'') <> '' AND element_type = 'simple' "
"GROUP BY label HAVING n > 1 ORDER BY n DESC, label",
},
"unlabelled_masters": {
"severity": "warning",
"note": "A master with no label cannot be told apart from its slaves' "
"cross-references.",
"sql": "SELECT folio, diagram_position, element_sub_type FROM "
"element_nomenclature_view WHERE element_type = 'master' AND COALESCE(label,'') = '' "
"ORDER BY folio, diagram_position",
},
"unnumbered_conductors": {
"severity": "info",
"note": "Conductors with no wire number -- empty, or QElectroTech's own "
"'_' placeholder, which is what a conductor gets when no "
"numbering is configured. If every conductor is unnumbered the "
"project simply does not use wire numbering; a few among many "
"numbered ones is the finding.",
"sql": "SELECT COUNT(*) AS unnumbered, (SELECT COUNT(*) FROM wiring_list_view) AS total "
"FROM wiring_list_view WHERE COALESCE(wire_number,'') IN ('', '_') "
"HAVING unnumbered > 0",
},
"empty_folios": {
"severity": "info",
"note": "Folios with no element on them. Often cover pages, sometimes "
"left behind by a deleted drawing.",
"sql": "SELECT p.pos AS position, p.title AS title FROM diagram d "
"JOIN project_summary_view p ON p.pos = d.pos "
"WHERE NOT EXISTS (SELECT 1 FROM element e WHERE e.diagram_uuid = d.uuid) "
"ORDER BY p.pos",
},
"crowded_terminals": {
"severity": "info",
"note": "Terminals with more than four wires: two double ferrules, one "
"either side of the screw, is already a lot for a real terminal "
"(discussion #1158). Cable, busbar and single-line symbols carry "
"more on purpose; element_type and label tell them apart.",
"sql": "SELECT d.pos AS folio, e.pos AS diagram_position, "
"COALESCE(ei.label,'') AS label, e.type AS element_type, "
"COALESCE(t.name,'') AS terminal, w.n AS wires FROM ("
"SELECT tu, eu, COUNT(*) AS n FROM ("
"SELECT terminal1_uuid AS tu, terminal1_element_uuid AS eu FROM conductor "
"UNION ALL SELECT terminal2_uuid, terminal2_element_uuid FROM conductor"
") GROUP BY tu, eu HAVING n > 4) AS w "
"JOIN element e ON e.uuid = w.eu "
"LEFT JOIN diagram d ON d.uuid = e.diagram_uuid "
"LEFT JOIN element_info ei ON ei.element_uuid = e.uuid "
"LEFT JOIN terminal t ON t.uuid = w.tu AND t.element_uuid = w.eu "
"WHERE e.type NOT IN ('next_report', 'previous_report') "
"ORDER BY w.n DESC, folio, diagram_position",
},
"reports_with_several_wires": {
"severity": "info",
"note": "Folio report arrows with more than one wire. A report is a "
"virtual point: one wire, continued on the other folio, says "
"where the wire really runs (discussion #1158). A third of the "
"shipped examples' reports have several, so this is a style, "
"not an error.",
"sql": "SELECT d.pos AS folio, e.pos AS diagram_position, "
"COALESCE(ei.label,'') AS label, w.n AS wires FROM ("
"SELECT tu, eu, COUNT(*) AS n FROM ("
"SELECT terminal1_uuid AS tu, terminal1_element_uuid AS eu FROM conductor "
"UNION ALL SELECT terminal2_uuid, terminal2_element_uuid FROM conductor"
") GROUP BY tu, eu HAVING n > 1) AS w "
"JOIN element e ON e.uuid = w.eu "
"LEFT JOIN diagram d ON d.uuid = e.diagram_uuid "
"LEFT JOIN element_info ei ON ei.element_uuid = e.uuid "
"WHERE e.type IN ('next_report', 'previous_report') "
"ORDER BY w.n DESC, folio, diagram_position",
},
"masters_without_manufacturer_reference": {
"severity": "info",
"note": "Masters that will show a blank article number in the BOM.",
"sql": "SELECT label, folio, diagram_position FROM element_nomenclature_view "
"WHERE element_type = 'master' AND COALESCE(manufacturer_reference,'') = '' "
"AND COALESCE(exclude_from_bom,'') IN ('', '0', 'false') "
"ORDER BY folio, diagram_position",
},
}
def tool_check(binary: str, project: str, checks: list | None = None,
sample: int = 10, elements_dir: str | None = None,
timeout: int = 180) -> dict:
"""Run design-rule checks over a project in one QElectroTech launch.
Every check is a read-only SELECT over the project database, so this is
qet_query with the questions already written down. It exists because the
useful questions are always the same handful and re-deriving them per
conversation is where the mistakes creep in -- the first draft of the
duplicate-label check counted slave contacts, which share their coil's
label by design and flagged nearly every relay.
"""
proj = Path(project).expanduser()
if not proj.is_file():
raise ValueError(f"no such project: {proj}")
chosen = list(CHECKS) if not checks else list(checks)
for name in chosen:
if name not in CHECKS:
raise ValueError(f"unknown check {name!r}; expected one of "
f"{', '.join(sorted(CHECKS))}")
if sample < 0:
raise ValueError("sample must be >= 0")
queries = {name: CHECKS[name]["sql"] for name in chosen}
script = ("var Q = %s;\nfor (var k in Q) {\n"
" var rows = qet.query(Q[k]);\n"
" qet.log(%s + JSON.stringify({kind: 'check', name: k, rows: rows, "
"error: qet.queryError()}));\n}\n" % (json.dumps(queries), json.dumps(_MARKER)))
result = _run_qet(binary, [str(proj)], timeout=timeout,
elements_dir=elements_dir, script=script, tail=2_000_000)
streams = result.get("stdout", "") + "\n" + result.get("stderr", "")
got = {}
for line in streams.splitlines():
idx = line.find(_MARKER)
if idx < 0:
continue
try:
rec = json.loads(line[idx + len(_MARKER):])
except json.JSONDecodeError:
continue
if rec.get("kind") == "check":
got[rec["name"]] = rec
findings, errors, passed = [], [], []
for name in chosen:
rec = got.get(name)
if rec is None:
errors.append({"check": name, "error": "no result came back"})
continue
if rec.get("error"):
errors.append({"check": name, "error": rec["error"]})
continue
rows = rec.get("rows") or []
if not rows:
passed.append(name)
continue
findings.append({"check": name, "severity": CHECKS[name]["severity"],
"count": len(rows), "note": CHECKS[name]["note"],
"rows": rows[:sample]})
order = {"error": 0, "warning": 1, "info": 2}
findings.sort(key=lambda f: (order[f["severity"]], f["check"]))
answer = {"ok": not errors and not any(f["severity"] == "error" for f in findings),
"summary": {"errors": sum(f["severity"] == "error" for f in findings),
"warnings": sum(f["severity"] == "warning" for f in findings),
"info": sum(f["severity"] == "info" for f in findings),
"passed": len(passed), "check_failures": len(errors)},
"findings": findings, "passed": passed, "check_failures": errors}
# This answer is built fresh rather than layered onto the launch result,
# so a reason the launch failed at all has to be carried across
# explicitly. Without it every check reads "no result came back", which
# is true and tells nobody why.
if result.get("hint"):
answer["ok"] = False
answer["hint"] = result["hint"]
answer["exit_code"] = result.get("exit_code")
return answer
# --------------------------------------------------------------------------
# Layout check: does the drawing read well?
# --------------------------------------------------------------------------
#
# A wire is straight only when its two terminals share an x or a y exactly.
# An assistant places a symbol by its origin, and its terminals sit at an
# offset from that origin it cannot see, so "under K1" lands a few pixels
# off and QElectroTech draws a jog. This check finds those, with the move
# that removes each one.
#
# The drawn path of a wire is not in the file: a wire on QElectroTech's
# default path is saved with no at all. So the geometry comes from
# QElectroTech, through the read calls of its scripting API, and is scored
# here. Nothing is saved.
LAYOUT_STYLES = ["auto", "iec", "nfpa"]
LAYOUT_GRID = 10.0 # Diagram::xGrid / yGrid
LAYOUT_RULES = {
"avoidable_bend": {
"severity": "warning",
"note": "The two terminals face each other along one axis but are a few "
"pixels out of line, so the wire jogs. Moving one symbol makes it "
"straight; \"fix\" is that symbol's move from \"fixes\". "
"\"conflict\": no move is offered, because both symbols are "
"already lined up by other wires along this axis or moving either "
"would put it on another symbol or across another wire.",
},
"extra_bends": {
"severity": "info",
"note": "The wire bends more often than its two terminals need. Often a "
"segment moved by hand; route_conductor redraws it.",
},
"wire_through_symbol": {
"severity": "warning",
"note": "A wire runs through a symbol it is not connected to. A symbol "
"drawn around one of the wire's own ends is a frame and is left "
"out, as the router does, and so is a symbol with no terminals. "
"Cable tags and shields are drawn across wires on purpose: ignore "
"the finding for those.",
},
"overlapping_symbols": {
"severity": "warning",
"note": "Two symbols overlap by more than one grid step (a symbol's box "
"is its declared size, so side-by-side symbols can share a few "
"pixels of it). Frames, drawn around another symbol, and symbols "
"with no terminals (tags, shields, label holders) are left out.",
},
"off_grid": {
"severity": "warning",
"note": "The symbol's origin is off the 10 px grid QElectroTech snaps "
"symbols to, so its terminals are off the grid the other symbols "
"are on. An axis a straight wire lines it up on is left alone. "
"\"fix\" moves it onto the grid.",
},
"crossing": {
"severity": "info",
"note": "Two wires cross. Counted so two drafts can be compared; some "
"crossings cannot be avoided, so they do not lower the score.",
},
}
_LAYOUT_SEG = re.compile(r"^\s*\d+:\s*\(([^,]+),([^)]+)\)-\(([^,]+),([^)]+)\)")
# One launch, read-only. Per folio: every symbol's geometry and terminals,
# every wire's ends and drawn path. conductorPath() reads any wire by its
# uuid; a build without it reads a wire through one of its ends, which
# conductorSegments() refuses on a terminal carrying a second wire.
_LAYOUT_JS = r"""
var only = @FOLIO@;
var byUuid = typeof qet.conductorPath === 'function';
for (var f = 0; f < qet.folioCount(); f++) {
if (only >= 0 && f !== only) continue;
var els = qet.elementUuids(f), E = [];
for (var i = 0; i < els.length; i++) {
E.push({uuid: els[i], name: qet.elementName(f, els[i]),
label: qet.elementLabel(f, els[i]), g: qet.elementGeometry(f, els[i]),
terminals: qet.elementTerminals(f, els[i]).length});
}
var cu = qet.conductorUuids(f), lines = qet.conductors(f), C = [];
for (var j = 0; j < cu.length; j++) {
var ends = qet.conductorEnds(f, cu[j]), path = null, segs = null;
if (byUuid) {
path = qet.conductorPath(f, cu[j]);
} else if (ends.length === 2) {
for (var k = 0; k < 2 && segs === null; k++) {
if (ends[k] === '?') continue;
var n = 0;
for (var l = 0; l < lines.length; l++) {
var p = lines[l].split(' : ')[0].split(' -- ');
if (p[0] === ends[k] || p[1] === ends[k]) n++;
}
if (n !== 1) continue;
var m = ends[k].split(' terminal ');
segs = qet.conductorSegments(f, m[0], parseInt(m[1], 10));
}
}
C.push({uuid: cu[j], ends: ends, path: path, segs: segs});
}
qet.log(@MARKER@ + JSON.stringify({kind: 'layout', folio: f, elements: E,
conductors: C}));
}
"""
def _layout_points(wire: dict) -> list | None:
"""The wire's drawn path as points, from either read call; None if it
could not be read."""
if wire.get("path"):
try:
return [(float(p["x"]), float(p["y"])) for p in wire["path"]]
except (KeyError, TypeError, ValueError):
return None
segs = wire.get("segs")
if not segs:
return None
pts = []
for line in segs:
mt = _LAYOUT_SEG.match(line)
if not mt:
return None
x1, y1, x2, y2 = (float(v) for v in mt.groups())
if not pts:
pts.append((x1, y1))
pts.append((x2, y2))
return pts if len(pts) >= 2 else None
def _simplify(pts: list) -> list:
"""Drop zero-length steps and merge straight runs, so what is left has a
corner at every inner point."""
out = []
for p in pts:
if out and abs(p[0] - out[-1][0]) < 1e-6 and abs(p[1] - out[-1][1]) < 1e-6:
continue
if len(out) >= 2:
a, b = out[-2], out[-1]
if ((abs(a[0] - b[0]) < 1e-6 and abs(b[0] - p[0]) < 1e-6)
or (abs(a[1] - b[1]) < 1e-6 and abs(b[1] - p[1]) < 1e-6)):
out[-1] = p
continue
out.append(p)
return out
def _facing(dock: tuple, nxt: tuple) -> tuple | None:
"""Which way a terminal sends its wire: the unit step from the dock point
to the next distinct point of the path, or None if that is not along an
axis."""
dx, dy = nxt[0] - dock[0], nxt[1] - dock[1]
if abs(dx) < 1e-6 and abs(dy) > 1e-6:
return (0, 1 if dy > 0 else -1)
if abs(dy) < 1e-6 and abs(dx) > 1e-6:
return (1 if dx > 0 else -1, 0)
return None
def _box(g: dict) -> tuple | None:
try:
return (float(g["left"]), float(g["top"]), float(g["right"]), float(g["bottom"]))
except (KeyError, TypeError, ValueError):
return None
def _contains(outer: tuple, inner: tuple) -> bool:
return (outer[0] <= inner[0] and outer[1] <= inner[1]
and outer[2] >= inner[2] and outer[3] >= inner[3] and outer != inner)
def _overlap(a: tuple, b: tuple, margin: float = 1.0) -> bool:
return (min(a[2], b[2]) - max(a[0], b[0]) > margin
and min(a[3], b[3]) - max(a[1], b[1]) > margin)
def _segment_through(p: tuple, q: tuple, box: tuple, margin: float = 1.0) -> bool:
"""Does the axis-aligned segment p-q run through the inside of box?"""
l, t, r, b = box[0] + margin, box[1] + margin, box[2] - margin, box[3] - margin
if l >= r or t >= b:
return False
if abs(p[1] - q[1]) < 1e-6: # horizontal
lo, hi = sorted((p[0], q[0]))
return t < p[1] < b and min(hi, r) - max(lo, l) > 1e-6
if abs(p[0] - q[0]) < 1e-6: # vertical
lo, hi = sorted((p[1], q[1]))
return l < p[0] < r and min(hi, b) - max(lo, t) > 1e-6
return False
def _crosses(a: tuple, b: tuple, c: tuple, d: tuple) -> bool:
"""Do a horizontal and a vertical segment cross inside both?"""
if abs(a[1] - b[1]) < 1e-6 and abs(c[0] - d[0]) < 1e-6:
h, v = (a, b), (c, d)
elif abs(a[0] - b[0]) < 1e-6 and abs(c[1] - d[1]) < 1e-6:
h, v = (c, d), (a, b)
else:
return False
x, y = v[0][0], h[0][1]
hx = sorted((h[0][0], h[1][0]))
vy = sorted((v[0][1], v[1][1]))
return hx[0] + 1e-6 < x < hx[1] - 1e-6 and vy[0] + 1e-6 < y < vy[1] - 1e-6
def _end_element(end: str) -> str:
return end.split(" terminal ")[0] if " terminal " in end else ""
def _grid_offset(v: float) -> float:
"""How far v must move to reach the nearest grid line."""
return round(v / LAYOUT_GRID) * LAYOUT_GRID - v
def _layout_folio(data: dict, max_shift: float) -> dict:
"""Score one folio's dump and plan the moves that fix it. Pure: no
QElectroTech, so every rule is testable with made-up geometry.
Fixes are planned together, one move per symbol, because they interact:
two jogs can ask one symbol to move two ways, and snapping a symbol to
the grid can bend a straight wire. So straight wires are taken first and
pin their two symbols on their axis; jogs then move a symbol not yet
pinned, preferring a move that lands it on the grid; grid snaps come
last and only on an axis nothing pinned.
Applying all of "fixes" at once is therefore consistent; applying each
finding's fix on its own, one after the other, is not.
"""
folio = int(data.get("folio", 0))
symbols = {}
for el in data.get("elements") or []:
box = _box(el.get("g") or {})
if box is None:
continue
g = el["g"]
symbols[el["uuid"]] = {"uuid": el["uuid"], "name": el.get("name", ""),
"label": el.get("label", ""),
# No terminals: a label holder or a drawing
# aid, put on top of other symbols on purpose.
"annotation": int(el.get("terminals", 1) or 0) == 0,
"x": float(g.get("x", 0)), "y": float(g.get("y", 0)),
"xy": (float(g.get("x", 0)), float(g.get("y", 0))),
"box": box, "docks": []}
wire_count = {}
wires, unread = [], []
for w in data.get("conductors") or []:
ends = [_end_element(e) for e in (w.get("ends") or [])]
for e in ends:
if e:
wire_count[e] = wire_count.get(e, 0) + 1
pts = _layout_points(w)
if pts is None:
unread.append(w.get("uuid", ""))
continue
ends = (ends + ["", ""])[:2]
# The path runs from the first end's terminal to the second's.
for end, dock in ((ends[0], pts[0]), (ends[1], pts[-1])):
if end in symbols:
symbols[end]["docks"].append(dock)
wires.append({"uuid": w.get("uuid", ""), "ends": ends, "pts": _simplify(pts)})
findings = []
dirty_wires, dirty_symbols = set(), set()
length = {"vertical": 0.0, "horizontal": 0.0}
move = {} # symbol uuid -> [dx, dy], the one planned move
locked = set() # (symbol uuid, axis) a jog fix already decided
def add(rule, **fields):
findings.append({"rule": rule, "severity": LAYOUT_RULES[rule]["severity"],
"folio": folio + 1, **fields})
def on_grid(v):
return abs(_grid_offset(v)) < 1e-6
solid = [x for x in symbols.values() if not x["annotation"]]
# Symbols lined up on an axis by straight wires (as planned) form a
# group; the grid snap moves a group together so it stays in line.
parent = {}
def find(k):
parent.setdefault(k, k)
while parent[k] != k:
parent[k] = parent[parent[k]]
k = parent[k]
return k
def union(a, b):
parent[find(a)] = find(b)
def shifted(box, d):
return (box[0] + d[0], box[1] + d[1], box[2] + d[0], box[3] + d[1])
def blocked(uuid, total):
"""Would moving this symbol by total (its whole planned move) put it
on another symbol, or across a wire it is not on, where it was not
before? A fix must not trade a jog for a collision."""
me = symbols[uuid]
if me["annotation"]:
return False
old, new = me["box"], shifted(me["box"], total)
for o in solid:
if o["uuid"] == uuid:
continue
ob = shifted(o["box"], move.get(o["uuid"], [0, 0]))
if _contains(ob, old) or _contains(old, ob) or _contains(ob, new) or _contains(new, ob):
continue
if _overlap(new, ob, margin=LAYOUT_GRID) and not _overlap(old, o["box"], margin=LAYOUT_GRID):
return True
for w in wires:
if uuid in w["ends"]:
continue
for p, q in zip(w["pts"], w["pts"][1:]):
if _segment_through(p, q, new) and not _segment_through(p, q, old):
return True
return False
for w in wires:
pts = w["pts"]
for p, q in zip(pts, pts[1:]):
length["vertical" if abs(p[0] - q[0]) < 1e-6 else "horizontal"] += (
abs(p[0] - q[0]) + abs(p[1] - q[1]))
w["bends"] = max(0, len(pts) - 2)
# Every wire whose terminals face each other along one axis: straight
# ones first, so they pin their symbols on that axis (a fix must not
# trade one straight wire for another), then jogs, smallest first.
in_line = []
for w in wires:
pts, bends = w["pts"], w["bends"]
if len(pts) < 2:
continue
a, b = pts[0], pts[-1]
fa, fb = _facing(a, pts[1]), _facing(b, pts[-2])
if not (fa and fb):
continue
if fa[0] == 0 and fb[0] == 0: # both vertical
axis, i = "x", 0
facing = fa[1] == (1 if b[1] > a[1] else -1) and fb[1] == -fa[1]
elif fa[1] == 0 and fb[1] == 0: # both horizontal
axis, i = "y", 1
facing = fa[0] == (1 if b[0] > a[0] else -1) and fb[0] == -fa[0]
else: # an L at best
if bends > 1:
add("extra_bends", conductor=w["uuid"], bends=bends, needed=1,
note=LAYOUT_RULES["extra_bends"]["note"],
fix={"op": "route_conductor", "folio": folio, "conductor": w["uuid"]})
dirty_wires.add(w["uuid"])
continue
offset = b[i] - a[i]
straight = facing and abs(offset) < 1e-6
jog = facing and 1e-6 <= abs(offset) <= max_shift and bends > 0
if straight or jog:
in_line.append((0 if straight else 1, abs(offset), w, axis, i, a, b, jog))
need = 0 if straight else 2
if not jog and bends > need:
add("extra_bends", conductor=w["uuid"], bends=bends, needed=need,
note=LAYOUT_RULES["extra_bends"]["note"],
fix={"op": "route_conductor", "folio": folio, "conductor": w["uuid"]})
dirty_wires.add(w["uuid"])
for _, _, w, axis, i, a, b, jog in sorted(in_line, key=lambda t: t[:2]):
ea, eb = w["ends"]
# As it will be once the moves planned so far are applied.
left = (b[i] + move.get(eb, [0, 0])[i]) - (a[i] + move.get(ea, [0, 0])[i])
mover = None
if abs(left) > 1e-6:
# Prefer the end that lands on the grid, then the one with fewer
# other wires (on a tie the second, so the first anchors a chain).
def rank(e):
delta = -left if e == eb else left
origin = symbols[e]["xy"][i] + move.get(e, [0, 0])[i] + delta
return (not on_grid(origin), wire_count.get(e, 0), e != eb)
free = [e for e in (ea, eb) if e in symbols and (e, axis) not in locked]
for cand in sorted(free, key=rank):
total = list(move.get(cand, [0.0, 0.0]))
total[i] += -left if cand == eb else left
if blocked(cand, total):
continue
mover = cand
move[cand] = total
break
locked.update((e, axis) for e in (ea, eb) if e in symbols)
if (abs(left) < 1e-6 or mover is not None) and ea in symbols and eb in symbols:
union((ea, axis), (eb, axis))
if jog:
w["mover"] = mover
w["conflict"] = abs(left) > 1e-6 and mover is None
add("avoidable_bend", conductor=w["uuid"],
offset=round(abs(b[i] - a[i]), 3), bends=w["bends"],
note=LAYOUT_RULES["avoidable_bend"]["note"])
dirty_wires.add(w["uuid"])
# Off the grid: the symbol's origin, which is what QElectroTech's own
# grid snaps. A symbol lined up with others by straight wires moves only
# with its whole group, and only when they are all off by the same
# amount -- straight wires matter more than the grid, and snapping one
# member alone would bend them, so the next run would undo it.
groups = {}
for (uuid, axis) in locked:
groups.setdefault(find((uuid, axis)), set()).add(uuid)
def offset(u, i):
return _grid_offset(symbols[u]["xy"][i] + move.get(u, [0, 0])[i])
step = {u: [0.0, 0.0] for u in symbols}
for i, axis in ((0, "x"), (1, "y")):
for u in symbols:
if (u, axis) not in locked:
step[u][i] = offset(u, i)
for root, members in groups.items():
if root[1] != axis:
continue
offs = {round(offset(u, i), 6) for u in members}
if len(offs) == 1:
d = offs.pop()
for u in members:
step[u][i] = d
def total(u):
m = move.get(u, [0.0, 0.0])
return [m[0] + step[u][0], m[1] + step[u][1]]
# A blocked member holds its whole group back on that axis.
for u in [u for u in symbols if any(abs(v) > 1e-6 for v in step[u])]:
if not blocked(u, total(u)):
continue
symbols[u]["blocked"] = True
for i, axis in ((0, "x"), (1, "y")):
if (u, axis) in locked and abs(step[u][i]) > 1e-6:
for v in groups.get(find((u, axis)), {u}):
step[v][i] = 0.0
snapped = []
for u, s in symbols.items():
if abs(step[u][0]) > 1e-6 or abs(step[u][1]) > 1e-6:
if blocked(u, total(u)):
s["blocked"] = True
else:
s["blocked"] = False
move[u] = total(u)
snapped.append(s)
elif s.get("blocked"):
snapped.append(s)
def move_op(uuid):
if uuid not in move:
return None
dx, dy = move[uuid]
return {"op": "move_element", "folio": folio, "element": uuid,
"dx": round(dx, 3) + 0.0, "dy": round(dy, 3) + 0.0}
by_wire = {w["uuid"]: w for w in wires}
for f in findings:
if f["rule"] == "avoidable_bend":
w = by_wire[f["conductor"]]
f["fix"] = move_op(w.get("mover"))
if w.get("conflict"):
f["conflict"] = True
for s in snapped:
extra = {"conflict": True} if s.get("blocked") else {}
add("off_grid", element=s["uuid"], label=s["label"], name=s["name"],
x=s["x"], y=s["y"], note=LAYOUT_RULES["off_grid"]["note"],
fix=None if s.get("blocked") else move_op(s["uuid"]), **extra)
dirty_symbols.add(s["uuid"])
# Wires through symbols. A symbol drawn around either end's own symbol
# is a frame (conductorrouter.cpp), not an obstacle.
for w in wires:
own = [symbols[e]["box"] for e in w["ends"] if e in symbols]
for uuid, s in symbols.items():
if (s["annotation"] or uuid in w["ends"]
or any(_contains(s["box"], o) for o in own)):
continue
pts = w["pts"]
if any(_segment_through(p, q, s["box"]) for p, q in zip(pts, pts[1:])):
add("wire_through_symbol", conductor=w["uuid"], element=uuid,
label=s["label"], name=s["name"],
note=LAYOUT_RULES["wire_through_symbol"]["note"],
fix={"op": "route_conductor", "folio": folio, "conductor": w["uuid"]})
dirty_wires.add(w["uuid"])
# A symbol's box is its declared size, rounded up to the grid, so two
# symbols drawn side by side can share up to one grid step of it.
for i, s in enumerate(solid):
for t in solid[i + 1:]:
if _contains(s["box"], t["box"]) or _contains(t["box"], s["box"]):
continue
if _overlap(s["box"], t["box"], margin=LAYOUT_GRID):
add("overlapping_symbols", elements=[s["uuid"], t["uuid"]],
labels=[s["label"], t["label"]], names=[s["name"], t["name"]],
note=LAYOUT_RULES["overlapping_symbols"]["note"])
dirty_symbols.update((s["uuid"], t["uuid"]))
crossings = 0
for i, w in enumerate(wires):
sw = list(zip(w["pts"], w["pts"][1:]))
for v in wires[i + 1:]:
n = sum(1 for a, b in sw for c, d in zip(v["pts"], v["pts"][1:])
if _crosses(a, b, c, d))
if n:
crossings += n
add("crossing", conductors=[w["uuid"], v["uuid"]], count=n,
note=LAYOUT_RULES["crossing"]["note"])
return {"folio": folio, "symbols": len(symbols), "wires": len(wires),
"unread": unread, "findings": findings, "dirty_wires": dirty_wires,
"dirty_symbols": dirty_symbols, "length": length, "crossings": crossings,
"straight": sum(1 for w in wires if w.get("bends") == 0),
"fixes": [move_op(u) for u in move]}
def _layout_answer(folios: list, style: str, limit: int) -> dict:
"""Combine per-folio results into the tool's answer."""
symbols = sum(f["symbols"] for f in folios)
wires = sum(f["wires"] for f in folios)
vertical = sum(f["length"]["vertical"] for f in folios)
horizontal = sum(f["length"]["horizontal"] for f in folios)
total = vertical + horizontal
if style == "auto":
style = "nfpa" if horizontal > vertical else "iec"
findings = [x for f in folios for x in f["findings"]]
order = {"error": 0, "warning": 1, "info": 2}
findings.sort(key=lambda x: (order[x["severity"]], x["folio"], x["rule"]))
count = {r: sum(1 for x in findings if x["rule"] == r) for r in LAYOUT_RULES}
clean_wires = wires - sum(len(f["dirty_wires"]) for f in folios)
clean_symbols = symbols - sum(len(f["dirty_symbols"]) for f in folios)
score = 100.0 * (0.6 * (clean_wires / wires if wires else 1.0)
+ 0.4 * (clean_symbols / symbols if symbols else 1.0))
unread = [u for f in folios for u in f["unread"]]
answer = {
"ok": True,
"style": style,
"score": round(score),
"summary": {
"folios": len(folios), "symbols": symbols, "wires": wires,
"unread_wires": len(unread),
"straight_wires": sum(f["straight"] for f in folios),
"avoidable_bends": count["avoidable_bend"],
"extra_bends": count["extra_bends"],
"wires_through_symbols": count["wire_through_symbol"],
"overlaps": count["overlapping_symbols"],
"off_grid": count["off_grid"],
"crossings": sum(f["crossings"] for f in folios),
"flow": {"vertical": round(vertical / total, 3) if total else 0.0,
"horizontal": round(horizontal / total, 3) if total else 0.0},
},
"findings": findings[:limit],
# One move per symbol, all findings' moves combined: apply them
# together in one qet_edit call.
"fixes": [op for f in folios for op in f["fixes"]],
}
if len(findings) > limit:
answer["truncated"] = len(findings) - limit
if unread:
answer["unread_wires"] = unread[:limit]
answer["note"] = (f"{len(unread)} wire(s) could not be read: both of their "
"terminals carry other wires too, and this QElectroTech build "
"has no conductorPath() to read them by uuid. They are left "
"out of the score.")
return answer
def tool_layout_check(binary: str, project: str, folio: int | None = None,
style: str = "auto", max_shift: float = 40, limit: int = 50,
elements_dir: str | None = None, timeout: int = 180) -> dict:
"""Score how well a project's drawing reads: straight wires, symbols in
line and on the grid, nothing overlapping. Read-only."""
proj = Path(project).expanduser()
if not proj.is_file():
raise ValueError(f"no such project: {proj}")
if style not in LAYOUT_STYLES:
raise ValueError(f"unknown style {style!r}; expected one of {', '.join(LAYOUT_STYLES)}")
if (isinstance(max_shift, bool) or not isinstance(max_shift, (int, float))
or max_shift < 0):
raise ValueError("max_shift must be a number >= 0")
if isinstance(limit, bool) or not isinstance(limit, int) or limit < 0:
raise ValueError("limit must be an integer >= 0")
if folio is not None and (isinstance(folio, bool) or not isinstance(folio, int)
or folio < 1):
raise ValueError("folio is counted from 1, as qet_elements numbers them")
script = (_LAYOUT_JS.replace("@FOLIO@", str(folio - 1 if folio else -1))
.replace("@MARKER@", json.dumps(_MARKER)))
result = _run_qet(binary, [str(proj)], timeout=timeout,
elements_dir=elements_dir, script=script, tail=20_000_000)
folios = []
for line in (result.get("stdout", "") + "\n" + result.get("stderr", "")).splitlines():
idx = line.find(_MARKER)
if idx < 0:
continue
try:
rec = json.loads(line[idx + len(_MARKER):])
except json.JSONDecodeError:
continue
if rec.get("kind") == "layout":
folios.append(_layout_folio(rec, float(max_shift)))
if not folios:
answer = {"ok": False,
"hint": result.get("hint") or (
"no layout came back: the folio does not exist, or this build's "
"scripting API predates the read calls this check needs")}
if result.get("exit_code") is not None:
answer["exit_code"] = result["exit_code"]
return answer
answer = _layout_answer(folios, style, limit)
if result.get("hint"):
answer["ok"] = False
answer["hint"] = result["hint"]
return answer
def tool_project_new(binary: str, output: str, title: str = "Untitled",
folios=1, author: str = "", overwrite: bool = False,
elements_dir: str | None = None, timeout: int = 180) -> dict:
"""Create a new, empty project so qet_edit has something to start from.
Every other edit tool needs an existing .qet, which made building a
schematic from nothing impossible. The obvious candidate,
examples/Projet_vierge.qet, is not blank: it is a 600 KB real project
with 23 elements and 15 conductors.
So this writes the smallest project QElectroTech will open -- one
element with a title, no folios -- and then has QElectroTech itself
add the folios and save. What is left on disk is QElectroTech's own
canonical output, not the hand-written skeleton, which is why this is
not the "write .qet XML directly" route that was rejected: the
skeleton never reaches the result, and the result is checked by
reading it back.
folios is a count, or a list of folio titles.
"""
out = Path(output).expanduser()
if out.exists() and not overwrite:
raise ValueError(f"{out} already exists; pass overwrite=true to replace it")
if isinstance(folios, bool) or not isinstance(folios, (int, list)):
raise ValueError("folios must be a count or a list of titles")
titles = ([""] * folios) if isinstance(folios, int) else [str(t) for t in folios]
if not 0 <= len(titles) <= 200:
raise ValueError("folios must be between 0 and 200")
if not isinstance(title, str) or not title.strip():
raise ValueError("title must be a non-empty string")
from xml.sax.saxutils import quoteattr
script = ["var t = %s;" % json.dumps(titles), "var made = [];",
"for (var i = 0; i < t.length; i++) {",
" var f = qet.addFolio(); made.push(f);",
" if (f >= 0 && t[i]) qet.setFolioTitle(f, t[i]);",
" if (f >= 0 && %s) qet.setFolioProperty(f, 'author', %s);" %
(json.dumps(bool(author)), json.dumps(author)),
"}",
"var saved = qet.save(%s);" % json.dumps(str(out)),
"qet.log(%s + JSON.stringify({kind: 'new', folios: made, saved: saved}));"
% json.dumps(_MARKER)]
out.parent.mkdir(parents=True, exist_ok=True)
with tempfile.TemporaryDirectory(prefix="qet-mcp-new-") as tmp:
skeleton = Path(tmp) / "skeleton.qet"
skeleton.write_text('\n\n'
% quoteattr(title), encoding="utf-8")
result = _run_qet(binary, [str(skeleton)], timeout=timeout,
elements_dir=elements_dir, script="\n".join(script), tail=200_000)
rec = None
for line in (result.get("stdout", "") + "\n" + result.get("stderr", "")).splitlines():
idx = line.find(_MARKER)
if idx >= 0:
try:
r = json.loads(line[idx + len(_MARKER):])
except json.JSONDecodeError:
continue
if r.get("kind") == "new":
rec = r
for key in ("stdout", "stderr"):
result[key] = "\n".join(l for l in result.get(key, "").splitlines()
if _MARKER not in l)[-2000:]
if rec is None or not rec.get("saved") or not out.is_file():
result["ok"] = False
#setdefault: _run_qet() may already have said something more
#specific than this guess -- notably that scripting is switched
#off, in which case "your build is too old" sends the reader
#looking for the wrong thing entirely.
result.setdefault("hint",
"QElectroTech did not write the project; this build's scripting "
"API may predate addFolio()/save()")
return result
if any(f < 0 for f in rec["folios"]):
result["ok"] = False
result["hint"] = "a folio could not be added"
return result
# Read back what is actually on disk rather than report what was asked for.
info = tool_project_info(str(out))
if info["title"] != title or info["folio_count"] != len(titles):
result["ok"] = False
result["hint"] = (f"the file on disk has title {info['title']!r} and "
f"{info['folio_count']} folio(s), not what was requested")
result["output"] = str(out)
result["project"] = info
return result
def _wrong_folio(project: str, op: dict) -> str:
"""Explain a failed op that named an element on the wrong folio.
qet_elements and qet_project_info number folios from 1, as the UI does;
qet_edit passes "folio" straight to the scripting API, which counts from
0. Passing the number qet_elements showed therefore addresses the next
folio, and the op fails with nothing but "false". When the element the
op names is in the project on some other folio, say which index to use.
"""
folio = op.get("folio")
spec = OPS.get(op.get("op"), (None, []))[1]
uuids = [op[key] for key, kind in spec
if kind == "elmt" and isinstance(op.get(key), str)
and not op[key].startswith("$")]
if not isinstance(folio, int) or not uuids:
return ""
try:
where = {el.get("uuid"): i for i, el in _elements(_root(project))}
except (OSError, ET.ParseError):
return ""
for uuid in uuids:
number = where.get(uuid)
if number is not None and number - 1 != folio:
return (f"Element {uuid} is on folio {number} as qet_elements numbers "
f"it, which is \"folio\": {number - 1} here: qet_edit counts "
"folios from 0.")
return ""
def tool_edit(binary: str, project: str, operations: list, output: str,
elements_dir: str | None = None, timeout: int = 180) -> dict:
"""Apply edits through the scripting API and report what actually changed.
The point is the last part. The scripting API returns a bool per call,
which says the call was accepted, not that the file came out the way
anyone intended -- so this runs qet_diff between the input project and
the saved result and puts that in the answer. A caller that trusts
"addConductor -> true" and stops there is back to trusting the
screenshot.
The project is never written in place: output is a separate file, and
the original is what the diff is taken against.
"""
proj = Path(project).expanduser()
if not proj.is_file():
raise ValueError(f"no such project: {proj}")
if not isinstance(operations, list) or not operations:
raise ValueError("operations must be a non-empty list")
out = Path(output).expanduser()
if out.resolve() == proj.resolve():
raise ValueError("output must differ from project; this tool does not "
"edit a project in place")
script = _build_script(operations, str(out))
# QET interrupts a script at 30 s (kScriptTimeoutMs in qetscripting.cpp),
# independently of this timeout. Leaving room above it means a script
# that hits the engine's limit comes back as a script error we can
# report, rather than as our own opaque process timeout.
result = _run_qet(binary, [str(proj)], timeout=timeout,
elements_dir=elements_dir, script=script, tail=200_000)
streams = result.get("stdout", "") + "\n" + result.get("stderr", "")
result.update(_parse_script_output(streams))
# The marker lines have been parsed into "operations"; leaving them in
# the reported streams as well just doubles the size of the answer.
for key in ("stdout", "stderr"):
kept = [ln for ln in result.get(key, "").splitlines() if _MARKER not in ln]
result[key] = "\n".join(kept)[-4000:]
result["output"] = str(out)
result["output_exists"] = out.exists()
if result.get("missing_methods") is None and not result.get("timed_out"):
# The script's first act is to report which methods exist. No report
# means the script never ran -- a binary with no --run support, one
# that exited early, or the wrong executable -- and exit code 0 from
# something that did nothing is not success.
result["ok"] = False
#setdefault, for the same reason as in tool_project_new(): a
#refusal to run scripts at all also produces no capability
#report, and "is it a build with --run support?" is then the
#wrong question.
result.setdefault("hint",
"the binary never ran the script (no capability report came "
"back), so nothing was changed. Is it a QElectroTech build with "
"--run support?")
result["script"] = script
return result
missing = result.get("missing_methods")
if missing:
result["ok"] = False
result["hint"] = (
"this build's scripting API lacks " + ", ".join(missing) +
" -- it predates the drawing verbs, so nothing was changed")
result["script"] = script
return result
for record in result.get("operations", []):
if not record["succeeded"]:
result["ok"] = False
hint = (f"operation {record['index']} ({record['op']}) returned "
f"{record['result']!r}; later operations were skipped. "
"qet.log lines in stderr/stdout say why.")
if 0 <= record["index"] < len(operations):
wrong = _wrong_folio(str(proj), operations[record["index"]])
if wrong:
hint += " " + wrong
result.setdefault("hint", hint)
break
if result.get("saved") is False:
result["ok"] = False
result.setdefault("hint", "the edits were made but save() failed")
if out.is_file():
result["output_bytes"] = out.stat().st_size
try:
result["diff"] = tool_diff(str(proj), str(out))
except ET.ParseError as exc: # a truncated or unwritten save
result["ok"] = False
result["diff_error"] = str(exc)
if not result.get("ok"):
result["script"] = script
return result
# --------------------------------------------------------------------------
# Stored scripts: the buttons in Projet > Scripts and on the Scripts toolbar
# --------------------------------------------------------------------------
#
# QElectroTech turns every .js file in one folder into a command with an
# icon, read from a // ==QETScript== header at the top of the file. The
# folder is the whole contract: a person writing a script by hand and an
# assistant using these tools both end with a file there, and QElectroTech
# notices it without a restart. So installing is writing a file, and these
# tools never talk to a running QElectroTech.
#
# The folder is not the client's to choose -- scripts_dir() finds it the
# way QElectroTech does -- and writing to it needs the same consent as
# editing a project: QET_ENABLE_SCRIPTING=1 in this server's environment.
# A stored script runs, with the user's rights, when they click its button.
_SCRIPT_CONTEXTS = ("canvas", "selection", "conductor")
_SCRIPT_ID = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$")
_SCRIPT_MAX_BYTES = 256 * 1024
_ICON_MAX_BYTES = 64 * 1024
SCRIPT_HEADER_HELP = (
"A stored script is one .js file; its first lines say how its button "
"looks:\n"
"// ==QETScript==\n"
"// @name Add revision note (required)\n"
"// @icon .svg (optional: a file next to the "
"script, or builtin:; a tile with the initials otherwise)\n"
"// @tooltip Puts a note on the folio on screen\n"
"// @shortcut Ctrl+Alt+R (optional default shortcut)\n"
"// @context canvas (canvas: always enabled; "
"selection: something selected; conductor: a conductor selected)\n"
"// @api 1\n"
"// ==/QETScript==\n"
"The script sees one global, qet. qet.currentFolio() is the folio on "
"screen; folio indexes count from 0. A click is one undo step, so do "
"not call qet.undo() in a stored script. Report with qet.log(); "
"qet.showMessage() opens a dialog the user has to close.")
def default_scripts_dir(os_name: str, platform: str, env, home) -> PurePath:
"""QETApp::dataDir() + "/scripts" for a given platform.
dataDir() is Qt's AppDataLocation with organisation and application
both "QElectroTech" (main.cpp). Pure, so the Windows and macOS answers
are tested on any machine.
"""
if os_name == "nt":
appdata = env.get("APPDATA")
base = (PureWindowsPath(appdata) if appdata
else PureWindowsPath(home, "AppData", "Roaming"))
elif platform == "darwin":
base = PurePosixPath(home, "Library", "Application Support")
else:
base = PurePosixPath(env.get("XDG_DATA_HOME") or PurePosixPath(home, ".local", "share"))
return base / "QElectroTech" / "QElectroTech" / "scripts"
def assistant_info_file() -> Path:
"""Where QElectroTech writes qet-assistant.json.
QET_MCP_INFO_FILE if set; next to a QET_MCP_SCRIPTS_DIR if that is set
(one folder up, as in QElectroTech's own layout); else the platform's
standard data folder, where QElectroTech always writes it, even when
--data-dir moves the rest.
"""
explicit = os.environ.get("QET_MCP_INFO_FILE", "").strip()
if explicit:
return Path(explicit).expanduser()
scripts = os.environ.get("QET_MCP_SCRIPTS_DIR", "").strip()
if scripts:
return Path(scripts).expanduser().parent / "qet-assistant.json"
default = default_scripts_dir(os.name, sys.platform, os.environ, str(Path.home()))
return Path(str(default.parent)) / "qet-assistant.json"
def assistant_info() -> dict | None:
"""qet-assistant.json as QElectroTech last wrote it, or None.
QElectroTech rewrites it whenever an editor opens and whenever its
stored scripts, settings or live channel change: its folders, features,
the calls a script can make, the stored scripts, and the live channel
while one is open.
"""
path = assistant_info_file()
try:
return json.loads(path.read_text(encoding="utf-8"))
except (OSError, ValueError):
return None
def scripts_dir() -> Path:
"""The folder QElectroTech reads stored scripts from.
QET_MCP_SCRIPTS_DIR first (a test, or a deliberate choice); then the
folder qet-assistant.json names; else the platform's default.
"""
env = os.environ.get("QET_MCP_SCRIPTS_DIR", "").strip()
if env:
return Path(env).expanduser()
info = assistant_info() or {}
named = (info.get("folders") or {}).get("scripts")
if isinstance(named, str) and named:
return Path(named)
return Path(str(default_scripts_dir(os.name, sys.platform, os.environ, str(Path.home()))))
SERVER_INSTRUCTIONS = (
"This server works with QElectroTech, a free editor for electrical "
"diagrams. A project (.qet) holds folios (sheets) of symbols "
"(elements) joined by wires (conductors). Folio indexes count from 0.\n"
"Start with qet_about: where QElectroTech keeps things, what is "
"switched on, and the stored scripts.\n"
"Two ways of working. HEADLESS (qet_* and qet_script_*): read, check "
"and edit .qet files and store script buttons; nothing the user has "
"open is touched. To make a button: qet_script_api for the calls, "
"qet_script_test on a copy until the diff is right, then "
"qet_script_install with test_project set. LIVE (qet_live_*): act on "
"the project open in the user's QElectroTech while they watch; only "
"when they switched live mode on and accepted its warning at this "
"start, each action one undo step, scripts written on the spot shown "
"to them first.\n"
"MACRO RECORDINGS: the user records something by hand in QElectroTech "
"and pastes you a request naming it; qet_recording_read it, write a "
"general script, qet_recording_check it until it matches, then "
"qet_script_install it.\n"
"Verify edits by reading the result (qet_diff, qet_elements), not by "
"assuming them. After drawing, run qet_layout_check and apply its fixes: "
"a wire is straight only when its two terminals are exactly in line.")
def tool_about() -> dict:
"""What this server and the QElectroTech it works with look like now."""
info = assistant_info()
binary = resolve_binary()
out = {
"server": {
"version": SERVER_VERSION,
"qelectrotech_binary": str(binary) if binary else None,
"workspace": [str(r) for r in workspace_roots()] or "any path (QET_MCP_ALLOW_ANY_PATH=1)",
"scripting_allowed_here": os.environ.get("QET_ENABLE_SCRIPTING") == "1",
},
"info_file": str(assistant_info_file()),
"scripts_folder": str(scripts_dir()),
}
if info is None:
out["found"] = False
out["note"] = ("QElectroTech writes this file when an editor window opens; "
"start it once (a version with script buttons) to fill it in. "
"Until then folders are this server's own guess.")
return out
live = info.get("live")
out.update({
"found": True,
"qelectrotech": {k: info.get(k) for k in
("qelectrotech_version", "program", "running", "written")},
"folders": info.get("folders"),
"features": info.get("features"),
"stored_scripts": info.get("stored_scripts"),
"refused_scripts": info.get("refused_scripts"),
"script_api": info.get("script_api"),
# Never the token: it is for the live tools, not the conversation.
"live": {"open": bool(live), "pid": (live or {}).get("pid")},
})
return out
def parse_script_header(text: str, script_id: str) -> dict:
"""The same rules as QElectroTech's ScriptHeader::parse().
Returns the header's fields, with "error" set when QElectroTech would
refuse it (and so show no button for it).
"""
h = {"id": script_id, "name": "", "icon": "", "tooltip": "", "shortcut": "",
"context": "canvas", "api": 1,
"action_id": "diagrameditor.script." + script_id}
m = re.search(r"//\s*==QETScript==\s*\n(.*?)//\s*==/QETScript==", text, re.S)
if not m:
h["error"] = "no // ==QETScript== header"
return h
for line in m.group(1).split("\n"):
lm = re.match(r"^\s*//\s*@(\w+)\s+(.*?)\s*$", line)
if not lm:
continue
key, value = lm.groups()
if key == "api":
try:
h["api"] = int(value)
except ValueError:
h["api"] = 0
elif key in ("name", "icon", "tooltip", "shortcut", "context"):
h[key] = value
else:
h["error"] = f"unknown header key @{key}"
return h
if not h["name"]:
h["error"] = "@name is required"
elif h["context"] not in _SCRIPT_CONTEXTS:
h["error"] = "@context must be one of: " + ", ".join(_SCRIPT_CONTEXTS)
elif h["api"] != 1:
h["error"] = f"@api {h['api']} is not supported by this version (1 is)"
return h
def _script_id(script_id) -> str:
if not isinstance(script_id, str) or not _SCRIPT_ID.match(script_id):
raise ValueError("'id' must be 1-64 characters of a-z, 0-9, '-' and '_', "
"starting with a letter or digit: it is the file name")
return script_id
def _require_script_consent() -> None:
if os.environ.get("QET_ENABLE_SCRIPTING") != "1":
raise ValueError(
"storing a script needs the same consent as editing a project: "
"QET_ENABLE_SCRIPTING=1 in the environment this server is started "
"in. A stored script runs with the user's rights when they click it.")
def _check_icon_svg(svg: str) -> None:
if not isinstance(svg, str) or len(svg.encode("utf-8")) > _ICON_MAX_BYTES:
raise ValueError(f"'icon_svg' must be SVG text under {_ICON_MAX_BYTES // 1024} KB")
try:
root = ET.fromstring(svg)
except ET.ParseError as exc:
raise ValueError(f"'icon_svg' is not well-formed XML: {exc}") from None
if root.tag.rsplit("}", 1)[-1] != "svg":
raise ValueError("'icon_svg' must have