Handle the alternate synchronize iterables format.

Make iterables validation more robust.

Author: FredLoney

nipype/pipeline/tests/test_engine.py

@@ -269,6 +269,65 @@ def test_itersource_expansion():
     # => 3 * 14 = 42 nodes in the group
     yield assert_equal, len(pe.generate_expanded_graph(wf3._flatgraph).nodes()), 42
 
+def test_itersource_synchronize1_expansion():
+    import nipype.pipeline.engine as pe
+    wf1 = pe.Workflow(name='test')
+    node1 = pe.Node(TestInterface(),name='node1')
+    node1.iterables = [('input1',[1,2]), ('input2',[3,4])]
+    node1.synchronize = True
+    node2 = pe.Node(TestInterface(),name='node2')
+    wf1.connect(node1,'output1', node2, 'input1')
+    node3 = pe.Node(TestInterface(),name='node3')
+    node3.itersource = ('node1', ['input1', 'input2'])
+    node3.iterables = [('input1', {(1,3):[5,6]}), ('input2', {(1,3):[7,8], (2,4): [9]})]
+    wf1.connect(node2,'output1', node3, 'input1')
+    node4 = pe.Node(TestInterface(),name='node4')
+    wf1.connect(node3,'output1', node4, 'input1')
+    wf3 = pe.Workflow(name='group')
+    for i in [0,1,2]:
+        wf3.add_nodes([wf1.clone(name='test%d'%i)])
+    wf3._flatgraph = wf3._create_flat_graph()
+    
+    # each expanded graph clone has:
+    # 2 node1 expansion nodes,
+    # 1 node2 per node1 replicate,
+    # 2 node3 replicates for the node1 input1 value 1,
+    # 3 node3 replicates for the node1 input1 value 2 and
+    # 1 node4 successor per node3 replicate
+    # => 2 + 2 + (2 + 3) + 5 = 14 nodes per expanded graph clone
+    # => 3 * 14 = 42 nodes in the group
+    yield assert_equal, len(pe.generate_expanded_graph(wf3._flatgraph).nodes()), 42
+
+def test_itersource_synchronize2_expansion():
+    import nipype.pipeline.engine as pe
+    wf1 = pe.Workflow(name='test')
+    node1 = pe.Node(TestInterface(),name='node1')
+    node1.iterables = [('input1',[1,2]), ('input2',[3,4])]
+    node1.synchronize = True
+    node2 = pe.Node(TestInterface(),name='node2')
+    wf1.connect(node1,'output1', node2, 'input1')
+    node3 = pe.Node(TestInterface(),name='node3')
+    node3.itersource = ('node1', ['input1', 'input2'])
+    node3.synchronize = True
+    node3.iterables = [('input1', 'input2'), {(1,3):[(5,7), (6,8)], (2,4):[(None,9)]}]
+    wf1.connect(node2,'output1', node3, 'input1')
+    node4 = pe.Node(TestInterface(),name='node4')
+    wf1.connect(node3,'output1', node4, 'input1')
+    wf3 = pe.Workflow(name='group')
+    for i in [0,1,2]:
+        wf3.add_nodes([wf1.clone(name='test%d'%i)])
+    wf3._flatgraph = wf3._create_flat_graph()
+    
+    # each expanded graph clone has:
+    # 2 node1 expansion nodes,
+    # 1 node2 per node1 replicate,
+    # 2 node3 replicates for the node1 input1 value 1,
+    # 1 node3 replicates for the node1 input1 value 2 and
+    # 1 node4 successor per node3 replicate
+    # => 2 + 2 + (2 + 1) + 3 = 10 nodes per expanded graph clone
+    # => 3 * 10 = 30 nodes in the group
+    yield assert_equal, len(pe.generate_expanded_graph(wf3._flatgraph).nodes()), 30
+
 def test_disconnect():
     import nipype.pipeline.engine as pe
     from nipype.interfaces.utility import IdentityInterface

nipype/pipeline/utils.py

@@ -645,14 +645,15 @@ def generate_expanded_graph(graph_in):
                 key = src_values[0]
             else:
                 key = tuple(src_values)
-            # the iterables is a {field: lambda} dictionary, where the
-            # lambda returns a {source key: iteration list} dictionary
-            iterables = {}
-            for field, func in inode.iterables.iteritems():
-                # the {source key: iteration list} dictionary
-                lookup = func()
-                if lookup.has_key(key):
-                    iterables[field] = lambda: lookup[key]
+            # The itersource iterables is a {field: lookup} dictionary, where the
+            # lookup is a {source key: iteration list} dictionary. Look up the
+            # current iterable value using the predecessor itersource input values.
+            iter_dict = {field: lookup[key] for field, lookup in inode.iterables
+                         if lookup.has_key(key)}
+            # convert the iterables to the standard {field: function} format
+            iter_items = map(lambda(field, value): (field, lambda: value),
+                             iter_dict.iteritems())
+            iterables = dict(iter_items)
         else:
             iterables = inode.iterables.copy()
         inode.iterables = None
