spacr.batch

Workflow inputs and outputs

Batch Runner

Validate and execute module jobs in dependency order. Each job writes its own module outputs and log.

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.

spacr.batch — stack arbitrary module+settings jobs into one queue and run them unattended.

The Plate Queue (spacr.qt.plate_queue) chains plates through one pipeline with one settings dict.

This module is the other axis. It holds a queue of arbitrary (module, settings) jobs in any order: Mask → Measure → Classify (CV) → Classify (ML), then the same four again with a different diameter, then a fifth plate’s Mask.

A queued job is a spacr-run invocation. Nothing here re-implements module dispatch or settings loading: spacr.cli owns the module registry, the settings loader, the --set coercion and the exit-code contract (0 ok, 1 the module raised, 2 bad arguments or settings), and this module drives it. The queue’s own result is a spacr.errors.RunLedger, stamped next to the queue file, and each job’s spacr.errors.read_run_status() stamp is read back so a job that exited 0 having silently skipped 40 fields is reported as partial rather than as a success.

Behaviour for unattended runs:

• Every job is validated when it is added, not when it runs. validate_queue() reports all problems at once and run_queue() refuses to start while any of them is an error.

• State is persisted after every transition, atomically. A machine that reboots mid-queue is resumed with resume_queue(), not restarted. The file is written to a temp name and os.replaced, so a crash cannot leave a truncated queue file.

• A job whose dependency failed is skipped, not run. Measure after a failed Mask produces a database that looks like a real result. skipped (deliberately not run), failed and not_run (the queue halted before reaching it) are three different things and stay that way.

• Repeated failures halt the queue. Under continue-on-error, max_consecutive_failures (3 by default) halts the queue after that many failures in a row and records the reason.

• Per-job logs go to their own file — one interleaved log from twelve overnight jobs is unreadable — and the path is on the job record.

Concurrency: jobs run strictly one at a time. They compete for one GPU, and two Cellpose jobs on one card can run out of CUDA memory. There is no max_workers option; to use two cards, run two queues in two processes.

Import cost: this module imports only the standard library plus spacr.cli, spacr.errors and spacr.validate, all of which are torch-free. A queue file can be read, validated and planned on a login node without a GPU stack — the pipeline itself is only imported inside the subprocess that runs a job.

Typical use:

from spacr.batch import Job, Queue, run_queue, save_queue

q = Queue(name='overnight')
q.add(Job(module='mask',    settings='/data/p1/settings/mask.csv'))
q.add(Job(module='measure', settings='/data/p1/settings/measure.csv',
          depends_on=['mask-1']))
save_queue(q, '/data/overnight.queue.json')
result = run_queue(q, path='/data/overnight.queue.json')
print(result.summary())

Exceptions

QueueError

The queue itself is wrong — a bad job, a cycle, a refused start.

Classes

Job	One spacr-run invocation, with its place in the queue's history.
Problem	One thing wrong with a job or with the queue.
Progress	One incremental progress report, handed to on_progress.
Queue	An ordered list of Jobs, run one at a time, top to bottom.
QueueResult	Outcome of a queue run: job states, timings and ledger.

Functions

classify_failure(→ Tuple[str, str])	Turn an exit code plus a log into (kind, one-line message).
fmt_duration(→ str)	Format a duration in seconds as a short string.
format_problems(→ str)	Render every problem at once, errors first.
inprocess_runner(→ int)	Run one job in this interpreter, with its output tee'd to its log.
job_command(→ List[str])	Return the exact spacr-run command line for job.
load_queue(→ Queue)	Read a queue file written by save_queue(), or hand-written.
plan(→ str)	Describe what the queue would do, without doing any of it.
resolve_job_settings(→ Dict[str, Any])	Build the settings dict this job's module will actually receive.
resume_queue(→ QueueResult)	Pick a queue up where a crash, a reboot or a Stop left it.
run_queue(→ QueueResult)	Run the queue, one job at a time, and report on all of it.
save_queue(→ pathlib.Path)	Write queue to path atomically, as indented JSON.
subprocess_runner(→ int)	Run one job as its own spacr-run process. The default runner.
validate_job(→ List[Problem])	Check one job the way spacr-run --dry-run would, plus queue rules.
validate_queue(→ List[Problem])	Validate every job, and the queue's own structure, all at once.

