Utils

utils_bids

create_datasource(output_query, data_dir, nipype_dir, subjects=None, sessions=None, acquisitions=None, derivative=None, name='bids_datasource', extra_derivatives=None, save_db=False)

Create a datasource node that have iterables following BIDS format. By default, from a BIDSLayout, lists all the subjects (<sub>), finds their session numbers (<ses>, if any) and their acquisition type (<acq>, if any), and builds an iterable of tuples (sub, ses, acq) with all valid combinations.

If a list of subjects/sessions/acquisitions is provided, the BIDSLayout is not queried and the provided subjects/sessions/acquisitions are used as is.

If derivative is not None, the BIDSLayout will be queried for the specified derivative.

For example, if provided with subjects=["sub01", "sub02"], sessions=["01"], acq=["haste", "tru"], the datagrabber will attempt at retrieving all of the following combinations:

[("sub01", "01", "haste"), ("sub01", "01","tru"),
 ("sub02", "01", "haste"), ("sub02", "01","tru")]

Parameters:

Name	Type	Description	Default
output_query	dict	A dictionary specifying the output query for the BIDSDataGrabber.	required
data_dir	str	The base directory of the BIDS dataset.	required
subjects	list	List of subject IDs to include. If None, all subjects in the dataset are included.	None
sessions	list	List of session IDs to include. If None, all sessions for each subject are included.	None
acquisitions	list	List of acquisition types to include. If None, all acquisitions for each subject/session are included.	None
derivative	str	The name of the derivative to query. If None, no derivative is queried.	None
name	str	Name for the datasource node. Defaults to "bids_datasource".	'bids_datasource'
extra_derivatives	list or str	Additional derivatives to include. If provided, these will be added to the BIDSDataGrabber.	None

Returns: pe.Node: A configured BIDSDataGrabber node that retrieves data according to the specified parameters.

Source code in fetpype/utils/utils_bids.py

def create_datasource(
    output_query,
    data_dir,
    nipype_dir,
    subjects=None,
    sessions=None,
    acquisitions=None,
    derivative=None,
    name="bids_datasource",
    extra_derivatives=None,
    save_db=False,
):
    """Create a datasource node that have iterables following BIDS format.
    By default, from a BIDSLayout, lists all the subjects (`<sub>`),
    finds their session numbers (`<ses>`, if any) and their acquisition
    type (`<acq>`, if any), and builds an iterable of tuples
    (sub, ses, acq) with all valid combinations.

    If a list of subjects/sessions/acquisitions is provided, the
    BIDSLayout is not queried and the provided
    subjects/sessions/acquisitions are used as is.

    If derivative is not None, the BIDSLayout will be queried for
    the specified derivative.

    For example, if provided with subjects=["sub01", "sub02"],
    sessions=["01"], acq=["haste", "tru"], the datagrabber will
    attempt at retrieving all of the following combinations:
    ```
    [("sub01", "01", "haste"), ("sub01", "01","tru"),
     ("sub02", "01", "haste"), ("sub02", "01","tru")]
    ```

    Args:
        output_query (dict): A dictionary specifying the output query
            for the BIDSDataGrabber.
        data_dir (str): The base directory of the BIDS dataset.
        subjects (list, optional): List of subject IDs to include.
            If None, all subjects in the dataset are included.
        sessions (list, optional): List of session IDs to include.
            If None, all sessions for each subject are included.
        acquisitions (list, optional): List of acquisition types to include.
            If None, all acquisitions for each subject/session are included.
        derivative (str, optional): The name of the derivative to query.
            If None, no derivative is queried.
        name (str, optional): Name for the datasource node. Defaults to
                            "bids_datasource".
        extra_derivatives (list or str, optional): Additional
            derivatives to include. If provided, these will be
            added to the BIDSDataGrabber.
    Returns:
        pe.Node: A configured BIDSDataGrabber node that retrieves data
        according to the specified parameters.
    """

    bids_datasource = pe.Node(
        interface=nio.BIDSDataGrabber(),
        name=name,
        synchronize=True,
    )

    bids_datasource.inputs.base_dir = data_dir
    if extra_derivatives is not None:
        bids_datasource.inputs.index_derivatives = True
        if isinstance(extra_derivatives, str):
            extra_derivatives = [extra_derivatives]
        bids_datasource.inputs.extra_derivatives = extra_derivatives
    bids_datasource.inputs.output_query = output_query

    layout = BIDSLayout(data_dir, validate=False)

    # Needed as the layout uses validate which
    # does not work with our segmentation files.

    if save_db:
        layout_db = os.path.join(nipype_dir, "layout_db")
        if os.path.exists(layout_db):
            os.remove(os.path.join(layout_db, "layout_index.sqlite"))
        layout.save(layout_db)

        bids_datasource.inputs.load_layout = layout_db

    # Verbose
    print("BIDS layout:", layout)
    print("\t", layout.get_subjects())
    print("\t", layout.get_sessions())
    iterables = [("subject", "session", "acquisition"), []]

    existing_sub = layout.get_subjects()
    if subjects is None:
        subjects = existing_sub

    for sub in subjects:
        if sub not in existing_sub:
            raise ValueError(
                f"Requested subject {sub} was not found in the "
                f"folder {data_dir}."
            )

        existing_ses = layout.get_sessions(subject=sub)
        if sessions is None:
            sessions_subj = existing_ses
        else:
            sessions_subj = sessions
        # If no sessions are found, it is possible that there is no session.
        sessions_subj = [None] if len(sessions_subj) == 0 else sessions_subj
        for ses in sessions_subj:
            if ses is not None and ses not in existing_ses:
                print(
                    f"WARNING: Session {ses} was not found for subject {sub}."
                )
            existing_acq = layout.get_acquisition(subject=sub, session=ses)
            if acquisitions is None:
                acquisitions_subj = existing_acq
            else:
                acquisitions_subj = acquisitions
            # If there is no acquisition found, maybe the acquisition
            # tag was not specified.
            acquisitions_subj = (
                [None] if len(acquisitions_subj) == 0 else acquisitions_subj
            )
            for acq in acquisitions_subj:
                if acq is not None and acq not in existing_acq:
                    print(
                        f"WARNING: Acquisition {acq} was not found for "
                        f"subject {sub} session {ses}."
                    )

                iterables[1] += [(sub, ses, acq)]

    bids_datasource.iterables = iterables

    return bids_datasource

