voxelops.runners package

Procedure runner functions. Each runner validates inputs, builds a command, executes it (via Docker or direct Python call), and returns an execution record dict.

Shared Utilities

Base utilities for all procedure runners.

run_docker(cmd: List[str], tool_name: str, participant: str, log_dir: Path | None = None, capture_output: bool = True) → Dict[str, Any][source]

Execute Docker command and return execution record.

Parameters:

  • cmd (List[str]) – Docker command as list of strings.

  • tool_name (str) – Name of the tool being run (for logging).

  • participant (str) – Participant label.

  • log_dir (Optional[Path], optional) – Directory to save execution log JSON, by default None.

  • capture_output (bool, optional) – Whether to capture stdout/stderr, by default True.

Returns:

A dictionary with execution details:

  • tool: Tool name

  • participant: Participant label

  • command: Full command that was executed

  • exit_code: Process exit code

  • start_time: ISO format timestamp

  • end_time: ISO format timestamp

  • duration_seconds: Duration in seconds

  • duration_human: Human-readable duration

  • success: Boolean success status

  • log_file: Path to JSON log (if log_dir provided)

  • stdout: Process stdout (if captured)

  • stderr: Process stderr (if captured)

  • error: Error message (if failed)

Return type:

Dict[str, Any]

Raises:

ProcedureExecutionError – If command fails.

validate_input_dir(input_dir: Path, dir_type: str = 'Input') → None[source]

Validate that an input directory exists.

Parameters:

  • input_dir (Path) – Directory to validate.

  • dir_type (str, optional) – Type of directory for error message, by default “Input”.

Raises:

InputValidationError – If directory doesn’t exist.

validate_participant(input_dir: Path, participant: str, prefix: str = 'sub-') → None[source]

Validate that a participant exists in the input directory.

Parameters:

  • input_dir (Path) – Directory containing participant data.

  • participant (str) – Participant label (without prefix).

  • prefix (str, optional) – Expected prefix, by default “sub-“.

Raises:

InputValidationError – If participant not found.

HeudiConv

HeudiConv DICOM to BIDS converter runner.

run_heudiconv(inputs: HeudiconvInputs, config: HeudiconvDefaults | None = None, **overrides) → Dict[str, Any][source]

Convert DICOM to BIDS using HeudiConv.

Parameters:

  • inputs (HeudiconvInputs) – Required inputs (dicom_dir, participant, etc.).

  • config (Optional[HeudiconvDefaults], optional) – Configuration (uses defaults if not provided), by default None.

  • **overrides – Override any config parameter.

Returns:

Execution record with:

  • tool: “heudiconv”

  • participant: Participant label

  • command: Full Docker command executed

  • exit_code: Process exit code

  • start_time, end_time: ISO format timestamps

  • duration_seconds, duration_human: Execution duration

  • success: Boolean success status

  • log_file: Path to JSON log

  • inputs: HeudiconvInputs instance

  • config: HeudiconvDefaults instance

  • expected_outputs: HeudiconvOutputs instance

Return type:

Dict[str, Any]

Raises:

Examples

>>> inputs = HeudiconvInputs(
...     dicom_dir=Path("/data/dicoms"),
...     participant="01",
... )
>>> config = HeudiconvDefaults(
...     heuristic=Path("/code/heuristic.py"),
... )
>>> result = run_heudiconv(inputs, config)
>>> print(result['expected_outputs'].bids_dir)
PosixPath('/data/bids')

QSIPrep

QSIPrep diffusion preprocessing runner.

run_qsiprep(inputs: QSIPrepInputs, config: QSIPrepDefaults | None = None, **overrides) → Dict[str, Any][source]

Run QSIPrep diffusion MRI preprocessing.

Parameters:

  • inputs (QSIPrepInputs) – Required inputs (bids_dir, participant, etc.).

  • config (Optional[QSIPrepDefaults], optional) – Configuration (uses brain bank defaults if not provided), by default None.

  • **overrides – Override any config parameter (e.g., nprocs=16).

Returns:

Execution record with:

  • tool: “qsiprep”

  • participant: Participant label

  • command: Full Docker command executed

  • exit_code: Process exit code

  • start_time, end_time: ISO format timestamps

  • duration_seconds, duration_human: Execution duration

  • success: Boolean success status

  • log_file: Path to JSON log

  • inputs: QSIPrepInputs instance

  • config: QSIPrepDefaults instance

  • expected_outputs: QSIPrepOutputs instance

Return type:

Dict[str, Any]

Raises:

Examples

>>> inputs = QSIPrepInputs(
...     bids_dir=Path("/data/bids"),
...     participant="01",
... )
>>> result = run_qsiprep(inputs, nprocs=16)  # Override default nprocs
>>> print(f"Completed in {result['duration_human']}")
>>> print(f"Outputs in: {result['expected_outputs'].qsiprep_dir}")

QSIRecon

QSIRecon diffusion reconstruction runner.

run_qsirecon(inputs: QSIReconInputs, config: QSIReconDefaults | None = None, **overrides) → Dict[str, Any][source]

Run QSIRecon diffusion reconstruction and connectivity.

Parameters:

  • inputs (QSIReconInputs) – Required inputs (qsiprep_dir, participant, etc.).

  • config (Optional[QSIReconDefaults], optional) – Configuration (uses brain bank defaults if not provided), by default None.

  • **overrides – Override any config parameter.

Returns:

Execution record with:

  • tool: “qsirecon”

  • participant: Participant label

  • command: Full Docker command executed

  • exit_code: Process exit code

  • start_time, end_time: ISO format timestamps

  • duration_seconds, duration_human: Execution duration

  • success: Boolean success status

  • log_file: Path to JSON log

  • inputs: QSIReconInputs instance

  • config: QSIReconDefaults instance

  • expected_outputs: QSIReconOutputs instance

Return type:

Dict[str, Any]

Raises:

Examples

>>> inputs = QSIReconInputs(
...     qsiprep_dir=Path("/data/derivatives/qsiprep"),
...     participant="01",
... )
>>> result = run_qsirecon(inputs, atlases=["schaefer100"])
>>> print(result['expected_outputs'].qsirecon_dir)

QSIParc

QSIParc parcellation runner using parcellate package.

run_qsiparc(inputs: QSIParcInputs, config: QSIParcDefaults | None = None, **overrides) → Dict[str, Any][source]

Run parcellation on QSIRecon outputs using parcellate.

Atlases are auto-discovered from the QSIRecon derivatives directory (BIDS dseg files). No manual atlas list is needed.

Parameters:

  • inputs (QSIParcInputs) – Required inputs (qsirecon_dir, participant, etc.).

  • config (Optional[QSIParcDefaults], optional) – Configuration (uses brain bank defaults if not provided), by default None.

  • **overrides – Override any config parameter.

Returns:

Execution record with:

  • tool: “qsiparc”

  • participant: Participant label

  • start_time, end_time: ISO format timestamps

  • duration_seconds, duration_human: Execution duration

  • success: Boolean success status

  • output_files: List of output TSV paths

  • inputs: QSIParcInputs instance

  • config: QSIParcDefaults instance

  • expected_outputs: QSIParcOutputs instance

Return type:

Dict[str, Any]

Raises:

Examples

>>> inputs = QSIParcInputs(
...     qsirecon_dir=Path("/data/derivatives/qsirecon"),
...     participant="01",
... )
>>> result = run_qsiparc(inputs)
>>> print(result['output_files'])

Module contents

Procedure runners for brain bank neuroimaging pipelines.