Batch

optari.batch.plan

Resolve a batch preset plus a dataset folder into a concrete, checked analysis plan.

A batch preset names the per-step presets rather than copying their contents, so the presets the user tuned in the docks stay the single source of truth and the plan itself stays small enough to read, share and cite.

Everything here is headless: no Qt, no napari, no viewer. Building a plan and telling the user what is wrong with it must work before anything is loaded.

BatchJob dataclass

BatchJob(study_path, scan_path, scan_info)

One scan to process, and where its outputs belong.

key property

key

Identifier for this scan's outputs, unique across the dataset.

scan_name property

scan_name

The scan's internal name from its acquisition metadata, or empty if it has none.

scan_stem property

scan_stem

Scan_3, whether the scan is an HDF5 file or an iThera folder.

BatchPlan dataclass

BatchPlan(jobs, output_dir, preset=dict(), reconstruction=None, unmixing=None, segmentation=None, roi=None, roi_from_scan=False, source=None, frame='spectral_motion', measure=MeasureScope(), outputs=OutputSpec())

A dataset, a set of resolved step settings, and where the results go.

measures property

measures

Whether this plan measures ROIs at all.

report_path property

report_path

Where the run's per-scan status report is written (batch_report.xlsx).

roi_placement property

roi_placement

How the ROI preset asks to be placed. Owned by the preset, not the plan.

step_names property

step_names

The analysis steps this plan will run, in execution order.

table_path property

table_path

Where the run's ROI measurements are written (batch_roi_table.xlsx).

OutputSpec dataclass

OutputSpec(xlsx=True, hdf5=False, ipasc=False, overlay_png=True, overlay_layer=None)

Which outputs a batch run writes per scan: ROI table, HDF5/IPASC export, and overlay PNG.

build_plan

build_plan(*, root, batch_preset, output_dir)

Resolve batch_preset against the dataset at root.

Raises ValueError, message safe to show the user, when the preset cannot be resolved at all. Problems that need the assembled plan to spot are reported by :func:validate_plan instead, so the dialog can list them together.

Source code in src/optari/batch/plan.py

def build_plan(
    *, root: Path, batch_preset: dict, output_dir: Path
) -> BatchPlan:
    """Resolve *batch_preset* against the dataset at *root*.

    Raises ValueError, message safe to show the user, when the preset cannot be
    resolved at all. Problems that need the assembled plan to spot are reported by
    :func:`validate_plan` instead, so the dialog can list them together.
    """
    root = Path(root)
    steps = batch_preset.get("steps") or {}
    if not isinstance(steps, dict):
        raise ValueError("Batch preset 'steps' must be an object.")

    jobs = [
        BatchJob(study_path=study, scan_path=scan, scan_info=info)
        for study, scans in discover_studies(root).items()
        for scan, info in scans.items()
    ]

    reconstruction = unmixing = segmentation = None
    name = _step_preset_name(steps, "reconstruction")
    if name is not None:
        reconstruction = _load_named(
            PresetStore(get_user_reconstruction_presets_dir()),
            name,
            "reconstruction",
        )
    name = _step_preset_name(steps, "unmixing")
    if name is not None:
        unmixing = _load_named(
            PresetStore(get_user_unmixing_presets_dir()), name, "unmixing"
        )
    name = _step_preset_name(steps, "segmentation")
    if name is not None:
        segmentation = _load_named(
            PresetStore(get_user_segmentation_presets_dir()),
            name,
            "segmentation",
        )

    roi = None
    name = _step_preset_name(steps, "roi")
    roi_from_scan = name == ROI_FROM_SCAN
    if name is not None and not roi_from_scan:
        try:
            roi = RoiPresetStore(get_user_roi_presets_dir()).get(name)
        except (TypeError, ValueError, OSError) as exc:
            raise ValueError(
                f"Could not load ROI preset '{name}': {exc}"
            ) from exc

    outputs_spec = batch_preset.get("outputs") or {}
    return BatchPlan(
        jobs=jobs,
        output_dir=Path(output_dir),
        preset=batch_preset,
        reconstruction=reconstruction,
        unmixing=unmixing,
        segmentation=segmentation,
        roi=roi,
        roi_from_scan=roi_from_scan,
        source=(batch_preset.get("source") or None),
        frame=_frame_selector(batch_preset.get("frame")),
        # the scan's ROIs sit on their own frames, so steps and measurement need them all
        measure=_measure_scope(
            batch_preset.get("measure") or {},
            bool(batch_preset.get("all_frames", False)) or roi_from_scan,
        ),
        outputs=OutputSpec(
            xlsx=bool(outputs_spec.get("xlsx", True)),
            hdf5=bool(outputs_spec.get("hdf5", False)),
            ipasc=bool(outputs_spec.get("ipasc", False)),
            overlay_png=bool(outputs_spec.get("overlay_png", True)),
            overlay_layer=outputs_spec.get("overlay_layer") or None,
        ),
    )

