diff --git a/misc/qet-mcp/README.md b/misc/qet-mcp/README.md index 98586dc52..1de4752d1 100644 --- a/misc/qet-mcp/README.md +++ b/misc/qet-mcp/README.md @@ -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 | diff --git a/misc/qet-mcp/qet_mcp.py b/misc/qet-mcp/qet_mcp.py index f1157ea3c..58177e01a 100755 --- a/misc/qet-mcp/qet_mcp.py +++ b/misc/qet-mcp/qet_mcp.py @@ -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 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 carry a + """Sweep every .qet and .qetz in a directory, counting how many 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 " diff --git a/misc/qet-mcp/test_qet_mcp.py b/misc/qet-mcp/test_qet_mcp.py index ea85bf798..30492aa4b 100644 --- a/misc/qet-mcp/test_qet_mcp.py +++ b/misc/qet-mcp/test_qet_mcp.py @@ -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)'}")