Parallel state flow

.github/workflows/python-package.yml

@@ -14,7 +14,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: [ 3.6, 3.7, 3.8, 3.9]
+        python-version: [ 3.11]
 
     steps:
     - uses: actions/checkout@v2

CHANGELOG.md

@@ -1,4 +1,5 @@
 # Changelog
+- **[Changed]** Added flow in and out for `ParallelState`. Flow in is copied and passed to all children. Flow out is a list of each child's flow out.
 - **[Changed]** Propagated the name for `State` being optional throughout the project. Now name is the first optional parameters.
 - **[Changed]** Updated all test to make it work with name being an optional parameter
 - **[Changed]** `SetBoardState`'s key and val are no longer optional variables

behavior_machine/library/parallel_state.py

@@ -2,6 +2,7 @@
 from ..core import StateStatus, State, NestedState, Board
 import typing
 import sys
+import copy
 
 
 class ParallelState(NestedState):
@@ -39,8 +40,11 @@ def _interrupt_running_children(self, timeout=None) -> bool:
         failed = False
         for child in self._children:
             if not child.interrupt(timeout):
-                print(f"ERROR {self.get_debug_name()} unable to complete Interrupt Action \
-                    for child {child.get_debug_name()} Zombie threads likely", file=sys.stderr)
+                print(
+                    f"ERROR {self.get_debug_name()} unable to complete Interrupt Action \
+                    for child {child.get_debug_name()} Zombie threads likely",
+                    file=sys.stderr,
+                )
                 failed = True
             if child.check_status(StateStatus.EXCEPTION):
                 # this is the child that thrown an exception
@@ -78,10 +82,13 @@ def execute(self, board: Board) -> StateStatus:
         # start each child.
         for child in self._children:
             # because each child starts their own thread, no extra management required.
-            child.start(board)
+            child.start(board, flow_in=copy.deepcopy(self.flow_in))
 
         # we wait for when this state should be completed
         self._state_complete_event.wait()
+        self.flow_out = []
+        for child in self._children:
+            self.flow_out.append(child.flow_out)
         # we now interrupt and stop all remaining running state
         self._interrupt_running_children()
         # if we were interrupted
@@ -133,7 +140,7 @@ def tick(self, board: Board):
     def get_debug_info(self) -> typing.Dict[str, typing.Any]:
 
         self_info = super().get_debug_info()
-        self_info['children'] = []
+        self_info["children"] = []
         for child in self._children:
-            self_info['children'].append(child.get_debug_info())
+            self_info["children"].append(child.get_debug_info())
         return self_info

behavior_machine/visualization/visualize.py

@@ -9,8 +9,12 @@
 
 
 
-def _recursive_visualize_state(state: State, graph: pgv.AGraph, visited_list: typing.List[str], leaf_list: typing.Mapping[str, str], root: bool = False) -> str:
-
+def _recursive_visualize_state(state: State, graph: pgv.AGraph, visited_list: typing.List[str], leaf_list: typing.Mapping[str, str], global_visited: typing.Set[str] = None, global_leaf_list: typing.Mapping[str, str] = None, root: bool = False) -> str:
+
+    if global_visited is None:
+        global_visited = set()
+    if global_leaf_list is None:
+        global_leaf_list = {}
 
     # get name of state
     state_name = state._name