describe_plan

describe_plan(plan)

The resolved settings, as shown in the batch window before the run.

Source code in src/optari/batch/plan.py

def describe_plan(plan: BatchPlan) -> str:
    """The resolved settings, as shown in the batch window before the run."""
    lines = [
        f"Scans:      {len(plan.jobs)} in {len({job.study_path for job in plan.jobs})} study folder(s)",
        f"Steps:      {' -> '.join(plan.step_names) or 'none (conversion only)'}",
        f"Frame:      {plan.frame}",
    ]
    if plan.reconstruction is not None:
        lines.append(
            "Recon:      "
            f"{plan.reconstruction.get('RECONSTRUCTION_ALGORITHM', 'unknown')} @ "
            f"{plan.reconstruction.get('RECONSTRUCTION_SPEED_OF_SOUND', '?')} m/s"
        )
    if plan.unmixing is not None:
        lines.append(
            f"Unmixing:   {', '.join(plan.unmixing.get('SPECTRA', []))}"
        )
    if plan.segmentation is not None:
        lines.append(f"Model:      {plan.segmentation.get('model_id', '?')}")
    if plan.roi is not None:
        lines.append(
            f"ROI:        {plan.roi.name} ({plan.roi_placement}, "
            f"tissue class '{plan.roi.geometry.tissue_class}')"
        )
    if plan.roi_from_scan:
        lines.append(
            "ROI:        the scan's own, each on the frames it is drawn on"
        )

    # Nothing is analysed or measured without an ROI, so saying how would be noise.
    if plan.measures:
        if plan.source:
            source = f"layer starting with '{plan.source}'"
        elif plan.reconstruction is not None:
            source = "the reconstruction produced by this plan"
        else:
            source = "the scan's default PA layer"
        lines.append(f"Analyse:    {source}")

        scope = plan.measure
        layers = (
            "every PA layer in the scan"
            if scope.all_layers
            else "that layer and what is unmixed from it"
        )
        channels = (
            "all channels"
            if scope.all_layers or scope.all_channels
            else "current wavelength (derived layers: all channels)"
        )
        lines.append(
            "Measure:    "
            f"{layers}, "
            f"{'all frames' if scope.all_frames else 'analysis frame'}, "
            f"{channels}"
        )
    wanted = [
        name
        for name, on in (
            ("xlsx", plan.outputs.xlsx),
            ("hdf5", plan.outputs.hdf5),
            ("ipasc", plan.outputs.ipasc),
            ("overlay png", plan.outputs.overlay_png),
        )
        if on
    ]
    lines.append(f"Outputs:    {', '.join(wanted) or 'none'}")
    lines.append(f"Folder:     {plan.output_dir}")
    return "\n".join(lines)

plan_warnings

plan_warnings(plan)

Things worth saying out loud that are not reasons to refuse the run.

Source code in src/optari/batch/plan.py

def plan_warnings(plan: BatchPlan) -> list[str]:
    """Things worth saying out loud that are not reasons to refuse the run."""
    if plan.source or plan.reconstruction is not None:
        return []
    if plan.unmixing is None and not plan.measures:
        return []
    return [
        "No reconstruction step and no 'source', so the scan's default PA layer will "
        "be analysed. Set 'source' to a layer-name prefix to choose deliberately."
    ]

validate_plan

validate_plan(plan)

Everything that would make this run fail or be pointless, reported together.

Only cross-checks that can be made before loading a scan live here. Per-scan failures (wavelengths not acquired, ROI outside the FOV, class absent from a frame) are the run's own business and are recorded in its report.

