[ENH] Add interfaces wrapping DIPY worflows

doc/index.rst

@@ -16,7 +16,7 @@
         these packages within a single workflow. Nipype provides an environment
         that encourages interactive exploration of algorithms from different
         packages (e.g., ANTS_, SPM_, FSL_, FreeSurfer_, Camino_, MRtrix_, MNE_, AFNI_,
-        Slicer_), eases the design of workflows within and between packages, and
+        Slicer_, DIPY_), eases the design of workflows within and between packages, and
         reduces the learning curve necessary to use different packages. Nipype is
         creating a collaborative platform for neuroimaging software development
         in a high-level language and addressing limitations of existing pipeline

doc/links_names.txt

@@ -97,6 +97,7 @@
 .. _MRtrix3: http://www.mrtrix.org/
 .. _MNE: https://martinos.org/mne/index.html
 .. _ANTS: http://stnava.github.io/ANTs/
+.. _DIPY: http://dipy.org
 
 .. General software
 .. _gcc: http://gcc.gnu.org

nipype/interfaces/dipy/base.py

@@ -4,14 +4,16 @@
                         absolute_import)
 
 import os.path as op
+import inspect
 import numpy as np
 from ... import logging
 from ..base import (traits, File, isdefined, LibraryBaseInterface,
-                    BaseInterfaceInputSpec)
+                    BaseInterfaceInputSpec, TraitedSpec)
 
 HAVE_DIPY = True
 try:
     import dipy
+    from dipy.workflows.base import IntrospectiveArgumentParser
 except ImportError:
     HAVE_DIPY = False
 
@@ -75,3 +77,137 @@ def _gen_filename(self, name, ext=None):
             ext = fext
 
         return out_prefix + '_' + name + ext
+
+
+def convert_to_traits_type(dipy_type, is_file=False):
+    """Convert DIPY type to Traits type."""
+    dipy_type = dipy_type.lower()
+    is_mandatory = bool("optional" not in dipy_type)
+    if "variable" in dipy_type and "string" in dipy_type:
+        return traits.ListStr, is_mandatory
+    elif "variable" in dipy_type and "int" in dipy_type:
+        return traits.ListInt, is_mandatory
+    elif "variable" in dipy_type and "float" in dipy_type:
+        return traits.ListFloat, is_mandatory
+    elif "variable" in dipy_type and "bool" in dipy_type:
+        return traits.ListBool, is_mandatory
+    elif "variable" in dipy_type and "complex" in dipy_type:
+        return traits.ListComplex, is_mandatory
+    elif "string" in dipy_type and not is_file:
+        return traits.Str, is_mandatory
+    elif "string" in dipy_type and is_file:
+        return traits.File, is_mandatory
+    elif "int" in dipy_type:
+        return traits.Int, is_mandatory
+    elif "float" in dipy_type:
+        return traits.Float, is_mandatory
+    elif "bool" in dipy_type:
+        return traits.Bool, is_mandatory
+    elif "complex" in dipy_type:
+        return traits.Complex, is_mandatory
+    else:
+        msg = "Error during convert_to_traits_type({0}).".format(dipy_type) + \
+              "Unknown DIPY type."
+        raise IOError(msg)
+
+
+def create_interface_specs(class_name, params=None, BaseClass=TraitedSpec):
+    """Create IN/Out interface specifications dynamically.
+
+    Parameters
+    ----------
+    class_name: str
+        The future class name(e.g, (MyClassInSpec))
+    params: list of tuple
+        dipy argument list
+    BaseClass: TraitedSpec object
+        parent class
+
+    Returns
+    -------
+    newclass: object
+        new nipype interface specification class
+
+    """
+    attr = {}
+    if params is not None:
+        for p in params:
+            name, dipy_type, desc = p[0], p[1], p[2]
+            is_file = bool("files" in name or "out_" in name)
+            traits_type, is_mandatory = convert_to_traits_type(dipy_type,
+                                                               is_file)
+            # print(name, dipy_type, desc, is_file, traits_type, is_mandatory)
+            if BaseClass.__name__ == BaseInterfaceInputSpec.__name__:
+                if len(p) > 3:
+                    attr[name] = traits_type(p[3], desc=desc[-1],
+                                             usedefault=True,
+                                             mandatory=is_mandatory)
+                else:
+                    attr[name] = traits_type(desc=desc[-1],
+                                             mandatory=is_mandatory)
+            else:
+                attr[name] = traits_type(p[3], desc=desc[-1], exists=True,
+                                         usedefault=True,)
+
+    newclass = type(str(class_name), (BaseClass, ), attr)
+    return newclass
+
+
+def dipy_to_nipype_interface(cls_name, dipy_flow, BaseClass=DipyBaseInterface):
+    """Construct a class in order to respect nipype interface specifications.
+
+    This convenient class factory convert a DIPY Workflow to a nipype
+    interface.
+
+    Parameters
+    ----------
+    cls_name: string
+        new class name
+    dipy_flow: Workflow class type.
+        It should be any children class of `dipy.workflows.workflow.Worflow`
+    BaseClass: object
+        nipype instance object
+
+    Returns
+    -------
+    newclass: object
+        new nipype interface specification class
+
+    """
+    parser = IntrospectiveArgumentParser()
+    flow = dipy_flow()
+    parser.add_workflow(flow)
+    default_values = inspect.getargspec(flow.run).defaults
+    optional_params = [args + (val,) for args, val in zip(parser.optional_parameters, default_values)]
+    start = len(parser.optional_parameters) - len(parser.output_parameters)
+
+    output_parameters = [args + (val,) for args, val in zip(parser.output_parameters, default_values[start:])]
+    input_parameters = parser.positional_parameters + optional_params
+
+    input_spec = create_interface_specs("{}InputSpec".format(cls_name),
+                                        input_parameters,
+                                        BaseClass=BaseInterfaceInputSpec)
+
+    output_spec = create_interface_specs("{}OutputSpec".format(cls_name),
+                                         output_parameters,
+                                         BaseClass=TraitedSpec)
+
+    def _run_interface(self, runtime):
+        flow = dipy_flow()
+        args = self.inputs.get()
+        flow.run(**args)
+
+    def _list_outputs(self):
+        outputs = self._outputs().get()
+        out_dir = outputs.get("out_dir", ".")
+        for key, values in outputs.items():
+            outputs[key] = op.join(out_dir, values)
+
+        return outputs
+
+    newclass = type(str(cls_name), (BaseClass, ),
+                    {"input_spec": input_spec,
+                     "output_spec": output_spec,
+                     "_run_interface": _run_interface,
+                     "_list_outputs:": _list_outputs})
+    return newclass

