API Reference#

This page documents the main public modules in launchcontainers using the package docstrings and function docstrings pulled directly from the source.

Core CLI#

prepare(lc_config: str = typer.Option, sub_ses_list: str = typer.Option, container_specific_config: str | None = typer.Option, quiet: bool = typer.Option, verbose: bool = typer.Option, debug: bool = typer.Option)#
run(workdir: str = typer.Option, run_lc: bool = typer.Option, quiet: bool = typer.Option, verbose: bool = typer.Option, debug: bool = typer.Option)#
qc(workdir: str = typer.Option, log_dir: str = typer.Option, debug: bool = typer.Option)#
main()#
main(parse_namespace) → tuple[bool, str | None]#

Run the full prepare workflow.

For both DWI and GLM pipelines, creates an analysis directory under BIDS/derivatives/fMRI-GLM-{version}/analysis-{analysis_name} (legacy) or fMRI-GLM-{version}_{analysis_name} (new layout), copies configs into it, and delegates to the appropriate prepare helpers.

If the container_specific.<container> section is absent for GLM, an example config is written and the function returns early.

Parameters:

parse_namespace (argparse.Namespace) – Parsed CLI arguments (lc_config, sub_ses_list, container_specific_config).

Returns:

tuple[bool, str | None] – (success, analysis_dir) where analysis_dir is None only when the GLM config section is missing.

write_job_script(job_script, script_dir, job_script_fname)#

Write a generated scheduler script to disk and make it executable.

Parameters:

  • job_script (str) – Full script text to write.

  • script_dir (str) – Directory where the script should be created.

  • job_script_fname (str) – Output filename for the script.

Returns:

str – Path to the written script file.

launch_jobs(parse_namespace, df_subses, job_script_dir, run_lc)#

Generate launch commands and either print or submit them.

Depending on the launchcontainers configuration this function performs a dry run, submits an array job through SLURM or SGE, or runs the commands locally in parallel using concurrent.futures.

Parameters:

  • parse_namespace (argparse.Namespace) – Parsed CLI arguments for run mode.

  • df_subses (pandas.DataFrame) – Filtered subject/session rows to launch.

  • job_script_dir (str) – Directory used to store generated scripts and batch command files.

  • run_lc (bool) – If True, submit jobs for execution. If False, only print the generated launch information.

main(workdir: str, run_lc: bool = False)#

Validate a prepared analysis directory and launch the requested jobs.

The run workflow reloads the prepared configuration files, validates that the analysis directory contains the expected inputs, shows an example input folder structure to the user, asks for confirmation, and then calls launch_jobs().

Parameters:

  • workdir (str) – Working directory for run mode.

  • run_lc (bool, default=False) – Whether to run launchcontainers.

qc()#

Placeholder QC entrypoint.

The package currently routes QC work through main(); this function is kept only as a stub for future expansion.

main(workdir: str, log_dir: str = './logs', debug: bool = False)#

Run the package-level QC workflow for a prepared analysis directory.

Parameters:

  • workdir (str) – Working directory for QC.

  • log_dir (str, default=”./logs”) – Directory to write logs.

  • debug (bool, default=False) – Debug mode.

Utilities#

parse_hms(ts: str) → str#

Normalise any time string to zero-padded HH:MM:SS. Handles ISO datetime, sub-seconds, single-digit hours.

times_match(t1: str, t2: str, max_diff_sec: int = 30) → bool#

Return True if |t1 - t2| <= max_diff_sec.

die(*args)#

Log an error message and terminate the current process.

Parameters:

*args – Positional message parts forwarded to console.print().

read_yaml(path_to_config_file)#

Read a YAML configuration file.

Parameters:

path_to_config_file (str or path-like) – Path to the YAML file to load.

Returns:

dict – Parsed YAML contents.

read_df(path_to_df_file)#

Read a CSV file into a dataframe and count runnable rows.

This helper is primarily used for subseslist files, where the RUN column marks which rows should be scheduled.

Parameters:

path_to_df_file (str or path-like) – Path to the CSV file to read.

Returns:

tuple[pandas.DataFrame, int | None] – The loaded dataframe and the number of rows where RUN == 'True'. If the file does not contain a RUN column, the count is None.

copy_file(src_file, dst_file, force)#

Copy a file to a destination path with optional overwrite behavior.

Parameters:

  • src_file (str or path-like) – Source file to copy.

  • dst_file (str or path-like) – Destination file path.

  • force (bool) – If True, overwrite an existing destination file.

Returns:

str – Destination path.

Raises:

