fixing _ordering and small fix to the state class (using left_combiner_all instead of left_combiner)

Author: djarecka

pydra/engine/helpers_state.py

@@ -59,8 +59,8 @@ def _ordering(
                     splitter=other_states[node_nm][0].splitter_final, name=node_nm
                 )
                 el = (splitter_mod, el[1])
-            if other_states[node_nm][0].other_states:
-                other_states.update(other_states[node_nm][0].other_states)
+                if other_states[node_nm][0].other_states:
+                    other_states.update(other_states[node_nm][0].other_states)
         if type(el[1]) is str and el[1].startswith("_"):
             node_nm = el[1][1:]
             if node_nm not in other_states and state_fields:
@@ -74,8 +74,8 @@ def _ordering(
                     splitter=other_states[node_nm][0].splitter_final, name=node_nm
                 )
                 el = (el[0], splitter_mod)
-            if other_states[node_nm][0].other_states:
-                other_states.update(other_states[node_nm][0].other_states)
+                if other_states[node_nm][0].other_states:
+                    other_states.update(other_states[node_nm][0].other_states)
         _iterate_list(
             el,
             ".",

pydra/engine/state.py

@@ -136,7 +136,6 @@ def splitter_final(self):
 
     @property
     def splitter_rpn(self):
-        # TODO NOW: this also changes other_states (properly, but should be explicite)
         _splitter_rpn = hlpst.splitter2rpn(
             self.splitter, other_states=self.other_states
         )
@@ -215,11 +214,6 @@ def left_combiner(self):
         else:
             return list(set(self.combiner) - set(self.right_combiner))
 
-    # TODO: should i have setter
-    @left_combiner.setter
-    def left_combiner(self, left_combiner):
-        self._left_combiner = left_combiner
-
     @property
     def right_combiner_all(self):
         if hasattr(self, "_right_combiner_all"):
@@ -232,11 +226,11 @@ def left_combiner_all(self):
         if hasattr(self, "_left_combiner_all"):
             return self._left_combiner_all
         else:
-            return []
+            return self.left_combiner
 
     @left_combiner_all.setter
     def left_combiner_all(self, left_combiner_all):
-        self._left_combiner_all = left_combiner_all
+        self._left_combiner_all = list(set(left_combiner_all))
 
     @property
     def other_states(self):
@@ -395,8 +389,7 @@ def _merge_previous_states(self):
         self.group_for_inputs_final = {}
         self.keys_final = []
         if self.left_combiner:
-            # TODO: should I update left_combiner somewhere else? (it changes when scallar splitter)
-            _, _, _, self.left_combiner = hlpst.splits_groups(
+            _, _, _, self.left_combiner_all = hlpst.splits_groups(
                 self.left_splitter_rpn, combiner=self.left_combiner
             )
         for i, left_nm in enumerate(self.left_splitter_rpn_compact):
@@ -414,7 +407,7 @@ def _merge_previous_states(self):
             st = self.other_states[left_nm[1:]][0]
             # checking if left combiner contains any element from the st splitter
             st_combiner = [
-                comb for comb in self.left_combiner if comb in st.splitter_rpn_final
+                comb for comb in self.left_combiner_all if comb in st.splitter_rpn_final
             ]
             if st_combiner:
                 # keys and groups from previous states

pydra/engine/tests/test_state.py

@@ -1143,6 +1143,8 @@ def test_state_connect_combine_3():
     assert st2.splitter == ["_NA", "NB.d"]
     assert st2.splitter_rpn == ["NA.b", "NB.d", "*"]
     assert st2.splitter_rpn_final == ["NA.b"]
+    assert st2.left_combiner_all == st2.left_combiner == []
+    assert st2.right_combiner_all == st2.right_combiner == ["NB.d"]
     assert st2.group_for_inputs_final == {"NA.b": 0}
     assert st2.groups_stack_final == [[0]]
 
@@ -1203,6 +1205,8 @@ def test_state_connect_innerspl_combine_1():
     assert st2.splitter_rpn == ["NA.a", "NB.c", "NB.b", "*", "*"]
     assert st2.splitter_final == ["NA.a", "NB.c"]
     assert st2.splitter_rpn_final == ["NA.a", "NB.c", "*"]
+    assert st2.left_combiner_all == st2.left_combiner == []
+    assert st2.right_combiner_all == st2.right_combiner == ["NB.b"]
     assert st2.group_for_inputs_final == {"NA.a": 0, "NB.c": 1}
     assert st2.groups_stack_final == [[0], [1]]
     # TODO: i think at the end I should merge [0] and [1], because there are no inner splitters anymore
@@ -1377,6 +1381,8 @@ def test_state_connect_combine_left_2():
     assert st2.splitter == "_NA"
     assert st2.splitter_rpn == ["NA.a", "NA.b", "*"]
     assert st2.combiner == ["NA.a"]
+    assert st2.left_combiner_all == st2.left_combiner == ["NA.a"]
+    assert st2.right_combiner_all == st2.right_combiner == []
     assert st2.splitter_rpn_final == ["NA.b"]
 
     assert st2.group_for_inputs_final == {"NA.b": 0}
@@ -1454,6 +1460,8 @@ def test_state_connect_combine_left_4():
     assert st3.splitter == ["_NA", "_NB"]
     assert st3.splitter_rpn == ["NA.a", "NB.a", "*"]
     assert st3.splitter_rpn_final == ["NB.a"]
+    assert st3.left_combiner_all == st3.left_combiner == ["NA.a"]
+    assert st3.right_combiner_all == st3.right_combiner == []
     assert st3.group_for_inputs_final == {"NB.a": 0}
     assert st3.groups_stack_final == [[0]]
 
@@ -1499,6 +1507,9 @@ def test_state_connect_combine_left_5():
     assert st3.splitter == ("_NA", "_NB")
     assert st3.splitter_rpn == ["NA.a", "NB.a", "."]
     assert st3.splitter_rpn_final == []
+    assert set(st3.left_combiner_all) == set(["NA.a", "NB.a"])
+    assert st3.left_combiner == ["NA.a"]
+    assert st3.right_combiner_all == st3.right_combiner == []
     assert st3.group_for_inputs_final == {}
     assert st3.groups_stack_final == []
 
@@ -1525,6 +1536,8 @@ def test_state_connect_combine_left_6():
     assert st2.splitter == ["_NA", "NB.c"]
     assert st2.splitter_rpn == ["NA.a", "NA.b", "*", "NB.c", "*"]
     assert st2.combiner == ["NA.a"]
+    assert st2.left_combiner_all == st2.left_combiner == ["NA.a"]
+    assert st2.right_combiner_all == st2.right_combiner == []
     assert st2.splitter_rpn_final == ["NA.b", "NB.c", "*"]
 
     assert st2.group_for_inputs_final == {"NA.b": 0, "NB.c": 1}

pydra/engine/tests/test_workflow.py

@@ -1165,6 +1165,33 @@ def test_wf_3nd_ndst_5(plugin):
     assert wf.output_dir.exists()
 
 
+@pytest.mark.parametrize("plugin", Plugins)
+def test_wf_3nd_ndst_6(plugin):
+    """ workflow with three tasks, third one connected to two previous tasks,
+        the third one uses scalar splitter from the previous ones and a combiner
+    """
+    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(
+        multiply(name="mult", x=wf.add2x.lzout.out, y=wf.add2y.lzout.out)
+        .split(("_add2x", "_add2y"))
+        .combine("add2y.x")
+    )
+    wf.inputs.x = [1, 2]
+    wf.inputs.y = [11, 12]
+    wf.set_output([("out", wf.mult.lzout.out)])
+    wf.plugin = plugin
+
+    with Submitter(plugin=plugin) as sub:
+        sub(wf)
+
+    results = wf.result()
+    assert results.output.out == [39, 56]
+    # checking the output directory
+    assert wf.output_dir.exists()
+
+
 # workflows with Left and Right part in splitters A -> B (L&R parts of the splitter)