create_bids_datasink(out_dir, pipeline_name, strip_dir, datatype='anat', name=None, rec_label=None, seg_label=None, surf_label=None, desc_label=None, custom_subs=None, custom_regex_subs=None)

Creates a BIDS-compatible datasink using parameterization and regex substitutions. Organizes outputs into: /derivatives//sub-/[ses-/] /

Source code in fetpype/utils/utils_bids.py

def create_bids_datasink(
    out_dir,
    pipeline_name,
    strip_dir,
    datatype="anat",
    name=None,
    rec_label=None,
    seg_label=None,
    surf_label=None,
    desc_label=None,
    custom_subs=None,
    custom_regex_subs=None,
):
    """
    Creates a BIDS-compatible datasink using parameterization and
    regex substitutions.
    Organizes outputs into:
    <out_dir>/derivatives/<pipeline_name>/sub-<ID>/[ses-<ID>/]
    <datatype>/<BIDS_filename>
    """
    if not strip_dir:
        raise ValueError(
            "`strip_dir` (Nipype work dir base path) is required."
        )
    if name is None:
        name = f"{pipeline_name}_datasink"

    datasink = pe.Node(
        nio.DataSink(
            base_directory=out_dir, parameterization=True, strip_dir=strip_dir
        ),
        name=name,
    )

    regex_subs = []

    bids_derivatives_root = out_dir  # we already pass the bids derivatives
    escaped_bids_derivatives_root = re.escape(out_dir)

    if pipeline_name == "preprocessing":
        # ** Rule 1: Preprocessing Stacks (Denoised) **
        if desc_label == "denoised":
            # with session
            regex_subs.append(
                (
                    (
                        rf"^{escaped_bids_derivatives_root}/"
                        rf".*?_?session_([^/]+)"
                        rf"_subject_([^/]+).*?/?_denoising.*/"
                        rf"(sub-[^_]+_ses-[^_]+(?:_run-\d+))?"
                        rf"_T2w_noise_corrected(\.nii(?:\.gz)?)$"
                    ),
                    (
                        rf"{bids_derivatives_root}/sub-\2/ses-\1/{datatype}"
                        rf"/\3_desc-denoised_T2w\4"
                    ),
                )
            )
            # without session
            regex_subs.append(
                (
                    (
                        rf"^{escaped_bids_derivatives_root}/"
                        rf"(?!.*?_?session_[^/]+).*?_?subject_([^/]+).*?/"
                        rf"?_denoising.*/"
                        rf"(sub-[^_]+(?:_run-\d+)?)?_?"
                        rf"T2w_noise_corrected(\.nii(?:\.gz)?)$"
                    ),
                    (
                        rf"{bids_derivatives_root}/sub-\1/{datatype}"
                        rf"/\2_desc-denoised_T2w\3"
                    ),
                )
            )

        # ** Rule 2: Preprocessing Masks (Cropped) **
        if desc_label == "cropped":
            # with session
            regex_subs.append(
                (
                    (
                        rf"^{escaped_bids_derivatives_root}/.*?_session_"
                        rf"([^/]+)_subject_([^/]+).*/?_cropping.*/"
                        rf"(sub-[^_]+_ses-[^_]+(?:_run-\d+)?)_"
                        rf"mask(\.nii(?:\.gz)?)$"
                    ),
                    (
                        rf"{bids_derivatives_root}/sub-\2/ses-\1/{datatype}"
                        rf"/\3_desc-cropped_mask\4"
                    ),
                )
            )
            # without session
            regex_subs.append(
                (
                    (
                        rf"^{escaped_bids_derivatives_root}/"
                        rf"(?!.*?_session_[^/]+).*?_subject_"
                        rf"([^/]+).*/?_cropping.*/"
                        rf"(sub-[^_]+(?:_run-\d+)?)_mask(\.nii(?:\.gz)?)$"
                    ),
                    (
                        rf"{bids_derivatives_root}/sub-\1/{datatype}"
                        rf"/\2_desc-cropped_mask\3"
                    ),
                )
            )

    # ** Rule 3: Reconstruction Output **
    if rec_label and not seg_label and pipeline_name != "preprocessing":
        # with session
        regex_subs.append(
            (
                (
                    rf"^{escaped_bids_derivatives_root}/.*?_?session_([^/]+)"
                    rf"_subject_([^/]+).*/(?:[^/]+)(\.nii(?:\.gz)?)$"
                ),
                (
                    rf"{bids_derivatives_root}/sub-\2/ses-\1/"
                    rf"{datatype}/sub-\2_ses-\1_rec-{rec_label}_T2w\3"
                ),
            )
        )
        # without session
        regex_subs.append(
            (
                (
                    rf"^{escaped_bids_derivatives_root}/(?!.*?_?session_[^/]+)"
                    rf".*?_?subject_"
                    rf"([^/]+).*/(?:[^/]+)(\.nii(?:\.gz)?)$"
                ),
                (
                    rf"{bids_derivatives_root}/sub-\1/"
                    rf"{datatype}/sub-\1_rec-{rec_label}_T2w\2"
                ),
            )
        )

    # ** Rule 4: Segmentation Output **
    if seg_label and rec_label and pipeline_name != "preprocessing":
        # with session
        regex_subs.append(
            (
                (
                    rf"^{escaped_bids_derivatives_root}/"
                    rf".*?_?session_([^/]+)_subject_([^/]+).*/"
                    rf"input_srr-mask-brain_bounti-19(\.nii(?:\.gz)?)$"
                ),
                (
                    rf"{bids_derivatives_root}/sub-\2/ses-\1/{datatype}/"
                    rf"sub-\2_ses-\1_rec-{rec_label}_seg-{seg_label}_dseg\3"
                ),
            )
        )
        # without session
        regex_subs.append(
            (
                (
                    rf"^{escaped_bids_derivatives_root}/(?!.*?_?session_[^/]+)"
                    rf".*?_?subject_([^/]+).*/"
                    rf"input_srr-mask-brain_bounti-19(\.nii(?:\.gz)?)$"
                ),
                (
                    rf"{bids_derivatives_root}/sub-\1/{datatype}/"
                    rf"sub-\1_rec-{rec_label}_seg-{seg_label}_dseg\2"
                ),
            )
        )

    # ** Rule 5: Surface outputs (.gii) **
    if surf_label:
        label = f"_rec-{rec_label}" if rec_label else ""
        label += f"_seg-{seg_label}" if seg_label else ""
        # with session
        regex_subs.append(
            (
                (
                    rf"^{escaped_bids_derivatives_root}/"
                    rf".*?_?session_([^/]+)_subject_([^/]+).*/"
                    rf"([^/]+)\.gii$"
                ),
                (
                    rf"{bids_derivatives_root}/sub-\2/ses-\1/{datatype}/"
                    rf"sub-\2_ses-\1{label}_\3.gii"
                ),
            )
        )
        # without session
        regex_subs.append(
            (
                (
                    rf"^{escaped_bids_derivatives_root}/(?!.*?_?session_[^/]+)"
                    rf".*?_?subject_([^/]+).*/"
                    rf"([^/]+)\.gii$"
                ),
                (
                    rf"{bids_derivatives_root}/sub-\1/{datatype}/"
                    rf"sub-\1{label}_\2.gii"
                ),
            )
        )

        # ** Rule 6: Surface outputs (.stl) **
        # with session
        regex_subs.append(
            (
                (
                    rf"^{escaped_bids_derivatives_root}/"
                    rf".*?_?session_([^/]+)_subject_([^/]+).*/"
                    rf"([^/]+)\.stl$"
                ),
                (
                    rf"{bids_derivatives_root}/sub-\2/ses-\1/{datatype}/"
                    rf"sub-\2_ses-\1{label}_\3.stl"
                ),
            )
        )
        # without session
        regex_subs.append(
            (
                (
                    rf"^{escaped_bids_derivatives_root}/(?!.*?_?session_[^/]+)"
                    rf".*?_?subject_([^/]+).*/"
                    rf"([^/]+)\.stl$"
                ),
                (
                    rf"{bids_derivatives_root}/sub-\1/{datatype}/"
                    rf"sub-\1{label}_\2.stl"
                ),
            )
        )

    # --- Generic cleanup rules (unchanged, but keep them
    # after the mapping rules) ---
    regex_subs.extend(
        [
            (r"sub-sub-", r"sub-"),
            (r"ses-ses-", r"ses-"),
            (r"_+", "_"),
            (r"(/)_", r"\1"),
            (r"(_)\.", r"\."),
            (r"-+", "-"),
            (r"//+", "/"),
            (r"[\\/]$", ""),
            (r"_ses-None", ""),  # in case something injected a None
            (r"(\.nii\.gz)\1+$", r"\1"),
            (r"(\.nii)\1+$", r"\1"),
        ]
    )

    # Add custom regex substitutions
    if custom_regex_subs:
        regex_subs.extend(custom_regex_subs)

    datasink.inputs.regexp_substitutions = regex_subs

    # Add custom simple substitutions
    final_subs = []
    if custom_subs:
        final_subs.extend(custom_subs)
    datasink.inputs.substitutions = final_subs

    return datasink