FileExistsError – If the source file does not exist.

copy_configs(output_path, force=True)#

Copy packaged example configuration files into a target directory.

Parameters:

  • output_path (str or path-like) – Directory where the example configuration files should be copied.

  • force (bool, default=True) – If True, overwrite existing files in output_path.

Create or refresh a symbolic link and validate the result.

Parameters:

  • file1 (str or path-like) – Link target, usually an output from a previous processing step.

  • file2 (str or path-like) – Link path to create for the current container input.

  • force (bool) – If True, replace an existing link at file2.

Raises:

OSError – If the source is missing or the link cannot be created.

Validate that a path is an existing symbolic link target.

Parameters:

path (str) – Link path to inspect.

Raises:

FileNotFoundError – If path is a symlink but its target does not exist.

Shared Rich console, color-scheme constants, and verbosity control for launchcontainers.

All modules import console from here so that the single instance can be redirected to a log file once (in cli.py) and every console.print() call across the package ends up in that file automatically.

Verbosity levels#

Call setup_verbosity() once at CLI startup to configure what gets printed:

  • default : INFO, WARNING, ERROR, CRITICAL are shown

  • --quiet : only ERROR and CRITICAL are shown

  • --verbose: same as default (INFO and above)

  • --debug : everything including DEBUG messages

Use the helper functions instead of console.print() directly so that the verbosity level is respected automatically:

from launchcontainers.log_setup import log_debug, log_info, log_warning, log_error, log_critical

log_info("Starting prepare phase")
log_warning("Config key missing, using default")
log_error("File not found")
log_critical("Launching now")
log_debug("Internal state: x = 42")

Color scheme (mirrors color_codes.txt)#

  • INFO / DEBUG "cyan" General information messages

  • WARNING "yellow" Warning messages

  • ERROR "red" Error messages

  • CRITICAL "bold red" Critical messages and important notifications

  • BLUE_INFO "blue" Specific info (e.g. container layout)

setup_verbosity(quiet: bool = False, verbose: bool = False, debug: bool = False) → None#

Configure the verbosity level for all console output.

Call this once at the start of each CLI command (prepare, run, qc) before any other output is produced. All subsequent calls to log_info, log_debug, etc. will respect the level set here.

Parameters:

  • quiet (bool) – Suppress INFO and WARNING output; only ERROR and CRITICAL are shown.

  • verbose (bool) – Alias for the default level (INFO and above). Provided for CLI symmetry; no additional output is added beyond the default.

  • debug (bool) – Show DEBUG messages in addition to all other levels.

set_log_files(log_fpath: str, err_fpath: str) → None#

Attach two FileHandlers to the launchcontainers logger.

Parameters:

  • log_fpath (str) – .log file — captures everything (DEBUG and above).

  • err_fpath (str) – .err file — captures WARNING and above only.

log_debug(msg: str) → None#

Print a debug message (green). Only shown when –debug is set.

log_info(msg: str) → None#

Print an info message (cyan). Suppressed in –quiet mode.

log_warning(msg: str) → None#

Print a warning message (yellow). Always logged to file.

log_error(msg: str) → None#

Print an error message (red). Always shown and logged.

log_critical(msg: str) → None#

Print a critical message (bold red). Always shown and logged.

Job-script generation#

gen_launch_cmd(parse_namespace, df_subses, batch_command_file)#

Generate one launch command per requested subject/session row.

Routes to the appropriate command generator based on the job type configured in lc_config:

The command list is also written to batch_command_file so scheduler array jobs can read one command per task index.

Parameters:

  • parse_namespace (argparse.Namespace) – Parsed CLI arguments for run mode.

  • df_subses (pandas.DataFrame) – Filtered subject/session rows to launch.

  • batch_command_file (str or path-like) – Output text file that stores the generated command list.

Returns:

list[str] – Launch commands in the same order as df_subses.

gen_cmd_prefix(lc_config)#

Generate the common Apptainer command prefix for a configured host.

The prefix includes optional module loading and the shared bind mounts required to run launchcontainers analyses inside an Apptainer/Singularity container.

Parameters:

lc_config (dict) – Parsed launchcontainers YAML configuration.

Returns:

str – Shell command prefix used by container-specific launch commands.

gen_RTP2_cmd(lc_config, sub, ses, analysis_dir)#

Build the launch command for one subject/session container run.

The generated command binds the prepared input and output directories into the Flywheel-style container layout expected by the supported DWI-related containers and redirects stdout/stderr into timestamped log files.

