Add supported for nested lists in iterfields.

Author: chrisgorgo

doc/users/mapnode_and_iterables.rst

@@ -92,6 +92,11 @@ is almost the same as running
 	
 It is a rarely used feature, but you can sometimes find it useful.
 
+In more advanced applications it is useful to be able to iterate over items
+of nested lists (for example [[1,2],[3,4]]). MapNode allows you to do this
+with the "nested=True" parameter. Outputs will preserve the same nested
+structure as the inputs.
+
 Iterables
 =========
 

nipype/algorithms/modelgen.py

@@ -165,11 +165,11 @@ def gen_info(run_event_files):
 
 
 class SpecifyModelInputSpec(BaseInterfaceInputSpec):
-    subject_info = InputMultiPath(Bunch, mandatory=True, xor=['event_files'],
+    subject_info = InputMultiPath(Bunch, mandatory=True, xor=['subject_info', 'event_files'],
           desc=("Bunch or List(Bunch) subject specific condition information. "
                 "see :ref:`SpecifyModel` or SpecifyModel.__doc__ for details"))
     event_files = InputMultiPath(traits.List(File(exists=True)), mandatory=True,
-                                 xor=['subject_info'],
+                                 xor=['subject_info', 'event_files'],
           desc=('list of event description files 1, 2 or 3 column format '
                 'corresponding to onsets, durations and amplitudes'))
     realignment_parameters = InputMultiPath(File(exists=True),

nipype/pipeline/engine.py

@@ -13,6 +13,7 @@
 """
 
 from datetime import datetime
+from nipype.utils.misc import flatten, unflatten
 try:
     from collections import OrderedDict
 except ImportError:
@@ -2027,7 +2028,7 @@ class MapNode(Node):
 
     """
 
-    def __init__(self, interface, iterfield, name, serial=False, **kwargs):
+    def __init__(self, interface, iterfield, name, serial=False, nested=False, **kwargs):
         """
 
         Parameters
@@ -2043,6 +2044,9 @@ def __init__(self, interface, iterfield, name, serial=False, **kwargs):
             node specific name
         serial : boolean
             flag to enforce executing the jobs of the mapnode in a serial manner rather than parallel
+        nested : boolea
+            support for nested lists, if set the input list will be flattened before running, and the 
+            nested list structure of the outputs will be resored 
         See Node docstring for additional keyword arguments.
         """
 
@@ -2051,6 +2055,7 @@ def __init__(self, interface, iterfield, name, serial=False, **kwargs):
         if isinstance(iterfield, six.string_types):
             iterfield = [iterfield]
         self.iterfield = iterfield
+        self.nested = nested
         self._inputs = self._create_dynamic_traits(self._interface.inputs,
                                                    fields=self.iterfield)
         self._inputs.on_trait_change(self._set_mapnode_input)
@@ -2066,7 +2071,10 @@ def _create_dynamic_traits(self, basetraits, fields=None, nitems=None):
         for name, spec in basetraits.items():
             if name in fields and ((nitems is None) or (nitems > 1)):
                 logger.debug('adding multipath trait: %s' % name)
-                output.add_trait(name, InputMultiPath(spec.trait_type))
+                if self.nested:
+                    output.add_trait(name, InputMultiPath(traits.Any()))
+                else:
+                    output.add_trait(name, InputMultiPath(spec.trait_type))
             else:
                 output.add_trait(name, traits.Trait(spec))
             setattr(output, name, Undefined)
@@ -2110,7 +2118,10 @@ def _get_hashval(self):
                     self._interface.inputs.traits()[name].trait_type))
             logger.debug('setting hashinput %s-> %s' %
                          (name, getattr(self._inputs, name)))
-            setattr(hashinputs, name, getattr(self._inputs, name))
+            if self.nested:
+                setattr(hashinputs, name, flatten(getattr(self._inputs, name)))
+            else:
+                setattr(hashinputs, name, getattr(self._inputs, name))
         hashed_inputs, hashvalue = hashinputs.get_hashval(
             hash_method=self.config['execution']['hash_method'])
         rm_extra = self.config['execution']['remove_unnecessary_outputs']
@@ -2137,7 +2148,10 @@ def outputs(self):
     def _make_nodes(self, cwd=None):
         if cwd is None:
             cwd = self.output_dir()