Source code in src/optari/batch/plan.py

def validate_plan(plan: BatchPlan) -> list[str]:
    """Everything that would make this run fail or be pointless, reported together.

    Only cross-checks that can be made before loading a scan live here. Per-scan
    failures (wavelengths not acquired, ROI outside the FOV, class absent from a
    frame) are the run's own business and are recorded in its report.
    """
    problems: list[str] = []

    if not plan.jobs:
        problems.append(
            "No scans found. Pick a folder holding Study_*/Scan_* data."
        )
    if not plan.step_names and not (plan.outputs.hdf5 or plan.outputs.ipasc):
        problems.append(
            "The plan has no analysis steps and no file export, so it would do nothing."
        )

    problems += _validate_reconstruction(plan)
    problems += _validate_segmentation(plan)
    problems += _validate_unmixing(plan)
    problems += _validate_output_dir(plan)
    return problems

optari.batch.runner

Drive a :class:BatchPlan over a dataset, one scan at a time.

The whole run is a single generator. Work that has to touch the viewer (loading a scan, placing an ROI, measuring, screenshots) happens inline on the main thread; each heavy stage is yielded as a :class:BackgroundStep that the runner hands to a worker thread and resumes from when it finishes. Writing it as a generator is what makes "log this scan's failure and carry on" an ordinary try/except around the per-scan body rather than a state machine.

Resuming happens on the worker's finished signal, not returned. superqt emits returned first and finished second, and the task helper clears the UI gate in its on_done; resuming any earlier would let the finishing worker tear down the progress bar and gate that the next step has already set up.

BatchCancelled

Bases: Exception

The user stopped the run. Nothing further should start.

BatchRunner

BatchRunner(controller, plan, *, on_progress=None, on_finished=None)

Runs a plan against the live viewer, reporting progress as it goes.

Set up the report, ROI table writer, and progress callbacks for plan.

Parameters:

Name	Type	Description	Default
controller	OptariController	OptariController driving the viewer this run operates on.	required
plan	BatchPlan	Resolved plan to execute.	required
on_progress	Callable[[BatchJob, str], None] | None	optional callback(job, status), called as each scan starts and ends.	None
on_finished	Callable[[], None] | None	optional callback(), called once the whole run has finished.	None

Source code in src/optari/batch/runner.py

def __init__(
    self,
    controller: OptariController,
    plan: BatchPlan,
    *,
    on_progress: Callable[[BatchJob, str], None] | None = None,
    on_finished: Callable[[], None] | None = None,
) -> None:
    """Set up the report, ROI table writer, and progress callbacks for *plan*.

    Args:
        controller: OptariController driving the viewer this run operates on.
        plan: Resolved plan to execute.
        on_progress: optional callback(job, status), called as each scan starts and ends.
        on_finished: optional callback(), called once the whole run has finished.
    """
    self.controller = controller
    self.viewer = controller.viewer
    self.plan = plan
    self.report = BatchReport(plan.report_path, plan_source=plan.preset)
    self.saved_table = SavedRoiTable(
        plan.table_path if plan.outputs.xlsx else None
    )
    self._on_progress = on_progress or (lambda job, status: None)
    self._on_finished = on_finished or (lambda: None)
    self._cancel_requested = False
    self._worker = None
    self._gen = None
    self._resources = ExitStack()

cancel

cancel()

Stop the run after the current step, and ask a running background worker to quit early.

Source code in src/optari/batch/runner.py

def cancel(self) -> None:
    """Stop the run after the current step, and ask a running background worker to quit early."""
    self._cancel_requested = True
    if self._worker is not None:
        self._worker.quit()

start

start()

Open the run's log file and begin the plan.

Source code in src/optari/batch/runner.py

