From f71982b898820f33e495cbcd72768b15082f1011 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 7 Nov 2026 13:02:13 +0000 Subject: [PATCH] Add the description model, canonical writer and section digests Stage 1 design PR 4: Description model built from reader-accepted documents, canonical bytes of experiment-v1 section 9.1, round trip and idempotence on every valid and transition fixture, section digests (v18 = v01; v16 and v21 digests equal v01). Not used by any run yet. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Scma6W3LC11s3qjXHZuGDm --- AGENTS.md | 3 + CHANGELOG.md | 10 + pyproject.toml | 2 + src/morphobricks_preprocessing/description.py | 187 ++++++++++++++++ .../description_write.py | 166 ++++++++++++++ tests/test_description_write.py | 209 ++++++++++++++++++ 6 files changed, 577 insertions(+) create mode 100644 src/morphobricks_preprocessing/description.py create mode 100644 src/morphobricks_preprocessing/description_write.py create mode 100644 tests/test_description_write.py diff --git a/AGENTS.md b/AGENTS.md index f26c1df6..c415933e 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -104,6 +104,9 @@ src/morphobricks_preprocessing/ RGB rules, display names, used_ids, one folder one ID, reference graph; description_warnings.py the three warnings; description_walk.py members/units (read_description_bytes in description_read.py runs phases 1-5 and warnings) + description.py the model (Description, Condition, Channel, Repeat, Member, Unit, Mask) + built from a reader-accepted document; description_from_bytes -> DescriptionError + description_write.py the canonical writer (Spec experiment-v1 9.1) and section digests (9.3) image_type.py image type (P4): FLUORESCENCE/RGB, review_image_type (suggestion from TiffMetadata.rgb_samples and the confirmed roles, whether RGB is possible via the brick's check_rgb), image_type_issues (rgb_not_possible) diff --git a/CHANGELOG.md b/CHANGELOG.md index d1fb4398..a0f325ab 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,16 @@ are in [docs/plan.md](docs/plan.md) §12. Upgrading from v0.3: see the ## Unreleased +- **Experiment-description model, canonical writer and section digests (Stage 1, PR 4).** New + modules `description` (the `Description` model; `description_from_bytes` reads with the strict + reader and raises `DescriptionError` with every issue of the first failing phase) and + `description_write` (`canonical_bytes`, `section_digests`): deterministic bytes of Spec + experiment-v1 section 9.1, the round trip `parse(write(parse(F))) = parse(F)` on every valid and + transition fixture, idempotent writing, and the SHA-256 of the `conditions`, `channels`, + `repeats` and `masks` sections with `display_name`, `target` and `treatment_key` omitted + (`v18` = `v01`; `v16` and `v21` have the digests of `v01`). Section 9.3 does not define the + byte form of one section; here it is the section's table blocks as the file writes them. Not + used by any run, notebook or CLI command yet. - **Experiment-description reader, phases 4-5 and warnings (Stage 1, PR 3).** New modules `description_identity`, `description_semantics`, `description_warnings` and `description_walk`; `description_read.read_description_bytes(data: bytes)` runs all five diff --git a/pyproject.toml b/pyproject.toml index 0b4eb3dd..bd553997 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -173,6 +173,8 @@ baseline = "tools/code_rules_baseline.json" "morphobricks_preprocessing.design" = "io" "morphobricks_preprocessing.design_values" = "pure" "morphobricks_preprocessing.description_identity" = "pure" +"morphobricks_preprocessing.description" = "pure" +"morphobricks_preprocessing.description_write" = "pure" "morphobricks_preprocessing.description_read" = "pure" "morphobricks_preprocessing.description_semantics" = "pure" "morphobricks_preprocessing.description_walk" = "pure" diff --git a/src/morphobricks_preprocessing/description.py b/src/morphobricks_preprocessing/description.py new file mode 100644 index 00000000..d353c84a --- /dev/null +++ b/src/morphobricks_preprocessing/description.py @@ -0,0 +1,187 @@ +"""The experiment description as a value: the model of ``morphobricks.toml`` (experiment-v1). + +The model is built only from a document the strict reader accepted (``description_read``); +nothing here validates, repairs, defaults or converts. Arrays whose order is not semantic +(``header_names``, ``source_channels``, ``used_ids``) are held sorted in code point order, so +two models are equal exactly when the files mean the same (section 9.2); a unit's ``channels`` +keep their declared order. An optional field is ``None`` when the file did not write it, and +an optional array keeps the difference between absent and ``[]``. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass, field +from typing import Any + +from .description_read import read_description_bytes +from .description_report import DescriptionIssue +from .description_rules import SCHEMA_ID, USED_IDS + +Table = Mapping[str, Any] +USED_KINDS = USED_IDS.fields +"""The kinds of ``used_ids``, in the order the file writes them.""" + + +class DescriptionError(Exception): + """Raised when bytes are not a valid description; ``issues`` are the first failing phase's.""" + + def __init__(self, issues: tuple[DescriptionIssue, ...], phase: int) -> None: + self.issues = issues + self.phase = phase + shown = "; ".join(f"{i.code.value} at {i.path}" for i in issues[:5]) + super().__init__(f"not a valid experiment description (phase {phase}): {shown}") + + +@dataclass(frozen=True) +class Condition: + """One experimental condition: its source folder and descriptive or reference fields.""" + + source_folder: str + display_name: str | None = None + reference: str | None = None + target: str | None = None + treatment_key: str | None = None + + +@dataclass(frozen=True) +class Channel: + """One declared channel; ``header_names`` is None when the file wrote none.""" + + display_name: str | None = None + header_names: tuple[str, ...] | None = None + + +@dataclass(frozen=True) +class Unit: + """One imaging unit (a technical repeat of a member); ``channels`` is C0, C1, ... in order.""" + + source_folder: str + channels: tuple[str, ...] + display_name: str | None = None + image_type: str | None = None + + +@dataclass(frozen=True) +class Member: + """One source biological-repeat folder of one condition, with its imaging units.""" + + source_repeat: str + image_count: int + technical_repeats: Mapping[str, Unit] + + +@dataclass(frozen=True) +class Repeat: + """One canonical biological repeat: the user's confirmation and its members by condition.""" + + confirmed: bool + confirmed_on: str + members: Mapping[str, Member] + display_name: str | None = None + + +@dataclass(frozen=True) +class Mask: + """One declared mask: its source type and the representation the user confirmed.""" + + source_type: str + representation: str + source_channels: tuple[str, ...] | None = None + display_name: str | None = None + + +@dataclass(frozen=True) +class Description: + """A whole experiment description. ``masks`` is empty when the file has none.""" + + conditions: Mapping[str, Condition] + channels: Mapping[str, Channel] + repeats: Mapping[str, Repeat] + used_ids: Mapping[str, tuple[str, ...]] + masks: Mapping[str, Mask] = field(default_factory=dict) + schema_version: int = 1 + schema: str = SCHEMA_ID + + +def _sorted(table: Table, key: str) -> tuple[str, ...] | None: + return tuple(sorted(table[key])) if key in table else None + + +def _unit(table: Table) -> Unit: + return Unit( + table["source_folder"], + tuple(table["channels"]), + table.get("display_name"), + table.get("image_type"), + ) + + +def _member(table: Table) -> Member: + units = {key: _unit(value) for key, value in table["technical_repeats"].items()} + return Member(table["source_repeat"], table["image_count"], units) + + +def _repeat(table: Table) -> Repeat: + members = {key: _member(value) for key, value in table["members"].items()} + return Repeat( + table["confirmed"], + table["confirmed_on"], + members, + table.get("display_name"), + ) + + +def from_document(doc: Table) -> Description: + """Return the model of a document that passed phases 1 to 5 of the reader. + + A document the reader refused is not a description: calling this on one may raise + ``KeyError`` or ``TypeError``, or build a model that means nothing. Use + ``description_from_bytes`` to read and refuse in one step. + """ + conditions = { + key: Condition( + value["source_folder"], + value.get("display_name"), + value.get("reference"), + value.get("target"), + value.get("treatment_key"), + ) + for key, value in doc["conditions"].items() + } + channels = { + key: Channel(value.get("display_name"), _sorted(value, "header_names")) + for key, value in doc["channels"].items() + } + masks = { + key: Mask( + value["source_type"], + value["representation"], + _sorted(value, "source_channels"), + value.get("display_name"), + ) + for key, value in doc.get("masks", {}).items() + } + return Description( + conditions, + channels, + {key: _repeat(value) for key, value in doc["repeats"].items()}, + {kind: tuple(sorted(doc["used_ids"][kind])) for kind in USED_KINDS}, + masks, + doc["schema_version"], + doc["schema"], + ) + + +def description_from_bytes(data: bytes) -> Description: + """Read the raw bytes of a file with the strict reader and return its model. + + Raises + ------ + DescriptionError + With every error of the first failing phase, when the bytes are not valid. + """ + report = read_description_bytes(data) + if not report.ok: + raise DescriptionError(report.errors, report.phase) + return from_document(report.document) diff --git a/src/morphobricks_preprocessing/description_write.py b/src/morphobricks_preprocessing/description_write.py new file mode 100644 index 00000000..770fa335 --- /dev/null +++ b/src/morphobricks_preprocessing/description_write.py @@ -0,0 +1,166 @@ +"""The canonical writer of ``morphobricks.toml`` and its section digests (experiment-v1 section 9). + +Writers produce deterministic bytes: UTF-8, no byte-order mark, LF, a final newline, no +comments. Order: ``schema``, ``schema_version``, ``conditions``, ``channels``, ``repeats``, +``masks`` (omitted when empty), ``used_ids``; entries of each table sorted by key in code point +order; fields in the order of the Spec's tables (``description_rules``); every table a +standard ``[header]``; arrays inline; keys bare when they match ``[A-Za-z0-9_-]+``, else +basic quoted strings; strings basic strings with the minimal escapes. Nothing is normalized, +defaulted or added: ``confirmed_on`` is written as the model holds it (section 4.4a). + +A section digest is the SHA-256 of that section's tables written as above, with every +``display_name``, ``target`` and ``treatment_key`` omitted (section 9.3). ``used_ids`` is not a +section. Section 9.3 does not define the byte form of a section alone: here it is exactly the +text the file holds for the section (its table blocks, each followed by a blank line), nothing +else; an empty section is the empty byte string. +""" + +from __future__ import annotations + +import hashlib +import re +from collections.abc import Callable, Iterable, Mapping + +from .description import USED_KINDS, Description +from .description_rules import CHANNEL, CONDITION, MASK, MEMBER, REPEAT, UNIT + +BARE_KEY = re.compile(r"[A-Za-z0-9_-]+") +DESCRIPTIVE = frozenset({"display_name", "target", "treatment_key"}) +"""The fields no digest covers (experiment-v1 sections 4.2a and 9.3).""" +UNORDERED = frozenset({"header_names", "source_channels"}) +"""Arrays whose order is not semantic: always written sorted (section 9.1).""" +SECTIONS = ("conditions", "channels", "repeats", "masks") +_ESCAPES = { + '"': '\\"', + "\\": "\\\\", + "\b": "\\b", + "\t": "\\t", + "\n": "\\n", + "\f": "\\f", + "\r": "\\r", +} + + +def toml_key(key: str) -> str: + """Return a table or dotted key as the canonical writer spells it.""" + return key if BARE_KEY.fullmatch(key) else toml_string(key) + + +def toml_string(text: str) -> str: + """Return a TOML basic string, escaping only what must be (control characters by number).""" + out = [] + for char in text: + if char in _ESCAPES: + out.append(_ESCAPES[char]) + elif ord(char) < 0x20 or ord(char) == 0x7F: + out.append(f"\\u{ord(char):04X}") + else: + out.append(char) + return '"' + "".join(out) + '"' + + +def toml_value(value: object) -> str: + """Return a bool, int, str or list of them as inline TOML.""" + if isinstance(value, bool): + return "true" if value else "false" + if isinstance(value, int): + return str(value) + if isinstance(value, str): + return toml_string(value) + if isinstance(value, (list, tuple)): + return "[" + ", ".join(toml_value(item) for item in value) + "]" + raise TypeError(f"cannot write a {type(value).__name__} to the description") + + +def _block(header: str, item: object, order: Iterable[str], *, descriptive: bool) -> list[str]: + """Return one ``[header]`` block: the fields of ``item`` set (not None), in ``order``.""" + lines = [f"[{header}]"] + for name in order: + value = getattr(item, name, None) + if value is None or isinstance(value, Mapping): # a table is its own block + continue + if not descriptive and name in DESCRIPTIVE: + continue + if name in UNORDERED: + value = sorted(value) + lines.append(f"{name} = {toml_value(value)}") + lines.append("") + return lines + + +def _keyed(base: str, key: str) -> str: + return f"{base}.{toml_key(key)}" + + +def _flat( + base: str, table: Mapping[str, object], order: Iterable[str], descriptive: bool +) -> list[str]: + lines: list[str] = [] + for key in sorted(table): + lines += _block(_keyed(base, key), table[key], order, descriptive=descriptive) + return lines + + +def _repeats(description: Description, descriptive: bool) -> list[str]: + """Return the repeat, member and unit blocks, each level sorted by key.""" + lines: list[str] = [] + for key in sorted(description.repeats): + repeat = description.repeats[key] + head = _keyed("repeats", key) + lines += _block(head, repeat, REPEAT.fields, descriptive=descriptive) + for condition in sorted(repeat.members): + member = repeat.members[condition] + mhead = _keyed(f"{head}.members", condition) + lines += _block(mhead, member, MEMBER.fields, descriptive=descriptive) + lines += _flat( + f"{mhead}.technical_repeats", member.technical_repeats, UNIT.fields, descriptive + ) + return lines + + +def _section(name: str, description: Description, descriptive: bool) -> list[str]: + builders: dict[str, Callable[[], list[str]]] = { + "conditions": lambda: _flat( + "conditions", description.conditions, CONDITION.fields, descriptive + ), + "channels": lambda: _flat("channels", description.channels, CHANNEL.fields, descriptive), + "repeats": lambda: _repeats(description, descriptive), + "masks": lambda: _flat("masks", description.masks, MASK.fields, descriptive), + } + return builders[name]() + + +def _text(lines: list[str]) -> bytes: + return "".join(f"{line}\n" for line in lines).encode("utf-8") + + +def canonical_bytes(description: Description) -> bytes: + """Return the canonical bytes of ``description`` (experiment-v1 section 9.1). + + The writer is given a model, never a document: it writes what the model holds and nothing + else, so a model that the reader would refuse is written as it is (the caller reads the + bytes back with the strict reader before publishing). + """ + lines = [ + f"schema = {toml_string(description.schema)}", + f"schema_version = {description.schema_version}", + "", + ] + for name in SECTIONS: + lines += _section(name, description, True) + lines.append("[used_ids]") + for kind in USED_KINDS: + lines.append(f"{kind} = {toml_value(sorted(description.used_ids[kind]))}") + return _text(lines) + + +def section_digests(description: Description) -> dict[str, str]: + """Return the SHA-256 (hex) of each of ``conditions``, ``channels``, ``repeats``, ``masks``. + + A change to a display name, a target, a treatment key or ``used_ids`` changes none of + them; ``source_folder`` is part of ``conditions`` and ``repeats``. + """ + return { + name: hashlib.sha256(_text(_section(name, description, False))).hexdigest() + for name in SECTIONS + } diff --git a/tests/test_description_write.py b/tests/test_description_write.py new file mode 100644 index 00000000..2ebd3b3f --- /dev/null +++ b/tests/test_description_write.py @@ -0,0 +1,209 @@ +"""The canonical writer, the round trip and the section digests on the Spec fixtures. + +Experiment-v1 section 9: every valid and transition file survives ``parse(write(parse(F)))``, +the writer is idempotent and its bytes follow section 9.1, ``v18`` means what ``v01`` means, +and ``v16`` and ``v21`` have the four section digests of ``v01``. +""" + +from __future__ import annotations + +import re +import tomllib + +import pytest + +from morphobricks_preprocessing.description import ( + Description, + DescriptionError, + description_from_bytes, +) +from morphobricks_preprocessing.description_read import read_description_bytes +from morphobricks_preprocessing.description_write import ( + SECTIONS, + canonical_bytes, + section_digests, + toml_key, + toml_string, +) +from spec_pin import fixtures + +VALID = {item.name.split("_")[0]: item for item in fixtures("valid")} +TRANSITIONS = fixtures("transitions") +ROUND_TRIP = [*fixtures("valid"), *TRANSITIONS] + + +def _model(name: str) -> Description: + return description_from_bytes(VALID[name].data) + + +def _meaning(doc: dict) -> dict: + """The parsed file with unordered arrays sorted and an empty optional table dropped.""" + + def walk(value, key=None): + if isinstance(value, dict): + return {k: walk(v, k) for k, v in value.items()} + if isinstance(value, list) and key in {"header_names", "source_channels", *doc["used_ids"]}: + return sorted(value) + return value + + out = walk(doc) + if out.get("masks") == {}: + del out["masks"] + return out + + +@pytest.mark.parametrize("item", ROUND_TRIP, ids=lambda item: item.name) +def test_parse_of_write_of_parse_is_parse(item): + first = description_from_bytes(item.data) + written = canonical_bytes(first) + assert description_from_bytes(written) == first + assert _meaning(tomllib.loads(written.decode("utf-8"))) == _meaning( + tomllib.loads(item.data.decode("utf-8")) + ) + + +@pytest.mark.parametrize("item", ROUND_TRIP, ids=lambda item: item.name) +def test_the_writer_is_idempotent_and_its_bytes_are_valid(item): + written = canonical_bytes(description_from_bytes(item.data)) + assert canonical_bytes(description_from_bytes(written)) == written + report = read_description_bytes(written) + assert report.errors == () + original = read_description_bytes(item.data) + assert report.warning_codes == original.warning_codes + + +@pytest.mark.parametrize("item", ROUND_TRIP, ids=lambda item: item.name) +def test_the_bytes_follow_the_layout_of_section_9_1(item): + written = canonical_bytes(description_from_bytes(item.data)) + text = written.decode("utf-8") + assert not written.startswith(b"\xef\xbb\xbf") + assert b"\r" not in written and text.endswith("\n") and not text.endswith("\n\n") + assert not [line for line in text.splitlines() if line.lstrip().startswith("#")] + heads = re.findall(r"^\[([^\]]+)\]$", text, flags=re.MULTILINE) + assert heads[-1] == "used_ids" + first = text.splitlines()[:2] + assert first == ['schema = "morphobricks-experiment"', "schema_version = 1"] + order = [head.split(".")[0] for head in heads] + assert order == sorted( + order, key=["conditions", "channels", "repeats", "masks", "used_ids"].index + ) + + +def test_v18_is_v01(): + assert _model("v18") == _model("v01") + assert VALID["v18"].data != VALID["v01"].data # another spelling of the same file + assert canonical_bytes(_model("v18")) == canonical_bytes(_model("v01")) + + +@pytest.mark.parametrize("name", ["v16", "v21"]) +def test_a_display_name_target_or_treatment_key_changes_no_digest(name): + assert _model(name) != _model("v01") + assert section_digests(_model(name)) == section_digests(_model("v01")) + assert canonical_bytes(_model(name)) != canonical_bytes(_model("v01")) + + +def test_the_digests_cover_the_four_sections_and_a_semantic_change_moves_them(): + base = _model("v01") + digests = section_digests(base) + assert tuple(digests) == SECTIONS + assert all(re.fullmatch(r"[0-9a-f]{64}", digest) for digest in digests.values()) + # source_folder is in conditions and repeats; a unit's channels are in repeats only. + folder = Description( + { + k: type(c)(c.source_folder + "x", c.display_name, c.reference) + for k, c in base.conditions.items() + }, + base.channels, + base.repeats, + base.used_ids, + base.masks, + ) + moved = {name for name, digest in section_digests(folder).items() if digest != digests[name]} + assert moved == {"conditions"} + + +def test_a_changed_unit_channel_list_changes_the_repeats_digest_only(): + base = _model("v01") + repeat_id, repeat = next(iter(base.repeats.items())) + condition, member = next(iter(repeat.members.items())) + unit_id, unit = next(iter(member.technical_repeats.items())) + other = type(unit)(unit.source_folder, tuple(reversed(unit.channels)), unit.display_name) + member2 = type(member)( + member.source_repeat, member.image_count, {**member.technical_repeats, unit_id: other} + ) + repeat2 = type(repeat)( + repeat.confirmed, + repeat.confirmed_on, + {**repeat.members, condition: member2}, + repeat.display_name, + ) + changed = Description( + base.conditions, + base.channels, + {**base.repeats, repeat_id: repeat2}, + base.used_ids, + base.masks, + ) + assert len(unit.channels) > 1 + before, after = section_digests(base), section_digests(changed) + assert {name for name in SECTIONS if before[name] != after[name]} == {"repeats"} + + +def test_reference_and_used_ids_are_handled_as_section_9_3_says(): + base = _model("v01") + ref = next(k for k, c in base.conditions.items() if c.reference) + other = { + **base.conditions, + ref: type(base.conditions[ref])(base.conditions[ref].source_folder, None, None), + } + changed = Description(other, base.channels, base.repeats, base.used_ids, base.masks) + after = section_digests(changed) + before = section_digests(base) + assert before["channels"] == after["channels"] and before["repeats"] == after["repeats"] + longer = {**base.used_ids, "channels": (*base.used_ids["channels"], "ZZZ")} + unused = Description(base.conditions, base.channels, base.repeats, longer, base.masks) + assert section_digests(unused) == before + + +def test_confirmed_on_is_written_as_the_model_holds_it(): + base = _model("v01") + repeat_id, repeat = next(iter(base.repeats.items())) + stamped = type(repeat)(repeat.confirmed, "1999-12-31", repeat.members, repeat.display_name) + text = canonical_bytes( + Description( + base.conditions, + base.channels, + {**base.repeats, repeat_id: stamped}, + base.used_ids, + base.masks, + ) + ).decode("utf-8") + assert 'confirmed_on = "1999-12-31"' in text and "1999-12-31" in text.split("[used_ids]")[0] + + +def test_a_refused_file_is_not_a_model(): + with pytest.raises(DescriptionError) as error: + description_from_bytes(b"schema = 1\n") + assert error.value.issues and error.value.phase == 2 + + +@pytest.mark.parametrize( + ("key", "spelled"), + [ + ("DNA", "DNA"), + ("a-b_c1", "a-b_c1"), + ("DAPI 405", '"DAPI 405"'), + ("é", '"é"'), + ('a"b', '"a\\"b"'), + ], +) +def test_keys_are_bare_when_they_can_be(key, spelled): + assert toml_key(key) == spelled + + +def test_strings_use_the_minimal_escapes(): + assert toml_string('a"b\\c\n\t\x01\x7f é') == '"a\\"b\\\\c\\n\\t\\u0001\\u007F é"' + assert ( + tomllib.loads(f"x = {toml_string(chr(1) + chr(127) + chr(8) + chr(12))}")["x"] + == "\x01\x7f\b\f" + )