def test_identity_connect_first_field(): import nipype.pipeline.engine as pe from nipype.interfaces.utility import IdentityInterface workflow = pe.Workflow(name='test') a = pe.Node(IdentityInterface(fields=['a']),name='a') b = pe.Node(IdentityInterface(fields=['a']),name='b') workflow.connect(a, 'a', b, 'a') yield assert_true, hasattr(workflow.inputs.a, 'a') yield assert_true, hasattr(b.inputs, 'a') yield assert_true, hasattr(workflow.inputs.b, 'a')
workflow.add_nodes([a, b])
import nipype.pipeline.engine as pefrom nipype.interfaces.utility import (IdentityInterface, Merge)outer_wf = pe.Workflow(name='test')a = pe.Node(IdentityInterface(fields=['a']),name='a')b = pe.Node(IdentityInterface(fields=['a']),name='b')c = pe.Node(Merge(2), name='c')inner_wf = pe.Workflow(name='inner')inner_wf.add_nodes([b])outer_wf.connect(a, 'a', inner_wf, 'b.a')outer_wf.connect(a, 'a', c, 'in1')outer_wf.connect(inner_wf, 'b.a', c, 'in2')
b = pe.Node(IdentityInterface(fields=['dummy', 'a']),name='b')
I encountered an odd problem with IdentityInterface. The following test fails in the last line:
def test_identity_connect_first_field():import nipype.pipeline.engine as pefrom nipype.interfaces.utility import IdentityInterfaceworkflow = pe.Workflow(name='test')a = pe.Node(IdentityInterface(fields=['a']),name='a')b = pe.Node(IdentityInterface(fields=['a']),name='b')workflow.connect(a, 'a', b, 'a')yield assert_true, hasattr(workflow.inputs.a, 'a')yield assert_true, hasattr(b.inputs, 'a')yield assert_true, hasattr(workflow.inputs.b, 'a')
Does this failure occur for others?
Interestingly, the problem does not occur if the connect line is replace with:
workflow.add_nodes([a, b])
The missing attribute causes other failures in practice, e.g. in the following which fails on the last line:
import nipype.pipeline.engine as pefrom nipype.interfaces.utility import (IdentityInterface, Merge)outer_wf = pe.Workflow(name='test')a = pe.Node(IdentityInterface(fields=['a']),name='a')b = pe.Node(IdentityInterface(fields=['a']),name='b')c = pe.Node(Merge(2), name='c')inner_wf = pe.Workflow(name='inner')inner_wf.add_nodes([b])outer_wf.connect(a, 'a', inner_wf, 'b.a')outer_wf.connect(a, 'a', c, 'in1')outer_wf.connect(inner_wf, 'b.a', c, 'in2')