Support a Set join field.

Author: FredLoney

nipype/pipeline/engine.py

@@ -1855,9 +1855,10 @@ def _override_join_traits(self, basetraits, fields):
  raise ValueError("The JoinNode %s does not have a field"
  " named %s" % (self.name, field))
  for name, trait in basetraits.items():
- # if a join field is a list, then the item trait is the list
- # inner trait. Otherwise, the item trait is a new Any trait.
- if name in fields and isinstance(trait.trait_type, traits.List):
+ # if a join field has a single inner trait, then the item
+ # trait is that inner trait. Otherwise, the item trait is
+ # a new Any trait.
+ if name in fields and len(trait.inner_traits) == 1:
  item_trait = trait.inner_traits[0]
  dyntraits.add_trait(name, item_trait)
  logger.debug("Converted the join node %s field %s"
@@ -1880,13 +1881,17 @@ def _collate_join_field_inputs(self):
  for field in self.joinfield:
  val = self._collate_input_value(field)
  setattr(self._interface.inputs, field, val)
- logger.debug("Collated %d inputs into each %s node join field"
+ logger.debug("Collated %d inputs into the %s node join fields"
  % (self._next_slot_index, self))
 
  def _collate_input_value(self, field):
- return [getattr(self._inputs, self._join_item_field_name(field, idx))
+ val = [getattr(self._inputs, self._join_item_field_name(field, idx))
  for idx in range(self._next_slot_index)]
-
+ basetrait = self._interface.inputs.trait(field)
+ if isinstance(basetrait.trait_type, traits.Set):
+ return set(val)
+ else:
+ return val
 
 class MapNode(Node):
  """Wraps interface objects that need to be iterated on a list of inputs.

nipype/pipeline/tests/test_join.py

@@ -63,6 +63,31 @@ def _list_outputs(self):
  return outputs
 
 
+_set_len = None
+"""The Set interface execution result."""
+
+class SetInputSpec(nib.TraitedSpec):
+ input1 = nib.traits.Set(nib.traits.Int, mandatory=True, desc='input')
+
+class SetOutputSpec(nib.TraitedSpec):
+ output1 = nib.traits.Int(desc='ouput')
+
+class SetInterface(nib.BaseInterface):
+ input_spec = SetInputSpec
+ output_spec = SetOutputSpec
+
+ def _run_interface(self, runtime):
+ runtime.returncode = 0
+ return runtime
+
+ def _list_outputs(self):
+ global _set_len
+ outputs = self._outputs().get()
+ _set_len = outputs['output1'] = len(self.inputs.input1)
+ print ">>>>>>>>SET LEN: %d" % _set_len 
+ return outputs
+
+
 _products = []
 """The Products interface execution results."""
 
@@ -136,6 +161,32 @@ def test_join_expansion():
  os.chdir(cwd)
  rmtree(wd)
 
+def test_set_join_node():
+ cwd = os.getcwd()
+ wd = mkdtemp()
+ os.chdir(wd)
+
+ # Make the workflow.
+ wf = pe.Workflow(name='test')
+ # the iterated input node
+ inputspec = pe.Node(IdentityInterface(fields=['n']), name='inputspec')
+ inputspec.iterables = [('n', [1, 2, 1, 3, 2])]
+ # a pre-join node in the iterated path
+ pre_join1 = pe.Node(IncrementInterface(), name='pre_join1')
+ wf.connect(inputspec, 'n', pre_join1, 'input1')
+ # the set join node
+ join = pe.JoinNode(SetInterface(), joinsource='inputspec',
+ joinfield='input1', name='join')
+ wf.connect(pre_join1, 'output1', join, 'input1')
+ 
+ wf.run()
+ 
+ # the join length is the number of unique inputs
+ assert_equal(_set_len, 3, "The join Set output value is incorrect: %s." % _set_len)
+
+ os.chdir(cwd)
+ rmtree(wd)
+
 
 if __name__ == "__main__":
  import nose