create_datasink(iterables, name='output', params_subs={}, params_regex_subs={})

Deprecated. Creates a data sink node for reformatting and organizing relevant outputs.

From: https://github.com/Macatools/macapype (adapted)

Parameters:

Name	Type	Description	Default
iterables	list or tuple	A collection of iterables, containing subject and session information.	required
name	str	The name for the data sink container. Defaults to "output".	'output'
params_subs	dict	A dictionary of parameter substitutions to apply to output paths. Defaults to an empty dictionary.	{}
params_regex_subs	dict	A dictionary of regular expression-based substitutions to apply to output paths. Defaults to an empty dictionary.	{}

Returns:

Type	Description
	pe.Node: A Pipeline Engine Node representing the configured datasink.

Source code in fetpype/utils/utils_bids.py

def create_datasink(
    iterables, name="output", params_subs={}, params_regex_subs={}
):
    """
    Deprecated. Creates a data sink node for reformatting and organizing
    relevant outputs.

    From: https://github.com/Macatools/macapype (adapted)

    Args:
        iterables (list or tuple): A collection of iterables, containing
                                   subject and session information.
        name (str, optional): The name for the data sink container.
                              Defaults to "output".
        params_subs (dict, optional): A dictionary of parameter substitutions
                                      to apply to output paths. Defaults to
                                      an empty dictionary.
        params_regex_subs (dict, optional): A dictionary of regular
                                            expression-based substitutions
                                            to apply to output paths.
                                            Defaults to an empty dictionary.

    Returns:
        pe.Node: A Pipeline Engine Node representing the configured datasink.
    """

    print("Datasink name: ", name)

    # Create the datasink node
    datasink = pe.Node(nio.DataSink(), name=name)

    # Generate subject folders with session and subject information
    subjFolders = [
        (
            "_acquisition_%s_session_%s_subject_%s" % (acq, ses, sub),
            "sub-%s/ses-%s/anat" % (sub, ses),
        )
        for (sub, ses, acq) in iterables[1]  # doublecheck
    ]

    print("subjFolders: ", subjFolders)

    # Load parameter substitutions from the 'subs.json' file
    json_subs = op.join(op.dirname(op.abspath(__file__)), "subs.json")
    dict_subs = json.load(open(json_subs, encoding="utf-8"))
    dict_subs.update(params_subs)  # Override with any provided substitutions

    subs = [(key, value) for key, value in dict_subs.items()]
    subjFolders.extend(subs)  # Add parameter substitutions to folders

    datasink.inputs.substitutions = subjFolders

    # Load regex-based substitutions from the 'regex_subs.json' file
    json_regex_subs = op.join(
        op.dirname(op.abspath(__file__)), "regex_subs.json"
    )
    dict_regex_subs = json.load(open(json_regex_subs, encoding="utf-8"))

    # Update with provided regex substitutions
    dict_regex_subs.update(params_regex_subs)

    regex_subs = [(key, value) for key, value in dict_regex_subs.items()]
    datasink.inputs.regexp_substitutions = regex_subs

    return datasink

