"""Active-learning queue — order unannotated crops by model uncertainty.
Annotation is the bottleneck in every screen. A crop the model already
calls with 0.999 teaches the model nothing when a human labels it; the
crops worth a person's afternoon are the ones sitting on the decision
boundary. This module turns an already-scored ``measurements.db`` into a
work queue ordered so the informative crops come first.
Public API
----------
``least_confidence(probs)`` / ``margin(probs)`` / ``entropy(probs)``
Per-row uncertainty scores. Pure numpy — no torch, no database.
``disagreement(prob_sets)``
Spread across an ensemble or a set of MC-dropout passes.
``rank_by_uncertainty(probs, measure=…)``
Row indices, most uncertain first, deterministically tie-broken.
``build_queue(db_path, annotation_column, …)``
The queue itself, read straight out of ``png_list``.
``queue_rows(queue)``
The queue as ``[(png_path, None), …]`` — the shape the Annotate
screen already paginates (see :func:`spacr.qt.annotate_engine.fetch_page`).
``format_queue_summary(queue)``
The queue's shape, class balance and caveats, as text.
``probabilities_from_logits(logits)`` / ``as_probabilities(scores)``
Raw-head-output → probability matrix, both head shapes handled.
``predict_probabilities(model, batches)``
Optional live-model bridge; the only function that touches torch.
Things this module refuses to get wrong
---------------------------------------
**A softmax is not a probability.** Everything here is called an
uncertainty *score*, never a confidence. Modern networks are badly
calibrated — typically over-confident, and more so the deeper they are
(Guo et al., 2017) — so a 0.87 from the head is not "87 % sure". The
scores are used for one thing only: putting crops in an order. Nothing
in this module reports a calibrated probability, and neither should
anything built on it. See :data:`CALIBRATION_NOTE`.
**Already-annotated crops never enter the queue.** ``NULL`` in the
annotation column is the abstention marker — the crop has not been
looked at. ``0`` is a real class, and a queue that re-serves it wastes
exactly the resource this module exists to save. The two are
distinguished by ``IS NULL``, never by falsiness. (Same convention as
:mod:`spacr.agreement`.)
**The two head shapes are handled separately.** The classifier head
emits either a single logit (binary; needs a sigmoid) or C logits
(multiclass; needs a softmax) — see
:func:`spacr.deep_spacr.apply_model_to_tar`. Pushing a single-logit
column through a softmax yields a column of 1.0 and destroys the
ordering; pushing C logits through a sigmoid can *invert* it. Both are
silent failures that produce a confident-looking queue full of nonsense,
so :func:`probabilities_from_logits` branches on the shape and the tests
pin both directions.
``margin`` and ``least_confidence`` are the same ranking on two
classes. For C = 2, ``1 − (p₁ − p₂) = 2·min(p, 1−p) = 2·(1 − max p)``:
a linear transform, so identical order and identical ties. They are not
two independent choices on a binary screen — they differ only from three
classes up, where ``margin`` looks at the top *two* classes while
``least_confidence`` looks only at the top one and ``entropy`` looks at
the whole distribution.
**Pure uncertainty ranking collapses onto one region of feature space.**
The 100 most uncertain crops on a real plate are routinely 100 near
copies from the same two wells — one ambiguity, labelled a hundred
times, for almost the information of labelling it once. So the queue is
diversified by default (round-robin across wells) and
:func:`format_queue_summary` prints how many wells the queue actually
covers. What that costs is stated plainly in :func:`build_queue`.
**Uncertainty sampling skews toward the majority class.** On a screen
that is 98 % negative, the decision boundary is mostly populated by
negatives, so the queue will be too. That is not a bug to hide — it is
reported as a class-balance table so the annotator can see it and cap or
rebalance if they want to.
**Determinism.** Same inputs, same seed, same order — including ties.
Ties break on row index by default; ``seed`` swaps in a seeded
permutation, which is still reproducible. Nothing here consults an
unseeded RNG.
Nothing in this module imports torch at module scope; the ranking maths
is numpy only, so the Qt screen can build a queue without waking a
multi-second import chain. :func:`predict_probabilities` imports torch
lazily and is the only entry point that needs it.
"""
from __future__ import annotations
import os
import sqlite3
from collections import OrderedDict
from typing import (Any, Callable, Dict, Iterable, List, Mapping, Optional,
Sequence, Tuple)
from urllib.parse import quote as _urlquote
import numpy as np
import pandas as pd
from . import schema
from .agreement import PNG_KEY, PNG_TABLE
#: ``png_list``'s per-crop id column, and the object type it means.
#:
#: ``filepaths_to_database`` writes exactly one of these per row — the one for
#: the crop mode it was called with — so which column holds a label *is* the
#: crop's object type. Derived from :data:`spacr.schema.OBJECT_TYPES` rather
#: than written out, which keeps it identical to
#: :data:`spacr.utils.PNG_CROP_MODE_BY_ID_COLUMN` without importing
#: :mod:`spacr.utils` and its multi-second chain into this module (see the
#: module docstring). ``tests/test_active_learning_loop.py`` pins the two
#: together.
PNG_ID_COLUMN_TYPES: Dict[str, str] = {
f"{object_type}_id": object_type for object_type in schema.OBJECT_TYPES
}
__all__ = [
"CALIBRATION_NOTE",
"DEFAULT_MEASURE",
"DIVERSITY_GROUPS",
"PNG_ID_COLUMN_TYPES",
"PNG_KEY",
"PNG_TABLE",
"PRED_COLUMN_CANDIDATES",
"ROUND_LOG_TABLE",
"ROUND_PRED_PREFIX",
"ROUND_TABLE",
"UNCERTAINTY_MEASURES",
"RoundResult",
"StoppingVerdict",
"annotation_coverage",
"as_probabilities",
"build_queue",
"crops_for_object_keys",
"disagreement",
"ensure_round_tables",
"entropy",
"format_coverage_summary",
"format_learning_curve",
"format_queue_summary",
"holdout_report",
"label_rounds",
"learning_curve",
"least_confidence",
"margin",
"next_round",
"predict_probabilities",
"probabilities_from_logits",
"queue_rows",
"rank_by_uncertainty",
"record_labels",
"record_round",
"resolve_measure",
"retrain_round",
"round_features",
"should_stop",
"uncertainty_scores",
]
#: Printed under every queue summary, and the reason no function here
#: returns anything called a confidence.
CALIBRATION_NOTE = (
"Uncertainty scores rank crops; they are not calibrated probabilities. "
"A neural network's sigmoid/softmax output is systematically "
"over-confident (Guo et al., 2017), so a score of 0.5 does not mean "
"'a coin flip' and a 0.99 does not mean '99 % sure'. Read the ORDER, "
"not the numbers."
)
#: Default measure. Entropy uses the whole distribution rather than only
#: the top one or two classes, which is the safer default once C > 2 and
#: is equivalent-in-spirit at C = 2.
DEFAULT_MEASURE = "entropy"
#: Column names :func:`build_queue` looks for when ``pred_column`` is not
#: given, best first. ``pred`` is what the CV classifier writes via
#: :func:`spacr.predictions.merge_cv_predictions` (the positive-class
#: probability, one REAL column); ``ml_pred`` is the same quantity from the
#: classical-ML classifier (:func:`spacr.predictions.merge_ml_predictions`).
#: The CV score comes first because a database carrying both was scored by a
#: model trained on crops, which is what the queue is picking crops for.
PRED_COLUMN_CANDIDATES: Tuple[str, ...] = ("pred", "ml_pred", "prediction", "score")
#: Metadata columns each named diversity strategy stratifies over, in the
#: order they are combined into a group key.
DIVERSITY_GROUPS: Dict[str, Tuple[str, ...]] = {
"plate": ("plateID",),
"well": ("plateID", "rowID", "columnID"),
"row": ("plateID", "rowID"),
"column": ("plateID", "columnID"),
"field": ("plateID", "rowID", "columnID", "fieldID"),
}
#: Columns copied into the queue frame when ``png_list`` has them, so the
#: caller can see where each crop came from without a second query.
_METADATA_COLUMNS: Tuple[str, ...] = (
"plateID", "rowID", "columnID", "fieldID", "prc", "prcfo", "file_name",
"cell_id", "nucleus_id", "pathogen_id", "cytoplasm_id",
)
#: Prefix of the per-class probability columns :func:`retrain_round` writes
#: back into ``png_list``. One column per class (``al_prob_0``, ``al_prob_1``,
#: …) rather than one positive-class score, so a three-class screen re-ranks
#: on the full distribution rather than on a collapsed binary proxy.
ROUND_PRED_PREFIX = "al_prob_"
#: Per-label provenance: which round each annotation was made in.
ROUND_TABLE = "annotation_rounds"
#: Per-round record: the learning curve, one row per retrain.
ROUND_LOG_TABLE = "annotation_round_log"
def _to_numpy(values: Any) -> np.ndarray:
"""Return ``values`` as a float array, accepting torch tensors.
Tensors are converted by duck-typing (``detach``/``cpu``/``numpy``)
rather than by importing torch, so the ranking maths stays usable in
a process that has never loaded it.
"""
obj = values
if hasattr(obj, "detach"):
obj = obj.detach()
if hasattr(obj, "cpu"):
obj = obj.cpu()
if not isinstance(obj, np.ndarray) and hasattr(obj, "numpy"):
obj = obj.numpy()
return np.asarray(obj, dtype=float)
def _sigmoid(x: np.ndarray) -> np.ndarray:
"""Numerically stable logistic function."""
out = np.empty_like(x, dtype=float)
pos = x >= 0
out[pos] = 1.0 / (1.0 + np.exp(-x[pos]))
exp_x = np.exp(x[~pos])
out[~pos] = exp_x / (1.0 + exp_x)
return out
def _softmax(x: np.ndarray) -> np.ndarray:
"""Row-wise softmax, max-subtracted so large logits do not overflow."""
shifted = x - np.max(x, axis=1, keepdims=True)
exp = np.exp(shifted)
total = exp.sum(axis=1, keepdims=True)
with np.errstate(invalid="ignore", divide="ignore"):
return exp / total
def _as_matrix(values: Any) -> np.ndarray:
"""Coerce to a 2-D ``(N, C)`` float array.
``(N,)`` becomes ``(N, 1)``. A scalar or a 3-D array is a caller
error, not something to guess about.
"""
arr = _to_numpy(values)
if arr.ndim == 0:
raise ValueError(
"Uncertainty needs a per-crop score array, got a scalar. Pass "
"shape (N,) for a single-logit/binary head or (N, C) for a "
"C-class head.")
if arr.ndim == 1:
arr = arr.reshape(-1, 1)
if arr.ndim != 2:
raise ValueError(
f"Expected a (N,) or (N, C) array of scores, got shape "
f"{arr.shape}. For an ensemble of score sets use "
f"disagreement(), which takes a list of them.")
if arr.shape[1] == 0:
raise ValueError(
f"Got {arr.shape[0]} rows with no columns: there are no class "
f"scores to be uncertain about.")
return arr
[docs]
def probabilities_from_logits(logits: Any) -> np.ndarray:
"""Convert raw classifier-head outputs to an ``(N, C)`` probability matrix.
The head shape decides the link function, and getting this wrong is
silent:
* ``(N,)`` or ``(N, 1)`` — a **single-logit binary head**. Sigmoid,
then expanded to ``[1 − p, p]`` so every measure sees two classes.
Pushing this through a softmax instead would return a column of
1.0 — every crop maximally certain, the whole ordering gone.
* ``(N, C)`` with C ≥ 2 — a **C-logit head**. Row-wise softmax.
Sigmoiding these instead reads each logit in isolation and can
invert the order: ``[5, 5]`` is a perfect 50/50 tie under softmax
but looks like a confident 0.993 under a sigmoid of column 1.
Both branches are exactly what :func:`spacr.deep_spacr.apply_model_to_tar`
and :func:`spacr.deep_spacr.evaluate_model_performance` do at
inference time.
:param logits: array-like or torch tensor of raw head outputs.
:returns: ``(N, C)`` float array whose rows sum to 1 (C ≥ 2).
:raises ValueError: for a scalar or a 3-D input.
.. note::
The output is a probability *vector*, not a calibrated
probability. See :data:`CALIBRATION_NOTE`.
"""
arr = _as_matrix(logits)
if arr.shape[1] == 1:
p = _sigmoid(arr[:, 0])
return np.column_stack([1.0 - p, p])
return _softmax(arr)
def _coerce_probabilities(scores: Any) -> Tuple[np.ndarray, List[str]]:
"""Normalise stored probabilities and report what had to be fixed.
:returns: ``(probs, notes)``. Rows that cannot be read as a
distribution come back as all-NaN, which every measure turns
into a NaN score and :func:`build_queue` drops with a count.
"""
arr = _as_matrix(scores)
notes: List[str] = []
n, c = arr.shape
if n == 0:
return np.zeros((0, max(c, 2)), dtype=float), notes
if c == 1:
p = arr[:, 0]
out_of_range = np.isfinite(p) & ((p < 0.0) | (p > 1.0))
if out_of_range.any():
lo, hi = float(np.nanmin(p)), float(np.nanmax(p))
notes.append(
f"{int(out_of_range.sum())} of {n} values in the prediction "
f"column fall outside [0, 1] (range {lo:.4g} … {hi:.4g}): "
f"that column holds logits, not probabilities. Those rows "
f"are dropped — score them with probabilities_from_logits() "
f"instead, or re-run the merge step.")
p = np.where(out_of_range, np.nan, p)
probs = np.column_stack([1.0 - p, p])
return probs, notes
probs = arr.astype(float, copy=True)
negative = np.isfinite(probs) & (probs < 0.0)
if negative.any():
rows = np.unique(np.nonzero(negative)[0])
notes.append(
f"{len(rows)} of {n} rows contain negative values, which no "
f"probability vector can: those columns are logits or a broken "
f"merge. Those rows are dropped.")
probs[rows, :] = np.nan
totals = probs.sum(axis=1)
valid = np.isfinite(totals)
zero = valid & np.isclose(totals, 0.0)
if zero.any():
notes.append(
f"{int(zero.sum())} of {n} rows sum to 0 and carry no "
f"distribution at all; they are dropped.")
probs[zero, :] = np.nan
totals = probs.sum(axis=1)
valid = np.isfinite(totals)
off = valid & ~np.isclose(totals, 1.0, atol=1e-6)
if off.any():
worst = float(np.nanmax(np.abs(totals[off] - 1.0)))
notes.append(
f"{int(off.sum())} of {n} probability rows did not sum to 1 "
f"(largest deviation {worst:.4g}); they were renormalised. "
f"Uncertainty is only meaningful on a normalised distribution, "
f"so this is a fix, not a warning to ignore — check the column "
f"really holds per-class probabilities.")
with np.errstate(invalid="ignore", divide="ignore"):
probs = np.where(valid[:, None], probs / totals[:, None], np.nan)
return probs, notes
[docs]
def as_probabilities(scores: Any) -> np.ndarray:
"""Read stored scores as an ``(N, C)`` probability matrix.
Unlike :func:`probabilities_from_logits` this assumes the values are
*already* probabilities — which is what the ``pred`` column of
``png_list`` holds, because
:func:`spacr.deep_spacr.apply_model_to_tar` applied the sigmoid or
softmax before writing it.
* a single column is read as the positive-class probability of a
binary problem and expanded to ``[1 − p, p]``;
* ``(N, C)`` rows are renormalised if they do not sum to 1;
* rows that cannot be a distribution (negatives, all-zero, values
outside [0, 1] in the single-column case) become all-NaN, so they
score NaN and get excluded rather than silently mis-ranked.
:param scores: array-like of stored probabilities.
:returns: ``(N, C)`` float array.
"""
return _coerce_probabilities(scores)[0]
[docs]
def least_confidence(probs: Any, normalize: bool = False) -> np.ndarray:
"""``1 − max_c p_c`` — how much probability mass is *not* on the winner.
Minimum 0 at a one-hot row; maximum ``1 − 1/C`` at the uniform row.
It looks only at the top class, so on 3+ classes it cannot tell
``[0.5, 0.5, 0.0]`` from ``[0.5, 0.25, 0.25]``; :func:`entropy` can.
:param probs: ``(N,)`` positive-class probabilities or ``(N, C)``
rows. Coerced with :func:`as_probabilities`.
:param normalize: rescale by ``C / (C − 1)`` so the maximum is 1.
A monotone rescaling — it never changes the ranking, only how
the number reads.
:returns: ``(N,)`` uncertainty scores; NaN for unusable rows.
"""
p, _ = _coerce_probabilities(probs)
if p.shape[0] == 0:
return np.zeros(0, dtype=float)
bad = np.all(~np.isfinite(p), axis=1)
score = 1.0 - np.max(np.where(np.isfinite(p), p, -np.inf), axis=1)
score = np.where(bad, np.nan, score)
if normalize and p.shape[1] > 1:
score = score * (p.shape[1] / (p.shape[1] - 1.0))
return score
[docs]
def margin(probs: Any) -> np.ndarray:
"""``1 − (p₁ − p₂)`` — closeness of the top two classes, as *uncertainty*.
This returns an uncertainty score, **not** the margin: a small margin
(the two leading classes neck and neck) is a large return value, so
it is oriented like every other measure here. Minimum 0 at a one-hot
row, maximum 1 at any row whose top two classes tie.
On **two classes this is a linear transform of**
:func:`least_confidence` — ``1 − (p₁ − p₂) = 2·(1 − max p)`` — so the
two produce the identical order and the identical ties. They are one
choice, not two, until C ≥ 3, where margin ignores everything below
the runner-up and least-confidence ignores everything below the
winner.
:param probs: ``(N,)`` or ``(N, C)``; coerced with :func:`as_probabilities`.
:returns: ``(N,)`` uncertainty scores in [0, 1]; NaN for unusable rows.
"""
p, _ = _coerce_probabilities(probs)
if p.shape[0] == 0:
return np.zeros(0, dtype=float)
bad = np.all(~np.isfinite(p), axis=1)
filled = np.sort(np.where(np.isfinite(p), p, -np.inf), axis=1)
gap = np.zeros(p.shape[0], dtype=float)
np.subtract(
filled[:, -1],
filled[:, -2],
out=gap,
where=~bad,
)
return np.where(bad, np.nan, 1.0 - gap)
[docs]
def entropy(probs: Any, base: Optional[float] = None,
normalize: bool = False) -> np.ndarray:
"""Shannon entropy ``−Σ p log p`` of each row.
The only measure here that uses the whole distribution. Minimum 0 at
a one-hot row; maximum ``log C`` at the uniform row (``log 2 ≈
0.6931`` for two classes in nats). ``0 · log 0`` is taken as 0.
:param probs: ``(N,)`` or ``(N, C)``; coerced with :func:`as_probabilities`.
:param base: logarithm base. ``None`` (default) means natural log,
so the units are nats; pass ``2`` for bits.
:param normalize: divide by ``log C`` so the maximum is 1. Monotone,
so the ranking is unchanged.
:returns: ``(N,)`` uncertainty scores; NaN for unusable rows.
"""
p, _ = _coerce_probabilities(probs)
if p.shape[0] == 0:
return np.zeros(0, dtype=float)
bad = np.all(~np.isfinite(p), axis=1)
safe = np.where(np.isfinite(p) & (p > 0.0), p, 1.0)
with np.errstate(divide="ignore", invalid="ignore"):
terms = np.where(np.isfinite(p) & (p > 0.0), -safe * np.log(safe), 0.0)
score = terms.sum(axis=1)
n_classes = p.shape[1]
if base is not None:
if base <= 0 or base == 1:
raise ValueError(f"Log base must be positive and != 1, got {base!r}.")
score = score / np.log(base)
if normalize and n_classes > 1:
denom = np.log(n_classes)
if base is not None:
denom = denom / np.log(base)
score = score / denom
return np.where(bad, np.nan, score)
[docs]
def disagreement(prob_sets: Any, method: str = "variance") -> np.ndarray:
"""How much several score sets disagree about each crop.
The measures above see one model's opinion and call a 50/50 output
"uncertain" whether the model is genuinely torn or merely
mis-calibrated. An ensemble — several checkpoints, several folds, or
several MC-dropout passes of one model — separates those: crops the
members *disagree* about are where the model class itself is
undecided (epistemic uncertainty), which is what a new label
actually fixes.
:param prob_sets: sequence of M score arrays, each ``(N,)`` or
``(N, C)`` over the SAME N crops in the same order (a 3-D
``(M, N, C)`` array works too). Each member is coerced with
:func:`as_probabilities`.
:param method:
* ``'variance'`` (default) — mean across classes of the
across-member variance (population variance, ddof=0). 0 when
the members agree exactly.
* ``'bald'`` — mutual information ``H(mean p) − mean H(p)``
(Houlsby et al., 2011). 0 when the members agree, regardless
of how uncertain they jointly are — so unlike ``entropy`` it
does not fire on crops that are ambiguous.
:returns: ``(N,)`` scores; NaN where any member is unusable for that
crop.
:raises ValueError: for fewer than one set, or sets of different
shapes — a length mismatch means the members are not aligned to
the same crops, and averaging them would be meaningless.
A single set returns all zeros: one opinion cannot disagree with
itself. That is a real answer, not a failure, but it means the queue
would be in row order, so check the count before using it.
"""
if isinstance(prob_sets, np.ndarray) and prob_sets.ndim == 3:
members = [prob_sets[i] for i in range(prob_sets.shape[0])]
elif isinstance(prob_sets, (list, tuple)):
members = list(prob_sets)
else:
arr = _to_numpy(prob_sets)
if arr.ndim == 3:
members = [arr[i] for i in range(arr.shape[0])]
else:
members = [arr]
if not members:
raise ValueError(
"disagreement() needs at least one set of scores; got none.")
coerced = [_coerce_probabilities(m)[0] for m in members]
shapes = {m.shape for m in coerced}
if len(shapes) != 1:
raise ValueError(
f"Every member must score the same crops in the same order; got "
f"shapes {sorted(shapes)}. Differing row counts mean the members "
f"are not aligned and the comparison would be meaningless.")
stack = np.stack(coerced, axis=0)
bad = ~np.all(np.isfinite(stack), axis=(0, 2))
if stack.shape[0] == 1:
return np.where(bad, np.nan, np.zeros(stack.shape[1]))
if method == "variance":
score = np.var(stack, axis=0).mean(axis=1)
elif method == "bald":
mean = stack.mean(axis=0)
h_mean = entropy(mean)
h_each = np.stack([entropy(m) for m in coerced], axis=0)
score = h_mean - np.nanmean(h_each, axis=0)
score = np.maximum(score, 0.0)
else:
raise ValueError(
f"Unknown disagreement method {method!r}; use 'variance' or 'bald'.")
return np.where(bad, np.nan, score)
#: The single-model measures, by name. All are ``f(probs) -> (N,)`` with
#: larger meaning less certain, so they are interchangeable everywhere a
#: ``measure=`` argument is taken.
#:
#: :func:`disagreement` is deliberately NOT in here: it takes a *list* of
#: score sets, not one, so it cannot be swapped in behind the same
#: argument without silently mis-reading its input.
UNCERTAINTY_MEASURES: Dict[str, Callable[..., np.ndarray]] = {
"least_confidence": least_confidence,
"margin": margin,
"entropy": entropy,
}
[docs]
def resolve_measure(measure: Any) -> Tuple[str, Callable[..., np.ndarray]]:
"""Turn a measure name (or callable) into ``(name, function)``.
:param measure: a key of :data:`UNCERTAINTY_MEASURES`, or any
callable ``f(probs) -> (N,)``.
:raises ValueError: for an unknown name, listing the valid ones.
"""
if callable(measure):
return getattr(measure, "__name__", "custom"), measure
name = str(measure)
if name not in UNCERTAINTY_MEASURES:
raise ValueError(
f"Unknown uncertainty measure {measure!r}. Available: "
f"{', '.join(sorted(UNCERTAINTY_MEASURES))}. (On two classes "
f"'margin' and 'least_confidence' give the same order.)")
return name, UNCERTAINTY_MEASURES[name]
[docs]
def uncertainty_scores(probs: Any, measure: Any = DEFAULT_MEASURE) -> np.ndarray:
"""Score every row with ``measure``.
:param probs: ``(N,)`` or ``(N, C)`` probabilities.
:param measure: name from :data:`UNCERTAINTY_MEASURES` or a callable.
:returns: ``(N,)`` uncertainty scores, larger = less certain.
"""
_, fn = resolve_measure(measure)
return np.asarray(fn(probs), dtype=float).reshape(-1)
[docs]
def rank_by_uncertainty(probs: Any, measure: Any = DEFAULT_MEASURE,
limit: Optional[int] = None,
seed: Optional[int] = None,
scores: Optional[Any] = None) -> np.ndarray:
"""Row indices ordered most-uncertain-first.
Ordering is total and reproducible:
* primary key — the uncertainty score, descending;
* ties — row index ascending by default, or a permutation seeded
with ``seed`` when one is given. Both are deterministic: the same
inputs and the same seed always give the same order. Ties are the
normal case on a screen where thousands of crops score exactly
0.5, so an unseeded shuffle there would reshuffle the annotator's
queue on every refresh.
* NaN scores sort last, always, and never in front of a real one.
They are ranked rather than dropped so the returned indices stay a
permutation of ``range(N)``; :func:`build_queue` drops them and
says how many.
:param probs: ``(N,)`` or ``(N, C)`` probabilities.
:param measure: name from :data:`UNCERTAINTY_MEASURES` or a callable.
:param limit: keep only the first ``limit`` indices.
:param seed: seed for tie-breaking; ``None`` breaks ties on index.
:param scores: pre-computed scores to rank instead of recomputing
from ``probs`` (used by :func:`build_queue`, and by anything
ranking a :func:`disagreement` score).
:returns: ``(N,)`` (or ``(limit,)``) int array of row indices.
"""
if scores is None:
values = uncertainty_scores(probs, measure)
else:
values = np.asarray(scores, dtype=float).reshape(-1)
n = values.shape[0]
if n == 0:
return np.zeros(0, dtype=int)
if seed is None:
tie = np.arange(n)
else:
tie = np.random.default_rng(seed).permutation(n)
finite = np.isfinite(values)
primary = np.where(finite, -values, np.inf)
order = np.lexsort((tie, primary))
if limit is not None:
order = order[:max(0, int(limit))]
return order.astype(int)
[docs]
def predict_probabilities(model: Callable[[Any], Any], batches: Iterable[Any],
device: Any = None,
from_logits: bool = True) -> np.ndarray:
"""Run ``model`` over ``batches`` and return an ``(N, C)`` probability matrix.
A convenience for scoring crops that are not in the database yet.
The queue does **not** need this — :func:`build_queue` works from the
``pred`` column that :func:`spacr.deep_spacr.merge_predictions_into_db`
already wrote, which is the normal path.
torch is imported inside this function, and only to get
``no_grad``/``device`` handling; if it is not importable the batches
are iterated and the model is called directly. Nothing else in
this module touches torch.
:param model: any callable mapping a batch to raw head outputs. A
``torch.nn.Module`` is put in ``eval()`` mode first if it has one.
:param batches: iterable of batches. A batch that is a ``(inputs,
…)`` tuple has its first element passed to the model, matching
the loaders in :mod:`spacr.deep_spacr`.
:param device: optional torch device to move inputs/model to.
:param from_logits: treat outputs as raw logits and apply
:func:`probabilities_from_logits` (the default — a model head
emits logits). Set False if the model already outputs
probabilities, which then go through :func:`as_probabilities`.
:returns: ``(N, C)`` probability matrix in batch order.
"""
try:
import torch
except Exception:
torch = None
if torch is not None and hasattr(model, "eval"):
model.eval()
if device is not None and hasattr(model, "to"):
model = model.to(device)
outputs: List[np.ndarray] = []
def _run() -> None:
"""Score captured batches in order and append normalized matrices.
Loader tuples contribute their input element, and movable inputs are
transferred to the selected device before the model is called.
"""
for batch in batches:
inputs = batch[0] if isinstance(batch, (tuple, list)) else batch
if (torch is not None and device is not None
and hasattr(inputs, "to")):
inputs = inputs.to(device)
outputs.append(_as_matrix(model(inputs)))
if torch is not None:
with torch.no_grad():
_run()
else:
_run()
if not outputs:
return np.zeros((0, 2), dtype=float)
widths = {o.shape[1] for o in outputs}
if len(widths) != 1:
raise ValueError(
f"Batches produced different head widths {sorted(widths)}; the "
f"model cannot be both a single-logit binary head and a "
f"multiclass one.")
raw = np.concatenate(outputs, axis=0)
return probabilities_from_logits(raw) if from_logits else as_probabilities(raw)
def _quote_ident(name: str) -> str:
"""Double-quote a SQL identifier (already schema-validated)."""
return '"' + str(name).replace('"', '""') + '"'
def _read_only_uri(path: str) -> str:
"""``file:…?mode=ro`` URI — SQLite itself then refuses every write."""
return "file:" + _urlquote(str(path).replace("\\", "/"), safe="/:") + "?mode=ro"
def _connect(db_path: str) -> sqlite3.Connection:
"""Open ``db_path`` read-only. Building a queue never writes.
:raises FileNotFoundError: when the file is not there — sqlite's own
"unable to open database file" says nothing about which file.
"""
if not db_path or not str(db_path).strip():
raise ValueError("No database path given.")
path = os.path.abspath(os.path.expanduser(str(db_path).strip()))
if not os.path.isfile(path):
raise FileNotFoundError(f"No such database: {path}")
from .database_concurrency import connect as _connect_database
con = _connect_database(path, readonly=True)
con.execute("PRAGMA query_only = ON")
return con
def _table_columns(con: sqlite3.Connection, table: str,
db_path: str) -> List[str]:
"""Column names of ``table``, in declaration order.
:raises ValueError: when the database has no such table — sqlite's
own "no such table" does not say what the file *does* contain.
"""
names = [r[0] for r in con.execute(
"SELECT name FROM sqlite_master WHERE type IN ('table','view')"
).fetchall()]
if table not in names:
raise ValueError(
f"{os.path.basename(str(db_path))} has no {table!r} table "
f"(found: {', '.join(names) or 'nothing'}). Crops and their "
f"predictions live in {PNG_TABLE!r}; run Measure with "
f"save_png=True first.")
return [r[1] for r in con.execute(
f"PRAGMA table_info({_quote_ident(table)})").fetchall()]
def _resolve_pred_columns(pred_column: Any, columns: Sequence[str],
table: str, db_path: str) -> List[str]:
"""Work out which column(s) hold the model's scores.
:raises ValueError: when none can be found, naming what to run.
"""
available = list(columns)
if pred_column is not None:
wanted = ([pred_column] if isinstance(pred_column, str)
else list(pred_column))
missing = [c for c in wanted if c not in available]
if missing:
raise ValueError(
f"{table!r} has no column(s) {', '.join(map(repr, missing))}. "
f"Available: {', '.join(available)}.")
if not wanted:
raise ValueError("pred_column was an empty list.")
return [str(c) for c in wanted]
for prefix in (ROUND_PRED_PREFIX, "pred_", "prob_", "score_"):
numbered = sorted(
(c for c in available
if c.startswith(prefix) and c[len(prefix):].isdigit()),
key=lambda c, p=prefix: int(c[len(p):]))
if len(numbered) >= 2:
return numbered
for candidate in PRED_COLUMN_CANDIDATES:
if candidate in available:
return [candidate]
raise ValueError(
f"{os.path.basename(str(db_path))} [{table}] has no prediction "
f"column — there is nothing to be uncertain about, so no queue can "
f"be built. Columns present: {', '.join(available) or 'none'}. Run a "
f"model over the crops first (spacr.deep_spacr.deep_spacr with "
f"apply_model_to_dataset=True, which writes 'pred' via "
f"merge_predictions_into_db), or pass pred_column= if the scores "
f"live under another name.")
def _group_columns_for(diversity: Any, columns: Sequence[str],
group_columns: Optional[Sequence[str]],
notes: List[str]) -> Tuple[str, List[str]]:
"""Resolve the diversity strategy to concrete, present columns.
Falls back to ``'none'`` with a note when the metadata is not there —
a missing plate map is a reason to stop diversifying, not to crash.
"""
if group_columns:
wanted = [str(c) for c in group_columns]
name = "custom"
elif diversity in (None, False, "none", "off"):
return "none", []
else:
key = "well" if diversity in (True, "auto") else str(diversity)
if key not in DIVERSITY_GROUPS:
raise ValueError(
f"Unknown diversity strategy {diversity!r}; use one of "
f"{', '.join(sorted(DIVERSITY_GROUPS))}, 'none', or pass "
f"group_columns=[…] explicitly.")
wanted, name = list(DIVERSITY_GROUPS[key]), key
present = [c for c in wanted if c in columns]
if not present:
notes.append(
f"Diversity was requested over {', '.join(wanted)}, but "
f"png_list has none of those columns, so the queue is ordered by "
f"pure uncertainty. Expect it to cluster: the most uncertain "
f"crops usually come from a handful of wells.")
return "none", []
if len(present) < len(wanted):
notes.append(
f"Diversity over {', '.join(wanted)} fell back to "
f"{', '.join(present)} — the rest are not in png_list.")
return name, present
def _group_key(frame: pd.DataFrame, cols: Sequence[str]) -> np.ndarray:
"""One string key per row, joining ``cols``. Empty-frame safe."""
if not len(frame):
return np.zeros(0, dtype=object)
return frame[list(cols)].astype(str).apply("_".join, axis=1).to_numpy()
def _round_robin(order: np.ndarray, group_keys: np.ndarray) -> np.ndarray:
"""Spread a ranked index array across groups, one per group per pass.
Groups are visited in order of their most uncertain member, so
position 1 of the queue is still the single most uncertain crop
overall; position 2 is the most uncertain crop in the *next* group,
and so on. Deterministic: no RNG, and the group order is fixed by
the (already deterministic) input ranking.
"""
buckets: "OrderedDict[Any, List[int]]" = OrderedDict()
for idx in order:
buckets.setdefault(group_keys[idx], []).append(int(idx))
out: List[int] = []
while buckets:
for key in list(buckets):
out.append(buckets[key].pop(0))
if not buckets[key]:
del buckets[key]
return np.asarray(out, dtype=int)
def _class_balance(labels: Sequence[Any]) -> Dict[Any, int]:
"""Counts per class as a plain, JSON-friendly dict, sorted by class.
A class read out of a REAL column arrives as ``1.0``; it is reported
as ``1``, because "class 1.0" is not a thing anyone annotated.
"""
series = pd.Series(list(labels)).dropna()
if series.empty:
return {}
counts = series.value_counts()
out: Dict[Any, int] = {}
for key in sorted(counts.index, key=lambda v: (str(type(v)), v)):
value = key
if isinstance(value, float) and float(value).is_integer():
value = int(value)
out[value] = int(counts[key])
return out
[docs]
def build_queue(db_path: str, annotation_column: str = "annotate",
pred_column: Any = None, table: str = PNG_TABLE,
key: str = PNG_KEY, measure: Any = DEFAULT_MEASURE,
limit: Optional[int] = None, diversity: Any = "well",
group_columns: Optional[Sequence[str]] = None,
seed: Optional[int] = None,
image_type: Optional[str] = None) -> pd.DataFrame:
"""Build the annotation queue: unlabelled crops, most uncertain first.
Reads ``png_list`` read-only and returns one row per crop still
waiting for a label, ordered so the crops that would teach the model
most come first.
**Already-annotated crops are excluded.** ``NULL`` in
``annotation_column`` means "not looked at"; anything else — *including
0* — means a human committed to a class. If the column does not exist
at all, nothing has been annotated yet and every crop is queued (with
a note saying so). This is the same abstention convention as
:mod:`spacr.agreement`.
**The queue is diversified by default, and that costs something.**
With ``diversity='well'`` the ranked crops are dealt round-robin
across wells: position 1 is still the single most uncertain crop, but
position 2 is the most uncertain crop in a *different* well, which may
be materially less uncertain than the runner-up overall. You give up
some per-item uncertainty to stop the annotator labelling the same
ambiguity a hundred times — the failure mode of pure uncertainty
sampling, where the top 100 crops routinely come from two wells. If
``limit`` is smaller than the number of wells, the queue will contain
roughly one crop from each of ``limit`` wells and none from the rest.
Pass ``diversity='none'`` for the pure order, ``'field'``/``'plate'``
for other strata, or ``group_columns=[…]`` for your own — including a
cluster id you computed from features, which is the more thorough
diversification this trades away for not needing a feature matrix.
:param db_path: path to ``measurements.db``.
:param annotation_column: column the Annotate app writes into
(default ``'annotate'``).
:param pred_column: name, or list of names, of the model-score
column(s). ``None`` (default) auto-detects: ``pred_0, pred_1, …``
style columns for multiclass, else ``pred``. A single column is
read as the positive-class probability of a binary problem.
:param table: table holding the crops (default ``png_list``).
:param key: row key (default ``png_path``).
:param measure: ``'entropy'`` (default), ``'least_confidence'``,
``'margin'``, or a callable. On two classes the last two give the
same order.
:param limit: keep at most this many crops.
:param diversity: ``'well'`` (default), ``'plate'``, ``'row'``,
``'column'``, ``'field'``, or ``'none'``.
:param group_columns: explicit columns to stratify over, overriding
``diversity``.
:param seed: tie-breaking seed; ``None`` breaks ties on row order.
Either way the result is reproducible.
:param image_type: substring filter on ``png_path`` — matches the
Annotate screen's own ``image_type`` filter (e.g. ``'cell'``).
:returns: DataFrame with ``rank`` (1-based), the key, ``uncertainty``,
``predicted_class``, the probability columns and whatever crop
metadata ``png_list`` has. Empty (with the same columns) when
there is nothing to annotate. Diagnostics live in
``queue.attrs['spacr_active_learning']``; render them with
:func:`format_queue_summary`.
:raises ValueError: when the table or the prediction column is
missing — both mean there is no queue to build, and guessing
would produce a plausible-looking wrong order.
:raises FileNotFoundError: when the database is not there.
.. warning::
``uncertainty`` is a ranking score, not a calibrated confidence.
See :data:`CALIBRATION_NOTE`.
"""
notes: List[str] = []
con = _connect(db_path)
try:
columns = _table_columns(con, table, db_path)
if key not in columns:
raise ValueError(
f"{table!r} has no {key!r} column — there is no way to say "
f"which crop a queue entry refers to.")
pred_cols = _resolve_pred_columns(pred_column, columns, table, db_path)
measure_name, measure_fn = resolve_measure(measure)
has_annotation = annotation_column in columns
if not has_annotation:
notes.append(
f"png_list has no {annotation_column!r} column yet, so nothing "
f"has been annotated: every scored crop is in the queue. The "
f"Annotate app creates the column the first time it saves.")
meta_cols = [c for c in _METADATA_COLUMNS if c in columns]
div_name, div_cols = _group_columns_for(diversity, columns,
group_columns, notes)
for c in div_cols:
if c not in meta_cols:
meta_cols.append(c)
spread_cols = div_cols or [c for c in DIVERSITY_GROUPS["well"]
if c in columns]
select_cols = [key] + ([annotation_column] if has_annotation else []) \
+ pred_cols + meta_cols
select_cols = list(dict.fromkeys(select_cols))
rows = con.execute(
f"SELECT {', '.join(_quote_ident(c) for c in select_cols)} "
f"FROM {_quote_ident(table)}").fetchall()
finally:
con.close()
frame = pd.DataFrame(rows, columns=select_cols)
n_rows = len(frame)
if image_type:
keep = frame[key].astype(str).str.contains(str(image_type), regex=False)
n_filtered = int((~keep).sum())
frame = frame[keep]
if n_filtered:
notes.append(
f"image_type={image_type!r} excluded {n_filtered} of {n_rows} "
f"crops before scoring.")
if has_annotation:
labelled_mask = frame[annotation_column].notna()
else:
labelled_mask = pd.Series(False, index=frame.index)
labelled_balance = _class_balance(
frame.loc[labelled_mask, annotation_column].tolist()
if has_annotation else [])
n_annotated = int(labelled_mask.sum())
pool = frame.loc[~labelled_mask].reset_index(drop=True)
n_unlabelled = len(pool)
raw = pool[pred_cols].to_numpy(dtype=float) if n_unlabelled else \
np.zeros((0, len(pred_cols)))
probs, prob_notes = _coerce_probabilities(raw)
notes.extend(prob_notes)
scores = np.asarray(measure_fn(probs), dtype=float).reshape(-1) \
if n_unlabelled else np.zeros(0)
usable = np.isfinite(scores)
n_dropped = int((~usable).sum())
if n_dropped:
notes.append(
f"{n_dropped} of {n_unlabelled} unlabelled crops have no usable "
f"score (NULL or NaN in {', '.join(pred_cols)}) and were left out "
f"of the queue. A crop the model never scored cannot be ranked by "
f"how unsure the model is about it — score them, or annotate them "
f"in the normal page order.")
pool = pool.loc[usable].reset_index(drop=True)
probs = probs[usable]
scores = scores[usable]
n_scored = len(pool)
predicted = (np.argmax(probs, axis=1).astype(int) if n_scored
else np.zeros(0, dtype=int))
pool_balance = _class_balance(predicted.tolist())
if n_scored > 1 and np.ptp(scores) == 0.0:
notes.append(
f"Every unlabelled crop scored exactly {float(scores[0]):.6g}, so "
f"the {measure_name} ordering carries no information from the "
f"model — the queue below is just row order. Check that the "
f"prediction column was written by a model that actually saw "
f"these crops.")
if probs.shape[1] == 2 and n_scored:
distinct = np.unique(np.round(probs[:, 1], 12))
if distinct.size == 1:
notes.append(
"The prediction column holds a single distinct value: the "
"model put every crop in the same place, which is a "
"one-class output, not a ranking.")
order = rank_by_uncertainty(None, measure=measure_fn, seed=seed,
scores=scores)
pool_keys = _group_key(pool, div_cols) if div_cols else np.zeros(0)
if div_cols and n_scored:
order = _round_robin(order, pool_keys)
if limit is not None:
order = order[:max(0, int(limit))]
out = pool.iloc[order].reset_index(drop=True)
out.insert(0, "rank", np.arange(1, len(out) + 1))
out["uncertainty"] = scores[order]
out["predicted_class"] = predicted[order]
if has_annotation and annotation_column in out.columns:
out = out.drop(columns=[annotation_column])
queue_balance = _class_balance(out["predicted_class"].tolist())
spread_keys = pool_keys if div_cols else (
_group_key(pool, spread_cols) if spread_cols else np.zeros(0))
n_groups_pool = int(pd.Series(spread_keys).nunique()) if spread_cols else 0
n_groups_queue = (int(pd.Series(spread_keys[order]).nunique())
if spread_cols else 0)
if not len(out):
if n_rows and n_annotated == n_rows:
notes.append(
f"The queue is empty: all {n_rows} crops in {table} are "
f"already annotated in {annotation_column!r}. There is "
f"nothing left to label — start a new annotation column, or "
f"retrain on what you have.")
elif not n_rows:
notes.append(
f"{table} is empty — no crops have been written to the "
f"database. Run Measure with save_png=True first.")
elif not n_scored:
notes.append(
"The queue is empty: no unlabelled crop has a usable score.")
out.attrs["spacr_active_learning"] = {
"db_path": str(db_path),
"table": table,
"key": key,
"annotation_column": annotation_column,
"annotation_column_present": bool(has_annotation),
"pred_columns": list(pred_cols),
"n_classes": int(probs.shape[1]) if probs.size else 0,
"measure": measure_name,
"diversity": div_name,
"diversity_columns": list(div_cols),
"spread_columns": list(spread_cols),
"seed": seed,
"limit": limit,
"image_type": image_type,
"n_rows": int(n_rows),
"n_annotated": int(n_annotated),
"n_unlabelled": int(n_unlabelled),
"n_scored": int(n_scored),
"n_unscorable": int(n_dropped),
"n_queued": int(len(out)),
"n_groups_pool": n_groups_pool,
"n_groups_queued": n_groups_queue,
"queue_class_balance": queue_balance,
"pool_class_balance": pool_balance,
"labelled_class_balance": labelled_balance,
"notes": notes,
"calibration": CALIBRATION_NOTE,
}
return out
[docs]
def queue_rows(queue: pd.DataFrame,
key: str = PNG_KEY) -> List[Tuple[str, Optional[int]]]:
"""The queue as ``[(png_path, None), …]``.
Exactly the shape :func:`spacr.qt.annotate_engine.fetch_page` and
:func:`spacr.qt.annotate_engine.fetch_filtered_paths` return, so the
Annotate screen can page through a queue with no other change. The
annotation is always ``None`` — every crop in a queue is unlabelled
by construction.
:param queue: frame from :func:`build_queue`.
:param key: the path column (default ``png_path``).
:returns: list of ``(path, None)`` tuples in queue order.
"""
if key not in queue.columns:
raise ValueError(
f"Queue has no {key!r} column; got {', '.join(queue.columns)}.")
return [(str(p), None) for p in queue[key].tolist()]
def _share(count: int, total: int) -> str:
"""``count`` as a percent of ``total``, or ``'—'`` when there is no total."""
if not total:
return "—"
return f"{100.0 * count / total:5.1f}%"
def _well_key(frame: pd.DataFrame, columns: Sequence[str]) -> pd.Series:
"""Readable ``plate/row/column`` key per row, or ``'(unknown)'``."""
present = [c for c in columns if c in frame.columns]
if not present or not len(frame):
return pd.Series(["(unknown)"] * len(frame), index=frame.index,
dtype=object)
return frame[present].astype(str).apply("/".join, axis=1)
def _concentration(counts: "pd.Series") -> Dict[str, Any]:
"""How lopsided a count distribution is, in three numbers.
:param counts: labels per group, any order.
:returns: ``n``, ``n_groups``, ``top``, ``top_n``, ``top_share`` and
``hhi`` — the Herfindahl index, the sum of squared shares. ``hhi``
is 1.0 when every label came from one group and ``1/k`` when they
are spread evenly over ``k``; it is the single number that says
"these 200 labels are really 1.06 wells' worth" without needing the
whole table.
"""
total = int(counts.sum())
if not total:
return {"n": 0, "n_groups": 0, "top": None, "top_n": 0,
"top_share": 0.0, "hhi": 0.0, "effective_groups": 0.0}
ordered = counts.sort_values(ascending=False)
shares = (ordered / total).to_numpy(dtype=float)
hhi = float(np.sum(shares ** 2))
return {
"n": total,
"n_groups": int((ordered > 0).sum()),
"top": str(ordered.index[0]),
"top_n": int(ordered.iloc[0]),
"top_share": float(ordered.iloc[0]) / total,
"hhi": hhi,
"effective_groups": (1.0 / hhi) if hhi else 0.0,
}
[docs]
def annotation_coverage(db_path: str, annotation_column: str = "annotate",
table: str = PNG_TABLE, key: str = PNG_KEY,
image_type: Optional[str] = None) -> pd.DataFrame:
"""Summarize annotation coverage by class, well, plate and acquisition.
The distribution of labels across experimental units determines whether
a classifier can generalize beyond acquisition-specific staining, focus
and confluency. A label set concentrated within one well can therefore
yield optimistic performance under an object-level random split. This
function exposes such concentration before model training or evaluation.
Reads ``png_list`` read-only, plus :data:`ROUND_TABLE` when it is there,
so labels can also be attributed to the active-learning round that
surfaced them.
:param db_path: path to ``measurements.db``.
:param annotation_column: the column the Annotate app writes into.
:param table: crop table (default ``png_list``).
:param key: row key (default ``png_path``).
:param image_type: substring filter on the key, matching the Annotate
screen's own filter. Every count below it, ``n_rows`` included, is
over the crops that matched — a denominator taken from the whole
table would put the numerator and the denominator on two different
populations. ``n_rows_unfiltered`` keeps the total, and a note says
how many were excluded.
:returns: one row per ``(plateID, rowID, columnID, class)`` that has at
least one annotation, with ``n`` and ``share`` — plus the whole
breakdown in ``attrs['spacr_annotation_coverage']``:
``by_class``, ``by_plate``, ``by_well``, ``by_class_plate``,
``by_class_well``, ``by_round``, ``by_source``, ``concentration``
(per class) and ``notes``.
:raises ValueError: when the table has no such column — an empty result
would read as "nothing annotated yet", which is a different fact.
:raises FileNotFoundError: when the database is not there.
"""
notes: List[str] = []
con = _connect(db_path)
try:
columns = _table_columns(con, table, db_path)
if key not in columns:
raise ValueError(
f"{table!r} has no {key!r} column, so an annotation cannot be "
f"attributed to a crop.")
if annotation_column not in columns:
raise ValueError(
f"{table!r} has no {annotation_column!r} column — nothing has "
f"been annotated into it yet. The Annotate app creates the "
f"column the first time it saves. Columns present: "
f"{', '.join(columns)}.")
meta = [c for c in ("plateID", "rowID", "columnID", "fieldID", "prc")
if c in columns]
select = list(dict.fromkeys([key, annotation_column] + meta))
rows = con.execute(
f"SELECT {', '.join(_quote_ident(c) for c in select)} "
f"FROM {_quote_ident(table)}").fetchall()
rounds = _read_rounds(con, annotation_column)
finally:
con.close()
frame = pd.DataFrame(rows, columns=select)
n_unfiltered = len(frame)
if image_type:
keep = frame[key].astype(str).str.contains(str(image_type), regex=False)
n_filtered = int((~keep).sum())
frame = frame[keep]
if n_filtered:
notes.append(
f"image_type={image_type!r} excluded {n_filtered} of "
f"{n_unfiltered} crops before counting; every number below "
f"describes the {len(frame)} that matched.")
n_rows = len(frame)
well_cols = [c for c in ("plateID", "rowID", "columnID") if c in frame.columns]
frame = frame.assign(_well=_well_key(frame, well_cols))
if "plateID" in frame.columns:
frame = frame.assign(_plate=frame["plateID"].astype(str))
else:
frame = frame.assign(_plate="(unknown)")
notes.append(
f"{table} has no plateID column, so nothing here can be attributed "
f"to a plate; every label is reported under '(unknown)'.")
wells_total = int(frame["_well"].nunique()) if len(frame) else 0
labelled = frame[frame[annotation_column].notna()].copy()
n_annotated = len(labelled)
if n_annotated:
labelled["_class"] = labelled[annotation_column].map(_class_name)
if not rounds.empty and n_annotated:
labelled = labelled.merge(
rounds, how="left", left_on=key, right_on="png_path")
labelled["round"] = labelled["round"].fillna(-1).astype(int)
labelled["source"] = labelled["source"].fillna("unrecorded")
else:
labelled["round"] = -1
labelled["source"] = "unrecorded"
if n_annotated:
notes.append(
f"No {ROUND_TABLE} rows for {annotation_column!r}, so no label "
f"can be attributed to an annotation round. Rounds are "
f"recorded from the Annotate screen; labels written before "
f"that read as 'unrecorded'.")
if not n_annotated:
out = pd.DataFrame(columns=["plateID", "rowID", "columnID", "class",
"n", "share"])
out.attrs["spacr_annotation_coverage"] = {
"db_path": str(db_path), "table": table,
"annotation_column": annotation_column,
"image_type": image_type,
"n_rows": n_rows, "n_rows_unfiltered": int(n_unfiltered),
"n_annotated": 0, "n_classes": 0,
"wells_total": wells_total, "wells_annotated": 0,
"plates_total": int(frame["_plate"].nunique()) if len(frame) else 0,
"plates_annotated": 0,
"by_class": {}, "by_plate": {}, "by_well": {},
"by_class_plate": {}, "by_class_well": {},
"by_round": {}, "by_source": {}, "concentration": {},
"notes": notes + [
f"Nothing is annotated in {annotation_column!r} yet "
f"({n_rows} crops in {table})."],
}
return out
group_cols = [c for c in ("plateID", "rowID", "columnID") if c in labelled.columns]
grouped = (labelled.groupby(group_cols + ["_class"], dropna=False)
.size().reset_index(name="n")
if group_cols else
labelled.groupby(["_class"]).size().reset_index(name="n"))
grouped = grouped.rename(columns={"_class": "class"})
grouped["share"] = grouped["n"] / float(n_annotated)
grouped = grouped.sort_values(["class", "n"], ascending=[True, False]
).reset_index(drop=True)
by_class = {k: int(v) for k, v in
labelled["_class"].value_counts().sort_index().items()}
by_plate = {str(k): int(v) for k, v in
labelled["_plate"].value_counts().sort_index().items()}
by_well = {str(k): int(v) for k, v in
labelled["_well"].value_counts().sort_index().items()}
by_class_plate = {
str(cls): {str(p): int(n) for p, n in
sub["_plate"].value_counts().sort_index().items()}
for cls, sub in labelled.groupby("_class")}
by_class_well = {
str(cls): {str(w): int(n) for w, n in
sub["_well"].value_counts().sort_index().items()}
for cls, sub in labelled.groupby("_class")}
concentration = {
str(cls): _concentration(sub["_well"].value_counts())
for cls, sub in labelled.groupby("_class")}
concentration["__all__"] = _concentration(labelled["_well"].value_counts())
for cls, stats in concentration.items():
if cls == "__all__" or stats["n"] < 10 or stats["n_groups"] < 1:
continue
if stats["top_share"] >= 0.5 and stats["n_groups"] > 1:
notes.append(
f"Class {cls}: {stats['top_n']} of {stats['n']} labels "
f"({stats['top_share']:.0%}) come from one well "
f"({stats['top']}). A classifier trained on this learns that "
f"well as much as it learns the class.")
elif stats["n_groups"] == 1:
notes.append(
f"Class {cls}: all {stats['n']} labels come from the single "
f"well {stats['top']}. There is no way to tell the class "
f"apart from the well, and a random train/test split of these "
f"objects will report an accuracy that does not transfer.")
counts = pd.Series(by_class)
if len(counts) > 1 and int(counts.min()) * 5 < int(counts.max()):
notes.append(
f"Class balance is {':'.join(str(int(v)) for v in counts)} "
f"({', '.join(map(str, counts.index))}). The smallest class has "
f"{int(counts.min())} labels; per-class accuracy, not the "
f"aggregate, is the number to read on a model trained here.")
out = grouped
out.attrs["spacr_annotation_coverage"] = {
"db_path": str(db_path), "table": table,
"annotation_column": annotation_column,
"image_type": image_type,
"n_rows": n_rows, "n_rows_unfiltered": int(n_unfiltered),
"n_annotated": int(n_annotated),
"n_classes": len(by_class),
"wells_total": wells_total,
"wells_annotated": int(labelled["_well"].nunique()),
"plates_total": int(frame["_plate"].nunique()),
"plates_annotated": int(labelled["_plate"].nunique()),
"by_class": by_class,
"by_plate": by_plate,
"by_well": by_well,
"by_class_plate": by_class_plate,
"by_class_well": by_class_well,
"by_round": {int(k): int(v) for k, v in
labelled["round"].value_counts().sort_index().items()},
"by_source": {str(k): int(v) for k, v in
labelled["source"].value_counts().sort_index().items()},
"concentration": concentration,
"notes": notes,
}
return out
def _class_name(value: Any) -> str:
"""``1.0`` and ``1`` are the same class; report it as ``'1'``."""
if isinstance(value, float) and float(value).is_integer():
return str(int(value))
return str(value)
def _object_label(value: Any) -> str:
"""``'o5'``/``5``/``5.0`` → ``'5'``; anything unusable → ``''``.
``png_list`` stores the object id as TEXT ``'o5'`` while every object
table stores an integer, and the sentinels ``'omulti'`` / ``'onone'`` /
``'error'`` are real values in that column. Both facts are load-bearing
here: a naive ``int(value)`` raises on the sentinels, and a naive string
compare never matches an integer key.
"""
if value is None:
return ""
text = str(value).strip()
if not text or text.lower() in ("nan", "none", "null"):
return ""
if text[:1] in ("o", "O"):
text = text[1:]
if not text:
return ""
try:
return str(int(float(text)))
except (TypeError, ValueError):
return ""
[docs]
def crops_for_object_keys(db_path: str, keys: Sequence[str], *,
table: str = PNG_TABLE, key: str = PNG_KEY,
annotation_column: Optional[str] = None,
timelapse: bool = False,
image_type: Optional[str] = None
) -> List[Tuple[str, Optional[int]]]:
"""Resolve object keys to crop rows, **in the caller's order**.
The database half of the object-routing contract in
:mod:`spacr.qt.linked_selection`: a scatter plot or a confusion-matrix
cell names objects by :data:`spacr.selection.OBJECT_KEY_COLUMNS` key,
and the Annotate screen has to turn those into the crops it paginates.
Kept here rather than in the Qt screen so it is testable without a
display, and so a second consumer does not have to reimplement the
``'o5'``-versus-``5`` trap in :func:`_object_label`.
Input order is preserved so priority rankings such as ``worst errors
first`` remain unchanged. Keys without a corresponding crop are omitted.
Typed keys distinguish objects with the same numeric label in one field.
``png_list`` identifies the object type through the populated
``<type>_id`` column (:data:`PNG_ID_COLUMN_TYPES`), so nucleus 1 and
pathogen 1 resolve independently. An untyped key selects the first matching
crop. If the table does not expose object-type columns, typed lookup falls
back to the corresponding untyped key.
Escaped metadata components are resolved in both encoded and raw form.
For example, :func:`spacr.selection.object_keys` represents a ``fieldID``
of ``'f_1'`` as ``'f%5F1'``, while a key assembled from crop-table columns
contains the raw underscore.
:param db_path: path to ``measurements.db``.
:param keys: object keys, typed or not. A ``png_path``, a ``prcfo`` or a
``file_name`` is also accepted, so a caller working from a crop table
rather than a measurement table does not need a translation step.
:param table: crop table.
:param key: crop key column.
:param annotation_column: read the existing label too, so an already
annotated crop renders with its colour rather than blank.
:param timelapse: the keys carry a timepoint.
:param image_type: substring filter on the crop key.
:returns: ``[(png_path, annotation or None), …]`` in the keys' order.
"""
wanted = [str(k) for k in keys]
if not wanted:
return []
con = _connect(db_path)
try:
columns = _table_columns(con, table, db_path)
if key not in columns:
raise ValueError(
f"{table!r} has no {key!r} column, so an object key cannot be "
f"resolved to a crop.")
has_annotation = bool(annotation_column
and annotation_column in columns)
select = [key]
for extra in ("prcfo", "file_name", "plateID", "rowID", "columnID",
"fieldID", "timeID", "cell_id", "nucleus_id",
"pathogen_id", "cytoplasm_id", "organelle_id"):
if extra in columns and extra not in select:
select.append(extra)
if has_annotation and annotation_column not in select:
select.append(annotation_column)
rows = con.execute(
f"SELECT {', '.join(_quote_ident(c) for c in select)} "
f"FROM {_quote_ident(table)}").fetchall()
finally:
con.close()
from .selection import (KEY_ESCAPED_CHARACTERS, escape_key_component,
key_object_type, untyped_object_key)
index = {c: i for i, c in enumerate(select)}
id_columns = [c for c in PNG_ID_COLUMN_TYPES if c in index]
meta_columns = ["plateID", "rowID", "columnID", "fieldID"]
if timelapse:
meta_columns.append("timeID")
by_key: Dict[str, Tuple[str, Optional[int]]] = {}
by_escaped_key: Dict[str, Tuple[str, Optional[int]]] = {}
untyped_rows: Dict[str, Tuple[str, Optional[int]]] = {}
escaped_untyped_rows: Dict[str, Tuple[str, Optional[int]]] = {}
def _register(target: Dict[str, Tuple[str, Optional[int]]],
composed: List[str], label: str, object_type: Optional[str],
entry: Tuple[str, Optional[int]]) -> None:
"""Register first-wins untyped and, when known, typed object keys."""
target.setdefault("_".join(composed + [label]), entry)
if object_type is not None:
target.setdefault(
"_".join(composed + [f"{object_type}{label}"]), entry)
for row in rows:
path = str(row[index[key]])
if image_type and str(image_type) not in path:
continue
annotation = None
if has_annotation:
raw = row[index[annotation_column]]
annotation = None if raw is None else int(raw)
entry = (path, annotation)
stated = [(column, _object_label(row[index[column]]))
for column in id_columns]
stated = [(column, value) for column, value in stated if value]
declared = [column for column in id_columns
if row[index[column]] is not None
and str(row[index[column]]).strip().lower()
not in ("", "nan", "none", "null")]
object_type = None
label = ""
if len(declared) == 1:
object_type = PNG_ID_COLUMN_TYPES[declared[0]]
if stated:
label = stated[0][1]
prcfo = (str(row[index["prcfo"]])
if "prcfo" in index and row[index["prcfo"]] is not None
else "")
if not label and prcfo:
label = _object_label(prcfo.rsplit("_", 1)[-1])
if label and all(c in index for c in meta_columns):
parts = [str(row[index[c]]) for c in meta_columns]
_register(by_key, parts, label, object_type, entry)
if not declared:
_register(untyped_rows, parts, label, None, entry)
if any(c in p for p in parts for c in KEY_ESCAPED_CHARACTERS):
_register(by_escaped_key,
[escape_key_component(p) for p in parts],
label, object_type, entry)
if not declared:
_register(escaped_untyped_rows,
[escape_key_component(p) for p in parts],
label, None, entry)
file_name = (str(row[index["file_name"]])
if "file_name" in index and
row[index["file_name"]] is not None else "")
for candidate in (path, prcfo, file_name):
if candidate:
candidate_type = key_object_type(candidate)
if (candidate_type is not None and declared
and candidate_type != object_type):
continue
by_key.setdefault(candidate, entry)
if not declared:
untyped_rows.setdefault(candidate, entry)
def _resolve(name: str) -> Optional[Tuple[str, Optional[int]]]:
"""The escaped spelling first — it is the one a producer emits today."""
found = by_escaped_key.get(name)
return by_key.get(name) if found is None else found
out: List[Tuple[str, Optional[int]]] = []
seen = set()
for wanted_key in wanted:
entry = _resolve(wanted_key)
if entry is None:
reduced = untyped_object_key(wanted_key)
if reduced != wanted_key:
entry = escaped_untyped_rows.get(reduced)
if entry is None:
entry = untyped_rows.get(reduced)
if entry is None or entry[0] in seen:
continue
seen.add(entry[0])
out.append(entry)
return out
def _utc_now() -> str:
"""ISO-8601 UTC, to the second — the stamp every round row carries."""
from datetime import datetime, timezone
return datetime.now(timezone.utc).isoformat(timespec="seconds")
def _write_connection(db_path: str) -> sqlite3.Connection:
"""A writable connection to ``db_path``, for the round tables only.
Separate from :func:`_connect`, which is read-only on purpose: building a
queue must never be able to write, and the two connections being visibly
different is what keeps that true.
"""
if not db_path or not str(db_path).strip():
raise ValueError("No database path given.")
path = os.path.abspath(os.path.expanduser(str(db_path).strip()))
if not os.path.isfile(path):
raise FileNotFoundError(f"No such database: {path}")
return sqlite3.connect(path, timeout=30)
[docs]
def ensure_round_tables(db_path: str) -> None:
"""Create :data:`ROUND_TABLE` and :data:`ROUND_LOG_TABLE` if absent.
:param db_path: existing ``measurements.db`` in which to create the tables.
Two tables, not one. Per-label provenance and per-round metrics have
different cardinalities and different lifetimes: a label keeps its round
forever, a round's held-out accuracy is rewritten if the round is re-fit.
"""
con = _write_connection(db_path)
try:
con.execute(f"""
CREATE TABLE IF NOT EXISTS {_quote_ident(ROUND_TABLE)} (
png_path TEXT NOT NULL,
annotation_column TEXT NOT NULL,
round INTEGER NOT NULL,
first_round INTEGER NOT NULL,
label INTEGER,
source TEXT NOT NULL DEFAULT 'manual',
labelled_utc TEXT NOT NULL,
PRIMARY KEY (png_path, annotation_column)
)""")
con.execute(f"""
CREATE TABLE IF NOT EXISTS {_quote_ident(ROUND_LOG_TABLE)} (
annotation_column TEXT NOT NULL,
round INTEGER NOT NULL,
finished_utc TEXT NOT NULL,
n_labels INTEGER NOT NULL DEFAULT 0,
n_new_labels INTEGER NOT NULL DEFAULT 0,
n_holdout INTEGER NOT NULL DEFAULT 0,
holdout_accuracy REAL,
holdout_f1_macro REAL,
per_class_json TEXT,
split_rule TEXT,
model_type TEXT,
model_path TEXT,
card_path TEXT,
measure TEXT,
diversity TEXT,
notes_json TEXT,
PRIMARY KEY (annotation_column, round)
)""")
con.commit()
finally:
con.close()
def _read_rounds(con: sqlite3.Connection,
annotation_column: str) -> pd.DataFrame:
"""Per-label round provenance, or an empty frame when unrecorded."""
names = {r[0] for r in con.execute(
"SELECT name FROM sqlite_master WHERE type='table'").fetchall()}
if ROUND_TABLE not in names:
return pd.DataFrame(columns=["png_path", "round", "first_round",
"source", "labelled_utc"])
rows = con.execute(
f"SELECT png_path, round, first_round, source, labelled_utc "
f"FROM {_quote_ident(ROUND_TABLE)} WHERE annotation_column = ?",
(str(annotation_column),)).fetchall()
return pd.DataFrame(rows, columns=["png_path", "round", "first_round",
"source", "labelled_utc"])
[docs]
def record_labels(db_path: str, annotation_column: str,
labels: Dict[str, Any], round_index: int,
source: str = "manual") -> int:
"""Stamp each label with the round it was made in.
Called by the Annotate screen every time it flushes a batch. The round a
label came from is what makes early-round bias auditable: the first
round's labels are drawn from whatever ordering existed before any model
had seen this screen, and if 80 % of a class's labels carry round 0 then
the "active learning" was mostly not active.
``first_round`` is preserved across re-labelling while ``round`` follows
the current value, so both "when was this crop first looked at" and
"which round set the label it has now" survive a correction.
:param db_path: path to ``measurements.db``.
:param annotation_column: the column the labels were written into.
:param labels: ``{png_path: class or None}``. ``None`` is a *cleared*
label and is recorded as such rather than dropped — a crop that was
looked at and deliberately left blank is not the same as one never
seen.
:param round_index: the round these labels belong to.
:param source: how the crop reached the annotator — ``'manual'``,
``'queue'``, or a caller's own tag.
:returns: number of rows written.
"""
if not labels:
return 0
ensure_round_tables(db_path)
stamp = _utc_now()
payload = [
(str(path), str(annotation_column), int(round_index),
int(round_index),
(None if value is None else int(value)), str(source), stamp)
for path, value in labels.items()]
con = _write_connection(db_path)
try:
con.executemany(
f"INSERT INTO {_quote_ident(ROUND_TABLE)} "
f"(png_path, annotation_column, round, first_round, label, "
f"source, labelled_utc) VALUES (?, ?, ?, ?, ?, ?, ?) "
f"ON CONFLICT(png_path, annotation_column) DO UPDATE SET "
f"round=excluded.round, label=excluded.label, "
f"source=excluded.source, labelled_utc=excluded.labelled_utc",
payload)
con.commit()
finally:
con.close()
return len(payload)
[docs]
def label_rounds(db_path: str,
annotation_column: str = "annotate") -> pd.DataFrame:
"""Per-label round provenance as a frame (empty when never recorded).
:param db_path: path to the ``measurements.db`` to read.
"""
con = _connect(db_path)
try:
return _read_rounds(con, annotation_column)
finally:
con.close()
[docs]
def next_round(db_path: str, annotation_column: str = "annotate") -> int:
"""The round number the next batch of labels belongs to.
:param db_path: path to the ``measurements.db`` whose round log is queried.
Round 0 is "before any model was retrained from inside Annotate" — the
labels that seeded the loop. The first retrain produces round 1.
"""
try:
con = _connect(db_path)
except (FileNotFoundError, ValueError):
return 0
try:
names = {r[0] for r in con.execute(
"SELECT name FROM sqlite_master WHERE type='table'").fetchall()}
if ROUND_LOG_TABLE not in names:
return 0
row = con.execute(
f"SELECT MAX(round) FROM {_quote_ident(ROUND_LOG_TABLE)} "
f"WHERE annotation_column = ?",
(str(annotation_column),)).fetchone()
finally:
con.close()
return int(row[0]) + 1 if row and row[0] is not None else 0
[docs]
def record_round(db_path: str, annotation_column: str, round_index: int,
**fields: Any) -> None:
"""Append (or replace) one row of the learning curve.
:param db_path: path to ``measurements.db``.
:param annotation_column: the column this round labelled into.
:param round_index: the round number.
:param fields: any of ``n_labels``, ``n_new_labels``, ``n_holdout``,
``holdout_accuracy``, ``holdout_f1_macro``, ``per_class`` (dict),
``split_rule``, ``model_type``, ``model_path``, ``card_path``,
``measure``, ``diversity``, ``notes`` (list).
"""
import json
ensure_round_tables(db_path)
values = {
"annotation_column": str(annotation_column),
"round": int(round_index),
"finished_utc": _utc_now(),
"n_labels": int(fields.get("n_labels", 0) or 0),
"n_new_labels": int(fields.get("n_new_labels", 0) or 0),
"n_holdout": int(fields.get("n_holdout", 0) or 0),
"holdout_accuracy": _as_float_or_none(fields.get("holdout_accuracy")),
"holdout_f1_macro": _as_float_or_none(fields.get("holdout_f1_macro")),
"per_class_json": json.dumps(fields.get("per_class") or {}),
"split_rule": str(fields.get("split_rule") or ""),
"model_type": str(fields.get("model_type") or ""),
"model_path": str(fields.get("model_path") or ""),
"card_path": str(fields.get("card_path") or ""),
"measure": str(fields.get("measure") or ""),
"diversity": str(fields.get("diversity") or ""),
"notes_json": json.dumps(list(fields.get("notes") or [])),
}
con = _write_connection(db_path)
try:
con.execute(
f"INSERT OR REPLACE INTO {_quote_ident(ROUND_LOG_TABLE)} "
f"({', '.join(_quote_ident(c) for c in values)}) "
f"VALUES ({', '.join('?' * len(values))})",
tuple(values.values()))
con.commit()
finally:
con.close()
def _as_float_or_none(value: Any) -> Optional[float]:
"""``float(value)`` unless it is None or non-finite."""
if value is None:
return None
try:
out = float(value)
except (TypeError, ValueError):
return None
return out if np.isfinite(out) else None
[docs]
def learning_curve(db_path: str,
annotation_column: str = "annotate") -> pd.DataFrame:
"""Held-out accuracy per round, oldest first — the curve to watch flatten.
:param db_path: path to ``measurements.db``.
:param annotation_column: the column the rounds labelled into.
:returns: a frame with one row per round: ``round``, ``finished_utc``,
``n_labels``, ``n_new_labels``, ``n_holdout``, ``holdout_accuracy``,
``holdout_f1_macro``, ``per_class`` (dict), ``split_rule``,
``model_type``, ``model_path``, ``card_path``, ``notes`` (list), and
the derived ``gain`` (accuracy change since the previous round).
Empty (with those columns) when no round has been recorded.
"""
import json
columns = ["round", "finished_utc", "n_labels", "n_new_labels",
"n_holdout", "holdout_accuracy", "holdout_f1_macro",
"per_class", "split_rule", "model_type", "model_path",
"card_path", "measure", "diversity", "notes", "gain"]
try:
con = _connect(db_path)
except (FileNotFoundError, ValueError):
return pd.DataFrame(columns=columns)
try:
names = {r[0] for r in con.execute(
"SELECT name FROM sqlite_master WHERE type='table'").fetchall()}
if ROUND_LOG_TABLE not in names:
return pd.DataFrame(columns=columns)
rows = con.execute(
f"SELECT round, finished_utc, n_labels, n_new_labels, n_holdout, "
f"holdout_accuracy, holdout_f1_macro, per_class_json, split_rule, "
f"model_type, model_path, card_path, measure, diversity, "
f"notes_json FROM {_quote_ident(ROUND_LOG_TABLE)} "
f"WHERE annotation_column = ? ORDER BY round",
(str(annotation_column),)).fetchall()
finally:
con.close()
if not rows:
return pd.DataFrame(columns=columns)
frame = pd.DataFrame(rows, columns=[
"round", "finished_utc", "n_labels", "n_new_labels", "n_holdout",
"holdout_accuracy", "holdout_f1_macro", "per_class_json",
"split_rule", "model_type", "model_path", "card_path", "measure",
"diversity", "notes_json"])
frame["per_class"] = [json.loads(v or "{}") for v in frame.pop("per_class_json")]
frame["notes"] = [json.loads(v or "[]") for v in frame.pop("notes_json")]
frame["gain"] = frame["holdout_accuracy"].astype(float).diff()
return frame[columns]
[docs]
class StoppingVerdict:
"""Whether the last stretch of annotation bought anything measurable.
:param stop: the recommendation.
:param reason: one sentence, in the words the screen shows.
:param gain: held-out accuracy change over the window examined.
:param labels_in_window: how many labels that change is attributed to.
:param window_from: the round the window opened at.
:param confident: whether ``gain`` is larger than one standard error of
the held-out accuracy itself. When it is *not*, "flat" and "we
cannot tell" look identical from the numbers, and this says which
you have.
:param noise: one standard error of the latest held-out accuracy,
``sqrt(p(1-p)/n)``.
:param trend: ``'rising'``, ``'flat'``, ``'falling'`` or ``'unknown'``.
"""
__slots__ = ("stop", "reason", "gain", "labels_in_window", "window_from",
"confident", "noise", "trend")
def __init__(self, stop: bool, reason: str, *, gain: Optional[float] = None,
labels_in_window: int = 0, window_from: Optional[int] = None,
confident: bool = False, noise: Optional[float] = None,
trend: str = "unknown"):
"""Normalize and store the recommendation and its measured evidence."""
self.stop = bool(stop)
self.reason = str(reason)
self.gain = gain
self.labels_in_window = int(labels_in_window)
self.window_from = window_from
self.confident = bool(confident)
self.noise = noise
self.trend = str(trend)
[docs]
def __bool__(self) -> bool:
"""True when the recommendation is to stop."""
return self.stop
[docs]
def __repr__(self) -> str:
"""Return stop, trend, gain, and labels-in-window for diagnostics."""
return (f"StoppingVerdict(stop={self.stop!r}, trend={self.trend!r}, "
f"gain={self.gain!r}, labels_in_window="
f"{self.labels_in_window!r})")
[docs]
def to_dict(self) -> Dict[str, Any]:
"""A JSON-friendly copy, for a card or a log."""
return {name: getattr(self, name) for name in self.__slots__}
[docs]
def should_stop(curve: pd.DataFrame, *, label_window: int = 50,
min_gain: float = 0.003,
min_rounds: int = 2) -> StoppingVerdict:
"""Has the last ``label_window`` labels moved held-out accuracy at all?
The rule, in one line: look back over whole rounds until at least
``label_window`` new labels have accumulated, and compare held-out
accuracy at the two ends. If it moved by less than ``min_gain``, stop.
Why this rule and not another:
* **Labels, not rounds, are the unit of cost.** A round is whatever size
the annotator felt like; "no improvement for 3 rounds" says nothing
when the rounds were 5, 300 and 8 labels. The thing being spent is
human attention, one crop at a time, so the window is measured in
crops.
* **It looks *back over* rounds, not *at* the last one.** A single round
that happened to land flat is noise; the question is whether the last
fifty labels — however they were divided up — bought anything.
* **It refuses to answer early.** Below ``label_window`` labels since the
first recorded round there is no window to measure, and a rule that
fired anyway would tell people to stop after their first twelve labels.
* **It distinguishes "flat" from "unmeasurable".** Held-out accuracy on
80 objects has a standard error near 0.05; a 0.3 % change is inside
the noise, and :attr:`StoppingVerdict.confident` says so instead of
dressing it up. Flat *is* still the recommendation — if more labels
are not moving a number you can measure, they are not buying anything
you can demonstrate — but the reason says the held-out set is too
small to prove convergence, which is a different piece of work.
* **A falling curve stops too, and says so.** Accuracy going down is not
convergence; it usually means the newest labels disagree with the
earlier ones, or that the held-out split moved. Either way, more of
the same is the wrong next move.
:param curve: the frame from :func:`learning_curve`.
:param label_window: how many labels the window must cover.
:param min_gain: accuracy change below which the window counts as flat.
:param min_rounds: rounds required before any verdict is given.
:returns: a :class:`StoppingVerdict`; ``bool(verdict)`` is the answer.
"""
if curve is None or not len(curve):
return StoppingVerdict(
False, "No round has been recorded yet — retrain once to start "
"the curve.")
usable = curve[curve["holdout_accuracy"].notna()].reset_index(drop=True)
if len(usable) < max(1, int(min_rounds)):
return StoppingVerdict(
False,
f"Only {len(usable)} round(s) with a held-out score; "
f"{int(min_rounds)} are needed before a stopping rule means "
f"anything.",
trend="unknown")
latest = usable.iloc[-1]
accuracy = float(latest["holdout_accuracy"])
n_holdout = int(latest.get("n_holdout", 0) or 0)
noise = (float(np.sqrt(max(accuracy * (1.0 - accuracy), 0.0) / n_holdout))
if n_holdout > 0 else None)
accumulated = 0
index = len(usable) - 1
while index > 0 and accumulated < int(label_window):
accumulated += int(usable.iloc[index].get("n_new_labels", 0) or 0)
index -= 1
baseline = usable.iloc[index]
gain = accuracy - float(baseline["holdout_accuracy"])
window_from = int(baseline["round"])
if accumulated < int(label_window):
return StoppingVerdict(
False,
f"Only {accumulated} labels since round {window_from}; the rule "
f"waits for {int(label_window)} before calling a plateau. "
f"Held-out accuracy has moved {gain:+.3f} so far.",
gain=gain, labels_in_window=accumulated, window_from=window_from,
noise=noise, trend="unknown")
confident = bool(noise is not None and abs(gain) > noise)
if gain < -max(float(min_gain), 0.0):
return StoppingVerdict(
True,
f"Held-out accuracy FELL {abs(gain):.1%} over the last "
f"{accumulated} labels (round {window_from} → "
f"{int(latest['round'])}). That is not convergence: check "
f"whether the newest labels disagree with the earlier ones "
f"before adding more.",
gain=gain, labels_in_window=accumulated, window_from=window_from,
confident=confident, noise=noise, trend="falling")
if gain < float(min_gain):
reason = (
f"The last {accumulated} labels moved held-out accuracy by "
f"{gain:+.1%} (round {window_from} → {int(latest['round'])}, now "
f"{accuracy:.1%}). That is below the {float(min_gain):.1%} "
f"threshold — annotating more of the same is not buying "
f"measurable accuracy.")
if noise is not None and abs(gain) <= noise:
reason += (
f" Note that the held-out set is only {n_holdout} objects, so "
f"one standard error is {noise:.1%}: this says the gain is "
f"unmeasurable, not that it is provably zero. A larger "
f"held-out set is the way to tell those apart.")
return StoppingVerdict(
True, reason, gain=gain, labels_in_window=accumulated,
window_from=window_from, confident=confident, noise=noise,
trend="flat")
return StoppingVerdict(
False,
f"Still learning: the last {accumulated} labels moved held-out "
f"accuracy {gain:+.1%} (round {window_from} → "
f"{int(latest['round'])}, now {accuracy:.1%}). Keep going.",
gain=gain, labels_in_window=accumulated, window_from=window_from,
confident=confident, noise=noise, trend="rising")
[docs]
def holdout_report(y_true: Any, probs: Any,
classes: Optional[Sequence[Any]] = None) -> Dict[str, Any]:
"""Held-out metrics, with the confusion matrix they were derived from.
Torch-free, so both the classical-ML round here and
:func:`spacr.deep_spacr.held_out_report` can use one implementation and
a model card written by either says the same thing in the same shape.
Every derived figure is exactly the standard function of the matrix —
``accuracy = trace / total``, ``per_class[c] = M[c, c] / M[c, :].sum()``
— so a reader can recompute the card rather than trust it.
``n`` **is the number of rows the matrix actually contains**, and the
supports sum to it. A row whose true class the head has no column for is
counted as an error rather than dropped: the matrix grows to hold
it, and the missing column stays empty because the head can never predict
that class. Reporting ``n`` over one population and ``accuracy`` over
another is the one thing a model card must not do — a three-class
held-out set scored by a binary head used to report a perfect score on a
set the model got two of three right, and the ``accuracy == trace /
total`` invariant still checked out because the row was missing from
both sides of it.
:param y_true: integer class ids, shape ``(N,)``. Negative ids are not
classes; they are excluded, counted in ``n_unscored`` and explained
in ``notes`` rather than silently folded into the total.
:param probs: ``(N,)`` positive-class probabilities, or ``(N, C)`` rows.
:param classes: class names in head order.
:returns: ``n``, ``n_unscored``, ``num_classes``, ``head_classes``,
``classes``, ``accuracy``, ``f1_macro``, ``per_class_accuracy``,
``class_support``, ``predicted_support``, ``confusion_matrix``,
``notes``.
:raises ValueError: when there are not as many score rows as labels —
the two are not aligned, and every figure below would be a
comparison of one object's label with another's prediction.
"""
y_true = np.asarray(y_true, dtype=int).reshape(-1)
matrix_in = np.asarray(probs, dtype=float)
if matrix_in.ndim == 1:
matrix_in = np.column_stack([1.0 - matrix_in, matrix_in])
elif matrix_in.ndim == 2 and matrix_in.shape[1] == 1:
col = matrix_in[:, 0]
matrix_in = np.column_stack([1.0 - col, col])
head_classes = int(matrix_in.shape[1]) if matrix_in.size else \
int(max(2, (y_true.max() + 1) if y_true.size else 2))
preds = (matrix_in.argmax(axis=1).astype(int) if matrix_in.size
else np.zeros(0, dtype=int))
if len(preds) != len(y_true):
raise ValueError(
f"{len(y_true)} held-out labels but {len(preds)} rows of scores; "
f"they name different objects, so no confusion matrix built from "
f"them would mean anything.")
notes: List[str] = []
scorable = y_true >= 0
n_unscored = int((~scorable).sum())
if n_unscored:
notes.append(
f"{n_unscored} of {len(y_true)} held-out rows carry a negative "
f"class id, which is not a class: they are excluded from every "
f"number here, including n.")
truth_classes = (int(y_true[scorable].max()) + 1) if scorable.any() else 0
n_classes = int(max(head_classes, truth_classes))
if truth_classes > head_classes:
notes.append(
f"The held-out labels name {truth_classes} classes but the "
f"probability matrix has {head_classes} columns, so classes "
f"{head_classes}..{truth_classes - 1} can never be predicted and "
f"every object in them counts as an error. Score this set with "
f"the head it was labelled for.")
matrix = np.zeros((n_classes, n_classes), dtype=np.int64)
if y_true.size:
np.add.at(matrix, (y_true[scorable], preds[scorable]), 1)
row_sums = matrix.sum(axis=1)
per_class = np.where(row_sums > 0,
np.diag(matrix) / np.maximum(row_sums, 1), 0.0)
total = int(matrix.sum())
col_sums = matrix.sum(axis=0)
diag = np.diag(matrix).astype(float)
with np.errstate(divide="ignore", invalid="ignore"):
precision = np.where(col_sums > 0, diag / np.maximum(col_sums, 1), 0.0)
recall = np.where(row_sums > 0, diag / np.maximum(row_sums, 1), 0.0)
f1 = np.where((precision + recall) > 0,
2 * precision * recall / np.maximum(precision + recall, 1e-12),
0.0)
present = row_sums + col_sums > 0
f1_macro = float(f1[present].mean()) if present.any() else float("nan")
names = ([str(c) for c in classes] if classes is not None
and len(classes) == n_classes
else [f"class_{i}" for i in range(n_classes)])
return {
"n": total,
"n_unscored": n_unscored,
"num_classes": n_classes,
"head_classes": int(head_classes),
"classes": names,
"accuracy": (float(np.trace(matrix)) / total) if total else float("nan"),
"f1_macro": f1_macro,
"per_class_accuracy": [float(v) for v in per_class],
"class_support": [int(v) for v in row_sums],
"predicted_support": [int(v) for v in col_sums],
"confusion_matrix": [[int(v) for v in row] for row in matrix],
"notes": notes,
}
[docs]
def round_features(db_path: str, table: str = PNG_TABLE,
key: str = PNG_KEY,
tables: Sequence[str] = ("cell", "nucleus", "pathogen",
"cytoplasm"),
nuclei_limit: int = 10,
pathogen_limit: int = 10) -> pd.DataFrame:
"""The measurement features for every crop, indexed by ``png_path``.
The feature matrix an in-screen retrain fits on. Measurement features
rather than pixels on purpose: the point of retraining from inside
Annotate is to get a fresh ranking in seconds, on the machine the
annotator is sitting at, without a GPU and without leaving the screen.
A CNN retrain is the right thing to do at the *end* of the loop, not
between two pages of crops.
:param db_path: path to ``measurements.db``.
:param table: crop table carrying ``png_path`` and ``prcfo``.
:param key: the crop key column.
:param tables: object tables to merge features from; missing ones are
skipped.
:param nuclei_limit: passed through to :func:`spacr.io._read_and_merge_data`.
:param pathogen_limit: likewise.
:returns: numeric features indexed by ``png_path``.
:raises ValueError: when no object table with features could be read.
"""
from .io import _read_and_merge_data, _read_db
con = _connect(db_path)
try:
columns = _table_columns(con, table, db_path)
if "prcfo" not in columns:
raise ValueError(
f"{table!r} has no 'prcfo' column, so a crop cannot be matched "
f"to its measurements. Re-run Measure with save_png=True.")
rows = con.execute(
f"SELECT {_quote_ident(key)}, \"prcfo\" FROM {_quote_ident(table)}"
).fetchall()
available = {r[0] for r in con.execute(
"SELECT name FROM sqlite_master WHERE type IN ('table','view')"
).fetchall()}
finally:
con.close()
wanted = [t for t in tables if t in available]
if not wanted:
raise ValueError(
f"{os.path.basename(str(db_path))} has none of the object tables "
f"{', '.join(tables)}, so there are no features to fit on. Run "
f"Measure first, or pass features= explicitly.")
merged, _ = _read_and_merge_data([str(db_path)], wanted, False,
nuclei_limit=nuclei_limit,
pathogen_limit=pathogen_limit)
crops = pd.DataFrame(rows, columns=[key, "prcfo"]).dropna(subset=["prcfo"])
numeric = merged.select_dtypes(include=[np.number])
joined = crops.join(numeric, on="prcfo", how="inner")
return joined.drop(columns=["prcfo"]).set_index(key)
_SIMILAR_K = 100
_SIMILAR_ID_NAMES = frozenset({
"row", "col", "column", "field", "plate", "well", "rowid", "columnid",
"fieldid", "plateid", "time", "timeid", "frame", "label", "object_label",
"track_id", "index", "level_0", "prcfo", "prcf",
})
_SIMILAR_ID_PARTS = ("centroid", "bbox", "coords")
def _similarity_columns(columns: Sequence[str]) -> List[str]:
"""The columns of a feature matrix that describe a cell, not its place.
Position and bookkeeping columns (well, field, frame, object label,
centroid, bounding box, anything ending in ``_id``) are numeric, so
they survive a numeric filter, and a search over them returns the
query's neighbours on the plate rather than cells that look like it.
:param columns: candidate column names.
:returns: the names kept, in their original order.
"""
kept = []
for name in columns:
low = str(name).lower()
if low in _SIMILAR_ID_NAMES or low.endswith("_id"):
continue
if any(part in low for part in _SIMILAR_ID_PARTS):
continue
kept.append(name)
return kept
class _SimilarityIndex:
"""Cosine nearest-neighbour search over one feature vector per crop.
Columns are centred on their median and scaled by their standard
deviation so no single measurement's units dominate; a missing value
becomes the column's median, and constant or empty columns are dropped.
Rows are then scaled to unit length, so the inner product is the cosine
similarity and 1.0 means the same direction in feature space.
The search runs in FAISS (an exact inner-product index, moved to every
visible GPU when FAISS was built with GPU support) when FAISS is
installed, and otherwise as a blocked matrix product in NumPy. Both are
exact, so they return the same neighbours; FAISS is faster on large
query batches.
:param features: numeric features indexed by crop key, one row per
crop. Measurement features and embedding columns both work.
:param backend: ``'auto'`` (FAISS when importable, else NumPy),
``'faiss'`` or ``'numpy'``.
:param block: rows scored per NumPy block, which bounds memory.
:raises ValueError: on an empty matrix or one with no usable column.
:raises ImportError: for ``backend='faiss'`` without FAISS installed.
"""
def __init__(self, features: pd.DataFrame, *, backend: str = "auto",
block: int = 262144):
"""Standardise ``features``, normalise its rows and build the index."""
numeric = features.select_dtypes(include=[np.number])
numeric = numeric.loc[~numeric.index.duplicated(keep="first")]
if numeric.empty:
raise ValueError(
"No crop has numeric features, so there is nothing to compare "
"cells by. Run Measure first.")
matrix = numeric.to_numpy(dtype=np.float32, copy=True)
matrix[~np.isfinite(matrix)] = np.nan
import warnings
with np.errstate(all="ignore"), warnings.catch_warnings():
warnings.simplefilter("ignore", RuntimeWarning)
if np.isnan(matrix).any():
centre = np.nanmedian(matrix, axis=0)
spread = np.nanstd(matrix, axis=0)
else:
centre = np.median(matrix, axis=0)
spread = matrix.std(axis=0)
usable = np.isfinite(centre) & np.isfinite(spread) & (spread > 0)
if not usable.any():
raise ValueError(
"Every feature column is constant or empty, so no two cells "
"can be told apart.")
matrix = matrix[:, usable]
centre = centre[usable]
spread = spread[usable]
missing = np.isnan(matrix)
if missing.any():
matrix[missing] = np.take(centre, np.nonzero(missing)[1])
matrix -= centre
matrix /= spread
norms = np.linalg.norm(matrix, axis=1, keepdims=True)
norms[norms == 0] = 1.0
matrix /= norms
self.keys = np.asarray([str(k) for k in numeric.index], dtype=object)
self.columns = [str(c) for c, ok in zip(numeric.columns, usable) if ok]
self._position = {key: i for i, key in enumerate(self.keys)}
self._matrix = np.ascontiguousarray(matrix, dtype=np.float32)
self._block = max(1, int(block))
self._faiss = None
self.backend = self._build(str(backend or "auto").lower())
def _build(self, backend: str) -> str:
"""Build the FAISS index when asked or available; name the backend."""
if backend not in ("auto", "faiss", "numpy"):
raise ValueError(
f"unknown similarity backend {backend!r}; expected 'auto', "
f"'faiss' or 'numpy'")
if backend == "numpy":
return "numpy"
try:
import faiss
except ImportError:
if backend == "faiss":
raise ImportError(
"FAISS is not installed. Install it with "
"'pip install faiss-cpu' (or 'conda install -c pytorch "
"faiss-gpu' for the GPU build), or use backend='numpy'.")
return "numpy"
index = faiss.IndexFlatIP(self._matrix.shape[1])
name = "faiss"
try:
if (os.environ.get("CUDA_VISIBLE_DEVICES", None) != ""
and hasattr(faiss, "get_num_gpus")
and faiss.get_num_gpus() > 0):
index = faiss.index_cpu_to_all_gpus(index)
name = "faiss-gpu"
except Exception:
name = "faiss"
index.add(self._matrix)
self._faiss = index
return name
def __len__(self) -> int:
"""Return the number of crops indexed."""
return int(self._matrix.shape[0])
def __contains__(self, key: Any) -> bool:
"""Whether crop ``key`` has a row in the index."""
return str(key) in self._position
def vector(self, key: Any) -> np.ndarray:
"""The normalised feature vector of crop ``key``.
:raises KeyError: when the crop has no row in the index.
"""
at = self._position.get(str(key))
if at is None:
raise KeyError(
f"{key!r} has no measured features, so there is nothing to "
f"compare it by")
return self._matrix[at]
def search(self, queries: np.ndarray, k: int
) -> Tuple[np.ndarray, np.ndarray]:
"""The ``k`` most similar rows for each query vector, best first.
:param queries: ``(n, d)`` or ``(d,)`` normalised vectors, as
:meth:`vector` returns them.
:param k: neighbours per query, capped at the index size.
:returns: ``(similarity, row)`` arrays of shape ``(n, k)``.
"""
queries = np.ascontiguousarray(
np.atleast_2d(np.asarray(queries, dtype=np.float32)))
k = max(1, min(int(k), len(self)))
if self._faiss is not None:
scores, rows = self._faiss.search(queries, k)
return scores, rows.astype(np.int64)
best_s = np.full((queries.shape[0], 0), -np.inf, dtype=np.float32)
best_i = np.zeros((queries.shape[0], 0), dtype=np.int64)
for start in range(0, len(self), self._block):
chunk = self._matrix[start:start + self._block]
scores = queries @ chunk.T
take = min(k, scores.shape[1])
part = np.argpartition(-scores, take - 1, axis=1)[:, :take]
best_s = np.concatenate(
[best_s, np.take_along_axis(scores, part, axis=1)], axis=1)
best_i = np.concatenate([best_i, part + start], axis=1)
if best_s.shape[1] > k:
keep = np.argpartition(-best_s, k - 1, axis=1)[:, :k]
best_s = np.take_along_axis(best_s, keep, axis=1)
best_i = np.take_along_axis(best_i, keep, axis=1)
order = np.argsort(-best_s, axis=1, kind="stable")
return (np.take_along_axis(best_s, order, axis=1),
np.take_along_axis(best_i, order, axis=1))
def like(self, key: Any, k: int = _SIMILAR_K, *,
exclude: Optional[Iterable[Any]] = None) -> pd.DataFrame:
"""The crops most like crop ``key``, most similar first.
:param key: the query crop.
:param k: how many to return, the query itself not counted.
:param exclude: crop keys to leave out of the answer, such as the
ones already annotated.
:returns: a frame with ``key``, ``similarity`` (cosine, 1.0 is the
same direction) and ``rank`` (1 is the closest).
:raises KeyError: when ``key`` has no row in the index.
"""
skip = {str(key)}
skip.update(str(x) for x in (() if exclude is None else exclude))
want = int(k) + len(skip)
scores, rows = self.search(self.vector(key), want)
out = [(self.keys[r], float(s)) for s, r in zip(scores[0], rows[0])
if r >= 0 and self.keys[r] not in skip][:max(0, int(k))]
frame = pd.DataFrame(out, columns=["key", "similarity"])
frame["rank"] = np.arange(1, len(frame) + 1)
return frame
_EMBEDDING_TABLE = "crop_embedding"
def _store_crop_embeddings(db_path: str, prcfo: Sequence[Any],
embedding: Any, *, encoder_entry=None) -> int:
"""Save crop embeddings in ``measurements.db``, one row per object.
The table is keyed by ``prcfo``, so it joins to the crop table the same
way the measurements do. Rows for the same ``prcfo`` are replaced; a
new embedding with different columns or a different specification
replaces the whole table. A specified result keeps its fingerprint and
complete specification beside each row; unspecified vectors cannot be
merged into that table because their encoder identity is unknown.
:param db_path: path to ``measurements.db``.
:param prcfo: one object key per row of ``embedding``.
:param embedding: an :class:`spacr.embeddings.EmbeddingResult`, or a
numeric ``(n, d)`` array or frame.
:returns: the number of rows written.
:raises ValueError: when the keys and rows do not match in number.
"""
from . import tabular
from dataclasses import asdict
import json
columns = getattr(embedding, "columns", None)
values = getattr(embedding, "values", embedding)
if isinstance(embedding, pd.DataFrame):
columns, values = list(embedding.columns), embedding.to_numpy()
values = np.asarray(values, dtype=np.float32)
keys = [str(k) for k in prcfo]
if values.ndim != 2 or len(keys) != values.shape[0]:
raise ValueError(
f"{len(keys)} object keys for {values.shape[0] if values.ndim else 0} "
f"embedding rows; pass one prcfo per crop.")
if columns is None:
columns = [f"emb_{i}" for i in range(values.shape[1])]
frame = pd.DataFrame(values, columns=[str(c) for c in columns])
frame.insert(0, "prcfo", keys)
frame = frame.drop_duplicates("prcfo", keep="last")
spec = getattr(embedding, "spec", None)
fingerprint = None if spec is None else spec.fingerprint()
if fingerprint is not None:
frame["_embedding_fingerprint"] = fingerprint
frame["_embedding_spec"] = json.dumps(asdict(spec), sort_keys=True)
if encoder_entry is not None:
frame["_embedding_weights_sha256"] = str(
getattr(encoder_entry, "sha256", "") or "").lower()
frame["_embedding_encoder_key"] = str(
getattr(encoder_entry, "key", "") or "")
frame["_embedding_encoder_source"] = str(
getattr(encoder_entry, "source", "") or "")
frame["_embedding_encoder_backbone"] = str(
getattr(spec, "backbone", "") or "")
existing = _stored_embedding_frame(db_path)
compatible = (existing is not None
and list(existing.columns) == list(frame.columns))
if compatible and fingerprint is not None:
compatible = bool(existing["_embedding_fingerprint"].eq(
fingerprint).all())
if compatible and encoder_entry is not None:
for column in ("_embedding_weights_sha256", "_embedding_encoder_key",
"_embedding_encoder_source", "_embedding_encoder_backbone"):
if not existing[column].eq(frame[column].iloc[0]).all():
compatible = False
break
if compatible:
existing = existing[~existing["prcfo"].isin(frame["prcfo"])]
if not existing.empty:
frame = pd.concat([existing, frame], ignore_index=True)
tabular.write_database(frame, db_path, _EMBEDDING_TABLE,
if_exists="replace", canonicalise=False)
return len(frame)
def _stored_embedding_frame(db_path: str) -> Optional[pd.DataFrame]:
"""The stored crop-embedding table, or ``None`` when there is none."""
from . import tabular
if not os.path.exists(str(db_path)):
return None
if _EMBEDDING_TABLE not in tabular.database_tables(db_path):
return None
return tabular.read_table(db_path, table=_EMBEDDING_TABLE,
canonicalise=False, report=None,
repair_plate_ids=False)
def _stored_embeddings(db_path: str, table: str = PNG_TABLE,
key: str = PNG_KEY) -> Optional[pd.DataFrame]:
"""Stored crop embeddings indexed by the crop key, or ``None``.
:param db_path: path to ``measurements.db``.
:param table: crop table carrying ``key`` and ``prcfo``.
:param key: the crop key column.
:returns: the embedding columns for every crop with a stored vector,
or ``None`` when no embedding is stored or none joins a crop.
"""
stored = _stored_embedding_frame(db_path)
if stored is None or stored.empty:
return None
con = _connect(db_path)
try:
if "prcfo" not in _table_columns(con, table, db_path):
return None
rows = con.execute(
f"SELECT {_quote_ident(key)}, \"prcfo\" FROM {_quote_ident(table)}"
).fetchall()
finally:
con.close()
crops = pd.DataFrame(rows, columns=[key, "prcfo"]).dropna(subset=["prcfo"])
crops["prcfo"] = crops["prcfo"].astype(str)
stored["prcfo"] = stored["prcfo"].astype(str)
joined = crops.merge(stored, on="prcfo", how="inner")
if joined.empty:
return None
metadata = [column for column in joined.columns
if str(column).startswith("_embedding_")]
return joined.drop(columns=["prcfo", *metadata]).set_index(key)
def _similarity_index(db_path: str, *, features: Optional[pd.DataFrame] = None,
image_type: Optional[str] = None,
backend: str = "auto") -> _SimilarityIndex:
"""Index every crop of a database for "find cells like this".
:param db_path: path to ``measurements.db``.
:param features: the vectors to compare by, indexed by ``png_path``,
such as an embedding from :func:`spacr.embeddings.embed_array`.
When omitted, the crop embeddings stored in the database by
:func:`_store_crop_embeddings` are used if present, otherwise the
measurement features from :func:`round_features`.
:param image_type: substring filter on the crop key.
:param backend: passed to :class:`_SimilarityIndex`.
:returns: the index.
:raises ValueError: when no crop has usable features.
"""
if features is None:
features = _stored_embeddings(db_path)
if features is None:
features = round_features(db_path)
features = features[_similarity_columns(features.columns)]
if image_type:
keep = features.index.astype(str).str.contains(str(image_type),
regex=False)
features = features.loc[keep]
return _SimilarityIndex(features, backend=backend)
class _MultiSimilarityIndex:
"""Search compatible crops across databases without losing source identity."""
def __init__(self, frames: Sequence[Tuple[str, pd.DataFrame]], *,
backend: str = "auto"):
"""Give every crop a private index key and retain its database and key."""
sources = {}
indexed = []
offset = 0
for db_path, features in frames:
unique = features.loc[~features.index.duplicated()].copy(deep=False)
identifiers = [str(i) for i in range(offset, offset + len(unique))]
sources.update(zip(identifiers,
((db_path, str(key)) for key in unique.index)))
unique.index = identifiers
indexed.append(unique)
offset += len(unique)
combined = pd.concat(indexed, copy=False)
self._index = _SimilarityIndex(combined, backend=backend)
self.sources = sources
self._identifiers = {source: identifier
for identifier, source in sources.items()}
self.keys = tuple(sources[identifier] for identifier in self._index.keys)
self.columns = self._index.columns
self.backend = self._index.backend
def __len__(self) -> int:
"""Return the number of source-qualified crops."""
return len(self._index)
def like(self, db_path: str, key: Any, k: int = _SIMILAR_K, *,
exclude: Optional[Iterable[Tuple[str, Any]]] = None) -> pd.DataFrame:
"""Find neighbours while excluding only the exact queried source row."""
source = (os.path.abspath(str(db_path)), str(key))
identifier = self._identifiers.get(source)
if identifier is None:
raise KeyError(f"{key!r} has no compatible features in {db_path!r}")
excluded = [self._identifiers[item] for item in (exclude or ())
if item in self._identifiers]
hits = self._index.like(identifier, k, exclude=excluded)
hits.insert(0, "db_path", [self.sources[row][0] for row in hits["key"]])
hits["key"] = [self.sources[row][1] for row in hits["key"]]
return hits
def _multi_similarity_index(db_paths: Sequence[str], *,
image_type: Optional[str] = None,
backend: str = "auto",
feature_kind: str = "auto") -> _MultiSimilarityIndex:
"""Build one CPU index only when every plate has the same feature space.
Stored embeddings must carry the same nonempty specification fingerprint
on every row of every plate. Measurement matrices must have the same
feature names. Mixing these two feature kinds is refused rather than
filling missing or incompatible dimensions with zeroes.
"""
paths = list(dict.fromkeys(os.path.abspath(str(path)) for path in db_paths))
if len(paths) < 2 or len({os.path.realpath(path) for path in paths}) < len(paths):
raise ValueError("Choose at least two different plate databases.")
if feature_kind not in ("auto", "embeddings", "measurements"):
raise ValueError("Choose embeddings or measurements as the feature kind.")
kind = fingerprint = provenance = columns = None
frames = []
for path in paths:
if not os.path.isfile(path):
raise ValueError(f"Plate database does not exist: {path}")
stored = (_stored_embedding_frame(path)
if feature_kind != "measurements" else None)
if stored is not None:
current_kind = "embedding"
if stored.empty or "_embedding_fingerprint" not in stored:
raise ValueError(f"Stored embeddings lack a model fingerprint: {path}")
fingerprints = stored["_embedding_fingerprint"].dropna().unique()
if (len(fingerprints) != 1 or not str(fingerprints[0]).strip()
or stored["_embedding_fingerprint"].isna().any()):
raise ValueError(f"Stored embeddings have mixed or missing model fingerprints: {path}")
current_fingerprint = str(fingerprints[0])
fields = ("_embedding_weights_sha256", "_embedding_encoder_key",
"_embedding_encoder_source", "_embedding_encoder_backbone")
if any(field not in stored for field in fields):
raise ValueError(f"Stored embeddings lack actual encoder provenance: {path}")
values = []
for field in fields:
unique = stored[field].dropna().astype(str).unique()
if (len(unique) != 1 or not unique[0].strip()
or stored[field].isna().any()):
raise ValueError(f"Stored embeddings have mixed or missing encoder provenance: {path}")
values.append(unique[0])
current_provenance = tuple(values)
digest = current_provenance[0]
if len(digest) != 64 or any(c not in "0123456789abcdef" for c in digest):
raise ValueError(f"Stored embeddings lack a verified weights SHA-256: {path}")
features = _stored_embeddings(path)
if features is None:
raise ValueError(f"No stored embeddings join crop rows in {path}")
else:
if feature_kind == "embeddings":
raise ValueError(f"No stored embeddings were found in {path}")
current_kind = "measurement"
current_fingerprint = None
current_provenance = None
features = round_features(path)
features = features[_similarity_columns(features.columns)]
if image_type:
keep = features.index.astype(str).str.contains(str(image_type),
regex=False)
features = features.loc[keep]
features = features.select_dtypes(include=[np.number])
if features.empty:
raise ValueError(f"No compatible crops were found in {path}")
current_columns = list(features.columns)
if kind is None:
kind, fingerprint, provenance, columns = (
current_kind, current_fingerprint, current_provenance,
current_columns)
elif current_kind != kind:
raise ValueError("Cannot compare stored embeddings with measurement features.")
elif current_fingerprint != fingerprint:
raise ValueError("Plate embeddings were made by different models or settings.")
elif current_provenance != provenance:
raise ValueError("Plate embeddings have different encoder weights or provenance.")
elif set(current_columns) != set(columns) or len(current_columns) != len(columns):
raise ValueError("Plate feature columns differ; measure or embed them consistently.")
frames.append((path, features.loc[:, columns]))
return _MultiSimilarityIndex(frames, backend=backend)
def _similarity_agreement(index: _SimilarityIndex, labels: Mapping[Any, Any],
k: int = 10) -> pd.DataFrame:
"""How often a labelled crop's nearest neighbours share its label.
The check that similarity search finds the same kind of cell rather
than the same well: every labelled crop is a query, its ``k`` nearest
labelled neighbours (itself excluded) are compared with its own label,
and the share that agree is set beside the share a random draw would
give, the class's frequency among the labelled crops.
:param index: the index to evaluate.
:param labels: crop key to class label; suggestions and blanks should
already be removed.
:param k: neighbours per query.
:returns: one row per class with ``n``, ``precision_at_k``,
``chance`` and ``lift``, then an ``all`` row, micro-averaged.
:raises ValueError: when fewer than two labelled crops are indexed.
"""
pairs = [(str(key), value) for key, value in labels.items()
if str(key) in index and value is not None]
if len(pairs) < 2:
raise ValueError(
"Fewer than two labelled crops have features, so agreement "
"cannot be measured.")
keys = np.asarray([p[0] for p in pairs], dtype=object)
classes = np.asarray([str(p[1]) for p in pairs], dtype=object)
rows = np.asarray([index._position[key] for key in keys], dtype=np.int64)
sub = index._matrix[rows]
k = max(1, min(int(k), len(keys) - 1))
hits = np.empty(len(keys), dtype=np.float64)
for start in range(0, len(keys), 2048):
scores = sub[start:start + 2048] @ sub.T
own = np.arange(scores.shape[0])
scores[own, own + start] = -np.inf
near = np.argpartition(-scores, k - 1, axis=1)[:, :k]
hits[start:start + scores.shape[0]] = (
classes[near] == classes[start:start + scores.shape[0], None]
).mean(axis=1)
counts = pd.Series(classes).value_counts()
records = []
for name in sorted(counts.index, key=str):
mask = classes == name
chance = float((counts[name] - 1) / max(1, len(keys) - 1))
precision = float(hits[mask].mean())
records.append({"class": name, "n": int(mask.sum()),
"precision_at_k": precision, "chance": chance,
"lift": precision / chance if chance > 0 else np.nan})
chance_all = float(((counts * (counts - 1)).sum())
/ max(1, len(keys) * (len(keys) - 1)))
records.append({"class": "all", "n": int(len(keys)),
"precision_at_k": float(hits.mean()),
"chance": chance_all,
"lift": (float(hits.mean()) / chance_all
if chance_all > 0 else np.nan)})
return pd.DataFrame(records)
[docs]
class RoundResult:
"""What one retrain round produced.
:param fields: named values for the round fields below; omitted list and
mapping fields are normalized to empty containers.
:param round_index: the round number recorded.
:param n_labels: labels the model was fitted on.
:param n_new_labels: labels added since the previous round.
:param report: the :func:`holdout_report` for this round.
:param split_rule: how the held-out set was drawn, in words.
:param scored: how many crops were re-scored in the database.
:param score_columns: the columns written back.
:param model_path: where the fitted model was saved, if it was.
:param card_path: the model card beside it, if one was written.
:param verdict: the :class:`StoppingVerdict` after this round.
:param notes: anything the round wants the annotator to know.
"""
__slots__ = ("round_index", "n_labels", "n_new_labels", "report",
"split_rule", "scored", "score_columns", "model_path",
"card_path", "verdict", "notes", "classes", "model_type")
def __init__(self, **fields: Any):
"""Populate supported fields and normalize missing containers."""
for name in self.__slots__:
setattr(self, name, fields.get(name))
self.notes = list(self.notes or [])
self.score_columns = list(self.score_columns or [])
self.report = dict(self.report or {})
@property
[docs]
def accuracy(self) -> float:
"""Held-out accuracy of this round."""
return float(self.report.get("accuracy", float("nan")))
@property
[docs]
def per_class(self) -> Dict[str, float]:
"""``{class name: held-out accuracy}`` for this round."""
names = self.report.get("classes") or []
accs = self.report.get("per_class_accuracy") or []
return {str(names[i]) if i < len(names) else f"class_{i}": float(a)
for i, a in enumerate(accs)}
[docs]
def __repr__(self) -> str:
"""Return the round, label count, and four-decimal held-out accuracy."""
return (f"RoundResult(round={self.round_index!r}, "
f"n_labels={self.n_labels!r}, accuracy={self.accuracy:.4f})")
[docs]
def summary(self) -> str:
"""One paragraph: the round, its numbers and what to do next."""
lines = [f"Round {self.round_index}: fitted on {self.n_labels} labels "
f"({self.n_new_labels} new), held out "
f"{self.report.get('n', 0)}."]
lines.append(f"Split: {self.split_rule}")
lines.append(f"Held-out accuracy {self.accuracy:.3f} · macro-F1 "
f"{float(self.report.get('f1_macro', float('nan'))):.3f}")
per_class = self.per_class
if per_class:
lines.append("Per class: " + " · ".join(
f"{k} {v:.3f}" for k, v in per_class.items()))
worst = min(per_class, key=per_class.get)
if per_class[worst] < 0.6:
lines.append(
f"Weakest class is {worst} at {per_class[worst]:.3f} — the "
f"aggregate above is not describing it.")
lines.append(f"Re-scored {self.scored} crops into "
f"{', '.join(self.score_columns) or 'nothing'}; the queue "
f"re-ranks on the next rebuild.")
if self.verdict is not None:
lines.append(("STOP — " if self.verdict.stop else "CONTINUE — ")
+ self.verdict.reason)
for note in self.notes:
lines.append(f"! {note}")
return "\n".join(lines)
[docs]
def retrain_round(db_path: str, annotation_column: str = "annotate", *,
features: Optional[pd.DataFrame] = None,
model_type: str = "logistic_regression",
group_by: str = "well", holdout: float = 0.25,
seed: int = 0, min_labels: int = 8,
round_index: Optional[int] = None,
table: str = PNG_TABLE, key: str = PNG_KEY,
image_type: Optional[str] = None,
write_scores: bool = True, save_model: bool = True,
model_dir: Optional[str] = None,
write_card: bool = True,
label_window: int = 50,
min_gain: float = 0.003,
measure: Any = DEFAULT_MEASURE,
diversity: Any = "well",
balance: str = "none",
synthetic_negatives: Optional[int] = None,
rejections: Optional[Mapping[Any, Any]] = None
) -> RoundResult:
"""Fit a model on the labels so far, score every crop, close the loop.
This is the half of active learning that has been missing: the queue put
the informative crops in front of the annotator, and then nothing
happened. Annotating without retraining is not active learning — it is
ordinary annotation in a clever order, and the order goes stale after the
first few dozen labels because it still reflects a model that has not
seen any of them.
One call does all five things the loop needs:
1. fits a model on every label in ``annotation_column``;
2. scores it on a **grouped** held-out split, so the number is not an
artefact of 190 labels coming from one well;
3. writes per-class probabilities back into ``png_list`` as
:data:`ROUND_PRED_PREFIX` columns, which :func:`build_queue` prefers
over the older ``pred`` — so the next queue is genuinely re-ranked;
4. records the round, giving :func:`learning_curve` another point;
5. returns the :class:`StoppingVerdict` for the curve so far.
:param db_path: path to ``measurements.db``.
:param annotation_column: the column holding the labels.
:param features: feature matrix indexed by the crop key. Omitted, it is
read from the measurement tables with :func:`round_features`.
:param model_type: ``'logistic_regression'`` (default — it is the one
that behaves at 20 labels), ``'random_forest'`` or
``'gradient_boosting'``.
:param group_by: ``'well'`` (default), ``'plate'``, ``'field'`` or
``'none'``. What the held-out split refuses to share. Matched
exactly and in lower case; anything else raises, the way an unknown
``diversity=`` does in :func:`build_queue`. When the strategy is
``'none'``, or the crop table has none of the columns it needs, the
split is a stratified random one and ``split_rule`` says ``NOT
grouped`` — it never claims a grouping it did not perform.
:param holdout: fraction held out.
:param seed: makes the split and the fit reproducible.
:param min_labels: refuse to fit below this many labels.
:param round_index: override the round number; defaults to
:func:`next_round`.
:param table: crop table.
:param key: crop key column.
:param image_type: substring filter on the crop key.
:param write_scores: write the new probabilities back into the database.
:param save_model: joblib-dump the fitted model beside the database.
:param model_dir: where to put it; defaults to ``<db dir>/active_learning``.
:param write_card: write a model card beside the saved model.
:param label_window: passed to :func:`should_stop`.
:param min_gain: passed to :func:`should_stop`.
:param measure: recorded with the round, for the queue that follows.
:param diversity: likewise.
:param balance: ``'none'`` (default) leaves imbalance to the estimator's
own ``class_weight='balanced'``; ``'downsample'`` cuts every class to
the size of the smallest BEFORE the grouped split.
THE TWO ARE NOT THE SAME ANSWER. Reweighting and downsampling produce
different probabilities from the same crops, and
:func:`spacr.suggest.suggest_from_scores` sorts on those
probabilities -- so the round records which was in force, in
``notes`` and on the model card. The smaller
class ("if there is class imbalance use the class with fewer").
:param rejections: ``{crop key: rejected class}`` -- suggestions the
annotator REJECTED (:func:`spacr.suggest.rejected_suggestions`). In
a two-class column a rejection of class 1 is an example of class 2,
and it is fitted as one; a crop that has since been labelled is
left to its label, and in a column with any other classes the
rejection cannot be turned into a label and is counted in the notes
instead. A rejection is information, not silence: without this the
model that proposed the wrong class would be fitted on exactly the
same evidence next round and propose it again.
:param synthetic_negatives: how many unannotated crops to draw at random
and fit as the ABSENT class when only one class has been annotated.
``None`` (default) refuses instead, as before.
THIS IS A DELIBERATE LIE AND THE ROUND SAYS SO. A random draw from
the unannotated pool is mostly-negative, not negative, so what comes
back is a ranking rather than a verdict; the count reaches the model
card, because a card that does not say the negatives were invented
describes a model that does not exist. Defined for the binary classes
1 and 2 only -- see :func:`_absent_binary_class`.
:returns: a :class:`RoundResult`.
:raises ValueError: below ``min_labels`` labels, with fewer than two
classes annotated — neither is something to paper over with a model
that will produce a confident-looking ranking out of nothing — or
for an unrecognised ``group_by``.
"""
notes: List[str] = []
con = _connect(db_path)
try:
columns = _table_columns(con, table, db_path)
if annotation_column not in columns:
raise ValueError(
f"{table!r} has no {annotation_column!r} column — there are no "
f"labels to retrain on yet.")
meta = [c for c in ("plateID", "rowID", "columnID", "fieldID")
if c in columns]
select = list(dict.fromkeys([key, annotation_column] + meta))
rows = con.execute(
f"SELECT {', '.join(_quote_ident(c) for c in select)} "
f"FROM {_quote_ident(table)}").fetchall()
finally:
con.close()
crops = pd.DataFrame(rows, columns=select)
if image_type:
crops = crops[crops[key].astype(str).str.contains(
str(image_type), regex=False)]
crops = crops.set_index(key)
crops = crops.loc[~crops.index.duplicated(keep="first")]
if features is None:
features = round_features(db_path, table=table, key=key)
features = features.select_dtypes(include=[np.number])
features = features.loc[~features.index.duplicated(keep="first")]
shared = crops.index.intersection(features.index)
if not len(shared):
raise ValueError(
f"No crop in {table} has a row in the feature matrix, so nothing "
f"can be fitted. Check that Measure and the crop export ran over "
f"the same objects.")
crops = crops.loc[shared]
matrix = features.loc[shared]
labelled_mask = (
crops[annotation_column].notna()
& ~crops[annotation_column].map(_is_suggestion)
)
n_labels = int(labelled_mask.sum())
if n_labels < int(min_labels):
raise ValueError(
f"Only {n_labels} labels in {annotation_column!r} (need at least "
f"{int(min_labels)}). A model fitted on fewer will still emit a "
f"confident-looking ranking, and it will be noise.")
train_index = crops.index[labelled_mask.to_numpy()]
raw_labels = list(crops.loc[train_index, annotation_column].to_numpy())
class_values = sorted({_class_value(v) for v in raw_labels})
rejected_index: List[Any] = []
if rejections:
rejected_index, rejected_labels, unusable = _rejections_as_labels(
rejections, class_values, set(train_index), set(matrix.index))
if rejected_index:
train_index = train_index.append(pd.Index(rejected_index))
raw_labels = raw_labels + rejected_labels
class_values = sorted({_class_value(v) for v in raw_labels})
notes.append(
f"{len(rejected_index)} rejected suggestions were fitted as "
f"the other class: in a two-class column a rejection of "
f"class 1 is an example of class 2.")
if unusable:
notes.append(
f"{unusable} rejected suggestions could not be fitted: only "
f"a column whose classes are 1 and 2 has an 'other class' "
f"to fit a rejection as.")
synthetic_index: List[Any] = []
if len(class_values) == 1 and synthetic_negatives:
present = class_values[0]
absent = _absent_binary_class(present)
if absent is None:
raise ValueError(
f"Only class {present!r} is annotated in "
f"{annotation_column!r}, and synthetic negatives are defined "
f"for the binary classes 1 and 2 only. Annotate an example "
f"of the other class instead.")
already = set(rejected_index)
pool = [i for i in crops.index[crops[annotation_column].isna()]
if i in matrix.index and i not in already]
if len(pool) < int(synthetic_negatives):
raise ValueError(
f"Asked for {int(synthetic_negatives)} synthetic negatives "
f"and only {len(pool)} unannotated crops carry features. "
f"Annotate less, or ask for fewer.")
drawn = np.random.default_rng(int(seed)).choice(
np.asarray(pool, dtype=object), size=int(synthetic_negatives),
replace=False)
synthetic_index = list(drawn)
train_index = train_index.append(pd.Index(synthetic_index))
raw_labels = raw_labels + [absent] * len(synthetic_index)
class_values = sorted({_class_value(v) for v in raw_labels})
notes.append(
f"{len(synthetic_index)} negatives were INVENTED: drawn at "
f"random from the unannotated pool and fitted as class "
f"{absent}, because only class {present} had been annotated. "
f"They are mostly-negative, not negative, so this round is a "
f"RANKING and not a verdict.")
if len(class_values) < 2:
raise ValueError(
f"Every label in {annotation_column!r} is class "
f"{class_values[0] if class_values else 'none'}. A classifier "
f"needs at least two classes; keep annotating until the other "
f"one appears.")
if str(balance).lower() == "downsample":
train_index, raw_labels, dropped = _downsample_to_smallest(
train_index, raw_labels, int(seed))
if dropped:
notes.append(
f"balance=downsample: {dropped} rows of the larger class "
f"were dropped so both classes are the size of the smaller. "
f"The estimator's own `class_weight='balanced'` is therefore "
f"acting on an already-even set.")
else:
notes.append(
"balance=downsample: the classes were already even, so "
"nothing was dropped.")
else:
notes.append(
"balance=none: class imbalance is handled by the estimator's "
"`class_weight='balanced'`, which reweights rather than drops.")
n_labels = len(raw_labels)
class_index = {value: i for i, value in enumerate(class_values)}
y = np.array([class_index[_class_value(v)] for v in raw_labels], dtype=int)
train_matrix = matrix.loc[train_index]
x = np.nan_to_num(train_matrix.to_numpy(dtype=float), nan=0.0,
posinf=0.0, neginf=0.0)
labelled_crops = crops.loc[train_index]
from .classifier_evaluation import (
grouped_split as _shared_grouped_split,
split_group_values as _shared_split_group_values,
)
group_name, groups = _shared_split_group_values(
group_by=group_by, frame=labelled_crops, table=table)
train_idx, test_idx, split_provenance = _shared_grouped_split(
groups, y, holdout, int(seed), group_by=group_name)
split_rule = split_provenance.summary()
notes.append(split_rule)
model = _build_round_model(model_type, int(seed), len(class_values))
model.fit(x[train_idx], y[train_idx])
test_probs = _predict_proba(model, x[test_idx], len(class_values))
report = holdout_report(y[test_idx], test_probs,
[str(v) for v in class_values])
all_x = np.nan_to_num(matrix.to_numpy(dtype=float), nan=0.0,
posinf=0.0, neginf=0.0)
all_probs = _predict_proba(model, all_x, len(class_values))
if round_index is None:
round_index = next_round(db_path, annotation_column)
previous = learning_curve(db_path, annotation_column)
prior_labels = (int(previous["n_labels"].iloc[-1]) if len(previous)
else 0)
n_new = max(0, n_labels - prior_labels)
score_columns: List[str] = []
scored = 0
if write_scores:
score_columns = [f"{ROUND_PRED_PREFIX}{i}"
for i in range(len(class_values))]
scored = _write_round_scores(db_path, matrix.index, all_probs,
score_columns, table=table, key=key)
model_path = ""
card_path = ""
if save_model:
target_dir = model_dir or os.path.join(
os.path.dirname(os.path.abspath(str(db_path))), "active_learning")
os.makedirs(target_dir, exist_ok=True)
model_path = os.path.join(
target_dir, f"round_{int(round_index):03d}_{model_type}.joblib")
try:
import joblib
joblib.dump({"model": model, "classes": class_values,
"features": list(matrix.columns)}, model_path)
except Exception as exc:
notes.append(f"Could not save the round model ({exc}).")
model_path = ""
if model_path and write_card:
card_path = _write_round_card(
model_path, report, split_rule, round_index,
annotation_column, db_path, class_values, matrix.columns,
model_type, n_labels, n_new, notes,
{"balance": str(balance),
"synthetic_negatives": len(synthetic_index),
"rejections_fitted": len(rejected_index),
"class_weight_balanced": _model_reweights(model_type)},
table=table, key=key, image_type=image_type)
per_class = {str(name): float(acc) for name, acc in
zip(report["classes"], report["per_class_accuracy"])}
record_round(db_path, annotation_column, int(round_index),
n_labels=n_labels, n_new_labels=n_new,
n_holdout=report["n"],
holdout_accuracy=report["accuracy"],
holdout_f1_macro=report["f1_macro"],
per_class=per_class, split_rule=split_rule,
model_type=model_type, model_path=model_path,
card_path=card_path,
measure=str(measure), diversity=str(diversity),
notes=notes)
verdict = should_stop(learning_curve(db_path, annotation_column),
label_window=label_window, min_gain=min_gain)
return RoundResult(round_index=int(round_index), n_labels=n_labels,
n_new_labels=n_new, report=report,
split_rule=split_rule, scored=scored,
score_columns=score_columns, model_path=model_path,
card_path=card_path, verdict=verdict, notes=notes,
classes=[str(v) for v in class_values],
model_type=model_type)
def _is_suggestion(value: Any) -> bool:
"""Whether a stored annotation is a SUGGESTION rather than a human answer.
:func:`spacr.suggest.write_suggestions` records a proposal in the same
column as the answers, offset by :data:`spacr.suggest.SUGGESTION_OFFSET`
so the two cannot collide: a suggested 1 is stored as 11. Anything at or
above the offset is therefore the model's own previous output.
Imported inside the call because `spacr.suggest` imports this module for
its estimator, and the dependency is one-way by design.
:param value: a raw value from the annotation column.
:returns: whether it is a suggestion and must not be fitted as a label.
"""
from .suggest import SUGGESTION_OFFSET
normalised = _class_value(value)
return isinstance(normalised, int) and normalised >= SUGGESTION_OFFSET
def _class_value(value: Any) -> Any:
"""``1.0`` and ``1`` are one class; normalise to int where possible."""
if isinstance(value, float) and float(value).is_integer():
return int(value)
if isinstance(value, (int, np.integer)):
return int(value)
return value
def _model_reweights(model_type: str) -> bool:
"""Whether this estimator applies ``class_weight="balanced"`` itself.
The card records it beside ``balance`` because the two compose: a
downsampled set fitted by a reweighting estimator is a third thing
again, and "balanced" on its own does not say which of the three was
run.
:param model_type: the estimator name `_build_round_model` resolves.
:returns: True when the built estimator reweights its classes.
"""
name = str(model_type).lower().replace("-", "_")
return name in ("logistic_regression", "logistic", "lr",
"random_forest", "rf")
def _rejections_as_labels(rejections: Mapping[Any, Any],
class_values: Sequence[Any],
labelled: set, scored: set):
"""Turn rejected suggestions into rows the round can fit.
:param rejections: ``{crop key: rejected class}``.
:param class_values: the classes the annotator's own labels hold.
:param labelled: crop keys that already carry a label. Those win: the
label is the stronger statement and is already in the fit.
:param scored: crop keys that have a row in the feature matrix.
:returns: ``(keys, labels, unusable)`` -- the crops to add, the class
each is fitted as, and how many rejections had no "other class".
"""
binary = set(_class_value(v) for v in class_values) <= {1, 2}
keys: List[Any] = []
labels: List[Any] = []
unusable = 0
seen = set()
for key, refused in dict(rejections).items():
if key in labelled or key not in scored or key in seen:
continue
other = _absent_binary_class(refused) if binary else None
if other is None:
unusable += 1
continue
seen.add(key)
keys.append(key)
labels.append(other)
return keys, labels, unusable
def _absent_binary_class(present: Any) -> Optional[int]:
"""The other of the two binary annotation classes, or ``None``.
spaCR's annotation column holds small integers and the binary case the
request was written for is 1 against 2. Anything else -- class 3, a
string, a float that is not 1 or 2 -- has no defensible "other class"
to invent, and guessing one would fit a model against a class the
caller never named.
:param present: the single class that has been annotated.
:returns: 2 for 1, 1 for 2, and ``None`` for everything else.
"""
value = _class_value(present)
if value == 1:
return 2
if value == 2:
return 1
return None
def _downsample_to_smallest(index, labels: List[Any], seed: int):
"""Cut every class to the size of the smallest, deterministically.
:param index: the training rows, aligned with ``labels``.
:param labels: one raw class value per row.
:param seed: seeds the draw, so a round is reproducible.
:returns: ``(index, labels, dropped)`` -- the kept rows, their labels,
and how many rows were dropped.
DROPPED RATHER THAN REWEIGHTED, which is the instruction
("if there is class imbalance use the class with fewer") and also what
keeps the fitted probability readable as a confidence: a reweighted
fit's probability is a function of the weights as much as of the crop,
and `spacr.suggest` sorts on it.
"""
by_class: Dict[Any, List[int]] = {}
for position, value in enumerate(labels):
by_class.setdefault(_class_value(value), []).append(position)
if len(by_class) < 2:
return index, labels, 0
smallest = min(len(rows) for rows in by_class.values())
rng = np.random.default_rng(int(seed))
keep: List[int] = []
for value in sorted(by_class, key=str):
rows = by_class[value]
if len(rows) > smallest:
rows = list(rng.choice(np.asarray(rows), size=smallest,
replace=False))
keep.extend(int(r) for r in rows)
keep.sort()
dropped = len(labels) - len(keep)
return index[keep], [labels[i] for i in keep], dropped
def _build_round_model(model_type: str, seed: int, n_classes: int):
"""The estimator a round fits. Small-data-first, no torch, no GPU."""
name = str(model_type).lower().replace("-", "_")
if name in ("logistic_regression", "logistic", "lr"):
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
return Pipeline([
("scale", StandardScaler()),
("model", LogisticRegression(max_iter=2000, random_state=seed,
class_weight="balanced")),
])
if name in ("random_forest", "rf"):
from sklearn.ensemble import RandomForestClassifier
return RandomForestClassifier(n_estimators=300, random_state=seed,
n_jobs=-1, class_weight="balanced")
if name in ("gradient_boosting", "hist_gradient_boosting", "gb"):
from sklearn.ensemble import HistGradientBoostingClassifier
return HistGradientBoostingClassifier(random_state=seed)
if name in ("xgboost", "xgb"):
try:
from xgboost import XGBClassifier
except ImportError as exc:
raise ValueError(
"model_type 'xgboost' needs the xgboost package, which "
"spaCR does not install: `pip install xgboost`. The same "
"algorithm is available now as "
"model_type='gradient_boosting', which is scikit-learn's "
"HistGradientBoostingClassifier and needs nothing extra."
) from exc
return XGBClassifier(
random_state=seed,
objective=("binary:logistic" if n_classes <= 2
else "multi:softprob"),
eval_metric="logloss",
)
raise ValueError(
f"Unknown model_type {model_type!r}; use 'logistic_regression', "
f"'random_forest', 'gradient_boosting' or 'xgboost'.")
def _predict_proba(model: Any, x: np.ndarray, n_classes: int) -> np.ndarray:
"""``(N, n_classes)`` probabilities, padding classes the fit never saw.
A fold that happened to contain only two of three classes leaves an
estimator whose ``predict_proba`` has two columns. Returning that
unpadded would silently renumber the classes downstream — the queue
would call class 2 "class 1" — so the missing columns are filled with
zeros in the right places instead.
"""
if not len(x):
return np.zeros((0, n_classes))
probs = np.asarray(model.predict_proba(x), dtype=float)
seen = np.asarray(getattr(model, "classes_", np.arange(probs.shape[1])),
dtype=int)
if probs.shape[1] == n_classes and np.array_equal(seen,
np.arange(n_classes)):
return probs
out = np.zeros((probs.shape[0], n_classes), dtype=float)
for column, class_id in enumerate(seen):
if 0 <= int(class_id) < n_classes:
out[:, int(class_id)] = probs[:, column]
return out
def _write_round_scores(db_path: str, keys: Any, probs: np.ndarray,
columns: Sequence[str], table: str = PNG_TABLE,
key: str = PNG_KEY) -> int:
"""Write per-class probabilities back into ``table``; rows updated.
Goes through :func:`spacr.predictions.merge_prediction_results`, which
already knows that ``png_list`` has a real column called ``rowID`` that
shadows SQLite's ``rowid`` — the exact trap a hand-rolled UPDATE here
would fall into.
"""
from .predictions import merge_prediction_results
frame = pd.DataFrame(np.asarray(probs, dtype=float),
columns=list(columns))
frame[key] = list(keys)
report = merge_prediction_results(
frame, db_path, {c: (c, "REAL") for c in columns},
table=table, key=key, verbose=False)
return int(getattr(report, "matched_rows", 0) or 0)
def _write_round_card(model_path: str, report: Dict[str, Any],
split_rule: str, round_index: int,
annotation_column: str, db_path: str,
class_values: Sequence[Any], feature_columns: Any,
model_type: str, n_labels: int, n_new: int,
notes: List[str], balancing: Dict[str, Any],
table: str = PNG_TABLE,
key: str = PNG_KEY,
image_type: Optional[str] = None) -> str:
"""Write the model card for one round's model. Never fatal.
The coverage block is read back with the round's own ``table``, ``key``
and ``image_type``, so the card describes the crops the round was fitted
on rather than everything the database happens to hold.
"""
try:
from .deep_spacr import model_card
coverage = annotation_coverage(db_path, annotation_column,
table=table, key=key,
image_type=image_type)
coverage_meta = dict(
coverage.attrs.get("spacr_annotation_coverage", {}))
card, card_path, _artifact = model_card(
model_path,
settings={"annotation_column": annotation_column,
"model_type": model_type, "round": int(round_index),
"db_path": str(db_path)},
classes=[str(v) for v in class_values],
split_rule=split_rule,
held_out=report,
class_balance={"annotated": coverage_meta.get("by_class", {})},
dataset_src=os.path.dirname(os.path.abspath(str(db_path))),
module="active_learning",
extra={
"round": int(round_index),
"n_labels": int(n_labels),
"n_new_labels": int(n_new),
"n_features": int(len(list(feature_columns))),
"annotation_coverage": {
k: coverage_meta.get(k) for k in
("by_class", "by_plate", "by_well", "by_round",
"concentration", "wells_annotated", "plates_annotated")},
"balancing": dict(balancing),
},
)
return card_path
except Exception as exc:
notes.append(f"Round model card could not be written ({exc}).")
return ""
_TTA_TRANSFORMS: Tuple[str, ...] = ("identity", "flip_lr", "flip_ud", "rot90")
_UNCERTAINTY_MATCH_IOU = 0.5
_UNCERTAIN_PIXEL = 0.5
_NEAR_MISS_MARGIN = 2.0
_NEAR_MISS_RING = 2
_FLOW_ERROR_LIMIT = 0.4
def _spatial_axes(image: np.ndarray) -> Tuple[int, int]:
"""The two image axes a flip or a rotation acts on.
A channel-first stack, a leading axis of four or fewer planes in front
of a larger last axis, is turned on its last two axes; every other
array on its first two, which covers a plain field and a channel-last
one.
:param image: a 2-D field, or a 3-D field with a channel axis.
:returns: the pair of axes.
"""
if image.ndim == 3 and image.shape[0] <= 4 < image.shape[-1]:
return (1, 2)
return (0, 1)
def _tta_forward(image: np.ndarray, name: str) -> np.ndarray:
"""``image`` with one test-time transform applied.
:param image: the field.
:param name: one of ``identity``, ``flip_lr``, ``flip_ud`` or ``rot90``.
:returns: the transformed field, as a contiguous copy.
:raises ValueError: for a transform name not listed.
"""
rows, cols = _spatial_axes(image)
if name == "identity":
out = image
elif name == "flip_lr":
out = np.flip(image, axis=cols)
elif name == "flip_ud":
out = np.flip(image, axis=rows)
elif name == "rot90":
out = np.rot90(image, 1, axes=(rows, cols))
else:
raise ValueError(f"unknown test-time transform {name!r}; expected "
f"one of {list(_TTA_TRANSFORMS)}")
return np.ascontiguousarray(out)
def _tta_inverse(labels: np.ndarray, name: str) -> np.ndarray:
"""A label image segmented under ``name``, turned back onto the field.
:param labels: the 2-D label image the model returned.
:param name: the transform the image was segmented under.
:returns: the label image in the field's own orientation.
"""
labels = np.asarray(labels)
if name == "identity":
return labels
if name == "flip_lr":
return np.ascontiguousarray(labels[:, ::-1])
if name == "flip_ud":
return np.ascontiguousarray(labels[::-1, :])
if name == "rot90":
return np.ascontiguousarray(np.rot90(labels, -1))
raise ValueError(f"unknown test-time transform {name!r}")
def _tta_label_sets(image: Any, segment: Callable[[np.ndarray], Any],
transforms: Sequence[str] = _TTA_TRANSFORMS
) -> List[np.ndarray]:
"""Segment one field once per test-time transform, all in its frame.
Flips and a quarter turn change nothing about the biology, so a model
that is sure of an object draws it the same way every time; where the
passes disagree, the model was guessing. The first transform is the
reference whose objects are scored, so it should be ``identity``.
:param image: the field, 2-D or with one channel axis.
:param segment: ``image -> labels``, any segmenter at all (Cellpose, a
threshold, another backend), called once per transform.
:param transforms: the transforms to run, from ``identity``,
``flip_lr``, ``flip_ud`` and ``rot90``.
:returns: one int32 label image per transform, each in the field's own
orientation.
"""
return _tta_passes(image, segment, transforms)["labels"]
def _tta_passes(image: Any, segment: Callable[[np.ndarray], Any],
transforms: Sequence[str] = _TTA_TRANSFORMS
) -> Dict[str, Any]:
"""Every test-time pass of one field, with Cellpose's maps when given.
As :func:`_tta_label_sets`, but a segmenter that returns
``(labels, cell probability)`` or ``(labels, cell probability, flow
vectors)`` has those kept too: each probability map turned back onto
the field, and the flow vectors of the first pass only, since turning
vectors back would also mean turning the vectors themselves.
:param image: the field, 2-D or with one channel axis.
:param segment: ``image -> labels`` or a tuple as above.
:param transforms: the transforms to run; the first is the reference.
:returns: ``{"labels": [...], "probabilities": [...] or None,
"vectors": (2, H, W) array or None}``.
"""
field = np.asarray(image)
labels_out: List[np.ndarray] = []
probabilities: List[np.ndarray] = []
vectors = None
for index, name in enumerate(transforms):
output = segment(_tta_forward(field, name))
parts = output if isinstance(output, tuple) else (output,)
labels = _tta_inverse(np.asarray(parts[0]), name).astype(np.int32)
labels_out.append(labels)
if len(parts) > 1 and parts[1] is not None:
probability = _tta_inverse(np.asarray(parts[1], np.float32), name)
if probability.shape == labels.shape:
probabilities.append(probability)
if index == 0 and len(parts) > 2 and parts[2] is not None:
flow = np.asarray(parts[2], np.float32)
if flow.shape == (2,) + labels.shape:
vectors = flow
return {"labels": labels_out,
"probabilities": (probabilities if len(probabilities)
== len(labels_out) else None),
"vectors": vectors}
def _flow_errors(labels: np.ndarray, vectors: np.ndarray
) -> Dict[int, float]:
"""Cellpose's own flow error of every object, by label.
The flows an object's outline implies are compared with the flows the
network predicted, as Cellpose does before discarding an object at its
flow threshold (0.4 by default).
:param labels: the reference label image.
:param vectors: the network's ``(2, H, W)`` flow vectors for it.
:returns: ``{label: error}``, empty when Cellpose is not importable or
there are no objects.
"""
ids = np.unique(labels)
ids = ids[ids != 0]
if not ids.size:
return {}
try:
import torch
from cellpose.dynamics import flow_error
except Exception: # noqa: BLE001
return {}
compact = np.searchsorted(np.concatenate([[0], ids]), labels)
errors, _ = flow_error(compact.astype(np.int32), np.asarray(vectors),
device=torch.device("cpu"))
return {int(label): float(error) for label, error in zip(ids, errors)}
def _segmentation_uncertainty(label_sets: Sequence[Any],
match_iou: float = _UNCERTAINTY_MATCH_IOU,
probabilities: Optional[Sequence[Any]] = None,
vectors: Optional[Any] = None,
probability_threshold: float = 0.0
) -> Dict[str, Any]:
"""How much repeated segmentations of one field disagree.
The passes may be test-time transforms of one model or several models
(an ensemble); a model can repeat an error under every transform, and
only another model, or its own cell probability, can show it.
``map``
Per pixel, in ``[0, 1]``: the largest of how split the passes are
on foreground against background (``4 p (1 - p)``, 1 when half the
passes call a pixel an object), the uncertainty of the reference
object the pixel belongs to and, with ``probabilities``, ``4 q (1 -
q)`` of the cell probability ``q`` of a near miss (below).
``objects``
Per object of the first (reference) pass: one minus its mean best
IoU with any object of each other pass (0 is an object every pass
drew identically, 1 one no other pass drew), or, with ``vectors``,
its Cellpose flow error divided by 0.4 (Cellpose's own discard
threshold, capped at 1) when that is larger.
``area``
The fraction of the foreground (every pixel any pass drew) whose
disagreement, the first two readings of ``map``, is at least 0.5.
``missed``
With ``probabilities`` (one cell-probability logit map per pass),
the pixels whose mean logit is above ``probability_threshold``
less 2 and that lie more than 2 pixels from anything a pass drew,
as a fraction of the foreground (capped at 1): objects every pass
missed the same way, which disagreement cannot see. 0 without
probabilities.
``field``
``area + missed``, the field score the curation queue sorts on.
``spread``
One minus the mean panoptic quality of each other pass against the
reference, matching objects one to one at ``match_iou``.
These are orderings, not probabilities of error: use them to rank
objects and fields for review.
:param label_sets: two or more label images of one field, same shape.
:param match_iou: the IoU at which two passes' objects are one object.
:param probabilities: optional cell-probability logit maps, one per pass
or any number, each the field's shape.
:param vectors: optional ``(2, H, W)`` Cellpose flow vectors of the
reference pass.
:param probability_threshold: the cell-probability threshold the passes
were drawn at. Default 0.0.
:returns: ``{"map", "objects", "area", "missed", "field", "spread",
"flow_errors", "n_objects", "n_passes"}``, ``objects`` and
``flow_errors`` being ``{label: value}`` for the reference pass.
:raises ValueError: for fewer than two passes or mismatched shapes.
"""
from scipy import ndimage as ndi
from .scorecard import iou_matrix, match_objects
sets = [np.asarray(labels) for labels in label_sets]
if len(sets) < 2:
raise ValueError("segmentation uncertainty needs at least two passes")
shape = sets[0].shape
if any(labels.shape != shape for labels in sets):
raise ValueError("every pass must have the field's shape")
reference = sets[0]
others = sets[1:]
foreground = np.mean([labels > 0 for labels in sets], axis=0)
pixel = (4.0 * foreground * (1.0 - foreground)).astype(np.float32)
ref_ids = np.unique(reference)
ref_ids = ref_ids[ref_ids != 0]
best = np.zeros((ref_ids.size, len(others)), dtype=float)
qualities = []
for column, other in enumerate(others):
ious, t_ids, _p_ids = iou_matrix(reference, other)
if ious.size and t_ids.size:
best[:, column] = ious.max(axis=1)
match = match_objects(reference, other, threshold=match_iou)
matched = len(match.pairs)
misses = (match.n_truth - matched) + (match.n_pred - matched)
denominator = matched + 0.5 * misses
qualities.append(sum(match.ious) / denominator if denominator
else 1.0)
per_object = 1.0 - best.mean(axis=1) if ref_ids.size else np.zeros(0)
def _paint(values: np.ndarray) -> np.ndarray:
""":param values: one value per reference object, in id order.
:returns: each object's pixels set to its value, 0 elsewhere.
"""
if not ref_ids.size:
return np.zeros(shape, dtype=np.float32)
lookup = np.zeros(int(reference.max()) + 1, dtype=np.float32)
lookup[ref_ids] = values
return lookup[np.where(reference > 0, reference, 0)] * (reference > 0)
disagreement = np.maximum(pixel, _paint(per_object))
drawn = foreground > 0
extent = int(drawn.sum())
area = (float((disagreement >= _UNCERTAIN_PIXEL).sum() / extent)
if extent else 0.0)
flows = _flow_errors(reference, vectors) if vectors is not None else {}
if flows:
flow_score = np.array([min(1.0, flows.get(int(label), 0.0)
/ _FLOW_ERROR_LIMIT) for label in ref_ids])
per_object = np.maximum(per_object, flow_score)
objects = {int(label): float(value)
for label, value in zip(ref_ids, per_object)}
uncertainty = np.maximum(pixel, _paint(per_object))
missed = 0.0
if probabilities is not None and len(probabilities):
logit = np.mean([np.asarray(p, np.float32) for p in probabilities],
axis=0)
if logit.shape != shape:
raise ValueError("every probability map must have the field's "
"shape")
cut = float(probability_threshold)
near = ((logit > cut - _NEAR_MISS_MARGIN)
& ~ndi.binary_dilation(drawn, iterations=_NEAR_MISS_RING))
count = int(near.sum())
if count:
missed = float(count / max(extent, count))
q = 1.0 / (1.0 + np.exp(-(logit - cut)))
uncertainty = np.where(
near, np.maximum(uncertainty, 4.0 * q * (1.0 - q)),
uncertainty).astype(np.float32)
return {
"map": uncertainty.astype(np.float32),
"objects": objects,
"area": area,
"missed": missed,
"field": float(area + missed),
"spread": float(1.0 - np.mean(qualities)),
"flow_errors": flows,
"n_objects": int(ref_ids.size),
"n_passes": len(sets),
}