Module Contents

exception spacr.batch.QueueError

Bases: spacr.errors.SpacrError

The queue itself is wrong — a bad job, a cycle, a refused start.

A subclass of spacr.errors.SpacrError so except SpacrError catches it alongside everything else spaCR raises deliberately.

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

class spacr.batch.Job

One spacr-run invocation, with its place in the queue’s history.

Parameters:

• module – module key or alias understood by spacr.cli.resolve_module() — 'mask', 'measure', 'ml_analyze', …

• settings – path to a settings CSV/JSON (what spacr-run --settings takes), or an inline settings dict for a job that has no file of its own.

• id – unique, human-typable identifier. Left empty, Queue.add() assigns one like 'mask-1'.

• label – what to show a human; defaults to module @ src.

• overrides – key=value strings applied on top of settings, exactly like spacr-run --set. This is how “the same four jobs again with a different diameter” is written without copying four settings files. A mapping is accepted and normalised.

• depends_on – ids of jobs that must succeed first. A job whose dependency failed is skipped, never run.

• status – one of ALL_STATUSES.

• started – ISO-8601 UTC start time, or ''.

• finished – ISO-8601 UTC end time, or ''.

• exit_code – the process exit code — 0 ok, 1 the module raised, 2 bad settings (spacr.cli’s contract).

• error – one-line explanation of a failure or a skip.

• log_path – this job’s own log file.

• run_status – the job’s spacr.errors.read_run_status() verdict, summarised — None when the job stamped nothing.

copy(**changes: Any) → Job

Return a fresh, never-run copy of this job with changes applied.

Used by the GUI’s “Duplicate” button.

default_label() → str

A readable label derived from the module and its src.

classmethod from_dict(data: Mapping[str, Any]) → Job

Rebuild a job from to_dict(), tolerating a hand-edited file.

Every field has a default and unknown keys are ignored, so a user can delete the bookkeeping (status, started, …) from a queue file and still have it load.

Parameters:

data – one entry of the queue file’s jobs list.

Raises:

QueueError – when data is not a mapping with a module.

reset() → Job

Clear the bookkeeping so the job can run again. Returns self.

to_dict() → Dict[str, Any]

Return the job as a JSON-serialisable dict, in a stable key order.

property duration_s: float | None

Wall-clock seconds the job took, or None if it never finished.

property elapsed_s: float | None

Seconds the job has taken, counting up while it is still running.

Returns:

duration_s once it has finished, the time since it started while it is running, else None.

property is_partial: bool

True when the job exited 0 but its own ledger says items failed.

Example: a Measure run that processed 344 of 384 wells exits 0.

property override_args: List[str]

overrides normalised to the key=value strings --set takes.

property settings_path: str

The settings file path, or '' when the job carries a dict.

class spacr.batch.Problem

One thing wrong with a job or with the queue.

Deliberately the same shape as spacr.validate.Problem — severity, message, fix — plus the job_id needed to say which of twelve jobs is at fault. Problems produced by spacr.validate.validate_settings() are wrapped into this, not re-invented.

Parameters:

• job_id – id of the offending job; '' for a queue-level problem.

• severity – 'error' (the queue must not start) or 'warning'.

• message – what is wrong, in the user’s terms.

• fix – the corrective action.

• setting – the settings key at fault, when there is one.

__str__() → str

Render the problem message followed by its corrective action.

property is_error: bool

True when this problem must stop the queue from starting.

class spacr.batch.Progress

One incremental progress report, handed to on_progress.

Every transition emits one of these, so a GUI can show progress while the queue runs.

Parameters:

• event – 'queue_started', 'job_started', 'job_finished', 'job_skipped', 'queue_stopped' or 'queue_finished'.

• job_id – the job this is about, or '' for queue-level events.

• index – 1-based position of the job in the queue, 0 when N/A.