get_gestational_age(bids_dir, T2)

Retrieve the gestational age for a specific subject from a BIDS dataset.

Parameters:

Name	Type	Description	Default
bids_dir		The file path to the root of the BIDS dataset, which must contain a 'participants.tsv' file.	required
T2		The path of the image. We can get the subject id from there if it follows a BIDS format.	required

Returns: gestational_age : The gestational age of the subject.

Source code in fetpype/utils/utils_bids.py

def get_gestational_age(bids_dir, T2):
    """
    Retrieve the gestational age for a specific subject from a BIDS dataset.

    Args:
        bids_dir : The file path to the root of the BIDS dataset,
            which must contain a 'participants.tsv' file.
        T2 : The path of the image. We can get the subject id from there if
            it follows a BIDS format.
    Returns:
        gestational_age : The gestational age of the subject.

    """
    import pandas as pd
    import os

    participants_path = f"{bids_dir}/participants.tsv"

    try:
        df = pd.read_csv(participants_path, delimiter="\t")
    except FileNotFoundError:
        raise FileNotFoundError(f"participants.tsv not found in {bids_dir}")

    # TODO This T2[0] not really clean
    subject_id = os.path.basename(T2).split("_")[0]
    try:
        gestational_age = df.loc[
            df["participant_id"] == f"{subject_id}", "gestational_age"
        ].values[0]
    except KeyError:
        raise KeyError(
            "Column 'gestational_age' not found in participants.tsv"
        )
    except IndexError:
        raise IndexError(
            f"Subject sub-{subject_id} not found in participants.tsv"
        )

    return gestational_age

