spacr.qt.annotate_engine

Pure-Python backend for the Qt annotate screen.

The image-processing pipeline (normalize / channel-filter / outline / colored border) and the SQLite-backed page fetch + background save worker are all Tk-free. The Qt screen wraps this with a QWidget UI.

Semantics mirror spacr.gui_elements.AnnotateApp so annotations made in either GUI are read/written the same way from the same measurements/measurements.db.

Exceptions

OutlineCancelled

Raised when a requested cancellation stops outline generation.

Classes

AnnotateSettings	Every knob the Annotate screen exposes, packed into one dataclass.
SaveWorker	Runs in a daemon thread; consumes {png_path: annotation} batches

Functions

add_colored_border(→ PIL.Image.Image)	Return img with an inset colored border of width px.
annotation_batch(→ Dict[str, Optional[int]])	Turn a path list into the batch SaveWorker.submit() takes.
cache_budget_entries()	Measured records for decoded outline arrays retained between draws.
class_counts(→ List[Tuple[int, int]])	Return sorted list of (class_value, count) for annotated rows.
clear_column(→ None)	Null every value in annotation_column of png_list.
count_rows(→ int)	Return the number of png_list rows, optionally filtered by image_type.
drop_cache_budget_entry(→ bool)	Evict one decoded array selected by the global memory policy.
empty_object_filters(→ Dict[str, ...)	Return all object-filter bounds in their disabled state.
ensure_annotation_column(→ None)	Add column INTEGER to table if missing and index png_path.
fetch_filtered_paths(→ List[Tuple[str, Optional[int]]])	Return ALL (png_path, annotation) rows matching every one of the
fetch_page(→ List[Tuple[str, Optional[int]]])	Read one page of (png_path, annotation) rows in insertion order.
filter_bound(→ Optional[float])	Parse a filter bound, returning None for empty or invalid input.
filter_channels_pil(→ PIL.Image.Image)	Zero out channels not present in channels (e.g. ['r','g']).
filter_key(→ str)	Return the settings key for a channel and measurement pair.
find_last_annotated_offset(→ Optional[int])	Return the page-aligned offset of the last annotated row, or None.
forget_outline_masks(→ None)	Drop every cached mask. For tests, and for a caller changing plates.
gate_paths(→ List[str])	png_paths surviving a chain of spacr.qt.widgets.gate_spec.Gate.
label_to_hex(→ Optional[str])	Map an annotation value to a hex border color.
load_crop_image(→ PIL.Image.Image)	Open one object crop PNG as an 8-bit RGB image in display order.
metadata_values(→ List[str])	The distinct values of one png_list metadata column, sorted.
normalize_object_filters(→ Dict[str, ...)	Normalise object-filter bounds and migrate legacy size limits.
normalize_pil(, normalize_channels)	Normalize the given PIL image per-channel using percentile stretch.
outline_image(, outline_method, object_filters[, ...])	Overlay per-channel object outlines on base_img.
parse_image_type(→ Tuple[str, List[str]])	Turn an image-type expression into a SQL fragment and its parameters.
paths_by_measurements(→ List[str])	png_paths satisfying EVERY {column, threshold, direction} rule.
paths_by_metadata(→ List[str])	png_paths whose column is one of values.

Module Contents

exception spacr.qt.annotate_engine.OutlineCancelled

Bases: Exception

Raised when a requested cancellation stops outline generation.

Cellpose model construction and inference cannot be interrupted safely. Cancellation is therefore checked between native calls, and this exception unwinds the current page of crops.

Initialize self. See help(type(self)) for accurate signature.

class spacr.qt.annotate_engine.AnnotateSettings

Every knob the Annotate screen exposes, packed into one dataclass.

Sensible defaults let callers instantiate AnnotateSettings() and override just the handful of fields they care about.

property page_size: int

Number of thumbnails per page (grid_rows * grid_cols, min 1).

class spacr.qt.annotate_engine.SaveWorker(db_path: str, annotation_column: str, *, table: str = DEFAULT_PNG_TABLE)

Runs in a daemon thread; consumes {png_path: annotation} batches from a Queue and commits them to the DB in coalesced transactions.

Prepare an idle worker; call start() to spawn its thread.

Parameters:

• db_path – path to the SQLite measurements.db.

• annotation_column – column in table to write into.

• table – the crop table being annotated. A generated set lands under png_list_2 and onwards, and a writer pointed at the wrong one would put a user’s labels on somebody else’s rows.

start() → None

Spawn the daemon writer thread if it isn’t already running.

stop(wait: bool = True) → None

Drain queued writes and stop the writer.

A bounded five-second join used to let the screen disappear while the daemon thread still owned a live SQLite connection. That is unsafe at application shutdown: CPython can finalize the sqlite extension while the thread is still inside it. SQLite already bounds lock waits with its 30-second connection timeout, so a requested blocking stop waits for the thread to close its cursor and connection completely.

submit(batch: dict, column: str | None = None) → None

Enqueue a copy of the batch for saving.

Parameters:

• batch – {png_path: value}; None clears.

• column – the column to write; the annotation column when omitted. The Annotate screen’s judgements go to the <column>_verdict column through this same writer rather than through a second connection: one writer, one queue, one order. A batch for another column that cannot be written is kept in _failed_extra under its column, apart from _failed_batch, so the annotation batch keeps the shape every reader of it expects.

property busy: bool

True while the writer thread is inside a commit.

property is_alive: bool

Whether the SQLite writer thread is still running.

property last_error: str | None

Actionable message for the latest writer failure, if any.

property last_save_ts: float | None

POSIX timestamp of the most recent successful commit, or None.

property pending_batches: int

Number of submitted-but-not-yet-committed batches.

spacr.qt.annotate_engine.add_colored_border(img: PIL.Image.Image, width: int, color: str) → PIL.Image.Image

Return img with an inset colored border of width px.

Kept for parity with the Tk AnnotateApp (and for callers that want a bordered image out of the pipeline). The Qt grid does NOT use it: its tiles paint their borders in _Thumbnail.paintEvent so recolouring one costs a repaint instead of a rebuilt pixmap.

Parameters:

• img – image to frame; it is pasted unchanged into an RGB canvas.

• width – border thickness in pixels on each side, so the result is 2 * width larger in both dimensions.

• color – PIL colour for the border, e.g. a name or "#rrggbb".

spacr.qt.annotate_engine.annotation_batch(paths: Iterable[str], value: int | None) → Dict[str, int | None]

Turn a path list into the batch SaveWorker.submit() takes.

Trivial, and it exists so every auto-annotation source ends at the same call. None clears, exactly as it does for a keystroke.

Parameters:

• paths – png_paths to label.

• value – the class number, or None to clear.

Returns:

{png_path: value}.

spacr.qt.annotate_engine.cache_budget_entries()

Measured records for decoded outline arrays retained between draws.

spacr.qt.annotate_engine.class_counts(db_path: str, annotation_column: str, *, table: str = DEFAULT_PNG_TABLE) → List[Tuple[int, int]]

Return sorted list of (class_value, count) for annotated rows.

Parameters:

• db_path – path to the SQLite measurement database; a missing file gives an empty list.

• annotation_column – integer annotation column in table. Values at or above spacr.suggest.SUGGESTION_OFFSET (model suggestions) and NULLs are not counted.

spacr.qt.annotate_engine.clear_column(db_path: str, annotation_column: str, *, table: str = DEFAULT_PNG_TABLE) → None

Null every value in annotation_column of png_list.

Parameters:

• db_path – path to measurements.db; missing files are ignored.

• annotation_column – column to reset.

spacr.qt.annotate_engine.count_rows(db_path: str, image_type: str | None = None, *, table: str = DEFAULT_PNG_TABLE) → int

Return the number of png_list rows, optionally filtered by image_type.

Parameters:

• db_path – path to measurements.db; missing files count as 0.

• image_type – optional substring to filter png_path on.

spacr.qt.annotate_engine.drop_cache_budget_entry(record_key) → bool

Evict one decoded array selected by the global memory policy.

Parameters:

record_key – (kind, key) pair from cache_budget_entries(); kind is "mask", "edge" or "model" (which releases the cached Cellpose outline model instead of an array).

Returns:

whether anything was evicted.

spacr.qt.annotate_engine.empty_object_filters() → Dict[str, Tuple[float | None, float | None]]

Return all object-filter bounds in their disabled state.

spacr.qt.annotate_engine.ensure_annotation_column(db_path: str, column: str, *, table: str = DEFAULT_PNG_TABLE) → None

Add column INTEGER to table if missing and index png_path.

Parameters:

• db_path – path to the SQLite measurement database; nothing happens if the file does not exist.

• column – name of the annotation column to create; an empty name does nothing.

spacr.qt.annotate_engine.fetch_filtered_paths(db_path: str, annotation_column: str, measurements: List[str], thresholds: List[float], directions: List[str], image_type: str | None = None, *, table: str = DEFAULT_PNG_TABLE) → List[Tuple[str, int | None]]

Return ALL (png_path, annotation) rows matching every one of the measurement/threshold/direction triples.

Rows come from a merge of png_list with the measurement tables (via spacr.io._read_and_join_tables) — same code path as the Tk app — filtered with the same bound SQL image-type expression as normal browsing. NOT, AND, OR, parentheses and SQLite LIKE semantics apply equally when measurement thresholds are enabled. Invalid expressions raise ValueError. Callers paginate the returned list themselves.

Parameters:

• db_path – path to the SQLite measurement database; a missing file gives an empty list.

• annotation_column – annotation column returned beside each path; filled with None if the joined tables lack it.

• measurements – measurement column names to threshold; an empty list gives an empty result. Columns the tables lack are skipped.

• thresholds – one cut per measurement, or a single value applied to all of them; an empty list gives an empty result.

• directions – one direction per measurement, or a single string or one-item list applied to all; "higher" keeps rows above the cut, "lower" keeps rows below it, and any other value applies no cut. Mismatched lengths raise ValueError.

spacr.qt.annotate_engine.fetch_page(db_path: str, annotation_column: str, offset: int, page_size: int, image_type: str | None = None, *, table: str = DEFAULT_PNG_TABLE) → List[Tuple[str, int | None]]

Read one page of (png_path, annotation) rows in insertion order.

Parameters:

• db_path – path to the SQLite measurement database; a missing file gives an empty list.

• annotation_column – annotation column read beside png_path.

• offset – number of matching rows to skip (SQL OFFSET).

• page_size – maximum number of rows to return (SQL LIMIT).

spacr.qt.annotate_engine.filter_bound(value) → float | None

Parse a filter bound, returning None for empty or invalid input.

Parameters:

value – user-entered bound, typically a number or numeric string; None, a blank string, anything float() rejects and NaN all give None.

spacr.qt.annotate_engine.filter_channels_pil(img: PIL.Image.Image, channels: Iterable[str] | None = None) → PIL.Image.Image

Zero out channels not present in channels (e.g. [‘r’,’g’]).

Parameters:

img – RGB PIL image; it must split into exactly three bands.

spacr.qt.annotate_engine.filter_key(channel: str, measure: str) → str

Return the settings key for a channel and measurement pair.

Parameters:

• channel – channel name; stripped and lower-cased to form the prefix of "<channel>_<measure>".

• measure – measurement name; stripped and lower-cased to form the suffix.

spacr.qt.annotate_engine.find_last_annotated_offset(db_path: str, annotation_column: str, page_size: int, image_type: str | None = None, *, table: str = DEFAULT_PNG_TABLE) → int | None

Return the page-aligned offset of the last annotated row, or None.

Parameters:

• db_path – path to the SQLite measurement database; a missing file gives None.

• annotation_column – annotation column scanned; any value other than NULL or 0 counts as annotated.

• page_size – rows per page; the last annotated row’s index is rounded down to a multiple of it.

spacr.qt.annotate_engine.forget_outline_masks() → None

Drop every cached mask. For tests, and for a caller changing plates.

spacr.qt.annotate_engine.gate_paths(db_path: str, gates: Sequence[Any], *, table: str = DEFAULT_PNG_TABLE) → List[str]

png_paths surviving a chain of spacr.qt.widgets.gate_spec.Gate.

The route the Gate Editor was missing. The gate maths is NOT reproduced here – GateClause evaluates the chain, exactly as it does when the same gates filter a plot, so a population gated on screen and a population annotated from it are the same population by construction.

Parameters:

• db_path – the measurements database.

• gates – the gate chain, outermost first.

Returns:

matching png_path strings.

spacr.qt.annotate_engine.label_to_hex(val: int | None, dark: bool = True) → str | None

Map an annotation value to a hex border color.

None / 0 / non-int -> None (no border). 1 -> blue, 2 -> red, 3+ -> golden-ratio hue rotation.

THE COLOURS DEPEND ON THE THEME, because contrast does. The original palette was tuned against a dark tile and measured 3.5-12.0 against #1e1e1e – all comfortable. Against a light tile (#f5f5f5) the SAME colours measure 1.28-4.34, with five of the first six below the 3.0 readability floor:

class 1 #3ea6ff 6.43 on dark 2.38 on light class 4 #55f2d8 11.97 on dark 1.28 on light

That is issue #6 – “labels do not appear with good contrast in the annotation app like they do on Linux machines” – and it is a theme difference rather than a platform one: macOS defaults to the light appearance far more often.

HUE IS PRESERVED so a class keeps its identity across themes; only saturation and value move, deepening the colour until it reads against a pale background.

Parameters:

• val – the annotation value.

• dark – True for the dark theme’s palette, False for the light theme’s deepened one.

spacr.qt.annotate_engine.load_crop_image(path: str, db_path: str | None = None, stored_channel_order: str = 'auto', display_order: str = 'rgb', display_primaries: str = 'rgb') → PIL.Image.Image

Open one object crop PNG as an 8-bit RGB image in display order.

spacr.crops.read_crop_png() resolves the stored format from the sidecar marker, database, or legacy fallback before applying the requested display order. Sixteen-bit single-channel images are narrowed consistently instead of being clipped by an RGB conversion.

Two different questions are kept separate rather than combined into one control:

stored_channel_order Physical channel order in the file, resolved

from its sidecar marker or database. 'auto' is recommended when metadata is available.

display_order Preferred on-screen order, independent of file

storage. Defaults to 'rgb'.

Parameters:

• path – the crop PNG.

• db_path – optional measurements.db, consulted when the crop folder carries no sidecar marker.

• display_order – one of spacr.crops.DISPLAY_ORDERS. Applied AFTER the format is corrected, so the two never fight.

Returns:

PIL Image in RGB mode.

spacr.qt.annotate_engine.metadata_values(db_path: str, column: str, *, table: str = DEFAULT_PNG_TABLE) → List[str]

The distinct values of one png_list metadata column, sorted.

Read from the database rather than guessed from a naming convention: plates are named by whoever ran them, and a picker offering rows A-H to someone whose plate is numbered is a picker they cannot use.

Parameters:

• db_path – the measurements database.

• column – one of METADATA_COLUMNS.

Returns:

the distinct values, as strings, sorted; empty when the column or the database is missing.

Raises:

ValueError – a column outside METADATA_COLUMNS, which would otherwise interpolate an arbitrary name into SQL.

spacr.qt.annotate_engine.normalize_object_filters(object_filters: Mapping | None = None, object_size=None) → Dict[str, Tuple[float | None, float | None]]

Normalise object-filter bounds and migrate legacy size limits.

Legacy object_size limits are applied to the area filter for each colour channel, with non-positive legacy limits treated as disabled. Explicit object_filters values take precedence; invalid or empty explicit bounds are disabled, while zero remains a valid explicit bound.

Parameters:

• object_filters – {'r_area': (min, max), ...}; partial maps are accepted and unknown keys are ignored.

• object_size – Legacy (min, max) area limits in pixels.

Returns:

A new dictionary containing every supported key and a pair of floats or None.

spacr.qt.annotate_engine.normalize_pil(img: PIL.Image.Image, percentiles: Tuple[float, float] = (1.0, 99.0), normalize_channels: Iterable[str] | None = None) → PIL.Image.Image

Normalize the given PIL image per-channel using percentile stretch.

If normalize_channels is None or empty, the image is returned unchanged (aside from clipping to 8-bit range).

Parameters:

img – grayscale or RGB PIL image; pixel values are clipped to 0-255 before any stretch.

spacr.qt.annotate_engine.outline_image(base_img: PIL.Image.Image, full_img: PIL.Image.Image, outline_channels: Iterable[str] | None = None, edge_sigma: float = 1.0, edge_thickness: float = 1.0, edge_transparency: float = 100.0, edge_image: bool = False, outline_threshold_factor: float = 1.0, object_size: Tuple[int, int] = (0, 0), outline_method: str = 'otsu', object_filters: Mapping | None = None, should_stop=None) → PIL.Image.Image

Overlay per-channel object outlines on base_img.

Mirrors AnnotateApp.outline_image (Tk) semantics: for every channel in outline_channels, compute an Otsu-thresholded foreground mask on the corresponding channel of full_img, extract the boundary, optionally dilate it, then alpha-blend it over the channel in base_img with edge_transparency/100 opacity. Peak-normalized so thin edges stay visible.

WHICH objects get an outline is decided per plane by object_filters – an area window and a mean-intensity window for each of red, green and blue. object_size is the one-window-for-every-plane setting those replaced and is still honoured: it is migrated onto the three area rows by normalize_object_filters(), so a caller that passes only it gets exactly what it always got.

Parameters:

• base_img – RGB display image to receive the blended outlines. Its current channel filtering is preserved except where an outlined channel is deliberately blanked in outline-only mode.

• full_img – unfiltered RGB image supplying the channel intensities used to detect objects, aligned pixel-for-pixel with base_img.

• should_stop – optional callable asked before each channel’s Cellpose model construction and forward pass. When it answers True the work is abandoned by raising OutlineCancelled rather than finishing a page nobody is waiting for; 'otsu' outlines are fast enough that they are never interrupted mid-channel.

spacr.qt.annotate_engine.parse_image_type(expression: str | None) → Tuple[str, List[str]]

Turn an image-type expression into a SQL fragment and its parameters.

The filter used to be one substring matched with LIKE %x%, which can only ever say what a path MUST contain. There was no way to ask for the complement – “the cells with no pathogen crop” – which is half of most comparisons (issue #7).

The grammar is small and deliberately close to what someone would type:

pathogen contains “pathogen” !pathogen does NOT contain it NOT pathogen the same, spelled out cell AND nucleus contains both cell OR nucleus contains either cell AND NOT pathogen mixes them

AND binds tighter than OR, as everywhere else. Terms are matched case-insensitively, since LIKE is already case-insensitive for ASCII in SQLite and a user typing “Pathogen” means the same thing.

EVERY TERM IS A BOUND PARAMETER. Nothing the user types is interpolated into SQL, so a path fragment containing a quote is a path fragment and not an injection.

Parameters:

expression – the user’s filter, or None/empty for “no filter”.

Returns:

(sql, params) where sql is a bracketed boolean expression over png_path, or ("", []) when there is nothing to filter.

Raises:

ValueError – on an expression that cannot be read, naming what was wrong – an empty NOT, a dangling operator, unbalanced parentheses.

spacr.qt.annotate_engine.paths_by_measurements(db_path: str, annotation_column: str, rules: Sequence[Mapping[str, Any]]) → List[str]

png_paths satisfying EVERY {column, threshold, direction} rule.

Several measurements at once is the point: one threshold is a gate, not a population. The rules are ANDed, which is what fetch_filtered_paths() already does for the settings-panel filter – reused here rather than re-derived, so the auto-annotator and the filter can never disagree about what a threshold means.

Parameters:

• db_path – the measurements database.

• annotation_column – the column being written (needed by the join).

• rules – mappings with column, threshold and direction ('higher' or 'lower').

Returns:

matching png_path strings.

Raises:

ValueError – a rule missing a field, or an unknown direction.

spacr.qt.annotate_engine.paths_by_metadata(db_path: str, column: str, values: Sequence[str], *, table: str = DEFAULT_PNG_TABLE) → List[str]

png_paths whose column is one of values.

Parameters:

• db_path – the measurements database.

• column – one of METADATA_COLUMNS.

• values – the values to select.

Returns:

matching png_path strings.

Raises:

ValueError – a column outside METADATA_COLUMNS.