• total – number of jobs in the queue.

• status – the job’s status at the moment of the event.

• message – one line fit to put in a status bar.

class spacr.batch.Queue

An ordered list of Jobs, run one at a time, top to bottom.

Parameters:

• jobs – the jobs, in the order they will run.

• created – ISO-8601 UTC creation time.

• name – what to call this queue in the summary and log folder.

__iter__()

Iterate over the jobs in queue order.

__len__() → int

Return the number of jobs currently in the queue.

add(job: Job, validate: bool = True) → Job

Validate job and append it to the queue.

Parameters:

• job – the job to add; its id and label are filled in when empty.

• validate – set False only when deliberately building an invalid queue (loading a hand-edited file, for instance, which reports its problems through validate_queue() instead of raising).

Returns:

the job, now owned by this queue.

Raises:

QueueError – when the job cannot run — an unknown or GUI-only module, an unreadable settings file, a bad override, a duplicate id, or a dependency that is not already in the queue.

counts() → OrderedDict[str, int]

Jobs per status, in ALL_STATUSES order.

find(job_id: str) → Job | None

Return the job with job_id, or None.

Parameters:

job_id – unique queue job identifier to find.

classmethod from_dict(data: Mapping[str, Any]) → Queue

Rebuild a queue from to_dict() or from a hand-written file.

Parameters:

data – serialized or hand-written queue mapping to rebuild.

Raises:

QueueError – when the document is not a queue, is a format from the future, or holds a job entry that cannot be read.

index(job_id: str) → int

Position of job_id in run order, or -1.

Parameters:

job_id – unique queue job identifier to locate.

mint_id(module: str) → str

Return an unused, human-typable id for a job of module.

Parameters:

module – module name used as the identifier’s readable stem.

move(job_id: str, offset: int) → int

Move job_id offset places (negative is earlier).

Parameters:

• job_id – unique identifier of the job to reposition.

• offset – relative number of queue positions to move.

Returns:

the job’s new index, or -1 when it is not in the queue.

remove(job_id: str) → bool

Remove job_id and drop it from every other job’s depends_on.

Parameters:

job_id – unique identifier of the job to remove.

Leaving a dangling dependency behind would silently skip the jobs that referred to it, so the reference is cleaned up here.

Returns:

True when a job was removed.

reset() → Queue

Clear every job’s bookkeeping so the whole queue runs again.

to_dict() → Dict[str, Any]

Return the whole queue as a JSON-serialisable dict.

property ids: List[str]

Every job id, in run order.

class spacr.batch.QueueResult

Outcome of a queue run: job states, timings and ledger.

Parameters:

• queue – the queue, with every job’s final status on it.

• log_dir – folder holding the per-job logs.

• started – ISO-8601 UTC start of the whole queue.

• finished – ISO-8601 UTC end of the whole queue.

• stopped_reason – why the queue halted early, or ''.

• ledger – the queue’s own spacr.errors.RunLedger — one recorded item per job, which is what groups identical failures and what gets stamped next to the queue file.

• path – the queue file that was kept up to date, or ''.

jobs_with(status: str) → List[Job]

Every job that ended in status.

Parameters:

status – job status value to select.

summary() → str

Render the end-of-queue report.

What ran, what failed and why (identical failures grouped), what was skipped and because of which upstream job, what is only partial, and how long each took.

property duration_s: float | None

Wall-clock seconds the queue took.

property failed: List[Job]

Jobs that ran and failed.

property not_run: List[Job]

Jobs the queue halted before reaching.

property ok: bool

True when everything ran, nothing failed and nothing is partial.

property partial: List[Job]

Jobs that exited 0 but whose own ledger recorded failed items.

property skipped: List[Job]

Jobs deliberately not run because an upstream job did not succeed.

property succeeded: List[Job]

Jobs that exited 0 — including the ones that are only partial.

spacr.batch.classify_failure(exit_code: int, log_path: str) → Tuple[str, str]

Turn an exit code plus a log into (kind, one-line message).

The kind is what groups identical failures: twelve jobs that all died on the same missing share are one problem worth one line, not twelve.

