Repository: PennLINC/xcp_d

Issue:
Error in finding parcellation atlases when running pipeline in parallel
## Summary
<!--What is the nature of the bug?-->

Now the output atlases folder is moved out of the subject directory, which is very reasonable. But for some unknown reason, when running pipeline for several subjects in parallel, some sub-tasks will result in an error in finding corresponding parcellation atlas file. The following is crash report:

```
Node: xcpd_wf.single_subject_SICNU028_wf.cifti_postprocess_11_wf.connectivity_wf.parcellate_reho
Working directory: /seastor/CAMP/tmp/xcpd_wf/single_subject_SICNU028_wf/cifti_postprocess_11_wf/connectivity_wf/parcellate_reho

Node inputs:

atlas = ['/seastor/CAMP/derivatives/xcpd_no_gsr/xcp_d/atlases/atlas-4S1056Parcels/space-fsLR_atlas-4S1056Parcels_den-91k_dseg.dlabel.nii', '/seastor/CAMP/derivatives/xcpd_no_gsr/xcp_d/atlases/atlas-4S156Parcels/space-fsLR_atlas-4S156Parcels_den-91k_dseg.dlabel.nii', '/seastor/CAMP/derivatives/xcpd_no_gsr/xcp_d/atlases/atlas-4S256Parcels/space-fsLR_atlas-4S256Parcels_den-91k_dseg.dlabel.nii', '/seastor/CAMP/derivatives/xcpd_no_gsr/xcp_d/atlases/atlas-4S356Parcels/space-fsLR_atlas-4S356Parcels_den-91k_dseg.dlabel.nii', '/seastor/CAMP/derivatives/xcpd_no_gsr/xcp_d/atlases/atlas-4S456Parcels/space-fsLR_atlas-4S456Parcels_den-91k_dseg.dlabel.nii', '/seastor/CAMP/derivatives/xcpd_no_gsr/xcp_d/atlases/atlas-4S556Parcels/space-fsLR_atlas-4S556Parcels_den-91k_dseg.dlabel.nii', '/seastor/CAMP/derivatives/xcpd_no_gsr/xcp_d/atlases/atlas-4S656Parcels/space-fsLR_atlas-4S656Parcels_den-91k_dseg.dlabel.nii', '/seastor/CAMP/derivatives/xcpd_no_gsr/xcp_d/atlases/atlas-4S756Parcels/space-fsLR_atlas-4S756Parcels_den-91k_dseg.dlabel.nii', '/seastor/CAMP/derivatives/xcpd_no_gsr/xcp_d/atlases/atlas-4S856Parcels/space-fsLR_atlas-4S856Parcels_den-91k_dseg.dlabel.nii', '/seastor/CAMP/derivatives/xcpd_no_gsr/xcp_d/atlases/atlas-4S956Parcels/space-fsLR_atlas-4S956Parcels_den-91k_dseg.dlabel.nii', '/seastor/CAMP/derivatives/xcpd_no_gsr/xcp_d/atlases/atlas-Glasser/space-fsLR_atlas-Glasser_den-91k_dseg.dlabel.nii', '/seastor/CAMP/derivatives/xcpd_no_gsr/xcp_d/atlases/atlas-Gordon/space-fsLR_atlas-Gordon_den-91k_dseg.dlabel.nii', '/seastor/CAMP/derivatives/xcpd_no_gsr/xcp_d/atlases/atlas-HCP/space-fsLR_atlas-HCP_den-91k_dseg.dlabel.nii', '/seastor/CAMP/derivatives/xcpd_no_gsr/xcp_d/atlases/atlas-Tian/space-fsLR_atlas-Tian_den-91k_dseg.dlabel.nii']
atlas_labels = <undefined>
data_file = /seastor/CAMP/tmp/xcpd_wf/single_subject_SICNU028_wf/cifti_postprocess_11_wf/reho_wf/merge_cifti/reho_converted.dscalar.nii
min_coverage = 0.0
parcellated_atlas = <undefined>

Traceback (most recent call last):
  File "/usr/local/miniconda/lib/python3.10/site-packages/nipype/pipeline/plugins/multiproc.py", line 292, in _send_procs_to_workers
    num_subnodes = self.procs[jobid].num_subnodes()
  File "/usr/local/miniconda/lib/python3.10/site-packages/nipype/pipeline/engine/nodes.py", line 1308, in num_subnodes
    self._get_inputs()
  File "/usr/local/miniconda/lib/python3.10/site-packages/nipype/pipeline/engine/nodes.py", line 1321, in _get_inputs
    self._inputs.trait_set(**old_inputs)
  File "/usr/local/miniconda/lib/python3.10/site-packages/traits/has_traits.py", line 1520, in trait_set
    setattr(self, name, value)
  File "/usr/local/miniconda/lib/python3.10/site-packages/nipype/interfaces/base/traits_extension.py", line 424, in validate
    value = super(MultiObject, self).validate(objekt, name, newvalue)
  File "/usr/local/miniconda/lib/python3.10/site-packages/traits/trait_types.py", line 2699, in validate
    return TraitListObject(self, object, name, value)
  File "/usr/local/miniconda/lib/python3.10/site-packages/traits/trait_list_object.py", line 582, in __init__
    super().__init__(
  File "/usr/local/miniconda/lib/python3.10/site-packages/traits/trait_list_object.py", line 213, in __init__
    super().__init__(self.item_validator(item) for item in iterable)
  File "/usr/local/miniconda/lib/python3.10/site-packages/traits/trait_list_object.py", line 213, in <genexpr>
    super().__init__(self.item_validator(item) for item in iterable)
  File "/usr/local/miniconda/lib/python3.10/site-packages/traits/trait_list_object.py", line 865, in _item_validator
    return trait_validator(object, self.name, value)
  File "/usr/local/miniconda/lib/python3.10/site-packages/nipype/interfaces/base/traits_extension.py", line 330, in validate
    value = super(File, self).validate(objekt, name, value, return_pathlike=True)
  File "/usr/local/miniconda/lib/python3.10/site-packages/nipype/interfaces/base/traits_extension.py", line 135, in validate
    self.error(objekt, name, str(value))
  File "/usr/local/miniconda/lib/python3.10/site-packages/traits/base_trait_handler.py", line 74, in error
    raise TraitError(
traits.trait_errors.TraitError: Each element of the 'atlas_labels' trait of a DynamicTraitedSpec instance must be a pathlike object or string representing an existing file, but a value of '/seastor/CAMP/derivatives/xcpd_no_gsr/xcp_d/atlases/atlas-4S1056Parcels/atlas-4S1056Parcels_dseg.tsv' <class 'str'> was specified.


When creating this crashfile, the results file corresponding
to the node could not be found.
```

