WIP: Interfaced Workflows and GraftWorkflow

CHANGES

@@ -1,6 +1,7 @@
 Next release
 ============
 
+* ENH: [Interfaced|Graft]Workflows (https://github.com/nipy/nipype/pull/882)
 * TST: Cache APT in CircleCI (https://github.com/nipy/nipype/pull/1333)
 * ENH: Add new flags to the BRAINSABC for new features (https://github.com/nipy/nipype/pull/1322)
 * ENH: Provides a Nipype wrapper for ANTs DenoiseImage (https://github.com/nipy/nipype/pull/1291)

doc/users/graft_workflow.rst

@@ -0,0 +1,87 @@
+.. _graft_workflow:
+
+=====================================================
+Interfaced workflows and GraftWorkflow (experimental)
+=====================================================
+
+:class:`nipype.pipeline.engine.InterfacedWorkflow` provides automatic input/output
+nodes generation, with some other utilities such as fast connection (avoiding
+to specify the connecting fields).
+
+:class:`nipype.pipeline.engine.GraftWorkflow` is intended to create evaluation workflows,
+where all the inputs are the same but several different methods are to be compared, stacking
+the outputs in lists.
+
+
+Interfaced workflows
+--------------------
+
+:class:`~nipype.pipeline.engine.InterfacedWorkflow` generates workflows with default
+``inputnode`` and ``outputnode``. It also exposes the fields without the ``inputnode.`` and
+``outputnode.`` prefix.
+
+Let's create a very simple workflow with a segmentation node. Please, notice the fundamental
+differences with a standard :class:`~nipype.pipeline.engine.Workflow`:
+1) No need for ``inputnode`` and ``outputnode``; 2) fast connection of fields.
+::
+
+    import nipype.pipeline.engine as pe
+    from nipype.interfaces import fsl
+    segm0 = pe.Node(fsl.FAST(number_classes=3, probability_maps=True),
+                    name='FSLFAST')
+    ifwf0 = pe.InterfacedWorkflow(name='testname0', input_names=['in_t1w'],
+                                 output_names=['out_tpm'])
+    ifwf0.connect([
+        ('in', segm0,  [('in_t1w', 'in_files')]),
+        (segm0, 'out', [('probability_maps', 'out_tpm')])
+    ])
+
+
+We can connect an input to this workflow as usual
+::
+
+    import nipype.interfaces.io as nio
+    ds = pe.Node(nio.DataGrabber(base_directory=os.getcwd(), template='t1.nii'),
+                 name='DataSource')
+    mywf = pe.Workflow(name='FullWorkflow')
+    mywf.connect(ds, 't1', ifwf0, 'inputnode.in_t1w')
+
+
+The InterfacedWorkflow is useful to create several segmentation alternatives that always take one input
+named ``in_t1w`` and return one output named ``out_tpm``. Independently,
+:class:`InterfacedWorkflows <nipype.pipeline.engine.InterfacedWorkflow>` do not add much value
+to the conventional :class:`Workflows <nipype.pipeline.engine.Workflow>`, but they are interesting as units inside
+:class:`GraftWorkflows <nipype.pipeline.engine.GraftWorkflow>`.
+
+
+
+Workflows to run cross-comparisons of methods
+---------------------------------------------
+
+Say we want to compare segmentation algorithms: FAST from FSL, and Atropos from ANTS.
+We want all the comparing methods to have the same names and number of inputs and outputs.
+
+We first create the :class:`~nipype.pipeline.engine.GraftWorkflow`, using a existing workflow
+as reference.
+
+::
+
+    compare_wf = pe.GraftWorkflow(name='Comparison', fields_from=ifwf0)
+
+We create the alternate segmentation workflow::
+
+    from nipype.interfaces import ants
+    segm1 = pe.Node(ants.Atropos(dimension=3, number_of_tissue_classes=3),
+                    name='Atropos')
+    ifwf1 = pe.InterfacedWorkflow(name='testname1', input_names=['in_t1w'],
+                                 output_names=['out_tpm'])
+    ifwf1.connect([
+        ('in', segm1,  [('in_t1w', 'intensity_images')]),
+        (segm1, 'out', [('posteriors', 'out_tpm')])
+    ])
+
+Finally, our workflows under comparison are inserted in the :class:`~nipype.pipeline.engine.GraftWorkflow` using
+the ``insert()`` method::
+
+    compare_wf.insert([ifwf0, ifwf1])
+

doc/users/index.rst

@@ -35,6 +35,7 @@
    joinnode_and_itersource
    model_specification
    saving_workflows
+   graft_workflow
    spmmcr
    mipav
    nipypecmd

nipype/interfaces/tests/test_auto_CollateInterface.py

@@ -0,0 +1,25 @@
+# AUTO-GENERATED by tools/checkspecs.py - DO NOT EDIT
+from nipype.testing import assert_equal
+from nipype.interfaces.utility import CollateInterface
+
+def test_CollateInterface_inputs():
+    input_map = dict(_outputs=dict(usedefault=True,
+    ),
+    ignore_exception=dict(nohash=True,
+    usedefault=True,
+    ),
+    )
+    inputs = CollateInterface.input_spec()
+
+    for key, metadata in input_map.items():
+        for metakey, value in metadata.items():
+            yield assert_equal, getattr(inputs.traits()[key], metakey), value
+
+def test_CollateInterface_outputs():
+    output_map = dict()
+    outputs = CollateInterface.output_spec()
+
+    for key, metadata in output_map.items():
+        for metakey, value in metadata.items():
+            yield assert_equal, getattr(outputs.traits()[key], metakey), value
+

nipype/interfaces/tests/test_auto_MultipleSelectInterface.py

@@ -0,0 +1,22 @@
+# AUTO-GENERATED by tools/checkspecs.py - DO NOT EDIT
+from nipype.testing import assert_equal
+from nipype.interfaces.utility import MultipleSelectInterface
+
+def test_MultipleSelectInterface_inputs():
+    input_map = dict(index=dict(mandatory=True,
+    ),
+    )
+    inputs = MultipleSelectInterface.input_spec()
+
+    for key, metadata in input_map.items():
+        for metakey, value in metadata.items():
+            yield assert_equal, getattr(inputs.traits()[key], metakey), value
+
+def test_MultipleSelectInterface_outputs():
+    output_map = dict()
+    outputs = MultipleSelectInterface.output_spec()
+
+    for key, metadata in output_map.items():
+        for metakey, value in metadata.items():
+            yield assert_equal, getattr(outputs.traits()[key], metakey), value
+

nipype/interfaces/utility.py

@@ -488,6 +488,163 @@ def _run_interface(self, runtime):
         return runtime
 
 
+class CollateInterfaceInputSpec(DynamicTraitedSpec, BaseInterfaceInputSpec):
+    _outputs = traits.Dict(traits.Any, value={}, usedefault=True)
+
+    def __setattr__(self, key, value):
+        if key not in self.copyable_trait_names():
+            if not isdefined(value):
+                super(CollateInterfaceInputSpec, self).__setattr__(key, value)
+            self._outputs[key] = value
+        else:
+            if key in self._outputs:
+                self._outputs[key] = value
+            super(CollateInterfaceInputSpec, self).__setattr__(key, value)
+
+
+class CollateInterface(IOBase):
+    """
+    A simple interface to multiplex inputs through a unique output node.
+    Channel is defined by the prefix of the fields. In order to avoid
+    inconsistencies, output fields should be defined forehand at initialization..
+
+    Example
+    -------
+
+    >>> from nipype.interfaces.utility import CollateInterface
+    >>> coll = CollateInterface(fields=['file','miscdata'])
+    >>> coll.inputs.src1_file = 'exfile1.csv'
+    >>> coll.inputs.src2_file = 'exfile2.csv'
+    >>> coll.inputs.src1_miscdata = 1.0
+    >>> coll.inputs.src2_miscdata = 2.0
+    >>> res = coll.run()
+    >>> print(res.outputs.file)
+    ['exfile1.csv', 'exfile2.csv']
+    >>> print(res.outputs.miscdata)
+    [1.0, 2.0]
+    """
+    input_spec = CollateInterfaceInputSpec
+    output_spec = DynamicTraitedSpec
+
+    def __init__(self, fields=None, fill_missing=False, **kwargs):
+        super(CollateInterface, self).__init__(**kwargs)
+
+        if fields is None or not fields:
+            raise ValueError('CollateInterface fields must be a non-empty list')
+        # Each input must be in the fields.
+        self._fields = fields
+        self._fill_missing = fill_missing
+
+    def _add_output_traits(self, base):
+        undefined_traits = {}
+        for key in self._fields:
+            base.add_trait(key, traits.Any)
+            undefined_traits[key] = Undefined
+        base.trait_set(trait_change_notify=False, **undefined_traits)
+        return base
+
+    def _list_outputs(self):
+        #manual mandatory inputs check
+        valuedict = dict( (key, {}) for key in self._fields)
+        nodekeys = []
+
+        for inputkey, inputval in self.inputs._outputs.items():
+            for key in self._fields:
+                if inputkey.endswith(key):
+                    nodekey = inputkey[::-1].replace(key[::-1], '', 1)[::-1]
+                    nodekeys.append(nodekey)
+
+                    if nodekey in valuedict[key].keys():
+                        msg = ('Trying to add field from existing node')
+                        raise ValueError(msg)
+                    valuedict[key][nodekey] = inputval
+
+        nodekeys = sorted(set(nodekeys))
+        outputs = self._outputs().get()
+        for key in self._fields:
+            outputs[key] = []
+            for nk in nodekeys:
+
+                if nk in valuedict[key]:
+                    val = valuedict[key][nk]
+                else:
+                    if self._fill_missing:
+                        val = None
+                    else:
+                        raise RuntimeError('Input missing for field to collate.')
+                outputs[key].append(val)
+
+        return outputs
+
+
+class MultipleSelectInputSpec(DynamicTraitedSpec):
+    index = InputMultiPath(traits.Int, mandatory=True,
+                           desc='0-based indices of values to choose')
+
+
+class MultipleSelectInterface(IOBase):
+    """
+    Basic interface that demultiplexes lists generated by CollateInterface
+
+    Example
+    -------
+
+    >>> from nipype.interfaces.utility import MultipleSelectInterface
+    >>> demux = MultipleSelectInterface(fields=['file','miscdata'], index=0)
+    >>> demux.inputs.file = ['exfile1.csv', 'exfile2.csv']
+    >>> demux.inputs.miscdata = [1.0, 2.0]
+    >>> res = demux.run()
+    >>> print(res.outputs.file)
+    exfile1.csv
+    >>> print(res.outputs.miscdata)
+    1.0
+    """
+    input_spec = MultipleSelectInputSpec
+    output_spec = DynamicTraitedSpec
+
+    def __init__(self, fields=None, mandatory_inputs=True, **inputs):
+        super(MultipleSelectInterface, self).__init__(**inputs)
+        if fields is None or not fields:
+            raise ValueError('Identity Interface fields must be a non-empty list')
+        # Each input must be in the fields.
+        for in_field in inputs:
+            if in_field not in fields and in_field != 'index':
+                raise ValueError('Identity Interface input is not in the fields: %s' % in_field)
+        self._fields = fields
+        self._mandatory_inputs = mandatory_inputs
+        add_traits(self.inputs, fields)
+        # Adding any traits wipes out all input values set in superclass initialization,
+        # even it the trait is not in the add_traits argument. The work-around is to reset
+        # the values after adding the traits.
+        self.inputs.set(**inputs)
+
+    def _add_output_traits(self, base):
+        undefined_traits = {}
+        for key in self._fields:
+            base.add_trait(key, traits.Any)
+            undefined_traits[key] = Undefined
+        base.trait_set(trait_change_notify=False, **undefined_traits)
+        return base
+
+    def _list_outputs(self):
+        #manual mandatory inputs check
+        if self._fields and self._mandatory_inputs:
+            for key in self._fields:
+                value = getattr(self.inputs, key)
+                if not isdefined(value):
+                    msg = "%s requires a value for input '%s' because it was listed in 'fields'. \
+                    You can turn off mandatory inputs checking by passing mandatory_inputs = False to the constructor." % \
+                    (self.__class__.__name__, key)
+                    raise ValueError(msg)
+
+        outputs = self._outputs().get()
+        for key in self._fields:
+            val = getattr(self.inputs, key)
+            if isdefined(val):
+                outputs[key] = np.squeeze(np.array(val)[np.array(self.inputs.index)]).tolist()
+        return outputs
+
+
 class CSVReaderInputSpec(DynamicTraitedSpec, TraitedSpec):
     in_file = File(exists=True, mandatory=True, desc='Input comma-seperated value (CSV) file')
     header = traits.Bool(False, usedefault=True, desc='True if the first line is a column header')
@@ -566,3 +723,4 @@ def _list_outputs(self):
                 entry = self._parse_line(line)
                 outputs = self._append_entry(outputs, entry)
         return outputs
+

nipype/pipeline/__init__.py

@@ -7,4 +7,5 @@
 
 from __future__ import absolute_import
 __docformat__ = 'restructuredtext'
-from .engine import Node, MapNode, JoinNode, Workflow
+from .engine import (Node, MapNode, JoinNode, Workflow,
+                     InterfacedWorkflow, GraftWorkflow)

nipype/pipeline/engine/__init__.py

@@ -9,6 +9,6 @@
 
 from __future__ import absolute_import
 __docformat__ = 'restructuredtext'
-from .workflows import Workflow
+from .workflows import Workflow, InterfacedWorkflow, GraftWorkflow
 from .nodes import Node, MapNode, JoinNode
 from .utils import generate_expanded_graph