create_description_file(out_dir, algo, prev_desc=None, cfg=None)

Create a dataset_description.json file in the derivatives folder. TODO: should look for the extra parameters and also add them

Parameters:

Name	Type	Description	Default
args		Dictionary containing the arguments passed to the script.	required
container_type		Type of container used to run the algorithm.	required

Source code in fetpype/utils/utils_bids.py

def create_description_file(out_dir, algo, prev_desc=None, cfg=None):
    """Create a dataset_description.json file in the derivatives folder.
    TODO: should look for the extra parameters and also add them

    Args:
        args : Dictionary containing the arguments passed to the script.
        container_type : Type of container used to run the algorithm.
    """
    if not os.path.exists(os.path.join(out_dir, "dataset_description.json")):
        description = {
            "Name": algo,
            "Version": "1.0",
            "BIDSVersion": "1.7.0",
            "PipelineDescription": {
                "Name": algo,
            },
            "GeneratedBy": [
                {
                    "Name": algo,
                }
            ],
        }

        if prev_desc is not None:
            with open(prev_desc, "r") as f:
                prev_desc = json.load(f)
                description["GeneratedBy"].append({"Name": prev_desc["Name"]})
        if cfg is not None:
            description["Config"] = OmegaConf.to_container(cfg, resolve=True)
        with open(
            os.path.join(
                out_dir,
                "dataset_description.json",
            ),
            "w",
            encoding="utf-8",
        ) as outfile:
            json.dump(description, outfile, indent=4)

utils_docker

flatten_cfg(cfg, base='')

