diff --git a/CHANGELOG.md b/CHANGELOG.md
index a6761754..07a9d1ff 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -8,6 +8,15 @@ are in [docs/plan.md](docs/plan.md) §12. Upgrading from v0.3: see the
## Unreleased
+- **Notebook: your channels, labels first then tag per unit (Stage 1 PR 15, draft).** After the axis
+ roles are confirmed, *Which labels did you image?* lists what was stained (prefilled only from
+ header names that are valid IDs as they are and not generic such as `Ch1`, `C0`, `AF488`; near
+ spellings are grouped for the user to decide, never merged; a case collision is refused at once),
+ and *Tag each channel* shows one row per unit with a safe thumbnail, the quoted header name and a
+ dropdown of the labels per channel, plus a per-unit image type with nothing preselected. Labels
+ used in only some units say so (`A_SCOPE_INCOMPLETE`). Every answer is saved to the draft, which is
+ now version 3 (a unit's `channels` may hold `null` for not tagged, never `""`: every position needs a declared label, e.g. `unused`;
+ `image_type` may be `channels`; versions 1 and 2 are refused); contract §18.
- **Contract: the final mask check is as late as practicable, not a lock (#222, docs only).** Source masks are checked right before the manifest replace and ordinary concurrent edits are detected; the files are not locked, so a change between that check and the replace is not detected, and the manifest binds the bytes as they were at the final check (owner decision of 8 November 2026: documented, not snapshotted). `docs/contract.md` §9.4.
- **Notebook Step 1 cards: names, IDs, pairing and references (Stage 1 PR 14, draft).** After the
labels are confirmed, *Names for your experiment* proposes the folder names as IDs (only where
diff --git a/docs/contract.md b/docs/contract.md
index 5f43b5ac..cd691b98 100644
--- a/docs/contract.md
+++ b/docs/contract.md
@@ -665,15 +665,15 @@ One JSON object, UTF-8, two-space indent, sorted keys, LF, a final newline, no d
| Key | Value |
|---|---|
| `format` | `"morphobricks-description-draft"` |
-| `version` | `2` (an integer; version 1, written before `pairing` existed, is refused as an unknown version; any other value is refused as an unknown version, never read as a newer or older one) |
+| `version` | `3` (an integer; versions 1 and 2 are refused as unknown versions: version 2 had no tagging states; any other value is refused as an unknown version, never read as a newer or older one) |
| `base_sha256` | The lower-case hex SHA-256 of the exact bytes of the published `morphobricks.toml` the draft was derived from, or `"absent"`. Missing or any other value: `R_DRAFT_BASE_MISSING` |
| `spec_commit` | The 40-digit MorphoBricks-Spec commit the rules are pinned to (`SPEC_COMMIT`, equal to `tests/spec_fixtures/PIN.toml`) |
| `conditions` | Condition **source folder** → object with only the keys that were answered, from `id`, `display_name`, `reference` (the source folder of the condition it is compared with, or `""` when the user answered that it is compared with nothing; absent means not answered), `target`, `treatment_key` |
| `technical_repeats` | Technical-repeat source folder (`single` for none) → its confirmed ID |
| `repeats` | The pairing label of the draft's repeat (the folder name, or the legacy `repeat_group`) → object with only the answered keys, from `id`, `confirmed_on`, `display_name` |
| `pairing` | Condition source folder → source repeat folder → the pairing label of the repeat the user paired that folder with (only folders not in the repeat of their own name; applied to the draft first, so the `repeats` keys are the paired labels; refused with `pairing_after_confirmation` once a repeat is named or confirmed, `unknown_key` for a folder the draft lacks). A reference to the condition itself or one that closes a loop is refused when saved (`E_REFERENCE_SELF`, `E_REFERENCE_CYCLE`). |
-| `units` | `condition folder` → `source repeat folder` → `technical-repeat folder` → `{channels, image_type?, display_name?}`; `channels` lists the channel ID of C0, C1, ... in order |
-| `channels` | Channel ID → `{display_name?, header_names?}` (the aliases the user confirmed) |
+| `units` | `condition folder` → `source repeat folder` → `technical-repeat folder` → `{channels, image_type?, display_name?}`; `channels` lists, for C0, C1, ... in order, a channel ID (every position needs a label the user declared, for example `unused`; an empty string is refused) or `null` (not tagged yet); `image_type` is `"rgb"` or `"channels"` (an ordinary channel image; absent means not answered), any other value is refused |
+| `channels` | Channel ID → `{display_name?, header_names?}` (the label list of the notebook's *Your channels* screen; `header_names` are the raw header names of every cell the user tagged with that label, recomputed at each save: the tag is the alias confirmation) |
| `masks` | Mask ID → `{source_type, representation, source_channels?, display_name?}` |
Rules.
@@ -682,5 +682,5 @@ Rules.
- **Saving** (`save_draft`) is atomic (temporary file, one replace) and under the output folder's lock, after each answer. The answers are applied to the draft and the exact bytes are read back with the strict reader first, so an answer that would be refused on resume (a `spec_commit` other than the pinned one, `spec_commit_mismatch`; an ID the reader would refuse, a folder the draft does not have, a date that is not a real day) is refused when saved and nothing is written. An incomplete draft can be saved. The file is replaced, not merged. A symbolic link at the draft's path is refused and never replaced.
- **Reading** is strict and never repairs: bytes that are not UTF-8 JSON, a byte-order mark, a duplicate key, `NaN`, another `format`, an unknown `version`, an unknown key at any level, a missing key, a value of the wrong type or a date that is not a real `YYYY-MM-DD` day are `DraftFileError` (`code`: `malformed`, `unknown_version`, `unknown_key`, `R_DRAFT_BASE_MISSING`, `R_FILE_SYMLINK`). Reading accepts any 40-digit `spec_commit`; whether it is the current one is decided at resume and save, below.
- **Resume** (`load_draft`): the file's `spec_commit` must equal the pinned `SPEC_COMMIT`, else `DraftFileError` with code `spec_commit_mismatch` (owner decision, 2026-11-08): the confirmations in the file were given under other rules, so none is carried forward. The message names both commits, says the published `morphobricks.toml` is untouched and says to start a fresh draft from the folders. Library code never rewrites the stored commit and never deletes or replaces the stale draft; the caller derives a fresh draft and saves it, which replaces the file. `publish_draft` resumes first, so it refuses the same way and writes nothing. Then the file's `base_sha256` must equal the SHA-256 of the file published now (or `absent` if there is none), else `R_PUBLISHED_FILE_CHANGED` (`PublicationRefused`); the draft is never merged with the new file, it is derived again from it. An answer keyed by a folder or repeat the draft no longer has is `DraftError` (`unknown_key`), never moved to another folder.
-- **Draft to description** (`description_from_draft(draft, answers)`, pure): refuses with `DraftIncomplete`, listing **every** missing confirmation once (`id_pending`, `repeat_not_confirmed`, `confirmed_on_missing`, `control_unresolved`, `legacy_folder_missing`, `unit_channels_missing` for a unit with no channel list, `channel_undeclared` for a channel a unit names that is not declared), otherwise returns the `Description`. Nothing is confirmed for the user. `used_ids` is the draft's retired IDs plus every ID confirmed now. The description is then validated by the strict reader at publication; the function does not repeat the reader's rules.
+- **Draft to description** (`description_from_draft(draft, answers)`, pure): refuses with `DraftIncomplete`, listing **every** missing confirmation once (`id_pending`, `repeat_not_confirmed`, `confirmed_on_missing`, `control_unresolved`, `legacy_folder_missing`, `unit_channels_missing` for a unit with no channel list, `unit_channel_untagged` for a position still `null`, `channel_undeclared` for a channel a unit names that is not declared), otherwise returns the `Description`. Nothing is confirmed for the user. `used_ids` is the draft's retired IDs plus every ID confirmed now. The description is then validated by the strict reader at publication; the function does not repeat the reader's rules.
- **Publication** (`publish_draft`): under one hold of the lock the draft is resumed, the description built, published with all the checks of `publish_description` (symbolic link, base, transition, strict reader on the exact bytes), and then the draft file is deleted. Any refusal writes nothing: the draft and the published file stay byte for byte as they were. If the draft cannot be deleted after a successful publication, the error is raised and the publication stands; the leftover draft records a base that no longer matches and is refused on resume.
diff --git a/notebooks/MorphoBricks_Preprocessing.ipynb b/notebooks/MorphoBricks_Preprocessing.ipynb
index 039d6baf..616a52b0 100644
--- a/notebooks/MorphoBricks_Preprocessing.ipynb
+++ b/notebooks/MorphoBricks_Preprocessing.ipynb
@@ -317,6 +317,7 @@
" notebook.welcome(),\n",
" notebook.folder_step(),\n",
" notebook.axis_step(),\n",
+ " notebook.channels_step(),\n",
" notebook.run_step(),\n",
" notebook.results_step(),\n",
" ],\n",
diff --git a/pyproject.toml b/pyproject.toml
index ccca357b..2d257ad3 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -138,6 +138,12 @@ baseline = "tools/code_rules_baseline.json"
"morphobricks_preprocessing.mask_binding" = "io"
"morphobricks_preprocessing.mask_record" = "io"
"morphobricks_preprocessing.notebook_axes" = "ui"
+"morphobricks_preprocessing.notebook_channel_thumbs" = "ui"
+"morphobricks_preprocessing.notebook_channels" = "ui"
+"morphobricks_preprocessing.notebook_channels_grid" = "ui"
+"morphobricks_preprocessing.notebook_channels_labels" = "ui"
+"morphobricks_preprocessing.notebook_channels_type" = "ui"
+"morphobricks_preprocessing.notebook_channels_view" = "ui"
"morphobricks_preprocessing.notebook_axis_preview" = "ui"
"morphobricks_preprocessing.notebook_axis_step" = "ui"
"morphobricks_preprocessing.notebook_description" = "ui"
@@ -194,6 +200,9 @@ baseline = "tools/code_rules_baseline.json"
"morphobricks_preprocessing.design_derive" = "io"
"morphobricks_preprocessing.description_identity" = "pure"
"morphobricks_preprocessing.draft_ids" = "pure"
+"morphobricks_preprocessing.channel_labels" = "pure"
+"morphobricks_preprocessing.channel_units" = "pure"
+"morphobricks_preprocessing.channel_answers" = "pure"
"morphobricks_preprocessing.draft" = "io"
"morphobricks_preprocessing.draft_answers" = "io"
"morphobricks_preprocessing.draft_format" = "io"
diff --git a/src/morphobricks_preprocessing/channel_answers.py b/src/morphobricks_preprocessing/channel_answers.py
new file mode 100644
index 00000000..4409fdcb
--- /dev/null
+++ b/src/morphobricks_preprocessing/channel_answers.py
@@ -0,0 +1,127 @@
+"""Changes to the draft answers made by the *Your channels* cards, each returning new answers.
+
+The label list (``channels``), the tags and the image type of each unit (``units``) are all in the
+draft (:mod:`.draft_answers`); these functions change one slice and keep the rest exactly:
+
+- :func:`with_labels` replaces the label list (the colour tokens ``R``, ``G``, ``B`` of declared
+ colour units are kept) and takes back every tag of a label that is no longer in it;
+- :func:`with_tags` and :func:`with_type` change one unit; a colour unit is tagged ``R``, ``G``,
+ ``B`` and nothing else, an ordinary one starts untagged;
+- :func:`synced` recomputes what is derived from the tags: the header names each label is
+ evidence for (the tag is the user's confirmation of the alias) and the declared colour tokens.
+
+Nothing is guessed: a position is tagged only by a tag the user gave.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Collection, Mapping, Sequence
+from dataclasses import replace
+
+from .channel_labels import RGB_IDS, UnitLayout, aliases
+from .draft_answers import ChannelAnswer, DraftAnswers, UnitAnswer, UnitKey
+
+
+def label_ids(answers: DraftAnswers) -> list[str]:
+ """Return the declared labels, sorted: the channels that are not a colour token."""
+ return sorted(k for k in answers.channels if k not in RGB_IDS)
+
+
+def _unit(answers: DraftAnswers, layout: UnitLayout) -> UnitAnswer:
+ saved = answers.units.get(layout.key)
+ if saved is not None and len(saved.channels) == layout.channels:
+ return saved
+ return UnitAnswer(
+ (None,) * layout.channels,
+ None if saved is None else saved.image_type,
+ None if saved is None else saved.display_name,
+ )
+
+
+def _store(answers: DraftAnswers, key: UnitKey, unit: UnitAnswer) -> DraftAnswers:
+ units = dict(answers.units)
+ if all(t is None for t in unit.channels) and unit.image_type is None and not unit.display_name:
+ units.pop(key, None)
+ else:
+ units[key] = unit
+ return replace(answers, units=units)
+
+
+def with_tags(
+ answers: DraftAnswers, layout: UnitLayout, tags: Sequence[str | None]
+) -> DraftAnswers:
+ """Return the answers with the tags of one unit set (one entry per C index)."""
+ if len(tags) != layout.channels:
+ raise ValueError(f"{layout.key} has {layout.channels} channels, not {len(tags)} tags")
+ return _store(answers, layout.key, replace(_unit(answers, layout), channels=tuple(tags)))
+
+
+def with_type(answers: DraftAnswers, layout: UnitLayout, image_type: str | None) -> DraftAnswers:
+ """Return the answers with one unit's image type set.
+
+ A colour unit's channels are ``R``, ``G``, ``B``; going back to ordinary channels (or to no
+ answer) clears them, since nothing says which label each position then is.
+ """
+ unit = _unit(answers, layout)
+ if image_type == "rgb":
+ if layout.channels != len(RGB_IDS):
+ raise ValueError(f"a colour image has 3 channels, {layout.key} has {layout.channels}")
+ tags: tuple[str | None, ...] = RGB_IDS
+ elif unit.image_type == "rgb":
+ tags = (None,) * layout.channels
+ else:
+ tags = tuple(unit.channels)
+ return _store(answers, layout.key, replace(unit, channels=tags, image_type=image_type))
+
+
+def with_labels(answers: DraftAnswers, labels: Mapping[str, str | None]) -> DraftAnswers:
+ """Return the answers with exactly these labels (ID -> shown-as name or None).
+
+ A tag of a label that is gone becomes untagged (never moved to another label).
+ The colour tokens of declared colour units stay.
+ """
+ kept = {k: v for k, v in answers.channels.items() if k in RGB_IDS}
+ channels = {
+ label: replace(answers.channels.get(label, ChannelAnswer()), display_name=name or None)
+ for label, name in labels.items()
+ }
+ units = {}
+ for key, unit in answers.units.items():
+ if unit.image_type == "rgb":
+ units[key] = unit
+ continue
+ tags = tuple(t if t in labels else None for t in unit.channels)
+ units[key] = replace(unit, channels=tags)
+ return replace(answers, channels={**kept, **channels}, units=units)
+
+
+def untagged_by(answers: DraftAnswers, labels: Collection[str]) -> int:
+ """Return how many saved tags name a label that is not in ``labels``."""
+ return sum(
+ 1
+ for unit in answers.units.values()
+ if unit.image_type != "rgb"
+ for t in unit.channels
+ if t and t not in labels
+ )
+
+
+def synced(answers: DraftAnswers, layouts: Sequence[UnitLayout]) -> DraftAnswers:
+ """Return the answers with the derived parts recomputed from the saved tags.
+
+ Each label's ``header_names`` become the raw header names of every cell tagged with it (None
+ when none has a name), and ``R``, ``G``, ``B`` are declared exactly when some unit is a
+ colour unit.
+ """
+ found = aliases(layouts, answers.units)
+ channels = {
+ k: replace(v, header_names=found.get(k))
+ for k, v in answers.channels.items()
+ if k not in RGB_IDS
+ }
+ if any(u.image_type == "rgb" for u in answers.units.values()):
+ channels.update({token: ChannelAnswer() for token in RGB_IDS})
+ return replace(answers, channels=channels)
+
+
+__all__ = ["label_ids", "synced", "untagged_by", "with_labels", "with_tags", "with_type"]
diff --git a/src/morphobricks_preprocessing/channel_labels.py b/src/morphobricks_preprocessing/channel_labels.py
new file mode 100644
index 00000000..dc28e2d5
--- /dev/null
+++ b/src/morphobricks_preprocessing/channel_labels.py
@@ -0,0 +1,373 @@
+"""The rules behind the notebook's *Your channels* screen: labels, look-alikes and usage.
+
+A **label** is the ID of a channel and names *what was stained* (``DNA``, ``ACTIN``), never the
+dye or the microscope's slot. Nothing here reads a file or a widget; the cards of
+:mod:`.notebook_channels_labels` and :mod:`.notebook_channels_grid` only show what these
+functions decide.
+
+- :func:`is_generic` and :data:`GENERIC_PATTERNS`: header names that say nothing about the stain
+ (``Ch1``, ``C0``, ``Channel 2``, ``AF488``, ``DAPI``, ``GFP``, ``488 nm``). They are never
+ proposed as labels; the user may still type one.
+- :func:`propose_labels`: the labels a set of raw header names can offer (valid IDs as they are,
+ not generic, never repaired), and the **look-alike groups** (same letters ignoring case,
+ spaces, ``-``/``_``, or one character different) that the user decides on. Nothing is merged.
+- :func:`label_problems`: the refusals of a typed label list, a case collision included
+ (``E_ID_CASE_COLLISION``), at once.
+- :func:`usage`, :func:`positions`, :func:`aliases`, :func:`remaining`: the counts and the
+ header evidence derived from the tags the user gave.
+"""
+
+from __future__ import annotations
+
+import re
+from collections import Counter
+from collections.abc import Collection, Iterable, Mapping, Sequence
+from dataclasses import dataclass, field
+
+from .description_identity import nfc
+from .draft_answers import UnitAnswer, UnitKey
+from .draft_ids import id_problems
+
+RGB_IDS = ("R", "G", "B")
+"""The fixed channel IDs of a colour (RGB) unit (experiment-v1 section 4.4b): never labels."""
+
+GENERIC_PATTERNS: tuple[tuple[str, str], ...] = (
+ (r"(?:ch|chan|channel|c)[\s_\-=:.]*\d+", "a slot name of the microscope"),
+ (r"(?:track|detector|laser|line|pmt|ex|em)[\s_\-=:.]*\d+(?:\s*nm)?", "a slot name"),
+ (r"\d{3,4}\s*(?:nm)?(?:[\s_\-]*(?:ex|em|laser))?", "a wavelength"),
+ (
+ r"(?:af|alexa(?:[\s_\-]*fluor)?|cy|atto|dy|dylight|cf|ir[\s_\-]*dye)[\s_\-]*\d{1,3}\w*",
+ "a dye",
+ ),
+ (
+ r"dapi|hoechst(?:[\s_\-]*\d+)?|draq\d|fitc|tritc|texas[\s_\-]*red|"
+ r"[ey]?gfp|[ecy]fp|rfp|mcherry|mscarlet|tdtomato",
+ "a dye or fluorescent protein",
+ ),
+)
+"""Small, documented list of header names that are not labels: slot names and dyes.
+
+Each entry is a full-match, case-insensitive regular expression and the reason said to the user.
+The list is deliberately short; a name it misses is only proposed, never adopted, and the user
+confirms every label.
+"""
+_GENERIC = tuple((re.compile(p, re.IGNORECASE), why) for p, why in GENERIC_PATTERNS)
+_SEPARATORS = re.compile(r"[\s\-_]+")
+GENERIC = "generic"
+NOT_VALID = "not_valid"
+
+
+def generic_reason(name: str) -> str | None:
+ """Return why ``name`` is a generic header name (a slot or a dye), or None if it is not."""
+ text = nfc(name).strip()
+ return next((why for pattern, why in _GENERIC if pattern.fullmatch(text)), None)
+
+
+def is_generic(name: str) -> bool:
+ """Return True for a header name that names a slot or a dye, not what was stained."""
+ return generic_reason(name) is not None
+
+
+def _fold(name: str) -> str:
+ return nfc(name).casefold()
+
+
+def _squashed(name: str) -> str:
+ return _SEPARATORS.sub("", _fold(name))
+
+
+def _one_edit(a: str, b: str) -> bool:
+ """Return True when ``a`` and ``b`` differ by exactly one inserted, deleted or changed char."""
+ if a == b or abs(len(a) - len(b)) > 1:
+ return False
+ if len(a) > len(b):
+ a, b = b, a
+ i = 0
+ while i < len(a) and a[i] == b[i]:
+ i += 1
+ if len(a) == len(b):
+ return a[i + 1 :] == b[i + 1 :]
+ return a[i:] == b[i + 1 :]
+
+
+def look_alike(a: str, b: str) -> bool:
+ """Return True when two different spellings could be one label.
+
+ Same letters ignoring case, spaces, ``-`` and ``_``, or one character different (names of
+ at least three characters, so ``A`` and ``B`` are not look-alikes). The user decides; the
+ answer is never used to merge anything.
+ """
+ if a == b:
+ return False
+ if _fold(a) == _fold(b) or _squashed(a) == _squashed(b):
+ return True
+ return min(len(a.strip()), len(b.strip())) >= 3 and _one_edit(_fold(a), _fold(b))
+
+
+def case_only(a: str, b: str) -> bool:
+ """Return True when two different IDs are the same ignoring capital letters."""
+ return a != b and _fold(a) == _fold(b)
+
+
+@dataclass(frozen=True)
+class LabelProposal:
+ """What the header names can offer as labels.
+
+ ``single`` are names with no look-alike, valid as an ID as they stand; ``groups`` are sets of
+ look-alike spellings (two or more) for the user to decide, each member listed whether it
+ could be an ID or not; ``skipped`` are ``(name, reason)`` for names not offered: reason
+ :data:`GENERIC` (a slot or a dye) or :data:`NOT_VALID` (not an ID as it stands).
+ """
+
+ single: tuple[str, ...] = ()
+ groups: tuple[tuple[str, ...], ...] = ()
+ skipped: tuple[tuple[str, str], ...] = ()
+
+
+def _components(names: Sequence[str]) -> list[list[str]]:
+ """Return the connected sets of look-alike names (union-find), in the order first seen."""
+ parent = {name: name for name in names}
+
+ def find(name: str) -> str:
+ while parent[name] != name:
+ parent[name] = parent[parent[name]]
+ name = parent[name]
+ return name
+
+ for i, a in enumerate(names):
+ for b in names[i + 1 :]:
+ if look_alike(a, b):
+ parent[find(a)] = find(b)
+ found: dict[str, list[str]] = {}
+ for name in names:
+ found.setdefault(find(name), []).append(name)
+ return list(found.values())
+
+
+def usable_as_label(name: str, used: Collection[str] = ()) -> bool:
+ """Return True when ``name`` is a valid channel ID as it stands, not an RGB token."""
+ return not id_problems("channels", name, used) and name.casefold() not in {
+ t.casefold() for t in RGB_IDS
+ }
+
+
+def propose_labels(header_names: Iterable[str | None], used: Collection[str] = ()) -> LabelProposal:
+ """Return the labels the raw header names can offer, and the look-alikes to decide on.
+
+ A name is offered only if it is a valid channel ID exactly as written and is not generic
+ (:func:`is_generic`); nothing is trimmed, re-cased or repaired. Names that look alike
+ (:func:`look_alike`) are not offered one by one: they come back as a group, so the user says
+ whether they are one label (and which spelling is the ID) or different ones.
+ """
+ names = list(dict.fromkeys(n for n in header_names if n))
+ skipped = [(n, GENERIC) for n in names if is_generic(n)]
+ left = [n for n in names if not is_generic(n)]
+ single: list[str] = []
+ groups: list[tuple[str, ...]] = []
+ for members in _components(left):
+ if len(members) > 1:
+ groups.append(tuple(sorted(members)))
+ elif usable_as_label(members[0], used):
+ single.append(members[0])
+ else:
+ skipped.append((members[0], NOT_VALID))
+ return LabelProposal(tuple(single), tuple(groups), tuple(skipped))
+
+
+def label_problems(labels: Sequence[str], used: Collection[str] = ()) -> dict[int, tuple[str, ...]]:
+ """Return, by position in ``labels``, the codes that refuse that entry; empty ones omitted.
+
+ The reader's channel-ID rules (:func:`.draft_ids.id_problems`), plus a label given twice
+ (``E_DUPLICATE_VALUE``), two labels that differ only by capital letters
+ (``E_ID_CASE_COLLISION``, reported on both) and ``R``, ``G``, ``B`` (``E_RGB_TOKEN_MISUSE``:
+ the fixed names of colour images, not stains).
+ """
+ out: dict[int, tuple[str, ...]] = {}
+ for index, label in enumerate(labels):
+ codes = list(id_problems("channels", label, used))
+ others = [o for i, o in enumerate(labels) if i != index]
+ if label in others:
+ codes.append("E_DUPLICATE_VALUE")
+ if any(case_only(label, other) for other in others):
+ codes.append("E_ID_CASE_COLLISION")
+ if label.casefold() in {t.casefold() for t in RGB_IDS}:
+ codes.append("E_RGB_TOKEN_MISUSE")
+ if codes:
+ out[index] = tuple(dict.fromkeys(codes))
+ return out
+
+
+@dataclass(frozen=True)
+class UnitLayout:
+ """One imaging unit as the channel screen needs it, read from its representative image.
+
+ ``image`` is the first image of the unit by sorted path, whose header gives ``channels`` (the
+ number of C indices, 3 for a colour image) and ``names`` (the raw header names by C index,
+ None where it has none). ``problem`` is why the unit cannot be tagged (no readable first
+ image, roles that do not fit it); then ``channels`` is 0.
+ """
+
+ key: UnitKey
+ image: str | None
+ channels: int = 0
+ names: tuple[str | None, ...] = ()
+ rgb: bool = False
+ problem: str = ""
+ images: int = 0
+
+ @property
+ def usable(self) -> bool:
+ """True when the unit's channels are known and can be tagged."""
+ return not self.problem and self.channels > 0
+
+ @property
+ def signature(self) -> tuple[int, tuple[str | None, ...], bool]:
+ """What units must share to be one layout: count, raw names and colour."""
+ return (self.channels, self.names, self.rgb)
+
+ def name(self, index: int) -> str | None:
+ """Return the raw header name at C index ``index``, None when the header has none."""
+ return self.names[index] if index < len(self.names) and self.names[index] else None
+
+
+def tags_of(layout: UnitLayout, answer: UnitAnswer | None) -> tuple[str | None, ...]:
+ """Return the saved tags of a unit as one entry per C index; None where not tagged.
+
+ A saved list whose length is not the image's channel count no longer fits it and is not
+ used: nothing is stretched, cut or moved to another position.
+ """
+ saved = () if answer is None else tuple(answer.channels)
+ return saved if len(saved) == layout.channels else (None,) * layout.channels
+
+
+def fits(layout: UnitLayout, answer: UnitAnswer | None) -> bool:
+ """Return True when there is no saved tag list, or it has this image's channel count."""
+ return answer is None or not answer.channels or len(answer.channels) == layout.channels
+
+
+def proposal_for(layout: UnitLayout, labels: Collection[str]) -> tuple[str, ...] | None:
+ """Return the tags a unit's header names propose, or None when they do not all match.
+
+ Only when **every** channel has a header name equal, exactly (NFC, no trimming, no case
+ folding), to a different label of ``labels``. A colour image proposes nothing.
+ """
+ if not layout.usable or layout.rgb or len(layout.names) != layout.channels:
+ return None
+ names = [n for n in layout.names if n]
+ if len(names) != layout.channels or len(set(names)) != len(names):
+ return None
+ known = {nfc(label): label for label in labels}
+ found = [known.get(nfc(name)) for name in names]
+ return None if None in found else tuple(f for f in found if f)
+
+
+def usage(
+ layouts: Sequence[UnitLayout], units: Mapping[UnitKey, UnitAnswer], labels: Iterable[str]
+) -> dict[str, tuple[UnitKey, ...]]:
+ """Return, for each label, the units that tag at least one channel with it (saved tags)."""
+ found: dict[str, list[UnitKey]] = {label: [] for label in labels}
+ for layout in layouts:
+ tags = set(tags_of(layout, units.get(layout.key)))
+ for label in found:
+ if label in tags and units[layout.key].image_type != "rgb":
+ found[label].append(layout.key)
+ return {label: tuple(keys) for label, keys in found.items()}
+
+
+def positions(
+ layouts: Sequence[UnitLayout], units: Mapping[UnitKey, UnitAnswer], labels: Iterable[str]
+) -> dict[str, Counter]:
+ """Return, for each label, how many units tag it at each C index (saved tags)."""
+ out: dict[str, Counter] = {label: Counter() for label in labels}
+ for layout in layouts:
+ for index, tag in enumerate(tags_of(layout, units.get(layout.key))):
+ if tag in out:
+ out[tag][index] += 1
+ return out
+
+
+def aliases(
+ layouts: Sequence[UnitLayout], units: Mapping[UnitKey, UnitAnswer]
+) -> dict[str, tuple[str, ...]]:
+ """Return the raw header names of every cell tagged with a label, by label, sorted.
+
+ The tag is the user's confirmation that the header name is evidence for the label
+ (experiment-v1 section 4.3); names are kept exactly as read. A colour unit has none (``R``,
+ ``G`` and ``B`` take no aliases), and a label whose cells have no header name is omitted.
+ """
+ found: dict[str, set[str]] = {}
+ for layout in layouts:
+ answer = units.get(layout.key)
+ if answer is None or answer.image_type == "rgb":
+ continue
+ for index, tag in enumerate(tags_of(layout, answer)):
+ name = layout.name(index)
+ if tag and name:
+ found.setdefault(tag, set()).add(name)
+ return {label: tuple(sorted(names)) for label, names in found.items()}
+
+
+@dataclass(frozen=True)
+class Remaining:
+ """What is still open on the tag grid: cells, image types and units that cannot be tagged."""
+
+ cells: int = 0
+ types: int = 0
+ blocked: int = 0
+ units: tuple[UnitKey, ...] = field(default=())
+
+ @property
+ def done(self) -> bool:
+ """True when every position of every unit is tagged and every type answered."""
+ return not (self.cells or self.types or self.blocked)
+
+
+def remaining(
+ layouts: Sequence[UnitLayout], units: Mapping[UnitKey, UnitAnswer], labels: Collection[str]
+) -> Remaining:
+ """Return what the user still has to answer: every position of every unit, every type."""
+ cells = types = blocked = 0
+ open_units: list[UnitKey] = []
+ for layout in layouts:
+ answer = units.get(layout.key)
+ if not layout.usable:
+ blocked += 1
+ open_units.append(layout.key)
+ continue
+ left = 0
+ if answer is None or answer.image_type is None:
+ left += 1
+ types += 1
+ if answer is None or answer.image_type != "rgb":
+ tags = tags_of(layout, answer)
+ missing = sum(1 for t in tags if t is None or t not in labels)
+ cells += missing
+ left += missing
+ if left:
+ open_units.append(layout.key)
+ return Remaining(cells, types, blocked, tuple(open_units))
+
+
+__all__ = [
+ "GENERIC",
+ "GENERIC_PATTERNS",
+ "NOT_VALID",
+ "RGB_IDS",
+ "LabelProposal",
+ "Remaining",
+ "UnitLayout",
+ "aliases",
+ "case_only",
+ "fits",
+ "generic_reason",
+ "is_generic",
+ "label_problems",
+ "look_alike",
+ "positions",
+ "propose_labels",
+ "proposal_for",
+ "remaining",
+ "tags_of",
+ "usable_as_label",
+ "usage",
+]
diff --git a/src/morphobricks_preprocessing/channel_units.py b/src/morphobricks_preprocessing/channel_units.py
new file mode 100644
index 00000000..17417097
--- /dev/null
+++ b/src/morphobricks_preprocessing/channel_units.py
@@ -0,0 +1,101 @@
+"""The imaging units of an experiment as the channel screen shows them (headers only).
+
+One :class:`.channel_labels.UnitLayout` per unit: the first image of the unit by sorted path is its
+representative, the channel count comes from the axis roles the user confirmed (a colour image
+has three samples), and the raw header names are read exactly as the header holds them. No pixel
+is read here and nothing is inferred from the image content.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Collection, Iterable
+
+from .axes import AxisReview
+from .channel_labels import UnitLayout
+from .discovery import DiscoveryResult
+from .draft_answers import UnitKey
+from .metadata import MetadataReport, TiffMetadata, header_channel_count
+
+
+def channel_count(meta: TiffMetadata, review: AxisReview) -> int | None:
+ """Return the channels of one image under the confirmed roles; None when they do not fit.
+
+ A colour (RGB) image has its three samples; otherwise the length of the ``C`` role, and one
+ channel when the file has no ``C`` axis.
+ """
+ mapping = review.file_mappings.get(meta.relative_path)
+ if mapping is None or meta.relative_path in review.axis_error_files:
+ return None
+ if meta.rgb_samples:
+ return header_channel_count(meta)
+ index = mapping.index_of("C")
+ return 1 if index is None else int(meta.shape[index])
+
+
+def slot_count(metadata: MetadataReport, review: AxisReview) -> int:
+ """Return the largest channel count of any non-colour image (0 if none)."""
+ counts = [
+ channel_count(f, review) or 0 for f in metadata.files if f.readable and not f.rgb_samples
+ ]
+ return max(counts, default=0)
+
+
+def header_names(metadata: MetadataReport) -> list[str]:
+ """Return every raw channel name any readable image's header gives, once each, in order."""
+ names = (n for f in metadata.files if f.readable for n in f.raw_channel_names if n)
+ return list(dict.fromkeys(names))
+
+
+def unit_images(
+ discovery: DiscoveryResult, excluded: Collection[str]
+) -> dict[UnitKey, tuple[str, ...]]:
+ """Return the images to analyse of every unit, sorted by path."""
+ found: dict[UnitKey, list[str]] = {}
+ for leaf in discovery.leaves:
+ key = (leaf.condition, leaf.biological_repeat, leaf.technical_repeat)
+ found.setdefault(key, []).extend(i for i in leaf.images if i not in excluded)
+ return {key: tuple(sorted(images)) for key, images in found.items() if images}
+
+
+def unit_layouts(
+ keys: Iterable[UnitKey],
+ discovery: DiscoveryResult,
+ excluded: Collection[str],
+ metadata: MetadataReport,
+ review: AxisReview,
+) -> tuple[UnitLayout, ...]:
+ """Return one layout per unit key, from the unit's first image by sorted path.
+
+ A unit whose first image cannot be read, or does not fit the confirmed roles, is returned
+ with a ``problem`` and no channels: the next image is **not** used instead.
+ """
+ images = unit_images(discovery, excluded)
+ headers = {f.relative_path: f for f in metadata.files}
+ out = []
+ for key in sorted(keys):
+ found = images.get(key, ())
+ if not found:
+ out.append(UnitLayout(key, None, problem="the unit has no image to analyse"))
+ continue
+ first = found[0]
+ meta = headers.get(first)
+ count = channel_count(meta, review) if meta is not None and meta.readable else None
+ if count is None:
+ out.append(
+ UnitLayout(
+ key,
+ first,
+ problem=f"{first} cannot be read or does not fit the confirmed axis roles",
+ images=len(found),
+ )
+ )
+ continue
+ out.append(
+ UnitLayout(
+ key, first, count, tuple(meta.raw_channel_names), meta.rgb_samples, "", len(found)
+ )
+ )
+ return tuple(out)
+
+
+__all__ = ["channel_count", "header_names", "slot_count", "unit_images", "unit_layouts"]
diff --git a/src/morphobricks_preprocessing/draft_answers.py b/src/morphobricks_preprocessing/draft_answers.py
index 51289fc7..1d2ce9a6 100644
--- a/src/morphobricks_preprocessing/draft_answers.py
+++ b/src/morphobricks_preprocessing/draft_answers.py
@@ -57,9 +57,15 @@ class RepeatAnswer:
@dataclass(frozen=True)
class UnitAnswer:
- """The channel ID of each C index of one imaging unit, in order, and its optional type."""
+ """The channel ID of each C index of one imaging unit, in order, and its image type.
- channels: tuple[str, ...]
+ An entry of ``channels`` is a declared channel ID or None while that position is not
+ tagged (every position needs a label the user declared, for example ``unused``).
+ ``image_type`` is ``"rgb"``, ``"channels"`` (an ordinary channel image,
+ the description's absent type) or None while not answered.
+ """
+
+ channels: tuple[str | None, ...]
image_type: str | None = None
display_name: str | None = None
@@ -289,8 +295,9 @@ def missing_confirmations(draft: Draft, answers: DraftAnswers) -> tuple[OpenQues
"""Return every confirmation still missing for ``draft`` with ``answers`` applied.
The draft's own open questions (IDs, repeats, controls, legacy folders), plus the channels
- of every imaging unit that has none (``unit_channels_missing``) and every channel a unit
- names that is not declared (``channel_undeclared``).
+ of every imaging unit that has none (``unit_channels_missing``), every position not tagged
+ (``unit_channel_untagged``) and every channel a unit names that is not declared
+ (``channel_undeclared``).
"""
out = list(draft.open_questions())
declared = set(answers.channels)
@@ -300,8 +307,14 @@ def missing_confirmations(draft: Draft, answers: DraftAnswers) -> tuple[OpenQues
answer = answers.units.get(key)
if answer is None or not answer.channels:
out.append(OpenQuestion("unit_channels_missing", " / ".join(key)))
- else:
- undeclared.update(c for c in answer.channels if c not in declared)
+ continue
+ where = " / ".join(key)
+ out += [
+ OpenQuestion("unit_channel_untagged", f"{where} / C{i}")
+ for i, c in enumerate(answer.channels)
+ if c is None
+ ]
+ undeclared.update(c for c in answer.channels if c and c not in declared)
return tuple(out + [OpenQuestion("channel_undeclared", c) for c in sorted(undeclared)])
@@ -333,8 +346,9 @@ def _repeat(draft: Draft, label: str, answers: DraftAnswers) -> Repeat:
units = {}
for u in member.units:
given = answers.units[(member.condition, member.source_repeat, u.source_folder)]
+ kind = None if given.image_type == "channels" else given.image_type
units[draft.technical_repeats[u.source_folder].confirmed] = Unit(
- u.source_folder, given.channels, given.display_name, given.image_type
+ u.source_folder, tuple(given.channels), given.display_name, kind
)
members[cid] = Member(member.source_repeat, member.image_count, units)
shown = answers.repeats.get(label, RepeatAnswer()).display_name
diff --git a/src/morphobricks_preprocessing/draft_format.py b/src/morphobricks_preprocessing/draft_format.py
index af074ee6..2e942500 100644
--- a/src/morphobricks_preprocessing/draft_format.py
+++ b/src/morphobricks_preprocessing/draft_format.py
@@ -32,7 +32,9 @@
from .run.description_file import ABSENT
DRAFT_FORMAT = "morphobricks-description-draft"
-DRAFT_VERSION = 2
+DRAFT_VERSION = 3
+IMAGE_TYPES = ("rgb", "channels")
+"""The image types a draft may hold: ``channels`` is the answer "ordinary channel image"."""
SPEC_COMMIT = "c8f68526c08633526395ca0180fd6bd1a182fc84"
"""The MorphoBricks-Spec commit the rules of this brick are pinned to (``tests/spec_fixtures``)."""
BASE_MISSING = "R_DRAFT_BASE_MISSING"
@@ -107,6 +109,15 @@ def _texts(value: object, where: str) -> tuple[str, ...]:
return tuple(_text(item, where) for item in value)
+def _tags(value: object, where: str) -> tuple[str | None, ...]:
+ """Read the channels of a unit: a channel ID or null (not tagged yet)."""
+ if not isinstance(value, list):
+ raise _bad(where, "a list is expected")
+ if not all(item is None or (isinstance(item, str) and item) for item in value):
+ raise _bad(where, "each entry is a channel ID (text) or null")
+ return tuple(value)
+
+
def _optional_texts(table: dict, key: str, where: str) -> tuple[str, ...] | None:
return _texts(table[key], f"{where}.{key}") if key in table else None
@@ -136,9 +147,12 @@ def _units(value: object) -> dict[tuple[str, str, str], UnitAnswer]:
for folder, raw in _entries(technical, f"units.{condition}.{repeat}").items():
where = f"units.{condition}.{repeat}.{folder}"
item = _table(raw, where, UNIT_KEYS, ("channels",))
+ kind = _optional(item, "image_type", where)
+ if kind is not None and kind not in IMAGE_TYPES:
+ raise _bad(f"{where}.image_type", f"one of {IMAGE_TYPES} is expected")
out[(condition, repeat, folder)] = UnitAnswer(
- _texts(item["channels"], f"{where}.channels"),
- _optional(item, "image_type", where),
+ _tags(item["channels"], f"{where}.channels"),
+ kind,
_optional(item, "display_name", where),
)
return out
@@ -318,6 +332,7 @@ def answers_to_bytes(answers: DraftAnswers) -> bytes:
"BASE_MISSING",
"DRAFT_FORMAT",
"DRAFT_VERSION",
+ "IMAGE_TYPES",
"SPEC_COMMIT",
"DraftFileError",
"answers_from_bytes",
diff --git a/src/morphobricks_preprocessing/notebook_axis_preview.py b/src/morphobricks_preprocessing/notebook_axis_preview.py
index 8761bc18..4e3c2474 100644
--- a/src/morphobricks_preprocessing/notebook_axis_preview.py
+++ b/src/morphobricks_preprocessing/notebook_axis_preview.py
@@ -169,6 +169,7 @@ def _toggle_confirmation(self) -> None:
self._render_preview_if_shown()
self._show_calibration()
self._refresh_run_step()
+ self._refresh_channels()
def _open_image_type(self) -> None:
"""Suggest the image type for the confirmed roles and ask the user to confirm it."""
diff --git a/src/morphobricks_preprocessing/notebook_channel_thumbs.py b/src/morphobricks_preprocessing/notebook_channel_thumbs.py
new file mode 100644
index 00000000..7bbf2592
--- /dev/null
+++ b/src/morphobricks_preprocessing/notebook_channel_thumbs.py
@@ -0,0 +1,84 @@
+"""Safe thumbnails of the channels of each unit's representative image (PR 13 rules).
+
+Every picture goes through :func:`.preview.channel_strip`: reduced first (every n-th pixel),
+then float32, then its own percentile contrast, flat images grey and non-finite pixels magenta,
+and only through the axis roles the user confirmed. A picture never names a channel. A picture
+that cannot be shown is replaced by the reason, never by a guess.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+
+from .axes import AxisReview
+from .channel_labels import UnitLayout
+from .preview import (
+ DETAIL_SIDE,
+ THUMBNAIL_SIDE,
+ PreviewCache,
+ PreviewRefused,
+ StripOptions,
+ channel_strip,
+)
+from .preview_render import png_bytes
+
+CONTRAST_TEXT = (
+ "Each picture is stretched on its own, from the 0.5th to the 99.5th percentile of its own "
+ "pixels (display only: the images are not changed), so brightness cannot be compared between "
+ "pictures. A flat picture is grey and a non-finite pixel is magenta. The pictures only help "
+ "you check the labels you choose: nothing is guessed from them."
+)
+
+
+@dataclass(frozen=True)
+class Thumb:
+ """One channel picture: PNG bytes (None when it cannot be shown) and a short note."""
+
+ png: bytes | None
+ note: str = ""
+
+
+class Thumbnails:
+ """Reads and keeps the channel pictures of the units' representative images."""
+
+ def __init__(self, root: Path, review: AxisReview, memory_limit: int) -> None:
+ self.root, self.review, self.memory_limit = root, review, memory_limit
+ self._cache = PreviewCache()
+ self._strips: dict[tuple[str, int], tuple[Thumb, ...] | Thumb] = {}
+ self.where: dict[str, str] = {}
+
+ def _read(self, layout: UnitLayout, side: int) -> tuple[Thumb, ...] | Thumb:
+ """Read every channel of the unit's image once per size; a refusal becomes one note."""
+ key = (layout.image or "", side)
+ if key not in self._strips:
+ options = StripOptions(side=side, cache=self._cache, memory_limit=self.memory_limit)
+ try:
+ strip = channel_strip(self.root / key[0], key[0], self.review, options)
+ except (PreviewRefused, OSError, ValueError, IndexError) as exc:
+ self._strips[key] = Thumb(None, str(exc))
+ else:
+ z = f"depth slice {strip.z}" if strip.z is not None else "no depth axis"
+ t = f"time point {strip.t}" if strip.t is not None else "no time axis"
+ self.where[key[0]] = f"{z}, {t}"
+ self._strips[key] = tuple(
+ Thumb(png_bytes(p.panel.pixels), p.panel.note) for p in strip.panels
+ )
+ return self._strips[key]
+
+ def get(self, layout: UnitLayout, c: int, *, detail: bool = False) -> Thumb:
+ """Return channel ``c`` of the unit's representative image, 256 px or 1024 px detail."""
+ if not layout.image or not layout.usable:
+ return Thumb(None, layout.problem)
+ found = self._read(layout, DETAIL_SIDE if detail else THUMBNAIL_SIDE)
+ if isinstance(found, Thumb):
+ return found
+ return found[c] if c < len(found) else Thumb(None, f"the image has no channel {c}")
+
+ def plane(self, layout: UnitLayout) -> str:
+ """Return which plane the pictures of this unit show ("depth slice 3, time point 0")."""
+ self.get(layout, 0)
+ return self.where.get(layout.image or "", "")
+
+
+__all__ = ["CONTRAST_TEXT", "Thumb", "Thumbnails"]
diff --git a/src/morphobricks_preprocessing/notebook_channels.py b/src/morphobricks_preprocessing/notebook_channels.py
new file mode 100644
index 00000000..e491e857
--- /dev/null
+++ b/src/morphobricks_preprocessing/notebook_channels.py
@@ -0,0 +1,157 @@
+"""The notebook's *Your channels* screen: step A (labels) and step B (tags) assembled.
+
+A thin layout over three self-contained pieces, each with its own saving and resuming, so a later
+layout can put them on separate steps without rewriting any logic:
+
+- :class:`.notebook_channels_labels.LabelsCard` (step A, the label list),
+- :class:`.notebook_channels_grid.TagGrid` (step B, the tag grid),
+- :class:`.notebook_channels_type.ImageTypeColumn` (the per-unit image type, placed in the grid).
+
+They read and save the draft answers through :class:`_NotebookStore`, the notebook's own
+``_answer``. The screen needs the axis roles the user confirmed (the pictures are read only
+through them, nothing is guessed from the shape of an image) and the checked folder's draft.
+"""
+
+from __future__ import annotations
+
+import ipywidgets as widgets
+
+from . import notebook_style as ui
+from .axis_review import review_axes
+from .channel_labels import UnitLayout
+from .channel_units import header_names, slot_count, unit_layouts
+from .draft_answers import DraftAnswers
+from .notebook_channel_thumbs import Thumbnails
+from .notebook_channels_grid import TagGrid
+from .notebook_channels_labels import LabelsCard
+from .notebook_channels_type import ImageTypeColumn
+
+INTRO = (
+ "What happens here: first name what you stained (the labels), then say, for every "
+ "unit, which label each channel is, looking at its picture. Nothing is chosen for you: "
+ "names found in the image headers are offered only as suggestions, and each one needs your "
+ "click."
+)
+
+
+class _NotebookStore:
+ """The cards' view of the notebook's draft answers (the notebook's own save)."""
+
+ def __init__(self, notebook: object) -> None:
+ self._nb = notebook
+
+ def answers(self) -> DraftAnswers:
+ """Return the current answers."""
+ return self._nb.draft_answers
+
+ def update(self, new: DraftAnswers) -> bool:
+ """Save the answers as the description cards do; False when they are refused."""
+ return self._nb._answer(new, rebuild=False)
+
+ def refusal(self) -> str:
+ """Return the reason the last answer was refused."""
+ return str(self._nb._refusal or "the answers were refused")
+
+ def used_channels(self) -> tuple[str, ...]:
+ """Return the channel IDs the lineage retired."""
+ base = self._nb._draft_base
+ return tuple(base.used_ids.get("channels", ())) if base is not None else ()
+
+
+class _ChannelsStep:
+ """The *Your channels* screen of the notebook."""
+
+ def _build_channels_step(self) -> None:
+ """Build the label card, the tag grid and the image-type column (hidden until needed)."""
+ store = _NotebookStore(self)
+ self._channels_key: tuple | None = None
+ self._channel_layouts: tuple[UnitLayout, ...] = ()
+ self._channel_thumbs: Thumbnails | None = None
+ self.channel_types = ImageTypeColumn(store, self._types_changed)
+ self.channel_grid = TagGrid(store, self.channel_types, self._channels_changed)
+ self.channel_labels = LabelsCard(store, self._channel_labels_changed)
+ self.channels_waiting = widgets.HTML()
+ self.channels_box = widgets.VBox(
+ [self.channels_waiting, self.channel_labels.widget, self.channel_grid.widget]
+ )
+
+ def channels_step(self, title: str | None = None, purpose: str | None = None) -> widgets.Widget:
+ """Return the channels screen: labels, then tags and image type per unit."""
+ return ui.page(
+ "Your channels" if title is None else title,
+ "Name what you stained, then tag every channel of every unit."
+ if purpose is None
+ else purpose,
+ [ui.intro(INTRO), self.channels_box],
+ )
+
+ # Opening and closing -----------------------------------------------------------------
+
+ def _hide_channels(self, why: str = "") -> None:
+ self._channels_key = None
+ self._channel_thumbs = None
+ self.channel_labels.close()
+ self.channel_grid.close()
+ self.channels_waiting.value = why
+
+ def _refresh_channels(self) -> None:
+ """Open the screen for the checked folder once its draft and axis roles are known."""
+ pre = self.preflight
+ if pre is None or self.draft is None or self.draft_answers is None:
+ self._hide_channels(
+ ui.status("info", "Waiting for step 1.", ["Check the folder and answer its cards."])
+ )
+ return
+ if self.confirmed_mapping is None:
+ self._hide_channels(
+ ui.status(
+ "info",
+ "Waiting for the axis roles.",
+ [
+ "Confirm the axis roles in step 2 first: the pictures are read only "
+ "through the roles you confirmed, never guessed from the shape."
+ ],
+ )
+ )
+ return
+ review = review_axes(
+ pre.metadata,
+ self.confirmed_mapping,
+ missing_as_singleton=self.missing_as_singleton.value,
+ )
+ keys = [
+ (r.condition, r.biological_repeat, r.technical_repeat) for r in self.draft.unit_rows()
+ ]
+ layouts = unit_layouts(keys, pre.discovery, pre.excluded, pre.metadata, review)
+ slots, names = slot_count(pre.metadata, review), header_names(pre.metadata)
+ key = (str(pre.discovery.root), self.confirmed_mapping, layouts, slots, tuple(names))
+ if key == self._channels_key:
+ return
+ self._channels_key = key
+ self._channel_layouts = layouts
+ self._channel_thumbs = Thumbnails(pre.discovery.root, review, self._memory_limit())
+ self.channels_waiting.value = ""
+ self.channel_labels.open(slots, names, layouts)
+ self._channel_labels_changed()
+
+ def _channel_labels_changed(self) -> None:
+ """Open the tag grid for the confirmed labels, or close it while they are being changed."""
+ if self.channel_labels.confirmed and self._channels_key is not None:
+ self.channel_grid.open(
+ self._channel_layouts, self.channel_labels.labels, self._channel_thumbs
+ )
+ else:
+ self.channel_grid.close()
+
+ def _types_changed(self) -> None:
+ """Redraw the grid after an image type (its rows hold the type cells)."""
+ self.channel_grid.refresh()
+ self._channels_changed()
+
+ def _channels_changed(self) -> None:
+ """Take the notebook's later steps into account after a tag or an image type."""
+ if self.confirmed_mapping is not None:
+ self._refresh_axis_next()
+
+
+__all__ = ["_ChannelsStep"]
diff --git a/src/morphobricks_preprocessing/notebook_channels_grid.py b/src/morphobricks_preprocessing/notebook_channels_grid.py
new file mode 100644
index 00000000..8a10795e
--- /dev/null
+++ b/src/morphobricks_preprocessing/notebook_channels_grid.py
@@ -0,0 +1,401 @@
+"""Step B of *Your channels*: "Tag each channel" (a self-contained card).
+
+An image-first grid: one row per imaging unit (condition / biological repeat / technical repeat),
+one column per C index. Each cell is a safe thumbnail of the unit's representative image (about
+160 px, **Enlarge** for the 1024 px view), the C index, the header name quoted exactly as read and
+a dropdown limited to the labels of step A (an unused slot needs a label too, for example
+*unused*); there is no free text. A
+unit whose header names equal the labels exactly is pre-tagged, but the pre-tag is only a
+suggestion until the user confirms it once per layout (units with the same header names), beside
+the thumbnails. Step B is complete only when every channel position of every unit is tagged and
+every image type answered. Labels used in only some units say so, with the consequence.
+
+The card reads and saves through an :class:`.notebook_channels_view.AnswerStore`; the image-type
+column is a separate piece (:class:`.notebook_channels_type.ImageTypeColumn`) the card places in
+each row.
+"""
+
+from __future__ import annotations
+
+import html
+from collections.abc import Callable, Sequence
+
+import ipywidgets as widgets
+
+from . import notebook_style as ui
+from .channel_answers import synced, with_tags
+from .channel_labels import (
+ UnitLayout,
+ positions,
+ proposal_for,
+ remaining,
+ tags_of,
+ usage,
+)
+from .draft_answers import UnitKey
+from .notebook_channel_thumbs import CONTRAST_TEXT, Thumbnails
+from .notebook_channels_type import ImageTypeColumn
+from .notebook_channels_view import (
+ CHOOSE,
+ AnswerStore,
+ blank_chip,
+ chip,
+ columns,
+ quoted,
+ unit_name,
+)
+
+THUMB_PX = "160px"
+
+
+def partial_text(label: str, used: int, total: int) -> str:
+ """Return the sentence for a label used in only some units, with its consequence."""
+ return (
+ f"used in {used} of {total} units: an analysis of {html.escape(label)} will be "
+ "refused for the other units"
+ )
+
+
+class TagGrid:
+ """The tag grid: rows of units, a dropdown per channel, one confirm per layout."""
+
+ def __init__(
+ self,
+ store: AnswerStore,
+ types: ImageTypeColumn,
+ on_change: Callable[[], None] = lambda: None,
+ ) -> None:
+ self.store, self.types, self._on_change = store, types, on_change
+ self.layouts: Sequence[UnitLayout] = ()
+ self.labels: list[str] = []
+ self.thumbs: Thumbnails | None = None
+ self.cells: dict[tuple[UnitKey, int], widgets.Dropdown] = {}
+ self.confirm_buttons: list[widgets.Button] = []
+ self.message = ""
+ self._building = False
+ self.blocks = widgets.VBox()
+ self.usage_html = widgets.HTML()
+ self.status = widgets.HTML()
+ self.problem = widgets.HTML()
+ self.detail = widgets.VBox()
+ self.contrast = widgets.HTML(f"{CONTRAST_TEXT}
")
+ self.plane = widgets.HTML()
+ self.widget = ui.card(
+ "B. Tag each channel",
+ [
+ ui.help_text(
+ "For every unit, say which label each channel is, looking at its picture. "
+ "The same label should be in the same column in every row; a label used in "
+ "only some units is shown in orange. Every channel and every image type "
+ "must be answered."
+ ),
+ self.contrast,
+ self.plane,
+ self.blocks,
+ self.detail,
+ self.problem,
+ self.usage_html,
+ self.status,
+ ],
+ visible=False,
+ )
+
+ # Opening -----------------------------------------------------------------------------
+
+ def open(self, layouts: Sequence[UnitLayout], labels: list[str], thumbs: Thumbnails) -> None:
+ """Show the grid for these units and labels (the confirmed label list of step A)."""
+ self.layouts, self.labels, self.thumbs = layouts, labels, thumbs
+ self.types.open(layouts)
+ self.detail.children = []
+ self.refresh()
+ ui.set_visible(self.widget, True)
+
+ def close(self) -> None:
+ """Hide the grid."""
+ ui.set_visible(self.widget, False)
+
+ @property
+ def complete(self) -> bool:
+ """True when every position of every unit is tagged and every type answered."""
+ return remaining(self.layouts, self.store.answers().units, self.labels).done
+
+ # Tags --------------------------------------------------------------------------------
+
+ def _proposal(self, layout: UnitLayout) -> tuple[str, ...] | None:
+ return proposal_for(layout, self.labels)
+
+ def effective(self, layout: UnitLayout) -> tuple[tuple[str | None, ...], set[int]]:
+ """Return the tags a unit shows and which of them are only proposals.
+
+ The user's saved tags first; a position not tagged takes the header proposal if that
+ label is not already tagged elsewhere in the unit.
+ """
+ saved = list(tags_of(layout, self.store.answers().units.get(layout.key)))
+ proposal = self._proposal(layout)
+ proposed: set[int] = set()
+ if proposal is not None and self.types.saved(layout.key) != "rgb":
+ for index, tag in enumerate(saved):
+ if tag is None and proposal[index] not in saved:
+ saved[index] = proposal[index]
+ proposed.add(index)
+ return tuple(saved), proposed
+
+ def _tagged(self, layout: UnitLayout, index: int, value: str | None) -> None:
+ """Save one cell's tag; a label already tagged elsewhere in the unit is refused."""
+ if self._building:
+ return
+ answers = self.store.answers()
+ tags = list(tags_of(layout, answers.units.get(layout.key)))
+ if value and value in [t for i, t in enumerate(tags) if i != index]:
+ other = tags.index(value) + 1
+ self.message = ui.status(
+ "error",
+ f"{value} was not tagged here.",
+ [
+ f"{html.escape(value)} is already channel {other} of "
+ f"{html.escape(unit_name(layout.key))}: one label cannot stand at two "
+ "positions of a unit. Change that one first."
+ ],
+ raw=True,
+ )
+ self.refresh()
+ return
+ tags[index] = value
+ if self.store.update(synced(with_tags(answers, layout, tags), self.layouts)):
+ self.message = ""
+ else:
+ self.message = ui.status("error", "Not saved.", [self.store.refusal()])
+ self.refresh()
+ self._on_change()
+
+ def _confirm_layout(self, units: Sequence[UnitLayout]) -> None:
+ """Save the header's proposal (tags and colour pre-tick) of these units as the answer."""
+ answers = self.store.answers()
+ rgb = [u.key for u in units if self.types.proposed(u)]
+ for layout in units:
+ proposal = self._proposal(layout)
+ if proposal is None:
+ continue
+ tags = tags_of(layout, answers.units.get(layout.key))
+ filled = tuple(t if t is not None else proposal[i] for i, t in enumerate(tags))
+ answers = with_tags(answers, layout, filled)
+ ok = self.store.update(synced(answers, self.layouts))
+ self.message = "" if ok else ui.status("error", "Not saved.", [self.store.refusal()])
+ if ok and rgb:
+ self.types.confirm_header(rgb)
+ self.refresh()
+ self._on_change()
+
+ # Building ----------------------------------------------------------------------------
+
+ def refresh(self) -> None:
+ """Rebuild the rows from the saved answers (pictures are kept, not read again)."""
+ self._building = True
+ try:
+ self.cells, self.confirm_buttons = {}, []
+ self.types.open(self.layouts)
+ width = max((u.channels for u in self.layouts if not u.rgb), default=3)
+ width = max(width, 3 if any(u.rgb for u in self.layouts) else 1)
+ groups: dict[tuple, list[UnitLayout]] = {}
+ for layout in sorted(self.layouts, key=lambda u: u.key):
+ groups.setdefault(layout.signature if layout.usable else (0,), []).append(layout)
+ blocks: list[widgets.Widget] = []
+ header = widgets.GridBox(
+ [widgets.HTML("unit")]
+ + [widgets.HTML(f"channel {i + 1} (C{i})") for i in range(width)]
+ + [widgets.HTML("image type")],
+ layout=columns(width),
+ )
+ blocks.append(header)
+ for units in groups.values():
+ blocks.append(self._group(units, width))
+ self.blocks.children = blocks
+ finally:
+ self._building = False
+ self._summary()
+
+ def _group(self, units: Sequence[UnitLayout], width: int) -> widgets.Widget:
+ """Build the rows of units with one layout, and their one confirm button."""
+ first = units[0]
+ out: list[widgets.Widget] = []
+ if first.usable:
+ names = " · ".join(quoted(first.name(i)) for i in range(first.channels))
+ what = f"Header names: {names}" if not first.rgb else "Header says: colour (RGB)"
+ text = f"{what} · {len(units)} unit(s) with this layout."
+ proposal = self._proposal(first)
+ pending = [u for u in units if self._pending(u)]
+ if proposal is not None:
+ text += (
+ " They equal your labels exactly: pre-tagged below, not confirmed yet."
+ if pending
+ else " They equal your labels exactly."
+ )
+ out.append(widgets.HTML(f"{text}
"))
+ if pending:
+ button = ui.button(
+ f" Yes, these are right for these {len(pending)} unit(s)",
+ style="success",
+ icon="check",
+ tooltip="Save the pre-ticked answers of these units",
+ )
+ button.on_click(lambda _, p=pending: self._confirm_layout(p))
+ self.confirm_buttons.append(button)
+ out.append(button)
+ out += [self._row(u, width) for u in units]
+ return widgets.VBox(out)
+
+ def _pending(self, layout: UnitLayout) -> bool:
+ """Return True when the unit shows a proposal (tags or colour) not saved yet."""
+ _, proposed = self.effective(layout)
+ return bool(proposed) or self.types.proposed(layout)
+
+ def _row(self, layout: UnitLayout, width: int) -> widgets.Widget:
+ """Build the row of one unit: its name, a cell per C index and its image type."""
+ info = f"{html.escape(unit_name(layout.key))}"
+ if layout.usable and self.thumbs is not None:
+ info += (
+ f"first image by path: {html.escape(layout.image or '')}"
+ f"
1 of {layout.images} image(s); {html.escape(self.thumbs.plane(layout))}"
+ "
"
+ )
+ if not layout.usable:
+ info += ui.status("warn", "Cannot be tagged.", [html.escape(layout.problem)], raw=True)
+ shown = self.types.shown(layout)
+ tags, proposed = self.effective(layout)
+ cells: list[widgets.Widget] = [widgets.HTML(info)]
+ for index in range(width):
+ if not layout.usable or index >= layout.channels:
+ text = "" if not layout.usable else f"no channel {index + 1} in this image"
+ cells.append(widgets.HTML(f"{text}"))
+ else:
+ cells.append(self._cell(layout, index, tags[index], index in proposed, shown))
+ cells.append(self.types.cell(layout) if layout.usable else widgets.HTML(""))
+ return widgets.GridBox(cells, layout=columns(width))
+
+ def _picture(self, layout: UnitLayout, index: int) -> widgets.Widget:
+ """Build the thumbnail of one channel, or the reason there is none."""
+ thumb = self.thumbs.get(layout, index) if self.thumbs is not None else None
+ if thumb is None or thumb.png is None:
+ note = "" if thumb is None else html.escape(thumb.note)
+ return widgets.HTML(
+ f""
+ f"no picture: {note}
"
+ )
+ box = widgets.Image(
+ value=thumb.png,
+ format="png",
+ layout=widgets.Layout(width="99%", max_width=THUMB_PX, border="1px solid #CBD5E1"),
+ )
+ return box
+
+ def _cell(
+ self, layout: UnitLayout, index: int, tag: str | None, proposed: bool, shown: str | None
+ ) -> widgets.Widget:
+ """Build one cell: picture, C index and header name, the tag and a zoom button."""
+ header = f"C{index} {quoted(layout.name(index))}"
+ children: list[widgets.Widget] = [self._picture(layout, index)]
+ thumb = self.thumbs.get(layout, index) if self.thumbs is not None else None
+ if thumb is not None and thumb.note:
+ header += f"{html.escape(thumb.note)}
"
+ children.append(widgets.HTML(header))
+ if shown == "rgb":
+ children.append(widgets.HTML(chip("RGB"[index], list("RGB"))))
+ else:
+ children.append(self._dropdown(layout, index, tag, proposed))
+ zoom = ui.button(" Enlarge", icon="search-plus", tooltip="Show the 1024 px detail view")
+ zoom.layout = widgets.Layout(width="99%")
+ zoom.on_click(lambda _, u=layout, i=index: self._enlarge(u, i))
+ children.append(zoom)
+ return widgets.VBox(children)
+
+ def _dropdown(
+ self, layout: UnitLayout, index: int, tag: str | None, proposed: bool
+ ) -> widgets.Widget:
+ """Build the dropdown of one cell: the labels and what is shown now."""
+ options = [(CHOOSE, None), *((label, label) for label in self.labels)]
+ value = tag if tag in {o[1] for o in options} else None
+ drop = widgets.Dropdown(options=options, value=value, layout=widgets.Layout(width="99%"))
+ drop.observe(lambda change, u=layout, i=index: self._tagged(u, i, change["new"]), "value")
+ self.cells[(layout.key, index)] = drop
+ if tag is None:
+ mark = blank_chip("not tagged")
+ else:
+ mark = chip(tag, self.labels, self._shown_as(tag))
+ if proposed:
+ mark += " proposed, not confirmed"
+ return widgets.VBox([drop, widgets.HTML(mark)])
+
+ def _enlarge(self, layout: UnitLayout, index: int) -> None:
+ """Show the 1024 px picture of one channel under the grid."""
+ thumb = self.thumbs.get(layout, index, detail=True)
+ close = ui.button(" Close", icon="times")
+ close.on_click(lambda _: setattr(self.detail, "children", []))
+ title = widgets.HTML(
+ f"{html.escape(unit_name(layout.key))} · C{index} "
+ f"{quoted(layout.name(index))} · {html.escape(layout.image or '')}"
+ )
+ picture: widgets.Widget = (
+ widgets.Image(value=thumb.png, format="png", layout=widgets.Layout(max_width="1024px"))
+ if thumb.png
+ else widgets.HTML(ui.status("warn", "No picture.", [html.escape(thumb.note)], raw=True))
+ )
+ self.detail.children = [title, picture, close]
+
+ # Summary -----------------------------------------------------------------------------
+
+ def _summary(self) -> None:
+ """Say where each label is used and whether tagging is finished."""
+ answers = self.store.answers()
+ total = len(self.layouts)
+ found = usage(self.layouts, answers.units, self.labels)
+ lines = [self._usage_line(label, len(found[label]), total) for label in self.labels]
+ for label, places in positions(self.layouts, answers.units, self.labels).items():
+ if len(places) > 1:
+ where = ", ".join(
+ f"channel {i + 1} in {n} unit(s)" for i, n in sorted(places.items())
+ )
+ lines.append(
+ f"{html.escape(label)} is not always in the same "
+ f"column: {where}.
"
+ )
+ self.usage_html.value = (
+ "Where each label is used
" + "".join(lines)
+ )
+ self.problem.value = self.message
+ self.status.value = self._progress(total)
+
+ def _shown_as(self, label: str) -> str | None:
+ """Return the shown-as name of a saved label (the chip's tooltip), if it has one."""
+ saved = self.store.answers().channels.get(label)
+ return None if saved is None else saved.display_name
+
+ def _usage_line(self, label: str, used: int, total: int) -> str:
+ """Return the usage line of a label; one used in only some units is orange."""
+ mark = chip(label, self.labels, self._shown_as(label))
+ if used == total:
+ return f"{mark} used in all {total} units
"
+ if used == 0:
+ return f"{mark} not used in any unit yet
"
+ return (
+ "{mark} "
+ f"{partial_text(label, used, total)} (A_SCOPE_INCOMPLETE)
"
+ )
+
+ def _progress(self, total: int) -> str:
+ """Return the status: what is still to answer, or that every channel is tagged."""
+ answers = self.store.answers()
+ left = remaining(self.layouts, answers.units, self.labels)
+ if left.done:
+ return ui.status("ok", f"Every channel of every one of the {total} units is tagged.")
+ parts = []
+ if left.cells:
+ parts.append(f"{left.cells} channel(s) still to tag")
+ if left.types:
+ parts.append(f"{left.types} image type(s) still to choose")
+ if left.blocked:
+ parts.append(f"{left.blocked} unit(s) cannot be tagged")
+ return ui.status("warn", "Not finished: " + ", ".join(parts) + ".")
+
+
+__all__ = ["TagGrid", "partial_text"]
diff --git a/src/morphobricks_preprocessing/notebook_channels_labels.py b/src/morphobricks_preprocessing/notebook_channels_labels.py
new file mode 100644
index 00000000..9e45d095
--- /dev/null
+++ b/src/morphobricks_preprocessing/notebook_channels_labels.py
@@ -0,0 +1,358 @@
+"""Step A of *Your channels*: "Which labels did you image?" (a self-contained card).
+
+One list of channel IDs naming **what was stained** (the target, not the dye or the microscope's
+slot), each with an optional shown-as name. The list starts with as many slots as the largest
+channel count found in any image and is prefilled only from header names that are valid IDs as
+they stand and not generic (:func:`.channel_labels.propose_labels`). Near-identical header
+spellings are shown as a group for the user to decide, never merged; a typed ID that differs from
+another only by capital letters is refused at once; **Add label** adds a slot. The labels are saved
+(to ``channels`` of the draft) only when the user confirms the list.
+
+The card reads and saves through an :class:`.notebook_channels_view.AnswerStore` and tells its
+owner when the confirmed list changed (``on_change``); it knows nothing of the tag grid.
+"""
+
+from __future__ import annotations
+
+import html
+from collections.abc import Callable, Sequence
+from dataclasses import dataclass
+
+import ipywidgets as widgets
+
+from . import notebook_style as ui
+from .channel_answers import label_ids, synced, untagged_by, with_labels
+from .channel_labels import (
+ GENERIC,
+ NOT_VALID,
+ LabelProposal,
+ UnitLayout,
+ case_only,
+ generic_reason,
+ label_problems,
+ propose_labels,
+ usable_as_label,
+)
+from .notebook_channels_view import AnswerStore, chip
+from .notebook_description_view import plain
+
+ROW_COLUMNS = "minmax(150px, 1fr) minmax(150px, 1fr) minmax(200px, 2fr)"
+QUESTION = "Are these the labels you imaged?"
+EXPLANATION = (
+ "A label is the short name of what was stained, for example DNA or "
+ "ACTIN, not the dye (AF488) or the microscope's slot (Ch1). The same "
+ "label must be used for the same stain in every unit, and different stains must never share "
+ "a label. Letters and digits, no space at the start or end; capital letters count as the "
+ "same label."
+)
+
+
+def problem_text(codes: Sequence[str]) -> str:
+ """Return the plain words for the codes that refuse a label (the code is always named)."""
+ out = []
+ for code in codes:
+ if code == "E_DUPLICATE_VALUE":
+ reason = "is listed twice"
+ elif code == "E_RGB_TOKEN_MISUSE":
+ reason = "R, G and B are the fixed names of colour images: use another label"
+ elif code == "E_ID_CASE_COLLISION":
+ reason = "differs from another label only by capital letters: refused"
+ else:
+ reason = plain(code)
+ out.append(f"{reason} ({code})")
+ return "; ".join(out)
+
+
+@dataclass
+class _Row:
+ """One label slot: its ID box, shown-as box and what is said beside it."""
+
+ id_box: widgets.Text
+ name_box: widgets.Text
+ state: widgets.HTML
+ widget: widgets.GridBox
+
+
+class LabelsCard:
+ """The label list: slots, proposals, look-alike groups, **Add label** and the confirmation."""
+
+ def __init__(self, store: AnswerStore, on_change: Callable[[], None]) -> None:
+ self.store, self._on_change = store, on_change
+ self.slots = 1
+ self.header_names: list[str] = []
+ self.rows: list[_Row] = []
+ self.groups: list[tuple[str, ...]] = []
+ self.proposal = LabelProposal()
+ self.layouts: Sequence[UnitLayout] = ()
+ self._building = False
+ self._ok = False
+ self.note = widgets.HTML()
+ self.rows_box = widgets.VBox()
+ self.groups_box = widgets.VBox()
+ self.skipped = widgets.HTML()
+ self.add_button = ui.button(
+ " Add label", style="info", icon="plus", tooltip="Add one more label slot"
+ )
+ self.add_button.on_click(lambda _: self._add_row())
+ self.confirm = ui.ConfirmToggle(
+ QUESTION,
+ EXPLANATION,
+ "Yes, these are the labels",
+ "Confirmed: these are the labels of the experiment.",
+ "Change the labels",
+ )
+ self.confirm.observe(self._toggled)
+ self.status = widgets.HTML()
+ self.problem = widgets.HTML()
+ self.widget = ui.card(
+ "A. Which labels did you image?",
+ [
+ ui.help_text(
+ "Name what was stained, once per stain. The number of slots is the largest "
+ "number of channels found in any image; Add label adds more. Header "
+ "names are offered only when they are valid labels as they are and not "
+ "generic. "
+ "Every channel position of every unit needs a label, so an unused slot "
+ "needs one too: declare it here (for example unused)."
+ ),
+ self.note,
+ self.groups_box,
+ self.rows_box,
+ self.add_button,
+ self.skipped,
+ self.problem,
+ self.status,
+ self.confirm.widget,
+ ],
+ visible=False,
+ )
+
+ @property
+ def labels(self) -> list[str]:
+ """Return the saved labels, sorted (what the grid's dropdowns offer)."""
+ return label_ids(self.store.answers())
+
+ @property
+ def confirmed(self) -> bool:
+ """True while the list is confirmed (saved and not being changed)."""
+ return self.confirm.value
+
+ def display_names(self) -> dict[str, str | None]:
+ """Return the shown-as name of each saved label."""
+ return {k: v.display_name for k, v in self.store.answers().channels.items()}
+
+ # Opening -----------------------------------------------------------------------------
+
+ def open(self, slots: int, header_names: list[str], layouts: Sequence[UnitLayout]) -> None:
+ """Show the card for an experiment: its slot count, header names and units; resume."""
+ self.slots, self.header_names, self.layouts = max(slots, 1), header_names, layouts
+ saved = self.labels
+ self.proposal = (
+ LabelProposal() if saved else propose_labels(header_names, self.store.used_channels())
+ )
+ self.groups = list(self.proposal.groups)
+ names = self.display_names()
+ self._building = True
+ try:
+ texts = saved or list(self.proposal.single)
+ self.rows = []
+ for text in texts:
+ self.rows.append(self._row(text, names.get(text) or ""))
+ while len(self.rows) < self.slots:
+ self.rows.append(self._row("", ""))
+ self.confirm.set(bool(saved), notify=False)
+ finally:
+ self._building = False
+ self._render()
+ ui.set_visible(self.widget, True)
+
+ def close(self) -> None:
+ """Hide the card."""
+ ui.set_visible(self.widget, False)
+
+ # Rows --------------------------------------------------------------------------------
+
+ def _row(self, text: str, name: str) -> _Row:
+ """Build one label slot with its ID box, shown-as box and state."""
+ id_box = widgets.Text(
+ value=text,
+ placeholder="what you stained, not the dye",
+ layout=widgets.Layout(width="99%"),
+ )
+ name_box = widgets.Text(
+ value=name, placeholder="shown as (optional)", layout=widgets.Layout(width="99%")
+ )
+ state = widgets.HTML()
+ row = _Row(
+ id_box,
+ name_box,
+ state,
+ widgets.GridBox(
+ [id_box, name_box, state],
+ layout=widgets.Layout(
+ grid_template_columns=ROW_COLUMNS,
+ grid_gap="6px 12px",
+ width="99%",
+ align_items="center",
+ ),
+ ),
+ )
+ id_box.observe(lambda _: self._typed(), names="value")
+ name_box.observe(lambda _: self._typed(), names="value")
+ return row
+
+ def _add_row(self) -> None:
+ if self.confirm.value:
+ return
+ self.rows.append(self._row("", ""))
+ self._render()
+
+ def _typed(self) -> None:
+ if not self._building:
+ self._validate()
+
+ def _render(self) -> None:
+ self.rows_box.children = [r.widget for r in self.rows]
+ self._render_groups()
+ self._validate()
+ locked = self.confirm.value
+ for row in self.rows:
+ row.id_box.disabled = row.name_box.disabled = locked
+ self.add_button.disabled = locked
+ self._render_skipped()
+
+ def _validate(self) -> None:
+ """Say, per slot, whether its label is allowed; refuse a case collision at once."""
+ texts = [r.id_box.value for r in self.rows]
+ filled = [(i, t) for i, t in enumerate(texts) if t != ""]
+ problems = label_problems([t for _, t in filled], self.store.used_channels())
+ by_slot = {filled[i][0]: codes for i, codes in problems.items()}
+ everything = [t for _, t in filled]
+ for index, row in enumerate(self.rows):
+ text = texts[index]
+ if index in by_slot:
+ row.state.value = ui.status("error", f"“{text}” {problem_text(by_slot[index])}")
+ elif text:
+ row.state.value = chip(text, everything, row.name_box.value or None)
+ else:
+ row.state.value = "empty slot"
+ ok = not by_slot and bool(filled)
+ self._ok = ok
+ self.confirm.disabled = not ok and not self.confirm.value
+ self.problem.value = (
+ ui.status("error", "These labels cannot be used yet.", ["Fix the red entries."])
+ if by_slot
+ else ""
+ )
+
+ # Look-alike groups -------------------------------------------------------------------
+
+ def _render_groups(self) -> None:
+ boxes: list[widgets.Widget] = []
+ for group in self.groups:
+ boxes.append(self._group_box(group))
+ self.groups_box.children = boxes
+
+ def _group_box(self, group: tuple[str, ...]) -> widgets.Widget:
+ """Build the question about one group of look-alike header names."""
+ found = ", ".join(f"‘{n}’" for n in group)
+ usable = [n for n in group if usable_as_label(n, self.store.used_channels())]
+ clash = any(case_only(a, b) for a in usable for b in usable)
+ pick = widgets.Dropdown(
+ options=usable, layout=widgets.Layout(width="220px"), disabled=not usable
+ )
+ same = ui.button(" Same label", style="success", icon="link", tooltip="One label")
+ same.disabled = not usable or self.confirm.value
+ same.on_click(lambda _: self._decide(group, [pick.value]))
+ different = ui.button(" Different labels", style="", icon="unlink")
+ different.disabled = len(usable) < 2 or clash or self.confirm.value
+ different.on_click(lambda _: self._decide(group, usable))
+ dismiss = ui.button(" Decide later", style="", icon="times")
+ dismiss.on_click(lambda _: self._decide(group, []))
+ why = ""
+ if clash:
+ why = (
+ "Some of these differ only by capital letters: they cannot be different labels "
+ "(E_ID_CASE_COLLISION). "
+ )
+ if not usable:
+ why += "None of them can be a label as it is written: type the label yourself. "
+ text = widgets.HTML(
+ ui.status(
+ "warn",
+ f"These header names look alike: {found}. Same label?",
+ [
+ why + "Nothing is merged for you: choose the spelling to use if they are "
+ "one label, or say they are different."
+ ],
+ raw=True,
+ )
+ )
+ return widgets.VBox([text, widgets.HBox([pick, same, different, dismiss])])
+
+ def _decide(self, group: tuple[str, ...], chosen: list[str]) -> None:
+ """Put the chosen spellings into the first empty slots; the group is then settled."""
+ self.groups = [g for g in self.groups if g != group]
+ for text in chosen:
+ free = next((r for r in self.rows if not r.id_box.value), None)
+ if free is None:
+ free = self._row("", "")
+ self.rows.append(free)
+ self._building = True
+ free.id_box.value = text
+ self._building = False
+ self._render()
+
+ def _render_skipped(self) -> None:
+ """Say which header names were not offered as labels, and why."""
+ said = []
+ for name, why in self.proposal.skipped:
+ if why == GENERIC:
+ said.append(f"{html.escape(name)} ({generic_reason(name)})")
+ elif why == NOT_VALID:
+ said.append(f"{html.escape(name)} (not valid as a label as written)")
+ self.skipped.value = (
+ "Not offered as labels: " + "; ".join(said) + ". Type a label "
+ "if you stained something these names stand for.
"
+ if said
+ else ""
+ )
+
+ # Confirming --------------------------------------------------------------------------
+
+ def _toggled(self, confirmed: bool) -> None:
+ if self._building:
+ return
+ if confirmed:
+ if not self._ok or not self._save():
+ self.confirm.set(False, notify=False)
+ self._render()
+ self._on_change()
+
+ def _save(self) -> bool:
+ """Save the label list; False when it is refused. Tags of a gone label are taken back."""
+ labels: dict[str, str | None] = {}
+ for row in self.rows:
+ if row.id_box.value:
+ labels[row.id_box.value] = row.name_box.value or None
+ answers = self.store.answers()
+ gone = untagged_by(answers, labels)
+ new = synced(with_labels(answers, labels), self.layouts)
+ if not self.store.update(new):
+ self.status.value = ui.status(
+ "error", "Not saved: the labels were refused.", [self.store.refusal()]
+ )
+ return False
+ self.status.value = (
+ ui.status(
+ "warn",
+ f"{gone} tag(s) of a label that is no longer in the list were taken back.",
+ ["Tag those channels again."],
+ )
+ if gone
+ else ""
+ )
+ return True
+
+
+__all__ = ["LabelsCard", "problem_text"]
diff --git a/src/morphobricks_preprocessing/notebook_channels_type.py b/src/morphobricks_preprocessing/notebook_channels_type.py
new file mode 100644
index 00000000..af4bd11f
--- /dev/null
+++ b/src/morphobricks_preprocessing/notebook_channels_type.py
@@ -0,0 +1,130 @@
+"""The image-type column of *Your channels* (a self-contained piece).
+
+One answer per unit, **nothing preselected**: *separate channels* (an ordinary channel image) or
+*colour (RGB)*. "Colour (RGB)" is pre-ticked only for a unit whose image header says it is a
+colour image (``rgb_samples``); that pre-tick is a proposal, saved only when the user confirms the
+layout (:meth:`ImageTypeColumn.confirm_header`) or ticks the box. The answer is declared by the
+user and never inferred from pixels or channel counts (experiment-v1 section 4.4b); a choice that
+contradicts the header is allowed here and said in words, since the run refuses it
+(``R_IMAGE_TYPE_CONTRADICTS_HEADER``). Saved to ``units.*.image_type`` of the draft.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable, Sequence
+
+import ipywidgets as widgets
+
+from . import notebook_style as ui
+from .channel_answers import synced, with_type
+from .channel_labels import UnitLayout
+from .draft_answers import UnitKey
+from .notebook_channels_view import AnswerStore, blank_chip
+
+OPTIONS = [("separate channels", "channels"), ("colour (RGB)", "rgb")]
+
+
+class ImageTypeColumn:
+ """The per-unit image-type choice; the owner places the cells where it likes."""
+
+ def __init__(self, store: AnswerStore, on_change: Callable[[], None]) -> None:
+ self.store, self._on_change = store, on_change
+ self.layouts: Sequence[UnitLayout] = ()
+ self.radios: dict[UnitKey, widgets.RadioButtons] = {}
+ self.notes: dict[UnitKey, widgets.HTML] = {}
+ self.message = ""
+ self._building = False
+
+ def open(self, layouts: Sequence[UnitLayout]) -> None:
+ """Take the units; cells are built by :meth:`cell`."""
+ self.layouts = layouts
+ self.radios, self.notes = {}, {}
+
+ def saved(self, key: UnitKey) -> str | None:
+ """Return the type the user answered for a unit (``channels``, ``rgb``), else None."""
+ unit = self.store.answers().units.get(key)
+ return None if unit is None else unit.image_type
+
+ def shown(self, layout: UnitLayout) -> str | None:
+ """Return what the cell shows: the answer, else ``rgb`` when the header says colour."""
+ return self.saved(layout.key) or ("rgb" if layout.usable and layout.rgb else None)
+
+ def proposed(self, layout: UnitLayout) -> bool:
+ """Return True while the cell shows the header's colour pre-tick, not saved yet."""
+ return self.saved(layout.key) is None and self.shown(layout) == "rgb"
+
+ def cell(self, layout: UnitLayout) -> widgets.Widget:
+ """Build the radio buttons and the note of one unit."""
+ radio = widgets.RadioButtons(
+ options=OPTIONS, value=self.shown(layout), layout=widgets.Layout(width="99%")
+ )
+ radio.disabled = not layout.usable
+ note = widgets.HTML(self._note(layout))
+ radio.observe(lambda change, u=layout: self._chosen(u, change["new"]), names="value")
+ self.radios[layout.key], self.notes[layout.key] = radio, note
+ return widgets.VBox([radio, note])
+
+ def _note(self, layout: UnitLayout) -> str:
+ """Return what is said under a unit's radio buttons."""
+ saved, shown = self.saved(layout.key), self.shown(layout)
+ if not layout.usable:
+ return ""
+ if saved is None and shown == "rgb":
+ return (
+ "Pre-ticked: the image header says colour. Not saved "
+ "until you confirm it."
+ )
+ if saved is None:
+ return blank_chip("choose one")
+ if saved == "channels" and layout.rgb:
+ return (
+ "The header says colour (RGB): the "
+ "run refuses a unit declared otherwise."
+ )
+ if saved == "rgb" and not layout.rgb:
+ return (
+ "The header does not say colour: the "
+ "run refuses this unit unless its header does."
+ )
+ return ""
+
+ def _chosen(self, layout: UnitLayout, value: str | None) -> None:
+ """Save the type the user ticked; a refusal is said and the tick is taken back."""
+ if self._building or value is None or value == self.saved(layout.key):
+ return
+ try:
+ new = with_type(self.store.answers(), layout, value)
+ except ValueError as exc:
+ self.message = ui.status("error", "Not taken.", [str(exc)])
+ self._revert(layout)
+ else:
+ if self.store.update(synced(new, self.layouts)):
+ self.message = ""
+ else:
+ self.message = ui.status("error", "Not saved.", [self.store.refusal()])
+ self._revert(layout)
+ self._on_change()
+
+ def _revert(self, layout: UnitLayout) -> None:
+ radio = self.radios.get(layout.key)
+ if radio is not None:
+ self._building = True
+ radio.value = self.shown(layout)
+ self._building = False
+
+ def confirm_header(self, keys: Sequence[UnitKey]) -> bool:
+ """Save the header's colour pre-tick of these units as the answer; False if refused."""
+ answers = self.store.answers()
+ by_key = {u.key: u for u in self.layouts}
+ for key in keys:
+ layout = by_key[key]
+ if self.proposed(layout):
+ answers = with_type(answers, layout, "rgb")
+ ok = self.store.update(synced(answers, self.layouts))
+ if not ok:
+ self.message = ui.status("error", "Not saved.", [self.store.refusal()])
+ self._on_change()
+ return ok
+
+
+__all__ = ["OPTIONS", "ImageTypeColumn"]
diff --git a/src/morphobricks_preprocessing/notebook_channels_view.py b/src/morphobricks_preprocessing/notebook_channels_view.py
new file mode 100644
index 00000000..cb2bda85
--- /dev/null
+++ b/src/morphobricks_preprocessing/notebook_channels_view.py
@@ -0,0 +1,120 @@
+"""Colours and small HTML pieces shared by the *Your channels* cards.
+
+Each label has one colour chip, taken from its place in the sorted label list, used by the label
+list, the column of the grid and every cell. Orange is kept for what needs attention (an untagged
+cell, a label used in only some units) and is not in the palette.
+"""
+
+from __future__ import annotations
+
+import html
+from collections.abc import Sequence
+from typing import Protocol
+
+import ipywidgets as widgets
+
+from .draft_answers import DraftAnswers
+
+PALETTE = (
+ "#1D4ED8",
+ "#0F766E",
+ "#7E22CE",
+ "#BE185D",
+ "#4D7C0F",
+ "#0E7490",
+ "#4338CA",
+ "#475569",
+ "#A21CAF",
+ "#166534",
+)
+ORANGE = "#D97706"
+ORANGE_FILL = "#FFFBEB"
+CHOOSE = "— tag this channel —"
+COLUMNS_PREFIX = "minmax(150px, 190px)"
+
+
+class AnswerStore(Protocol):
+ """Where the cards read and save the draft answers (the notebook gives one).
+
+ Each card reads its own slice from :meth:`answers` and saves a changed copy with
+ :meth:`update`; they know nothing of each other, so a later layout can put them on separate
+ steps.
+ """
+
+ def answers(self) -> DraftAnswers:
+ """Return the current answers."""
+ ...
+
+ def update(self, new: DraftAnswers) -> bool:
+ """Save the new answers; False when they are refused (see :meth:`refusal`)."""
+ ...
+
+ def refusal(self) -> str:
+ """Return the reason the last update was refused, in words."""
+ ...
+
+ def used_channels(self) -> tuple[str, ...]:
+ """Return the channel IDs the experiment's lineage retired (never reusable)."""
+ ...
+
+
+def colour(label: str, labels: Sequence[str]) -> str:
+ """Return the chip colour of a label: its place in the sorted list, cycling the palette."""
+ order = sorted(labels)
+ return PALETTE[(order.index(label) if label in order else 0) % len(PALETTE)]
+
+
+def chip(label: str, labels: Sequence[str], display: str | None = None) -> str:
+ """Return the colour chip of a label as HTML: always the ID, the shown-as name as tooltip."""
+ tip = f" title='{html.escape(display, quote=True)}'" if display else ""
+ return (
+ f"{html.escape(label)}"
+ )
+
+
+def blank_chip(text: str) -> str:
+ """Return an orange outlined chip for something not answered."""
+ return (
+ f""
+ f"{html.escape(text)}"
+ )
+
+
+def quoted(name: str | None) -> str:
+ """Return a raw header name quoted so outer spaces show, or a note that there is none."""
+ if not name:
+ return "no header name"
+ return f"‘{html.escape(name).replace(' ', ' ')}’"
+
+
+def columns(count: int) -> widgets.Layout:
+ """Return the grid layout of the tag grid: unit, one column per C index, image type."""
+ return widgets.Layout(
+ grid_template_columns=f"{COLUMNS_PREFIX} repeat({max(count, 1)}, minmax(120px, 1fr)) "
+ "minmax(150px, 190px)",
+ grid_gap="8px 10px",
+ width="99%",
+ align_items="flex-start",
+ )
+
+
+def unit_name(key: tuple[str, str, str]) -> str:
+ """Return ``condition / repeat / technical repeat`` of a unit; ``single`` is left out."""
+ return " / ".join(part for part in key if part != "single") or "single"
+
+
+__all__ = [
+ "CHOOSE",
+ "AnswerStore",
+ "ORANGE",
+ "ORANGE_FILL",
+ "PALETTE",
+ "blank_chip",
+ "chip",
+ "colour",
+ "columns",
+ "quoted",
+ "unit_name",
+]
diff --git a/src/morphobricks_preprocessing/notebook_description.py b/src/morphobricks_preprocessing/notebook_description.py
index 01821451..7843675d 100644
--- a/src/morphobricks_preprocessing/notebook_description.py
+++ b/src/morphobricks_preprocessing/notebook_description.py
@@ -135,6 +135,7 @@ def _hide_description(self) -> None:
self.draft = self.draft_answers = self._draft_base = None
ui.set_visible(self.names_card, False)
ui.set_visible(self.references_card, False)
+ self._hide_channels()
def _design_pairing(self) -> dict[tuple[str, str], str]:
"""Return the folders the design pairs under another name than their own."""
@@ -291,6 +292,7 @@ def _refresh_description(self) -> None:
self._refreshing = False
ui.set_visible(self.names_card, True)
ui.set_visible(self.references_card, len(self.draft.conditions) > 1)
+ self._refresh_channels()
def _build_names(self) -> None:
"""Rebuild the ID and shown-as boxes, keeping what was typed and not yet confirmed."""
diff --git a/src/morphobricks_preprocessing/notebook_ui.py b/src/morphobricks_preprocessing/notebook_ui.py
index fe81906a..a9a03cc7 100644
--- a/src/morphobricks_preprocessing/notebook_ui.py
+++ b/src/morphobricks_preprocessing/notebook_ui.py
@@ -52,6 +52,7 @@
from .notebook_axis_preview import _AxisPreview
from .notebook_axis_step import _AxisStep
from .notebook_calibration import _CalibrationCard
+from .notebook_channels import _ChannelsStep
from .notebook_convert import _ConvertCard
from .notebook_description import _DescriptionCard
from .notebook_design import _DesignCard
@@ -124,6 +125,7 @@ class PreprocessingNotebook(
_MasksCard,
_DesignCard,
_DescriptionCard,
+ _ChannelsStep,
_CalibrationCard,
_OutputCards,
_AxisStep,
@@ -188,6 +190,7 @@ def __init__(self, experiment_folder: str = "", output_folder: str = ""): # noq
self._build_output_cards()
self._build_calibration_card()
self._build_masks_cards()
+ self._build_channels_step()
self._build_folder_step(str(experiment_folder), str(output_folder))
self._build_axis_step()
self._build_convert_card()
diff --git a/tests/channel_images.py b/tests/channel_images.py
new file mode 100644
index 00000000..6ebb662e
--- /dev/null
+++ b/tests/channel_images.py
@@ -0,0 +1,82 @@
+"""Small experiment folders with named channels and pictures that have visible structure.
+
+``blobs`` draws nucleus-like discs, ``filaments`` a tangle of line segments and ``cells`` a
+smooth cell-shaped haze, so a thumbnail shows something recognisable (screenshots, tests), never
+blank or noise-only arrays.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import numpy as np
+import tifffile
+
+SIZE = 64
+
+
+def blobs(seed: int, size: int = SIZE) -> np.ndarray:
+ """Return nucleus-like bright discs of varied size on a dim background (uint16)."""
+ rng = np.random.default_rng(seed)
+ yy, xx = np.mgrid[:size, :size]
+ img = rng.normal(300, 20, (size, size))
+ for _ in range(7):
+ cy, cx, r = rng.uniform(8, size - 8), rng.uniform(8, size - 8), rng.uniform(4, 8)
+ img += 3000 * np.exp(-((((yy - cy) ** 2 + (xx - cx) ** 2) / (2 * (r / 1.6) ** 2)) ** 1.5))
+ return np.clip(img, 0, 65535).astype(np.uint16)
+
+
+def filaments(seed: int, size: int = SIZE) -> np.ndarray:
+ """Return a tangle of bright line segments (uint16)."""
+ rng = np.random.default_rng(seed)
+ img = rng.normal(250, 15, (size, size))
+ for _ in range(14):
+ y0, x0 = rng.uniform(0, size, 2)
+ angle, length = rng.uniform(0, np.pi), rng.uniform(18, 40)
+ for t in np.linspace(0, length, int(length * 3)):
+ y, x = int(y0 + t * np.sin(angle)), int(x0 + t * np.cos(angle))
+ if 0 <= y < size and 0 <= x < size:
+ img[max(y - 1, 0) : y + 1, max(x - 1, 0) : x + 1] += 1800
+ return np.clip(img, 0, 65535).astype(np.uint16)
+
+
+def cells(seed: int, size: int = SIZE) -> np.ndarray:
+ """Return a smooth cell-shaped haze with a few brighter regions (uint16)."""
+ rng = np.random.default_rng(seed)
+ yy, xx = np.mgrid[:size, :size]
+ img = rng.normal(400, 20, (size, size))
+ for _ in range(4):
+ cy, cx = rng.uniform(12, size - 12, 2)
+ r = rng.uniform(10, 16)
+ img += 1500 * np.exp(-(((yy - cy) ** 2 + (xx - cx) ** 2) / (2 * r**2)))
+ return np.clip(img, 0, 65535).astype(np.uint16)
+
+
+KINDS = {"blobs": blobs, "filaments": filaments, "cells": cells}
+
+
+def write_channels(
+ path: str | Path,
+ names: list[str | None] | None,
+ kinds: list[str],
+ seed: int = 1,
+ size: int = SIZE,
+):
+ """Write a CYX OME-TIFF whose channel ``i`` shows ``kinds[i]`` and has header ``names[i]``."""
+ path = Path(path)
+ path.parent.mkdir(parents=True, exist_ok=True)
+ data = np.stack([KINDS[k](seed + i, size) for i, k in enumerate(kinds)])
+ metadata: dict = {"axes": "CYX"}
+ if names is not None:
+ metadata["Channel"] = {"Name": ["" if n is None else n for n in names]}
+ tifffile.imwrite(path, data, ome=True, photometric="minisblack", metadata=metadata)
+ return path
+
+
+def write_rgb(path: str | Path, seed: int = 1):
+ """Write a colour (RGB) TIFF: nuclei in blue, filaments in green, haze in red."""
+ path = Path(path)
+ path.parent.mkdir(parents=True, exist_ok=True)
+ data = np.stack([cells(seed) // 256, filaments(seed) // 256, blobs(seed) // 256], axis=-1)
+ tifffile.imwrite(path, data.astype(np.uint8), photometric="rgb")
+ return path
diff --git a/tests/draft_file_support.py b/tests/draft_file_support.py
index 8d4b0280..fcad27d3 100644
--- a/tests/draft_file_support.py
+++ b/tests/draft_file_support.py
@@ -70,15 +70,16 @@ def broken(change):
"list": (b"[]", "malformed"),
"nan": (b'{"format": NaN}', "malformed"),
"duplicate": (
- answers_to_bytes(complete()).replace(b'"version": 2,', b'"version": 2, "version": 2,'),
+ answers_to_bytes(complete()).replace(b'"version": 3,', b'"version": 3, "version": 3,'),
"malformed",
),
"wrong_format": (broken(lambda d: d.update(format="other")), "malformed"),
"version_1": (broken(lambda d: d.update(version=1)), "unknown_version"),
- "version_3": (broken(lambda d: d.update(version=3)), "unknown_version"),
- "version_text": (broken(lambda d: d.update(version="2")), "unknown_version"),
+ "version_4": (broken(lambda d: d.update(version=4)), "unknown_version"),
+ "version_2": (broken(lambda d: d.update(version=2)), "unknown_version"),
+ "version_text": (broken(lambda d: d.update(version="3")), "unknown_version"),
"version_bool": (broken(lambda d: d.update(version=True)), "unknown_version"),
- "version_float": (broken(lambda d: d.update(version=2.0)), "unknown_version"),
+ "version_float": (broken(lambda d: d.update(version=3.0)), "unknown_version"),
"version_missing": (broken(lambda d: d.pop("version")), "unknown_version"),
"extra_top": (broken(lambda d: d.update(extra=1)), "unknown_key"),
"extra_condition": (
diff --git a/tests/test_channel_labels.py b/tests/test_channel_labels.py
new file mode 100644
index 00000000..c42deb37
--- /dev/null
+++ b/tests/test_channel_labels.py
@@ -0,0 +1,286 @@
+"""Labels, look-alikes, tags and the draft file's tagging states (Stage 1, PR 15)."""
+
+from dataclasses import replace
+
+import pytest
+
+from draft_file_support import UNITS, complete
+from morphobricks_preprocessing import discover_experiment
+from morphobricks_preprocessing.channel_answers import (
+ synced,
+ with_labels,
+ with_tags,
+ with_type,
+)
+from morphobricks_preprocessing.channel_labels import (
+ UnitLayout,
+ aliases,
+ is_generic,
+ label_problems,
+ look_alike,
+ proposal_for,
+ propose_labels,
+ remaining,
+ usage,
+)
+from morphobricks_preprocessing.draft import draft_from_discovery
+from morphobricks_preprocessing.draft_answers import (
+ DraftAnswers,
+ DraftIncomplete,
+ UnitAnswer,
+ carry_over,
+ description_from_draft,
+ missing_confirmations,
+)
+from morphobricks_preprocessing.draft_format import (
+ ABSENT,
+ SPEC_COMMIT,
+ DraftFileError,
+ answers_from_bytes,
+ answers_to_bytes,
+)
+from tiff_tree import make_tree
+
+
+def layout(key=("a", "r1", "single"), names=("DAPI", "Actin"), rgb=False, n=None):
+ return UnitLayout(key, "x.tif", n or len(names), tuple(names), rgb, "", 1)
+
+
+@pytest.mark.parametrize(
+ "name",
+ [
+ "Ch1",
+ "C0",
+ "c=0",
+ "Channel 2",
+ "channel_3",
+ "AF488",
+ "Alexa Fluor 568",
+ "Cy5",
+ "DAPI",
+ "Hoechst 33342",
+ "FITC",
+ "EGFP",
+ "mCherry",
+ "488",
+ "561 nm",
+ "Track 1",
+ "Detector 2",
+ ],
+)
+def test_generic_header_names_are_recognised(name):
+ assert is_generic(name)
+
+
+@pytest.mark.parametrize("name", ["Actin", "Tubulin", "LaminB1", "PHASE", "Nucleus", "Golgi"])
+def test_stain_names_are_not_generic(name):
+ assert not is_generic(name)
+
+
+def test_proposals_leave_out_generic_and_invalid_names_and_never_repair():
+ found = propose_labels(["DAPI", "Ch1", "AF488", "Actin", " Tubulin", None, "Actin"])
+ assert found.single == ("Actin",)
+ assert dict(found.skipped) == {
+ "DAPI": "generic",
+ "Ch1": "generic",
+ "AF488": "generic",
+ " Tubulin": "not_valid",
+ }
+
+
+@pytest.mark.parametrize(
+ ("a", "b"),
+ [
+ ("DAPI", "Dapi"),
+ ("Lamin B", "LaminB"),
+ ("Lamin-B", "Lamin_B"),
+ ("Actin", "Actim"),
+ ("Actin", "Actn"),
+ ],
+)
+def test_look_alikes(a, b):
+ assert look_alike(a, b)
+
+
+@pytest.mark.parametrize(("a", "b"), [("Actin", "Tubulin"), ("A", "B"), ("Actin", "Actin")])
+def test_different_names_are_not_look_alikes(a, b):
+ assert not look_alike(a, b)
+
+
+def test_look_alikes_come_back_grouped_and_are_not_offered_one_by_one():
+ found = propose_labels(["Lamin", "LaminB", "Actin", "Actin "])
+ assert sorted(found.groups) == [("Actin", "Actin "), ("Lamin", "LaminB")]
+ assert found.single == ()
+
+
+def test_label_problems_refuse_a_case_collision_on_both_entries():
+ out = label_problems(["Actin", "ACTIN", "DNA", "r"])
+ assert out[0] == out[1] == ("E_ID_CASE_COLLISION",)
+ assert 2 not in out
+ assert "E_RGB_TOKEN_MISUSE" in out[3]
+ assert "E_DUPLICATE_VALUE" in label_problems(["DNA", "DNA"])[0]
+
+
+def test_a_header_proposal_needs_every_channel_to_equal_a_label_exactly():
+ labels = ["Actin", "DNA"]
+ assert proposal_for(layout(names=("DNA", "Actin")), labels) == ("DNA", "Actin")
+ assert proposal_for(layout(names=("DNA", "actin")), labels) is None # no case folding
+ assert proposal_for(layout(names=("DNA", None)), labels) is None
+ assert proposal_for(layout(names=("DNA", "DNA")), labels) is None
+ assert proposal_for(layout(names=("R", "G", "B"), rgb=True), ["R", "G", "B"]) is None
+
+
+def test_usage_and_aliases_come_from_the_saved_tags_only():
+ lay = [layout(("a", "r1", "single")), layout(("a", "r2", "single"), names=("DAPI", None))]
+ units = {
+ ("a", "r1", "single"): UnitAnswer(("DNA", "ACTIN"), "channels"),
+ ("a", "r2", "single"): UnitAnswer(("DNA", None), "channels"),
+ }
+ assert usage(lay, units, ["DNA", "ACTIN"]) == {
+ "DNA": (("a", "r1", "single"), ("a", "r2", "single")),
+ "ACTIN": (("a", "r1", "single"),),
+ }
+ assert aliases(lay, units) == {"DNA": ("DAPI",), "ACTIN": ("Actin",)}
+
+
+def test_remaining_counts_untagged_cells_and_unanswered_types():
+ lay = [layout()]
+ assert remaining(lay, {}, ["DNA"]).cells == 2
+ half = {lay[0].key: UnitAnswer(("DNA", None), None)}
+ left = remaining(lay, half, ["DNA"])
+ assert (left.cells, left.types, left.done) == (1, 1, False)
+ full = {lay[0].key: UnitAnswer(("DNA", "unused"), "channels")}
+ assert remaining(lay, full, ["DNA", "unused"]).done # an unused slot has a label too
+ assert not remaining(lay, {lay[0].key: UnitAnswer(("GONE", "DNA"), "channels")}, ["DNA"]).done
+
+
+def test_tags_labels_and_types_change_one_slice_and_keep_the_rest():
+ lay = layout()
+ base = DraftAnswers(ABSENT, SPEC_COMMIT)
+ tagged = with_tags(with_labels(base, {"DNA": "DAPI", "ACTIN": None}), lay, ["DNA", None])
+ assert tagged.units[lay.key] == UnitAnswer(("DNA", None))
+ assert tagged.channels["DNA"].display_name == "DAPI"
+ assert synced(tagged, [lay]).channels["DNA"].header_names == ("DAPI",)
+ # a label that is gone takes its tags back, never moves them to another label
+ gone = with_labels(tagged, {"ACTIN": None})
+ assert gone.units[lay.key].channels == (None, None)
+ # colour: R, G, B and nothing else, declared once a unit is colour
+ rgb = layout(names=(None, None, None), rgb=True)
+ coloured = synced(with_type(base, rgb, "rgb"), [rgb])
+ assert coloured.units[rgb.key] == UnitAnswer(("R", "G", "B"), "rgb")
+ assert {"R", "G", "B"} <= set(coloured.channels)
+ with pytest.raises(ValueError, match="3 channels"):
+ with_type(base, lay, "rgb")
+
+
+# --- the draft file ---------------------------------------------------------------------------
+
+
+def tagged_answers():
+ answers = complete()
+ units = dict(answers.units)
+ key = sorted(units)[0]
+ units[key] = UnitAnswer(("DNA", None), "channels")
+ units[sorted(units)[1]] = UnitAnswer((None, "ACTIN"), "rgb")
+ return replace(answers, units=units)
+
+
+def test_the_file_holds_untagged_positions_and_the_image_type():
+ answers = tagged_answers()
+ data = answers_to_bytes(answers)
+ assert b"null" in data
+ back = answers_from_bytes(data)
+ assert back.units == answers.units
+ assert answers_to_bytes(back) == data
+
+
+def test_an_empty_text_tag_is_refused_in_the_file():
+ data = answers_to_bytes(complete())
+ bad = data.replace(b'"DNA",\n', b'"",\n', 1)
+ assert bad != data
+ with pytest.raises(DraftFileError, match="channel ID"):
+ answers_from_bytes(bad)
+
+
+def test_the_file_is_version_3_and_older_versions_are_refused():
+ data = answers_to_bytes(complete())
+ assert b'"version": 3' in data
+ with pytest.raises(DraftFileError) as refused:
+ answers_from_bytes(data.replace(b'"version": 3', b'"version": 2'))
+ assert refused.value.code == "unknown_version"
+
+
+def test_an_image_type_the_format_does_not_have_is_refused():
+ data = answers_to_bytes(complete()).replace(b'"channels"\n', b'"channels"\n')
+ answers = complete()
+ key = sorted(answers.units)[0]
+ odd = replace(
+ answers, units={**answers.units, key: UnitAnswer(("DNA", "ACTIN"), "fluorescence")}
+ )
+ with pytest.raises(DraftFileError, match="image_type"):
+ answers_from_bytes(answers_to_bytes(odd))
+ assert data
+
+
+def test_a_tag_that_is_neither_text_nor_null_is_refused():
+ data = answers_to_bytes(complete())
+ bad = data.replace(b'"DNA",\n', b"7,\n", 1)
+ assert bad != data
+ with pytest.raises(DraftFileError, match="channel ID"):
+ answers_from_bytes(bad)
+
+
+@pytest.fixture
+def exp(tmp_path):
+ from draft_file_support import PATHS
+
+ make_tree(tmp_path / "exp", PATHS)
+ return discover_experiment(tmp_path / "exp")
+
+
+def test_an_untagged_position_blocks_the_description(exp):
+ base = draft_from_discovery(exp)
+ answers = complete()
+ key = sorted(UNITS)[0]
+ for tags, code in ((("DNA", None), "unit_channel_untagged"),):
+ units = {**answers.units, key: UnitAnswer(tags)}
+ new = replace(answers, units=units)
+ codes = {q.code for q in missing_confirmations(base, new)}
+ assert code in codes
+ with pytest.raises(DraftIncomplete):
+ description_from_draft(base, new)
+
+
+def test_channels_image_type_is_an_ordinary_unit_in_the_description(exp):
+ answers = complete()
+ key = sorted(UNITS)[0]
+ answers = replace(
+ answers, units={**answers.units, key: UnitAnswer(("DNA", "ACTIN"), "channels")}
+ )
+ description = description_from_draft(draft_from_discovery(exp), answers)
+ units = [
+ u
+ for r in description.repeats.values()
+ for m in r.members.values()
+ for u in m.technical_repeats.values()
+ ]
+ assert all(u.image_type is None for u in units)
+
+
+def test_carry_over_keeps_the_tags_of_the_units_that_remain(exp):
+ base = draft_from_discovery(exp)
+ answers = tagged_answers()
+ keep, drop = sorted(answers.units)[0], sorted(answers.units)[-1]
+ units = {keep: answers.units[keep], drop: UnitAnswer(("DNA", None), "channels")}
+ saved = replace(answers, units=units)
+ trimmed = discover_experiment(exp.root)
+ gone = tuple(
+ leaf
+ for leaf in trimmed.leaves
+ if (leaf.condition, leaf.biological_repeat, leaf.technical_repeat) != drop
+ )
+ now = draft_from_discovery(replace(trimmed, leaves=gone))
+ kept, dropped = carry_over(now, saved, {})
+ assert kept.units == {keep: answers.units[keep]}
+ assert any(" / ".join(drop) in line for line in dropped)
+ assert base is not None
diff --git a/tests/test_draft_pairing.py b/tests/test_draft_pairing.py
index 764687a3..3ed52b1e 100644
--- a/tests/test_draft_pairing.py
+++ b/tests/test_draft_pairing.py
@@ -102,7 +102,7 @@ def test_the_pairing_is_saved_and_resumed_from_the_draft_file(base, tmp_path):
def test_the_file_of_an_older_version_without_pairing_is_refused(base):
- old = answers_to_bytes(answers()).replace(b'"version": 2', b'"version": 1')
+ old = answers_to_bytes(answers()).replace(b'"version": 3', b'"version": 1')
with pytest.raises(DraftFileError) as refused:
answers_from_bytes(old)
assert refused.value.code == "unknown_version"
diff --git a/tests/test_embedding_api.py b/tests/test_embedding_api.py
index 7be84088..5eb5fff0 100644
--- a/tests/test_embedding_api.py
+++ b/tests/test_embedding_api.py
@@ -59,6 +59,10 @@ def _parameters(function):
("intro", KEYWORD, None),
("show_next", KEYWORD, True),
],
+ "channels_step": [
+ ("title", POSITIONAL, None),
+ ("purpose", POSITIONAL, None),
+ ],
"run_step": [],
"results_step": [],
"check_folder": [],
diff --git a/tests/test_notebook_channels.py b/tests/test_notebook_channels.py
new file mode 100644
index 00000000..a762e0ef
--- /dev/null
+++ b/tests/test_notebook_channels.py
@@ -0,0 +1,481 @@
+"""The *Your channels* screen: labels first, then tags and image type per unit (Stage 1, PR 15).
+
+Widgets are driven headless, as in ``test_notebook_description``. Nothing is saved or confirmed
+until the user answers; every answer is in ``morphobricks.draft.json``.
+"""
+
+import shutil
+
+import pytest
+
+from channel_images import write_channels, write_rgb
+from morphobricks_preprocessing.description_file import DRAFT_FILENAME
+from morphobricks_preprocessing.draft_format import answers_from_bytes
+
+pytest.importorskip("ipywidgets")
+pytest.importorskip("matplotlib")
+
+from morphobricks_preprocessing import notebook_style as ui # noqa: E402
+from morphobricks_preprocessing.notebook_ui import PreprocessingNotebook # noqa: E402
+
+FOLDERS = ["ctrl/rep1", "ctrl/rep2", "drug/rep1", "drug/rep2"]
+KINDS = ["blobs", "filaments", "cells"]
+DNA, ACTIN, PHASE = "DNA", "ACTIN", "PHASE"
+
+
+def tree(base, names_of, folders=FOLDERS):
+ """Write one named-channel image per folder; ``names_of(folder)`` gives its header names."""
+ for i, folder in enumerate(folders):
+ write_channels(base / "exp" / folder / "a.tif", names_of(folder), KINDS, seed=i + 1)
+ return base / "exp"
+
+
+def open_screen(root, out):
+ """Open the notebook on ``root``, check the folder and confirm the axis roles."""
+ nb = PreprocessingNotebook(str(root), output_folder=str(out))
+ for step in (nb.welcome, nb.folder_step, nb.axis_step, nb.channels_step, nb.run_step):
+ step()
+ nb.check_button.click()
+ nb.labels_confirmed.button.click()
+ nb.preview_checked.set(True)
+ nb.show_preview()
+ nb.confirm_button.click()
+ assert nb.confirmed_mapping is not None
+ return nb
+
+
+def set_labels(nb, labels):
+ """Type the labels into the slots (adding slots as needed) and confirm the list."""
+ card = nb.channel_labels
+ while len(card.rows) < len(labels):
+ card.add_button.click()
+ for row, text in zip(card.rows, labels + [""] * len(card.rows), strict=False):
+ row.id_box.value = text
+ card.confirm.button.click()
+ assert card.confirmed
+
+
+def saved(out):
+ return answers_from_bytes((out / DRAFT_FILENAME).read_bytes())
+
+
+def tag_all(nb, tags=(DNA, ACTIN, PHASE), *, skip=()):
+ """Tag every cell of every unit (not those of ``skip``) and answer every image type."""
+ grid = nb.channel_grid
+ for key, index in list(grid.cells):
+ if key not in skip:
+ grid.cells[(key, index)].value = tags[index]
+ for layout in grid.layouts:
+ if layout.key not in skip:
+ nb.channel_types.radios[layout.key].value = "channels"
+
+
+@pytest.fixture
+def out(tmp_path):
+ return tmp_path / "out"
+
+
+@pytest.fixture
+def plain(tmp_path):
+ """Four units whose headers say DAPI, Actin and Ch3: nothing a label could be."""
+ return tree(tmp_path, lambda f: ["DAPI", "Actin", "Ch3"])
+
+
+# --- step A: the label list ----------------------------------------------------------------------
+
+
+def test_generic_header_names_are_not_suggested_as_labels(plain, out):
+ nb = open_screen(plain, out)
+ card = nb.channel_labels
+ assert [r.id_box.value for r in card.rows] == ["Actin", "", ""] # one slot per channel
+ assert "DAPI" in card.skipped.value and "Ch3" in card.skipped.value
+ assert not any(r.id_box.value in {"DAPI", "Ch3"} for r in card.rows)
+ assert not card.confirmed and not (out / DRAFT_FILENAME).exists()
+ assert not ui.is_visible(nb.channel_grid.widget) # step B waits for the labels
+
+
+def test_the_slots_are_the_channel_count_of_the_images(tmp_path, out):
+ root = tmp_path / "exp"
+ for folder in FOLDERS:
+ write_channels(root / folder / "a.tif", ["Actin", "Tubulin"], KINDS[:2])
+ nb = open_screen(root, out)
+ assert nb.channel_labels.slots == 2 and len(nb.channel_labels.rows) == 2
+
+
+def test_near_identical_spellings_are_grouped_and_never_merged(tmp_path, out):
+ root = tree(tmp_path, lambda f: ["Lamin", "LaminB", "Actin"])
+ nb = open_screen(root, out)
+ card = nb.channel_labels
+ assert card.groups == [("Lamin", "LaminB")]
+ assert [r.id_box.value for r in card.rows] == ["Actin", "", ""] # neither is a label yet
+ assert "look alike" in card.groups_box.children[0].children[0].value
+ assert "Lamin" in card.groups_box.children[0].children[0].value
+ assert not (out / DRAFT_FILENAME).exists()
+ # the user decides: "Same label" puts the chosen spelling in a slot, and only that one
+ _, buttons = card.groups_box.children[0].children
+ pick, same, different, _ = buttons.children
+ pick.value = "LaminB"
+ same.click()
+ assert [r.id_box.value for r in card.rows] == ["Actin", "LaminB", ""]
+ assert card.groups == []
+
+
+def test_different_labels_puts_each_spelling_in_a_slot(tmp_path, out):
+ root = tree(tmp_path, lambda f: ["Lamin", "LaminB", "Actin"])
+ card = open_screen(root, out).channel_labels
+ card.groups_box.children[0].children[1].children[2].click()
+ assert sorted(r.id_box.value for r in card.rows) == ["Actin", "Lamin", "LaminB"]
+
+
+def test_spellings_that_differ_only_by_case_cannot_be_different_labels(tmp_path, out):
+ root = tree(tmp_path, lambda f: ["Actin", "ACTIN", "Golgi"])
+ card = open_screen(root, out).channel_labels
+ assert card.groups == [("ACTIN", "Actin")]
+ different = card.groups_box.children[0].children[1].children[2]
+ assert different.disabled
+ assert "E_ID_CASE_COLLISION" in card.groups_box.children[0].children[0].value
+
+
+def test_a_case_collision_is_refused_at_once_and_nothing_is_saved(plain, out):
+ nb = open_screen(plain, out)
+ card = nb.channel_labels
+ card.rows[0].id_box.value = "Actin"
+ card.rows[1].id_box.value = "ACTIN"
+ assert "E_ID_CASE_COLLISION" in card.rows[0].state.value
+ assert "E_ID_CASE_COLLISION" in card.rows[1].state.value
+ assert card.confirm.disabled
+ card.confirm.button.click()
+ assert not card.confirmed and not (out / DRAFT_FILENAME).exists()
+
+
+def test_add_label_adds_a_slot(plain, out):
+ nb = open_screen(plain, out)
+ card = nb.channel_labels
+ before = len(card.rows)
+ card.add_button.click()
+ assert len(card.rows) == before + 1
+ set_labels(nb, ["DNA", "ACTIN", "PHASE", "LAMIN"])
+ assert sorted(saved(out).channels) == ["ACTIN", "DNA", "LAMIN", "PHASE"]
+
+
+def test_the_label_list_is_saved_with_shown_as_names_only_on_confirmation(plain, out):
+ nb = open_screen(plain, out)
+ card = nb.channel_labels
+ card.rows[0].id_box.value, card.rows[0].name_box.value = DNA, "DAPI"
+ assert not (out / DRAFT_FILENAME).exists()
+ set_labels(nb, [DNA, ACTIN, PHASE])
+ assert saved(out).channels[DNA].display_name == "DAPI"
+
+
+# --- step B: the tag grid ------------------------------------------------------------------------
+
+
+def test_each_dropdown_offers_the_labels_only(plain, out):
+ nb = open_screen(plain, out)
+ set_labels(nb, [DNA, ACTIN, PHASE])
+ grid = nb.channel_grid
+ assert ui.is_visible(grid.widget)
+ assert len(grid.cells) == 4 * 3 # one dropdown per unit and C index
+ for drop in grid.cells.values():
+ assert [value for _, value in drop.options] == [None, ACTIN, DNA, PHASE]
+ assert all(drop.value is None for drop in grid.cells.values()) # nothing chosen for you
+
+
+def test_each_unit_has_a_thumbnail_a_c_index_and_its_quoted_header_name(plain, out):
+ nb = open_screen(plain, out)
+ set_labels(nb, [DNA, ACTIN, PHASE])
+ text = repr(nb.channel_grid.blocks.children)
+ assert "‘DAPI’" in text or "‘DAPI’" in text
+ assert "first image by path: ctrl/rep1/a.tif" in text
+ pictures = [w for w in _all(nb.channel_grid.blocks) if w.__class__.__name__ == "Image"]
+ assert len(pictures) == 12 and all(p.value[:4] == b"\x89PNG" for p in pictures)
+
+
+def _all(widget):
+ yield widget
+ for child in getattr(widget, "children", ()):
+ yield from _all(child)
+
+
+def test_incomplete_tagging_blocks_completion(plain, out):
+ nb = open_screen(plain, out)
+ set_labels(nb, [DNA, ACTIN, PHASE])
+ grid = nb.channel_grid
+ assert not grid.complete and "Not finished" in grid.status.value
+ tag_all(nb, skip=[("drug", "rep2", "single")])
+ assert not grid.complete
+ assert "3 channel(s) still to tag" in grid.status.value
+ assert "1 image type(s) still to choose" in grid.status.value
+ tag_all(nb)
+ assert grid.complete and "Every channel of every one of the 4 units" in grid.status.value
+
+
+def test_the_dropdown_has_no_empty_option_and_an_unused_slot_needs_a_label(plain, out):
+ nb = open_screen(plain, out)
+ set_labels(nb, [DNA, ACTIN, "unused"])
+ grid = nb.channel_grid
+ for drop in grid.cells.values():
+ assert [label for _, label in drop.options] == [None, ACTIN, DNA, "unused"]
+ assert "unused" in nb.channel_labels.widget.children[1].value # the Step A hint
+ tag_all(nb, (DNA, ACTIN, "unused"))
+ assert grid.complete
+ assert all(u.channels == (DNA, ACTIN, "unused") for u in saved(out).units.values())
+
+
+def test_the_step_b_is_incomplete_while_one_position_has_no_label(plain, out):
+ nb = open_screen(plain, out)
+ set_labels(nb, [DNA, ACTIN, "unused"])
+ tag_all(nb, (DNA, ACTIN, "unused"), skip=[("drug", "rep2", "single")])
+ assert not nb.channel_grid.complete
+
+
+def test_a_chip_shows_the_id_and_the_shown_as_name_only_as_tooltip(plain, out):
+ nb = open_screen(plain, out)
+ card = nb.channel_labels
+ while len(card.rows) < 3:
+ card.add_button.click()
+ for row, text in zip(card.rows, [DNA, ACTIN, "unused"], strict=True):
+ row.id_box.value = text
+ card.rows[0].name_box.value = "DAPI stain"
+ assert ">DNA" in card.rows[0].state.value
+ assert "DAPI stain" not in card.rows[0].state.value
+ assert "title='DAPI stain'" in card.rows[0].state.value
+ card.confirm.button.click()
+ tag_all(nb, (DNA, ACTIN, "unused"))
+ html_texts = [w.value for w in _all(nb.channel_grid.widget) if hasattr(w, "value")]
+ assert any(
+ ">DNA" in v and "title='DAPI stain'" in v for v in html_texts if isinstance(v, str)
+ )
+ assert not any("DAPI stain" in v for v in html_texts if isinstance(v, str))
+
+
+def test_a_label_cannot_stand_at_two_positions_of_a_unit(plain, out):
+ nb = open_screen(plain, out)
+ set_labels(nb, [DNA, ACTIN, PHASE])
+ grid = nb.channel_grid
+ key = ("ctrl", "rep1", "single")
+ grid.cells[(key, 0)].value = DNA
+ grid.cells[(key, 1)].value = DNA
+ assert "one label cannot stand at two positions" in grid.problem.value
+ assert saved(out).units[key].channels == (DNA, None, None)
+
+
+def test_a_label_used_in_only_some_units_says_what_will_be_refused(plain, out):
+ nb = open_screen(plain, out)
+ set_labels(nb, [DNA, ACTIN, PHASE])
+ grid = nb.channel_grid
+ tag_all(nb, skip=[("drug", "rep2", "single")])
+ text = grid.usage_html.value
+ assert "used in 3 of 4 units: an analysis of DNA will be refused for the other units" in text
+ assert "A_SCOPE_INCOMPLETE" in text
+ tag_all(nb)
+ assert "used in all 4 units" in grid.usage_html.value
+ assert "will be refused" not in grid.usage_html.value
+
+
+def test_a_label_in_another_column_in_one_unit_is_pointed_out(plain, out):
+ nb = open_screen(plain, out)
+ set_labels(nb, [DNA, ACTIN, PHASE])
+ tag_all(nb, skip=[("drug", "rep2", "single")])
+ key = ("drug", "rep2", "single")
+ grid = nb.channel_grid
+ for index, tag in enumerate((ACTIN, DNA, PHASE)):
+ grid.cells[(key, index)].value = tag
+ assert "DNA is not always in the same column" in grid.usage_html.value
+
+
+def test_a_pre_tag_from_the_header_is_only_a_suggestion_until_confirmed(tmp_path, out):
+ root = tree(tmp_path, lambda f: ["DNA", "ACTIN", "PHASE"])
+ nb = open_screen(root, out)
+ assert [r.id_box.value for r in nb.channel_labels.rows] == ["DNA", "ACTIN", "PHASE"]
+ set_labels(nb, ["ACTIN", "DNA", "PHASE"])
+ grid = nb.channel_grid
+ assert {k[1]: d.value for k, d in grid.cells.items() if k[0] == ("ctrl", "rep1", "single")} == {
+ 0: DNA,
+ 1: ACTIN,
+ 2: PHASE,
+ }
+ assert "proposed, not confirmed" in repr(grid.blocks.children)
+ assert not saved(out).units # nothing is the user's answer yet
+ assert not grid.complete
+ assert len(grid.confirm_buttons) == 1 # one confirm for the one layout
+ assert "4 unit(s)" in grid.confirm_buttons[0].description
+ grid.confirm_buttons[0].click()
+ assert all(u.channels == (DNA, ACTIN, PHASE) for u in saved(out).units.values())
+ assert "proposed, not confirmed" not in repr(grid.blocks.children)
+ assert not grid.confirm_buttons
+ assert not grid.complete # the image type is still to answer
+
+
+def test_units_with_different_header_names_are_different_layouts(tmp_path, out):
+ root = tree(
+ tmp_path, lambda f: ["DNA", "ACTIN", "PHASE"] if "ctrl" in f else ["DNA", "ACTIN", "x"]
+ )
+ nb = open_screen(root, out)
+ set_labels(nb, ["ACTIN", "DNA", "PHASE"])
+ grid = nb.channel_grid
+ assert len(grid.confirm_buttons) == 1 # only the layout that equals the labels exactly
+ assert "2 unit(s)" in grid.confirm_buttons[0].description
+ assert grid.cells[(("drug", "rep1", "single"), 0)].value is None
+
+
+# --- the image type ------------------------------------------------------------------------------
+
+
+def test_the_image_type_is_not_preselected(plain, out):
+ nb = open_screen(plain, out)
+ set_labels(nb, [DNA, ACTIN, PHASE])
+ radios = nb.channel_types.radios
+ assert len(radios) == 4 and all(r.value is None for r in radios.values())
+ assert [value for _, value in next(iter(radios.values())).options] == ["channels", "rgb"]
+ assert not (saved(out).units)
+
+
+def rgb_tree(base):
+ for i, folder in enumerate(FOLDERS):
+ write_rgb(base / "exp" / folder / "a.tif", seed=i + 1)
+ return base / "exp"
+
+
+def test_colour_is_pre_ticked_only_when_the_header_says_rgb(tmp_path, out):
+ nb = open_screen(rgb_tree(tmp_path), out)
+ assert nb.channel_labels.slots == 1 # colour images have no stain slots
+ set_labels(nb, ["DNA"])
+ radios = nb.channel_types.radios
+ assert len(radios) == 4 and all(r.value == "rgb" for r in radios.values())
+ key = ("drug", "rep2", "single")
+ assert "Not saved until you confirm" in nb.channel_types.notes[key].value
+ assert not saved(out).units # the pre-tick is not an answer
+ assert not nb.channel_grid.complete
+
+
+def test_confirming_the_colour_layout_declares_r_g_b_and_completes_every_unit(tmp_path, out):
+ nb = open_screen(rgb_tree(tmp_path), out)
+ set_labels(nb, ["DNA"])
+ grid = nb.channel_grid
+ assert len(grid.confirm_buttons) == 1 and "4 unit(s)" in grid.confirm_buttons[0].description
+ grid.confirm_buttons[0].click()
+ answers = saved(out)
+ assert all(
+ (u.channels, u.image_type) == (("R", "G", "B"), "rgb") for u in answers.units.values()
+ )
+ assert {"R", "G", "B", "DNA"} == set(answers.channels)
+ assert all(answers.channels[t].header_names is None for t in "RGB")
+ assert not grid.cells # a colour unit has no dropdown: its channels are R, G, B
+ assert grid.complete
+
+
+def test_a_type_that_contradicts_the_header_is_allowed_and_said(plain, out):
+ nb = open_screen(plain, out)
+ set_labels(nb, [DNA, ACTIN, PHASE])
+ key = ("ctrl", "rep1", "single")
+ nb.channel_types.radios[key].value = "rgb" # three channels: allowed, the run will refuse
+ assert "header does not say colour" in nb.channel_types.notes[key].value
+ assert saved(out).units[key].image_type == "rgb"
+
+
+# --- saving and resuming ---------------------------------------------------------------------
+
+
+def test_every_answer_is_saved_and_a_reopened_notebook_shows_it(plain, out):
+ nb = open_screen(plain, out)
+ set_labels(nb, [DNA, ACTIN, PHASE])
+ tag_all(nb)
+ answers = saved(out)
+ assert {c: a.header_names for c, a in answers.channels.items()} == {
+ ACTIN: ("Actin",),
+ DNA: ("DAPI",),
+ PHASE: ("Ch3",),
+ } # the tag is the confirmation of each header name as evidence
+ again = open_screen(plain, out)
+ card, grid = again.channel_labels, again.channel_grid
+ assert card.confirmed and card.labels == [ACTIN, DNA, PHASE]
+ assert ui.is_visible(grid.widget) and grid.complete
+ assert all(d.value == (DNA, ACTIN, PHASE)[k[1]] for k, d in grid.cells.items())
+ assert all(r.value == "channels" for r in again.channel_types.radios.values())
+ assert "Every channel of every one of the 4 units" in grid.status.value
+
+
+def test_changing_the_labels_takes_back_the_tags_of_a_label_that_is_gone(plain, out):
+ nb = open_screen(plain, out)
+ set_labels(nb, [DNA, ACTIN, PHASE])
+ tag_all(nb)
+ card = nb.channel_labels
+ card.confirm.button.click() # change the labels
+ assert not card.confirmed and not ui.is_visible(nb.channel_grid.widget)
+ ids = [r.id_box.value for r in card.rows]
+ card.rows[ids.index(PHASE)].id_box.value = "GOLGI"
+ card.confirm.button.click()
+ assert "tag(s) of a label that is no longer in the list were taken back" in card.status.value
+ assert all(u.channels == (DNA, ACTIN, None) for u in saved(out).units.values())
+ assert not nb.channel_grid.complete
+
+
+def test_the_other_cards_answers_are_not_touched_by_the_channel_answers(plain, out):
+ nb = open_screen(plain, out)
+ nb.id_inputs[("conditions", "ctrl")].value = "ctrl"
+ nb.ids_confirmed.button.click()
+ before = saved(out)
+ set_labels(nb, [DNA, ACTIN, PHASE])
+ after = saved(out)
+ assert after.conditions == before.conditions and after.repeats == before.repeats
+
+
+def test_a_removed_unit_drops_only_its_own_answer(plain, out):
+ nb = open_screen(plain, out)
+ set_labels(nb, [DNA, ACTIN, PHASE])
+ tag_all(nb)
+ assert len(saved(out).units) == 4
+ shutil.rmtree(plain / "drug" / "rep2")
+ again = open_screen(plain, out)
+ assert "drug / rep2 / single" in again.names_status.value
+ grid = again.channel_grid
+ assert grid.complete and len(grid.layouts) == 3
+ assert all(d.value == (DNA, ACTIN, PHASE)[k[1]] for k, d in grid.cells.items())
+ assert len(saved(out).units) == 4 # the file is read, not rewritten, until the next answer
+ grid.cells[(("ctrl", "rep1", "single"), 2)].value = None # the next answer saves what remains
+ assert set(saved(out).units) == {
+ ("ctrl", "rep1", "single"),
+ ("ctrl", "rep2", "single"),
+ ("drug", "rep1", "single"),
+ }
+
+
+def test_a_unit_whose_channel_count_changed_is_tagged_again_not_stretched(plain, out):
+ nb = open_screen(plain, out)
+ set_labels(nb, [DNA, ACTIN, PHASE])
+ tag_all(nb)
+ for folder in FOLDERS:
+ write_channels(plain / folder / "a.tif", ["DAPI", "Actin"], KINDS[:2])
+ again = open_screen(plain, out)
+ grid = again.channel_grid
+ assert sorted({k[1] for k in grid.cells}) == [0, 1]
+ assert all(d.value is None for d in grid.cells.values())
+ assert not grid.complete
+
+
+# --- the screen waits for what it needs ---------------------------------------------------------
+
+
+def test_the_screen_waits_for_the_axis_roles(plain, out):
+ nb = PreprocessingNotebook(str(plain), output_folder=str(out))
+ nb.channels_step()
+ assert not ui.is_visible(nb.channel_labels.widget)
+ nb.check_button.click()
+ nb.labels_confirmed.button.click()
+ assert "Waiting for the axis roles" in nb.channels_waiting.value
+ assert not ui.is_visible(nb.channel_labels.widget)
+
+
+def test_the_enlarged_view_shows_the_1024_px_picture(plain, out):
+ nb = open_screen(plain, out)
+ set_labels(nb, [DNA, ACTIN, PHASE])
+ grid = nb.channel_grid
+ zoom = next(
+ w
+ for w in _all(grid.blocks)
+ if w.__class__.__name__ == "Button" and "Enlarge" in w.description
+ )
+ zoom.click()
+ assert len(grid.detail.children) == 3
+ assert grid.detail.children[1].value[:4] == b"\x89PNG"
diff --git a/tests/test_notebook_setup.py b/tests/test_notebook_setup.py
index df72fa48..21fbc882 100644
--- a/tests/test_notebook_setup.py
+++ b/tests/test_notebook_setup.py
@@ -108,20 +108,8 @@ def import_interface(name):
import ipywidgets as widgets
class FakeNotebook:
- def welcome(self):
- return widgets.HTML("welcome")
-
- def folder_step(self):
- return widgets.HTML("folder")
-
- def axis_step(self):
- return widgets.HTML("axes")
-
- def run_step(self):
- return widgets.HTML("run")
-
- def results_step(self):
- return widgets.HTML("results")
+ def __getattr__(self, name): # welcome, folder_step, axis_step, channels_step, ...
+ return lambda: widgets.HTML(name)
return SimpleNamespace(PreprocessingNotebook=FakeNotebook)