## Additional details
<!--Please fill in the following details-->
- xcp_d version: 0.6.1
- Docker version: N/A
- Singularity version: 3.7.0

### What were you trying to do?

Running several subjects in parallel.

### What did you expect to happen?

Atlases should be okay for all subjects.

### What actually happened?

Some subject pipeline cannot find atlas file.

## Reproducing the bug
<!--Please share any steps you performed that revealed the bug-->
<!--Please include any code snippets.
Enclose them in triple back-ticks (```)
Like this:

```
<code>
```
-->

None. The pipeline is set up as the default setting.

Discussion:

Patch:
diff --git a/xcp_d/utils/atlas.py b/xcp_d/utils/atlas.py
index 3e39d5a..561973b 100644
--- a/xcp_d/utils/atlas.py
+++ b/xcp_d/utils/atlas.py
@@ -160,6 +160,24 @@ def get_atlas_cifti(atlas):
 def copy_atlas(name_source, in_file, output_dir, atlas):
     """Copy atlas file to output directory.
 
+    Parameters
+    ----------
+    name_source : :obj:`str`
+        The source name of the atlas file.
+    in_file : :obj:`str`
+        The atlas file to copy.
+    output_dir : :obj:`str`
+        The output directory.
+    atlas : :obj:`str`
+        The name of the atlas.
+
+    Returns
+    -------
+    out_file : :obj:`str`
+        The path to the copied atlas file.
+
+    Notes
+    -----
     I can't use DerivativesDataSink because it has a problem with dlabel CIFTI files.
     It gives the following error:
     "AttributeError: 'Cifti2Header' object has no attribute 'set_data_dtype'"
@@ -167,31 +185,38 @@ def copy_atlas(name_source, in_file, output_dir, atlas):
     I can't override the CIFTI atlas's data dtype ahead of time because setting it to int8 or int16
     somehow converts all of the values in the data array to weird floats.
     This could be a version-specific nibabel issue.
+
+    I've also updated this function to handle JSON and TSV files as well.
     """
     import os
     import shutil
 
     from xcp_d.utils.bids import get_entity
 
-    extension = ".nii.gz" if name_source.endswith(".nii.gz") else ".dlabel.nii"
-    space = get_entity(name_source, "space")
-    res = get_entity(name_source, "res")
-    den = get_entity(name_source, "den")
-    cohort = get_entity(name_source, "cohort")
-
-    cohort_str = f"_cohort-{cohort}" if cohort else ""
-    res_str = f"_res-{res}" if res else ""
-    den_str = f"_den-{den}" if den else ""
-    if extension == ".dlabel.nii":
-        atlas_basename = f"space-{space}_atlas-{atlas}{den_str}{cohort_str}_dseg{extension}"
-    elif extension == ".nii.gz":
-        atlas_basename = f"space-{space}_atlas-{atlas}{res_str}{cohort_str}_dseg{extension}"
+    if in_file.endswith(".json"):
+        out_basename = f"atlas-{atlas}_dseg.json"
+    elif in_file.endswith(".tsv"):
+        out_basename = f"atlas-{atlas}_dseg.tsv"
+    else:
+        extension = ".nii.gz" if name_source.endswith(".nii.gz") else ".dlabel.nii"
+        space = get_entity(name_source, "space")
+        res = get_entity(name_source, "res")
+        den = get_entity(name_source, "den")
+        cohort = get_entity(name_source, "cohort")
+
+        cohort_str = f"_cohort-{cohort}" if cohort else ""
+        res_str = f"_res-{res}" if res else ""
+        den_str = f"_den-{den}" if den else ""
+        if extension == ".dlabel.nii":
+            out_basename = f"space-{space}_atlas-{atlas}{den_str}{cohort_str}_dseg{extension}"
+        elif extension == ".nii.gz":
+            out_basename = f"space-{space}_atlas-{atlas}{res_str}{cohort_str}_dseg{extension}"
 
     atlas_out_dir = os.path.join(output_dir, f"xcp_d/atlases/atlas-{atlas}")
     os.makedirs(atlas_out_dir, exist_ok=True)
-    out_atlas_file = os.path.join(atlas_out_dir, atlas_basename)
+    out_file = os.path.join(atlas_out_dir, out_basename)
     # Don't copy the file if it exists, to prevent any race conditions between parallel processes.
-    if not os.path.isfile(out_atlas_file):
-        shutil.copyfile(in_file, out_atlas_file)
+    if not os.path.isfile(out_file):
+        shutil.copyfile(in_file, out_file)
 
-    return out_atlas_file
+    return out_file
diff --git a/xcp_d/workflows/connectivity.py b/xcp_d/workflows/connectivity.py
index 005349f..0535d49 100644
--- a/xcp_d/workflows/connectivity.py
+++ b/xcp_d/workflows/connectivity.py
@@ -127,11 +127,9 @@ def init_load_atlases_wf(
             name="get_transforms_to_bold_space",
         )
 
-        # fmt:off
         workflow.connect([
             (inputnode, get_transforms_to_bold_space, [("name_source", "bold_file")]),
-        ])
-        # fmt:on
+        ])  # fmt:skip
 
         # ApplyTransforms needs a 3D image for the reference image.
         grab_first_volume = pe.Node(
@@ -154,7 +152,6 @@ def init_load_atlases_wf(
             n_procs=omp_nthreads,
         )
 
-        # fmt:off
         workflow.connect([
             (grab_first_volume, warp_atlases_to_bold_space, [("out_file", "reference_image")]),
             (atlas_file_grabber, warp_atlases_to_bold_space, [("atlas_file", "input_image")]),
@@ -162,8 +159,7 @@ def init_load_atlases_wf(
                 ("transformfile", "transforms"),
             ]),
             (warp_atlases_to_bold_space, atlas_buffer, [("output_image", "atlas_file")]),
-        ])
-        # fmt:on
+        ])  # fmt:skip
 
     else:
         # Add empty vertices to atlas for locations in data, but not in atlas
@@ -175,13 +171,11 @@ def init_load_atlases_wf(
             iterfield=["label"],
         )
 
-        # fmt:off
         workflow.connect([
             (inputnode, resample_atlas_to_data, [("bold_file", "template_cifti")]),
             (atlas_file_grabber, resample_atlas_to_data, [("atlas_file", "label")]),
             (resample_atlas_to_data, atlas_buffer, [("cifti_out", "atlas_file")]),
-        ])
-        # fmt:on
+        ])  # fmt:skip
 
         # Change the atlas to a scalar file.
         convert_to_dscalar = pe.MapNode(
@@ -195,11 +189,9 @@ def init_load_atlases_wf(
             n_procs=omp_nthreads,
             iterfield=["data_cifti"],
         )
-        # fmt:off
         workflow.connect([
             (resample_atlas_to_data, convert_to_dscalar, [("cifti_out", "data_cifti")]),
-        ])
-        # fmt:on
+        ])  # fmt:skip
 
         # Convert atlas from dlabel to pscalar format.
         # The pscalar version of the atlas is later used for its ParcelAxis.
@@ -215,13 +207,11 @@ def init_load_atlases_wf(
             iterfield=["in_file", "atlas_label"],
         )
 
-        # fmt:off
         workflow.connect([
             (atlas_file_grabber, parcellate_atlas, [("atlas_file", "atlas_label")]),
             (convert_to_dscalar, parcellate_atlas, [("cifti_out", "in_file")]),
             (parcellate_atlas, outputnode, [("out_file", "parcellated_atlas_files")]),
-        ])
-        # fmt:on
+        ])  # fmt:skip
 
     ds_atlas = pe.MapNode(
         Function(
@@ -241,80 +231,58 @@ def init_load_atlases_wf(
     ds_atlas.inputs.output_dir = output_dir
     ds_atlas.inputs.atlas = atlases
 
-    # fmt:off
     workflow.connect([
         (inputnode, ds_atlas, [("name_source", "name_source")]),
         (atlas_buffer, ds_atlas, [("atlas_file", "in_file")]),
         (ds_atlas, outputnode, [("out_file", "atlas_files")]),
-    ])
-    # fmt:on
+    ])  # fmt:skip
 
     ds_atlas_labels_file = pe.MapNode(
-        DerivativesDataSink(
-            base_directory=output_dir,
-            check_hdr=False,
-            dismiss_entities=[
-                "datatype",
-                "subject",
-                "session",
-                "task",
-                "run",
-                "desc",
-                "space",
-                "res",
-                "den",
-                "cohort",
+        Function(
+            function=copy_atlas,
+            input_names=[
+                "name_source",
+                "in_file",
+                "output_dir",
+                "atlas",
             ],
-            allowed_entities=["atlas"],
-            suffix="dseg",
-            extension=".tsv",
+            output_names=["out_file"],
         ),
         name="ds_atlas_labels_file",
-        iterfield=["atlas", "in_file"],
+        iterfield=["in_file", "atlas"],
         run_without_submitting=True,
     )
+    ds_atlas_labels_file.inputs.output_dir = output_dir
     ds_atlas_labels_file.inputs.atlas = atlases
 
-    # fmt:off
     workflow.connect([
-        (inputnode, ds_atlas_labels_file, [("name_source", "source_file")]),
+        (inputnode, ds_atlas_labels_file, [("name_source", "name_source")]),
         (atlas_file_grabber, ds_atlas_labels_file, [("atlas_labels_file", "in_file")]),
         (ds_atlas_labels_file, outputnode, [("out_file", "atlas_labels_files")]),
-    ])
-    # fmt:on
+    ])  # fmt:skip
 
     ds_atlas_metadata = pe.MapNode(
-        DerivativesDataSink(
-            base_directory=output_dir,
-            check_hdr=False,
-            dismiss_entities=[
-                "datatype",
-                "subject",
-                "session",
-                "task",
-                "run",
-                "desc",
-                "space",
-                "res",
-                "den",
-                "cohort",
+        Function(
+            function=copy_atlas,
+            input_names=[
+                "name_source",
+                "in_file",
+                "output_dir",
+                "atlas",
             ],
-            allowed_entities=["atlas"],
-            suffix="dseg",
-            extension=".json",
+            output_names=["out_file"],
         ),
         name="ds_atlas_metadata",
-        iterfield=["atlas", "in_file"],
+        iterfield=["in_file", "atlas"],
         run_without_submitting=True,
     )
+    ds_atlas_metadata.inputs.output_dir = output_dir
     ds_atlas_metadata.inputs.atlas = atlases
 
-    # fmt:off
     workflow.connect([
-        (inputnode, ds_atlas_metadata, [("name_source", "source_file")]),
+        (inputnode, ds_atlas_metadata, [("name_source", "name_source")]),
         (atlas_file_grabber, ds_atlas_metadata, [("atlas_metadata_file", "in_file")]),
-    ])
-    # fmt:on
+    ])  # fmt:skip
 
     return workflow
 


Test patch:
diff --git a/xcp_d/tests/test_utils_atlas.py b/xcp_d/tests/test_utils_atlas.py
index 32a2527..43ffdf7 100644
--- a/xcp_d/tests/test_utils_atlas.py
+++ b/xcp_d/tests/test_utils_atlas.py
@@ -62,10 +62,40 @@ def test_copy_atlas(tmp_path_factory):
     assert os.path.basename(out_file) == "space-MNI152NLin2009cAsym_atlas-Y_res-2_dseg.nii.gz"
 
     # CIFTI
-    atlas_file, _, _ = atlas.get_atlas_cifti("Gordon")
+    atlas_file, atlas_labels_file, atlas_metadata_file = atlas.get_atlas_cifti("Gordon")
     name_source = "sub-01_task-imagery_run-01_space-fsLR_den-91k_desc-denoised_bold.dtseries.nii"
     out_file = atlas.copy_atlas(
         name_source=name_source, in_file=atlas_file, output_dir=tmpdir, atlas="Y"
     )
     assert os.path.isfile(out_file)
     assert os.path.basename(out_file) == "space-fsLR_atlas-Y_den-91k_dseg.dlabel.nii"
+
+    # TSV
+    name_source = "sub-01_task-imagery_run-01_space-fsLR_den-91k_desc-denoised_bold.dtseries.nii"
+    out_file = atlas.copy_atlas(
+        name_source=name_source, in_file=atlas_labels_file, output_dir=tmpdir, atlas="Y"
+    )
+    assert os.path.isfile(out_file)
+    assert os.path.basename(out_file) == "atlas-Y_dseg.tsv"
+
+    # JSON
+    name_source = "sub-01_task-imagery_run-01_space-fsLR_den-91k_desc-denoised_bold.dtseries.nii"
+    out_file = atlas.copy_atlas(
+        name_source=name_source, in_file=atlas_metadata_file, output_dir=tmpdir, atlas="Y"
+    )
+    assert os.path.isfile(out_file)
+    assert os.path.basename(out_file) == "atlas-Y_dseg.json"
+
+    # Ensure that out_file isn't overwritten if it already exists
+    fake_in_file = os.path.join(tmpdir, "fake.json")
+    with open(fake_in_file, "w") as fo:
+        fo.write("fake")
+
+    out_file = atlas.copy_atlas(
+        name_source=name_source, in_file=fake_in_file, output_dir=tmpdir, atlas="Y"
+    )
+    assert os.path.isfile(out_file)
+    assert os.path.basename(out_file) == "atlas-Y_dseg.json"
+    # The file should not be overwritten, so the contents shouldn't be "fake"
+    with open(out_file, "r") as fo:
+        assert fo.read() != "fake"