@@ -24,10 +28,6 @@ def _recursive_visualize_state(state: State, graph: pgv.AGraph, visited_list: ty
     global_name = f"{parent_name}_{state_name}"
     valid_node_name = ""
 
-    # [Base Case]  ignore if we already visited it.
-    if state_name in visited_list:
-        return
-
     # Check if its a nest class or machine
     sub_graph = None
     if not hasattr(state, "_children") and not isinstance(state, Machine):
@@ -39,7 +39,7 @@ def _recursive_visualize_state(state: State, graph: pgv.AGraph, visited_list: ty
         if isinstance(state, Machine):
             state: Machine
             sub_graph.graph_attr["pencolor"] = "red"
-            valid_node_name = _recursive_visualize_state(state._root, sub_graph, [], {}, True)
+            valid_node_name = _recursive_visualize_state(state._root, sub_graph, [], {}, global_visited, global_leaf_list, True)
         else:
             if isinstance(state, SequentialState) or isinstance(state, SelectorState):
                 sub_graph.graph_attr["pencolor"] = "green"
@@ -48,18 +48,20 @@ def _recursive_visualize_state(state: State, graph: pgv.AGraph, visited_list: ty
 
                 prev_valid_name = ""
                 for children in state._children:
-                    valid_node_name = _recursive_visualize_state(children, sub_graph, [], {})
+                    valid_node_name = _recursive_visualize_state(children, sub_graph, [], {}, global_visited, global_leaf_list)
                     if prev_valid_name != "":
                         sub_graph.add_edge(prev_valid_name, valid_node_name)
                     prev_valid_name = valid_node_name
             else:
                 sub_graph.graph_attr["pencolor"] = "blue"
                 for children in state._children:
-                    valid_node_name = _recursive_visualize_state(children, sub_graph, [], {})
+                    valid_node_name = _recursive_visualize_state(children, sub_graph, [], {}, global_visited, global_leaf_list)
 
-    # registered as visited
+    # registered as visited (both locally and globally)
     visited_list.append(state_name)
+    global_visited.add(state_name)
     leaf_list[state_name] = valid_node_name
+    global_leaf_list[state_name] = valid_node_name
 
     # Create connection between each states.
     for transition in state._transitions:    
@@ -68,27 +70,28 @@ def _recursive_visualize_state(state: State, graph: pgv.AGraph, visited_list: ty
         nxt_state = transition[1]
         nxt_state_name = nxt_state._name
 
-        # run the recursive visualization function if not visted:
-        if nxt_state_name not in visited_list:
-            nxt_valid_node_name = _recursive_visualize_state(nxt_state, graph, visited_list, leaf_list)
+        # run the recursive visualization function if not visited globally:
+        if nxt_state_name not in global_visited:
+            nxt_valid_node_name = _recursive_visualize_state(nxt_state, graph, visited_list, leaf_list, global_visited, global_leaf_list)
         else:
             # we get a valid node to connect the clusters.
-            nxt_valid_node_name = leaf_list[nxt_state_name]
+            nxt_valid_node_name = global_leaf_list.get(nxt_state_name, "")
 
         # construct the name if its a cluster
         nxt_cluster_name = f"cluster_{parent_name}_{nxt_state_name}"
         # HACK: graphviz require nodes to be linked up and cannot use cluster. Their way is to just hide it
-        if graph.get_subgraph(nxt_cluster_name) is None:
-            if sub_graph is None:
-                graph.add_edge(valid_node_name, nxt_valid_node_name)
-            else:
-                graph.add_edge(valid_node_name, nxt_valid_node_name, ltail=sub_graph.name)
-        else:
-            nxt_graph = graph.get_subgraph(nxt_cluster_name)
-            if sub_graph is None:
-                graph.add_edge(valid_node_name, nxt_valid_node_name, lhead=nxt_graph.name)
+        if nxt_valid_node_name:  # Only add edge if we have a valid node name
+            if graph.get_subgraph(nxt_cluster_name) is None:
+                if sub_graph is None:
+                    graph.add_edge(valid_node_name, nxt_valid_node_name)
+                else:
+                    graph.add_edge(valid_node_name, nxt_valid_node_name, ltail=sub_graph.name)
             else:
-                graph.add_edge(valid_node_name, nxt_valid_node_name, ltail=sub_graph.name, lhead=nxt_cluster_name)
+                nxt_graph = graph.get_subgraph(nxt_cluster_name)
+                if sub_graph is None:
+                    graph.add_edge(valid_node_name, nxt_valid_node_name, lhead=nxt_graph.name)
+                else:
+                    graph.add_edge(valid_node_name, nxt_valid_node_name, ltail=sub_graph.name, lhead=nxt_cluster_name)
 
     return valid_node_name
 

tests/core/performance_test.py

@@ -22,13 +22,13 @@ def execute(self, board: Board) -> StateStatus:
     ds2.add_transition_on_success(ds3)
     ds3.add_transition_on_success(ds1)
 
-    exe = Machine(ds1, rate=60)
+    exe = Machine(ds1, rate=30)
     exe.start(None)
     time.sleep(2)
     exe.interrupt()
     # the performance of the computer might change this.
-    assert counter >= (60 * 2) - 2
-    assert counter <= (60 * 2) + 1
+    assert counter >= (30 * 2) * 0.9
+    assert counter <= (30 * 2) * 1.1
 
 def test_validate_transition_immediate():
 
@@ -47,13 +47,13 @@ def execute(self, board: Board) -> StateStatus:
     ds2.add_transition(lambda s, b: True, ds3)
     ds3.add_transition(lambda s, b: True, ds1)
 
-    exe = Machine(ds1, rate=60)
+    exe = Machine(ds1, rate=30)
     exe.start(None)
     time.sleep(2)
     exe.interrupt()
     # the performance of the computer might change this.
-    assert counter >= (60 * 2) - 2
-    assert counter <= (60 * 2) + 1
+    assert counter >= (30 * 2) * 0.9
+    assert counter <= (30 * 2) * 1.1
 
 
 def test_multiple_parallel_states():
@@ -67,11 +67,11 @@ def execute(self, board: Board) -> StateStatus:
     for i in range(0, num_parallel):
         child_states.append(CompleteState(f"I{i}"))
 
-    pp = ParallelState("parallel", child_states)
+    pp = ParallelState(child_states, name="parallel")
     exe = Machine(pp, end_state_ids=['parallel'], rate=100)
     start_time = time.time()
     exe.start(None)
     exe.wait()
     elapsed_time = time.time() - start_time
-    assert elapsed_time < (1/10)
+    assert elapsed_time < 0.1
 

tests/core/statemachine_test.py

@@ -58,9 +58,9 @@ class NothingState(State):
         def execute(self, board):
             print("hello")
 
-    ds1 = DummyState("d1")
-    ns = NothingState("ns")
-    ds2 = DummyState("ds2")
+    ds1 = DummyState(name="d1")
+    ns = NothingState(name="ns")
+    ds2 = DummyState(name="ds2")
 
     ds1.add_transition_on_success(ns)
     ns.add_transition_on_complete(ds2)
@@ -207,12 +207,12 @@ def test_machine_with_exception(capsys):
 
 def test_machine_with_exception_in_transition(capsys):
 
-    is1 = DummyState('d1')
-    is2 = DummyState('d2')
+    is1 = DummyState(name='d1')
+    is2 = DummyState(name='d2')
 
     is1.add_transition(lambda s, b: s.unknown(), is2)
 
-    mac = Machine(is1, ["is2"])
+    mac = Machine(is1, end_state_ids=["is2"])
     mac.run()
 
     assert mac._status == StateStatus.EXCEPTION
@@ -224,11 +224,11 @@ def test_machine_with_exception_in_transition(capsys):
 def test_machine_with_exception_in_transition_with_zombie_states(capsys):
 
     ws1 = WaitState(10)
-    is2 = DummyState('d2')
+    is2 = DummyState(name='d2')
 
     ws1.add_transition(lambda s, b: s.unknown(), is2)
 
-    mac = Machine(ws1, ["is2"])
+    mac = Machine(ws1, end_state_ids=["is2"])
     mac.run()
     assert mac._status == StateStatus.EXCEPTION
     # this is an interrupted, because exception happen at higher level
@@ -259,9 +259,9 @@ def test_debugging_machine(caplog):
 
 def test_interrupt_machine(capsys):
     s1 = WaitState(1.1)
-    s2 = DummyState('s2')
+    s2 = DummyState(name='s2')
     s1.add_transition_on_success(s2)
-    mac = Machine(s1, ["s2"], debug=True, rate=1)
+    mac = Machine(s1, end_state_ids=["s2"], debug=True, rate=1)
     mac.start(None)
     assert mac.interrupt()