Merge pull request #159 from djarecka/empty_input_list

fixing issues with state input fields that are empty lists

Author: djarecka

pydra/engine/core.py

@@ -427,6 +427,14 @@ def done(self):
             # TODO: only check for needed state result
             if self.result() and all(self.result()):
                 return True
+            # checking if self.result() is not an empty list only because
+            # the states_ind is an empty list (input field might be an empty list)
+            elif (
+                self.result() == []
+                and hasattr(self.state, "states_ind")
+                and self.state.states_ind == []
+            ):
+                return True
         else:
             if self.result():
                 return True

pydra/engine/specs.py

@@ -487,7 +487,7 @@ def get_value(self, wf, state_index=None):
             node = getattr(wf, self.name)
             result = node.result(state_index=state_index)
             if isinstance(result, list):
-                if isinstance(result[0], list):
+                if len(result) and isinstance(result[0], list):
                     results_new = []
                     for res_l in result:
                         if self.field == "all_":
@@ -534,6 +534,6 @@ def reset(self):
 def path_to_string(value):
     if isinstance(value, Path):
         value = str(value)
-    elif isinstance(value, list) and isinstance(value[0], Path):
+    elif isinstance(value, list) and len(value) and isinstance(value[0], Path):
         value = [str(val) for val in value]
     return value

pydra/engine/task.py

@@ -159,7 +159,7 @@ def _run_task(self):
         del inputs["_func"]
         self.output_ = None
         output = cp.loads(self.inputs._func)(**inputs)
-        if output:
+        if output is not None:
             output_names = [el[0] for el in self.output_spec.fields]
             self.output_ = {}
             if len(output_names) > 1:

pydra/engine/tests/test_node_task.py

@@ -269,6 +269,20 @@ def test_task_init_5c():
     assert nn.state.splitter_rpn_final == ["NA.a"]
 
 
+def test_task_init_6():
+    """ task with splitter, but the input is an empty list"""
+    nn = fun_addtwo(name="NA", a=[])
+    nn.split(splitter="a")
+    assert nn.inputs.a == []
+
+    assert nn.state.splitter == "NA.a"
+    assert nn.state.splitter_rpn == ["NA.a"]
+
+    nn.state.prepare_states(nn.inputs)
+    assert nn.state.states_ind == []
+    assert nn.state.states_val == []
+
+
 def test_task_error():
     func = fun_div(name="div", a=1, b=0)
     with pytest.raises(ZeroDivisionError):
@@ -607,6 +621,27 @@ def test_task_state_singl_1(plugin):
         assert odir.exists()
 
 
+@pytest.mark.parametrize("plugin", Plugins)
+def test_task_state_2(plugin):
+    """ task with the simplest splitter, the input is an empty list"""
+    nn = fun_addtwo(name="NA").split(splitter="a", a=[])
+
+    assert nn.state.splitter == "NA.a"
+    assert nn.state.splitter_rpn == ["NA.a"]
+    assert nn.inputs.a == []
+
+    with Submitter(plugin=plugin) as sub:
+        sub(nn)
+
+    # checking the results
+    results = nn.result()
+    expected = []
+    for i, res in enumerate(expected):
+        assert results[i].output.out == res[1]
+    # checking the output_dir
+    assert nn.output_dir == []
+
+
 @pytest.mark.parametrize("plugin", Plugins)
 def test_task_state_comb_1(plugin):
     """ task with the simplest splitter and combiner"""
@@ -775,6 +810,27 @@ def test_task_state_comb_singl_1(plugin):
         assert odir.exists()
 
 
