[ENH/WIP] WorkflowInterface

CHANGES

@@ -1,6 +1,7 @@
 Upcoming release
 ================
 
+* ENH: New WorkflowInterface (https://github.com/nipy/nipype/pull/1835)
 * ENH: Improve terminal_output feature (https://github.com/nipy/nipype/pull/2209)
 * ENH: Simple interface to FSL std2imgcoords (https://github.com/nipy/nipype/pull/2209, prev #1398)
 * ENH: Centralize virtual/physical $DISPLAYs (https://github.com/nipy/nipype/pull/#2203)

doc/users/function_interface.rst

@@ -1,17 +1,23 @@
 .. _function_interface:
 
-======================
-The Function Interface
-======================
+==============================================
+Wrapping Python functions and Nipype Workflows
+==============================================
 
 Most Nipype interfaces provide access to external programs, such as FSL
-binaries or SPM routines. However, a special interface,
-:class:`nipype.interfaces.utility.Function`,
-allows you to wrap arbitrary Python code in the Interface framework and
-seamlessly integrate it into your workflows.
+binaries or SPM routines. However, Nipype has two special interfaces
+that allow you to wrap arbitrary Python code 
+(:class:`nipype.interfaces.utility.wrappers.Function`)
+or full Nipype workflows (:class:`nipype.interfaces.utility.wrappers.WorkflowInterface`)
+in the Interface framework and seamlessly integrate them
+into your workflows.
+
+
+The Function Interface
+----------------------
 
 A Simple Function Interface
----------------------------
+~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
 The most important component of a working Function interface is a Python
 function. There are several ways to associate a function with a Function
@@ -50,8 +56,9 @@ Note that, if you are working interactively, the Function interface is
 unable to use functions that are defined within your interpreter session.
 (Specifically, it can't use functions that live in the ``__main__`` namespace).
 
+
 Using External Packages
------------------------
+~~~~~~~~~~~~~~~~~~~~~~~
 
 Chances are, you will want to write functions that do more complicated
 processing, particularly using the growing stack of Python packages
@@ -77,9 +84,7 @@ constructor. This allows for the use of external functions that do not
 import all external definitions inside the function body.
 
 Hello World - Function interface in a workflow
-----------------------------------------------
-
-Contributed by: Hänel Nikolaus Valentin
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
 The following snippet of code demonstrates the use of the function interface in
 the context of a workflow. Note the use of ``import os`` within the function as
@@ -88,7 +93,9 @@ is necessary because functions are coded as strings and do not have to be on the
 PYTHONPATH. However any function called by this function has to be available on
 the PYTHONPATH. The `absolute path` is necessary because all workflow nodes are
 executed in their own directory and therefore there is no way of determining
-that the input file came from a different directory::
+that the input file came from a different directory:
+
+  ::
 
     import nipype.pipeline.engine as pe
     from nipype.interfaces.utility import Function
@@ -128,7 +135,7 @@ that the input file came from a different directory::
 
 
 Advanced Use
-------------
+~~~~~~~~~~~~
 
 To use an existing function object (as we have been doing so far) with a Function
 interface, it must be passed to the constructor. However, it is also possible
@@ -148,4 +155,87 @@ meaning that you could write a function that outputs different function
 strings depending on some run-time contingencies, and connect that output
 the the ``function_str`` input of a downstream Function interface.
 
+The Workflow Interface
+----------------------
+
+Along the same lines of the Function interface, it is possible to wrap
+nipype workflows inside the
+:class:`nipype.interfaces.utility.wrappers.WorkflowInterface`.
+Wrapping workflows enables subworkflows to dinamically iterate over
+inputs using ``MapNode`` (`#819 <https://github.com/nipy/nipype/issues/819>`_),
+and allows for the  conditional execution of subworkflows
+(see for instance, `#878 <https://github.com/nipy/nipype/issues/878>`_,
+`#1081 <https://github.com/nipy/nipype/issues/1081>`_, and
+`#1299 <https://github.com/nipy/nipype/issues/1299>`_).
+
+A use case
+~~~~~~~~~~
+
+Say we have the following workflow:
+
+  ::
+
+    def merge_workflow(name='MergeWorkflow'):
+        inputnode = pe.Node(niu.IdentityInterface(
+            fields=['a', 'b']), name='inputnode')
+
+        mergenode = pe.Node(niu.Merge(2), name='mergenode')
+
+        outputnode = pe.Node(niu.IdentityInterface(
+            fields=['c']), name='outputnode')
+
+        wf = pe.Workflow(name=name)
+        wf.connect([
+            (inputnode, mergenode, [('a', 'in1'), ('b', 'in2')]),
+            (mergenode, outputnode, [('out', 'c')])
+        ])
+
+        return wf
+
+
+We want to run the workflow over a number of combinations of
+the inputs ``a`` and ``b``. However, these combinations should
+be generated by a previous workflow. Therefore, we cannot
+use the ``iterables`` feature of Nipype ``Node``. In this case,
+we need to use a ``MapNode``, with our workflow wrapped within
+a ``WorkflowInterface``.
+This ``wrapped`` map node can be used in a workflow, and set the inputs
+``'a'`` and ``'b'`` the same way it is done in a regular interface:
+
+  ::
+
+      outerworkflow = pe.Workflow(name='OuterWorkflow')
+      outerinput = pe.Node(niu.IdentityInterface(
+                           fields=['a', 'b']), name='outerinput')
+      wrapped = pe.MapNode(niu.WorkflowInterface(workflow=merge_workflow),
+                           iterfield=['a', 'b'], name='dyniterable')
+      outeroutput = pe.Node(niu.IdentityInterface(
+                            fields=['c']), name='outeroutput')
+
+      outerworkflow.connect([
+          (outerinput, wrapped, [('a', 'a'), ('b', 'b')]),
+          (wrapped, outeroutput, [('out', 'c')])
+      ])
+      outerworkflow.inputs.outerinput.a = [10, 12, 14]
+      outerworkflow.inputs.outerinput.b = [9, -1, 10]
+
+      outerworkflow.run()
+
+
+
+Both inputs will work as expected when connected from other nodes in
+a Nipype Workflow.
+
+
+Limitations
+~~~~~~~~~~~
+
+The main limitation of :class:`nipype.interfaces.utility.wrappers.WorkflowInterface`
+is that the workflow can only be run internally using the ``Linear`` or ``MultiProc``
+plugins. Further work could be directed towards running using other synchronous
+plugins. The only restriction for the plugin used to run the inner interface is
+that it should be synchronous, in order to collect the workflow outputs after execution.
+
+
+
 .. include:: ../links_names.txt

nipype/interfaces/tests/test_auto_SimpleInterface.py

@@ -0,0 +1,16 @@
+# AUTO-GENERATED by tools/checkspecs.py - DO NOT EDIT
+from __future__ import unicode_literals
+from ..base import SimpleInterface
+
+
+def test_SimpleInterface_inputs():
+    input_map = dict(ignore_exception=dict(nohash=True,
+    usedefault=True,
+    ),
+    )
+    inputs = SimpleInterface.input_spec()
+
+    for key, metadata in list(input_map.items()):
+        for metakey, value in list(metadata.items()):
+            assert getattr(inputs.traits()[key], metakey) == value
+

nipype/interfaces/utility/__init__.py

@@ -10,4 +10,4 @@
 from .base import (IdentityInterface, Rename, Select, Split, Merge,
                    AssertEqual)
 from .csv import CSVReader
-from .wrappers import Function
+from .wrappers import Function, WorkflowInterface

nipype/interfaces/utility/tests/test_auto_WorkflowInterface.py

@@ -0,0 +1,27 @@
+# AUTO-GENERATED by tools/checkspecs.py - DO NOT EDIT
+from __future__ import unicode_literals
+from ..wrappers import WorkflowInterface
+
+
+def test_WorkflowInterface_inputs():
+    input_map = dict(ignore_exception=dict(nohash=True,
+    usedefault=True,
+    ),
+    nprocs=dict(mandatory=True,
+    usedefault=True,
+    ),
+    )
+    inputs = WorkflowInterface.input_spec()
+
+    for key, metadata in list(input_map.items()):
+        for metakey, value in list(metadata.items()):
+            assert getattr(inputs.traits()[key], metakey) == value
+
+
+def test_WorkflowInterface_outputs():
+    output_map = dict()
+    outputs = WorkflowInterface.output_spec()
+
+    for key, metadata in list(output_map.items()):
+        for metakey, value in list(metadata.items()):
+            assert getattr(outputs.traits()[key], metakey) == value