Support MCRIBS derivatives (PennLINC#1029)

xcp_d/__main__.py

@@ -1,4 +1,5 @@
 """A method for calling the command-line interface."""
+
 from xcp_d.cli.run import main
 
 if __name__ == "__main__":

xcp_d/cli/parser_utils.py

@@ -1,4 +1,5 @@
 """Utility functions for xcp_d command-line interfaces."""
+
 import argparse
 import json
 import logging

xcp_d/ingression/__init__.py

@@ -1,4 +1,5 @@
 """Tools for converting derivatives from various pipelines to an fMRIPrep-like format."""
+
 from xcp_d.ingression import abcdbids, hcpya, ukbiobank, utils
 
 __all__ = [

xcp_d/ingression/ukbiobank.py

@@ -1,4 +1,5 @@
 """Functions to convert preprocessed UK Biobank BOLD data to BIDS derivatives format."""
+
 import glob
 import json
 import os

xcp_d/interfaces/ants.py

@@ -1,4 +1,5 @@
 """ANTS interfaces."""
+
 import logging
 import os
 

xcp_d/interfaces/bids.py

@@ -1,9 +1,18 @@
 """Adapted interfaces from Niworkflows."""
+
 from json import loads
 from pathlib import Path
 
 from bids.layout import Config
 from nipype import logging
+from nipype.interfaces.base import (
+    BaseInterfaceInputSpec,
+    Directory,
+    File,
+    SimpleInterface,
+    TraitedSpec,
+    traits,
+)
 from niworkflows.interfaces.bids import DerivativesDataSink as BaseDerivativesDataSink
 from pkg_resources import resource_filename as pkgrf
 
@@ -33,3 +42,151 @@ class DerivativesDataSink(BaseDerivativesDataSink):
     _config_entities = config_entities
     _config_entities_dict = merged_entities
     _file_patterns = xcp_d_spec["default_path_patterns"]
+
+
+class _CollectRegistrationFilesInputSpec(BaseInterfaceInputSpec):
+    segmentation_dir = Directory(
+        exists=True,
+        required=True,
+        desc="Path to FreeSurfer or MCRIBS derivatives.",
+    )
+    software = traits.Enum(
+        "FreeSurfer",
+        "MCRIBS",
+        required=True,
+        desc="The software used for segmentation.",
+    )
+    hemisphere = traits.Enum(
+        "L",
+        "R",
+        required=True,
+        desc="The hemisphere being used.",
+    )
+    participant_id = traits.Str(
+        required=True,
+        desc="Participant ID. Used to select the subdirectory of the FreeSurfer derivatives.",
+    )
+
+
+class _CollectRegistrationFilesOutputSpec(TraitedSpec):
+    subject_sphere = File(
+        exists=True,
+        desc="Subject-space sphere.",
+    )
+    source_sphere = File(
+        exists=True,
+        desc="Source-space sphere (namely, fsaverage).",
+    )
+    target_sphere = File(
+        exists=True,
+        desc="Target-space sphere (fsLR for FreeSurfer, dHCP-in-fsLR for MCRIBS).",
+    )
+    sphere_to_sphere = File(
+        exists=True,
+        desc="Warp file going from source space to target space.",
+    )
+
+
+class CollectRegistrationFiles(SimpleInterface):
+    """Collect registration files for fsnative-to-fsLR transformation."""
+
+    input_spec = _CollectRegistrationFilesInputSpec
+    output_spec = _CollectRegistrationFilesOutputSpec
+
+    def _run_interface(self, runtime):
+        import os
+
+        from pkg_resources import resource_filename as pkgrf
+        from templateflow.api import get as get_template
+
+        hemisphere = self.inputs.hemisphere
+        hstr = f"{hemisphere.lower()}h"
+        participant_id = self.inputs.participant_id
+        if not participant_id.startswith("sub-"):
+            participant_id = f"sub-{participant_id}"
+
+        if self.inputs.software == "FreeSurfer":
+            # Find the subject's sphere in the FreeSurfer derivatives.
+            # TODO: Collect from the preprocessing derivatives if they're a compliant version.
+            # Namely, fMRIPrep >= 23.1.2, Nibabies >= 24.0.0a1.
+            self._results["subject_sphere"] = os.path.join(
+                self.inputs.segmentation_dir,
+                participant_id,
+                "surf",
+                f"{hstr}.sphere.reg",
+            )
+
+            # Load the fsaverage-164k sphere
+            # FreeSurfer: tpl-fsaverage_hemi-?_den-164k_sphere.surf.gii
+            self._results["source_sphere"] = str(
+                get_template(
+                    template="fsaverage",
+                    space=None,
+                    hemi=hemisphere,
+                    density="164k",
+                    desc=None,
+                    suffix="sphere",
+                )
+            )
+
+            # TODO: Collect from templateflow once it's uploaded.
+            # FreeSurfer: fs_?/fs_?-to-fs_LR_fsaverage.?_LR.spherical_std.164k_fs_?.surf.gii
+            self._results["sphere_to_sphere"] = pkgrf(
+                "xcp_d",
+                (
+                    f"data/standard_mesh_atlases/fs_{hemisphere}/"
+                    f"fs_{hemisphere}-to-fs_LR_fsaverage.{hemisphere}_LR.spherical_std."
+                    f"164k_fs_{hemisphere}.surf.gii"
+                ),
+            )
+
+            # FreeSurfer: tpl-fsLR_hemi-?_den-32k_sphere.surf.gii
+            self._results["target_sphere"] = str(
+                get_template(
+                    template="fsLR",
+                    space=None,
+                    hemi=hemisphere,
+                    density="32k",
+                    desc=None,
+                    suffix="sphere",
+                )
+            )
+
+        elif self.inputs.software == "MCRIBS":
+            # Find the subject's sphere in the MCRIBS derivatives.
+            # TODO: Collect from the preprocessing derivatives if they're a compliant version.
+            # Namely, fMRIPrep >= 23.1.2, Nibabies >= 24.0.0a1.
+            self._results["subject_sphere"] = os.path.join(
+                self.inputs.segmentation_dir,
+                participant_id,
+                "freesurfer",
+                participant_id,
+                "surf",
+                f"{hstr}.sphere.reg2",
+            )
+
+            # TODO: Collect from templateflow once it's uploaded.
+            # MCRIBS: tpl-fsaverage_hemi-?_den-41k_desc-reg_sphere.surf.gii
+            self._results["source_sphere"] = os.path.join(
+                self.inputs.segmentation_dir,
+                "templates_fsLR",
+                f"tpl-fsaverage_hemi-{hemisphere}_den-41k_desc-reg_sphere.surf.gii",
+            )
+
+            # TODO: Collect from templateflow once it's uploaded.
+            # MCRIBS: tpl-dHCP_space-fsaverage_hemi-?_den-41k_desc-reg_sphere.surf.gii
+            self._results["sphere_to_sphere"] = os.path.join(
+                self.inputs.segmentation_dir,
+                "templates_fsLR",
+                f"tpl-dHCP_space-fsaverage_hemi-{hemisphere}_den-41k_desc-reg_sphere.surf.gii",
+            )
+
+            # TODO: Collect from templateflow once it's uploaded.
+            # MCRIBS: tpl-dHCP_space-fsLR_hemi-?_den-32k_desc-week42_sphere.surf.gii
+            self._results["target_sphere"] = os.path.join(
+                self.inputs.segmentation_dir,
+                "templates_fsLR",
+                f"tpl-dHCP_space-fsLR_hemi-{hemisphere}_den-32k_desc-week42_sphere.surf.gii",
+            )
+
+        return runtime

xcp_d/interfaces/censoring.py

@@ -1,4 +1,5 @@
 """Interfaces for the post-processing workflows."""
+
 import os
 
 import nibabel as nb
@@ -104,9 +105,9 @@ def _run_interface(self, runtime):
         if dummy_scans == 0:
             # write the output out
             self._results["bold_file_dropped_TR"] = self.inputs.bold_file
-            self._results[
-                "fmriprep_confounds_file_dropped_TR"
-            ] = self.inputs.fmriprep_confounds_file
+            self._results["fmriprep_confounds_file_dropped_TR"] = (
+                self.inputs.fmriprep_confounds_file
+            )
             self._results["confounds_file_dropped_TR"] = self.inputs.confounds_file
             self._results["motion_file_dropped_TR"] = self.inputs.motion_file
             self._results["temporal_mask_dropped_TR"] = self.inputs.temporal_mask

xcp_d/interfaces/concatenation.py

@@ -1,4 +1,5 @@
 """Interfaces for the concatenation workflow."""
+
 import itertools
 import os
 import re

xcp_d/interfaces/nilearn.py

@@ -1,4 +1,5 @@
 """Interfaces for Nilearn code."""
+
 import os
 
 from nilearn import maskers

xcp_d/interfaces/utils.py

@@ -1,4 +1,5 @@
 """Miscellaneous utility interfaces."""
+
 from nipype import logging
 from nipype.interfaces.base import (
     BaseInterfaceInputSpec,

xcp_d/interfaces/workbench.py

@@ -1,4 +1,5 @@
 """Custom wb_command interfaces."""
+
 import os
 
 import nibabel as nb

xcp_d/tests/conftest.py

@@ -1,4 +1,5 @@
 """Fixtures for the CircleCI tests."""
+
 import base64
 import os
 

xcp_d/tests/test_TR.py

@@ -5,6 +5,7 @@
 Arguments have to be passed to these functions because the data may be
 mounted in a container somewhere unintuitively.
 """
+
 import os.path as op
 
 import nibabel as nb

xcp_d/tests/test_cli.py

@@ -1,4 +1,5 @@
 """Command-line interface tests."""
+
 import os
 import shutil
 

xcp_d/tests/test_cli_run.py

@@ -1,4 +1,5 @@
 """Tests for functions in the cli.run module."""
+
 import logging
 import os
 from copy import deepcopy

xcp_d/tests/test_cli_utils.py

@@ -1,4 +1,5 @@
 """Tests for the xcp_d.cli.parser_utils module."""
+
 from argparse import ArgumentTypeError
 
 import pytest

xcp_d/tests/test_confounds.py

@@ -1,4 +1,5 @@
 """Test confounds handling."""
+
 import os
 
 import numpy as np

xcp_d/tests/test_despike.py

@@ -6,6 +6,7 @@
 Arguments have to be passed to these functions because the data may be
 mounted in a container somewhere unintuitively.
 """
+
 import os
 
 import nibabel as nb

xcp_d/tests/test_filtering.py

@@ -1,4 +1,5 @@
 """Tests for filtering methods."""
+
 import re
 
 import numpy as np

xcp_d/tests/test_interfaces_censoring.py

@@ -1,4 +1,5 @@
 """Tests for framewise displacement calculation."""
+
 import json
 import os
 

xcp_d/tests/test_interfaces_concatenation.py

@@ -1,4 +1,5 @@
 """Tests for the xcp_d.interfaces.concatenation module."""
+
 import os
 
 from nipype.interfaces.base import Undefined, isdefined

xcp_d/tests/test_interfaces_nilearn.py

@@ -1,4 +1,5 @@
 """Tests for the xcp_d.interfaces.nilearn module."""
+
 import os
 
 import nibabel as nb

xcp_d/tests/test_interfaces_utils.py

@@ -1,4 +1,5 @@
 """Tests for xcp_d.interfaces.utils module."""
+
 import os
 
 import nibabel as nb

xcp_d/tests/test_smoothing.py

@@ -1,4 +1,5 @@
 """Tests for smoothing methods."""
+
 import os
 import re
 import tempfile

xcp_d/tests/test_utils_atlas.py

@@ -1,4 +1,5 @@
 """Tests for the xcp_d.utils.atlas module."""
+
 import os
 
 import pytest

xcp_d/tests/test_utils_bids.py

@@ -1,4 +1,5 @@
 """Tests for the xcp_d.utils.bids module."""
+
 import json
 import os
 
@@ -196,31 +197,25 @@ def test_get_tr(ds001419_data):
 
 
 def test_get_freesurfer_dir(datasets):
-    """Test get_freesurfer_dir and get_freesurfer_sphere."""
-    with pytest.raises(NotADirectoryError, match="No FreeSurfer derivatives found."):
+    """Test get_freesurfer_dir."""
+    with pytest.raises(NotADirectoryError, match="No FreeSurfer/MCRIBS derivatives found"):
         xbids.get_freesurfer_dir(".")
 
-    fs_dir = xbids.get_freesurfer_dir(datasets["nibabies"])
+    fs_dir, software = xbids.get_freesurfer_dir(datasets["nibabies"])
     assert os.path.isdir(fs_dir)
+    assert software == "FreeSurfer"
 
-    # Create fake FreeSurfer folder so there are two possible folders
-    tmp_fs_dir = os.path.join(os.path.dirname(fs_dir), "freesurfer-fail")
-    os.mkdir(tmp_fs_dir)
-    with pytest.raises(ValueError, match="More than one candidate"):
-        xbids.get_freesurfer_dir(datasets["nibabies"])
+    # Create fake FreeSurfer folder so there are two possible folders and it grabs the closest
+    tmp_fs_dir = os.path.join(datasets["nibabies"], "sourcedata/mcribs")
+    os.makedirs(tmp_fs_dir, exist_ok=True)
+    fs_dir, software = xbids.get_freesurfer_dir(datasets["nibabies"])
+    assert os.path.isdir(fs_dir)
+    assert software == "MCRIBS"
     os.rmdir(tmp_fs_dir)
 
-    fs_dir = xbids.get_freesurfer_dir(datasets["pnc"])
+    fs_dir, software = xbids.get_freesurfer_dir(datasets["pnc"])
     assert os.path.isdir(fs_dir)
-
-    sphere_file = xbids.get_freesurfer_sphere(fs_dir, "1648798153", "L")
-    assert os.path.isfile(sphere_file)
-
-    sphere_file = xbids.get_freesurfer_sphere(fs_dir, "sub-1648798153", "L")
-    assert os.path.isfile(sphere_file)
-
-    with pytest.raises(FileNotFoundError, match="Sphere file not found at"):
-        sphere_file = xbids.get_freesurfer_sphere(fs_dir, "fail", "L")
+    assert software == "FreeSurfer"
 
 
 def test_get_entity(datasets):

xcp_d/tests/test_utils_concatenation.py

@@ -1,4 +1,5 @@
 """Tests for the xcp_d.utils.concatenation module."""
+
 import os
 
 import nibabel as nb

xcp_d/tests/test_utils_doc.py

@@ -1,4 +1,5 @@
 """Tests for the xcp_d.utils.doc module."""
+
 import os
 
 from xcp_d.utils import doc

xcp_d/tests/test_utils_execsummary.py

@@ -1,4 +1,5 @@
 """Test functions in xcp_d.utils.execsummary."""
+
 import os
 
 import matplotlib.pyplot as plt