nipoppy.workflows.base

Workflow utilities.

Module Contents

class nipoppy.workflows.base.BaseWorkflow(dpath_root, name, fpath_layout=None, logger=None, dry_run=False)

Bases: nipoppy.base.Base, abc.ABC

Base class with logging/subprocess utilities.

Parameters:

  • dpath_root (nipoppy.env.StrOrPathLike)

  • name (str)

  • fpath_layout (Optional[nipoppy.env.StrOrPathLike])

  • logger (Optional[logging.Logger])

log_prefix_run = '[RUN]'
log_prefix_run_stderr = '[RUN STDERR]'
log_prefix_run_stdout = '[RUN STDOUT]'
path_sep = '-'
pipeline_name = '[[NIPOPPY_PIPELINE_NAME]]'
pipeline_version = '[[NIPOPPY_PIPELINE_VERSION]]'
validate_layout = True
config()

Load the configuration.

Return type:

nipoppy.config.main.Config

copy(path_source, path_dest, log_level=logging.INFO, **kwargs)

Copy a file or directory.

Create a symlink to another path.

dicom_dir_map()

Get the DICOM directory mapping.

Return type:

nipoppy.tabular.dicom_dir_map.DicomDirMap

doughnut()

Load the doughnut.

Return type:

nipoppy.tabular.doughnut.Doughnut

generate_fpath_log(dnames_parent=None, fname_stem=None)

Generate a log file path.

Parameters:

  • dnames_parent (Optional[str | list[str]])

  • fname_stem (Optional[str])

Return type:

pathlib.Path

log_command(command)

Write a command to the log with a special prefix.

Parameters:

command (str)

manifest()

Load the manifest.

Return type:

nipoppy.tabular.manifest.Manifest

mkdir(dpath, log_level=logging.INFO, **kwargs)

Create a directory (by default including parents).

Do nothing if the directory already exists.

rm(path, log_level=logging.INFO, **kwargs)

Remove a file or directory.

run()

Run the workflow.

run_cleanup()

Run the cleanup part of the workflow.

run_command(command_or_args, check=True, **kwargs)

Run a command in a subprocess.

The command’s stdout and stderr outputs are written to the log with special prefixes.

If in “dry run” mode, the command is not executed, and the method returns the command string. Otherwise, the subprocess.Popen object is returned unless capture_output is True.

Parameters:

  • command_or_args (Sequence[str] | str) – The command to run.

  • check (bool, optional) – If True, raise an error if the process exits with a non-zero code, by default True

  • **kwargs – Passed to subprocess.Popen.

Return type:

subprocess.Popen or str

abstract run_main()

Run the main part of the workflow.

run_setup()

Run the setup part of the workflow.

save_tabular_file(tabular, fpath)

Save a tabular file.

Parameters:
nipoppy.workflows.base.LOG_SUFFIX = '.log'