Merge remote-tracking branch 'origin/pr/851'

Author: chrisgorgo

nipype/pipeline/engine.py

@@ -2008,7 +2008,7 @@ class MapNode(Node):
 
     """
 
-    def __init__(self, interface, iterfield, name, **kwargs):
+    def __init__(self, interface, iterfield, name, serial=False, **kwargs):
         """
 
         Parameters
@@ -2022,9 +2022,12 @@ def __init__(self, interface, iterfield, name, **kwargs):
             paired (i.e. it does not compute a combinatorial product).
         name : alphanumeric string
             node specific name
-
+        serial : boolean
+            flag to enforce executing the jobs of the mapnode in a serial manner rather than parallel
         See Node docstring for additional keyword arguments.
         """
+
+
         super(MapNode, self).__init__(interface, name, **kwargs)
         if isinstance(iterfield, str):
             iterfield = [iterfield]
@@ -2033,6 +2036,8 @@ def __init__(self, interface, iterfield, name, **kwargs):
                                                    fields=self.iterfield)
         self._inputs.on_trait_change(self._set_mapnode_input)
         self._got_inputs = False
+        
+        self._serial = serial
 
     def _create_dynamic_traits(self, basetraits, fields=None, nitems=None):
         """Convert specific fields of a trait to accept multiple inputs
@@ -2223,7 +2228,10 @@ def num_subnodes(self):
             self._get_inputs()
             self._got_inputs = True
         self._check_iterfield()
-        return len(filename_to_list(getattr(self.inputs, self.iterfield[0])))
+        if self._serial :
+            return 1
+        else:
+            return len(filename_to_list(getattr(self.inputs, self.iterfield[0])))
 
     def _get_inputs(self):
         old_inputs = self._inputs.get()

nipype/pipeline/tests/test_engine.py

@@ -10,8 +10,7 @@
 
 import networkx as nx
 
-from nipype.testing import (assert_raises, assert_equal, assert_true,
-                            assert_false)
+from nipype.testing import (assert_raises, assert_equal, assert_true, assert_false)
 import nipype.interfaces.base as nib
 import nipype.pipeline.engine as pe
 from nipype import logging
@@ -551,7 +550,7 @@ def func2(a):
     w1.connect(n1, ('a', modify), n2,'a')
     w1.base_dir = wd
 
-    w1.config = {'crashdump_dir': wd}
+    w1.config['execution']['crashdump_dir'] = wd
     # generate outputs
     error_raised = False
     try:
@@ -581,7 +580,7 @@ def func1(in1):
     n1.inputs.in1 = [1]
     w1 = Workflow(name='test')
     w1.base_dir = wd
-    w1.config = {'crashdump_dir': wd}
+    w1.config['execution']['crashdump_dir'] = wd
     w1.add_nodes([n1])
     w1.run()
     n1.inputs.in1 = [2]
@@ -606,3 +605,54 @@ def func1(in1):
     yield assert_false, error_raised
     os.chdir(cwd)
     rmtree(wd)
+
+def test_serial_input():
+    cwd = os.getcwd()
+    wd = mkdtemp()
+    os.chdir(wd)
+    from nipype import MapNode, Function, Workflow
+    def func1(in1):
+        return in1
+    n1 = MapNode(Function(input_names=['in1'],
+                          output_names=['out'],
+                          function=func1),
+                 iterfield=['in1'],
+                 name='n1')
+    n1.inputs.in1 = [1,2,3]
+    
+    
+    w1 = Workflow(name='test')
+    w1.base_dir = wd
+    w1.add_nodes([n1])
+    # set local check
+    w1.config['execution'] = {'stop_on_first_crash': 'true',
+                              'local_hash_check': 'true',
+                              'crashdump_dir': wd}
+    
+    # test output of num_subnodes method when serial is default (False)
+    yield assert_equal, n1.num_subnodes(), len(n1.inputs.in1)
+    
+    # test running the workflow on default conditions
+    error_raised = False
+    try:
+        w1.run(plugin='MultiProc')
+    except Exception, e:
+        pe.logger.info('Exception: %s' % str(e))
+        error_raised = True
+    yield assert_false, error_raised
+    
+    # test output of num_subnodes method when serial is True
+    n1._serial=True
+    yield assert_equal, n1.num_subnodes(), 1
+    
+    # test running the workflow on serial conditions
+    error_raised = False
+    try:
+        w1.run(plugin='MultiProc')
+    except Exception, e:
+        pe.logger.info('Exception: %s' % str(e))
+        error_raised = True
+    yield assert_false, error_raised
+    
+    os.chdir(cwd)
+    rmtree(wd)