Flatten a nested configuration dictionary into a flat dictionary with keys as paths and values as the corresponding values. Args: cfg (dict): The configuration dictionary to flatten. base (str): The base path to prepend to the keys. Returns: generator: A generator that yields tuples of (path, value).

Source code in fetpype/utils/utils_docker.py

def flatten_cfg(cfg, base=""):
    """
    Flatten a nested configuration dictionary into a flat dictionary
    with keys as paths and values as the corresponding values.
    Args:
        cfg (dict): The configuration dictionary to flatten.
        base (str): The base path to prepend to the keys.
    Returns:
        generator: A generator that yields tuples of (path, value).
    """
    for k, v in cfg.items():
        if isinstance(v, dict):
            yield from flatten_cfg(v, "/".join([base, k]))
        else:
            yield ("/".join([base, k]), v)

is_available_container(container_type, container_name)

Check if the container is available on the system. Args: container_type (str): The type of container, either 'docker' or 'singularity' container_name (str): The name of the container to check. Returns: bool: True if the container is available, False otherwise.

Source code in fetpype/utils/utils_docker.py

def is_available_container(container_type, container_name):
    """
    Check if the container is available on the system.
    Args:
        container_type (str):   The type of container, either 'docker'
                                or 'singularity'
        container_name (str): The name of the container to check.
    Returns:
        bool: True if the container is available, False otherwise.
    """
    if container_type == "docker":
        try:
            subprocess.run(
                [container_type, "inspect", container_name],
                check=True,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
            )
        except subprocess.CalledProcessError:
            return False
        else:
            return True
    elif container_type == "singularity":
        if os.path.isfile(container_name):
            return True
        else:
            return False
    else:
        raise ValueError(
            f"Container type {container_type} not supported. "
            "Please use 'docker' or 'singularity'."
        )

retrieve_container(container_type, container_name)

Retrieve the container from the registry.

Args: container_type (str): The type of container, either 'docker' or 'singularity' container_name (str): The name of the container to retrieve.

Source code in fetpype/utils/utils_docker.py

def retrieve_container(container_type, container_name):
    """
        Retrieve the container from the registry.
    Args:
        container_type (str):   The type of container, either 'docker' or
                                'singularity'
        container_name (str): The name of the container to retrieve.

    """
    if container_type == "docker":

        cmd = [container_type, "pull", container_name]
        print(f"Running {' '.join(cmd)}")
        process = subprocess.Popen(
            cmd,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
        )
        for line in process.stdout:
            print(line, end="")

        process.wait()
        if process.returncode != 0:
            for line in process.stderr:
                print(line, end="")
            raise subprocess.CalledProcessError(
                process.returncode,
                cmd,
                output=process.stdout.read(),
                stderr=process.stderr.read(),
            )
    elif container_type == "singularity":
        raise NotImplementedError
    else:
        raise ValueError(
            f"Container type {container_type} not supported. "
            "Please use 'docker' or 'singularity'."
        )

check_container_commands(container_type, cfg)

Check if the required docker or singularity images are available on the system.

Parameters:

Name	Type	Description	Default
container_type	str	The type of container, either 'docker' or 'singularity'	required
cfg	dict	The configuration dictionary containing the container names.	required

Source code in fetpype/utils/utils_docker.py

def check_container_commands(container_type, cfg):
    """
    Check if the required docker or singularity images are available
    on the system.

    Args:
        container_type (str):   The type of container, either 'docker' or
                                'singularity'
        cfg (dict): The configuration dictionary containing the
                    container names.

    """
    # Check if the container_type is valid
    if container_type not in ["docker", "singularity"]:
        raise ValueError(
            f"Container type {container_type} not supported. "
            "Please use 'docker' or 'singularity'."
        )

    # Iterate the nested config dictionary
    cfg_dict = dict(flatten_cfg(cfg))
    container_names = {}
    for k, v in cfg_dict.items():
        # Return the word that is after the <mount> tag in the string v
        if container_type == "docker" and "docker" in k:
            docker_name = v.split("<mount>")[-1].split()[0]
            container_names[k] = docker_name
        elif container_type == "singularity" and "singularity" in k:
            # Find a string that ends with .sif
            if v is None:
                continue
            print("CHECKING PATH IN", k, v)
            singularity_name = [s for s in v.split(" ") if s.endswith(".sif")][
                0
            ]
            container_names[k] = singularity_name

    container_names_list = defaultdict(list)
    for k, v in container_names.items():
        container_names_list[v].append(k)

    # Check which containers are missing
    missing_containers = []
    for k, v in container_names_list.items():
        print(f"Checking {container_type} {k} -- Used by {', '.join(v)}")
        if not is_available_container(container_type, k):
            print("\tSTATUS: NOT FOUND")
            missing_containers.append(k)
        else:
            print("\tSTATUS: AVAILABLE")

    # Retrieve the missing containers
    if len(missing_containers) > 0 and container_type == "docker":
        var = input(
            f"Would you like me to retrieve the missing containers "
            f"{', '.join(missing_containers)}? (y/n) "
        )
        if var == "y":
            for k in missing_containers:
                retrieve_container(container_type, k)

        else:
            print("Exiting...")
            sys.exit(1)
    elif len(missing_containers) > 0 and container_type == "singularity":
        raise NotImplementedError(
            "Automated container retrieval for singularity is "
            "not implemented yet."
        )
        sys.exit(1)

