Batch 34b: preview-gated one-click axes, header evidence on reopen, NaN crop

src/morphobricks_preprocessing/description_reconcile.py

@@ -65,11 +65,17 @@ class ObservedUnit:
 
     ``channel_count`` and ``header_names`` (raw names by C index, None where there is none) are
     None when the headers were not read or do not state them: that is not evidence of a change.
+    ``uniform`` is False when the unit's headers disagree with each other (it is no longer one
+    layout): its channel evidence then no longer fits its declared channel list. ``rgb`` is
+    whether the headers state a colour (RGB) image, None when they were not read or do not say
+    (OME-Zarr): a declared image type the headers now contradict is a channel change.
     """
 
     image_count: int
     channel_count: int | None = None
     header_names: tuple[str | None, ...] | None = None
+    uniform: bool = True
+    rgb: bool | None = None
 
     def __post_init__(self) -> None:
         if self.image_count < 1:
@@ -224,6 +230,10 @@ def _channels_differ(
 ) -> bool:
     """Return whether the observed channel evidence no longer fits the declared channel list."""
     unit = published.unit
+    if not seen.uniform:
+        return True
+    if seen.rgb is not None and seen.rgb != (unit.image_type == "rgb"):
+        return True  # as the run's checks compare it (IMAGE_TYPE_CONTRADICTS_HEADER)
     if seen.channel_count is not None and seen.channel_count != len(unit.channels):
         return True
     if seen.header_names is None or unit.image_type == "rgb":

src/morphobricks_preprocessing/description_reopen.py

@@ -9,7 +9,7 @@
 
 from __future__ import annotations
 
-from collections.abc import Collection
+from collections.abc import Collection, Iterable
 from dataclasses import dataclass
 from datetime import date
 from pathlib import Path
@@ -19,6 +19,7 @@
 from .description_reconcile import Reconciliation, reconcile
 from .description_removal import RemovalQuestion, removal_questions
 from .discovery import DiscoveryResult
+from .metadata import TiffMetadata
 from .reconcile_observe import observe_folders
 
 
@@ -38,16 +39,27 @@ class PublishedState:
 
     @property
     def changed(self) -> bool:
-        """True when a folder is new or removed, or a member's image count changed."""
+        """True when the folders or headers differ from the saved file.
+
+        A folder is new or removed, a member's image count changed, or a unit's headers no
+        longer fit its saved channels (any reason that reopens a repeat).
+        """
         rec = self.reconciliation
