From 3a03180a6fd07ef12ad3fc498c14b0b89466c329 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 7 Nov 2026 14:07:52 +0000 Subject: [PATCH 1/2] Add the physical R_* checks of an experiment description (Stage 1, PR 8) check_physical compares a description with the folder and the image headers: units found and declared, per-unit homogeneity (replacing the count and header checks of its unit), header evidence on raw names, image type against the header, and declared mask types. Pixels are never read; undecidable images are listed as unchecked. layout_groups marks which layouts can be confirmed once (header-named only). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Scma6W3LC11s3qjXHZuGDm --- AGENTS.md | 3 + CHANGELOG.md | 14 + pyproject.toml | 1 + .../physical_check.py | 376 +++++++++++++++++ tests/test_physical_check.py | 390 ++++++++++++++++++ 5 files changed, 784 insertions(+) create mode 100644 src/morphobricks_preprocessing/physical_check.py create mode 100644 tests/test_physical_check.py diff --git a/AGENTS.md b/AGENTS.md index 518c5586..3dee961c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -107,6 +107,9 @@ src/morphobricks_preprocessing/ 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) + physical_check.py the physical R_* checks of a description against the folder and image headers + (units found/declared, homogeneity, header evidence, image type, mask types + declared); layout_groups (header-named layouts are batch-confirmable, Q3) description_transition.py the R_* comparison of a candidate with the published file (4.4a, 4.6, 4.9), Reconfirmation (the user's recorded acceptance of a changed ID) description_file.py read_description(path): symlink refusal (R_FILE_SYMLINK), bytes -> Description diff --git a/CHANGELOG.md b/CHANGELOG.md index 80f42cdb..5f242ab8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,20 @@ are in [docs/plan.md](docs/plan.md) §12. Upgrading from v0.3: see the ## Unreleased +- **Physical `R_*` checks of an experiment description (Stage 1, PR 8; stacked on the + discovery and header additions, which provide the raw header names).** New module + `physical_check`: `check_physical(description, discovery, metadata, ...)` reports + `R_UNIT_NOT_FOUND`, `R_UNIT_NOT_DECLARED`, `R_UNIT_NOT_HOMOGENEOUS` (names the FOV, path and C + index; it replaces the count and header checks of its unit), `R_CHANNEL_COUNT_MISMATCH`, + `R_HEADER_CONTRADICTION` (raw names, NFC of both sides, no trimming, no case folding), + `R_IMAGE_TYPE_CONTRADICTS_HEADER` (both directions, header only; OME-Zarr states no colour + interpretation and is recorded as not header-verified) and `R_MASK_TYPE_NOT_DECLARED`. Pixels + are never read and no mask representation is defaulted or inferred. Homogeneity compares the + axes with `T` left out, so time-lapses of different lengths are one layout. An image whose + channel count the header does not state (a plain stack without confirmed roles) or whose header + cannot be read is listed in `unchecked`, not passed. `layout_groups` groups units by identical + layout; only groups whose channels are all header-named are batch-confirmable. Not used by any + run, notebook or CLI command yet. - **Discovery and header additions for the experiment description (Stage 1, PR 6).** Additive; a run without a description behaves as before. (1) `preflight` and `remasked` take `with_description=False`; with it, a condition, repeat or technical-repeat folder that diff --git a/pyproject.toml b/pyproject.toml index 9e32ea12..ce059f5a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -176,6 +176,7 @@ baseline = "tools/code_rules_baseline.json" "morphobricks_preprocessing.design" = "io" "morphobricks_preprocessing.design_values" = "pure" "morphobricks_preprocessing.description_identity" = "pure" +"morphobricks_preprocessing.physical_check" = "io" "morphobricks_preprocessing.description" = "pure" "morphobricks_preprocessing.description_write" = "pure" "morphobricks_preprocessing.description_transition" = "pure" diff --git a/src/morphobricks_preprocessing/physical_check.py b/src/morphobricks_preprocessing/physical_check.py new file mode 100644 index 00000000..5c0224c1 --- /dev/null +++ b/src/morphobricks_preprocessing/physical_check.py @@ -0,0 +1,376 @@ +"""The physical ``R_*`` checks of an experiment description against the folder (experiment-v1 §8). + +What the reader cannot see from one file: do the declared imaging units exist, is each unit +homogeneous, do the image headers agree with the declared channels and image type, are the +mask types declared. Everything is decided from the image **headers** (and the folder walk): +pixels are never examined, nothing is guessed, trimmed or repaired, and a check that cannot be +decided is listed as unchecked instead of passing. + +Order within one unit (§8): ``R_UNIT_NOT_HOMOGENEOUS`` is reported first and replaces +``R_CHANNEL_COUNT_MISMATCH`` and ``R_HEADER_CONTRADICTION`` for that unit. An image type that +contradicts the header also replaces them for its unit: the layout cannot be read until the +declaration is right. Homogeneity compares the interpreted axes with ``T`` left out (a +time-lapse may have any number of time points, owner decision Q7), the channel count, the image +type and the raw header name at each C index (equal after NFC, never trimmed). +""" + +from __future__ import annotations + +import unicodedata +from collections import Counter +from collections.abc import Collection, Iterable, Mapping, Sequence +from dataclasses import dataclass +from types import MappingProxyType + +from .axis_names import OPEN_AXIS +from .description import Description, Unit +from .discovery import DiscoveryResult +from .image_names import position_id +from .metadata import OME_ZARR, TiffMetadata, header_channel_count, named_axes + +UNIT_NOT_FOUND = "R_UNIT_NOT_FOUND" +UNIT_NOT_DECLARED = "R_UNIT_NOT_DECLARED" +CHANNEL_COUNT_MISMATCH = "R_CHANNEL_COUNT_MISMATCH" +UNIT_NOT_HOMOGENEOUS = "R_UNIT_NOT_HOMOGENEOUS" +HEADER_CONTRADICTION = "R_HEADER_CONTRADICTION" +IMAGE_TYPE_CONTRADICTS_HEADER = "R_IMAGE_TYPE_CONTRADICTS_HEADER" +MASK_TYPE_NOT_DECLARED = "R_MASK_TYPE_NOT_DECLARED" +PHYSICAL_CODES = frozenset( + { + UNIT_NOT_FOUND, + UNIT_NOT_DECLARED, + CHANNEL_COUNT_MISMATCH, + UNIT_NOT_HOMOGENEOUS, + HEADER_CONTRADICTION, + IMAGE_TYPE_CONTRADICTS_HEADER, + MASK_TYPE_NOT_DECLARED, + } +) +"""The codes of this module (the rest of §8's ``R_*`` belong to publication and the run).""" + +RGB = "rgb" + + +@dataclass(frozen=True) +class UnitKey: + """An imaging unit by its source folder labels (never by canonical IDs).""" + + condition: str + biological_repeat: str + technical_repeat: str + + +@dataclass(frozen=True) +class PhysicalIssue: + """One refusal: its code, the unit, and where it was found. + + ``path`` is the image (relative to the experiment root) and ``fov`` its position ID; + ``c_index`` the source channel index, None when the issue is not about one channel. For + ``R_MASK_TYPE_NOT_DECLARED`` ``unit`` is None and ``detail`` is the mask type. + """ + + code: str + unit: UnitKey | None + path: str | None = None + fov: str | None = None + c_index: int | None = None + detail: str = "" + + +@dataclass(frozen=True) +class Layout: + """What one image's header says about its channels: the unit's signature when all agree. + + ``channels`` is None when the header does not name the channel axis and no confirmed + roles were given, so the count is not known. ``names`` are the raw header names by C + index, as read (None where there is none). + """ + + channels: int | None + names: tuple[str | None, ...] + rgb: bool + axes: str + + @property + def named(self) -> bool: + """True when every channel has a header name.""" + return bool(self.channels) and len(self.names) == self.channels and all(self.names) + + +@dataclass(frozen=True) +class LayoutGroup: + """Units with one identical layout, for the channel question (owner decision Q3). + + A group whose channels all have header names is confirmed once for all its units. A group + without header names, or with some missing, is **not** batch-confirmable: each unit needs + its own confirmation beside its own preview. + """ + + layout: Layout + units: tuple[UnitKey, ...] + + @property + def batch_confirmable(self) -> bool: + """True only for a layout whose channels are all named by the header.""" + return self.layout.named + + +@dataclass(frozen=True) +class PhysicalReport: + """The outcome of the physical checks. + + ``issues`` are the refusals. ``layouts`` holds the common layout of each unit that is + homogeneous and whose image type agrees with its declaration. ``image_type_verified`` + says per unit whether the declared image type was checked against a header that states a + colour interpretation (TIFF), False for a format that states none (OME-Zarr: it stands on + the user's confirmation). ``unchecked`` lists images whose header could not be read or + whose channel count is not stated by the header: they were not checked, which is not a pass. + """ + + issues: tuple[PhysicalIssue, ...] = () + layouts: Mapping[UnitKey, Layout] = MappingProxyType({}) + image_type_verified: Mapping[UnitKey, bool] = MappingProxyType({}) + unchecked: tuple[str, ...] = () + + @property + def codes(self) -> frozenset[str]: + """The set of codes reported (what conformance compares).""" + return frozenset(issue.code for issue in self.issues) + + @property + def ok(self) -> bool: + """True when nothing is refused. Unchecked images are listed in ``unchecked``.""" + return not self.issues + + +def nfc(text: str) -> str: + """Return the NFC form of ``text``, a comparison key only (never stored).""" + return unicodedata.normalize("NFC", text) + + +def _name(name: str | None) -> str | None: + return nfc(name) if name else None + + +def layout_of(meta: TiffMetadata, roles: Sequence[str | None] | None = None) -> Layout: + """Return the layout one readable image's header states. + + The channel count is the length of the confirmed ``C`` role when ``roles`` are given, else + what the header names (the ``C`` axis, or the samples of an RGB image); None when the header + leaves an axis open and no roles say which dimension is the channel. + """ + axes = named_axes(meta).replace("T", "") + if meta.rgb_samples: + count: int | None = header_channel_count(meta) + elif roles is not None: + count = dict(zip(roles, meta.shape, strict=True)).get("C", 1) + elif OPEN_AXIS in named_axes(meta): + count = None + else: + count = header_channel_count(meta) + return Layout(count, tuple(meta.raw_channel_names), meta.rgb_samples, axes) + + +def _units(description: Description) -> dict[UnitKey, tuple[str, Unit]]: + """Return the declared units by source folders, with the repeat ID they belong to.""" + found: dict[UnitKey, tuple[str, Unit]] = {} + for repeat_id, repeat in description.repeats.items(): + for condition_id, member in repeat.members.items(): + folder = description.conditions[condition_id].source_folder + for unit in member.technical_repeats.values(): + found[UnitKey(folder, member.source_repeat, unit.source_folder)] = ( + repeat_id, + unit, + ) + return found + + +def _discovered(discovery: DiscoveryResult, excluded: Collection[str]) -> dict[UnitKey, list[str]]: + found: dict[UnitKey, list[str]] = {} + for leaf in discovery.leaves: + included = [image for image in leaf.images if image not in excluded] + if included: # a folder whose images are all excluded is not a unit + found.setdefault(UnitKey(*leaf.labels), []).extend(included) + return found + + +def _differences(reference: Layout, other: Layout) -> list[tuple[int | None, str]]: + """Return ``(C index or None, what differs)`` of ``other`` from ``reference``.""" + out: list[tuple[int | None, str]] = [] + if reference.channels != other.channels: + out.append((None, f"{other.channels} channels, the unit has {reference.channels}")) + if reference.rgb != other.rgb: + out.append((None, "image type (colour interpretation) differs")) + if reference.axes != other.axes: + out.append((None, f"axes {other.axes!r}, the unit has {reference.axes!r}")) + for index in range(max(len(reference.names), len(other.names))): + ours = reference.names[index] if index < len(reference.names) else None + theirs = other.names[index] if index < len(other.names) else None + if _name(ours) != _name(theirs): + out.append((index, f"header name {theirs!r}, the unit has {ours!r}")) + return out + + +def _homogeneity(key: UnitKey, layouts: Mapping[str, Layout]) -> tuple[Layout, list[PhysicalIssue]]: + """Return the unit's common layout and one issue per image and difference from it. + + The common layout is the most frequent one (the first in path order on a tie). + """ + counts = Counter(layouts.values()) + top = max(counts.values()) + reference = next(layout for layout in layouts.values() if counts[layout] == top) + issues = [ + PhysicalIssue(UNIT_NOT_HOMOGENEOUS, key, path, position_id(path), index, what) + for path, layout in layouts.items() + for index, what in _differences(reference, layout) + ] + return reference, issues + + +def _type_issues( + key: UnitKey, declared: Unit, layouts: Mapping[str, Layout], metas: Mapping[str, TiffMetadata] +) -> list[PhysicalIssue]: + """Compare the declared image type with each header that states a colour interpretation.""" + return [ + PhysicalIssue( + IMAGE_TYPE_CONTRADICTS_HEADER, + key, + path, + position_id(path), + None, + f"declared {declared.image_type or 'not rgb'}, the header says " + f"{'rgb' if layout.rgb else 'not rgb'}", + ) + for path, layout in layouts.items() + if metas[path].kind != OME_ZARR and layout.rgb != (declared.image_type == RGB) + ] + + +def _evidence_issues( + key: UnitKey, unit: Unit, description: Description, layouts: Mapping[str, Layout] +) -> list[PhysicalIssue]: + """Return ``R_CHANNEL_COUNT_MISMATCH`` and ``R_HEADER_CONTRADICTION`` for a unit.""" + out = [] + for path, layout in layouts.items(): + fov = position_id(path) + if layout.channels is not None and layout.channels != len(unit.channels): + what = f"{layout.channels} channels, the description lists {len(unit.channels)}" + out.append(PhysicalIssue(CHANNEL_COUNT_MISMATCH, key, path, fov, None, what)) + continue + if unit.image_type == RGB: # R, G and B take no header names (§4.4b) + continue + for index, (channel, name) in enumerate(zip(unit.channels, layout.names, strict=False)): + known = description.channels[channel].header_names or () + if name and nfc(name) not in {nfc(entry) for entry in known}: + what = f"header name {name!r} is not a confirmed name of channel {channel!r}" + out.append(PhysicalIssue(HEADER_CONTRADICTION, key, path, fov, index, what)) + return out + + +def _check_unit( + key: UnitKey, + repeat_unit: tuple[str, Unit], + description: Description, + paths: Sequence[str], + metas: Mapping[str, TiffMetadata], + roles: Mapping[str, Sequence[str | None]], +) -> tuple[list[PhysicalIssue], Layout | None, bool, list[str]]: + """Check one declared unit that has images; return issues, layout, verified, unchecked.""" + _, unit = repeat_unit + readable = [p for p in paths if p in metas and metas[p].readable] + layouts = {p: layout_of(metas[p], roles.get(p)) for p in readable} + unchecked = [p for p in paths if p not in layouts] + verified = bool(readable) and all(metas[p].kind != OME_ZARR for p in readable) + if not layouts: + return [], None, verified, unchecked + reference, issues = _homogeneity(key, layouts) + if issues: + return issues, None, verified, unchecked + issues = _type_issues(key, unit, layouts, metas) + if issues: + return issues, None, verified, unchecked + issues = _evidence_issues(key, unit, description, layouts) + if reference.channels is None: # the header does not say; never a guessed count + unchecked += [p for p in layouts if p not in unchecked] + return issues, None if issues else reference, verified, unchecked + + +def check_physical( + description: Description, + discovery: DiscoveryResult, + metadata: Iterable[TiffMetadata], + *, + excluded: Iterable[str] = (), + roles: Mapping[str, Sequence[str | None]] | None = None, + mask_types: Iterable[str] | None = None, +) -> PhysicalReport: + """Check a description against the experiment folder and the image headers. + + Parameters + ---------- + description + A description the strict reader accepted. + discovery + The folder walk (:func:`.discovery.discover_experiment`). + metadata + The image headers (:func:`.metadata.inspect_images`), with raw channel names. + excluded + Images the user left out; they belong to no unit. + roles + The confirmed role of each axis by relative path (:attr:`.axes.AxisReview.file_mappings`); + without them the channel count is read only from headers that name every axis. + mask_types + The mask types discovered in the masks folder, or None when masks were not scanned + (then ``R_MASK_TYPE_NOT_DECLARED`` is not examined). A declared mask with no files is + valid, and nothing here reads mask pixels (§4.5a). + + Returns + ------- + PhysicalReport + Every refusal (the unit's ``R_UNIT_NOT_HOMOGENEOUS`` first, replacing the count and + header checks of that unit), the common layout of each clean unit, and what was not + checked. + """ + metas = {m.relative_path: m for m in metadata} + declared = _units(description) + found = _discovered(discovery, set(excluded)) + issues = [PhysicalIssue(UNIT_NOT_FOUND, key) for key in declared if key not in found] + issues += [PhysicalIssue(UNIT_NOT_DECLARED, key) for key in found if key not in declared] + layouts: dict[UnitKey, Layout] = {} + verified: dict[UnitKey, bool] = {} + unchecked: list[str] = [] + for key, paths in found.items(): + if key not in declared: + continue + found_issues, layout, ok, missing = _check_unit( + key, declared[key], description, sorted(paths), metas, roles or {} + ) + issues += found_issues + unchecked += missing + verified[key] = ok + if layout is not None: + layouts[key] = layout + if mask_types is not None: + known = {mask.source_type for mask in description.masks.values()} + issues += [ + PhysicalIssue(MASK_TYPE_NOT_DECLARED, None, detail=name) + for name in sorted(set(mask_types) - known) + ] + return PhysicalReport( + tuple(issues), + MappingProxyType(layouts), + MappingProxyType(verified), + tuple(dict.fromkeys(unchecked)), + ) + + +def layout_groups(layouts: Mapping[UnitKey, Layout]) -> tuple[LayoutGroup, ...]: + """Group units by identical layout (channel count, raw header names, image type, axes). + + Header names are compared exactly as read: ``DAPI`` and ``dapi`` are two layouts. Only a + group whose channels are all header-named is batch-confirmable (:class:`LayoutGroup`, Q3). + """ + groups: dict[Layout, list[UnitKey]] = {} + for key, layout in sorted(layouts.items(), key=lambda item: tuple(vars(item[0]).values())): + groups.setdefault(layout, []).append(key) + return tuple(LayoutGroup(layout, tuple(keys)) for layout, keys in groups.items()) diff --git a/tests/test_physical_check.py b/tests/test_physical_check.py new file mode 100644 index 00000000..6f538eaa --- /dev/null +++ b/tests/test_physical_check.py @@ -0,0 +1,390 @@ +"""The physical ``R_*`` checks on the Spec fixtures: conformance P01-P03, P10, P12, P13, P23, P26. + +Experiment-v1 section 8. The folder is built from a vendored valid fixture (one small image +per unit position, header names taken from the fixture's confirmed names), then one image or +one declaration is changed. The owner's answers: Q3 (no batch confirmation of unnamed +layouts), Q5 (no default mask representation: nothing here reads or infers one), Q7 +(time-lapses are allowed; homogeneity compares axis semantics, not T counts). +""" + +from dataclasses import replace + +import numpy as np +import pytest +import tifffile + +from morphobricks_preprocessing.description import Description, description_from_bytes +from morphobricks_preprocessing.discovery import discover_experiment +from morphobricks_preprocessing.metadata import OME_ZARR, inspect_images +from morphobricks_preprocessing.physical_check import ( + CHANNEL_COUNT_MISMATCH, + HEADER_CONTRADICTION, + IMAGE_TYPE_CONTRADICTS_HEADER, + MASK_TYPE_NOT_DECLARED, + PHYSICAL_CODES, + UNIT_NOT_DECLARED, + UNIT_NOT_FOUND, + UNIT_NOT_HOMOGENEOUS, + UnitKey, + check_physical, + layout_groups, +) +from spec_pin import fixtures + +VALID = {item.name.split("_")[0]: item for item in fixtures("valid")} +DECOMPOSED = "Café" +COMPOSED = "Café" + + +def model(name): + return description_from_bytes(VALID[name].data) + + +def write_plane(path, names, *, rgb=False, shape=None, axes=None): + """Write one small image: an OME file with channel ``names`` (None: no names), or RGB.""" + path.parent.mkdir(parents=True, exist_ok=True) + if rgb: + tifffile.imwrite(path, np.zeros((6, 7, 3), np.uint8), photometric="rgb") + elif names is None: + count = shape or 2 + tifffile.imwrite( + path, + np.zeros((count, 6, 7), np.uint16), + photometric="minisblack", + metadata={"axes": "CYX"}, + ) + else: + tifffile.imwrite( + path, + np.zeros((*(shape or (len(names),)), 6, 7), np.uint16), + ome=True, + photometric="minisblack", + metadata={"axes": axes or "CYX", "Channel": {"Name": list(names)}}, + ) + return path + + +def confirmed_names(description: Description, unit): + """The first confirmed header name of each channel of the unit (None if it has none).""" + names = [(description.channels[c].header_names or (None,))[0] for c in unit.channels] + return None if None in names else names + + +def build(root, description, *, images=2): + """Write every unit of the description as a folder of ``images`` conforming images.""" + for repeat in description.repeats.values(): + for condition_id, member in repeat.members.items(): + folder = description.conditions[condition_id].source_folder + for unit in member.technical_repeats.values(): + names = confirmed_names(description, unit) + parts = [ + p for p in (folder, member.source_repeat, unit.source_folder) if p != "single" + ] + for index in range(1, images + 1): + path = root.joinpath(*parts, f"FOV{index:03d}.ome.tif") + write_plane(path, names, rgb=unit.image_type == "rgb") + return root + + +def check(root, description, *, excluded=(), **kwargs): + discovery = discover_experiment(root) + files = inspect_images(root, discovery.images).files + return check_physical(description, discovery, files, excluded=excluded, **kwargs) + + +def issues_of(report, code): + return [i for i in report.issues if i.code == code] + + +def unit_dir(root, *parts): + return root.joinpath(*parts) + + +@pytest.fixture +def v01(tmp_path): + description = model("v01") + return build(tmp_path / "exp", description), description + + +def test_a_folder_that_matches_its_description_passes(v01): + root, description = v01 + report = check(root, description) + assert report.ok and not report.unchecked + assert report.issues == () + layout = report.layouts[UnitKey("CTRL", "rep1", "single")] + assert (layout.channels, layout.names, layout.rgb) == (2, ("DAPI", "Phalloidin-488"), False) + assert report.image_type_verified[UnitKey("CTRL", "rep1", "single")] is True + + +def test_the_codes_are_the_ones_of_section_8_and_only_those(v01): + root, description = v01 + assert PHYSICAL_CODES == { + "R_UNIT_NOT_FOUND", + "R_UNIT_NOT_DECLARED", + "R_CHANNEL_COUNT_MISMATCH", + "R_UNIT_NOT_HOMOGENEOUS", + "R_HEADER_CONTRADICTION", + "R_IMAGE_TYPE_CONTRADICTS_HEADER", + "R_MASK_TYPE_NOT_DECLARED", + } + + +# P01, P02 ------------------------------------------------------------------------------ + + +def test_p01_an_image_with_another_channel_order_is_not_homogeneous(v01): + root, description = v01 + write_plane(root / "CTRL" / "rep1" / "FOV003.ome.tif", ["Phalloidin-488", "DAPI"]) + report = check(root, description) + found = issues_of(report, UNIT_NOT_HOMOGENEOUS) + assert report.codes == {UNIT_NOT_HOMOGENEOUS} # replaces the header contradiction + assert {(i.fov, i.path, i.c_index) for i in found} == { + ("FOV003", "CTRL/rep1/FOV003.ome.tif", 0), + ("FOV003", "CTRL/rep1/FOV003.ome.tif", 1), + } + assert UnitKey("CTRL", "rep1", "single") not in report.layouts # the unit is blocked + + +def test_p02_an_image_with_one_channel_is_not_homogeneous_not_a_count_mismatch(v01): + root, description = v01 + write_plane(root / "CTRL" / "rep1" / "FOV004.ome.tif", ["DAPI"]) + report = check(root, description) + assert report.codes == {UNIT_NOT_HOMOGENEOUS} + assert {i.fov for i in report.issues} == {"FOV004"} + + +def test_a_whole_unit_with_the_wrong_count_is_a_count_mismatch(v01): + root, description = v01 + for path in (root / "CTRL" / "rep1").glob("*.tif"): + write_plane(path, ["DAPI", "Phalloidin-488", "Extra"]) + report = check(root, description) + assert report.codes == {CHANNEL_COUNT_MISMATCH} + assert {i.unit for i in report.issues} == {UnitKey("CTRL", "rep1", "single")} + + +# P03, P26 ------------------------------------------------------------------------------ + + +def rewrite_unit(folder, names): + for path in folder.glob("*.tif"): + write_plane(path, names) + + +@pytest.mark.parametrize("name", ["hoechst 33342", "Hoechst 33342"]) +def test_p03_no_case_folding_and_no_similarity(tmp_path, name): + description = model("v11") + root = build(tmp_path / "exp", description) + rewrite_unit(root / "CTRL" / "rep1" / "acquisition_1", [name, "ACTIN"]) + report = check(root, description) + if name == "Hoechst 33342": + assert report.ok + else: + found = issues_of(report, HEADER_CONTRADICTION) + assert {(i.c_index, i.unit) for i in found} == { + (0, UnitKey("CTRL", "rep1", "acquisition_1")) + } + assert report.codes == {HEADER_CONTRADICTION} + + +def test_p26_outer_whitespace_is_not_trimmed_against_v11_but_a_confirmed_alias_passes_v31(tmp_path): + for fixture, expected in (("v11", {HEADER_CONTRADICTION}), ("v31", set())): + description = model(fixture) + root = build(tmp_path / fixture, description) + for folder in (root / "CTRL").rglob("rep1"): + rewrite_unit(folder, [" DAPI", "ACTIN" if fixture == "v11" else f"{DECOMPOSED} 488"]) + for sub in folder.iterdir(): + if sub.is_dir(): + rewrite_unit(sub, [" DAPI", "ACTIN"]) + assert check(root, description).codes == expected + + +@pytest.mark.parametrize(("header", "entry"), [(DECOMPOSED, COMPOSED), (COMPOSED, DECOMPOSED)]) +def test_p26_header_names_match_after_nfc_of_both_sides(tmp_path, header, entry): + description = model("v31") + channels = dict(description.channels) + channels["ACTIN"] = replace(channels["ACTIN"], header_names=(f"{entry} 488",)) + description = replace(description, channels=channels) + root = build(tmp_path / "exp", description) + rewrite_unit(root / "CTRL" / "rep1", ["DAPI", f"{header} 488"]) + report = check(root, description) + assert report.ok + assert report.layouts[UnitKey("CTRL", "rep1", "single")].names[1] == f"{header} 488" # as read + + +def test_an_image_with_names_when_the_channel_has_none_contradicts(tmp_path): + description = model("v12") + root = build(tmp_path / "exp", description) + assert check(root, description).ok # unnamed images need no evidence + first = next(iter(sorted(root.rglob("*.tif")))) + names = ["DAPI", "Phalloidin"] + for path in first.parent.glob("*.tif"): + write_plane(path, names) + assert check(root, description).codes == {HEADER_CONTRADICTION} + + +# P10, P12 ------------------------------------------------------------------------------ + + +def test_p10_an_undeclared_mask_type_is_refused_and_a_declared_empty_one_is_valid(tmp_path): + description = model("v07") + root = build(tmp_path / "exp", description) + declared = sorted({m.source_type for m in description.masks.values()}) + assert declared + report = check(root, description, mask_types=[*declared, "membrane"]) + (found,) = issues_of(report, MASK_TYPE_NOT_DECLARED) + assert found.detail == "membrane" and found.unit is None + assert check(root, description, mask_types=[]).ok # a declared mask with no files is valid + assert check(root, description, mask_types=declared).ok + assert check(root, description).ok # masks not scanned: not examined, never a default + + +def test_p12_a_declared_unit_with_every_image_excluded_is_not_found(v01): + root, description = v01 + images = [p.relative_to(root).as_posix() for p in (root / "DrugA" / "rep2").glob("*.tif")] + report = check(root, description, excluded=images) + assert report.codes == {UNIT_NOT_FOUND} + assert issues_of(report, UNIT_NOT_FOUND)[0].unit == UnitKey("DrugA", "rep2", "single") + + +def test_a_discovered_unit_that_is_not_in_the_file_is_not_declared(v01): + root, description = v01 + write_plane(root / "DrugA" / "rep3" / "FOV001.ome.tif", ["DAPI", "Phalloidin-488"]) + report = check(root, description) + assert report.codes == {UNIT_NOT_DECLARED} + assert issues_of(report, UNIT_NOT_DECLARED)[0].unit == UnitKey("DrugA", "rep3", "single") + + +# P13, P23 ------------------------------------------------------------------------------ + + +def test_p13_rgb_units_declared_rgb_match_the_header_and_are_header_verified(tmp_path): + description = model("v28") + root = build(tmp_path / "exp", description) + report = check(root, description) + assert report.ok + assert all(report.image_type_verified.values()) and report.image_type_verified + layout = report.layouts[UnitKey("CTRL", "rep1", "single")] + assert (layout.channels, layout.rgb) == (3, True) + + +def test_p23_a_unit_declared_rgb_whose_images_are_two_channel_grayscale(tmp_path): + description = model("v28") + root = build(tmp_path / "exp", description) + for path in (root / "CTRL" / "rep1").glob("*.tif"): + write_plane(path, None) + report = check(root, description) + assert report.codes == {IMAGE_TYPE_CONTRADICTS_HEADER} + assert {i.unit for i in report.issues} == {UnitKey("CTRL", "rep1", "single")} + + +def test_p23_an_rgb_tiff_in_a_unit_that_does_not_declare_it(v01): + root, description = v01 + for path in (root / "CTRL" / "rep1").glob("*.tif"): + write_plane(path, None, rgb=True) + assert check(root, description).codes == {IMAGE_TYPE_CONTRADICTS_HEADER} + + +def test_p23_pixels_that_look_like_rgb_are_never_examined(tmp_path): + description = model("v28") + description = replace( + description, + repeats={ + key: replace( + repeat, + members={ + c: replace( + m, + technical_repeats={ + k: replace(u, image_type=None, channels=("R", "G", "B")) + for k, u in m.technical_repeats.items() + }, + ) + for c, m in repeat.members.items() + }, + ) + for key, repeat in description.repeats.items() + }, + ) + root = tmp_path / "exp" + for path in ("CTRL/rep1/a.tif", "DrugA/rep1/a.tif", "CTRL/rep2/a.tif", "DrugA/rep2/a.tif"): + data = np.zeros((3, 6, 7), np.uint8) + data[0], data[1], data[2] = 255, 128, 0 # looks like colour + (root / path).parent.mkdir(parents=True, exist_ok=True) + tifffile.imwrite(root / path, data, photometric="minisblack", metadata={"axes": "CYX"}) + assert check(root, description).ok + + +def test_a_format_without_a_colour_interpretation_is_recorded_as_not_header_verified(tmp_path): + description = model("v28") + root = build(tmp_path / "exp", description) + discovery = discover_experiment(root) + files = [ + replace(f, kind=OME_ZARR, rgb_samples=False) + for f in inspect_images(root, discovery.images).files + ] + report = check_physical(description, discovery, files) + assert IMAGE_TYPE_CONTRADICTS_HEADER not in report.codes # it cannot contradict + assert not any(report.image_type_verified.values()) + + +# Q7, unchecked ------------------------------------------------------------------------- + + +def test_q7_time_lapses_with_different_numbers_of_time_points_are_homogeneous(tmp_path): + description = model("v11") + root = build(tmp_path / "exp", description) + folder = root / "CTRL" / "rep1" / "acquisition_1" + names = ["DAPI", "ACTIN"] + write_plane(folder / "FOV001.ome.tif", names, shape=(3, 2), axes="TCYX") + write_plane(folder / "FOV002.ome.tif", names, shape=(5, 2), axes="TCYX") + write_plane(folder / "FOV003.ome.tif", names) # one time point, no T axis + report = check(root, description) + assert UNIT_NOT_HOMOGENEOUS not in report.codes + assert report.ok + + +def test_an_unreadable_image_and_an_unnamed_axis_are_listed_not_passed(tmp_path): + description = model("v11") + root = build(tmp_path / "exp", description) + folder = root / "CTRL" / "rep1" / "acquisition_1" + for path in folder.glob("*.tif"): + path.unlink() + (folder / "FOV009.tif").write_bytes(b"not an image") + tifffile.imwrite( + folder / "FOV010.tif", np.zeros((2, 6, 7), np.uint16), photometric="minisblack" + ) + report = check(root, description) + assert "CTRL/rep1/acquisition_1/FOV009.tif" in report.unchecked + assert "CTRL/rep1/acquisition_1/FOV010.tif" in report.unchecked # plain stack: no roles yet + roles = {"CTRL/rep1/acquisition_1/FOV010.tif": ("C", "Y", "X")} + with_roles = check(root, description, roles=roles) + assert "CTRL/rep1/acquisition_1/FOV010.tif" not in with_roles.unchecked + + +# Q3 ------------------------------------------------------------------------------------- + + +def test_q3_unnamed_layouts_are_not_batch_confirmable(tmp_path): + description = model("v11") + root = build(tmp_path / "exp", description) # two named units, same names + report = check(root, description) + (group,) = layout_groups(report.layouts) + assert len(group.units) == 2 and group.batch_confirmable + for path in (root / "CTRL" / "rep1" / "acquisition_2").glob("*.tif"): + write_plane(path, None) + other = check(root, description) + groups = layout_groups(other.layouts) + assert [g.batch_confirmable for g in groups] == [False, True] or [ + g.batch_confirmable for g in groups + ] == [True, False] + unnamed = next(g for g in groups if not g.batch_confirmable) + assert unnamed.units == (UnitKey("CTRL", "rep1", "acquisition_2"),) + assert not unnamed.layout.named + + +def test_header_names_group_exactly_as_read(tmp_path): + description = model("v11") + root = build(tmp_path / "exp", description) + rewrite_unit(root / "CTRL" / "rep1" / "acquisition_2", ["Hoechst 33342", "ACTIN"]) + groups = layout_groups(check(root, description).layouts) + assert len(groups) == 2 and all(g.batch_confirmable for g in groups) From b457b22c609375a571b1c7e7820b23bd4e3457b6 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 7 Nov 2026 18:10:58 +0000 Subject: [PATCH 2/2] Use default_factory for the mapping defaults (Python 3.11) Python 3.11 refuses an unhashable mappingproxy as a dataclass default, so tests/test_physical_check.py failed at collection on every 3.11 CI leg. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Scma6W3LC11s3qjXHZuGDm --- src/morphobricks_preprocessing/physical_check.py | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/src/morphobricks_preprocessing/physical_check.py b/src/morphobricks_preprocessing/physical_check.py index 5c0224c1..cd497bd4 100644 --- a/src/morphobricks_preprocessing/physical_check.py +++ b/src/morphobricks_preprocessing/physical_check.py @@ -19,7 +19,7 @@ import unicodedata from collections import Counter from collections.abc import Collection, Iterable, Mapping, Sequence -from dataclasses import dataclass +from dataclasses import dataclass, field from types import MappingProxyType from .axis_names import OPEN_AXIS @@ -128,8 +128,11 @@ class PhysicalReport: """ issues: tuple[PhysicalIssue, ...] = () - layouts: Mapping[UnitKey, Layout] = MappingProxyType({}) - image_type_verified: Mapping[UnitKey, bool] = MappingProxyType({}) + # default_factory: Python 3.11 refuses an unhashable mappingproxy as a dataclass default. + layouts: Mapping[UnitKey, Layout] = field(default_factory=lambda: MappingProxyType({})) + image_type_verified: Mapping[UnitKey, bool] = field( + default_factory=lambda: MappingProxyType({}) + ) unchecked: tuple[str, ...] = () @property