def start(self) -> None:
    """Open the run's log file and begin the plan."""
    stamp = dt.datetime.now(dt.timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    log_path = self.plan.output_dir / f"batch_{stamp}.log"
    self._resources.enter_context(run_log_file(log_path))
    logger.info(
        "batch run starting: %s scan(s), steps: %s",
        len(self.plan.jobs),
        " -> ".join(self.plan.step_names) or "none",
    )
    self._gen = self._run()
    self._advance(("ok", None))

BatchStepError

BatchStepError(message, step='')

Bases: Exception

This scan cannot be processed. Record it and move on to the next one.

Source code in src/optari/batch/runner.py

def __init__(self, message: str, step: str = "") -> None:
    super().__init__(message)
    self.step = step

Tick dataclass

Tick()

Hand control back to the event loop, so Cancel stays responsive between scans.

optari.batch.report

Per-scan outcome record for a batch run.

Written out after every scan rather than at the end: a run over a whole study can take hours, and a crash four hours in must not take the record of the first three with it. The same reason the ROI table autosaves.

BatchReport

BatchReport(destination=None, plan_source=None)

What happened to every scan in a run, mirrored to destination as it goes.

Track scan outcomes, mirroring the report to destination after each update.

Parameters:

Name	Type	Description	Default
destination	Path | None	Path to write batch_report.xlsx to, or None to keep the report in memory only.	None
plan_source	dict | None	The batch preset this run came from, recorded in the report's provenance sheet.	None

Source code in src/optari/batch/report.py

def __init__(
    self, destination: Path | None = None, plan_source: dict | None = None
):
    """Track scan outcomes, mirroring the report to *destination* after each update.

    Args:
        destination: Path to write batch_report.xlsx to, or None to keep the report in memory only.
        plan_source: The batch preset this run came from, recorded in the report's provenance sheet.
    """
    self._entries: dict[str, _Entry] = {}
    self._destination = destination
    self._plan_source = plan_source or {}

failures property

failures

Entries for scans that ended with status "failed".

rows property

rows

All recorded scans as a DataFrame, in REPORT_COLUMNS order.

__len__

__len__()

Number of scans recorded so far, regardless of outcome.

Source code in src/optari/batch/report.py

def __len__(self) -> int:
    """Number of scans recorded so far, regardless of outcome."""
    return len(self._entries)

cancel

cancel(job)

Mark job as cancelled by the user.

Source code in src/optari/batch/report.py

def cancel(self, job) -> None:
    """Mark *job* as cancelled by the user."""
    self._finish(job, "cancelled")

counts

counts()

How many scans ended in each status, for the run summary.

Source code in src/optari/batch/report.py

def counts(self) -> dict[str, int]:
    """How many scans ended in each status, for the run summary."""
    counts: dict[str, int] = {}
    for entry in self._entries.values():
        counts[entry.status] = counts.get(entry.status, 0) + 1
    return counts

fail

fail(job, message, step='')

Mark job as failed, recording the reason and which step it failed on.

Source code in src/optari/batch/report.py

def fail(self, job, message: str, step: str = "") -> None:
    """Mark *job* as failed, recording the reason and which step it failed on."""
    self._finish(job, "failed", message=message, failed_step=step)

note

note(job, **fields)

Attach details to a scan still being processed (frame, output paths, row count).

Source code in src/optari/batch/report.py

def note(self, job, **fields) -> None:
    """Attach details to a scan still being processed (frame, output paths, row count)."""
    entry = self._entries.get(job.key)
    if entry is None:
        return
    for key, value in fields.items():
        # A typo here would otherwise vanish silently: asdict only reports fields.
        if not hasattr(entry, key):
            raise AttributeError(f"batch report has no field '{key}'")
        setattr(entry, key, value)

start

start(job)

Begin tracking job, recording its start time, and mirror the report to disk.

Source code in src/optari/batch/report.py

def start(self, job) -> None:
    """Begin tracking *job*, recording its start time, and mirror the report to disk."""
    self._entries[job.key] = _Entry(
        study=job.study_path.name,
        scan=job.scan_path.name,
        scan_name=job.scan_name,
        kind=job.scan_info.kind,
        started_at=dt.datetime.now(),
    )
    self._write()

succeed

succeed(job)

Mark job as completed successfully.

Source code in src/optari/batch/report.py

def succeed(self, job) -> None:
    """Mark *job* as completed successfully."""
    self._finish(job, "ok")

summary

summary()

One line for the batch window once the run ends.

Source code in src/optari/batch/report.py

def summary(self) -> str:
    """One line for the batch window once the run ends."""
    counts = self.counts()
    parts = [
        f"{count} {status}" for status, count in sorted(counts.items())
    ]
    return (
        f"{len(self)} scan(s): {', '.join(parts)}"
        if parts
        else "nothing to do"
    )