Merge pull request #1712 from mgxd/fix/join

fix: joinnode connection bugfix

Author: satra

nipype/pipeline/engine/base.py

@@ -69,6 +69,13 @@ def fullname(self):
             fullname = self._hierarchy + '.' + self.name
         return fullname
 
+    @property
+    def itername(self):
+        itername = self._id
+        if self._hierarchy:
+            itername = self._hierarchy + '.' + self._id
+        return itername
+
     def clone(self, name):
         """Clone an EngineBase object
 

nipype/pipeline/engine/tests/test_join.py

@@ -602,6 +602,46 @@ def test_set_join_node_file_input():
     os.chdir(cwd)
     rmtree(wd)
 
+def test_nested_workflow_join():
+    """Test collecting join inputs within a nested workflow"""
+    cwd = os.getcwd()
+    wd = mkdtemp()
+    os.chdir(wd)
+
+    # Make the nested workflow
+    def nested_wf(i, name='smallwf'):
+        #iterables with list of nums
+        inputspec = pe.Node(IdentityInterface(fields=['n']), name='inputspec')
+        inputspec.iterables = [('n', i)]
+        # increment each iterable before joining
+        pre_join = pe.Node(IncrementInterface(),
+                           name='pre_join')
+        # rejoin nums into list
+        join = pe.JoinNode(IdentityInterface(fields=['n']),
+                          joinsource='inputspec',
+                          joinfield='n',
+                          name='join')
+        #define and connect nested workflow
+        wf = pe.Workflow(name='wf_%d'%i[0])
+        wf.connect(inputspec, 'n', pre_join, 'input1')
+        wf.connect(pre_join, 'output1', join, 'n')
+        return wf
+    # master wf
+    meta_wf = pe.Workflow(name='meta', base_dir='.')
+    # add each mini-workflow to master
+    for i in [[1,3],[2,4]]:
+        mini_wf = nested_wf(i)
+        meta_wf.add_nodes([mini_wf])
+
+    result = meta_wf.run()
+    
+    # there should be six nodes in total
+    assert_equal(len(result.nodes()), 6,
+             "The number of expanded nodes is incorrect.")
+    
+    os.chdir(cwd)
+    rmtree(wd)
+
 
 if __name__ == "__main__":
     import nose

nipype/pipeline/engine/utils.py

@@ -711,7 +711,7 @@ def generate_expanded_graph(graph_in):
             in_edges = jedge_dict[jnode] = {}
             edges2remove = []
             for src, dest, data in graph_in.in_edges_iter(jnode, True):
-                in_edges[src._id] = data
+                in_edges[src.itername] = data
                 edges2remove.append((src, dest))
 
             for src, dest in edges2remove:
@@ -796,7 +796,7 @@ def make_field_func(*pair):
             expansions = defaultdict(list)
             for node in graph_in.nodes_iter():
                 for src_id, edge_data in list(old_edge_dict.items()):
-                    if node._id.startswith(src_id):
+                    if node.itername.startswith(src_id):
                         expansions[src_id].append(node)
             for in_id, in_nodes in list(expansions.items()):
                 logger.debug("The join node %s input %s was expanded"