logging

StdToLogger

Redirect a stream (stdout/stderr) into a logger (line-buffered).

Parameters:

Name	Type	Description	Default
logger	Logger	The logger to which the stream will be redirected.	required
level	int	The logging level (e.g., logging.INFO, logging.ERROR).	required

Source code in fetpype/utils/logging.py

class StdToLogger:
    """Redirect a stream (stdout/stderr) into a logger (line-buffered).

    Args:
        logger (logging.Logger): The logger to which the stream will
            be redirected.
        level (int): The logging level (e.g., logging.INFO, logging.ERROR).

    """

    def __init__(self, logger, level):
        self.logger = logger
        self.level = level
        self._buf = ""

    def write(self, s):
        if not s:
            return
        s = s.replace("\r\n", "\n")
        parts = (self._buf + s).split("\n")
        for line in parts[:-1]:
            if line:
                self.logger.log(self.level, line)
        self._buf = parts[-1]

    def flush(self):
        if self._buf:
            self.logger.log(self.level, self._buf)
            self._buf = ""

setup_logging(base_dir, debug=False, verbose=False, capture_prints=True, container_logger_name='nipype.container')

Set up logging for the Nipype workflow. Ensures the possibility of limited console output while providing detailed logging to a file. Args: base_dir (str): The base directory for the workflow. debug (bool): Enable debug logging. Logging to will be at the DEBUG level. verbose (bool): Enable verbose logging. Logging level to console will be set to INFO if debug is False and to DEBUG if debug is True. capture_prints (bool): Capture print statements. container_logger_name (str): The name of the container logger.

Source code in fetpype/utils/logging.py

