"""Display and annotate image crops in the Qt interface.
Displays a paginated grid of clickable image thumbnails backed by
``png_list`` in ``measurements/measurements.db``. Left-click assigns value 1,
right-click assigns value 2, and clicking the assigned value again clears it.
Annotations are persisted through
:class:`spacr.qt.annotate_engine.SaveWorker`.
**Where it sits in the workflow.** Annotate is the third step of the
pipeline. It needs the crops Measure lists in ``png_list``, and the labels it
writes into an annotation column of that table are what Classify trains on
when ``dataset_mode`` is ``annotation``.
A keyboard-only rapid-annotation layer sits on top of the same write
path (see :meth:`AnnotateScreen.handle_key`): ``1``–``9`` assign a class
and auto-advance to the next unlabelled crop, ``0`` clears, arrows /
``hjkl`` move focus, ``Space``/``Backspace`` step without labelling,
``u`` undoes and ``Enter`` commits the page.
Every crop is drawn as a rounded square (see :class:`_Thumbnail`) with
two independent bands of colour, so its three states stay readable
together rather than overwriting each other:
* **resting** — a thin gray ring hugging the image
* **classified** — that same ring in the class colour
(:func:`~spacr.qt.annotate_engine.label_to_hex`, the app's one
class→colour map)
* **current** — an *extra* white ring outside it on the single tile the
next click or keystroke will hit
The cursor and the keyboard move the same current tile: entering a tile
makes it current, and an arrow key moves it away. There is no second
"hovered" highlight that could point somewhere else.
A PAGE IS WHAT FITS. The grid holds exactly the crops that fit
the room the crop pane gives it, at the crop size the settings ask for, and
it never scrolls: opening the console, folding a pane, changing the GUI
scale or resizing the window recomputes the page and pushes the rest to
the next one, the page counter follows, and the first crop on screen stays
where it was. See :func:`grid_that_fits` and
:meth:`AnnotateScreen._refit_grid`.
A SUGGESTION IS JUDGED, NOT RELABELLED. A crop Suggest proposed a
class for wears an amber ``?`` badge beside its dashed ring. A click or ``Y``
confirms it (it becomes an ordinary label and wears a green tick); a
right-click or ``N`` rejects it (it goes back to unanswered and wears a red
cross); ``U`` undoes either. The bar above the key legend says so, and
counts what the page and the column hold. Each judgement is written to the
``<column>_verdict`` column as ``+c`` or ``-c``, so it survives a restart,
and the next Suggest round is handed the rejections to train on. See
:meth:`AnnotateScreen._judge` and :func:`spacr.suggest.verdict_column`.
Shift + left click blows one crop up to fill the grid's container, drawn in
front of the tiles rather than reflowing them (:class:`_ZoomOverlay`); a
click beside it, or ``Escape``, folds it back. It is an EXTRA gesture --
plain left click goes on labelling, because that is what this screen is
for, and the class keys still land while a crop is open.
Annotator Agreement is folded onto this screen's masthead rather than
carrying a tile of its own: κ between annotation columns is a question
about the labels this screen writes, asked by the person who wrote them
while the crops are still in front of them. The button is that module's
own icon, lit on hover in its maturity colour, and it opens the agreement
screen itself as a PAGE beside the grid rather than a window over it — a
window is the last resort for a fold, and nothing here needs one. See
:data:`FOLDED_APPS`.
The Qt screen does not currently provide the UMAP window, Deep spaCR training
launcher, or measurement-threshold filtering. A threshold can be entered in
settings, but page queries do not apply it.
"""
from __future__ import annotations
import inspect
import logging
import os
import re
import threading
import time
import weakref
from copy import deepcopy
from collections import deque
from functools import partial
from typing import Any, Deque, Dict, List, Optional, Sequence, Tuple
from PySide6.QtCore import (
Qt,
QEvent,
QLocale,
QPointF,
QRect,
QRectF,
QSize,
QThread,
QTimer,
Signal,
Slot,
)
from PIL import Image
from PIL.ImageQt import ImageQt
from PySide6.QtGui import (QColor, QDoubleValidator, QFont, QImage,
QPainter, QPainterPath, QPen, QPixmap)
from PySide6.QtWidgets import (
QCheckBox,
QComboBox,
QDialog,
QDialogButtonBox,
QDoubleSpinBox,
QFileDialog,
QFormLayout,
QGridLayout,
QHBoxLayout,
QInputDialog,
QLabel,
QLineEdit,
QMenu,
QMessageBox,
QPlainTextEdit,
QPushButton,
QRubberBand,
QScrollArea,
QSizePolicy,
QSpinBox,
QStackedWidget,
QToolButton,
QVBoxLayout,
QWidget,
)
from ..widgets.percentile_pair import (
DECIMALS as PERCENTILE_DECIMALS)
from ..widgets.toggle import Toggle
from ..i18n import tr
from ..bridge import drain_thread
from ..job_runner import JobRunner
from ..annotate_engine import (
FILTER_CHANNELS,
FILTER_MEASURES,
AnnotateSettings,
OutlineCancelled,
SaveWorker,
class_counts,
clear_column,
count_rows,
ensure_annotation_column,
fetch_filtered_paths,
fetch_page,
filter_bound,
filter_channels_pil,
filter_key,
find_last_annotated_offset,
label_to_hex,
load_crop_image,
normalize_object_filters,
normalize_pil,
outline_image,
)
from ..linked_selection import (register_object_opener,
unregister_object_opener)
from .. import iconset, path_probe, prefs
from ..theme import SPACING, palette_for, register_widget_qss
from ..widgets.column_picker import attach_column_picker
from ..widgets import Divider, EmptyState
from ..widgets.fold_strip import FoldStrip
from ..widgets.collapsible_splitter import CollapsibleSplitter
from .map_barcodes import FoldOpener, restate_fold_button
from ..widgets.test_data_chooser import TestDataChooser # noqa: F401
#: Imported at module level rather than inside the handler because the tile
#: chrome reads it on every repaint. `spacr.suggest` pulls in pandas and
#: numpy and nothing heavier -- both are already imported by the time this
#: screen exists -- so this costs nothing at launch.
from ...suggest import SUGGESTION_OFFSET
LOG = logging.getLogger(__name__)
#: Registry keys of the modules folded onto this screen's masthead, in
#: the order the strip draws them. Annotator Agreement measures Cohen's
#: and Fleiss' κ between the annotation columns this screen writes and
#: walks the disagreements, so it is the same visit as annotating: the
#: person who wants it is already here, looking at the crops it disagrees
#: about.
FOLDED_APPS = ("agreement",)
#: Registry key of the screen this module hangs its strip on.
#:
#: MISSING UNTIL 2026-09-03, AND IT COST THE FOLD ITS PLACE EVERYWHERE.
#: `app.folded_children()` builds the one host-to-children map the dock, the
#: spaCR menu and the generated API page all read, and it identifies a host
#: by `APP_KEY` or `HOST_KEY`. This module declared neither, so it was
#: skipped -- `FOLDED_APPS` above was read by the fold STRIP, which is handed
#: the key by the screen, and by nothing else. Annotator Agreement therefore
#: had a button on this masthead and no nested row in the dock, no entry
#: under Annotate in the menu, and no line on the API page, while every other
#: host's children had all three.
#:
#: Found while wiring the nesting into the API homepage.
#: It is the same shape as the bug in the same commit's other half: one
#: mapping read by three surfaces, and a host that quietly is not in it.
HOST_KEY = "annotate"
#: How long a real answer about a folder is still worth acting on.
#:
#: An answer is proof the mount was awake when it was taken, not that it still
#: is -- an idle `autofs` share goes back to sleep (a typical one has
#: `timeout=600`), and a picker started in a folder that has since dozed off
#: is the original freeze again. Two minutes is comfortably inside any
#: plausible automount timeout and comfortably longer than opening Settings,
#: reading the form and pressing Browse, which is the sequence the head start
#: exists for. An expired answer is not wrong, only old: it stops being acted
#: on and a fresh check is queued.
VOUCH_TTL_S = 120.0
#: Most folders checked at once. These are stat calls on a handful of paths --
#: the open source, the one remembered from last session, whatever has been
#: typed into the settings dialog's source field -- so the cap is a guard
#: against a pathological caller rather than a throughput knob. A check turned
#: away by it costs the user the picker's head start and nothing else.
VOUCH_WORKERS = 4
#: Real answers only, keyed on the path: ``{path: (monotonic, is_dir)}``.
#: Written by the checking threads, read by the GUI thread, hence the lock.
_VOUCHED: Dict[str, Tuple[float, bool]] = {}
#: Paths a check is running for right now, so two presses of Browse do not
#: park two threads on the same sleeping mount.
_VOUCHING: set = set()
_VOUCH_LOCK = threading.Lock()
def _vouch_worker(path: str) -> None:
"""Stat one folder and record the answer, on a throwaway thread.
:param path: the folder to ask about.
:returns: nothing; :data:`_VOUCHED` is the output.
This is the only place in this module that touches a filesystem, and it
is never called on the GUI thread. It may park for as long as the kernel
takes -- a daemon thread stuck in a sleeping automount costs a stack,
where a stuck GUI thread costs the application -- and whenever it does
park, the absence of an entry in :data:`_VOUCHED` is itself the answer
:func:`_vouched_dir` needs.
"""
answer = False
try:
answer = os.path.isdir(path)
except OSError:
answer = False
finally:
with _VOUCH_LOCK:
_VOUCHED[path] = (time.monotonic(), bool(answer))
_VOUCHING.discard(path)
def _vouch_later(path) -> None:
"""Queue a truthful check of ``path``, and return immediately.
Call this wherever a folder is about to become a candidate for a file
dialog -- when a source is opened, when the settings dialog is built,
when its source field stops being edited -- so that the answer is in by
the time anybody presses a button. :func:`_vouched_dir` is the read.
:param path: the folder to ask about. Anything falsy is ignored.
:returns: nothing.
Never stats on the calling thread. A path whose answer is still fresh is
not asked again, and neither is one already being asked about.
"""
text = str(path or "").strip()
if not text:
return
now = time.monotonic()
with _VOUCH_LOCK:
if text in _VOUCHING:
return
held = _VOUCHED.get(text)
if held is not None and (now - held[0]) < VOUCH_TTL_S:
return
if len(_VOUCHING) >= VOUCH_WORKERS:
return
_VOUCHING.add(text)
threading.Thread(target=_vouch_worker, args=(text,), daemon=True,
name=f"spacr-annotate-vouch:{text[:40]}").start()
def _vouched_dir(path) -> bool:
"""True only for a folder a real stat came back and confirmed, recently.
The gate in front of every ``QFileDialog`` on this screen, and the reason
the module comment above says `path_probe` cannot serve here: a folder the
probe only ASSUMED was there, because its stat never returned, is the one
case where opening the picker in it reproduces the freeze.
:param path: the folder being considered as a starting directory.
:returns: whether it may be handed to a file dialog.
Reads a dict and the clock; it never stats and never blocks. A folder it
cannot vouch for costs the user the head start and nothing else -- the
picker opens at the working directory, which is local by construction, and
they can navigate wherever they like from there. A stale answer is
re-asked in the background so a folder in daily use keeps its head start
instead of losing it permanently the first time a mount was slow.
"""
text = str(path or "").strip()
if not text:
return False
with _VOUCH_LOCK:
held = _VOUCHED.get(text)
if held is None:
_vouch_later(text)
return False
when, answer = held
if (time.monotonic() - when) >= VOUCH_TTL_S:
_vouch_later(text)
return False
return bool(answer)
def _probe_isdir(path) -> bool:
""":func:`spacr.qt.path_probe.isdir`, for the CHEAP question only.
Whether to name a folder in the subtitle. Answers from `path_probe`'s
shared cache, never stats on the calling thread, and an unseen path comes
back ``False`` with a check queued -- so a caller that gates on it must
also subscribe to `path_probe.probes.answered` and run again, which
:meth:`AnnotateScreen._follow_path_probes` is.
Deliberately NOT the gate in front of a file dialog; see
:func:`_vouched_dir` and the module comment above for why the two
questions cannot share an answer.
"""
text = str(path or "").strip()
if not text:
return False
return path_probe.isdir(text)
def _ask_about_the_folder(path) -> None:
"""Put both questions about ``path`` to their own threads, and return.
The one call every warm-up site makes, so that no site has to remember
that there are two caches and which of them a button will read. Both
halves return immediately and neither stats on the calling thread.
Two moments call it. ONE, where the mount is demonstrably awake -- a
picker has just listed the folder, or a source has just been opened in
it -- so the next press of a Browse button starts there instead of at
the working directory. TWO, ahead of a button nobody has pressed yet --
the settings dialog is built, its source field stops being edited -- so
the answer is in by the time anybody reaches it. The second is why this
is a question and not a recording: `path_probe.prime` would be the call
if the answer were already in hand, and here it is not.
:param path: the folder to ask about. Anything falsy is ignored.
:returns: nothing.
"""
text = str(path or "").strip()
if not text:
return
_probe_isdir(text)
_vouch_later(text)
def _build_agreement(host_window) -> QWidget:
"""Annotator Agreement's own screen, unchanged.
It arrives as the screen the sidebar row used to open rather than as
a summary of it: the κ table, the pairwise matrix and the
disagreement review are all capabilities of that screen, and a fold
that reimplemented one of them would be a fold that lost the others.
:param host_window: the main window, unused -- this screen reaches
nothing outside itself.
"""
from .agreement import AgreementScreen
return AgreementScreen()
#: One builder per folded module, the same shape the settings-driven fold
#: hosts use — see :func:`spacr.qt.screens.map_barcodes.install_fold_strip`.
FOLD_BUILDERS = {"agreement": _build_agreement}
#: ``objectName`` of the canvas the crops are laid out on, and the name its
#: QSS block is registered under.
GRID_OBJECT_NAME = "AnnotateGrid"
#: ``objectName`` of the scroll viewport the canvas sits in.
#:
#: IT NEEDS A NAME BECAUSE A WIDGET-LOCAL SHEET IS INHERITED. A rule set on
#: the viewport with ``setStyleSheet("background: transparent")`` applies to
#: the viewport *and every descendant*, and a widget-local rule outranks the
#: application sheet — so the unselectored version of it silently blanked the
#: canvas inside it. Named, the rule can say "this widget" and mean it.
GRID_VIEWPORT_NAME = "AnnotateGridViewport"
#: ``objectName`` of the Console + AI switch on the bottom row.
CONSOLE_SWITCH_NAME = "AnnotateConsoleSwitch"
def _grid_backdrop_qss(palette, opacity=None) -> str:
"""The panel the thumbnails sit on: a page surface with round corners.
The corner is :data:`spacr.qt.theme.RADIUS`'s ``md``, which is what the
panels either side of it use — the settings cards, the tab panes and the
chart frames all round at the same number, and a square-cornered slab in
the middle of them reads as unfinished rather than as deliberate.
"""
from ..theme import RADIUS, block_surface
return f"""
QWidget#{GRID_OBJECT_NAME} {{
background: {block_surface("surface_alt", palette.get("theme"), opacity)};
border-radius: {RADIUS["md"]}px;
}}
QWidget#{GRID_VIEWPORT_NAME} {{
background: transparent;
}}
"""
def _console_switch_qss(palette, opacity=None) -> str:
"""The Console switch: text on the page, lit while the pane is open.
No plate. A ``QToolButton`` with no rule of its own is drawn by the
widget style from the palette's Button role, which is a dark slab behind
the caption — the "black box". Written as text alone, in the theme's own
foreground colour (white on the dark themes), and it takes the accent
while the console is open.
The lit state is a STATE and not a hover: it holds while nobody is
touching the button, the same way a checkable fold button holds its
stage fill (:mod:`spacr.qt.widgets.fold_strip`).
"""
return f"""
QToolButton#{CONSOLE_SWITCH_NAME} {{
background: transparent;
border: none;
padding: 3px 6px;
color: {palette["fg"]};
}}
QToolButton#{CONSOLE_SWITCH_NAME}:hover {{
color: {palette.get("accent_hi", palette["accent"])};
}}
QToolButton#{CONSOLE_SWITCH_NAME}:checked {{
color: {palette["accent"]};
}}
"""
register_widget_qss(GRID_OBJECT_NAME, _grid_backdrop_qss, replace=True)
register_widget_qss(CONSOLE_SWITCH_NAME, _console_switch_qss, replace=True)
BORDER_WIDTH = 2
HOVER_RING_WIDTH = 3
TILE_INSET = HOVER_RING_WIDTH + BORDER_WIDTH
TILE_RADIUS = 10
IMAGE_RADIUS = max(1, TILE_RADIUS - TILE_INSET)
UNDO_LIMIT = 128
CROP_PANE_MIN_HEIGHT = 140
#: How long `closeEvent` waits for a native worker before parking it, in ms.
#: Generous, because a Cellpose/PyTorch page decode or an sklearn fit really
#: can take this long, and interrupting one mid-write is the SIGSEGV this
#: screen's teardown is arranged to avoid. Bounded, because the alternative --
#: `QThread.wait()` with no argument -- is ULONG_MAX milliseconds, and a
#: wedged worker then hangs the close forever with the window still on screen.
CLOSE_DRAIN_MS = 15000
[docs]
def on_dark_theme() -> bool:
"""Is the app currently showing the dark theme?
The Annotate grid paints raw colours rather than being QSS-styled, so it
resolves this itself -- see :func:`tile_palette`, which does the same for
the tile chrome.
"""
try:
from ..preferences import resolve_effective_theme
return str(resolve_effective_theme()).lower() != "light"
except Exception:
return True
[docs]
def tile_palette() -> Dict[str, str]:
"""Palette for the theme the app is actually showing right now.
The Annotate grid paints raw colours (it is not QSS-styled), so it has
to resolve dark/light itself instead of importing the dark ``PALETTE``
at module scope — a hard-coded gray is invisible on one of the two
themes.
"""
try:
from ..preferences import resolve_effective_theme
return palette_for(resolve_effective_theme())
except Exception:
return palette_for("dark")
[docs]
def resting_border_color() -> str:
"""The thin gray line every unlabelled crop carries."""
return tile_palette()["border"]
[docs]
def current_ring_color() -> str:
"""Colour of the "this is the tile you are on" ring.
``fg`` is pure white on the (default) dark theme — exactly the white
border the feature asks for — and flips to the near-black foreground
on the light theme, where white would vanish into the background.
Either way it is a colour :func:`label_to_hex` can never produce
(class 1 is blue, 2 red, 3+ are HSV rotations at saturation 0.65), so
the current ring is never mistaken for a class.
"""
return tile_palette()["fg"]
JUDGEMENT_STATES = ("suggested", "confirmed", "rejected")
[docs]
def badge_colors(state: str, dark: Optional[bool] = None) -> Tuple[str, str]:
"""``(fill, glyph)`` for the corner badge a judged or judgeable crop wears.
A suggestion is amber, a confirmed suggestion green and a
rejected one red -- the theme's own warning, success and error colours,
so the badge follows a theme change the way the rest of the chrome does.
The glyph is drawn in the theme's background colour, which is black on
the dark theme and near-white on the light one; each of those fills was
chosen against it and reads at better than 4.5:1 in both. None of the
three is a class colour, so a badge is never read as a class.
:param state: one of :data:`JUDGEMENT_STATES`.
:param dark: which theme to answer for; the one on screen when None.
:returns: ``(fill, glyph)`` hex colours.
"""
if dark is None:
palette = tile_palette()
else:
palette = palette_for("dark" if dark else "light")
fill = {
"suggested": palette["warning"],
"confirmed": palette["success"],
"rejected": palette["error"],
}.get(str(state), palette["warning"])
return fill, palette["bg"]
[docs]
def verdict_contradicts(verdict: Optional[int], value: Optional[int]) -> bool:
"""True when a recorded judgement no longer agrees with the label.
A confirmation of class c stands while the crop is labelled c; a
rejection of class c stands until the crop is labelled c after all.
Relabelling a confirmed crop, or labelling a rejected one as the class
that was rejected, withdraws the judgement, so the verdict column never
says something the annotation column contradicts.
:param verdict: ``+c``, ``-c`` or None.
:param value: the crop's label, a suggestion or None.
:returns: True when the verdict should be withdrawn.
"""
if verdict is None:
return False
try:
verdict = int(verdict)
except (TypeError, ValueError):
return True
if verdict > 0:
return value is None or value != verdict
if verdict < 0:
return value is not None and value == -verdict
return True
[docs]
def grid_that_fits(width: int, height: int, tile_w: int, tile_h: int, *,
gap: int = 0, margin: int = 0) -> Tuple[int, int]:
"""``(rows, cols)`` of ``tile_w`` x ``tile_h`` tiles that fit a room.
Pure arithmetic, so a page size can be asserted without a window.
``margin`` comes off every edge of the ``width`` x ``height`` room,
``gap`` sits between tiles and not after the last one, and there is
always at least one row and one column: a room smaller than a tile
shows one crop rather than none.
:param width: the room's width, in the same pixels as the tiles.
:param height: the room's height.
:param tile_w: one tile's width, rings included.
:param tile_h: one tile's height, rings included.
:param gap: the layout's spacing between tiles.
:param margin: the layout's margin on each edge.
:returns: ``(rows, cols)``.
"""
cols = ((int(width) - 2 * int(margin) + int(gap))
// max(1, int(tile_w) + int(gap)))
rows = ((int(height) - 2 * int(margin) + int(gap))
// max(1, int(tile_h) + int(gap)))
return max(1, int(rows)), max(1, int(cols))
_TEXT_TOKENS = {
"left": "left", "right": "right", "up": "up", "down": "down",
"h": "left", "j": "down", "k": "up", "l": "right",
"space": "space",
"backspace": "backspace", "back": "backspace",
"u": "undo", "undo": "undo",
"enter": "enter", "return": "enter",
"?": "help", "help": "help",
"escape": "escape", "esc": "escape",
"y": "confirm", "n": "reject",
}
_QT_NAME_TOKENS = (
("Key_Left", "left"), ("Key_Right", "right"), ("Key_Up", "up"),
("Key_Down", "down"), ("Key_Space", "space"),
("Key_Backspace", "backspace"), ("Key_Return", "enter"),
("Key_Enter", "enter"), ("Key_Question", "help"),
("Key_Escape", "escape"),
)
def _qt_code_tokens() -> Dict[int, str]:
"""Build {Qt key code -> token} once, tolerating enum-shape differences."""
out: Dict[int, str] = {}
for name, token in _QT_NAME_TOKENS:
code = getattr(Qt, name, None)
if code is None:
code = getattr(getattr(Qt, "Key", None), name, None)
if code is None:
continue
try:
out[int(code)] = token
except (TypeError, ValueError):
continue
return out
_QT_CODE_TOKENS: Dict[int, str] = _qt_code_tokens()
def _token_from_text(text: str) -> Optional[str]:
"""Map a literal character or key name onto a canonical token."""
if text == " ":
return "space"
low = text.strip().lower()
if not low:
return None
if low in _TEXT_TOKENS:
return _TEXT_TOKENS[low]
if len(low) == 1 and low.isdigit():
return low
return None
[docs]
def key_token(key, text: str = "") -> Optional[str]:
"""Normalise ``key`` (Qt code, key name or character) to an action token.
Returns ``None`` for anything the annotate screen does not bind, so
callers can fall through to the default Qt handling.
:param key: Qt key code (an int or int-like value), a key name such as
``"Left"``, or a literal character such as ``"1"``.
"""
if isinstance(key, str):
token = _token_from_text(key)
return token if token is not None else (_token_from_text(text)
if text else None)
code: Optional[int]
try:
code = int(key)
except (TypeError, ValueError):
code = None
if code is not None:
token = _QT_CODE_TOKENS.get(code)
if token:
return token
if 0x30 <= code <= 0x39:
return chr(code)
if 0x41 <= code <= 0x5A:
token = _token_from_text(chr(code))
if token:
return token
return _token_from_text(text) if text else None
class _PageLoadWorker(QThread):
"""Loads + processes a page of thumbnail images OFF the GUI thread.
``_load_thumb_image`` (normalise + optional Otsu/Cellpose outline) is
expensive; running it inline froze the UI when settings changed. This runs
the whole page in a worker and emits the finished (PIL image, annotation)
list back to the main thread, which does only the cheap pixmap conversion.
``gen`` lets the screen ignore results from a superseded load.
"""
done = Signal(int, object)
def __init__(self, gen: int, paths: list, load_fn, parent=None):
"""Load one page of crops off the GUI thread.
:param gen: the page generation this worker belongs to. Carried back
with the result so a page the user has already left can be
discarded rather than drawn over the one they are looking at.
:param paths: the crops to load, in grid order.
:param load_fn: the callable that loads one crop. Whether it accepts
``should_stop`` is asked ONCE here rather than per crop, so the
answer stays out of the loop.
:param parent: parent object.
"""
super().__init__(parent)
self._gen = gen
self._paths = paths
self._load_fn = load_fn
try:
self._load_fn_stops = "should_stop" in inspect.signature(
load_fn).parameters
except (TypeError, ValueError):
self._load_fn_stops = False
def _stop_requested(self) -> bool:
"""Whether this page has been abandoned.
Handed DOWN into the outline code as ``should_stop`` rather than only
being consulted by the loop here. One crop's Cellpose outline is a
model construction plus one forward pass per channel, all of it native
and none of it interruptible, so a per-crop check answers minutes too
late: ``closeEvent`` gave up waiting and parked a QThread that was
still running, and destroying that wrapper aborts the process.
A destroyed C++ half reads as "stop" instead of raising, because a
worker whose wrapper has gone has certainly been abandoned.
"""
try:
return bool(self.isInterruptionRequested())
except RuntimeError:
return True
def run(self):
"""Decode one page of crops and hand them back, if the screen still exists.
THE EMIT IS INSIDE THE GUARD, AND THAT IS THE WHOLE POINT. ``emit`` and
``isInterruptionRequested`` are calls into this worker's C++ half, and
by the time a page finishes decoding the screen may be gone -- Qt
destroys the C++ object with its parent while this thread is still in
PIL. Both then raise, and raised here, outside any ``try``, the
exception escapes a ``QThread.run`` override: PySide6 prints "Error
calling Python override of QThread::run()" and the process aborts.
Caught in the full suite mid-``Image.resize``.
Nothing is lost by swallowing it: the only thing that branch does is
hand results to a screen that no longer exists. A page abandoned
mid-crop returns without emitting for the same reason -- the partial
list describes a page the screen has already moved off.
"""
try:
loaded = []
for row in self._paths:
if self._stop_requested():
return
if self._load_fn_stops:
loaded.append(
self._load_fn(row, should_stop=self._stop_requested))
else:
loaded.append(self._load_fn(row))
except OutlineCancelled:
return
except Exception:
loaded = []
try:
if not self.isInterruptionRequested():
self.done.emit(self._gen, loaded)
except RuntimeError:
pass
class _RetrainWorker(QThread):
"""Fit one active-learning round off the GUI thread.
``spacr.active_learning.retrain_round`` reads the whole feature matrix,
fits an estimator and writes scores back — seconds to a minute on a real
plate. Run inline it would freeze the grid mid-annotation, which is the
one thing this screen cannot afford, so it lives here and reports back
through signals.
``done``/``failed`` are ordinary signals connected to bound methods of
the screen, so Qt queues them onto the GUI thread. Nothing here touches a
widget.
"""
done = Signal(object)
failed = Signal(str)
def __init__(self, db_path: str, annotation_column: str,
options: Dict[str, object], parent=None):
"""Carry one retraining round's inputs onto a worker thread.
:param db_path: the annotation database to retrain from.
:param annotation_column: which column holds the labels.
:param options: keyword arguments passed through to
``active_learning.retrain_round``. COPIED, not referenced: the
caller's dict belongs to a widget that may be edited while this
runs, and a worker reading it mid-round would train on settings
nobody chose.
:param parent: parent object.
"""
super().__init__(parent)
self._db_path = db_path
self._column = annotation_column
self._options = dict(options)
def run(self):
"""Run one active-learning round and hand back the result.
Guarded at both ends for the reason ``_PageLoadWorker.run`` sets out: a
signal emitted at a destroyed C++ object raises out of ``run``, and an
exception out of a ``QThread.run`` override aborts the process. The
failure is surfaced rather than eaten -- a retrain that quietly did
nothing is worse than one that says why it could not.
"""
try:
from ... import active_learning as al
result = al.retrain_round(self._db_path, self._column,
**self._options)
except Exception as exc:
try:
self.failed.emit(f"{type(exc).__name__}: {exc}")
except RuntimeError:
pass
return
try:
if self.isInterruptionRequested():
return
self.done.emit(result)
except RuntimeError:
pass
def _similarity_source_stamp(paths):
"""Identify database and WAL changes that make a cached index stale."""
stamp = []
for path in paths:
for name in (str(path), f"{path}-wal"):
try:
stat = os.stat(name)
stamp.append((name, stat.st_dev, stat.st_ino,
stat.st_mtime_ns, stat.st_ctime_ns, stat.st_size))
except FileNotFoundError:
stamp.append((name, None, None, None, None, None))
return tuple(stamp)
class _SimilarityWorker(QThread):
"""Find the crops most like one crop, off the GUI thread.
The first search on a source reads every measurement table into one
feature matrix, which takes seconds on a real plate; the index it builds
is handed back with the answer so the screen can keep it and every later
search on the same source costs only the lookup.
"""
done = Signal(object)
failed = Signal(str)
def __init__(self, db_path: str, image_type: Optional[str], key: str,
index: Any = None, k: int = 100, parent=None, *,
unlabelled_only: bool = False, annotation_column: str = "annotate",
png_table: str = "png_list", pending_labels=None, writer=None,
db_paths=None, feature_kind: str = "auto"):
"""Carry one search's inputs onto a worker thread.
:param db_path: the database whose crops are searched.
:param image_type: substring filter on the crop key.
:param key: the ``png_path`` of the crop to match.
:param index: an index built earlier for the same source, or
``None`` to build one.
:param k: how many similar crops to return, excluding the query.
:param parent: parent object.
:param unlabelled_only: exclude committed labels, retaining cleared and proposed labels.
:param annotation_column: label column currently edited by Annotate.
:param png_table: crop table currently selected in Annotate.
:param pending_labels: unsaved local labels overriding stored values.
:param writer: existing save worker whose submitted batches must settle before reading.
:param db_paths: the open database and explicitly added plates.
:param feature_kind: auto, stored embeddings, or measurements.
"""
super().__init__(parent)
self._db_path = db_path
self._image_type = image_type
self._key = key
self._index = index
self._k = int(k)
self._unlabelled_only = bool(unlabelled_only)
self._annotation_column = annotation_column
self._png_table = png_table
self._pending_labels = dict(pending_labels or {})
self._writer = writer
self._db_paths = tuple(dict.fromkeys(
os.path.abspath(str(path)) for path in (db_paths or (db_path,))))
self._feature_kind = feature_kind
def _excluded_labels(self, index):
"""Read fresh human-label state without caching it with the feature index.
Annotate represents cleared labels by NULL or zero and unanswered model
proposals above SUGGESTION_OFFSET; none of these is a human answer.
"""
import sqlite3
from urllib.parse import quote
if not self._unlabelled_only:
return None
deadline = time.monotonic() + 30
while self._writer is not None and self._writer.pending_batches:
if self.isInterruptionRequested():
return None
if self._writer.last_error:
raise ValueError("Labels could not be saved; resolve the save error before searching unlabelled crops.")
if time.monotonic() >= deadline:
raise ValueError("Labels are still being saved; try the unlabelled search after saving finishes.")
time.sleep(0.02)
if self._writer is not None and self._writer.last_error:
raise ValueError("Labels could not be saved; resolve the save error before searching unlabelled crops.")
table = '"' + self._png_table.replace('"', '""') + '"'
column = '"' + self._annotation_column.replace('"', '""') + '"'
excluded = set()
multi = hasattr(index, "sources")
for path in self._db_paths if multi else (self._db_path,):
uri = f"file:{quote(os.path.abspath(path), safe='/')}?mode=ro"
with sqlite3.connect(uri, uri=True, timeout=30) as db:
fields = {row[1] for row in db.execute(f'PRAGMA table_info({table})')}
value = column if self._annotation_column in fields else 'NULL'
labels = dict(db.execute(f'SELECT png_path, {value} FROM {table}'))
if os.path.abspath(path) == os.path.abspath(self._db_path):
labels.update(self._pending_labels)
keys = (key for source, key in index.keys if source == path) if multi else index.keys
for key in keys:
value = labels.get(str(key))
if (str(key) not in labels or
value is not None and int(value) != 0
and int(value) <= SUGGESTION_OFFSET):
excluded.add((path, str(key)) if multi else str(key))
return excluded
def run(self):
"""Build the index if needed, search it, and hand back the hits."""
try:
from ... import active_learning as al
initial_stamp = _similarity_source_stamp(self._db_paths)
index = self._index
if index is None:
if len(self._db_paths) > 1:
index = al._multi_similarity_index(
self._db_paths, image_type=self._image_type,
feature_kind=self._feature_kind)
elif self._feature_kind == "measurements":
measured = al.round_features(self._db_path)
measured = measured[al._similarity_columns(measured.columns)]
index = al._similarity_index(
self._db_path, features=measured,
image_type=self._image_type)
elif self._feature_kind == "embeddings":
stored = al._stored_embeddings(self._db_path)
if stored is None:
raise ValueError("No stored crop embeddings were found in this source.")
index = al._similarity_index(
self._db_path, features=stored,
image_type=self._image_type)
else:
index = al._similarity_index(
self._db_path, image_type=self._image_type)
started = time.perf_counter()
excluded = self._excluded_labels(index)
if self.isInterruptionRequested():
return
hits = (index.like(self._db_path, self._key, self._k,
exclude=excluded) if len(self._db_paths) > 1
else index.like(self._key, self._k, exclude=excluded))
seconds = time.perf_counter() - started
source_stamp = _similarity_source_stamp(self._db_paths)
if source_stamp != initial_stamp:
source_stamp = initial_stamp
except Exception as exc:
try:
self.failed.emit(f"{type(exc).__name__}: {exc}")
except RuntimeError:
pass
return
try:
if self.isInterruptionRequested():
return
self.done.emit({"index": index, "hits": hits, "key": self._key,
"db_path": self._db_path,
"db_paths": self._db_paths,
"source_stamp": source_stamp,
"feature_kind": self._feature_kind,
"image_type": self._image_type,
"annotation_column": self._annotation_column,
"png_table": self._png_table,
"unlabelled_only": self._unlabelled_only,
"requested_k": self._k,
"seconds": seconds})
except RuntimeError:
pass
class _SuggestCancelled(Exception):
"""Raised inside a suggestion run when Cancel was pressed."""
class _SuggestWorker(QThread):
"""Fit a round, then write its opinion down as proposed labels.
BOTH HALVES OFF THE GUI THREAD, and the first half is why. Fitting is
the same work ``_RetrainWorker`` does -- seconds to a minute on a real
plate -- and a thirty-second block on the GUI thread is what the desktop
offers to force-quit; that happened on 2026-09-05 over a `urlopen` and
is not repeating over a model fit. The second half reads the whole crop
table into pandas, which is not free either.
IT FITS RATHER THAN READING A STALE SCORE. The request is "train a model
on your annotated images", and the labels made in the last ten minutes
are the ones that matter most; suggesting from the round before them
would propose labels the annotator has already moved past. Fitting here
also keeps ONE model behind both the suggestion and the queue's ranking,
because the round it fits is the round that writes the scores
:func:`spacr.suggest.suggest_from_scores` then reads.
FIVE STEPS, EACH ANNOUNCED. ``progress`` carries ``(step, STEPS, stage)``
before each of: clearing the outstanding suggestions, reading the
measurements, fitting, ranking, writing. Cancel is
``requestInterruption()``; it is honoured at the next step boundary and
the run then emits ``cancelled`` instead of ``done``. A cancel before
the writing step writes no suggestion; a cancel after the fit leaves
that round's scores, which only reorder the queue.
A SUGGESTION DOES NOT NEED A WELL-SEPARATED SCORE. The fit's held-out
check keeps wells apart, and labels from one page often put a whole
class in a single well, where no well-separated split exists. That is a
reason to distrust the round's accuracy, not to withhold the
suggestions, so the round is refitted on a random split, the round's
split rule says it is not grouped, and ``split_relaxed`` carries the
refusal so the screen can say why. The Retrain button keeps the refusal:
its product is the accuracy.
Nothing here touches a widget: the signals are ordinary signals
connected to bound methods of the screen, so Qt queues them onto the
GUI thread.
"""
STEPS = 5
done = Signal(object)
failed = Signal(str)
progress = Signal(int, int, str)
cancelled = Signal()
split_relaxed = Signal(str)
def __init__(self, db_path: str, annotation_column: str,
options: Dict[str, object], *,
png_table: str = "png_list",
only_paths: Optional[Sequence[str]] = None,
parent=None):
"""Carry one suggestion run's inputs onto a worker thread.
:param db_path: the annotation database to fit and write in.
:param annotation_column: which column holds the labels.
:param options: keyword arguments for ``retrain_round``. COPIED for
the reason ``_RetrainWorker.__init__`` gives: the caller's dict
belongs to a widget that may be edited while this runs.
:param png_table: the crop table.
:param only_paths: restrict the suggestions written to these crops,
which is the "the images on screen" scope. None means every
unannotated crop. COPIED, and for the same reason: the screen's
page turns while this runs.
:param parent: parent object.
"""
super().__init__(parent)
self._db_path = db_path
self._column = annotation_column
self._options = dict(options)
self._png_table = png_table
self._only = None if only_paths is None else [str(p)
for p in only_paths]
def _step(self, step: int, stage: str) -> None:
"""Stop here if Cancel was pressed, else announce the next step.
:param step: the step about to start, counted from 1.
:param stage: its name: ``clear``, ``features``, ``fit``, ``rank``
or ``write``.
:raises _SuggestCancelled: when interruption was requested.
"""
if self.isInterruptionRequested():
raise _SuggestCancelled()
try:
self.progress.emit(int(step), self.STEPS, str(stage))
except RuntimeError:
pass
def _fit(self, al, round_kwargs: Dict[str, object]) -> None:
"""Fit the round, on a random split when wells cannot be kept apart.
:param al: the ``spacr.active_learning`` module.
:param round_kwargs: keyword arguments for ``retrain_round``.
"""
from ...classifier_evaluation import _GroupedSplitImpossible
try:
al.retrain_round(self._db_path, self._column, **round_kwargs)
return
except _GroupedSplitImpossible as exc:
if str(round_kwargs.get("group_by", "well")).lower() in ("none",
"cell"):
raise
reason = str(exc)
relaxed = dict(round_kwargs)
relaxed["group_by"] = "none"
al.retrain_round(self._db_path, self._column, **relaxed)
try:
self.split_relaxed.emit(reason)
except RuntimeError:
pass
def run(self):
"""Fit, propose, write, and hand back what was proposed.
Guarded at both ends for the reason ``_RetrainWorker.run`` sets out:
an exception out of a ``QThread.run`` override aborts the process,
and a signal emitted at a destroyed C++ object raises out of ``run``.
Cancellation is honored before the write step. Once its transaction
commits, the saved result is reported even if Cancel arrived during the
write, since reporting a cancel would contradict the database.
"""
try:
from ... import active_learning as al
from ...suggest import (rejected_suggestions, resolve_suggestions,
suggest_from_scores, write_suggestions)
self._step(1, "clear")
resolve_suggestions(self._db_path, self._column, keep=False,
png_table=self._png_table)
rejections = rejected_suggestions(
self._db_path, self._column, png_table=self._png_table)
options = dict(self._options)
if rejections:
options["rejections"] = rejections
round_kwargs = {**options}
self._step(2, "features")
if round_kwargs.get("features") is None:
round_kwargs["features"] = al.round_features(
self._db_path,
table=str(round_kwargs.get("table", al.PNG_TABLE)),
key=str(round_kwargs.get("key", al.PNG_KEY)))
self._step(3, "fit")
self._fit(al, round_kwargs)
self._step(4, "rank")
proposal = suggest_from_scores(
self._db_path, self._column, png_table=self._png_table)
frame = proposal.frame
if self._only is not None and not frame.empty:
frame = frame[frame["png_path"].isin(set(self._only))]
frame = frame.reset_index(drop=True)
proposal.frame = frame
proposal.scored = int(len(frame))
self._step(5, "write")
written = 0
if not frame.empty:
written = write_suggestions(
self._db_path, self._column, frame,
png_table=self._png_table)
except _SuggestCancelled:
try:
self.cancelled.emit()
except RuntimeError:
pass
return
except Exception as exc:
try:
self.failed.emit(f"{type(exc).__name__}: {exc}")
except RuntimeError:
pass
return
try:
self.done.emit((proposal, written, len(rejections)))
except RuntimeError:
pass
def _retire(obj) -> bool:
"""Hand ``obj`` back to Qt for deletion, tolerating one already gone.
Every caller is a retirement path that runs LATER than the thread it is
retiring: a queued ``finished`` slot, or ``closeEvent`` after a bounded
drain. By then the worker's C++ half may already have been destroyed with
its parent, and ``deleteLater`` on that shell raises ``RuntimeError:
Internal C++ object already deleted``.
The disconnects beside each call site were already guarded against exactly
that state; the delete on the next line was not, so the guard stopped one
line short of the object it was written for and the exception surfaced out
of the Qt event loop instead -- a traceback with nothing the user can do
about it, in a slot whose whole job is tidying up.
:returns: True when Qt was asked to delete it, False when there was
nothing left to delete.
"""
try:
obj.deleteLater()
except RuntimeError:
return False
return True
class _TextReportDialog(QDialog):
"""A monospaced, scrollable, copyable text report.
The coverage table and the learning curve are wide, aligned text that a
``QMessageBox`` reflows into unreadable soup, and both are things a user
wants to paste into a lab notebook.
SHOWN, NEVER ``exec``-ED. ``QDialog.exec`` runs a NESTED event loop, and
``QCoreApplication.quit`` unwinds only the outermost one: closing the main
window while a report was open left the process alive with no window and
the GUI thread parked inside the report call for good. The same nested
loop also lets this dialog's own parent be destroyed while its ``exec``
is still on the stack, which is a delete of the object running the loop.
:meth:`AnnotateScreen._show_report` opens it as a plain window instead.
"""
def __init__(self, title: str, body: str, parent: Optional[QWidget] = None):
"""Show a report beside the grid rather than over it.
:param title: the window title.
:param body: the report text.
:param parent: parent widget, used for ownership only -- this opens
as a WINDOW in its own right so it can be moved and kept open
while annotating continues behind it.
"""
super().__init__(parent)
self.setWindowFlag(Qt.Window, True)
self.setWindowTitle(title)
self.resize(920, 620)
layout = QVBoxLayout(self)
view = QPlainTextEdit(self)
view.setReadOnly(True)
view.setLineWrapMode(QPlainTextEdit.NoWrap)
font = QFont("monospace")
font.setStyleHint(QFont.Monospace)
view.setFont(font)
view.setPlainText(body)
view.setProperty("i18nSkipText", True)
layout.addWidget(view, 1)
buttons = QDialogButtonBox(QDialogButtonBox.Close)
buttons.rejected.connect(self.reject)
buttons.accepted.connect(self.accept)
layout.addWidget(buttons)
self._view = view
def set_body(self, body: str) -> None:
"""Replace the report text, keeping the window where the user put it.
Pressing Coverage twice is asking for a fresher answer, not for a
second window: the reports are one per subject, so the open one is
rewritten rather than stacked behind a new one.
"""
self._view.setPlainText(body)
class _FieldQCDialog(QDialog):
"""Label whole fields for the learned image-quality classifier.
Steps through the raw ``.npy`` fields of a folder one at a time, shows a
percentile-stretched maximum projection, and records good or any of the
defect classes. Ticks are pre-filled from the classifier probabilities in
``<folder>/../qc/image_quality.json`` (or ``<folder>/qc``) when a screened
report is there. Save writes a field and label table, read back by the
``image_qc_classifier_labels`` setting to fine-tune and benchmark the
classifier.
"""
def __init__(self, parent: Optional[QWidget] = None):
"""Build an empty view; :meth:`load_folder` fills it.
:param parent: owning widget.
"""
from ...image_quality import _QC_CLASSES
super().__init__(parent)
self.setObjectName("AnnotateFieldQCDialog")
self.setWindowFlag(Qt.Window, True)
self.setWindowTitle(tr("Field quality labels"))
self.resize(560, 620)
self._classes = tuple(_QC_CLASSES)
self._fields: List[str] = []
self._labels: Dict[str, set] = {}
self._suggested: Dict[str, Dict[str, float]] = {}
self._index = 0
self._folder = ""
layout = QVBoxLayout(self)
self._image = QLabel(self)
self._image.setAlignment(Qt.AlignCenter)
self._image.setMinimumSize(320, 320)
layout.addWidget(self._image, 1)
self._caption = QLabel(self)
self._caption.setProperty("i18nSkipText", True)
layout.addWidget(self._caption)
names = {'out_of_focus': tr("Out of focus"), 'saturated': tr("Saturated"),
'debris': tr("Debris"), 'bubble': tr("Bubble"), 'empty': tr("Empty")}
self._good = QCheckBox(tr("Good"), self)
self._good.toggled.connect(self._on_good)
layout.addWidget(self._good)
self._boxes: Dict[str, QCheckBox] = {}
for name in self._classes:
box = QCheckBox(names.get(name, name), self)
box.toggled.connect(self._on_defect)
self._boxes[name] = box
layout.addWidget(box)
row = QHBoxLayout()
self._back = QPushButton(tr("Back"), self)
self._back.clicked.connect(lambda: self.show_field(self._index - 1))
self._next = QPushButton(tr("Next"), self)
self._next.clicked.connect(lambda: self.show_field(self._index + 1))
self._save = QPushButton(tr("Save labels"), self)
self._save.clicked.connect(self.save)
for button in (self._back, self._next, self._save):
row.addWidget(button)
layout.addLayout(row)
self._status = QLabel(self)
self._status.setProperty("i18nSkipText", True)
layout.addWidget(self._status)
self._syncing = False
def labels_path(self) -> str:
"""Where :meth:`save` writes the table: ``<folder>/qc/image_qc_labels.csv``."""
return os.path.join(self._folder, "qc", "image_qc_labels.csv")
def load_folder(self, folder: str) -> int:
"""List the folder's ``.npy`` fields and read saved labels and suggestions.
:param folder: a folder of raw fields, such as a plate's ``stack``.
:returns: how many fields were found.
"""
import json
from pathlib import Path
from ...image_quality import REPORT, _parse_qc_labels
from ...tabular import read_table
self._folder = str(folder)
self._fields = sorted(p.name for p in Path(folder).glob("*.npy"))
self._labels, self._suggested = {}, {}
for root in (Path(folder).parent, Path(folder)):
report = root / REPORT
if not report.is_file():
continue
try:
fields = json.loads(report.read_text(encoding="utf-8"))["fields"]
except (OSError, ValueError, KeyError):
continue
for entry in fields:
for record in entry.get("channels", [])[:1]:
scores = {n: float(record[f"p_{n}"]) for n in self._classes
if f"p_{n}" in record}
if scores:
self._suggested[entry["field"]] = scores
break
if os.path.isfile(self.labels_path()):
table = read_table(self.labels_path(), report=None)
table.columns = [str(c).strip().lower() for c in table.columns]
table = table.rename(columns={"fieldid": "field"})
for row in table.to_dict("records"):
self._labels[str(row["field"])] = _parse_qc_labels(row["label"])
self.show_field(0)
return len(self._fields)
def show_field(self, index: int) -> None:
"""Show field ``index`` with its saved or suggested labels ticked.
:param index: position in the folder's sorted field list; clamped.
"""
if not self._fields:
self._caption.setText(tr("No .npy fields in this folder."))
return
self._index = max(0, min(int(index), len(self._fields) - 1))
name = self._fields[self._index]
from ..hidpi import scaled_for
self._image.setPixmap(scaled_for(
QPixmap.fromImage(self._field_image(name)), self._image,
self._image.minimumSize()))
scores = self._suggested.get(name, {})
chosen = self._labels.get(name)
if chosen is None:
from ...image_quality import DEFAULTS
threshold = DEFAULTS["image_qc_classifier_threshold"]
chosen = {n for n, p in scores.items() if p >= threshold}
self._syncing = True
for n, box in self._boxes.items():
box.setChecked(n in chosen)
self._good.setChecked(not chosen)
self._syncing = False
hint = ", ".join(f"{n} {p:.2f}" for n, p in scores.items())
self._caption.setText(f"{self._index + 1}/{len(self._fields)} {name}"
+ (f" ({hint})" if hint else ""))
def _field_image(self, name: str) -> QImage:
"""A percentile-stretched 8-bit maximum projection of one field."""
import numpy as np
array = np.load(os.path.join(self._folder, name), mmap_mode="r",
allow_pickle=False)
plane = np.asarray(array, dtype=np.float32)
while plane.ndim > 2:
plane = plane.max(axis=-1)
low, high = np.percentile(plane, (1, 99.5))
plane = np.clip((plane - low) / max(float(high - low), 1e-6), 0, 1)
data = np.ascontiguousarray((plane * 255).astype(np.uint8))
image = QImage(data.data, data.shape[1], data.shape[0], data.strides[0],
QImage.Format_Grayscale8)
return image.copy()
def current_labels(self) -> set:
"""The defect classes ticked for the shown field; empty means good."""
return {n for n, box in self._boxes.items() if box.isChecked()}
def _remember(self) -> None:
"""Store the ticks of the shown field."""
if self._fields and not self._syncing:
self._labels[self._fields[self._index]] = self.current_labels()
def _on_good(self, checked: bool) -> None:
"""Good clears every defect tick."""
if checked and not self._syncing:
self._syncing = True
for box in self._boxes.values():
box.setChecked(False)
self._syncing = False
self._remember()
def _on_defect(self, checked: bool) -> None:
"""A defect tick clears Good; no tick at all means good again."""
if not self._syncing:
self._syncing = True
self._good.setChecked(not self.current_labels())
self._syncing = False
self._remember()
def save(self) -> str:
"""Write every labelled field as a field and label table.
:returns: the written path, or ``""`` with nothing labelled.
"""
import pandas as pd
from ...tabular import write_table
rows = [dict(field=name, label=";".join(sorted(chosen)) or "good")
for name, chosen in sorted(self._labels.items())]
if not rows:
self._status.setText(tr("Nothing labelled yet."))
return ""
os.makedirs(os.path.dirname(self.labels_path()), exist_ok=True)
write_table(pd.DataFrame(rows), self.labels_path())
self._status.setText(tr("Saved {n} field labels to {path}").format(
n=len(rows), path=self.labels_path()))
return self.labels_path()
class _Thumbnail(QLabel):
"""One crop in the grid: a rounded square wearing up to two rings.
Everything is drawn in :meth:`paintEvent`, which is what makes the
borders cheap: changing a border is one ``update()`` on ONE widget, not
a rebuilt pixmap. The pixmap handed to ``setPixmap`` is the bare crop —
the rounded corners come from clipping it here, so the corner is
actually round instead of a rounded frame sitting on a square image.
Three visual states, drawn in two separate bands so they compose
instead of overwriting each other:
* resting — thin ``resting_border_color()`` gray ring
* classified — the same ring, recoloured to the class colour
* current (cursor is on it, or the keyboard is) — an ADDITIONAL white
ring outside the first one, leaving the class colour untouched
"""
left_clicked = Signal(int)
right_clicked = Signal(int)
#: Shift + left click. An EXTRA gesture: plain left click still labels,
#: because annotating is the primary action and a crop the user wanted to
#: look at closely is the exception rather than the rule.
shift_clicked = Signal(int)
hover_changed = Signal(int, bool)
def __init__(self, slot: int, parent: Optional[QWidget] = None,
border_color: Optional[str] = None,
ring_color: Optional[str] = None):
"""Build one grid cell.
:param slot: this cell's fixed position in the grid, which is how
the screen addresses it when a page of crops arrives.
:param parent: parent widget.
:param border_color: the resting border, or ``None`` to look it up.
:param ring_color: the ring drawn on the current cell, or ``None``
to look it up.
BOTH COLOURS ARE PASSED DOWN, not looked up per cell: the screen
resolves them once per grid rebuild so the hover path never touches
a palette.
"""
super().__init__(parent)
self.slot = slot
self._border_color = border_color or resting_border_color()
self._ring_color = ring_color or current_ring_color()
self._current = False
self._occupied = False
#: A machine's proposal rather than the annotator's answer. Drawn as
#: the SAME class colour with a dashed ring, never as a colour of its
#: own: a third colour on a two-class screen reads as a third class,
#: which is the one thing a suggestion must not look like.
self._suggested = False
self._badge: Optional[str] = None
self._badge_colors: Tuple[str, str] = ("", "")
self.setAlignment(Qt.AlignCenter)
self.setSizePolicy(QSizePolicy.Preferred, QSizePolicy.Preferred)
self.setStyleSheet("background: transparent;")
self.setProperty("kbdFocused", False)
def border_color(self) -> str:
"""Colour of the ring hugging the image: resting gray or class colour."""
return self._border_color
def ring_color(self) -> Optional[str]:
"""The current-tile ring colour, or ``None`` when this isn't it."""
return self._ring_color if self._current else None
def outline_color(self) -> str:
"""Outermost colour drawn — what a user would call "the border"."""
return self._ring_color if self._current else self._border_color
def is_current(self) -> bool:
"""True when this is the one tile the next action applies to."""
return self._current
def is_occupied(self) -> bool:
"""True when this cell holds a crop (empty cells draw nothing)."""
return self._occupied
def set_border_color(self, color: Optional[str]) -> bool:
"""Recolour the state ring; returns True when a repaint was needed."""
color = str(color or resting_border_color())
if color.lower() == self._border_color.lower():
return False
self._border_color = color
self.update()
return True
def set_current(self, on: bool) -> bool:
"""Add/remove the current-tile ring; returns True when it changed."""
on = bool(on)
self.setProperty("kbdFocused", on)
if on == self._current:
return False
self._current = on
self.update()
return True
def set_occupied(self, on: bool) -> bool:
"""Mark whether this cell holds a crop; empty cells paint nothing."""
on = bool(on)
if on == self._occupied:
return False
self._occupied = on
self.update()
return True
def is_suggested(self) -> bool:
"""True when this cell's label was proposed, not decided."""
return self._suggested
def set_suggested(self, on: bool) -> bool:
"""Draw this cell's ring dashed; returns True when it changed.
Mirrored onto a Qt property as well as the field so a stylesheet and
a test can both ask, the way ``set_current`` does -- and so "is this
a suggestion" has exactly one answer per tile.
"""
on = bool(on)
self.setProperty("suggested", on)
if on == self._suggested:
return False
self._suggested = on
self.update()
return True
def badge(self) -> Optional[str]:
"""The judgement badge this cell wears, or None (item 512).
``"suggested"`` on a proposal still to judge, ``"confirmed"`` or
``"rejected"`` once the annotator has judged it.
"""
return self._badge
def set_badge(self, state: Optional[str],
colors: Optional[Tuple[str, str]] = None) -> bool:
"""Show, change or remove the judgement badge; True when it changed.
Mirrored onto the ``judgement`` Qt property the way
:meth:`set_suggested` mirrors ``suggested``, so a test and a
stylesheet ask the same question.
:param state: one of :data:`JUDGEMENT_STATES`, or None for no badge.
:param colors: ``(fill, glyph)``; looked up from the theme on screen
when omitted. Passed down by the screen so a theme change reaches
the badge on the next repaint of the cell.
"""
state = state if state in JUDGEMENT_STATES else None
colors = tuple(colors) if colors else (
badge_colors(state) if state else ("", ""))
self.setProperty("judgement", state)
if state == self._badge and colors == self._badge_colors:
return False
self._badge = state
self._badge_colors = colors
self.update()
return True
def badge_rect(self) -> QRectF:
"""Where the badge is drawn: the crop's top right corner, inside it."""
w = float(self.width())
h = float(self.height())
size = max(12.0, min(22.0, min(w, h) * 0.3))
inset = float(TILE_INSET) + 2.0
return QRectF(w - inset - size, inset, size, size)
def _paint_badge(self, painter: QPainter) -> None:
"""Draw the judgement badge: a disc with a tick, a cross or a ``?``.
The marks are stroked rather than typed, so a font without a tick
glyph cannot turn a confirmation into a missing-glyph box.
:param painter: the painter :meth:`paintEvent` is drawing with.
"""
state = self._badge
if state is None:
return
fill, glyph = self._badge_colors
box = self.badge_rect()
if box.width() <= 0 or self.width() < box.width() * 2:
return
outline = QPen(QColor(glyph))
outline.setWidthF(1.5)
painter.setPen(outline)
painter.setBrush(QColor(fill))
painter.drawEllipse(box)
pen = QPen(QColor(glyph))
pen.setWidthF(max(1.6, box.width() / 8.0))
pen.setCapStyle(Qt.RoundCap)
pen.setJoinStyle(Qt.RoundJoin)
painter.setPen(pen)
painter.setBrush(Qt.NoBrush)
c = box.center()
r = box.width() * 0.24
if state == "confirmed":
path = QPainterPath()
path.moveTo(c.x() - r, c.y() + r * 0.05)
path.lineTo(c.x() - r * 0.3, c.y() + r * 0.7)
path.lineTo(c.x() + r, c.y() - r * 0.65)
painter.drawPath(path)
elif state == "rejected":
painter.drawLine(QPointF(c.x() - r * 0.8, c.y() - r * 0.8),
QPointF(c.x() + r * 0.8, c.y() + r * 0.8))
painter.drawLine(QPointF(c.x() - r * 0.8, c.y() + r * 0.8),
QPointF(c.x() + r * 0.8, c.y() - r * 0.8))
else:
path = QPainterPath()
path.moveTo(c.x() - r * 0.6, c.y() - r * 0.45)
path.cubicTo(c.x() - r * 0.6, c.y() - r * 1.15,
c.x() + r * 0.6, c.y() - r * 1.15,
c.x() + r * 0.6, c.y() - r * 0.45)
path.cubicTo(c.x() + r * 0.6, c.y() - r * 0.05,
c.x(), c.y() - r * 0.05,
c.x(), c.y() + r * 0.35)
painter.drawPath(path)
painter.setBrush(QColor(glyph))
painter.setPen(Qt.NoPen)
dot = max(1.2, pen.widthF() * 0.65)
painter.drawEllipse(QPointF(c.x(), c.y() + r * 0.85), dot, dot)
def paintEvent(self, event): # noqa: N802 (Qt naming)
"""Draw the clipped crop, the state ring, the ring and the badge."""
if not self._occupied:
return
w = float(self.width())
h = float(self.height())
painter = QPainter(self)
try:
painter.setRenderHint(QPainter.Antialiasing, True)
painter.setBrush(Qt.NoBrush)
inner = QRectF(TILE_INSET, TILE_INSET,
max(0.0, w - 2 * TILE_INSET),
max(0.0, h - 2 * TILE_INSET))
pm = self.pixmap()
if pm is not None and not pm.isNull() \
and inner.width() > 0 and inner.height() > 0:
clip = QPainterPath()
clip.addRoundedRect(inner, IMAGE_RADIUS, IMAGE_RADIUS)
painter.save()
painter.setClipPath(clip)
painter.drawPixmap(_cover_rect(pm, inner), pm,
QRectF(pm.rect()))
painter.restore()
self._stroke(painter, self._border_color, BORDER_WIDTH,
HOVER_RING_WIDTH + BORDER_WIDTH / 2.0, w, h,
dashed=self._suggested)
if self._current:
self._stroke(painter, self._ring_color, HOVER_RING_WIDTH,
HOVER_RING_WIDTH / 2.0, w, h)
self._paint_badge(painter)
finally:
painter.end()
@staticmethod
def _stroke(painter: QPainter, color: str, width: int, inset: float,
w: float, h: float, dashed: bool = False) -> None:
"""Stroke one rounded rect inset by ``inset`` from the widget edge.
``dashed`` marks a suggested label. The COLOUR is unchanged -- the
dash pattern is the whole difference -- because a suggested 1 is a
proposal about class 1 and must read as one. Giving it its own
colour would put a third swatch on a two-class screen and invite the
reading that there is a third class.
"""
if w - 2 * inset <= 0 or h - 2 * inset <= 0:
return
pen = QPen(QColor(color))
pen.setWidth(width)
if dashed:
pen.setStyle(Qt.CustomDashLine)
pen.setDashPattern([2.0, 2.0])
painter.setPen(pen)
radius = max(1.0, TILE_RADIUS - inset)
painter.drawRoundedRect(
QRectF(inset, inset, w - 2 * inset, h - 2 * inset),
radius, radius)
def mousePressEvent(self, event):
"""Route the mouse to typed signals; ignore buttons with no meaning.
Shift is read before the plain left click rather than after, so the
zoom gesture cannot also drop a label on the crop it opens.
"""
if event.button() == Qt.LeftButton:
if event.modifiers() & Qt.ShiftModifier:
self.shift_clicked.emit(self.slot)
else:
self.left_clicked.emit(self.slot)
elif event.button() == Qt.RightButton:
self.right_clicked.emit(self.slot)
else:
super().mousePressEvent(event)
def enterEvent(self, event): # noqa: N802 (Qt naming)
"""Cursor arrived — tell the screen this tile is now the current one."""
self.hover_changed.emit(self.slot, True)
super().enterEvent(event)
def leaveEvent(self, event): # noqa: N802 (Qt naming)
"""Cursor left — the screen drops the hover if it still points here."""
self.hover_changed.emit(self.slot, False)
super().leaveEvent(event)
class _ZoomOverlay(QWidget):
"""One crop blown up to fill the grid's container, in front of the grid.
IN FRONT, not instead of. The overlay is a child of the scroll
viewport and is raised above the canvas the tiles are laid out on, so
nothing reflows: the grid keeps its geometry, the page keeps its
scroll position, and folding the crop back is a repaint rather than a
rebuild.
Clicks are split by where they land. Inside the picture they are
swallowed — the annotator is looking at it, and a stray click there
must not label the crop underneath or dismiss the thing they opened.
Outside it the overlay folds back, which is the whole way out of the
gesture along with ``Escape``.
:param parent: parent widget; ownership only.
"""
#: A click landed off the picture: the caller should fold it back.
dismissed = Signal()
def __init__(self, parent: Optional[QWidget] = None):
"""Build an empty overlay, bound to no slot yet."""
super().__init__(parent)
self._pixmap: Optional[QPixmap] = None
self.slot: int = -1
self.setCursor(Qt.PointingHandCursor)
self.hide()
def show_pixmap(self, pixmap: QPixmap, slot: int) -> None:
"""Take over the container with ``pixmap``, drawn for ``slot``."""
self._pixmap = pixmap
self.slot = int(slot)
self.show()
self.raise_()
self.update()
def picture_rect(self) -> QRectF:
"""Where the crop is actually drawn, in this widget's coordinates.
The whole container minus a margin, at the crop's own aspect ratio.
The margin is what a "click outside" has to land in, so it is real
space rather than nothing: an overlay drawn edge to edge would leave
a user holding a picture with no way out but the keyboard.
"""
pm = self._pixmap
if pm is None or pm.isNull():
return QRectF()
margin = float(SPACING["md"])
box_w = max(1.0, self.width() - 2 * margin)
box_h = max(1.0, self.height() - 2 * margin)
scale = min(box_w / pm.width(), box_h / pm.height())
w = pm.width() * scale
h = pm.height() * scale
return QRectF(margin + (box_w - w) / 2.0,
margin + (box_h - h) / 2.0, w, h)
def paintEvent(self, event): # noqa: N802 (Qt naming)
"""Dim the grid, then draw the crop over it with round corners."""
pm = self._pixmap
if pm is None or pm.isNull():
return
painter = QPainter(self)
try:
painter.setRenderHint(QPainter.Antialiasing, True)
painter.setRenderHint(QPainter.SmoothPixmapTransform, True)
painter.fillRect(self.rect(), QColor(0, 0, 0, 170))
box = self.picture_rect()
if box.isEmpty():
return
clip = QPainterPath()
clip.addRoundedRect(box, TILE_RADIUS, TILE_RADIUS)
painter.setClipPath(clip)
painter.drawPixmap(box, pm, QRectF(pm.rect()))
painter.setClipping(False)
pen = QPen(QColor(current_ring_color()))
pen.setWidth(BORDER_WIDTH)
painter.setPen(pen)
painter.setBrush(Qt.NoBrush)
painter.drawRoundedRect(box, TILE_RADIUS, TILE_RADIUS)
finally:
painter.end()
def mousePressEvent(self, event): # noqa: N802 (Qt naming)
"""Swallow clicks on the picture; fold back on the ones beside it."""
if not self.picture_rect().contains(event.position()):
self.dismissed.emit()
event.accept()
def keyPressEvent(self, event): # noqa: N802 (Qt naming)
"""Escape folds the crop back; everything else goes to the grid."""
from ..shortcuts import _screen_event_key
owner = self.parentWidget()
while owner is not None and getattr(owner, "_spacr_screen_scope", None) != "Annotate":
owner = owner.parentWidget()
key = _screen_event_key(owner, event) if owner is not None else event.key()
if key == Qt.Key_Escape:
self.dismissed.emit()
event.accept()
return
event.ignore()
def _csv_to_list(text: str) -> Optional[List[str]]:
"""Parse a comma-separated string into a stripped list, or ``None`` when empty."""
parts = [p.strip() for p in text.split(",") if p.strip()]
return parts or None
def _list_to_csv(vals: Optional[List[str]]) -> str:
"""Format a list as a comma-separated string; empty/None becomes ``""``."""
return ", ".join(str(v) for v in vals) if vals else ""
def _filter_text(value) -> str:
"""One filter bound as a user would type it; ``None`` becomes empty.
``%g`` rather than ``str``: the bounds are floats internally and a field
that read back ``200.0`` after the user typed ``200`` looks like the
dialog corrected them.
"""
if value is None:
return ""
try:
return f"{float(value):g}"
except (TypeError, ValueError):
return ""
#: What each of the six filter rows is called on the settings form. The
#: colour leads, because that is what the annotator is choosing between --
#: the plane holding the nuclei, not the second row of the third group.
FILTER_ROW_LABELS = {
("r", "area"): "Red area",
("r", "intensity"): "Red intensity",
("g", "area"): "Green area",
("g", "intensity"): "Green intensity",
("b", "area"): "Blue area",
("b", "intensity"): "Blue intensity",
}
#: The sentence every one of the twelve fields carries. EMPTY IS THE POINT:
#: it is the only way to say "no bound on this side", and a user who cannot
#: find the off switch fills a zero in instead, which is a different filter.
FILTER_FIELD_TIP = (
"The window an object must fall inside to be outlined in this colour. "
"Area is the object's size in pixels; intensity is its mean brightness "
"in that colour, 0\u2013255.\n\nLEAVE A FIELD EMPTY for no bound on "
"that side. An empty field is how half a filter is turned off, and it "
"is not the same as a zero.")
def _cover_rect(pm: QPixmap, box: QRectF) -> QRectF:
"""Rect to draw ``pm`` into so it fills ``box`` at its own aspect ratio.
Crops rather than letterboxes (the clip path trims the overflow), so a
tile is always a complete rounded square with no canvas showing through
at the edges. Crops normally arrive already resized to the box, in
which case this is the identity.
"""
pw = float(pm.width())
ph = float(pm.height())
if pw <= 0 or ph <= 0:
return box
scale = max(box.width() / pw, box.height() / ph)
w = pw * scale
h = ph * scale
return QRectF(box.x() + (box.width() - w) / 2.0,
box.y() + (box.height() - h) / 2.0, w, h)
def _reanchor_png_path(path: str, db_path: str) -> str:
"""Re-anchor a stored ``png_path`` against the opened database's location.
The measurements DB records absolute png paths built at measure time. If the
dataset was moved (or measure ran with a relative ``src``), those paths no
longer resolve and the Annotate grid shows grey placeholders instead of
images. The DB always sits at ``<root>/measurements/measurements.db`` beside
the ``<root>/data/...`` crops, so when the stored path fails we rebuild it
from the ``/data/`` segment onward under this DB's own root.
"""
if not path or os.path.isfile(path):
return path
if not db_path:
return path
root = os.path.dirname(os.path.dirname(os.path.abspath(db_path)))
norm = str(path).replace("\\", "/")
i = norm.rfind("/data/")
if i != -1:
cand = os.path.join(root, norm[i + 1:])
if os.path.isfile(cand):
return cand
if norm.startswith("data/"):
cand = os.path.join(root, norm)
if os.path.isfile(cand):
return cand
return path
def _load_thumb_image_worker(row, src, settings, should_stop=None):
"""Load one thumbnail from an immutable page-request snapshot.
This function deliberately receives no :class:`AnnotateScreen`. Calling a
bound QWidget method from a worker kept the screen wrapper alive after its
C++ object had been destroyed and let background threads read settings
while the GUI thread replaced them.
:param should_stop: passed through to :func:`outline_image`, which asks it
before every Cellpose call. A Cellpose outline is native work that
cannot be stopped once it has started, and a page of them went on
running for minutes after the screen had asked its worker to stop --
long enough that the close gave up waiting and parked a QThread that
was still running. Handing the interruption flag down here is what
lets a page be abandoned between calls instead.
"""
if isinstance(row, dict):
annotation = row.get("annotation")
else:
path, annotation = row
row = {"png_path": path}
s = settings
if src is not None and getattr(src, "kind", "png") == "merged":
try:
img = Image.fromarray(src.get(row)).convert("RGB")
except Exception:
return Image.new("RGB", s.image_size, (30, 30, 30)), annotation
else:
path = _reanchor_png_path(row.get("png_path"), s.db_path)
if not path or not os.path.isfile(path):
return Image.new("RGB", s.image_size, color=(20, 20, 20)), annotation
try:
img = load_crop_image(
path, db_path=s.db_path,
stored_channel_order=getattr(
s, "stored_channel_order", "rgb"),
display_order=getattr(s, "display_order", "rgb"),
display_primaries=getattr(s, "display_primaries", "rgb"))
except Exception:
return Image.new("RGB", s.image_size, (30, 30, 30)), annotation
img = normalize_pil(img, s.percentiles, s.normalize_channels)
full_img = img
img = filter_channels_pil(img, s.channels)
if s.outline:
try:
img = outline_image(
base_img=img,
full_img=full_img,
outline_channels=s.outline,
edge_sigma=s.outline_sigma,
edge_thickness=s.edge_thickness,
edge_transparency=s.edge_transparency,
edge_image=s.edge_image,
outline_threshold_factor=s.outline_threshold_factor,
object_size=s.object_size,
object_filters=getattr(s, "object_filters", None),
outline_method=getattr(s, "outline_method", "otsu"),
should_stop=should_stop,
)
except OutlineCancelled:
raise
except Exception:
pass
return img.resize(s.image_size), annotation
def _compute_total(s: AnnotateSettings, filter_active: bool) -> dict:
"""How many objects the grid is paging through, and which ones.
Runs on a worker thread, and is module-level rather than a method so that
it *cannot* reach a widget: everything it needs arrives in ``s``, and
everything it produces goes back as a plain dict for
:meth:`AnnotateScreen._apply_total` to paint.
The expensive branch is the last one. ``fetch_filtered_paths`` joins every
measurement table in the database into a single pandas frame through
:func:`spacr.io._read_and_join_tables` before it can apply a threshold --
2.6 s on a 60 000-object database, measured, and it used to run inline on
every settings apply.
:param s: a frozen copy of the screen's settings.
:param filter_active: whether a measurement/threshold filter is set,
decided by the screen because it is a question about its own controls.
:returns: ``filtered_rows``, ``total``, ``queue_summary`` and an optional
``note`` for the page label.
"""
if s.queue_by_uncertainty:
from ... import active_learning as al
try:
queue = al.build_queue(
s.db_path, s.annotation_column,
measure=s.queue_measure,
diversity=(s.queue_diversity or "none"),
limit=(s.queue_limit or None),
image_type=s.image_type, seed=0)
except (FileNotFoundError, ValueError) as exc:
return {"filtered_rows": None,
"total": count_rows(s.db_path, s.image_type, table=s.png_table),
"queue_summary": "",
"note": f"Uncertainty queue unavailable: {exc}"}
rows = al.queue_rows(queue)
return {"filtered_rows": rows, "total": len(rows),
"queue_summary": al.format_queue_summary(queue), "note": ""}
if filter_active:
rows = fetch_filtered_paths(
s.db_path,
s.annotation_column,
s.measurement if isinstance(s.measurement, list)
else [s.measurement],
s.threshold if isinstance(s.threshold, list) else [s.threshold],
s.threshold_direction if isinstance(s.threshold_direction, list)
else [s.threshold_direction],
s.image_type,
)
return {"filtered_rows": rows, "total": len(rows),
"queue_summary": "", "note": ""}
return {"filtered_rows": None,
"total": count_rows(s.db_path, s.image_type, table=s.png_table),
"queue_summary": "", "note": ""}
def _blind_order(rows, rank: Dict[str, int]) -> list:
"""``rows`` in a blinding key's shuffled order.
:param rows: ``(png_path, annotation)`` pairs.
:param rank: each path's position in the key's order. A path the key
does not know, one added to the table after blinding started, goes
after the known ones, ordered by a hash of its path so it is not
next to its own well either.
:returns: the rows, reordered.
"""
import hashlib
def position(row):
"""Order known crops by the blind key and unseen crops by path hash."""
path = str(row[0])
at = rank.get(path)
if at is not None:
return (0, at, "")
return (1, 0, hashlib.sha256(path.encode("utf-8")).hexdigest())
return sorted(rows, key=position)
def _blinded_total(outcome: dict, s: AnnotateSettings,
rank: Optional[Dict[str, int]]) -> dict:
"""A population count with its rows put in the blinding key's order.
Runs on the same worker as :func:`_compute_total`. The unfiltered
population has no row list of its own, so while blinded it is read
whole, because a page can then only be cut from the shuffled list.
:param outcome: what :func:`_compute_total` returned.
:param s: a frozen copy of the screen's settings.
:param rank: the key's order, or ``None`` when not blinded.
:returns: ``outcome``, reordered when blinded.
"""
if rank is None:
return outcome
rows = outcome.get("filtered_rows")
if rows is None:
rows = fetch_page(s.db_path, s.annotation_column, 0, -1,
s.image_type, table=s.png_table)
ordered = _blind_order(list(rows), rank)
return dict(outcome, filtered_rows=ordered, total=len(ordered), note="")
#: What separates one identifier from the next in a message: whitespace,
#: quotes, brackets and commas. Colons and dots are kept inside the token so
#: a Windows drive or a file suffix stays part of the path it belongs to.
_BLIND_TOKEN = re.compile(r"[^\s'\"()\[\]{}<>,;]+")
def _blind_lookup(codes: Dict[str, str], src: str) -> Dict[str, str]:
"""Every name a blinded crop or its source can be written as, and its stand-in.
:param codes: the key's ``{png_path: code}``.
:param src: the experiment folder; its own name is often the condition.
:returns: ``{identifier: replacement}``. A crop's full path, file name
and stem map to its code; a name two crops share, and the source
folder's name, map to the neutral word "Blind".
"""
lookup: Dict[str, str] = {}
for path, code in codes.items():
base = os.path.basename(str(path))
for identifier in (str(path), os.path.abspath(str(path)), base,
os.path.splitext(base)[0]):
if not identifier:
continue
previous = lookup.get(identifier, code)
lookup[identifier] = code if previous == code else tr("Blind")
name = os.path.basename(os.path.normpath(src)) if src else ""
if name:
lookup[name] = tr("Blind")
return lookup
def _blind_scrub(text: str, lookup: Dict[str, str],
folders: Sequence[str]) -> str:
"""``text`` with every crop, folder and source name replaced for blinding.
Tokens are looked up whole rather than searched for inside the text,
because a population is often hundreds of thousands of crops -- one
alternation over all of them would be slow to build and slower to run --
and a short stem searched as a substring would eat ordinary words.
Folders are few and hold a separator, so they are replaced as substrings.
:param text: any message about to be shown.
:param lookup: what :func:`_blind_lookup` built.
:param folders: folder paths to hide wherever they appear.
:returns: the scrubbed text.
"""
text = str(text or "")
if not text:
return text
def token(match):
"""One token's stand-in, keeping trailing punctuation outside it."""
word = match.group()
core = word.rstrip(".:!?")
tail = word[len(core):]
hit = lookup.get(core)
if hit is None and ("/" in core or "\\" in core):
leaf = core.replace("\\", "/").rstrip("/").rsplit("/", 1)[-1]
hit = lookup.get(leaf)
return word if hit is None else hit + tail
text = _BLIND_TOKEN.sub(token, text)
for folder in sorted({f for f in folders if f and len(f) > 1},
key=len, reverse=True):
text = text.replace(folder, tr("Blind"))
return text
class _BlindStatusLabel(QLabel):
"""The status line, which a blinded screen scrubs of crop and folder names.
Messages reach the status line from many places -- a failed save, a
suggestion run, a search -- and any of them may quote an exception that
names a file. Scrubbing where the text lands, rather than at each caller,
is what keeps a message added later from reopening the leak.
:ivar _blind_owner: a ``weakref`` to the screen, set once it is built.
"""
def setText(self, text: str) -> None:
"""Show ``text``, scrubbed while the owning screen is blinded."""
owner_ref = getattr(self, "_blind_owner", None)
owner = owner_ref() if owner_ref is not None else None
if owner is not None:
text = owner._blind_text(text)
super().setText(text)
def _read_example_settings(path) -> Dict[str, str]:
"""Read a settings CSV that shipped with a dataset, as ``key -> value``.
Shared by the settings form, which fills its widgets from it, and by the
screen's Load test data, which fills the live settings from it before
opening the plate.
:returns: an empty dict when the file is missing or unreadable.
"""
import csv
from pathlib import Path
path = Path(path)
if not path.is_file():
return {}
try:
with path.open(newline="") as handle:
return {str(row[0]).strip(): str(row[1]).strip()
for row in csv.reader(handle)
if len(row) >= 2 and row[0] != "Key"}
except OSError:
LOG.debug("could not read %s", path, exc_info=True)
return {}
def _plate_of_source(src: str) -> str:
"""The plate folder a typed source names.
The screen's ``src`` is the plate folder and the database is derived from
it as ``measurements/measurements.db``. The published example settings
name the DATABASE instead, and a source given that way used to be joined
as it stood, so the form looked for
``.../measurements.db/measurements/measurements.db`` and OK found nothing
to page. A path to that database is read as the plate that holds it,
whether it is given as ``measurements/measurements.db`` or, as the
shipped file spells it, as ``measurements.db`` beside the plate's folders.
"""
text = str(src or "").strip()
if not text:
return ""
path = os.path.normpath(text)
if os.path.basename(path) != "measurements.db":
return text
parent = os.path.dirname(path)
if os.path.basename(parent) == "measurements":
return os.path.dirname(parent)
return parent
class _SettingsDialog(QDialog):
"""Modal dialog that edits an :class:`AnnotateSettings` in place.
:param settings: the :class:`AnnotateSettings` this edits IN PLACE. The
dialog mutates the object it was handed rather than returning a new
one, so a caller that wants the old values back on Cancel has to
keep its own copy.
:param parent: parent widget; ownership only.
"""
def __init__(self, settings: AnnotateSettings, parent: Optional[QWidget] = None):
"""Build the form, detached from the window manager."""
super().__init__(parent)
from ..dialogs import RESIZABLE, detach_from_window_manager
detach_from_window_manager(self)
self.setProperty(RESIZABLE, True)
self.setWindowTitle("Annotate — Settings")
from ..preferences import scaled_px
self._settings = settings
form = QFormLayout()
self._src_edit = QLineEdit(settings.src)
if settings.src:
_ask_about_the_folder(settings.src)
self._src_edit.editingFinished.connect(self._probe_the_source_field)
src_row = QHBoxLayout()
src_row.setContentsMargins(0, 0, 0, 0)
src_row.addWidget(self._src_edit, 1)
src_btn = QPushButton("Browse…")
src_btn.clicked.connect(self._pick_src)
src_row.addWidget(src_btn)
src_wrap = QWidget(); src_wrap.setLayout(src_row)
form.addRow("Source folder", src_wrap)
self._ann_col = QLineEdit(settings.annotation_column)
form.addRow("Annotation column", self._ann_col)
attach_column_picker(self._ann_col, self._picker_db_path, "png_list",
layout=form)
self._img_size = QSpinBox()
self._img_size.setRange(48, 800)
self._img_size.setValue(settings.image_size[0])
form.addRow("Crop size (px)", self._img_size)
from ...crops import (LOAD_IMAGES, LOAD_IMAGES_LABEL, STREAM_IMAGES,
STREAM_IMAGES_LABEL)
self._crop_source = QComboBox()
self._crop_source.addItem(
f"{LOAD_IMAGES_LABEL} \u2014 crops already in data/", LOAD_IMAGES)
self._crop_source.addItem(
f"{STREAM_IMAGES_LABEL} \u2014 cut from merged/*.npy",
STREAM_IMAGES)
self._crop_source.setToolTip(
"LOAD IMAGES reads the crops the measure step exported under "
"data/. STREAM IMAGES cuts them out of merged/*.npy as the page "
"is drawn, which needs no export. Either mode falls back to the "
"other when its folder is missing, and says which route drew.")
stored = str(getattr(settings, "crop_source", "") or "").strip().lower()
self._crop_source.setCurrentIndex(
1 if stored in ("merged", "stream", "stream_images", "on_demand")
else 0)
form.addRow("Image source", self._crop_source)
self._image_type = QLineEdit(settings.image_type or "")
self._image_type.setPlaceholderText("e.g. cell (blank = all types)")
form.addRow("Image type filter", self._image_type)
self._channels = QLineEdit(_list_to_csv(settings.channels))
self._channels.setPlaceholderText("r, g, b (blank = all)")
form.addRow("Show channels", self._channels)
self._stored_channel_order = QComboBox()
self._stored_channel_order.addItem("RGB (standard)", "rgb")
self._stored_channel_order.addItem(
"Auto (use spaCR format marker)", "auto")
self._stored_channel_order.addItem(
"Legacy BGR (old unmarked crops)", "legacy_bgr")
current_order = str(
getattr(settings, "stored_channel_order", "rgb")).lower()
order_index = self._stored_channel_order.findData(current_order)
self._stored_channel_order.setCurrentIndex(max(0, order_index))
self._stored_channel_order.setToolTip(
"Order stored in the PNG file. RGB keeps standard PNG channels "
"unchanged. Auto uses spaCR's sidecar/database format marker. "
"Legacy BGR repairs crops written by older cv2-based releases. "
"After decoding, Annotate always uses RGB arrays.")
form.addRow("Stored PNG order", self._stored_channel_order)
from ...crops import DISPLAY_ORDERS
self._display_order = QComboBox()
for order in DISPLAY_ORDERS:
label = " ".join(order.upper())
self._display_order.addItem(
f"{label}" + (" (unchanged)" if order == "rgb" else ""),
order)
current_display = str(
getattr(settings, "display_order", "rgb")).lower()
display_index = self._display_order.findData(current_display)
self._display_order.setCurrentIndex(max(0, display_index))
self._display_order.setToolTip(
"Which source channel is drawn in each colour slot. This is a "
"VIEW setting and changes nothing on disk and nothing measured "
"\u2014 it does not say how the file was written, which is the "
"row above. Use it when a project authored before the crop-format "
"fix should be seen in the colours it was authored for: B G R "
"restores that picture without marking the folder as a format it "
"is not.")
form.addRow("Display order", self._display_order)
from ...crops import DISPLAY_PRIMARIES
self._display_primaries = QComboBox()
_PRIMARY_LABELS = {
"rgb": "RGB (as acquired)",
"cmy": "CMY (publication style)",
"deuteranope": "Colourblind \u2014 deuteranope (red-green)",
"protanope": "Colourblind \u2014 protanope (red-green)",
"tritanope": "Colourblind \u2014 tritanope (blue-yellow)",
}
for mode in DISPLAY_PRIMARIES:
self._display_primaries.addItem(_PRIMARY_LABELS[mode], mode)
current_primaries = str(
getattr(settings, "display_primaries", "") or "").lower()
if current_primaries in ("", "rgb"):
try:
from ..preferences import image_display_primaries
current_primaries = image_display_primaries()
except Exception:
current_primaries = "rgb"
primaries_index = self._display_primaries.findData(current_primaries)
self._display_primaries.setCurrentIndex(max(0, primaries_index))
self._display_primaries.setToolTip(
"What colours the channels are drawn in. A VIEW setting: it "
"changes nothing on disk and nothing measured.\n\n"
"CMY is the publication style most multichannel micrographs use "
"now. The three colourblind modes are separate because which "
"PAIR of colours collapses depends on the deficiency \u2014 "
"red-green for a deuteranope or protanope, blue-yellow for a "
"tritanope \u2014 so one setting cannot serve all three. CMY is "
"NOT one of them: measured against a red-green deficiency it "
"separates the channels less well than plain RGB.")
form.addRow("Channel colours", self._display_primaries)
self._norm_channels = QLineEdit(_list_to_csv(settings.normalize_channels))
self._norm_channels.setPlaceholderText("r, g, b (blank = off)")
form.addRow("Normalize channels", self._norm_channels)
self._pct_lo = QDoubleSpinBox()
self._pct_lo.setDecimals(PERCENTILE_DECIMALS)
self._pct_lo.setRange(0.0, 100.0)
self._pct_lo.setSingleStep(0.01)
self._pct_lo.setValue(float(settings.percentiles[0]))
self._pct_hi = QDoubleSpinBox()
self._pct_hi.setDecimals(PERCENTILE_DECIMALS)
self._pct_hi.setRange(0.0, 100.0)
self._pct_hi.setSingleStep(0.01)
self._pct_hi.setValue(float(settings.percentiles[1]))
pct_row = QHBoxLayout(); pct_row.setContentsMargins(0, 0, 0, 0)
pct_row.addWidget(self._pct_lo); pct_row.addWidget(QLabel("–"))
pct_row.addWidget(self._pct_hi)
pct_wrap = QWidget(); pct_wrap.setLayout(pct_row)
form.addRow("Percentiles", pct_wrap)
self._outline = QLineEdit(_list_to_csv(settings.outline))
self._outline.setPlaceholderText("channels to outline, e.g. g")
form.addRow("Outline channels", self._outline)
self._outline_method = QComboBox()
self._outline_method.addItems(["otsu", "cellpose"])
self._outline_method.setCurrentText(
getattr(settings, "outline_method", "otsu"))
self._outline_method.setToolTip(
"How object outlines are found: 'otsu' (fast threshold) or "
"'cellpose' (a small Cellpose model — cleaner, slower).")
form.addRow("Outline method", self._outline_method)
self._out_factor = QDoubleSpinBox()
self._out_factor.setRange(0.0, 100.0)
self._out_factor.setValue(float(settings.outline_threshold_factor))
form.addRow("Outline threshold factor", self._out_factor)
self._out_sigma = QDoubleSpinBox()
self._out_sigma.setRange(0.0, 100.0)
self._out_sigma.setValue(float(settings.outline_sigma))
form.addRow("Outline sigma", self._out_sigma)
self._edge_thick = QDoubleSpinBox()
self._edge_thick.setRange(0.0, 20.0)
self._edge_thick.setDecimals(2)
self._edge_thick.setValue(float(settings.edge_thickness))
form.addRow("Edge thickness", self._edge_thick)
self._edge_transp = QDoubleSpinBox()
self._edge_transp.setRange(0.0, 100.0)
self._edge_transp.setValue(float(settings.edge_transparency))
form.addRow("Edge transparency", self._edge_transp)
self._edge_image = Toggle("Show original image under outline")
self._edge_image.setChecked(bool(settings.edge_image))
form.addRow("", self._edge_image)
self._object_filter_fields: Dict[
Tuple[str, str], Tuple[QLineEdit, QLineEdit]] = {}
current_filters = normalize_object_filters(
getattr(settings, "object_filters", None),
getattr(settings, "object_size", None))
for channel in FILTER_CHANNELS:
for measure in FILTER_MEASURES:
low, high = current_filters[filter_key(channel, measure)]
filter_row = QHBoxLayout()
filter_row.setContentsMargins(0, 0, 0, 0)
pair: List[QLineEdit] = []
for value, hint in ((low, "min"), (high, "max")):
edit = QLineEdit(_filter_text(value))
edit.setPlaceholderText(hint)
validator = QDoubleValidator(edit)
validator.setNotation(QDoubleValidator.StandardNotation)
validator.setBottom(0.0)
validator.setLocale(QLocale.c())
edit.setValidator(validator)
edit.setToolTip(FILTER_FIELD_TIP)
filter_row.addWidget(edit)
pair.append(edit)
filter_wrap = QWidget(); filter_wrap.setLayout(filter_row)
form.addRow(FILTER_ROW_LABELS[(channel, measure)],
filter_wrap)
self._object_filter_fields[(channel, measure)] = (
pair[0], pair[1])
self._measurement = QLineEdit(
", ".join(settings.measurement) if isinstance(settings.measurement, (list, tuple))
else (str(settings.measurement) if settings.measurement else "")
)
self._measurement.setPlaceholderText("e.g. cell_area (blank = off)")
form.addRow("Measurement column(s)", self._measurement)
attach_column_picker(self._measurement, self._picker_db_path,
layout=form, multi=True)
self._threshold = QLineEdit(
", ".join(str(x) for x in settings.threshold) if isinstance(settings.threshold, (list, tuple))
else (str(settings.threshold) if settings.threshold is not None else "")
)
self._threshold.setPlaceholderText("e.g. 500 (comma-separated to match)")
form.addRow("Threshold(s)", self._threshold)
self._threshold_dir = QComboBox()
for d in ("higher", "lower"):
self._threshold_dir.addItem(d)
idx = 0
if settings.threshold_direction == "lower":
idx = 1
elif isinstance(settings.threshold_direction, (list, tuple)) \
and settings.threshold_direction \
and str(settings.threshold_direction[0]).lower() == "lower":
idx = 1
self._threshold_dir.setCurrentIndex(idx)
form.addRow("Direction", self._threshold_dir)
self._queue_on = Toggle("Order by model uncertainty")
self._queue_on.setChecked(bool(getattr(settings, "queue_by_uncertainty", False)))
self._queue_on.setToolTip(
"Show the unlabelled crops the classifier is least sure about "
"first. Needs model scores in png_list, so run Classify (CV) "
"before turning this on.")
form.addRow("Queue", self._queue_on)
self._queue_measure = QComboBox()
for m in ("entropy", "least_confidence", "margin"):
self._queue_measure.addItem(m)
self._queue_measure.setCurrentText(
str(getattr(settings, "queue_measure", "entropy")))
self._queue_measure.setToolTip(
"How uncertainty is scored. With two classes, margin and "
"least_confidence give the identical ranking; they only diverge "
"at three classes or more.")
form.addRow("Uncertainty measure", self._queue_measure)
self._queue_diversity = QComboBox()
for d in ("well", "field", "plate", "none"):
self._queue_diversity.addItem(d)
self._queue_diversity.setCurrentText(
str(getattr(settings, "queue_diversity", "well")))
self._queue_diversity.setToolTip(
"Spread the queue across wells rather than serving the most "
"uncertain crops in ranked order. Pure uncertainty collapses onto "
"one or two wells, so you end up labelling the same ambiguity a "
"hundred times. 'none' turns that protection off.")
form.addRow("Queue diversity", self._queue_diversity)
self._queue_limit = QSpinBox()
self._queue_limit.setRange(0, 1_000_000)
self._queue_limit.setValue(int(getattr(settings, "queue_limit", 0) or 0))
self._queue_limit.setSpecialValueText("all unlabelled")
form.addRow("Queue length", self._queue_limit)
form_widget = QWidget()
form_widget.setLayout(form)
self._form_scroll = QScrollArea()
self._form_scroll.setFrameShape(QScrollArea.NoFrame)
self._form_scroll.setWidgetResizable(True)
self._form_scroll.setWidget(form_widget)
self.setLayout(QVBoxLayout())
self.layout().addWidget(self._form_scroll, 1)
buttons = QDialogButtonBox(QDialogButtonBox.Ok | QDialogButtonBox.Cancel)
buttons.accepted.connect(self.accept)
buttons.rejected.connect(self.reject)
self.layout().addWidget(buttons)
from ..dialogs import give_it_a_size_grip
from ..hidpi import screen_for_widget
give_it_a_size_grip(self)
available = screen_for_widget(self).availableGeometry()
self.setMinimumWidth(min(scaled_px(480), available.width()))
self.resize(min(scaled_px(640), available.width()),
min(scaled_px(720), int(available.height() * 0.9)))
from .settings_model import install_api_tooltips
install_api_tooltips(self, "annotate", {
self._src_edit: "src",
self._ann_col: "annotation_column",
self._img_size: "crop_size",
self._image_type: "image_type",
self._channels: "channels",
self._stored_channel_order: "stored_channel_order",
self._norm_channels: "normalize_channels",
self._pct_lo: "lower_percentile",
self._pct_hi: "upper_percentile",
self._outline: "outline",
self._outline_method: "outline_method",
self._out_factor: "outline_threshold_factor",
self._out_sigma: "outline_sigma",
self._edge_thick: "edge_thickness",
self._edge_transp: "edge_transparency",
self._edge_image: "edge_image",
self._measurement: "measurement",
self._threshold: "threshold",
self._threshold_dir: "threshold_direction",
self._queue_on: "queue_by_uncertainty",
self._queue_measure: "queue_measure",
self._queue_diversity: "queue_diversity",
self._queue_limit: "queue_limit",
})
def _probe_the_source_field(self) -> None:
"""Queue a background check of whatever the source field now holds.
Never blocks: :func:`_ask_about_the_folder` hands both questions to
other threads and returns. The answers are deliberately discarded
here -- they are read by :meth:`_pick_src`, later, through
:func:`_vouched_dir`, and this is only what puts them in the caches
in time to be read.
``editingFinished``, so one check per editing session rather than one
per keystroke: a check is a stat, and a sleeping mount parks the
thread that makes it.
"""
text = self._src_edit.text().strip()
if text:
_ask_about_the_folder(text)
def _picker_db_path(self) -> str:
"""Where the SQL picker looks — the src folder as it reads right now.
A callable rather than a captured string: users routinely set the
source folder and the annotation column in the same visit to this
dialog, and a value captured at construction would point at the
previous folder.
"""
return self._src_edit.text().strip()
def example_destination(self):
"""The shared example plate folder.
Shared with Classify because it is one dataset, and with Mask and
Measure because `data/`, `merged/` and `measurements/measurements.db`
are parts of one plate that are needed together.
"""
from ..hf_download import example_plate_folder
return example_plate_folder()
def _load_the_example_data(self, *, ask=None) -> str:
"""Fetch the example set and point this screen at it.
:returns: the source that was set, or ``""``.
"""
destination = self.example_destination()
destination.mkdir(parents=True, exist_ok=True)
if (destination / "measurements" / "measurements.db").is_file():
return self._use_the_example_data(destination)
button = getattr(self, "_example_btn", None)
if button is not None:
button.setEnabled(False)
button.setText("Fetching…")
def _done(result, error):
"""Restore the button whether the load worked or failed."""
if button is not None:
button.setEnabled(True)
button.setText("Example…")
if result is None:
LOG.info("annotation example not downloaded: %s", error)
return
self._use_the_example_data(destination)
download = ask
if download is None:
from ..hf_download import download_annotate_example as download
download(self, destination, _done)
return ""
def _load_the_streaming_example(self, *, ask=None) -> str:
"""Fetch the merged arrays, for the streaming strategy.
The same plate folder as the crops, so the two compose: a user who
presses both ends up with a plate that either strategy can work from.
"""
destination = self.example_destination()
destination.mkdir(parents=True, exist_ok=True)
merged = destination / "merged"
if merged.is_dir() and any(merged.glob("*.npy")):
return self._use_the_example_data(destination)
button = getattr(self, "_stream_btn", None)
if button is not None:
button.setEnabled(False)
button.setText("Fetching…")
def _done(result, error):
"""Restore the button whether the load worked or failed."""
if button is not None:
button.setEnabled(True)
button.setText("Example (streaming)")
if result is None:
LOG.info("streaming example not downloaded: %s", error)
return
self._use_the_example_data(destination)
download = ask
if download is None:
from ..hf_download import download_measure_example as download
download(self, destination, _done)
return ""
def _use_the_example_data(self, destination) -> str:
"""Point the source field at the downloaded example.
The DATABASE, not the folder: this screen reads `png_list`, and the
published settings file names the database for the same reason.
"""
from pathlib import Path
destination = Path(destination)
database = destination / "measurements" / "measurements.db"
source = str(database if database.is_file() else destination)
self._src_edit.setText(source)
self._apply_example_settings(destination / "settings"
/ "annotate_settings.csv")
if hasattr(self, "_ann_col") and not self._ann_col.text().strip():
self._ann_col.setText("infected")
return source
#: Settings keys the dialog can fill in, and the widget each one drives.
#:
#: Named rather than derived: the dialog holds far more widgets than the
#: published file sets, and a loop over every attribute would quietly pick
#: up whichever ones happened to share a name.
_EXAMPLE_SETTING_WIDGETS = {
"annotation_column": "_ann_col",
"crop_size": "_img_size",
"channels": "_channels",
"image_type": "_image_type",
"measurement": "_measurement",
"threshold": "_threshold",
"normalize_channels": "_norm_channels",
"outline": "_outline",
}
#: Other spellings the same question has been written under.
#:
#: `crop_size` IS THE CURRENT NAME of what the factory used to call
#: `img_size`. Every settings CSV written before the rename says
#: `img_size`, including the example dataset's,
#: so the old spelling is still read here when the new one is absent.
#:
#: `image_size` IS NOT ACCEPTED, and that is the point of the rename.
#: It is the MODEL's input crop -- default 224, read by training and
#: inference -- while this field is how large each cell is drawn,
#: default 200. This screen used to read `image_size` as its own
#: spelling, and the example dataset's `annotate_settings.csv`, as
#: downloaded, carries both -- `img_size,200` and `image_size,224` -- so
#: the example drew its cells at the model's resolution. `AnnotateSettings.image_size` keeps its name: it is the
#: dataclass field this value lands in, not a settings key.
_ALSO_SPELT = {
"crop_size": ("img_size",),
}
def _apply_example_settings(self, path) -> int:
"""Fill the form from a settings CSV that shipped with a dataset.
Each field independently: one unusable value must not cost the rest of
the form, which is the whole reason a settings file is worth shipping.
:returns: how many fields were set.
"""
rows = _read_example_settings(path)
if not rows:
return 0
applied = 0
for key, attribute in self._EXAMPLE_SETTING_WIDGETS.items():
value = rows.get(key)
for spelling in self._ALSO_SPELT.get(key, ()):
if value in (None, "", "None"):
value = rows.get(spelling)
widget = getattr(self, attribute, None)
if widget is None or value in (None, "", "None"):
continue
try:
if hasattr(widget, "setValue"):
widget.setValue(int(float(value)))
elif hasattr(widget, "setCurrentText"):
widget.setCurrentText(value)
else:
widget.setText(value)
applied += 1
except Exception: # noqa: BLE001
LOG.debug("example settings: %r is not usable for %s",
value, key, exc_info=True)
LOG.info("applied %d settings from %s", applied, path)
return applied
def _pick_src(self):
"""Ask for the experiment source, starting where the field points.
...but only when the field points somewhere a real stat has come back
and confirmed. This is the same defect as the subtitle stat on the
screen that owns this dialog: `QFileDialog.getExistingDirectory` stats
and then lists its starting directory ON THE GUI THREAD, and the field
routinely holds the folder of the last source opened -- so a user
whose plate lives on a sleeping `autofs` mount pressed Browse and got
the twenty-second freeze that `spacr.qt.path_probe` exists to prevent.
:func:`_vouched_dir` rather than the probe's own answer, and the
difference matters: the probe reports an unanswered stat as PRESENT
after its timeout, so gating on it alone would reopen this exact hole
five seconds late. The check is queued when the dialog opens and again
when the field is edited (see :meth:`_probe_the_source_field`), so a
folder that is really there has almost always answered by the time
this runs. One that has not opens the picker at the working directory,
which is local by construction; nothing is lost but the head start,
and the user can still navigate from there.
"""
start = self._src_edit.text().strip()
if not _vouched_dir(start):
start = os.getcwd()
d = QFileDialog.getExistingDirectory(self, "Pick experiment source",
start)
if d:
self._src_edit.setText(d)
_ask_about_the_folder(d)
def accept(self): # noqa: D401 - Qt slot
"""Refuse OK when the threshold filter cannot be applied as typed.
"Measurement column(s)" and "Threshold(s)" are two independent
free-text line edits, so three columns and two thresholds is a typo
away -- and the engine used to `zip` them, silently dropping the
third filter and hand-labelling a population the user never asked
for.
`fetch_filtered_paths` now raises on the mismatch, but that runs on a
worker thread whose failure signal has no receiver, so the raise
alone would be invisible. This is the half the user can act on: they
are standing in the dialog with both fields in front of them.
One threshold for several columns is a documented shorthand and stays
allowed; it is only the in-between case that is refused.
"""
from PySide6.QtWidgets import QMessageBox
measurements = _csv_to_list(self._measurement.text().strip())
raw = [p.strip() for p in self._threshold.text().strip().split(",")
if p.strip()]
if measurements and raw and len(raw) not in (1, len(measurements)):
QMessageBox.warning(
self, "One threshold per measurement",
f"You have given {len(measurements)} measurement column(s) "
f"and {len(raw)} threshold(s).\n\nGive one threshold per "
f"column, or a single threshold to apply to all of them. "
f"Anything in between has no defined pairing, and guessing "
f"one would filter on a population you did not ask for.")
self._threshold.setFocus()
return
super().accept()
def collect(self) -> AnnotateSettings:
"""Read every editor and return the updated settings object."""
s = self._settings
s.src = _plate_of_source(self._src_edit.text())
s.db_path = os.path.join(s.src, "measurements", "measurements.db")
s.annotation_column = self._ann_col.text().strip() or "annotate"
size = int(self._img_size.value())
s.image_size = (size, size)
s.crop_source = str(self._crop_source.currentData() or "png")
s.image_type = self._image_type.text().strip() or None
s.channels = _csv_to_list(self._channels.text())
s.stored_channel_order = str(
self._stored_channel_order.currentData() or "rgb")
s.display_order = str(self._display_order.currentData() or "rgb")
s.display_primaries = str(
self._display_primaries.currentData() or "rgb")
s.normalize_channels = _csv_to_list(self._norm_channels.text())
s.percentiles = (float(self._pct_lo.value()), float(self._pct_hi.value()))
s.outline = _csv_to_list(self._outline.text())
s.outline_method = self._outline_method.currentText()
s.outline_threshold_factor = float(self._out_factor.value())
s.outline_sigma = float(self._out_sigma.value())
s.edge_thickness = float(self._edge_thick.value())
s.edge_transparency = float(self._edge_transp.value())
s.edge_image = bool(self._edge_image.isChecked())
s.object_filters = {
filter_key(channel, measure): (filter_bound(low.text()),
filter_bound(high.text()))
for (channel, measure), (low, high)
in self._object_filter_fields.items()}
s.object_size = (0, 0)
meas_txt = self._measurement.text().strip()
s.measurement = _csv_to_list(meas_txt)
thr_txt = self._threshold.text().strip()
if thr_txt:
parts = [p.strip() for p in thr_txt.split(",") if p.strip()]
parsed: List[float] = []
for p in parts:
try:
parsed.append(float(p))
except ValueError:
pass
s.threshold = parsed or None
else:
s.threshold = None
s.threshold_direction = self._threshold_dir.currentText() \
if (s.measurement and s.threshold) else None
s.queue_by_uncertainty = bool(self._queue_on.isChecked())
s.queue_measure = self._queue_measure.currentText()
s.queue_diversity = self._queue_diversity.currentText()
s.queue_limit = int(self._queue_limit.value())
return s
class _GenerateAnnotationDatabaseDialog(QDialog):
"""Build a set of crops to annotate, without measuring the plate again.
Everything the run needs is already on disk twice over: the object masks
inside the merged arrays, and the coordinate columns in measurements.db.
This chooses between them, filters, and registers the result as a new
``png_list`` table. See :mod:`spacr.annotation_dataset`.
THE TWO SOURCES ARE NOT INTERCHANGEABLE and the difference is not
guessable, so it is written on the form rather than left to the manual:
the database stores coordinates and nothing else, so it can only ever cut
a bounding box.
:param settings: the annotation settings the form is filled from and
written back to.
:param parent: parent widget; ownership only.
"""
def __init__(self, settings, parent=None):
"""Build the form for generating an annotation database."""
super().__init__(parent)
self.setWindowTitle("Generate annotation database")
self._settings = settings
self._written = ""
outer = QVBoxLayout(self)
form = QFormLayout()
self._source = QComboBox(self)
from spacr.annotation_dataset import STREAM_SOURCES
for value, label in STREAM_SOURCES:
self._source.addItem(label, value)
self._source.currentIndexChanged.connect(self._refresh_gates)
form.addRow("Read the objects from", self._source)
self._object = QComboBox(self)
self._object.addItems(["cell", "nucleus", "pathogen", "cytoplasm"])
form.addRow("Object", self._object)
self._channels = QLineEdit("0,1,2", self)
self._channels.setToolTip(
"Image channels to put in the crop, in order. Three make an RGB "
"picture; one makes greyscale.")
form.addRow("Channels", self._channels)
self._bounding_box = QCheckBox("Crop to the bounding box", self)
self._bounding_box.setChecked(True)
self._bounding_box.setToolTip(
"On, the crop is the rectangle around the object and keeps "
"whatever else falls inside it. Off, everything outside the mask "
"is removed -- which only the array source can do.")
form.addRow("", self._bounding_box)
self._min_size = QSpinBox(self)
self._min_size.setRange(0, 100_000_000)
self._min_size.setToolTip("0 means no lower bound.")
form.addRow("Minimum object area", self._min_size)
self._max_size = QSpinBox(self)
self._max_size.setRange(0, 100_000_000)
self._max_size.setToolTip("0 means no upper bound.")
form.addRow("Maximum object area", self._max_size)
self._max_objects = QSpinBox(self)
self._max_objects.setRange(0, 10_000_000)
self._max_objects.setToolTip(
"0 means every object. A cap is applied last and in the table's "
"existing order, so the same settings always give the same set.")
form.addRow("Most objects to take", self._max_objects)
outer.addLayout(form)
self._note = QLabel("", self)
self._note.setWordWrap(True)
self._note.setObjectName("SubtitleSmall")
outer.addWidget(self._note)
self._status = QLabel("", self)
self._status.setWordWrap(True)
outer.addWidget(self._status)
buttons = QDialogButtonBox(QDialogButtonBox.Close)
self._generate = QPushButton("Generate")
self._generate.setDefault(True)
self._generate.clicked.connect(self._on_generate)
buttons.addButton(self._generate, QDialogButtonBox.ActionRole)
buttons.rejected.connect(self.reject)
outer.addWidget(buttons)
self._refresh_gates()
def written_table(self) -> str:
"""The table that was written, or ``""``."""
return self._written
def _refresh_gates(self, *_args) -> None:
"""Say what the chosen source can and cannot do."""
source = self._source.currentData()
if source == "database":
self._bounding_box.setChecked(True)
self._bounding_box.setEnabled(False)
self._note.setText(
"The database stores coordinates and no masks, so this source "
"can only cut a bounding box. Choose the arrays to cut to the "
"object itself.")
else:
self._bounding_box.setEnabled(True)
self._note.setText(
"The arrays carry the object masks, so this source can cut to "
"the object or to its bounding box.")
def _collected(self) -> dict:
"""The settings the generator will be given."""
object_type = self._object.currentText()
channels = [int(part) for part in
str(self._channels.text()).replace(" ", "").split(",")
if part]
return {
"src": str(getattr(self._settings, "src", "") or ""),
"database": str(getattr(self._settings, "db_path", "") or ""),
"stream_source": self._source.currentData(),
"object_array": object_type,
"channel_arrays": channels or [0, 1, 2],
"bounding_box": self._bounding_box.isChecked(),
f"{object_type}_min_size": int(self._min_size.value()),
f"{object_type}_max_size": int(self._max_size.value()),
"max_objects": int(self._max_objects.value()),
}
def _on_generate(self) -> None:
"""Run it, and say what happened either way."""
from spacr.annotation_dataset import generate_annotation_dataset
self._generate.setEnabled(False)
self._status.setText("Working…")
try:
report = generate_annotation_dataset(self._collected())
except Exception as error: # noqa: BLE001
LOG.warning("could not generate the annotation set", exc_info=True)
self._status.setText(f"Failed: {error}")
self._generate.setEnabled(True)
return
self._generate.setEnabled(True)
self._written = str(report.get("table") or "")
trouble = "; ".join(str(t) for t in report.get("trouble") or [])
if not report.get("written"):
self._status.setText(
f"Nothing was written. {trouble}" if trouble
else "Nothing was written.")
return
message = (f"Wrote {report['written']} crops from "
f"{report['fields']} field(s) into {self._written}.")
if trouble:
message += f"\n\nAlso: {trouble}"
self._status.setText(message)
self.accept()
class _AutoAnnotateDialog(QDialog):
"""Label a whole population at once, from metadata or measurements.
Four sources were asked for. Two are here, and two are hand-offs, which
is a deliberate split rather than an unfinished one:
* **metadata** and **measurement** are implemented, because nothing else
in spaCR turns "column 2" or "cell_area > 500" into an annotation.
* the **Gate Editor** and the **Image UMAP** already select populations
and already write annotations. Reimplementing either would put a
second copy of the gate maths, or of the clustering, on a divergent
path from the one the user sees on screen. What was missing was the
route, so those buttons open the real thing.
**Nothing is written until the count has been shown.** Annotating
thousands of rows is not undoable through the grid -- the undo stack only
holds the slots on this page -- so the preview is the safety, and the
Apply button stays disabled until one has been taken.
:param settings: the :class:`AnnotateSettings` the preview and the apply
both run against.
:param parent: parent widget; ownership only.
"""
def __init__(self, settings: AnnotateSettings,
parent: Optional[QWidget] = None):
"""Build the form, with Apply disabled until a preview is taken."""
super().__init__(parent)
from ..dialogs import detach_from_window_manager
detach_from_window_manager(self)
self.setWindowTitle("Annotate — auto-annotate")
from ..preferences import scaled_px
self.setMinimumWidth(scaled_px(520))
self._settings = settings
self._matched: List[str] = []
outer = QVBoxLayout(self)
form = QFormLayout()
self._source = QComboBox()
self._source.addItem("Metadata (plate, row, column, field)", "metadata")
self._source.addItem("Measurement thresholds", "measurement")
self._source.currentIndexChanged.connect(self._on_source_changed)
form.addRow("Select by", self._source)
self._column = QComboBox()
self._column.currentTextChanged.connect(self._on_metadata_column)
form.addRow("Metadata column", self._column)
self._values = QLineEdit()
self._values.setPlaceholderText("e.g. c1,c2 — blank means every value")
form.addRow("Values (comma separated)", self._values)
self._rules = QPlainTextEdit()
self._rules.setPlaceholderText(
"One rule per line: cell_area > 500\n"
"nucleus_area < 200\n\n"
"Rules are ANDed. Use more than one — a single threshold is a "
"gate, not a population.")
self._rules.setFixedHeight(90)
form.addRow("Measurement rules", self._rules)
self._value = QSpinBox()
self._value.setRange(0, 999)
self._value.setValue(1)
form.addRow("Annotate as class", self._value)
outer.addLayout(form)
self._preview_label = QLabel("Preview to see how many objects match.")
self._preview_label.setObjectName("SubtitleSmall")
self._preview_label.setWordWrap(True)
outer.addWidget(self._preview_label)
row = QHBoxLayout()
self._btn_preview = QPushButton("Preview")
self._btn_preview.clicked.connect(self._on_preview)
row.addWidget(self._btn_preview)
self._btn_gate = QPushButton("Gate Editor…")
self._btn_gate.setToolTip(
"Draw gates on a scatter of your measurements, then annotate "
"what falls inside. Opens the real Gate Editor.")
self._btn_gate.clicked.connect(self._on_open_gate_editor)
row.addWidget(self._btn_gate)
self._btn_umap = QPushButton("Image UMAP…")
self._btn_umap.setToolTip(
"Cluster the objects in a UMAP and write the cluster labels as "
"annotations. Opens the Image UMAP module, which does the write "
"itself.")
self._btn_umap.clicked.connect(self._on_open_umap)
row.addWidget(self._btn_umap)
row.addStretch(1)
outer.addLayout(row)
self._buttons = QDialogButtonBox(
QDialogButtonBox.Apply | QDialogButtonBox.Close)
self._apply = self._buttons.button(QDialogButtonBox.Apply)
self._apply.setEnabled(False)
self._apply.clicked.connect(self.accept)
self._buttons.rejected.connect(self.reject)
outer.addWidget(self._buttons)
self._load_metadata_columns()
self._on_source_changed()
def source(self) -> str:
"""The annotation source the user chose."""
return str(self._source.currentData())
def value(self) -> int:
"""The value to write for the chosen source."""
return int(self._value.value())
def matched_paths(self) -> List[str]:
"""The population the last preview found. Empty until one is taken."""
return list(self._matched)
def parsed_rules(self) -> List[Dict[str, Any]]:
"""Parse the rule box into ``{column, threshold, direction}`` dicts.
:raises ValueError: a line that is not ``<column> <op> <number>``.
Refused rather than skipped: a typo that silently dropped a rule
would widen the population and label objects the user never
asked for.
"""
rules: List[Dict[str, Any]] = []
for raw in self._rules.toPlainText().splitlines():
line = raw.strip()
if not line:
continue
parts = line.replace(">=", ">").replace("<=", "<").split()
if len(parts) != 3 or parts[1] not in (">", "<"):
raise ValueError(
f"cannot read rule {line!r}; expected "
f"'<measurement> > <number>' or '< <number>'")
try:
threshold = float(parts[2])
except ValueError:
raise ValueError(
f"{parts[2]!r} in rule {line!r} is not a number")
rules.append({
"column": parts[0],
"threshold": threshold,
"direction": "higher" if parts[1] == ">" else "lower",
})
return rules
def _load_metadata_columns(self) -> None:
"""Offer the metadata columns the engine knows about.
Taken from the engine rather than listed here, so a column added there
appears without a second edit.
"""
from ..annotate_engine import METADATA_COLUMNS
self._column.addItems(list(METADATA_COLUMNS))
def _on_source_changed(self, *_args) -> None:
"""Enable the column and value fields only for a metadata source."""
metadata = self.source() == "metadata"
for widget in (self._column, self._values):
widget.setEnabled(metadata)
self._rules.setEnabled(not metadata)
self._invalidate_preview()
def _on_metadata_column(self, *_args) -> None:
"""Show the values this column actually holds.
Read from the database rather than guessed: a picker offering rows
A-H to someone whose plate is numbered is a picker they cannot use.
"""
self._invalidate_preview()
if self.source() != "metadata" or not self._settings.db_path:
return
try:
from ..annotate_engine import metadata_values
values = metadata_values(self._settings.db_path,
self._column.currentText())
except Exception:
return
if values:
shown = ", ".join(values[:12])
more = "" if len(values) <= 12 else f" (+{len(values) - 12} more)"
self._values.setPlaceholderText(f"{shown}{more}")
def _invalidate_preview(self, *_args) -> None:
"""Throw the preview away and disable Apply.
Called whenever anything the preview was computed FROM changes. Applying
against a stale preview would annotate rows the user never saw, and
annotating thousands of rows is not undoable through the grid.
"""
self._matched = []
self._apply.setEnabled(False)
def _on_preview(self) -> None:
"""Match rows and show how many, without changing anything.
The preview is the SAFETY: applying annotates thousands of rows and the
undo stack only holds the slots on this page, so Apply stays disabled
until this has run.
"""
from ..annotate_engine import paths_by_measurements, paths_by_metadata
if not self._settings.db_path:
self._preview_label.setText("Open a source first.")
return
try:
if self.source() == "metadata":
raw = self._values.text().strip()
values = [v.strip() for v in raw.split(",") if v.strip()]
if not values:
from ..annotate_engine import metadata_values
values = metadata_values(self._settings.db_path,
self._column.currentText())
paths = paths_by_metadata(
self._settings.db_path, self._column.currentText(), values)
else:
rules = self.parsed_rules()
if not rules:
self._preview_label.setText("Add at least one rule.")
self._invalidate_preview()
return
paths = paths_by_measurements(
self._settings.db_path,
self._settings.annotation_column, rules)
except Exception as exc:
self._preview_label.setText(f"Could not preview: {exc}")
self._invalidate_preview()
return
self._matched = list(dict.fromkeys(paths))
self._apply.setEnabled(bool(self._matched))
self._preview_label.setText(
f"{len(self._matched):,} object(s) match. Apply writes class "
f"{self.value()} to \"{self._settings.annotation_column}\"."
if self._matched else "Nothing matches.")
def _on_open_gate_editor(self) -> None:
"""Close with the code that asks the caller to open the Gate Editor."""
self.done(_AUTO_ANNOTATE_OPEN_GATE)
def _on_open_umap(self) -> None:
"""Close with the code that asks the caller to open the UMAP."""
self.done(_AUTO_ANNOTATE_OPEN_UMAP)
#: Dialog result codes for the two hand-offs. Distinct from Accepted and
#: Rejected so the caller can tell "annotate this" from "take me there".
_AUTO_ANNOTATE_OPEN_GATE = 100
_AUTO_ANNOTATE_OPEN_UMAP = 101
[docs]
class AnnotateScreen(QWidget):
"""Main Qt widget for the annotate app.
:param parent: parent widget.
"""
train_requested = Signal(str, dict)
def __init__(self, parent: Optional[QWidget] = None):
"""Build the annotation grid, its controls and its shortcuts.
:param parent: parent widget.
"""
super().__init__(parent)
from ..theme import ensure_widget_qss_applied
ensure_widget_qss_applied(
GRID_OBJECT_NAME, CONSOLE_SWITCH_NAME, root=self)
self._settings = AnnotateSettings()
self._offset = 0
self._total = 0
self._page_paths: List[Tuple[str, Optional[int]]] = []
self._filtered_rows: Optional[List[Tuple[str, Optional[int]]]] = None
#: rendered spread/class-balance summary when the uncertainty queue is on
self._queue_summary: str = ""
self._round_index = 0
self._retrain_worker: Optional[_RetrainWorker] = None
self._similar_worker: Optional[_SimilarityWorker] = None
self._similar_cache: Optional[Tuple[Any, ...]] = None
self._similar_notice_until = 0.0
self._similar_sources: List[str] = []
self._similar_navigation: Optional[Dict[str, Any]] = None
#: The Suggest run, kept separate from the retrain above so
#: one can be running while the other is retired. They fit
#: the same kind of model and must not be the same slot.
self._suggest_worker: Optional[_SuggestWorker] = None
#: True when the outstanding suggestions came from a fit whose
#: negatives were INVENTED -- see `_synthetic_negatives_needed`.
#: Such a run is a RANKING and not a verdict, so the bulk KEEP is
#: withheld: accepting two thousand mostly-negative guesses in one
#: click is the worst thing this button could do.
self._suggestions_are_a_ranking = False
self._last_round = None
self._stop_verdict = None
self._object_request = None
self._object_rows: Optional[List[Tuple[str, Optional[int]]]] = None
#: the routed request's reason, kept on the header through page loads
self._request_note = ""
self._blind: Optional[Dict[str, Any]] = None
self._object_opener = self.open_object_request
self._pending_updates: Dict[str, Optional[int]] = {}
self._page_verdicts: Dict[str, int] = {}
self._local_verdicts: Dict[str, Optional[int]] = {}
self._pending_verdicts: Dict[str, Optional[int]] = {}
self._judge_totals: Dict[str, int] = {
"left": 0, "confirmed": 0, "rejected": 0}
self._verdict_ready: Optional[Tuple[str, str, str]] = None
self._worker: Optional[SaveWorker] = None
self._thumbs: List[_Thumbnail] = []
self._thumb_pixmaps: List[Optional[QPixmap]] = []
self._raw_thumb_images: List[Optional[Image.Image]] = []
self._page_worker: Optional[_PageLoadWorker] = None
self._pending_page_load = None
self._page_gen = 0
self._closing = False
self._settings_dialog: Optional[_SettingsDialog] = None
self._total_jobs = JobRunner(self, app_key="annotate count")
self._report_jobs = JobRunner(self, app_key="annotate report",
user_visible=False)
self._resize_timer = QTimer(self)
self._resize_timer.setSingleShot(True)
self._resize_timer.setInterval(150)
self._resize_timer.timeout.connect(self._reload_after_resize)
self._built_dims: Tuple[int, int] = (0, 0)
self._suggested_source = prefs.get_last_source("annotate")
self._focus_slot = 0
self._hover_slot: Optional[int] = None
#: Rubber-band selection. `_band_origin` is None whenever no drag is
#: in progress, and is what every band handler gates on.
self._band = None
self._band_origin = None
self._undo_stack: Deque[Tuple[int, str, Optional[int]]] = deque(
maxlen=UNDO_LIMIT)
self._redo_stack: Deque[Tuple[int, str, Optional[int]]] = deque(
maxlen=UNDO_LIMIT)
self._legend_expanded = False
self._build_ui()
self._install_shortcuts()
self.setFocusPolicy(Qt.StrongFocus)
try:
from ..gui_scale import add_listener
add_listener(self._on_gui_scale_changed)
except Exception:
LOG.debug("The GUI scale listener could not be installed",
exc_info=True)
try:
from ..dnd import install_dropzone
from ..dnd_handlers import AnnotateDropHandler
install_dropzone(self, AnnotateDropHandler(), self)
except Exception:
pass
self._status_timer = QTimer(self)
self._status_timer.setInterval(500)
self._status_timer.timeout.connect(self._refresh_status_label)
self._status_timer.start()
register_object_opener("annotate", self._object_opener)
self._follow_path_probes()
self._apply_suggested_source()
if self._suggested_source:
_vouch_later(self._suggested_source)
def _apply_suggested_source(self) -> None:
"""Offer the last-used source in the subtitle, if it is still there.
:func:`_probe_isdir` answers from a cache and never stats on the
calling thread -- see `spacr.qt.path_probe`'s module docstring for the
twenty-second `os.path.exists` that made it necessary. An unknown path
comes back ABSENT, which is the right way round for this label:
leaving the placeholder standing for a moment and then offering the
suggestion is a screen that learned something, while offering a folder
that turns out to be gone and then retracting it is one that lied.
`_follow_path_probes` is the half that supplies the answer.
The probe's own answer is enough HERE, without the stricter
:func:`_vouched_dir` the pickers use: naming a folder in a subtitle
costs nothing if the mount is asleep, while opening a file dialog in
it freezes the window. The two questions are asked separately and
answered separately -- ``__init__`` puts the harder one -- because
one answer genuinely cannot serve both; see the module comment beside
:func:`_vouched_dir`.
Split out of ``__init__`` precisely so the probe can re-run this
same decision later without duplicating it.
"""
if self._settings.src:
return
if self._suggested_source and _probe_isdir(self._suggested_source):
self._src_label.setText(
f"Suggested (last used): {self._suggested_source}"
)
self._src_label.setProperty("i18nSkipText", True)
def _follow_path_probes(self) -> None:
"""Fill the suggestion in when the background stat finally answers.
The connection is to `path_probe.probes`, which is process-wide and
outlives every screen ever built on it, so the slot is a plain
closure kept on the instance and withdrawn twice over: in
``closeEvent`` for the ordinary path, and on ``destroyed`` for a
screen that is deleted without being closed. It is guarded on
RuntimeError besides: a queued emission can land after Qt has
deleted the C++ half of a screen whose Python wrapper is still
alive, and touching the label then is an abort, not an exception
that anything upstream would catch.
"""
def landed(path: str, answer: bool) -> None:
"""Re-run the suggestion when the probe answers about its path.
:param path: the path whose answer just changed. Every probe in
the process arrives here, so anything but this screen's own
remembered source is ignored.
:param answer: what the probe now says. Only True is acted on --
a folder that is gone leaves the placeholder standing.
:returns: nothing; the subtitle is the output.
"""
try:
if self._closing:
return
if answer and path == self._suggested_source:
self._apply_suggested_source()
except RuntimeError:
pass
self._path_probe_landed = landed
signal = path_probe.probes.answered
signal.connect(landed)
withdrawn: List[bool] = []
def let_go(*_args) -> None:
"""Withdraw this screen from the process-wide probe signal.
:param _args: whatever the caller passes. ``destroyed`` carries
the dying QObject and ``closeEvent`` passes nothing, and this
reads neither.
:returns: nothing. Safe to call twice: the second call is a
no-op rather than a disconnect of an absent slot.
"""
if withdrawn:
return
withdrawn.append(True)
try:
signal.disconnect(landed)
except (RuntimeError, TypeError):
pass
self._release_path_probe = let_go
self.destroyed.connect(let_go)
def _install_folds(self, header, row) -> Optional[FoldStrip]:
"""Put the folded modules' buttons on the masthead ``row``.
One button per entry in :data:`FOLDED_APPS`, drawn as that
module's own icon and lit on hover in its maturity colour. The
strip is added past the row's stretch, so it stays right-aligned
however wide the title column gets.
Guarded on its own: a screen that opens without its fold buttons
is a smaller screen, while an exception raised here would be no
annotation screen at all.
:returns: the strip, or None if it could not be built.
"""
self._fold_page_title = "Annotate"
try:
openers = [FoldOpener(self, key, FOLD_BUILDERS[key])
for key in FOLDED_APPS]
strip = FoldStrip([(o.key, o.open) for o in openers], header)
for opener in openers:
restate_fold_button(strip.button_for(opener.key), opener.key)
row.addWidget(strip)
except Exception:
LOG.debug("Could not install the annotate fold strip",
exc_info=True)
return None
self._fold_openers = openers
self._fold_strip = strip
return strip
def _build_ui(self):
"""Lay out the thumbnail grid over the class and navigation rows.
The crop pane's minimum height is pinned to
:data:`CROP_PANE_MIN_HEIGHT` rather than left to the stack, whose
own minimum is the empty state's full height (321 px). That minimum
is what the window had to grow by when the console opened: at
1366 x 768 the screen's minimum was taller than the screen. The
grid does not need it -- a page is what fits (item 512) -- so the
crop pane gives room up to the console before the window grows.
"""
PALETTE = tile_palette()
outer = QVBoxLayout(self)
outer.setContentsMargins(SPACING["lg"], SPACING["lg"],
SPACING["lg"], SPACING["lg"])
outer.setSpacing(SPACING["md"])
header = QWidget()
head_row = QHBoxLayout(header)
head_row.setContentsMargins(0, 0, 0, 0)
head_row.setSpacing(SPACING["lg"])
title_column = QWidget(header)
hbox = QVBoxLayout(title_column)
hbox.setContentsMargins(0, 0, 0, 0)
hbox.setSpacing(4)
title = QLabel("Annotate")
title.setObjectName("TitleHeading")
hbox.addWidget(title)
self._src_label = QLabel(tr("No source selected — click Open source…"))
self._src_label.setObjectName("SubtitleSmall")
hbox.addWidget(self._src_label)
head_row.addWidget(title_column)
head_row.addStretch(1)
self._install_folds(header, head_row)
outer.addWidget(header)
outer.addWidget(Divider())
from ..widgets.flow import FlowHost, FlowLayout
toolbar = FlowHost()
row = FlowLayout(toolbar, spacing=SPACING["sm"])
row.setContentsMargins(0, 0, 0, 0)
self._btn_open = QPushButton("Open source…")
self._btn_open.setObjectName("PrimaryButton")
self._btn_open.setIcon(iconset.contrast_icon("open"))
self._btn_open.setCursor(Qt.PointingHandCursor)
self._btn_open.clicked.connect(self._on_pick_source)
row.addWidget(self._btn_open)
self._btn_settings = QPushButton("Settings")
self._btn_settings.setIcon(iconset.icon("settings"))
self._btn_settings.setCursor(Qt.PointingHandCursor)
self._btn_settings.clicked.connect(self._on_open_settings)
row.addWidget(self._btn_settings)
row.addWidget(self._build_blind_toggle())
row.addWidget(self._build_similar_button())
row.addWidget(self._build_similar_options())
row.addWidget(self._build_field_qc_button())
self._btn_prev = QPushButton("Back")
self._btn_prev.setIcon(iconset.icon("prev"))
self._btn_prev.setCursor(Qt.PointingHandCursor)
self._btn_prev.clicked.connect(self._on_prev)
row.addWidget(self._btn_prev)
self._btn_next = QPushButton("Next")
self._btn_next.setIcon(iconset.icon("next"))
self._btn_next.setLayoutDirection(Qt.RightToLeft)
self._btn_next.setCursor(Qt.PointingHandCursor)
self._btn_next.clicked.connect(self._on_next)
row.addWidget(self._btn_next)
self._btn_skip = QPushButton("Skip to last annotated")
self._btn_skip.setIcon(iconset.icon("skip"))
self._btn_skip.setCursor(Qt.PointingHandCursor)
self._btn_skip.clicked.connect(self._on_skip)
row.addWidget(self._btn_skip)
self._btn_count = QPushButton("Class counts")
self._btn_count.setIcon(iconset.icon("chart"))
self._btn_count.setCursor(Qt.PointingHandCursor)
self._btn_count.clicked.connect(self._on_class_counts)
row.addWidget(self._btn_count)
self._btn_coverage = QPushButton("Coverage")
self._btn_coverage.setIcon(iconset.icon("chart"))
self._btn_coverage.setCursor(Qt.PointingHandCursor)
self._btn_coverage.setToolTip(
"Summarize annotation provenance by class, well, plate and "
"active-learning round. This identifies label sets concentrated "
"within a small number of wells or plates."
)
self._btn_coverage.clicked.connect(self._on_coverage)
row.addWidget(self._btn_coverage)
self._btn_retrain = QPushButton("Retrain")
self._btn_retrain.setIcon(iconset.icon("classify"))
self._btn_retrain.setCursor(Qt.PointingHandCursor)
self._btn_retrain.setToolTip(
"Fit a model on the labels made so far, score every crop with "
"it, and re-rank the uncertainty queue — without leaving this "
"screen. Held out by well, so the accuracy is not an artefact of "
"labelling one field of view. Each round writes a model card."
)
self._btn_retrain.clicked.connect(self._on_retrain)
row.addWidget(self._btn_retrain)
self._btn_suggest = QPushButton("Suggest…")
self._btn_suggest.setIcon(iconset.icon("classify"))
self._btn_suggest.setCursor(Qt.PointingHandCursor)
self._btn_suggest.setToolTip(
"Fit boosted trees on the labels made so far and write the "
"model's proposed label onto every crop you have not answered "
"yet — most confident first, dashed rather than solid, and only "
"where you have not already decided. Then keep them all, throw "
"them all away, or answer the ones it got wrong one at a time."
)
self._btn_suggest.clicked.connect(self._on_suggest_menu)
row.addWidget(self._btn_suggest)
self._btn_suggest_cancel = QPushButton(tr("Cancel"))
self._btn_suggest_cancel.setObjectName("AnnotateSuggestCancel")
self._btn_suggest_cancel.setCursor(Qt.PointingHandCursor)
self._btn_suggest_cancel.setToolTip(tr(
"Stop the suggestion run at its next step. No new suggestions "
"are written if it stops before the writing step; earlier "
"suggestions may have been cleared and round scores may have "
"been updated. Your annotations are unchanged. Shown only "
"while a run is going."))
self._btn_suggest_cancel.clicked.connect(self._cancel_suggest)
self._btn_suggest_cancel.hide()
row.addWidget(self._btn_suggest_cancel)
self._btn_curve = QPushButton("Rounds")
self._btn_curve.setIcon(iconset.icon("chart"))
self._btn_curve.setCursor(Qt.PointingHandCursor)
self._btn_curve.setToolTip(
"Held-out accuracy per round, and whether the last stretch of "
"labelling moved it. This is how you find out you can stop."
)
self._btn_curve.clicked.connect(self._on_learning_curve)
row.addWidget(self._btn_curve)
self._btn_train = QPushButton("Train…")
self._btn_train.setIcon(iconset.icon("classify"))
self._btn_train.setCursor(Qt.PointingHandCursor)
self._btn_train.setToolTip(
"Train a model on the current annotations, on the images or on "
"the measured features, and apply it to the whole dataset."
)
train_menu = QMenu(self._btn_train)
action_cv = train_menu.addAction(
iconset.icon("classify"), "On the images (CNN / Transformer)…")
action_cv.setToolTip(
"Generate a training dataset from the current annotations "
"and train a Torch CNN / Transformer classifier, then apply "
"it to the full dataset. Opens the Classify screen with "
"this source pre-selected."
)
action_cv.triggered.connect(self._on_train_cv)
action_xg = train_menu.addAction(
iconset.icon("chart"), "On the measured features (XGBoost)…")
action_xg.setToolTip(
"Train an XGBoost model on the measurement features "
"using the current annotations as class labels, then apply "
"it to score the full dataset. Opens the ML Analyze screen "
"with this source pre-selected."
)
action_xg.triggered.connect(self._on_train_xg)
self._btn_train.setMenu(train_menu)
self._train_menu = train_menu
self._btn_train_cv = self._btn_train
self._btn_train_xg = self._btn_train
row.addWidget(self._btn_train)
self._btn_auto = QPushButton("Auto-annotate…")
self._btn_auto.setCursor(Qt.PointingHandCursor)
self._btn_auto.setToolTip(
"Label a whole population at once — by metadata, by measurement "
"thresholds, by a gate, or by UMAP cluster.")
self._btn_auto.clicked.connect(self._on_auto_annotate)
row.addWidget(self._btn_auto)
self._btn_browse_db = QPushButton("View in database")
self._btn_browse_db.setCursor(Qt.PointingHandCursor)
self._btn_browse_db.setToolTip(
"Open png_list in the Database Browser to see the annotation "
"column as it is stored.")
self._btn_browse_db.clicked.connect(self._on_browse_db)
row.addWidget(self._btn_browse_db)
self._btn_annotate_page = QPushButton("Annotate page")
self._btn_annotate_page.setCursor(Qt.PointingHandCursor)
self._btn_annotate_page.setToolTip(
"Give every image on this page the same class. Asks which.")
self._btn_annotate_page.clicked.connect(self._on_annotate_page)
row.addWidget(self._btn_annotate_page)
self._btn_clear_page = QPushButton("Clear page")
self._btn_clear_page.setCursor(Qt.PointingHandCursor)
self._btn_clear_page.setToolTip(
"Remove the annotations on this page only. Undoable.")
self._btn_clear_page.clicked.connect(self._on_clear_page)
row.addWidget(self._btn_clear_page)
self._btn_clear = QPushButton("Clear column")
self._btn_clear.setObjectName("DangerButton")
self._btn_clear.setIcon(iconset.icon("clear", color=PALETTE["error"]))
self._btn_clear.setCursor(Qt.PointingHandCursor)
self._btn_clear.clicked.connect(self._on_clear_column)
row.addWidget(self._btn_clear)
self._btn_test_data = QPushButton(tr("Load test data"))
self._btn_test_data.setCursor(Qt.PointingHandCursor)
self._btn_test_data.setToolTip(tr(
"Download an example plate to annotate, and fill in the settings "
"that go with it."))
self._btn_test_data.clicked.connect(self._choose_the_test_data)
row.addWidget(self._btn_test_data)
self._btn_generate = QPushButton("Generate annotation database…")
self._btn_generate.setCursor(Qt.PointingHandCursor)
self._btn_generate.setToolTip(
"Build a set of crops to annotate by streaming them from the "
"merged arrays or from the coordinates in measurements.db, "
"without measuring the plate again. The result is registered as "
"a new png_list table.")
self._btn_generate.clicked.connect(self._on_generate_annotation_db)
row.addWidget(self._btn_generate)
self._page_label = QLabel("")
self._page_label.setObjectName("SubtitleSmall")
self._page_label.setProperty("i18nSkipText", True)
row.addWidget(self._page_label)
outer.addWidget(toolbar)
self._al_label = QLabel("")
self._al_label.setObjectName("SubtitleSmall")
self._al_label.setProperty("i18nSkipText", True)
self._al_label.setWordWrap(True)
self._al_label.hide()
outer.addWidget(self._al_label)
outer.addWidget(self._build_judge_bar())
outer.addWidget(self._build_key_legend())
self._content_stack = QStackedWidget()
self._content_stack.setMinimumHeight(CROP_PANE_MIN_HEIGHT)
self._empty_state = EmptyState(
title="Open an experiment to start annotating",
subtitle=(
"Pick a folder that contains "
"`measurements/measurements.db`. Left-click an image to "
"assign class 1, right-click for class 2, click again to "
"clear — or go keyboard-only: 1–9 label and jump to the "
"next crop. Annotations save in the background."
),
icon=iconset.accent_icon("tag"),
cta_label="Open source…",
on_action=self._on_pick_source,
)
self._content_stack.addWidget(self._empty_state)
self._grid_scroll = QScrollArea()
self._grid_scroll.setWidgetResizable(True)
self._grid_scroll.setFrameShape(QScrollArea.NoFrame)
self._grid_scroll.setVerticalScrollBarPolicy(Qt.ScrollBarAlwaysOff)
self._grid_scroll.setHorizontalScrollBarPolicy(Qt.ScrollBarAlwaysOff)
self._grid_scroll.viewport().setAutoFillBackground(False)
self._grid_scroll.viewport().setObjectName(GRID_VIEWPORT_NAME)
self._grid_holder = QWidget()
self._grid_holder.setObjectName(GRID_OBJECT_NAME)
self._grid_layout = QGridLayout(self._grid_holder)
self._grid_layout.setSpacing(SPACING["sm"])
self._grid_layout.setContentsMargins(SPACING["sm"], SPACING["sm"],
SPACING["sm"], SPACING["sm"])
self._grid_scroll.setWidget(self._grid_holder)
self._grid_holder.setAutoFillBackground(False)
self._zoom_overlay = _ZoomOverlay(self._grid_scroll.viewport())
self._zoom_overlay.dismissed.connect(self._fold_zoom_back)
self._grid_scroll.installEventFilter(self)
self._grid_scroll.viewport().installEventFilter(self)
self._grid_holder.installEventFilter(self)
self._content_stack.addWidget(self._grid_scroll)
self._content_stack.setCurrentWidget(self._empty_state)
self._runtime_splitter = CollapsibleSplitter(
Qt.Vertical, self, persist_key="annotate::runtime")
self._runtime_splitter.add_section(
self._content_stack, "Crops", persist_key="annotate/Crops",
stretch=4)
self._btn_copy_console = QPushButton(tr("Copy console"))
self._btn_copy_console.setObjectName("GhostButton")
self._btn_copy_console.setCursor(Qt.PointingHandCursor)
self._btn_copy_console.setToolTip(tr(
"Copy everything in the console, section headers included."))
self._btn_copy_console.clicked.connect(self._on_copy_console)
self._btn_file_issue = QPushButton(tr("File as issue"))
self._btn_file_issue.setObjectName("GhostButton")
self._btn_file_issue.setCursor(Qt.PointingHandCursor)
self._btn_file_issue.setToolTip(tr(
"Open a pre-filled GitHub issue with the last traceback and "
"environment. You review it before submitting."))
self._btn_file_issue.setVisible(False)
self._btn_file_issue.setEnabled(False)
self._btn_file_issue.clicked.connect(self._on_file_issue)
from ..widgets import ConsolePanel
self._console = ConsolePanel(active_app_label="Annotate")
self._console.setMinimumHeight(180)
_original_append_error = self._console.append_error
def _append_error_and_offer_the_report(text, *args, **kwargs):
"""Show the error as before, then offer to file it.
WRAPS rather than replaces, so the original behaviour is unchanged even
if the offer fails -- an error message must still appear when the thing
that would report it is broken.
"""
try:
return _original_append_error(text, *args, **kwargs)
finally:
self.note_console_error()
self._console.append_error = _append_error_and_offer_the_report
self._console_wrap = self._runtime_splitter.add_section(
self._console, "Console + AI", persist_key="annotate/Console + AI",
stretch=2,
actions=[self._btn_copy_console, self._btn_file_issue])
self._console_wrap.hide()
outer.addWidget(self._runtime_splitter, 1)
bottom = QWidget(self)
bottom_row = QHBoxLayout(bottom)
bottom_row.setContentsMargins(0, 0, 0, 0)
bottom_row.setSpacing(SPACING["sm"])
self._status_label = _BlindStatusLabel(tr("Ready."))
self._status_label._blind_owner = weakref.ref(self)
self._status_label.setObjectName("SubtitleSmall")
bottom_row.addWidget(self._status_label, 1)
self._console_switch = QToolButton(self)
self._console_switch.setObjectName(CONSOLE_SWITCH_NAME)
self._console_switch.setProperty("i18nSkipText", True)
self._set_console_switch_text(False)
self._console_switch.setCheckable(True)
self._console_switch.setCursor(Qt.PointingHandCursor)
self._console_switch.setFocusPolicy(Qt.NoFocus)
console_tip = "Show or hide the Console + AI pane."
self._console_switch.setProperty("_spacr_i18n_tooltip", console_tip)
self._console_switch.setToolTip(tr(console_tip))
self._console_switch.toggled.connect(self._on_console_switch)
bottom_row.addWidget(self._console_switch)
from ..widgets import AiToggleLabel
self._ai_switch = AiToggleLabel()
self._ai_switch.toggled.connect(self._on_ai_switch)
bottom_row.addWidget(self._ai_switch)
outer.addWidget(bottom)
self._rebuild_grid()
def _on_console_switch(self, on: bool) -> None:
"""Expand or collapse Annotate's merged Console + AI pane.
Opened again, the pane takes the height the user last dragged it to
when there is one, and otherwise about a third of the column.
"""
if self._blind is not None:
self._console_wrap.hide()
return
self._console_wrap.setVisible(on)
self._set_console_switch_text(on)
pane = self._runtime_splitter.pane("Console + AI")
if on and pane is not None and pane.extent > 0:
self._runtime_splitter.rebalance(grow=pane)
elif on:
height = max(480, self._runtime_splitter.height())
self._runtime_splitter.setSizes(
[max(240, int(height * 0.62)), max(180, int(height * 0.38))])
def _choose_the_test_data(self, *, chooser=None, ask=None) -> str:
"""Ask which half of the example plate to use, fetch it, and open it.
The two routes need different halves of the plate and differ in size,
so the choice is made in a dialog that can describe both before either
starts -- see :class:`TestDataChooser`.
Whichever route is taken, the press ends with the plate OPEN: the
crops on the screen, the save worker running, and the settings filled
in from the pack that shipped with the data. Data already on disk is
opened at once; a missing half is downloaded first, with the shared
progress dialog, and the plate is opened from the download's
completion callback. Nothing waits on a timer, so a download that
takes a minute on a busy machine ends in the same state as one that
takes a second.
Load needs the crops and the measurements database. Stream needs the
merged arrays AND that database, because the database is what lists
the objects and where each one sits; the database half is fetched
first when it is missing, then the arrays.
:param chooser: replaces the dialog, for tests.
:param ask: replaces every downloader, for tests. Called as
``ask(parent, destination, on_done)`` once per missing half.
:returns: the source that was opened, or ``""`` when nothing was
chosen or a download is still running.
"""
dialog = chooser if chooser is not None else TestDataChooser(self)
if hasattr(dialog, "exec"):
dialog.exec()
route = str(getattr(dialog, "chosen", "") or "")
if not route:
return ""
from .. import hf_download
destination = hf_download.example_plate_folder()
destination.mkdir(parents=True, exist_ok=True)
missing = []
if not (destination / "measurements" / "measurements.db").is_file():
missing.append(ask or hf_download.download_annotate_example)
if route == "stream":
merged = destination / "merged"
if not (merged.is_dir() and any(merged.glob("*.npy"))):
missing.append(ask or hf_download.download_measure_example)
if not missing:
return self._use_the_test_data(destination, route)
was = self._btn_test_data.text()
self._btn_test_data.setEnabled(False)
self._btn_test_data.setText(tr("Fetching test data…"))
def _restore():
"""Give the button back, whether the download worked or not."""
try:
self._btn_test_data.setEnabled(True)
self._btn_test_data.setText(was)
except RuntimeError:
pass
def _next(result=True, error=""):
"""Start the next missing half, or open the plate after the last."""
if self._closing:
return
if result is None:
_restore()
LOG.info("test data not downloaded: %s", error)
self._console.append_notice(
"Test data was not downloaded: {detail}\n",
detail=error or "cancelled")
return
if missing:
download = missing.pop(0)
download(self, destination, _next)
return
_restore()
self._use_the_test_data(destination, route)
_next()
return ""
def _use_the_test_data(self, destination, route: str) -> str:
"""Fill the settings from the plate's own pack, then open the plate.
The pack's display settings are taken -- the annotation column the
labels live in, the crop size, the channels, the image type -- and its
PATHS are not: they were written on the publisher's machine, so
``src`` is always the LOCAL plate folder, set after the pack so that
nothing from the pack can be the last write. It is the FOLDER, not the
database: the settings form derives the database from the folder, and
a source naming the database file sent it looking for
``measurements.db/measurements/measurements.db``.
Opening goes through :meth:`_open_source`, the same path as picking
the folder by hand, which counts and reads the crops off the GUI
thread.
:returns: the source that was opened, or ``""`` when the plate has
no database to open.
"""
from pathlib import Path
from ...crops import LOAD_IMAGES, STREAM_IMAGES
destination = Path(destination)
database = destination / "measurements" / "measurements.db"
self._apply_test_data_settings(
_read_example_settings(destination / "settings"
/ "annotate_settings.csv"))
self._settings.crop_source = (
STREAM_IMAGES if route == "stream" else LOAD_IMAGES)
source = str(destination)
self._settings.src = source
self._settings.db_path = str(database)
if not database.is_file():
self._console.append_notice(
"The test data has no measurements database at {path}\n",
path=str(database))
return ""
try:
self._open_source(source)
except Exception as exc: # noqa: BLE001
LOG.warning("could not open the test data", exc_info=True)
self._console.append_notice(
"Could not open the test data: {detail}\n", detail=exc)
return ""
self._console.append_notice("Test data ready: {path}\n", path=source)
return source
def _apply_test_data_settings(self, rows: Dict[str, str]) -> None:
"""Copy the display settings a dataset shipped with onto the screen.
Field by field, and a value that does not parse leaves that field as
it was. ``annotation_column`` falls back to ``infected``, the column
the example's labels are stored in, so the published labels show even
when the pack does not name it.
:param rows: the pack, as :func:`_read_example_settings` returns it.
"""
s = self._settings
s.annotation_column = (rows.get("annotation_column")
or "infected").strip()
size = rows.get("crop_size") or rows.get("img_size")
try:
if size:
value = int(float(size))
s.image_size = (value, value)
except ValueError:
LOG.debug("example settings: crop size %r is not usable", size)
channels = rows.get("channels")
if channels:
s.channels = _csv_to_list(channels)
image_type = rows.get("image_type")
if image_type and image_type != "None":
s.image_type = image_type
def _on_copy_console(self) -> None:
"""Copy the whole console, and say so.
A clipboard write is silent, so a button that appears to do nothing is
indistinguishable from one that failed. The caption reports it, and it
is TRANSLATED at the moment of writing: the language pass ran when this
screen was built and does not run again, so an English literal set by a
handler would stay English for the session.
THE BUTTON IS THE TIMER'S CONTEXT, and that third argument is the
whole of a bug rather than tidiness. `QTimer.singleShot(msec,
functor)` belongs to nothing: it fires 1.2 seconds later whatever has
happened in the meantime, and the functor holds this screen, so the
deleted thing it reaches for is the C++ QPushButton:
RuntimeError: libshiboken: Internal C++ object
(PySide6.QtWidgets.QPushButton) already deleted.
A user who copies the console and leaves the screen inside 1.2
seconds gets that in the event loop. In the suite it was worse than
an error in the wrong place -- it was an error in the wrong TEST:
`test_annotate_console_can_be_copied_and_reported.py` presses this
button twice, its screens are torn down at the end of those tests,
and the two strays landed in whichever test happened to be spinning
an event loop 1.2 seconds later. Measured:
`test_annotate_console_can_be_copied_and_reported.py` plus
`test_annotate_keyboard.py` reported the error against
`test_auto_advance_skips_already_labelled_crops`, which touches
neither the console nor a button.
Passing the button as the context object makes Qt cancel the pending
call when the button is destroyed, which is the guarantee the
two-argument form does not give. Same pair before and after:
102 passed and 1 error, then 103 passed.
"""
if self._blind is not None:
return
try:
text = self._console.copy_all()
except Exception as exc: # noqa: BLE001
self._console.append_notice(
"Could not copy the console: {detail}\n", detail=exc)
return
button = self._btn_copy_console
button.setText(tr("Copied"))
QTimer.singleShot(1200, button,
lambda: button.setText(tr("Copy console")))
LOG.debug("copied %d console characters", len(text))
[docs]
def note_console_error(self) -> None:
"""Reveal "File as issue" because something went wrong.
The button appears in this console when there is an issue, and is
hidden until then rather than always present,
because a permanently visible report button invites reports with no
traceback attached, which are the ones nobody can act on.
Still gated on the user's opt-in, the same gate the module screens use.
Opting in reveals the ACTION; nothing is ever submitted in response to
the failure itself.
"""
button = getattr(self, "_btn_file_issue", None)
if button is None:
return
try:
from ..ai import settings as _ai_settings
allowed = _ai_settings.get_auto_file_issues()
except Exception: # noqa: BLE001
allowed = False
button.setVisible(bool(allowed))
button.setEnabled(bool(allowed))
def _on_file_issue(self) -> None:
"""Open a pre-filled GitHub issue for what the console is holding.
THE CALL WAS WRONG AND THE BUTTON HAD NEVER FILED ANYTHING.
``file_issue(self, {"screen": "annotate"}, body)`` handed the screen
where the traceback goes, a dict where the app id goes and the
console text where the settings go, so the first thing the reporter
did was ``sanitize_path(<AnnotateScreen>)`` and the user got
``Could not file the issue: 'AnnotateScreen' object has no attribute
'replace'``. The ``except TypeError`` around it caught a signature
mismatch that Python never raised: every argument was positional and
the arity was right. It went unnoticed because the button was hidden
behind an opt-in that shipped off; `auto_file_issues` now defaults on
and this button is part of the default experience.
The console text is read here, on the GUI thread, because reading a
widget is the one part that must happen here. Everything after it --
resolving a token through ``gh auth token`` and POSTing to
api.github.com -- is what the module screens measured at up to 28
seconds on a bad network, so it goes to this screen's reporting
runner rather than running inline.
The preview is shown whatever the reporting mode is, because the
button's own tooltip promises it ("You review it before submitting")
and because a press is already the affirmative act that 'always'
exists to avoid asking for. 'never' files nothing and says so.
"""
if self._blind is not None:
return
try:
body = self._console.copy_all()
except Exception: # noqa: BLE001
body = ""
if not str(body).strip():
self._console.append_notice(
"There is nothing in the console to report.\n")
return
try:
from PySide6.QtWidgets import QDialog
from ..ai.issue_preview import IssuePreviewDialog
from ..ai.issue_report import build_report, submit_report
from ..preferences import (ISSUE_PROMPT_NEVER,
get_issue_prompt_mode,
get_share_diagnostic_logs)
except Exception as exc: # noqa: BLE001
self._console.append_notice(
"Issue reporting is unavailable: {detail}\n", detail=exc)
return
try:
if get_issue_prompt_mode() == ISSUE_PROMPT_NEVER:
self._console.append_notice(
"Not filing a report: issue reporting is set to 'never' "
"in Preferences.\n")
return
report = build_report(
body, active_app="annotate",
include_log_tail=bool(get_share_diagnostic_logs()))
preview = IssuePreviewDialog(report, self, console=self._console,
traceback_text=body)
if preview.exec() != QDialog.Accepted:
self._console.append_notice(
"The report was not sent.\n")
return
approved = preview.approved_report()
except Exception as exc: # noqa: BLE001
self._console.append_notice(
"Could not file the issue: {detail}\n", detail=exc)
return
def _send():
"""Post the approved report. Off the GUI thread; never raises."""
try:
return {"url": submit_report(approved)}
except Exception as exc: # noqa: BLE001 - reported, not hidden
return {"error": f"{type(exc).__name__}: {exc}"}
self._console.append_notice("Sending the approved report to GitHub…\n")
if not self._report_jobs.submit(_send, self._on_issue_filed):
self._console.append_notice(
"Could not file the issue: the reporter would not start.\n")
def _on_issue_filed(self, outcome: dict) -> None:
"""Say where the console's report went, or why it did not. GUI thread.
:param outcome: ``{"url": ...}`` or ``{"error": ...}`` from the
worker, which returns its failure as data because an ``except``
around the caller can no longer see it.
"""
error = (outcome or {}).get("error")
if error:
self._console.append_notice(
"Could not file the issue: {detail}\n", detail=error)
return
self._console.append_notice(
"The report was sent: {url}\n",
url=str((outcome or {}).get("url") or "")[:200])
def _set_console_switch_text(self, expanded: bool,
language: Optional[str] = None) -> None:
"""Render the localized caption without losing the arrow state.
Only the word is looked up. The arrow is state rather than prose, and
a caption composed with it first asks the catalog for a key that can
never exist.
"""
arrow = "▴" if expanded else "▾"
self._console_switch.setText(f"{tr('Console', language)} {arrow}")
[docs]
def retranslate_dynamic_content(
self, language: Optional[str] = None,
) -> None:
"""Re-render the captions this screen composes from live state."""
self._set_console_switch_text(
self._console_switch.isChecked(), language)
def _on_ai_switch(self, on: bool) -> None:
"""Enable chat routing and reveal the console when AI is selected."""
self._console.set_ai_active(on)
if not on:
return
if not self._console_switch.isChecked():
self._console_switch.setChecked(True)
from .. import ai as ai_module
if not self._console._current_provider_name:
configured = ai_module.configured_providers()
if configured:
self._console.set_ai_provider(
self._wanted_provider() or configured[0].name)
else:
self._console.append_notice(
"[AI] No vendor CLI installed. "
"Preferences → AI → Providers…\n")
self._ai_switch.setChecked(False)
def _wanted_provider(self) -> str:
"""The provider Preferences asks for, if it is actually installed.
:returns: the chosen provider's name, or ``""`` to let the console
take the first available one.
"""
from .. import ai as ai_module
from ..preferences import get_preferred_provider
wanted = get_preferred_provider()
if not wanted:
return ""
try:
names = {p.name for p in ai_module.configured_providers()}
except Exception: # noqa: BLE001
return ""
return wanted if wanted in names else ""
LEGEND_COMPACT = (
"<b>1</b>–<b>9</b> label + advance · <b>0</b> clear "
" · <b>← ↑ ↓ →</b> / <b>hjkl</b> move · "
"<b>Space</b> skip · <b>Backspace</b> back "
" · <b>u</b> undo · <b>Enter</b> next batch"
)
LEGEND_FULL = (
"<b>1</b>–<b>9</b> assign that class to the focused crop and jump to "
"the next unlabelled one · <b>0</b> clear the focused "
"crop · <b>← ↑ ↓ →</b> or <b>h j k l</b> move focus "
"without labelling · <b>Space</b> skip forward one "
" · <b>Backspace</b> step back one · "
"<b>u</b> undo the last label · <b>Enter</b> save this "
"page and load the next batch · mouse still works: "
"left-click = class 1, right-click = class 2."
)
def _build_key_legend(self) -> QWidget:
"""Build the always-visible keyboard cheat strip.
Nothing in it accepts focus — a legend that stole focus would break
the very keyboard flow it documents.
"""
PALETTE = tile_palette()
legend = QWidget()
legend.setObjectName("AnnotateKeyLegend")
legend.setFocusPolicy(Qt.NoFocus)
from ..theme import pane_surface
legend.setStyleSheet(
f"QWidget#AnnotateKeyLegend {{ background: {pane_surface('surface')};"
f" border: 1px solid {PALETTE['border_soft']};"
f" border-radius: 6px; }}"
)
lay = QHBoxLayout(legend)
lay.setContentsMargins(SPACING["sm"], SPACING["xs"],
SPACING["sm"], SPACING["xs"])
lay.setSpacing(SPACING["sm"])
self._legend_label = QLabel(self.LEGEND_COMPACT)
self._legend_label.setObjectName("SubtitleSmall")
self._legend_label.setTextFormat(Qt.RichText)
self._legend_label.setWordWrap(True)
self._legend_label.setFocusPolicy(Qt.NoFocus)
lay.addWidget(self._legend_label, 1)
self._kbd_hint = QLabel("")
self._kbd_hint.setObjectName("SubtitleSmall")
self._kbd_hint.setFocusPolicy(Qt.NoFocus)
self._kbd_hint.setStyleSheet(f"color: {PALETTE['warning']};")
lay.addWidget(self._kbd_hint, 0)
self._legend_toggle = QPushButton("?")
self._legend_toggle.setFocusPolicy(Qt.NoFocus)
self._legend_toggle.setCursor(Qt.PointingHandCursor)
self._legend_toggle.setMinimumWidth(max(28, self._legend_toggle.sizeHint().width()))
self._legend_toggle.setMaximumWidth(max(28, self._legend_toggle.sizeHint().width()))
self._legend_toggle.setToolTip("Show the full keyboard reference")
self._legend_toggle.clicked.connect(self._toggle_legend)
lay.addWidget(self._legend_toggle, 0)
self._legend = legend
return legend
def _build_judge_bar(self) -> QWidget:
"""The strip that says how to judge suggestions, and how many are left.
Item 512: "a clear and easy mechanism to annotate the suggestions as
correct or wrong with clear instructions and visual feedback". The
instruction names the click and the key for each verdict, the counts
say what this page and the whole column hold, and two buttons judge
what is left on the page at once. Hidden until the column holds a
suggestion or a judgement, so a screen that never used Suggest looks
as it always did. Nothing in it takes focus, for the reason the key
legend gives.
"""
PALETTE = tile_palette()
bar = QWidget()
bar.setObjectName("AnnotateJudgeBar")
bar.setFocusPolicy(Qt.NoFocus)
from ..theme import pane_surface
bar.setStyleSheet(
f"QWidget#AnnotateJudgeBar {{ background: {pane_surface('surface')};"
f" border: 1px solid {PALETTE['warning']};"
f" border-radius: 6px; }}"
)
lay = QHBoxLayout(bar)
lay.setContentsMargins(SPACING["sm"], SPACING["xs"],
SPACING["sm"], SPACING["xs"])
lay.setSpacing(SPACING["sm"])
text_column = QVBoxLayout()
text_column.setContentsMargins(0, 0, 0, 0)
text_column.setSpacing(2)
self._judge_hint = QLabel(tr(
"Suggested crops wear a ? badge. Click one or press Y to confirm "
"it; right-click or press N to reject it; U undoes."))
self._judge_hint.setObjectName("SubtitleSmall")
self._judge_hint.setWordWrap(True)
self._judge_hint.setFocusPolicy(Qt.NoFocus)
text_column.addWidget(self._judge_hint)
self._judge_label = QLabel("")
self._judge_label.setObjectName("SubtitleSmall")
self._judge_label.setProperty("i18nSkipText", True)
self._judge_label.setWordWrap(True)
self._judge_label.setFocusPolicy(Qt.NoFocus)
text_column.addWidget(self._judge_label)
lay.addLayout(text_column, 1)
self._btn_confirm_page = QPushButton(tr("Confirm the rest of the page"))
self._btn_confirm_page.setFocusPolicy(Qt.NoFocus)
self._btn_confirm_page.setCursor(Qt.PointingHandCursor)
self._btn_confirm_page.setToolTip(tr(
"Confirm every suggestion on this page you have not judged yet. "
"Undoable with U."))
self._btn_confirm_page.clicked.connect(
lambda: self._judge_page(confirm=True))
lay.addWidget(self._btn_confirm_page, 0)
self._btn_reject_page = QPushButton(tr("Reject the rest of the page"))
self._btn_reject_page.setFocusPolicy(Qt.NoFocus)
self._btn_reject_page.setCursor(Qt.PointingHandCursor)
self._btn_reject_page.setToolTip(tr(
"Reject every suggestion on this page you have not judged yet. "
"Undoable with U."))
self._btn_reject_page.clicked.connect(
lambda: self._judge_page(confirm=False))
lay.addWidget(self._btn_reject_page, 0)
self._judge_bar = bar
bar.hide()
return bar
def _toggle_legend(self) -> bool:
"""Flip the legend between the compact strip and the full reference."""
self._legend_expanded = not self._legend_expanded
from ..shortcuts import _refresh_screen_hints
_refresh_screen_hints(self)
return True
def _set_kbd_hint(self, text: str = "") -> None:
"""Show (or clear) the transient keyboard-mode message."""
if getattr(self, "_kbd_hint", None) is not None:
self._kbd_hint.setText(text)
def _install_shortcuts(self):
"""Bind the number keys, the arrows and undo.
THE KEYBOARD IS THE INTERFACE HERE. Annotation is thousands of
one-key decisions, and a hand that has to reach for the mouse between
each one does a fraction as many in an hour.
"""
from ..shortcuts import _bind_screen_key
for key, callback in (("PageUp", self._on_prev), ("PageDown", self._on_next),
("Alt+Left", self._on_prev), ("Alt+Right", self._on_next),
("Ctrl+Z", self._kbd_undo), ("Ctrl+Y", self._kbd_redo),
("Ctrl+Shift+Z", self._kbd_redo)):
_bind_screen_key(self, "Annotate", key, callback)
def _grid_area(self) -> Optional[QSize]:
"""The room the crops have, in device pixels, or None before layout.
Measured on the crop pane (the content stack) rather than on the
scroll viewport. They are the same rectangle -- the scroll area has
no frame and, since item 512, no scrollbars -- but the stack has
its size while the grid is still behind the empty state, which is
when the first fit after opening a source is computed, and a
viewport that has never been shown still reports its pre-layout
default.
"""
scroll = getattr(self, "_grid_scroll", None)
stack = getattr(self, "_content_stack", None)
if scroll is None or stack is None:
return None
try:
if not stack.isVisible():
return None
rect = stack.contentsRect()
except RuntimeError:
return None
if rect.width() <= 0 or rect.height() <= 0:
return None
return QSize(rect.width(), rect.height())
def _compute_grid_dims(self):
"""Fit as many ``image_size`` crops as the crop pane holds, no more.
THE PAGE IS WHAT FITS (item 512). Rows and columns come from the
room the pane gives and the size a tile is drawn at -- the crop
size plus its rings, the layout's gap and margins, all at the GUI
scale in force -- so the grid is never taller or wider than its
viewport and there is nothing left to scroll to; what does not fit
is the next page. Before the pane has been laid out (nothing shown
yet, or a test that pins the shape) the previous shape is kept,
because sizing from a 100x30 default would give one enormous cell.
"""
w, h = self._settings.image_size
area = self._grid_area()
if area is None:
cols = max(1, self._settings.grid_cols or 5)
rows = max(1, self._settings.grid_rows or 5)
else:
from ..gui_scale import scale_int
pad = TILE_INSET * 2
rows, cols = grid_that_fits(
area.width(), area.height(),
scale_int(w + pad), scale_int(h + pad),
gap=scale_int(SPACING["sm"]), margin=scale_int(SPACING["sm"]))
self._settings.grid_cols = cols
self._settings.grid_rows = rows
def _rebuild_grid(self):
"""Regenerate empty thumbnail widgets sized for current settings."""
self._fold_zoom_back()
self._compute_grid_dims()
for w in self._thumbs:
w.setParent(None)
w.deleteLater()
self._thumbs.clear()
self._hover_slot = None
self._thumb_pixmaps = [None] * (self._settings.grid_rows *
self._settings.grid_cols)
self._raw_thumb_images = [None] * len(self._thumb_pixmaps)
cols = self._settings.grid_cols
rows = self._settings.grid_rows
w, h = self._settings.image_size
pad = TILE_INSET * 2
resting = resting_border_color()
ring = current_ring_color()
for i in range(rows * cols):
thumb = _Thumbnail(i, border_color=resting, ring_color=ring)
thumb.setFixedSize(w + pad, h + pad)
thumb.left_clicked.connect(self._on_thumb_left)
thumb.right_clicked.connect(self._on_thumb_right)
thumb.shift_clicked.connect(self._on_thumb_shift)
thumb.hover_changed.connect(self._on_thumb_hover)
self._grid_layout.addWidget(thumb, i // cols, i % cols)
self._thumbs.append(thumb)
self._built_dims = (rows, cols)
self._focus_slot = max(0, min(self._focus_slot, len(self._thumbs) - 1))
self._refresh_focus_marks()
[docs]
def resizeEvent(self, event):
"""Re-fit the thumbnail grid after resize activity settles.
:param event: the resize event, passed to the base class; the grid is
re-measured from the widget's new size.
"""
super().resizeEvent(event)
self._refit_grid()
def _refit_grid(self) -> None:
"""Recompute the page from the room the crops have; reload if it moved.
Called from every route by which that room changes: the window
resizing, the console opening or a pane folding (the crop pane's
own Resize, caught in :meth:`eventFilter`) and the GUI scale
changing. The reload waits for the burst to settle and is skipped
when the grid on screen already has the new shape (``_built_dims``,
the rows and columns the grid was last built with), so a refit that
finds nothing to change costs no page load. The first crop on
screen stays: ``_offset`` is not touched, so the page that comes
back starts where this one did and only its length changes.
"""
if not getattr(self, "_grid_scroll", None):
return
prev = (self._settings.grid_rows, self._settings.grid_cols)
self._compute_grid_dims()
new = (self._settings.grid_rows, self._settings.grid_cols)
moved = new != prev or new != self._built_dims
if moved and self._worker is not None:
self._resize_timer.start()
def _on_gui_scale_changed(self, _scale: float) -> None:
"""The tiles just changed size in the same room: fit them again.
:param _scale: the new GUI scale; the refit reads it from
:mod:`spacr.qt.gui_scale` itself.
"""
if getattr(self, "_closing", False):
return
self._refit_grid()
@Slot()
def _reload_after_resize(self):
"""Apply the final geometry after a burst of resize events."""
if self._closing or self._worker is None:
return
self._compute_grid_dims()
wanted = (self._settings.grid_rows, self._settings.grid_cols)
built = len(self._thumbs) == wanted[0] * wanted[1]
if wanted == self._built_dims and built:
return
self._flush_pending()
self._rebuild_grid()
self._refresh_total(then=self._load_page)
def _starting_folder(self) -> str:
"""Where the source picker opens, decided without stat-ing anything.
`QFileDialog.getExistingDirectory` stats and then LISTS its starting
directory on the GUI thread, so handing it a path on a sleeping
`autofs` mount is the same twenty-second freeze as the subtitle stat
was, moved to the click that opens the picker.
BOTH candidates are paths the user supplied -- the source now open
and the one remembered from last session -- so both go through
:func:`_vouched_dir`, which offers a folder only when a real stat came
back and confirmed it. `path_probe`'s own answer would not do: it
reports an unanswered stat as PRESENT once its timeout is up, which is
the freeze back five seconds later. Neither candidate is asked for the
first time here: the open source was queued by :meth:`_open_source`
when it was opened and the remembered one during construction, so by
the time anybody reaches this button the answer is almost always
already in hand. One that is not, or one that has gone stale, is
re-asked in the background by :func:`_vouched_dir` itself and is
offered again on the press after next.
While blinded the picker opens in the home folder instead: opened in
the source, its path bar and listing would name the plate.
:returns: a folder to open the picker in, falling back to the
working directory, which is local by construction.
"""
if getattr(self, "_blind", None) is not None:
return os.path.expanduser("~")
for candidate in (self._settings.src, self._suggested_source):
if _vouched_dir(candidate):
return candidate
return os.getcwd()
def _on_pick_source(self):
"""Ask for a project to annotate."""
d = QFileDialog.getExistingDirectory(self, "Pick experiment source",
self._starting_folder())
if not d:
return
self._open_source(d)
def _open_source(self, src: str, *, then=None,
preserve_similarity: bool = False):
"""Open a project and load its first page of crops.
:param src: the project folder.
:param then: optional callback after its population count is ready.
:param preserve_similarity: retain cross-plate results during navigation.
"""
db_path = os.path.join(src, "measurements", "measurements.db")
if not os.path.isfile(db_path):
answer = QMessageBox.question(
self, "Database not found",
f"No file at:\n{db_path}\n\nUse it anyway?",
)
if answer != QMessageBox.Yes:
return
self._flush_pending()
self._leave_blind_unopened("another source was opened")
if self._worker:
self._worker.stop(wait=True)
self._worker = None
self._settings.src = src
self._settings.db_path = db_path
_ask_about_the_folder(src)
ensure_annotation_column(db_path, self._settings.annotation_column,
table=self._settings.png_table)
self._pending_verdicts.clear()
self._verdict_ready = None
self._ensure_verdict_column()
self._recount_judgements()
self._worker = SaveWorker(db_path, self._settings.annotation_column,
table=self._settings.png_table)
self._worker.start()
self._offset = 0
self._src_label.setText(f"{src} → {db_path}")
self._src_label.setProperty("i18nSkipText", True)
self._console.append_notice("Opened {name}\n", name=db_path)
prefs.push_recent_source("annotate", src)
self._content_stack.setCurrentWidget(self._grid_scroll)
self.setFocus(Qt.OtherFocusReason)
self._object_request = None
self._object_rows = None
self._similar_notice_until = 0.0
if not preserve_similarity:
self._similar_cache = None
self._similar_navigation = None
self._similar_result_plate.clear()
self._last_round = None
self._refresh_round_state()
self._refresh_total(then=then or self._rebuild_and_load)
def _rebuild_and_load(self):
"""Rebuild the grid against the (now realized) viewport, then load."""
self._rebuild_grid()
self._load_page()
def _build_blind_toggle(self) -> QPushButton:
"""The Blind switch: score the crops without knowing where they are from.
On, the crops of the population on screen are shuffled under a
blinding key (:func:`spacr.run_journal.start_blinding`), the source
line and a routed request's reason are hidden, and the tools that
show plates, wells or conditions (Coverage, Auto-annotate, View in
database) are disabled. Off asks first, then unblinds through
:func:`spacr.run_journal.unblind`, which records who did it and
when. An alpha feature, registered as ``AnnotateBlindToggle`` in
:data:`spacr.settings.ALPHA_FEATURES`.
:returns: the checkable button.
"""
from ..preferences import _apply_alpha_widgets
button = QPushButton(tr("Blind"), self)
button.setObjectName("AnnotateBlindToggle")
button.setCheckable(True)
button.setCursor(Qt.PointingHandCursor)
button.setToolTip(tr(
"Score blind: hide the source, plates, wells, conditions and file "
"names, and show the crops in a shuffled order. The key is kept "
"beside the run journal, outside the data folder. Turning it off "
"unblinds, and the journal records who unblinded and when. "
"Default off."))
button.toggled.connect(self._on_blind_toggled)
_apply_alpha_widgets(button)
self._btn_blind = button
return button
def _build_field_qc_button(self) -> QPushButton:
"""The Field QC button: label whole raw fields for the image-quality classifier.
Opens :class:`_FieldQCDialog` on a chosen folder of ``.npy`` fields.
An alpha feature, registered as ``AnnotateFieldQCButton`` in
:data:`spacr.settings.ALPHA_FEATURES`.
:returns: the button.
"""
from ..preferences import _apply_alpha_widgets
button = QPushButton(tr("Field QC…"), self)
button.setObjectName("AnnotateFieldQCButton")
button.setCursor(Qt.PointingHandCursor)
button.setToolTip(tr(
"Label whole raw fields as good, out of focus, saturated, debris, "
"bubble or empty, one at a time, with the classifier's guesses "
"ticked when a screened report exists. Saves qc/image_qc_labels.csv "
"for the image_qc_classifier_labels setting. Default not open."))
button.clicked.connect(self._on_field_qc)
_apply_alpha_widgets(button)
self._btn_field_qc = button
return button
def _on_field_qc(self) -> None:
"""Ask for a folder of raw fields and open the field-quality view on it."""
folder = QFileDialog.getExistingDirectory(
self, tr("Choose a folder of raw .npy fields"))
if not folder:
return
dialog = _FieldQCDialog(self)
dialog.load_folder(folder)
dialog.show()
self._field_qc_dialog = dialog
def _build_similar_button(self) -> QPushButton:
"""The Like this button: show the crops most like the current one.
Searches every crop of the open source by its measured features
(:func:`spacr.active_learning._similarity_index`) and pins the grid
to the query and its closest matches, most similar first, through
:meth:`open_object_request`, so a rare class found once can be
labelled many times. An alpha feature, registered as
``AnnotateFindSimilar`` in :data:`spacr.settings.ALPHA_FEATURES`.
:returns: the button.
"""
from ..preferences import _apply_alpha_widgets
button = QPushButton(tr("Like this"), self)
button.setObjectName("AnnotateFindSimilar")
button.setIcon(iconset.icon("classify"))
button.setCursor(Qt.PointingHandCursor)
button.setToolTip(tr(
"Show the requested number of crops whose measurements are most like the "
"selected crop, the one with the ring, most similar first, so a "
"rare class found once can be labelled many times. The first "
"search on a source reads its measurements and takes a few "
"seconds; later ones are instant. Back to all crops by opening "
"the source again. Default not run."))
button.clicked.connect(self._on_find_similar)
_apply_alpha_widgets(button)
self._btn_similar = button
return button
def _build_similar_options(self) -> QWidget:
"""Expose neighbour count and unanswered-only filtering beside Like this."""
from ..preferences import _apply_alpha_widgets
panel = QWidget(self)
panel.setObjectName("AnnotateSimilarityOptions")
layout = QHBoxLayout(panel)
layout.setContentsMargins(0, 0, 0, 0)
label = QLabel(tr("Similar crops"), panel)
self._similar_k = QSpinBox(panel)
self._similar_k.setObjectName("AnnotateSimilarCount")
self._similar_k.setRange(1, 1000000)
self._similar_k.setValue(100)
label.setToolTip(tr("Maximum number of similar crops; the reference crop is shown separately."))
label.setBuddy(self._similar_k)
layout.addWidget(label)
layout.addWidget(self._similar_k)
self._similar_unlabelled = QCheckBox(tr("Unlabelled only"), panel)
self._similar_unlabelled.setToolTip(tr(
"Only retrieve crops without a human class label in the current annotation column. "
"Cleared labels (blank or 0) and unanswered model suggestions remain eligible. "
"The selected reference crop stays visible even when labelled."))
layout.addWidget(self._similar_unlabelled)
self._similar_feature_kind = QComboBox(panel)
self._similar_feature_kind.setObjectName("AnnotateSimilarFeatureKind")
self._similar_feature_kind.addItem(tr("Auto features"), "auto")
self._similar_feature_kind.addItem(tr("Embeddings"), "embeddings")
self._similar_feature_kind.addItem(tr("Measurements"), "measurements")
self._similar_feature_kind.setToolTip(tr(
"All selected plates must use the same feature columns. Stored "
"embeddings must also have the same model settings and weight checksum."))
layout.addWidget(self._similar_feature_kind)
self._btn_similar_add_plate = QPushButton(tr("Add plate…"), panel)
self._btn_similar_add_plate.setObjectName("AnnotateSimilarAddPlate")
self._btn_similar_add_plate.setToolTip(tr(
"Choose another measurements.db to include in the next similarity search."))
self._btn_similar_add_plate.clicked.connect(self._on_add_similarity_plate)
layout.addWidget(self._btn_similar_add_plate)
self._btn_similar_clear_plates = QPushButton(tr("Clear plates"), panel)
self._btn_similar_clear_plates.setObjectName("AnnotateSimilarClearPlates")
self._btn_similar_clear_plates.clicked.connect(self._on_clear_similarity_plates)
layout.addWidget(self._btn_similar_clear_plates)
self._similar_result_plate = QComboBox(panel)
self._similar_result_plate.setObjectName("AnnotateSimilarResultPlate")
self._similar_result_plate.setPlaceholderText(tr("Results by plate"))
self._similar_result_plate.setMinimumWidth(90)
self._similar_result_plate.setToolTip(tr(
"Open matching crops from this plate in Annotate before assigning labels."))
self._similar_result_plate.currentIndexChanged.connect(
self._on_similar_result_plate)
layout.addWidget(self._similar_result_plate)
_apply_alpha_widgets(panel)
return panel
def _on_add_similarity_plate(self, path=None) -> bool:
"""Add an explicit database; the worker validates its feature space."""
if isinstance(path, bool):
path = None
if path is None:
path, _ = QFileDialog.getOpenFileName(
self, tr("Choose another plate database"),
self._starting_folder(), tr("SQLite databases (*.db)"))
if not path:
return False
path = os.path.abspath(str(path))
if (os.path.basename(path) != "measurements.db" or
os.path.basename(os.path.dirname(path)) != "measurements"):
self._status_label.setText(tr(
"Choose a plate's measurements/measurements.db file."))
return False
if path not in self._similar_sources and path != os.path.abspath(
self._settings.db_path or ""):
self._similar_sources.append(path)
self._similar_cache = None
self._status_label.setText(tr(
"{n} additional plate(s) selected for the next search.",
n=len(self._similar_sources)))
return True
def _on_clear_similarity_plates(self) -> None:
"""Return subsequent searches to the open source alone."""
self._similar_sources.clear()
self._similar_cache = None
self._similar_navigation = None
self._similar_result_plate.clear()
self._status_label.setText(tr("Similarity search will use the open source."))
def _similar_paths(self) -> Tuple[str, ...]:
"""The open database followed by each distinct explicitly added plate."""
return tuple(dict.fromkeys([os.path.abspath(self._settings.db_path),
*self._similar_sources]))
def _similar_query_key(self) -> Optional[str]:
"""The ``png_path`` of the selected crop, ``None`` on an empty page."""
slot = self._focus_slot
if not self._slot_is_valid(slot):
return None
return str(self._page_paths[slot][0])
def _on_find_similar(self) -> None:
"""Search for the crops most like the current one, off the GUI thread."""
if not self._settings.db_path:
QMessageBox.information(
self, tr("Open a source first"),
tr("Open an experiment source before searching it."))
return
if self._similar_worker is not None:
self._status_label.setText(tr("A search is already running."))
return
key = self._similar_query_key()
if key is None:
self._status_label.setText(tr("No crop is selected to match."))
return
paths = self._similar_paths()
feature_kind = self._similar_feature_kind.currentData()
if len(paths) > 1 and self._blind is not None:
self._status_label.setText(tr(
"Leave blind mode before searching across plates."))
return
cache = self._similar_cache
self._similar_notice_until = 0.0
index = None
if (cache is not None and cache[:2] == (frozenset(paths), self._settings.image_type)
and cache[3] == feature_kind
and cache[4] == _similarity_source_stamp(paths)):
index = cache[2]
self._btn_similar.setEnabled(False)
self._status_label.setText(
tr("Finding crops like this one…") if index is not None
else tr("Reading the measurements to compare crops by…"))
worker = _SimilarityWorker(self._settings.db_path,
self._settings.image_type, key,
index=index, k=self._similar_k.value(), parent=self,
unlabelled_only=self._similar_unlabelled.isChecked(),
annotation_column=self._settings.annotation_column,
png_table=self._settings.png_table,
pending_labels=self._pending_updates, writer=self._worker,
db_paths=paths, feature_kind=feature_kind)
worker.done.connect(self._on_similar_done)
worker.failed.connect(self._on_similar_failed)
worker.finished.connect(self._on_similar_finished)
self._similar_worker = worker
worker.start()
@Slot(object)
def _on_similar_done(self, result) -> None:
"""Keep the index and pin the grid to the query and its matches."""
from ...selection import ObjectRequest
if (result["db_path"] != self._settings.db_path or
tuple(result.get("db_paths", (os.path.abspath(result["db_path"]),))) != self._similar_paths() or
result.get("feature_kind", self._similar_feature_kind.currentData()) != self._similar_feature_kind.currentData() or
result.get("image_type", self._settings.image_type) != self._settings.image_type or
result.get("annotation_column", self._settings.annotation_column) != self._settings.annotation_column or
result.get("png_table", self._settings.png_table) != self._settings.png_table):
return
current_stamp = _similarity_source_stamp(self._similar_paths())
self._similar_cache = (frozenset(self._similar_paths()), result["image_type"],
result["index"], result.get("feature_kind", "auto"),
result.get("source_stamp", current_stamp))
hits = result["hits"]
if "db_path" in hits.columns:
self._similar_navigation = result
self._similar_result_plate.blockSignals(True)
self._similar_result_plate.clear()
for path in self._similar_paths():
count = int(hits["db_path"].eq(path).sum())
if path == os.path.abspath(result["db_path"]) or count:
src = os.path.dirname(os.path.dirname(path))
self._similar_result_plate.addItem(
tr("{name} · {n} matches", name=os.path.basename(src),
n=count), path)
self._similar_result_plate.setItemData(
self._similar_result_plate.count() - 1, path, Qt.ToolTipRole)
self._similar_result_plate.setCurrentIndex(0)
self._similar_result_plate.blockSignals(False)
self._present_similarity_source(os.path.abspath(result["db_path"]))
return
self._similar_navigation = None
self._similar_result_plate.clear()
keys = [result["key"]] + [str(k) for k in hits["key"]]
name = os.path.basename(result["key"])
request = ObjectRequest(
keys=keys,
reason=tr("{name} and the {n} crops most like it, most similar "
"first").format(name=name, n=len(hits)),
source="similarity",
context={"similarity": dict(zip(hits["key"], hits["similarity"])),
"unlabelled_only": result.get("unlabelled_only", False),
"requested_k": result.get("requested_k", 100)})
self.open_object_request(request)
self._status_label.setText(
tr("Searched {n} crops in {ms} ms.").format(
n=f"{len(result['index']):,}",
ms=f"{result['seconds'] * 1000:.0f}"))
if result.get("unlabelled_only"):
self._status_label.setText(self._status_label.text() + " " +
tr("{n} unlabelled matches; selected reference kept separately.", n=len(hits)))
self._similar_notice_until = time.monotonic() + 3.0
def _on_similar_result_plate(self, index: int) -> None:
"""Open a matching plate before showing crops that its writer owns."""
if index < 0 or self._similar_navigation is None:
return
result = self._similar_navigation
if (self._blind is not None or
result["image_type"] != self._settings.image_type or
result["annotation_column"] != self._settings.annotation_column or
result["png_table"] != self._settings.png_table or
result["feature_kind"] != self._similar_feature_kind.currentData()):
self._status_label.setText(tr(
"Search settings changed; run Like this again before opening another plate."))
return
path = self._similar_result_plate.itemData(index)
if not path:
return
if path == os.path.abspath(self._settings.db_path):
self._present_similarity_source(path)
return
if not os.path.isfile(path):
self._status_label.setText(tr("The selected plate database is no longer available."))
return
previous = os.path.abspath(self._settings.db_path)
if previous not in self._similar_sources:
self._similar_sources.append(previous)
source = os.path.dirname(os.path.dirname(path))
def show_source_hits():
"""Rebuild the new plate's grid before presenting its matches."""
self._rebuild_grid()
self._present_similarity_source(path)
self._open_source(source, preserve_similarity=True,
then=show_source_hits)
def _present_similarity_source(self, path: str) -> None:
"""Pin only this database's result rows in its own annotation grid."""
from ...selection import ObjectRequest
result = self._similar_navigation
if result is None or os.path.abspath(self._settings.db_path) != path:
return
hits = result["hits"]
local = hits.loc[hits["db_path"].eq(path)]
keys = [str(key) for key in local["key"]]
if path == os.path.abspath(result["db_path"]):
keys.insert(0, result["key"])
source = os.path.basename(os.path.dirname(os.path.dirname(path)))
request = ObjectRequest(
keys=keys,
reason=tr("{n} similar crops in {source}; results stay with their source database.",
n=len(keys), source=source),
source="similarity",
context={"similarity": dict(zip(local["key"], local["similarity"])),
"source_db": path,
"query_db": result["db_path"],
"unlabelled_only": result.get("unlabelled_only", False),
"requested_k": result.get("requested_k", 100)})
self.open_object_request(request)
self._status_label.setText(tr(
"Searched {n} crops across {plates} plates in {ms} ms.",
n=f"{len(result['index']):,}", plates=len(result["db_paths"]),
ms=f"{result['seconds'] * 1000:.0f}"))
if result.get("unlabelled_only"):
self._status_label.setText(self._status_label.text() + " " +
tr("{n} unlabelled matches; selected reference kept separately.",
n=len(hits)))
self._similar_notice_until = time.monotonic() + 3.0
@Slot(str)
def _on_similar_failed(self, message: str) -> None:
"""Say why the search could not run."""
self._status_label.setText(
tr("Could not search for similar crops: {msg}").format(
msg=message))
self._similar_notice_until = time.monotonic() + 3.0
@Slot()
def _on_similar_finished(self) -> None:
"""Retire the search thread on the GUI thread."""
worker = self._similar_worker
self._similar_worker = None
self._btn_similar.setEnabled(True)
if worker is None:
return
try:
worker.done.disconnect(self._on_similar_done)
worker.failed.disconnect(self._on_similar_failed)
worker.finished.disconnect(self._on_similar_finished)
except (RuntimeError, TypeError):
pass
_retire(worker)
def _set_blind_checked(self, on: bool) -> None:
"""Move the Blind switch without asking it to act."""
button = getattr(self, "_btn_blind", None)
if button is None:
return
button.blockSignals(True)
button.setChecked(bool(on))
button.blockSignals(False)
def _on_blind_toggled(self, checked: bool) -> None:
"""Start blinding, or ask to unblind; undo the click if refused."""
if checked and self._blind is None:
if not self._start_blind():
self._set_blind_checked(False)
elif not checked and self._blind is not None:
if not self._end_blind():
self._set_blind_checked(True)
def _start_blind(self) -> bool:
"""Shuffle the population on screen under a new blinding key.
:returns: whether blinding started; not without an open source.
"""
if not self._settings.db_path:
QMessageBox.information(
self, tr("Open a source first"),
tr("Open an experiment source before scoring it blind."))
return False
self._flush_pending()
rows = (list(self._filtered_rows) if self._filtered_rows is not None
else fetch_page(self._settings.db_path,
self._settings.annotation_column, 0, -1,
self._settings.image_type,
table=self._settings.png_table))
from ...run_journal import start_blinding
src = self._settings.src or os.path.dirname(
os.path.dirname(self._settings.db_path))
key = start_blinding([row[0] for row in rows], scope="annotate",
src=src)
rank = {item: index for index, item in enumerate(key["order"])}
self._blind = {"key_id": key["key_id"], "rank": rank,
"codes": key["codes"]}
self._filtered_rows = _blind_order(rows, rank)
self._total = len(self._filtered_rows)
self._offset = 0
self._apply_blind_chrome(True)
self._console.append_notice(
"Blinded {count} crops under key {key}.\n",
count=len(rows), key=key["key_id"])
self._load_page()
return True
def _end_blind(self, *, ask=None) -> bool:
"""Unblind, after asking, and record who did it and when.
:param ask: returns whether to go ahead; a Yes/No question when
omitted.
:returns: whether the session was unblinded.
"""
if self._blind is None:
return True
if ask is None:
def ask():
"""Confirm revealing crop origins and recording the unblind event."""
return QMessageBox.question(
self, tr("Unblind?"),
tr("Unblinding shows where every crop is from again, "
"and the run journal records who unblinded and "
"when. An analysis lock on this folder treats any "
"later change as post-hoc. Unblind now?"),
QMessageBox.Yes | QMessageBox.No,
QMessageBox.No) == QMessageBox.Yes
if not ask():
return False
from ...run_journal import unblind
key_id = self._blind["key_id"]
self._flush_pending()
unblind(key_id, reason="annotate")
self._blind = None
self._apply_blind_chrome(False)
self._console.append_notice(
"Unblinded key {key}; the journal recorded who and when.\n",
key=key_id)
self._offset = 0
self._refresh_total(then=self._load_page)
return True
def _leave_blind_unopened(self, reason: str) -> None:
"""End a blinded session without unblinding it, and log that it ended.
:param reason: why it ended, kept in the key's log.
"""
if self._blind is None:
return
from ...run_journal import _close_blinding
_close_blinding(self._blind["key_id"], reason=reason)
self._blind = None
self._apply_blind_chrome(False)
def _blind_text(self, text: str) -> str:
"""``text`` as it may be shown: scrubbed of crop and folder names while blinded.
:param text: a status line or a failure message.
:returns: ``text`` unchanged when not blinded; otherwise every crop
path, file name and stem replaced by its code, and the source,
database and crop folders by the word "Blind".
"""
blind = getattr(self, "_blind", None)
if blind is None:
return str(text or "")
if "lookup" not in blind:
src = self._settings.src or ""
db_path = self._settings.db_path or ""
folders = {src, db_path, os.path.dirname(db_path)}
folders.update(os.path.dirname(str(path))
for path in blind["codes"])
blind["lookup"] = _blind_lookup(blind["codes"], src)
blind["folders"] = sorted(f for f in folders if f)
return _blind_scrub(text, blind["lookup"], blind["folders"])
def _blind_warning(self, title: str, text: str) -> None:
"""A warning box whose text is scrubbed while blinded.
Failures quote their exception, and an exception quotes the file it
failed on.
:param title: the box's title.
:param text: the message.
"""
QMessageBox.warning(self, title, self._blind_text(text))
def _apply_blind_chrome(self, on: bool) -> None:
"""Hide or restore what on this screen says where the crops are from.
:param on: true while blinded.
Train hands the source path to Classify or ML Analyze, and Generate
writes a table named after the source and reports its folder, so both
would carry the source onto a screen while it is hidden here.
"""
self._set_blind_checked(on)
for button in (self._btn_coverage, self._btn_auto,
self._btn_browse_db, self._btn_settings,
self._console_switch, self._ai_switch,
self._btn_train, self._btn_generate):
if on:
button.setProperty("_spacr_blind_was", button.isEnabled())
button.setEnabled(False)
elif button.property("_spacr_blind_was") is not None:
button.setEnabled(bool(button.property("_spacr_blind_was")))
button.setProperty("_spacr_blind_was", None)
if on:
dialog = getattr(self, "_settings_dialog", None)
if dialog is not None:
dialog.reject()
self._console_wrap.setProperty(
"_spacr_blind_visible", not self._console_wrap.isHidden())
self._console_wrap.hide()
self._src_label.setText(tr(
"Blinded: the source, plates, wells, conditions and file "
"names are hidden, and the crops are in a shuffled order."))
elif self._settings.db_path:
self._src_label.setText(
f"{self._settings.src} → {self._settings.db_path}")
else:
self._src_label.setText(
tr("No source selected — click Open source…"))
if not on:
self._console_wrap.setVisible(bool(
self._console_wrap.property("_spacr_blind_visible")))
self._console_wrap.setProperty("_spacr_blind_visible", None)
def _on_open_settings(self):
"""Open the annotation settings dialog, and apply it on OK.
OK OPENS THE SOURCE WHENEVER NONE IS OPEN, not only when the source
field changed. A source can be filled in without being opened -- by
a remembered session, or by the test data before this was fixed --
and comparing old against new then found nothing changed, repainted
an empty screen, and OK appeared to do nothing.
"""
if self._blind is not None:
return
dlg = _SettingsDialog(self._settings, self)
self._settings_dialog = dlg
dlg.destroyed.connect(self._on_settings_dialog_destroyed)
try:
if dlg.exec() != QDialog.Accepted:
return
old_src = self._settings.src
old_col = self._settings.annotation_column
nothing_open = self._worker is None
self._settings = dlg.collect()
self._rebuild_grid()
if self._settings.src and (
nothing_open
or self._settings.src != old_src
or self._settings.annotation_column != old_col):
self._open_source(self._settings.src)
else:
self._refresh_total(then=self._load_page)
finally:
try:
dlg.deleteLater()
except RuntimeError:
self._settings_dialog = None
@Slot(object)
def _on_settings_dialog_destroyed(self, _dialog=None):
"""Release the modal only after Qt destroyed it on the GUI thread."""
self._settings_dialog = None
def _on_next(self):
"""Go to the next page of crops."""
self._flush_pending()
page = self._settings.page_size
if self._offset + page < max(self._total, 1):
self._offset += page
self._load_page()
def _on_prev(self):
"""Go to the previous page."""
self._flush_pending()
page = self._settings.page_size
self._offset = max(0, self._offset - page)
self._load_page()
def _on_skip(self):
"""Leave this page unannotated and move on."""
self._flush_pending()
offset = self._last_annotated_offset()
if offset is None:
self._status_label.setText("No annotated images found.")
return
self._offset = offset
self._load_page()
def _last_annotated_offset(self) -> Optional[int]:
"""Page offset of the last annotated crop, in the population on screen.
``_offset`` indexes whatever :meth:`_load_page` is paging through, and
that is NOT always the table: a threshold filter, the uncertainty
queue and a routed request each replace it with ``_filtered_rows``.
Asking the database for the answer in those states returns an index
into the table and hands it to a list of a different length, so Skip
landed on an empty page -- with no message, and with Next/Prev unable
to get back to the crop the button was pressed to find.
So the row set that is on screen is the one that gets counted, and
only the unfiltered case asks the database.
"""
page = self._settings.page_size
rows = self._filtered_rows
if rows is None:
return find_last_annotated_offset(self._settings.db_path,
self._settings.annotation_column,
page,
self._settings.image_type,table=self._settings.png_table)
last = None
for index, (_path, value) in enumerate(rows):
if value is not None and value != 0:
last = index
if last is None:
return None
return (last // page) * page
def _on_class_counts(self):
"""Show how many objects have been given each class so far.
THE NUMBER THAT SAYS WHEN TO STOP. A classifier needs a balance, not
a total, and the count per class is the only thing that shows it.
"""
rows = class_counts(self._settings.db_path, self._settings.annotation_column, table=self._settings.png_table)
if not rows:
QMessageBox.information(self, "Class counts", "No annotated rows yet.")
return
lines = ["Class Count Color"]
for cls, cnt in rows:
lines.append(f"{cls:>5} {cnt:>7} {label_to_hex(cls, dark=on_dark_theme()) or ''}")
try:
from ...suggest import pending_suggestions
waiting = pending_suggestions(
self._settings.db_path, self._settings.annotation_column,
png_table=self._settings.png_table)
except Exception: # noqa: BLE001
waiting = 0
if waiting:
lines.append("")
lines.append(f"{waiting:,} suggested, not yet kept or thrown "
f"away — not counted above.")
QMessageBox.information(self, "Class counts", "\n".join(lines))
def _label_source(self) -> str:
"""How the crops on screen reached the annotator, for provenance."""
if self._object_request is not None:
return "object_request"
return "queue" if self._settings.queue_by_uncertainty else "manual"
def _refresh_round_state(self) -> None:
"""Re-read the round counter and the curve, and repaint the strip."""
curve = None
if not self._settings.db_path or not os.path.isfile(
self._settings.db_path):
self._round_index = 0
self._stop_verdict = None
self._al_label.hide()
return
try:
from ... import active_learning as al
self._round_index = al.next_round(
self._settings.db_path, self._settings.annotation_column)
curve = al.learning_curve(self._settings.db_path,
self._settings.annotation_column)
self._stop_verdict = al.should_stop(curve) if len(curve) else None
except Exception:
self._round_index = 0
self._stop_verdict = None
curve = None
self._refresh_al_label(curve)
def _refresh_al_label(self, curve=None) -> None:
"""Repaint the one-line active-learning strip."""
parts = [f"Round {self._round_index}"]
result = self._last_round
if result is not None:
parts.append(f"{result.n_labels} labels")
parts.append(f"held-out {result.accuracy:.3f}")
per_class = result.per_class
if per_class:
worst = min(per_class, key=per_class.get)
parts.append(f"worst class {worst} {per_class[worst]:.3f}")
elif curve is not None and len(curve):
last = curve.iloc[-1]
parts.append(f"{int(last['n_labels'])} labels")
accuracy = last["holdout_accuracy"]
if accuracy is not None:
parts.append(f"held-out {float(accuracy):.3f}")
verdict = self._stop_verdict
if verdict is not None:
parts.append(("STOP — " if verdict.stop else "keep going — ")
+ verdict.reason)
elif result is None:
parts.append("no model fitted from here yet — press Retrain to "
"start the curve")
self._al_label.setText(" · ".join(parts))
self._al_label.setVisible(bool(self._settings.db_path))
def _show_report(self, title: str, body: str):
"""Put a text report on screen WITHOUT entering a nested event loop.
``QDialog.exec`` is the obvious way to show one of these and it wedges
the application. ``exec`` runs a nested event loop;
``QCoreApplication.quit`` — which is what closing the last window
does — unwinds only the outermost loop. Close the main window while a
report is up and the window disappears, the process stays alive, and
the GUI thread sits inside this call for ever with nothing left to
click. Reproduced in a child process: the main thread's only frame is
``_on_coverage``, and it never returns.
The nested loop is dangerous a second way. The report is parented to
this screen so that it dies with it, and a nested loop is exactly the
window in which the screen CAN die — leaving Qt to delete the object
whose ``exec`` is still on the stack.
So the report is opened as a window that owns its own end:
``WA_DeleteOnClose`` retires it when the user closes it, the Qt parent
retires it when this screen goes, and no event loop of ours is ever on
the stack. Reports are more useful this way too — a coverage table can
stay open while the annotating continues underneath it.
:param title: window title, and the key one report is remembered
under so a second press rewrites it instead of stacking a copy.
:param body: the pre-formatted, column-aligned text.
:returns: the window, so a caller can assert on what it shows.
"""
open_reports = getattr(self, "_reports", None)
if open_reports is None:
open_reports = {}
self._reports = open_reports
for key, report in list(open_reports.items()):
try:
report.isVisible()
except RuntimeError:
open_reports.pop(key, None)
existing = open_reports.get(title)
if existing is not None:
existing.set_body(body)
existing.show()
existing.raise_()
existing.activateWindow()
return existing
dialog = _TextReportDialog(title, body, self)
dialog.setAttribute(Qt.WA_DeleteOnClose, True)
open_reports[title] = dialog
dialog.show()
dialog.raise_()
dialog.activateWindow()
return dialog
def _on_coverage(self):
"""Show where the annotations actually came from."""
if not self._settings.db_path:
QMessageBox.information(
self, "Open a source first",
"Open an experiment source to see annotation coverage.")
return
self._flush_pending()
try:
from ... import active_learning as al
coverage = al.annotation_coverage(
self._settings.db_path, self._settings.annotation_column)
body = al.format_coverage_summary(coverage)
except Exception as exc:
self._blind_warning("Coverage unavailable",
f"{type(exc).__name__}: {exc}")
return
self._show_report("Annotation coverage", body)
def _on_learning_curve(self):
"""Show held-out accuracy per round and the stopping verdict."""
if not self._settings.db_path:
QMessageBox.information(
self, "Open a source first",
"Open an experiment source to see the learning curve.")
return
try:
from ... import active_learning as al
curve = al.learning_curve(self._settings.db_path,
self._settings.annotation_column)
verdict = al.should_stop(curve)
body = al.format_learning_curve(curve, verdict)
except Exception as exc:
self._blind_warning("Learning curve unavailable",
f"{type(exc).__name__}: {exc}")
return
self._show_report("Active-learning rounds", body)
def _on_retrain(self):
"""Fit a round on the labels so far, then re-rank without leaving."""
if not self._settings.db_path:
QMessageBox.information(
self, "Open a source first",
"Open an experiment source before retraining.")
return
if self._retrain_worker is not None:
self._status_label.setText("A retrain is already running.")
return
self._flush_pending()
if self._worker is not None:
self._worker.stop(wait=True)
self._worker = SaveWorker(self._settings.db_path,
self._settings.annotation_column,
table=self._settings.png_table)
self._worker.start()
self._btn_retrain.setEnabled(False)
self._status_label.setText("Retraining on the labels so far…")
self._console.append_notice(
"Retraining round {round} on the labels so far…\n",
round=str(self._round_index))
worker = _RetrainWorker(
self._settings.db_path, self._settings.annotation_column,
{"round_index": self._round_index,
"measure": self._settings.queue_measure,
"diversity": self._settings.queue_diversity,
"image_type": self._settings.image_type},
parent=self)
worker.done.connect(self._on_retrain_done)
worker.failed.connect(self._on_retrain_failed)
worker.finished.connect(self._on_retrain_finished)
self._retrain_worker = worker
worker.start()
@Slot(object)
def _on_retrain_done(self, result) -> None:
"""A round finished: record it, re-rank, and say what it means."""
self._last_round = result
self._stop_verdict = result.verdict
self._round_index = int(result.round_index) + 1
self._console.append_notice("{report}\n", report=result.summary())
self._refresh_al_label()
self._status_label.setText(
f"Round {result.round_index}: held-out "
f"{result.accuracy:.3f} on {result.report.get('n', 0)} objects.")
if self._object_request is None:
self._offset = 0
self._refresh_total(then=self._load_page)
@Slot(str)
def _on_retrain_failed(self, message: str) -> None:
"""A round could not be fitted — say why rather than going quiet."""
self._console.append_notice("Retrain failed: {msg}\n", msg=message)
self._status_label.setText(f"Retrain failed — {message}")
self._blind_warning(
"Retrain failed",
f"{message}\n\nThe annotations are untouched. The usual causes "
f"are too few labels, only one class annotated so far, or no "
f"measurement tables to build features from.")
@Slot()
def _on_retrain_finished(self) -> None:
"""Retire the retrain thread on the GUI thread."""
worker = self._retrain_worker
self._retrain_worker = None
self._btn_retrain.setEnabled(True)
if worker is None:
return
try:
worker.done.disconnect(self._on_retrain_done)
worker.failed.disconnect(self._on_retrain_failed)
worker.finished.disconnect(self._on_retrain_finished)
except (RuntimeError, TypeError):
pass
_retire(worker)
[docs]
def open_object_request(self, request):
"""Show exactly the crops ``request`` names, in its order.
The one method this screen grows for the whole routing contract. A
UMAP point, a confusion-matrix cell and anything added later all
arrive here as an :class:`~spacr.selection.ObjectRequest`; none of
them imports this module and this module grows no method per caller.
The subset *pins* the grid: pagination, the uncertainty queue and
the threshold filter all defer to it until it is cleared, because a
request that quietly got replaced by the next queue rebuild would
show the user a different population under the same heading. Opening
a source, or :meth:`clear_object_request`, clears it.
:param request: the routed request. ``request.reason`` becomes the
line above the grid — a grid of twelve crops that does not say
why reads as the whole screen.
:returns: this screen, so the caller can raise or focus it.
"""
self._object_request = request
keys = list(request.keys)
if not self._settings.db_path:
self._object_rows = []
self._request_note = request.describe()
self._set_page_label(
"no source is open, so none of them can be shown. Open the "
"experiment first.")
return self
try:
from ... import active_learning as al
rows = al.crops_for_object_keys(
self._settings.db_path, keys,
annotation_column=self._settings.annotation_column,
timelapse=bool(request.timelapse),
image_type=self._settings.image_type)
except Exception as exc:
self._object_request = None
self._object_rows = None
self._request_note = ""
self._set_page_label(
f"Could not open {len(keys)} objects: "
f"{type(exc).__name__}: {exc}")
return self
self._flush_pending()
self._object_rows = rows
if self._blind is not None:
rows = _blind_order(rows, self._blind["rank"])
self._filtered_rows = rows
self._total = len(rows)
self._offset = 0
self._content_stack.setCurrentWidget(self._grid_scroll)
missing = len(keys) - len(rows)
suffix = (f" · {missing} of them are not in this database"
if missing > 0 else "")
self._request_note = request.describe() + suffix
self._load_page()
return self
[docs]
def clear_object_request(self) -> None:
"""Unpin a routed subset and go back to the ordinary population."""
if self._object_request is None:
return
self._object_request = None
self._object_rows = None
self._request_note = ""
self._offset = 0
self._refresh_total(then=self._load_page)
def _synthetic_negatives_needed(self):
"""How many negatives to invent, or ``None`` when both classes exist.
The rule is "the same number of images as is annotated
for the other class", so the count is the number of answers already
made -- read from the column rather than assumed, because the whole
point is that the user has been labelling one class only.
Returns ``None`` whenever two or more classes are present, which is
the ordinary case and the one that needs no lie told about it.
"""
from ..annotate_engine import class_counts
try:
rows = class_counts(self._settings.db_path,
self._settings.annotation_column,
table=self._settings.png_table)
except Exception: # noqa: BLE001
return None
answers = [(cls, count) for cls, count in rows
if cls < SUGGESTION_OFFSET]
if len(answers) != 1:
return None
return int(answers[0][1])
def _on_suggest_menu(self):
"""Build the Suggest menu and drop it under the button."""
menu = self._build_suggest_menu()
if menu is None:
return
menu.exec(self._btn_suggest.mapToGlobal(
self._btn_suggest.rect().bottomLeft()))
def _build_suggest_menu(self):
"""The Suggest menu, or None when there is no source to suggest for.
SEPARATE FROM SHOWING IT, so what the menu offers can be asserted
without a modal. `QMenu.exec` blocks in C++ and does not come back
for a monkeypatch on a Shiboken type; a test that tried hung until
its timeout killed it, which is how this seam got found.
BUILT ON EVERY PRESS rather than once at construction, because the
two verdict entries carry a live count and a menu that said "Keep 0
suggestions" would be worse than no menu. The count is one COUNT(*)
on an indexed integer column, which is cheap enough to pay for on a
button press and not cheap enough to pay for on every repaint.
"""
if not self._settings.db_path:
QMessageBox.information(
self, "Open a source first",
"Open an experiment source before suggesting labels.")
return None
from ...suggest import pending_suggestions
try:
waiting = pending_suggestions(
self._settings.db_path, self._settings.annotation_column,
png_table=self._settings.png_table)
except Exception: # noqa: BLE001
waiting = 0
menu = QMenu(self._btn_suggest)
page = menu.addAction(iconset.icon("classify"),
"Suggest for the images on this page")
page.setToolTip(
"Fit on every label you have made, but write suggestions only "
"onto the crops in front of you. The model is the same either "
"way — this narrows what you have to review, not what it learns."
)
page.triggered.connect(lambda: self._start_suggest(this_page=True))
every = menu.addAction(iconset.icon("chart"),
"Suggest for every unanswered image")
every.setToolTip(
"Write a suggestion onto every crop with no answer yet, most "
"confident first. Nothing you have already annotated is touched."
)
every.triggered.connect(lambda: self._start_suggest(this_page=False))
if waiting:
menu.addSeparator()
if self._suggestions_are_a_ranking:
caveat = menu.addAction(
"Only one class was annotated — review these one by one")
caveat.setEnabled(False)
else:
keep = menu.addAction(f"Keep all {waiting:,} suggestions")
keep.setToolTip(
"Turn every outstanding suggestion into an ordinary "
"annotation, indistinguishable from one you made by hand."
)
keep.triggered.connect(
lambda: self._resolve_suggestions(True))
throw = menu.addAction(f"Throw away all {waiting:,} suggestions")
throw.setToolTip(
"Clear every outstanding suggestion back to unanswered. "
"Your own annotations are not touched."
)
throw.triggered.connect(lambda: self._resolve_suggestions(False))
return menu
def _drain_saves(self) -> None:
"""Queue what is unsaved, wait for the writer to finish, restart it.
For the bulk actions that write the column on a connection of their
own: without the wait, a batch still in the writer's queue could
land after them and undo part of what they did.
"""
self._flush_pending()
if self._worker is None:
return
self._worker.stop(wait=True)
self._worker = SaveWorker(self._settings.db_path,
self._settings.annotation_column,
table=self._settings.png_table)
self._worker.start()
def _start_suggest(self, *, this_page: bool) -> None:
"""Fit a round and write its proposals, off the GUI thread."""
if self._suggest_worker is not None:
self._status_label.setText("A suggestion run is already going.")
return
self._flush_pending()
if self._worker is not None:
self._worker.stop(wait=True)
self._worker = SaveWorker(self._settings.db_path,
self._settings.annotation_column,
table=self._settings.png_table)
self._worker.start()
synthetic = self._synthetic_negatives_needed()
only = None
if this_page:
only = [str(path) for path, _ in self._page_paths]
if not only:
self._status_label.setText(
"There is nothing on this page to suggest for.")
return
self._suggestions_are_a_ranking = bool(synthetic)
self._btn_suggest.setEnabled(False)
self._btn_suggest_cancel.setEnabled(True)
self._btn_suggest_cancel.show()
self._status_label.setText(
"Fitting on the labels so far, then suggesting…")
self._console.append_notice(
"Suggesting labels from a model fitted on the labels so far…\n")
worker = _SuggestWorker(
self._settings.db_path, self._settings.annotation_column,
{"round_index": self._round_index,
"model_type": "gradient_boosting",
"balance": "downsample",
"measure": self._settings.queue_measure,
"diversity": self._settings.queue_diversity,
"image_type": self._settings.image_type,
"synthetic_negatives": synthetic},
png_table=self._settings.png_table,
only_paths=only, parent=self)
worker.done.connect(self._on_suggest_done)
worker.failed.connect(self._on_suggest_failed)
worker.finished.connect(self._on_suggest_finished)
for signal, slot in self._suggest_extra_slots(worker):
signal.connect(slot)
self._suggest_worker = worker
worker.start()
def _suggest_extra_slots(self, worker):
"""The progress, cancel and split signals a suggestion run carries.
:param worker: the run.
:returns: ``(signal, slot)`` pairs; a stand-in worker without the
signals yields none.
"""
pairs = []
for name, slot in (("progress", self._on_suggest_progress),
("cancelled", self._on_suggest_cancelled),
("split_relaxed", self._on_suggest_split_relaxed)):
signal = getattr(worker, name, None)
if signal is not None and hasattr(signal, "connect"):
pairs.append((signal, slot))
return pairs
def _cancel_suggest(self) -> None:
"""Ask the running suggestion run to stop at its next step."""
worker = self._suggest_worker
if worker is None:
return
try:
worker.requestInterruption()
except (RuntimeError, AttributeError):
return
self._btn_suggest_cancel.setEnabled(False)
self._status_label.setText(tr(
"Cancelling the suggestion run after its current step…"))
@Slot(int, int, str)
def _on_suggest_progress(self, step: int, total: int, stage: str) -> None:
"""Say which step of the run is going, as n of N.
:param step: the step that started, counted from 1.
:param total: how many steps the run has.
:param stage: the step's name.
"""
what = {
"clear": tr("clearing the outstanding suggestions"),
"features": tr("reading the measurements"),
"fit": tr("fitting on the labels so far"),
"rank": tr("ranking the proposals"),
"write": tr("writing the suggestions"),
}.get(str(stage), str(stage))
self._status_label.setText(tr(
"Suggest: step {n} of {total} — {what}…",
n=int(step), total=int(total), what=what))
@Slot()
def _on_suggest_cancelled(self) -> None:
"""The run stopped on Cancel: say so, and show what is there now."""
self._console.append_notice(
"Suggest cancelled before writing new suggestions. Earlier "
"suggestions may have been cleared and round scores may have "
"been updated.\n")
self._status_label.setText(tr("Suggest cancelled."))
if self._worker is not None:
self._recount_judgements()
self._refresh_total(then=self._load_page)
@Slot(str)
def _on_suggest_split_relaxed(self, reason: str) -> None:
"""Say that this round's accuracy comes from a random split.
:param reason: the grouped split's refusal.
"""
self._console.append_notice(
"This classifier used a random split because too few laboratory "
"wells have labels. ")
self._console.append_notice("Its accuracy may be overestimated. ")
self._console.append_notice("The suggestions are unaffected. ")
self._console.append_notice(
"Label crops from more wells for validation with independent wells. ({why})\n",
why=self._blind_text(str(reason)))
@Slot(object)
def _on_suggest_done(self, payload) -> None:
"""Suggestions are in the column: say how many, and show them.
:param payload: ``(proposal, written)``, or ``(proposal, written,
rejections)`` where the last is how many rejected suggestions
the round was handed to train on (item 512).
"""
proposal, written = payload[0], payload[1]
rejections = int(payload[2]) if len(payload) > 2 else 0
note = getattr(proposal, "note", "") or ""
if not written:
message = note or "nothing was left to suggest a label for"
self._console.append_notice(
"No suggestions written — {why}.\n", why=message)
self._status_label.setText(f"No suggestions — {message}.")
if self._worker is not None:
self._recount_judgements()
self._refresh_total(then=self._load_page)
return
self._console.append_notice(
"Wrote {n} suggestions, most confident first. They are drawn "
"with a dashed ring: answer the ones it got wrong, then keep or "
"throw away the rest from the Suggest menu.\n",
n=f"{written:,}")
if self._suggestions_are_a_ranking:
self._console.append_notice(
"These came from a model with INVENTED negatives — only one "
"class was annotated, so the other was drawn at random from "
"unlabelled crops. Treat this as a ranking to review, not as "
"answers: accepting them in bulk is not offered.\n")
if rejections:
self._console.append_notice(
"{n} rejected suggestions were handed to this round's fit: a "
"rejection is an example of the other class.\n",
n=f"{rejections:,}")
self._console.append_notice(
"Judge them one at a time: click or press Y to confirm a "
"suggestion, right-click or press N to reject it.\n")
self._status_label.setText(
f"{written:,} suggestions written — dashed rings are proposals, "
f"not answers.")
self._recount_judgements()
self._refresh_total(then=self._load_page)
@Slot(str)
def _on_suggest_failed(self, message: str) -> None:
"""A suggestion run could not fit — say why, and what is missing."""
self._console.append_notice(
"Suggest failed: {msg}\n", msg=message)
self._status_label.setText(f"Suggest failed — {message}")
box = QMessageBox(
QMessageBox.Warning, "Suggest failed",
self._blind_text(
f"{message}\n\nNo new suggestions from this round were saved. "
f"Earlier suggestions may have been cleared and round scores "
f"may have been updated. Your annotations are unchanged. "
f"The usual causes are too few labels, only one "
f"class annotated so far — a classifier needs an example of "
f"both — or no measurement tables to build features from."),
QMessageBox.Ok, self)
box.setObjectName("AnnotateSuggestFailedBox")
box.setAttribute(Qt.WA_DeleteOnClose)
box.open()
@Slot()
def _on_suggest_finished(self) -> None:
"""Retire the suggestion thread on the GUI thread."""
worker = self._suggest_worker
self._suggest_worker = None
try:
self._btn_suggest.setEnabled(True)
self._btn_suggest_cancel.hide()
except RuntimeError:
return
if worker is None:
return
for signal, slot in ((worker.done, self._on_suggest_done),
(worker.failed, self._on_suggest_failed),
(worker.finished, self._on_suggest_finished),
*self._suggest_extra_slots(worker)):
try:
signal.disconnect(slot)
except (RuntimeError, TypeError):
pass
_retire(worker)
def _resolve_suggestions(self, keep: bool) -> None:
"""Accept every outstanding suggestion, or clear them all away.
Confirmed before either, because both are bulk and one of them is a
bulk WRITE of labels a machine chose. `resolve_suggestions` acts only
on offset values, so a human's annotation caught by the same query is
untouched whichever way this goes -- but the annotator should still
get to see the number before it happens.
"""
from ...suggest import pending_suggestions, resolve_suggestions
db_path = self._settings.db_path
column = self._settings.annotation_column
if not db_path:
return
waiting = pending_suggestions(db_path, column,
png_table=self._settings.png_table)
if not waiting:
self._status_label.setText("There are no suggestions waiting.")
return
if keep:
question = (f"Accept all {waiting:,} suggestions as annotations "
f'in column "{column}"?\n\nThey become ordinary '
f"annotations and can no longer be told apart from "
f"the ones you made by hand.")
else:
question = (f"Throw away all {waiting:,} suggestions in column "
f'"{column}"?\n\nThose crops go back to unanswered. '
f"Your own annotations are not touched.")
if QMessageBox.question(
self, "Keep suggestions" if keep else "Throw away suggestions",
question) != QMessageBox.Yes:
return
self._drain_saves()
changed = resolve_suggestions(
db_path, column, keep=keep,
png_table=self._settings.png_table)
if not keep:
self._suggestions_are_a_ranking = False
verb = "accepted" if keep else "thrown away"
self._console.append_notice(
"{n} suggestions {verb}.\n", n=f"{changed:,}", verb=verb)
self._status_label.setText(f"{changed:,} suggestions {verb}.")
self._recount_judgements()
self._refresh_total(then=self._load_page)
def _on_train_cv(self):
"""Save any pending annotations, then hand off to Classify."""
if not self._settings.src:
QMessageBox.information(
self, "Open a source first",
"Open an experiment source before training a classifier.",
)
return
self._flush_pending()
seed = {
"src": self._settings.src,
"annotation_column": self._settings.annotation_column,
"dataset_mode": "annotation",
"generate_training_dataset": True,
"train": True,
"apply_model_to_dataset": True,
}
self.train_requested.emit("classify", seed)
def _on_train_xg(self):
"""Save any pending annotations, then hand off to ML Analyze."""
if not self._settings.src:
QMessageBox.information(
self, "Open a source first",
"Open an experiment source before training an XGBoost model.",
)
return
self._flush_pending()
seed = {
"src": self._settings.src,
"annotation_column": self._settings.annotation_column,
"model_type": "xgboost",
}
self.train_requested.emit("ml_analyze", seed)
def _on_clear_column(self):
"""Clear every annotation in one column, after confirming."""
col = self._settings.annotation_column
answer = QMessageBox.question(
self, "Confirm clear",
f'Clear ALL annotations in column "{col}"?\nThis cannot be undone.',
)
if answer != QMessageBox.Yes:
return
self._pending_updates.clear()
self._pending_verdicts.clear()
self._drain_saves()
clear_column(self._settings.db_path, col, table=self._settings.png_table)
self._recount_judgements()
self._refresh_total(then=self._load_page)
def _on_auto_annotate(self):
"""Label a population chosen by metadata, measurement, gate or UMAP.
The write goes through `_worker` -- the SaveWorker this screen
already owns -- not through a new connection. A second sqlite writer
on measurements.db is a known hazard, and routing bulk writes through
the existing one means they land in the same place, in the same
order, as annotations made by hand.
"""
if not self._settings.db_path:
QMessageBox.information(
self, "Open a source first",
"Open an experiment source before auto-annotating.")
return
dlg = _AutoAnnotateDialog(self._settings, self)
result = dlg.exec()
if result == _AUTO_ANNOTATE_OPEN_GATE:
self._flush_pending()
self.train_requested.emit("gate_editor", {
"src": self._settings.src,
"annotation_column": self._settings.annotation_column,
})
return
if result == _AUTO_ANNOTATE_OPEN_UMAP:
self._flush_pending()
self.train_requested.emit("umap", {
"src": self._settings.src,
"annotation_column": self._settings.annotation_column,
})
return
if result != QDialog.Accepted:
return
paths = dlg.matched_paths()
if not paths:
return
self._apply_bulk_annotation(paths, dlg.value())
def _apply_bulk_annotation(self, paths, value) -> int:
"""Write ``value`` to every path, through the existing save worker.
The column is created first if it is missing -- the same
`ensure_annotation_column` the Annotate screen calls when it opens a
source, so an auto-annotation into a fresh column behaves like a
hand annotation into one.
Slots on the current page are updated in place as well, so the grid
agrees with the database without a reload. That is not cosmetic: a
user who auto-annotates and then labels by hand would otherwise be
looking at stale borders while writing on top of them.
:param paths: png_paths to label.
:param value: the class number, or None to clear.
:returns: how many rows were submitted.
"""
from ..annotate_engine import annotation_batch, ensure_annotation_column
column = self._settings.annotation_column
try:
ensure_annotation_column(self._settings.db_path, column,
table=self._settings.png_table)
except Exception as exc:
self._blind_warning(
"Could not write",
f"The annotation column {column!r} could not be created:\n{exc}")
return 0
batch = annotation_batch(paths, value)
if self._worker is not None:
self._worker.submit(dict(batch))
else:
self._pending_updates.update(batch)
wanted = set(batch)
for slot, (path, _current) in enumerate(self._page_paths):
if path in wanted:
self._page_paths[slot] = (path, value)
self._repaint_slot(slot)
self._set_kbd_hint(
f"Auto-annotated {len(batch):,} object(s) as {value}.")
self._refresh_total()
return len(batch)
def _on_browse_db(self):
"""Show png_list in the Database Browser.
Pending annotations are flushed first. Sending the user to look at
the table while this screen is still holding unwritten labels would
show them a table that disagrees with the grid they just left, and
the obvious conclusion -- "the annotations are not being saved" -- is
wrong and alarming.
"""
if not self._settings.db_path:
QMessageBox.information(
self, "Open a source first",
"Open an experiment source before browsing its database.")
return
self._flush_pending()
self.train_requested.emit("db_browser", {
"db_path": self._settings.db_path,
"table": "png_list",
"column": self._settings.annotation_column,
})
def _ask_class(self, count: int, what: str) -> Optional[int]:
"""Prompt for the class number to give ``count`` images.
``None`` when the user cancels, and cancelling must leave nothing
changed -- so every caller asks BEFORE it writes anything.
:param count: how many images will be labelled, named in the prompt
because "annotate 240 images" and "annotate 2" are different
decisions and the button looks the same for both.
:param what: the selection being labelled, for the prompt text.
:returns: the class number, or None.
"""
value, ok = QInputDialog.getInt(
self, "Annotate", f"Class number for {count} image(s) on {what}:",
1, 0, 999, 1)
return int(value) if ok else None
def _slots_on_page(self) -> List[int]:
"""Every slot on this page that actually holds a crop.
Not ``range(self._slot_count())``: the last page is usually short and
the trailing slots are empty. Labelling those would write annotations
against whatever path was last in them.
"""
return [slot for slot in range(self._slot_count())
if self._slot_is_valid(slot)]
def _apply_to_slots(self, slots: Sequence[int],
value: Optional[int]) -> int:
"""Set ``value`` on ``slots``, undoably. Returns how many changed.
The single write path for every bulk action -- the page buttons and
the rubber-band selection both come here -- so they cannot drift
apart in what they record for undo or in what they leave in
``_pending_updates`` for the save worker.
A slot already holding ``value`` is skipped rather than rewritten,
so undo does not fill up with entries that change nothing.
"""
changed = 0
for slot in slots:
if not self._slot_is_valid(slot):
continue
previous = self._current_value(slot)
if previous == value:
continue
path, _ = self._page_paths[slot]
self._push_undo(slot, path, previous)
if self._set_annotation(slot, value):
changed += 1
return changed
def _on_annotate_page(self):
"""Give every crop on this page the same class."""
slots = self._slots_on_page()
if not slots:
self._set_kbd_hint("Nothing on this page to annotate.")
return
value = self._ask_class(len(slots), "this page")
if value is None:
return
changed = self._apply_to_slots(slots, value)
self._set_kbd_hint(f"Annotated {changed} image(s) as {value}.")
def _on_clear_page(self):
"""Remove the annotations on this page only.
No confirmation, unlike Clear column: this is undoable and bounded to
what the user can see. Clear column is neither.
"""
slots = self._slots_on_page()
changed = self._apply_to_slots(slots, None)
self._set_kbd_hint(f"Cleared {changed} image(s) on this page.")
def _on_generate_annotation_db(self) -> None:
"""Open the streaming generator, and adopt what it makes."""
dialog = _GenerateAnnotationDatabaseDialog(self._settings, self)
if dialog.exec() != QDialog.Accepted:
return
from spacr.annotation_dataset import crops_folder_for
table = dialog.written_table()
if not table:
return
self._settings.png_table = table
try:
self._reload()
except Exception: # noqa: BLE001
LOG.debug("could not open the generated set", exc_info=True)
QMessageBox.information(
self, "Annotation set ready",
f"Wrote {table} and opened it. Its crops are in "
f"{crops_folder_for(table)}/ beside the database.")
def _on_thumb_left(self, slot: int):
"""Assign class 1 to one crop, or confirm it if it is a suggestion.
On a suggested crop a click is a JUDGEMENT, not a label (item 512):
the proposal is confirmed as the class it proposes, whichever that
is. Everywhere else the click labels, as it always did.
:param slot: which grid position was clicked.
"""
if self._is_suggested_slot(slot):
self._judge(slot, confirm=True, advance=False)
return
self._toggle_annotation(slot, 1)
def _on_thumb_right(self, slot: int):
"""Assign class 2 to one crop, or reject it if it is a suggestion.
:param slot: which grid position was clicked.
"""
if self._is_suggested_slot(slot):
self._judge(slot, confirm=False, advance=False)
return
self._toggle_annotation(slot, 2)
def _on_thumb_shift(self, slot: int):
"""Blow ``slot``'s crop up to fill the grid's container.
Does nothing for a cell with no crop in it: there is nothing to
look at, and an empty black overlay a user then has to dismiss is
worse than the gesture appearing not to have fired.
"""
overlay = getattr(self, "_zoom_overlay", None)
if overlay is None:
return
if not (0 <= slot < len(self._thumb_pixmaps)):
return
pixmap = self._thumb_pixmaps[slot]
if pixmap is None or pixmap.isNull():
return
self._fit_zoom_overlay()
overlay.show_pixmap(pixmap, slot)
overlay.setFocus(Qt.OtherFocusReason)
def _fold_zoom_back(self) -> None:
"""Put the zoomed crop back in its place in the grid."""
overlay = getattr(self, "_zoom_overlay", None)
if overlay is None:
return
overlay.hide()
overlay.slot = -1
self.setFocus(Qt.OtherFocusReason)
def _zoom_is_open(self) -> bool:
"""True while one crop is blown up over the grid."""
overlay = getattr(self, "_zoom_overlay", None)
try:
return overlay is not None and overlay.isVisible()
except RuntimeError:
return False
def _fit_zoom_overlay(self) -> None:
"""Keep the overlay exactly the size of the container it fills."""
overlay = getattr(self, "_zoom_overlay", None)
scroll = getattr(self, "_grid_scroll", None)
if overlay is None or scroll is None:
return
try:
overlay.setGeometry(scroll.viewport().rect())
except RuntimeError:
return
def _filter_active(self) -> bool:
"""Whether a filter is narrowing what the grid shows.
:returns: True when filtered.
"""
s = self._settings
return bool(s.measurement and s.threshold and s.threshold_direction)
[docs]
def active_jobs(self) -> int:
"""How many population counts are still winding down."""
return self._total_jobs.active_jobs()
[docs]
def is_busy(self) -> bool:
"""True while a population count has not delivered its result."""
return self._total_jobs.is_busy()
def _refresh_total(self, then=None):
"""Recount the population, off the GUI thread, then run ``then``.
This was the heaviest single GUI-thread call in the application. With
a threshold filter set it reaches ``fetch_filtered_paths``, which
calls :func:`spacr.io._read_and_join_tables` -- every measurement
table in the database joined into one pandas frame -- and it ran on
every source open, every settings apply, every retrain and every
clear. Measured with the event-loop watchdog on a 60 000-object
database: **2.6 s of frozen window**, and a real screen's database is
larger than that.
The counting now happens in :func:`_compute_total`, which is a
module-level function precisely so it cannot reach a widget. What is
left here is the part that must be on the GUI thread: reading the
settings, and painting the result.
:param then: called on the GUI thread once the new totals are in
place. Every caller previously followed ``_refresh_total()`` with
``_load_page()`` on the next line; that is what this is for, and
passing it is how the ordering survives becoming asynchronous.
"""
rank = self._blind["rank"] if self._blind is not None else None
if self._object_rows is not None:
self._filtered_rows = (self._object_rows if rank is None
else _blind_order(self._object_rows, rank))
self._total = len(self._object_rows)
self._queue_summary = ""
if then is not None:
then()
return
settings = deepcopy(self._settings)
filter_active = self._filter_active()
self._total_jobs.cancel()
self._total_jobs.submit(
lambda: _blinded_total(_compute_total(settings, filter_active),
settings, rank),
lambda outcome, _then=then: self._apply_total(outcome, _then))
def _apply_total(self, outcome: dict, then=None) -> None:
"""Install a worker-computed population count. GUI thread only."""
if self._closing:
return
self._filtered_rows = outcome.get("filtered_rows")
self._total = int(outcome.get("total") or 0)
self._queue_summary = outcome.get("queue_summary") or ""
note = outcome.get("note")
if note:
self._page_label.setText(note)
if then is not None:
then()
def _load_page(self):
"""Load the current page, on a worker."""
if self._closing:
return
page = self._settings.page_size
if self._filtered_rows is not None:
self._page_paths = list(self._filtered_rows[self._offset:self._offset + page])
else:
self._page_paths = fetch_page(self._settings.db_path,
self._settings.annotation_column,
self._offset,
page,
self._settings.image_type,table=self._settings.png_table)
self._fetch_page_verdicts()
self._fold_zoom_back()
for i in range(len(self._thumbs)):
self._set_slot_image(i, None)
self._undo_stack.clear()
self._redo_stack.clear()
self._set_kbd_hint("")
self._revalidate_hover()
first = self._next_unannotated(0)
self._set_focus_slot(first if first is not None else 0)
for i in range(len(self._thumbs)):
self._repaint_slot(i)
self._refresh_judge_bar()
self._page_gen += 1
self._set_page_label(f"Loading {len(self._page_paths)} images…")
crop_src = self._crop_source()
settings = deepcopy(self._settings)
request = (self._page_gen, list(self._page_paths), crop_src, settings)
self._queue_page_load(request)
def _queue_page_load(self, request):
"""Run at most one page QThread, retaining only the newest request."""
if self._closing:
return
worker = self._page_worker
if worker is not None:
self._pending_page_load = request
return
self._start_page_worker(request)
def _start_page_worker(self, request):
"""Read one page of crops off the GUI thread.
OFF THE GUI THREAD BECAUSE THE CROPS COME FROM A DATABASE, and a page
read from a sleeping network share would otherwise freeze the window
for as long as the share takes to wake.
:param request: which page to read.
"""
gen, paths, crop_src, settings = request
load_fn = partial(
_load_thumb_image_worker,
src=crop_src,
settings=settings,
)
worker = _PageLoadWorker(gen, paths, load_fn)
worker.setObjectName(f"annotate-page-{gen}")
worker.done.connect(self._on_page_loaded, Qt.QueuedConnection)
worker.finished.connect(
self._on_page_worker_finished,
Qt.QueuedConnection,
)
self._page_worker = worker
worker.start()
@Slot()
def _on_page_worker_finished(self):
"""Retire the QThread on the GUI thread and launch the newest request."""
worker = self._page_worker
self._page_worker = None
if worker is not None:
_retire(worker)
if self._pending_page_load is not None and not self._closing:
request = self._pending_page_load
self._pending_page_load = None
self._start_page_worker(request)
@Slot(int, object)
def _on_page_loaded(self, gen: int, loaded):
"""Show a page that has come back, unless it is stale.
THE GENERATION IS CHECKED. A user who pages quickly has several reads
in flight, and an older one landing last would replace the page they
are now looking at.
:param gen: the generation this read belongs to.
:param loaded: the crops it returned.
"""
if gen != self._page_gen:
return
page = self._settings.page_size
for i, (img, _annotation) in enumerate(loaded):
if i >= len(self._thumbs):
break
self._set_slot_image(i, img)
self._repaint_slot(i)
self._set_page_label(self._page_counter_text())
def _page_counter_text(self) -> str:
"""``Page N of M`` for the page on screen, and which crops it holds.
The page size is whatever fits (item 512), so M moves when the
console opens or the window shrinks; N is the page the first crop
on screen falls on at that size. The crop range is one-based, for
people rather than for the query.
"""
page = max(1, int(self._settings.page_size))
total = max(0, int(self._total))
first = max(0, int(self._offset))
last = min(first + page, total)
number = first // page + 1
pages = max(number, -(-total // page)) if total else 1
return tr("Page {page} of {pages} · crops {first}–{last} of {total}",
page=number, pages=pages,
first=(first + 1) if total else 0, last=last, total=total)
def _set_page_label(self, text: str) -> None:
"""Write the line above the grid, keeping any routed request's reason.
The page counter is rewritten twice per page load, and it used to be
the only writer — so a routed subset's "12 objects · predicted
infected, annotated uninfected" was replaced by "Page rows 0–12 / 12"
before the crops had finished decoding, and twelve crops with a page
counter over them read as the whole screen. The reason is the part
that must not be lost.
"""
if getattr(self, "_blind", None) is not None:
self._page_label.setText(tr("Blinded · {page}", page=text))
return
note = getattr(self, "_request_note", "")
self._page_label.setText(f"{note} — {text}" if note else text)
def _crop_source(self):
"""Resolve the crop source once per settings change, then cache it.
'auto' keeps today's behaviour: the PNG folder is used whenever one
exists, and only a project without one falls back to cutting crops out
of merged/*.npy. See spacr.crops.resolve_crop_source.
"""
s = self._settings
key = (s.db_path, getattr(s, "crop_source", "auto"), s.image_type)
if getattr(self, "_cropsrc_key", None) != key:
from ...crops import resolve_crop_source
root = os.path.dirname(os.path.dirname(s.db_path or ""))
obj = (s.image_type or "cell_png").replace("_png", "") or "cell"
try:
self._cropsrc = resolve_crop_source(
{"src": root, "crop_source": getattr(s, "crop_source", "auto")},
object_type=obj)
except Exception:
self._cropsrc = None
self._cropsrc_key = key
return self._cropsrc
def _load_thumb_image(self, row, src=None, settings=None):
"""Compatibility wrapper for direct callers and older tests.
Page workers call :func:`_load_thumb_image_worker` with snapshots and
never retain this bound QWidget method.
"""
if src is None:
src = self._crop_source()
if settings is None:
settings = deepcopy(self._settings)
return _load_thumb_image_worker(row, src, settings)
def _image_to_pixmap(self, img: Image.Image) -> QPixmap:
"""Convert one decoded crop to a bare pixmap.
No border and no corner rounding are baked in — both are painted by
:class:`_Thumbnail`, so changing either never costs a conversion.
"""
qimg = ImageQt(img.convert("RGB"))
return QPixmap.fromImage(QImage(qimg))
def _slot_is_valid(self, slot: int) -> bool:
"""True when ``slot`` addresses a crop present on the current page."""
return 0 <= slot < self._slot_count()
def _slot_count(self) -> int:
"""Number of keyboard-navigable crops on this page."""
return min(len(self._page_paths), len(self._thumbs))
def _current_value(self, slot: int) -> Optional[int]:
"""The label ``slot`` carries right now, pending writes included."""
if not (0 <= slot < len(self._page_paths)):
return None
path, current = self._page_paths[slot]
if path in self._pending_updates:
return self._pending_updates[path]
return current
def _is_annotated(self, slot: int) -> bool:
"""True when ``slot`` already carries a non-zero class label.
A SUGGESTION IS NOT AN ANSWER, and this is where that matters most.
Both callers walk to the next crop the annotator has not decided,
and "Suggest for every unanswered image" gives every crop a value --
so counting proposals as answers would leave the keyboard flow with
nowhere to go on a page it had just filled with things to review.
The dashed ring says "look at me"; this is what lets Tab get there.
"""
value = self._current_value(slot)
if value is None or value == 0:
return False
return not self._is_suggested_slot(slot)
def _set_annotation(self, slot: int, value: Optional[int]) -> bool:
"""Record ``value`` (or ``None`` to clear) as ``slot``'s label.
A judgement the new label contradicts is withdrawn with it
(:func:`verdict_contradicts`), and the judgement counts move by
whatever this change moved them by -- so every path that labels a
crop, the keyboard, the mouse, the page buttons and undo, keeps the
counts true without counting anything itself.
"""
if not (0 <= slot < len(self._page_paths)):
return False
path, _ = self._page_paths[slot]
before = self._judge_contribution(slot)
self._pending_updates[path] = value
self._page_paths[slot] = (path, value)
if verdict_contradicts(self._verdict_value(slot), value):
self._store_verdict(path, None)
self._account_judgement(before, self._judge_contribution(slot))
self._repaint_slot(slot)
return True
def _set_slot_image(self, slot: int, img: Optional[Image.Image]) -> None:
"""Install (or clear) the decoded crop for ``slot``.
This is the ONLY place a pixmap is built. Border and ring changes go
through :meth:`_repaint_slot`, which never touches pixels.
"""
if not (0 <= slot < len(self._thumbs)):
return
self._raw_thumb_images[slot] = img
if img is None:
self._thumb_pixmaps[slot] = None
self._thumbs[slot].setPixmap(QPixmap())
return
pm = self._image_to_pixmap(img)
self._thumb_pixmaps[slot] = pm
self._thumbs[slot].setPixmap(pm)
def _border_color_for(self, slot: int) -> str:
"""The state-ring colour for ``slot``: its class colour, else gray.
``label_to_hex`` is the app's one class→colour map (the Class counts
dialog reads the same function), so the border can never disagree
with the colour shown anywhere else.
THE THEME IS PASSED THROUGH. The palette is tuned against a dark
tile; on a light one the same colours measured 1.28-4.34 contrast,
five of the first six below the readability floor. That is issue #6 --
reported as a macOS problem, and really a light-theme one, since
macOS defaults to the light appearance far more often than Linux.
A SUGGESTION RESOLVES TO ITS OWN CLASS'S COLOUR. Suggestions are
stored offset (a suggested 1 is an 11 -- see
:data:`spacr.suggest.SUGGESTION_OFFSET`), and handing that raw to
``label_to_hex`` would either fall off the end of the palette or,
worse, land on the colour of a class the model never proposed. The
offset is removed here; what distinguishes a suggestion is the
dashed ring :meth:`_is_suggested_slot` turns on, not the colour.
"""
return (label_to_hex(self._displayed_class(slot), dark=on_dark_theme())
or resting_border_color())
def _displayed_class(self, slot: int):
"""``slot``'s class as the palette knows it, suggestion or answer."""
value = self._current_value(slot)
if value is None:
return None
try:
value = int(value)
except (TypeError, ValueError):
return value
if value > SUGGESTION_OFFSET:
return value - SUGGESTION_OFFSET
return value
def _is_suggested_slot(self, slot: int) -> bool:
"""True when ``slot`` carries a proposal rather than an answer."""
value = self._current_value(slot)
try:
return value is not None and int(value) > SUGGESTION_OFFSET
except (TypeError, ValueError):
return False
def _repaint_slot(self, slot: int) -> None:
"""Sync one tile's chrome with the model. Cheap: no pixmap work.
Sets the state ring (class colour or resting gray) and whether this
is the current tile. Both setters no-op when nothing changed, so a
redundant call costs nothing and a real change costs one
``update()`` on one widget.
"""
if not (0 <= slot < len(self._thumbs)):
return
thumb = self._thumbs[slot]
thumb.set_occupied(slot < len(self._page_paths))
thumb.set_border_color(self._border_color_for(slot))
thumb.set_suggested(self._is_suggested_slot(slot))
state = self._judgement_of(slot)
thumb.set_badge(state, badge_colors(state) if state else None)
thumb.set_current(slot == self._focus_slot)
def _toggle_annotation(self, slot: int, new_value: int):
"""Mouse semantics: same class again clears, otherwise assign."""
if not self._slot_is_valid(slot):
return
existing = self._current_value(slot)
resolved = None if existing == new_value else new_value
if self._set_annotation(slot, resolved):
path = str(self._page_paths[slot][0])
path = path.replace("\r", r"\r").replace("\n", r"\n")
self._console.append_stdout(
f"path={path} | annotation={resolved}\n"
)
def _verdict_of_path(self, path: str) -> Optional[int]:
"""The judgement recorded for ``path``: this session's, else stored.
:param path: the crop's ``png_path``.
:returns: ``+c``, ``-c`` or None.
"""
if path in self._local_verdicts:
return self._local_verdicts[path]
return self._page_verdicts.get(path)
def _verdict_value(self, slot: int) -> Optional[int]:
"""``slot``'s judgement: ``+c`` confirmed, ``-c`` rejected, or None.
:param slot: the grid position asked about.
"""
if not (0 <= slot < len(self._page_paths)):
return None
return self._verdict_of_path(self._page_paths[slot][0])
def _store_verdict(self, path: str, verdict: Optional[int]) -> None:
"""Remember a judgement and queue it for the verdict column.
Kept in ``_local_verdicts`` for the rest of the session as well as
in ``_pending_verdicts`` for the writer: the writer is asynchronous,
so a page read back from the database straight after a flush could
otherwise show the judgement the user just made as not made yet.
:param path: the crop's ``png_path``.
:param verdict: ``+c``, ``-c``, or None to withdraw a judgement.
"""
verdict = None if verdict is None else int(verdict)
self._local_verdicts[path] = verdict
self._pending_verdicts[path] = verdict
def _set_verdict(self, slot: int, verdict: Optional[int]) -> bool:
"""Record ``slot``'s judgement, keeping the counts and the badge true.
:param slot: the grid position judged.
:param verdict: ``+c``, ``-c``, or None to withdraw a judgement.
:returns: True when the judgement changed.
"""
if not (0 <= slot < len(self._page_paths)):
return False
if self._verdict_value(slot) == verdict:
return False
before = self._judge_contribution(slot)
self._store_verdict(self._page_paths[slot][0], verdict)
self._account_judgement(before, self._judge_contribution(slot))
self._repaint_slot(slot)
return True
def _judgement_of(self, slot: int) -> Optional[str]:
"""Which badge ``slot`` wears: suggested, confirmed, rejected or none.
A proposal on the crop wins -- a crop rejected in an earlier round
and proposed again is something to judge again. A recorded
judgement shows only while the label still agrees with it.
:param slot: the grid position asked about.
:returns: one of :data:`JUDGEMENT_STATES`, or None.
"""
if not self._slot_is_valid(slot):
return None
if self._is_suggested_slot(slot):
return "suggested"
verdict = self._verdict_value(slot)
if verdict is None:
return None
value = self._current_value(slot)
if verdict > 0 and value == verdict:
return "confirmed"
if verdict < 0 and value is None:
return "rejected"
return None
def _judge_contribution(self, slot: int) -> Tuple[int, int, int]:
"""``slot``'s share of the column's (left, confirmed, rejected).
:param slot: the grid position asked about.
:returns: three 0-or-1 counts.
"""
if not (0 <= slot < len(self._page_paths)):
return (0, 0, 0)
verdict = self._verdict_value(slot)
return (int(self._is_suggested_slot(slot)),
int(verdict is not None and verdict > 0),
int(verdict is not None and verdict < 0))
def _account_judgement(self, before: Tuple[int, int, int],
after: Tuple[int, int, int]) -> None:
"""Move the column's judgement counts by one crop's change.
:param before: the crop's :meth:`_judge_contribution` before it.
:param after: the same, after it.
"""
if before == after:
return
totals = self._judge_totals
for key, old, new in zip(("left", "confirmed", "rejected"),
before, after):
totals[key] = max(0, int(totals.get(key, 0)) + new - old)
self._refresh_judge_bar()
def _page_judgements(self) -> Dict[str, int]:
"""What this page holds: suggested, confirmed, rejected and left.
``suggested`` is every crop on the page Suggest put a proposal on
that is still proposed or has been judged; ``left`` is the part of
it nobody has judged yet.
"""
counts = {"suggested": 0, "confirmed": 0, "rejected": 0, "left": 0}
for slot in range(self._slot_count()):
state = self._judgement_of(slot)
if state is None:
continue
counts["left" if state == "suggested" else state] += 1
counts["suggested"] += 1
return counts
def _recount_judgements(self) -> None:
"""Read the column's judgement counts back from the database.
Called where the database was just changed behind the screen -- a
source opened, a Suggest run written, suggestions kept or thrown in
bulk, the column cleared -- and the counts kept by
:meth:`_account_judgement` would otherwise be counting from a
column that is no longer there.
"""
self._local_verdicts.clear()
self._judge_totals = {"left": 0, "confirmed": 0, "rejected": 0}
db_path = self._settings.db_path
if db_path and os.path.isfile(db_path):
try:
from ...suggest import judgement_counts
self._judge_totals.update(judgement_counts(
db_path, self._settings.annotation_column,
png_table=self._settings.png_table))
except Exception:
LOG.debug("The judgement counts could not be read",
exc_info=True)
self._refresh_judge_bar()
def _fetch_page_verdicts(self) -> None:
"""Read the stored judgements of the crops on this page."""
self._page_verdicts = {}
db_path = self._settings.db_path
if not db_path or not self._page_paths or not os.path.isfile(db_path):
return
try:
from ...suggest import fetch_verdicts
self._page_verdicts = fetch_verdicts(
db_path, self._settings.annotation_column,
[str(path) for path, _ in self._page_paths],
png_table=self._settings.png_table)
except Exception:
LOG.debug("The page's judgements could not be read",
exc_info=True)
def _ensure_verdict_column(self) -> bool:
"""Give the crop table its verdict column, once per source and column.
This is the migration: a table made before item 512 has no
``<column>_verdict`` and gains one the first time it is opened here.
"""
db_path = self._settings.db_path
key = (str(db_path), str(self._settings.annotation_column),
str(self._settings.png_table))
if self._verdict_ready == key:
return True
if not db_path or not os.path.isfile(db_path):
return False
try:
from ...suggest import ensure_verdict_column
ready = ensure_verdict_column(
db_path, self._settings.annotation_column,
png_table=self._settings.png_table)
except Exception:
LOG.debug("The verdict column could not be added", exc_info=True)
ready = False
if ready:
self._verdict_ready = key
return ready
def _refresh_judge_bar(self) -> None:
"""Rewrite the judgement counts, and show the bar when it has any."""
bar = getattr(self, "_judge_bar", None)
if bar is None:
return
page = self._page_judgements()
totals = self._judge_totals
any_here = any(page.values()) or any(totals.values())
self._judge_label.setText(
tr("This page: {suggested} suggested · {confirmed} confirmed · "
"{rejected} rejected · {left} left. Whole column: {tleft} "
"left to judge · {tconfirmed} confirmed · {trejected} "
"rejected.",
suggested=page["suggested"], confirmed=page["confirmed"],
rejected=page["rejected"], left=page["left"],
tleft=f"{totals.get('left', 0):,}",
tconfirmed=f"{totals.get('confirmed', 0):,}",
trejected=f"{totals.get('rejected', 0):,}"))
self._btn_confirm_page.setEnabled(page["left"] > 0)
self._btn_reject_page.setEnabled(page["left"] > 0)
if bar.isVisibleTo(self) != any_here:
bar.setVisible(any_here)
def _judge(self, slot: int, confirm: bool, *, advance: bool = True,
quiet: bool = False) -> bool:
"""Confirm or reject the suggestion on ``slot`` (item 512).
CONFIRMING turns the proposal into an ordinary label of the class it
proposed; REJECTING clears it. Both are written down beside the
column as ``+c`` or ``-c`` so the judgement survives a restart and
reaches the next round -- where a rejection of class 1 in a two-class
column is fitted as an example of class 2.
:param slot: the crop to judge.
:param confirm: True to confirm, False to reject.
:param advance: move the focus on to the next suggestion.
:param quiet: leave the keyboard hint alone (the page buttons say
one thing for the whole page instead).
:returns: True when a suggestion was judged.
"""
if not self._slot_is_valid(slot):
if not quiet:
self._set_kbd_hint(tr("Nothing to judge — open a source first."))
return False
if self._suggest_worker is not None:
if not quiet:
self._set_kbd_hint(tr(
"A suggestion run is going; judge its suggestions when it "
"has finished."))
return False
if not self._is_suggested_slot(slot):
if not quiet:
self._set_kbd_hint(tr(
"This crop is not a suggestion; a number key labels it."))
return False
cls = self._displayed_class(slot)
path = str(self._page_paths[slot][0])
self._push_undo(slot, path, self._current_value(slot))
self._set_annotation(slot, cls if confirm else None)
self._set_verdict(slot, cls if confirm else -cls)
self._refresh_judge_bar()
shown = path.replace("\r", r"\r").replace("\n", r"\n")
self._console.append_stdout(
f"path={shown} | {'confirmed' if confirm else 'rejected'}={cls}\n")
if quiet:
return True
if confirm:
hint = tr("Confirmed: class {cls}.", cls=cls)
else:
hint = tr("Rejected: not class {cls}.", cls=cls)
if advance:
nxt = self._next_suggested(slot + 1)
if nxt is None:
nxt = self._next_suggested(0)
if nxt is not None:
self._set_focus_slot(nxt)
else:
hint = tr("{done} Every suggestion on this page is judged; "
"Enter loads the next page.", done=hint)
self._set_kbd_hint(hint)
return True
def _next_suggested(self, start: int) -> Optional[int]:
"""First slot at or after ``start`` still waiting to be judged.
:param start: the first grid position to look at.
:returns: the slot, or None when none is left on the page.
"""
for i in range(max(0, start), self._slot_count()):
if self._is_suggested_slot(i):
return i
return None
def _kbd_judge(self, *, confirm: bool) -> bool:
"""Y / N: judge the focused crop and move on to the next suggestion.
:param confirm: True for Y (confirm), False for N (reject).
:returns: True, always -- the key is bound whether or not the
focused crop was a suggestion, and the hint says which.
"""
self._judge(self._focus_slot, confirm)
return True
def _judge_page(self, *, confirm: bool) -> int:
"""Confirm or reject every suggestion left on this page.
:param confirm: True to confirm them, False to reject them.
:returns: how many were judged. Each is undoable on its own with U.
"""
judged = 0
for slot in range(self._slot_count()):
if self._is_suggested_slot(slot) and self._judge(
slot, confirm, advance=False, quiet=True):
judged += 1
if confirm:
self._set_kbd_hint(tr(
"Confirmed {n} suggestions on this page.", n=judged))
else:
self._set_kbd_hint(tr(
"Rejected {n} suggestions on this page.", n=judged))
return judged
def _refresh_focus_marks(self) -> None:
"""Re-apply the current-tile marker across every thumbnail."""
for i in range(len(self._thumbs)):
self._repaint_slot(i)
def _set_focus_slot(self, slot: int, ensure_visible: bool = True) -> None:
"""Move the current tile to ``slot``, repainting the old and new cells.
``ensure_visible`` is False on the hover path: the tile the cursor is
inside is visible by definition, and scrolling under the cursor could
drag a fresh tile under it and set off a feedback loop.
"""
if not self._thumbs:
self._focus_slot = max(0, slot)
return
slot = max(0, min(int(slot), len(self._thumbs) - 1))
previous = self._focus_slot
self._focus_slot = slot
if self._hover_slot is not None and self._hover_slot != slot:
self._hover_slot = None
if previous != slot and 0 <= previous < len(self._thumbs):
self._repaint_slot(previous)
self._repaint_slot(slot)
if ensure_visible:
try:
self._grid_scroll.ensureWidgetVisible(self._thumbs[slot])
except Exception:
pass
def _on_thumb_hover(self, slot: int, entered: bool) -> None:
"""Handle a tile's Enter/Leave. Runs once per boundary crossed."""
if entered:
self._set_hover_slot(slot)
elif self._hover_slot == slot:
self._set_hover_slot(None)
def _set_hover_slot(self, slot: Optional[int]) -> None:
"""Record which tile the cursor is inside and follow it.
Entering a tile makes it the current tile, so the white ring is
always on the crop the next click or keystroke will hit. Leaving
only forgets the cursor position: the ring stays where it is,
because the keyboard still targets that crop.
"""
if slot is not None:
slot = int(slot)
if not (0 <= slot < self._slot_count()):
slot = None
if slot == self._hover_slot:
return
self._hover_slot = slot
if slot is not None:
self._set_focus_slot(slot, ensure_visible=False)
def _revalidate_hover(self) -> None:
"""Drop a hover the cursor is no longer actually inside.
Called after a page load: the widgets stay put but the crops under
them change, and Qt only re-sends Enter/Leave when the cursor
crosses a boundary. ``underMouse`` is the authority on where the
cursor really is.
"""
slot = self._hover_slot
if slot is None:
return
if not (0 <= slot < self._slot_count()) \
or not self._thumbs[slot].underMouse():
self._hover_slot = None
@property
[docs]
def focus_slot(self) -> int:
"""Index of the crop the keyboard currently acts on."""
return self._focus_slot
@property
[docs]
def hover_slot(self) -> Optional[int]:
"""Index of the tile the cursor is inside, or ``None``."""
return self._hover_slot
@property
[docs]
def current_slot(self) -> int:
"""The one tile wearing the white ring — mouse and keyboard agree."""
return self._focus_slot
def _next_unannotated(self, start: int) -> Optional[int]:
"""First unlabelled slot at or after ``start``; ``None`` if there is none."""
for i in range(max(0, start), self._slot_count()):
if not self._is_annotated(i):
return i
return None
def _push_undo(self, slot: int, path: str, previous: Optional[int]) -> None:
"""Record one annotation so it can be taken back.
The crop's judgement is recorded with its label, because a change
of label can withdraw a judgement and a judgement always changes
the label: undo has to put both back to be an undo.
:param slot: which grid position changed.
:param path: the object that was annotated.
:param previous: what it was before.
"""
self._undo_stack.append(
(slot, path, previous, self._verdict_of_path(path)))
self._redo_stack.clear()
[docs]
def handle_key(self, key, text: str = "") -> bool:
"""Run the annotate keybinding for ``key``.
``key`` may be a Qt key code, a Qt key name (``"Left"``) or a literal
character (``"1"``, ``"h"``). Returns True when the key is bound —
unbound keys return False and are left for Qt's default handling.
This is the single entry point for the whole keyboard feature so it
can be driven directly, without synthesising key events.
:param key: Qt key code, Qt key name or literal character, normalised
with :func:`key_token`.
"""
token = key_token(key, text)
if token is None:
return False
if token.isdigit():
value = int(token)
return self._kbd_clear() if value == 0 else self._kbd_assign(value)
if token in ("left", "right", "up", "down"):
return self._kbd_move(token)
if token == "space":
return self._kbd_step(+1)
if token == "backspace":
return self._kbd_step(-1)
if token == "undo":
return self._kbd_undo()
if token == "enter":
return self._kbd_commit_page()
if token == "confirm":
return self._kbd_judge(confirm=True)
if token == "reject":
return self._kbd_judge(confirm=False)
if token == "help":
return self._toggle_legend()
if token == "escape":
if self._legend_expanded:
return self._toggle_legend()
return False
return False
def _kbd_assign(self, value: int) -> bool:
"""Label the focused crop with ``value`` and advance."""
slot = self._focus_slot
if not self._slot_is_valid(slot):
self._set_kbd_hint("Nothing to annotate — open a source first.")
return True
path = self._page_paths[slot][0]
previous = self._current_value(slot)
self._set_annotation(slot, value)
self._push_undo(slot, path, previous)
self._advance_after_assign()
return True
def _kbd_clear(self) -> bool:
"""Clear the focused crop's label, staying put so it can be re-keyed."""
slot = self._focus_slot
if not self._slot_is_valid(slot):
self._set_kbd_hint("Nothing to annotate — open a source first.")
return True
path = self._page_paths[slot][0]
previous = self._current_value(slot)
self._set_annotation(slot, None)
self._push_undo(slot, path, previous)
self._set_kbd_hint("Cleared.")
return True
def _advance_after_assign(self) -> bool:
"""Jump to the next unlabelled crop; never wrap without saying so."""
nxt = self._next_unannotated(self._focus_slot + 1)
if nxt is not None:
self._set_focus_slot(nxt)
self._set_kbd_hint("")
return True
behind = sum(1 for i in range(self._focus_slot)
if not self._is_annotated(i))
if behind:
self._set_kbd_hint(
f"End of page — {behind} unlabelled crop(s) above. "
"Press Enter for the next batch."
)
else:
self._set_kbd_hint(
"End of page — all crops labelled. "
"Press Enter to load the next batch."
)
return False
def _kbd_move(self, token: str) -> bool:
"""Move focus one cell in ``token``'s direction, clamped to the grid."""
count = self._slot_count()
if count <= 0:
self._set_kbd_hint("Nothing to annotate — open a source first.")
return True
cols = max(1, int(self._settings.grid_cols))
slot = self._focus_slot
target = slot
if token == "left":
if slot % cols > 0:
target = slot - 1
elif token == "right":
if slot % cols < cols - 1 and slot + 1 < count:
target = slot + 1
elif token == "up":
if slot - cols >= 0:
target = slot - cols
elif token == "down":
if slot + cols < count:
target = slot + cols
if target != slot:
self._set_focus_slot(target)
self._set_kbd_hint("")
return True
def _kbd_step(self, delta: int) -> bool:
"""Step focus by ``delta`` in reading order without touching labels."""
count = self._slot_count()
if count <= 0:
self._set_kbd_hint("Nothing to annotate — open a source first.")
return True
target = self._focus_slot + delta
if target < 0:
self._set_kbd_hint("Start of page.")
return True
if target >= count:
self._set_kbd_hint(
"End of page — press Enter to load the next batch.")
return True
self._set_focus_slot(target)
self._set_kbd_hint("")
return True
def _kbd_undo(self) -> bool:
"""Walk back the most recent keyboard label assignment."""
while self._undo_stack:
entry = self._undo_stack.pop()
slot, path, previous = entry[:3]
if slot < len(self._page_paths) and self._page_paths[slot][0] == path:
self._redo_stack.append((slot, path, self._current_value(slot),
self._verdict_of_path(path)))
self._set_annotation(slot, previous)
if len(entry) > 3:
self._set_verdict(slot, entry[3])
self._refresh_judge_bar()
self._set_focus_slot(slot)
self._set_kbd_hint("Undone.")
return True
self._set_kbd_hint("Nothing to undo.")
return True
def _kbd_redo(self) -> bool:
"""Put back the label and judgement the last undo took away."""
while self._redo_stack:
slot, path, value, verdict = self._redo_stack.pop()
if slot < len(self._page_paths) and self._page_paths[slot][0] == path:
self._undo_stack.append((slot, path, self._current_value(slot),
self._verdict_of_path(path)))
self._set_annotation(slot, value)
self._set_verdict(slot, verdict)
self._refresh_judge_bar()
self._set_focus_slot(slot)
self._set_kbd_hint(tr("Redone."))
return True
self._set_kbd_hint(tr("Nothing to redo."))
return True
def _kbd_commit_page(self) -> bool:
"""Save this page and load the next batch — same as the Next button."""
before = self._offset
self._on_next()
if self._offset == before:
self._set_kbd_hint("Saved — this is the last page.")
return True
[docs]
def keyPressEvent(self, event):
"""Route keystrokes through :meth:`handle_key` before Qt's default.
:param event: the key event; its key code and text are read, and it is
accepted when the key is bound.
"""
from ..shortcuts import _screen_event_key
key = _screen_event_key(self, event)
if key is not None and self.handle_key(key, event.text() if key == event.key() else ""):
event.accept()
return
super().keyPressEvent(event)
[docs]
def eventFilter(self, obj, event): # noqa: N802 (Qt naming)
"""Catch keys landing on the scroll area so arrows don't just scroll.
Also catches the cursor leaving the grid as a whole. A tile's own
Leave normally clears the hover, but the cursor can quit the grid
without one (window hidden, cursor warped), and a hover nobody is
pointing at any more must not survive.
:param obj: the watched object: the grid scroll area, its viewport or
the grid holder. Resizes count only on the viewport, and mouse
presses, moves and releases drive the selection band only on the
grid holder.
:param event: the event; key presses go to :meth:`handle_key`, and
``Leave`` clears the hover.
"""
if getattr(self, "_closing", False):
return False
try:
etype = event.type()
except Exception:
return False
if etype == QEvent.KeyPress:
from ..shortcuts import _screen_event_key
key = _screen_event_key(self, event)
if key == Qt.Key_Escape and self._zoom_is_open():
self._fold_zoom_back()
return True
if key is not None and self.handle_key(key, event.text() if key == event.key() else ""):
return True
if etype == QEvent.Leave:
self._set_hover_slot(None)
if etype == QEvent.Resize:
scroll = getattr(self, "_grid_scroll", None)
if scroll is not None and obj is scroll:
self._refit_grid()
zooming = self._zoom_is_open() and scroll is not None
if zooming and obj is scroll.viewport():
self._fit_zoom_overlay()
grid_holder = getattr(self, "_grid_holder", None)
if obj is grid_holder and grid_holder is not None and self._band_event(
etype, event
):
return True
return super().eventFilter(obj, event)
def _detach_event_filters(self) -> None:
"""Stop observed grid widgets from calling back during teardown."""
observed = []
scroll = getattr(self, "_grid_scroll", None)
if scroll is not None:
observed.append(scroll)
try:
observed.append(scroll.viewport())
except RuntimeError:
pass
holder = getattr(self, "_grid_holder", None)
if holder is not None:
observed.append(holder)
for widget in observed:
try:
widget.removeEventFilter(self)
except RuntimeError:
pass
def _band_event(self, etype, event) -> bool:
"""Drive the selection band. True when the event was consumed."""
if etype == QEvent.MouseButtonPress:
if event.button() != Qt.LeftButton:
return False
self._band_origin = event.position().toPoint()
if self._band is None:
self._band = QRubberBand(QRubberBand.Rectangle,
self._grid_holder)
self._band.setGeometry(QRect(self._band_origin, QSize()))
self._band.show()
return True
if etype == QEvent.MouseMove and self._band_origin is not None:
rect = QRect(self._band_origin,
event.position().toPoint()).normalized()
self._band.setGeometry(rect)
return True
if etype == QEvent.MouseButtonRelease and self._band_origin is not None:
rect = QRect(self._band_origin,
event.position().toPoint()).normalized()
self._band_origin = None
if self._band is not None:
self._band.hide()
self._apply_band(rect)
return True
return False
def _slots_in_rect(self, rect) -> List[int]:
"""Slots whose tile intersects ``rect``, in reading order.
Intersection, not containment: a band dragged across a row is meant
to take the tiles it crosses, and requiring a tile to be wholly
inside would make the gesture miss the ones at either end -- which
reads as the selection silently dropping images.
"""
hits = []
for slot, thumb in enumerate(self._thumbs):
if not self._slot_is_valid(slot):
continue
if rect.intersects(thumb.geometry()):
hits.append(slot)
return hits
def _apply_band(self, rect) -> None:
"""Label everything the band touched, after asking what to call it."""
if rect.width() < 4 and rect.height() < 4:
return
slots = self._slots_in_rect(rect)
if not slots:
self._set_kbd_hint("Selection was empty.")
return
value = self._ask_class(len(slots), "the selection")
if value is None:
return
changed = self._apply_to_slots(slots, value)
self._set_kbd_hint(f"Annotated {changed} selected image(s) as {value}.")
def _flush_pending(self):
"""Write the queued annotations to the database.
BATCHED RATHER THAN PER CLICK: annotation is fast and the database is
not, so a write per keystroke would make the grid stutter under
exactly the rhythm it is designed for.
"""
if self._worker is None:
return
if self._pending_verdicts and self._ensure_verdict_column():
from ...suggest import verdict_column
self._worker.submit(
dict(self._pending_verdicts),
column=verdict_column(self._settings.annotation_column))
self._pending_verdicts.clear()
if not self._pending_updates:
return
batch = dict(self._pending_updates)
self._worker.submit(self._pending_updates)
self._pending_updates.clear()
if self._blind is not None and self._filtered_rows is not None:
self._filtered_rows = [
(path, batch[path]) if path in batch else (path, value)
for path, value in self._filtered_rows]
self._record_round_provenance(batch)
def _record_round_provenance(self, batch: Dict[str, Optional[int]]) -> None:
"""Stamp a saved batch with the round it was made in.
Without this, "which labels came from before the model had seen
anything" is unanswerable, and an early-round bias — the first
hundred labels drawn from raw page order, all from plate 1 — stays
invisible for the life of the dataset.
Never fatal: a provenance write that fails must not cost the labels
themselves, which are on their way to the database on another
thread regardless.
"""
if not batch or not self._settings.db_path:
return
try:
from ... import active_learning as al
al.record_labels(self._settings.db_path,
self._settings.annotation_column, batch,
round_index=self._round_index,
source=self._label_source())
except Exception as exc:
self._console.append_notice(
"Round provenance not recorded for {n} label(s): {err}\n",
n=str(len(batch)), err=f"{type(exc).__name__}: {exc}")
def _refresh_status_label(self):
"""Say what the save worker is doing, or what stopped it.
A failure wins over everything else: unsaved counts and a "saving…"
note beside an error would read as progress.
"""
w = self._worker
if w is None:
if (self._similar_worker is not None or
time.monotonic() < self._similar_notice_until):
return
self._status_label.setText(tr("Ready."))
return
parts = []
if w.last_error:
self._status_label.setText(f"Save failed — {w.last_error}")
return
if self._pending_updates:
parts.append(f"{len(self._pending_updates)} unsaved change(s)")
if w.busy:
parts.append("saving…")
elif w.pending_batches > 0:
parts.append(f"{w.pending_batches} batch queued")
if (not parts and (self._similar_worker is not None or
time.monotonic() < self._similar_notice_until)):
return
if w.last_save_ts is not None and not parts:
parts.append("saved")
self._status_label.setText(
" · ".join(parts) if parts else tr("Ready."))
[docs]
def closeEvent(self, event):
"""Drain every native/Python worker before Qt destroys this screen.
:param event: the close event, passed on to the base class after
every worker has been stopped or drained.
"""
self._closing = True
self._detach_event_filters()
for report in list(getattr(self, "_reports", {}).values()):
try:
report.close()
except RuntimeError:
pass
if getattr(self, "_reports", None) is not None:
self._reports.clear()
unregister_object_opener("annotate", self._object_opener)
release = getattr(self, "_release_path_probe", None)
if release is not None:
release()
self._path_probe_landed = None
self._resize_timer.stop()
self._pending_page_load = None
self._flush_pending()
try:
self._leave_blind_unopened("the screen was closed")
except Exception:
LOG.debug("could not log the end of a blinded session",
exc_info=True)
self._total_jobs.shutdown()
self._report_jobs.shutdown()
similar = self._similar_worker
if similar is not None:
similar.requestInterruption()
stopped = drain_thread(similar, timeout_ms=CLOSE_DRAIN_MS)
self._similar_worker = None
try:
similar.done.disconnect(self._on_similar_done)
similar.failed.disconnect(self._on_similar_failed)
similar.finished.disconnect(self._on_similar_finished)
except (RuntimeError, TypeError):
pass
if stopped:
_retire(similar)
else:
similar.setParent(None)
retrain = self._retrain_worker
if retrain is not None:
retrain.requestInterruption()
stopped = drain_thread(retrain, timeout_ms=CLOSE_DRAIN_MS)
self._retrain_worker = None
try:
retrain.done.disconnect(self._on_retrain_done)
retrain.failed.disconnect(self._on_retrain_failed)
retrain.finished.disconnect(self._on_retrain_finished)
except (RuntimeError, TypeError):
pass
if stopped:
_retire(retrain)
else:
retrain.setParent(None)
suggest = self._suggest_worker
if suggest is not None:
suggest.requestInterruption()
stopped = drain_thread(suggest, timeout_ms=CLOSE_DRAIN_MS)
self._suggest_worker = None
for signal, slot in ((suggest.done, self._on_suggest_done),
(suggest.failed, self._on_suggest_failed),
(suggest.finished,
self._on_suggest_finished),
*self._suggest_extra_slots(suggest)):
try:
signal.disconnect(slot)
except (RuntimeError, TypeError):
pass
if stopped:
_retire(suggest)
else:
suggest.setParent(None)
if self._worker:
self._worker.stop(wait=True)
self._worker = None
self._page_gen += 1
worker = self._page_worker
if worker is not None:
worker.requestInterruption()
stopped = drain_thread(worker, timeout_ms=CLOSE_DRAIN_MS)
self._page_worker = None
try:
worker.done.disconnect(self._on_page_loaded)
worker.finished.disconnect(self._on_page_worker_finished)
except (RuntimeError, TypeError):
pass
if stopped:
_retire(worker)
try:
self._console.shutdown()
except Exception:
pass
super().closeEvent(event)