+@pytest.mark.parametrize("plugin", Plugins)
+def test_task_state_comb_2(plugin):
+    """ task with the simplest splitter, the input is an empty list"""
+    nn = fun_addtwo(name="NA").split(splitter="a", a=[]).combine(combiner=["a"])
+
+    assert nn.state.splitter == "NA.a"
+    assert nn.state.splitter_rpn == ["NA.a"]
+    assert nn.inputs.a == []
+
+    with Submitter(plugin=plugin) as sub:
+        sub(nn)
+
+    # checking the results
+    results = nn.result()
+    expected = []
+    for i, res in enumerate(expected):
+        assert results[i].output.out == res[1]
+    # checking the output_dir
+    assert nn.output_dir == []
+
+
 @pytest.mark.parametrize("plugin", Plugins)
 def test_task_state_comb_order(plugin):
     """ tasks with an outer splitter and various combiner;

pydra/engine/tests/test_workflow.py

@@ -302,6 +302,28 @@ def test_wf_st_1_call_plug(plugin):
         assert odir.exists()
 
 
+@pytest.mark.parametrize("plugin", Plugins)
+def test_wf_st_noinput_1(plugin):
+    """ Workflow with one task, a splitter for the workflow"""
+    wf = Workflow(name="wf_spl_1", input_spec=["x"])
+    wf.add(add2(name="add2", x=wf.lzin.x))
+
+    wf.split(("x"))
+    wf.inputs.x = []
+    wf.set_output([("out", wf.add2.lzout.out)])
+    wf.plugin = plugin
+
+    checksum_before = wf.checksum
+    with Submitter(plugin=plugin) as sub:
+        sub(wf)
+
+    assert wf.checksum == checksum_before
+    results = wf.result()
+    assert results == []
+    # checking all directories
+    assert wf.output_dir == []
+
+
 @pytest.mark.parametrize("plugin", Plugins)
 def test_wf_ndst_1(plugin):
     """ workflow with one task, a splitter on the task level"""
@@ -394,6 +416,26 @@ def test_wf_ndst_updateinp_1(plugin):
     assert wf.output_dir.exists()
 
 
+@pytest.mark.parametrize("plugin", Plugins)
+def test_wf_ndst_noinput_1(plugin):
+    """ workflow with one task, a splitter on the task level"""
+    wf = Workflow(name="wf_spl_1", input_spec=["x"])
+    wf.add(add2(name="add2", x=wf.lzin.x).split("x"))
+    wf.inputs.x = []
+    wf.set_output([("out", wf.add2.lzout.out)])
+    wf.plugin = plugin
+
+    checksum_before = wf.checksum
+    with Submitter(plugin=plugin) as sub:
+        sub(wf)
+
+    assert wf.checksum == checksum_before
+    results = wf.result()
+
+    assert results.output.out == []
+    assert wf.output_dir.exists()
+
+
 @pytest.mark.parametrize("plugin", Plugins)
 def test_wf_st_2(plugin):
     """ workflow with one task, splitters and combiner for workflow"""
@@ -538,12 +580,60 @@ def test_wf_ndst_4(plugin):
 
 @pytest.mark.parametrize("plugin", Plugins)
 def test_wf_st_5(plugin):
-    """ workflow with two tasks, outer splitter and combiner for the workflow"""
+    """ workflow with two tasks, outer splitter and no combiner"""
     wf = Workflow(name="wf_st_5", input_spec=["x", "y"])
     wf.add(multiply(name="mult", x=wf.lzin.x, y=wf.lzin.y))
     wf.add(add2(name="add2", x=wf.mult.lzout.out))
 
     wf.split(["x", "y"], x=[1, 2], y=[11, 12])
+    wf.set_output([("out", wf.add2.lzout.out)])
+    wf.plugin = plugin
+
+    with Submitter(plugin=plugin) as sub:
+        sub(wf)
+
+    results = wf.result()
+    assert results[0].output.out == 13
+    assert results[1].output.out == 14
+    assert results[2].output.out == 24
+    assert results[3].output.out == 26
+    # checking all directories
+    assert wf.output_dir
+    for odir in wf.output_dir:
+        assert odir.exists()
+
+
+@pytest.mark.parametrize("plugin", Plugins)
+def test_wf_ndst_5(plugin):
+    """ workflow with two tasks, outer splitter on tasks level and no combiner"""
+    wf = Workflow(name="wf_ndst_5", input_spec=["x", "y"])
+    wf.add(multiply(name="mult", x=wf.lzin.x, y=wf.lzin.y).split(["x", "y"]))
+    wf.add(add2(name="add2", x=wf.mult.lzout.out))
+    wf.inputs.x = [1, 2]
+    wf.inputs.y = [11, 12]
+    wf.set_output([("out", wf.add2.lzout.out)])
+    wf.plugin = plugin
+
+    with Submitter(plugin=plugin) as sub:
+        sub(wf)
+
+    results = wf.result()
+    assert results.output.out[0] == 13
+    assert results.output.out[1] == 14
+    assert results.output.out[2] == 24
+    assert results.output.out[3] == 26
+    # checking the output directory
+    assert wf.output_dir.exists()
+
+
+@pytest.mark.parametrize("plugin", Plugins)
+def test_wf_st_6(plugin):
+    """ workflow with two tasks, outer splitter and combiner for the workflow"""
+    wf = Workflow(name="wf_st_6", input_spec=["x", "y"])
+    wf.add(multiply(name="mult", x=wf.lzin.x, y=wf.lzin.y))
+    wf.add(add2(name="add2", x=wf.mult.lzout.out))
+
+    wf.split(["x", "y"], x=[1, 2, 3], y=[11, 12])
     wf.combine("x")
     wf.set_output([("out", wf.add2.lzout.out)])
     wf.plugin = plugin
@@ -554,21 +644,23 @@ def test_wf_st_5(plugin):
     results = wf.result()
     assert results[0][0].output.out == 13
     assert results[0][1].output.out == 24
+    assert results[0][2].output.out == 35
     assert results[1][0].output.out == 14
     assert results[1][1].output.out == 26
+    assert results[1][2].output.out == 38
     # checking all directories
     assert wf.output_dir
     for odir in wf.output_dir:
         assert odir.exists()
 
 
 @pytest.mark.parametrize("plugin", Plugins)
-def test_wf_ndst_5(plugin):
+def test_wf_ndst_6(plugin):
     """ workflow with two tasks, outer splitter and combiner on tasks level"""
-    wf = Workflow(name="wf_ndst_5", input_spec=["x", "y"])
+    wf = Workflow(name="wf_ndst_6", input_spec=["x", "y"])
     wf.add(multiply(name="mult", x=wf.lzin.x, y=wf.lzin.y).split(["x", "y"]))
     wf.add(add2(name="add2", x=wf.mult.lzout.out).combine("mult.x"))
-    wf.inputs.x = [1, 2]
+    wf.inputs.x = [1, 2, 3]
     wf.inputs.y = [11, 12]
     wf.set_output([("out", wf.add2.lzout.out)])
     wf.plugin = plugin
@@ -577,8 +669,9 @@ def test_wf_ndst_5(plugin):
         sub(wf)
 
     results = wf.result()
-    assert results.output.out[0] == [13, 24]
-    assert results.output.out[1] == [14, 26]
+    assert results.output.out[0] == [13, 24, 35]
+    assert results.output.out[1] == [14, 26, 38]
+
     # checking the output directory
     assert wf.output_dir.exists()
 
@@ -587,11 +680,11 @@ def test_wf_ndst_5(plugin):
 
 
 @pytest.mark.parametrize("plugin", Plugins)
-def test_wf_st_6(plugin):
+def test_wf_st_7(plugin):
     """ workflow with three tasks, third one connected to two previous tasks,
         splitter on the workflow level
     """
-    wf = Workflow(name="wf_st_6", input_spec=["x", "y"])
+    wf = Workflow(name="wf_st_7", input_spec=["x", "y"])
     wf.add(add2(name="add2x", x=wf.lzin.x))
     wf.add(add2(name="add2y", x=wf.lzin.y))
     wf.add(multiply(name="mult", x=wf.add2x.lzout.out, y=wf.add2y.lzout.out))
@@ -615,11 +708,11 @@ def test_wf_st_6(plugin):
 
 
 @pytest.mark.parametrize("plugin", Plugins)
-def test_wf_ndst_6(plugin):
+def test_wf_ndst_7(plugin):
     """ workflow with three tasks, third one connected to two previous tasks,
         splitter on the tasks levels
     """
-    wf = Workflow(name="wf_ndst_6", input_spec=["x", "y"])
+    wf = Workflow(name="wf_ndst_7", input_spec=["x", "y"])
     wf.add(add2(name="add2x", x=wf.lzin.x).split("x"))
     wf.add(add2(name="add2y", x=wf.lzin.y).split("x"))
     wf.add(multiply(name="mult", x=wf.add2x.lzout.out, y=wf.add2y.lzout.out))
@@ -639,11 +732,11 @@ def test_wf_ndst_6(plugin):
 
 
 @pytest.mark.parametrize("plugin", Plugins)
-def test_wf_st_7(plugin):
+def test_wf_st_8(plugin):
     """ workflow with three tasks, third one connected to two previous tasks,
         splitter and partial combiner on the workflow level
     """
-    wf = Workflow(name="wf_st_7", input_spec=["x", "y"])
+    wf = Workflow(name="wf_st_8", input_spec=["x", "y"])
     wf.add(add2(name="add2x", x=wf.lzin.x))
     wf.add(add2(name="add2y", x=wf.lzin.y))
     wf.add(multiply(name="mult", x=wf.add2x.lzout.out, y=wf.add2y.lzout.out))
@@ -670,11 +763,11 @@ def test_wf_st_7(plugin):
 
 
 @pytest.mark.parametrize("plugin", Plugins)
-def test_wf_ndst_7(plugin):
+def test_wf_ndst_8(plugin):
     """ workflow with three tasks, third one connected to two previous tasks,
         splitter and partial combiner on the tasks levels
     """
-    wf = Workflow(name="wf_ndst_7", input_spec=["x", "y"])
+    wf = Workflow(name="wf_ndst_8", input_spec=["x", "y"])
     wf.add(add2(name="add2x", x=wf.lzin.x).split("x"))
     wf.add(add2(name="add2y", x=wf.lzin.y).split("x"))
     wf.add(
@@ -699,11 +792,11 @@ def test_wf_ndst_7(plugin):
 
 
 @pytest.mark.parametrize("plugin", Plugins)
-def test_wf_st_8(plugin):
+def test_wf_st_9(plugin):
     """ workflow with three tasks, third one connected to two previous tasks,
         splitter and partial combiner (from the second task) on the workflow level
     """
-    wf = Workflow(name="wf_st_8", input_spec=["x", "y"])
+    wf = Workflow(name="wf_st_9", input_spec=["x", "y"])
     wf.add(add2(name="add2x", x=wf.lzin.x))
     wf.add(add2(name="add2y", x=wf.lzin.y))
     wf.add(multiply(name="mult", x=wf.add2x.lzout.out, y=wf.add2y.lzout.out))
@@ -730,11 +823,11 @@ def test_wf_st_8(plugin):
 
 
 @pytest.mark.parametrize("plugin", Plugins)
-def test_wf_ndst_8(plugin):
+def test_wf_ndst_9(plugin):
     """ workflow with three tasks, third one connected to two previous tasks,
         splitter and partial combiner (from the second task) on the tasks levels
     """
-    wf = Workflow(name="wf_ndst_8", input_spec=["x", "y"])
+    wf = Workflow(name="wf_ndst_9", input_spec=["x", "y"])
     wf.add(add2(name="add2x", x=wf.lzin.x).split("x"))
     wf.add(add2(name="add2y", x=wf.lzin.y).split("x"))
     wf.add(
@@ -760,11 +853,11 @@ def test_wf_ndst_8(plugin):
 
 
 @pytest.mark.parametrize("plugin", Plugins)
-def test_wf_st_9(plugin):
+def test_wf_st_10(plugin):
     """ workflow with three tasks, third one connected to two previous tasks,
         splitter and full combiner on the workflow level
     """
-    wf = Workflow(name="wf_st_9", input_spec=["x", "y"])
+    wf = Workflow(name="wf_st_10", input_spec=["x", "y"])
     wf.add(add2(name="add2x", x=wf.lzin.x))
     wf.add(add2(name="add2y", x=wf.lzin.y))
     wf.add(multiply(name="mult", x=wf.add2x.lzout.out, y=wf.add2y.lzout.out))
@@ -790,11 +883,11 @@ def test_wf_st_9(plugin):
 
 
 @pytest.mark.parametrize("plugin", Plugins)
-def test_wf_ndst_9(plugin):
+def test_wf_ndst_10(plugin):
     """ workflow with three tasks, third one connected to two previous tasks,
         splitter and full combiner on the tasks levels
     """
-    wf = Workflow(name="wf_ndst_9", input_spec=["x", "y"])
+    wf = Workflow(name="wf_ndst_10", input_spec=["x", "y"])
     wf.add(add2(name="add2x", x=wf.lzin.x).split("x"))
     wf.add(add2(name="add2y", x=wf.lzin.y).split("x"))
     wf.add(