nipoppy.workflows.base.BaseWorkflow

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

Base class with logging/subprocess utilities.

Parameters:

• dpath_root (nipoppy.env.StrOrPathLike)

• name (str)

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

• verbose (bool)

• dry_run (bool)

• _skip_logging (bool)

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

Copy a file or directory.

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

Copy directory tree.

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

Create a symlink to another path.

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)

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

Create a directory (by default including parents).

Do nothing if the directory already exists.

movetree(path_source, path_dest, kwargs_mkdir=None, kwargs_move=None, log_level=logging.INFO)

Move directory tree.

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, quiet=False, **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

• quiet (bool, optional) – If True, do not log the command, by default False

• **kwargs – Passed to subprocess.Popen.

Return type:

subprocess.Popen or str

abstractmethod 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:

• tabular (nipoppy.tabular.base.BaseTabular)

• fpath (pathlib.Path)

property config: nipoppy.config.main.Config

Load the configuration.

Raise error if not found.

Return type:

nipoppy.config.main.Config

property curation_status_table: nipoppy.tabular.curation_status.CurationStatusTable

Load the curation status file if it exists.

Otherwise, generate a new one.

Return type:

nipoppy.tabular.curation_status.CurationStatusTable

property dicom_dir_map: nipoppy.tabular.dicom_dir_map.DicomDirMap

Get the DICOM directory mapping.

Return type:

nipoppy.tabular.dicom_dir_map.DicomDirMap

dpath_root

dry_run = False

fpath_layout = None

layout

log_prefix_run = '[RUN]'

log_prefix_run_stderr = '[RUN STDERR]'

log_prefix_run_stdout = '[RUN STDOUT]'

logger

property manifest: nipoppy.tabular.manifest.Manifest

Load the manifest.

Raise error if not found.

Return type:

nipoppy.tabular.manifest.Manifest

name

path_sep = '-'

property processing_status_table: nipoppy.tabular.processing_status.ProcessingStatusTable

Load the processing status file it it exists.

Otherwise, return an empty processing status table.

Return type:

nipoppy.tabular.processing_status.ProcessingStatusTable

return_code = 0

validate_layout = True

verbose = False