@@ -800,56 +801,82 @@ def _standardize_iterables(node):
     iterables = node.iterables
     # The candidate iterable fields
     fields = set(node.inputs.copyable_trait_names())
-    
-    # Synchronize iterables can be in [fields, value tuples] format
-    # rather than [(field, value list), (field, value list), ...]
-    if node.synchronize and len(iterables) == 2:
-        first, last = iterables
-        if all((isinstance(item, str) and item in fields
-                for item in first)):
-            iterables = _transpose_iterables(first, last)
-            
+    # Flag indicating whether the iterables are in the alternate
+    # synchronize form and are not converted to a standard format.
+    synchronize = False
+    # A synchronize iterables node without an itersource can be in
+    # [fields, value tuples] format rather than
+    # [(field, value list), (field, value list), ...]
+    if node.synchronize:
+        if len(iterables) == 2:
+            first, last = iterables
+            if all((isinstance(item, str) and item in fields
+                    for item in first)):
+                iterables = _transpose_iterables(first, last)
+
     # Convert a tuple to a list
     if isinstance(iterables, tuple):
         iterables = [iterables]
+    # Validate the standard [(field, values)] format
+    _validate_iterables(node, iterables, fields)
     # Convert a list to a dictionary
     if isinstance(iterables, list):
-        # Validate the format
-        for item in iterables:
-            try:
-                if len(item) != 2:
-                    raise ValueError("The %s iterables do not consist of"
-                                     " (field, values) pairs" % node.name)
-            except TypeError, e:
-                raise TypeError("The %s iterables is not iterable: %s"
-                                % (node.name, e))
-        # Convert the values to functions. This is a legacy Nipype
-        # requirement with unknown rationale.
-        iter_items = map(lambda(field, value): (field, lambda: value),
-                         iterables)
-        # Make the iterables dictionary
-        iterables = dict(iter_items)
-    elif not isinstance(iterables, dict):
+        # Convert a values list to a function. This is a legacy
+        # Nipype requirement with unknown rationale.
+        if not node.itersource:
+            iter_items = map(lambda(field, value): (field, lambda: value),
+                             iterables)
+            iterables = dict(iter_items)
+    node.iterables = iterables
+
+def _validate_iterables(node, iterables, fields):
+    """
+    Raise TypeError if an iterables member is not iterable.
+
+    Raise ValueError if an iterables member is not a (field, values) pair.
+
+    Raise ValueError if an iterable field is not in the inputs.
+    """
+    # The iterables can be a {field: value list} dictionary.
+    if isinstance(iterables, dict):
+        iterables = iterables.items()
+    elif not isinstance(iterables, tuple) and not isinstance(iterables, list):
         raise ValueError("The %s iterables type is not a list or a dictionary:"
                          " %s" % (node.name, iterables.__class__))
-    
-    # Validate the iterable fields
-    for field in iterables.iterkeys():
+    for item in iterables:
+        try:
+            if len(item) != 2:
+                raise ValueError("The %s iterables is not a [(field, values)]"
+                                 " list" % node.name)
+        except TypeError, e:
+            raise TypeError("A %s iterables member is not iterable: %s"
+                            % (node.name, e))
+        field, _ = item
         if field not in fields:
             raise ValueError("The %s iterables field is unrecognized: %s"
                              % (node.name, field))
-    
-    # Assign to the standard form
-    node.iterables = iterables
 
 def _transpose_iterables(fields, values):
     """
-    Converts the given fields and tuple values into a list of
-    iterable (field: value list) pairs, suitable for setting
-    a node iterables property.
+    Converts the given fields and tuple values into a standardized
+    iterables value.
+    
+    If the input values is a synchronize iterables dictionary, then
+    the result is a (field, {key: values}) list.
+    
+    Otherwise, the result is a list of (field: value list) pairs.
     """
-    return zip(fields, [filter(lambda(v): v != None, list(transpose))
-                        for transpose in zip(*values)])
+    if isinstance(values, dict):
+        transposed = {field: defaultdict(list) for field in fields}
+        for key, tuples in values.iteritems():
+            for kvals in tuples:
+                for idx, val in enumerate(kvals):
+                    if val != None:
+                        transposed[fields[idx]][key].append(val)
+        return transposed.items()
+    else:
+        return zip(fields, [filter(lambda(v): v != None, list(transpose))
+                            for transpose in zip(*values)])
 
 def export_graph(graph_in, base_dir=None, show=False, use_execgraph=False,
                  show_connectinfo=False, dotfilename='graph.dot', format='png',