Parameters:

  • lc_config (dict) – Parsed launchcontainers YAML configuration.

  • sub (str) – Subject identifier without the sub- prefix.

  • ses (str) – Session identifier without the ses- prefix.

  • analysis_dir (str) – Prepared analysis directory containing per-session input/output trees.

Returns:

str – Full shell command used to run the container for one session.

Raises:

ValueError – If no command template can be selected for the configured container.

gen_matlab_cmd(lc_config, sub, ses, analysis_dir)#

Build a MATLAB launch command for one subject/session.

Parameters:

  • lc_config (dict) – Parsed launchcontainers YAML configuration.

  • sub (str) – Subject identifier without the sub- prefix.

  • ses (str) – Session identifier without the ses- prefix.

  • analysis_dir (str) – Prepared analysis directory containing per-session input/output trees.

Returns:

str – Full shell command to invoke the MATLAB script for one session.

gen_py_cmd(lc_config, sub, ses, analysis_dir)#

Build a Python script launch command for one subject/session.

Parameters:

  • lc_config (dict) – Parsed launchcontainers YAML configuration.

  • sub (str) – Subject identifier without the sub- prefix.

  • ses (str) – Session identifier without the ses- prefix.

  • analysis_dir (str) – Prepared analysis directory containing per-session input/output trees.

Returns:

str – Full shell command to invoke the Python script for one session.

Preparation#

For full documentation of the prepare pipelines see the tutorial pages:

Schedulers#

launch_serial(cmds: list[str]) → list[int]#

Execute a list of shell commands one by one in order.

Parameters:

cmds (list[str]) – Shell commands to run sequentially.

Returns:

list[int] – Return codes in submission order.

launch_parallel(cmds: list[str], max_workers: int | None = None, mem_per_job: str | None = None) → list[int]#

Execute a list of shell commands in parallel using ProcessPoolExecutor.

Each worker process is initialised with an optional memory ceiling via resource.setrlimit, so the apptainer container it spawns cannot exceed that limit. CPU concurrency is controlled by max_workers; set it to floor(total_cores / cpus_per_job) in the yaml to keep total CPU usage within budget.

Parameters:

  • cmds (list[str]) – Shell commands to run in parallel.

  • max_workers (int or None) – Maximum number of parallel workers. None uses the number of CPUs.

  • mem_per_job (str or None) – Memory limit per worker, e.g. '32g'. None means no limit.

Returns:

list[int] – Return codes in completion order.

gen_slurm_array_job_script(parse_namespace, log_dir, n_jobs)#

Build the SLURM array-job script used for batch container launches.

Parameters:

  • parse_namespace (argparse.Namespace) – Parsed CLI arguments for run mode.

  • log_dir (str) – Directory where scheduler stdout/stderr logs should be written.

  • n_jobs (int) – Number of array tasks to request.

  • Returns –

    str

    Full SLURM batch script text.

gen_sge_array_job_script(parse_namespace, log_dir, n_jobs)#

Build the SGE array-job script used for batch container launches.

Parameters:

  • parse_namespace (argparse.Namespace) – Parsed CLI arguments for run mode.

  • log_dir (str) – Directory where scheduler stdout/stderr logs should be written.

  • n_jobs (int) – Number of array tasks to request.

  • Returns –

    str

    Full SGE batch script text.

Checks And QC#

MIT License

Copyright (c) 2020-2023 Garikoitz Lerma-Usabiaga Copyright (c) 2020-2022 Mengxing Liu Copyright (c) 2022-2024 Leandro Lecca Copyright (c) 2022-2023 Yongning Lei Copyright (c) 2023 David Linhardt Copyright (c) 2023 Iñigo Tellaetxe

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the “Software”), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

check_tractparam(lc_config, sub, ses, tractparam_df)#

Verify that all ROIs referenced in tractparams exist in fs.zip.

Parameters:

  • lc_config (dict) – Parsed launchcontainers YAML configuration.

  • sub (str) – Subject identifier without the sub- prefix.

  • ses (str) – Session identifier without the ses- prefix.

  • tractparam_df (pandas.DataFrame) – Tract parameter table used by rtp-pipeline or rtp2-pipeline.

Returns:

bool – True if every required ROI file is present in the anatomical derivative fs.zip archive.

Raises:

FileNotFoundError – If one or more required ROI files are missing.

check_dwi_analysis_folder(parse_namespace, container)#

Validate the analysis-level configuration files for a prepared run.

Parameters:

  • parse_namespace (argparse.Namespace) – Parsed CLI arguments for run mode.

  • container (str) – Container name used to resolve the expected JSON filename.