-        return bool(rec.new or rec.removed or rec.counts)
+        return bool(rec.new or rec.removed or rec.counts or rec.reopened)
 
 
 def published_state(
-    output: Path, discovery: DiscoveryResult, excluded: Collection[str] = ()
+    output: Path,
+    discovery: DiscoveryResult,
+    excluded: Collection[str] = (),
+    metadata: Iterable[TiffMetadata] | None = None,
 ) -> tuple[PublishedState | None, str]:
     """Return the published file compared with the folders, or None and why it cannot be used.
 
+    ``metadata`` are the image headers already read (the preflight's): with them each unit's
+    channel count and header names are compared with its saved channels, so a header changed
+    under the same files is a change, never "nothing changed".
+
     Returns
     -------
     tuple
@@ -65,7 +77,7 @@ def published_state(
         return None, f"{DESCRIPTION_FILENAME} cannot be used ({codes})"
     except OSError as exc:
         return None, f"{DESCRIPTION_FILENAME} cannot be read: {exc}"
-    rec = reconcile(published, observe_folders(discovery, excluded))
+    rec = reconcile(published, observe_folders(discovery, excluded, metadata))
     day = date.fromtimestamp(path.stat().st_mtime).isoformat()
     return PublishedState(
         published, sha256_hex(data), day, rec, removal_questions(published, rec)

src/morphobricks_preprocessing/notebook_axes_quick.py

@@ -25,6 +25,7 @@
 from .axes import AxisMapping, axis_name
 from .axis_check import ConfirmedAxes
 from .axis_review import review_axes
+from .description_reconcile import CHANNELS_CHANGED
 from .fingerprint import unchanged_confirmation
 
 DETAILS_OPEN, DETAILS_CLOSED = " Hide the role boxes", " Look closer, or change the roles"
@@ -133,9 +134,25 @@ def _refresh_quick_axes(self) -> None:
                 self._quick_filling = False
         self.quick_text.value = self._quick_text(offer)
         self.show_preview()
-        self.quick_image.value = self.preview_image.value
-        ui.set_visible(self.quick_image, bool(self.preview_image.value))
-        self.quick_caption.value = self.preview_caption.value
+        shown = self._quick_preview_shown(offer)
+        self.quick_image.value = self.preview_image.value if shown else b""
+        ui.set_visible(self.quick_image, shown)
+        self.quick_confirm.disabled = not shown
+        self.quick_caption.value = self.preview_caption.value + (
+            ""
+            if shown
+            else ui.status(
+                "warn",
+                "Confirm these axes is not offered: no preview of these roles is shown.",
+                ["Open the role boxes below to look closer or change the roles."],
+            )
+        )
+
+    def _quick_preview_shown(self, offer: AxisMapping) -> bool:
+        """Return whether the preview now shows a plane drawn with exactly the offered roles."""
+        drawn = getattr(self, "_preview_drawn", None)
+        same = drawn is not None and (drawn.axes, drawn.roles) == (offer.axes, offer.roles)
+        return bool(self.preview_image.value) and same
 
     def _quick_text(self, offer: AxisMapping) -> str:
         """Return what the card says: the roles, where they come from, what the click does."""
@@ -153,13 +170,25 @@ def _quick_text(self, offer: AxisMapping) -> str:
             "info",
             f"Suggested: {roles}.",
             [
+                *self._headers_changed_line(),
                 f"These are {source}; the axis check finds nothing to ask about them.",
                 "Look at the preview: if it looks like a normal image of your sample, confirm. "
                 "Nothing is confirmed until you click.",
             ],
             raw=True,
         )
 
+    def _headers_changed_line(self) -> list[str]:
+        """Say that the headers changed since the save, so the earlier roles are not offered."""
+        state = self._published() if self.preflight is not None else None
+        units = () if state is None else state.reconciliation.units
+        if not any(CHANNELS_CHANGED in unit.reasons for unit in units):
+            return []
+        return [
+            "Some image headers changed since the experiment was saved: the roles the earlier "
+            "run recorded are not offered. Review these afresh."
+        ]
+
     # The click ---------------------------------------------------------------------------
 
     def _confirm_quick_axes(self) -> bool:
@@ -176,6 +205,8 @@ def _confirm_quick_axes(self) -> bool:
         if self._current_mapping() != offer:
             self._refresh_quick_axes()
             return False
+        if not self._quick_preview_shown(offer):
+            return False  # no preview of these roles was shown: the click cannot vouch for it
         self.preview_checked.set(True)  # the user's click says the preview looks right
         self._toggle_confirmation()
         return self.confirmed_mapping is not None

src/morphobricks_preprocessing/notebook_axis_preview.py

@@ -11,7 +11,8 @@
 from .notebook_options import IMAGE_TYPE_NAMES
 from .notebook_plots import plane_png
 from .notebook_render import preview_caption
-from .planes import PlaneReadError, read_plane
+from .planes import Plane, PlaneReadError, read_plane
+from .preview import PreviewRefused, check_unchanged, checked_fingerprint
 from .validation import plural
 
 
@@ -68,9 +69,16 @@ def _render_preview_if_shown(self) -> None:
             self._render_preview()
 
     def _render_preview(self) -> None:
-        """Draw the chosen plane with the roles shown, labelled confirmed or not (PP-10)."""
+        """Draw the chosen plane with the roles shown, labelled confirmed or not (PP-10).
+
+        ``_preview_drawn`` is the mapping of the plane now shown, None when none is (no image
+        fits, or the plane cannot be read): the one-click confirmation needs it (BLOCKER 3).
+        """
+        self._preview_drawn = None
         header = self._preview_header()
         if header is None:
+            self.preview_image.value = b""
+            ui.set_visible(self.preview_image, False)
             self.preview_caption.value = ui.status("warn", "No image fits these axis roles.")
             return
         file_mapping = file_mapping_for(header, self.mapping)
@@ -80,15 +88,8 @@ def _render_preview(self) -> None:
             if file_mapping.index_of(role) is not None
         }
         try:
-            with self.log.attached():
-                plane = read_plane(
-                    self.preflight.discovery.root / header.relative_path,
-                    file_mapping,
-                    memory_limit=self._memory_limit(),
-                    preview=True,
-                    **coords,
-                )
-        except (PlaneReadError, IndexError, ValueError) as exc:
+            plane = self._bound_plane(header.relative_path, file_mapping, coords)
+        except (PreviewRefused, PlaneReadError, IndexError, ValueError, OSError) as exc:
             self.preview_image.value = b""
             ui.set_visible(self.preview_image, False)
             self.preview_caption.value = ui.status(
@@ -97,11 +98,36 @@ def _render_preview(self) -> None:
             return
         self.preview_image.value = plane_png(plane)
         ui.set_visible(self.preview_image, True)
+        self._preview_drawn = self.mapping
         self.preview_caption.value = self._preview_label() + preview_caption(
             header.relative_path, plane
         )
         self._show_log()
 
+    def _bound_plane(self, rel: str, file_mapping: AxisMapping, coords: dict) -> Plane:
+        """Read the preview plane of the image only as the folder check read its header.
+
+        The fingerprint the check bound to the header must hold before and after the read: an
+        image changed since the check (same name, same shape or not) is refused, never shown as
+        the image checked, so the one-click confirmation is not offered on it. The binding is
+        the one every other picture uses (:func:`.preview.checked_fingerprint`).
+
+        Raises
+        ------
+        PreviewRefused
+            When the image is not bound, or is no longer that revision ("check the folder
+            again").
+        """
+        path = self.preflight.discovery.root / rel
+        expected = checked_fingerprint(self.preflight.header_fingerprints, rel)
+        check_unchanged(path, expected, rel)
+        with self.log.attached():
+            plane = read_plane(
+                path, file_mapping, memory_limit=self._memory_limit(), preview=True, **coords
+            )
+        check_unchanged(path, expected, rel)
+        return plane
+
     def _preview_label(self) -> str:
         """Say whether the preview shows the confirmed roles or roles still being edited."""
         if self.confirmed_mapping is not None:

src/morphobricks_preprocessing/notebook_axis_step.py

@@ -156,6 +156,7 @@ def _refresh_axis_step(self) -> None:
         pre = self.preflight
         self.axis_next.value = ""
         self._preview_shown = False
+        self._preview_drawn = None
         self._close_image_type()
         waiting = self._axis_waiting()
         if waiting:

src/morphobricks_preprocessing/notebook_check_save.py

@@ -162,6 +162,13 @@ def _run_allowed(self) -> bool:
         check = self._published_check()
         return check is not None and check.can_run
 
+    def _published_refused(self) -> bool:
+        """Return whether the run's checks ran on the published file and refuse it."""
+        if self.confirmed_mapping is None:
+            return False  # the physical checks wait for the roles: nothing is decided yet
+        check = self._published_check()
+        return check is not None and check.physical_run and not check.can_run
+
     def _published_check(self) -> CandidateCheck | None:
         """Return the run's checks of the published file (None without one)."""
         state = self._published()
@@ -345,7 +352,9 @@ def _show_removals(self, state: object) -> None:
     def _show_saved_state(self) -> None:
         """Show the green saved line (folded or not) and the run gate."""
         state = self._published()
-        if self._saved_current():
+        if self._published_refused():
+            self._folded = False  # a refusal of the run's checks is never folded away
+        if self._saved_current() and not self._published_refused():
             when = self._saved_on or (state.saved_on if state else "")
             self.save_status.value = ui.status("ok", f"Saved {when}. The run (step 12) can start.")
             self.saved_line.value = ui.status("ok", f"Saved on {when}, nothing changed.")
@@ -354,6 +363,7 @@ def _show_saved_state(self) -> None:
             if self.save_status.value.find("Saved") >= 0:
                 self.save_status.value = ""
         ui.set_visible(self.fold_bar, self._folded)
+        ui.set_visible(self.check_card, not self._folded)
         for name in FOLDED:
             ui.set_visible(getattr(self, name), not self._folded)
         self._refresh_run_step()

src/morphobricks_preprocessing/notebook_check_view.py

@@ -11,7 +11,7 @@
 
 from . import notebook_style as ui
 from .description_candidate import CandidateCheck
-from .description_reconcile import Reconciliation, UnitFolders
+from .description_reconcile import CHANNELS_CHANGED, Reconciliation, UnitFolders
 from .description_removal import RemovalQuestion
 from .description_review import MemberView, ReplicateView, unit_text
 from .draft import OpenQuestion
@@ -187,13 +187,19 @@ def _folders(unit: UnitFolders) -> str:
 
 
 def changes_html(rec: Reconciliation) -> str:
-    """Say in orange what changed in the folders since the description was saved."""
+    """Say in orange what changed in the folders (or their headers) since the save."""
     lines = [f"new: {_folders(u.folders)} · {plural(u.image_count or 0, 'image')}" for u in rec.new]
     lines += [f"removed: {_folders(u.folders)}" for u in rec.removed]
     lines += [
         f"{c.repeat_id} / {c.condition_id}: {c.before} images saved, {c.after} now"
         for c in rec.counts
     ]
+    lines += [
+        f"headers changed: {_folders(u.folders)} · its image headers no longer fit the saved "
+        "channels (the run's checks say which)"
+        for u in rec.unchanged
+        if CHANNELS_CHANGED in u.reasons
+    ]
     if not lines:
         return ""
     return ui.status(

src/morphobricks_preprocessing/notebook_described.py

@@ -64,7 +64,7 @@ def _published(self) -> PublishedState | None:
         if getattr(self, "_published_key", None) != key:
             self._published_key = key
             self._published_cache, self._published_problem = published_state(
-                output, pre.discovery, pre.excluded
+                output, pre.discovery, pre.excluded, pre.metadata.files
             )
         return self._published_cache
 

src/morphobricks_preprocessing/notebook_step_states.py

@@ -9,6 +9,7 @@
 from __future__ import annotations
 
 from .channel_labels import remaining
+from .description_reconcile import CHANNELS_CHANGED
 from .mask_answers import mask_kinds, masks_unanswered
 from .notebook_stepper import StepState
 from .notebook_values import natural_key, size_rows
@@ -68,6 +69,9 @@ def _images_state(self) -> StepState:
         if state is not None and state.changed:
             rec = state.reconciliation
             text += f" · since saved: {len(rec.new)} new, {len(rec.removed)} gone"
+            headers = sum(CHANNELS_CHANGED in u.reasons for u in rec.unchanged)
+            if headers:
+                text += f", {plural(headers, 'folder')} with changed headers"
         if not pre.ok:
             text += " · problems to fix"
         return StepState(pre.ok, text)

src/morphobricks_preprocessing/notebook_ui.py

@@ -198,6 +198,7 @@ def __init__(
         self.log = _LogCapture()
         self._checked_folder = ""
         self._preview_shown = False
+        self._preview_drawn = None
         self._show_output = True
         self._show_design = True
         self._layouts_cache: tuple[tuple, dict[str, PlaneLayout]] | None = None