-        nitems = len(filename_to_list(getattr(self.inputs, self.iterfield[0])))
+        if self.nested:
+            nitems = len(flatten(filename_to_list(getattr(self.inputs, self.iterfield[0]))))
+        else:    
+            nitems = len(filename_to_list(getattr(self.inputs, self.iterfield[0])))
         for i in range(nitems):
             nodename = '_' + self.name + str(i)
             node = Node(deepcopy(self._interface), name=nodename)
@@ -2147,7 +2161,10 @@ def _make_nodes(self, cwd=None):
             node._interface.inputs.set(
                 **deepcopy(self._interface.inputs.get()))
             for field in self.iterfield:
-                fieldvals = filename_to_list(getattr(self.inputs, field))
+                if self.nested:
+                    fieldvals = flatten(filename_to_list(getattr(self.inputs, field)))
+                else:
+                    fieldvals = filename_to_list(getattr(self.inputs, field))
                 logger.debug('setting input %d %s %s' % (i, field,
                                                          fieldvals[i]))
                 setattr(node.inputs, field,
@@ -2199,6 +2216,14 @@ def _collate_results(self, nodes):
                     defined_vals = [isdefined(val) for val in values]
                     if any(defined_vals) and self._result.outputs:
                         setattr(self._result.outputs, key, values)
+        
+        if self.nested:
+            for key, _ in self.outputs.items():
+                values = getattr(self._result.outputs, key)
+                if isdefined(values):
+                    values = unflatten(values, filename_to_list(getattr(self.inputs, self.iterfield[0])))
+                setattr(self._result.outputs, key, values)
+                        
         if returncode and any([code is not None for code in returncode]):
             msg = []
             for i, code in enumerate(returncode):
@@ -2249,7 +2274,10 @@ def num_subnodes(self):
         if self._serial :
             return 1
         else:
-            return len(filename_to_list(getattr(self.inputs, self.iterfield[0])))
+            if self.nested:
+                return len(filename_to_list(flatten(getattr(self.inputs, self.iterfield[0]))))
+            else:
+                return len(filename_to_list(getattr(self.inputs, self.iterfield[0])))
 
     def _get_inputs(self):
         old_inputs = self._inputs.get()
@@ -2289,8 +2317,12 @@ def _run_interface(self, execute=True, updatehash=False):
         os.chdir(cwd)
         self._check_iterfield()
         if execute:
-            nitems = len(filename_to_list(getattr(self.inputs,
-                                                  self.iterfield[0])))
+            if self.nested:
+                nitems = len(filename_to_list(flatten(getattr(self.inputs,
+                                                      self.iterfield[0]))))
+            else:
+                nitems = len(filename_to_list(getattr(self.inputs,
+                                                      self.iterfield[0])))
             nodenames = ['_' + self.name + str(i) for i in range(nitems)]
             # map-reduce formulation
             self._collate_results(self._node_runner(self._make_nodes(cwd),

nipype/pipeline/tests/test_engine.py

@@ -470,6 +470,39 @@ def test_mapnode_iterfield_check():
     yield assert_raises, ValueError, mod1._check_iterfield
 
 
+def test_mapnode_nested():
+    cwd = os.getcwd()
+    wd = mkdtemp()
+    os.chdir(wd)
+    from nipype import MapNode, Function
+    def func1(in1):
+        return in1 + 1
+    n1 = MapNode(Function(input_names=['in1'],
+                          output_names=['out'],
+                          function=func1),
+                 iterfield=['in1'],
+                 nested=True,
+                 name='n1')
+    n1.inputs.in1 = [[1,[2]],3,[4,5]]
+    n1.run()
+    print n1.get_output('out')
+    yield assert_equal, n1.get_output('out'), [[2,[3]],4,[5,6]]
+    
+    n2 = MapNode(Function(input_names=['in1'],
+                          output_names=['out'],
+                          function=func1),
+                 iterfield=['in1'],
+                 nested=False,
+                 name='n1')
+    n2.inputs.in1 = [[1,[2]],3,[4,5]]
+    error_raised = False
+    try:
+        n2.run()
+    except Exception, e:
+        pe.logger.info('Exception: %s' % str(e))
+        error_raised = True
+    yield assert_true, error_raised
+
 def test_node_hash():
     cwd = os.getcwd()
     wd = mkdtemp()