Returns:

bool – True if the expected analysis-level configuration files exist.

Raises:

FileNotFoundError – If any required analysis-level configuration file is missing.

backup_old_rtp2pipeline_log(parse_namespace, df_subses)#

Rename pre-existing RTP log files before launching a new run.

Parameters:

  • parse_namespace (argparse.Namespace) – Parsed CLI arguments for run mode.

  • df_subses (pandas.DataFrame) – Subject/session rows whose output directories should be inspected.

MIT License

Copyright (c) 2020-2023 Garikoitz Lerma-Usabiaga Copyright (c) 2020-2022 Mengxing Liu Copyright (c) 2022-2024 Leandro Lecca Copyright (c) 2022-2023 Yongning Lei Copyright (c) 2023 David Linhardt Copyright (c) 2023 Iñigo Tellaetxe

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the “Software”), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

cli_show_folder_struc(analysis_dir, sub, ses)#

Show an example prepared input tree for one subject/session.

Parameters:

  • analysis_dir (str) – Prepared analysis directory.

  • sub (str) – Subject identifier without the sub- prefix.

  • ses (str) – Session identifier without the ses- prefix.

print_option_for_review(num_of_jobs, lc_config, container, bidsdir_name)#

Log the most important launch settings for manual review.

Parameters:

  • num_of_jobs (int) – Number of subject/session rows that will be launched.

  • lc_config (dict) – Parsed launchcontainers YAML configuration.

  • container (str) – Container name selected for the run.

  • bidsdir_name (str) – Name of the BIDS dataset directory under the configured base path.

Raises:

FileNotFoundError – If the configured container image is not present in containerdir.

process_tracts(sub, ses, analysis_dir)#

Unpack and reorganize tract outputs for one subject/session.

Parameters:

  • sub (str) – Subject identifier without the sub- prefix.

  • ses (str) – Session identifier without the ses- prefix.

  • analysis_dir (str) – Analysis directory containing per-session output folders.

find_subseslist(analysis_dir)#

Locate subseslist.txt somewhere under an analysis directory.

Parameters:

analysis_dir (str or path-like) – Root directory to search.

Returns:

str – Full path to the discovered subseslist.txt file.

main(analysis_dir)#

Reprocess tract outputs in parallel for every runnable DWI session.

Parameters:

analysis_dir (str) – Analysis directory containing the prepared subseslist.txt file.

The output QC will only related to the analysis dir, so it is very good! TODO: feature, give a analysis name and version, it will look for the things under BIDS/derivatives using pybids

For rtp2preproc, it seems that the only thing we need to check is that if the DWI file was prepared.

find_newest_log(log_dir: Path, exts: tuple[str, ...] = ('.err', '.e')) → Path | None#

Return the newest log file in a directory.

The function first looks for files whose names end with a timestamp of the form YYYY-MM-DD_HH-MM-SS followed by one of the requested extensions. If no timestamped filename is found, it falls back to non-timestamped log files and chooses the only match or the most recently modified one.

Parameters:

  • log_dir (pathlib.Path) – Directory containing log files.

  • exts (tuple[str, …], default=(‘.err’, ‘.e’)) – File extensions that should be considered valid log files.

Returns:

pathlib.Path | None – Path to the newest matching log file, or None if no candidate log file exists.

find_subseslist(analysis_dir)#

Locate subseslist.txt somewhere under an analysis directory.

Parameters:

analysis_dir (str or path-like) – Root directory to search.

Returns:

str – Full path to the discovered subseslist.txt file.

check_rtp_preproc_logs(analysis_dir)#

Report sessions whose expected preprocessed DWI output is missing.

Parameters:

analysis_dir (str or path-like) – Analysis directory containing session output folders and subseslist.txt.

Tract processing and validation functions

load_tract_params(tractparams_file)#

Load tract names from a tract parameter CSV file.

Parameters:

tractparams_file (str or path-like) – Path to the tract parameter CSV file.

Returns:

pandas.Series – Derived tract names that should be present in the output.

get_expected_tract_files()#

Return the file suffixes expected for each tract output.

Returns:

list[str] – Suffixes that define a complete tract result.

check_tract_completeness(subses_outputdir)#

Check whether each expected tract has all required output files.

Parameters:

subses_outputdir (str or path-like) – Output directory for one subject/session.

Returns:

tuple[list[str], list[str], dict] – Complete tracts, missing tracts, and a per-tract status dictionary.

find_subseslist(analysis_dir)#