nipype/interfaces/dipy/reconstruction.py

@@ -13,13 +13,30 @@
 
 import numpy as np
 import nibabel as nb
+from distutils.version import LooseVersion
 
 from ... import logging
 from ..base import TraitedSpec, File, traits, isdefined
-from .base import DipyDiffusionInterface, DipyBaseInterfaceInputSpec
+from .base import (DipyDiffusionInterface, DipyBaseInterfaceInputSpec,
+                   HAVE_DIPY, dipy_version, dipy_to_nipype_interface)
+
 
 IFLOGGER = logging.getLogger('nipype.interface')
 
+if HAVE_DIPY and LooseVersion(dipy_version()) >= LooseVersion('0.15'):
+    from dipy.workflows.reconst import (ReconstDkiFlow, ReconstCSAFlow,
+                                        ReconstCSDFlow, ReconstMAPMRIFlow,
+                                        ReconstDtiFlow)
+
+    DKIModel = dipy_to_nipype_interface("DKIModel", ReconstDkiFlow)
+    MapmriModel = dipy_to_nipype_interface("MapmriModel", ReconstMAPMRIFlow)
+    DTIModel = dipy_to_nipype_interface("DTIModel", ReconstDtiFlow)
+    CSAModel = dipy_to_nipype_interface("CSAModel", ReconstCSAFlow)
+    CSDModel = dipy_to_nipype_interface("CSDModel", ReconstCSDFlow)
+else:
+    IFLOGGER.info("We advise you to upgrade DIPY version. This upgrade will"
+                  " activate DKIModel, MapmriModel, DTIModel, CSAModel.")
+
 
 class RESTOREInputSpec(DipyBaseInterfaceInputSpec):
     in_mask = File(exists=True, desc=('input mask in which compute tensors'))

nipype/interfaces/dipy/registration.py

@@ -0,0 +1,18 @@
+
+from distutils.version import LooseVersion
+from ... import logging
+from .base import HAVE_DIPY, dipy_version, dipy_to_nipype_interface
+
+IFLOGGER = logging.getLogger('nipype.interface')
+
+if HAVE_DIPY and LooseVersion(dipy_version()) >= LooseVersion('0.15'):
+
+    from dipy.workflows.align import ResliceFlow, SlrWithQbxFlow
+
+    Reslice = dipy_to_nipype_interface("Reslice", ResliceFlow)
+    StreamlineRegistration = dipy_to_nipype_interface("StreamlineRegistration",
+                                                      SlrWithQbxFlow)
+
+else:
+    IFLOGGER.info("We advise you to upgrade DIPY version. This upgrade will"
+                  " activate Reslice, StreamlineRegistration.")

nipype/interfaces/dipy/tests/test_base.py