Parameters:

• exit_code – the runner’s exit code.

• log_path – the job’s log file.

Returns:

(kind, message).

spacr.batch.fmt_duration(seconds: float | None) → str

Format a duration in seconds as a short string.

Parameters:

seconds – elapsed seconds, or None when the job never finished.

Returns:

'—', '42.1s', '7m 12s' or '7h 41m'.

spacr.batch.format_problems(problems: Sequence[Problem], title: str = 'queue check') → str

Render every problem at once, errors first.

All problems are listed, not only the first.

Parameters:

• problems – what validate_queue() returned.

• title – heading for the block.

Returns:

the report as one string, no trailing newline.

spacr.batch.inprocess_runner(job: Job, settings_path: str, log_path: str) → int

Run one job in this interpreter, with its output tee’d to its log.

Parameters:

• job – queue job whose module and overrides are executed.

• settings_path – resolved settings file supplied to the job command.

• log_path – destination file for redirected output.

Same argv, same exit-code contract as subprocess_runner() — it calls spacr.cli.main() directly — but with none of the isolation: a segfault here takes the queue with it. Use it only where spawning is not possible (a frozen single-file build), and never from the GUI.

Returns:

the exit code spacr.cli.main() returned.

spacr.batch.job_command(job: Job, settings_path: str, python: str | None = None) → List[str]

Return the exact spacr-run command line for job.

Spelled as <this python> -m spacr.cli ... rather than the spacr-run console script so the job runs in the interpreter the queue is running in — the environment spaCR is installed in — even when PATH says otherwise.

Parameters:

• job – the job.

• settings_path – settings file the job will be given.

• python – interpreter to use; sys.executable by default.

Returns:

the argv list.

spacr.batch.load_queue(path: str | os.PathLike) → Queue

Read a queue file written by save_queue(), or hand-written.

Parameters:

path – the queue file.

Returns:

the Queue.

Raises:

QueueError – when the file is missing or is not a queue; the message is one sentence.

spacr.batch.plan(queue: Queue, detail: bool = False) → str

Describe what the queue would do, without doing any of it.

Parameters:

• queue – the queue.

• detail – also render spacr.validate.describe_plan() for every job — the full per-job plan. Off by default because it lists directories, which is slow over NFS.

Returns:

the plan as one string.

spacr.batch.resolve_job_settings(job: Job) → Dict[str, Any]

Build the settings dict this job’s module will actually receive.

Layered exactly as spacr-run layers them — module defaults, then the settings file (or the inline dict) read under today’s setting names, then the --set overrides — using spacr.cli’s own loader, migration and coercion so a value the CLI accepts is a value the queue accepts.

Parameters:

job – the job.

Returns:

the resolved settings.

Raises:

SettingsError – on an unknown module, an unreadable settings file, an unknown override key or an uncoercible override value.

spacr.batch.resume_queue(path: str | os.PathLike, retry_failed: bool = False, **kwargs: Any) → QueueResult

Pick a queue up where a crash, a reboot or a Stop left it.

Jobs that already succeeded are left alone. Jobs that were not_run (the queue halted before reaching them) or still pending are run. A job that was running when the machine went down is reset and run again. For Mask, Measure, and Format Converter jobs the rerun receives resume=True automatically, so their verified field checkpoints are reused instead of repeating the whole plate. Failed and skipped jobs stay as they are unless retry_failed is set.

Parameters:

• path – the queue file run_queue() was persisting to.

• retry_failed – also re-run jobs that failed, and un-skip the jobs that were skipped because of them.

• kwargs – forwarded to run_queue().

Returns:

the QueueResult for the resumed run.

spacr.batch.run_queue(queue: Queue, path: str | os.PathLike | None = None, on_error: str = 'continue', log_dir: str | os.PathLike | None = None, max_consecutive_failures: int | None = 3, on_progress: Callable[[Progress], None] | None = None, runner: Callable[[Job, str, str], int] | None = None, stop_flag: Callable[[], bool] | None = None, force: bool = False, echo: bool = True) → QueueResult

