diff --git a/AGENTS.md b/AGENTS.md index 3b4fe069..4f3c2bf1 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -120,6 +120,10 @@ src/morphobricks_preprocessing/ description_file.py read_description(path): symlink refusal (R_FILE_SYMLINK), bytes -> Description run/description_file.py publish_description: lock, base SHA-256, strict re-read, transitions, atomic write, append-only morphobricks.records.jsonl (Spec experiment-v1 4.6, 10) + draft_ids.py proposed IDs: the folder label or raw header name only when the reader would + accept it as written, else none (never repaired); draft.py the draft + (IdSlot proposed/confirmed, open questions, unit_rows with condition_id, + repeat_group, technical_repeat_id) and migrate_legacy_design 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 148ddd3f..28a61943 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -100,6 +100,24 @@ are in [docs/plan.md](docs/plan.md) ยง12. Upgrading from v0.3: see the a changed image count alone is refused (new FOVs do not reopen a confirmation, section 10); the Spec's validator accepts it. Copying the records into the run manifest is a later pull request. Not used by any run, notebook or CLI command yet. +- **Draft description, proposed IDs and legacy design migration (Stage 1, PR 7).** New + modules `draft_ids` and `draft`, and `design.parse_design_bytes` (the + former private `_parse_strict`, now public: the strict parse, no confirmation required). A folder label is proposed as a condition, + technical-repeat or biological-repeat ID only when the reader would accept it as written + (Spec 4.6a; the same rule for repeat IDs): `control`, `files` or `z` get no proposal, nothing is + repaired, derived or renamed. A raw header name is proposed as a channel ID only when it is a + valid ID as it stands (never trimmed or repaired; `R`, `G`, `B` never). A draft keeps IDs as + *proposed* until `confirm_id`, lists its open questions, and offers the added `condition_id`, + `repeat_group` and `technical_repeat_id` beside the unchanged source labels. + `migrate_legacy_design` reads `experiment_design.csv` strictly (`yes`/`no`/empty refused), + carries `repeat_group` as the proposed repeat ID, takes the latest `confirmed_on` of a group + with several dates and lists the original dates in a note (never the clock), maps 0, 1 or 2+ + marked controls to no reference, one reference per condition or an open question, and refuses a + design that breaks a Spec rule with the reader's code. A legacy `repeat_group` that is not an + allowed repeat ID (`z`, `files`) is not refused: it migrates with no proposed ID and an open + question, so the user names the repeat (Spec 4.6a; owner decision of 7 November 2026). Where the Spec says "another allowed + ID" for a forbidden folder there is no proposal: no rule for making one was given. Nothing is + written and no run, notebook or CLI command uses it yet. - **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 diff --git a/pyproject.toml b/pyproject.toml index c52838ae..9e7ec139 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -178,6 +178,8 @@ baseline = "tools/code_rules_baseline.json" "morphobricks_preprocessing.design" = "io" "morphobricks_preprocessing.design_values" = "pure" "morphobricks_preprocessing.description_identity" = "pure" +"morphobricks_preprocessing.draft_ids" = "pure" +"morphobricks_preprocessing.draft" = "io" "morphobricks_preprocessing.physical_check" = "io" "morphobricks_preprocessing.description" = "pure" "morphobricks_preprocessing.description_write" = "pure" diff --git a/src/morphobricks_preprocessing/design.py b/src/morphobricks_preprocessing/design.py index 67f74c80..ddcd7a0b 100644 --- a/src/morphobricks_preprocessing/design.py +++ b/src/morphobricks_preprocessing/design.py @@ -608,7 +608,7 @@ def read_confirmed_design( ) try: data = path.read_bytes() - design = _parse_strict(data, path.name) + design = parse_design_bytes(data, path.name) except (OSError, DesignFormatError) as exc: raise _unreadable(path, exc, rerun) from exc issues = design_issues(design) @@ -740,8 +740,8 @@ def _unreadable(path: Path, exc: Exception, action: str) -> DesignCheckError: ) -def _parse_strict(data: bytes, name: str) -> ExperimentDesign: - """Parse a design as :func:`write_experiment_design` writes it, refusing anything else.""" +def parse_design_bytes(data: bytes, name: str = DESIGN_FILENAME) -> ExperimentDesign: + """Parse a design as :func:`write_experiment_design` writes it; confirmation not required.""" if data.startswith(b"\xef\xbb\xbf"): raise DesignFormatError( f"{name} starts with a byte-order mark: it was saved by another program, not " diff --git a/src/morphobricks_preprocessing/draft.py b/src/morphobricks_preprocessing/draft.py new file mode 100644 index 00000000..8f16762a --- /dev/null +++ b/src/morphobricks_preprocessing/draft.py @@ -0,0 +1,518 @@ +"""The draft of an experiment description, and the migration of a legacy ``experiment_design.csv``. + +A draft is what discovery can say before the user has answered (experiment-v1 section 12): the +folders found, the IDs proposed for them, the pairing of biological repeats, and what is still +open. It is **not** a description and is never read as one: an ID is *proposed* until the user +confirms it, ``confirmed_on`` is a date the user (or a legacy file) gave, never a clock read, +and a draft with an open question cannot be published. Nothing here writes a file, reads an +image or reads the clock. + +Source folders keep their labels in every key (section 3.4): a draft names conditions and +technical repeats by their source folder, the IDs are slots beside them, and ``unit_rows`` gives +the added ``condition_id``, ``repeat_group`` and ``technical_repeat_id`` next to the unchanged +folder labels. +""" + +from __future__ import annotations + +from collections.abc import Collection, Iterable, Mapping +from dataclasses import dataclass, field, replace +from types import MappingProxyType + +from morphobricks_normalization.records import SINGLE + +from .description_identity import nfc +from .description_read import PHASES +from .description_report import DescriptionIssue +from .description_rules import SCHEMA_ID +from .design import ExperimentDesign, design_issues, parse_design_bytes +from .discovery import DiscoveryResult +from .draft_ids import IdKind, id_problems, propose_id + +PLACEHOLDER_DATE = "2000-01-01" +"""Stands in for a missing date in the rule-check document only; it never leaves that function.""" + + +class DraftError(ValueError): + """Raised when a draft step is refused; ``codes`` are the reasons (Spec codes if any).""" + + def __init__(self, message: str, codes: tuple[str, ...] = ()) -> None: + self.codes = codes + super().__init__(message) + + +class LegacyMigrationRefused(DraftError): + """Raised when a legacy design violates experiment-v1: refused, never repaired (section 7). + + ``issues`` are the reader's issues (code, table path, value) of the first failing phase; + the paths use stand-in condition IDs where the folder is not an allowed ID. + """ + + def __init__(self, message: str, issues: tuple[DescriptionIssue, ...] = ()) -> None: + self.issues = issues + codes = tuple(dict.fromkeys(issue.code.value for issue in issues)) or ( + "repeat_group_conflict", + ) + super().__init__(message, codes) + + +@dataclass(frozen=True) +class IdSlot: + """An ID that may be proposed and is adopted only when the user confirms one. + + ``proposed`` is the folder label when it is allowed (None: no proposal, the user names + it). ``confirmed`` is None until the user confirmed an ID, the proposed one or another. + """ + + proposed: str | None = None + confirmed: str | None = None + + @property + def pending(self) -> bool: + """True while no ID is confirmed.""" + return self.confirmed is None + + +@dataclass(frozen=True) +class DraftUnit: + """One imaging unit (a technical-repeat folder, or ``single``) of a member.""" + + source_folder: str + image_count: int + + +@dataclass(frozen=True) +class DraftMember: + """One biological-repeat folder of one condition, with its units.""" + + condition: str + source_repeat: str + image_count: int + units: tuple[DraftUnit, ...] + + @property + def key(self) -> tuple[str, str]: + """``(condition folder, source repeat)``.""" + return (self.condition, self.source_repeat) + + +@dataclass(frozen=True) +class DraftRepeat: + """One biological repeat: its ID slot, its member folders and the user's confirmation. + + ``confirmed`` is the N35 answer, ``confirmed_on`` its date (``""`` while none is known). + """ + + id: IdSlot + members: tuple[DraftMember, ...] + confirmed: bool = False + confirmed_on: str = "" + + +@dataclass(frozen=True) +class MigrationNote: + """The original per-folder dates of a legacy group that had several (AQ12).""" + + repeat_group: str + dates: tuple[tuple[str, str, str], ...] + """``(condition folder, source repeat, date)``, ``""`` for a folder with no date.""" + kept: str + """The date the draft carries: the latest of them, never the clock.""" + + +@dataclass(frozen=True) +class OpenQuestion: + """Something the user must answer before the draft can be published.""" + + code: str + subject: str + + +@dataclass(frozen=True) +class UnitRow: + """The key of one unit with the canonical IDs added beside it (section 3.4). + + ``condition``, ``biological_repeat`` and ``technical_repeat`` stay the source folder + labels; ``condition_id``, ``repeat_group`` and ``technical_repeat_id`` are the confirmed IDs, + None while not confirmed. + """ + + condition: str + biological_repeat: str + technical_repeat: str + condition_id: str | None + repeat_group: str | None + technical_repeat_id: str | None + image_count: int + + +@dataclass(frozen=True) +class Draft: + """A draft description: IDs beside the source folders, and the open questions. + + ``conditions`` and ``technical_repeats`` are keyed by source folder (one folder, one ID); + ``repeats`` by the pairing label (the folder name for the default pairing, the legacy + ``repeat_group`` for a migration). ``references`` maps a condition folder to the folder of + the condition it is compared against; ``unresolved_controls`` lists the folders a legacy + design marked as controls when there were two or more (no name-stem guessing), and + ``legacy_missing`` the legacy folders no longer found. ``origin`` is ``"discovery"`` or + ``"legacy"``. + """ + + conditions: Mapping[str, IdSlot] + technical_repeats: Mapping[str, IdSlot] + repeats: Mapping[str, DraftRepeat] + references: Mapping[str, str] = field(default_factory=dict) + unresolved_controls: tuple[str, ...] = () + legacy_missing: tuple[tuple[str, str], ...] = () + notes: tuple[MigrationNote, ...] = () + used_ids: Mapping[str, tuple[str, ...]] = field(default_factory=dict) + origin: str = "discovery" + + def open_questions(self) -> tuple[OpenQuestion, ...]: + """Return every question still open; a draft with any cannot be published.""" + out = [ + OpenQuestion("id_pending", f"condition {folder}") + for folder, slot in sorted(self.conditions.items()) + if slot.pending + ] + out += [ + OpenQuestion("id_pending", f"technical repeat {folder}") + for folder, slot in sorted(self.technical_repeats.items()) + if slot.pending + ] + for group, repeat in sorted(self.repeats.items()): + if repeat.id.pending: + out.append(OpenQuestion("id_pending", f"repeat {group}")) + if not repeat.confirmed: + out.append(OpenQuestion("repeat_not_confirmed", f"repeat {group}")) + elif not repeat.confirmed_on: + out.append(OpenQuestion("confirmed_on_missing", f"repeat {group}")) + out += [OpenQuestion("control_unresolved", folder) for folder in self.unresolved_controls] + out += [OpenQuestion("legacy_folder_missing", f"{c} / {r}") for c, r in self.legacy_missing] + return tuple(out) + + @property + def publishable(self) -> bool: + """True when no question is open. Publication still validates and asks the user.""" + return not self.open_questions() + + def confirm_id(self, kind: IdKind, key: str, value: str) -> Draft: + """Return the draft with the user's ID for a condition, technical-repeat folder or repeat. + + ``value`` is what the user accepted or typed, adopted exactly as given. It is refused + with the codes of the reader's rules when the reader would refuse it as an ID, or when + it is already used by another entity of this draft or by ``used_ids``. + + Raises + ------ + DraftError + With the codes, or ``unknown_key`` when the draft has no such folder or repeat. + """ + table = { + "conditions": self.conditions, + "technical_repeats": self.technical_repeats, + "repeats": {group: repeat.id for group, repeat in self.repeats.items()}, + }.get(kind) + if table is None or key not in table: + raise DraftError(f"the draft has no {kind} entry {key!r}", ("unknown_key",)) + codes = list(id_problems(kind, value, self.used_ids.get(kind, ()))) + for name, slot in table.items(): # another entry of this draft already has the ID + if name != key and slot.confirmed is not None: + if slot.confirmed == value: + codes.append("id_in_use_in_draft") + elif nfc(slot.confirmed).casefold() == nfc(value).casefold(): + codes.append("E_ID_CASE_COLLISION") + if kind in ("conditions", "technical_repeats") and (value == SINGLE) != (key == SINGLE): + codes.append("E_SINGLE_MISUSE") + if codes: + raise DraftError( + f"{value!r} cannot be the ID of {kind} {key!r}: {', '.join(codes)}", + tuple(dict.fromkeys(codes)), + ) + if kind == "repeats": + repeats = dict(self.repeats) + repeats[key] = replace(repeats[key], id=replace(repeats[key].id, confirmed=value)) + return replace(self, repeats=MappingProxyType(repeats)) + slots = dict(table) + slots[key] = replace(slots[key], confirmed=value) + return replace(self, **{kind: MappingProxyType(slots)}) + + def unit_rows(self) -> tuple[UnitRow, ...]: + """Return one row per unit: the source labels and the confirmed IDs beside them.""" + rows = [ + UnitRow( + member.condition, + member.source_repeat, + unit.source_folder, + self.conditions[member.condition].confirmed, + repeat.id.confirmed, + self.technical_repeats[unit.source_folder].confirmed, + unit.image_count, + ) + for repeat in self.repeats.values() + for member in repeat.members + for unit in member.units + ] + return tuple( + sorted(rows, key=lambda r: (r.condition, r.biological_repeat, r.technical_repeat)) + ) + + +def _slots(kind: IdKind, labels: Iterable[str], used: Collection[str]) -> dict[str, IdSlot]: + """Propose each label as its own ID where allowed, never twice for one case-folded name.""" + slots: dict[str, IdSlot] = {} + taken = list(used) + for label in sorted(set(labels)): + proposal = propose_id(kind, label, taken) + if proposal is not None: + taken.append(proposal) + slots[label] = IdSlot(proposal) + return slots + + +def _members(discovery: DiscoveryResult, excluded: Collection[str]) -> list[DraftMember]: + """Return the biological-repeat folders with images to analyse, with their units.""" + folders: dict[tuple[str, str], dict[str, int]] = {} + for leaf in discovery.leaves: + included = sum(image not in excluded for image in leaf.images) + if included: # a folder whose images are all excluded is not analysed + units = folders.setdefault((leaf.condition, leaf.biological_repeat), {}) + units[leaf.technical_repeat] = units.get(leaf.technical_repeat, 0) + included + return [ + DraftMember( + condition, + repeat, + sum(units.values()), + tuple(DraftUnit(folder, count) for folder, count in sorted(units.items())), + ) + for (condition, repeat), units in sorted(folders.items()) + ] + + +def _draft( + members: list[DraftMember], + groups: Mapping[tuple[str, str], str], + used: Mapping[str, tuple[str, ...]], +) -> Draft: + """Build a draft from members and the repeat group of each (default: the folder name).""" + paired: dict[str, list[DraftMember]] = {} + for member in members: + paired.setdefault(groups.get(member.key, member.source_repeat), []).append(member) + repeat_slots = _slots("repeats", paired, used.get("repeats", ())) + return Draft( + MappingProxyType( + _slots("conditions", (m.condition for m in members), used.get("conditions", ())) + ), + MappingProxyType( + _slots( + "technical_repeats", + (u.source_folder for m in members for u in m.units), + used.get("technical_repeats", ()), + ) + ), + MappingProxyType( + { + group: DraftRepeat(repeat_slots[group], tuple(found)) + for group, found in sorted(paired.items()) + } + ), + used_ids=MappingProxyType(dict(used)), + ) + + +def draft_from_discovery( + discovery: DiscoveryResult, + excluded: Iterable[str] = (), + *, + used_ids: Mapping[str, Iterable[str]] | None = None, +) -> Draft: + """Build the draft of a folder before any answer: proposed IDs, units, counts, pairing. + + Biological-repeat folders with the same name are one repeat, as in the default design + (:func:`.design.default_design`); nothing is confirmed. IDs are proposed as the folder + labels where allowed (Q6: the same rule for conditions, technical repeats and repeats), and + ``used_ids`` (the IDs a published file of this lineage lists) are never proposed again. + No image is opened, no folder renamed and no file written. + """ + used = {kind: tuple(values) for kind, values in (used_ids or {}).items()} + return _draft(_members(discovery, set(excluded)), {}, used) + + +def _legacy_groups(legacy: ExperimentDesign) -> dict[tuple[str, str], str]: + issues = [issue for issue in design_issues(legacy) if issue.code == "repeat_group_conflict"] + if issues: + raise LegacyMigrationRefused(issues[0].message) + return legacy.repeat_groups() + + +def _probe(draft: Draft) -> dict[str, object]: + """Return the draft as a document for the reader's rules only; stand-ins fill missing IDs. + + Condition, technical-repeat and repeat folders that are not allowed IDs get a stand-in ID + (their question stays open, section 4.6a), and one channel stands in for the layouts the + draft does not know. Nothing of it is stored. + """ + stand_in = {} + repeat_slots = {group: repeat.id for group, repeat in draft.repeats.items()} + for kind, slots in ( + ("conditions", draft.conditions), + ("technical", draft.technical_repeats), + ("repeats", repeat_slots), + ): + for folder, slot in slots.items(): + name = folder if folder == SINGLE and kind != "repeats" else slot.proposed + number = len(stand_in) + 1 + while name is None or (kind, name) in stand_in.values(): + name = f"{kind} {number}" + number += 1 + stand_in[(kind, folder)] = (kind, name) + cid = {folder: name for (kind, folder), (_, name) in stand_in.items() if kind == "conditions"} + tid = {folder: name for (kind, folder), (_, name) in stand_in.items() if kind == "technical"} + rid = {folder: name for (kind, folder), (_, name) in stand_in.items() if kind == "repeats"} + return { + "schema": SCHEMA_ID, + "schema_version": 1, + "conditions": {cid[f]: {"source_folder": f} for f in draft.conditions}, + "channels": {"Probe": {}}, + "repeats": { + rid[group]: { + "confirmed": True, + "confirmed_on": repeat.confirmed_on or PLACEHOLDER_DATE, + "members": { + cid[m.condition]: { + "source_repeat": m.source_repeat, + "image_count": m.image_count, + "technical_repeats": { + tid[u.source_folder]: { + "source_folder": u.source_folder, + "channels": ["Probe"], + } + for u in m.units + }, + } + for m in repeat.members + }, + } + for group, repeat in draft.repeats.items() + }, + "used_ids": { + "conditions": sorted(cid.values()), + "channels": ["Probe"], + "repeats": sorted(rid.values()), + "technical_repeats": sorted(tid.values()), + }, + } + + +def _check_legacy_rules(draft: Draft) -> None: + """Refuse a migrated draft that experiment-v1 would refuse, with the reader's codes.""" + doc = _probe(draft) + for _, check in PHASES: + issues = check(doc) + if issues: + codes = ", ".join(sorted({i.code.value for i in issues.items})) + raise LegacyMigrationRefused( + f"the legacy design breaks the rules of the experiment description ({codes}); " + "it is not repaired: change the repeat groups, or start a new design", + tuple(issues.items), + ) + + +def _migrated_repeat( + group: str, repeat: DraftRepeat, legacy: ExperimentDesign +) -> tuple[DraftRepeat, MigrationNote | None]: + """Carry a legacy group's confirmation: only if every folder is confirmed, latest date.""" + rows = [legacy.row(member.key) for member in repeat.members] + known = [row for row in rows if row is not None] + dates = sorted({row.confirmed_on for row in known if row.confirmed_on}) + confirmed = len(known) == len(rows) and all(row.confirmed for row in known) + kept = dates[-1] if confirmed and dates else "" + note = None + if len(dates) > 1: + listed = tuple( + (m.condition, m.source_repeat, row.confirmed_on if row else "") + for m, row in zip(repeat.members, rows, strict=True) + ) + note = MigrationNote(group, listed, kept) + return replace(repeat, confirmed=confirmed, confirmed_on=kept), note + + +def migrate_legacy_design( + data: bytes, + discovery: DiscoveryResult, + excluded: Iterable[str] = (), + *, + name: str = "experiment_design.csv", + used_ids: Mapping[str, Iterable[str]] | None = None, +) -> Draft: + """Migrate the bytes of a legacy ``experiment_design.csv`` to a draft (experiment-v1 section 7). + + The file is read strictly (:func:`.design.parse_design_bytes`: ``yes``, ``no`` and empty + booleans are refused, not coerced). Its ``repeat_group`` becomes the proposed repeat ID, + ``biological_repeat`` the source repeat, and the ``condition`` label the condition's source + folder, whose ID is proposed as that label when allowed. Technical repeats and counts come + from ``discovery``; channels are not part of a legacy file. A group is confirmed only if + every one of its folders is; with several dates it takes the latest and a + :class:`MigrationNote` lists each folder's date; the clock is never read, and a confirmed + group without any date stays open (``confirmed_on_missing``). A ``repeat_group`` that is not + an allowed repeat ID (for example ``z``) is not refused: it migrates with no proposed ID and + an open question, so the user names the repeat (section 4.6a). Controls: none marked, no + reference; one marked, every other condition references it; two or more, an unresolved + question and no reference. A folder found now but not in the file is a new pending member + of the group named after it, which therefore is not confirmed. A legacy folder no longer + found is listed in ``legacy_missing`` and blocks publication. IDs stay proposals: the user + confirms them. + + Raises + ------ + DesignFormatError + If the bytes are not a design this brick writes (:func:`.design.parse_design_bytes`). + LegacyMigrationRefused + If a repeat group holds two folders of one condition, or the migrated design breaks a + rule of the experiment description (for example ``E_REPEAT_ID_SHADOWS_SOURCE``): it is + refused with the code, never repaired. + """ + legacy = parse_design_bytes(data, name) + used = {kind: tuple(values) for kind, values in (used_ids or {}).items()} + members = _members(discovery, set(excluded)) + draft = _draft(members, _legacy_groups(legacy), used) + notes, repeats = [], {} + for group, repeat in draft.repeats.items(): + migrated, note = _migrated_repeat(group, repeat, legacy) + proposal = group if not id_problems("repeats", group, used.get("repeats", ())) else None + repeats[group] = replace(migrated, id=IdSlot(proposal)) + notes += [note] if note else [] + present = {m.condition for m in members} + marked = sorted(c for c in legacy.controls if c in present) + references = ( + {c: marked[0] for c in sorted(present) if c != marked[0]} if len(marked) == 1 else {} + ) + found = {m.key for m in members} + draft = replace( + draft, + repeats=MappingProxyType(repeats), + references=MappingProxyType(references), + unresolved_controls=tuple(marked) if len(marked) > 1 else (), + legacy_missing=tuple(sorted(row.key for row in legacy.rows if row.key not in found)), + notes=tuple(notes), + origin="legacy", + ) + _check_legacy_rules(draft) + return draft + + +__all__ = [ + "Draft", + "DraftError", + "DraftMember", + "DraftRepeat", + "DraftUnit", + "IdSlot", + "LegacyMigrationRefused", + "MigrationNote", + "OpenQuestion", + "UnitRow", + "draft_from_discovery", + "migrate_legacy_design", +] diff --git a/src/morphobricks_preprocessing/draft_ids.py b/src/morphobricks_preprocessing/draft_ids.py new file mode 100644 index 00000000..8255ac0b --- /dev/null +++ b/src/morphobricks_preprocessing/draft_ids.py @@ -0,0 +1,114 @@ +"""Proposed IDs for a draft experiment description (experiment-v1 sections 3.2, 4.6a and 7). + +A proposal is the folder label (or the raw header name of a channel) **as it is**, offered only +when the reader would accept it as an ID; otherwise there is no proposal and the user names the +ID. Nothing is trimmed, normalized, re-cased, numbered or derived (section 3.3: no ID is ever +repaired), and a proposal is never adopted without the user's confirmation. The rules applied are +the reader's own (``description_identity``, ``description_rules``): this module holds the +decision "propose or not" and no second copy of the rules. + +The Spec says a condition or technical-repeat folder that is not allowed (``control``, ``files``) +gets "a different ID that is allowed". It does not say how to make one, and deriving one from +the label would be a repair, so here such a folder has no proposal either. +""" + +from __future__ import annotations + +from collections.abc import Collection, Sequence +from dataclasses import dataclass +from typing import Literal + +from .description_identity import Identities, nfc +from .description_report import Issues +from .description_rules import FORBIDDEN, RGB_TOKENS, Code + +IdKind = Literal["conditions", "channels", "repeats", "technical_repeats"] +REUSED = "R_USED_ID_REUSED" +"""Reported for a value that is already in ``used_ids``: it names a retired or declared ID.""" + + +def id_problems(kind: IdKind, value: str, used: Collection[str] = ()) -> tuple[str, ...]: + """Return the codes under which ``value`` is refused as an ID of ``kind``, empty if allowed. + + The Label rules, NFC and the ``single`` rule for every kind; the forbidden names (section + 4.7, compared exactly) for every ID, since each is a table key; the Token rule and the + reserved list for channels. ``used`` holds the IDs of that kind the lineage already lists + (``used_ids``): a value equal ignoring case to one is ``E_ID_CASE_COLLISION``, and equal + exactly is ``R_USED_ID_REUSED`` (section 4.9). + + The ``single`` rule that ties a condition or technical-repeat ID to its folder is not + checked here: it compares the ID with a folder, which this function does not know. + """ + out = Issues() + ids = Identities(out) + if kind == "channels": + ids.token(value, "id") + ids.nfc(value, "id") + ids.single(value, "id") + ids.reserved(value, "id") + else: + ids.label_id(value, "id", kind) + codes = [issue.code.value for issue in out.items] + if value in FORBIDDEN: + codes.append(Code.E_FORBIDDEN_FIELD.value) + if value in used: + codes.append(REUSED) + elif any(nfc(other).casefold() == nfc(value).casefold() for other in used): + codes.append(Code.E_ID_CASE_COLLISION.value) + return tuple(dict.fromkeys(codes)) + + +def propose_id(kind: IdKind, label: str, used: Collection[str] = ()) -> str | None: + """Return the folder ``label`` as the proposed ID of ``kind``, or None when it is not allowed. + + Applies to conditions, technical repeats and, as the owner decided (Q6), biological repeats: + for ``control``, ``files`` or ``z`` there is no proposal. Never a repaired spelling. + """ + if kind == "channels": + raise ValueError("a channel ID is proposed from a header name: use propose_channel_ids") + return None if id_problems(kind, label, used) else label + + +@dataclass(frozen=True) +class ChannelProposal: + """What is proposed for the channel at one source index C*i* of a layout. + + ``header_name`` is the name as the header holds it (None when it names none), and + ``proposed_id`` the same text when it is a valid channel ID as it stands, else None. + """ + + index: int + header_name: str | None + proposed_id: str | None + + +def propose_channel_ids( + header_names: Sequence[str | None], used: Collection[str] = () +) -> tuple[ChannelProposal, ...]: + """Propose a channel ID for each C index of a layout from its raw header names (Q4). + + A header name is proposed as the channel ID only if it is a valid channel ID exactly as + written (Token rule, not reserved, not retired in ``used``, no outer whitespace); otherwise + the ID is left blank for the user. Names that are equal ignoring case within the layout + get no proposal, since one channel cannot stand at two positions of a unit. ``R``, ``G`` + and ``B`` are never proposed: they are the fixed IDs of a declared RGB unit and take no + header names (section 4.4b), so a header called ``R`` is no evidence of an RGB unit. + A proposal is not an alias: the user decides whether the name is also recorded as + ``header_names`` evidence. + """ + candidates = [ + name if name and not id_problems("channels", name, used) else None for name in header_names + ] + folded = [nfc(name).casefold() for name in candidates if name] + return tuple( + ChannelProposal( + index, + header, + candidate + if candidate + and folded.count(nfc(candidate).casefold()) == 1 + and candidate.casefold() not in {token.casefold() for token in RGB_TOKENS} + else None, + ) + for index, (header, candidate) in enumerate(zip(header_names, candidates, strict=True)) + ) diff --git a/tests/test_draft.py b/tests/test_draft.py new file mode 100644 index 00000000..e90581a0 --- /dev/null +++ b/tests/test_draft.py @@ -0,0 +1,380 @@ +"""The draft of an experiment description: proposed IDs and the legacy design migration. + +Experiment-v1 sections 4.6a and 7, conformance scenarios P30 and P32 (the proposal part). The +owner's answers: Q4 a header name is proposed as a channel ID only when it is valid as it +stands, Q6 the proposal rule covers repeat IDs and the condition ``single`` rule, Q8 the added +fields are ``condition_id``, ``repeat_group`` and ``technical_repeat_id``. +""" + +import pytest + +from morphobricks_preprocessing import DesignFormatError, discover_experiment +from morphobricks_preprocessing.design import DesignRow, ExperimentDesign, write_experiment_design +from morphobricks_preprocessing.draft import ( + DraftError, + LegacyMigrationRefused, + draft_from_discovery, + migrate_legacy_design, +) +from morphobricks_preprocessing.draft_ids import ( + id_problems, + propose_channel_ids, + propose_id, +) +from tiff_tree import make_tree + + +def folders(root, *paths): + """Write two images into each folder path and return the discovery of the root.""" + make_tree(root, [f"{p}/{name}.tif" for p in paths for name in ("a", "b")]) + return discover_experiment(root) + + +def legacy(tmp_path, rows): + """Return the bytes of a legacy design with ``rows`` written by the brick's own writer.""" + path = tmp_path / "legacy.csv" + write_experiment_design(ExperimentDesign(tuple(rows)), path) + return path.read_bytes() + + +def row(condition, repeat, group=None, *, control=False, on="2026-10-30", confirmed=True): + return DesignRow( + condition, + repeat, + group or repeat, + control=control, + image_count=2, + confirmed=confirmed, + confirmed_on=on if confirmed else "", + ) + + +# --- proposed IDs (P32, Q4, Q6) --------------------------------------------------------------- + + +def test_the_folder_label_is_proposed_when_the_reader_would_accept_it(): + assert propose_id("conditions", "CTRL") == "CTRL" + assert propose_id("technical_repeats", "acquisition_1") == "acquisition_1" + assert propose_id("repeats", "rep1") == "rep1" + assert propose_id("conditions", "single") == "single" + assert propose_id("repeats", "R999") == "R999" # no width or prefix rule + + +@pytest.mark.parametrize( + ("kind", "label"), + [ + ("conditions", "control"), + ("technical_repeats", "files"), + ("technical_repeats", "overrides"), + ("repeats", "z"), # Q6: a repeat folder gets no proposal either + ("repeats", "paths"), + ("conditions", "NA"), + ("conditions", " CTRL"), + ("conditions", "Single"), + ], +) +def test_a_label_that_is_not_an_allowed_id_gets_no_proposal_and_is_never_repaired(kind, label): + assert id_problems(kind, label) + assert propose_id(kind, label) is None + + +def test_a_label_that_collides_with_a_used_id_gets_no_proposal(): + assert propose_id("repeats", "rep1", ["rep1"]) is None + assert id_problems("repeats", "rep1", ["rep1"]) == ("R_USED_ID_REUSED",) + assert id_problems("conditions", "ctrl", ["CTRL"]) == ("E_ID_CASE_COLLISION",) + assert propose_id("conditions", "ctrl", ["CTRL"]) is None + + +def test_p32_forbidden_folders_get_no_proposal_and_nothing_is_adopted(tmp_path): + root = tmp_path / "exp" + discovery = folders(root, "control/rep1/files", "control/rep1/overrides", "DrugA/rep1/files") + before = sorted(p.relative_to(tmp_path) for p in tmp_path.rglob("*")) + draft = draft_from_discovery(discovery) + assert draft.conditions["control"].proposed is None + assert draft.conditions["DrugA"].proposed == "DrugA" + assert draft.technical_repeats["files"].proposed is None + assert draft.technical_repeats["overrides"].proposed is None + assert all(slot.confirmed is None for slot in draft.conditions.values()) + assert not draft.publishable + assert {q.code for q in draft.open_questions()} >= {"id_pending", "repeat_not_confirmed"} + assert before == sorted(p.relative_to(tmp_path) for p in tmp_path.rglob("*")) # no file written + assert sorted(p.name for p in root.iterdir()) == ["DrugA", "control"] # no folder renamed + + +def test_p32_ordinary_folders_are_proposed_as_their_labels(tmp_path): + draft = draft_from_discovery( + folders(tmp_path / "exp", "CTRL/rep1/acquisition_1", "CTRL/rep2/acquisition_1") + ) + assert draft.conditions["CTRL"].proposed == "CTRL" + assert draft.technical_repeats["acquisition_1"].proposed == "acquisition_1" + assert {g: r.id.proposed for g, r in draft.repeats.items()} == {"rep1": "rep1", "rep2": "rep2"} + + +def test_a_repeat_folder_named_z_has_no_proposal_and_the_user_names_it(tmp_path): + draft = draft_from_discovery(folders(tmp_path / "exp", "A/z", "B/z", "A/rep2", "B/rep2")) + assert draft.repeats["z"].id.proposed is None + assert [m.condition for m in draft.repeats["z"].members] == ["A", "B"] + assert draft.repeats["z"].members[0].source_repeat == "z" # the source label is kept + with pytest.raises(DraftError) as refused: + draft.confirm_id("repeats", "z", "z") + assert "E_FORBIDDEN_FIELD" in refused.value.codes + named = draft.confirm_id("repeats", "z", "R1") + assert named.repeats["z"].id.confirmed == "R1" + assert draft.repeats["z"].id.confirmed is None # a draft is never changed in place + + +def test_a_flat_experiment_proposes_single_for_every_level(tmp_path): + draft = draft_from_discovery(folders(tmp_path / "exp", ".")) + assert draft.conditions["single"].proposed == "single" + assert draft.technical_repeats["single"].proposed == "single" + assert draft.repeats["single"].id.proposed == "single" + assert draft.unit_rows()[0].condition == "single" + + +def test_the_condition_single_rule_ties_the_id_to_its_folder(tmp_path): + draft = draft_from_discovery(folders(tmp_path / "exp", ".")) + with pytest.raises(DraftError) as refused: + draft.confirm_id("conditions", "single", "main") + assert refused.value.codes == ("E_SINGLE_MISUSE",) + draft = draft_from_discovery(folders(tmp_path / "two", "A/rep1", "B/rep1")) + with pytest.raises(DraftError) as refused: + draft.confirm_id("conditions", "A", "single") + assert refused.value.codes == ("E_SINGLE_MISUSE",) + + +def test_used_ids_are_never_proposed_again(tmp_path): + discovery = folders(tmp_path / "exp", "A/rep1", "A/rep2") + draft = draft_from_discovery(discovery, used_ids={"repeats": ["rep1"], "conditions": ["a"]}) + assert draft.repeats["rep1"].id.proposed is None + assert draft.repeats["rep2"].id.proposed == "rep2" + assert draft.conditions["A"].proposed is None # `a` is retired and collides ignoring case + + +def test_ids_in_the_draft_appear_beside_the_unchanged_source_labels(tmp_path): + discovery = folders(tmp_path / "exp", "control/rep1", "DrugA/rep1") + draft = draft_from_discovery(discovery) + draft = draft.confirm_id("conditions", "control", "ctrl") + draft = draft.confirm_id("conditions", "DrugA", "DrugA") + draft = draft.confirm_id("technical_repeats", "single", "single") + draft = draft.confirm_id("repeats", "rep1", "rep1") + rows = {r.condition: r for r in draft.unit_rows()} + assert rows["control"].condition_id == "ctrl" + assert rows["control"].biological_repeat == "rep1" + assert (rows["control"].repeat_group, rows["control"].technical_repeat_id) == ("rep1", "single") + assert rows["DrugA"].condition_id == "DrugA" + with pytest.raises(DraftError) as refused: + draft.confirm_id("conditions", "DrugA", "CTRL") # differs from `ctrl` only in case + assert refused.value.codes == ("E_ID_CASE_COLLISION",) + + +def test_counts_come_from_the_folders_and_excluded_images_are_not_counted(tmp_path): + discovery = folders(tmp_path / "exp", "A/rep1/t1", "A/rep1/t2") + draft = draft_from_discovery(discovery, excluded=["A/rep1/t2/a.tif", "A/rep1/t2/b.tif"]) + (member,) = draft.repeats["rep1"].members + assert member.image_count == 2 + assert [(u.source_folder, u.image_count) for u in member.units] == [("t1", 2)] + + +# --- channel IDs from raw header names (Q4) ---------------------------------------------------- + + +def proposed(names, used=()): + return [p.proposed_id for p in propose_channel_ids(names, used)] + + +def test_a_header_name_is_proposed_as_the_channel_id_only_when_valid_as_it_stands(): + assert proposed(["DAPI", "Phalloidin488", "ACTIN_2"]) == ["DAPI", "Phalloidin488", "ACTIN_2"] + assert proposed(["DAPI 405"]) == [None] # never repaired to DAPI_405 + assert proposed([" DAPI"]) == [None] # never trimmed + assert proposed(["DAPI-405", "405nm", "a__b"]) == [None, None, None] + assert proposed(["cell", "Nucleus", "z"]) == [None, None, None] # reserved + assert proposed(["R", "g", "B"]) == [None, None, None] # fixed RGB IDs, no aliases + + +def test_unnamed_and_duplicate_header_names_get_no_proposal(): + assert proposed([None, "", "DAPI"]) == [None, None, "DAPI"] + assert proposed(["DAPI", "dapi"]) == [None, None] # one channel cannot stand at two indices + assert proposed(["DAPI"], used=["Dapi"]) == [None] # retired ID ignoring case + + +def test_a_proposal_keeps_the_raw_header_name_and_its_index(): + (first, second) = propose_channel_ids(["DAPI", " raw "]) + assert (first.index, first.header_name, first.proposed_id) == (0, "DAPI", "DAPI") + assert (second.index, second.header_name, second.proposed_id) == (1, " raw ", None) + + +# --- legacy migration (P30) -------------------------------------------------------------------- + + +def test_a_legacy_design_migrates_to_proposals_and_open_confirmations(tmp_path): + discovery = folders(tmp_path / "exp", "CTRL/rep1", "CTRL/rep2", "DrugA/rep1", "DrugA/rep2") + data = legacy( + tmp_path, + [ + row("CTRL", "rep1", control=True), + row("CTRL", "rep2", control=True), + row("DrugA", "rep1"), + row("DrugA", "rep2"), + ], + ) + draft = migrate_legacy_design(data, discovery) + assert draft.origin == "legacy" + assert {g: (r.id.proposed, r.id.confirmed) for g, r in draft.repeats.items()} == { + "rep1": ("rep1", None), + "rep2": ("rep2", None), + } + assert {r.confirmed_on for r in draft.repeats.values()} == {"2026-10-30"} + assert all(r.confirmed for r in draft.repeats.values()) + assert dict(draft.references) == {"DrugA": "CTRL"} + assert draft.notes == () + assert {q.code for q in draft.open_questions()} == {"id_pending"} # IDs not adopted + assert not draft.publishable + for kind, key in [ + ("conditions", "CTRL"), + ("conditions", "DrugA"), + ("technical_repeats", "single"), + ("repeats", "rep1"), + ("repeats", "rep2"), + ]: + draft = draft.confirm_id(kind, key, key) + assert draft.open_questions() == () + assert draft.publishable + + +def test_p30_a_group_with_two_dates_takes_the_latest_and_a_note_lists_both(tmp_path): + discovery = folders(tmp_path / "exp", "A/rep1", "B/rep1", "A/rep2", "B/rep2") + data = legacy( + tmp_path, + [ + row("A", "rep1", "G1", on="2026-10-01"), + row("B", "rep1", "G1", on="2026-10-15"), + row("A", "rep2", "G2", on="2026-10-20"), + row("B", "rep2", "G2", on="2026-10-20"), + ], + ) + draft = migrate_legacy_design(data, discovery) + assert draft.repeats["G1"].confirmed_on == "2026-10-15" + assert draft.repeats["G2"].confirmed_on == "2026-10-20" + (note,) = draft.notes + assert note.repeat_group == "G1" + assert note.kept == "2026-10-15" + assert set(note.dates) == {("A", "rep1", "2026-10-01"), ("B", "rep1", "2026-10-15")} + + +def test_a_confirmed_group_without_any_date_stays_open_and_takes_no_clock_date(tmp_path): + discovery = folders(tmp_path / "exp", "A/rep1", "B/rep1") + data = legacy(tmp_path, [row("A", "rep1", on=""), row("B", "rep1", on="")]) + draft = migrate_legacy_design(data, discovery) + assert draft.repeats["rep1"].confirmed is True + assert draft.repeats["rep1"].confirmed_on == "" + assert ("confirmed_on_missing", "repeat rep1") in { + (q.code, q.subject) for q in draft.open_questions() + } + + +def test_an_unconfirmed_legacy_group_stays_unconfirmed(tmp_path): + discovery = folders(tmp_path / "exp", "A/rep1", "B/rep1") + data = legacy(tmp_path, [row("A", "rep1", confirmed=False), row("B", "rep1")]) + repeat = migrate_legacy_design(data, discovery).repeats["rep1"] + assert (repeat.confirmed, repeat.confirmed_on) == (False, "") + + +@pytest.mark.parametrize( + ("marked", "references", "unresolved"), + [ + ((), {}, ()), + (("A",), {"B": "A", "C": "A"}, ()), + (("A", "B"), {}, ("A", "B")), + ], +) +def test_legacy_controls_give_no_reference_one_reference_or_an_open_question( + tmp_path, marked, references, unresolved +): + discovery = folders(tmp_path / "exp", "A/rep1", "B/rep1", "C/rep1") + data = legacy(tmp_path, [row(c, "rep1", control=c in marked) for c in "ABC"]) + draft = migrate_legacy_design(data, discovery) + assert dict(draft.references) == references + assert draft.unresolved_controls == unresolved + assert ( + ("control_unresolved", "A") in {(q.code, q.subject) for q in draft.open_questions()} + ) == (len(unresolved) > 1) + + +def test_a_legacy_condition_on_the_forbidden_list_migrates_with_no_proposed_id(tmp_path): + discovery = folders(tmp_path / "exp", "control/rep1", "DrugA/rep1") + data = legacy(tmp_path, [row("control", "rep1", control=True), row("DrugA", "rep1")]) + draft = migrate_legacy_design(data, discovery) # not refused + assert draft.conditions["control"].proposed is None + assert dict(draft.references) == {"DrugA": "control"} # by folder; the ID is confirmed later + assert ("id_pending", "condition control") in { + (q.code, q.subject) for q in draft.open_questions() + } + assert ( + draft.confirm_id("conditions", "control", "ctrl").conditions["control"].confirmed == "ctrl" + ) + + +def test_lenient_legacy_values_are_refused_not_coerced(tmp_path): + discovery = folders(tmp_path / "exp", "A/rep1") + data = legacy(tmp_path, [row("A", "rep1")]).replace(b",true,", b",yes,") + with pytest.raises(DesignFormatError): + migrate_legacy_design(data, discovery) + + +def test_a_legacy_design_that_breaks_a_rule_is_refused_with_the_code(tmp_path): + discovery = folders(tmp_path / "exp", "A/rep1", "A/rep2", "B/rep1", "B/rep2") + swapped = legacy( + tmp_path, + [ + row("A", "rep1", "rep2"), + row("A", "rep2", "rep1"), + row("B", "rep1", "rep2"), + row("B", "rep2", "rep1"), + ], + ) + with pytest.raises(LegacyMigrationRefused) as refused: + migrate_legacy_design(swapped, discovery) + assert refused.value.codes == ("E_REPEAT_ID_SHADOWS_SOURCE",) + twice = legacy(tmp_path, [row("A", "rep1", "G"), row("A", "rep2", "G")]) + with pytest.raises(LegacyMigrationRefused) as refused: + migrate_legacy_design(twice, discovery) + assert refused.value.codes == ("repeat_group_conflict",) + + +@pytest.mark.parametrize("group", ["z", "files", "overrides"]) +def test_a_legacy_repeat_group_that_is_not_an_allowed_id_opens_a_question(tmp_path, group): + """Owner decision 2026-11-07, Spec 4.6a: not refused, no proposal, never repaired.""" + discovery = folders(tmp_path / "exp", "A/rep1", "B/rep1") + data = legacy(tmp_path, [row("A", "rep1", group), row("B", "rep1", group)]) + draft = migrate_legacy_design(data, discovery) # not refused + repeat = draft.repeats[group] + assert repeat.id.proposed is None and repeat.id.confirmed is None + assert repeat.confirmed and repeat.confirmed_on == "2026-10-30" # the legacy answer is kept + assert [m.source_repeat for m in repeat.members] == ["rep1", "rep1"] + assert ("id_pending", f"repeat {group}") in { + (q.code, q.subject) for q in draft.open_questions() + } + assert not draft.publishable + with pytest.raises(DraftError) as refused: # the user cannot adopt it unchanged + draft.confirm_id("repeats", group, group) + assert "E_FORBIDDEN_FIELD" in refused.value.codes + named = draft.confirm_id("repeats", group, "R1") + assert named.repeats[group].id.confirmed == "R1" + + +def test_new_and_missing_folders_are_never_merged_silently(tmp_path): + discovery = folders(tmp_path / "exp", "A/rep1", "B/rep1", "B/rep2") + data = legacy(tmp_path, [row("A", "rep1"), row("B", "rep1"), row("B", "gone")]) + draft = migrate_legacy_design(data, discovery) + assert draft.legacy_missing == (("B", "gone"),) + assert draft.repeats["rep1"].confirmed is True + assert draft.repeats["rep2"].confirmed is False # a new folder is pending, not confirmed + codes = {(q.code, q.subject) for q in draft.open_questions()} + assert ("legacy_folder_missing", "B / gone") in codes + assert ("repeat_not_confirmed", "repeat rep2") in codes + joined = legacy(tmp_path, [row("A", "rep1"), row("B", "rep1")]) + rejoined = migrate_legacy_design( + joined, folders(tmp_path / "exp2", "A/rep1", "B/rep1", "C/rep1") + ) + assert rejoined.repeats["rep1"].confirmed is False # the new folder joined; asked again + assert rejoined.repeats["rep1"].confirmed_on == ""