spacr.qt.plate_queue

Workflow inputs and outputs

Plate Queue

Run the chosen pipeline across plates. Outputs are those of each queued module; retain each plate and its settings separately.

Open: the application’s Help/tools menus.

Inputs and outputs below include conditional alternatives. The guidance and handoff notes say which route applies.

Inputs

• Run queue — Saved module/plate/settings job definitions and dependency order.

Outputs

• Run history and artifacts — Project run records, settings, output paths, artifact provenance, status and logs.

API reference.

Module tutorial.

Plate queue — sequential execution of many pipelines.

Users often have 5–20 plates to segment, measure, or classify with the same settings. Running each one manually from the Mask app is fine for one plate and painful for twenty. This module lets them:

1. Enqueue a plate as (app_key, settings) (or import a batch of plates from a CSV).

2. Run the queue in the background — one item at a time — and see per-item status update live.

3. Pause between items, or stop cold.

4. Have every completed item show up in the run-journal history like a normal invocation, so nothing about downstream tooling changes.

The queue itself is a plain Python data structure — the Qt screen in spacr.qt.screens.queue renders it. Keeping the logic separate makes it unit-testable without a display.

Persistence: the queue serialises to ~/.spacr/queue.json on every mutation so a crash or restart doesn’t lose the plan.

Classes

PlateQueue	Ordered list of QueueItem objects with atomic on-disk snapshots.
QueueItem	One plate to process.
Status	Per-item lifecycle. Mirrors the run journal's terminology.

Functions

default_runner(→ None)	Execute item synchronously via the resolved pipeline entry
import_plates_from_csv(→ List[QueueItem])	Parse a CSV of plates into QueueItem objects.
run_queue(→ None)	Run every QUEUED item in queue sequentially.

Module Contents

class spacr.qt.plate_queue.PlateQueue(path: pathlib.Path | None = None)

Ordered list of QueueItem objects with atomic on-disk snapshots.

The queue is thread-agnostic — the Qt screen owns exclusive access. If two callers ever need to touch it concurrently, wrap each mutation in a lock at the call site.

Parameters:

path – where the snapshot is written. None uses spaCR’s own queue file, which is the ordinary case; a test passes a temporary path so it does not disturb the user’s real queue.

Open the queue, loading whatever is already on disk.

Parameters:

path – where the queue is stored; None uses the default location, so the queue survives a restart.

__iter__()

Iterate the queued items in order.

__len__() → int

Return the number of queued items.

add(item: QueueItem) → None

Append a plate and save immediately.

SAVED ON EVERY CHANGE, not on close: the queue is shared with other screens and read from disk, so an unsaved change is one another screen cannot see.

Parameters:

item – the plate to queue.

clear() → int

Remove EVERY item whatever its status; return count removed.

The counterpart to clear_finished(), which keeps what is still waiting. This one does not, so a RUNNING item goes too – and that is the whole reason to say so here: dropping the record does NOT stop the run. The worker holds its own settings and keeps going; what disappears is the queue’s knowledge of it, so its completion is never written back.

Only reachable from Clear on Home’s Queued panel, where the queue being wrong is what the user is trying to fix. A caller that wants to leave a live run alone wants clear_finished().

clear_finished() → int

Remove SUCCESS/FAILED/SKIPPED items; return count removed.

find(item_id: str) → QueueItem | None

Return the item with item_id or None.

Parameters:

item_id – the QueueItem.id to look for.

is_all_done() → bool

True iff no item is in QUEUED or RUNNING.

items() → List[QueueItem]

Return a shallow copy of the current items list.

load() → None

Read the queue from disk, replacing what is held.

next_queued() → QueueItem | None

Return the first Status.QUEUED item, or None.

remove(item_id: str) → bool

Drop one plate by id and save.

Parameters:

item_id – the plate’s id.

Returns:

True when it was there to remove.

save() → None

Write the queue to disk, unless it was opened read-only.

update(item_id: str, **fields) → None

Patch fields on the item with item_id. Saves on any change.

Parameters:

• item_id – the QueueItem.id of the item to patch; an unknown id is ignored.

• fields – attribute values to set on the item. Names the item does not have are ignored.

class spacr.qt.plate_queue.QueueItem

One plate to process.

Parameters:

• id – item identifier; build() mints eight hex characters.

• app_key – key of the app whose pipeline runs this plate, resolved by spacr.qt.bridge.resolve_pipeline_entry().

• settings – settings dict handed to that pipeline.

• status – lifecycle state of the item.

• start_ts – epoch seconds when the run started, or None.

• end_ts – epoch seconds when the run finished, or None.

• error – error message of a failed run, or None.

• run_dir – run folder recorded for the item, or None; stored and reloaded with the queue.

• label – display label shown for the item.

classmethod build(app_key: str, settings: Dict[str, Any], label: str = '') → QueueItem

Factory that mints an ID + resolves a display label.

Parameters:

• app_key – key of the app whose pipeline runs this plate.

• settings – settings for the run; copied into the item. Its src value becomes the label when none is given.

property elapsed_s: float | None

Wall-clock seconds if the item has both timestamps.

class spacr.qt.plate_queue.Status

Bases: str, enum.Enum

Per-item lifecycle. Mirrors the run journal’s terminology.

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

spacr.qt.plate_queue.default_runner(item: QueueItem) → None

Execute item synchronously via the resolved pipeline entry point. Intended for CLI use or tests — the Qt screen uses a QThread wrapper instead so the UI stays responsive.

Parameters:

item – the queue item to run; its app_key selects the pipeline, which is called with its settings. An app with no pipeline raises RuntimeError.

spacr.qt.plate_queue.import_plates_from_csv(csv_path: Any, base_settings: Dict[str, Any], app_key: str = 'mask') → List[QueueItem]

Parse a CSV of plates into QueueItem objects.

The CSV must have a header row. Each remaining row is one plate. Columns other than src are merged over base_settings; src becomes the item’s src (and label). Rows missing src are skipped.

Parameters:

• csv_path – path to a CSV with at least a src column.

• base_settings – settings dict applied to every row before the row’s own overrides.

• app_key – pipeline id for every generated item.

spacr.qt.plate_queue.run_queue(queue: PlateQueue, runner: RunnerFn = default_runner, stop_on_error: bool = False) → None

Run every QUEUED item in queue sequentially.

Each item’s status transitions QUEUED → RUNNING → SUCCESS/FAILED. If stop_on_error is True, the first failure halts the loop with remaining items left as QUEUED.

Not called by the Qt screen (which needs threads + signals) but exposed as a plain function for CLI / tests / scripting.

Parameters:

queue – the queue to drain; its QUEUED items run one at a time and their status, timestamps and errors are written back to it.