def setup_logging(
    base_dir,
    debug=False,
    verbose=False,
    capture_prints=True,
    container_logger_name="nipype.container",
):
    """
    Set up logging for the Nipype workflow.
    Ensures the possibility of limited console output
    while providing detailed logging to a file.
    Args:
        base_dir (str): The base directory for the workflow.
        debug (bool): Enable debug logging.
            Logging to <log_file> will be at the DEBUG level.
        verbose (bool): Enable verbose logging. Logging level to
            console will be set to INFO if `debug` is `False` and
            to DEBUG if `debug` is `True`.
        capture_prints (bool): Capture print statements.
        container_logger_name (str): The name of the container logger.
    """
    log_dir = os.path.join(base_dir, "logs", time.strftime("%Y%m%d-%H%M%S"))
    os.makedirs(log_dir, exist_ok=True)
    log_file = os.path.join(log_dir, "pypeline.log")
    try:
        os.remove(log_file)  # start fresh each run
    except FileNotFoundError:
        pass

    file_level = "DEBUG" if debug else "INFO"

    # Effective console level when verbose
    to_console = (
        "DEBUG" if (verbose and debug) else "INFO" if verbose else "ERROR"
    )

    # Tell Nipype to log to file
    config.update_config(
        {
            "logging": {
                "log_to_file": True,
                "log_directory": log_dir,
                "log_size": str(50 * 1024 * 1024),
                "log_rotate": "5",
                "workflow_level": file_level,
                "interface_level": file_level,
                "utils_level": file_level,
            },
            "execution": {
                "crashdump_dir": log_dir,  # ← fixed key name
                "crashfile_format": "txt",
            },
        }
    )
    nlogging.update_logging(config)

    # Parent 'nipype' console handler → quiet or verbose
    nipype_root = logging.getLogger("nipype")
    for h in list(nipype_root.handlers):
        if isinstance(h, logging.StreamHandler) and not isinstance(
            h, logging.FileHandler
        ):
            h.setLevel(getattr(logging, to_console))
            h.setFormatter(logging.Formatter("%(message)s"))
            h.stream = sys.__stdout__

    # Plain-lines logger to the SAME file (no header per line)
    container_log = logging.getLogger(container_logger_name)
    container_log.setLevel(getattr(logging, file_level))
    container_log.propagate = False  # don't bubble into 'nipype' handlers

    # Ensure we have exactly one rotating file handler to pypeline.log
    if not any(
        isinstance(h, RFH) and getattr(h, "baseFilename", None) == log_file
        for h in container_log.handlers
    ):
        ch = RFH(log_file, maxBytes=50 * 1024 * 1024, backupCount=5)
        ch.setLevel(getattr(logging, file_level))
        ch.setFormatter(logging.Formatter("%(message)s"))  # no header
        container_log.addHandler(ch)

    # In verbose mode, also mirror container lines to the console
    if verbose and not any(
        isinstance(h, logging.StreamHandler) for h in container_log.handlers
    ):
        sh = logging.StreamHandler(stream=sys.__stdout__)
        sh.setLevel(getattr(logging, to_console))  # INFO or DEBUG
        sh.setFormatter(logging.Formatter("%(message)s"))
        container_log.addHandler(sh)

    # Route print()/tracebacks to the Nipype workflow logger (file) if desired
    if capture_prints:
        nipype_workflow_log = logging.getLogger("nipype.workflow")
        nipype_workflow_log.propagate = True
        sys.stdout = StdToLogger(nipype_workflow_log, logging.INFO)
        sys.stderr = StdToLogger(nipype_workflow_log, logging.ERROR)

    # Where Nipype interfaces send their stdout/stderr
    # - verbose=True  → stream to terminal
    # - verbose=False → save to stdout.nipype / stderr.nipype in node workdirs
    CommandLine.set_default_terminal_output("stream" if verbose else "file")

    logging.captureWarnings(True)

    def _excepthook(exc_type, exc, tb):
        logging.getLogger("nipype.workflow").error(
            "Uncaught exception", exc_info=(exc_type, exc, tb)
        )
        sys.__excepthook__(exc_type, exc, tb)

    sys.excepthook = _excepthook

run_and_tee(cmd, *, prefix='')

Run a command, stream output live to terminal, and log every line. Returns the full combined output; raises RuntimeError on non-zero exit.

Parameters:

Name	Type	Description	Default
cmd	str	The command to run.	required
prefix	str	A prefix to add to each line of output.	''

Returns:

Name	Type	Description
str		The combined output of the command.

Source code in fetpype/utils/logging.py

def run_and_tee(cmd, *, prefix=""):
    """
    Run a command, stream output live to terminal, and log every line.
    Returns the full combined output; raises RuntimeError on non-zero exit.

    Args:
        cmd (str): The command to run.
        prefix (str): A prefix to add to each line of output.

    Returns:
        str: The combined output of the command.
    """

    log_plain = logging.getLogger("nipype.container")  # message-only
    log_evt = logging.getLogger("nipype.workflow")  # structured events

    env = os.environ.copy()
    env.setdefault("PYTHONUNBUFFERED", "1")  # flush Python in container
    env.setdefault("PYTHONIOENCODING", "utf-8")
    env.setdefault("TQDM_DISABLE", "1")  # avoid CR-based progress bars

    proc = subprocess.Popen(
        cmd,
        shell=True,  # keep if you're passing a single string
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,  # merge stderr -> stdout
        text=True,
        bufsize=1,  # line-buffered
        env=env,
    )

    log_evt.info("Running: %s", cmd)  # one structured line

    captured = []
    for line in proc.stdout:
        line = line.rstrip("\n")
        captured.append(line)
        log_plain.info("%s%s", prefix, line)  # file: plain line
        sys.__stdout__.write(prefix + line + "\n")  # console: live
        sys.__stdout__.flush()

    proc.stdout.close()
    rc = proc.wait()
    output = "".join(captured)

    if rc != 0:
        raise RuntimeError(
            f"Docker call failed with exit code {rc}.\n"
            f"Command: {cmd}\n"
            f"Output:\n{output.strip() or '<<no output>>'}"
        )

    return output