make new nipype interface

Author: oesteban

fmriprep/interfaces/__init__.py

@@ -16,5 +16,5 @@
 from .utils import TPM2ROI, AddTPMs, AddTSVHeader, ConcatAffines, JoinTSVColumns
 from .fmap import FieldEnhance, FieldToRadS, FieldToHz, Phasediff2Fieldmap
 from .confounds import GatherConfounds, ICAConfounds, FMRISummary
-from .itk import MCFLIRT2ITK, MultiApplyTransforms
+from .itk import MCFLIRT2ITK, Volreg2ITK, MultiApplyTransforms
 from .multiecho import FirstEcho

fmriprep/interfaces/itk.py

@@ -9,6 +9,7 @@
 
 """
 import os
+from pathlib import Path
 from mimetypes import guess_type
 from tempfile import TemporaryDirectory
 import numpy as np
@@ -78,6 +79,49 @@ def _run_interface(self, runtime):
         return runtime
 
 
+class Volreg2ITKInputSpec(BaseInterfaceInputSpec):
+    in_file = File(exists=True, mandatory=True,
+                   desc='mat file generated by AFNI\'s 3dVolreg')
+
+
+class Volreg2ITKOutputSpec(TraitedSpec):
+    out_file = File(desc='the output ITKTransform file')
+
+
+class Volreg2ITK(SimpleInterface):
+
+    """
+    Convert an AFNI's mat file into an ITK Transform file.
+    """
+    input_spec = Volreg2ITKInputSpec
+    output_spec = Volreg2ITKOutputSpec
+
+    def _run_interface(self, runtime):
+        # Load AFNI mat entries and append (0, 0, 0, 1) for homogeneous coordinates
+        orig_afni_mat = np.loadtxt(self.inputs.in_file)
+        afni_affines = [np.vstack((orig_afni_mat[i, :].reshape(3, 4, order='C'), [0, 0, 0, 1]))
+                        for i in range(orig_afni_mat.shape[0])]
+
+        out_file = Path(fname_presuffix(self.inputs.in_file,
+                        suffix='_mc4d_itk.txt', newpath=runtime.cwd))
+
+        fixed_params = 'FixedParameters: 0 0 0'  # Center of rotation does not change
+        lines = ["#Insight Transform File V1.0"]
+        for i, affine in enumerate(afni_affines):
+            lines.append("#Transform %d" % i)
+            lines.append("Transform: AffineTransform_double_3_3")
+
+            ants_affine_2d = np.hstack((affine[:3, :3].reshape(1, -1), affine[:3, 3].reshape(1, -1)))
+            params = ants_affine_2d.reshape(-1, 1).astype('float64')
+            params_list = ["%g" % i for i in params.tolist()]
+            lines.append("Parameters: %s" % ' '.join(params_list))
+            lines.append(fixed_params)
+
+        out_file.write_text('\n'.join(lines))
+        self._results['out_file'] = str(out_file)
+        return runtime
+
+
 class MultiApplyTransformsInputSpec(ApplyTransformsInputSpec):
     input_image = InputMultiPath(File(exists=True), mandatory=True,
                                  desc='input time-series as a list of volumes after splitting'

fmriprep/utils/misc.py

@@ -29,49 +29,6 @@ def split_and_rm_rotshear_func(in_file):
     return out_files
 
 
-def afni2itk_func(in_file):
-    import os
-    from numpy import loadtxt, hstack, vstack, zeros, float64
-
-    def read_afni_affine(input_file, debug=False):
-        orig_afni_mat = loadtxt(input_file)
-        if debug:
-            print(orig_afni_mat)
-
-        output = []
-        for i in range(orig_afni_mat.shape[0]):
-            output.append(vstack((orig_afni_mat[i, :].reshape(3, 4, order='C'), [0, 0, 0, 1])))
-        return output
-
-    def get_ants_dict(affine, debug=False):
-        out_dict = {}
-        ants_affine_2d = hstack((affine[:3, :3].reshape(1, -1), affine[:3, 3].reshape(1, -1)))
-        out_dict['AffineTransform_double_3_3'] = ants_affine_2d.reshape(-1, 1).astype(float64)
-        out_dict['fixed'] = zeros((3, 1))
-        if debug:
-            print(out_dict)
-
-        return out_dict
-
-    out_file = os.path.abspath('mc4d.txt')
-    with open(out_file, 'w') as fp:
-        fp.write("#Insight Transform File V1.0\n")
-        for i, affine in enumerate(read_afni_affine(in_file)):
-            fp.write("#Transform %d\n" % i)
-            fp.write("Transform: AffineTransform_double_3_3\n")
-            trans_dict = get_ants_dict(affine)
-
-            params_list = ["%g" % i for i in list(trans_dict['AffineTransform_double_3_3'])]
-            params_str = ' '.join(params_list)
-            fp.write("Parameters: " + params_str + "\n")
-
-            fixed_params_list = ["%g" % i for i in list(trans_dict['fixed'])]
-            fixed_params_str = ' '.join(fixed_params_list)
-            fp.write("FixedParameters: " + ' '.join(fixed_params_str) + "\n")
-
-    return out_file
-
-
 def fix_multi_T1w_source_name(in_files):
     """
     Make up a generic source name when there are multiple T1s

fmriprep/workflows/bold/hmc.py

@@ -12,7 +12,7 @@
 from nipype.pipeline import engine as pe
 from nipype.interfaces import utility as niu, fsl, afni
 from niworkflows.interfaces import NormalizeMotionParams
-from fmriprep.utils.misc import afni2itk_func
+from ...interfaces import Volreg2ITK
 from ...engine import Workflow
 
 DEFAULT_MEMORY_MIN_GB = 0.01
@@ -75,12 +75,7 @@ def init_bold_hmc_wf(mem_gb, omp_nthreads, name='bold_hmc_wf'):
     # Head motion correction (hmc)
     mc = pe.Node(afni.Volreg(args='-prefix NULL -twopass',
                              zpad=4, outputtype='NIFTI_GZ'), name="mc", mem_gb=mem_gb * 3)
-
-    afni2itk = pe.Node(niu.Function(function=afni2itk_func,
-                                    input_names=["in_file"],
-                                    output_names=["out_files"]), name='afni2itk',
-                       mem_gb=0.05)
-
+    afni2itk = pe.Node(Volreg2ITK(), name='afni2itk', mem_gb=0.05)
     normalize_motion = pe.Node(NormalizeMotionParams(format='AFNI'),
                                name="normalize_motion",
                                mem_gb=DEFAULT_MEMORY_MIN_GB)
@@ -90,7 +85,7 @@ def init_bold_hmc_wf(mem_gb, omp_nthreads, name='bold_hmc_wf'):
                          ('bold_file', 'in_file')]),
         (mc, afni2itk, [('oned_matrix_save', 'in_file')]),
         (mc, normalize_motion, [('oned_file', 'in_file')]),
-        (afni2itk, outputnode, [('out_files', 'xforms')]),
+        (afni2itk, outputnode, [('out_file', 'xforms')]),
         (normalize_motion, outputnode, [('out_file', 'movpar_file')]),
     ])