Source code for spacr.qt.widgets.organize_for_measure
"""Make Masks' "Organize for Measure…" popup: images and masks into merged/.
One popup puts images and their masks into the layout Measure reads -- Yokogawa-named channel folders, the masks,
and ``merged/*.npy`` -- through :mod:`spacr.channel_sorting` and
:mod:`spacr.folder_consolidation`, which hold every decision:
* a SOURCE folder (typed, browsed or dropped), optionally consolidated
first (its subfolders' images copied into one folder, named after their
folders);
* the REGEX box: Mask generation's own ``metadata_type`` field -- the same
grouped convention menu, example and "Test on my folder" -- with its
``custom_regex``, plus "Sort by regex", "Auto regex" and "Detect sets";
* the DIMENSIONS: which columns are image channels and which are masks,
each mask with its role (cell, nucleus, pathogen) and the channel whose
images it outlines;
* a TABLE with one column per channel and per mask and one row per field.
Every column takes files and folders dropped from the file manager (a
folder brings its images, recursively) and cells dragged from another
column. Dropped files sort themselves by name and are matched across the
columns the way Detect sets pairs images -- file-name tokens, then order,
never two images of different size (:func:`spacr.channel_sorting.
_match_columns`). A cell nothing matched stays empty and flagged.
Mode 1: fill the source, choose or test the regex, press "Sort by regex",
Apply. Mode 2: fill the columns by dropping, Apply. Mode 3, "Teach me…":
answer "Which channel is this?" for one image at a time while spaCR learns
a regex from the answers (:func:`spacr.channel_sorting._teach_step`), until
every image is placed. Each works without the others. A folder made by
folder consolidation is read under its files' ORIGINAL names, from its
``rename_manifest.csv``, because the copies' names may have lost what told
the channels apart. If any image cannot be copied during consolidation, the
popup reports the failed-file count and the manifest path and stops there:
the source folder, the consolidation choice and the table stay as they were,
so the problem can be fixed and the step retried. Apply builds a :class:`spacr.channel_sorting.SortPlan`; incomplete
rows block it (the regex box is where to fix them), RGB images and z-stacks
are offered for conversion, and detected sets are shown three at a time
before they are used. Nothing moves before Apply; the screen applies the
plan off the GUI thread and every move is written to the manifest.
"""
from __future__ import annotations
import json
import os
import sys
from collections import OrderedDict
from dataclasses import dataclass
from typing import Dict, Iterable, List, Optional, Sequence
import numpy as np
from PySide6.QtCore import (QAbstractTableModel, QByteArray,
QMimeData, QModelIndex, QRect, QSize, Qt, Signal,
QTimer, QUrl)
from PySide6.QtGui import (QBrush, QColor, QImage, QPainter, QPalette, QPen,
QPixmap)
from PySide6.QtWidgets import (
QAbstractItemView, QApplication, QCheckBox, QComboBox,
QDialog, QDialogButtonBox, QFileDialog, QFrame, QHBoxLayout, QHeaderView, QLabel, QLineEdit,
QMessageBox, QPushButton, QStyle, QStyledItemDelegate, QTableView, QTableWidget,
QToolTip, QVBoxLayout,
)
from shiboken6 import isValid as shiboken_valid
from ... import channel_sorting as cs
from ..hidpi import scaled_for
from ..i18n import tr
from .colour_picker import pick_colour
#: The drag format of cells dragged between the table's columns: JSON
#: ``{"column": int, "paths": [str]}``.
_CELLS_MIME = "application/x-spacr-organize-cells"
#: Subfolders consolidation leaves out: masks, raw originals, earlier sorts.
_NOT_CONSOLIDATED = ("masks", "orig", cs.DEFAULT_DEST_NAME)
#: The convention the regex box starts on: spaCR proposes a regex.
_DEFAULT_METADATA_TYPE = "auto"
#: How long a refreshed table waits before it asks for thumbnails: one
#: frame and a paint, so the reading threads start after the new rows are
#: on screen rather than while the GUI thread lays them out and draws them.
_THUMBS_AFTER_REFRESH_MS = 60
#: How long rows a sort installed wait before the table draws them: one
#: frame, so installing and drawing are two GUI-thread steps with the event
#: loop free between them, not one long one.
_DRAW_AFTER_INSTALL_MS = 16
#: The popup's modes: Make Masks' Measure layout, Mask Generation's images
#: only, and Import's images and masks with either layout.
_MODES = ("measure", "images", "import")
def _flatten_for_masks(plan: cs.SortPlan) -> cs.SortPlan:
"""Retarget ``plan`` to one folder of images, the input Mask Generation reads.
Each image goes to ``<dest>/<Yokogawa name>`` instead of its channel
folder, and its mask and curation ledger to ``<dest>/masks/``, where
Make Masks looks for an image's mask. The Yokogawa names carry the
channel, so no two images share a name.
:param plan: a plan from :func:`spacr.channel_sorting.build_plan`.
:returns: the same plan, retargeted.
"""
for row in plan.rows:
row.target_image = os.path.join(plan.dest,
os.path.basename(row.target_image))
if row.target_mask:
row.target_mask = os.path.join(plan.dest, "masks",
os.path.basename(row.target_mask))
if row.target_ledger:
row.target_ledger = row.target_mask + ".curation.json"
return plan
#: The most planes a multi-channel image may have for each to be a channel.
_MAX_SPLIT_CHANNELS = 8
def _split_channel_file(path: str, folder: str, base: str = "") -> List[str]:
"""Write each channel of a multi-channel image as its own 2-D TIFF.
The channel axis is the image's shortest axis, first or last, of at most
:data:`_MAX_SPLIT_CHANNELS` planes; ``(H, W, 3)`` RGB images split into
their three colours. The copies are named after the file's path below
``base`` with ``_ch1``, ``_ch2``... appended; the original is not touched.
:param path: the image.
:param folder: where the planes are written.
:param base: the folder whose relative path names the planes.
:returns: the plane files in channel order, or ``[]`` when the image is
not one 3-D multi-channel array or cannot be read.
"""
from ...tiff_io import write_tiff
try:
if path.lower().endswith((".tif", ".tiff")):
import tifffile
array = tifffile.imread(path)
else:
from PIL import Image
with Image.open(path) as image:
array = np.asarray(image)
except Exception:
return []
array = np.squeeze(np.asarray(array))
if array.ndim != 3:
return []
first, last = array.shape[0], array.shape[-1]
if last <= _MAX_SPLIT_CHANNELS and last <= first:
axis = 2
elif first <= _MAX_SPLIT_CHANNELS:
axis = 0
else:
return []
relative = os.path.relpath(path, base) if base else os.path.basename(path)
stem = cs.split_extension(relative)[0].replace(os.sep, "_")
os.makedirs(folder, exist_ok=True)
planes = []
for index in range(array.shape[axis]):
target = os.path.join(folder, f"{stem}_ch{index + 1}.tif")
write_tiff(target, np.ascontiguousarray(np.take(array, index, axis)))
planes.append(target)
return planes
def _apply_organized(plan: cs.SortPlan, target_layout: str = "measure",
log=None) -> cs.ApplyResult:
"""Move the files as planned into ``target_layout``.
:param plan: an ``ok`` plan.
:param target_layout: ``"measure"`` (channel folders, ``stack/`` and
``merged/``) or ``"mask"`` (:func:`_flatten_for_masks`, nothing
merged).
:param log: progress lines; default prints.
:returns: the :class:`spacr.channel_sorting.ApplyResult`.
"""
if target_layout == "mask":
return cs.apply_plan(_flatten_for_masks(plan), merge=False, log=log)
return cs.apply_plan(plan, log=log)
def _mask_generation_settings(plan: cs.SortPlan) -> Dict[str, object]:
"""The Mask Generation settings that read a folder written for it.
:param plan: the applied plan.
:returns: ``src``, the ``cellvoyager`` naming and one zero-based index
per channel.
"""
return {"src": plan.dest, "metadata_type": "cellvoyager",
"channels": list(range(len(plan.channels)))}
def _dialog_for(mode: str):
"""The popup class for ``mode``: Make Masks' own, or one of its modes.
:param mode: one of :data:`_MODES`.
:returns: a subclass of :class:`OrganizeForMeasureDialog` whose
``_mode`` is ``mode``.
:raises ValueError: for an unknown mode.
"""
if mode not in _MODES:
raise ValueError(f"mode must be one of {_MODES}, not {mode!r}")
if mode == "measure":
return OrganizeForMeasureDialog
return type(f"_Organize{mode.title()}Dialog", (OrganizeForMeasureDialog,),
{"_mode": mode, "__doc__": OrganizeForMeasureDialog.__doc__})
def _open_and_organize(host, mode: str, source: str = "", done=None):
"""Open the popup in ``mode`` and, once accepted, apply it off the GUI thread.
:param host: the screen opening it; it keeps the popup as
``_organize_dialog`` and the job as ``_organize_job``.
:param mode: ``"images"`` or ``"import"``.
:param source: the source folder to prefill.
:param done: called on the GUI thread as ``done(error, result, plan,
layout)`` when the move ends.
:returns: the popup.
"""
dialog = _dialog_for(mode)(source, host)
host._organize_dialog = dialog
if _headless() or dialog.exec() != QDialog.Accepted or dialog.plan is None:
return dialog
_start_organized(host, dialog.plan, dialog._target_layout, done)
return dialog
def _start_organized(host, plan: cs.SortPlan, target_layout: str, done=None):
"""Apply ``plan`` on a worker thread and report to ``done``.
:param host: the parent of the worker, which keeps it as
``_organize_job``.
:param plan: an ``ok`` plan.
:param target_layout: ``"measure"`` or ``"mask"``.
:param done: ``done(error, result, plan, layout)``, or None.
:returns: the started worker.
"""
import logging
from ..screens.make_masks import _FolderJobWorker
log = logging.getLogger(__name__).info
worker = _FolderJobWorker(
lambda: _apply_organized(plan, target_layout, log=log), host)
if done is not None:
worker.finished.connect(lambda: done(worker.error, worker.result,
plan, target_layout))
host._organize_job = worker
worker.start()
return worker
class _ConsolidationFailed(RuntimeError):
"""A reported partial copy that must not replace the current working table."""
def _headless() -> bool:
"""Whether no one can answer a modal box (offscreen/minimal platform)."""
from PySide6.QtGui import QGuiApplication
return QGuiApplication.platformName() in ("offscreen", "minimal")
def _mime_paths(mime: Optional[QMimeData]) -> List[str]:
"""The local files and folders a drag from the file manager carries.
:param mime: the drag's data.
:returns: absolute paths, in drag order.
"""
if mime is None or not mime.hasUrls():
return []
return [os.path.abspath(url.toLocalFile()) for url in mime.urls()
if url.isLocalFile() and url.toLocalFile()]
def _mime_cells(mime: Optional[QMimeData]):
"""The column and files a drag out of the table carries.
:param mime: the drag's data.
:returns: ``(column, [paths])``, or ``(None, [])`` for any other drag.
"""
if mime is None or not mime.hasFormat(_CELLS_MIME):
return None, []
try:
data = json.loads(bytes(mime.data(_CELLS_MIME)).decode("utf-8"))
return int(data["column"]), [str(p) for p in data["paths"]]
except (ValueError, KeyError, TypeError, UnicodeDecodeError):
return None, []
def _mime_slots(mime: Optional[QMimeData]) -> List[List[int]]:
"""The table slots a drag out of the table carries.
:param mime: the drag's data.
:returns: ``[[row, column], ...]``, empty for any other drag.
"""
if mime is None or not mime.hasFormat(_CELLS_MIME):
return []
try:
data = json.loads(bytes(mime.data(_CELLS_MIME)).decode("utf-8"))
return [[int(r), int(c)] for r, c in data.get("cells", [])]
except (ValueError, TypeError, AttributeError, UnicodeDecodeError):
return []
def _mime_anchor(mime: Optional[QMimeData]) -> Optional[List[int]]:
"""The slot a drag out of the table started on.
:param mime: the drag's data.
:returns: ``[row, column]``, or None for any other drag.
"""
if mime is None or not mime.hasFormat(_CELLS_MIME):
return None
try:
data = json.loads(bytes(mime.data(_CELLS_MIME)).decode("utf-8"))
anchor = data.get("anchor")
return [int(anchor[0]), int(anchor[1])] if anchor else None
except (ValueError, TypeError, AttributeError, IndexError, KeyError,
UnicodeDecodeError):
return None
def _images_in(paths: Iterable[str]):
"""Expand dropped files and folders into image files.
:param paths: files and folders.
:returns: ``([image paths], [names left out])``.
"""
from ... import drop_classification as dc
found: List[str] = []
skipped: List[str] = []
for path in paths:
path = os.path.abspath(os.fspath(path))
if os.path.isdir(path):
found.extend(dc._folder_images(path))
elif os.path.isfile(path) and path.lower().endswith(cs.IMAGE_EXTS):
found.append(path)
else:
skipped.append(os.path.basename(path) or path)
return list(dict.fromkeys(found)), skipped
def _alias_in(cache: Dict[str, Dict[str, str]], path: str) -> str:
"""The name the regex reads for ``path``, through a manifest cache.
:param cache: ``{folder: {copy name: original name}}``, filled as
folders are first met (:func:`spacr.channel_sorting._original_names`).
:param path: an image path.
:returns: the path under its folder with its original file name.
"""
folder = os.path.dirname(path)
if folder not in cache:
cache[folder] = cs._original_names(folder)
name = os.path.basename(path)
return os.path.join(folder, cache[folder].get(name, name))
def _regex_for_names(key: str, custom: str, names: Sequence[str]) -> str:
"""The regex a ``metadata_type`` and a custom regex stand for.
:param key: the convention; ``auto`` and ``custom`` are spaCR's own.
:param custom: the custom regex as typed, stripped.
:param names: file names or paths.
:returns: the regex, or ``""`` when there is none.
"""
from ... import regex_infer
if key == "auto":
return custom or cs.infer_regex(list(names)) or ""
if key == "custom":
return custom
extensions = sorted({os.path.splitext(n)[1].lstrip(".") or "tif"
for n in names}) or ["tif"]
try:
if len(extensions) == 1:
return regex_infer._metadata_pattern(key, extensions[0],
custom or None)
return regex_infer._metadata_pattern_any_extension(key, extensions)
except KeyError:
return ""
def _files_in(folder: str):
"""The names of the files in ``folder``, from one directory read.
:param folder: a folder.
:returns: a set of names; an empty one when the folder is missing; None
where file names ignore case, so callers ask the disk per name.
"""
if os.path.normcase("A") != "A" or sys.platform == "darwin":
return None
try:
with os.scandir(folder) as entries:
return {entry.name for entry in entries if entry.is_file()}
except OSError:
return set()
def _gather_sort_paths(source: str, masks_dir: Optional[str],
channel_files, mask_pairs, skipped):
"""The images a sort reads, their channels and their masks.
Reads the disk only, so it runs on a worker thread.
:param source: the source folder, or ``""`` to read the columns.
:param masks_dir: the source's masks folder, when not its ``masks/``.
:param channel_files: ``[(files, channel), ...]``, one per channel
column, read when there is no source.
:param mask_pairs: ``[(channel column files, mask column files), ...]``.
:param skipped: paths "Teach me" was told to skip.
:returns: ``(paths, {path: channel}, {image: mask})``.
"""
channels: Dict[str, int] = {}
masks: Dict[str, str] = {}
if source:
names = cs.list_folder_images(source)
paths = [os.path.join(source, n) for n in names]
saved = _files_in(cs.masks_folder(source, masks_dir))
for name, path in zip(names, paths):
stem = cs.split_extension(name)[0] + ".tif"
if saved is None:
mask = cs.mask_for(source, name, masks_dir)
else:
mask = (os.path.join(cs.masks_folder(source, masks_dir), stem)
if stem in saved else None)
if mask:
masks[path] = mask
else:
paths = []
for files, channel in channel_files:
for path in files:
paths.append(path)
channels[path] = channel
for images, mask_files in mask_pairs:
pairs = cs._match_columns([list(images), list(mask_files)])
masks.update({row[0]: row[1] for row in pairs
if row[0] and row[1]})
paths = [p for p in paths if p not in skipped]
return paths, channels, masks
def _regex_layout(paths: Sequence[str], channels: Dict[str, int],
masks: Dict[str, str], pattern: Optional[str], chan_map,
key: str, custom: str,
aliases: Dict[str, Dict[str, str]]) -> Dict[str, object]:
"""The table a regex makes of these images: columns, rows and report.
Pure apart from reading rename manifests into ``aliases``, so it runs
on a worker thread; :meth:`OrganizeForMeasureDialog.sort_by_regex`
describes the layout.
:param paths: the images.
:param channels: ``{path: channel}`` for images already in a column.
:param masks: ``{image: mask}`` found beside them.
:param pattern: the regex, or None for what ``key`` and ``custom`` give.
:param chan_map: ``{chanID value: label}`` from "Teach me", or None.
:param key: the regex box's convention.
:param custom: the custom regex as typed, stripped.
:param aliases: the rename-manifest cache, filled in place.
:returns: ``{"empty": True}`` when there are no images, else the
``columns``, ``rows``, ``row_keys``, set ``report`` and the masks
two images ``shared``.
"""
if not paths:
return {"empty": True}
alias_of = {p: _alias_in(aliases, p) for p in paths}
back = {alias_of[p]: p for p in paths}
if pattern is None:
pattern = _regex_for_names(key, custom, [alias_of[p] for p in paths])
parsed = cs.parse_names(
[alias_of[p] for p in paths], pattern,
{_alias_in(aliases, p): c for p, c in channels.items()} or None)
name_masks: Dict[object, list] = {}
if chan_map is not None:
compiled = cs.compile_regex(pattern)[0]
for item in parsed:
item.channel = None
if not item.matched or compiled is None:
continue
label = chan_map.get(item.groups.get(cs.CHANNEL_GROUP, ""))
if isinstance(label, tuple) and label[0] == "channel":
item.channel = int(label[1])
elif isinstance(label, tuple) and label[0] == "mask":
name_masks.setdefault(item.set_key, []).append(
(int(label[1]), label[2] if len(label) > 2 else "",
back[item.name]))
item.matched = False
taken_as_masks = {m for entries in name_masks.values()
for _c, _r, m in entries}
report = cs.check_sets([p for p in parsed
if back[p.name] not in taken_as_masks])
chosen: Dict[cs.SetKey, Dict[int, str]] = {}
for item in parsed:
if item.matched and item.channel is not None:
chosen.setdefault(item.set_key, {}).setdefault(
item.channel, back[item.name])
claimed: Dict[str, List[str]] = {}
for image, mask in masks.items():
claimed.setdefault(mask, []).append(image)
shared = sorted(m for m, images in claimed.items() if len(images) > 1)
masks = {i: m for i, m in masks.items() if m not in shared}
present = sorted({c for m in chosen.values() for c in m})
position = {c: i + 1 for i, c in enumerate(present)}
explicit = {}
for set_key, entries in name_masks.items():
for channel, role, mask in entries:
image = chosen.get(set_key, {}).get(channel)
if image is not None:
masks[image] = mask
if role:
explicit[channel] = role
roles = cs.default_mask_roles(
{c: [m[c] for m in chosen.values() if c in m] for c in present},
sorted({c for m in chosen.values() for c, p in m.items()
if p in masks}))
roles.update({c: r for c, r in explicit.items() if c in roles})
columns = [_Column("channel") for _c in present]
for channel in sorted(roles):
columns.append(_Column("mask", roles[channel], position[channel]))
rows: List[List[Optional[str]]] = []
row_keys: List[Optional[cs.SetKey]] = []
for set_key in sorted(chosen,
key=lambda k: cs.natural_key(cs.set_label(k))):
row: List[Optional[str]] = [None] * len(columns)
for index, channel in enumerate(present):
row[index] = chosen[set_key].get(channel)
for index, column in enumerate(columns):
if column.kind == "mask":
image = chosen[set_key].get(present[column.of_channel - 1])
row[index] = masks.get(image) if image else None
rows.append(row)
row_keys.append(set_key)
return {"columns": columns, "rows": rows, "row_keys": row_keys,
"report": report, "shared": shared}
def _name_shapes(rows, width: int, digits: bool = True) -> List[List[int]]:
"""Per column, one row for each shape of file name it holds.
A name's shape is the name with its digits all made ``0``: in a font
whose digits are one width, names of one shape are one width, so
measuring one of each finds the widest. Pure, so a sort worker works it
out beside the layout.
:param rows: one list per row, one path or None per column.
:param width: how many columns.
:param digits: fold digits; False keeps every distinct name.
:returns: ``[[row, ...], ...]``, one list per column.
"""
fold = str.maketrans("123456789", "000000000") if digits else None
out: List[List[int]] = []
for column in range(width):
shapes: Dict[str, int] = {}
for index, row in enumerate(rows):
path = row[column] if column < len(row) else None
name = os.path.basename(path) if path else "\u2014"
shapes.setdefault(name.translate(fold) if fold else name, index)
out.append(list(shapes.values()))
return out
def _sort_job(source: str, masks_dir: Optional[str], channel_files,
mask_pairs, skipped, pattern: Optional[str], chan_map,
key: str, custom: str,
aliases: Dict[str, Dict[str, str]]) -> Dict[str, object]:
"""Read the images and lay them out by the regex, off the GUI thread.
:func:`_gather_sort_paths`, then :func:`_regex_layout`, with the same
arguments; ``aliases`` is a copy the caller merges back.
:returns: :func:`_regex_layout`'s result plus the ``aliases`` read.
"""
paths, channels, masks = _gather_sort_paths(
source, masks_dir, channel_files, mask_pairs, skipped)
result = _regex_layout(paths, channels, masks, pattern, chan_map, key,
custom, aliases)
result["aliases"] = aliases
if "rows" in result:
result["shapes"] = _name_shapes(result["rows"],
len(result["columns"]))
return result
@dataclass
class _Column:
"""One table column: an image channel or a mask.
:ivar kind: ``"channel"`` or ``"mask"``.
:ivar role: a mask's role (cell, nucleus, pathogen); unused for a channel.
:ivar of_channel: a mask's channel -- the 1-based channel whose images
it outlines; unused for a channel.
"""
kind: str
role: str = ""
of_channel: int = 1
class _PathEdit(QLineEdit):
"""The source field, which also takes a folder dropped on it."""
def __init__(self, parent=None):
"""A line edit that accepts drops.
:param parent: the Qt parent.
"""
super().__init__(parent)
self.setAcceptDrops(True)
def dragEnterEvent(self, event) -> None:
"""Accept a drag holding a folder.
:param event: the drag event.
"""
if any(os.path.isdir(p) for p in _mime_paths(event.mimeData())):
event.acceptProposedAction()
else:
event.ignore()
def dragMoveEvent(self, event) -> None:
"""Keep accepting while the drag moves over the field.
:param event: the drag event.
"""
self.dragEnterEvent(event)
def dropEvent(self, event) -> None:
"""Put the first dropped folder in the field.
:param event: the drop event.
"""
folders = [p for p in _mime_paths(event.mimeData()) if os.path.isdir(p)]
if not folders:
event.ignore()
return
event.acceptProposedAction()
self.setText(folders[0])
class _VolumePathEdit(QLineEdit):
"""A volume input accepting one local file by drop or pasted path."""
def __init__(self, parent=None):
"""Build an empty path field that accepts dropped files.
:param parent: optional Qt parent widget.
"""
super().__init__(parent)
self.setAcceptDrops(True)
self.setPlaceholderText(tr("Drop a file or paste its path"))
def dragEnterEvent(self, event):
"""Accept a drag of exactly one local file onto an editable field.
:param event: the Qt drag-enter event.
"""
paths = _mime_paths(event.mimeData())
if not self.isReadOnly() and len(paths) == 1 and os.path.isfile(paths[0]):
event.acceptProposedAction()
else:
event.ignore()
def dragMoveEvent(self, event):
"""Keep accepting or refusing the drag as it moves over the field.
:param event: the Qt drag-move event.
"""
self.dragEnterEvent(event)
def dropEvent(self, event):
"""Put a dropped local file's path into an editable field.
:param event: the Qt drop event.
"""
paths = _mime_paths(event.mimeData())
if not self.isReadOnly() and len(paths) == 1 and os.path.isfile(paths[0]):
self.setText(paths[0])
event.acceptProposedAction()
else:
event.ignore()
def path(self):
"""Normalize a pasted path or local file URL without loading its data."""
value = self.text().strip().strip("\"'")
if not value:
return ""
if value.startswith("file:"):
value = QUrl(value).toLocalFile()
return os.path.abspath(os.path.expanduser(value)) if value else ""
[docs]
class MeasureVolumeDialog(QDialog):
"""Choose extra channels and image sets for the current edited 3D mask.
:param source_path: path of the current 3D image, fixed in the first row.
:param parent: optional Qt parent widget.
:param spacing: optional voxel distances displayed in axis order.
:param axes: spatial axis order used to label the displayed spacing.
:param unit: unit displayed beside the voxel spacing.
"""
def __init__(self, source_path, parent=None, *, spacing=None, axes="ZYX", unit="pixel"):
"""Build the dialog with the current image fixed in its first row."""
super().__init__(parent)
self.setObjectName("MakeMasksMeasureVolumeDialog")
self.setWindowTitle(tr("Measure 3D images"))
self.resize(960, 440)
layout = QVBoxLayout(self)
description = QLabel(tr(
"The first image uses its current edited labels and all loaded channels. "
"Add matching 3D channel files or more image sets below. Leave the mask "
"field blank to use the image's saved mask. Measurements are written to measurements.db "
"in each image folder; existing measurements for these sets are updated."), self)
description.setWordWrap(True)
layout.addWidget(description)
if spacing is not None:
calibration = QLabel(tr(
"Current image voxel spacing ({axes}): {spacing} {unit}.",
axes=axes, spacing=" × ".join(f"{value:g}" for value in spacing), unit=unit), self)
calibration.setObjectName("VolumeMeasureCalibration")
calibration.setWordWrap(True)
layout.addWidget(calibration)
role_row = QHBoxLayout()
role_label = QLabel(tr("Object type"), self)
self.object_type = QComboBox(self)
self.object_type.setObjectName("VolumeMeasureObjectType")
for label, value in ((tr("Cell"), "cell"), (tr("Nucleus"), "nucleus"),
(tr("Pathogen"), "pathogen"), (tr("Organelle"), "organelle")):
self.object_type.addItem(label, value)
role_label.setBuddy(self.object_type)
role_row.addWidget(role_label)
role_row.addWidget(self.object_type)
role_row.addStretch()
layout.addLayout(role_row)
self.table = QTableWidget(0, 4, self)
self.table.setObjectName("VolumeMeasureImageSets")
self.table.horizontalHeader().setSectionResizeMode(QHeaderView.Stretch)
self.table.setSelectionBehavior(QAbstractItemView.SelectRows)
self.table.setSelectionMode(QAbstractItemView.SingleSelection)
self._set_headers()
layout.addWidget(self.table, 1)
controls = QHBoxLayout()
self.add_row_button = QPushButton(tr("Add image set"), self)
self.add_row_button.setObjectName("VolumeMeasureAddImageSet")
self.add_row_button.clicked.connect(lambda: self.add_row())
self.add_channel_button = QPushButton(tr("Add channel"), self)
self.add_channel_button.setObjectName("VolumeMeasureAddChannel")
self.add_channel_button.clicked.connect(self.add_channel)
self.remove_row_button = QPushButton(tr("Remove image set"), self)
self.remove_row_button.clicked.connect(self.remove_row)
for button in (self.add_row_button, self.add_channel_button, self.remove_row_button):
controls.addWidget(button)
controls.addStretch()
layout.addLayout(controls)
self.error = QLabel(self)
self.error.setWordWrap(True)
self.error.setObjectName("VolumeMeasureInputError")
layout.addWidget(self.error)
self.buttons = QDialogButtonBox(QDialogButtonBox.Save | QDialogButtonBox.Cancel, self)
self.buttons.button(QDialogButtonBox.Save).setText(tr("Save measurements"))
self.buttons.accepted.connect(self.accept)
self.buttons.rejected.connect(self.reject)
layout.addWidget(self.buttons)
self.add_row()
self.table.cellWidget(0, 0).setText(str(source_path))
self.table.cellWidget(0, 0).setReadOnly(True)
self.table.cellWidget(0, 1).setText(tr("Current edited labels"))
self.table.cellWidget(0, 1).setReadOnly(True)
def _set_headers(self):
"""Label the image, mask and numbered extra-channel columns."""
self.table.setHorizontalHeaderLabels(
[tr("3D image"), tr("Saved mask (optional)")]
+ [tr("Extra channel {n}", n=index + 1)
for index in range(self.table.columnCount() - 2)])
def _set_path_cell(self, row, column):
"""Place a path field, named for its row and column, in one cell.
:param row: the image-set row.
:param column: the column the field fills.
"""
field = _VolumePathEdit(self.table)
field.setAccessibleName(tr("Image set {row}, {column}", row=row + 1,
column=self.table.horizontalHeaderItem(column).text()))
self.table.setCellWidget(row, column, field)
[docs]
def add_row(self):
"""Append an independently measured image set."""
row = self.table.rowCount()
self.table.insertRow(row)
for column in range(self.table.columnCount()):
self._set_path_cell(row, column)
self.table.setCurrentCell(row, 0)
[docs]
def add_channel(self):
"""Add an optional channel slot to every image set."""
column = self.table.columnCount()
self.table.insertColumn(column)
self._set_headers()
for row in range(self.table.rowCount()):
self._set_path_cell(row, column)
[docs]
def remove_row(self):
"""Remove the selected extra image set, preserving the current image row."""
row = self.table.currentRow()
if row > 0:
self.table.removeRow(row)
[docs]
def requests(self):
"""Validate local paths and preserve each row's channel ordering."""
rows = []
for row in range(self.table.rowCount()):
source = self.table.cellWidget(row, 0).path()
mask = self.table.cellWidget(row, 1).path() if row else ""
channels = [self.table.cellWidget(row, column).path()
for column in range(2, self.table.columnCount())]
if row and not source and not mask and not any(channels):
continue
if not source:
raise ValueError(tr("Choose a 3D image for image set {n}.", n=row + 1))
while channels and not channels[-1]:
channels.pop()
if any(not path for path in channels):
raise ValueError(tr("Fill extra channels from left to right in image set {n}.", n=row + 1))
for path in (source, mask, *channels):
if path and not os.path.isfile(path):
raise ValueError(tr("File not found: {path}", path=path))
rows.append({"source_path": source, "mask_path": mask or None,
"channel_paths": tuple(channels)})
return rows
[docs]
def accept(self):
"""Accept only valid image-set requests; display validation errors without closing."""
try:
self.requests()
except ValueError as exc:
self.error.setText(str(exc))
return
self.error.clear()
super().accept()
#: How the table shows a cell: its name, its thumbnail, or both.
_VIEWS = ("text", "image", "both")
#: The thumbnail side in the table's image views, in pixels.
_CELL_THUMB = 96
#: The thumbnail-size slider's range, in pixels.
_THUMB_RANGE = (32, 320)
#: The side of the × that clears a cell, in pixels.
_CLOSE_SIZE = 14
def _swap_cells(rows: List[List[Optional[str]]], mask_of: Dict[int, int],
source, target) -> List[List[Optional[str]]]:
"""Swap two table slots; the images' masks move with them.
Dropping on an empty slot is the same swap with nothing. A target row
one past the last makes a new row. When both slots are in channel
columns that have a mask column, the masks in those rows swap too, so
an image never leaves its mask behind.
:param rows: the table, one list per row, changed in place.
:param mask_of: ``{channel column: its mask column}``.
:param source: ``(row, column)`` dragged.
:param target: ``(row, column)`` dropped on.
:returns: ``rows``.
"""
(r1, c1), (r2, c2) = source, target
width = len(rows[0]) if rows else max(c1, c2) + 1
while r2 >= len(rows):
rows.append([None] * width)
rows[r1][c1], rows[r2][c2] = rows[r2][c2], rows[r1][c1]
if c1 in mask_of and c2 in mask_of and (r1, c1) != (r2, c2):
m1, m2 = mask_of[c1], mask_of[c2]
rows[r1][m1], rows[r2][m2] = rows[r2][m2], rows[r1][m1]
return rows
def _clear_cell(rows: List[List[Optional[str]]], mask_of: Dict[int, int],
cell) -> Optional[str]:
"""Empty one slot; an image's mask leaves the table with it.
:param rows: the table, changed in place.
:param mask_of: ``{channel column: its mask column}``.
:param cell: ``(row, column)``.
:returns: the file that was there, or None.
"""
row, column = cell
if not 0 <= row < len(rows):
return None
removed, rows[row][column] = rows[row][column], None
if removed and column in mask_of:
rows[row][mask_of[column]] = None
return removed
def _block_moves(cells, anchor, target, width: int,
mask_of: Dict[int, int]) -> Dict[tuple, tuple]:
"""Where each cell of a dragged block lands.
The block keeps its shape: every cell moves by the offset from the cell
the drag started on (``anchor``) to the slot it was dropped on. A cell
in a channel column with a mask column takes its mask along, into the
target channel's mask column, unless that mask slot is already spoken
for. A block that would leave the table sideways or above row 0 does
not move at all.
:param cells: ``[(row, column), ...]`` the dragged cells.
:param anchor: ``(row, column)`` the cell the drag started on.
:param target: ``(row, column)`` the slot it was dropped on.
:param width: the number of columns.
:param mask_of: ``{channel column: its mask column}``.
:returns: ``{source slot: target slot}``, empty when the block cannot
move.
"""
dr, dc = target[0] - anchor[0], target[1] - anchor[1]
moves: Dict[tuple, tuple] = {}
for row, column in cells:
to = (row + dr, column + dc)
if to[0] < 0 or not 0 <= to[1] < width:
return {}
moves[(row, column)] = to
taken = set(moves.values())
for (row, column), (to_row, to_column) in list(moves.items()):
if column in mask_of and to_column in mask_of:
source = (row, mask_of[column])
to = (to_row, mask_of[to_column])
if source not in moves and to not in taken and source != to:
moves[source] = to
taken.add(to)
return {s: t for s, t in moves.items() if s != t}
def _move_block(rows: List[List[Optional[str]]], moves: Dict[tuple, tuple]
) -> List[List[Optional[str]]]:
"""Move several slots at once; what they land on swaps back.
Every source's file goes to its target. A file already on a target that
is not itself moving goes to the slot freed at the start of that chain
of moves -- for a block dropped clear of itself that is exactly a
pairwise swap, and for a block shifted onto part of itself nothing is
lost. Rows past the end are added.
:param rows: the table, changed in place.
:param moves: ``{(row, column): (row, column)}`` from :func:`_block_moves`.
:returns: ``rows``.
"""
if not moves:
return rows
width = len(rows[0]) if rows else 1 + max(
c for pair in moves.items() for _r, c in pair)
last = max(r for pair in moves.items() for r, _c in pair)
while last >= len(rows):
rows.append([None] * width)
before = {slot: rows[slot[0]][slot[1]]
for pair in moves.items() for slot in pair}
back = {t: s for s, t in moves.items()}
for source in moves:
rows[source[0]][source[1]] = None
for source, to in moves.items():
rows[to[0]][to[1]] = before[source]
for to in moves.values():
if to in moves or before[to] is None:
continue
free = back[to]
while free in back:
free = back[free]
rows[free[0]][free[1]] = before[to]
return rows
def _selection_cells(indexes) -> List[List[int]]:
"""The filled cells among selected indexes, in reading order.
:param indexes: model indexes (or anything with ``row``, ``column`` and
``data``).
:returns: ``[[row, column], ...]`` of the cells holding a file.
"""
return sorted([index.row(), index.column()] for index in indexes
if index.data(Qt.UserRole))
def _close_rect(cell_rect):
"""Where a cell's × sits: its top-right corner.
:param cell_rect: the cell's QRect.
:returns: the × QRect.
"""
from PySide6.QtCore import QRect
return QRect(cell_rect.right() - _CLOSE_SIZE - 3, cell_rect.top() + 3,
_CLOSE_SIZE, _CLOSE_SIZE)
#: The ×'s outline and stroke pens, plain and hovered, made once.
_CLOSE_PENS: Dict[bool, tuple] = {}
def _close_pens(hovered: bool) -> tuple:
"""The ×'s ``(outline, stroke)`` pens: white, red when hovered.
:param hovered: whether the mouse is over the ×.
"""
hovered = bool(hovered)
pens = _CLOSE_PENS.get(hovered)
if pens is None:
pens = _CLOSE_PENS[hovered] = (
QPen(QColor(0, 0, 0, 170), 4, Qt.SolidLine, Qt.RoundCap),
QPen(QColor(220, 40, 40) if hovered else QColor("white"),
2, Qt.SolidLine, Qt.RoundCap))
return pens
class _CellDelegate(QStyledItemDelegate):
"""Draws a table cell as text, thumbnail or both, with its ×.
The × is painted, not a widget per cell: white with a dark outline, red
while the mouse is over it (the table tracks the mouse and the click).
"""
def __init__(self, table, parent=None):
"""Remember the table whose view, colour and thumbnails it reads.
:param table: the :class:`_OrganizeTable`.
:param parent: the Qt parent.
"""
super().__init__(parent)
self._table = table
self.view = "text"
self.text_color = QColor("white")
self.pixmaps: Dict[str, QPixmap] = OrderedDict()
#: The (row, column) whose × the mouse is over, or None.
self.hover = None
def paint(self, painter, option, index) -> None:
"""Draw the cell in the chosen view; the table draws the ×s after.
:param painter: the painter.
:param option: the style option.
:param index: the cell.
"""
model = index.model()
if not isinstance(model, _CellModel):
super().paint(painter, option, index)
return
path = model.path(index.row(), index.column())
pixmap = (self.pixmaps.get(path) if path and self.view != "text"
else None)
if pixmap is None:
self._paint_text(painter, option, index, model, path)
else:
painter.save()
if option.state & QStyle.State_Selected:
painter.fillRect(option.rect, option.palette.highlight())
scaled = scaled_for(pixmap, self._table,
option.rect.size() - QSize(4, 4))
shown = scaled.deviceIndependentSize()
x = option.rect.x() + int(option.rect.width() - shown.width()) // 2
y = option.rect.y() + int(option.rect.height() - shown.height()) // 2
painter.drawPixmap(x, y, scaled)
if self.view == "both":
text = os.path.basename(path)
box = QRect(x, y, int(shown.width()),
int(shown.height())).adjusted(3, 3, -3, -3)
painter.setPen(QColor(0, 0, 0, 200))
painter.drawText(box.translated(1, 1),
Qt.AlignBottom | Qt.AlignLeft | Qt.TextWordWrap,
text)
painter.setPen(self.text_color)
painter.drawText(box, Qt.AlignBottom | Qt.AlignLeft
| Qt.TextWordWrap, text)
painter.restore()
def _paint_text(self, painter, option, index, model, path) -> None:
"""Draw a cell's name as the plain delegate would, without asking.
``initStyleOption`` asks the model for nine roles a cell, each a
call into Python; this cell's name and flag colour are known, so
the option is filled from them and the style draws it as usual.
:param painter: the painter.
:param option: the style option.
:param index: the cell.
:param model: its :class:`_CellModel`.
:param path: its file, or None.
"""
from PySide6.QtWidgets import QStyleOptionViewItem
cell = QStyleOptionViewItem(option)
cell.index = index
cell.features |= QStyleOptionViewItem.HasDisplay
cell.text = os.path.basename(path) if path else "\u2014"
if not path and index.column() in model.flagged:
cell.backgroundBrush = model.flag
widget = option.widget
style = widget.style() if widget is not None else QApplication.style()
style.drawControl(QStyle.CE_ItemViewItem, cell, painter, widget)
@staticmethod
def _paint_close(painter, rect, hovered: bool) -> None:
"""Draw the × in ``rect``: white, red when hovered.
:param painter: the painter.
:param rect: where.
:param hovered: whether the mouse is over it.
"""
painter.save()
painter.setRenderHint(QPainter.Antialiasing, True)
inset = rect.adjusted(3, 3, -3, -3)
for pen in _close_pens(hovered):
painter.setPen(pen)
painter.drawLine(inset.topLeft(), inset.bottomRight())
painter.drawLine(inset.topRight(), inset.bottomLeft())
painter.restore()
def helpEvent(self, event, view, option, index) -> bool:
"""The ×'s tooltip.
:param event: the help event.
:param view: the view.
:param option: the style option.
:param index: the cell.
:returns: True when a tooltip was shown.
"""
if index.data(Qt.UserRole) and _close_rect(option.rect).contains(
event.pos()):
QToolTip.showText(event.globalPos(), tr(
"Remove from the table (Delete); the file is not touched."),
view)
return True
return super().helpEvent(event, view, option, index)
#: Where the table's view and overlay colour are remembered, per viewer.
_PREFS_VIEW = "organize_for_measure/view"
_PREFS_COLOR = "organize_for_measure/text_color"
def _load_view_prefs():
"""The remembered view and overlay colour.
:returns: ``(view, colour name)``; ``("text", "#ffffff")`` when nothing
is stored or the preferences cannot be read.
"""
try:
from ..prefs import _s
store = _s()
view = str(store.value(_PREFS_VIEW, "text") or "text")
color = str(store.value(_PREFS_COLOR, "#ffffff") or "#ffffff")
except Exception:
return "text", "#ffffff"
return (view if view in _VIEWS else "text"), color
def _save_view_prefs(view: str, color: str) -> None:
"""Remember the view and overlay colour for the next popup.
:param view: one of :data:`_VIEWS`.
:param color: a colour name.
"""
try:
from ..prefs import _s
store = _s()
store.setValue(_PREFS_VIEW, view)
store.setValue(_PREFS_COLOR, color)
except Exception:
pass
def _legible_backdrop(color: QColor, window: QColor) -> str:
"""A backdrop that keeps a caption in ``color`` readable on ``window``.
The "Text colour" button writes its caption in the chosen colour; white
or yellow on a light window, or black on a dark one, would vanish. Below
a 3:1 contrast ratio (WCAG relative luminance) a translucent chip of the
opposite shade goes behind it.
:param color: the caption's colour.
:param window: the colour behind the button.
:returns: a style-sheet colour, ``transparent`` when none is needed.
"""
def luminance(c: QColor) -> float:
"""The colour's relative luminance (WCAG 2), 0 for black to 1 for white."""
def linear(v: float) -> float:
"""One sRGB channel, 0..1, converted to linear light."""
return v / 12.92 if v <= 0.03928 else ((v + 0.055) / 1.055) ** 2.4
c = QColor(c)
return (0.2126 * linear(c.redF()) + 0.7152 * linear(c.greenF())
+ 0.0722 * linear(c.blueF()))
fore, back = luminance(color), luminance(window)
ratio = (max(fore, back) + 0.05) / (min(fore, back) + 0.05)
if ratio >= 3.0:
return "transparent"
if fore >= 0.18:
return "rgba(0, 0, 0, 160)"
return "rgba(255, 255, 255, 200)"
#: Where the thumbnail size is remembered.
_PREFS_THUMB = "organize_for_measure/thumb_size"
def _clamp_thumb(value) -> int:
"""A thumbnail size inside :data:`_THUMB_RANGE`.
:param value: anything ``int()`` takes; junk gives :data:`_CELL_THUMB`.
:returns: the size in pixels.
"""
try:
value = int(value)
except (TypeError, ValueError):
return _CELL_THUMB
return max(_THUMB_RANGE[0], min(_THUMB_RANGE[1], value))
def _load_thumb_pref() -> int:
"""The remembered thumbnail size, or :data:`_CELL_THUMB`."""
try:
from ..prefs import _s
return _clamp_thumb(_s().value(_PREFS_THUMB, _CELL_THUMB))
except Exception:
return _CELL_THUMB
def _save_thumb_pref(size: int) -> None:
"""Remember the thumbnail size for the next popup.
:param size: pixels.
"""
try:
from ..prefs import _s
_s().setValue(_PREFS_THUMB, _clamp_thumb(size))
except Exception:
pass
class _CellItem:
"""One cell of the table, read back the way a table-widget item was.
The table shows a fixed :class:`_CellModel`; this is a handle on one of
its cells for the code and tests that ask the table for ``item(row,
column)`` or its ``selectedItems()``.
:param table: the :class:`_OrganizeTable`.
:param row: the cell's row.
:param column: the cell's column.
"""
__slots__ = ("_table", "_row", "_column")
def __init__(self, table, row: int, column: int):
"""Point at one cell.
:param table: the table.
:param row: the row.
:param column: the column.
"""
self._table = table
self._row = row
self._column = column
def _index(self):
"""The cell's model index."""
return self._table.model().index(self._row, self._column)
def row(self) -> int:
"""The cell's row."""
return self._row
def column(self) -> int:
"""The cell's column."""
return self._column
def data(self, role: int = Qt.DisplayRole):
"""The cell's data for ``role``: its path for ``Qt.UserRole``.
:param role: the item data role.
"""
return self._index().data(role)
def text(self) -> str:
"""The text the cell shows."""
return self.data(Qt.DisplayRole)
def toolTip(self) -> str:
"""The cell's tooltip."""
return self.data(Qt.ToolTipRole)
def flags(self):
"""The cell's item flags."""
return self._table.model().flags(self._index())
def isSelected(self) -> bool:
"""Whether the cell is selected."""
return self._table.selectionModel().isSelected(self._index())
def setSelected(self, selected: bool) -> None:
"""Select or deselect the cell.
:param selected: the new state.
"""
from PySide6.QtCore import QItemSelectionModel
self._table.selectionModel().select(
self._index(), QItemSelectionModel.Select if selected
else QItemSelectionModel.Deselect)
#: The roles :class:`_CellModel` answers; a cell is asked for a dozen
#: others each time it is drawn, and those return at once.
_CELL_ROLES = frozenset(int(getattr(role, "value", role)) for role in (
Qt.DisplayRole, Qt.UserRole, Qt.ToolTipRole, Qt.BackgroundRole))
class _CellModel(QAbstractTableModel):
"""The table's cells: a fixed snapshot of the rows, swapped whole.
A model is never changed once the table shows it; a new layout builds a
new model and the table is pointed at it in one step
(:meth:`_OrganizeTable.show_cells`). Nothing is built per cell: the text,
tooltip and flag colour are worked out when a cell is drawn, so a
3,000-file sort costs the same to show as a ten-file one.
:param rows: one list per row, one path or None per column.
:param captions: the column headers.
:param flagged: the columns whose empty cells are flagged.
:param table: the table; its ``mimeData`` builds a drag.
"""
def __init__(self, rows, captions, flagged, table=None):
"""Hold the snapshot; it never changes afterwards.
:param rows: one list per row, one path or None per column.
:param captions: the column headers.
:param flagged: the columns whose empty cells are flagged.
:param table: the table, also the model's Qt parent.
"""
super().__init__(table)
self._table = table
self._rows = tuple(tuple(row) for row in rows)
self._captions = tuple(captions)
self.flagged = frozenset(flagged)
self.has_files = any(p for row in self._rows for p in row)
self._move_tip = tr("Drag it to another slot to move it (an occupied "
"slot swaps); the × or Delete removes it.")
self._empty_tip = tr("Nothing matched here")
self.flag = QBrush(QColor(200, 60, 60, 60))
self._slot_flags = (Qt.ItemIsEnabled | Qt.ItemIsSelectable
| Qt.ItemIsDropEnabled)
self._file_flags = self._slot_flags | Qt.ItemIsDragEnabled
def rowCount(self, parent=QModelIndex()) -> int:
"""How many rows; a cell has no children.
:param parent: the parent index; only the root has rows.
"""
return 0 if parent.isValid() else len(self._rows)
def columnCount(self, parent=QModelIndex()) -> int:
"""How many columns.
:param parent: the parent index; only the root has columns.
"""
return 0 if parent.isValid() else len(self._captions)
def path(self, row: int, column: int) -> Optional[str]:
"""The file in a cell, or None.
:param row: the row.
:param column: the column.
"""
if 0 <= row < len(self._rows) and 0 <= column < len(self._rows[row]):
return self._rows[row][column]
return None
def data(self, index, role=Qt.DisplayRole):
"""A cell's name, path, tooltip or flag colour.
:param index: the cell.
:param role: the item data role.
"""
if role not in _CELL_ROLES or not index.isValid():
return None
path = self.path(index.row(), index.column())
if role == Qt.DisplayRole:
return os.path.basename(path) if path else "—"
if role == Qt.UserRole:
return path
if role == Qt.ToolTipRole:
return path + "\n" + self._move_tip if path else self._empty_tip
if (role == Qt.BackgroundRole and not path
and index.column() in self.flagged):
return self.flag
return None
def headerData(self, section, orientation, role=Qt.DisplayRole):
"""The column captions, and row numbers from 1.
:param section: the column or row.
:param orientation: horizontal for columns.
:param role: the item data role.
"""
if role != Qt.DisplayRole:
return None
if orientation == Qt.Horizontal:
if 0 <= section < len(self._captions):
return self._captions[section]
return None
return str(section + 1)
def flags(self, index):
"""A cell takes drops; one holding a file can be dragged.
:param index: the cell.
"""
if not index.isValid():
return Qt.ItemIsDropEnabled
if self.path(index.row(), index.column()):
return self._file_flags
return self._slot_flags
def supportedDragActions(self):
"""Cells are moved, or copied to another popup."""
return Qt.CopyAction | Qt.MoveAction
def supportedDropActions(self):
"""Drops move cells or copy files in."""
return Qt.CopyAction | Qt.MoveAction
def mimeTypes(self) -> List[str]:
"""The cells format only."""
return [_CELLS_MIME]
def mimeData(self, indexes) -> QMimeData:
"""The drag of these cells, as the table builds it.
:param indexes: the dragged cells.
"""
table = self._table
if table is None:
return QMimeData()
return table.mimeData([_CellItem(table, i.row(), i.column())
for i in indexes if i.isValid()])
class _OrganizeTable(QTableView):
"""The channel/mask table; every column is a drop target.
:ivar dropped: ``(column, paths, from_column)`` -- files dropped on a
column; ``from_column`` is -1 for a drop from the file manager.
:ivar cell_moved: ``(from row, from column, to row, to column)`` -- one
cell dragged onto a slot; the dialog swaps the two.
:ivar clear_requested: ``([[row, column], ...])`` -- cells whose × was
clicked, or that were selected when Delete was pressed.
:ivar block_moved: ``([[row, column], ...], [anchor row, column],
[to row, column])`` -- several cells dragged together.
:ivar placeholder: the hint painted over the table while it holds no
file.
"""
dropped = Signal(int, list, int)
cell_moved = Signal(int, int, int, int)
clear_requested = Signal(list)
block_moved = Signal(list, list, list)
def __init__(self, parent=None):
"""A table that drags cells out and takes drops on any column.
:param parent: the Qt parent.
"""
super().__init__(parent)
self.setModel(_CellModel((), (), (), self))
self.setAcceptDrops(True)
self.setDragEnabled(True)
self.setDragDropMode(QAbstractItemView.DragDrop)
self.setDefaultDropAction(Qt.MoveAction)
self.setEditTriggers(QAbstractItemView.NoEditTriggers)
self.setSelectionMode(QAbstractItemView.ExtendedSelection)
self.setMouseTracking(True)
self._pressed_close = None
self.placeholder = tr("Drag images or folders here")
#: Where a rubber-band selection started, and its band.
self._band_origin = None
self._band = None
self._band_extend = False
self._drag_anchor = None
def _is_empty(self) -> bool:
"""Whether no cell holds a file (the placeholder is shown)."""
return not self.model().has_files
def show_cells(self, rows, captions, flagged) -> None:
"""Show new cells: a new model in one step, the old one released.
:param rows: one list per row, one path or None per column.
:param captions: the column headers.
:param flagged: the columns whose empty cells are flagged.
"""
old_model, old_selection = self.model(), self.selectionModel()
self.setModel(_CellModel(rows, captions, flagged, self))
for gone in (old_selection, old_model):
if gone is not None:
gone.deleteLater()
def rowCount(self) -> int:
"""How many rows the table shows."""
return self.model().rowCount()
def columnCount(self) -> int:
"""How many columns the table shows."""
return self.model().columnCount()
def item(self, row: int, column: int) -> Optional[_CellItem]:
"""A handle on one cell, or None outside the table.
:param row: the row.
:param column: the column.
"""
if 0 <= row < self.rowCount() and 0 <= column < self.columnCount():
return _CellItem(self, row, column)
return None
def selectedItems(self) -> List[_CellItem]:
"""Handles on the selected cells."""
return [_CellItem(self, i.row(), i.column())
for i in self.selectedIndexes()]
def paintEvent(self, event) -> None:
"""Paint the table, and the drop hint over it while it is empty.
:param event: the paint event.
"""
super().paintEvent(event)
if not self._is_empty():
self._paint_closes(event.rect())
return
painter = QPainter(self.viewport())
color = QColor(self.palette().color(self.foregroundRole()))
color.setAlpha(150)
painter.setPen(color)
font = painter.font()
font.setPointSizeF(font.pointSizeF() * 1.4)
painter.setFont(font)
painter.drawText(self.viewport().rect(), Qt.AlignCenter
| Qt.TextWordWrap, self.placeholder)
painter.end()
def _paint_closes(self, area) -> None:
"""Draw the × of every file cell in ``area``, in one pass.
One painter and two pens for all of them: drawn cell by cell, a
text view's × doubled the cost of painting the table.
:param area: the viewport rectangle being painted.
"""
model = self.model()
top = self.rowAt(area.top())
bottom = self.rowAt(area.bottom())
left = self.columnAt(area.left())
right = self.columnAt(area.right())
if top < 0 or left < 0:
return
bottom = model.rowCount() - 1 if bottom < 0 else bottom
right = model.columnCount() - 1 if right < 0 else right
hover = getattr(self.itemDelegate(), "hover", None)
marks = []
for row in range(top, bottom + 1):
y, height = self.rowViewportPosition(row), self.rowHeight(row)
for column in range(left, right + 1):
if model.path(row, column):
cell = QRect(self.columnViewportPosition(column), y,
self.columnWidth(column), height)
marks.append((_close_rect(cell).adjusted(3, 3, -3, -3),
hover == (row, column)))
if not marks:
return
painter = QPainter(self.viewport())
painter.setRenderHint(QPainter.Antialiasing, True)
for layer in (0, 1):
for inset, hovered in marks:
painter.setPen(_close_pens(hovered)[layer])
painter.drawLine(inset.topLeft(), inset.bottomRight())
painter.drawLine(inset.topRight(), inset.bottomLeft())
painter.end()
def _select_band(self, pos) -> None:
"""Select every cell the band from its origin to ``pos`` touches.
:param pos: a point in the viewport.
"""
from PySide6.QtCore import QItemSelectionModel, QRect
from PySide6.QtWidgets import QRubberBand
rect = QRect(self._band_origin, pos).normalized()
if self._band is None:
self._band = QRubberBand(QRubberBand.Rectangle, self.viewport())
self._band.setGeometry(rect)
self._band.show()
flag = (QItemSelectionModel.Select if self._band_extend
else QItemSelectionModel.ClearAndSelect)
self.setSelection(rect, flag)
def _end_band(self) -> bool:
"""Hide the rubber band and forget its origin.
:returns: whether a band was showing.
"""
showing = self._band is not None and self._band.isVisible()
self._band_origin = None
if self._band is not None:
self._band.hide()
return showing
def _close_hit(self, pos):
"""The cell whose × is under ``pos``, or None.
:param pos: a point in the viewport.
:returns: ``(row, column)`` or None.
"""
index = self.indexAt(pos)
if index.isValid() and index.data(Qt.UserRole) and _close_rect(
self.visualRect(index)).contains(pos):
return index.row(), index.column()
return None
def _set_hover(self, hit) -> None:
"""Turn the × under the mouse red, and the last one white again.
:param hit: ``(row, column)`` or None.
"""
delegate = self.itemDelegate()
if getattr(delegate, "hover", None) != hit:
delegate.hover = hit
self.viewport().update()
def mouseMoveEvent(self, event) -> None:
"""Follow the mouse over the ×s.
:param event: the mouse event.
"""
pos = event.position().toPoint()
self._set_hover(self._close_hit(pos))
if self._band_origin is not None and event.buttons() & Qt.LeftButton:
if (self._band is not None and self._band.isVisible()) or (
pos - self._band_origin).manhattanLength() >= \
QApplication.startDragDistance():
self._select_band(pos)
event.accept()
return
super().mouseMoveEvent(event)
def leaveEvent(self, event) -> None:
"""No × is hovered once the mouse leaves.
:param event: the leave event.
"""
self._set_hover(None)
super().leaveEvent(event)
def mousePressEvent(self, event) -> None:
"""A press on a × is the ×'s, not the start of a drag.
:param event: the mouse event.
A band selection starts from an empty slot, from outside the cells, or
with Ctrl or Shift held; a plain press on a file drags it.
"""
pos = event.position().toPoint()
hit = self._close_hit(pos)
self._pressed_close = hit
if hit is not None and event.button() == Qt.LeftButton:
event.accept()
return
index = self.indexAt(pos)
self._drag_anchor = ([index.row(), index.column()]
if index.isValid() else None)
extend = bool(event.modifiers()
& (Qt.ControlModifier | Qt.ShiftModifier))
if event.button() == Qt.LeftButton and (
extend or not index.isValid()
or not index.data(Qt.UserRole)):
self._band_origin = pos
self._band_extend = extend
super().mousePressEvent(event)
def mouseReleaseEvent(self, event) -> None:
"""Releasing on the × that was pressed clears its cell.
:param event: the mouse event.
"""
if self._end_band():
self._pressed_close = None
event.accept()
return
hit = self._close_hit(event.position().toPoint())
pressed, self._pressed_close = self._pressed_close, None
if hit is not None and hit == pressed:
event.accept()
chosen = _selection_cells(self.selectedIndexes())
self.clear_requested.emit(chosen if list(hit) in chosen
else [list(hit)])
return
super().mouseReleaseEvent(event)
def keyPressEvent(self, event) -> None:
"""Delete or Backspace clears the selected cells.
:param event: the key event.
"""
if event.key() in (Qt.Key_Delete, Qt.Key_Backspace):
cells = [[item.row(), item.column()]
for item in self.selectedItems()
if item.data(Qt.UserRole)]
if cells:
self.clear_requested.emit(cells)
event.accept()
return
super().keyPressEvent(event)
def mimeTypes(self) -> List[str]:
"""The formats a drag out of the table offers.
:returns: the cells format only.
"""
return [_CELLS_MIME]
def mimeData(self, items) -> QMimeData:
"""Put the dragged cells' files and slots into the drag.
``column`` is the column of the cell the drag started on (or of the
first cell) and ``paths`` that column's files, which a drop on the
"new channel" strip moves; ``cells`` are every dragged slot and
``anchor`` the one the drag started on, which a drop on the table
moves as a block.
:param items: the dragged items.
:returns: the drag's data.
"""
mime = QMimeData()
items = sorted((item for item in items if item.data(Qt.UserRole)),
key=lambda item: (item.row(), item.column()))
cells = [[item.row(), item.column()] for item in items]
anchor = self._drag_anchor if self._drag_anchor in cells else (
cells[0] if cells else None)
column = anchor[1] if anchor else -1
chosen = [item for item in items if item.column() == column]
mime.setData(_CELLS_MIME, QByteArray(json.dumps(
{"column": column,
"paths": [item.data(Qt.UserRole) for item in chosen],
"cells": cells, "anchor": anchor}
).encode("utf-8")))
return mime
def _accepts(self, event) -> bool:
"""Whether a drag carries files or cells, and there is a column.
:param event: the drag event.
"""
mime = event.mimeData()
if _mime_cells(mime)[1]:
return bool(self.columnCount())
return bool(_mime_paths(mime))
def dragEnterEvent(self, event) -> None:
"""Accept files from outside and cells from another column.
:param event: the drag event.
"""
if self._accepts(event):
event.acceptProposedAction()
else:
event.ignore()
def dragMoveEvent(self, event) -> None:
"""Keep accepting while the drag moves over a column.
:param event: the drag event.
"""
self.dragEnterEvent(event)
def dropEvent(self, event) -> None:
"""Hand the drop, and the column it landed in, to the dialog.
:param event: the drop event.
A drop on the empty table's placeholder makes the first channel column.
"""
if not self._accepts(event):
event.ignore()
return
event.setDropAction(Qt.CopyAction)
event.accept()
if not self.columnCount():
self.dropped.emit(-1, _mime_paths(event.mimeData()), -1)
return
column = self.columnAt(int(event.position().x()))
if column < 0:
column = self.columnCount() - 1
source, cells = _mime_cells(event.mimeData())
slots = _mime_slots(event.mimeData())
if slots:
row = self.rowAt(int(event.position().y()))
if row < 0:
row = self.rowCount()
if len(slots) == 1:
self.cell_moved.emit(slots[0][0], slots[0][1], row, column)
else:
anchor = _mime_anchor(event.mimeData()) or slots[0]
self.block_moved.emit(slots, anchor, [row, column])
return
if cells:
self.dropped.emit(column, cells, -1 if source is None else source)
else:
self.dropped.emit(column, _mime_paths(event.mimeData()), -1)
class _NewColumnZone(QFrame):
""""Drop here for a new channel": makes a channel column of the drop.
:ivar dropped: ``(paths, from_column)``.
"""
dropped = Signal(list, int)
def __init__(self, parent=None):
"""Build the strip.
:param parent: the Qt parent.
"""
super().__init__(parent)
self.setObjectName("OrganizeNewColumnZone")
self.setFrameShape(QFrame.StyledPanel)
self.setAcceptDrops(True)
layout = QVBoxLayout(self)
layout.setContentsMargins(6, 4, 6, 4)
label = QLabel(tr("Drop images or folders here for a new channel"))
label.setAlignment(Qt.AlignCenter)
label.setWordWrap(True)
layout.addWidget(label)
def dragEnterEvent(self, event) -> None:
"""Accept files from outside and cells from the table.
:param event: the drag event.
"""
mime = event.mimeData()
if _mime_paths(mime) or _mime_cells(mime)[1]:
event.acceptProposedAction()
else:
event.ignore()
def dragMoveEvent(self, event) -> None:
"""Keep accepting while the drag moves over the strip.
:param event: the drag event.
"""
self.dragEnterEvent(event)
def dropEvent(self, event) -> None:
"""Hand the drop to the dialog.
:param event: the drop event.
"""
mime = event.mimeData()
source, cells = _mime_cells(mime)
paths = cells or _mime_paths(mime)
if not paths:
event.ignore()
return
event.acceptProposedAction()
self.dropped.emit(paths, -1 if source is None else source)
class _TeachQuestion(QDialog):
""""Which channel is this?" -- one image, one answer, for "Teach me".
:ivar answer: ``("channel", n)``, ``("mask", n, role)``, ``"skip"`` or
``"stop"``.
"""
def __init__(self, path: str, known: List[object], parent=None):
"""Show the image and one button per possible answer.
:param path: the image.
:param known: labels already used, channels first.
:param parent: the Qt parent.
"""
super().__init__(parent)
self.setWindowTitle(tr("Which channel is this?"))
self.answer: object = "stop"
layout = QVBoxLayout(self)
picture = QLabel()
array = cs.thumbnail(path, 320)
if array is not None:
from PySide6.QtGui import QImage, QPixmap
array = np.ascontiguousarray(array)
image = QImage(array.data, array.shape[1], array.shape[0],
array.shape[1], QImage.Format_Grayscale8)
picture.setPixmap(QPixmap.fromImage(image.copy()))
layout.addWidget(picture)
caption = QLabel(os.path.basename(path))
caption.setWordWrap(True)
layout.addWidget(caption)
channels = sorted({lab[1] for lab in known if lab[0] == "channel"})
row = QHBoxLayout()
for n in channels + [max(channels, default=0) + 1]:
text = (tr("Channel {n}", n=n) if n in channels
else tr("New channel ({n})", n=n))
button = QPushButton(text)
button.clicked.connect(
lambda _c=False, n=n: self._choose(("channel", n)))
row.addWidget(button)
layout.addLayout(row)
mask_row = QHBoxLayout()
self.mask_channel = QComboBox()
for n in channels or [1]:
self.mask_channel.addItem(tr("of channel {n}", n=n), n)
self.mask_role = QComboBox()
for role in cs.MASK_ROLES:
self.mask_role.addItem(role, role)
mask_button = QPushButton(tr("It is a mask"))
mask_button.clicked.connect(lambda: self._choose((
"mask", int(self.mask_channel.currentData()),
self.mask_role.currentData())))
for widget in (mask_button, self.mask_role, self.mask_channel):
mask_row.addWidget(widget)
layout.addLayout(mask_row)
end_row = QHBoxLayout()
skip = QPushButton(tr("Skip this image"))
skip.clicked.connect(lambda: self._choose("skip"))
stop = QPushButton(tr("Stop: the rest are not needed"))
stop.clicked.connect(lambda: self._choose("stop"))
end_row.addWidget(skip)
end_row.addWidget(stop)
layout.addLayout(end_row)
def _choose(self, answer) -> None:
"""Keep the answer and close.
:param answer: the label chosen.
"""
self.answer = answer
self.accept()
[docs]
class OrganizeForMeasureDialog(QDialog):
"""Put images and masks into channel folders and ``merged/`` for Measure.
:ivar plan: the confirmed :class:`spacr.channel_sorting.SortPlan` once the
dialog is accepted, else None.
:ivar columns: the table's columns, in order.
:ivar rows: one list per field, one path or None per column.
"""
def __init__(self, source: str = "", parent=None,
channel_folders: Optional[Sequence[str]] = None,
masks_dir: Optional[str] = None):
"""Build the popup, prefilled from a drop when there was one.
:param source: the source folder, if known.
:param parent: the Qt parent.
:param channel_folders: folders to put in the table, one channel
column each (their masks from each folder's ``masks/``) -- the
Make Masks drop of several folders or of channel-like
subfolders.
:param masks_dir: the source's masks folder, when not its ``masks/``.
The layout puts the fields on aligned rows at the top, every action
button on one line below them, then the view controls over the table.
"""
super().__init__(parent)
mode = self._mode
if mode not in _MODES:
raise ValueError(f"mode must be one of {_MODES}, not {mode!r}")
self.mode = mode
self.setWindowTitle(tr("Organize for Measure") if mode == "measure"
else tr("Organize images"))
self.resize(1000, 760)
self.masks_dir = masks_dir
self.columns: List[_Column] = []
self.rows: List[List[Optional[str]]] = []
self.row_keys: List[Optional[cs.SetKey]] = []
self.plan: Optional[cs.SortPlan] = None
self._output_auto = True
self._note = ""
self._alias_cache: Dict[str, Dict[str, str]] = {}
#: The regex "Teach me" learned and its ``{marker: label}``.
self._taught: tuple = ("", None)
self._skipped: set = set()
#: Asks "Which channel is this?" for one image; replaced in tests.
self.ask_label = self._ask_label
#: Asks what a part of the names means; replaced in tests.
self.ask_marker = self._ask_marker
#: Answers yes/no questions; replaced in tests.
self.ask = self._ask
#: Shows three detected sets and asks; replaced in tests.
self.confirm_examples = self._ask_examples
layout = QVBoxLayout(self)
intro = QLabel({
"measure": tr(
"Organise images and their masks into the layout Measure reads. "
"Either give a source folder and sort it with a regex, or drop "
"files and folders into the columns below: each column is a "
"channel or a mask, each row one field, and dropped files are "
"matched across the columns by name. On Apply the images and "
"masks are MOVED into Yokogawa-named channel folders and merged "
"into merged/; every move is written to a manifest."),
"images": tr(
"Organise intensity images, however they are named or "
"foldered, into the folder Mask Generation reads. Either give "
"a source folder and sort it with a regex, or drop files and "
"folders into the columns below: each column is a channel, "
"each row one field. On Apply the images are MOVED into one "
"folder with Yokogawa names and src is set to it; every move "
"is written to a manifest."),
"import": tr(
"Organise intensity images and their masks into the layout "
"Mask Generation or Measure reads. Either give a source folder "
"and sort it with a regex, or drop files and folders into the "
"columns below: each column is a channel or a mask, each row "
"one field. On Apply the files are MOVED; every move is "
"written to a manifest."),
}[mode])
intro.setWordWrap(True)
layout.addWidget(intro)
from PySide6.QtWidgets import QGridLayout, QSlider
fields = QGridLayout()
fields.setColumnStretch(1, 1)
fields.addWidget(QLabel(tr("Source folder")), 0, 0)
self.source_edit = _PathEdit()
self.source_edit.setPlaceholderText(tr(
"A folder of images; drop one here"))
self.source_edit.setText(source or "")
self.source_edit.textChanged.connect(lambda _t: self._suggest_output())
fields.addWidget(self.source_edit, 0, 1)
self.browse_button = QPushButton(tr("Browse…"))
self.browse_button.clicked.connect(self._browse)
fields.addWidget(self.browse_button, 0, 2)
self.consolidate_check = QCheckBox(tr(
"Consolidate subfolders into filenames first (copies the images "
"into a new folder, each named after its folders; the originals "
"are not touched)"))
fields.addWidget(self.consolidate_check, 1, 1, 1, 2)
from ..screens.settings_model import _MetadataTypeField
regex_label = QLabel(tr("Regex"))
regex_label.setToolTip(tr(
"Regex: the same filename conventions as Mask generation. "
"'auto' lets spaCR propose one; 'custom' uses the regex below."))
fields.addWidget(regex_label, 2, 0)
self.metadata_field = _MetadataTypeField(
_DEFAULT_METADATA_TYPE,
source_folder=lambda: self.source_edit.text(),
custom_regex=lambda: self.custom_edit.text(), parent=self)
fields.addWidget(self.metadata_field, 2, 1, 1, 2)
fields.addWidget(QLabel(tr("custom_regex")), 3, 0)
self.custom_edit = QLineEdit()
self.custom_edit.setPlaceholderText(tr(
"Regex with named groups: chanID, and wellID, fieldID..."))
fields.addWidget(self.custom_edit, 3, 1, 1, 2)
fields.addWidget(QLabel(tr("Output folder")), 4, 0)
self.output_edit = QLineEdit()
self.output_edit.textEdited.connect(self._output_edited)
fields.addWidget(self.output_edit, 4, 1, 1, 2)
self.layout_box = QComboBox()
self.layout_box.setObjectName("OrganizeLayoutBox")
self.layout_box.addItem(tr("Measure: channel folders and merged/"),
"measure")
self.layout_box.addItem(tr("Mask Generation: one folder of images, "
"masks in masks/"), "mask")
self.layout_box.setCurrentIndex(1 if mode == "images" else 0)
if mode == "import":
layout_label = QLabel(tr("Write for"))
layout_label.setToolTip(tr(
"Measure: channel folders, the masks and merged/*.npy. Mask "
"Generation: one folder of Yokogawa-named images, its src, "
"with any masks in its masks/ folder."))
fields.addWidget(layout_label, 5, 0)
fields.addWidget(self.layout_box, 5, 1, 1, 2)
else:
self.layout_box.hide()
layout.addLayout(fields)
self.sort_button = QPushButton(tr("Sort by regex"))
self.sort_button.setToolTip(tr(
"Read the source folder (or the files in the table) with the "
"regex: chanID gives the column, the other groups the row."))
self.sort_button.clicked.connect(lambda: self._start_sort())
from ..job_runner import JobRunner
self._sort_jobs = JobRunner(self, app_key="organize images",
user_visible=False)
self._sort_jobs.job_failed.connect(self._sort_failed)
self._sort_token = 0
self._sorting = False
self._known_shapes = None
self.auto_button = QPushButton(tr("Auto regex"))
self.auto_button.setToolTip(tr(
"Find a regex that puts every image in exactly one complete set, "
"put it in custom_regex and sort by it."))
self.auto_button.clicked.connect(lambda: self.auto_regex())
self.detect_button = QPushButton(tr("Detect sets"))
self.detect_button.setToolTip(tr(
"Ignore the names' structure: match every column's files by "
"similarity, order and size, then show three sets to check."))
self.detect_button.clicked.connect(lambda: self.detect_sets())
self.teach_button = QPushButton(tr("Teach me…"))
self.teach_button.setToolTip(tr(
"Show one image at a time and ask which channel (or mask) it is, "
"learning a regex from the answers until every image is placed."))
self.teach_button.clicked.connect(lambda: self.teach())
self.add_channel_button = QPushButton(tr("Add channel"))
self.add_channel_button.clicked.connect(
lambda: self.add_column("channel"))
self.add_mask_button = QPushButton(tr("Add mask"))
self.add_mask_button.clicked.connect(lambda: self.add_column("mask"))
self.add_mask_button.setVisible(mode != "images")
self.remove_button = QPushButton(tr("Remove selected files"))
self.remove_button.setToolTip(tr(
"Take the selected cells' files out of the table; the files "
"themselves are not touched."))
self.remove_button.clicked.connect(self._remove_selected)
self.split_button = QPushButton(tr("Split multi-channel files"))
self.split_button.setObjectName("OrganizeSplitButton")
self.split_button.setToolTip(tr(
"Copy each channel of the table's multi-channel images into its "
"own file in split_channels/, one channel column per plane; the "
"originals are not touched."))
self.split_button.clicked.connect(
lambda _checked=False: self._split_multichannel())
#: The one line of action buttons, in order.
self.action_row = QHBoxLayout()
for button in (self.sort_button, self.auto_button, self.detect_button,
self.teach_button):
self.action_row.addWidget(button)
self.action_row.addSpacing(12)
extra = (self.split_button,) if mode != "measure" else ()
if not extra:
self.split_button.setParent(self)
self.split_button.hide()
for button in (self.add_channel_button, self.add_mask_button,
self.remove_button) + extra:
self.action_row.addWidget(button)
self.action_row.addStretch(1)
layout.addLayout(self.action_row)
view_row = QHBoxLayout()
view_row.addWidget(QLabel(tr("Channels and masks")))
view_row.addStretch(1)
self.color_button = QPushButton(tr("Text colour"))
self.color_button.setFlat(True)
self.color_button.setCursor(Qt.PointingHandCursor)
self.color_button.setToolTip(tr(
"The colour of the names written over the images."))
self.color_button.clicked.connect(lambda: self._pick_text_color())
view_row.addWidget(self.color_button)
show_label = QLabel(tr("Show"))
show_label.setToolTip(tr(
"Show each cell as its file name, its image, or its image with "
"the name written over it."))
view_row.addWidget(show_label)
self.view_box = QComboBox()
for value, caption in (("text", tr("Text")), ("image", tr("Image")),
("both", tr("Image + text"))):
self.view_box.addItem(caption, value)
view_row.addWidget(self.view_box)
size_label = QLabel(tr("Size"))
size_label.setToolTip(tr("The size of the images in the table."))
view_row.addWidget(size_label)
self.size_slider = QSlider(Qt.Horizontal)
self.size_slider.setRange(*_THUMB_RANGE)
self.size_slider.setFixedWidth(140)
self.size_slider.setToolTip(tr("The size of the images in the table."))
self.thumb_size = _load_thumb_pref()
self.size_slider.setValue(self.thumb_size)
self.size_slider.valueChanged.connect(self._set_thumb_size)
view_row.addWidget(self.size_slider)
from ..mask_thumbnail_quality import quality_combo, changes
view_row.addWidget(QLabel(tr("Thumbnail quality")))
self.quality_box = quality_combo(self)
view_row.addWidget(self.quality_box)
changes.changed.connect(self._quality_changed)
layout.addLayout(view_row)
self._editors_row = QHBoxLayout()
layout.addLayout(self._editors_row)
table_row = QHBoxLayout()
self.table = _OrganizeTable()
self.table.dropped.connect(self._on_table_drop)
self.table.cell_moved.connect(self._swap_slots)
self.table.block_moved.connect(self._move_slots)
self.table.clear_requested.connect(self._clear_slots)
self.delegate = _CellDelegate(self.table, self.table)
self.table.setItemDelegate(self.delegate)
self._thumb_workers: list = []
self._thumb_pending: set = set()
self._thumb_generation = 0
self._thumb_signatures = {}
self._thumb_requested = {}
self._thumb_closed = False
self._thumb_render_ratio = None
self._thumb_poll = QTimer(self)
self._thumb_poll.setInterval(750)
self._thumb_poll.timeout.connect(self._load_thumbnails)
self._thumb_poll.start()
self._thumb_kick = QTimer(self)
self._thumb_kick.setSingleShot(True)
self._thumb_kick.setInterval(_THUMBS_AFTER_REFRESH_MS)
self._thumb_kick.timeout.connect(self._load_thumbnails)
self.table.verticalScrollBar().valueChanged.connect(self._load_thumbnails)
self.table.horizontalScrollBar().valueChanged.connect(self._load_thumbnails)
self.finished.connect(lambda _code: self._stop_thumbs())
table_row.addWidget(self.table, 1)
self.new_zone = _NewColumnZone()
self.new_zone.setMaximumWidth(150)
self.new_zone.dropped.connect(self._on_new_zone_drop)
table_row.addWidget(self.new_zone)
layout.addLayout(table_row, 1)
self.status = QLabel()
self.status.setWordWrap(True)
layout.addWidget(self.status)
buttons = QHBoxLayout()
self.cancel_button = QPushButton(tr("Cancel"))
self.apply_button = QPushButton(tr("Apply"))
self.apply_button.setObjectName("PrimaryButton")
self.cancel_button.clicked.connect(self.reject)
self.apply_button.clicked.connect(self._on_apply)
buttons.addStretch(1)
buttons.addWidget(self.cancel_button)
buttons.addWidget(self.apply_button)
layout.addLayout(buttons)
for folder in channel_folders or ():
self.add_files(self.add_column("channel"),
[os.path.join(folder, n)
for n in cs.list_folder_images(folder)],
rematch=False)
if channel_folders:
self._rematch()
self._suggest_output()
stored = _load_view_prefs()
self._set_text_color(stored[1])
self.view_box.setCurrentIndex(max(0, self.view_box.findData(stored[0])))
self.view_box.currentIndexChanged.connect(
lambda _i: self._set_view(self.view_box.currentData()))
self._set_view(stored[0], remember=False)
self._refresh()
from ..screens.settings_model import retarget_field_tooltips
retarget_field_tooltips(self)
#: ``"measure"`` writes Measure's layout; ``"images"`` takes intensity
#: images only and writes one folder of Yokogawa-named images, the input
#: Mask Generation reads; ``"import"`` takes images and masks and offers
#: both layouts. Set by :func:`_dialog_for`.
_mode = "measure"
@property
def _target_layout(self) -> str:
"""``"measure"`` or ``"mask"``: the layout Apply writes."""
if self.mode == "measure":
return "measure"
return self.layout_box.currentData()
def _split_multichannel(self) -> int:
"""Split the table's multi-channel images into one channel each.
Each plane of a multi-channel image in a channel column is written to
``<base>/split_channels/`` by :func:`_split_channel_file` and put in
the channel column as many places right of the image's own as its
plane number, made when missing; the image leaves the table. The rows
are matched again afterwards.
:returns: how many images were split.
"""
base = self._base_folder()
folder = ""
targets: Dict[int, List[str]] = {}
split: List[str] = []
for column in self._channel_columns():
channel = self._channel_of(column)
for row in self.rows:
path = row[column]
if not path or len(cs.image_shape(path) or ()) != 3:
continue
if not folder:
folder = cs.unused_folder(base or os.path.dirname(path),
"split_channels")
planes = _split_channel_file(path, folder, base)
if not planes:
continue
split.append(path)
for index, plane in enumerate(planes):
targets.setdefault(channel + index, []).append(plane)
if not split:
self.status.setText(tr("No multi-channel images in the table."))
return 0
self._forget(split)
while len(self._channel_columns()) < max(targets):
self.add_column("channel")
for channel, planes in sorted(targets.items()):
self.add_files(self._column_for_channel(channel), planes,
rematch=False)
self._rematch()
self.status.setText(tr(
"Split {n} multi-channel image(s) into {folder}.", n=len(split),
folder=folder))
return len(split)
def _channel_columns(self) -> List[int]:
"""The indices of the channel columns, in order."""
return [i for i, c in enumerate(self.columns) if c.kind == "channel"]
def _channel_of(self, column: int) -> int:
"""The 1-based channel number of channel column ``column``.
:param column: a channel column's index.
"""
return self._channel_columns().index(column) + 1
def _column_for_channel(self, channel: int) -> Optional[int]:
"""The index of channel ``channel``'s column, or None.
:param channel: a 1-based channel.
"""
channels = self._channel_columns()
return channels[channel - 1] if 0 < channel <= len(channels) else None
def _mask_column(self, channel: int) -> Optional[int]:
"""The first mask column of ``channel``, or None.
:param channel: a 1-based channel.
"""
for index, column in enumerate(self.columns):
if column.kind == "mask" and column.of_channel == channel:
return index
return None
[docs]
def add_column(self, kind: str, role: str = "",
of_channel: Optional[int] = None) -> int:
"""Add a channel or mask column at the end.
:param kind: ``"channel"`` or ``"mask"``.
:param role: a mask's role; default the first role no mask has.
:param of_channel: a mask's channel; default the first channel with
no mask column, else 1.
:returns: the new column's index.
"""
column = _Column(kind)
if kind == "mask":
used = {c.role for c in self.columns if c.kind == "mask"}
column.role = role or next(
(r for r in cs.MASK_ROLES if r not in used), cs.MASK_ROLES[0])
if of_channel is None:
masked = {c.of_channel for c in self.columns if c.kind == "mask"}
of_channel = next(
(n for n in range(1, len(self._channel_columns()) + 1)
if n not in masked), 1)
column.of_channel = int(of_channel)
self.columns.append(column)
for row in self.rows:
row.append(None)
self._refresh()
return len(self.columns) - 1
[docs]
def remove_column(self, index: int) -> None:
"""Remove a column and its files from the table.
:param index: the column.
"""
if not 0 <= index < len(self.columns):
return
removed = self.columns.pop(index)
if removed.kind == "channel":
channel = len([i for i in range(index)
if self.columns[i].kind == "channel"]) + 1
for column in self.columns:
if column.kind == "mask" and column.of_channel > channel:
column.of_channel -= 1
for row in self.rows:
row.pop(index)
self._drop_empty_rows()
self._refresh()
def _column_files(self, index: int) -> List[str]:
"""The files in column ``index``, in row order.
:param index: a column.
"""
return [row[index] for row in self.rows if row[index]]
[docs]
def add_files(self, column: int, paths: Iterable[str],
rematch: bool = True) -> List[str]:
"""Put dropped files and folders into a column, and match the rows.
A folder brings its images, recursively, not its ``masks/``. Images
dropped into a channel column bring the masks Make Masks saved for
them (``<their folder>/masks/<stem>.tif``) into that channel's mask
column, which is made when there is none. A file already in the
table moves to this column. Nothing on disk changes.
:param column: the column index.
:param paths: files and folders.
:param rematch: match the rows afterwards (off while prefilling).
:returns: the image files added.
"""
images, skipped = _images_in(paths)
self._note = (tr("Not images, left out: {names}",
names=", ".join(skipped)) if skipped else "")
if not images or not 0 <= column < len(self.columns):
self._refresh()
return []
self._forget(images)
for path in images:
row = [None] * len(self.columns)
row[column] = path
self.rows.append(row)
self.row_keys.append(None)
self._bring_masks(images, column)
if not self.source_edit.text().strip():
self._suggest_output()
if rematch:
self._rematch()
else:
self._refresh()
return images
def _bring_masks(self, images: Sequence[str], column: int) -> None:
"""Put the saved masks of images in a channel column beside them.
Each image's mask where Make Masks saves it
(``<its folder>/masks/<stem>.tif``) goes into that channel's mask
column, which is made -- its role guessed from the names -- when
there is none. A mask already elsewhere in the table moves.
:param images: images just put in ``column``.
:param column: their column.
"""
if self.columns[column].kind != "channel":
return
channel = self._channel_of(column)
masks = [m for m in (cs.mask_for(os.path.dirname(p), p)
for p in images) if m]
if not masks:
return
mask_column = self._mask_column(channel)
if mask_column is None:
roles = cs.default_mask_roles({channel: list(images)}, [channel])
used = {c.role for c in self.columns if c.kind == "mask"}
role = roles.get(channel, "")
mask_column = self.add_column(
"mask", role if role not in used else "", channel)
self._forget(masks)
for mask in masks:
row = [None] * len(self.columns)
row[mask_column] = mask
self.rows.append(row)
self.row_keys.append(None)
def _forget(self, paths: Iterable[str]) -> None:
"""Take files out of whatever cells hold them.
:param paths: files.
"""
paths = set(paths)
for row in self.rows:
for index, value in enumerate(row):
if value in paths:
row[index] = None
self._drop_empty_rows()
def _drop_empty_rows(self) -> None:
"""Remove rows that hold no file at all."""
kept = [(row, key) for row, key in zip(self.rows, self.row_keys)
if any(row)]
self.rows = [row for row, _key in kept]
self.row_keys = [key for _row, key in kept]
def _move_files(self, paths: Iterable[str], column: int) -> None:
"""Move files already in the table into another column, and rematch.
:param paths: files in the table.
:param column: the column they go to.
"""
paths = [p for p in paths if p]
if not paths or not 0 <= column < len(self.columns):
return
self._forget(paths)
for path in paths:
row = [None] * len(self.columns)
row[column] = path
self.rows.append(row)
self.row_keys.append(None)
self._bring_masks(paths, column)
self._rematch()
def _remove_selected(self) -> int:
"""Take the selected cells' files out of the table.
:returns: how many files were taken out.
"""
paths = [item.data(Qt.UserRole) for item in self.table.selectedItems()
if item.data(Qt.UserRole)]
self._forget(paths)
self._refresh()
return len(paths)
def _on_table_drop(self, column: int, paths: list, source: int) -> None:
"""Files dropped on a column: from outside, or from another column.
:param column: the column dropped on.
:param paths: the files or folders.
:param source: the column they were dragged from, or -1.
"""
if column < 0:
self._on_new_zone_drop(paths, source)
return
if source >= 0:
if source != column:
self._move_files(paths, column)
return
self.add_files(column, paths)
def _on_new_zone_drop(self, paths: list, source: int) -> None:
"""Files dropped on "new channel": a channel column of their own.
:param paths: the files or folders.
:param source: the column they were dragged from, or -1.
"""
column = self.add_column("channel")
if source >= 0:
self._move_files(paths, column)
else:
self.add_files(column, paths)
def _partners(self) -> Dict[int, int]:
"""``{mask column: its channel's column}`` for matching.
:returns: the pairs whose channel column exists.
"""
partners = {}
for index, column in enumerate(self.columns):
if column.kind == "mask":
target = self._column_for_channel(column.of_channel)
if target is not None:
partners[index] = target
return partners
def _rematch(self) -> None:
"""Line every column's files up in rows by name, order and size."""
files = [self._column_files(i) for i in range(len(self.columns))]
self.rows = cs._match_columns(files, self._partners())
self.row_keys = [None] * len(self.rows)
self._refresh()
def _source(self) -> str:
"""The source folder as typed, or ``""``."""
return self.source_edit.text().strip()
def _regex_for(self, names: Sequence[str]) -> str:
"""The regex the regex box stands for, for these file names.
``auto`` is the custom regex when one is typed, else spaCR's proposal
(:func:`spacr.channel_sorting.infer_regex`); every other convention
is Mask generation's own pattern for the names' extensions.
:param names: file names or paths.
:returns: the regex, or ``""`` when there is none.
"""
return _regex_for_names(
self.metadata_field.get_value() or _DEFAULT_METADATA_TYPE,
self.custom_edit.text().strip(), names)
def _consolidate(self, source: str) -> str:
"""Copy the source's subfolders' images into one folder, if asked.
:param source: the source folder.
:returns: the folder to read: the consolidated copy, or ``source``.
"""
from ... import folder_consolidation as fc
if not self.consolidate_check.isChecked():
return source
files, _folders = fc.nested_file_count(source, cs.IMAGE_EXTS,
_NOT_CONSOLIDATED)
if not files:
return source
QApplication.setOverrideCursor(Qt.WaitCursor)
try:
result = fc.consolidate_folder(
source, extensions=cs.IMAGE_EXTS, skip_dirs=_NOT_CONSOLIDATED,
log=lambda _text: None)
finally:
QApplication.restoreOverrideCursor()
if result.failed:
self.status.setText("\n".join((
tr("Consolidation failed"),
tr("{n} file(s) could not be copied; see the manifest.", n=result.failed),
str(result.manifest),
)))
raise _ConsolidationFailed()
self.consolidate_check.setChecked(False)
self.source_edit.setText(str(result.output))
return str(result.output)
def _alias(self, path: str) -> str:
"""The name the regex reads for ``path``.
A file copied by folder consolidation is read under its ORIGINAL
name, from the folder's ``rename_manifest.csv``
(:func:`spacr.channel_sorting._original_names`): the copy's name may
have lost what told the channels apart.
:param path: an image path.
The alias keeps the file's folder, so two folders' field1.tif stay two
names; the regex reads the file name alone.
"""
return _alias_in(self._alias_cache, path)
def _sort_inputs(self):
"""What a sort reads, gathered from the widgets on the GUI thread.
The source is consolidated first when asked, as before any read.
:returns: the keyword arguments of :func:`_gather_sort_paths`, or
None when consolidation failed.
"""
source = self._source()
if source and os.path.isdir(source):
try:
source = self._consolidate(source)
except _ConsolidationFailed:
return None
return {"source": source, "masks_dir": self.masks_dir,
"channel_files": [], "mask_pairs": [],
"skipped": set(self._skipped)}
channel_files = [(self._column_files(column),
self._channel_of(column))
for column in self._channel_columns()]
partners = self._partners()
mask_pairs = [(self._column_files(partners[index]),
self._column_files(index))
for index, column in enumerate(self.columns)
if column.kind == "mask" and index in partners]
return {"source": "", "masks_dir": self.masks_dir,
"channel_files": channel_files, "mask_pairs": mask_pairs,
"skipped": set(self._skipped)}
def _paths_to_sort(self):
"""The images a regex or "Teach me" reads, their channels and masks.
:returns: ``(paths, {path: channel}, {image: mask})`` -- the source
folder's images and saved masks, or, without a source, the
channel columns' files and the masks matched to them.
"""
inputs = self._sort_inputs()
if inputs is None:
return None
return _gather_sort_paths(**inputs)
def _sort_arguments(self, pattern, chan_map):
"""The arguments of :func:`_sort_job` for one sort, or None.
:param pattern: the regex, or None for the regex box's.
:param chan_map: "Teach me"'s labels, or None for the taught ones
while the custom regex is still the taught one.
:returns: keyword arguments, or None when consolidation failed.
"""
inputs = self._sort_inputs()
if inputs is None:
return None
custom = self.custom_edit.text().strip()
if chan_map is None and pattern is None and custom == self._taught[0]:
chan_map = self._taught[1]
inputs.update(
pattern=pattern, chan_map=chan_map, custom=custom,
key=self.metadata_field.get_value() or _DEFAULT_METADATA_TYPE,
aliases={k: dict(v) for k, v in self._alias_cache.items()})
return inputs
def _install_sort(self, result) -> Optional[cs.SetReport]:
"""Show what :func:`_sort_job` laid out.
:param result: its result.
:returns: the set report, or None when there was nothing to read.
"""
self._alias_cache.update(result.get("aliases") or {})
if result.get("empty"):
self.status.setText(tr(
"Nothing to sort: give a source folder of images, or drop "
"images into the channel columns."))
return None
report = result["report"]
shared = result["shared"]
self.columns = result["columns"]
self.rows = result["rows"]
self.row_keys = result["row_keys"]
self._known_shapes = result.get("shapes")
self._note = report.summary()
if shared:
self._note += "\n" + tr(
"{n} mask(s) are shared by two images with one name stem and "
"were left out: {names}", n=len(shared),
names=", ".join(os.path.basename(m) for m in shared[:4]))
self._suggest_output()
self._refresh()
return report
[docs]
def sort_by_regex(self, pattern: Optional[str] = None,
chan_map: Optional[Dict[str, object]] = None
) -> Optional[cs.SetReport]:
"""Fill the table from the regex: chanID is the column, the rest the row.
With a source folder its images are read (consolidated first when
asked), and their masks found where Make Masks saves them; with none,
the files already in the channel columns are re-sorted, keeping the
column each is in. Masks go into a mask column per channel. A mask
two images share (two files with one stem) is left out and named.
:param pattern: the regex; default what the regex box stands for.
:param chan_map: ``{chanID value: label}`` learned by "Teach me",
a label being ``("channel", n)`` or ``("mask", n, role)``; by
default the regex's chanID values are channels in natural order.
:returns: the set report, or None when there was nothing to read.
"""
self._sort_token += 1
self._set_sorting(False)
arguments = self._sort_arguments(pattern, chan_map)
if arguments is None:
return None
return self._install_sort(_sort_job(**arguments))
def _start_sort(self) -> bool:
""""Sort by regex" pressed: read and lay out on a worker thread.
The folder listing, the mask lookups, the regex and the layout run
on :attr:`_sort_jobs`; the table is swapped in on the GUI thread
when they are done. A newer sort, a synchronous one or a closed
popup drops the result; an edit of the table meanwhile, when the
sort reads the columns, sorts again from the edited table.
:returns: whether a job was started.
"""
self._sort_token += 1
token = self._sort_token
arguments = self._sort_arguments(None, None)
if arguments is None:
self._set_sorting(False)
return False
snapshot = None if arguments["source"] else self._table_snapshot()
self._set_sorting(True)
self._sort_jobs.submit(
lambda: _sort_job(**arguments),
lambda result, _t=token, _s=snapshot: self._sort_done(
_t, _s, result))
return True
def _table_snapshot(self) -> tuple:
"""The table's files and columns, to tell whether it was edited."""
return (tuple(tuple(row) for row in self.rows),
tuple((c.kind, c.role, c.of_channel) for c in self.columns))
def _sort_done(self, token: int, snapshot, result) -> None:
"""Install a worker's sort unless something newer replaced it.
:param token: the job's :attr:`_sort_token`; a stale one is dropped.
:param snapshot: the table the job read, or None when it read the
source folder; a table edited since is sorted again.
:param result: what :func:`_sort_job` returned.
"""
if token != self._sort_token or self._thumb_closed:
return
if snapshot is not None and snapshot != self._table_snapshot():
self._start_sort()
return
self._set_sorting(False)
self.table.setUpdatesEnabled(False)
try:
self._install_sort(result)
finally:
QTimer.singleShot(_DRAW_AFTER_INSTALL_MS, self._show_table)
def _show_table(self) -> None:
"""Let the table draw the rows a sort installed, a turn later.
Installing the rows and drawing them are each a GUI-thread step of
15 to 30 ms on a large plate; run back to back they made one block
long enough to drop frames, so the drawing waits a frame.
"""
if shiboken_valid(self.table):
self.table.setUpdatesEnabled(True)
def _sort_failed(self, message: str) -> None:
"""A sort job raised: release the buttons and say why.
:param message: the error's one line.
"""
self._sort_token += 1
self._set_sorting(False)
self.status.setText(message)
def _set_sorting(self, busy: bool) -> None:
"""Hold the sort buttons, with a busy cursor, while a sort runs.
:param busy: whether a sort job is running.
"""
if busy == self._sorting:
return
self._sorting = busy
for button in (self.sort_button, self.auto_button,
self.detect_button, self.teach_button):
button.setEnabled(not busy)
if busy:
self.setCursor(Qt.BusyCursor)
else:
self.unsetCursor()
def _sort_busy(self) -> bool:
"""Whether a "Sort by regex" job is still running."""
return self._sorting
[docs]
def auto_regex(self) -> Optional[str]:
"""Propose a regex, put it in custom_regex, and sort by it.
:returns: the regex, or None when none was found.
"""
pending = self._paths_to_sort()
if pending is None:
return None
paths, by_path, _masks = pending
names = [self._alias(p) for p in paths]
channels = ({self._alias(p): c for p, c in by_path.items()}
if by_path else None)
pattern = cs.infer_regex(names, channels) if names else None
if not pattern:
self.status.setText(tr(
"spaCR found no regex that puts every image in exactly one "
"complete set. Drop the images into channel columns, or use "
"Detect sets."))
return None
self.metadata_field.set_value("custom")
self.custom_edit.setText(pattern)
self.sort_by_regex(pattern)
return pattern
[docs]
def teach(self) -> Optional[str]:
""""Teach me": ask what images are until a regex places all of them.
One image is shown at a time with "Which channel is this?"
(:attr:`ask_label`). From the answers
:func:`spacr.channel_sorting._teach_step` learns what in the names
marks each channel or mask and writes a regex; each new marker is
confirmed with :attr:`ask_marker` (a channel, a mask, or only part
of the field's name). The table fills as it goes, and the regex is
put in custom_regex, where it can be edited. It ends when every
image is placed, or when the user says stop.
:returns: the learned regex, or None when none was learned.
"""
pending = self._paths_to_sort()
if pending is None:
return None
paths, _channels, _masks = pending
if not paths:
self.status.setText(tr(
"Nothing to sort: give a source folder of images, or drop "
"images into the channel columns."))
return None
back = {self._alias(p): p for p in paths}
names = list(back)
answers: Dict[str, object] = {}
confirmed: Dict[str, object] = {}
pattern = None
by_marker: Dict[str, object] = {}
for _round in range(len(names) + 8):
pattern, by_marker, nxt = cs._teach_step(names, answers)
relabelled = False
for marker, label in sorted(by_marker.items()):
if marker in confirmed:
continue
verdict = self.ask_marker(marker, label)
if verdict == "field":
others = [lab for lab in by_marker.values() if lab != label]
for name, answer in list(answers.items()):
if answer == label and others:
answers[name] = others[0]
relabelled = True
break
confirmed[marker] = verdict or label
if verdict and verdict != label:
for name, answer in list(answers.items()):
if answer == label:
answers[name] = verdict
relabelled = True
break
if relabelled:
continue
if pattern:
self.metadata_field.set_value("custom")
self.custom_edit.setText(pattern)
self._taught = (pattern, dict(by_marker))
self.sort_by_regex(pattern, dict(by_marker))
if nxt is None:
break
answer = self.ask_label(back[nxt], self._known_labels(answers))
if answer == "stop" or answer is None:
break
if answer == "skip":
self._skipped.add(back[nxt])
answers[nxt] = answer
if pattern:
self.status.setText(self.status.text() + "\n" + tr(
"Learned regex: {regex}", regex=pattern))
return pattern
@staticmethod
def _known_labels(answers: Dict[str, object]) -> List[object]:
"""The labels answered so far, channels first.
:param answers: ``{name: label}``.
"""
labels = {a for a in answers.values() if isinstance(a, tuple)}
return sorted(labels, key=lambda lab: (lab[0] != "channel", lab[1:]))
def _ask_label(self, path: str, known: List[object]):
"""Show one image and ask which channel it is (a modal box).
:param path: the image.
:param known: labels already used.
:returns: ``("channel", n)``, ``("mask", n, role)``, ``"skip"`` or
``"stop"`` (always ``"stop"`` when no one can answer).
"""
if _headless():
return "stop"
dialog = _TeachQuestion(path, known, self)
dialog.exec()
return dialog.answer
def _ask_marker(self, marker: str, label):
"""Ask what the text that tells images apart means.
:param marker: the text (``""`` for nothing).
:param label: what the answers say it marks.
:returns: the label, another label, or ``"field"`` when the text is
only part of the field's name; headless, the label unchanged.
"""
if _headless():
return label
shown = marker or tr("(nothing)")
kind = (tr("channel {n}", n=label[1]) if label[0] == "channel"
else tr("the {role} mask of channel {n}", role=tr(label[2]),
n=label[1]))
box = QMessageBox(self)
box.setWindowTitle(tr("What does this part of the name mean?"))
box.setText(tr(
"Images whose names have “{marker}” here look like {kind}. "
"Does this part of the name mark a channel or a mask, or is it "
"only part of the field's name?", marker=shown, kind=kind))
keep = box.addButton(tr("It marks {kind}", kind=kind),
QMessageBox.AcceptRole)
box.addButton(tr("It is part of the field's name"),
QMessageBox.RejectRole)
box.exec()
return label if box.clickedButton() is keep else "field"
[docs]
def detect_sets(self) -> bool:
"""Match the columns' files by name, order and size, and check three.
With an empty table and a source folder, the source is sorted by the
regex first, for its channels. The sets are shown three at a time
(:class:`spacr.qt.widgets.channel_sort_dialog.ExampleSetsDialog`)
and kept only when the user says they are right.
:returns: whether the detected sets were kept.
"""
if not self.rows and self._source():
if self.sort_by_regex() is None:
return False
if len([c for c in self._channel_columns() if self._column_files(c)]) < 2:
self.status.setText(tr(
"Detect sets needs at least two channel columns with images: "
"drop them into the columns, or sort by a regex with chanID."))
return False
before = ([list(r) for r in self.rows], list(self.row_keys))
self._rematch()
sets = self._complete_sets(detected=True)
if not sets or not self.confirm_examples(sets):
self.rows, self.row_keys = before
self._refresh()
self.status.setText(tr("The detected sets were not used."))
return False
self.row_keys = [((cs.DETECTED_GROUP, f"{n + 1:06d}"),)
for n in range(len(self.rows))]
self._refresh()
return True
def _ask_examples(self, sets) -> bool:
"""Show three detected sets and ask; False when no one can answer.
:param sets: ``{set key: {channel: path}}``.
"""
if _headless():
return False
from .channel_sort_dialog import ExampleSetsDialog
dialog = ExampleSetsDialog(self._source() or "", sets, None, self)
return dialog.exec() == QDialog.Accepted
def _complete_sets(self, detected: bool = False):
"""``{set key: {channel: image}}`` for the rows with every channel.
:param detected: key every set by its number, as Detect sets does,
even when the rows came from a regex.
"""
channels = self._channel_columns()
rows = [(row, key) for row, key in zip(self.rows, self.row_keys)
if channels and all(row[c] for c in channels)]
use_keys = not detected and rows and all(key is not None
for _r, key in rows)
sets = {}
for number, (row, key) in enumerate(rows):
if not use_keys:
key = ((cs.DETECTED_GROUP, f"{number + 1:06d}"),)
sets[key] = {self._channel_of(c): row[c] for c in channels}
return sets
def _incomplete_rows(self) -> List[int]:
"""The rows missing an image in some channel column."""
channels = self._channel_columns()
return [i for i, row in enumerate(self.rows)
if any(row[c] is None for c in channels)]
def _base_folder(self) -> str:
"""The folder the output goes beside: the source, else the files'.
:returns: a folder, or ``""`` when there is nothing yet.
"""
source = self._source()
if source and os.path.isdir(source):
return source
folders = sorted({os.path.dirname(p) for row in self.rows
for p in row if p})
if not folders:
return ""
common = os.path.commonpath(folders)
if os.path.dirname(common) == common:
common = folders[0]
return common
def _suggest_output(self) -> None:
"""Propose ``<base>/sorted_channels`` until the user types their own."""
if not self._output_auto:
return
base = self._base_folder()
self.output_edit.setText(
cs.unused_folder(base, cs.DEFAULT_DEST_NAME) if base else "")
def _output_edited(self, _text: str) -> None:
"""The user typed an output folder: stop proposing one.
:param _text: the new text.
"""
self._output_auto = False
[docs]
def prepare_plan(self, convert: bool = False) -> Optional[cs.SortPlan]:
"""Build the plan from the table, or None while rows are incomplete.
:param convert: convert RGB images and z-stacks to one plane.
:returns: the plan, which may still carry problems.
"""
if not self._channel_columns() or not self.rows or self._incomplete_rows():
return None
sets = self._complete_sets()
masks: Dict[str, Optional[str]] = {}
roles: Dict[int, str] = {}
problems = []
for index, column in enumerate(self.columns):
if column.kind != "mask":
continue
target = self._column_for_channel(column.of_channel)
if target is None:
problems.append(tr(
"A {role} mask column belongs to channel {n}, which "
"there is not.", role=tr(column.role), n=column.of_channel))
continue
if column.of_channel in roles:
problems.append(tr(
"Channel {n} has two mask columns; a channel's images "
"have one mask each.", n=column.of_channel))
continue
roles[column.of_channel] = column.role
for row in self.rows:
if row[target] and row[index]:
masks[row[target]] = row[index]
base = self._base_folder()
plan = cs.build_plan(base, sets, masks=masks, mask_roles=roles,
dest=self.output_edit.text().strip() or None,
convert=convert)
plan.problems.extend(problems)
return plan
def _ask(self, title: str, text: str) -> bool:
"""Ask a yes/no question; no when no one can answer.
:param title: the box's title.
:param text: the question.
"""
if _headless():
return False
box = QMessageBox(QMessageBox.Question, title, text,
QMessageBox.Yes | QMessageBox.No, self)
box.button(QMessageBox.Yes).setText(tr("Yes"))
box.button(QMessageBox.No).setText(tr("No"))
return box.exec() == QMessageBox.Yes
def _plan_summary(self, plan: cs.SortPlan) -> str:
"""Describe the selected output layout and retain every plan diagnostic."""
masks = sum(1 for row in plan.rows if row.source_mask)
if self._target_layout == "measure":
summary = tr(
"Move {images} image(s) and {masks} mask(s) into {channels} "
"channel folder(s) under {dest}, as {sets} set(s) in "
"Yokogawa naming. Then build channel stacks and any merged "
"arrays supported by the available masks.",
images=len(plan.rows), masks=masks, channels=len(plan.channels),
dest=plan.dest, sets=plan.n_sets)
else:
summary = tr(
"Move {images} image(s) into {dest} using Yokogawa names, "
"with {masks} mask(s) in {masks_dir}. Do not write stacks "
"or merged arrays.", images=len(plan.rows), dest=plan.dest,
masks=masks, masks_dir=os.path.join(plan.dest, "masks"))
lines = [summary]
if plan.mask_roles:
roles = ", ".join(
tr("channel {n}: {role}", n=channel, role=tr(role))
for channel, role in sorted(plan.mask_roles.items()))
lines.append(tr("Mask roles: {roles}", roles=roles))
lines.extend(plan.warnings)
lines.extend(plan.problems)
return "\n".join(lines)
def _on_apply(self) -> None:
"""Plan, confirm and accept -- or say what is missing."""
plan = self.prepare_plan()
if plan is None:
missing = self._incomplete_rows()
self.status.setText(tr(
"{n} row(s) lack an image in some channel column, so they "
"cannot become Yokogawa sets. Change the regex (Sort by "
"regex, Auto regex) or use Detect sets, drop the missing "
"images into their columns, or remove those files.",
n=len(missing)) if missing else tr(
"Add images to at least one channel column first."))
self.custom_edit.setFocus()
return
if plan.convertible and self.mode != "measure" and self.ask(
tr("Split multi-channel files?"), tr(
"{n} image(s) hold more than one plane. Split each into "
"one file per channel, one channel column per plane? The "
"originals are not touched.", n=len(plan.convertible))):
self._split_multichannel()
return
if plan.convertible:
shown = "\n".join(os.path.basename(p) for p in plan.convertible[:8])
if self.ask(tr("Convert RGB images and z-stacks?"), tr(
"{n} image(s) are not a single grey plane:\n{names}\n\n"
"Convert them? RGB images become grey (the mean of their "
"colours) and z-stacks become their maximum projection. "
"The originals are kept in originals/ of the sorted "
"folder.", n=len(plan.convertible), names=shown)):
plan = self.prepare_plan(convert=True)
if not plan.ok:
self.status.setText(self._plan_summary(plan))
return
if not self.ask(tr("Move and merge?") if self._target_layout == "measure"
else tr("Move the images?"), self._plan_summary(plan)):
return
self.plan = plan
self.accept()
def _caption(self, index: int) -> str:
"""A column's header.
:param index: the column.
"""
column = self.columns[index]
if column.kind == "channel":
return tr("Channel {n}", n=self._channel_of(index))
return tr("{role} mask (channel {n})", role=tr(column.role),
n=column.of_channel)
def _rebuild_editors(self) -> None:
"""One small editor per column: its role and channel, and Remove."""
while self._editors_row.count():
widget = self._editors_row.takeAt(0).widget()
if widget is not None:
widget.hide()
widget.setParent(None)
widget.deleteLater()
channels = len(self._channel_columns())
for index, column in enumerate(self.columns):
box = QFrame()
row = QHBoxLayout(box)
row.setContentsMargins(2, 0, 2, 0)
row.addWidget(QLabel(self._caption(index)))
if column.kind == "mask":
role = QComboBox()
captions = {"cell": tr("cell"), "nucleus": tr("nucleus"),
"pathogen": tr("pathogen"),
"organelle": tr("organelle")}
for value in cs.MASK_ROLES:
role.addItem(captions[value], value)
role.setCurrentIndex(max(0, role.findData(column.role)))
role.currentIndexChanged.connect(
lambda _i, c=column, b=role: self._set_role(c, b))
row.addWidget(role)
owner = QComboBox()
for n in range(1, max(channels, 1) + 1):
owner.addItem(tr("of channel {n}", n=n), n)
owner.setCurrentIndex(max(0, owner.findData(column.of_channel)))
owner.currentIndexChanged.connect(
lambda _i, c=column, b=owner: self._set_owner(c, b))
row.addWidget(owner)
remove = QPushButton(tr("Remove"))
remove.setToolTip(tr("Remove this column and its files from the "
"table."))
remove.clicked.connect(
lambda _checked=False, c=column: self.remove_column(
self.columns.index(c)))
row.addWidget(remove)
self._editors_row.addWidget(box)
self._editors_row.addStretch(1)
def _set_role(self, column: _Column, box: QComboBox) -> None:
"""A mask column's role was chosen.
:param column: the column.
:param box: its role box.
"""
column.role = box.currentData()
self._refresh_table()
def _set_owner(self, column: _Column, box: QComboBox) -> None:
"""A mask column's channel was chosen: rematch its masks.
:param column: the column.
:param box: its channel box.
"""
column.of_channel = int(box.currentData())
self._rematch()
def _refresh(self) -> None:
"""Redraw the column editors, the table and the status line."""
self._rebuild_editors()
self._refresh_table()
def _mask_of(self) -> Dict[int, int]:
"""``{channel column: its mask column}``, for moving masks along."""
return {channel: mask for mask, channel in self._partners().items()}
def _swap_slots(self, from_row: int, from_column: int, to_row: int,
to_column: int) -> None:
"""Move one cell to another slot; an occupied slot swaps with it.
The images' masks move with them. Rows left with nothing are dropped.
:param from_row: the dragged cell's row.
:param from_column: its column.
:param to_row: the slot's row; one past the last makes a new row.
:param to_column: the slot's column.
"""
if not (0 <= from_row < len(self.rows)
and 0 <= from_column < len(self.columns)
and 0 <= to_column < len(self.columns) and to_row >= 0):
return
before = len(self.rows)
_swap_cells(self.rows, self._mask_of(), (from_row, from_column),
(to_row, to_column))
self.row_keys.extend([None] * (len(self.rows) - before))
self._drop_empty_rows()
self._refresh_table()
def _move_slots(self, cells, anchor, target) -> int:
"""Move several cells together as a block.
Each cell moves by the offset from ``anchor`` to ``target``; what
they land on swaps back into the slots they left, and images keep
their masks. Rows left with nothing are dropped.
:param cells: ``[[row, column], ...]``.
:param anchor: ``[row, column]`` the drag started on.
:param target: ``[row, column]`` dropped on.
:returns: how many slots moved.
"""
cells = [(int(r), int(c)) for r, c in cells
if 0 <= int(r) < len(self.rows)
and 0 <= int(c) < len(self.columns)]
moves = _block_moves(cells, tuple(anchor), tuple(target),
len(self.columns), self._mask_of())
before = len(self.rows)
_move_block(self.rows, moves)
self.row_keys.extend([None] * (len(self.rows) - before))
self._drop_empty_rows()
self._refresh_table()
return len(moves)
def _set_thumb_size(self, size: int, remember: bool = True) -> None:
"""Resize the table's images live (the Size slider).
:param size: pixels; clamped to :data:`_THUMB_RANGE`.
:param remember: store it in the preferences.
"""
self.thumb_size = _clamp_thumb(size)
if remember:
_save_thumb_pref(self.thumb_size)
self._size_cells()
self._quality_changed("")
def _size_cells(self) -> None:
"""Size the rows and columns for the view and the thumbnail size.
In the image views a column is never narrower than its heading.
"""
if self.delegate.view == "text":
self._fit_text_cells()
return
header = self.table.horizontalHeader()
for column in range(self.table.columnCount()):
self.table.setColumnWidth(column, max(
self.thumb_size + 24, header.sectionSizeHint(column)))
self.table.verticalHeader().setDefaultSectionSize(self.thumb_size + 8)
def _fit_text_cells(self) -> None:
"""Fit the text view's columns and rows to their contents.
What ``resizeColumnsToContents`` and ``resizeRowsToContents`` give,
measured on one cell per column: the one whose name is widest. When
the font's digits are all one width, names that differ only in their
digits are one width too, so one of each is measured: a plate's
thousands of names are a handful of shapes. Qt asks every cell for
its size hint, and a model cell answers in Python: 0.6 s for a
3,136-image plate.
"""
table = self.table
model = table.model()
if not model.rowCount() or not model.columnCount():
return
from PySide6.QtGui import QFontMetricsF
metrics = table.fontMetrics()
tabular = len({QFontMetricsF(table.font()).horizontalAdvance(d)
for d in "0123456789"}) == 1
known, self._known_shapes = self._known_shapes, None
if not tabular:
known = None
if known is None or len(known) != model.columnCount():
known = _name_shapes(self.rows, model.columnCount(), tabular)
header = table.horizontalHeader()
grid = 1 if table.showGrid() else 0
for column in range(model.columnCount()):
shapes = known[column]
widest, row_of = -1, 0
for row in shapes:
path = model.path(row, column)
width = metrics.horizontalAdvance(
os.path.basename(path) if path else "\u2014")
if width > widest:
widest, row_of = width, row
hint = table.sizeHintForIndex(model.index(row_of, column)).width()
table.setColumnWidth(column, max(hint + grid,
header.sectionSizeHint(column)))
rows = table.verticalHeader()
rows.setDefaultSectionSize(max(table.sizeHintForRow(0),
rows.sectionSizeHint(0)))
def _clear_slots(self, cells) -> int:
"""Empty cells (the × or Delete); their images' masks go too.
:param cells: ``[[row, column], ...]``.
:returns: how many files left the table.
"""
mask_of = self._mask_of()
removed = sum(1 for row, column in cells
if _clear_cell(self.rows, mask_of, (row, column)))
self._drop_empty_rows()
self._refresh_table()
return removed
def _set_view(self, view: str, remember: bool = True) -> None:
"""Show cells as ``text``, ``image`` or ``both``.
:param view: one of :data:`_VIEWS`.
:param remember: store it in the preferences.
"""
view = view if view in _VIEWS else "text"
self.delegate.view = view
self.color_button.setEnabled(view == "both")
self.size_slider.setEnabled(view != "text")
if remember:
_save_view_prefs(view, self.delegate.text_color.name())
self._refresh_table()
def _set_text_color(self, color, remember: bool = False) -> None:
"""The colour of the names written over the images.
:param color: a QColor or a colour name such as ``#ffff00``.
:param remember: store it in the preferences.
"""
color = QColor(color)
if not color.isValid():
color = QColor("white")
self.delegate.text_color = color
backdrop = _legible_backdrop(
color, self.palette().color(QPalette.Window))
self.color_button.setStyleSheet(
f"QPushButton {{ color: {color.name()}; border: none; "
f"background: {backdrop}; border-radius: 3px; "
f"padding: 2px 4px; }}"
f"QPushButton:disabled {{ color: {color.name()}80; }}")
if remember:
_save_view_prefs(self.delegate.view, color.name())
self.table.viewport().update()
def _pick_text_color(self) -> None:
"""Choose the overlay colour in a colour dialog (not headless)."""
if _headless():
return
color = pick_colour(self, self.delegate.text_color, tr("Text colour"))
if color.isValid():
self._set_text_color(color, remember=True)
def _quality_changed(self, _quality: str) -> None:
"""Discard display samples and asynchronously reload at the new quality.
:param _quality: the newly persisted quality key.
"""
self._thumb_generation += 1
for worker in self._thumb_workers:
worker.stop()
self._thumb_pending.clear()
self._thumb_signatures.clear()
self._thumb_requested.clear()
self.delegate.pixmaps.clear()
self.table.viewport().update()
self._load_thumbnails()
@staticmethod
def _thumbnail_signature(path):
"""Identify the current file contents for cache invalidation.
:param path: image or mask file.
:returns: modification time and size, or None for an unavailable file.
"""
try:
stat = os.stat(path)
return stat.st_mtime_ns, stat.st_size
except OSError:
return None
def _load_thumbnails(self, *_args) -> None:
"""Read only visible rows off-thread, with one bounded batch at a time."""
if (self._thumb_closed or self.delegate.view == "text"
or not self.rows or self._thumb_workers):
return
from ..hidpi import device_ratio
ratio = device_ratio(self.table)
previous_ratio, self._thumb_render_ratio = self._thumb_render_ratio, ratio
if previous_ratio is not None and previous_ratio != ratio:
self._quality_changed("")
return
viewport = self.table.viewport()
first = max(0, self.table.rowAt(0))
last = self.table.rowAt(max(0, viewport.height() - 1))
if last < 0:
last = len(self.rows) - 1
first_col = max(0, self.table.columnAt(0))
last_col = self.table.columnAt(max(0, viewport.width() - 1))
if last_col < 0:
last_col = len(self.columns) - 1
paths = list(dict.fromkeys(
p for row in self.rows[first:last + 1]
for p in row[first_col:last_col + 1] if p))
wanted = []
for path in paths:
signature = self._thumbnail_signature(path)
if (path in self.delegate.pixmaps
and self._thumb_signatures.get(path) == signature):
self.delegate.pixmaps.move_to_end(path)
continue
self.delegate.pixmaps.pop(path, None)
self._thumb_requested[path] = signature
wanted.append(path)
if not wanted:
return
from .channel_sort_dialog import _ThumbWorker
self._thumb_pending.update(wanted)
worker = _ThumbWorker("", wanted, None, self)
worker.generation = self._thumb_generation
worker.ready.connect(self._take_thumbnail)
worker.finished.connect(self._thumbnail_worker_finished)
self._thumb_workers.append(worker)
worker.start()
def _thumbnail_worker_finished(self) -> None:
"""Release completed workers and service the latest quality/viewport."""
worker = self.sender()
if worker in self._thumb_workers:
self._thumb_workers.remove(worker)
if worker is not None:
worker.deleteLater()
if (not self._thumb_closed and worker is not None
and worker.generation != self._thumb_generation):
QTimer.singleShot(0, self._load_thumbnails)
def _take_thumbnail(self, path: str, image, _mask) -> None:
"""Cache a current thumbnail, with a 64 MiB/512-entry display budget.
:param path: the image or mask file.
:param image: its 2-D uint8 source-sampled thumbnail, or None.
:param _mask: paired overlay, unused in this separate-column table.
"""
worker = self.sender()
if (self._thumb_closed or (worker is not None
and worker.generation != self._thumb_generation)):
return
self._thumb_pending.discard(path)
if image is None:
return
signature = self._thumb_requested.get(path)
if signature != self._thumbnail_signature(path):
return
array = np.ascontiguousarray(image)
qimage = QImage(array.data, array.shape[1], array.shape[0],
array.shape[1], QImage.Format_Grayscale8)
pixmap = QPixmap.fromImage(qimage.copy())
from ..hidpi import device_ratio, scaled_for
display_side = self.thumb_size + 24
display_cap = max(1, round(display_side * device_ratio(self.table)))
if max(pixmap.width(), pixmap.height()) > display_cap:
pixmap = scaled_for(pixmap, self.table, display_side)
self.delegate.pixmaps[path] = pixmap
self._thumb_signatures[path] = signature
cache = self.delegate.pixmaps
budget = 64 * 1024 * 1024
total = sum(p.width() * p.height() * 4 for p in cache.values())
while len(cache) > 512 or (total > budget and len(cache) > 1):
old, pixmap = cache.popitem(last=False)
total -= pixmap.width() * pixmap.height() * 4
self._thumb_signatures.pop(old, None)
self._thumb_requested.pop(old, None)
self.table.viewport().update()
def _stop_thumbs(self) -> None:
"""Stop polling and end thumbnail threads before destroying the popup."""
self._thumb_closed = True
self._thumb_poll.stop()
self._thumb_kick.stop()
self._sort_token += 1
self._sort_jobs.shutdown()
for worker in self._thumb_workers:
if worker.isRunning():
worker.stop()
worker.wait(5000)
def _refresh_table(self) -> None:
"""Fill the table from the rows, empty cells flagged.
The drop hint lives in the table while it is empty; the new-channel
strip appears only once there is something beside it.
"""
channels = set(self._channel_columns())
self.table.show_cells(
self.rows, [self._caption(i) for i in range(len(self.columns))],
channels)
self._size_cells()
self._known_shapes = None
if self.delegate.view != "text":
self._thumb_kick.start()
self.new_zone.setVisible(not self.table._is_empty())
self.table.viewport().update()
missing = len(self._incomplete_rows())
lines = [tr("{rows} row(s) in {channels} channel(s) and {masks} "
"mask column(s); {missing} row(s) incomplete.",
rows=len(self.rows), channels=len(channels),
masks=len(self.columns) - len(channels), missing=missing)]
if self._note:
lines.append(self._note)
self.status.setText("\n".join(lines))
def _browse(self) -> None:
"""Pick the source folder."""
folder = QFileDialog.getExistingDirectory(
self, tr("Pick the source folder"), self._source() or os.getcwd())
if folder:
self.source_edit.setText(folder)