"""Timelapse live preview — segment once, re-link live.
The Timelapse module is mask generation over a time series followed by
frame-to-frame linking. Those two halves cost wildly different amounts:
segmenting twelve frames with Cellpose is tens of seconds, re-linking the
*same* twelve label images with a new ``timelapse_displacement`` is
milliseconds. A preview that re-segments on every slider move is unusable,
so this panel splits them:
* **Per-frame masks are cached** under a *segmentation signature* — the
source path, the frame indices, and every setting that can change a
label image (model, channel, diameter, flow threshold, cell probability,
normalisation). Change a *tracking* setting and the signature is
unchanged, the cache hits, and only :func:`link_tracks` runs. Change a
segmentation setting and the signature moves, so the masks are rebuilt.
* **The sequence is read lazily.** :class:`FrameSequence` never
materialises the whole time series: a directory reads one file at a
time, a multi-page TIFF reads one page at a time, and an ``.npy`` stack
is memory-mapped and sliced. Only the frames the preview actually shows
are touched, and a small LRU keeps the scrubber responsive without
pinning the movie in RAM.
What the panel *shows* is chosen around the two failure modes people
actually tune a tracker against:
* **Fragmentation** — one object becoming several track ids. Surfaced as
track count, mean/median track length, the number of tracks shorter
than a live-settable N, and the count of tracks that start after the
first frame or end before the last one.
* **Identity swaps** — a track jumping to a different object. Surfaced as
the number of within-track steps longer than the displacement limit the
user is tuning, plus the largest single step. Masks are relabelled *by
track id* and drawn in a per-track colour, so a swap is visible as an
object changing colour mid-movie.
Both indicators are computed without ground truth and are labelled in the
UI as indicators, not measurements.
Optional backends (``trackastra``, ``ultrack``) are detected before they
are called: absent, the panel writes one inline line naming the package
and the install command. An ``ImportError`` traceback never reaches the
user.
"""
from __future__ import annotations
import logging
import os
import re
import hashlib
import sys
import tempfile
import threading
import time
import weakref
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple
import numpy as np
from PySide6.QtCore import Qt, QThread, QTimer, Signal
from PySide6.QtWidgets import (
QCheckBox, QComboBox, QDialog, QDoubleSpinBox, QFileDialog, QFormLayout,
QAbstractItemView, QGroupBox, QHBoxLayout, QLabel, QLineEdit, QPushButton, QSizePolicy,
QSlider, QSpinBox, QTableWidget, QVBoxLayout, QWidget,
)
from ..i18n import tr
from ..hidpi import scaled_for
from .sortable_table import install_sorting, table_item
from .preview_controls import (
DEFAULT_MAX_SETS, MAX_SETS_TOOLTIP, FlatButton, FlatComboBox, FlatSpinBox,
ImageSetSampler, apply_sample_to_combo, populate_channel_combo,
selected_channel, sibling_sources,
)
from .preview_contract import (
PREVIEW_CANCEL_TEXT, PREVIEW_RUN_TEXT, LivePreviewContract,
preview_cellpose_model, preview_failure_message,
)
from .toggle import Toggle
from .. import path_probe
from ..job_runner import JobRunner
from .live_preview import (
_boundary_mask,
_model_the_run_would_use,
_offer_the_run_model,
_select_channel,
_to_uint8,
_ZoomView,
numpy_to_qpixmap,
)
LOG = logging.getLogger("spacr.qt.timelapse_preview")
FRAME_SUFFIXES = (".tif", ".tiff", ".png", ".jpg", ".jpeg", ".npy")
#: Linking backends, in the order ``timelapse_mode`` documents them.
TRACK_MODES = ("trackastra", "ultrack", "trackpy", "btrack", "iou")
#: Backends that live behind an optional dependency, and the pip target that
#: installs them. Consulted before any import so the panel can report a
#: missing package as one inline line instead of an ImportError traceback.
OPTIONAL_BACKENDS: Dict[str, str] = {
"trackastra": "pip install trackastra",
"ultrack": "pip install spacr[ultrack]",
"btrack": "pip install btrack",
}
#: Distinct track colours, cycled by track id. Chosen to stay separable on
#: a dark micrograph and against each other.
TRACK_COLOURS: Tuple[Tuple[int, int, int], ...] = (
(32, 220, 32), (222, 82, 200), (32, 200, 220), (255, 220, 32),
(240, 100, 60), (120, 140, 255), (255, 150, 200), (150, 255, 190),
(255, 190, 110), (180, 120, 240), (110, 220, 255), (220, 220, 220),
)
_LIVE_FRAME_SEQUENCES: "weakref.WeakSet[FrameSequence]" = weakref.WeakSet()
_LIVE_PREVIEW_PANELS: "weakref.WeakSet[TimelapsePreviewPanel]" = \
weakref.WeakSet()
def _live_cache_owners():
"""Decoded-frame and derived-mask caches that still have real owners."""
return tuple(_LIVE_FRAME_SEQUENCES) + tuple(_LIVE_PREVIEW_PANELS)
def _ensure_cache_budget_sweep() -> None:
"""Arm the shared memory-budget sweep, if the cleanup module is loaded.
Looked up in ``sys.modules`` rather than imported: this runs at widget
construction, and importing the cleanup machinery in order to register
with it would pull it in on every preview whether or not anything else
wanted it.
"""
cleanup = sys.modules.get("spacr.qt.resource_cleanup")
install = getattr(cleanup, "install_budget_sweep", None)
if callable(install):
install()
[docs]
class TrackerUnavailable(RuntimeError):
"""A linking backend cannot run here, with an actionable reason.
Raised instead of letting an ``ImportError`` escape so the panel can put
the message inline. The string always names the package and the command
that fixes it.
"""
def _natural_key(name: str):
"""Sort key that orders ``f_2`` before ``f_10`` (frame indices are numbers)."""
return [int(p) if p.isdigit() else p.lower()
for p in re.split(r"(\d+)", str(name))]
[docs]
class FrameSequence:
"""A time series read one frame at a time.
Three layouts are understood, and none of them is ever read whole:
* a **directory** of per-frame image files — one file opened per access;
* a **multi-page TIFF** — one page decoded per access
(``tifffile.imread(path, key=i)``);
* an ``.npy`` **stack** whose first axis is time — memory-mapped, so a
slice touches only that plane's pages.
:param kind: ``"files"``, ``"tiff"`` or ``"npy"``.
:param source: the list of paths (``files``) or the single path.
:param n_available: how many frames exist on disk.
:param indices: the subset of frame indices this sequence exposes, so a
400-frame movie can be previewed as its first 12 frames.
:param label: what to call the sequence in the UI. Empty falls back to
``str(source)``, which for a file list is the list -- fine for one
path, unreadable for four hundred, so anything user-facing should
pass a name.
:param cache_size: how many decoded frames to keep in the LRU.
:ivar read_count: number of decodes actually performed — the instrument
the tests assert against to prove nothing is read eagerly.
"""
def __init__(self, kind: str, source, n_available: int,
indices: Sequence[int], label: str = "", cache_size: int = 6):
"""Hold one sequence of frames, read lazily and cached.
:param kind: what the frames are -- images, masks, or an overlay.
:param source: where to read them from.
:param n_available: how many frames exist.
:param indices: which of them this sequence shows.
:param label: the caption for this sequence.
:param cache_size: how many decoded frames to keep.
"""
self.kind = kind
self.source = source
self.n_available = int(n_available)
self.indices: List[int] = list(indices)
self.label = label or str(source)
self._cache: "dict[int, np.ndarray]" = {}
self._cache_order: List[int] = []
self._cache_last_used: Dict[int, float] = {}
self._cache_lock = threading.RLock()
self._cache_size = max(1, int(cache_size))
self._memmap = None
self.read_count = 0
@classmethod
[docs]
def open(cls, path, max_frames: int = 12) -> "FrameSequence":
"""Open ``path`` as a sequence, reading at most metadata to do so.
:param path: a directory of frames, a multi-page TIFF, or an ``.npy``
stack whose first axis is time.
:param max_frames: cap on the number of frames the preview exposes.
:raises FileNotFoundError: if the path does not exist.
:raises ValueError: if the path holds no usable time series.
"""
p = Path(path)
if not p.exists():
raise FileNotFoundError(f"No such file or directory: {p}")
cap = max(1, int(max_frames))
if p.is_dir():
files = sorted(
(f for f in p.iterdir()
if f.is_file() and f.suffix.lower() in FRAME_SUFFIXES),
key=lambda f: _natural_key(f.name),
)
if not files:
raise ValueError(
f"{p.name} holds no {'/'.join(FRAME_SUFFIXES)} frames.")
if len(files) < 2:
raise ValueError(
f"{p.name} holds a single frame — a timelapse preview "
"needs at least two.")
idx = list(range(min(len(files), cap)))
return cls("files", files, len(files), idx, label=str(p))
suf = p.suffix.lower()
if suf == ".npy":
arr = np.load(str(p), mmap_mode="r")
if arr.ndim < 3:
raise ValueError(
f"{p.name} has shape {arr.shape}; a timelapse stack needs "
"(T, H, W) or (T, H, W, C).")
n = int(arr.shape[0])
if n < 2:
raise ValueError(
f"{p.name} has {n} frame(s) — a timelapse preview needs "
"at least two.")
return cls("npy", arr, n, list(range(min(n, cap))), label=str(p))
if suf in (".tif", ".tiff"):
import tifffile
with tifffile.TiffFile(str(p)) as tf:
n_pages = len(tf.pages)
try:
shape = tuple(tf.series[0].shape)
except Exception:
shape = tuple(tf.pages[0].shape)
if n_pages > 1:
kind, n = "tiff", n_pages
else:
kind = "tiffmm"
n = int(shape[0]) if len(shape) >= 3 else 1
if n < 2:
raise ValueError(
f"{p.name} holds {n} frame(s) — a timelapse preview needs "
"a multi-frame TIFF or a folder of frames.")
return cls(kind, p, n, list(range(min(n, cap))), label=str(p))
raise ValueError(
f"{p.name}: unsupported input. Drop a folder of frames, a "
"multi-page TIFF, or a (T, H, W) .npy stack.")
[docs]
def __len__(self) -> int:
"""How many frames this sequence shows.
:returns: the frame count.
"""
return len(self.indices)
@property
[docs]
def truncated(self) -> bool:
"""Whether the preview is showing fewer frames than exist on disk."""
return len(self.indices) < self.n_available
[docs]
def frame(self, i: int) -> np.ndarray:
"""Return preview-frame ``i`` (0-based over :attr:`indices`).
:param i: preview-frame position; outside ``0`` to
``len(indices) - 1`` raises :class:`IndexError`. Frames read are
kept in a small cache.
"""
if not (0 <= i < len(self.indices)):
raise IndexError(i)
real = self.indices[i]
with self._cache_lock:
hit = self._cache.get(real)
if hit is not None:
self._cache_last_used[real] = time.time()
return hit
arr = self._read(real)
self.read_count += 1
with self._cache_lock:
self._cache[real] = arr
if real in self._cache_order:
self._cache_order.remove(real)
self._cache_order.append(real)
self._cache_last_used[real] = time.time()
while len(self._cache_order) > self._cache_size:
old = self._cache_order.pop(0)
self._cache.pop(old, None)
self._cache_last_used.pop(old, None)
return arr
def _cache_budget_entries(self):
"""Measured decoded-frame entries for the process-wide policy."""
if not self._cache_lock.acquire(blocking=False):
return []
now = time.time()
try:
return [
(real, max(0, int(frame.nbytes)),
float(self._cache_last_used.get(real, now)), False)
for real, frame in list(self._cache.items())
]
finally:
self._cache_lock.release()
def _register_cache_budget(self) -> None:
"""Publish this sequence after its worker hands it to the GUI thread."""
_LIVE_FRAME_SEQUENCES.add(self)
_ensure_cache_budget_sweep()
def _drop_cache_budget_entry(self, real) -> bool:
"""Forget one decoded frame; the source can reproduce it exactly."""
if not self._cache_lock.acquire(blocking=False):
return False
real = int(real)
try:
existed = real in self._cache
self._cache.pop(real, None)
self._cache_last_used.pop(real, None)
try:
self._cache_order.remove(real)
except ValueError:
pass
return existed
finally:
self._cache_lock.release()
def _read(self, real: int) -> np.ndarray:
"""Decode one frame, going to the source only on a cache miss.
CACHED BECAUSE SCRUBBING RE-READS. A user dragging the scrub bar asks
for the same frames repeatedly, and decoding each time makes the drag
as slow as the disk.
:param real: the frame's index in the source.
:returns: the decoded frame.
"""
if self.kind == "files":
path = self.source[real]
if path.suffix.lower() == ".npy":
return np.asarray(np.load(str(path)))
if path.suffix.lower() in (".tif", ".tiff"):
import tifffile
return np.asarray(tifffile.imread(str(path)))
from PIL import Image
with Image.open(path) as im:
return np.asarray(im)
if self.kind == "npy":
return np.asarray(self.source[real])
import tifffile
if self.kind == "tiffmm":
if self._memmap is None:
try:
self._memmap = tifffile.memmap(str(self.source))
except Exception:
LOG.debug("tifffile.memmap unavailable", exc_info=True)
self._memmap = tifffile.imread(str(self.source))
return np.asarray(self._memmap[real])
return np.asarray(tifffile.imread(str(self.source), key=real))
[docs]
def describe(self) -> str:
"""One-line summary for the status label."""
shown = len(self.indices)
if self.truncated:
return (f"{os.path.basename(self.label)} · showing {shown} of "
f"{self.n_available} frames")
return f"{os.path.basename(self.label)} · {shown} frames"
[docs]
def frame_channel(frame: np.ndarray, channel: int) -> np.ndarray:
"""Return a 2-D plane from a frame stored either (H, W, C) or (C, H, W).
A merged frame written by spaCR is channel-last, but a raw acquisition
TIFF page is often channel-first. Guessing wrong turns a 3-channel image
into a 3-pixel-tall one, so pick the axis that actually looks like a
channel axis (small, and not the only small axis).
:param frame: the frame array; a 2-D frame is returned as is and one
with other than three dimensions is squeezed.
:param channel: the channel index to take (wrapped modulo the channel
count on a channel-first frame).
"""
arr = np.asarray(frame)
if arr.ndim == 2:
return arr
if arr.ndim != 3:
return arr.squeeze()
if arr.shape[-1] <= 8 and arr.shape[0] > 8:
return _select_channel(arr, channel)
if arr.shape[0] <= 8 and arr.shape[-1] > 8:
return arr[int(channel) % arr.shape[0]]
return _select_channel(arr, channel)
[docs]
def segment_frame(image: np.ndarray, params: Dict[str, Any]) -> np.ndarray:
"""Segment one frame with Cellpose and return an ``int32`` label image.
Cellpose is imported inside the call so importing this module — as the
test suite does — costs nothing and needs no CUDA stack. The panel calls
this through the module global, which is also what lets a test swap in a
counting stub to prove that tuning a *tracking* setting never reaches
segmentation.
:param image: the frame, 2-D or with a channel axis.
:param params: segmentation settings; ``model``, ``channel``,
``normalise``, ``lo_pct``, ``hi_pct``, ``diameter``,
``flow_threshold`` and ``cellprob`` are read, each with a default.
A ``cellpose3:...`` model is segmented in the Cellpose 3 backend by
:func:`spacr.object._cellpose3_masks`, as the run segments it, and a
``cellpose_dino:<path>`` model in the Cellpose-DINO backend by
:func:`spacr.object._cellpose_dino_masks`.
"""
from ...object import _prefixed_model_route
name = str(params.get("model", "cpsam"))
route = _prefixed_model_route(name)
model = None if route else preview_cellpose_model(name)
plane = frame_channel(image, int(params.get("channel", 0)))
if params.get("normalise", True):
plane = _to_uint8(plane, normalise=True,
lo_pct=float(params.get("lo_pct", 2.0)),
hi_pct=float(params.get("hi_pct", 98.0)))
if route:
from ... import _segmentation_backends
backend, masks_of = route
diameter = float(params.get("diameter", 30.0))
settings = {
"cell_diameter": diameter or None,
"cell_flow_threshold": float(params.get("flow_threshold", 0.4)),
"cell_cellprob_threshold": float(params.get("cellprob", 0.0)),
}
masks, _flows = masks_of(
_segmentation_backends._load_backend(
backend, model_name=name, object_type="cell"),
[plane], settings, "cell", min_size=15,
default_diameter=diameter or 30.0)
return np.asarray(masks[0]).astype(np.int32)
result = model.eval(
plane,
diameter=float(params.get("diameter", 30.0)) or None,
flow_threshold=float(params.get("flow_threshold", 0.4)),
cellprob_threshold=float(params.get("cellprob", 0.0)),
)
mask = result[0] if isinstance(result, (list, tuple)) else result
if isinstance(mask, list):
mask = mask[0]
return np.asarray(mask).astype(np.int32)
def _as_label_stack(seq: "FrameSequence", channel: int = 0) -> np.ndarray:
"""Stack a mask sequence into (T, H, W) ``int32`` labels, frame by frame."""
planes = [np.asarray(frame_channel(seq.frame(i), channel)).astype(np.int32)
for i in range(len(seq))]
return np.stack(planes, axis=0)
[docs]
def segment_sequence(seq: "FrameSequence", params: Dict[str, Any]) -> np.ndarray:
"""Segment every preview frame of ``seq`` into a (T, H, W) label stack.
Reads and segments one frame at a time so the whole movie is never
resident, then stacks only the label images (which are far smaller than
the raw multi-channel frames).
:param seq: the frame sequence to segment.
:param params: segmentation settings, as :func:`segment_frame` takes
them.
"""
masks = [segment_frame(seq.frame(i), params) for i in range(len(seq))]
shapes = {m.shape for m in masks}
if len(shapes) != 1:
raise ValueError(
f"frames segmented to different shapes {sorted(shapes)}; the "
"sequence is not a single field of view.")
return np.stack(masks, axis=0).astype(np.int32)
[docs]
def backend_available(mode: str) -> Tuple[bool, str]:
"""Whether linking backend ``mode`` can run, and why not when it cannot.
Checks with :func:`importlib.util.find_spec`, so nothing heavy is
imported just to answer the question and a missing optional dependency
never raises.
:param mode: linking backend name, one of :data:`TRACK_MODES`
(case-insensitive); ``iou`` needs no optional package.
:returns: ``(True, "")`` when usable, else ``(False, message)`` where the
message names the package and the command that installs it.
"""
mode = (mode or "").lower()
if mode not in TRACK_MODES:
return False, f"Unknown linking mode {mode!r}."
pkg = {"trackastra": "trackastra", "ultrack": "ultrack",
"btrack": "btrack", "trackpy": "trackpy"}.get(mode)
if pkg is None:
return True, ""
import importlib.util
if importlib.util.find_spec(pkg) is not None:
return True, ""
fix = OPTIONAL_BACKENDS.get(mode, f"pip install {pkg}")
alt = ", ".join(m for m in ("trackpy", "iou") if m != mode)
return False, (
f"timelapse_mode='{mode}' needs the {pkg} package, which is not "
f"installed. Install it with `{fix}`, or preview with {alt}.")
def _tracks_from_features(tracks_df, features):
"""Attach centroids to a track table that only carries labels."""
cols = ["frame", "original_label", "x", "y"]
return tracks_df.merge(features[cols], on=["frame", "original_label"],
how="left", validate="many_to_one")
def _link_iou(masks: np.ndarray, iou_threshold: float):
"""Link by frame-to-frame IoU using spaCR's own linker."""
from spacr.timelapse import _prepare_for_tracking, _track_by_iou
df = _track_by_iou(masks, iou_threshold=float(iou_threshold))
return _tracks_from_features(df, _prepare_for_tracking(masks))
def _link_trackpy(masks: np.ndarray, displacement: float, memory: int):
"""Link with trackpy's nearest-neighbour linker.
Uses the same feature table the pipeline builds
(:func:`spacr.timelapse._prepare_for_tracking`) and the same two knobs
the user tunes — ``search_range`` (``timelapse_displacement``) and
``memory`` (``timelapse_memory``) — so what the preview shows is what a
run will do.
"""
import trackpy as tp
from spacr.timelapse import _prepare_for_tracking
features = _prepare_for_tracking(masks)
try:
tp.quiet()
except Exception:
LOG.debug("trackpy.quiet() unavailable", exc_info=True)
linked = tp.link_df(features, search_range=float(displacement),
memory=int(memory))
return linked.rename(columns={"particle": "track_id"})
def _link_btrack(masks: np.ndarray, displacement: float):
"""Link with btrack's Bayesian tracker using its stock cell motion model.
btrack ships no config file; ``btrack.datasets.cell_config()`` fetches
one on first use. In a preview that download is the difference between
instant and hung, so a failure is turned into an actionable message
rather than a stack trace.
"""
import btrack
from btrack import datasets as btrack_datasets
from spacr.timelapse import _prepare_for_tracking
try:
config = btrack_datasets.cell_config()
except Exception as exc:
raise TrackerUnavailable(
"btrack's motion-model config is not available offline "
f"({exc}). btrack.datasets.cell_config() downloads it once — run "
"the Timelapse module with timelapse_mode='btrack' to cache it, "
"or preview with trackpy / iou.") from exc
objects = btrack.utils.segmentation_to_objects(masks, properties=("area",))
with btrack.BayesianTracker() as tracker:
tracker.configure(config)
tracker.max_search_radius = float(displacement)
tracker.append(objects)
tracker.track()
rows = []
for tr in tracker.tracks:
for t, x, y in zip(tr["t"], tr["x"], tr["y"]):
rows.append({"frame": int(t), "track_id": int(tr["ID"]),
"x": float(x), "y": float(y)})
import pandas as pd
df = pd.DataFrame(rows, columns=["frame", "track_id", "x", "y"])
if df.empty:
return _empty_tracks(df)
features = _prepare_for_tracking(masks)
return _attach_labels_by_position(df, features)
def _empty_tracks(like):
"""An empty track table shaped like ``like`` plus ``original_label``."""
import pandas as pd
out = like.iloc[0:0].copy()
out["original_label"] = pd.Series(dtype="int64")
return out
def _attach_labels_by_position(df, features):
"""Give each tracked point the label of the nearest object in its frame.
btrack reports track coordinates, not the label they came from; the
overlay needs a label so masks can be recoloured by track id. Nearest
centroid within a frame is exact for the centroids btrack was fed.
"""
out = []
for frame, g in df.groupby("frame"):
f = features[features["frame"] == frame]
if f.empty:
continue
fx = f["x"].to_numpy(dtype=float)
fy = f["y"].to_numpy(dtype=float)
labels = f["original_label"].to_numpy()
g = g.copy()
idx = [int(np.argmin((fx - float(x)) ** 2 + (fy - float(y)) ** 2))
for x, y in zip(g["x"], g["y"])]
g["original_label"] = labels[idx]
out.append(g)
if not out:
return _empty_tracks(df)
import pandas as pd
return pd.concat(out, ignore_index=True)
def _link_trackastra(masks: np.ndarray, images: Optional[np.ndarray],
model_name: str, linking: str):
"""Link with Trackastra's pretrained transformer (no hyperparameters)."""
from trackastra.model import Trackastra
from trackastra.tracking import graph_to_ctc
from spacr.timelapse import _relabelled_stack_to_tracks_df
imgs = np.asarray(images) if images is not None else masks.astype(np.float32)
model = Trackastra.from_pretrained(str(model_name), device="automatic")
graph = model.track(imgs, masks, mode=str(linking))
_ctc, tracked = graph_to_ctc(graph, masks, outdir=None)
return _relabelled_stack_to_tracks_df(np.asarray(tracked))
def _link_ultrack(masks: np.ndarray, max_distance: float):
"""Link with Ultrack's joint segmentation/linking solver."""
from ultrack import MainConfig, track, to_tracks_layer, tracks_to_zarr
from ultrack import utils as ultrack_utils
from spacr.timelapse import (
_relabelled_stack_to_tracks_df, _ultrack_labels_to_contours,
_ultrack_set, _ultrack_track_kwargs,
)
import tempfile
labels_to_contours = _ultrack_labels_to_contours(ultrack_utils)
fg_kwarg, contours_kwarg, _accepts_images = _ultrack_track_kwargs(track)
foreground, contours = labels_to_contours(masks, sigma=0.0)
config = MainConfig()
work_dir = tempfile.mkdtemp(prefix="spacr_ultrack_preview_")
import pathlib
_ultrack_set(config.data_config, "working_dir", pathlib.Path(work_dir),
"ultrack_working_dir")
_ultrack_set(config.linking_config, "max_distance", float(max_distance),
"ultrack_max_distance")
track(config, **{fg_kwarg: foreground, contours_kwarg: contours})
tracks_layer, _graph = to_tracks_layer(config)
tracked = tracks_to_zarr(config, tracks_layer)
return _relabelled_stack_to_tracks_df(np.asarray(tracked))
[docs]
def link_tracks(masks: np.ndarray, mode: str = "iou",
displacement: float = 50.0, memory: int = 3,
iou_threshold: float = 0.1,
images: Optional[np.ndarray] = None,
trackastra_model: str = "general_2d",
trackastra_linking: str = "greedy"):
"""Link a (T, H, W) label stack into tracks with the chosen backend.
:param masks: label stack with at least two frames; any other shape
raises :class:`ValueError`.
:param mode: linking backend, one of :data:`TRACK_MODES`.
:param displacement: maximum displacement between frames, in pixels,
passed to the ``trackpy``, ``btrack`` and ``ultrack`` backends.
:param memory: frames an object may vanish for (``trackpy`` only).
:param iou_threshold: minimum overlap to link (``iou`` only).
:param images: intensity stack for ``trackastra``.
:param trackastra_model: ``trackastra`` model name.
:param trackastra_linking: ``trackastra`` linking mode.
:returns: a DataFrame with ``frame``, ``original_label``, ``track_id``,
``x`` and ``y`` — the same layout the pipeline's trackers emit.
:raises TrackerUnavailable: when the backend's package is missing, with
a message naming the package and the install command.
"""
masks = np.asarray(masks)
if masks.ndim != 3:
raise ValueError(f"masks must be (T, H, W); got shape {masks.shape}")
if masks.shape[0] < 2:
raise ValueError("a timelapse preview needs at least two frames.")
mode = (mode or "iou").lower()
ok, why = backend_available(mode)
if not ok:
raise TrackerUnavailable(why)
if mode == "iou":
return _link_iou(masks, iou_threshold)
if mode == "trackpy":
return _link_trackpy(masks, displacement, memory)
if mode == "btrack":
return _link_btrack(masks, displacement)
if mode == "trackastra":
return _link_trackastra(masks, images, trackastra_model,
trackastra_linking)
return _link_ultrack(masks, displacement)
[docs]
def relabel_by_track(masks: np.ndarray, tracks) -> np.ndarray:
"""Recolour a label stack by track id using spaCR's own relabeller.
This is what makes an identity swap visible: an object that keeps its
track id keeps its colour across frames, and one that is handed to
another track changes colour mid-movie.
:param masks: the (T, H, W) label stack the tracks were built from.
:param tracks: the track table from :func:`link_tracks`; ``None`` or an
empty table gives an all-zero stack.
"""
from spacr.timelapse import _relabel_masks_based_on_tracks
if tracks is None or len(tracks) == 0:
return np.zeros_like(masks)
return _relabel_masks_based_on_tracks(np.asarray(masks), tracks)
@dataclass
[docs]
class TrackStats:
"""What the user is actually tuning against.
``fragmentation_*`` and ``suspicious_jumps`` are *indicators* computed
without ground truth, not measurements — the panel labels them as such.
"""
n_frames: int = 0
n_tracks: int = 0
mean_length: float = 0.0
median_length: float = 0.0
n_short: int = 0
min_length: int = 3
starts_after_first: int = 0
ends_before_last: int = 0
suspicious_jumps: int = 0
max_step: float = 0.0
objects_per_frame: float = 0.0
displacement_limit: float = 0.0
@property
[docs]
def fragmentation_events(self) -> int:
"""Track starts after frame 0 plus track ends before the last frame."""
return self.starts_after_first + self.ends_before_last
[docs]
def summary(self) -> str:
"""The one-line status the panel pins under the canvas."""
return (
f"{self.n_tracks} tracks over {self.n_frames} frames · "
f"mean length {self.mean_length:.1f} "
f"(median {self.median_length:.0f}) · "
f"{self.n_short} shorter than {self.min_length} frames · "
f"fragmentation {self.fragmentation_events} "
f"({self.starts_after_first} late starts, "
f"{self.ends_before_last} early ends) · "
f"{self.suspicious_jumps} steps over "
f"{self.displacement_limit:.0f} px "
f"(max {self.max_step:.0f} px) — swap risk"
)
[docs]
def track_stats(tracks, n_frames: int, min_length: int = 3,
displacement_limit: float = 50.0) -> TrackStats:
"""Summarise a track table into the numbers that drive a tuning decision.
:param tracks: DataFrame with ``frame``, ``track_id`` and (for the swap
indicator) ``x``/``y``.
:param n_frames: how many frames the preview covered.
:param min_length: tracks shorter than this are counted as short — the
live-settable threshold that says what "too short" means here.
:param displacement_limit: the linking radius being tuned; a within-track
step longer than this is counted as a swap risk.
"""
st = TrackStats(n_frames=int(n_frames), min_length=int(min_length),
displacement_limit=float(displacement_limit))
if tracks is None or len(tracks) == 0:
return st
lengths = tracks.groupby("track_id")["frame"].nunique()
st.n_tracks = int(lengths.shape[0])
st.mean_length = float(lengths.mean())
st.median_length = float(lengths.median())
st.n_short = int((lengths < int(min_length)).sum())
st.objects_per_frame = float(len(tracks)) / max(1, int(n_frames))
firsts = tracks.groupby("track_id")["frame"].min()
lasts = tracks.groupby("track_id")["frame"].max()
st.starts_after_first = int((firsts > tracks["frame"].min()).sum())
st.ends_before_last = int((lasts < tracks["frame"].max()).sum())
if {"x", "y"}.issubset(tracks.columns):
jumps = 0
biggest = 0.0
for _tid, g in tracks.sort_values("frame").groupby("track_id"):
x = g["x"].to_numpy(dtype=float)
y = g["y"].to_numpy(dtype=float)
if x.size < 2:
continue
d = np.hypot(np.diff(x), np.diff(y))
d = d[np.isfinite(d)]
if d.size == 0:
continue
jumps += int((d > float(displacement_limit)).sum())
biggest = max(biggest, float(d.max()))
st.suspicious_jumps = jumps
st.max_step = biggest
return st
[docs]
def track_colour(track_id: int) -> Tuple[int, int, int]:
"""Deterministic colour for a track id, stable across frames and runs.
:param track_id: the track id, converted with ``int``; it indexes
:data:`TRACK_COLOURS` cyclically.
"""
return TRACK_COLOURS[int(track_id) % len(TRACK_COLOURS)]
def _draw_segment(rgb: np.ndarray, x0, y0, x1, y1, colour) -> None:
"""Draw a 1-px line into ``rgb`` by sampling along it (no cv2 needed)."""
h, w = rgb.shape[:2]
n = int(max(abs(x1 - x0), abs(y1 - y0))) + 1
xs = np.linspace(x0, x1, max(2, n)).astype(int)
ys = np.linspace(y0, y1, max(2, n)).astype(int)
keep = (xs >= 0) & (xs < w) & (ys >= 0) & (ys < h)
rgb[ys[keep], xs[keep]] = np.array(colour, dtype=np.uint8)
def _draw_dot(rgb: np.ndarray, x, y, colour, radius: int = 2) -> None:
"""Paint a small filled square onto an RGB array.
Clipped to the array, so a track leaving the field draws what is still
inside rather than raising.
:param rgb: the image to draw on, modified in place.
:param x: centre column.
:param y: centre row.
:param colour: the RGB triple to fill with.
:param radius: half-width in pixels; at least 1, so a dot is never
invisible.
"""
h, w = rgb.shape[:2]
x, y, r = int(x), int(y), max(1, int(radius))
y0, y1 = max(0, y - r), min(h, y + r + 1)
x0, x1 = max(0, x - r), min(w, x + r + 1)
if y1 > y0 and x1 > x0:
rgb[y0:y1, x0:x1] = np.array(colour, dtype=np.uint8)
[docs]
def render_frame(image: np.ndarray, labels: Optional[np.ndarray] = None,
tracks=None, frame: int = 0, tail: int = 12,
normalise: bool = True, lo_pct: float = 2.0,
hi_pct: float = 98.0, channel: int = 0) -> np.ndarray:
"""Render one preview frame: image, mask outlines, and track history.
Outlines are coloured **per track id** so a fragmented track shows up as
an object that changes colour partway through, and the trailing polyline
shows where each object came from over the last ``tail`` frames.
:param image: the frame to draw, 2-D or with a channel axis; the plane is
picked with :func:`frame_channel`.
:param labels: optional track-relabelled mask for this frame, outlined
per track colour.
:param tracks: optional track table with ``x``, ``y``, ``frame`` and
``track_id``; trails are drawn up to ``frame``.
:param frame: index of this frame in the track table.
:param tail: how many earlier frames each trail reaches back.
:param normalise: stretch the plane between the two percentiles.
:param lo_pct: lower percentile of the stretch.
:param hi_pct: upper percentile of the stretch.
:param channel: channel of ``image`` to draw.
:returns: an RGB ``uint8`` array.
"""
plane = frame_channel(image, channel)
base = _to_uint8(plane, normalise=normalise, lo_pct=lo_pct, hi_pct=hi_pct)
if base.ndim == 2:
rgb = np.stack([base, base, base], axis=-1)
else:
rgb = np.ascontiguousarray(base[..., :3])
if labels is not None and np.asarray(labels).any():
labels = np.asarray(labels).astype(np.int32)
boundary = _boundary_mask(labels)
edge = boundary & (labels > 0)
ids = np.unique(labels[edge]) if edge.any() else np.array([], dtype=int)
for tid in ids:
rgb[edge & (labels == tid)] = np.array(
track_colour(int(tid)), dtype=np.uint8)
_needed = {"x", "y", "frame", "track_id"}
if tracks is not None and len(tracks) and _needed.issubset(tracks.columns):
lo = max(0, int(frame) - int(tail))
window = tracks[(tracks["frame"] <= int(frame))
& (tracks["frame"] >= lo)]
for tid, g in window.sort_values("frame").groupby("track_id"):
colour = track_colour(int(tid))
xs = g["x"].to_numpy(dtype=float)
ys = g["y"].to_numpy(dtype=float)
ok = np.isfinite(xs) & np.isfinite(ys)
xs, ys = xs[ok], ys[ok]
for i in range(1, xs.size):
_draw_segment(rgb, xs[i - 1], ys[i - 1], xs[i], ys[i], colour)
if xs.size:
_draw_dot(rgb, xs[-1], ys[-1], colour, radius=2)
return rgb
@dataclass
[docs]
class TimelapseRequest:
"""One preview pass. ``cached_masks`` is what makes re-linking cheap."""
sequence: Optional[FrameSequence] = None
mask_sequence: Optional[FrameSequence] = None
cached_masks: Optional[np.ndarray] = None
cached_images: Optional[np.ndarray] = None
seg: Dict[str, Any] = field(default_factory=dict)
track: Dict[str, Any] = field(default_factory=dict)
include_images: bool = False
[docs]
class MovieFieldCancelled(RuntimeError):
"""A queued movie field was abandoned before it retained its arrays."""
[docs]
def movie_worker_interrupted() -> bool:
"""Whether the JobRunner thread executing this field was cancelled."""
try:
return bool(QThread.currentThread().isInterruptionRequested())
except RuntimeError:
return True
def _check_movie_cancelled(cancelled: Optional[Callable[[], bool]]) -> None:
"""Raise if the movie render has been cancelled.
:param cancelled: called to ask; ``None`` never cancels.
:raises MovieFieldCancelled: when it answers ``True``. Raised rather
than returned so the unwinding happens wherever the render happens
to be, without every step having to check a flag.
"""
if cancelled is not None and cancelled():
raise MovieFieldCancelled("movie field cancelled")
def _read_sequence_frames(
sequence: FrameSequence,
cancelled: Optional[Callable[[], bool]] = None) -> np.ndarray:
"""Read one sequence once, checking cancellation between every frame."""
frames = []
for index in range(len(sequence)):
_check_movie_cancelled(cancelled)
frames.append(np.asarray(sequence.frame(index)))
_check_movie_cancelled(cancelled)
return np.stack(frames, axis=0)
def _read_and_segment_sequence(
sequence: FrameSequence,
params: Dict[str, Any],
cancelled: Optional[Callable[[], bool]] = None,
) -> Tuple[np.ndarray, np.ndarray]:
"""Read and segment each frame once, returning raw images and masks."""
frames = []
masks = []
for index in range(len(sequence)):
_check_movie_cancelled(cancelled)
image = np.asarray(sequence.frame(index))
frames.append(image)
masks.append(segment_frame(image, params))
_check_movie_cancelled(cancelled)
shapes = {mask.shape for mask in masks}
if len(shapes) != 1:
raise ValueError(
f"frames segmented to different shapes {sorted(shapes)}; the "
"sequence is not a single field of view.")
return (np.stack(frames, axis=0),
np.stack(masks, axis=0).astype(np.int32))
[docs]
def run_preview_pass(req: TimelapseRequest) -> Dict[str, Any]:
"""Do the work of one preview: masks (maybe cached), then linking.
:param req: the request; its cached masks are used first, then its mask
sequence, then its image sequence is segmented with ``req.seg``, and
the masks are linked with ``req.track``. With none of these it
raises :class:`ValueError`.
"""
masks = req.cached_masks
images = req.cached_images
segmented = False
if masks is None:
if req.mask_sequence is not None:
masks = _as_label_stack(req.mask_sequence,
int(req.seg.get("mask_channel", 0)))
elif req.sequence is not None:
if req.include_images:
images, masks = _read_and_segment_sequence(
req.sequence, req.seg)
else:
masks = segment_sequence(req.sequence, req.seg)
segmented = True
else:
raise ValueError("Load a sequence first.")
masks = np.asarray(masks)
if req.include_images and images is None and req.sequence is not None:
images = _read_sequence_frames(req.sequence)
tracks = link_tracks(masks, **req.track)
return {"masks": masks, "tracks": tracks, "segmented": segmented,
"masks_built": req.cached_masks is None, "images": images}
[docs]
def build_movie_field(
path, *, max_frames: int, seg: Dict[str, Any],
track: Dict[str, Any], cached_masks: Optional[np.ndarray] = None,
cancelled: Optional[Callable[[], bool]] = None,
) -> Dict[str, Any]:
"""Open, segment and link one additional field without touching Qt UI.
Raw frames are retained for the movie, so a cache miss reads each frame
exactly once and hands that same array to segmentation. ``cancelled`` is
checked between frames and before linking; the production callback reads
the worker QThread's interruption flag, which lets lowering the Fields cap
stop an expensive sibling before it retains the rest of the sequence.
:param path: the field to open: a directory of frames, a multi-page TIFF,
or an ``.npy`` stack whose first axis is time.
:param max_frames: cap on the number of frames read.
:param seg: segmentation parameters, as :func:`segment_frame` takes
them; its ``channel`` is also returned.
:param track: keyword arguments for :func:`link_tracks`.
:param cached_masks: an existing (T, H, W) label stack to reuse instead
of segmenting; its frame count must match the sequence.
:param cancelled: a no-argument callable returning ``True`` to stop.
"""
source = Path(os.fspath(path))
sequence = FrameSequence.open(source, max_frames=max_frames)
_check_movie_cancelled(cancelled)
if cached_masks is None:
images, masks = _read_and_segment_sequence(
sequence, seg, cancelled=cancelled)
segmented = True
else:
images = _read_sequence_frames(sequence, cancelled=cancelled)
masks = np.asarray(cached_masks)
if int(masks.shape[0]) != len(sequence):
raise ValueError(
f"cached masks have {masks.shape[0]} frames but "
f"{source.name} now has {len(sequence)}")
segmented = False
_check_movie_cancelled(cancelled)
tracks = link_tracks(masks, **track)
_check_movie_cancelled(cancelled)
return {
"source": str(source),
"title": source.name or str(source),
"images": images,
"masks": masks,
"labels": relabel_by_track(masks, tracks),
"tracks": tracks,
"channel": int(seg.get("channel", 0)),
"segmented": segmented,
}
[docs]
def movie_field_payload(**kwargs) -> Dict[str, Any]:
"""Never let one bad sibling strand the remaining movie-field queue."""
try:
return build_movie_field(**kwargs)
except MovieFieldCancelled:
return {"cancelled": True, "source": str(kwargs.get("path", ""))}
except Exception as exc: # noqa: BLE001
LOG.info("timelapse movie field failed: %s", exc, exc_info=True)
return {"error": str(exc), "source": str(kwargs.get("path", ""))}
class _TimelapseWorker(QThread):
"""Runs one preview pass off the GUI thread.
Mirrors ``live_preview._PreviewWorker`` exactly: every exception is
caught inside :meth:`run` and re-emitted as an error *string*, so
nothing ever propagates out of a Qt thread's ``run()``.
"""
finished_result = Signal(object, str)
def __init__(self, request: TimelapseRequest, parent=None):
"""Prepare the worker.
:param request: everything the pass needs, READ ON THE WORKER THREAD
rather than here -- so it must not be mutated after the worker
is started; build a new request instead.
:param parent: parent object; ownership only. It does NOT keep the
thread alive across a parent's destruction, so the panel still
has to wait for the thread itself.
"""
super().__init__(parent)
self._request = request
def run(self):
"""Run one preview pass and emit its result, or the failure text.
A failure is emitted rather than raised: this runs on a worker thread,
where an exception has nobody to catch it.
"""
try:
self.finished_result.emit(run_preview_pass(self._request), "")
except Exception as e:
LOG.info("timelapse preview failed: %s", e, exc_info=True)
self.finished_result.emit(None, str(e))
#: Last resort if `spacr.settings` cannot be reached at all — a stub in
#: sys.modules, a partially-installed tree. A dropdown with nothing in it
#: is a dead end, so there is always something here.
_FALLBACK_MODELS = ("cpsam", "cyto3", "cyto2", "nuclei")
def _model_menu():
"""What the Cellpose model combo offers, read from the Cellpose API.
Delegates to :func:`spacr.settings.cellpose_model_menu`, which asks
``cellpose.models`` for its stock list plus any checkpoint the user
registered, then appends the accepted-but-mapped legacy spellings so a
saved preview setting still loads.
Wrapped because this is a *widget*: it must build even when
``spacr.settings`` is a stand-in (a test that stubs the descriptions
table does exactly that). It degrades to the shipped list rather than
to an empty combo.
"""
try:
from ...settings import cellpose_model_menu
menu = tuple(cellpose_model_menu())
except Exception:
return _FALLBACK_MODELS
return menu or _FALLBACK_MODELS
[docs]
def open_sequence_payload(path, max_frames: int = 12,
list_siblings: bool = True) -> Dict[str, Any]:
"""Open a sequence and list its neighbours. No Qt, so it runs on a worker.
Warms the sequence's own frame cache with frame 0, because
``_frame_channel_count`` decodes exactly that frame on the GUI thread to
fill the channel dropdown -- doing it here turns that read into a cache
hit rather than a second trip to disk.
:param path: sequence file or frame directory accepted by
:meth:`FrameSequence.open`; the same value is returned as text in the
payload even when opening fails.
:param list_siblings: ``False`` reuses the sampler's cached listing; the
FOV dropdown hands out a path it has already enumerated.
:returns: ``{path, sequence, siblings, error}``. When ``list_siblings``
is true ``siblings`` is ALWAYS a list, never ``None``, even if the
listing failed -- see the fallback below for why that matters.
"""
out: Dict[str, Any] = {"path": str(path), "sequence": None,
"siblings": None, "error": ""}
try:
seq = FrameSequence.open(path, max_frames=max_frames)
except Exception as exc:
out["error"] = f"Load failed: {exc}"
return out
try:
seq.frame(0)
except Exception:
LOG.debug("could not warm the first frame of %s", path, exc_info=True)
out["sequence"] = seq
if list_siblings:
target = Path(os.fspath(path))
try:
out["siblings"] = sibling_sources(
target, FRAME_SUFFIXES,
directories=(getattr(seq, "kind", "") == "files"))
except Exception:
LOG.exception("Could not list sequences beside %s", path)
out["siblings"] = [target]
return out
def _annotation_track_identity(path: Path) -> Tuple[str, str, str]:
"""Parse the detector's exact tracker, object and field filename."""
match = re.fullmatch(
r"(trackpy|trackastra|ultrack|timeflows|sam2|btrack)_tracks_"
r"(cell|nucleus|pathogen)_(.+)\.csv", path.name)
if match is None:
raise ValueError(tr("Choose a spaCR tracks CSV named backend_tracks_object_field.csv."))
return match.group(1), match.group(2), match.group(3)
def _annotation_digest(path: Path) -> Optional[str]:
"""Identify the exact annotation bytes last read before publication."""
if not path.exists():
return None
digest = hashlib.sha256()
with path.open("rb") as source:
for chunk in iter(lambda: source.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def _annotation_field_payload(tracks_path: str, sequence_path: str,
annotations_path: str) -> dict:
"""Open a tracked field without decoding the movie into memory."""
from spacr.tabular import read_table
import pandas as pd
track_file = Path(tracks_path).expanduser().resolve(strict=True)
sequence_file = Path(sequence_path).expanduser().resolve(strict=True)
target = Path(annotations_path).expanduser().resolve()
if target == track_file or target == sequence_file:
raise ValueError(tr("Choose an annotation CSV separate from the tracker and image source."))
backend, object_type, field_name = _annotation_track_identity(track_file)
sequence = FrameSequence.open(sequence_file, max_frames=2_147_483_647)
track_digest = _annotation_digest(track_file)
annotation_digest = _annotation_digest(target)
tracks = read_table(
str(track_file), canonicalise=False, report=None,
usecols=lambda name: name in ("frame", "track_id", "x", "y"))
if tracks.empty or not {"frame", "track_id", "x", "y"}.issubset(tracks.columns):
raise ValueError(tr("The tracks CSV needs frame and track_id rows with x and y positions."))
for column in ("frame", "track_id"):
values = pd.to_numeric(tracks[column], errors="raise")
if (not np.isfinite(values).all() or (values < 0).any()
or (values % 1 != 0).any()):
raise ValueError(tr("Track IDs and frames must be nonnegative integers."))
tracks[column] = values.astype("int64")
for column in ("x", "y"):
values = pd.to_numeric(tracks[column], errors="raise")
if not np.isfinite(values).all():
raise ValueError(tr("Track positions must be finite numbers."))
tracks[column] = values.astype(float)
if tracks.duplicated(["track_id", "frame"]).any():
raise ValueError(tr("The tracks CSV repeats a track at the same frame."))
if int(tracks["frame"].max()) >= len(sequence):
raise ValueError(tr("A tracked frame is outside the selected image sequence."))
frame = np.asarray(sequence.frame(0))
if frame.ndim == 3 and frame.shape[-1] <= 8 and frame.shape[0] > 8:
channels = int(frame.shape[-1])
elif frame.ndim == 3 and frame.shape[0] <= 8 and frame.shape[-1] > 8:
channels = int(frame.shape[0])
else:
channels = int(frame.shape[-1]) if frame.ndim == 3 else 1
columns = ["field", "track_id", "frame", "event", "object"]
existing = (read_table(str(target), canonicalise=False, report=None)
if target.exists() else pd.DataFrame(columns=columns))
if target.exists() and not set(columns[:4]).issubset(existing.columns):
raise ValueError(tr("The existing annotation table lacks field, track_id, frame or event."))
if not existing.empty:
if existing[columns[:4]].isna().any().any():
raise ValueError(tr("The existing annotation table has incomplete event rows."))
for column in ("track_id", "frame"):
values = pd.to_numeric(existing[column], errors="raise")
if (not np.isfinite(values).all() or (values < 0).any()
or (values % 1 != 0).any()):
raise ValueError(tr("Existing track IDs and frames must be nonnegative integers."))
existing[column] = values.astype("int64")
if any(not str(value).strip() or str(value).strip().lower() == "background"
or any(ord(char) < 32 for char in str(value))
for value in existing["event"]):
raise ValueError(tr("The existing annotation table has an invalid event name."))
same_field = existing["field"].astype(str) == field_name
if "object" not in existing:
if same_field.any():
raise ValueError(tr("Existing annotations for this field need an object column before editing."))
existing["object"] = ""
keys = existing[["field", "object", "track_id", "frame"]].copy()
keys["object"] = keys["object"].fillna("").astype(str)
if keys.duplicated().any():
raise ValueError(tr("The existing annotation table labels one track frame more than once."))
if (same_field & existing["object"].isna()).any():
raise ValueError(tr("Existing annotations for this field have no object identity."))
selected = same_field & (existing["object"].astype(str) == object_type)
mine = existing[selected].copy()
legacy_rows = False
for row in mine.to_dict("records"):
recorded_backend = row.get("tracker_backend")
recorded_digest = row.get("track_source_sha256")
has_backend = pd.notna(recorded_backend) and bool(str(recorded_backend).strip())
has_digest = pd.notna(recorded_digest) and bool(str(recorded_digest).strip())
if not has_backend and not has_digest:
legacy_rows = True
elif (not has_backend or not has_digest
or recorded_backend != backend or recorded_digest != track_digest):
raise ValueError(tr("Existing event labels belong to another tracker CSV."))
observed = set(zip(tracks["track_id"], tracks["frame"]))
events = []
for row in mine.to_dict("records"):
key = (int(row["track_id"]), int(row["frame"]))
name = str(row["event"]).strip().lower()
if key not in observed:
raise ValueError(tr("An existing event is not on an observed track frame."))
events.append({"track_id": key[0], "frame": key[1], "event": name,
"extra": {column: value for column, value in row.items()
if column not in columns}})
if _annotation_digest(track_file) != track_digest:
raise ValueError(tr("The tracks CSV changed while it was being read."))
if _annotation_digest(target) != annotation_digest:
raise ValueError(tr("Annotations changed while they were being read."))
return {"sequence": sequence, "tracks": tracks, "observed": observed,
"events": events, "other": existing[~selected].copy(),
"field": field_name, "object": object_type, "backend": backend,
"channels": channels, "target": target, "legacy_rows": legacy_rows,
"digest": annotation_digest,
"track_digest": track_digest,
"track_path": track_file, "sequence_path": sequence_file}
class _EventAnnotationDialog(QDialog):
"""Edit detector-ready event rows against real tracked observations."""
def __init__(self, parent=None, *, sequence_path=None, threaded=True):
"""Construct one field editor with a lazy sequence and bounded cache."""
super().__init__(parent)
self.setObjectName("TimelapseEventAnnotationDialog")
self.setWindowTitle(tr("Track event annotations"))
self.resize(760, 680)
self._jobs = JobRunner(self, threaded=threaded,
app_key=tr("event annotation load"),
user_visible=False)
self._jobs.job_failed.connect(self._load_failed)
self._field = None
self._events = []
self._shown_observation = None
self._saved_path = None
self._load_token = 0
root = QVBoxLayout(self)
form = QFormLayout()
self._tracks_path = QLineEdit(self)
self._tracks_path.setObjectName("TimelapseEventTracksPath")
self._sequence_path = QLineEdit(self)
self._sequence_path.setObjectName("TimelapseEventSequencePath")
if sequence_path is not None:
self._sequence_path.setText(os.fspath(sequence_path))
self._output_path = QLineEdit(self)
self._output_path.setObjectName("TimelapseEventOutputPath")
for control, caption, picker in (
(self._tracks_path, tr("Tracks CSV"), self._pick_tracks),
(self._sequence_path, tr("Image sequence"), self._pick_sequence),
(self._output_path, tr("Annotations CSV"), self._pick_output)):
row = QHBoxLayout()
row.addWidget(control, 1)
button = QPushButton(tr("Browse…"), self)
if control is self._tracks_path:
button.setObjectName("TimelapseEventBrowseTracks")
elif control is self._sequence_path:
button.setObjectName("TimelapseEventBrowseSequence")
else:
button.setObjectName("TimelapseEventBrowseOutput")
button.clicked.connect(picker)
row.addWidget(button)
if control is self._sequence_path:
folder = QPushButton(tr("Folder…"), self)
folder.setObjectName("TimelapseEventBrowseFolder")
folder.clicked.connect(self._pick_sequence_folder)
row.addWidget(folder)
form.addRow(caption, row)
root.addLayout(form)
self._open_button = QPushButton(tr("Open tracked field"), self)
self._open_button.setObjectName("TimelapseEventOpenField")
self._open_button.clicked.connect(self._load_field)
root.addWidget(self._open_button)
self._identity = QLabel(tr("Choose a tracks CSV and its matching image sequence."), self)
self._identity.setWordWrap(True)
root.addWidget(self._identity)
self._confirm = QCheckBox(
tr("I confirm this image sequence and existing event labels belong to this tracker CSV."), self)
self._confirm.setObjectName("TimelapseEventConfirmField")
self._confirm.setChecked(False)
root.addWidget(self._confirm)
self._tracks_path.textChanged.connect(self._source_changed)
self._sequence_path.textChanged.connect(self._source_changed)
self._preview = QLabel(self)
self._preview.setObjectName("TimelapseEventFramePreview")
self._preview.setMinimumHeight(220)
self._preview.setAlignment(Qt.AlignCenter)
root.addWidget(self._preview, 1)
controls = QHBoxLayout()
self._track = QComboBox(self)
self._track.setObjectName("TimelapseEventTrack")
self._track.currentIndexChanged.connect(self._show_frame)
self._frame = QSpinBox(self)
self._frame.setObjectName("TimelapseEventFrame")
self._frame.valueChanged.connect(self._show_frame)
self._channel = QSpinBox(self)
self._channel.setObjectName("TimelapseEventChannel")
self._channel.valueChanged.connect(self._show_frame)
self._event = QLineEdit(self)
self._event.setObjectName("TimelapseEventName")
self._event.setPlaceholderText(tr("Event name, for example mitosis"))
for caption, control in ((tr("Track"), self._track),
(tr("Frame (zero-based)"), self._frame),
(tr("Channel"), self._channel),
(tr("Event"), self._event)):
controls.addWidget(QLabel(caption, self))
controls.addWidget(control)
root.addLayout(controls)
self._rows = QTableWidget(0, 3, self)
self._rows.setObjectName("TimelapseEventRows")
self._rows.setHorizontalHeaderLabels(
[tr("Track ID"), tr("Frame"), tr("Event")])
self._rows.setSelectionBehavior(QAbstractItemView.SelectRows)
self._rows.setSelectionMode(QAbstractItemView.SingleSelection)
self._rows.setEditTriggers(QAbstractItemView.NoEditTriggers)
install_sorting(self._rows)
self._rows.itemSelectionChanged.connect(self._select_event)
root.addWidget(self._rows)
actions = QHBoxLayout()
self._add_button = QPushButton(tr("Add or update"), self)
self._add_button.setObjectName("TimelapseEventAdd")
self._add_button.clicked.connect(self._add_event)
self._remove_button = QPushButton(tr("Remove selected"), self)
self._remove_button.setObjectName("TimelapseEventRemove")
self._remove_button.clicked.connect(self._remove_event)
self._save_button = QPushButton(tr("Save annotations"), self)
self._save_button.setObjectName("TimelapseEventSave")
self._save_button.clicked.connect(self._save)
close_button = QPushButton(tr("Close"), self)
close_button.setObjectName("TimelapseEventClose")
close_button.clicked.connect(self.reject)
for button in (self._add_button, self._remove_button,
self._save_button, close_button):
actions.addWidget(button)
root.addLayout(actions)
self._status = QLabel("", self)
self._status.setWordWrap(True)
root.addWidget(self._status)
def _pick_tracks(self) -> None:
"""Choose an actual tracker export without opening a movie."""
path, _ = QFileDialog.getOpenFileName(
self, tr("Choose spaCR tracks CSV"), "", tr("CSV (*.csv)"))
if path:
self._tracks_path.setText(path)
self._output_path.setText(str(Path(path).parent / "events" /
"annotations.csv"))
def _pick_sequence(self) -> None:
"""Choose the matching stack or one folder of image frames."""
path, _ = QFileDialog.getOpenFileName(
self, tr("Choose image sequence"), "",
tr("Images (*.tif *.tiff *.npy)"))
if path:
self._sequence_path.setText(path)
def _pick_sequence_folder(self) -> None:
"""Choose a directory of frames through the ordinary folder picker."""
path = QFileDialog.getExistingDirectory(
self, tr("Choose image-frame folder"), "")
if path:
self._sequence_path.setText(path)
def _pick_output(self) -> None:
"""Choose an annotation CSV, including one already containing fields."""
path, _ = QFileDialog.getSaveFileName(
self, tr("Choose annotations CSV"),
self._output_path.text(), tr("CSV (*.csv)"))
if path:
self._output_path.setText(path)
def _source_changed(self, *_args) -> None:
"""Require renewed source confirmation after either identity changes."""
self._confirm.setChecked(False)
def _load_failed(self, message: str) -> None:
"""Report a failed file read without changing a prior field."""
self._status.setText(tr("Could not open tracked field: {reason}",
reason=message))
def _load_field(self) -> None:
"""Read table and sequence metadata on a worker, then adopt on GUI."""
paths = (self._tracks_path.text().strip(),
self._sequence_path.text().strip())
if not all(paths):
self._status.setText(tr("Choose both a tracks CSV and image sequence."))
return
target = self._output_path.text().strip()
if not target:
target = str(Path(paths[0]).parent / "events" / "annotations.csv")
self._output_path.setText(target)
if Path(target).suffix.lower() != ".csv":
self._status.setText(tr("Annotations must be saved as a CSV file."))
return
self._load_token += 1
token = self._load_token
self._jobs.cancel()
self._confirm.setChecked(False)
self._status.setText(tr("Opening tracked field…"))
self._jobs.submit(
lambda: _annotation_field_payload(paths[0], paths[1], target),
lambda result, _token=token: self._adopt_field(_token, result))
def _adopt_field(self, token: int, field: dict) -> None:
"""Install one validated field and keep only its lazy frame cache."""
if token != self._load_token:
return
self._field = field
self._events = list(field["events"])
self._shown_observation = None
field["sequence"]._register_cache_budget()
self._confirm.setChecked(False)
self._identity.setText(
tr("Field {field} · {object} · {backend} · source {source}",
field=field["field"], object=field["object"],
backend=field["backend"], source=str(field["sequence_path"])))
if field["legacy_rows"]:
self._status.setText(tr(
"Existing labels have no tracker identity; confirm this exact tracker CSV before saving."))
else:
self._status.setText(tr("Loaded tracked field; frames are read only when shown."))
self._track.clear()
for track_id in sorted(set(int(value) for value in field["tracks"]["track_id"])):
self._track.addItem(str(track_id), track_id)
self._frame.setRange(0, len(field["sequence"]) - 1)
self._channel.setRange(0, max(0, field["channels"] - 1))
self._refresh_rows()
self._show_frame()
def _show_frame(self, *_args) -> None:
"""Display one lazy-decoded frame and the selected track's position."""
field = self._field
self._shown_observation = None
if field is None:
return
frame = int(self._frame.value())
track_id = self._track.currentData()
if track_id is None:
return
try:
image = field["sequence"].frame(frame)
tracks = field["tracks"]
track = tracks[tracks["track_id"] == int(track_id)]
rgb = render_frame(image, tracks=track, frame=frame,
channel=int(self._channel.value()))
pixmap = numpy_to_qpixmap(rgb)
self._preview.setPixmap(scaled_for(pixmap, self._preview,
(640, 300)))
self._shown_observation = (int(track_id), frame)
if (int(track_id), frame) not in field["observed"]:
self._status.setText(tr("This track has no observation at this frame."))
except Exception as exc:
self._preview.clear()
self._status.setText(tr("Could not show frame: {reason}", reason=str(exc)))
def _select_event(self) -> None:
"""Copy the selected event into the edit controls."""
selected = self._rows.selectionModel().selectedRows()
cell = self._rows.item(selected[0].row(), 0) if selected else None
row = cell.data(Qt.UserRole) if cell is not None else -1
if not 0 <= row < len(self._events):
return
event = self._events[row]
self._track.setCurrentIndex(self._track.findData(event["track_id"]))
self._frame.setValue(event["frame"])
self._event.setText(event["event"])
def _refresh_rows(self) -> None:
"""Show in-memory edits without touching the annotation CSV."""
self._rows.setSortingEnabled(False)
self._rows.setRowCount(len(self._events))
for index, event in enumerate(self._events):
for column, key in enumerate(("track_id", "frame", "event")):
item = table_item(event[key])
if column == 0:
item.setData(Qt.UserRole, index)
self._rows.setItem(index, column, item)
self._rows.setSortingEnabled(True)
self._rows.clearSelection()
def _add_event(self) -> None:
"""Add or edit one event only at an observed zero-based track frame."""
field = self._field
if field is None or not self._confirm.isChecked():
self._status.setText(tr("Open a field and confirm its image sequence first."))
return
if (Path(self._tracks_path.text().strip()).expanduser().resolve()
!= field["track_path"] or
Path(self._sequence_path.text().strip()).expanduser().resolve()
!= field["sequence_path"]):
self._status.setText(tr("The source paths changed; reopen the tracked field."))
return
track_id = self._track.currentData()
frame = int(self._frame.value())
name = self._event.text().strip().lower()
if (track_id is None or (int(track_id), frame) not in field["observed"]
or self._shown_observation != (int(track_id), frame)):
self._status.setText(tr("Choose a frame where this track was observed."))
return
if not name or name == "background" or any(ord(char) < 32 for char in name):
self._status.setText(tr("Enter a nonempty event name other than background."))
return
event = {"track_id": int(track_id), "frame": frame, "event": name}
selection = self._rows.selectionModel().selectedRows()
cell = self._rows.item(selection[0].row(), 0) if selection else None
selected = cell.data(Qt.UserRole) if cell is not None else -1
if any(all(item[key] == event[key]
for key in ("track_id", "frame"))
for index, item in enumerate(self._events) if index != selected):
self._status.setText(tr("This track frame already has an event label."))
return
if 0 <= selected < len(self._events):
event["extra"] = self._events[selected].get("extra", {})
self._events[selected] = event
else:
self._events.append(event)
self._refresh_rows()
self._event.clear()
self._status.setText(tr("Event is staged; save to publish the table."))
def _remove_event(self) -> None:
"""Remove one selected staged row, leaving the file unchanged."""
selection = self._rows.selectionModel().selectedRows()
cell = self._rows.item(selection[0].row(), 0) if selection else None
selected = cell.data(Qt.UserRole) if cell is not None else -1
if not 0 <= selected < len(self._events):
return
del self._events[selected]
self._refresh_rows()
self._status.setText(tr("Event removed from staged edits."))
def _save(self) -> None:
"""Validate with the detector reader before atomically replacing CSV."""
import pandas as pd
from spacr.tabular import write_table
from spacr.timelapse import _event_read_annotations
field = self._field
if field is None or not self._confirm.isChecked():
self._status.setText(tr("Open a field and confirm its image sequence first."))
return
if (Path(self._tracks_path.text().strip()).expanduser().resolve()
!= field["track_path"] or
Path(self._sequence_path.text().strip()).expanduser().resolve()
!= field["sequence_path"]):
self._status.setText(tr("The source paths changed; reopen the tracked field."))
return
target = Path(self._output_path.text().strip()).expanduser().resolve()
if target.suffix.lower() != ".csv":
self._status.setText(tr("Annotations must be saved as a CSV file."))
return
same_target = target == field["target"]
try:
keys = [(event["track_id"], event["frame"])
for event in self._events]
if len(keys) != len(set(keys)):
raise ValueError(tr("One track frame cannot have multiple event labels."))
if _annotation_digest(field["track_path"]) != field["track_digest"]:
raise ValueError(tr("The tracks CSV changed; reopen the field before saving."))
digest = _annotation_digest(target)
if (same_target and digest != field["digest"]
or not same_target and digest is not None):
raise ValueError(tr("Annotations changed on disk; reopen the field or choose a new CSV."))
rows = [{"field": field["field"], "track_id": event["track_id"],
"frame": event["frame"], "event": event["event"],
"object": field["object"], **event.get("extra", {}),
"tracker_backend": field["backend"],
"track_source_sha256": field["track_digest"]}
for event in self._events]
current = (pd.DataFrame(rows) if rows else pd.DataFrame(
columns=["field", "track_id", "frame", "event", "object"]))
table = pd.concat([field["other"], current], ignore_index=True)
target.parent.mkdir(parents=True, exist_ok=True)
handle, pending = tempfile.mkstemp(
prefix=".event-annotations-", suffix=".csv", dir=target.parent)
os.close(handle)
try:
write_table(table, pending, canonicalise=False)
checked = _event_read_annotations(pending)
if len(checked) != len(table):
raise ValueError(tr("The detector reader did not retain every event row."))
if _annotation_digest(target) != digest:
raise ValueError(tr("Annotations changed during save; reopen the field."))
os.replace(pending, target)
finally:
if os.path.exists(pending):
os.unlink(pending)
field["target"] = target
field["digest"] = _annotation_digest(target)
self._saved_path = str(target)
self._status.setText(tr("Saved annotations to {path}", path=str(target)))
except Exception as exc:
self._status.setText(tr("Could not save annotations: {reason}", reason=str(exc)))
def _release_field(self) -> None:
"""Drop six-frame cache and retire the loader on either dialog exit."""
self._jobs.shutdown()
field = self._field
self._field = None
self._shown_observation = None
if field is not None:
sequence = field["sequence"]
for key in tuple(sequence._cache):
sequence._drop_cache_budget_entry(key)
self._preview.clear()
def done(self, result):
"""Release image and worker references when Save or Cancel exits."""
self._release_field()
super().done(result)
def closeEvent(self, event):
"""Also release references on a window-manager close."""
self._release_field()
super().closeEvent(event)
[docs]
class TimelapsePreviewPanel(LivePreviewContract, QWidget):
"""Interactive tracking preview — Timelapse module.
Same contract as :class:`~spacr.qt.widgets.live_preview.LivePreviewPanel`,
and now literally the same code for the shared half: a standalone
``QWidget``, a ``QThread`` worker that emits results over signals,
:class:`~spacr.qt.widgets.preview_contract.LivePreviewContract` for the
run/cancel/status protocol, :meth:`set_propagate_callback` to push tuned
values back into the main settings panel, and a ``build_*_card`` factory.
:param parent: parent widget.
:param threaded: whether the jobs run off the GUI thread. Opening a
sequence reads a TIFF header or memory-maps a stack and then lists
every sibling field of view, which is not GUI-thread work on a plate.
False runs each job inline, emitting the same signals in the same
order, so a test can drive this panel synchronously without the
behaviour diverging.
"""
#: Where this preview's section folds and sizes are remembered (item
#: 471): the settings groups, the movie and the track quality each fold by their heading and trade height by their
#: edge.
SECTION_KEY = "timelapse_preview"
preview_ready = Signal(object)
PREVIEW_SOURCE_HINT = "Load a sequence first."
def __init__(self, parent=None, *, threaded: bool = True):
"""Build the preview: its canvases, its scrub bar and its controls.
:param parent: parent widget.
:param threaded: whether work runs on a worker.
"""
super().__init__(parent)
self._jobs = JobRunner(self, threaded=threaded,
app_key="timelapse preview")
self._movie_jobs = JobRunner(
self, threaded=threaded, app_key="timelapse movie fields")
self._jobs.job_failed.connect(self._on_job_failed)
#: Bumped whenever a newer open supersedes the one in flight.
self._load_token = 0
#: The same, for mask opens. Separate, because loading masks does not
#: supersede an image sequence that is still on its way.
self._mask_load_token = 0
#: The placeholder currently on the status label, or None. Only a
#: line this panel wrote and still owns may be replaced by a failure.
self._transient_status: Optional[str] = None
self._sequence: Optional[FrameSequence] = None
self._mask_sequence: Optional[FrameSequence] = None
self._masks: Optional[np.ndarray] = None
self._movie_images: Optional[np.ndarray] = None
self._tracked: Optional[np.ndarray] = None
self._tracks = None
self._raw_tracks = None
self._stats: Optional[TrackStats] = None
self._mask_cache: Dict[tuple, np.ndarray] = {}
self._mask_cache_last_used: Dict[tuple, float] = {}
self._movie_fields: Dict[str, Dict[str, Any]] = {}
self._movie_sources: List[str] = []
self._movie_pending_path: Optional[str] = None
self._movie_pending_key: Optional[tuple] = None
self._movie_generation = 0
self._movie_seg_key: Optional[tuple] = None
self._movie_track_key: Optional[tuple] = None
self._movie_failures: Dict[str, tuple] = {}
self._worker: Optional[_TimelapseWorker] = None
self._retired_worker: Optional[_TimelapseWorker] = None
self._run_token = 0
self._pending_token = 0
self._pending_signature: Optional[tuple] = None
self._propagate_cb = None
self._settings: Dict[str, Any] = {}
self._sequence_path: Optional[Path] = None
self._loading_fov = False
self._sampler = ImageSetSampler(DEFAULT_MAX_SETS)
_LIVE_PREVIEW_PANELS.add(self)
_ensure_cache_budget_sweep()
self._play_timer = QTimer(self)
self._play_timer.timeout.connect(self._advance_frame)
self._build_ui()
from ..preferences import _is_alpha_visible
if not _is_alpha_visible("widgets", "TimelapseEventAnnotationButton"):
self._event_annotation_btn.setProperty("_spacr_alpha_hid", True)
self._event_annotation_btn.hide()
self.setAcceptDrops(True)
for v in (self._src_view, self._out_view):
v.setAcceptDrops(False)
from ..screens.settings_model import retarget_field_tooltips
retarget_field_tooltips(self)
def _build_ui(self):
"""Lay out the canvases over the scrub bar and the control row."""
root = QVBoxLayout(self)
root.setContentsMargins(8, 8, 8, 8)
root.setSpacing(6)
pick = QHBoxLayout()
self._pick_row = pick
self._path_label = QLabel(
"No sequence loaded — drop a folder of frames, a multi-page TIFF, "
"or a (T, H, W) .npy stack here")
self._path_label.setSizePolicy(QSizePolicy.Expanding,
QSizePolicy.Preferred)
self._max_sets_box = FlatSpinBox(self, value=DEFAULT_MAX_SETS,
tooltip=MAX_SETS_TOOLTIP)
self._max_sets_box.valueChanged.connect(self._on_max_sets_changed)
self._fov_box = FlatComboBox(
self,
tooltip=("Field of view. Lists a random sample of the sequences "
"sitting beside the loaded one; picking one loads it."))
self._fov_box.currentIndexChanged.connect(self._on_fov_changed)
self._channel_box = FlatComboBox(
self,
tooltip=("Channel shown and segmented. Bound to the segmentation "
"channel, so the frame you look at is the frame Cellpose "
"sees."))
self._channel_box.currentIndexChanged.connect(
self._on_display_channel_changed)
self._seq_btn = FlatButton("Choose sequence…", self)
self._seq_btn.clicked.connect(self._pick_sequence)
self._mask_btn = FlatButton(
"Masks…", self,
tooltip=("Optional: point at a folder or stack of ready-made "
"label images to skip segmentation entirely."))
self._mask_btn.clicked.connect(self._pick_masks)
pick.addWidget(self._path_label, 1)
pick.addWidget(self._max_sets_box)
pick.addWidget(self._fov_box)
pick.addWidget(self._channel_box)
pick.addWidget(self._seq_btn)
pick.addWidget(self._mask_btn)
root.addLayout(pick)
self._model_box = QComboBox(self)
self._model_box.addItems(list(_model_menu()))
self._model_box.setToolTip(
"(str) Cellpose model used to segment every frame. Changing this "
"re-segments — it is the expensive half of the preview.")
self._object_box = QComboBox(self)
self._object_box.addItems(["cell", "nucleus", "pathogen"])
self._object_box.setToolTip(
"(list) Which object is tracked across frames (timelapse_objects).")
self._channel = QSpinBox(self)
self._channel.setRange(0, 8)
self._channel.setToolTip(
"(int) Image channel index segmented for the tracked object.")
self._diameter = QDoubleSpinBox(self)
self._diameter.setRange(0, 400)
self._diameter.setValue(30.0)
self._diameter.setSuffix(" px")
self._diameter.setToolTip(
"(float, px) Expected object diameter. Ignored by Cellpose-SAM.")
self._flow = QDoubleSpinBox(self)
self._flow.setRange(-1, 3)
self._flow.setSingleStep(0.05)
self._flow.setValue(0.4)
self._flow.setToolTip("(float) Cellpose flow threshold.")
self._prob = QDoubleSpinBox(self)
self._prob.setRange(-6, 6)
self._prob.setSingleStep(0.1)
self._prob.setValue(0.0)
self._prob.setToolTip("(float) Cellpose cell-probability threshold.")
self._max_frames = QSpinBox(self)
self._max_frames.setRange(2, 500)
self._max_frames.setValue(12)
self._max_frames.setToolTip(
"(int) How many frames of the sequence the preview reads. The "
"rest of the movie is never loaded.")
self._normalise = Toggle("Normalise", self)
self._normalise.setChecked(True)
self._normalise.setToolTip(
"(bool) Percentile-stretch each frame for display + segmentation.")
self._mode_box = QComboBox(self)
self._mode_box.addItems(list(TRACK_MODES))
self._mode_box.setCurrentText("iou")
self._mode_box.setToolTip(
"(str) timelapse_mode — which backend links objects between "
"frames.")
self._displacement = QDoubleSpinBox(self)
self._displacement.setRange(1, 2000)
self._displacement.setValue(50.0)
self._displacement.setSuffix(" px")
self._displacement.setToolTip(
"(int, px) timelapse_displacement — the largest jump a linker "
"will accept. Too small fragments tracks; too large swaps "
"identities.")
self._memory = QSpinBox(self)
self._memory.setRange(0, 50)
self._memory.setValue(3)
self._memory.setToolTip(
"(int) timelapse_memory — how many frames an object may vanish "
"and still rejoin its track (trackpy only).")
self._iou = QDoubleSpinBox(self)
self._iou.setRange(0.0, 1.0)
self._iou.setSingleStep(0.05)
self._iou.setValue(0.1)
self._iou.setToolTip(
"(float) Minimum IoU accepted by the 'iou' linker.")
self._min_len = QSpinBox(self)
self._min_len.setRange(1, 500)
self._min_len.setValue(3)
self._min_len.setToolTip(
"(int) Tracks shorter than this are counted as fragments in the "
"indicators below.")
self._remove_transient = Toggle(
"Keep only full-length tracks", self)
self._remove_transient.setToolTip(
"(bool) timelapse_remove_transient — drop every track not present "
"in all frames.")
self._tail = QSpinBox(self)
self._tail.setRange(0, 200)
self._tail.setValue(12)
self._tail.setToolTip(
"(int) How many frames of track history are drawn behind each "
"object.")
seg_group = QGroupBox("Segmentation (cached — changing these re-segments)")
seg_form = QFormLayout(seg_group)
seg_form.addRow("Model", self._model_box)
seg_form.addRow("Tracked object", self._object_box)
seg_form.addRow("Channel", self._channel)
seg_form.addRow("Diameter", self._diameter)
seg_form.addRow("Flow threshold", self._flow)
seg_form.addRow("Cell probability", self._prob)
seg_form.addRow("Frames previewed", self._max_frames)
seg_form.addRow(self._normalise)
trk_group = QGroupBox("Tracking (live — changing these only re-links)")
trk_form = QFormLayout(trk_group)
trk_form.addRow("Mode", self._mode_box)
trk_form.addRow("Max displacement", self._displacement)
trk_form.addRow("Memory", self._memory)
trk_form.addRow("IoU threshold", self._iou)
trk_form.addRow("Min track length", self._min_len)
trk_form.addRow("Track tail", self._tail)
trk_form.addRow(self._remove_transient)
groups_host = QWidget(self)
groups = QHBoxLayout(groups_host)
groups.setContentsMargins(0, 0, 0, 0)
groups.addWidget(seg_group, 1)
groups.addWidget(trk_group, 1)
from .collapsible_splitter import CollapsibleSplitter
key = self.SECTION_KEY
self._section_split = CollapsibleSplitter(
Qt.Vertical, self, persist_key=f"{key}::sections")
self._sections = {}
self._sections["Preview settings"] = self._section_split.add_section(
groups_host, "Preview settings", stretch=0,
persist_key=f"{key}/Preview settings")
for w in (self._displacement, self._memory, self._iou):
w.valueChanged.connect(self._on_tracking_changed)
self._mode_box.currentTextChanged.connect(self._on_tracking_changed)
self._remove_transient.toggled.connect(self._on_tracking_changed)
self._min_len.valueChanged.connect(self._on_scoring_changed)
self._tail.valueChanged.connect(lambda *_: self._refresh_canvases())
self._normalise.toggled.connect(lambda *_: self._refresh_canvases())
self._channel.valueChanged.connect(self._sync_channel_combo_from_spin)
act = QHBoxLayout()
self._run_btn = QPushButton(PREVIEW_RUN_TEXT, self)
self._run_btn.clicked.connect(self.run_preview)
self._cancel_btn = QPushButton(PREVIEW_CANCEL_TEXT, self)
self._cancel_btn.setToolTip(
"Abandon the pass in flight. Cellpose cannot be interrupted, so "
"it finishes in the background and its result is dropped.")
self._cancel_btn.setEnabled(False)
self._cancel_btn.clicked.connect(self.cancel_preview)
self._relink_btn = QPushButton("Re-link", self)
self._relink_btn.setToolTip(
"Re-run only the tracker on the cached per-frame masks.")
self._relink_btn.clicked.connect(self.relink)
self._event_annotation_btn = QPushButton(tr("Event annotations…"), self)
self._event_annotation_btn.setObjectName("TimelapseEventAnnotationButton")
self._event_annotation_btn.setToolTip(
tr("Label events on an existing tracked field for event detection."))
self._event_annotation_btn.clicked.connect(self._open_event_annotations)
self._propagate_btn = QPushButton("Propagate settings", self)
self._propagate_btn.setObjectName("ToggleButton")
self._propagate_btn.setCheckable(True)
self._propagate_btn.setToolTip(
"When on, the settings tuned here are copied into the main "
"Timelapse settings so the run uses them.")
self._propagate_btn.toggled.connect(self._on_propagate_toggled)
self._status = QLabel("", self)
act.addWidget(self._run_btn)
act.addWidget(self._cancel_btn)
act.addWidget(self._relink_btn)
act.addWidget(self._event_annotation_btn)
act.addWidget(self._propagate_btn)
act.addWidget(self._status, 1)
from .preview_scale import install_preview_scale
self._scale_control = install_preview_scale(self, "timelapse", act)
root.addLayout(act)
movie = QWidget(self)
movie_col = QVBoxLayout(movie)
movie_col.setContentsMargins(0, 0, 0, 0)
canvas = QHBoxLayout()
self._src_view = _ZoomView(self)
self._src_view.setMinimumHeight(160)
self._out_view = _ZoomView(self)
self._out_view.setMinimumHeight(160)
self._src_view.set_peer(self._out_view)
self._out_view.set_peer(self._src_view)
canvas.addWidget(self._src_view, 1)
canvas.addWidget(self._out_view, 1)
movie_col.addLayout(canvas, 1)
scrub = QHBoxLayout()
scrub.addWidget(QLabel("Frame", self))
self._play_btn = QPushButton("Play", self)
self._play_btn.setEnabled(False)
self._play_btn.setToolTip(
"Play the preview as a loop. The source and tracked views stay "
"synchronised while zoomed or panned.")
self._play_btn.clicked.connect(self._toggle_playback)
scrub.addWidget(self._play_btn)
self._frame_slider = QSlider(Qt.Horizontal, self)
self._frame_slider.setMinimum(0)
self._frame_slider.setMaximum(0)
self._frame_slider.valueChanged.connect(self._on_scrub)
self._frame_label = QLabel("–", self)
self._frame_label.setStyleSheet("font-family: monospace;")
self._play_fps = QSpinBox(self)
self._play_fps.setRange(1, 30)
self._play_fps.setValue(8)
self._play_fps.setSuffix(" fps")
self._play_fps.setToolTip("Playback speed; this does not alter data.")
self._play_fps.valueChanged.connect(self._update_playback_interval)
scrub.addWidget(self._frame_slider, 1)
scrub.addWidget(self._frame_label)
scrub.addWidget(self._play_fps)
movie_col.addLayout(scrub)
self._sections["Movie"] = self._section_split.add_section(
movie, "Movie", stretch=1, persist_key=f"{key}/Movie")
self._stats_label = QLabel(
"Load a sequence and run the preview to see track quality.", self)
self._stats_label.setWordWrap(True)
self._stats_label.setStyleSheet("font-family: monospace;")
self._sections["Track quality"] = self._section_split.add_section(
self._stats_label, "Track quality", stretch=0,
persist_key=f"{key}/Track quality")
root.addWidget(self._section_split, 1)
def _open_event_annotations(self) -> None:
"""Open the alpha editor using real exported track IDs, not preview IDs."""
from ..preferences import _is_alpha_visible
if not _is_alpha_visible("widgets", "TimelapseEventAnnotationButton"):
return
dialog = _EventAnnotationDialog(
self, sequence_path=self._sequence_path,
threaded=self._jobs._threaded)
try:
dialog.exec()
if dialog._saved_path is not None and self._propagate_cb is not None:
self._propagate_cb({"timelapse_events_annotations":
dialog._saved_path})
finally:
dialog.close()
dialog.deleteLater()
def _dropped_path(self, event) -> Optional[str]:
"""The usable path out of a drop, or None.
:param event: the Qt drop event.
:returns: the path, or None.
"""
mime = event.mimeData()
if not mime.hasUrls():
return None
for url in mime.urls():
if not url.isLocalFile():
continue
p = Path(url.toLocalFile())
if (p.suffix.lower() in FRAME_SUFFIXES
or path_probe.isdir(str(p), default=not p.suffix)):
return str(p)
return None
[docs]
def dragEnterEvent(self, event): # noqa: N802 (Qt naming)
"""Accept a drag carrying a timelapse folder or one of its frames.
:param event: the Qt drag event.
"""
if self._dropped_path(event) is not None:
event.acceptProposedAction()
else:
event.ignore()
[docs]
def dragMoveEvent(self, event): # noqa: N802
"""Keep accepting while a timelapse folder or one of its frames stays over the panel.
:param event: the Qt drag event.
"""
if self._dropped_path(event) is not None:
event.acceptProposedAction()
else:
event.ignore()
[docs]
def dropEvent(self, event): # noqa: N802
"""Take the dropped input and preview it.
:param event: the Qt drop event.
"""
p = self._dropped_path(event)
if p is None:
event.ignore()
return
event.acceptProposedAction()
self.load_sequence_async(p)
@property
def _loads_in_flight(self) -> List[int]:
"""Outstanding opens, as a list so ``not ...`` reads naturally."""
runner = getattr(self, "_jobs", None)
return [] if runner is None else [0] * runner.pending_jobs()
def _set_transient_status(self, text: str) -> None:
"""Write a placeholder a worker is expected to replace.
Remembered as well as shown, so :meth:`_on_job_failed` can tell a
line it is allowed to overwrite from one the user has since been
given for a different reason.
"""
self._transient_status = text
self._status.setText(text)
def _set_status(self, text: str) -> None:
"""Write a settled line, retiring whatever placeholder it replaces."""
self._transient_status = None
self._status.setText(text)
def _on_job_failed(self, message: str) -> None:
"""Replace a placeholder whose job died before it could deliver.
``JobRunner._on_settled`` runs ``on_done`` only for a job that
SUCCEEDED, so without this an open that raised on the worker left
"Opening field3…" on screen forever and the panel looked hung
rather than broken.
``job_failed`` is not generation-guarded -- a superseded job's
failure arrives just the same -- so the placeholder itself is the
guard: the line is replaced only while it is still the one written
before a submit. A failure that arrives after a newer load has
already reported something is dropped rather than painted over it.
"""
placeholder = getattr(self, "_transient_status", None)
if placeholder is None:
return
try:
if self._status.text() != placeholder:
self._transient_status = None
return
self._transient_status = None
self._status.setText(f"Load failed: {message}")
except RuntimeError:
self._transient_status = None
[docs]
def load_sequence_async(self, path, *, list_siblings: bool = True) -> bool:
"""Open ``path`` on a worker, then install it on the GUI thread.
Every GUI entry point -- the drop handler, the Choose-sequence dialog
and the FOV dropdown -- comes through here.
:param path: the image sequence: a directory of frames, a multi-page TIFF, or an ``.npy``
stack whose first axis is time. ``None`` or empty
submits nothing.
:param list_siblings: list the sibling fields next to ``path`` for
the field dropdown; ``False`` reuses the cached listing.
:returns: ``True`` when a job was submitted.
"""
text = os.fspath(path).strip() if path is not None else ""
if not text:
return False
self._stop_playback()
self._load_token += 1
token = self._load_token
cap = int(self._max_frames.value())
self._set_transient_status(f"Opening {os.path.basename(text)}…")
self._jobs.submit(
lambda: open_sequence_payload(text, cap, list_siblings),
lambda payload, _t=token: self._on_sequence_loaded(_t, payload))
return True
def _on_sequence_loaded(self, token: int, payload) -> None:
"""Install an opened sequence. Always on the GUI thread."""
if token != self._load_token or not isinstance(payload, dict):
return
if payload.get("error"):
self._set_status(payload["error"])
return
seq = payload.get("sequence")
if seq is None:
return
siblings = payload.get("siblings")
if siblings is not None:
self._sampler.enumerate_paths(
Path(payload["path"]).parent, lambda: siblings, force=True)
self._install_sequence(payload["path"], seq)
[docs]
def shutdown(self) -> None:
"""Abandon anything in flight and leave no QThread behind."""
for name in ("_jobs", "_movie_jobs"):
runner = getattr(self, name, None)
if runner is not None:
runner.shutdown()
[docs]
def load_sequence(self, path) -> bool:
"""Synchronously open ``path`` as the preview sequence.
For programmatic callers and tests, mirroring
``LivePreviewPanel.load_image``. The GUI uses
:meth:`load_sequence_async`.
:param path: the image sequence: a directory of frames, a multi-page TIFF, or an ``.npy``
stack whose first axis is time.
:returns: False, with the reason in the status line, when it cannot
be opened.
"""
self._stop_playback()
self._load_token += 1
payload = open_sequence_payload(
path, int(self._max_frames.value()), list_siblings=False)
if payload["error"]:
self._set_status(payload["error"])
return False
self._install_sequence(path, payload["sequence"])
return True
def _install_sequence(self, path, seq) -> bool:
"""Adopt an already-opened sequence and redraw."""
self._reset_movie_fields(clear_panel=True)
seq._register_cache_budget()
self._sequence = seq
self._masks = None
self._movie_images = None
self._tracked = None
self._tracks = None
self._mask_cache.clear()
self._mask_cache_last_used.clear()
self._path_label.setText(seq.describe())
self._frame_slider.setMaximum(max(0, len(seq) - 1))
self._frame_slider.setValue(0)
self._play_btn.setEnabled(len(seq) > 1)
self._sequence_path = Path(os.fspath(path))
self._refresh_source_selectors()
note = self.sample_note()
self._set_status(
f"Loaded {seq.describe()} — run the preview to segment + link."
+ (f" ({note})" if note else ""))
self._refresh_canvases()
return True
def _frame_channel_count(self) -> int:
"""How many channels one frame of the loaded sequence holds."""
seq = self._sequence
if seq is None or not len(seq):
return 0
try:
frame = np.asarray(seq.frame(0))
except Exception:
return 0
if frame.ndim != 3:
return 1
if frame.shape[-1] <= 8 and frame.shape[0] > 8:
return int(frame.shape[-1])
if frame.shape[0] <= 8 and frame.shape[-1] > 8:
return int(frame.shape[0])
return int(frame.shape[-1])
def _loaded_source_is_a_folder(self, source) -> bool:
"""Whether the loaded field of view is a folder of frames.
WITHOUT A STAT, and that is the whole point of the method.
``source.is_dir()`` used to be called here and in
:meth:`_movie_source_paths`, both on the GUI thread, both on the path
the user chose. Under ``/nas_mnt`` -- an ``autofs`` mount with a
sleeping share -- one such stat had not returned after TWENTY SECONDS
when this was measured: and
:meth:`_on_max_sets_changed` runs this path on every click of the
sets spinner. See :mod:`spacr.qt.path_probe`.
The answer is already in hand: :meth:`FrameSequence.open` ran on a
worker and recorded the layout it found, and ``kind == "files"`` is
set from the directory branch and from nowhere else. Only when there
is no sequence to ask -- nothing installs a path without one, so this
is the belt-and-braces arm -- does it fall back to the probe cache,
which answers from the name until a background check replaces it.
:param source: the loaded sequence's path.
:returns: True when the field of view is a directory of frames.
"""
kind = getattr(getattr(self, "_sequence", None), "kind", None)
if kind is not None:
return kind == "files"
text = str(source)
return path_probe.isdir(text, default=not Path(text).suffix)
def _refresh_source_selectors(self) -> None:
"""Re-fill the sets and channel dropdowns for the loaded sequence.
A timelapse field of view is a whole folder of frames (or one stack),
so there is nothing to group — but a plate still holds thousands of
them, and the dropdown lists a bounded random sample rather than all
of them. The listing is cached per folder, so stepping through fields
re-lists nothing.
NOTHING HERE TOUCHES THE DISK ON THE GUI THREAD. The layout question
goes to :meth:`_loaded_source_is_a_folder`, which reads it off the
opened sequence, and the listing lambda is not called at all on the
asynchronous path: ``open_sequence_payload`` lists the siblings on
the worker and ``_on_sequence_loaded`` adopts that listing (always --
even a failed listing yields the field itself) before this runs, so
``enumerate_paths`` finds its cache key already set. The lambda is
reached only by the deliberately synchronous :meth:`load_sequence`,
whose contract is that it blocks its caller.
"""
source = getattr(self, "_sequence_path", None)
if source is not None:
directories = self._loaded_source_is_a_folder(source)
self._sampler.enumerate_paths(
source.parent,
lambda: sibling_sources(source, FRAME_SUFFIXES,
directories=directories))
self._sample_note = apply_sample_to_combo(
self._fov_box, self._max_sets_box, self._sampler, source,
tooltip="Field of view")
populate_channel_combo(
self._channel_box, self._frame_channel_count(), include_all=False,
keep=f"Ch {int(self._channel.value())}")
self._sync_channel_spin_from_combo()
def _sync_channel_spin_from_combo(self) -> None:
"""Push the dropdown's channel into the segmentation spinner."""
index = selected_channel(self._channel_box)
if index is None or int(self._channel.value()) == int(index):
return
self._channel.setValue(int(index))
def _sync_channel_combo_from_spin(self, *_args) -> None:
"""Reflect a spinner-side channel change in the dropdown."""
box = getattr(self, "_channel_box", None)
if box is None:
return
wanted = f"Ch {int(self._channel.value())}"
index = box.findText(wanted)
if index < 0 or index == box.currentIndex():
return
blocked = box.blockSignals(True)
try:
box.setCurrentIndex(index)
finally:
box.blockSignals(blocked)
[docs]
def sample_note(self) -> str:
"""The sentence stating this preview is a sample of N of M sets."""
return getattr(self, "_sample_note", "")
def _on_max_sets_changed(self, value: int) -> None:
"""Draw a new sample at the user's new cap — without re-listing."""
if not self._sampler.set_max(int(value)):
return
self._refresh_source_selectors()
if self.sample_note():
self._status.setText(
self.sample_note()[:1].upper() + self.sample_note()[1:])
def _on_fov_changed(self, *_args) -> None:
"""Load the field of view the user picked from the dropdown."""
if self._loading_fov:
return
path = self._fov_box.currentData()
current = getattr(self, "_sequence_path", None)
if not path or (current is not None and str(current) == str(path)):
return
self._loading_fov = True
try:
self.load_sequence_async(path, list_siblings=False)
finally:
self._loading_fov = False
[docs]
def display_channel(self) -> Optional[int]:
"""Channel index the canvases show, or ``None`` when unset."""
return selected_channel(self._channel_box)
def _on_display_channel_changed(self, *_args) -> None:
"""Show (and segment) the newly selected channel."""
if not hasattr(self, "_channel"):
return
self._sync_channel_spin_from_combo()
self._refresh_canvases()
[docs]
def load_masks_async(self, path) -> bool:
"""Open ``path`` as a mask sequence on a worker, then install it here.
THE GUI ENTRY POINT, and the reason it exists is that
:meth:`load_masks` opens the sequence inline.
``FrameSequence.open`` on a folder of label images is a stat, a
listing and one ``is_file()`` per entry -- hundreds of round trips
for a plate -- and the folder comes from the user, which on one such
workstation means it can be a sleeping ``/nas_mnt`` share
where a single stat had not returned after twenty seconds (measured;
see :mod:`spacr.qt.path_probe`). Run from
:meth:`_pick_masks` that froze the whole window the moment the file
dialog closed.
The same worker function as the image sequence, so the mask
sequence's first frame is warmed off the GUI thread too.
:param path: the label images: a directory of frames, a multi-page TIFF, or an ``.npy``
stack whose first axis is time. ``None`` or empty
submits nothing.
:returns: ``True`` when a job was submitted.
"""
text = os.fspath(path).strip() if path is not None else ""
if not text:
return False
self._stop_playback()
self._mask_load_token += 1
token = self._mask_load_token
cap = int(self._max_frames.value())
self._set_transient_status(
f"Opening masks from {os.path.basename(text)}…")
self._jobs.submit(
lambda: open_sequence_payload(text, cap, list_siblings=False),
lambda payload, _t=token: self._on_masks_loaded(_t, payload))
return True
def _on_masks_loaded(self, token: int, payload) -> None:
"""Install an opened mask sequence. Always on the GUI thread.
Generation-guarded on ``_mask_load_token``: two mask folders picked
in quick succession, or one picked and then abandoned, must not let
the slower open paint over the newer one.
"""
if token != self._mask_load_token or not isinstance(payload, dict):
return
if payload.get("error"):
self._set_status(f"Mask load failed: {payload['error']}")
return
seq = payload.get("sequence")
if seq is None:
return
self._install_masks(seq)
[docs]
def load_masks(self, path) -> bool:
"""Synchronously use ready-made label images instead of segmenting.
For programmatic callers and tests. The GUI uses
:meth:`load_masks_async`, because this one opens the folder on the
thread that calls it.
:param path: the label images: a directory of frames, a multi-page TIFF, or an ``.npy``
stack whose first axis is time.
:returns: False, with the reason in the status line, when it cannot
be opened.
"""
self._stop_playback()
self._mask_load_token += 1
try:
seq = FrameSequence.open(path, max_frames=self._max_frames.value())
except Exception as e:
self._set_status(f"Mask load failed: {e}")
return False
self._install_masks(seq)
return True
def _install_masks(self, seq) -> bool:
"""Adopt an already-opened mask sequence and redraw. GUI thread."""
self._mask_sequence = seq
seq._register_cache_budget()
self._masks = None
self._mask_cache.clear()
self._mask_cache_last_used.clear()
if self._sequence is None:
self._frame_slider.setMaximum(max(0, len(seq) - 1))
self._frame_slider.setValue(0)
self._play_btn.setEnabled(len(seq) > 1)
self._set_status(
f"Masks: {seq.describe()} — segmentation will be skipped.")
return True
[docs]
def set_propagate_callback(self, cb) -> None:
"""Register a ``callback(dict)`` that writes tuned values into the
main settings panel (wired by the AppScreen).
:param cb: the callable, or ``None`` to remove it.
"""
self._propagate_cb = cb
[docs]
def settings_for_propagation(self) -> dict:
"""Map the preview's widgets onto real Timelapse setting keys."""
obj = self._object_box.currentText()
return {
"timelapse_mode": self._mode_box.currentText(),
"timelapse_displacement": int(self._displacement.value()),
"timelapse_memory": int(self._memory.value()),
"timelapse_objects": [obj],
"timelapse_remove_transient": bool(
self._remove_transient.isChecked()),
"timelapse_frame_limits": [0, int(self._max_frames.value())],
f"{obj}_channel": int(self._channel.value()),
f"{obj}_diameter": float(self._diameter.value()),
"cell_flow_threshold": float(self._flow.value()),
"cell_cellprob_threshold": float(self._prob.value()),
"normalize": bool(self._normalise.isChecked()),
}
[docs]
def propagate_settings(self) -> None:
"""Push the current settings to the main panel, if wired."""
if self._propagate_cb is not None:
try:
self._propagate_cb(self.settings_for_propagation())
except Exception:
LOG.debug("propagate_settings failed", exc_info=True)
[docs]
def apply_settings(self, settings: dict) -> None:
"""Seed the preview from the main Timelapse settings dict.
:param settings: the Timelapse settings; a copy is kept, and the
linking mode, displacement, memory, object, transient filter and
that object's channel, diameter and model are read from it.
"""
self._settings = dict(settings or {})
try:
mode = settings.get("timelapse_mode")
if mode and self._mode_box.findText(str(mode)) >= 0:
self._mode_box.setCurrentText(str(mode))
disp = settings.get("timelapse_displacement")
if disp:
self._displacement.setValue(float(disp))
mem = settings.get("timelapse_memory")
if mem is not None:
self._memory.setValue(int(mem))
objs = settings.get("timelapse_objects")
if objs and self._object_box.findText(str(objs[0])) >= 0:
self._object_box.setCurrentText(str(objs[0]))
self._remove_transient.setChecked(
bool(settings.get("timelapse_remove_transient", False)))
obj = self._object_box.currentText()
ch = settings.get(f"{obj}_channel")
if ch is not None:
self._channel.setValue(int(ch))
diam = settings.get(f"{obj}_diameter")
if diam:
self._diameter.setValue(float(diam))
wanted, _key, here = _model_the_run_would_use(
self._settings, obj)
_offer_the_run_model(self._model_box, wanted, here)
except Exception:
LOG.debug("apply_settings failed", exc_info=True)
[docs]
def current_params(self) -> dict:
"""Snapshot for tests + external callers."""
return {
"model": self._model_box.currentText(),
"object": self._object_box.currentText(),
"channel": int(self._channel.value()),
"mode": self._mode_box.currentText(),
"displacement": float(self._displacement.value()),
"memory": int(self._memory.value()),
"iou_threshold": float(self._iou.value()),
"min_length": int(self._min_len.value()),
"max_frames": int(self._max_frames.value()),
"n_frames": len(self._sequence) if self._sequence else 0,
"display_channel": self.display_channel(),
"fov": self._fov_box.currentText(),
}
def _cache_budget_entries(self):
"""Derived mask stacks retained for re-linking under new settings.
The stack currently drawn, and one a worker is currently re-linking,
are pinned. Older segmentation signatures are reproducible caches and
can be evicted independently.
"""
now = time.time()
worker_key = (self._pending_signature
if self._worker is not None else None)
return [
(key, max(0, int(value.nbytes)),
float(self._mask_cache_last_used.get(key, now)),
value is self._masks or key == worker_key)
for key, value in list(self._mask_cache.items())
]
def _drop_cache_budget_entry(self, key) -> bool:
"""Drop one inactive segmentation result chosen by the policy."""
value = self._mask_cache.get(key)
if value is None or value is self._masks:
return False
if self._worker is not None and key == self._pending_signature:
return False
self._mask_cache.pop(key, None)
self._mask_cache_last_used.pop(key, None)
return True
def _segmentation_signature(self) -> tuple:
"""Everything that can change a *label image*, and nothing else.
Tracking settings are deliberately absent: that is what makes a
tracking change a cache hit and therefore free.
"""
seq = self._sequence
msk = self._mask_sequence
return (
getattr(msk, "label", None),
getattr(seq, "label", None),
tuple(getattr(seq, "indices", ())),
None if msk is not None else self._model_box.currentText(),
None if msk is not None else int(self._channel.value()),
None if msk is not None else float(self._diameter.value()),
None if msk is not None else float(self._flow.value()),
None if msk is not None else float(self._prob.value()),
None if msk is not None else bool(self._normalise.isChecked()),
)
def _seg_params(self) -> Dict[str, Any]:
"""The segmentation settings the controls currently describe.
:returns: the parameters.
"""
return {
"model": self._model_box.currentText(),
"channel": int(self._channel.value()),
"mask_channel": int(self._channel.value()),
"diameter": float(self._diameter.value()),
"flow_threshold": float(self._flow.value()),
"cellprob": float(self._prob.value()),
"normalise": bool(self._normalise.isChecked()),
}
def _track_params(self) -> Dict[str, Any]:
"""The tracking settings the controls currently describe.
:returns: the parameters.
"""
return {
"mode": self._mode_box.currentText(),
"displacement": float(self._displacement.value()),
"memory": int(self._memory.value()),
"iou_threshold": float(self._iou.value()),
"trackastra_model": str(
self._settings.get("trackastra_model", "general_2d")),
"trackastra_linking": str(
self._settings.get("trackastra_linking", "greedy")),
}
[docs]
def run_preview(self) -> None:
"""Segment (unless cached) then link, off the GUI thread."""
self._start(allow_segmentation=True)
[docs]
def relink(self) -> None:
"""Re-link the cached masks. Never re-segments."""
self._start(allow_segmentation=False)
def _on_tracking_changed(self, *_):
"""A linking knob moved — re-link if masks are already cached.
Silent when nothing has been segmented yet: the user is still
setting up, and a change should not kick off an expensive pass they
did not ask for.
"""
if self._masks is None:
return
self.relink()
def _on_scoring_changed(self, *_):
"""The fragment threshold moved — re-score, don't re-link."""
if self._masks is None or self._raw_tracks is None:
return
self._apply_tracks(self._raw_tracks, note="Re-scored (cached tracks)")
def _preview_blocked_reason(self) -> str:
"""Why this panel cannot track right now, or ``""``.
Two reasons, both of which the user can act on: nothing is loaded,
or the tracker they picked is not installed.
"""
if self._sequence is None and self._mask_sequence is None:
return self.PREVIEW_SOURCE_HINT
ok, why = backend_available(self._mode_box.currentText())
return "" if ok else str(why)
def _start(self, allow_segmentation: bool) -> None:
"""Run the preview, optionally segmenting first.
:param allow_segmentation: False to preview existing masks only.
"""
blocked = self.preview_blocked_reason()
if not self.begin_preview():
if blocked and blocked != self.PREVIEW_SOURCE_HINT:
self.preview_ready.emit(None)
return
sig = self._segmentation_signature()
cached = self._mask_cache.get(sig)
if cached is not None:
self._mask_cache_last_used[sig] = time.time()
if cached is None and not allow_segmentation:
self.set_preview_busy(False)
self._status.setText(
"No cached masks for these segmentation settings — "
"hit Run preview.")
return
self._pending_signature = sig
req = TimelapseRequest(
sequence=self._sequence,
mask_sequence=self._mask_sequence,
cached_masks=cached,
cached_images=self._movie_images,
seg=self._seg_params(),
track=self._track_params(),
include_images=getattr(self, "_movie_panel", None) is not None,
)
self._relink_btn.setEnabled(False)
if cached is not None:
note = "Re-linking cached masks…"
elif req.mask_sequence is not None:
note = "Reading the label images, then linking…"
else:
note = "Segmenting frames, then linking…"
self._status.setText(note)
self._release_worker()
worker = _TimelapseWorker(req, self)
worker.preview_token_value = self.preview_token()
self._pending_token = worker.preview_token_value
worker.finished_result.connect(self._on_worker_done)
worker.finished.connect(self._on_worker_finished)
self._worker = worker
worker.start()
def _release_worker(self) -> None:
"""Free a worker whose thread has already exited.
Parented to the panel, so C++ owns it and it would otherwise hold a
whole mask stack until the panel itself died. Unparenting hands
ownership back to Python. Mirrors ``LivePreviewPanel._release_worker``.
"""
old = self._retired_worker
self._retired_worker = None
if old is None:
return
try:
old.wait()
old.setParent(None)
except RuntimeError:
LOG.debug("worker already deleted", exc_info=True)
def _result_token(self):
"""The generation the result now landing belongs to."""
worker = self.sender()
token = getattr(worker, "preview_token_value", None)
return getattr(self, "_pending_token", 0) if token is None else token
def _on_worker_finished(self) -> None:
"""Relay for the worker thread's own ``finished`` signal.
A bound method on purpose (see :meth:`_start`). Returning the
buttons to the idle state here as well as in :meth:`_on_worker_done`
keeps them usable after a pass whose result was dropped as stale,
and this is the first moment at which the QThread may be freed.
"""
self.set_preview_busy(False)
self._relink_btn.setEnabled(True)
self._release_worker()
def _on_worker_done(self, result, err: str) -> None:
"""Adopt a finished pass. Runs on the GUI thread (queued signal)."""
if self.preview_stale(self._result_token()):
LOG.debug("dropping a superseded timelapse preview result")
return
if self._worker is not None:
self._retired_worker = self._worker
self._worker = None
self.set_preview_busy(False)
self._relink_btn.setEnabled(True)
if err:
self._status.setText(preview_failure_message(err))
self.preview_ready.emit(None)
return
if not result:
self._status.setText("Preview returned nothing.")
self.preview_ready.emit(None)
return
masks = result["masks"]
self._masks = masks
self._movie_images = result.get("images")
sig = getattr(self, "_pending_signature", None)
if sig is not None:
self._mask_cache[sig] = masks
self._mask_cache_last_used[sig] = time.time()
note = ("Masks built + linked" if result.get("masks_built")
else "Re-linked (cached masks)")
self._apply_tracks(result["tracks"], note=note)
def _apply_tracks(self, tracks, note: str = "Linked") -> None:
"""Filter, relabel, score and render a track table.
Called both after a worker pass and when only the scoring threshold
moved, which is why the status text is passed in.
"""
self._raw_tracks = tracks
n_frames = int(self._masks.shape[0])
if (tracks is not None and len(tracks)
and self._remove_transient.isChecked()):
full = tracks.groupby("track_id")["frame"].nunique() == n_frames
keep = set(full[full].index)
tracks = tracks[tracks["track_id"].isin(keep)]
self._tracks = tracks
self._tracked = relabel_by_track(self._masks, tracks)
self._stats = track_stats(
tracks, n_frames,
min_length=int(self._min_len.value()),
displacement_limit=float(self._displacement.value()))
self._frame_slider.setMaximum(max(0, n_frames - 1))
self._play_btn.setEnabled(n_frames > 1)
self._status.setText(f"{note} · {n_frames} frames")
self._stats_label.setText(
self._stats.summary()
+ "\nFragmentation and swap figures are indicators computed "
"without ground truth, not measurements.")
self._refresh_canvases()
self._push_to_movie()
if self._propagate_btn.isChecked():
self.propagate_settings()
self.preview_ready.emit(self._stats)
@staticmethod
def _freeze_movie_value(value):
"""Hash nested setting values without weakening their identity."""
if isinstance(value, dict):
return tuple(sorted(
(str(key), TimelapsePreviewPanel._freeze_movie_value(item))
for key, item in value.items()))
if isinstance(value, (list, tuple)):
return tuple(TimelapsePreviewPanel._freeze_movie_value(item)
for item in value)
return value
def _movie_setting_keys(self) -> Tuple[tuple, tuple]:
"""Segmentation and linking identities shared by every field."""
seg = dict(self._seg_params())
seg["max_frames"] = int(self._max_frames.value())
return (self._freeze_movie_value(seg),
self._freeze_movie_value(self._track_params()))
def _reset_movie_fields(self, *, clear_panel: bool = False) -> None:
"""Cancel sibling work and release every retained movie-field array."""
self._movie_generation += 1
runner = getattr(self, "_movie_jobs", None)
if runner is not None:
runner.cancel()
self._movie_pending_path = None
self._movie_pending_key = None
self._movie_fields.clear()
self._movie_sources.clear()
self._movie_failures.clear()
self._movie_seg_key = None
self._movie_track_key = None
if clear_panel:
movie = getattr(self, "_movie_panel", None)
if movie is not None:
movie.set_fields([])
def _movie_source_paths(self) -> List[str]:
"""Current field first, followed by its cached sibling listing.
``open_sequence_payload`` obtains that listing with
:func:`sibling_sources` on its worker and ``_on_sequence_loaded``
adopts it into ``ImageSetSampler`` -- ALWAYS, since a listing that
failed still yields the loaded field. So on every path a user can
drive, ``self._sampler.sets`` is non-empty by the time this runs and
the fallback below is not reached; it remains for the deliberately
synchronous :meth:`load_sequence`, and for a caller that has emptied
the sampler by hand, both of which accept a blocking listing.
The layout question, which used to be ``current_path.is_dir()``
here, is answered off the opened sequence instead -- it was a stat on
the GUI thread for a path the user chose, and it ran BEFORE the
listing whose result decides whether it was needed at all.
"""
current_path = getattr(self, "_sequence_path", None)
if current_path is None:
return []
current = str(current_path)
siblings = []
for item in self._sampler.sets:
try:
siblings.append(str(item.path()))
except Exception: # noqa: BLE001
continue
if not siblings:
siblings = [str(path) for path in sibling_sources(
current_path, FRAME_SUFFIXES,
directories=self._loaded_source_is_a_folder(current_path))]
return [current] + [path for path in siblings if path != current]
def _desired_movie_sources(self) -> List[str]:
"""Which sequences the current settings say should be shown.
:returns: the sources.
"""
movie = getattr(self, "_movie_panel", None)
if movie is None:
return []
return list(self._movie_sources[: max(1, int(movie.max_fields()))])
def _movie_entry_is_current(self, entry: Optional[dict]) -> bool:
"""Whether a loaded sequence still matches the settings.
CHECKED BEFORE REUSE, so a movie built under the previous settings is
not shown as though it answered the current ones.
:param entry: the loaded sequence.
:returns: True when it is still valid.
"""
return bool(
entry
and entry.get("_seg_key") == self._movie_seg_key
and entry.get("_track_key") == self._movie_track_key
and not entry.get("_needs_refresh"))
def _present_movie_fields(self) -> None:
"""Publish completed current-generation fields in source order."""
movie = getattr(self, "_movie_panel", None)
if movie is None:
return
ready = []
for source in self._desired_movie_sources():
entry = self._movie_fields.get(source)
if self._movie_entry_is_current(entry):
ready.append(entry)
movie.set_fields(ready)
def _cancel_pending_movie_field(self) -> None:
"""Abandon a field whose movie is still being built."""
self._movie_generation += 1
self._movie_pending_path = None
self._movie_pending_key = None
runner = getattr(self, "_movie_jobs", None)
if runner is not None:
runner.cancel()
def _refresh_movie_targets(self) -> None:
"""Trim to the live cap, then start at most one missing field."""
desired = self._desired_movie_sources()
desired_set = set(desired)
for source in list(self._movie_fields):
if source not in desired_set:
self._movie_fields.pop(source, None)
for source in list(self._movie_failures):
if source not in desired_set:
self._movie_failures.pop(source, None)
wanted_key = (self._movie_seg_key, self._movie_track_key)
pending = self._movie_pending_path
if pending is not None and (
pending not in desired_set
or self._movie_pending_key != wanted_key):
self._cancel_pending_movie_field()
pending = None
self._present_movie_fields()
if pending is not None:
return
source = None
cached_masks = None
for candidate in desired:
entry = self._movie_fields.get(candidate)
if self._movie_entry_is_current(entry):
continue
if self._movie_failures.get(candidate) == wanted_key:
continue
source = candidate
if (entry is not None
and entry.get("_seg_key") == self._movie_seg_key):
cached_masks = entry.get("masks")
break
if source is None:
return
generation = self._movie_generation
self._movie_pending_path = source
self._movie_pending_key = wanted_key
kwargs = {
"path": source,
"max_frames": int(self._max_frames.value()),
"seg": dict(self._seg_params()),
"track": dict(self._track_params()),
"cached_masks": cached_masks,
"cancelled": movie_worker_interrupted,
}
self._status.setText(
f"Loading movie field {desired.index(source) + 1} "
f"of {len(desired)}…")
self._movie_jobs.submit(
lambda _kwargs=kwargs: movie_field_payload(**_kwargs),
lambda result, _source=source, _generation=generation,
_key=wanted_key: self._on_movie_field_done(
_source, _generation, _key, result))
def _on_movie_field_done(self, source: str, generation: int,
wanted_key: tuple, result) -> None:
"""Install one sibling result on the GUI thread, then take the next."""
if (generation != self._movie_generation
or wanted_key != (self._movie_seg_key,
self._movie_track_key)):
return
self._movie_pending_path = None
self._movie_pending_key = None
if not isinstance(result, dict) or result.get("cancelled"):
self._refresh_movie_targets()
return
if result.get("error"):
self._movie_failures[source] = wanted_key
self._status.setText(
f"Movie field {Path(source).name} failed: {result['error']}")
self._refresh_movie_targets()
return
result["_seg_key"] = self._movie_seg_key
result["_track_key"] = self._movie_track_key
result["_needs_refresh"] = False
self._movie_fields[source] = result
self._present_movie_fields()
shown = len([
path for path in self._desired_movie_sources()
if self._movie_entry_is_current(self._movie_fields.get(path))])
self._status.setText(f"Movie ready · {shown} field(s)")
self._refresh_movie_targets()
def _on_movie_field_limit_changed(self, _count: int) -> None:
"""Apply a lower cap immediately; a higher one resumes the queue."""
self._refresh_movie_targets()
def _push_to_movie(self) -> None:
"""Publish the selected field, then stream sibling fields as ready.
Every ``labels`` stack is relabelled by track id. Additional fields
are opened, segmented and linked on ``_movie_jobs`` one at a time;
only the selected field is assembled here, and its raw frames normally
arrived with the preview worker result. If the movie was attached
after that result, masks are shown briefly while the raw frames are
read and re-linked off the GUI thread.
"""
movie = getattr(self, "_movie_panel", None)
if movie is None:
return
if self._masks is None or self._tracked is None:
movie.set_fields([])
return
source_path = getattr(self, "_sequence_path", None)
source = str(source_path) if source_path is not None else "Field"
seg_key, track_key = self._movie_setting_keys()
if (self._movie_seg_key is not None and seg_key != self._movie_seg_key):
self._cancel_pending_movie_field()
self._movie_fields.clear()
self._movie_failures.clear()
elif (self._movie_track_key is not None
and track_key != self._movie_track_key):
self._cancel_pending_movie_field()
self._movie_failures.clear()
self._movie_seg_key = seg_key
self._movie_track_key = track_key
self._movie_sources = self._movie_source_paths() or [source]
images = self._movie_images
needs_refresh = images is None and self._sequence is not None
if images is None:
images = self._masks
self._movie_fields[source] = {
"source": source,
"title": Path(source).name or "Field",
"images": images,
"masks": self._masks,
"labels": self._tracked,
"tracks": self._tracks,
"channel": int(self._channel.value()),
"_seg_key": seg_key,
"_track_key": track_key,
"_needs_refresh": needs_refresh,
}
self._refresh_movie_targets()
[docs]
def attach_movie_panel(self, movie) -> None:
"""Wire a :class:`TimelapseMoviePanel` to this preview.
Kept as a seam rather than a constructor argument so the movie is
optional: the panel is built by two different callers and a screen
that only wants the stats view should not pay for the frames.
:param movie: the movie panel; its ``max_fields_changed`` signal is
connected and a previously attached panel is disconnected.
"""
previous = getattr(self, "_movie_panel", None)
if previous is not None and previous is not movie:
try:
previous.max_fields_changed.disconnect(
self._on_movie_field_limit_changed)
except (RuntimeError, TypeError):
pass
self._movie_panel = movie
if previous is not movie:
movie.max_fields_changed.connect(
self._on_movie_field_limit_changed)
self._push_to_movie()
def _on_scrub(self, _value: int) -> None:
"""Show the frame the scrub bar now points at.
:param _value: the bar's position; re-read from the widget.
"""
self._refresh_canvases()
def _toggle_playback(self) -> None:
"""Start or pause looped playback of the loaded preview frames."""
if self._play_timer.isActive():
self._stop_playback()
return
if self._frame_slider.maximum() <= 0:
return
self._update_playback_interval()
self._play_timer.start()
self._play_btn.setText("Pause")
def _stop_playback(self) -> None:
"""Stop the playback timer."""
self._play_timer.stop()
if hasattr(self, "_play_btn"):
self._play_btn.setText("Play")
def _update_playback_interval(self, *_args) -> None:
"""Set the timer from the chosen frame rate."""
fps = max(1, int(self._play_fps.value()))
self._play_timer.setInterval(max(1, round(1000 / fps)))
def _advance_frame(self) -> None:
"""Advance one frame, wrapping at the end for continuous playback."""
last = self._frame_slider.maximum()
if last <= 0:
self._stop_playback()
return
current = self._frame_slider.value()
self._frame_slider.setValue(0 if current >= last else current + 1)
def _refresh_canvases(self) -> None:
"""Redraw every canvas for the current frame."""
seq = self._sequence
idx = int(self._frame_slider.value())
if seq is None:
if self._masks is None:
return
image = self._masks[min(idx, self._masks.shape[0] - 1)]
else:
if not len(seq):
return
image = seq.frame(min(idx, len(seq) - 1))
norm = self._normalise.isChecked()
plane = frame_channel(image, int(self._channel.value()))
self._src_view.set_pixmap(numpy_to_qpixmap(
_to_uint8(plane, normalise=norm)))
labels = None
if self._tracked is not None and idx < self._tracked.shape[0]:
labels = self._tracked[idx]
overlay = render_frame(
image, labels=labels, tracks=self._tracks, frame=idx,
tail=int(self._tail.value()), normalise=norm,
channel=int(self._channel.value()))
self._out_view.set_pixmap(numpy_to_qpixmap(overlay))
total = self._masks.shape[0] if self._masks is not None else (
len(seq) if seq is not None else 0)
self._frame_label.setText(f"{idx + 1}/{max(1, total)}")
def _on_propagate_toggled(self, on: bool) -> None:
"""Turn settings propagation on or off.
:param on: True to push settings to the run as they change.
"""
if on:
self.propagate_settings()
def _pick_sequence(self):
"""Ask for a sequence of frames to preview."""
path = QFileDialog.getExistingDirectory(
self, "Choose a folder of frames")
if path:
self.load_sequence_async(path)
def _pick_masks(self):
"""Ask for a set of masks to overlay."""
path = QFileDialog.getExistingDirectory(
self, "Choose a folder of label images")
if path:
self.load_masks_async(path)
[docs]
def closeEvent(self, event):
"""Let a running pass finish before the widget is torn down.
A ``QThread`` collected while it is still running aborts the whole
process, and this panel's worker outlives the emit that produced its
result by a few instructions.
:param event: the close event; passed on to the base class after
the workers finish (up to 5 s each).
"""
self._stop_playback()
self.shutdown()
for worker in (self._worker, getattr(self, "_retired_worker", None)):
if worker is not None:
try:
worker.wait(5000)
except RuntimeError:
LOG.debug("worker already deleted", exc_info=True)
super().closeEvent(event)
[docs]
def refresh_model_choices(self) -> None:
"""Re-read the Cellpose model list and add anything new.
`spacr.settings.cellpose_model_choices` only reads the API when
Cellpose is already imported, because importing it costs ~2.5 s and
this panel is built while a page is being laid out. That means the
first build usually gets the shipped fallback — so ask again every
time the panel is shown. After the first segmentation Cellpose is
loaded and a checkpoint the user registered appears here.
Additive on purpose: the current selection is never disturbed, and
an entry is never removed, so a value the user picked cannot vanish
under them because a probe came back thinner.
"""
wanted = _model_menu()
have = {self._model_box.itemText(i)
for i in range(self._model_box.count())}
for index, name in enumerate(wanted):
if name not in have:
self._model_box.insertItem(index, name)
[docs]
def showEvent(self, event): # noqa: N802 (Qt naming)
"""Refresh the model list whenever the panel comes back on screen.
:param event: the show event; passed on to the base class first.
"""
super().showEvent(event)
self.refresh_model_choices()
[docs]
def build_timelapse_preview_card(host, *, panel_later: bool = False):
"""Build the ``Track preview`` card + panel pair.
Mirrors ``spacr.qt.screens.hyperparam.build_hyperparam_card`` and
``app_screen._build_live_preview_card``: returns the pair without adding
it to any layout, so the host screen puts it in whatever splitter it
likes and starts it hidden behind the toggle.
:param host: the :class:`AppScreen` asking for the card.
:param panel_later: return ``None`` for the panel and leave the card
empty, for :func:`_fill_timelapse_preview_card` to fill the first
time it is shown -- how Mask carries this preview for its folded
Timelapse switch without building ~130 widgets nobody has asked to
see.
:returns: ``(panel, card)``.
"""
from .card import Card, _CardBuiltWhenShown
from .timelapse_movie import TimelapseMoviePanel # noqa: F401
card = (_CardBuiltWhenShown if panel_later else Card)(
title="Track preview")
if panel_later:
card.setMinimumHeight(320)
return None, card
panel = _fill_timelapse_preview_card(host, card)
card.setMinimumHeight(320)
return panel, card
def _fill_timelapse_preview_card(host, card):
"""Build the track preview and its movie panel into ``card``.
:param host: the screen the card belongs to; unused, and taken so every
preview's fill has the same shape.
:param card: a card from :func:`build_timelapse_preview_card`.
:returns: the preview panel.
"""
from .timelapse_movie import TimelapseMoviePanel
panel = TimelapsePreviewPanel(card)
card.body_layout.addWidget(panel)
movie = TimelapseMoviePanel(card)
card.body_layout.addWidget(movie, 1)
panel.attach_movie_panel(movie)
return panel