diff --git a/misc/qet-mcp/qet_mcp.py b/misc/qet-mcp/qet_mcp.py index 58177e01a..7de720d62 100755 --- a/misc/qet-mcp/qet_mcp.py +++ b/misc/qet-mcp/qet_mcp.py @@ -56,6 +56,7 @@ workspace policy applies exactly as it does over stdio. from __future__ import annotations +import atexit import datetime import json import re @@ -110,7 +111,21 @@ def _root(path: str) -> ET.Element: # 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). +# Only the newest few are kept, and the server removes them when it exits. _QETZ_COPIES: dict = {} +_QETZ_COPIES_MAX = 8 + + +def _forget_qetz_copy(key) -> None: + out = _QETZ_COPIES.pop(key, None) + if out is not None: + shutil.rmtree(out.parent, ignore_errors=True) + + +@atexit.register +def _forget_qetz_copies() -> None: + for key in list(_QETZ_COPIES): + _forget_qetz_copy(key) def _is_qetz(path) -> bool: @@ -129,6 +144,7 @@ def _as_qet(path) -> Path: 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(): + _QETZ_COPIES[key] = _QETZ_COPIES.pop(key) # newest last return cached binary = resolve_binary() if not binary: @@ -138,12 +154,21 @@ def _as_qet(path) -> Path: 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: + # A run that timed out or failed may have left part of a .qet behind + if not result.get("ok") or 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:] + if result.get("timed_out"): + detail = f"timed out after {result.get('timeout_s')} s" + else: + 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 "")) + # An older version of the same file is not read again + for old in [k for k in _QETZ_COPIES if k[0] == key[0]]: + _forget_qetz_copy(old) _QETZ_COPIES[key] = out + while len(_QETZ_COPIES) > _QETZ_COPIES_MAX: + _forget_qetz_copy(next(iter(_QETZ_COPIES))) return out @@ -840,12 +865,17 @@ def _num(v) -> float: def tool_scan(directory: str, tag: str = "conductor", - attribute: str = "cable", recursive: bool = True) -> dict: + attribute: str = "cable", recursive: bool = True, + max_qetz: int = 20) -> dict: """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 25 projects" is exactly one call to this tool. + + Each .qetz is read by QElectroTech saving it as a .qet, which can take + minutes for a big project, so only the first `max_qetz` are read; the + others are listed as skipped. """ d = Path(directory).expanduser() if not d.is_dir(): @@ -853,8 +883,14 @@ def tool_scan(directory: str, tag: str = "conductor", 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 = [], {}, [] + per_file, values, unreadable, skipped = [], {}, [], [] + qetz_read = 0 for f in files: + if _is_qetz(f): + if qetz_read >= max_qetz: + skipped.append(f.name) + continue + qetz_read += 1 try: root = ET.parse(_as_qet(f)).getroot() except (ET.ParseError, ValueError) as exc: @@ -872,6 +908,9 @@ def tool_scan(directory: str, tag: str = "conductor", per_file.append({"file": f.name, tag: n, "non_empty": ne}) return { "directory": str(d), "files": len(files), "unreadable": unreadable, + **({"skipped_qetz": skipped, + "note": f"only the first {max_qetz} .qetz were read (max_qetz)"} + if skipped else {}), "tag": tag, "attribute": attribute, "total": total, "non_empty": non_empty, "distinct_values": sorted(values.items(), key=lambda kv: -kv[1])[:25], @@ -4902,10 +4941,12 @@ TOOLS = [ }, { "name": "qet_scan", - "description": "Sweep every .qet in a directory and count how many nodes of " - "a given tag carry a non-empty attribute, with the distinct " - "values found. For corpus questions such as how many " - "conductors in the shipped examples have a cable value.", + "description": "Sweep every .qet and .qetz in a directory and count how many " + "nodes of a given tag carry a non-empty attribute, with the " + "distinct values found. For corpus questions such as how many " + "conductors in the shipped examples have a cable value. Each " + ".qetz is read through QElectroTech, which can take minutes " + "for a big project: only the first max_qetz are read.", "inputSchema": { "type": "object", "properties": { @@ -4913,12 +4954,14 @@ TOOLS = [ "tag": {"type": "string", "default": "conductor"}, "attribute": {"type": "string", "default": "cable"}, "recursive": {"type": "boolean", "default": True}, + "max_qetz": {"type": "integer", "default": 20}, }, "required": ["directory"], }, "handler": lambda a: tool_scan(a["directory"], a.get("tag", "conductor"), a.get("attribute", "cable"), - a.get("recursive", True)), + a.get("recursive", True), + a.get("max_qetz", 20)), }, { "name": "qet_element_info", diff --git a/misc/qet-mcp/test_qet_mcp.py b/misc/qet-mcp/test_qet_mcp.py index 30492aa4b..3ada780c3 100644 --- a/misc/qet-mcp/test_qet_mcp.py +++ b/misc/qet-mcp/test_qet_mcp.py @@ -6489,6 +6489,66 @@ class ZippedProjectWithoutBinary(unittest.TestCase): m.tool_project_info(str(z)) self.assertIn("QET_BINARY", str(caught.exception)) + QET = '' + + def fake_resave(self, ok=True, timed_out=False): + """A _run_qet that writes a small .qet where --resave was asked to""" + def run(binary, args, timeout=180, **kw): + Path(args[2]).write_text(self.QET) + if timed_out: + return {"ok": False, "timed_out": True, "timeout_s": timeout} + return {"ok": ok, "exit_code": 0 if ok else 1, "stderr": "", "stdout": ""} + return run + + def qetz_files(self, tmp, n): + files = [] + for i in range(n): + z = Path(tmp) / f"p{i}.qetz" + z.write_bytes(b"PK") + files.append(z) + return files + + def test_a_failed_or_timed_out_conversion_is_not_kept(self): + """A .qet left behind by a run that did not finish is not read""" + for kind in ({"ok": False}, {"timed_out": True}): + with self.subTest(**kind), tempfile.TemporaryDirectory() as tmp: + m._forget_qetz_copies() + z, = self.qetz_files(tmp, 1) + with mock.patch.object(m, "resolve_binary", return_value="qet"), \ + mock.patch.object(m, "_run_qet", side_effect=self.fake_resave(**kind)): + with self.assertRaises(ValueError): + m._as_qet(z) + self.assertEqual(m._QETZ_COPIES, {}) + + def test_only_the_newest_conversions_are_kept_on_disk(self): + with tempfile.TemporaryDirectory() as tmp: + m._forget_qetz_copies() + files = self.qetz_files(tmp, m._QETZ_COPIES_MAX + 2) + with mock.patch.object(m, "resolve_binary", return_value="qet"), \ + mock.patch.object(m, "_run_qet", side_effect=self.fake_resave()): + copies = [m._as_qet(z) for z in files] + self.assertEqual(len(m._QETZ_COPIES), m._QETZ_COPIES_MAX) + self.assertFalse(copies[0].parent.exists()) + self.assertTrue(copies[-1].is_file()) + # A new version of a file replaces its old copy + os.utime(files[-1], ns=(1, 1)) + newer = m._as_qet(files[-1]) + self.assertFalse(copies[-1].parent.exists()) + m._forget_qetz_copies() + self.assertFalse(newer.parent.exists()) + + def test_scan_reads_at_most_max_qetz(self): + with tempfile.TemporaryDirectory() as tmp: + m._forget_qetz_copies() + self.qetz_files(tmp, 3) + with mock.patch.object(m, "resolve_binary", return_value="qet"), \ + mock.patch.object(m, "_run_qet", side_effect=self.fake_resave()) as run: + scan = m.tool_scan(tmp, max_qetz=2) + m._forget_qetz_copies() + self.assertEqual(run.call_count, 2) + self.assertEqual((scan["total"], scan["non_empty"]), (2, 2)) + self.assertEqual(scan["skipped_qetz"], ["p2.qetz"]) + @needs_examples class ZippedProjects(unittest.TestCase): @@ -6561,8 +6621,9 @@ class ZippedProjects(unittest.TestCase): 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()) + rotated = m.tool_diff(str(self.qetz), str(out))["elements"]["rotated"] + self.assertEqual([r["uuid"] for r in rotated], [first["uuid"]]) + self.assertNotEqual(*rotated[0]["orientation"]) if __name__ == "__main__":