"""Classifier Evaluation workbench for out-of-fold result bundles.
Discovery and CSV/JSON parsing run away from the GUI thread. The screen only
renders already-loaded data, so a large OOF prediction table cannot freeze the
application while it is being read.
Opening the objects behind a confusion-matrix cell
--------------------------------------------------
Every cell of the matrix is clickable, and clicking one asks
:func:`spacr.qt.linked_selection.open_objects` for exactly the crops that cell
counted, so "43 uninfected called infected" opens those 43 crops. The analysis behind it is in :mod:`spacr.confusion`,
which has no Qt in it; this file is the table, the two lists and the buttons.
Two lists, not one, per cell. The model being **sure** and wrong is evidence
against the *annotation*; the model being **unsure** and wrong is evidence
about the *boundary*. They are different diagnoses with different fixes, so
they are opened separately and never blended into one confidence-sorted list
where the distinction disappears somewhere in the middle of a scroll.
Each cell is also broken down per well and per plate, which shows where the
errors came from: 43 errors spread over 20 wells point to the model, 43 errors
from one well point to that well.
"""
from __future__ import annotations
import json
import logging
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
import numpy as np
import pandas as pd
from PySide6.QtCore import Qt, QTimer, Signal
from PySide6.QtGui import (QDesktopServices, QDragEnterEvent, QDropEvent,
QStandardItem, QStandardItemModel)
from shiboken6 import isValid
from PySide6.QtCore import QUrl
from PySide6.QtWidgets import (
QAbstractItemView,
QApplication,
QComboBox,
QDoubleSpinBox,
QFileDialog,
QHBoxLayout,
QHeaderView,
QLabel,
QLineEdit,
QListWidget,
QListWidgetItem,
QPlainTextEdit,
QPushButton,
QTabWidget,
QTableView,
QTableWidget,
QTableWidgetItem,
QVBoxLayout,
QWidget,
)
from ... import confusion as cx
from ..bridge import make_thread
from ..iconset import icon
from ..i18n import tr
from ..linked_selection import (DEFAULT_OPEN_KIND, has_object_opener,
open_objects)
from ..theme import (SPACING, active_palette, page_tabs_qss,
register_widget_qss)
from ..widgets import Divider
from ..widgets.collapsible_splitter import CollapsibleSplitter, FoldSection
from ..widgets.sortable_table import (SortableProxyModel, _retire_table_cells,
_reuse_list_items, _spare_table_item,
install_sorting, table_item)
LOG = logging.getLogger(__name__)
[docs]
def find_evaluation_bundles(root: Any) -> List[Path]:
"""Discover evaluation manifests while keeping sklearn off GUI startup.
The implementation is imported only when a scan worker calls this seam.
Keeping the seam at module scope also lets tests and downstream wrappers
replace discovery without importing the scientific stack eagerly.
"""
from ...classifier_evaluation import find_evaluation_bundles as discover
return discover(root)
[docs]
def load_evaluation_bundle(path: Any) -> Dict[str, Any]:
"""Load one evaluation bundle without importing sklearn at GUI startup."""
from ...classifier_evaluation import load_evaluation_bundle as load
return load(path)
__all__ = [
"ClassifierEvaluationScreen",
"APP_KEY",
"APP_NAME",
"APP_SECTION",
"APP_INTRO",
"LINK_SOURCE",
]
#: What this screen calls itself when it routes objects somewhere, so the
#: receiving grid's header can say where the crops came from.
LINK_SOURCE = "classifier_evaluation"
#: How many objects a cell list shows before it stops adding rows. The list is
#: a preview of an ordering, not the work surface — the work surface is the
#: crop grid the button opens, and rendering ninety thousand QListWidgetItems
#: to look at the top twelve is how a click becomes a two-second freeze.
LIST_PREVIEW = 200
#: Milliseconds the prediction filter waits after the last keystroke before
#: it re-filters, so typing ``r5_c2`` filters once rather than five times.
_FILTER_DELAY_MS = 250
#: Most predictions the filtered table shows: the first matches in bundle
#: order, as before the table became a model view.
_ROW_LIMIT = 2000
APP_KEY = "classifier_evaluation"
APP_NAME = "Classifier Evaluation"
APP_SECTION = "Core"
APP_INTRO = (
"Inspect held-out predictions, grouped or nested cross-validation, "
"calibration, confusion matrices, per-plate performance and leakage checks."
)
class _DropPathEdit(QLineEdit):
"""A path field that accepts one dropped folder or manifest.
:param parent: parent widget; ownership only.
"""
path_dropped = Signal(str)
def __init__(self, parent=None):
"""Build the field and let it accept drops."""
super().__init__(parent)
self.setAcceptDrops(True)
def dragEnterEvent(self, event: QDragEnterEvent) -> None:
"""Accept local file/folder URLs."""
if any(url.isLocalFile() for url in event.mimeData().urls()):
event.acceptProposedAction()
else:
event.ignore()
def dropEvent(self, event: QDropEvent) -> None:
"""Place the first dropped local path in the field."""
paths = [
str(Path(url.toLocalFile()))
for url in event.mimeData().urls()
if url.isLocalFile()
]
if paths:
self.setText(paths[0])
self.path_dropped.emit(paths[0])
event.acceptProposedAction()
else:
event.ignore()
def _alive(obj: Any) -> bool:
"""Whether ``obj`` exists and its C++ half has not been destroyed.
:param obj: a Qt object, or ``None``.
:returns: ``False`` for ``None`` and for a wrapper whose C++ object is
gone, ``True`` otherwise.
"""
return obj is not None and isValid(obj)
def _cell_text(value: Any) -> str:
"""The text a result cell shows for ``value``.
:param value: one cell of a result table.
:returns: ``""`` for a missing value, five significant digits for a
float, and ``str(value)`` for anything else.
"""
if value is None or (isinstance(value, float) and pd.isna(value)):
return ""
if isinstance(value, float):
return f"{value:.5g}"
return str(value)
def _item(value: Any, table: Optional[QTableWidget] = None) -> QTableWidgetItem:
"""Return a read-only table item, reusing ``table``'s spare cells if given."""
item = (table_item(_cell_text(value)) if table is None
else _spare_table_item(table, _cell_text(value)))
item.setFlags(item.flags() & ~Qt.ItemIsEditable)
return item
def _row_search_text(frame: pd.DataFrame) -> pd.Series:
"""Each row of ``frame`` as one case-folded string, for the filter box.
The row's cells as text, joined by single spaces, so a filter term
matches anywhere in the row. Pure pandas: the load worker builds it
off the GUI thread.
:param frame: a prediction table.
:returns: one string per row, on ``frame``'s index.
"""
columns = [frame.iloc[:, index].astype(str).tolist()
for index in range(frame.shape[1])]
if columns:
rows = [" ".join(values).casefold() for values in zip(*columns)]
else:
rows = [""] * len(frame)
return pd.Series(rows, index=frame.index, dtype=object)
class _PredictionRows:
"""Every out-of-fold prediction of the loaded bundle, behind the grid.
A ``QStandardItemModel`` sized to the whole bundle once per load. The
filter box never resizes or clears it: typing only changes which rows
:class:`_PredictionFilter` lets through, so no cell is deleted while the
user types and assistive technology keeps stable indexes. A row's cells
are written the first time the filter lets it through, by ``setData``
on an empty cell, so Qt creates every cell itself and none of them ever
has a Python wrapper; painting reads them at C++ speed.
:param parent: Qt parent of the model.
"""
def __init__(self, parent=None):
"""Start empty: no columns and no rows."""
self.model = QStandardItemModel(parent)
prototype = QStandardItem()
prototype.setFlags(Qt.ItemIsSelectable | Qt.ItemIsEnabled)
self.model.setItemPrototype(prototype)
self.frame: pd.DataFrame = pd.DataFrame()
self._columns: List[Any] = []
self._written = np.zeros(0, dtype=bool)
def set_frame(self, frame: pd.DataFrame) -> None:
"""Size the model to ``frame``, with every cell still empty.
Views and proxies see one reset, after the model has its new shape.
:param frame: the bundle's prediction table; it is not modified.
"""
model = self.model
blocked = model.blockSignals(True)
try:
model.clear()
model.setColumnCount(frame.shape[1])
model.setRowCount(len(frame))
model.setHorizontalHeaderLabels(
[tr(str(column)) for column in frame.columns])
finally:
model.blockSignals(blocked)
self.frame = frame
self._columns = [frame.iloc[:, index].to_numpy(dtype=object)
for index in range(frame.shape[1])]
self._written = np.zeros(len(frame), dtype=bool)
model.beginResetModel()
model.endResetModel()
def write(self, rows: np.ndarray) -> None:
"""Fill in the cells of ``rows`` that have never been shown.
Formatted exactly as :func:`_cell_text` formats a table item. The
rows are not yet let through the filter, so nothing is told about
the change; the filter pass that follows shows them.
:param rows: source row numbers, in bundle order.
"""
rows = np.asarray(rows, dtype=np.int64)
todo = rows[~self._written[rows]]
if not len(todo):
return
model = self.model
blocked = model.blockSignals(True)
try:
for row in todo.tolist():
for column, values in enumerate(self._columns):
model.setData(model.index(row, column),
_cell_text(values[row]))
finally:
model.blockSignals(blocked)
self._written[todo] = True
class _PredictionFilter(SortableProxyModel):
"""The rows the filter box selects, sorted the way every table sorts.
:param parent: Qt parent.
"""
def __init__(self, parent=None):
"""Let no row through until a filter has been applied."""
super().__init__(parent)
self._accepted: List[bool] = []
def set_accepted(self, accepted: List[bool], *, quiet: bool = False) -> None:
"""Let through the source rows marked ``True``, and no others.
One reset of the proxy, not a removal per row: a filter that keeps
every third row would otherwise reach the view as hundreds of
separate removals, each re-laying out the header. The source rows
and their cells are untouched.
:param accepted: one flag per source row, in source order.
:param quiet: store the flags without re-filtering, for the moment
just before the source model resets and re-filters anyway.
"""
if quiet:
self._accepted = list(accepted)
return
self.beginResetModel()
try:
self._accepted = list(accepted)
finally:
self.endResetModel()
def filterAcceptsRow(self, source_row, source_parent) -> bool: # noqa: N802
"""Whether the filter keeps one source row.
:param source_row: the row in bundle order.
:param source_parent: the source parent index; unused by a table.
:returns: the row's flag from :meth:`set_accepted`.
"""
accepted = self._accepted
return 0 <= source_row < len(accepted) and accepted[source_row]
class _CellText:
"""One read-only cell, with the ``text()`` a table item would have.
:param text: what the cell shows.
"""
__slots__ = ("_text",)
def __init__(self, text: str):
"""Hold the cell's text."""
self._text = text
def text(self) -> str:
"""The cell's text.
:returns: what the cell shows.
"""
return self._text
class _PredictionTable(QTableView):
"""The predictions grid, readable the way the old ``QTableWidget`` was.
``rowCount``, ``columnCount``, ``item`` and ``horizontalHeaderItem``
answer from the rows on screen, filtered and sorted, so a test or a
tutorial driver that read the old grid reads this one the same way.
:param parent: Qt parent.
"""
def setModel(self, model) -> None: # noqa: N802
"""Show ``model`` with the shared three-state header sorting.
:param model: the filter model over the bundle's prediction rows.
"""
super().setModel(model)
install_sorting(self)
def rowCount(self) -> int: # noqa: N802
"""Rows on screen.
:returns: how many predictions the filter lets through.
"""
model = self.model()
return 0 if model is None else model.rowCount()
def columnCount(self) -> int: # noqa: N802
"""Columns on screen.
:returns: the column count.
"""
model = self.model()
return 0 if model is None else model.columnCount()
def item(self, row: int, column: int) -> Optional[_CellText]:
"""The cell at a row and column on screen.
:param row: the row as displayed, after filtering and sorting.
:param column: the column number.
:returns: the cell, or ``None`` outside the table.
"""
model = self.model()
if model is None:
return None
index = model.index(row, column)
if not index.isValid():
return None
value = model.data(index, Qt.ItemDataRole.DisplayRole)
return _CellText("" if value is None else str(value))
def horizontalHeaderItem(self, column: int) -> Optional[_CellText]: # noqa: N802
"""The header cell of one column.
:param column: the column number.
:returns: the header, or ``None`` outside the table.
"""
model = self.model()
if model is None or not 0 <= column < model.columnCount():
return None
value = model.headerData(column, Qt.Orientation.Horizontal,
Qt.ItemDataRole.DisplayRole)
return _CellText("" if value is None else str(value))
#: ``objectName`` of the tab strip, and the name its QSS block is
#: registered under. The tabs ARE the page on this screen, so they take
#: Home's treatment — rounded top corners, a dark-grey tab, the accent
#: blue on hover — at the page opacity, instead of the shipped rules'
#: opaque `surface`/`surface_alt` hex.
TABS_NAME = "ClassifierEvalTabs"
def _tabs_qss(palette: dict, opacity) -> str:
"""QSS for the tab strip, registered through the theme seam."""
return page_tabs_qss(TABS_NAME, palette, opacity)
register_widget_qss(TABS_NAME, _tabs_qss, replace=True)
[docs]
class ClassifierEvaluationScreen(QWidget):
"""Browse and inspect classifier evaluation bundles.
:param parent: Qt parent.
:param threaded: discover and parse bundles in a worker thread. Tests may
pass ``False`` to make refresh deterministic.
:ivar last_error: latest non-fatal source or parsing error.
:ivar bundles: manifests found by the most recent scan.
"""
evaluation_loaded = Signal(str)
def __init__(self, parent=None, threaded: bool = True):
"""Build the screen and arm its drop zone.
The confusion cell under inspection is held, so moving the confidence
threshold re-splits that cell rather than making the user click it
again.
:param parent: parent widget, or ``None``.
:param threaded: scan on a worker thread. Set ``False`` in tests so
``scan`` finishes before it returns.
"""
super().__init__(parent)
self._threaded = bool(threaded)
self._busy = False
self._jobs: List[tuple] = []
self._pending_error = ""
self._pending_bundles: Optional[List[Path]] = None
self._pending_bundle: Optional[Dict[str, Any]] = None
self.bundles: List[Path] = []
self.bundle: Optional[Dict[str, Any]] = None
self.last_error = ""
#: the confusion cell currently being inspected, or ``None``. Held so
#: that moving the confidence threshold re-splits *this* cell rather
#: than needing the user to click it again.
self._cell: Optional[cx.ConfusionCell] = None
#: the counts matrix the cells are read out of, for tests and for the
#: ranking line.
self._counts = pd.DataFrame()
self._rendering_predictions = False
self._predictions_stale = False
self._bundle_serial = 0
self._shown_predictions: Optional[tuple] = None
self._search_text: Optional[tuple] = None
self._pending_search: Optional[tuple] = None
self._build_ui()
self._set_status(
tr("Choose or drop a classifier run folder, then select Scan.")
)
from ..dnd import install_for
install_for(self, "classifier_evaluation")
from .settings_model import retarget_field_tooltips
retarget_field_tooltips(self)
def _build_ui(self) -> None:
"""Construct source controls and tabbed evaluation views."""
outer = QVBoxLayout(self)
outer.setContentsMargins(
SPACING["lg"], SPACING["lg"], SPACING["lg"], SPACING["lg"],
)
outer.setSpacing(SPACING["md"])
title = QLabel(tr(APP_NAME), self)
title.setObjectName("DisplayHeading")
outer.addWidget(title)
intro = QLabel(tr(APP_INTRO), self)
intro.setObjectName("Muted")
intro.setWordWrap(True)
outer.addWidget(intro)
outer.addWidget(Divider())
source_row = QHBoxLayout()
source_label = QLabel(tr("Results folder"), self)
self._source = _DropPathEdit(self)
self._source.setPlaceholderText(
tr("Drop a run/evaluation folder or evaluation_manifest.json")
)
self._source.returnPressed.connect(self.scan)
self._source.path_dropped.connect(lambda _path: self.scan())
self._browse = QPushButton(tr("Browse…"), self)
self._browse.setIcon(icon("folder"))
self._browse.clicked.connect(self._choose_source)
self._scan = QPushButton(tr("Scan"), self)
self._scan.setObjectName("PrimaryButton")
self._scan.setIcon(icon("redo"))
self._scan.clicked.connect(self.scan)
source_row.addWidget(source_label)
source_row.addWidget(self._source, 1)
source_row.addWidget(self._browse)
source_row.addWidget(self._scan)
outer.addLayout(source_row)
bundle_row = QHBoxLayout()
bundle_row.addWidget(QLabel(tr("Evaluation run"), self))
self._bundle_choice = QComboBox(self)
self._bundle_choice.currentIndexChanged.connect(
self._load_selected_bundle,
)
self._open_folder = QPushButton(tr("Open folder"), self)
self._open_folder.setIcon(icon("folder"))
self._open_folder.clicked.connect(self._open_current_folder)
self._copy_path = QPushButton(tr("Copy path"), self)
self._copy_path.clicked.connect(self._copy_current_path)
bundle_row.addWidget(self._bundle_choice, 1)
bundle_row.addWidget(self._open_folder)
bundle_row.addWidget(self._copy_path)
outer.addLayout(bundle_row)
self._tabs = QTabWidget(self)
self._tabs.setObjectName(TABS_NAME)
self._overview = QPlainTextEdit(self)
self._overview.setReadOnly(True)
confusion_page = self._build_confusion_page()
self._per_plate = self._table()
self._calibration = self._table()
predictions_page = QWidget(self)
predictions_layout = QVBoxLayout(predictions_page)
predictions_layout.setContentsMargins(0, 0, 0, 0)
filter_row = QHBoxLayout()
filter_row.addWidget(QLabel(tr("Filter predictions"), predictions_page))
self._prediction_filter = QLineEdit(predictions_page)
self._prediction_filter.setPlaceholderText(
tr("Plate, well, class, filename…")
)
self._prediction_filter.setClearButtonEnabled(True)
self._filter_timer = QTimer(self)
self._filter_timer.setSingleShot(True)
self._filter_timer.setInterval(_FILTER_DELAY_MS)
self._filter_timer.timeout.connect(self._render_predictions)
self._prediction_filter.textChanged.connect(
self._schedule_prediction_filter,
)
self._prediction_filter.returnPressed.connect(
self._apply_prediction_filter,
)
self._prediction_filter.editingFinished.connect(
self._apply_prediction_filter,
)
filter_row.addWidget(self._prediction_filter, 1)
predictions_layout.addLayout(filter_row)
self._prediction_rows = _PredictionRows(self)
self._prediction_view = _PredictionFilter(self)
self._prediction_view.setSourceModel(self._prediction_rows.model)
self._predictions = self._prediction_table()
predictions_layout.addWidget(self._predictions, 1)
self._leakage = QPlainTextEdit(self)
self._leakage.setReadOnly(True)
for label, widget in (
(tr("Summary"), self._overview),
(tr("Confusion matrix"), confusion_page),
(tr("Per-plate metrics"), self._per_plate),
(tr("Calibration"), self._calibration),
(tr("Predictions"), predictions_page),
(tr("Leakage audit"), self._leakage),
):
self._tabs.addTab(widget, label)
self._tabs_section = FoldSection(
self._tabs, "Evaluation results",
persist_key="classifier_evaluation/Evaluation results")
outer.addWidget(self._tabs_section, 1)
self._status = QLabel("", self)
self._status.setObjectName("Muted")
self._status.setWordWrap(True)
outer.addWidget(self._status)
self._clear_bundle()
def _prediction_table(self) -> _PredictionTable:
"""The predictions grid: a view over the bundle's rows.
Read-only, row-selecting, alternating and sortable like every other
table here, with no item of its own to delete when the filter
changes.
"""
table = _PredictionTable(self)
table.setModel(self._prediction_view)
table.setEditTriggers(QAbstractItemView.NoEditTriggers)
table.setSelectionBehavior(QAbstractItemView.SelectRows)
table.setAlternatingRowColors(True)
table.verticalHeader().setVisible(False)
return table
def _table(self) -> QTableWidget:
"""Return a read-only, row-selecting data table."""
table = QTableWidget(0, 0, self)
install_sorting(table)
table.setEditTriggers(QAbstractItemView.NoEditTriggers)
table.setSelectionBehavior(QAbstractItemView.SelectRows)
table.setAlternatingRowColors(True)
table.verticalHeader().setVisible(False)
return table
def _build_confusion_page(self) -> QWidget:
"""The matrix, the ranking in words, and the clicked cell's two lists.
Selection is per *cell*, not per row: a confusion is a (true,
predicted) pair and selecting whole rows would make "click the cell
with 43 in it" ambiguous between the four cells beside it.
"""
page = QWidget(self)
layout = QVBoxLayout(page)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(SPACING["sm"])
self._confusion_ranking = QLabel("", page)
self._confusion_ranking.setWordWrap(True)
layout.addWidget(self._confusion_ranking)
split = CollapsibleSplitter(
Qt.Vertical, page,
persist_key="classifier_evaluation::confusion")
self._confusion = self._table()
self._confusion.setSelectionBehavior(QAbstractItemView.SelectItems)
self._confusion.setSelectionMode(QAbstractItemView.SingleSelection)
self._confusion.cellClicked.connect(self._on_confusion_cell)
split.add_section(self._confusion, "Confusion table",
persist_key="classifier_evaluation/Confusion table")
inspector = QWidget()
column = QVBoxLayout(inspector)
column.setContentsMargins(0, 0, 0, 0)
column.setSpacing(SPACING["sm"])
controls = QHBoxLayout()
controls.addWidget(QLabel(tr("The model counts as sure at"), inspector))
self._threshold = QDoubleSpinBox(inspector)
self._threshold.setRange(0.0, 1.0)
self._threshold.setSingleStep(0.05)
self._threshold.setDecimals(2)
self._threshold.setValue(0.75)
self._threshold.setToolTip(tr(
"Errors at or above this confidence suggest a wrong label; below "
"it, a boundary the model has not learned."))
self._threshold.valueChanged.connect(self._on_threshold_changed)
controls.addWidget(self._threshold)
controls.addStretch(1)
column.addLayout(controls)
self._cell_summary = QLabel("", inspector)
self._cell_summary.setWordWrap(True)
column.addWidget(self._cell_summary)
self._cell_breakdown = QLabel("", inspector)
self._cell_breakdown.setWordWrap(True)
self._cell_breakdown.setObjectName("Muted")
column.addWidget(self._cell_breakdown)
lists = QHBoxLayout()
lists.setSpacing(SPACING["md"])
self._high_head, self._high_list, self._high_open = self._error_column(
inspector, tr("Sure and wrong — suspect the label"),
tr("Open these crops in Annotate, most confident first. If the "
"model is this certain and disagrees, look at the annotation."),
self._open_high)
lists.addLayout(self._high_head)
self._low_head, self._low_list, self._low_open = self._error_column(
inspector, tr("Unsure and wrong — suspect the boundary"),
tr("Open these crops in Annotate, least confident first. These sat "
"on the decision boundary; the label is probably right."),
self._open_low)
lists.addLayout(self._low_head)
column.addLayout(lists, 1)
split.add_section(inspector, "Error inspector",
persist_key="classifier_evaluation/Error inspector")
layout.addWidget(split, 1)
self._confusion_splitter = split
return page
def _error_column(self, parent: QWidget, title: str, tip: str,
on_open) -> Tuple[QVBoxLayout, QListWidget, QPushButton]:
"""One of the two lists: a heading, the objects, and one button."""
box = QVBoxLayout()
box.setSpacing(4)
heading = QLabel(title, parent)
heading.setWordWrap(True)
box.addWidget(heading)
listing = QListWidget(parent)
listing.setSelectionMode(QAbstractItemView.NoSelection)
listing.setToolTip(tip)
box.addWidget(listing, 1)
button = QPushButton(tr("Open in Annotate"), parent)
button.setToolTip(tip)
button.setEnabled(False)
button.clicked.connect(on_open)
box.addWidget(button)
return box, listing, button
def _choose_source(self) -> None:
"""Choose a folder containing one or more evaluation bundles."""
path = QFileDialog.getExistingDirectory(
self, tr("Choose classifier results folder"),
self._source.text().strip(),
)
if path:
self._source.setText(path)
self.scan()
[docs]
def scan(self) -> None:
"""Discover evaluation bundles without blocking the GUI thread."""
if self._busy:
return
source = self._source.text().strip()
if not source:
self._set_status(tr("Choose a results folder first."), error=True)
return
self._start_busy(tr("Searching for classifier evaluations…"))
self._pending_bundles = None
self._pending_bundle = None
self._pending_error = ""
def _work(_settings):
"""Find the evaluation bundles under a folder. Off the GUI thread."""
try:
self._pending_bundles = find_evaluation_bundles(source)
except Exception as exc:
self._pending_error = f"{type(exc).__name__}: {exc}"
LOG.exception("Classifier-evaluation scan failed")
if not self._threaded:
_work({})
self._finish_scan(not bool(self._pending_error))
return
thread, worker = make_thread(
_work, {}, app_key="classifier_evaluation_scan", journal=False,
)
self._keep_job(thread, worker, self._finish_scan)
def _finish_scan(self, ok: bool) -> None:
"""Populate the run selector after discovery."""
self._finish_busy()
self.last_error = self._pending_error
self.bundles = list(self._pending_bundles or []) if ok else []
self._bundle_choice.blockSignals(True)
self._bundle_choice.clear()
root = Path(self._source.text().strip()).expanduser()
for manifest in self.bundles:
try:
label = str(manifest.parent.relative_to(
root if root.is_dir() else root.parent,
))
except ValueError:
label = str(manifest.parent)
self._bundle_choice.addItem(label or manifest.parent.name, manifest)
self._bundle_choice.blockSignals(False)
if self.last_error:
self._clear_bundle()
self._set_status(
f"{tr('Could not scan classifier evaluations')}: "
f"{self.last_error}",
error=True,
)
elif not self.bundles:
self._clear_bundle()
self._set_status(
tr("No evaluation_manifest.json files were found."),
error=True,
)
else:
self._set_status(
f"{tr('Found')} {len(self.bundles)} "
f"{tr('classifier evaluation(s).')}"
)
self._bundle_choice.setCurrentIndex(0)
self._load_selected_bundle()
def _load_selected_bundle(self, *_args) -> None:
"""Parse the selected bundle in a worker."""
if self._busy:
return
manifest = self._bundle_choice.currentData()
if not manifest:
return
self._start_busy(tr("Loading evaluation tables…"))
self._pending_bundle = None
self._pending_search = None
self._pending_error = ""
def _work(_settings):
"""Load one evaluation bundle. Off the GUI thread."""
try:
bundle = load_evaluation_bundle(manifest)
predictions = bundle.get("predictions")
if isinstance(predictions, pd.DataFrame):
self._pending_search = (predictions,
_row_search_text(predictions))
self._pending_bundle = bundle
except Exception as exc:
self._pending_error = f"{type(exc).__name__}: {exc}"
LOG.exception("Classifier-evaluation load failed")
if not self._threaded:
_work({})
self._finish_load(not bool(self._pending_error))
return
thread, worker = make_thread(
_work, {}, app_key="classifier_evaluation_load", journal=False,
)
self._keep_job(thread, worker, self._finish_load)
def _finish_load(self, ok: bool) -> None:
"""Render a loaded bundle on the GUI thread."""
self._finish_busy()
self.last_error = self._pending_error
self.bundle = self._pending_bundle if ok else None
self._bundle_serial += 1
pending = self._pending_search
self._pending_search = None
self._search_text = (pending if pending is not None and self.bundle
is not None and pending[0] is
self.bundle.get("predictions") else None)
if self.bundle is None:
self._clear_bundle()
self._set_status(
f"{tr('Could not load classifier evaluation')}: "
f"{self.last_error or tr('worker failed')}",
error=True,
)
return
summary = dict(self.bundle.get("summary") or {})
manifest = dict(self.bundle.get("manifest") or {})
self._overview.setPlainText(json.dumps({
"results": str(self.bundle["path"]),
"summary": summary,
"bundle_warnings": manifest.get("warnings") or [],
}, indent=2, sort_keys=True, default=str))
self._render_confusion(summary)
self._render_frame(self._per_plate, self.bundle["per_plate"])
self._render_frame(self._calibration, self.bundle["calibration"])
self._leakage.setPlainText(json.dumps(
self.bundle.get("leakage") or {},
indent=2, sort_keys=True, default=str,
))
self._render_predictions()
self._open_folder.setEnabled(True)
self._copy_path.setEnabled(True)
count = len(self.bundle["predictions"])
passed = bool((self.bundle.get("leakage") or {}).get("passed", False))
self._set_status(
f"{tr('Loaded')} {count} {tr('held-out prediction(s)')} · "
f"{tr('leakage audit')}: "
f"{tr('passed') if passed else tr('review required')}."
)
self.evaluation_loaded.emit(str(self.bundle["path"]))
def _render_frame(
self,
table: QTableWidget,
frame: pd.DataFrame,
*,
include_index: bool = False,
row_limit: Optional[int] = None,
) -> None:
"""Render a dataframe without mutating it.
A table whose C++ half is already gone is left alone: a filter that
was waiting when the screen closed must not clear a deleted table.
"""
if not _alive(table):
return
shown = frame if row_limit is None else frame.head(row_limit)
columns = list(shown.columns)
if include_index:
columns = [shown.index.name or "true_class", *columns]
table.setSortingEnabled(False)
_retire_table_cells(table, headers=True)
table.setColumnCount(len(columns))
table.setHorizontalHeaderLabels([str(column) for column in columns])
table.setRowCount(len(shown))
for row, (index, values) in enumerate(shown.iterrows()):
offset = 0
if include_index:
table.setItem(row, 0, _item(index, table))
offset = 1
for column, value in enumerate(values):
table.setItem(row, column + offset, _item(value, table))
header = table.horizontalHeader()
header.setSectionResizeMode(QHeaderView.ResizeToContents)
if columns:
header.setSectionResizeMode(len(columns) - 1, QHeaderView.Stretch)
table.setSortingEnabled(True)
def _schedule_prediction_filter(self, text: str = "") -> None:
"""Re-filter once typing pauses, or at once when the box is emptied.
Every keystroke restarts the wait, so typing ``r5_c2`` filters the
table once instead of five times, and the pass reads the box when
it runs: the newest filter always wins. Emptying the box (its clear
button, or Ctrl+A then Backspace) is one deliberate action and
restores every row at once.
:param text: the filter as ``textChanged`` reports it.
"""
if not str(text or "").strip():
self._apply_prediction_filter()
return
if _alive(self._filter_timer):
self._filter_timer.start()
def _apply_prediction_filter(self) -> None:
"""Apply the filter now: Enter, leaving the box, or an emptied box."""
timer = getattr(self, "_filter_timer", None)
if _alive(timer):
timer.stop()
self._render_predictions()
def _render_predictions(self, *_args) -> None:
"""Show at most 2,000 OOF rows matching the filter, one pass at a time.
Filtering only changes which of the bundle's rows the view lets
through; no cell is created or deleted. A request that arrives while
a pass is running is not nested inside it; it marks the view stale,
and the running pass makes one more afterwards that reads the filter
as it is by then. A request for the filter already on screen does
nothing, so Enter after the pause, or leaving the box, does not
filter a second time.
"""
if self._rendering_predictions:
self._predictions_stale = True
return
if not (_alive(self) and _alive(self._predictions)
and _alive(self._prediction_filter)):
return
timer = getattr(self, "_filter_timer", None)
self._rendering_predictions = True
try:
self._predictions_stale = True
while self._predictions_stale:
self._predictions_stale = False
if not (_alive(self._predictions)
and _alive(self._prediction_filter)):
return
if _alive(timer):
timer.stop()
self._show_filtered_predictions()
finally:
self._rendering_predictions = False
def _show_filtered_predictions(self) -> None:
"""Let through the rows the filter box selects, unless already shown.
Every term must occur in the row's text, case-folded; the first
:data:`_ROW_LIMIT` matches in bundle order are shown.
"""
if self.bundle is None:
frame, terms = pd.DataFrame(), ()
else:
frame = self.bundle["predictions"]
terms = tuple(
term.casefold()
for term in self._prediction_filter.text().split()
if term
)
key = (self._bundle_serial, terms)
if key == self._shown_predictions:
return
if self._prediction_rows.frame is not frame:
self._set_prediction_frame(frame)
mask = np.ones(len(frame), dtype=bool)
if terms and not frame.empty:
text = self._prediction_search_text(frame)
for term in terms:
mask &= text.str.contains(
term, regex=False, na=False).to_numpy(dtype=bool)
shown = np.flatnonzero(mask)[:_ROW_LIMIT]
self._prediction_rows.write(shown)
accepted = np.zeros(len(frame), dtype=bool)
accepted[shown] = True
self._prediction_view.set_accepted(accepted.tolist())
self._shown_predictions = key
def _set_prediction_frame(self, frame: pd.DataFrame) -> None:
"""Put a new bundle's predictions behind the grid, showing none yet.
The only time the grid's rows are replaced: once per bundle, never
per keystroke. Every column fits its contents and the last one
takes the remaining width, as on the other tables.
:param frame: the bundle's prediction table, or an empty frame.
"""
self._prediction_view.set_accepted([], quiet=True)
self._prediction_rows.set_frame(frame)
if not _alive(self._predictions):
return
header = self._predictions.horizontalHeader()
header.setSectionResizeMode(QHeaderView.ResizeToContents)
if frame.shape[1]:
header.setSectionResizeMode(frame.shape[1] - 1, QHeaderView.Stretch)
def _prediction_search_text(self, frame: pd.DataFrame) -> pd.Series:
"""Each prediction row as one case-folded string, built once a bundle.
:param frame: the loaded out-of-fold table.
:returns: the searchable text of each row, on ``frame``'s index.
"""
cached = self._search_text
if cached is not None and cached[0] is frame:
return cached[1]
text = _row_search_text(frame)
self._search_text = (frame, text)
return text
def _render_confusion(self, summary: Dict[str, Any]) -> None:
"""Draw the matrix in COUNTS, and say the ranking out loud.
Counts, not the normalised shares the bundle also carries, because
this table is now a set of buttons: "43" is what tells you whether a
cell is worth opening, while "0.12" needs the row total to interpret.
The share is on every cell's tooltip, so nothing is lost.
Recomputed from the prediction table rather than read from
``confusion_counts.csv``. The two are the same number today, but only
one of them can also list the objects — and a matrix that disagreed
with the crops it opens is worse than no matrix.
"""
predictions = self._predictions_frame()
classes = summary.get("classes") or None
try:
counts = cx.confusion_counts(predictions, classes)
except cx.ConfusionError as exc:
LOG.info("cannot build a clickable confusion matrix: %s", exc)
self._render_frame(self._confusion,
self.bundle["confusion_normalized"],
include_index=True)
self._confusion_ranking.setText(
f"{tr('This bundle cannot be broken down by object')}: {exc}")
self._clear_cell()
return
self._counts = counts
self._render_frame(self._confusion, counts, include_index=True)
shares = counts.to_numpy(dtype=float)
row_totals = shares.sum(axis=1)
for row in range(self._confusion.rowCount()):
for column in range(1, self._confusion.columnCount()):
item = self._confusion.item(row, column)
if item is None:
continue
total = row_totals[row] if row < len(row_totals) else 0.0
value = shares[row, column - 1] if row < len(shares) else 0.0
item.setToolTip(
f"{int(value)} object(s) · {value / total:.1%} of this "
f"row" if total else f"{int(value)} object(s)")
n_classes = max(2, len(counts.index))
blocked = self._threshold.blockSignals(True)
try:
self._threshold.setValue(cx.confidence_threshold(n_classes))
finally:
self._threshold.blockSignals(blocked)
self._confusion_ranking.setText(cx.describe_confusions(counts))
self._clear_cell()
def _predictions_frame(self) -> pd.DataFrame:
"""The loaded out-of-fold table, or an empty frame."""
if self.bundle is None:
return pd.DataFrame()
frame = self.bundle.get("predictions")
return frame if isinstance(frame, pd.DataFrame) else pd.DataFrame()
def _on_confusion_cell(self, row: int, column: int) -> None:
"""A cell was clicked: resolve it to (true, predicted) and inspect it.
Read out of the widget rather than out of the frame by position. The
table is sortable, so visual row 0 is not model row 0 after a header
click — and a cell inspector that showed the wrong class after a sort
would be worse than one that showed nothing.
"""
if column <= 0:
self._set_status(tr(
"Click a cell rather than the class name: a confusion is a "
"(true, predicted) pair."))
return
true_item = self._confusion.item(row, 0)
header = self._confusion.horizontalHeaderItem(column)
if true_item is None or header is None:
return
self.show_cell(true_item.text(), header.text())
[docs]
def show_cell(self, true_class: str, predicted_class: str) -> None:
"""Inspect one confusion cell. The seam a test (or a link) goes through.
:param true_class: annotated class naming the confusion-matrix row.
:param predicted_class: model-predicted class naming the
confusion-matrix column.
:returns: nothing; the two lists, the breakdown and the buttons are
the result.
"""
predictions = self._predictions_frame()
if predictions.empty:
self._clear_cell()
return
try:
cell = cx.ConfusionCell.build(
predictions, true_class, predicted_class,
threshold=float(self._threshold.value()))
except cx.ConfusionError as exc:
self._clear_cell()
self._cell_summary.setText(str(exc))
return
self._cell = cell
self._cell_summary.setText(
f"{cell.reason()} — {cell.describe()}")
self._cell_breakdown.setText(self._describe_origin(cell))
self._fill_list(self._high_list, cell.high, cell.threshold)
self._fill_list(self._low_list, cell.low, cell.threshold)
openable = has_object_opener(DEFAULT_OPEN_KIND)
for button, frame, label in (
(self._high_open, cell.high, tr("sure and wrong")),
(self._low_open, cell.low, tr("unsure and wrong")),
):
button.setEnabled(openable and not frame.empty)
button.setText(
f"{tr('Open')} {len(frame)} {label}"
if not frame.empty else tr("Nothing to open"))
if not openable:
button.setToolTip(tr(
"Open the Annotate screen first — it is what shows crops."))
def _describe_origin(self, cell: "cx.ConfusionCell") -> str:
"""Per-well and per-plate breakdown of one cell, both lines.
Both levels, always, because they answer different questions and the
interesting case is when they disagree: concentrated in one well but
spread over plates is a pipetting error, concentrated in one plate but
spread over its wells is an imaging session.
"""
lines = []
for level in ("well", "plate"):
try:
lines.append(f"{level}: {cx.describe_breakdown(cell.rows, level)}")
except cx.ConfusionError as exc:
lines.append(f"{level}: {exc}")
return "\n".join(lines)
def _fill_list(self, listing: QListWidget, frame: pd.DataFrame,
threshold: float) -> None:
"""Show the head of one ordered list, and say when it is a head."""
if frame.empty:
_reuse_list_items(listing, 1)[0].setText(tr("(none)"))
return
try:
column = cx.object_key_column(frame)
except cx.ConfusionError:
column = None
shown = frame.head(LIST_PREVIEW)
more = len(frame) > len(shown)
items = _reuse_list_items(listing, len(shown) + (1 if more else 0))
for item, (_index, values) in zip(items, shown.iterrows()):
name = (str(values.get("basename") or values.get(column or "", ""))
if column else "")
confidence = values.get(cx.CONFIDENCE_COLUMN)
text = f"{float(confidence):.3f} {name}" if pd.notna(
confidence) else f" ? {name}"
item.setText(text)
item.setToolTip(str(values.get(column or "", name)))
if more:
items[-1].setText(
f"… {len(frame) - len(shown)} {tr('more; open them to see all')}")
def _on_threshold_changed(self, _value: float) -> None:
"""Re-split the open cell where "sure" now starts."""
if self._cell is not None:
self.show_cell(self._cell.true_class, self._cell.predicted_class)
def _open_high(self) -> Any:
"""Open the crops in the high-confidence half of the inspected cell.
:returns: whatever :meth:`open_cell` returns.
"""
return self.open_cell("high")
def _open_low(self) -> Any:
"""Open the crops in the low-confidence half of the inspected cell.
:returns: whatever :meth:`open_cell` returns.
"""
return self.open_cell("low")
[docs]
def open_cell(self, which: str) -> Any:
"""Route one half of the open cell to whatever shows crops.
Nothing here imports Annotate: the request travels through
:func:`spacr.qt.linked_selection.open_objects`, so a second
destination added later needs no change in this file.
The per-key confidences ride along in ``context`` so the receiver can
show *why* this order, and the threshold so it can say where the split
was made.
:param which: ``"high"`` for the confident (sure and wrong) half of the
open cell or ``"low"`` for the unconfident half; a value the cell
rejects is reported in the status line.
:returns: whatever the opener returned, or ``None`` when there was
nothing to open or nowhere to open it.
"""
cell = self._cell
if cell is None:
return None
try:
keys = cell.keys(which)
except cx.ConfusionError as exc:
self._set_status(str(exc), error=True)
return None
if not len(keys):
self._set_status(tr("That list is empty."))
return None
frame = cell.high if which == "high" else cell.low
scores = {}
try:
column = cx.object_key_column(frame)
scores = {
str(k): float(v)
for k, v in zip(frame[column], frame[cx.CONFIDENCE_COLUMN])
if pd.notna(v)
}
except Exception:
LOG.debug("no per-key confidences to send with the request",
exc_info=True)
try:
return open_objects(
keys, reason=cell.reason(which), source=LINK_SOURCE,
context={"scores": scores, "threshold": cell.threshold,
"true_class": cell.true_class,
"predicted_class": cell.predicted_class,
"which": which})
except Exception as exc:
LOG.exception("Could not open a confusion cell")
self._set_status(f"{tr('Could not open those crops')}: {exc}",
error=True)
return None
def _clear_cell(self) -> None:
"""No cell is open: say so rather than showing the last one's lists."""
self._cell = None
self._cell_summary.setText(tr(
"Click a cell of the matrix to see the objects it counted."))
self._cell_breakdown.setText("")
for listing in (self._high_list, self._low_list):
_reuse_list_items(listing, 0)
for button in (self._high_open, self._low_open):
button.setEnabled(False)
button.setText(tr("Open in Annotate"))
def _start_busy(self, text: str) -> None:
"""Disable source actions while a worker is active."""
self._busy = True
self._scan.setEnabled(False)
self._browse.setEnabled(False)
self._bundle_choice.setEnabled(False)
self._set_status(text)
def _finish_busy(self) -> None:
"""Restore source actions after worker completion."""
self._busy = False
self._scan.setEnabled(True)
self._browse.setEnabled(True)
self._bundle_choice.setEnabled(True)
def _keep_job(self, thread, worker, callback) -> None:
"""Retain a worker pair until its QThread exits."""
pair = (thread, worker)
self._jobs.append(pair)
worker.finished.connect(callback)
thread.finished.connect(self._retire_jobs)
thread.start()
def _retire_jobs(self) -> None:
"""Drop ownership of pairs whose QThread has stopped.
A bare ``isRunning()`` filter leaked every pair: by the time this
queued slot runs, ``thread.finished -> deleteLater`` has reaped the
QThread's C++ half and ``isRunning()`` raises ``RuntimeError`` out
of the slot, so the assignment never happens. See
:func:`spacr.qt.bridge.prune_job_pairs`.
"""
from ..bridge import prune_job_pairs
self._jobs = prune_job_pairs(self._jobs, self.sender())
[docs]
def closeEvent(self, event) -> None:
"""Drain scan/load workers before Qt destroys this screen.
A job left running here outlives its owner but stays in the
process-wide run registry, which ``MainWindow.closeEvent`` reads to
decide whether the application may quit.
:param event: the close event; it is not inspected, only passed on to
the base class once the workers are drained.
"""
from ..bridge import drain_thread
if _alive(self._filter_timer):
self._filter_timer.stop()
for thread, worker in list(self._jobs):
if worker is not None:
try:
worker.request_cancel("classifier-evaluation closed")
except Exception:
pass
drain_thread(thread, worker, timeout_ms=3000)
self._jobs.clear()
super().closeEvent(event)
def _clear_bundle(self) -> None:
"""Clear all views and disable bundle actions."""
self.bundle = None
self._bundle_serial += 1
self._shown_predictions = None
self._search_text = None
self._counts = pd.DataFrame()
self._overview.clear()
self._leakage.clear()
for table in (
self._confusion, self._per_plate, self._calibration,
):
self._render_frame(table, pd.DataFrame())
self._set_prediction_frame(pd.DataFrame())
self._confusion_ranking.setText("")
self._clear_cell()
self._open_folder.setEnabled(False)
self._copy_path.setEnabled(False)
def _open_current_folder(self) -> None:
"""Open the loaded evaluation folder."""
if self.bundle is not None:
QDesktopServices.openUrl(QUrl.fromLocalFile(
str(Path(self.bundle["path"]).parent.resolve())
))
def _copy_current_path(self) -> None:
"""Copy the loaded manifest path."""
if self.bundle is not None:
QApplication.clipboard().setText(str(self.bundle["path"]))
self._set_status(tr("Evaluation path copied."))
def _set_status(self, text: str, *, error: bool = False) -> None:
"""Show an inline status; failures are never silent or modal."""
self._status.setText(str(text))
palette = active_palette()
self._status.setStyleSheet(
f"color: {palette['error']};" if error else ""
)