"""Embeddings — a label-free vector for every object, beside the measured panel.
The upstream Cell-DINO, OpenPhenom and SubCell models expose Python
interfaces; this screen runs them without code. It also accepts an explicitly
mapped, digest-verified local official Cell-DINO checkpoint.
The screen runs the same embedding engine, :mod:`spacr.embeddings`, that the
Python API exposes, without requiring any Python.
**What it is for.** Giving every object a vector that no label went into.
The measured panel says how big a cell is and how bright each stain was;
an embedding describes what the cell looks like, in a few hundred numbers a
pretrained backbone produces without being told what to look for. It
complements the panel rather than replacing it: a phenotype that no
measured column captures can still be represented in the vector.
**What it needs.** Crops, and a choice about the channels. The crops come
from the row at the top of the screen, which is :mod:`spacr.crop_loader`:
either a ``measurements.db``, narrowed to one object class, one plate or
any condition its crop table supports, or a folder of crop PNGs. Whichever
it is, the pixels are read the way every other screen reads them --
:func:`spacr.crops.resolve_crop_source` decides between the crops already
under ``data/`` and cutting them from ``merged/*.npy``, and says which it
chose. Nothing has to be prepared first: a plate that has been through
Measure with crop output on is ready.
**What it produces.** One row per object and one column per dimension,
named ``emb_c<channel>_<dimension>`` under the per-channel policy and
``emb_<dimension>`` under projection, with the model-zoo entry that made
them -- backbone, policy and the checksum of the weights on this machine.
Two runs' dimension 17 are the same number and not the same thing unless
all three match, which is why the entry is part of the result rather than
a log line. The preview shows the first few dimensions of the first fifty
objects. The alpha Save for Similar crops action stores the complete matrix
in the database that supplied the loaded objects, so Annotate can search
those vectors. Folder and programmatically supplied crops have no database
object identity and cannot use this action.
**What to do next.** Treat the columns as a feature source, not as a
result. They are consumed exactly as the measured panel is -- a reduction
to see whether the objects separate, a regression against the perturbation,
a retrieval to find the objects most like one you picked. The screen repeats
the corresponding warning under the table: a single dimension is not a
phenotype, and these numbers can only be interpreted through an analysis
that uses the whole vector.
All the arithmetic is in :mod:`spacr.embeddings` and none is here. This module
is the surface, and three of the engine's decisions shape it:
**The channel policy is the first control on the screen.** How channels are
combined must not be decided implicitly by whatever the backbone expects.
Per-channel encoding keeps the channel in every column name and costs N
forward passes; projection mixes the channels into three and costs one. Those
produce columns of identical width and dtype that mean entirely different
things, so the screen makes the choice explicit and says what each one costs.
**A single dimension is not a phenotype, and the screen states this under the
table.** The table shows the first few dimensions so the user can see that
the embedding ran; the caption says they are not interpretable individually.
Showing `emb_c2_017 = 0.43` without that caption invites reading one
dimension as a phenotype.
**The encoder is identified with the checksum of its weights.** The run
summary carries the model-zoo entry from
:func:`spacr.embeddings.encoder_entry` -- backbone, policy and the digest of
the weights on this machine -- because two runs' dimension 17 are the same
number and not the same thing unless all three match.
The embedding runs off the GUI thread through
:class:`spacr.qt.job_runner.JobRunner`, like every other compute in the Qt
layer: a plate of crops through a backbone is minutes, and ``threaded=False``
runs the identical code inline so a test drives the shipped path.
:func:`register` is **not** called at import; read its docstring.
"""
from __future__ import annotations
import logging
import os
import threading
from typing import Any, Dict, List, Optional
import numpy as np
import pandas as pd
from PySide6.QtCore import Qt, QTimer, Signal
from PySide6.QtWidgets import (
QCheckBox, QComboBox, QDialog, QDialogButtonBox, QDoubleSpinBox, QFileDialog, QFormLayout,
QHBoxLayout, QLabel, QLineEdit, QProgressBar, QPushButton, QSpinBox,
QStackedWidget, QTableWidget, QVBoxLayout, QWidget,
)
from ...crop_loader import (CROP_SOURCE_DATABASE, CROP_SOURCES,
DEFAULT_CROP_LIMIT, DEFAULT_PAGE_SIZE)
from ..app_catalog import declared_app
from ..i18n import tr
from ..widgets.measurements_example import install_test_data_button
from ..job_runner import JobRunner
from ..theme import SPACING
from ..widgets.collapsible_splitter import FoldSection
from ..widgets.sortable_table import (
SORT_KEY_ROLE, _retire_table_cells, _spare_table_item, install_sorting,
table_item,
)
from .app_screen import ModuleHeader
LOG = logging.getLogger("spacr.qt.screens.embeddings")
__all__ = ["EmbeddingsScreen", "make_embeddings_screen", "register",
"APP_KEY", "APP_NAME", "APP_DESCRIPTION", "APP_INTRO",
"APP_CLI_NOTE", "APP_NAME_TRANSLATIONS", "PREVIEW_DIMENSIONS",
"DIMENSION_CAVEAT", "NOTHING_LOADED", "NOTHING_LOADED_HINT",
"MAX_CROP_LIMIT"]
#: The registry key. Chosen once and never renamed.
APP_KEY = "embeddings"
_CELL_DINO_CHOICE_KEY = "embeddings/cell_dino_checkpoint"
#: How many dimensions the preview table shows. Enough to see that the run
#: produced numbers, few enough that nobody mistakes the table for the result
#: -- the result is the whole vector and it is not readable.
PREVIEW_DIMENSIONS = 8
#: Shown under the table, every time. 386's own warning, in the one place a
#: user reading a dimension will actually be looking.
DIMENSION_CAVEAT = (
"A single dimension is not a phenotype and means nothing on its own. "
"These are only interpretable as a vector — through a reduction, a "
"retrieval, or an attribution."
)
#: The screen's opening state, before anything has been asked for. It is a
#: heading rather than an empty table, because an empty table reads as a run
#: that produced nothing.
NOTHING_LOADED = "No crops loaded"
#: And what to do about it. Every other state of this panel replaces both
#: lines, so the panel is never showing a stale instruction beside a result.
NOTHING_LOADED_HINT = (
"Choose a measurements database or a folder of crop PNGs above, then "
"press Load crops. A database can be narrowed to one object class, one "
"plate, or any condition its columns support."
)
#: The most crops the panel will let anybody ask for in one load. Not a
#: limit of the loader, which has none: it is the point past which a stack
#: of 96x96x3 crops stops fitting beside a backbone on an ordinary card, and
#: a spin box that offers a number nothing can hold is a trap.
MAX_CROP_LIMIT = 200_000
class _CropState(QWidget):
"""The panel shown where the preview goes until there is a preview.
NOT :class:`spacr.qt.widgets.empty_state.EmptyState`, and the reason is
the only reason: that widget takes its title and subtitle at
construction and offers no way to change them, while every sentence this
panel shows is different -- what to do first, what was loaded, and the
refusal a particular query earned. It carries the same object names, so
it is the same two type styles on the page.
"""
def __init__(self, parent=None):
"""Build the two labels and open on the nothing-loaded state."""
super().__init__(parent)
self.setObjectName("EmbeddingsCropState")
outer = QVBoxLayout(self)
outer.setContentsMargins(SPACING["xl"], SPACING["xl"],
SPACING["xl"], SPACING["xl"])
outer.setSpacing(SPACING["sm"])
outer.addStretch(1)
self._title = QLabel(NOTHING_LOADED, self)
self._title.setObjectName("TitleHeading")
self._title.setAlignment(Qt.AlignCenter)
outer.addWidget(self._title)
self._detail = QLabel(NOTHING_LOADED_HINT, self)
self._detail.setObjectName("SubtitleSmall")
self._detail.setAlignment(Qt.AlignCenter)
self._detail.setWordWrap(True)
outer.addWidget(self._detail)
outer.addStretch(1)
def say(self, title: str, detail: str) -> None:
"""Replace both lines at once.
Both, always, so a heading can never be left standing over the
explanation of a different state -- which is how "No crops" ends up
under a sentence telling the reader to press Embed.
:param title: the heading.
:param detail: the sentence under it.
"""
self._title.setText(str(title))
self._detail.setText(str(detail))
def title(self) -> str:
"""The heading currently shown."""
return self._title.text()
def detail(self) -> str:
"""The sentence currently shown under it."""
return self._detail.text()
[docs]
class EmbeddingsScreen(QWidget):
"""Pick a crop source and a channel policy, and embed every object.
The crop loader above the encoder controls is the screen's entry point:
:mod:`spacr.crop_loader` plans the selection, this screen pages through
it on a worker thread, and each finished load calls :meth:`set_crops`.
:param parent: the usual Qt parent.
:param threaded: ``False`` runs the embedding and the crop load inline,
emitting the same signals in the same order, so a test drives the
screen synchronously without the behaviour diverging.
"""
#: A page of crops has been read: ``(done, total)``. Emitted from the
#: loading thread, so it is queued onto the GUI thread by Qt; nothing
#: connected to it may assume it runs where it was emitted.
crops_progress = Signal(int, int)
#: A load finished, carrying how many crops it put on the screen.
crops_loaded = Signal(int)
_dino_progress = Signal(int, int, int, float)
_classifier_progress = Signal(int, int, int, float, float)
def __init__(self, parent=None, *, threaded: bool = True):
"""Build the screen: the controls above, the preview table below."""
super().__init__(parent)
self.setObjectName("EmbeddingsScreen")
self._frame: Optional[pd.DataFrame] = None
self._result = None
self._result_encoder = None
self._crop_identity = None
self._result_identity = None
self._crop_generation = 0
self._scale_record: dict = {}
self._crop_record: Dict[str, Any] = {}
self._plan = None
self._stop = threading.Event()
self._loading = False
self._read_path = ""
self._demo_active = False
self._subcell_channels: Optional[tuple[int, int, int, int]] = None
self._cell_dino_factory = ""
self._cell_dino_channels: Optional[tuple[int, ...]] = None
self._cell_dino_checkpoint = ""
self._cell_dino_digest = ""
self._policy_before_subcell: Optional[str] = None
self._jobs = JobRunner(self, threaded=threaded, app_key=APP_KEY)
self._jobs.job_failed.connect(self._on_job_failed)
self.crops_progress.connect(self._on_crops_progress)
self._dino_jobs = JobRunner(self, threaded=threaded, app_key=APP_KEY)
self._dino_jobs.job_failed.connect(self._on_dino_failed)
self._dino_jobs.job_finished.connect(self._on_dino_finished)
self._dino_progress.connect(self._on_dino_progress)
self._dino_running = False
self._dino_closed = False
self._dino_generation = 0
self._classifier_jobs = JobRunner(self, threaded=threaded, app_key=APP_KEY)
self._classifier_jobs.job_failed.connect(self._on_classifier_failed)
self._classifier_jobs.job_finished.connect(self._on_classifier_finished)
self._classifier_progress.connect(self._on_classifier_progress)
self._classifier_running = False
self._classifier_generation = 0
self._classifier_crop_generation = 0
self._classifier_stop = threading.Event()
self._classifier_apply_jobs = JobRunner(self, threaded=threaded, app_key=APP_KEY)
self._classifier_apply_jobs.job_failed.connect(self._on_classifier_application_failed)
self._classifier_apply_jobs.job_finished.connect(self._on_classifier_application_finished)
self._classifier_apply_jobs.busy_changed.connect(self._sync_classifier_application_controls)
self._classifier_apply_running = False
self._classifier_apply_delivery = False
self._classifier_apply_generation = 0
self._classifier_apply_crop_generation = 0
self._classifier_application_result = None
self._classifier_retire_timer = QTimer(self)
self._classifier_retire_timer.setInterval(25)
self._classifier_retire_timer.timeout.connect(self._sync_classifier_application_controls)
outer = QVBoxLayout(self)
outer.setContentsMargins(SPACING["md"], SPACING["md"],
SPACING["md"], SPACING["md"])
outer.setSpacing(SPACING["sm"])
head = QHBoxLayout()
head.setSpacing(SPACING["sm"])
self._header = ModuleHeader(
APP_NAME,
description=APP_DESCRIPTION,
instruction="Load crops, choose how the channels are encoded, "
"then embed.",
)
head.addWidget(self._header)
self._source = QLabel("no crops loaded", self)
self._source.setObjectName("EmbeddingsSourceLabel")
head.addWidget(self._source, 1)
outer.addLayout(head)
picker = QHBoxLayout()
picker.setSpacing(SPACING["sm"])
picker.addWidget(QLabel("Crops from:", self))
self._where = QComboBox(self)
self._where.setObjectName("EmbeddingsCropSourcePicker")
for value, label in CROP_SOURCES:
self._where.addItem(label, value)
self._where.setToolTip(
"A measurements database lets you choose which objects to embed. "
"A crop folder takes every PNG in it.")
self._where.currentIndexChanged.connect(self._on_source_changed)
picker.addWidget(self._where)
self._path = QLineEdit(self)
self._path.setObjectName("EmbeddingsCropPath")
self._path.setPlaceholderText(
"path to measurements.db, or to a *_png crop folder")
self._path.editingFinished.connect(self._on_path_changed)
picker.addWidget(self._path, 1)
self._browse = QPushButton("Browse…", self)
self._browse.setObjectName("EmbeddingsBrowseButton")
self._browse.setToolTip("Find the database or the crop folder")
self._browse.clicked.connect(self.choose_path)
picker.addWidget(self._browse)
example = install_test_data_button(
self, picker, lambda _folder, db: self._use_example_database(db),
say=lambda text: self._status.setText(text))
example.setObjectName("EmbeddingsTestDataButton")
self._demo_button = QPushButton(tr("Load synthetic demo"), self)
self._demo_button.setObjectName("EmbeddingsDemoButton")
self._demo_button.setToolTip(tr(
"Load 24 generated shapes with labels. DINOv2 runs on the CPU; "
"pretrained weights download only if not cached."))
self._demo_button.clicked.connect(self._load_demo_crops)
picker.addWidget(self._demo_button)
self._cell_example_button = QPushButton(tr("Load cell example"), self)
self._cell_example_button.setObjectName("EmbeddingsCellExampleButton")
self._cell_example_button.setToolTip(tr(
"Load the example plate's annotated single-cell crops with their "
"labels and run DINOv2 on the CPU. The plate downloads once, "
"about 280 MB."))
self._cell_example_button.clicked.connect(self._load_cell_example)
picker.addWidget(self._cell_example_button)
from ..preferences import _apply_alpha_widgets
_apply_alpha_widgets(self._demo_button)
_apply_alpha_widgets(self._cell_example_button)
outer.addLayout(picker)
selection = QHBoxLayout()
selection.setSpacing(SPACING["sm"])
self._class_label = QLabel("Object:", self)
selection.addWidget(self._class_label)
self._object = QComboBox(self)
self._object.setObjectName("EmbeddingsObjectClassPicker")
self._object.setToolTip(
"Which object class to embed. It picks the label column in "
"png_list and the mask plane a streamed crop is cut by, so "
"asking for nuclei out of a cell crop table returns nothing "
"rather than returning cells.")
selection.addWidget(self._object)
self._plate_label = QLabel("Plate:", self)
selection.addWidget(self._plate_label)
self._plate = QComboBox(self)
self._plate.setObjectName("EmbeddingsPlatePicker")
self._plate.addItem("every plate", "")
self._plate.setToolTip(
"One plate, or all of them. A scale is estimated per run, so a "
"run over two plates scales both by one number.")
selection.addWidget(self._plate)
self._filter_label = QLabel("Where:", self)
selection.addWidget(self._filter_label)
self._filter = QLineEdit(self)
self._filter.setObjectName("EmbeddingsCropFilter")
self._filter.setPlaceholderText("any condition, e.g. cell_area > 500")
self._filter.setToolTip(
"A condition over the crop table's own columns, in SQL. The "
"database is opened read-only, so nothing typed here can change "
"it.")
selection.addWidget(self._filter, 1)
self._limit_label = QLabel("At most:", self)
selection.addWidget(self._limit_label)
self._limit = QSpinBox(self)
self._limit.setObjectName("EmbeddingsCropLimit")
self._limit.setRange(1, MAX_CROP_LIMIT)
self._limit.setValue(DEFAULT_CROP_LIMIT)
self._limit.setToolTip(
"How many of the matching crops to load. The rest are counted "
"and named, not silently dropped.")
selection.addWidget(self._limit)
self._load = QPushButton("Load crops", self)
self._load.setObjectName("EmbeddingsLoadButton")
self._load.setToolTip("Choose a database or a folder first")
self._load.setEnabled(False)
self._load.clicked.connect(self.load_crops)
selection.addWidget(self._load)
outer.addLayout(selection)
controls = QHBoxLayout()
controls.setSpacing(SPACING["sm"])
controls.addWidget(QLabel("Channels:", self))
self._policy = QComboBox(self)
self._policy.setObjectName("EmbeddingsPolicyPicker")
self._policy.addItem("Per channel (one pass per stain)", "per_channel")
self._policy.addItem("Project to three (one pass)", "project")
self._policy.setToolTip(
"Encoding runs separately for every stain and names it in the "
"column. Projection mixes them into three and is faster.")
controls.addWidget(self._policy)
controls.addWidget(QLabel("Backbone:", self))
self._backbone = QComboBox(self)
self._backbone.setObjectName("EmbeddingsBackbonePicker")
self._backbone.setEditable(True)
self._backbone.setToolTip(
"Any encoder timm can resolve. The model-zoo entry records which "
"one, with the checksum of its weights: two runs' dimension 17 "
"are the same number and not the same thing unless the backbone, "
"the weights and the channel policy all match.")
controls.addWidget(self._backbone)
self._add_foundation_picker(controls)
controls.addWidget(QLabel("Batch:", self))
self._batch = QSpinBox(self)
self._batch.setObjectName("EmbeddingsBatchSize")
self._batch.setRange(1, 4096)
self._batch.setValue(64)
self._batch.setToolTip(
"Crops per forward pass. Lower it if the card runs out of "
"memory; it changes speed and nothing else.")
controls.addWidget(self._batch)
controls.addStretch(1)
self._run = QPushButton("Embed", self)
self._run.setObjectName("PrimaryButton")
self._run.setEnabled(False)
self._run.setToolTip("Load crops first")
self._run.clicked.connect(self.embed)
controls.addWidget(self._run)
self._add_well_mil_button(controls)
self._add_dino_button(controls)
self._add_use_for_pickers(controls)
outer.addLayout(controls)
outer.addWidget(self._dino_in_use)
self._save_similar = QPushButton(tr("Save for Similar crops"), self)
self._save_similar.setObjectName("EmbeddingsSaveForSimilarity")
self._save_similar.setToolTip(tr(
"Save these vectors in their source database for Annotate's "
"Like this search. Load crops from a database first."))
self._save_similar.setEnabled(False)
self._save_similar.clicked.connect(self._save_for_similarity)
outer.addWidget(self._save_similar)
from ..preferences import _apply_alpha_widgets
_apply_alpha_widgets(self._save_similar)
self._table = install_sorting(QTableWidget(0, 0, self))
self._table.setObjectName("EmbeddingsPreviewTable")
self._table.setAlternatingRowColors(True)
self._table.setEditTriggers(QTableWidget.NoEditTriggers)
self._state = _CropState(self)
self._pages = QStackedWidget(self)
self._pages.setObjectName("EmbeddingsPreviewPages")
self._pages.addWidget(self._state)
self._pages.addWidget(self._table)
self._pages.setCurrentWidget(self._state)
self._pages_section = FoldSection(
self._pages, "Crop preview",
persist_key="embeddings/Crop preview")
outer.addWidget(self._pages_section, 1)
self._caveat = QLabel(DIMENSION_CAVEAT, self)
self._caveat.setObjectName("EmbeddingsCaveatLabel")
self._caveat.setWordWrap(True)
outer.addWidget(self._caveat)
self._status = QLabel("", self)
self._status.setObjectName("EmbeddingsStatusLabel")
self._status.setWordWrap(True)
outer.addWidget(self._status)
self._fill_backbones()
self._fill_object_classes(())
self._on_source_changed()
from .settings_model import retarget_field_tooltips
retarget_field_tooltips(self)
def _add_foundation_picker(self, controls) -> None:
"""The Foundation model picker, alpha-gated, beside the backbone.
A single-cell foundation model chosen here replaces the backbone for
the run, whether the picker is shown or not; None keeps the backbone.
"""
from ...embeddings import _FOUNDATION_MODELS
from ..preferences import _apply_alpha_widgets
label = QLabel(tr("Foundation model:"), self)
label.setObjectName("EmbeddingsFoundationLabel")
controls.addWidget(label)
self._foundation = QComboBox(self)
self._foundation.setObjectName("EmbeddingsFoundationPicker")
self._foundation.addItem(tr("None (use the backbone)"), "")
self._foundation.addItem(tr("DINOv2 (Cell-DINO when installed)"),
"dino_auto")
self._foundation.setItemData(1, tr(
"Runs official Cell-DINO when its checkpoint is installed and "
"its planes are mapped; otherwise the open DINOv2 ViT-S/14. The "
"line below the controls says which one is in use."),
Qt.ItemDataRole.ToolTipRole)
for name, info in _FOUNDATION_MODELS.items():
self._foundation.addItem(tr(info["label"]), name)
self._foundation.setToolTip(tr(
"A model trained on microscopy rather than photographs. "
"OpenPhenom and ChAda-ViT take any number of stains. SubCell's "
"two-plane model takes DNA then protein; its four-plane model "
"needs explicit microtubules, ER, DNA and protein mapping. "
"Other weights download once. Cell-DINO requires a local "
"official checkpoint, declared SHA-256 and explicit model-plane "
"mapping. "
"Default None (use the "
"backbone)."))
controls.addWidget(self._foundation)
self._subcell_button = QPushButton(tr("Channels…"), self)
self._subcell_button.setObjectName("EmbeddingsSubCellChannelsButton")
self._subcell_button.setToolTip(tr(
"Map four crop channels to microtubules, ER, DNA and protein "
"before running SubCell's four-plane model."))
self._subcell_button.clicked.connect(self._choose_subcell_channels)
controls.addWidget(self._subcell_button)
self._cell_dino_button = QPushButton(tr("Checkpoint and channels…"), self)
self._cell_dino_button.setObjectName("EmbeddingsCellDinoConfigButton")
self._cell_dino_button.clicked.connect(self._choose_cell_dino)
controls.addWidget(self._cell_dino_button)
self._dino_in_use = QLabel("", self)
self._dino_in_use.setObjectName("EmbeddingsDinoInUseLabel")
self._dino_in_use.setWordWrap(True)
self._foundation.currentIndexChanged.connect(
self._sync_subcell_controls)
self._sync_subcell_controls()
_apply_alpha_widgets(label)
_apply_alpha_widgets(self._foundation)
def _sync_subcell_controls(self) -> None:
"""Show the four-plane choice and keep its policy caption truthful."""
from ..preferences import _get_show_alpha_features
selected = self._foundation.currentData() == "subcell_rybg"
auto = self._foundation.currentData() == "dino_auto"
cell_dino = auto or self._foundation.currentData() == "cell_dino"
project = self._policy.findData("project")
if selected or (cell_dino and not auto):
if self._policy_before_subcell is None:
self._policy_before_subcell = str(self._policy.currentData())
self._policy.setCurrentIndex(project)
self._policy.setItemText(
project, tr("Four mapped planes (one pass)") if selected else
tr("Mapped model planes (one pass)"))
self._policy.setEnabled(False)
else:
self._policy.setItemText(
project, tr("Project to three (one pass)"))
self._policy.setEnabled(True)
if self._policy_before_subcell is not None:
previous = self._policy.findData(self._policy_before_subcell)
self._policy.setCurrentIndex(max(previous, 0))
self._policy_before_subcell = None
crops = getattr(self, "_crops", None)
count = 0 if crops is None else int(crops.shape[-1])
self._subcell_button.setEnabled(selected and count >= 4)
if selected and count < 4:
self._subcell_button.setToolTip(tr(
"Load crops with at least four channels before mapping "
"SubCell's microtubules, ER, DNA and protein planes."))
else:
self._subcell_button.setToolTip(tr(
"Map four crop channels to microtubules, ER, DNA and protein "
"before running SubCell's four-plane model."))
alpha_on = _get_show_alpha_features()
self._subcell_button.setVisible(selected and alpha_on)
self._subcell_button.setProperty(
"_spacr_alpha_hid", bool(selected and not alpha_on))
self._cell_dino_button.setEnabled(cell_dino and count >= 1)
self._cell_dino_button.setVisible(cell_dino and alpha_on)
self._cell_dino_button.setProperty(
"_spacr_alpha_hid", bool(cell_dino and not alpha_on))
self._dino_in_use.setText(self._dino_sentence(
self._dino_note(verify=False)) if auto else "")
self._dino_in_use.setVisible(auto and alpha_on)
self._dino_in_use.setProperty(
"_spacr_alpha_hid", bool(auto and not alpha_on))
def _dino_note(self, verify: bool = True):
"""Decide between installed Cell-DINO and open DINOv2 for these crops."""
from ...embeddings import CHANNEL_PROJECT, EmbeddingSpec, _dino_choice
crops = getattr(self, "_crops", None)
available = 0 if crops is None else int(crops.shape[-1])
batch = int(self._batch.value()) if hasattr(self, "_batch") else 64
device = "cpu" if self._demo_active else None
if self._cell_dino_checkpoint and not self._cell_dino_mapping_error(
self._cell_dino_factory, self._cell_dino_channels, available,
self._cell_dino_checkpoint, self._cell_dino_digest):
return EmbeddingSpec(
backbone="cell_dino", channel_policy=CHANNEL_PROJECT,
channels=self._cell_dino_channels, batch_size=batch,
device=device, cell_dino_factory=self._cell_dino_factory,
checkpoint_path=self._cell_dino_checkpoint,
checkpoint_sha256=self._cell_dino_digest), {
"model": "cell_dino", "factory": self._cell_dino_factory,
"path": self._cell_dino_checkpoint}
from ..preferences import _settings
remembered = str(_settings().value(_CELL_DINO_CHOICE_KEY, "") or "")
return _dino_choice(
self._cell_dino_channels, available,
channel_policy=str(self._policy.currentData()),
batch_size=batch, device=device, verify=verify,
preferred=self._cell_dino_checkpoint or remembered)
def _dino_sentence(self, choice) -> str:
"""Say which DINO model runs and, for DINOv2, why not Cell-DINO."""
spec, note = choice
if note["model"] == "cell_dino":
digest = (spec.checkpoint_sha256 or "")[:12]
return tr(
"In use: official Cell-DINO ({factory}), checkpoint {name}"
"{digest}.").format(
factory=note["factory"],
name=os.path.basename(note["path"]),
digest=(" · " + digest + "…") if digest else "")
if note["reason"] == "unmapped":
why = tr(
"Cell-DINO checkpoint {name} is installed, but its {planes} "
"input planes are not mapped to crop channels. Map them with "
"{button}").format(button=tr("Checkpoint and channels…"),
name=os.path.basename(note["path"]),
planes=note["planes"])
elif note["reason"] == "unreadable":
why = tr(
"{path} is not a readable official Cell-DINO model state.").format(
path=note["path"])
else:
why = tr(
"No official Cell-DINO checkpoint is installed. Meta grants "
"access by request; put the file in {folder} to use it.").format(
folder=note["folder"])
return tr("In use: open DINOv2 ViT-S/14 (general-image weights).") + (
" " + why)
def _subcell_mapping_error(
self, channels: tuple[object, ...] | None,
available: int) -> str:
"""Explain an incomplete, repeated or stale four-plane mapping."""
if not isinstance(channels, tuple) or len(channels) != 4 or any(
not isinstance(channel, int) or isinstance(channel, bool)
for channel in channels):
return tr("Choose a crop channel for each SubCell plane: "
"microtubules, ER, DNA and protein.")
if len(set(channels)) != 4:
return tr("Choose four different crop channels for SubCell.")
if any(channel < 0 or channel >= available for channel in channels):
return tr("A selected SubCell channel is outside the loaded "
"crops. Open Channels… and map the four planes again.")
return ""
def _subcell_channels_dialog(self) -> QDialog:
"""Build four initially unchosen selectors in official R/Y/B/G order."""
crops = getattr(self, "_crops", None)
available = 0 if crops is None else int(crops.shape[-1])
dialog = QDialog(self)
dialog.setObjectName("EmbeddingsSubCellChannelsDialog")
dialog.setWindowTitle(tr("SubCell four-plane channels"))
layout = QVBoxLayout(dialog)
guidance = QLabel(tr(
"Assign four different crop channels in SubCell's official "
"order. No stain identity is guessed from channel position."),
dialog)
guidance.setWordWrap(True)
layout.addWidget(guidance)
form = QFormLayout()
microtubules = QComboBox(dialog)
microtubules.setObjectName("EmbeddingsSubCellMicrotubulesChannel")
er = QComboBox(dialog)
er.setObjectName("EmbeddingsSubCellErChannel")
dna = QComboBox(dialog)
dna.setObjectName("EmbeddingsSubCellDnaChannel")
protein = QComboBox(dialog)
protein.setObjectName("EmbeddingsSubCellProteinChannel")
selectors = (microtubules, er, dna, protein)
for role, (label, selector) in enumerate((
(tr("Microtubules (R)"), microtubules),
(tr("ER (Y)"), er),
(tr("DNA (B)"), dna),
(tr("Protein (G)"), protein),
)):
selector.addItem(tr("Choose channel…"), None)
for index in range(available):
selector.addItem(tr("Channel {position} (index {index})").format(
position=index + 1, index=index), index)
if self._subcell_channels is not None:
matching = selector.findData(self._subcell_channels[role])
selector.setCurrentIndex(max(matching, 0))
form.addRow(label, selector)
layout.addLayout(form)
problem = QLabel("", dialog)
problem.setObjectName("EmbeddingsSubCellChannelsProblem")
problem.setWordWrap(True)
layout.addWidget(problem)
actions = QDialogButtonBox(
QDialogButtonBox.Ok | QDialogButtonBox.Cancel, dialog)
layout.addWidget(actions)
def accept_mapping() -> None:
"""Store only a complete four-index choice, then close."""
chosen = tuple(selector.currentData() for selector in selectors)
current = getattr(self, "_crops", None)
count = 0 if current is None else int(current.shape[-1])
reason = self._subcell_mapping_error(chosen, count)
if reason:
problem.setText(reason)
return
self._subcell_channels = chosen
self._status.setText(tr(
"SubCell channel mapping saved in microtubules, ER, DNA, "
"protein order."))
dialog.accept()
actions.accepted.connect(accept_mapping)
actions.rejected.connect(dialog.reject)
return dialog
def _choose_subcell_channels(self) -> None:
"""Open the mapping form without retaining its Qt wrappers."""
dialog = self._subcell_channels_dialog()
try:
dialog.exec()
finally:
dialog.deleteLater()
def _cell_dino_mapping_error(
self, factory: str, channels: tuple[object, ...] | None,
available: int, path: str, digest: str) -> str:
"""Explain a missing checkpoint or ambiguous model-input order."""
from ...embeddings import _CELL_DINO_FACTORIES
if factory not in _CELL_DINO_FACTORIES:
return tr("Choose an official Cell-DINO model factory.")
width = _CELL_DINO_FACTORIES[factory][0]
if not width and isinstance(channels, tuple) and channels:
width = len(channels)
if (not isinstance(channels, tuple) or len(channels) != width
or any(not isinstance(channel, int)
or isinstance(channel, bool) for channel in channels)):
return tr("Choose a crop channel for every Cell-DINO input plane.")
if len(set(channels)) != width:
return tr("Choose different crop channels for every Cell-DINO plane.")
if any(channel < 0 or channel >= available for channel in channels):
return tr("A Cell-DINO channel is outside the loaded crops. "
"Reopen the mapping and choose valid planes.")
if not path or not os.path.isfile(os.path.expanduser(path)):
return tr("Choose a local official Cell-DINO checkpoint file.")
import re
if not re.fullmatch(r"[0-9a-fA-F]{64}", digest):
return tr("Enter the checkpoint's 64-character SHA-256 digest.")
return ""
def _cell_dino_dialog(self) -> QDialog:
"""Configure a pinned official factory, local weights and input order."""
from ...embeddings import (_CELL_DINO_FACTORIES, _CHANNEL_ADAPTIVE,
_cell_dino_checkpoints)
crops = getattr(self, "_crops", None)
available = 0 if crops is None else int(crops.shape[-1])
dialog = QDialog(self)
dialog.setObjectName("EmbeddingsCellDinoDialog")
dialog.setWindowTitle(tr("Cell-DINO checkpoint and model planes"))
layout = QVBoxLayout(dialog)
guidance = QLabel(tr(
"Use an official checkpoint obtained from Meta. Choose its exact "
"factory and SHA-256, then map crop channels in model-input "
"order. No stain identity is inferred."), dialog)
guidance.setWordWrap(True)
layout.addWidget(guidance)
form = QFormLayout()
factory = QComboBox(dialog)
factory.setObjectName("EmbeddingsCellDinoFactory")
factory.addItem(tr("Choose model…"), "")
factory.addItem(tr("HPA ViT-L/16 · four planes · 224 pixels"),
"cell_dino_hpa_vitl16")
factory.addItem(tr("HPA ViT-L/14 · four planes · 518 pixels"),
"cell_dino_hpa_vitl14")
factory.addItem(tr("Cell Painting ViT-S/8 · five planes · 128 pixels"),
"cell_dino_cp_vits8")
factory.addItem(tr("Channel-Adaptive DINO ViT-L/16 · any planes · "
"224 pixels"), _CHANNEL_ADAPTIVE)
factory.setCurrentIndex(max(factory.findData(self._cell_dino_factory), 0))
installed = QComboBox(dialog)
installed.setObjectName("EmbeddingsCellDinoInstalled")
installed.addItem(tr("Choose installed checkpoint…"), "")
for entry in _cell_dino_checkpoints():
model = (factory.itemText(factory.findData(entry["factory"]))
if entry.get("factory") else
tr("not a readable Cell-DINO model state"))
installed.addItem(os.path.basename(entry["path"]) + " · " + model,
entry["path"])
if not entry.get("factory"):
installed.model().item(installed.count() - 1).setEnabled(False)
installed.setCurrentIndex(max(installed.findData(os.path.realpath(
os.path.expanduser(self._cell_dino_checkpoint)))
if self._cell_dino_checkpoint else 0, 0))
form.addRow(tr("Installed checkpoint"), installed)
form.addRow(tr("Official model"), factory)
path = QLineEdit(dialog)
path.setObjectName("EmbeddingsCellDinoCheckpointPath")
path.setText(self._cell_dino_checkpoint)
browse = QPushButton(tr("Browse…"), dialog)
browse.setObjectName("EmbeddingsCellDinoBrowse")
def choose_file() -> None:
"""Choose local weights without downloading or accepting terms."""
chosen, _ = QFileDialog.getOpenFileName(
dialog, tr("Choose official Cell-DINO checkpoint"),
path.text(), tr("PyTorch checkpoints (*.pth *.pt);;All files (*)"))
if chosen:
path.setText(chosen)
browse.clicked.connect(choose_file)
path_row = QHBoxLayout()
path_row.addWidget(path)
path_row.addWidget(browse)
form.addRow(tr("Local checkpoint"), path_row)
digest = QLineEdit(dialog)
digest.setObjectName("EmbeddingsCellDinoDigest")
digest.setText(self._cell_dino_digest)
form.addRow(tr("Expected SHA-256"), digest)
def use_installed() -> None:
"""Fill the file and its detected model; the digest is read on OK."""
chosen = str(installed.currentData() or "")
entry = next((item for item in _cell_dino_checkpoints()
if item["path"] == chosen), {})
if entry.get("factory"):
path.setText(chosen)
factory.setCurrentIndex(factory.findData(entry["factory"]))
digest.clear()
installed.currentIndexChanged.connect(use_installed)
planes = []
plane1 = QComboBox(dialog)
plane1.setObjectName("EmbeddingsCellDinoPlane1")
plane2 = QComboBox(dialog)
plane2.setObjectName("EmbeddingsCellDinoPlane2")
plane3 = QComboBox(dialog)
plane3.setObjectName("EmbeddingsCellDinoPlane3")
plane4 = QComboBox(dialog)
plane4.setObjectName("EmbeddingsCellDinoPlane4")
plane5 = QComboBox(dialog)
plane5.setObjectName("EmbeddingsCellDinoPlane5")
planes.extend((plane1, plane2, plane3, plane4, plane5))
for index, selector in enumerate(planes):
selector.addItem(tr("Choose channel…"), None)
for channel in range(available):
selector.addItem(tr("Channel {position} (index {index})").format(
position=channel + 1, index=channel), channel)
if self._cell_dino_channels is not None and (
index < len(self._cell_dino_channels)):
selector.setCurrentIndex(max(selector.findData(
self._cell_dino_channels[index]), 0))
form.addRow(tr("Input plane {number}").format(number=index + 1),
selector)
adaptive_note = QLabel(tr(
"Channel-Adaptive DINO encodes every crop channel on its own; "
"no mapping is needed."), dialog)
adaptive_note.setObjectName("EmbeddingsCellDinoAdaptiveNote")
adaptive_note.setWordWrap(True)
def sync_width() -> None:
"""Show the planes the chosen factory takes; none when adaptive."""
adaptive = factory.currentData() == _CHANNEL_ADAPTIVE
for selector in planes:
visible = not adaptive and (
selector is not plane5
or factory.currentData() == "cell_dino_cp_vits8")
form.labelForField(selector).setVisible(visible)
selector.setVisible(visible)
adaptive_note.setVisible(adaptive)
factory.currentIndexChanged.connect(sync_width)
sync_width()
layout.addLayout(form)
layout.addWidget(adaptive_note)
problem = QLabel("", dialog)
problem.setObjectName("EmbeddingsCellDinoProblem")
problem.setWordWrap(True)
layout.addWidget(problem)
actions = QDialogButtonBox(
QDialogButtonBox.Ok | QDialogButtonBox.Cancel, dialog)
layout.addWidget(actions)
def accept_mapping() -> None:
"""Keep only plain configuration values after the form closes."""
from ...embeddings import _cell_dino_entry
selected = str(factory.currentData() or "")
width = _CELL_DINO_FACTORIES.get(selected, (0, 0))[0]
chosen = tuple(plane.currentData() for plane in planes[:width])
current = getattr(self, "_crops", None)
count = 0 if current is None else int(current.shape[-1])
if selected == _CHANNEL_ADAPTIVE:
chosen = tuple(range(count))
if not digest.text().strip() and path.text().strip():
digest.setText(_cell_dino_entry(
path.text().strip(), verify=True).get("sha256", ""))
reason = self._cell_dino_mapping_error(
selected, chosen, count, path.text().strip(),
digest.text().strip())
if reason:
problem.setText(reason)
return
self._cell_dino_factory = selected
self._cell_dino_channels = chosen
self._cell_dino_checkpoint = path.text().strip()
self._cell_dino_digest = digest.text().strip().lower()
from ..preferences import _settings
_settings().setValue(_CELL_DINO_CHOICE_KEY, os.path.realpath(
os.path.expanduser(self._cell_dino_checkpoint)))
self._status.setText(tr(
"Cell-DINO checkpoint and input-plane order saved. "
"The file digest will be checked before model loading."))
dialog.accept()
self._sync_subcell_controls()
actions.accepted.connect(accept_mapping)
actions.rejected.connect(dialog.reject)
return dialog
def _choose_cell_dino(self) -> None:
"""Open the Cell-DINO form and dispose its widgets on close."""
if not self._cell_dino_checkpoint:
from ...embeddings import _find_cell_dino
from ..preferences import _settings
crops = getattr(self, "_crops", None)
found = _find_cell_dino(
verify=True,
available=0 if crops is None else int(crops.shape[-1]),
preferred=str(_settings().value(_CELL_DINO_CHOICE_KEY, "") or ""))
if found.get("factory"):
self._cell_dino_factory = found["factory"]
self._cell_dino_checkpoint = found["path"]
self._cell_dino_digest = found.get("sha256", "")
dialog = self._cell_dino_dialog()
try:
dialog.exec()
finally:
dialog.deleteLater()
self._sync_subcell_controls()
def _add_use_for_pickers(self, controls) -> None:
"""Alpha controls that score the embedding and feed it onward.
A label table gives the retrieval scorecard of the chosen backbone;
the picker then runs an image UMAP or a classifier on the vectors.
"""
from ..preferences import _apply_alpha_widgets
self._labels: Dict[str, str] = {}
self._labels_button = QPushButton(tr("Labels…"), self)
self._labels_button.setObjectName("EmbeddingsLabelsButton")
self._labels_button.setToolTip(tr(
"Choose a table with one row per embedded crop, in crop order, "
"and a 'label' column (else the first column). The backbone is "
"scored on it: kNN accuracy, mean average precision and "
"precision at 10 against chance, and the classifier uses the "
"same labels. Default no labels."))
self._labels_button.clicked.connect(lambda: self._choose_labels())
controls.addWidget(self._labels_button)
label = QLabel(tr("Use for:"), self)
label.setObjectName("EmbeddingsUseForLabel")
controls.addWidget(label)
self._use_for = QComboBox(self)
self._use_for.setObjectName("EmbeddingsUseForPicker")
self._use_for.addItem(tr("Image UMAP"), "umap")
self._use_for.addItem(tr("Classifier"), "classifier")
self._use_for.setToolTip(tr(
"What to do with the embedding of the chosen backbone or "
"foundation model. Image UMAP maps every crop to two "
"dimensions; Classifier cross-validates a logistic regression "
"on the labels. Default Image UMAP."))
controls.addWidget(self._use_for)
self._use_for_run = QPushButton(tr("Run"), self)
self._use_for_run.setObjectName("EmbeddingsUseForRun")
self._use_for_run.setToolTip(tr(
"Run the chosen use on the last embedding. Default off until "
"an embedding exists."))
self._use_for_run.clicked.connect(lambda: self._use_embeddings())
controls.addWidget(self._use_for_run)
for widget in (self._labels_button, label, self._use_for,
self._use_for_run):
_apply_alpha_widgets(widget)
def _choose_labels(self, path: str = "") -> str:
"""Read a label table and score the last embedding against it.
:param path: the table; asks for one when empty.
:returns: the table used, or ``''`` when nothing was chosen or the
table does not fit the embedding.
"""
from ...tabular import read_table
if not path:
path, _filter = QFileDialog.getOpenFileName(
self, tr("Choose a label table"), "",
tr("Tables (*.csv *.tsv *.parquet *.feather *.xlsx)"))
if not path:
return ""
try:
frame = read_table(str(path), report=None)
column = "label" if "label" in frame.columns else frame.columns[0]
values = frame[column].tolist()
except (OSError, ValueError, IndexError, ImportError) as error:
self._status.setText(tr(
"Could not read the label table: {error}", error=str(error)))
return ""
embedded = getattr(self, "_frame", None)
if embedded is not None and len(values) != len(embedded):
self._status.setText(tr(
"The label table has {rows} rows but {crops} crops were "
"embedded.").format(rows=len(values), crops=len(embedded)))
return ""
self._labels = {str(i): ("" if pd.isna(v) else str(v))
for i, v in enumerate(values)}
if embedded is not None:
self._show_scorecard()
return str(path)
def _show_scorecard(self) -> None:
"""Measure the encoder's retrieval scorecard on the labels."""
from ...embeddings import _scored_encoder_entry
spec = getattr(self._result, "spec", None) or self.spec()
entry = _scored_encoder_entry(spec, self._frame, self._labels)
self._entry = entry
card = entry.metrics
if not card:
self._status.setText(tr(
"The labels cannot be scored: two classes with labelled "
"crops are needed."))
return
self._status.setText(tr(
"Encoder {name}: kNN accuracy {knn:.2f}, mAP {map:.2f} (chance "
"{chance:.2f}), precision at 10 {prec:.2f}, over {n} crops in "
"{classes} classes.").format(
name=entry.name, knn=card["knn_accuracy"], map=card["map"],
chance=card["chance_map"], prec=card["precision_at_k"],
n=int(card["n"]), classes=int(card["classes"])))
def _use_embeddings(self) -> None:
"""Run the picked use (image UMAP or classifier) on the embedding."""
frame = getattr(self, "_frame", None)
if frame is None:
self._status.setText(tr("Embed the crops first."))
return
use = str(self._use_for.currentData())
labels = dict(self._labels)
if use == "classifier" and not labels:
self._status.setText(tr("Choose labels first."))
return
def work():
"""Map or classify off the GUI thread."""
from ... import embeddings as emb
if use == "umap":
return use, emb._embedding_umap(frame)
return use, emb._embedding_classifier_scorecard(frame, labels)
self._status.setText(tr("Running on the embedding…"))
self._jobs.submit(work, self._on_used)
def _on_used(self, result) -> None:
"""Show the UMAP coordinates or the classifier scorecard."""
use, answer = result
spec = getattr(self._result, "spec", None) or self.spec()
if use == "umap":
self._umap = answer
self._fill_preview(answer)
self._status.setText(tr(
"Image UMAP of {n} crops from {name}.").format(
n=len(answer), name=spec.backbone))
return
self._status.setText(tr(
"Classifier on {name}: accuracy {acc:.2f} ± {sd:.2f} over "
"{folds} folds (chance {chance:.2f}), {n} crops in {classes} "
"classes.").format(
name=spec.backbone, acc=answer["accuracy"],
sd=answer["accuracy_sd"], folds=int(answer["folds"]),
chance=answer["chance"], n=int(answer["n"]),
classes=int(answer["classes"])))
def _add_well_mil_button(self, controls) -> None:
"""The alpha button that learns which cells carry a well label."""
from ..preferences import _apply_alpha_widgets
self._well_mil = QPushButton(tr("Learn from well labels…"), self)
self._well_mil.setObjectName("EmbeddingsWellMilButton")
self._well_mil.setToolTip(tr(
"Choose a per-cell table with a 'well' column, a 'well_label' "
"column (1 for treated or knockout wells, 0 for controls) and "
"embedding or numeric feature columns. An attention model learns "
"from the well labels alone which cells carry the phenotype. "
"Two tables are written beside the input: each cell's attention "
"and each well's probability. Runs on the CPU. Default 4-fold "
"cross-validation over wells."))
self._well_mil.clicked.connect(lambda: self._learn_from_well_labels())
controls.addWidget(self._well_mil)
_apply_alpha_widgets(self._well_mil)
def _mil_column_form(self, columns):
"""A form for choosing the well, label and feature columns.
:param columns: the table's column names.
:returns: ``(dialog, well, label, features)``: the dialog, the two
combo boxes and the feature list; the defaults are the ``wellID``
or ``well`` column, ``well_label`` and the ``emb_`` columns.
"""
from PySide6.QtWidgets import (QAbstractItemView, QComboBox,
QDialog, QDialogButtonBox,
QFormLayout, QListWidget)
columns = [str(c) for c in columns]
dialog = QDialog(self)
dialog.setObjectName("EmbeddingsWellMilForm")
dialog.setWindowTitle(tr("Learn from well labels"))
form = QFormLayout(dialog)
well = QComboBox(dialog)
well.setObjectName("EmbeddingsWellMilWellColumn")
well.addItems(columns)
for name in ("wellID", "well"):
if name in columns:
well.setCurrentText(name)
break
label = QComboBox(dialog)
label.setObjectName("EmbeddingsWellMilLabelColumn")
label.addItems(columns)
if "well_label" in columns:
label.setCurrentText("well_label")
features = QListWidget(dialog)
features.setObjectName("EmbeddingsWellMilFeatureColumns")
features.setSelectionMode(QAbstractItemView.MultiSelection)
features.addItems(columns)
wanted = [c for c in columns if c.startswith("emb_")]
for row, name in enumerate(columns):
if name in wanted:
features.item(row).setSelected(True)
form.addRow(tr("Well column"), well)
form.addRow(tr("Label column"), label)
form.addRow(tr("Feature columns"), features)
buttons = QDialogButtonBox(
QDialogButtonBox.Ok | QDialogButtonBox.Cancel, dialog)
buttons.accepted.connect(dialog.accept)
buttons.rejected.connect(dialog.reject)
form.addRow(buttons)
return dialog, well, label, features
def _ask_mil_columns(self, columns):
"""Ask for the columns; ``None`` when the form is dismissed."""
from PySide6.QtWidgets import QDialog
dialog, well, label, features = self._mil_column_form(columns)
if dialog.exec() != QDialog.Accepted:
return None
chosen = [i.text() for i in features.selectedItems()]
return {"well_column": well.currentText(),
"label_column": label.currentText(),
"feature_columns": chosen or None}
def _learn_from_well_labels(self, path: str = "",
columns=None, labels: str = "") -> str:
"""Train the well-label attention model on a per-cell table.
The table is read, a form asks for the well, label and feature
columns, the model is cross-validated and then fitted on every well
in the background, and ``<name>_mil_cells.csv`` and
``<name>_mil_wells.csv`` are written beside it. The scorecard and a
montage of the highest-attention crops are shown on the screen.
A ``measurements.db`` is read through its stored crop embeddings,
joined to object ids and wells; the well labels then come from a
second table with a well column and a ``well_label`` column.
:param path: the table or database; asks for one when empty.
:param columns: ``well_column``, ``label_column`` and
``feature_columns`` for the model; asks with a form when ``None``.
:param labels: the well-label table for a database; asks for one
when empty.
:returns: the table used, or ``''`` when a dialog was dismissed or
the database has no stored embeddings.
"""
from ...tabular import read_table
if not path:
path, _filter = QFileDialog.getOpenFileName(
self, tr("Choose a per-cell table with well labels"), "",
tr("Tables or measurements.db (*.csv *.tsv *.parquet "
"*.feather *.xlsx *.db)"))
if not path:
return ""
path = str(path)
if path.lower().endswith(".db"):
from ...embeddings import _mil_frame_from_db
if not labels:
labels, _filter = QFileDialog.getOpenFileName(
self, tr("Choose the well-label table for the stored "
"embeddings"), os.path.dirname(path),
tr("Tables (*.csv *.tsv *.parquet *.feather *.xlsx)"))
if not labels:
return ""
try:
frame = _mil_frame_from_db(path, str(labels))
except ValueError as exc:
self._status.setText(str(exc))
return ""
else:
frame = read_table(path, report=None)
if columns is None:
columns = self._ask_mil_columns(frame.columns)
if columns is None:
return ""
columns = dict(columns)
self._status.setText(tr("Learning from well labels…"))
def work():
"""Score and fit off the GUI thread."""
from ...embeddings import _mil_from_table
from ...tabular import write_table
cells, wells, card = _mil_from_table(frame, **columns)
stem = os.path.splitext(path)[0]
write_table(cells, stem + "_mil_cells.csv")
write_table(wells, stem + "_mil_wells.csv")
return card, cells
self._jobs.submit(work, self._on_well_mil_done)
return path
def _on_well_mil_done(self, result) -> None:
"""Show the scorecard and the highest-attention crops."""
card, cells = result
self._mil_card = dict(card)
self._status.setText(tr(
"Well-label model: held-out well AUROC {mil:.2f} (mean-feature "
"baseline {mean:.2f}) over {wells} wells. Cell attention and "
"well probabilities were written beside the table.").format(
mil=card["mil_auroc"], mean=card["mean_auroc"],
wells=int(card["wells"])))
self._show_mil_result(card, cells)
def _show_mil_result(self, card, cells, top: int = 24) -> None:
"""A window with the scorecard and a montage of the top crops.
The montage holds the ``top`` cells with the highest attention that
have a ``png_path``; without crop paths only the scorecard shows.
"""
from PySide6.QtCore import QPoint
from PySide6.QtGui import QPainter, QPixmap
from PySide6.QtWidgets import (QDialog, QLabel, QTableWidget,
QVBoxLayout)
from ..hidpi import scaled_for
dialog = QDialog(self)
dialog.setObjectName("EmbeddingsWellMilResult")
dialog.setWindowTitle(tr("Well-label model"))
layout = QVBoxLayout(dialog)
table = QTableWidget(len(card), 2, dialog)
table.setObjectName("EmbeddingsWellMilScorecard")
table.setHorizontalHeaderLabels([tr("Measure"), tr("Value")])
for row, (key, value) in enumerate(card.items()):
table.setItem(row, 0, table_item(str(key)))
text = (f"{value:.3f}" if isinstance(value, float)
else str(value))
table.setItem(row, 1, table_item(text))
install_sorting(table)
layout.addWidget(table)
montage = QLabel(dialog)
montage.setObjectName("EmbeddingsWellMilMontage")
self._mil_montage_count = 0
if "png_path" in cells.columns and "mil_attention" in cells.columns:
ranked = cells.sort_values("mil_attention", ascending=False)
tiles = []
for crop in ranked["png_path"].dropna().astype(str):
pixmap = QPixmap(crop)
if not pixmap.isNull():
tiles.append(scaled_for(pixmap, montage, 64))
if len(tiles) >= top:
break
if tiles:
across = min(len(tiles), 8)
down = -(-len(tiles) // across)
ratio = self.devicePixelRatioF()
sheet = QPixmap(int(across * 66 * ratio),
int(down * 66 * ratio))
sheet.setDevicePixelRatio(ratio)
sheet.fill(Qt.black)
painter = QPainter(sheet)
for i, tile in enumerate(tiles):
painter.drawPixmap(
QPoint((i % across) * 66 + 1, (i // across) * 66 + 1),
tile)
painter.end()
montage.setPixmap(sheet)
self._mil_montage_count = len(tiles)
if not self._mil_montage_count:
montage.setText(tr(
"No crop images: add a 'png_path' column to see the cells "
"with the highest attention."))
layout.addWidget(montage)
self._mil_result = dialog
dialog.show()
def _add_dino_button(self, controls) -> None:
"""Group the alpha checkpoint picker, epoch control and training progress."""
from ..preferences import _apply_alpha_widgets
self._dino_controls = QWidget(self)
self._dino_controls.setObjectName("EmbeddingsDinoControls")
layout = QVBoxLayout(self._dino_controls)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(SPACING["sm"])
buttons = QHBoxLayout()
self._dino_checkpoint = QPushButton(
tr("Use own DINO checkpoint…"), self._dino_controls)
self._dino_checkpoint.setObjectName("EmbeddingsDinoCheckpointButton")
self._dino_checkpoint.clicked.connect(self._choose_dino_checkpoint)
buttons.addWidget(self._dino_checkpoint)
self._dino = QPushButton(tr("Pretrain on these crops…"), self._dino_controls)
self._dino.setObjectName("EmbeddingsDinoPretrainButton")
self._dino.setToolTip(tr(
"Self-supervised (DINO) pretraining of a ResNet-18 on the loaded "
"crops, no labels needed, under the chosen channel policy. Choose "
"a checkpoint file; it is saved after every epoch, and choosing "
"the same file again resumes the run. When it finishes the "
"checkpoint is offered in the Foundation model picker. Slow "
"without a GPU. Default 20 epochs from random weights at 64 "
"px."))
self._dino.clicked.connect(
lambda: self._pretrain_dino(epochs=self._dino_epochs.value()))
buttons.addWidget(self._dino)
layout.addLayout(buttons)
settings = QHBoxLayout()
label = QLabel(tr("Epochs:"), self._dino_controls)
self._dino_epochs = QSpinBox(self._dino_controls)
self._dino_epochs.setObjectName("EmbeddingsDinoEpochs")
self._dino_epochs.setRange(1, 10000)
self._dino_epochs.setValue(20)
self._dino_epochs.setAccessibleName(tr("Epochs"))
self._dino_epochs.setToolTip(tr(
"Train up to this total number of epochs, including epochs "
"already saved in the checkpoint."))
label.setBuddy(self._dino_epochs)
settings.addWidget(label)
settings.addWidget(self._dino_epochs)
self._dino_progress_bar = QProgressBar(self._dino_controls)
self._dino_progress_bar.setObjectName("EmbeddingsDinoProgress")
self._dino_progress_bar.setAccessibleName(tr("Pretraining progress"))
self._dino_progress_bar.setRange(0, 20)
self._dino_progress_bar.setValue(0)
self._dino_progress_bar.setTextVisible(False)
settings.addWidget(self._dino_progress_bar)
layout.addLayout(settings)
self._dino_progress_label = QLabel(tr("Pretraining progress"), self._dino_controls)
self._dino_progress_label.setObjectName("EmbeddingsDinoProgressCaption")
self._dino_progress_label.setWordWrap(True)
self._dino_progress_label.setTextFormat(Qt.PlainText)
self._dino_progress_label.setBuddy(self._dino_progress_bar)
layout.addWidget(self._dino_progress_label)
classifier = QHBoxLayout()
self._classifier_train = QPushButton(tr("Train classifier…"), self._dino_controls)
self._classifier_train.setObjectName("EmbeddingsDinoClassifierButton")
self._classifier_train.clicked.connect(self._choose_dino_classifier)
classifier.addWidget(self._classifier_train)
self._classifier_cancel = QPushButton(tr("Stop classifier"), self._dino_controls)
self._classifier_cancel.setObjectName("EmbeddingsDinoClassifierStop")
self._classifier_cancel.setEnabled(False)
self._classifier_cancel.clicked.connect(self._cancel_dino_classifier)
classifier.addWidget(self._classifier_cancel)
self._classifier_progress_bar = QProgressBar(self._dino_controls)
self._classifier_progress_bar.setObjectName("EmbeddingsDinoClassifierProgress")
self._classifier_progress_bar.setAccessibleName(tr("Classifier training progress"))
self._classifier_progress_bar.setRange(0, 20)
self._classifier_progress_bar.setValue(0)
self._classifier_progress_bar.setTextVisible(False)
layout.addLayout(classifier)
layout.addWidget(self._classifier_progress_bar)
self._classifier_progress_label = QLabel(tr("Classifier training progress"), self._dino_controls)
self._classifier_progress_label.setObjectName("EmbeddingsDinoClassifierProgressCaption")
self._classifier_progress_label.setWordWrap(True)
self._classifier_progress_label.setTextFormat(Qt.PlainText)
self._classifier_progress_label.setBuddy(self._classifier_progress_bar)
layout.addWidget(self._classifier_progress_label)
application = QHBoxLayout()
self._classifier_apply = QPushButton(tr("Apply saved classifier…"), self._dino_controls)
self._classifier_apply.setObjectName("EmbeddingsDinoClassifierApply")
self._classifier_apply.clicked.connect(lambda: self._apply_dino_classifier())
application.addWidget(self._classifier_apply)
self._classifier_apply_cancel = QPushButton(tr("Cancel classifier result"), self._dino_controls)
self._classifier_apply_cancel.setObjectName("EmbeddingsDinoClassifierApplyCancel")
self._classifier_apply_cancel.setToolTip(tr(
"Discard this application result. The model may finish its current work; "
"it is not forcibly stopped."))
self._classifier_apply_cancel.setEnabled(False)
self._classifier_apply_cancel.clicked.connect(self._cancel_classifier_application)
application.addWidget(self._classifier_apply_cancel)
self._classifier_results = QPushButton(tr("Classifier results…"), self._dino_controls)
self._classifier_results.setObjectName("EmbeddingsDinoClassifierResults")
self._classifier_results.setEnabled(False)
self._classifier_results.clicked.connect(self._show_classifier_application)
application.addWidget(self._classifier_results)
layout.addLayout(application)
controls.addWidget(self._dino_controls)
_apply_alpha_widgets(self._dino_controls)
def _select_dino_checkpoint(self, path: str) -> None:
"""Select a checkpoint path without loading it, reusing an existing entry."""
from ...embeddings import _DINO_PREFIX
canonical = os.path.normcase(os.path.realpath(path))
index = -1
for candidate in range(self._foundation.count()):
data = str(self._foundation.itemData(candidate) or "")
if data.startswith(_DINO_PREFIX) and os.path.normcase(
os.path.realpath(data[len(_DINO_PREFIX):])) == canonical:
index = candidate
break
if index < 0:
self._foundation.addItem(
tr("Own DINO: {name}").format(name=os.path.basename(path)),
_DINO_PREFIX + path)
index = self._foundation.count() - 1
self._foundation.setCurrentIndex(index)
def _choose_dino_checkpoint(self) -> None:
"""Choose an existing spaCR DINO checkpoint path for the next Embed."""
if (self._dino_running or self._classifier_running or self._dino_closed
or self._classifier_apply_running or self._classifier_apply_jobs.active_jobs()):
return
path, _filter = QFileDialog.getOpenFileName(
self, tr("Choose a spaCR DINO checkpoint"), "",
tr("PyTorch checkpoints (*.pt)"))
if not path:
return
if not os.path.isfile(path):
self._status.setText(tr(
"Choose an existing spaCR DINO checkpoint file."))
return
self._select_dino_checkpoint(str(path))
self._status.setText(tr("Checkpoint selected. Press Embed to use it."))
def _set_dino_running(self, running: bool) -> None:
"""Keep training controls unavailable while this screen owns a DINO run."""
self._dino_running = bool(running)
self._sync_classifier_application_controls()
def _set_training_progress_text(self, bar, label, text: str) -> None:
"""Keep progress text visible outside the slim bar and accessible on it."""
bar.setFormat(text)
bar.setAccessibleDescription(text)
label.setText(text)
def _on_dino_progress(self, generation: int, epoch: int,
total: int, loss: float) -> None:
"""Apply reports on the GUI thread only while their run is current."""
if (self._dino_closed or not self._dino_running
or generation != self._dino_generation):
return
self._dino_progress_bar.setRange(0, max(1, total))
self._dino_progress_bar.setValue(epoch)
self._set_training_progress_text(
self._dino_progress_bar, self._dino_progress_label, tr(
"Epoch {epoch}/{total} — loss {loss:.3f}").format(
epoch=epoch, total=total, loss=loss))
def _on_dino_failed(self, message: str) -> None:
"""Restore this run's controls and retain its error instead of late progress."""
if self._dino_closed:
return
self._set_dino_running(False)
self._set_training_progress_text(
self._dino_progress_bar, self._dino_progress_label, tr("Pretraining failed."))
self._status.setText(str(message))
def _on_dino_finished(self, ok: bool) -> None:
"""Restore controls after cancellation without clearing a newer pending run."""
if (ok or self._dino_closed or not self._dino_running
or self._dino_jobs.pending_jobs()):
return
self._set_dino_running(False)
self._set_training_progress_text(
self._dino_progress_bar, self._dino_progress_label, tr("Pretraining stopped."))
self._status.setText(tr("Pretraining stopped."))
def _pretrain_dino(self, path: str = "", *, epochs: int = 20) -> str:
"""Pretrain a backbone on the loaded crops and offer it as an encoder.
Training runs in the background and writes a resumable checkpoint to
``path``; when it ends the checkpoint is added to the Foundation
model picker and chosen, so the next Embed uses it.
:param path: the checkpoint file; asks for one when empty.
:param epochs: epochs to train up to.
:returns: the checkpoint path, or ``''`` when nothing was started.
"""
if (self._dino_running or self._classifier_running or self._dino_closed
or self._classifier_apply_running or self._classifier_apply_jobs.active_jobs()):
return ""
crops = getattr(self, "_crops", None)
if crops is None:
self._status.setText(tr("Load crops first."))
return ""
if not path:
path, _filter = QFileDialog.getSaveFileName(
self, tr("Save the pretrained backbone as"), "",
tr("PyTorch checkpoints (*.pt)"))
if not path:
return ""
path = str(path)
policy = str(self._policy.currentData())
self._dino_generation += 1
generation = self._dino_generation
emit = self._dino_progress.emit
self._set_dino_running(True)
self._dino_progress_bar.setRange(0, max(1, int(epochs)))
self._dino_progress_bar.setValue(0)
self._set_training_progress_text(
self._dino_progress_bar, self._dino_progress_label, tr("Starting pretraining…"))
self._status.setText(tr("Pretraining on {n} crops…").format(
n=crops.shape[0]))
def _progress(epoch, total, loss):
"""Relay backend reports without touching widgets on the worker thread."""
try:
emit(generation, int(epoch), int(total), float(loss))
except RuntimeError:
pass
def work():
"""Train off the GUI thread."""
from ...embeddings import _dino_pretrain
return _dino_pretrain(crops, path, channel_policy=policy,
epochs=epochs, pretrained=False,
progress=_progress)
def _done(summary):
"""Accept only a completion belonging to this screen's current run."""
if not self._dino_closed and generation == self._dino_generation:
self._on_dino_done(summary)
try:
self._dino_jobs.submit(work, _done)
except Exception as exc:
self._on_dino_failed(str(exc))
return ""
return path
def _on_dino_done(self, summary) -> None:
"""Offer the new checkpoint in the picker and choose it."""
if self._dino_closed:
return
self._dino_summary = dict(summary)
self._select_dino_checkpoint(str(summary["path"]))
loss = summary["loss"][-1] if summary["loss"] else float("nan")
epochs = int(summary["epochs"])
self._on_dino_progress(self._dino_generation, epochs, epochs, loss)
self._set_dino_running(False)
self._status.setText(tr(
"Pretrained {epochs} epochs (last loss {loss:.3f}). The "
"checkpoint is chosen in the Foundation model picker; press "
"Embed to use it.").format(epochs=int(summary["epochs"]),
loss=loss))
def _dino_classifier_dialog(self) -> QDialog:
"""Configure supervised training without reading a checkpoint or starting work."""
dialog = QDialog(self)
dialog.setObjectName("EmbeddingsDinoClassifierDialog")
dialog.setWindowTitle(tr("Train an own-DINO classifier"))
layout = QVBoxLayout(dialog)
instruction = QLabel(tr(
"Choose a table with one row per loaded crop, in the same order, "
"and columns 'label' and 'acquisition_group'. Blank labels are "
"excluded. Each labelled crop needs its known acquisition-well "
"identity; complete groups are held out for validation."), dialog)
instruction.setWordWrap(True)
layout.addWidget(instruction)
form = QFormLayout()
table_row = QHBoxLayout()
table = QLineEdit(dialog)
table.setObjectName("EmbeddingsDinoClassifierTable")
table.setAccessibleName(tr("Training labels and acquisition groups"))
browse = QPushButton(tr("Browse…"), dialog)
browse.setObjectName("EmbeddingsDinoClassifierTableBrowse")
def choose_table():
"""Choose an explicit row-aligned label and acquisition-group table."""
path, _filter = QFileDialog.getOpenFileName(
dialog, tr("Choose training labels and acquisition groups"),
table.text(), tr("Tables (*.csv *.tsv *.parquet *.feather *.xlsx)"))
if path:
table.setText(path)
browse.clicked.connect(choose_table)
table_row.addWidget(table)
table_row.addWidget(browse)
form.addRow(tr("Training table"), table_row)
channels = QLineEdit(dialog)
channels.setObjectName("EmbeddingsDinoClassifierChannels")
channels.setAccessibleName(tr("Teacher channel order"))
channels.setToolTip(tr(
"Enter unique zero-based physical crop channels, separated by commas, "
"in the order represented by the teacher's saved scaling. This order "
"must be known; historical teacher checkpoints do not prove it."))
form.addRow(tr("Teacher channel order"), channels)
epochs = QSpinBox(dialog)
epochs.setObjectName("EmbeddingsDinoClassifierEpochs")
epochs.setRange(1, 10000)
epochs.setValue(20)
form.addRow(tr("Supervised epochs"), epochs)
batch = QSpinBox(dialog)
batch.setObjectName("EmbeddingsDinoClassifierBatch")
batch.setRange(1, 4096)
batch.setValue(32)
form.addRow(tr("Batch size"), batch)
rate = QDoubleSpinBox(dialog)
rate.setObjectName("EmbeddingsDinoClassifierLearningRate")
rate.setDecimals(7)
rate.setRange(0.0000001, 1.0)
rate.setValue(0.0001)
form.addRow(tr("Learning rate"), rate)
holdout = QDoubleSpinBox(dialog)
holdout.setObjectName("EmbeddingsDinoClassifierHoldout")
holdout.setRange(0.01, 0.99)
holdout.setSingleStep(0.05)
holdout.setValue(0.2)
form.addRow(tr("Validation fraction"), holdout)
seed = QSpinBox(dialog)
seed.setObjectName("EmbeddingsDinoClassifierSeed")
seed.setRange(0, 2147483647)
form.addRow(tr("Split and training seed"), seed)
freeze = QCheckBox(tr("Freeze the encoder; train only the classifier head"), dialog)
freeze.setObjectName("EmbeddingsDinoClassifierFreeze")
freeze.setChecked(True)
form.addRow(freeze)
normalize = QCheckBox(tr("Normalize features before the classifier head"), dialog)
normalize.setObjectName("EmbeddingsDinoClassifierNormalize")
normalize.setChecked(True)
form.addRow(normalize)
form.addRow(tr("Training device"), QLabel(tr("CPU"), dialog))
layout.addLayout(form)
error = QLabel(dialog)
error.setWordWrap(True)
layout.addWidget(error)
buttons = QDialogButtonBox(
QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel,
parent=dialog)
buttons.button(QDialogButtonBox.StandardButton.Ok).setText(tr("Train"))
buttons.button(QDialogButtonBox.StandardButton.Ok).setObjectName("EmbeddingsDinoClassifierStart")
buttons.rejected.connect(dialog.reject)
def start():
"""Retain the form when a path or metadata validation refuses the run."""
path = self._train_dino_classifier(
table_path=table.text(), source_channels=channels.text(),
epochs=epochs.value(), batch_size=batch.value(), lr=rate.value(),
holdout=holdout.value(), seed=seed.value(),
freeze_encoder=freeze.isChecked(), normalize_features=normalize.isChecked())
if path:
dialog.accept()
else:
error.setText(self._status.text())
buttons.accepted.connect(start)
layout.addWidget(buttons)
return dialog
def _choose_dino_classifier(self) -> None:
"""Open supervised settings only when this screen has no DINO work in flight."""
if (self._dino_running or self._classifier_running or self._dino_closed
or self._classifier_apply_running or self._classifier_apply_jobs.active_jobs()):
return
dialog = self._dino_classifier_dialog()
try:
dialog.exec()
finally:
dialog.deleteLater()
def _set_classifier_running(self, running: bool) -> None:
"""Keep pretraining and classifier actions mutually exclusive while working."""
self._classifier_running = bool(running)
self._set_dino_running(self._dino_running)
self._classifier_cancel.setEnabled(
self._classifier_running and not self._dino_closed
and not self._classifier_stop.is_set())
def _cancel_dino_classifier(self) -> None:
"""Request a cooperative stop at the trainer's next batch boundary."""
if self._classifier_running:
self._classifier_stop.set()
self._classifier_cancel.setEnabled(False)
self._status.setText(tr("Stopping classifier training…"))
def _on_classifier_progress(self, generation: int, epoch: int, total: int,
train_loss: float, validation_loss: float) -> None:
"""Apply only the current run's reports to its original crop selection."""
if (self._dino_closed or not self._classifier_running
or generation != self._classifier_generation
or self._classifier_crop_generation != self._crop_generation):
return
self._classifier_progress_bar.setRange(0, max(1, total))
self._classifier_progress_bar.setValue(epoch)
self._set_training_progress_text(
self._classifier_progress_bar, self._classifier_progress_label, tr(
"Epoch {epoch}/{total} — training {train:.3f}, validation {validation:.3f}").format(
epoch=epoch, total=total, train=train_loss, validation=validation_loss))
def _on_classifier_failed(self, message: str) -> None:
"""Restore controls after an error without changing a newer crop selection."""
if self._dino_closed:
return
self._set_classifier_running(False)
if self._classifier_crop_generation == self._crop_generation:
self._set_training_progress_text(
self._classifier_progress_bar, self._classifier_progress_label, tr("Classifier training stopped."))
self._status.setText(str(message))
def _on_classifier_finished(self, ok: bool) -> None:
"""Handle pre-start cancellation that finishes without an error callback."""
if (ok or self._dino_closed or not self._classifier_running
or self._classifier_jobs.pending_jobs()):
return
self._on_classifier_failed(tr("Classifier training stopped."))
def _train_dino_classifier(self, path: str = "", *, table_path: str = "",
source_channels: str = "", epochs: int = 20,
batch_size: int = 32, lr: float = 1e-4,
freeze_encoder: bool = True,
normalize_features: bool = True,
holdout: float = 0.2, seed: int = 0) -> str:
"""Train on explicitly labelled raw crops using the selected own teacher.
Rows with blank labels are excluded before calling the backend. Every
retained row must carry an explicit acquisition group. Crop planes are
not sliced or scaled by the GUI; the trainer uses the declared physical
channel order and the teacher's saved scale. The existing logistic
regression route and its labels are untouched.
:param path: new classifier artifact; asks for a path when empty.
:param table_path: row-aligned table with label and acquisition_group columns.
:param source_channels: comma-separated ordered physical channel indices.
:param epochs: supervised epochs, independent of pretraining progress.
:param batch_size: labelled crops per optimizer step.
:param lr: optimizer learning rate.
:param freeze_encoder: train only the head when true.
:param normalize_features: normalize encoder features before the head.
:param holdout: grouped validation fraction, checked by the backend.
:param seed: split and initialization seed.
:returns: output path when submitted, otherwise an empty string.
"""
from ...embeddings import _DINO_PREFIX
from ...tabular import read_table, table_format
if (self._dino_closed or self._dino_running or self._classifier_running
or self._dino_jobs.active_jobs() or self._classifier_jobs.active_jobs()
or self._classifier_apply_running or self._classifier_apply_jobs.active_jobs()):
return ""
crops = getattr(self, "_crops", None)
if crops is None:
self._status.setText(tr("Load crops first."))
return ""
selection = str(self._foundation.currentData() or "")
checkpoint_path = selection[len(_DINO_PREFIX):] if selection.startswith(_DINO_PREFIX) else ""
if not checkpoint_path or not os.path.isfile(checkpoint_path):
self._status.setText(tr("Choose an existing own-DINO teacher checkpoint first."))
return ""
try:
channels = tuple(int(value.strip()) for value in source_channels.split(","))
if (not channels or len(set(channels)) != len(channels)
or any(value < 0 or value >= crops.shape[-1] for value in channels)):
raise ValueError
except (TypeError, ValueError):
self._status.setText(tr("Enter unique physical channel indices in the teacher's saved order."))
return ""
try:
if not table_path:
raise ValueError(tr("Choose the training label and acquisition-group table."))
options = ({"dtype": str, "keep_default_na": False}
if table_format(str(table_path)) not in ("parquet", "feather") else {})
frame = read_table(str(table_path), canonicalise=False, report=None,
repair_plate_ids=False, **options)
if (not frame.columns.is_unique
or not {"label", "acquisition_group"}.issubset(frame.columns)):
raise ValueError(tr("The training table needs 'label' and 'acquisition_group' columns."))
if len(frame) != len(crops):
raise ValueError(tr("The training table must have one row per loaded crop, in crop order."))
indices, labels, groups = [], [], []
for index, (label, group) in enumerate(zip(frame["label"], frame["acquisition_group"])):
if pd.isna(label) or not str(label).strip():
continue
if (pd.isna(group) or str(group).strip().casefold()
in ("", "nan", "none", "unknown", "<na>")):
raise ValueError(tr("Every labelled crop needs a known acquisition-group identity."))
indices.append(index)
labels.append(str(label).strip())
groups.append(str(group).strip())
if len(set(labels)) < 2 or len(set(groups)) < 2:
raise ValueError(tr("Choose labelled crops from at least two classes and two acquisition groups."))
except Exception as exc:
self._status.setText(str(exc))
return ""
if not path:
path, _filter = QFileDialog.getSaveFileName(
self, tr("Save the trained classifier as"), "", tr("PyTorch checkpoints (*.pt)"))
if not path:
return ""
path = str(path)
if os.path.exists(path) or os.path.realpath(path) == os.path.realpath(checkpoint_path):
self._status.setText(tr("Choose a new classifier file; existing files are not replaced."))
return ""
self._classifier_generation += 1
generation = self._classifier_generation
self._classifier_crop_generation = self._crop_generation
self._classifier_stop = threading.Event()
stop = self._classifier_stop
emit = self._classifier_progress.emit
labels, groups, indices = tuple(labels), tuple(groups), tuple(indices)
self._set_classifier_running(True)
self._classifier_progress_bar.setRange(0, max(1, int(epochs)))
self._classifier_progress_bar.setValue(0)
self._set_training_progress_text(
self._classifier_progress_bar, self._classifier_progress_label, tr("Starting classifier training…"))
self._status.setText(tr("Training on {n} labelled crops…").format(n=len(indices)))
def progress(epoch, total, train_loss, validation_loss):
"""Relay worker progress without reading or changing GUI widgets."""
try:
emit(generation, int(epoch), int(total), float(train_loss), float(validation_loss))
except RuntimeError:
pass
def cancelled():
"""Honor both the application's worker cancellation and the local stop."""
from ..bridge import checkpoint
checkpoint()
return stop.is_set()
def work():
"""Filter the frozen row selection before calling the real trainer on raw planes."""
from ...embeddings import _dino_finetune_classifier
return _dino_finetune_classifier(
crops[list(indices)], checkpoint_path, path, labels=labels, groups=groups,
source_channels=channels, freeze_encoder=freeze_encoder,
normalize_features=normalize_features, epochs=epochs,
batch_size=batch_size, lr=lr, holdout=holdout, seed=seed,
device="cpu", progress=progress, cancelled=cancelled)
def done(summary):
"""Publish actual artifact metadata only to the run's original crop selection."""
if self._dino_closed or generation != self._classifier_generation:
return
self._set_classifier_running(False)
if self._classifier_crop_generation != self._crop_generation:
return
self._dino_classifier_summary = dict(summary)
self._status.setText(tr("Classifier saved to {path}. Classes: {classes}.").format(
path=summary["path"], classes=", ".join(summary["classes"])))
try:
self._classifier_jobs.submit(work, done)
except Exception as exc:
self._on_classifier_failed(str(exc))
return ""
return path
def _sync_classifier_application_controls(self, *_args) -> None:
"""Keep application and training mutually exclusive until worker retirement."""
active = bool(self._classifier_apply_jobs.active_jobs())
if active and not self._dino_closed:
self._classifier_retire_timer.start()
else:
self._classifier_retire_timer.stop()
busy = self._classifier_apply_running or active
enabled = not (self._dino_running or self._classifier_running or self._dino_closed or busy)
for widget in (self._dino, self._dino_checkpoint, self._dino_epochs, self._classifier_train):
widget.setEnabled(enabled)
self._classifier_apply.setEnabled(enabled and not self._loading)
self._classifier_apply_cancel.setEnabled(
self._classifier_apply_running and self._classifier_apply_delivery and not self._dino_closed)
def _cancel_classifier_application(self) -> None:
"""Discard result delivery without claiming to interrupt model computation."""
if not self._classifier_apply_running:
return
self._classifier_apply_delivery = False
self._classifier_apply_generation += 1
self._classifier_apply_running = False
self._classifier_apply_jobs.cancel()
self._sync_classifier_application_controls()
if not self._dino_closed:
self._status.setText(tr("Classifier result cancelled; background computation may finish."))
def _on_classifier_application_failed(self, message: str) -> None:
"""Restore controls and report errors only to the original crop selection."""
deliver = (self._classifier_apply_delivery and not self._dino_closed
and self._classifier_apply_crop_generation == self._crop_generation)
self._classifier_apply_running = False
self._classifier_apply_delivery = False
self._sync_classifier_application_controls()
if deliver and not self._loading:
self._status.setText(str(message))
def _on_classifier_application_finished(self, ok: bool) -> None:
"""Restore the form after cancellation before the adapter starts."""
if not ok and self._classifier_apply_running and not self._classifier_apply_jobs.pending_jobs():
self._on_classifier_application_failed(tr("Classifier application stopped."))
def _apply_dino_classifier(self, path: str = "", *, batch_size: int | None = None) -> str:
"""Apply a saved classifier to raw crops through its dedicated adapter.
Features, logits, probabilities and class labels retain backend order.
Teacher selection, logistic labels and existing embedding results are
untouched. Cancelling suppresses delivery; it does not interrupt the
adapter's computation.
"""
if (self._dino_closed or self._loading or self._dino_running or self._classifier_running
or self._classifier_apply_running or self._classifier_apply_jobs.active_jobs()
or self._dino_jobs.active_jobs() or self._classifier_jobs.active_jobs()):
return ""
crops = getattr(self, "_crops", None)
if crops is None:
self._status.setText(tr("Load crops first."))
return ""
if not path:
recent = getattr(self, "_dino_classifier_summary", {}).get("path", "")
path, _filter = QFileDialog.getOpenFileName(
self, tr("Choose a saved own-DINO classifier"), str(recent),
tr("PyTorch checkpoints (*.pt)"))
if not path:
return ""
path = str(path)
batch = self._batch.value() if batch_size is None else batch_size
self._classifier_apply_generation += 1
generation = self._classifier_apply_generation
self._classifier_apply_crop_generation = self._crop_generation
crop_generation = self._crop_generation
self._classifier_apply_running = True
self._classifier_apply_delivery = True
self._sync_classifier_application_controls()
self._status.setText(tr("Applying classifier to {n} crops…").format(n=len(crops)))
def work():
"""Apply the actual artifact to unchanged full raw channel-last crops."""
from ...embeddings import _dino_classifier_apply
answer = _dino_classifier_apply(crops, path, batch_size=batch, device="cpu")
features = np.asarray(answer["features"])
classes = list(answer["classes"])
if (features.ndim != 2 or features.shape[0] != len(crops)
or np.shape(answer["logits"]) != (len(crops), len(classes))
or np.shape(answer["probabilities"]) != (len(crops), len(classes))
or len(answer["predictions"]) != len(crops)):
raise ValueError(tr("Classifier output is not aligned with the loaded crops."))
return dict(answer, path=path)
def finished(answer):
"""Retain outputs only for the still-current, uncancelled crop selection."""
if self._dino_closed or generation != self._classifier_apply_generation:
return
self._classifier_apply_running = False
self._classifier_apply_delivery = False
self._sync_classifier_application_controls()
if crop_generation != self._crop_generation or self._loading:
return
self._classifier_application_result = answer
self._classifier_results.setEnabled(True)
self._status.setText(tr(
"Applied classifier to {n} crops. Classes: {classes}. Open Classifier results to inspect or export.").format(
n=len(answer["predictions"]), classes=", ".join(answer["classes"])))
try:
self._classifier_apply_jobs.submit(work, finished)
except Exception as exc:
self._on_classifier_application_failed(str(exc))
return ""
return path
def _show_classifier_application(self):
"""Inspect actual classifier outputs separately from teacher embeddings."""
answer = self._classifier_application_result
if answer is None:
return None
dialog = QDialog(self)
dialog.setObjectName("EmbeddingsDinoClassifierResultDialog")
dialog.setWindowTitle(tr("Classifier results"))
dialog.setAttribute(Qt.WA_DeleteOnClose)
layout = QVBoxLayout(dialog)
provenance = QLabel(tr("Artifact: {path}\nSHA-256: {digest}\nPhysical channels: {channels}").format(
path=answer["path"], digest=answer["checkpoint_sha256"],
channels=", ".join(map(str, answer["source_channels"]))), dialog)
provenance.setWordWrap(True)
provenance.setTextFormat(Qt.PlainText)
provenance.setTextInteractionFlags(Qt.TextSelectableByMouse)
layout.addWidget(provenance)
picker = QComboBox(dialog)
picker.setObjectName("EmbeddingsDinoClassifierResultView")
for label, key in ((tr("Predictions"), "predictions"), (tr("Features"), "features"),
(tr("Logits"), "logits"), (tr("Probabilities"), "probabilities")):
picker.addItem(label, key)
layout.addWidget(picker)
table = QTableWidget(dialog)
table.setObjectName("EmbeddingsDinoClassifierResultTable")
table.setEditTriggers(QTableWidget.NoEditTriggers)
layout.addWidget(table)
note = QLabel(tr("Preview shows up to {rows} crops and {columns} columns; export includes all values.").format(
rows=50, columns=PREVIEW_DIMENSIONS), dialog)
note.setWordWrap(True)
layout.addWidget(note)
save = QPushButton(tr("Export classifier results…"), dialog)
save.setObjectName("EmbeddingsDinoClassifierExport")
save.clicked.connect(lambda: self._save_classifier_application(answer=answer))
layout.addWidget(save)
def show_view():
"""Render the selected real output without rerunning or changing the model."""
key = picker.currentData()
values = np.asarray(answer[key])
if key == "predictions":
values = values.reshape(-1, 1)
headers = [tr("Prediction")]
elif key == "features":
headers = [f"feature_{index}" for index in range(values.shape[1])]
else:
headers = list(answer["classes"])
rows, columns = min(len(values), 50), min(values.shape[1], PREVIEW_DIMENSIONS)
table.setRowCount(rows)
table.setColumnCount(columns)
table.setHorizontalHeaderLabels(headers[:columns])
table.setVerticalHeaderLabels([str(index) for index in range(rows)])
for row in range(rows):
for column in range(columns):
value = values[row, column]
text = str(value) if key == "predictions" else f"{float(value):.6g}"
table.setItem(row, column, table_item(text, key=None if key == "predictions" else float(value)))
picker.currentIndexChanged.connect(show_view)
show_view()
dialog.resize(750, 500)
dialog.show()
return dialog
def _save_classifier_application(self, path: str = "", *, answer=None) -> str:
"""Export complete ordered outputs and actual artifact provenance via tabular."""
answer = self._classifier_application_result if answer is None else answer
if answer is None:
return ""
if not path:
path, _filter = QFileDialog.getSaveFileName(
self, tr("Export classifier results"), "", tr("CSV tables (*.csv)"))
if not path:
return ""
path = str(path)
def work():
"""Write every output column with saved class names and channel provenance."""
from ...tabular import write_table
columns = {"crop_index": np.arange(len(answer["predictions"])),
"prediction": answer["predictions"]}
for index, label in enumerate(answer["classes"]):
columns[f"probability:{label}"] = np.asarray(answer["probabilities"])[:, index]
columns[f"logit:{label}"] = np.asarray(answer["logits"])[:, index]
for index in range(np.shape(answer["features"])[1]):
columns[f"feature_{index}"] = np.asarray(answer["features"])[:, index]
frame = pd.DataFrame(columns)
frame["checkpoint_path"] = answer["path"]
frame["checkpoint_sha256"] = answer["checkpoint_sha256"]
frame["source_channels"] = ",".join(map(str, answer["source_channels"]))
return write_table(frame, path, canonicalise=False, index=False)
def finished(saved):
"""Report the exported snapshot only while its source result is current."""
if not self._dino_closed and self._classifier_application_result is answer:
self._status.setText(tr("Classifier results saved to {path}.").format(path=saved))
self._jobs.submit(work, finished)
return path
def _fill_backbones(self) -> None:
"""Offer the engine's default first, and never an empty list."""
from ...embeddings import DEFAULT_BACKBONE
for name in (DEFAULT_BACKBONE, "resnet50", "convnext_tiny",
"vit_base_patch16_224"):
if self._backbone.findText(name) < 0:
self._backbone.addItem(name)
self._backbone.setCurrentText(DEFAULT_BACKBONE)
def _fill_object_classes(self, classes) -> None:
"""Offer the classes this database has, or every class it could have.
With nothing chosen yet the full vocabulary is offered, because a
combo that is empty until a path is typed looks broken. Once a
database is open the list is what IS in it, so a class that can only
ever return nothing is not on offer -- and the current choice is kept
when the new database still has it, so re-pointing at a second plate
does not silently switch which object is being embedded.
"""
from ...png_list import PNG_LIST_ID_COLUMNS
wanted = self._object.currentData() or "cell"
names = tuple(classes) or tuple(PNG_LIST_ID_COLUMNS)
self._object.blockSignals(True)
try:
self._object.clear()
for name in names:
self._object.addItem(name, name)
index = self._object.findData(wanted)
self._object.setCurrentIndex(max(0, index))
finally:
self._object.blockSignals(False)
def _fill_plates(self, names) -> None:
"""Offer every plate in the database, with 'every plate' first."""
wanted = self._plate.currentData() or ""
self._plate.blockSignals(True)
try:
self._plate.clear()
self._plate.addItem("every plate", "")
for name in names:
self._plate.addItem(str(name), str(name))
index = self._plate.findData(wanted)
self._plate.setCurrentIndex(max(0, index))
finally:
self._plate.blockSignals(False)
def _use_example_database(self, database) -> None:
"""Point the crop source at the example plate's measurements database.
Selects the database source, fills the path and refreshes what it
offers; the crops are loaded when Load crops is pressed.
:param database: the example plate's ``measurements.db``.
"""
index = self._where.findData(CROP_SOURCE_DATABASE)
if index >= 0:
self._where.setCurrentIndex(index)
self._path.setText(str(database))
self._on_path_changed()
def _load_demo_crops(self) -> None:
"""Load the bundled generic shapes through the ordinary crop reader."""
if self._loading or self._dino_closed:
return
self._stop.clear()
self._set_loading(True)
self._status.setText(tr("Loading synthetic demo crops…"))
stop = self._stop
def work():
"""Unpack and read the small local example off the GUI thread."""
from ...crop_loader import (CROP_SOURCE_FOLDER, CropQuery,
load_crops, plan_crops)
from ...example_archives import _prepare_embeddings_example
example = _prepare_embeddings_example()
plan = plan_crops(CropQuery(
source=CROP_SOURCE_FOLDER, path=str(example["crops"])))
record = {}
crops = load_crops(plan, cancelled=stop.is_set, record=record)
return example, plan, crops, record
def finished(answer):
"""Select the CPU example and attach its row-aligned labels."""
if self._dino_closed:
return
if stop.is_set():
self._stopped(tr("Loading synthetic demo crops was stopped."))
return
example, plan, crops, record = answer
self._where.setCurrentIndex(self._where.findData(plan.query.source))
self._path.setText(str(example["crops"]))
self._filter.clear()
self._limit.setValue(DEFAULT_CROP_LIMIT)
self._select_dino_example()
self._plan = plan
self._on_crops_read((crops, record), demo=example)
self._jobs.submit(work, finished)
def _select_dino_example(self) -> None:
"""Pick the CPU DINO choice, one projected pass and small batches."""
self._foundation.setCurrentIndex(self._foundation.findData("dino_auto"))
self._policy.setCurrentIndex(self._policy.findData("project"))
self._batch.setValue(8)
def _load_cell_example(self) -> None:
"""Load the example plate's annotated cells and their labels.
The plate is fetched once into the shared example folder; the crops
annotated ``infected`` 1 or 2 load in database order, and a
label table with one row per crop is written beside the plate.
"""
from ..widgets.measurements_example import (example_measurements_db,
example_measurements_folder)
if self._loading or self._dino_closed:
return
database = example_measurements_db()
if database.is_file():
self._read_cell_example(database)
return
self._cell_example_button.setEnabled(False)
self._status.setText(tr("Fetching test data…"))
def fetched(result, error) -> None:
"""Read the plate once it is on disk, or say why it is not."""
self._cell_example_button.setEnabled(not self._loading)
if result is None or not database.is_file():
self._status.setText(tr(
"The test data could not be downloaded: {detail}",
detail=error or tr("unknown error")))
return
self._read_cell_example(database)
from ..hf_download import download_annotate_example
download_annotate_example(self, example_measurements_folder(), fetched)
def _read_cell_example(self, database) -> None:
"""Read the annotated example crops off the GUI thread."""
self._stop.clear()
self._set_loading(True)
self._status.setText(tr("Loading the annotated example cells…"))
stop = self._stop
def work():
"""Plan, read and label the annotated crops."""
from ...crop_loader import CropQuery, load_crops, plan_crops
from ...example_archives import example_plate_folder
from ...tabular import write_table
plan = plan_crops(CropQuery(
source=CROP_SOURCE_DATABASE, path=str(database),
object_type="cell", where="infected IN (1, 2)", limit=0))
record = {}
crops = load_crops(plan, cancelled=stop.is_set, record=record)
labels = [str(int(row["infected"])) for row in plan.rows]
folder = example_plate_folder().parent / "embeddings_cell_example"
path = folder / "labels.csv"
write_table(pd.DataFrame({
"prcfo": [row.get("prcfo") for row in plan.rows],
"label": labels}), path)
return plan, crops, record, {"folder": folder, "labels": path,
"cells": True}
def finished(answer):
"""Select the CPU DINO choice and attach the annotations."""
if self._dino_closed:
return
if stop.is_set():
self._stopped(tr("Loading the annotated example cells was "
"stopped."))
return
plan, crops, record, example = answer
self._where.setCurrentIndex(self._where.findData(plan.query.source))
self._path.setText(str(database))
self._filter.setText(plan.query.where)
self._limit.setValue(max(1, plan.count))
self._select_dino_example()
self._plan = plan
self._on_crops_read((crops, record), demo=example)
self._jobs.submit(work, finished)
[docs]
def crop_source(self) -> str:
"""Which kind of place the crops come from."""
return str(self._where.currentData() or CROP_SOURCE_DATABASE)
def _on_source_changed(self, *_args) -> None:
"""Grey the controls the chosen source does not read.
A folder has no classes, no plates and no columns to filter on. The
house rule from :mod:`spacr.crop_source` is that a control the user
can edit that changes nothing is worse than one that is not there,
so these are disabled rather than left to mislead.
"""
database = self.crop_source() == CROP_SOURCE_DATABASE
for widget in (self._class_label, self._object, self._plate_label,
self._plate, self._filter_label, self._filter):
widget.setEnabled(database)
self._path.setPlaceholderText(
"path to measurements.db" if database
else "path to a folder of crop PNGs")
self._on_path_changed()
def _on_path_changed(self, *_args) -> None:
"""Enable Load once there is a path, and read what the database offers.
NOTHING HERE TOUCHES THE FILESYSTEM. Whether the path is there is a
stat, and a stat against a hung or unmounting network share blocks
for as long as the mount takes to give up -- on this thread, which
is the GUI one. This method runs on every edit of the path box and
again at the end of every load, when the user did nothing at all, so
the freeze would arrive unprompted.
So Load is offered for any non-empty path and the answer comes from
pressing it: :func:`spacr.crop_loader.plan_crops` names the path it
could not find, which is a better sentence than a greyed button, and
it says it from a worker thread.
"""
path = str(self._path.text()).strip()
self._load.setEnabled(bool(path) and not self._loading)
self._load.setToolTip(
"Read these crops" if path
else "Choose a database or a folder first")
if not path or self.crop_source() != CROP_SOURCE_DATABASE:
return
full = os.path.expanduser(path)
if full == self._read_path:
return
self._read_path = full
self._read_database_choices(full)
def _read_database_choices(self, path: str) -> None:
"""Fill the class and plate pickers from ``path``, off the GUI thread.
Run once per database rather than once per edit: this is reached
from every path change AND from the end of every load, since both
put the controls back, and re-reading a database on a network mount
each time a load finishes is a pause nobody asked for.
"""
def work():
"""Ask the database what it holds. Two cheap indexed reads.
The stat runs here, on the worker, because it is the part that
blocks on a slow mount. A path that is not a file yet is not an
error -- it is one somebody is halfway through typing -- so it
answers with nothing to offer rather than raising, which would
put a refusal in the status line for every keystroke.
"""
from ...crop_loader import object_classes, plates
if not os.path.isfile(path):
return (), ()
return object_classes(path), plates(path)
def done(answer) -> None:
"""Put the answer in the two combos."""
classes, names = answer
self._fill_object_classes(classes)
self._fill_plates(names)
self._jobs.submit(work, done)
[docs]
def choose_path(self) -> str:
"""Ask for the database or the folder, and remember the answer.
:returns: what was chosen, or ``''`` when the dialog was dismissed,
so a caller can tell a cancel from a choice.
"""
if self.crop_source() == CROP_SOURCE_DATABASE:
path, _filter = QFileDialog.getOpenFileName(
self, "Choose a measurements database", self._path.text(),
"SQLite databases (*.db *.sqlite);;Every file (*)")
else:
path = QFileDialog.getExistingDirectory(
self, "Choose a folder of crop PNGs", self._path.text())
if path:
self._path.setText(str(path))
self._on_path_changed()
return str(path or "")
[docs]
def crop_query(self):
"""The :class:`spacr.crop_loader.CropQuery` the controls describe.
:returns: the query a press of Load would run, which is what makes
the panel testable without pressing anything.
"""
from ...crop_loader import CropQuery
return CropQuery(
source=self.crop_source(),
path=os.path.expanduser(str(self._path.text()).strip()),
object_type=str(self._object.currentData() or "cell"),
plate=str(self._plate.currentData() or ""),
where=str(self._filter.text()).strip(),
limit=int(self._limit.value()),
page_size=DEFAULT_PAGE_SIZE,
)
[docs]
def is_loading(self) -> bool:
"""Whether a crop load is in flight."""
return bool(self._loading)
[docs]
def load_crops(self) -> None:
"""Plan the selection, then read it a page at a time.
The load runs as two jobs. Planning is a count and a select, so it
comes back in milliseconds and the screen can say what was matched --
or why nothing was -- before committing to read a single pixel. Only
then does the second job start, and it reports after every page.
"""
if self._loading:
self.stop_loading()
return
query = self.crop_query()
if not query.path:
self._refuse("No crops loaded",
"Choose a measurements database or a folder of crop "
"PNGs first.")
return
self._stop.clear()
self._set_loading(True)
self._status.setText(f"Looking for {query.describe()}…")
def work():
"""Ask what matches. No pixels are read here."""
from ...crop_loader import plan_crops
return plan_crops(query)
self._jobs.submit(work, self._on_planned)
[docs]
def stop_loading(self) -> None:
"""Stop a load between pages, keeping the pages already read.
The flag is checked before each page rather than inside one, so a
stop costs at most one page and never leaves a half-written crop in
the stack.
"""
self._stop.set()
self._status.setText("Stopping after this page…")
def _set_loading(self, loading: bool) -> None:
"""Turn Load into Stop for the duration, and park the other controls.
EMBED IS PUT BACK THE WAY IT WAS, not left off. A load that matched
nothing unloads nothing, so the crops from the previous load are
still there and still encodable; leaving Embed disabled would make
the panel's own sentence about them false, and there is no way for a
user to get it back without loading again.
"""
self._loading = bool(loading)
self._load.setText("Stop" if loading else "Load crops")
self._load.setEnabled(True if loading else bool(
str(self._path.text()).strip()))
for widget in (self._where, self._path, self._browse, self._object,
self._plate, self._filter, self._limit,
self._demo_button, self._cell_example_button,
self._test_data_button):
widget.setEnabled(not loading)
self._run.setEnabled(
False if loading else getattr(self, "_crops", None) is not None)
self._save_similar.setEnabled(
not loading and self._result_identity is not None
and self._result is not None)
self._sync_classifier_application_controls()
if not loading:
self._on_source_changed()
def _on_planned(self, plan) -> None:
"""Say what matched, and start reading it -- or say why nothing did.
A STOP PRESSED DURING PLANNING IS HONOURED HERE. The plan job cannot
be interrupted -- it is one count and one select -- so a user who
presses Stop while it is in flight is answered when it returns, by
not starting the read at all. Submitting it anyway would read a page
and then refuse, which is slower and says the wrong thing.
"""
self._plan = plan
if plan.is_empty:
self._set_loading(False)
self._refuse("No crops", plan.empty_reason)
return
if self._stop.is_set():
self._stopped(f"Loading {plan.describe()} was stopped before any "
f"crop was read.")
return
self._status.setText(f"Loading {plan.describe()}…")
self._state.say(f"Loading {plan.count:,} crops",
plan.describe())
stop = self._stop
emit = self.crops_progress.emit
def work():
"""Read the pages on a worker thread.
``emit`` is a bound signal and the only thing here that reaches
the GUI: Qt queues it onto the receiving thread, so the progress
line moves without this function ever touching a widget.
"""
from ...crop_loader import load_crops
record: Dict[str, Any] = {}
crops = load_crops(plan, progress=emit,
cancelled=stop.is_set, record=record)
return crops, record
self._jobs.submit(work, self._on_crops_read)
def _on_crops_progress(self, done: int, total: int) -> None:
"""One page in. Runs on the GUI thread, however it was emitted."""
self._status.setText(f"Loaded {done:,} of {total:,} crops…")
def _on_crops_read(self, answer, *, demo=None) -> None:
"""Hand the stack to the screen and say what it cost."""
crops, record = answer
self._crop_record = dict(record)
self._set_loading(False)
self.set_crops(crops, label=self._loaded_label(record))
plan = self._plan
if plan is not None and plan.query.source == CROP_SOURCE_DATABASE:
rows = plan.rows[:int(crops.shape[0])]
keys = tuple(row.get("prcfo") for row in rows)
if (len(keys) == len(crops) and all(
key is not None and not pd.isna(key)
and str(key).strip() for key in keys)
and len(set(map(str, keys))) == len(keys)):
self._crop_identity = (
os.path.abspath(os.path.expanduser(plan.query.path)),
tuple(str(key) for key in keys))
self._status.setText(self._loaded_sentence(record))
if demo is not None:
self._demo_active = True
self._caveat.setText(DIMENSION_CAVEAT + " " + (tr(
"These example cells carry the plate's own 'infected' "
"annotation (classes 1 and 2). They show the workflow on "
"real crops; so few labels are not a scientific "
"comparison.") if demo.get("cells") else tr(
"Synthetic shapes demonstrate the workflow only. Their "
"labels and scores are not biological validation.")))
if self._choose_labels(str(demo["labels"])):
if demo.get("cells"):
self._status.setText(tr(
"Loaded {count} annotated example cells and their "
"labels. Press {embed}, then choose {classifier} and "
"{run}. Labels: {path}").format(
count=int(crops.shape[0]), embed=tr("Embed"),
classifier=tr("Classifier"), run=tr("Run"),
path=demo["labels"]))
else:
self._status.setText(tr(
"Loaded 24 synthetic crops and labels. Press {embed}, then "
"choose {classifier} and {run}. Example files: {folder}").format(
embed=tr("Embed"), classifier=tr("Classifier"),
run=tr("Run"), folder=demo["folder"]))
self.crops_loaded.emit(int(crops.shape[0]))
def _loaded_label(self, record) -> str:
"""The short form, for the label beside the module header."""
shape = record.get("crop_shape") or []
size = "x".join(str(v) for v in shape[:2]) if shape else "?"
channels = shape[2] if len(shape) > 2 else "?"
return (f"{int(record.get('loaded', 0)):,} objects, {size}, "
f"{channels} channels")
def _loaded_sentence(self, record) -> str:
"""The long form: what was taken, out of what, and what was changed.
Every clause here is a thing a reader would otherwise have to guess
at, and three of them change what the numbers mean: a capped load is
a subset of the plate, a dropped row is a crop that does not exist to
be loaded, and a conformed crop has been padded or trimmed.
THE CAP CLAUSE IS DRIVEN BY ``capped``, NOT BY ``loaded < matched``.
The two are not the same on the merged route, where rows that cannot
be cut leave ``loaded`` below ``matched`` with no limit involved, and
"raise 'At most' to take more" is then an instruction that returns
the same crops and prints the same sentence again.
"""
loaded = int(record.get("loaded", 0))
matched = int(record.get("matched", loaded))
dropped = int(record.get("dropped", 0))
parts = [f"{loaded:,} crops loaded"]
if record.get("stopped"):
parts.append(f"stopped early; {matched:,} matched")
elif record.get("capped"):
parts.append(f"the first of {matched:,} that matched — raise "
f"'At most' to take more")
if dropped:
parts.append(f"{dropped:,} more matched but could not be cut "
f"from merged/*.npy")
conformed = int(record.get("conformed", 0))
if conformed:
shape = record.get("crop_shape") or []
frame = "x".join(str(v) for v in shape[:2])
parts.append(f"{conformed:,} padded or trimmed to {frame}")
parts.append(str(record.get("source", "")))
return ". ".join(part for part in parts if part) + "."
def _refuse(self, title: str, detail: str) -> None:
"""Put a refusal where the result would have been, and in the status.
Both places, because they are read at different moments: the panel is
where somebody looking for the crops is looking, and the status line
is what survives a scroll.
A REFUSAL UNLOADS NOTHING, and the panel has to say so or it
contradicts the rest of the screen. A query that matched nothing
leaves the previous stack exactly where it was, with Embed still
enabled and the header still naming it, and a panel reading "No
crops" over that is the screen telling a reader two things at once.
"""
crops = getattr(self, "_crops", None)
if crops is not None:
detail = (f"{detail} The {crops.shape[0]:,} crops already loaded "
f"are untouched, and Embed still encodes them.")
self._state.say(title, detail)
self._pages.setCurrentWidget(self._state)
self._status.setText(detail)
LOG.info("%s: %s", title, detail)
[docs]
def crop_record(self) -> Dict[str, Any]:
"""What the last load actually read, as the loader recorded it."""
return dict(self._crop_record)
[docs]
def set_crops(self, crops: np.ndarray, *, label: str = "") -> None:
"""Hand the screen an ``(objects, height, width, channels)`` stack.
The channel axis comes last. Getting it backwards does not raise
here; it reaches the model and fails there with a message about
tensor shapes.
That layout is what :mod:`spacr.crops` produces and what
:func:`spacr.embeddings.embed_array` documents.
:param crops: the object stack, channels last. Not copied: the
screen keeps this array and hands it to the backbone, so a
caller that mutates it afterwards changes what gets embedded.
:raises ValueError: when the stack is not four-dimensional.
"""
crops = np.asarray(crops)
if crops.ndim != 4:
raise ValueError(
f"crops are (objects, height, width, channels); got shape "
f"{crops.shape}. spacr.crops produces this stack, and "
f"spacr.embeddings.embed_array expects channels last.")
if self._classifier_running:
self._classifier_stop.set()
self._classifier_cancel.setEnabled(False)
self._cancel_classifier_application()
self._classifier_application_result = None
self._classifier_results.setEnabled(False)
self._crops = crops
if self._demo_active:
self._labels = {}
self._demo_active = False
self._caveat.setText(DIMENSION_CAVEAT)
self._crop_generation += 1
self._crop_identity = None
self._result_identity = None
self._result = None
self._frame = None
self._result_encoder = None
self._save_similar.setEnabled(False)
self._scale_record = {}
self._source.setText(
label or f"{crops.shape[0]} objects x {crops.shape[-1]} channels")
if (self._subcell_channels is not None
and self._subcell_mapping_error(
self._subcell_channels, int(crops.shape[-1]))):
self._subcell_channels = None
self._status.setText(tr(
"The new crops have fewer channels. Reopen Channels… to "
"map SubCell's four planes."))
if (self._cell_dino_channels is not None
and any(channel >= int(crops.shape[-1])
for channel in self._cell_dino_channels)):
self._cell_dino_channels = None
self._status.setText(tr(
"The new crops have fewer channels. Reopen Cell-DINO "
"checkpoint and channels to map its input planes."))
self._sync_subcell_controls()
self._run.setEnabled(True)
self._run.setToolTip("Encode every object")
self._state.say(
f"{crops.shape[0]:,} crops loaded",
"Choose how the channels are encoded, then press Embed. The "
"table fills with the first few dimensions of each object.")
self._pages.setCurrentWidget(self._state)
[docs]
def spec(self):
"""The :class:`spacr.embeddings.EmbeddingSpec` the controls describe."""
from ...embeddings import CHANNEL_PROJECT, EmbeddingError, EmbeddingSpec
foundation = str(self._foundation.currentData() or "")
device = "cpu" if self._demo_active else None
if foundation == "dino_auto":
return self._dino_note()[0]
if foundation == "subcell_rybg":
if self._subcell_channels is None:
raise EmbeddingError(tr(
"Choose four distinct crop channels before running "
"SubCell's four-plane model."))
return EmbeddingSpec(
backbone=foundation,
channel_policy=CHANNEL_PROJECT,
channels=self._subcell_channels,
batch_size=int(self._batch.value()),
device=device,
normalize=False,
)
if foundation == "cell_dino":
crops = getattr(self, "_crops", None)
count = 0 if crops is None else int(crops.shape[-1])
reason = self._cell_dino_mapping_error(
self._cell_dino_factory, self._cell_dino_channels,
count, self._cell_dino_checkpoint, self._cell_dino_digest)
if reason:
raise EmbeddingError(reason)
return EmbeddingSpec(
backbone=foundation,
channel_policy=CHANNEL_PROJECT,
channels=self._cell_dino_channels,
batch_size=int(self._batch.value()),
device=device,
cell_dino_factory=self._cell_dino_factory,
checkpoint_path=self._cell_dino_checkpoint,
checkpoint_sha256=self._cell_dino_digest,
)
return EmbeddingSpec(
backbone=foundation or str(self._backbone.currentText()).strip(),
channel_policy=str(self._policy.currentData()),
batch_size=int(self._batch.value()),
device=device,
)
[docs]
def embed(self) -> None:
"""Runs the encoder in the background and shows the result."""
crops = getattr(self, "_crops", None)
if crops is None:
self._status.setText("Load crops first.")
return
if self._foundation.currentData() == "subcell_rybg":
if crops.shape[-1] < 4:
self._status.setText(tr(
"SubCell's four-plane model needs crops with at least "
"four channels. Load a suitable crop source first."))
return
reason = self._subcell_mapping_error(
self._subcell_channels, int(crops.shape[-1]))
if reason:
self._status.setText(reason)
return
if self._foundation.currentData() == "cell_dino":
reason = self._cell_dino_mapping_error(
self._cell_dino_factory, self._cell_dino_channels,
int(crops.shape[-1]), self._cell_dino_checkpoint,
self._cell_dino_digest)
if reason:
self._status.setText(reason)
return
auto = self._foundation.currentData() == "dino_auto"
spec = None if auto else self.spec()
record = self._scale_record
generation = self._crop_generation
identity = self._crop_identity
self._status.setText(f"Embedding {crops.shape[0]} objects…")
choose = self._dino_note
def work():
"""Run the backbone off the GUI thread.
The import is inside because it pulls torch in: a user who never
opens this screen should not pay for it, and a user who does
should pay for it once, here, rather than at launch.
"""
from ...embeddings import _embed_plate, encoder_entry
try:
run = choose() if auto else (spec, None)
result = _embed_plate(crops, run[0], record=record)
entry = encoder_entry(getattr(result, "spec", None) or run[0])
return result, entry, run
except Exception as exc:
return None, None, f"{type(exc).__name__}: {exc}"
def finished(answer):
"""Publish only to the crop selection that produced the vectors."""
if generation != self._crop_generation:
return
result, entry, error = answer
if isinstance(error, str):
if not self._loading:
self._status.setText(error)
return
self._result_identity = identity
self._on_embedded(result, entry=entry)
if error[1] is not None and not self._loading:
self._dino_in_use.setText(self._dino_sentence(error))
self._status.setText(
self._status.text() + " " + self._dino_sentence(error))
self._jobs.submit(work, finished)
def _on_embedded(self, result, *, entry=None) -> None:
"""Fill the preview and say which encoder produced it."""
self._result = result
self._result_encoder = entry
self._save_similar.setEnabled(
not self._loading and self._result_identity is not None)
frame = pd.DataFrame(np.asarray(result.values),
columns=list(result.columns))
self._frame = frame
if self._loading:
return
self._fill_preview(frame)
from ...embeddings import encoder_entry
if entry is None:
entry = encoder_entry(getattr(result, "spec", None) or self.spec())
self._result_encoder = entry
digest = entry.sha256[:12] + "…" if entry.sha256 else "no checksum"
self._status.setText(
f"{len(frame)} objects x {len(frame.columns)} dimensions. "
f"Encoder {entry.name}, weights {digest}.")
if self._labels and len(self._labels) == len(frame):
self._show_scorecard()
def _save_for_similarity(self) -> None:
"""Write the actual result against its frozen source object keys."""
identity, result = self._result_identity, self._result
if identity is None or result is None:
self._status.setText(tr(
"Load database crops and embed them before saving vectors."))
return
database, keys = identity
entry = self._result_encoder
self._save_similar.setEnabled(False)
self._status.setText(tr("Saving crop embeddings…"))
def work():
"""Persist vectors without blocking the graphical thread."""
from ...active_learning import _store_crop_embeddings
try:
_store_crop_embeddings(database, keys, result,
encoder_entry=entry)
return len(keys), ""
except Exception as exc:
return 0, f"{type(exc).__name__}: {exc}"
def finished(answer):
"""A newer crop selection keeps its own status and controls."""
if self._result is result and not self._loading:
count, error = answer
self._save_similar.setEnabled(True)
if error:
self._status.setText(error)
return
self._status.setText(tr(
"Saved {count} crop embeddings. Open this database in "
"Annotate and use Like this.").format(count=count))
self._jobs.submit(work, finished)
def _fill_preview(self, frame: pd.DataFrame) -> None:
"""Show the first few dimensions, and only the first few.
The whole matrix is the result and it is not readable; a table of
2,048 columns would invite scrolling through it as though a column
meant something.
"""
columns = list(frame.columns)[:PREVIEW_DIMENSIONS]
rows = min(len(frame), 50)
self._pages.setCurrentWidget(self._table)
self._table.setSortingEnabled(False)
try:
_retire_table_cells(self._table, headers=True)
self._table.setRowCount(0)
self._table.setColumnCount(len(columns))
self._table.setHorizontalHeaderLabels(columns)
self._table.setRowCount(rows)
for row in range(rows):
for index, column in enumerate(columns):
value = float(frame.iloc[row][column])
item = _spare_table_item(
self._table, f"{value:.4f}",
make=lambda text, key=value: table_item(text, key=key))
item.setData(SORT_KEY_ROLE, float(value))
item.setTextAlignment(Qt.AlignRight | Qt.AlignVCenter)
self._table.setItem(row, index, item)
finally:
self._table.setSortingEnabled(True)
def _on_job_failed(self, message: str) -> None:
"""A refusal is a sentence on the screen, never a silent empty table.
One handler for both jobs, because both fail the same way as far as
the reader is concerned. A load that failed also has to put the
controls back: leaving Load reading "Stop" after the job it would
have stopped has already died is a button that does nothing.
A STOP IS NOT A FAILURE, and the one case that arrives here is a
stop taken before the first page finished -- the loader has no
crops to return and says so by raising. Heading that "Crops could
not be loaded" tells a user their machine refused the thing they
themselves just cancelled, so the stop is headed as a stop. It is
told apart by the flag the user set, not by the message.
"""
text = str(message)
if self._loading and self._stop.is_set():
self._stopped(text)
elif self._loading:
self._set_loading(False)
self._refuse("Crops could not be loaded", text)
else:
self._status.setText(text)
self._save_similar.setEnabled(
self._result_identity is not None and self._result is not None)
LOG.warning("embeddings job ended: %s", text)
def _stopped(self, detail: str) -> None:
"""Put the controls back after a stop, and head it as one.
Through :meth:`_refuse` for the body, because a stop unloads nothing
either: whatever was on the screen before is still there, still
encodable, and the panel has to say so.
"""
self._set_loading(False)
self._refuse("Loading stopped", detail)
[docs]
def is_busy(self) -> bool:
"""Whether crop loading, embedding, training or classifier application is in flight.
Includes the training and application runners until their workers have retired,
even after their final results have reached the screen.
"""
return (bool(self._jobs.active_jobs()) or self._loading
or self._dino_running or bool(self._dino_jobs.active_jobs())
or self._classifier_running or bool(self._classifier_jobs.active_jobs())
or self._classifier_apply_running or bool(self._classifier_apply_jobs.active_jobs()))
[docs]
def closeEvent(self, event): # noqa: N802 - Qt name
"""Discard pending results and drain all runners before the widget goes.
Crop loading and classifier training are asked to stop first; further
training progress is ignored.
Each ``shutdown`` waits a bounded time, then parks any worker that
outlasts it. A parked training worker can finish writing its checkpoint
without updating this closed GUI view; it is not forcibly terminated.
:param event: Qt's close event, passed to the base class unchanged.
The close is accepted even if a worker must be parked.
"""
self._stop.set()
self._classifier_stop.set()
self._dino_closed = True
self._cancel_classifier_application()
self._classifier_generation += 1
self._set_classifier_running(False)
self._dino_generation += 1
self._set_dino_running(False)
try:
self._classifier_apply_jobs.shutdown()
except Exception:
pass
try:
self._classifier_jobs.shutdown()
except Exception:
pass
try:
self._dino_jobs.shutdown()
except Exception:
pass
try:
self._jobs.shutdown()
except Exception: # noqa: BLE001
pass
super().closeEvent(event)
[docs]
def make_embeddings_screen(app_key: Optional[str] = None) -> QWidget:
"""Factory handed to :func:`spacr.qt.app.register_app`."""
return EmbeddingsScreen()
_ROW = declared_app(APP_KEY)
APP_NAME = _ROW.name
APP_DESCRIPTION = _ROW.desc
APP_INTRO = _ROW.intro
APP_CLI_NOTE = _ROW.cli_note
APP_NAME_TRANSLATIONS = _ROW.translations
[docs]
def register() -> bool:
"""Put Embeddings in the app registry. Idempotent.
Called from :data:`spacr.qt.SELF_REGISTERING_MODULES`, not at import, so
importing this module to reach :class:`EmbeddingsScreen` from a test or a
notebook does not mutate process-wide state.
``SECTION_DATA``, beside the other things that turn a table into numbers.
An embedding is a feature source, not an analysis: it produces columns
that the reductions, the regressions and the hit calling then consume
exactly as they consume the measured panel.
"""
from ..app import register_app
return register_app(APP_KEY, make_embeddings_screen)