nipoppy.workflows.pipeline

Base class for pipeline workflows.

Module Contents

class nipoppy.workflows.pipeline.BasePipelineWorkflow(dpath_root, name, pipeline_name, pipeline_version=None, pipeline_step=None, participant_id=None, session_id=None, fpath_layout=None, logger=None, dry_run=False)

Bases: nipoppy.workflows.base.BaseWorkflow, abc.ABC

A workflow for a pipeline that has a Boutiques descriptor.

Parameters:

  • dpath_root (nipoppy.env.StrOrPathLike)

  • name (str)

  • pipeline_name (str)

  • pipeline_version (Optional[str])

  • pipeline_step (Optional[str])

  • participant_id (str)

  • session_id (str)

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

  • logger (Optional[logging.Logger])

boutiques_config()

Get the Boutiques configuration.

check_dir(dpath)

Create directory if it does not exist.

Parameters:

dpath (pathlib.Path)

check_pipeline_version()

Set the pipeline version based on the config if it is not given.

descriptor()

Load the pipeline step’s Boutiques descriptor.

Return type:

dict

dpath_pipeline()

Return the path to the pipeline’s derivatives directory.

Return type:

pathlib.Path

dpath_pipeline_bids_db()

Return the path to the pipeline’s BIDS database directory.

Return type:

pathlib.Path

dpath_pipeline_output()

Return the path to the pipeline’s output directory.

Return type:

pathlib.Path

dpath_pipeline_work()

Return the path to the pipeline’s working directory.

Return type:

pathlib.Path

dpaths_to_check()

Directory paths to create if needed during the setup phase.

Return type:

list[pathlib.Path]

fpath_container()

Return the full path to the pipeline’s container.

Return type:

pathlib.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

abstract get_participants_sessions_to_run(participant_id, session_id)

Return participant-session pairs to loop over with run_single().

This is an abstract method that should be defined explicitly in subclasses.

Parameters:

  • participant_id (Optional[str])

  • session_id (Optional[str])

invocation()

Load the pipeline step’s Boutiques invocation.

Return type:

dict

pipeline_config()

Get the user config for the pipeline.

Return type:

nipoppy.config.pipeline.ProcPipelineConfig

process_template_json(template_json, participant_id, session_id, bids_participant=None, bids_session=None, objs=None, return_str=False, **kwargs)

Replace template strings in a JSON object.

Parameters:

  • template_json (dict)

  • participant_id (Optional[str])

  • session_id (Optional[str])

  • bids_participant (Optional[str])

  • bids_session (Optional[str])

  • objs (Optional[list])

  • return_str (bool)

pybids_ignore_patterns()

Load the pipeline step’s PyBIDS ignore pattern list.

Note: this does not apply any substitutions, since the subject/session patterns are always added.

Return type:

list[str]

run_cleanup()

Log a summary message.

run_main()

Run the pipeline.

run_setup()

Run pipeline setup.

abstract run_single(participant_id, session_id)

Run on a single participant/session.

This is an abstract method that should be defined explicitly in subclasses.

Parameters:

  • participant_id (Optional[str])

  • session_id (Optional[str])

set_up_bids_db(dpath_bids_db, participant_id=None, session_id=None)

Set up the BIDS database.

Parameters:

  • dpath_bids_db (nipoppy.env.StrOrPathLike)

  • participant_id (Optional[str])

  • session_id (Optional[str])

Return type:

bids.BIDSLayout

nipoppy.workflows.pipeline.apply_analysis_level(participants_sessions, analysis_level)

Filter participant-session pairs to run based on the analysis level.

Parameters:
Return type:

Tuple[str, str]