From a1e0badb63eac261d1fb60f600da11ffdd03f473 Mon Sep 17 00:00:00 2001 From: Blaise deB Frederick Date: Fri, 28 Aug 2026 15:27:58 -0400 Subject: [PATCH 1/2] Create scans in an archive, and check the result against reality A scanner accepts archives built this way: four added scans loaded, kept their running order, and the one carrying an edit kept its own protocol -- 19 distinct protocols in and 19 out. Generation is not blocked by the format. Getting there cost two defects, and both are the reason validate.py exists rather than a comment. The first was rejected outright. EdfProgramContent describes the running order in five maps keyed by step id -- LinksFrom, LinksTo, Ranks, RelationsFrom, RelationsTo -- and the first attempt updated one. A step present in only some leaves the console unable to build the program, which it reports by showing the folder tree with no protocols in it. That is indistinguishable from an archive exported off an empty folder node, which is how it was first misread. The second loaded and was silently wrong. A protocol and a label each carry ParentElementId pointing at their own step, and the console resolves a step's protocol through that reverse link rather than through the step's Children blob. Copies that kept the source's pointer were served the source's protocol. Three of the four duplicates looked perfect, because they duplicate content identical to their source by design; only the copy carrying a deliberate edit could show it, returning its source's TR. A generated object must differ from its template in some observable way or the test cannot tell "created correctly" from "aliased to the original". Both archives were internally coherent throughout: every reference resolved, every id was unique. So the checks in validate.py are stated as relationships that hold in every console-authored archive and re-derived from the archive under test, not as self-consistency. Four negative tests reintroduce each defect and require it to be reported, because a checker that never fires carries the appearance of assurance without the fact. Writing those found a flaw in the checker itself: "does the chain cover every step" compared the chain against archive.steps, which is derived by walking the chain. A chain stopping early agreed with itself. It now counts EdfMeasurementStep instances independently. A validator built from the same model as the thing it validates inherits its blind spots, which is how the original defects got past. Also: Instance.Tags on a protocol carries #ContentHash|, which is neither the ContentHash column nor the stored blob's hash, and matches all 18 protocols in the reference archive. replace_content recomputes it. A stale one is tolerated on load -- two scans in the NAV option scan shared one and both kept their values -- but it is derivable and describes the content. Co-Authored-By: Claude Opus 5 --- CLAUDE.md | 30 +++ src/siemens_protocol/exar/__init__.py | 4 + src/siemens_protocol/exar/archive.py | 34 +++ src/siemens_protocol/exar/generate.py | 319 ++++++++++++++++++++++++++ src/siemens_protocol/exar/validate.py | 234 +++++++++++++++++++ tests/test_exar_generate.py | 205 +++++++++++++++++ 6 files changed, 826 insertions(+) create mode 100644 src/siemens_protocol/exar/generate.py create mode 100644 src/siemens_protocol/exar/validate.py create mode 100644 tests/test_exar_generate.py diff --git a/CLAUDE.md b/CLAUDE.md index d47d019..252f1a4 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -458,6 +458,36 @@ before/after pair is the only way to learn where a printed value is stored -- files then differ from what we wrote in exactly one respect: the console recompresses (its DEFLATE is tighter -- 770 KB against our 934 KB) and otherwise the ASCCONV text is identical field for field, churn included. +- **A scan can be added to an archive, and the scanner accepts it.** Four + duplicated scans loaded, kept their running order, and one carrying an edit + kept its own protocol -- 19 distinct protocols in and 19 out. So generation + is not blocked by the format. Adding one scan means three new instances + (step, protocol, label) with fresh ids in all three GUID spaces, an `Element` + row and an `InstanceChangeSet` row for each, three pairs appended to the head + changeset's `ElementToInstanceMap`, and the step's element appended to the + program's `Children`. +- **`EdfProgramContent` has five parallel maps keyed by step id, not one.** + `LinksFrom` (outgoing), `LinksTo` (incoming, as `$ref` back-pointers into the + link objects `LinksFrom` defines), `Ranks` (`{Rank: 0..N, StepId}`, the + running order), and `RelationsFrom`/`RelationsTo` (an empty list each). A step + present in only some of them leaves the console unable to build the program, + which it reports by showing the folder tree with no protocols in it -- the + same symptom as an archive exported from an empty folder node, and the first + duplication attempt was rejected exactly that way. +- **A protocol and a label each carry `ParentElementId` pointing at their own + step**, and only the step points at the program. The console resolves a + step's protocol through that reverse link rather than through the step's + `Children` blob, so a copy that keeps the source's pointer is served the + source's protocol. That is invisible while the copy is identical and silently + discards every edit once it is not: the second attempt loaded all 22 scans + and returned the edited copy with its source's TR. Any generated scan needs + one deliberately edited parameter or the test cannot fail. +- **`Instance.Tags` on a protocol carries `#ContentHash|`**, which is neither the `ContentHash` column nor the stored blob's + hash. It matches all 18 protocols in the reference archive. + `replace_content` recomputes it. A stale one is tolerated on load -- two + scans in the NAV option scan shared one and both kept their values -- but it + is derivable and describes the content. - **Import and re-export is not the same as editing.** `tCheckUUID`, the GUID leading `sWipMemBlock.tFree` and the date hidden in `sSpecPara.lFinalMatrixSize*` are regenerated when a protocol is *edited on the console*, and left alone diff --git a/src/siemens_protocol/exar/__init__.py b/src/siemens_protocol/exar/__init__.py index 87e73bb..49e7dc5 100644 --- a/src/siemens_protocol/exar/__init__.py +++ b/src/siemens_protocol/exar/__init__.py @@ -14,6 +14,7 @@ from __future__ import annotations +from . import generate, patch, validate from .archive import ( Archive, Instance, @@ -28,6 +29,9 @@ __all__ = [ "Archive", + "generate", + "patch", + "validate", "Envelope", "Instance", "PreviewEntry", diff --git a/src/siemens_protocol/exar/archive.py b/src/siemens_protocol/exar/archive.py index 8b2a7cd..ab7dac6 100644 --- a/src/siemens_protocol/exar/archive.py +++ b/src/siemens_protocol/exar/archive.py @@ -28,6 +28,8 @@ from __future__ import annotations +import hashlib +import re import uuid from dataclasses import dataclass, field from typing import Any, Iterator @@ -493,6 +495,10 @@ def replace_content(self, instance: Instance, document: dict[str, Any]) -> str: rows = self.container.tables["Instance"] for position in rows.find("Id", instance.id): rows.set(position, "ContentHash", digest) + tags = rows.rows[position][rows.index_of("Tags")] + refreshed = _refresh_content_tag(tags, document) + if refreshed != tags: + rows.set(position, "Tags", refreshed) instance.content_hash = digest return digest @@ -511,6 +517,34 @@ def write(self, path: str) -> None: store.write(self.container, path) +def _refresh_content_tag(tags: Any, document: dict[str, Any]) -> Any: + """Keep a protocol's ``#ContentHash`` tag in step with its XProtocol text. + + ``Instance.Tags`` carries a fingerprint of the protocol -- the SHA-1 of the + ``Data`` string, which is not the ``ContentHash`` column and not the hash of + the stored blob. The console recomputes it on save, and a stale one was + tolerated on load, but it is derivable and describes the content, so it is + kept correct rather than left to drift. + + Parameters + ---------- + tags : Any + The instance's current ``Tags`` value, which may be ``None``. + document : dict + The replacement document. Only protocol content carries ``Data``. + + Returns + ------- + Any + The updated tags, or the original when there is nothing to update. + """ + data = document.get("Data") if isinstance(document, dict) else None + if not isinstance(tags, str) or not isinstance(data, str): + return tags + digest = hashlib.sha1(data.encode("utf-8")).hexdigest() + return re.sub(r"(#ContentHash\|)[0-9a-f]+", rf"\g<1>{digest}", tags) + + def _head_branch(container: store.Container) -> tuple[str, str]: """Choose which branch head to read the archive at. diff --git a/src/siemens_protocol/exar/generate.py b/src/siemens_protocol/exar/generate.py new file mode 100644 index 0000000..9e64b5c --- /dev/null +++ b/src/siemens_protocol/exar/generate.py @@ -0,0 +1,319 @@ +"""Create nodes in an ``.exar1`` archive, rather than only editing them. + +A scanner has accepted archives built by this module: four added scans loaded, +kept their running order, and the one carrying an edit kept its own protocol. +Getting there cost two defects worth naming, because neither is visible to a +reader and both look like success. + +The first is that :data:`STEP_KEYED_MAPS` is five maps, not one. +``EdfProgramContent`` describes the running order in ``LinksFrom``, +``LinksTo``, ``Ranks``, ``RelationsFrom`` and ``RelationsTo``, all keyed by +step id. A step present in only some of them leaves the console unable to +build the program, which it reports by showing the folder tree with no +protocols in it -- indistinguishable, from the outside, from an archive +exported off an empty folder node. + +The second is that a protocol and a label each carry a ``ParentElementId`` +pointing at *their own step*, and the console resolves a step's protocol +through that reverse link rather than through the step's ``Children`` blob. A +copy that keeps the source's pointer is quietly served the source's protocol. +While the copy is identical that is invisible; once it is edited, the edit +disappears. Any test of this must therefore change something in the copy, or +it cannot tell "created correctly" from "aliased to the original". +""" + +from __future__ import annotations + +import uuid +from typing import Any + +from . import envelope +from .archive import Archive, Step, pack_guids, unpack_guids + +#: The maps in ``EdfProgramContent`` keyed by measurement-step id. Every step +#: must appear in all of them. +STEP_KEYED_MAPS = ("LinksFrom", "LinksTo", "Ranks", "RelationsFrom", "RelationsTo") + +#: .NET type moniker Newtonsoft writes on a link between two steps. +LINK_TYPE = "syngo.MR.ExamDataFoundation.Data.EdfProgramLink, syngo.MR.ExamDataFoundation.Data" + +#: The all-zero GUID, which these payloads use for "no condition". +NO_GUID = "00000000-0000-0000-0000-000000000000" + + +def renumber_references(document: Any) -> Any: + """Renumber Newtonsoft ``$id`` values sequentially, remapping ``$ref``. + + Json.NET numbers references ``1..N`` in the order it writes them, so an + invented key is legal JSON and not what the console produces. Verified by + reproducing an untouched console document exactly, which is what shows the + walk order matches Newtonsoft's serialization order rather than merely + looking plausible. + + Parameters + ---------- + document : Any + A decoded JSON document. + + Returns + ------- + Any + The document with ``$id`` renumbered and every ``$ref`` following it. + """ + mapping: dict[str, str] = {} + counter = 0 + + def collect(node: Any) -> None: + nonlocal counter + if isinstance(node, dict): + if "$id" in node: + counter += 1 + mapping[str(node["$id"])] = str(counter) + for value in node.values(): + collect(value) + elif isinstance(node, list): + for value in node: + collect(value) + + def rewrite(node: Any) -> Any: + if isinstance(node, dict): + out: dict[str, Any] = {} + for key, value in node.items(): + if key == "$id": + out[key] = mapping[str(value)] + elif key == "$ref": + out[key] = mapping.get(str(value), str(value)) + else: + out[key] = rewrite(value) + return out + if isinstance(node, list): + return [rewrite(value) for value in node] + return node + + collect(document) + return rewrite(document) + + +def duplicate_step(archive: Archive, step: Step, name: str) -> str: + """Append a copy of ``step`` to the archive's running order. + + The copy gets its own identity in all three GUID spaces and its own label. + Its protocol keeps the source's ``ContentHash``, which is correct rather + than lazy: the content is identical and the store is addressed by content, + exactly as the console shares one protocol between identical scans. Patch + the copy afterwards to give it content of its own. + + Parameters + ---------- + archive : Archive + The archive to extend. Modified in place; call + :meth:`Archive.write` to save, and re-read before using the new step, + since the live instance map is rebuilt on read. + step : Step + The step to copy. + name : str + Displayed name for the copy. + + Returns + ------- + str + The new step's object id. + + Raises + ------ + ValueError + If the archive has no program node to attach the step to. + """ + program = archive.program + if program is None: + raise ValueError("archive has no program to append a step to") + + tables = archive.container.tables + instances = tables["Instance"] + source_label = archive.by_element[step.instance.label_element_id] + fresh = { + tag: (str(uuid.uuid4()), str(uuid.uuid4()), str(uuid.uuid4())) + for tag in ("step", "protocol", "label") + } + + label_hash = _store_label(archive, source_label, name) + _add_instances(archive, step, source_label, fresh, label_hash) + _extend_map(archive, fresh) + _attach_to_program(archive, program, fresh["step"]) + return fresh["step"][2] + + +def _store_label(archive: Archive, source: Any, name: str) -> str: + """Write a locale table holding ``name`` and return its content hash. + + Parameters + ---------- + archive : Archive + The archive to store the content in. + source : Instance + The label node being copied, whose locale keys are reused. + name : str + The displayed name. + + Returns + ------- + str + Content hash of the stored label. + """ + document = dict(archive.document(source)) + texts = document.get("Texts", {}) + document["Texts"] = {k: (v if k == "$id" else name) for k, v in texts.items()} + content = archive.contents[source.content_hash].replace(document) + if content.hash not in archive.contents: + archive.contents[content.hash] = content + archive.container.tables["Content"].append( + {"Hash": content.hash, "Data": content.to_stored(), "Format": envelope.STORED_FORMAT} + ) + return content.hash + + +def _add_instances( + archive: Archive, + step: Step, + source_label: Any, + fresh: dict[str, tuple[str, str, str]], + label_hash: str, +) -> None: + """Add the three instance rows, their elements and their changeset rows. + + Parameters + ---------- + archive : Archive + The archive being extended. + step : Step + The step being copied. + source_label : Instance + The label node being copied. + fresh : dict + New ``(Id, Element_id, ObjectId)`` per node. + label_hash : str + Content hash of the copy's label. + + Returns + ------- + None + """ + tables = archive.container.tables + instances, elements = tables["Instance"], tables["Element"] + changes = tables["InstanceChangeSet"] + sources = {"step": step.instance, "protocol": step.protocol.instance, "label": source_label} + for tag, source in sources.items(): + identifier, element, obj = fresh[tag] + row = dict(zip(instances.columns, instances.rows[instances.find("Id", source.id)[0]])) + row["Id"], row["Element_id"], row["ObjectId"] = identifier, element, obj + if tag == "step": + row["Children"] = pack_guids([fresh["protocol"][1]]) + row["LabelElement_id"] = fresh["label"][1] + else: + # A protocol and a label both parent to their own step. Keeping the + # source's pointer makes the console serve this copy the source's + # protocol, and any edit to it silently disappears. + row["ParentElementId"] = fresh["step"][1] + if tag == "label": + row["ContentHash"] = label_hash + instances.append(row) + source_element = elements.rows[elements.find("Id", source.element_id)[0]] + elements.append(dict(zip(elements.columns, (element, source_element[1], None)))) + changes.append( + { + "InstanceId": identifier, + "ChangeSetId": archive.head, + "ElementId": element, + "State": 0, + } + ) + + +def _extend_map(archive: Archive, fresh: dict[str, tuple[str, str, str]]) -> None: + """Add the new nodes to the head changeset's checkout index. + + Parameters + ---------- + archive : Archive + The archive being extended. + fresh : dict + New ``(Id, Element_id, ObjectId)`` per node. + + Returns + ------- + None + """ + maps = archive.container.tables["ElementToInstanceMap"] + wanted = _head_map_id(archive) + position = maps.find("Id", wanted)[0] + blob = maps.rows[position][maps.index_of("Data")] + for identifier, element, _obj in fresh.values(): + blob += uuid.UUID(element).bytes_le + uuid.UUID(identifier).bytes_le + maps.set(position, "Data", blob) + + +def _head_map_id(archive: Archive) -> str: + """Return the element map the head changeset resolves through. + + Parameters + ---------- + archive : Archive + The archive to inspect. + + Returns + ------- + str + The map's id. + """ + for row in archive.container.rows("ChangeSet"): + if str(row["Id"]) != archive.head: + continue + delta = str(row["DeltaElementMapId"]) + return delta if delta != NO_GUID else str(row["BaseElementMapId"]) + raise ValueError(f"no changeset {archive.head}") + + +def _attach_to_program(archive: Archive, program: Any, step_ids: tuple[str, str, str]) -> None: + """Put the new step in the program's children and in all five maps. + + Parameters + ---------- + archive : Archive + The archive being extended. + program : Instance + The program node. + step_ids : tuple + The new step's ``(Id, Element_id, ObjectId)``. + + Returns + ------- + None + """ + instances = archive.container.tables["Instance"] + position = instances.find("Id", program.id)[0] + children = unpack_guids(instances.rows[position][instances.index_of("Children")]) + instances.set(position, "Children", pack_guids(children + [step_ids[1]])) + + document = archive.document(program) + last, new = document["LastStepId"], step_ids[2] + link = f"link-{new}" + document["LinksFrom"].setdefault(last, {"$id": f"lf-{last}", "$values": []}) + document["LinksFrom"][last]["$values"].append( + { + "$id": link, + "$type": LINK_TYPE, + "ConditionId": NO_GUID, + "SelectionId": NO_GUID, + "SourceId": last, + "TargetId": new, + } + ) + document["LinksFrom"][new] = {"$id": f"lf-{new}", "$values": []} + # The incoming edge is the same link object, referenced rather than repeated. + document["LinksTo"][new] = {"$id": f"lt-{new}", "$values": [{"$ref": link}]} + rank = max(v["Rank"] for k, v in document["Ranks"].items() if k != "$id") + 1 + document["Ranks"][new] = {"$id": f"rk-{new}", "Rank": rank, "StepId": new} + document["RelationsFrom"][new] = {"$id": f"rf-{new}", "$values": []} + document["RelationsTo"][new] = {"$id": f"rt-{new}", "$values": []} + document["LastStepId"] = new + archive.replace_content(program, renumber_references(document)) diff --git a/src/siemens_protocol/exar/validate.py b/src/siemens_protocol/exar/validate.py new file mode 100644 index 0000000..33703f3 --- /dev/null +++ b/src/siemens_protocol/exar/validate.py @@ -0,0 +1,234 @@ +"""Check an archive against the structure a console-authored one has. + +Self-consistency is not enough, and this module exists because of two defects +that proved it. Both times the archive was internally coherent -- every +reference resolved, every id was unique -- and both times the console rejected +or silently corrupted it. A field can be populated, well-formed and wrong, +because what it means is relational: ``ParentElementId`` on a copied protocol +pointed at a real step that simply was not its own. + +So the checks here are stated as *relationships that hold in every archive a +console wrote*, and are re-derived from the archive under test rather than +assumed. :func:`problems` returns what does not hold, most specific first, and +an empty list is the only passing result. +""" + +from __future__ import annotations + +import re +import uuid +from typing import Any + +from . import envelope +from .archive import MEASUREMENT_STEP, Archive +from .generate import NO_GUID, STEP_KEYED_MAPS + +#: Matches a GUID as these payloads spell one. +GUID = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$") + + +def problems(archive: Archive) -> list[str]: + """Return every structural rule the archive breaks. + + Parameters + ---------- + archive : Archive + The archive to check. + + Returns + ------- + list of str + One line per broken rule, empty when the archive is sound. + """ + found: list[str] = [] + program = archive.program + if program is None: + return ["archive has no program node"] + document = archive.document(program) + found += _program_maps(archive, document) + found += _running_order(archive, document) + found += _references(document) + found += _parents(archive) + found += _identity(archive) + return found + + +def _program_maps(archive: Archive, document: dict[str, Any]) -> list[str]: + """Every step must appear in every step-keyed map. + + Parameters + ---------- + archive : Archive + The archive under test. + document : dict + The decoded program content. + + Returns + ------- + list of str + Broken rules. + """ + steps = {step.instance.object_id for step in archive.steps} + found = [] + for name in STEP_KEYED_MAPS: + table = document.get(name) + if not isinstance(table, dict): + found.append(f"program content has no {name} map") + continue + keys = {k for k in table if k != "$id"} + missing = steps - keys + if missing: + found.append(f"{name} is missing {len(missing)} step(s): {sorted(missing)[:3]}") + stray = {k for k in keys if GUID.match(k)} - steps + if stray: + found.append(f"{name} names {len(stray)} step(s) that do not exist") + return found + + +def _running_order(archive: Archive, document: dict[str, Any]) -> list[str]: + """Ranks number the steps 0..N, and the chain spans them all. + + Parameters + ---------- + archive : Archive + The archive under test. + document : dict + The decoded program content. + + Returns + ------- + list of str + Broken rules. + """ + found = [] + order = archive.step_order() + # Count the step nodes independently of the chain. ``archive.steps`` is + # derived by walking the chain, so comparing the two would be circular: a + # chain that stops early would agree with itself and the break would show + # up only as unrelated complaints about the maps. + existing = sum(1 for i in archive.instances.values() if i.kind == MEASUREMENT_STEP) + if len(order) != existing: + found.append(f"link chain covers {len(order)} steps but {existing} exist") + if order and document.get("FirstStepId") != order[0]: + found.append("FirstStepId is not where the chain starts") + if order and document.get("LastStepId") != order[-1]: + found.append("LastStepId is not where the chain ends") + ranks = sorted(v["Rank"] for k, v in document.get("Ranks", {}).items() if k != "$id") + if ranks and ranks != list(range(len(ranks))): + found.append(f"Ranks are not 0..{len(ranks) - 1} without gaps") + if len(archive.program.children) != len(archive.steps): + found.append( + f"program lists {len(archive.program.children)} children " + f"for {len(archive.steps)} steps" + ) + return found + + +def _references(document: dict[str, Any]) -> list[str]: + """Newtonsoft ``$ref`` values resolve, and ``$id`` values are unique. + + Parameters + ---------- + document : dict + The decoded program content. + + Returns + ------- + list of str + Broken rules. + """ + ids: list[str] = [] + refs: list[str] = [] + + def walk(node: Any) -> None: + if isinstance(node, dict): + if "$id" in node: + ids.append(str(node["$id"])) + if "$ref" in node: + refs.append(str(node["$ref"])) + for value in node.values(): + walk(value) + elif isinstance(node, list): + for value in node: + walk(value) + + walk(document) + found = [] + if len(ids) != len(set(ids)): + found.append("program content repeats a $id") + dangling = set(refs) - set(ids) + if dangling: + found.append(f"program content has {len(dangling)} unresolved $ref") + return found + + +def _parents(archive: Archive) -> list[str]: + """A protocol and a label parent to their step; a step to the program. + + This is the rule a duplicated scan breaks by inheriting its source's + pointer, and the console resolves a step's protocol through it -- so the + copy is served the original's protocol and any edit to it disappears. + + Parameters + ---------- + archive : Archive + The archive under test. + + Returns + ------- + list of str + Broken rules. + """ + rows = {str(row["Id"]): row for row in archive.container.rows("Instance")} + program = archive.program + found = [] + for step in archive.steps: + expected = { + "protocol": (step.protocol.instance, step.instance.element_id), + "step": (step.instance, program.element_id), + } + holder = archive.by_element.get(step.instance.label_element_id) + if holder is not None: + expected["label"] = (holder, step.instance.element_id) + for tag, (node, wanted) in expected.items(): + actual = rows[node.id]["ParentElementId"] + if actual is not None and str(actual) != wanted: + found.append(f"{step.name}: its {tag} parents to another node, not {tag}'s own") + return found + + +def _identity(archive: Archive) -> list[str]: + """Ids are unique, the checkout index resolves, content hashes hold. + + Parameters + ---------- + archive : Archive + The archive under test. + + Returns + ------- + list of str + Broken rules. + """ + rows = archive.container.rows("Instance") + ids = [str(row["Id"]) for row in rows] + elements = {str(row["Id"]) for row in archive.container.rows("Element")} + found = [] + if len(ids) != len(set(ids)): + found.append("Instance.Id is not unique") + absent = {str(row["Element_id"]) for row in rows} - elements + if absent: + found.append(f"{len(absent)} instance(s) reference a missing Element row") + for row in rows: + digest = row["ContentHash"] + if digest is not None and str(digest) not in archive.contents: + found.append(f"instance {str(row['Id'])[:8]} points at missing content") + break + for stored, content in archive.contents.items(): + rebuilt = envelope.Envelope( + content_type=content.content_type, payload=envelope.dumps(content.decode()) + ) + if rebuilt.hash != stored: + found.append(f"{content.kind} does not re-encode to its own address") + break + return found diff --git a/tests/test_exar_generate.py b/tests/test_exar_generate.py new file mode 100644 index 0000000..76652af --- /dev/null +++ b/tests/test_exar_generate.py @@ -0,0 +1,205 @@ +"""Tests for creating scans in an ``.exar1`` archive, and for validating one. + +A scanner has accepted archives built this way, which is the only authority +that counts. Getting there cost two defects, and both are represented here as +negative tests: a validator that never fires would be worse than none, since +it would carry the appearance of assurance. + +The positive tests assert that generation produces an archive indistinguishable +from a console-authored one on every rule :mod:`validate` knows. The negative +tests reintroduce each defect and require it to be reported. +""" + +from __future__ import annotations + +import pathlib + +import pytest + +from conftest import find_exar, protocol_archive_path, requires_exar # noqa: F401 +from siemens_protocol.exar import generate, patch, read, validate + + +@requires_exar +def test_every_console_archive_passes_the_structural_rules(protocol_archive_path: str) -> None: + """The rules are stated from what console archives do, so they must hold. + + Parameters + ---------- + protocol_archive_path : str + One archive from the corpus that carries protocols. + + Returns + ------- + None + """ + assert validate.problems(read(protocol_archive_path)) == [] + + +@requires_exar +def test_a_duplicated_scan_produces_a_sound_archive(tmp_path: pathlib.Path) -> None: + """Adding a scan leaves every structural rule intact. + + Parameters + ---------- + tmp_path : pathlib.Path + Destination for the written archive. + + Returns + ------- + None + """ + archive = read(find_exar("Potpourri_P1.exar1")) + before = [step.name for step in archive.steps] + generate.duplicate_step(archive, archive.steps[0], "COPY_of_first") + + written = tmp_path / "grown.exar1" + archive.write(str(written)) + grown = read(str(written)) + + assert validate.problems(grown) == [] + names = [step.name for step in grown.steps] + assert names == before + ["COPY_of_first"], "the copy is not last in running order" + assert len(grown.program.children) == len(before) + 1 + + +@requires_exar +def test_a_copy_can_be_given_content_of_its_own(tmp_path: pathlib.Path) -> None: + """A duplicated scan patched afterwards keeps its own protocol. + + This is the case that matters and the one that failed on a scanner: while + a copy is identical to its source, being served the source's protocol is + indistinguishable from success. Only an edited copy can tell the two + apart, so this test edits one. + + Parameters + ---------- + tmp_path : pathlib.Path + Destination for the written archive. + + Returns + ------- + None + """ + source = read(find_exar("Potpourri_P1.exar1")) + original = {s.name: s for s in source.steps}["Minn_CMRR_2.3mm_S8_rest_6min"] + generate.duplicate_step(source, original, "COPY_with_its_own_TR") + first = tmp_path / "grown.exar1" + source.write(str(first)) + + # Re-read: the live instance map is rebuilt on read, so the new step is + # only addressable afterwards. + grown = read(str(first)) + copy = {s.name: s for s in grown.steps}["COPY_with_its_own_TR"] + document, applied, skipped = patch.patch_document(copy.protocol, {"TR": 652.0}) + assert applied and not skipped + grown.replace_content(copy.protocol.instance, document) + second = tmp_path / "patched.exar1" + grown.write(str(second)) + + final = read(str(second)) + assert validate.problems(final) == [] + steps = {s.name: s for s in final.steps} + assert steps["COPY_with_its_own_TR"].protocol.preview["sub.0.msr.tr.0"].value == 652.0 + assert steps["Minn_CMRR_2.3mm_S8_rest_6min"].protocol.preview["sub.0.msr.tr.0"].value == 650.0 + assert ( + steps["COPY_with_its_own_TR"].protocol.instance.content_hash + != steps["Minn_CMRR_2.3mm_S8_rest_6min"].protocol.instance.content_hash + ), "the copy is sharing its source's protocol, so an edit to it would be lost" + + +@requires_exar +def test_renumbering_reproduces_an_untouched_console_document() -> None: + """The walk order matches Newtonsoft's, which is what makes it canonical. + + Returns + ------- + None + """ + archive = read(find_exar("Potpourri_P1.exar1")) + document = archive.document(archive.program) + assert generate.renumber_references(document) == document + + +# -------------------------------------------------------------------------- +# The defects that motivated the validator. Each must be reported. +# -------------------------------------------------------------------------- + + +@requires_exar +def test_a_step_missing_from_the_other_maps_is_reported() -> None: + """The console rejects such an archive outright; the checker must not. + + ``EdfProgramContent`` describes the running order in five maps keyed by + step id. A step in only some of them leaves the console unable to build + the program, which it reports by showing the folder and no protocols -- + the same symptom as an archive exported off an empty folder node. + + Returns + ------- + None + """ + archive = read(find_exar("Potpourri_P1.exar1")) + document = archive.document(archive.program) + victim = archive.steps[-1].instance.object_id + for name in ("LinksTo", "Ranks", "RelationsFrom", "RelationsTo"): + document[name].pop(victim, None) + archive.replace_content(archive.program, document) + + reported = validate.problems(archive) + assert reported, "a step missing from four maps was not reported" + for name in ("LinksTo", "Ranks", "RelationsFrom", "RelationsTo"): + assert any(line.startswith(f"{name} is missing") for line in reported), name + + +@requires_exar +def test_a_protocol_parenting_to_another_step_is_reported() -> None: + """The failure this catches is silent, which is why it is worth catching. + + The console resolves a step's protocol through ``ParentElementId``, so a + copy that keeps its source's pointer is served the source's protocol. On a + scanner that showed up as an edited copy returning its source's TR. + + Returns + ------- + None + """ + archive = read(find_exar("Potpourri_P1.exar1")) + rows = archive.container.tables["Instance"] + target = archive.steps[3] + elsewhere = archive.steps[0].instance.element_id + rows.set(rows.find("Id", target.protocol.instance.id)[0], "ParentElementId", elsewhere) + + reported = validate.problems(archive) + assert any("its protocol parents to another node" in line for line in reported) + + +@requires_exar +def test_a_dangling_reference_is_reported() -> None: + """Newtonsoft resolves ``$ref`` against ``$id``; an unresolved one is broken. + + Returns + ------- + None + """ + archive = read(find_exar("Potpourri_P1.exar1")) + document = archive.document(archive.program) + document["LinksTo"][document["LastStepId"]]["$values"] = [{"$ref": "nonexistent"}] + archive.replace_content(archive.program, document) + assert any("unresolved $ref" in line for line in validate.problems(archive)) + + +@requires_exar +def test_a_broken_running_order_is_reported() -> None: + """A chain that does not span the steps means scans would go unseen. + + Returns + ------- + None + """ + archive = read(find_exar("Potpourri_P1.exar1")) + document = archive.document(archive.program) + document["LinksFrom"][document["FirstStepId"]]["$values"] = [] + archive.replace_content(archive.program, document) + reported = validate.problems(archive) + assert any("link chain covers" in line for line in reported) From 4ee347fd07ce1dd830c8d51c2cc9aba43ef82f5c Mon Sep 17 00:00:00 2001 From: Blaise deB Frederick Date: Fri, 28 Aug 2026 15:39:28 -0400 Subject: [PATCH 2/2] Drive an archive from a parsed PDF, and state what it could not write Adds `siemens-protocol-tool exar `: take a template .exar1, write every parameter with a verified mapping, and report the rest. The manifest is as much the point as the output. Roughly a tenth of what a protocol prints is mapped, so a built archive is mostly the template it started from. A tool reporting only its successes would describe a small part of the result and imply the whole, so the report states the fraction, counts inherited values, and names the unmapped parameters by frequency -- which is what says where the next mapping is worth deriving rather than guessing at one. Driving an archive from its *own* PDF must write nothing, and that single check exercises units, scales, the derived basis, sparse arrays and change detection together. It caught two defects. Change detection compared the displayed value, which the Special card does not have -- every ASCCONV-only mapping therefore counted as a write and rewrote content that had not moved. And a printed "0.00" against an assignment a sparse array omits was reported as a change to "0.0" when writing zero into a sparse array in fact removes the line; records now say what is actually stored, for which "(absent)" is a value rather than a gap. patch.resolve now falls back to the label Preview prints. A multi-echo scan prints "TE 1" where a single-echo one prints "TE", and the preview entry is labelled the same way, so resolving through it follows the printout instead of duplicating every spelling in the table. Without it the driver silently skipped TE on exactly the scans that print it differently -- 2 fields out of 1353 wrong, which is the kind of miss that reads as a rounding difference. With it, driving the unmodified archive from the changed protocol's PDF lands on the console's own values in all 1353 mapped fields. Scans are matched to the template by name, and an unmatched one is reported rather than guessed at: the PDF names a sequence by kernel and the archive by sequence file, so there is no reliable way to choose a donor to copy. generate.duplicate_step is there for a caller who knows which scan to copy; the driver will not choose one. The GUI exposes the command through the one description in gui/commands.py, as its own test requires. Co-Authored-By: Claude Opus 5 --- CLAUDE.md | 20 +++ src/siemens_protocol/cli.py | 66 +++++++ src/siemens_protocol/exar/build.py | 258 +++++++++++++++++++++++++++ src/siemens_protocol/exar/patch.py | 28 ++- src/siemens_protocol/gui/commands.py | 65 +++++++ tests/test_exar_generate.py | 158 +++++++++++++++- 6 files changed, 592 insertions(+), 3 deletions(-) create mode 100644 src/siemens_protocol/exar/build.py diff --git a/CLAUDE.md b/CLAUDE.md index 252f1a4..b78ba65 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -525,6 +525,26 @@ before/after pair is the only way to learn where a printed value is stored -- the corpus, and the sweeps that need scans take `protocol_archive_path` while the structural ones (envelope, hashing, GUID layout) still take `archive_path` and are exercised by it. +- **`siemens-protocol-tool exar ` is the driver**, and its + manifest is as much the point as its output. Roughly a tenth of what a + protocol prints has a verified mapping, so a built archive is mostly the + template it started from; the report states that fraction, counts inherited + values and names the unmapped parameters by frequency, which is what says + where the next mapping is worth deriving. Driving an archive from its *own* + PDF must write nothing -- that one check exercises units, scales, the derived + basis, sparse arrays and change detection at once, and it caught two spurious + writes where a printed `0.00` met an assignment a sparse array omits. +- **Scans are matched to the template by name, and an unmatched one is + reported rather than guessed at.** The PDF names a sequence by kernel + (`epfid`) and the archive by sequence file (`cmrr_mbep2d_bold`), so there is + no reliable way to pick a donor scan to copy. `generate.duplicate_step` is + available to a caller who knows which scan to copy; the driver will not + choose one. +- **`patch.resolve` falls back to the label `Preview` prints.** A multi-echo + scan prints `TE 1` where a single-echo one prints `TE`, and the preview entry + is labelled the same way, so resolving through it follows the printout + instead of duplicating every spelling in the table. Without that the driver + silently skipped TE on exactly the scans that print it differently. - Everything so far is XA60 (`VA60A`). No XA30 archive has been seen, so the claim that the model is release-independent is untested -- that is the first thing to check when one arrives. diff --git a/src/siemens_protocol/cli.py b/src/siemens_protocol/cli.py index d32b27e..0da9b92 100644 --- a/src/siemens_protocol/cli.py +++ b/src/siemens_protocol/cli.py @@ -341,6 +341,27 @@ def build_parser() -> argparse.ArgumentParser: ) vocab_suggest.add_argument("--vocabulary", metavar="DIR", help="an overlay directory") + exar_cmd = sub.add_parser( + "exar", + help="write a protocol PDF's parameters into an XA .exar1 archive", + description=( + "Take a template .exar1 archive and a protocol PDF, write every " + "parameter that has a verified mapping, and report what could not " + "be written. Only a fraction of what a protocol prints is mapped, " + "so the result is mostly the template it started from -- the " + "manifest says how much, and names the parameters no mapping " + "covers so the gap is visible rather than implied. Nothing is " + "written without --out." + ), + ) + exar_cmd.add_argument("archive", help="the template .exar1 archive") + exar_cmd.add_argument("input", help="a protocol PDF, or a previously parsed JSON file") + add_release_option(exar_cmd, "force a Siemens release profile for a PDF input (default: auto)") + exar_cmd.add_argument("--out", help="write the resulting archive here") + exar_cmd.add_argument( + "--show", type=int, default=12, metavar="N", help="how many entries to list (default: 12)" + ) + versions_cmd = sub.add_parser("versions", help="list the known version profiles") versions_cmd.set_defaults(command="versions") @@ -1268,6 +1289,48 @@ def use_utf8_output() -> None: pass +def _run_exar(args: argparse.Namespace) -> int: + """Write a PDF's mapped parameters into a template archive. + + Parameters + ---------- + args : argparse.Namespace + Parsed arguments carrying ``archive``, ``input``, ``out`` and ``show``. + + Returns + ------- + int + Process exit status. + """ + from .exar import build as exar_build + from .exar import read as read_exar + from .exar import validate as exar_validate + + try: + archive = read_exar(args.archive) + protocol = _load_protocol(args.input, getattr(args, "release", "auto")) + except (OSError, ValueError) as exc: + print(f"{exc}", file=sys.stderr) + return 1 + + report = exar_build.apply_protocol(archive, protocol) + print(report.report(limit=args.show)) + + problems = exar_validate.problems(archive) + if problems: + print("\nthe result is not structurally sound:", file=sys.stderr) + for line in problems: + print(f" {line}", file=sys.stderr) + return 1 + + if args.out: + archive.write(args.out) + print(f"\nwrote {args.out}") + else: + print("\n(no --out given, so nothing was written)") + return 0 + + def main(argv: list[str] | None = None) -> int: """Run the command line interface. @@ -1300,6 +1363,9 @@ def main(argv: list[str] | None = None) -> int: if args.command == "list": return _run_list(args) + if args.command == "exar": + return _run_exar(args) + if args.command == "sequences": return _run_sequences(args) diff --git a/src/siemens_protocol/exar/build.py b/src/siemens_protocol/exar/build.py new file mode 100644 index 0000000..cea9c34 --- /dev/null +++ b/src/siemens_protocol/exar/build.py @@ -0,0 +1,258 @@ +"""Drive an archive from a parsed PDF: apply what is mapped, report the rest. + +This is the layer a person actually uses. Give it a template archive and a +protocol parsed out of a PDF, and it writes every parameter both sides agree +on, then says plainly what it could not write. + +The reporting is the point as much as the writing. Only a fraction of what a +protocol prints has a verified mapping, so an archive this produces is mostly +the template it started from. A tool that reported only its successes would +be describing a small part of the result and implying the whole; the manifest +therefore counts inherited values, names the printed parameters no mapping +covers, and orders them by how often they actually differ between scans -- +which is what says where the next mapping is worth deriving. + +Scans are matched to the template by name. The PDF's sequence field is the +kernel (``epfid``) and the archive's is the sequence file (``cmrr_mbep2d_bold``), +so an unmatched scan cannot be paired with a donor to copy without guessing, +and this module does not guess: it reports the scan as unmatched and leaves +adding it to a caller who knows which template scan it resembles. +""" + +from __future__ import annotations + +import collections +import re +from dataclasses import dataclass, field +from typing import Any +from typing import Mapping as MappingType + +from . import patch +from .archive import Archive + +#: Units the card prints beside a value and the protocol does not store. +UNIT_SUFFIX = re.compile( + r"\s*(ms|s|mm|cm|deg|degree|degrees|Hz|Hz/Px|kHz|%|TRs|TR|min|sec|mT/m|ppm)\s*$", + re.I, +) + + +def printed_value(text: Any) -> Any: + """Strip the unit a printout appends, leaving what the protocol stores. + + Parameters + ---------- + text : Any + A value as the PDF prints it, for example ``"650.0 ms"``. + + Returns + ------- + Any + The value without its unit, unchanged when it carries none. + """ + if not isinstance(text, str): + return text + # Some values carry two: "5.0 mm" is one, "24 TRs" is one, and a printed + # angle is "360 degrees". Strip repeatedly rather than assume a single unit. + stripped = text.strip() + while True: + shorter = UNIT_SUFFIX.sub("", stripped).strip() + if shorter == stripped: + return stripped + stripped = shorter + + +def printed_parameters(scan: MappingType[str, Any]) -> dict[str, Any]: + """Flatten one parsed scan to ``{parameter: printed value}``. + + Parameters + ---------- + scan : mapping + One entry of a parsed protocol's ``scans``. + + Returns + ------- + dict + Every parameter the scan prints, keyed by printed name. + """ + return { + key: (item.get("value") if isinstance(item, dict) else item) + for key, item in (scan.get("flat") or {}).items() + } + + +@dataclass +class BuildReport: + """What a build wrote, refused, and left as it found it. + + Attributes + ---------- + applied : list + Values written, as :class:`patch.Applied` records. + skipped : list + Values a mapping claimed and could not write, with reasons. + unchanged : int + Mapped parameters whose printed value already matched the template. + inherited : collections.Counter + Printed parameters no mapping covers, counted across matched scans. + matched : list of str + Scan names present in both the PDF and the template. + unmatched : list of str + Scans the PDF prints that the template does not hold. + untouched : list of str + Template scans the PDF does not mention, left exactly as they were. + """ + + applied: list[patch.Applied] = field(default_factory=list) + skipped: list[patch.Skipped] = field(default_factory=list) + unchanged: int = 0 + inherited: collections.Counter = field(default_factory=collections.Counter) + matched: list[str] = field(default_factory=list) + unmatched: list[str] = field(default_factory=list) + untouched: list[str] = field(default_factory=list) + + @property + def coverage(self) -> tuple[int, int]: + """Return how much of what the PDF prints this build can write. + + Returns + ------- + tuple of int + Parameters written or confirmed, and parameters printed in total. + """ + writable = len(self.applied) + self.unchanged + return (writable, writable + sum(self.inherited.values())) + + def report(self, limit: int = 12) -> str: + """Render the manifest for a person to read. + + Parameters + ---------- + limit : int, optional + How many unmapped parameters to name. + + Returns + ------- + str + The manifest. + """ + written, total = self.coverage + share = f"{100 * written / total:.0f}%" if total else "n/a" + lines = [ + f"matched {len(self.matched)} scan(s); " + f"{len(self.unmatched)} in the PDF are not in the template; " + f"{len(self.untouched)} template scan(s) untouched", + f"wrote {len(self.applied)} value(s), {self.unchanged} already matched, " + f"{len(self.skipped)} refused", + f"coverage: {written} of {total} printed parameters ({share}) have a mapping", + ] + for one in self.applied[:limit]: + lines.append(f" set {one.step}: {one.label} {one.previous} -> {one.value}") + if len(self.applied) > limit: + lines.append(f" ... and {len(self.applied) - limit} more") + for miss in self.skipped[:limit]: + lines.append(f" skip {miss.step}: {miss.label} -- {miss.reason}") + if self.unmatched: + lines.append("scans with no template counterpart: " + ", ".join(self.unmatched[:8])) + if self.inherited: + lines.append("most common printed parameters with no mapping:") + for name, count in self.inherited.most_common(limit): + lines.append(f" {count:4d}x {name}") + return "\n".join(lines) + + +def apply_protocol(archive: Archive, parsed: MappingType[str, Any]) -> BuildReport: + """Write every mapped parameter a parsed PDF and a template agree on. + + The archive is edited in memory; call :meth:`Archive.write` to save, and + :func:`validate.problems` to check the result before trusting it. + + Parameters + ---------- + archive : Archive + Template archive, modified in place. + parsed : mapping + A protocol as ``siemens_protocol`` parses it, with a ``scans`` list. + + Returns + ------- + BuildReport + What was written, refused and inherited. + """ + report = BuildReport() + steps: dict[str, list[Any]] = {} + for step in archive.steps: + steps.setdefault(step.name, []).append(step) + + seen: set[str] = set() + for scan in parsed.get("scans", []): + name = scan.get("name", "") + found = steps.get(name, []) + if len(found) != 1: + report.unmatched.append(name) + continue + seen.add(name) + report.matched.append(name) + _apply_scan(archive, found[0], scan, report) + report.untouched = [n for n in steps if n not in seen] + return report + + +def _moved(record: patch.Applied) -> bool: + """Return whether a written record actually changed anything. + + Parameters + ---------- + record : patch.Applied + One written value. + + Returns + ------- + bool + ``True`` when the stored value differs from what was there. + """ + if record.previous is None: + return record.ascconv_previous != record.ascconv_value + return str(record.previous) != str(record.value) + + +def _apply_scan( + archive: Archive, step: Any, scan: MappingType[str, Any], report: BuildReport +) -> None: + """Write one scan's mapped parameters and account for the rest. + + Parameters + ---------- + archive : Archive + The archive being built. + step : Step + The template step matching this scan. + scan : mapping + The parsed scan. + report : BuildReport + Accumulates the outcome. + + Returns + ------- + None + """ + requests: dict[str, Any] = {} + for label, value in printed_parameters(scan).items(): + mapping, _reason = patch.resolve(step.protocol, label) + if mapping is None: + report.inherited[label] += 1 + continue + requests[label] = printed_value(value) + if not requests: + return + document, applied, skipped = patch.patch_document(step.protocol, requests, step=step.name) + # A value the template already holds is confirmation, not a write. The + # Special card has no preview side, so its records carry no previous + # displayed value at all -- comparing that would count every one of them + # as a change and rewrite content that did not move. + changed = [a for a in applied if _moved(a)] + report.unchanged += len(applied) - len(changed) + report.applied.extend(changed) + report.skipped.extend(skipped) + if changed: + archive.replace_content(step.protocol.instance, document) diff --git a/src/siemens_protocol/exar/patch.py b/src/siemens_protocol/exar/patch.py index 699e1ee..e722c12 100644 --- a/src/siemens_protocol/exar/patch.py +++ b/src/siemens_protocol/exar/patch.py @@ -45,6 +45,10 @@ from .archive import Archive, Protocol, Step +#: How a record spells an assignment that is not present. A sparse array omits +#: an element holding zero, so "absent" is a value rather than a gap. +ABSENT = "(absent)" + #: Delimiters of the ASCCONV block inside the XProtocol text. ASCCONV_BEGIN = "### ASCCONV BEGIN" ASCCONV_END = "### ASCCONV END" @@ -920,6 +924,20 @@ def resolve(protocol: Protocol, name: str) -> tuple[Mapping | None, str]: hits = [m for m in in_scope if m.preview_path == name] if len(hits) == 1: return (hits[0], "") + if not hits: + # The card does not always print a parameter under the name a mapping + # carries: a multi-echo scan prints "TE 1" where a single-echo one + # prints "TE", and Preview labels it the same way. Resolving through + # the preview entry follows the printout rather than duplicating every + # spelling in the table. + paths = { + entry.path + for entry in protocol.preview.values() + if entry.label.strip().casefold() == wanted + } + hits = [m for m in in_scope if m.preview_path in paths] + if len(hits) == 1: + return (hits[0], "") if len(hits) > 1: keys = ", ".join(sorted(m.ascconv_key for m in hits)) return (None, f"label {name!r} maps to several parameters: {keys}") @@ -1122,7 +1140,8 @@ def refused(why: str) -> tuple[Skipped, str]: # is a legitimate state, not a missing target. literal = set_bit(existing, mapping.bit, bool(number)) if not first_before: - first_before, first_after = existing or "0", literal + first_before = existing or "0" + first_after = ABSENT if (sparse and _is_zero(literal)) else literal text = _store(text, key, literal, existing, sparse) continue if existing is None and not sparse: @@ -1136,7 +1155,12 @@ def refused(why: str) -> tuple[Skipped, str]: written *= float(basis) literal = format_like(written, existing if existing is not None else _model(key)) if not first_before: - first_before, first_after = existing or "(absent)", literal + # Report what will actually be stored. Writing zero into a sparse + # array removes the assignment, so an absent element asked to hold + # zero does not change -- and saying otherwise makes a no-op run + # look like it wrote something. + first_before = existing or ABSENT + first_after = ABSENT if (sparse and _is_zero(literal)) else literal text = _store(text, key, literal, existing, sparse) previous = None diff --git a/src/siemens_protocol/gui/commands.py b/src/siemens_protocol/gui/commands.py index 488cff6..45fb936 100644 --- a/src/siemens_protocol/gui/commands.py +++ b/src/siemens_protocol/gui/commands.py @@ -773,6 +773,70 @@ def _vocab_commands() -> tuple[Command, ...]: ) +def _exar_command() -> Command: + """Describe the archive-writing command. + + Returns + ------- + Command + The command's form and argument construction. + """ + return Command( + name="exar", + group="Archive", + title="Write into an .exar1 archive", + summary=( + "Take a template .exar1 archive and a protocol PDF, write every parameter " + "that has a verified mapping, and report what could not be written. Only a " + "fraction of what a protocol prints is mapped, so the result is mostly the " + "template it started from -- the manifest says how much, and names the " + "parameters no mapping covers. Nothing is written without a destination." + ), + argv=("exar",), + fields=( + Field( + name="archive", + kind="path", + label="Template archive", + help="The .exar1 archive to write into. It is not modified in place.", + picker="file", + accept=(".exar1",), + required=True, + ), + Field( + name="input", + kind="path", + label="Protocol", + help="A PDF, or JSON this tool wrote earlier.", + picker="file", + accept=(".pdf", ".json"), + required=True, + ), + _release_field("Force a Siemens release profile for a PDF input."), + Field( + name="out", + kind="path", + label="Write archive to", + help=( + "Where to write the result. Left empty, the manifest is shown and " + "no archive is written." + ), + flag="--out", + picker="save", + accept=(".exar1",), + ), + Field( + name="show", + kind="int", + label="Entries to list", + help="How many written and unmapped entries the manifest names.", + flag="--show", + default=12, + ), + ), + ) + + def _versions_command() -> Command: """Describe the ``versions`` subcommand. @@ -810,6 +874,7 @@ def command_specs() -> tuple[Command, ...]: _diff_command(), _check_command(), _list_command(), + _exar_command(), _sequences_command(), *_vocab_commands(), _versions_command(), diff --git a/tests/test_exar_generate.py b/tests/test_exar_generate.py index 76652af..8466349 100644 --- a/tests/test_exar_generate.py +++ b/tests/test_exar_generate.py @@ -16,7 +16,12 @@ import pytest -from conftest import find_exar, protocol_archive_path, requires_exar # noqa: F401 +from conftest import ( # noqa: F401 + find_exar, + find_pdf, + protocol_archive_path, + requires_exar, +) from siemens_protocol.exar import generate, patch, read, validate @@ -203,3 +208,154 @@ def test_a_broken_running_order_is_reported() -> None: archive.replace_content(archive.program, document) reported = validate.problems(archive) assert any("link chain covers" in line for line in reported) + + +# -------------------------------------------------------------------------- +# The driver: a template archive plus a parsed PDF +# -------------------------------------------------------------------------- + + +def _parse(pdf: str) -> dict: + """Parse one example PDF through the CLI, as the driver's callers do. + + Parameters + ---------- + pdf : str + Path to the PDF. + + Returns + ------- + dict + The parsed protocol. + """ + import json + import subprocess + import sys + + done = subprocess.run( + [sys.executable, "-m", "siemens_protocol.cli", "parse", pdf, "--stdout"], + capture_output=True, + text=True, + check=True, + ) + return json.loads(done.stdout) + + +@requires_exar +def test_driving_an_archive_from_its_own_pdf_writes_nothing() -> None: + """A protocol told what it already says must not change. + + This is the sharpest cheap check on the whole chain: units, scales, the + derived basis, sparse arrays and change detection all have to be right or + something reports a spurious write. An earlier version wrote two values + here, both a printed ``0.00`` against an assignment a sparse array omits. + + Returns + ------- + None + """ + from siemens_protocol.exar import build + + archive = read(find_exar("Potpourri_P1.exar1")) + report = build.apply_protocol(archive, _parse(find_pdf("Potpourri_P1.pdf"))) + assert report.applied == [], [f"{a.step}: {a.label}" for a in report.applied] + assert report.unchanged > 100, "nothing was compared, so this proves nothing" + assert report.unmatched == [] + + +@requires_exar +def test_driving_an_archive_reproduces_the_console_edit(tmp_path: pathlib.Path) -> None: + """Given the changed PDF, the driver writes what the console wrote. + + ``Potpourri_P1_changed`` is the same protocol after the console changed + many parameters across five scans. Driving the unmodified archive from + that PDF must land on the same values in every mapped field. + + Parameters + ---------- + tmp_path : pathlib.Path + Destination for the built archive. + + Returns + ------- + None + """ + from siemens_protocol.exar import build + + archive = read(find_exar("Potpourri_P1.exar1")) + report = build.apply_protocol(archive, _parse(find_pdf("Potpourri_P1_changed.pdf"))) + assert report.applied, "the changed PDF should have moved something" + assert validate.problems(archive) == [] + + written = tmp_path / "built.exar1" + archive.write(str(written)) + ours = {s.name: s for s in read(str(written)).steps} + theirs = read(find_exar("Potpourri_P1_changed.exar1")) + + compared = 0 + for step in theirs.steps: + mine = ours[step.name] + for mapping in patch.MAPPINGS: + if not patch.applies_to(mapping, step.protocol): + continue + for key, _index in patch.expand(mapping.ascconv_key, step.protocol.xprotocol): + got = patch.read_ascconv(mine.protocol.xprotocol, key) + want = patch.read_ascconv(step.protocol.xprotocol, key) + if got is None and want is None: + continue + compared += 1 + if mapping.bit is not None: + assert (int(got or 0) >> mapping.bit & 1) == ( + int(want or 0) >> mapping.bit & 1 + ), f"{step.name}: {mapping.label}" + elif mapping.basis is not None: + # FOV Phase is quantised by the console; see patch.Manifest. + assert abs(float(got) - float(want)) <= 5e-4 * max(1.0, abs(float(want))) + else: + assert got == want, f"{step.name}: {mapping.label}" + assert compared > 1000, f"only {compared} fields compared" + + +@requires_exar +def test_the_report_counts_what_it_could_not_write() -> None: + """Coverage is stated, not implied. + + Most of what a protocol prints has no mapping, so a manifest that listed + only successes would describe a small part of the result as if it were the + whole. The unmapped parameters are named and counted. + + Returns + ------- + None + """ + from siemens_protocol.exar import build + + archive = read(find_exar("Potpourri_P1.exar1")) + report = build.apply_protocol(archive, _parse(find_pdf("Potpourri_P1.pdf"))) + written, total = report.coverage + assert 0 < written < total, "coverage should be a real fraction, not all or nothing" + assert report.inherited, "no unmapped parameters were recorded" + text = report.report() + assert "coverage:" in text and "no mapping" in text + + +@requires_exar +def test_a_scan_the_template_lacks_is_reported_not_invented() -> None: + """An unmatched scan surfaces rather than being guessed at. + + The PDF names a sequence by kernel and the archive by sequence file, so a + donor to copy cannot be chosen without guessing. The driver says so. + + Returns + ------- + None + """ + from siemens_protocol.exar import build + + archive = read(find_exar("Potpourri_P1.exar1")) + parsed = _parse(find_pdf("Potpourri_P1.pdf")) + parsed["scans"] = list(parsed["scans"]) + [ + {"name": "a_scan_no_template_has", "flat": {"TR": {"value": "1000 ms"}}} + ] + report = build.apply_protocol(archive, parsed) + assert report.unmatched == ["a_scan_no_template_has"]