@@ -0,0 +1,142 @@
+import pytest
+from collections import namedtuple
+from ...base import traits, TraitedSpec, BaseInterfaceInputSpec
+from ..base import (convert_to_traits_type, create_interface_specs,
+                    dipy_to_nipype_interface, DipyBaseInterface, no_dipy)
+
+
+def test_convert_to_traits_type():
+    Params = namedtuple("Params", "traits_type is_file")
+    Res = namedtuple("Res", "traits_type is_mandatory")
+    l_entries = [Params('variable string', False),
+                 Params('variable int', False),
+                 Params('variable float', False),
+                 Params('variable bool', False),
+                 Params('variable complex', False),
+                 Params('variable int, optional', False),
+                 Params('variable string, optional', False),
+                 Params('variable float, optional', False),
+                 Params('variable bool, optional', False),
+                 Params('variable complex, optional', False),
+                 Params('string', False), Params('int', False),
+                 Params('string', True), Params('float', False),
+                 Params('bool', False), Params('complex', False),
+                 Params('string, optional', False),
+                 Params('int, optional', False),
+                 Params('string, optional', True),
+                 Params('float, optional', False),
+                 Params('bool, optional', False),
+                 Params('complex, optional', False),
+                 ]
+    l_expected = [Res(traits.ListStr, True), Res(traits.ListInt, True),
+                  Res(traits.ListFloat, True), Res(traits.ListBool, True),
+                  Res(traits.ListComplex, True), Res(traits.ListInt, False),
+                  Res(traits.ListStr, False), Res(traits.ListFloat, False),
+                  Res(traits.ListBool, False), Res(traits.ListComplex, False),
+                  Res(traits.Str, True), Res(traits.Int, True),
+                  Res(traits.File, True), Res(traits.Float, True),
+                  Res(traits.Bool, True), Res(traits.Complex, True),
+                  Res(traits.Str, False), Res(traits.Int, False),
+                  Res(traits.File, False), Res(traits.Float, False),
+                  Res(traits.Bool, False), Res(traits.Complex, False),
+                  ]
+
+    for entry, res in zip(l_entries, l_expected):
+        traits_type, is_mandatory = convert_to_traits_type(entry.traits_type,
+                                                           entry.is_file)
+        assert traits_type == res.traits_type
+        assert is_mandatory == res.is_mandatory
+
+    with pytest.raises(IOError):
+        convert_to_traits_type("file, optional")
+
+
+def test_create_interface_specs():
+    new_interface = create_interface_specs("MyInterface")
+
+    assert new_interface.__base__ == TraitedSpec
+    assert isinstance(new_interface(), TraitedSpec)
+    assert new_interface.__name__ == "MyInterface"
+    assert not new_interface().get()
+
+    new_interface = create_interface_specs("MyInterface",
+                                           BaseClass=BaseInterfaceInputSpec)
+    assert new_interface.__base__ == BaseInterfaceInputSpec
+    assert isinstance(new_interface(), BaseInterfaceInputSpec)
+    assert new_interface.__name__ == "MyInterface"
+    assert not new_interface().get()
+
+    params = [("params1", "string", ["my description"]), ("params2_files", "string", ["my description @"]),
+              ("params3", "int, optional", ["useful option"]), ("out_params", "string", ["my out description"])]
+
+    new_interface = create_interface_specs("MyInterface", params=params,
+                                           BaseClass=BaseInterfaceInputSpec)
+
+    assert new_interface.__base__ == BaseInterfaceInputSpec
+    assert isinstance(new_interface(), BaseInterfaceInputSpec)
+    assert new_interface.__name__ == "MyInterface"
+    current_params = new_interface().get()
+    assert len(current_params) == 4
+    assert 'params1' in current_params.keys()
+    assert 'params2_files' in current_params.keys()
+    assert 'params3' in current_params.keys()
+    assert 'out_params' in current_params.keys()
+
+
+@pytest.mark.skipif(no_dipy(), reason="DIPY is not installed")
+def test_dipy_to_nipype_interface():
+    from dipy.workflows.workflow import Workflow
+
+    class DummyWorkflow(Workflow):
+
+        @classmethod
+        def get_short_name(cls):
+            return 'dwf1'
+
+        def run(self, in_files, param1=1, out_dir='', out_ref='out1.txt'):
+            """Workflow used to test basic workflows.
+
+            Parameters
+            ----------
+            in_files : string
+                fake input string param
+            param1 : int, optional
+                fake positional param (default 1)
+            out_dir : string, optional
+                fake output directory (default '')
+            out_ref : string, optional
+                fake out file (default out1.txt)
+
+            References
+            -----------
+            dummy references
+
+            """
+            return param1
+
+    new_specs = dipy_to_nipype_interface("MyModelSpec", DummyWorkflow)
+    assert new_specs.__base__ == DipyBaseInterface
+    assert isinstance(new_specs(), DipyBaseInterface)
+    assert new_specs.__name__ == "MyModelSpec"
+    assert hasattr(new_specs, 'input_spec')
+    assert new_specs().input_spec.__base__ == BaseInterfaceInputSpec
+    assert hasattr(new_specs, 'output_spec')
+    assert new_specs().output_spec.__base__ == TraitedSpec
+    assert hasattr(new_specs, '_run_interface')
+    assert hasattr(new_specs, '_list_outputs')
+    params_in = new_specs().inputs.get()
+    params_out = new_specs()._outputs().get()
+    assert len(params_in) == 4
+    assert 'in_files' in params_in.keys()
+    assert 'param1' in params_in.keys()
+    assert 'out_dir' in params_out.keys()
+    assert 'out_ref' in params_out.keys()
+
+    with pytest.raises(ValueError):
+        new_specs().run()
+
+
+if __name__ == "__main__":
+    test_convert_to_traits_type()
+    test_create_interface_specs()
+    test_dipy_to_nipype_interface()