Description model, canonical writer and section digests (Stage 1, PR 4)

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)

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.
 - **Spec pin moved to MorphoBricks-Spec `745259b` (Stage 1, PR A).** The vendored fixtures,
   `experiment-v1.md` and the reference validator are the Spec's at that commit (new fixtures
   `i131`-`i134`; 36 valid, 134 invalid, 12 transition). The reader now applies the condition

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"

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)

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
+    }