Run the queue, one job at a time, and report on all of it.

Jobs run sequentially and only sequentially — they compete for one GPU. The queue is validated in full before the first job starts, its state is written to path after every transition, and a job whose dependency did not succeed is skipped rather than run against a half-written input.

Parameters:

• queue – the queue to run. Mutated in place: each job’s status, timestamps, exit code, log path and run_status are filled in.

• path – queue file kept up to date after every transition. Without it the run cannot be resumed.

• on_error – 'continue' (default) runs the jobs that do not depend on the failed one; 'stop' halts the queue immediately.

• log_dir – folder for per-job logs; defaults to <queue file>_logs or a temp folder.

• max_consecutive_failures – halt after this many failures in a row. None or 0 disables the check.

• on_progress – called with a Progress on every transition, so a GUI can show the queue moving. Exceptions from it are logged and swallowed.

• runner – (job, settings_path, log_path) -> exit_code; subprocess_runner() by default. Injected by the tests, and by anyone who wants to submit jobs to a scheduler instead.

• stop_flag – polled between jobs; return True to stop the queue after the running job finishes (the GUI’s Stop button).

• force – run even though validation found errors. Use it only when a validation error does not apply to your data.

• echo – print the summary at the end. Always logged either way.

Returns:

the QueueResult.

Raises:

QueueError – when validation found errors and force is False — with every problem in the message, not just the first.

spacr.batch.save_queue(queue: Queue, path: str | os.PathLike) → pathlib.Path

Write queue to path atomically, as indented JSON.

JSON rather than a bespoke format because it is stdlib (no PyYAML on a compute node), because spacr.errors already persists its stamps this way, and because indented JSON can be edited in a text editor.

Parameters:

• queue – the queue to persist.

• path – destination file.

Returns:

the written path.

spacr.batch.subprocess_runner(job: Job, settings_path: str, log_path: str) → int

Run one job as its own spacr-run process. The default runner.

A separate process isolates failures: cellpose segfaulting or the CUDA driver wedging kills that job, not the other eleven and not the GUI. The exit code comes straight from spacr.cli — 0 ok, 1 the module raised, 2 bad settings — and stdout+stderr go to this job’s own log file.

Parameters:

• job – the job to run.

• settings_path – settings file to pass to --settings.

• log_path – file to write this job’s output to.

Returns:

the process exit code.

spacr.batch.validate_job(job: Job, queue: Queue | None = None) → List[Problem]

Check one job the way spacr-run --dry-run would, plus queue rules.

Every check that can be made without running anything is made here, so it is made when the job is added:

• the module resolves, and is not one of the GUI-only apps (spacr.cli.INTERACTIVE_ONLY) that has no headless callable;

• the settings file exists, parses, and the overrides name real settings with coercible values;

• spacr.validate.validate_settings() agrees the settings are runnable against the data on disk — with data that an upstream job in this queue has not produced yet deferred to a warning (see _deferrable());

• dependencies exist and come earlier in the queue.

Parameters:

• job – the job to check.

• queue – the queue it belongs to, needed for the dependency rules.

Returns:

every problem found, errors and warnings mixed. Never raises.

spacr.batch.validate_queue(queue: Queue) → List[Problem]

Validate every job, and the queue’s own structure, all at once.

All problems are returned at once: duplicate ids, cycles, unknown dependencies, and each job’s own settings problems.

Parameters:

queue – the queue to check.

Returns:

every problem found. [p for p in problems if p.is_error] being empty is exactly the condition run_queue() requires.

Nested helpers

_cycle_problems.walk(job_id: str, trail: List[str]) → None

Depth-first search one captured dependency subgraph for cycles.

Parameters:

• job_id – queue job whose known dependencies should be visited.

• trail – active ancestor IDs used to reconstruct a back-edge cycle.

Returns:

None. Captured state records active and completed jobs, and a back edge appends one problem; missing dependency IDs are ignored here because ordinary dependency validation reports those typos.

spacr/batch.py:959

run_queue.persist() → None

Write the queue after a transition. Never fatal.

spacr/batch.py:1620