kotschak/qelectrotech-source-mirror
0
0
Fork 0
mirror of https://github.com/qelectrotech/qelectrotech-source-mirror.git synced 2026-10-14 17:44:13 +02:00
Code Issues Actions Packages Projects Releases Wiki Activity

Read and write .qetz in the qet MCP server

Browse Source 
Every tool now takes a .qetz as it takes a .qet. The tools that read a
project's XML read the .qet the configured QElectroTech saves from the
.qetz, made once per version of the file (path, size, modification
time); without a QElectroTech, they say to set QET_BINARY. The tools that
launch QElectroTech hand it the .qetz, and write one when the output path
ends in .qetz. qet_scan counts .qetz files too. The server's description
and README say so.

Reading through QElectroTech instead of the file's own layout keeps the
server right while the zipped format still changes; the first read of a
big .qetz takes as long as QElectroTech needs to save it.

test_qet_mcp.py: the read tools answer the same for a .qetz and its .qet,
qet_diff finds nothing between them, qet_scan counts both, a .qetz is
converted once, qet_query reads it, qet_edit writes one (four fail when
_root() parses the file itself); without QElectroTech the answer names
QET_BINARY. Whole suite: 430 tests pass with QET_ENABLE_SCRIPTING=1.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
ispyisail
2026-10-12 21:46:37 +13:00
parent a6d81a21a8
commit b8390c853c
3 changed files with 151 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
misc/qet-mcp/README.md
+7
View File
@@ -25,6 +25,13 @@ here read the model.
## Tools
Every tool takes a `.qet` or a `.qetz` (the same project, zipped). The
tools that read a project's XML read the `.qet` QElectroTech saves from a
`.qetz`, made once per version of the file with the configured
QElectroTech (`QET_BINARY`); the first read of a big `.qetz` takes as long
as QElectroTech needs to save it. The tools that launch QElectroTech hand
it the `.qetz` itself, and write one when the output path ends in `.qetz`.
| Tool | What it answers |
|---|---|
| `qet_project_info` | title, format version, folios with their uuids, element and conductor counts |
misc/qet-mcp/qet_mcp.py
+55 -7
View File
@@ -97,11 +97,56 @@ def _root(path: str) -> ET.Element:
if not p.is_file():
raise ValueError(f"no such file: {p}")
try:
return ET.parse(p).getroot()
return ET.parse(_as_qet(p)).getroot()
except ET.ParseError as exc:
raise ValueError(f"{p.name} is not parseable XML: {exc}") from exc
# --------------------------------------------------------------------------
# zipped projects (.qetz)
# --------------------------------------------------------------------------
# A .qetz is QElectroTech's zipped project: a database and one file per
# folio. The tools that read a project's XML read the .qet QElectroTech
# saves from it, so they read what QElectroTech reads, whatever the zipped
# format holds; one copy per version of the file (path, size, time).
_QETZ_COPIES: dict = {}
def _is_qetz(path) -> bool:
return str(path).lower().endswith(".qetz")
def _as_qet(path) -> Path:
"""The file the XML tools read for `path`: itself, or for a .qetz the
.qet QElectroTech saves from it (made with the configured binary)."""
p = Path(path).expanduser()
if not _is_qetz(p):
return p
if not p.is_file():
raise ValueError(f"no such file: {p}")
st = p.stat()
key = (str(p.resolve()), st.st_size, st.st_mtime_ns)
cached = _QETZ_COPIES.get(key)
if cached is not None and cached.is_file():
return cached
binary = resolve_binary()
if not binary:
raise ValueError(f"{p.name} is a zipped project (.qetz): reading it needs "
"QElectroTech; set QET_BINARY")
folder = Path(tempfile.mkdtemp(prefix="qet-mcp-qetz-"))
out = folder / (p.stem + ".qet")
# Saving as .qet builds every folio: allow a big project its time
result = _run_qet(str(binary), ["--resave", str(p), str(out)], timeout=900)
if not out.is_file() or out.stat().st_size == 0:
shutil.rmtree(folder, ignore_errors=True)
detail = (result.get("stderr") or result.get("stdout") or "").strip()[-400:]
raise ValueError(f"QElectroTech could not read {p.name}"
+ (f": {detail}" if detail else ""))
_QETZ_COPIES[key] = out
return out
def _element_info(el: ET.Element) -> dict:
"""The <elementInformations> bag, as a plain dict."""
out = {}
@@ -796,7 +841,7 @@ def _num(v) -> float:
def tool_scan(directory: str, tag: str = "conductor",
attribute: str = "cable", recursive: bool = True) -> dict:
"""Sweep every .qet in a directory, counting how many <tag> carry a
"""Sweep every .qet and .qetz in a directory, counting how many <tag> carry a
non-empty `attribute`.
This is the corpus question: "3190 conductors, 0 cable values across
@@ -805,13 +850,14 @@ def tool_scan(directory: str, tag: str = "conductor",
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"))
files = sorted([f for pattern in ("*.qet", "*.qetz")
for f in (d.rglob(pattern) if recursive else d.glob(pattern))])
total = non_empty = 0
per_file, values, unreadable = [], {}, []
for f in files:
try:
root = ET.parse(f).getroot()
except ET.ParseError as exc:
root = ET.parse(_as_qet(f)).getroot()
except (ET.ParseError, ValueError) as exc:
unreadable.append({"file": f.name, "error": str(exc)})
continue
n = ne = 0
@@ -3919,8 +3965,10 @@ def scripts_dir() -> Path:
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"
"diagrams. A project (.qet, or .qetz: the same project zipped) holds "
"folios (sheets) of symbols (elements) joined by wires (conductors). "
"Every tool takes either; a .qetz is read through QElectroTech, so the "
"first read of a big one takes a while. Folio indexes count from 0.\n"
"Start with qet_about: where QElectroTech keeps things, what is "
"switched on, the stored scripts, and house_style -- this "
"installation's own drawing conventions if the user set any. Follow "
misc/qet-mcp/test_qet_mcp.py
+89
View File
@@ -6476,6 +6476,95 @@ class CorpusIntegration(unittest.TestCase):
sb.close()
class ZippedProjectWithoutBinary(unittest.TestCase):
def test_reading_a_qetz_without_qelectrotech_says_what_to_set(self):
"""A .qetz is read through QElectroTech (m._as_qet()): with none
configured, the answer names QET_BINARY rather than failing to
parse a zip as XML."""
with tempfile.TemporaryDirectory() as tmp:
z = Path(tmp) / "p.qetz"
z.write_bytes(b"PK\x03\x04 not read")
with mock.patch.object(m, "resolve_binary", return_value=None):
with self.assertRaises(ValueError) as caught:
m.tool_project_info(str(z))
self.assertIn("QET_BINARY", str(caught.exception))
@needs_examples
class ZippedProjects(unittest.TestCase):
"""Every tool reads and writes QElectroTech's zipped project (.qetz) as
it does the .qet: the XML tools read the .qet QElectroTech saves from
it, the tools that launch QElectroTech hand it the .qetz."""
@classmethod
def setUpClass(cls):
cls.tmp = tempfile.TemporaryDirectory()
root = Path(cls.tmp.name)
cls.env = mock.patch.dict(os.environ, {"QET_BINARY": BINARY, "QET_ENABLE_SCRIPTING": "1",
"QET_MCP_SCRIPTS_DIR": str(root / "scripts"),
"QET_MCP_INFO_FILE": str(root / "none.json")})
cls.env.start()
cls.qet, cls.qetz = root / "p.qet", root / "p.qetz"
source = str(Path(EXAMPLES) / "industrial.qet")
for out in (cls.qet, cls.qetz):
r = m._run_qet(BINARY, ["--resave", source, str(out)], timeout=600)
assert out.is_file(), r
assert cls.qetz.read_bytes()[:2] == b"PK"
@classmethod
def tearDownClass(cls):
cls.env.stop()
cls.tmp.cleanup()
def test_read_tools_answer_as_for_the_qet(self):
def strip(d):
return {k: v for k, v in d.items() if k not in ("path", "file")}
for tool in (m.tool_project_info, m.tool_elements, m.tool_conductors, m.tool_items):
self.assertEqual(strip(tool(str(self.qetz))), strip(tool(str(self.qet))), tool.__name__)
def test_diff_finds_nothing_between_the_qet_and_the_qetz(self):
d = m.tool_diff(str(self.qet), str(self.qetz))
self.assertTrue(d.get("identical") or not any(
v.get("added") or v.get("removed") or v.get("changed")
for v in d.values() if isinstance(v, dict)), d)
def test_scan_counts_a_qetz(self):
folder = Path(self.tmp.name) / "scan"
folder.mkdir(exist_ok=True)
for f in (self.qet, self.qetz):
shutil.copy2(f, folder / f.name)
scan = m.tool_scan(str(folder), "conductor", "cable", recursive=False)
self.assertEqual(scan["files"], 2)
self.assertEqual(scan["unreadable"], [])
self.assertEqual(scan["per_file"][0]["conductor"], scan["per_file"][1]["conductor"])
def test_a_qetz_is_converted_once_while_it_does_not_change(self):
m._QETZ_COPIES.clear()
real = m._run_qet
with mock.patch.object(m, "_run_qet", side_effect=real) as run:
m.tool_project_info(str(self.qetz))
m.tool_elements(str(self.qetz))
self.assertEqual(run.call_count, 1)
def test_query_reads_the_qetz_itself(self):
sql = "SELECT COUNT(*) AS n FROM element"
self.assertEqual(m.tool_query(BINARY, str(self.qetz), sql)["rows"],
m.tool_query(BINARY, str(self.qet), sql)["rows"])
def test_edit_writes_a_qetz(self):
first = m.tool_elements(str(self.qetz))["elements"][0]
out = Path(self.tmp.name) / "edited.qetz"
r = m.tool_edit(BINARY, str(self.qetz),
[{"op": "rotate_element", "folio": first["folio"] - 1,
"element": first["uuid"], "angle": 90}],
str(out), ELEMENTS or None)
self.assertTrue(r.get("ok"), r)
self.assertEqual(out.read_bytes()[:2], b"PK")
# The rotation is what the diff of the two .qetz finds
diff = json.dumps(m.tool_diff(str(self.qetz), str(out)))
self.assertIn(m._uuid_key(first["uuid"]) if hasattr(m, "_uuid_key") else first["uuid"], diff.lower())
if __name__ == "__main__":
print(f"binary : {BINARY or '(none: integration tests skipped)'}")
print(f"elements : {ELEMENTS or '(none)'}")