Locate subseslist.txt somewhere under an analysis directory.

Parameters:

analysis_dir (str or path-like) – Root directory to search.

Returns:

str – Full path to the discovered subseslist.txt file.

check_and_tract_dir(subses_outputdir)#

Check whether tract output directories or archives are present.

Parameters:

subses_outputdir (str or path-like) – Output directory for one subject/session.

Returns:

tuple[bool, bool, bool, str] – Presence of the tract directory, presence of the tract archive, whether the state is acceptable, and an optional warning message.

check_all_sub(analysis_dir)#

Print missing tract summaries for all runnable DWI sessions.

Parameters:

analysis_dir (str or path-like) – Analysis directory containing subseslist.txt and output folders.

find_subseslist(analysis_dir)#

Locate subseslist.txt somewhere under an analysis directory. This is limited to RTP2 pipeline outputs, which should always have this file. TODO: make other pipelines also stores a subseslist.txt in the same format, and then this function can be used for all pipelines

flatten_directory(directory)#

Flatten a single nested directory produced by archive extraction.

Parameters:

directory (str or path-like) – Directory that may contain one unnecessary top-level subdirectory.

unzip_subses_output(subses_outputdir: Path, force)#

Ensure that one session output has an extracted RTP output directory.

Parameters:

  • subses_outputdir (pathlib.Path) – Output directory for one subject/session.

  • force (bool) – If True, re-extract the archive even when an output directory already exists.

Returns:

tuple[bool, bool, bool, str] – Presence of the extracted directory, presence of the zip archive, extraction success, and an optional warning message.

check_subses_output(subses_outputdir)#

Check whether one session output is ready for tract inspection.

Parameters:

subses_outputdir (str or path-like) – Output directory for one subject/session.

Returns:

tuple[bool, bool, bool, str] – Presence of the extracted directory, presence of the zip archive, whether additional extraction is required, and an optional warning.

process_single_subject(analysis_dir, sub, ses, run, dwi, force)#

Unzip tract outputs for one subject/session if the row should run.

unzipping_tracts_parallel(analysis_dir, force, n_workers=65)#

Unzip all runnable tract archives in parallel.

Parameters:

  • analysis_dir (str or path-like) – Analysis directory containing subseslist.txt and output folders.

  • force (bool) – If True, re-extract existing output directories.

  • n_workers (int, default=65) – Maximum number of worker threads.

Returns:

list[dict] – Per-session processing results.

save_processing_summary(analysis_dir, results)#

Write a JSON summary of the tract-unzipping batch run.

check_and_unzip_tract_wrapper(output_dir, sub, ses)#

Wrap unzip_subses_output() for executor-based parallel processing.

rsync_single_subject(analysis_dir, sub, ses)#

Synchronize tract files for one subject/session using rsync.

Parameters:

  • analysis_dir (str or path-like) – Analysis directory containing session outputs.

  • sub (str) – Subject identifier without the sub- prefix.

  • ses (str) – Session identifier without the ses- prefix.

Returns:

str – Human-readable status message for the transfer.

find_subseslist(analysis_dir)#

Locate subseslist.txt somewhere under an analysis directory.

batch_rsync_tracts(analysis_dir)#

Run tract synchronization across all runnable DWI sessions in parallel.

Parameters:

analysis_dir (str or path-like) – Analysis directory containing subseslist.txt and output folders.

Returns:

list[str] – Status messages returned by rsync_single_subject().

Helper Scripts#

copy_example_configs(copy_configs)#

Copy packaged example configuration files into a destination directory.

Parameters:

copy_configs (str) – Destination directory that will receive the example files.

main()#

Run the copy_configs helper from the shared launchcontainers parser.

main()#

Create a fake BIDS tree and matching derivative skeleton for testing.

The helper reads a small configuration YAML plus a subject/session list and creates empty files and folders that mimic the inputs expected by the rest of the package.

gen_subseslist(basedir: str, output_name: str, output_dir=None) → None#

Generate a simple subseslist file from a BIDS-style directory tree.

Parameters:

  • basedir (str) – Root directory containing sub-* folders.

  • output_name (str) – Output filename stem without the .txt suffix.

  • output_dir (str, optional) – Destination directory for the generated file. If omitted, basedir is used.

main()#

Run the gen_subses helper from the shared launchcontainers parser.

get_git_root()#

Return the repository root detected from Git.

Returns:

pathlib.Path | None – Repository root path, or None if the command is not run inside a Git working tree.

do_zip_configs()#

Package the repository example configuration directory as a zip archive.