From 698b3f49877cf312ceed044fb432a03e9f1e84e6 Mon Sep 17 00:00:00 2001 From: Arushi55 Date: Thu, 13 Nov 2025 15:25:22 -0500 Subject: [PATCH 1/7] Added flow_out list to parallel state --- behavior_machine/library/parallel_state.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/behavior_machine/library/parallel_state.py b/behavior_machine/library/parallel_state.py index 7f6ee17..eb098c1 100644 --- a/behavior_machine/library/parallel_state.py +++ b/behavior_machine/library/parallel_state.py @@ -82,6 +82,9 @@ def execute(self, board: Board) -> StateStatus: # 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 From 4d257e1a62a3d1b5fd196f71fda43c60b32e17df Mon Sep 17 00:00:00 2001 From: Arushi55 Date: Fri, 14 Nov 2025 09:26:02 -0500 Subject: [PATCH 2/7] testing flow additions to parallel state --- behavior_machine/library/parallel_state.py | 2 +- tests/library/parallel_state_test.py | 46 ++++++++++++++++++++++ 2 files changed, 47 insertions(+), 1 deletion(-) diff --git a/behavior_machine/library/parallel_state.py b/behavior_machine/library/parallel_state.py index eb098c1..cbfeedd 100644 --- a/behavior_machine/library/parallel_state.py +++ b/behavior_machine/library/parallel_state.py @@ -78,7 +78,7 @@ 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=self.flow_in) # we wait for when this state should be completed self._state_complete_event.wait() diff --git a/tests/library/parallel_state_test.py b/tests/library/parallel_state_test.py index 45df0e8..6d97729 100644 --- a/tests/library/parallel_state_test.py +++ b/tests/library/parallel_state_test.py @@ -237,3 +237,49 @@ def test_parallel_debug_info(): assert info['children'][1]['name'] == 'w2' assert info['children'][1]['status'] == StateStatus.RUNNING pm.interrupt() + +def test_parallel_state_flow(): + class FirstState(State): + def execute(self, board): + self.flow_out = "first-done" + return StateStatus.SUCCESS + + class PreStateOne(State): + def execute(self, board): + self.flow_out = "pre-one-done" + return StateStatus.SUCCESS + + class PreStateTwo(State): + def execute(self, board): + self.flow_out = "pre-two-done" + return StateStatus.SUCCESS + + class PreStateThree(State): + def execute(self, board): + return StateStatus.SUCCESS + + class ReceiveState(State): + def execute(self, board): + assert self.flow_in[0] == "pre-one-done" + assert self.flow_in[1] == "pre-two-done" + assert self.flow_in[2] == None + return StateStatus.SUCCESS + + ps = FirstState("ps") + ps1 = PreStateOne("ps1") + ps2 = PreStateTwo("ps2") + ps3 = PreStateThree("ps3") + pm = ParallelState(children=[ps1, ps2, ps3], name="pm") + rs = ReceiveState("rs") + ps.add_transition_on_success(pm) + pm.add_transition_on_success(rs) + exe = Machine(ps, end_state_ids=['rs'], rate=10) + exe.start(None) + exe.wait() + assert exe.check_status(StateStatus.SUCCESS) + assert exe._curr_state == rs + assert exe.flow_out == None + assert pm.flow_in == "first-done" + assert pm._children[0].flow_in == "first-done" + assert pm._children[1].flow_in == "first-done" + assert pm._children[2].flow_in == "first-done" \ No newline at end of file From 3377c881370ff4c06f4807df4af7d79a1d175b7e Mon Sep 17 00:00:00 2001 From: Arushi55 Date: Fri, 14 Nov 2025 11:26:11 -0500 Subject: [PATCH 3/7] Fixed tests, updated changelog --- CHANGELOG.md | 1 + tests/core/performance_test.py | 2 +- tests/core/statemachine_test.py | 20 ++++++++++---------- 3 files changed, 12 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index cb6ab88..5c808c3 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/tests/core/performance_test.py b/tests/core/performance_test.py index cf7e1aa..bbbbfb1 100644 --- a/tests/core/performance_test.py +++ b/tests/core/performance_test.py @@ -67,7 +67,7 @@ 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) diff --git a/tests/core/statemachine_test.py b/tests/core/statemachine_test.py index af05415..345574f 100644 --- a/tests/core/statemachine_test.py +++ b/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() From 4380f3626703b32285b58c27f407b5bb038396f3 Mon Sep 17 00:00:00 2001 From: Zhi Tan Date: Wed, 19 Nov 2025 22:36:45 -0500 Subject: [PATCH 4/7] changed the test parameters. --- .github/workflows/python-package.yml | 2 +- tests/core/performance_test.py | 14 +++++++------- 2 files changed, 8 insertions(+), 8 deletions(-) diff --git a/.github/workflows/python-package.yml b/.github/workflows/python-package.yml index a235c54..16f4276 100644 --- a/.github/workflows/python-package.yml +++ b/.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 diff --git a/tests/core/performance_test.py b/tests/core/performance_test.py index bbbbfb1..1e53cae 100644 --- a/tests/core/performance_test.py +++ b/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(): @@ -62,7 +62,7 @@ class CompleteState(State): def execute(self, board: Board) -> StateStatus: return StateStatus.SUCCESS - num_parallel = 500 + num_parallel = 1000 child_states = [] for i in range(0, num_parallel): child_states.append(CompleteState(f"I{i}")) From 3c15d1e59a1d16fe62e6420132258e130e05b083 Mon Sep 17 00:00:00 2001 From: Zhi Tan Date: Wed, 19 Nov 2025 22:38:59 -0500 Subject: [PATCH 5/7] make performance test easier for github actions. --- tests/core/performance_test.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/core/performance_test.py b/tests/core/performance_test.py index 1e53cae..7c5c7cf 100644 --- a/tests/core/performance_test.py +++ b/tests/core/performance_test.py @@ -62,7 +62,7 @@ class CompleteState(State): def execute(self, board: Board) -> StateStatus: return StateStatus.SUCCESS - num_parallel = 1000 + num_parallel = 500 child_states = [] for i in range(0, num_parallel): child_states.append(CompleteState(f"I{i}")) @@ -73,5 +73,5 @@ def execute(self, board: Board) -> StateStatus: exe.start(None) exe.wait() elapsed_time = time.time() - start_time - assert elapsed_time < (1/10) + assert elapsed_time < 0.1 From 99930fc812fdc4073031ee6f5d6c9102f0f3d593 Mon Sep 17 00:00:00 2001 From: Arushi55 Date: Thu, 20 Nov 2025 14:36:57 -0500 Subject: [PATCH 6/7] deepcopy flow_in to parallel children and tested --- behavior_machine/library/parallel_state.py | 14 +-- tests/library/parallel_state_test.py | 99 ++++++++++++++++------ 2 files changed, 82 insertions(+), 31 deletions(-) diff --git a/behavior_machine/library/parallel_state.py b/behavior_machine/library/parallel_state.py index cbfeedd..6d8164a 100644 --- a/behavior_machine/library/parallel_state.py +++ b/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,7 +82,7 @@ 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, flow_in=self.flow_in) + child.start(board, flow_in=copy.deepcopy(self.flow_in)) # we wait for when this state should be completed self._state_complete_event.wait() @@ -136,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 diff --git a/tests/library/parallel_state_test.py b/tests/library/parallel_state_test.py index 6d97729..a26a2f3 100644 --- a/tests/library/parallel_state_test.py +++ b/tests/library/parallel_state_test.py @@ -11,7 +11,7 @@ def __init__(self, timing): def execute(self, board): start_time = time.time() - while ((time.time() - start_time) < self._timing): + while (time.time() - start_time) < self._timing: if self.is_interrupted(): return StateStatus.INTERRUPTED time.sleep(0.01) @@ -47,7 +47,7 @@ def test_parallel_state_in_machine(capsys): es = IdleState("es") pm = ParallelState([ws, ws2]) pm.add_transition_on_success(es) - exe = Machine(pm, end_state_ids=['es'], rate=10) + exe = Machine(pm, end_state_ids=["es"], rate=10) # run machine and see how it reacts exe.start(None) # wait for one second @@ -68,18 +68,17 @@ def test_parallel_one_state_fails(capsys): fs = FailAfterSecState(1) es = IdleState("es") fes = IdleState("fs-terminal") - pm = ParallelState([ws, fs], name='pm') + pm = ParallelState([ws, fs], name="pm") pm.add_transition_on_success(es) pm.add_transition_on_failed(fes) - exe = Machine(pm, end_state_ids=[ - 'es', 'fs-terminal'], rate=10) + exe = Machine(pm, end_state_ids=["es", "fs-terminal"], rate=10) # run machine and see how it reacts exe.start(None) # wait for one second assert not exe.wait(0.5) assert exe.check_status(StateStatus.RUNNING) - assert exe._curr_state.check_name('pm') + assert exe._curr_state.check_name("pm") # at this point ws should be done but ws2 is still going # wait another one seconds assert exe.wait(2) @@ -118,15 +117,14 @@ def execute(self, board): pm = ParallelState(children=[ws, fs], name="pm") pm.add_transition_on_success(es) pm.add_transition_on_failed(fes) - exe = Machine(pm, end_state_ids=[ - 'es', 'fs-terminal'], rate=10) + exe = Machine(pm, end_state_ids=["es", "fs-terminal"], rate=10) # run machine and see how it reacts exe.start(None) # wait for 0.5 second to see it is still running assert not exe.wait(0.5) assert exe.check_status(StateStatus.RUNNING) - assert exe._curr_state.check_name('pm') + assert exe._curr_state.check_name("pm") # at this point, it should throw or raise the exception # wait another 1.5 seconds assert exe.wait(1.5) @@ -149,15 +147,14 @@ def execute(self, board): pm = ParallelState(children=[ws, fs], name="pm") pm.add_transition_on_success(es) pm.add_transition_on_failed(fes) - exe = Machine(pm, end_state_ids=[ - 'es', 'fs-terminal'], rate=10) + exe = Machine(pm, end_state_ids=["es", "fs-terminal"], rate=10) # run machine and see how it reacts exe.start(None) # wait for 0.5 second to see it is still running assert not exe.wait(0.5) assert exe.check_status(StateStatus.RUNNING) - assert exe._curr_state.check_name('pm') + assert exe._curr_state.check_name("pm") # at this point, it should throw or raise the exception # wait another 1.5 seconds assert exe.wait(1.5) @@ -178,7 +175,7 @@ def execute(self, board): pm = ParallelState([ws, re], name="pm") es = IdleState("es") pm.add_transition_on_success(es) - exe = Machine(pm, name='xe', end_state_ids=['es', 'onException'], rate=10) + exe = Machine(pm, name="xe", end_state_ids=["es", "onException"], rate=10) # run machine and see how it reacted exe.start(None) exe.wait(0.5) @@ -196,7 +193,7 @@ def test_interrupt_in_parallel_state(capsys): pm = ParallelState(children=[ws, ws2]) pm.add_transition_on_success(es) pm.add_transition_after_elapsed(es, 0.1) - exe = Machine(pm, end_state_ids=['es'], rate=10) + exe = Machine(pm, end_state_ids=["es"], rate=10) # run machine exe.start(None) # because of the elapsed transition, the machine will immediate transition to the end state in 0.1 seconds @@ -230,14 +227,15 @@ def test_parallel_debug_info(): pm.start(None) pm.wait(0.1) info = pm.get_debug_info() - assert info['name'] == 'pm' - assert len(info['children']) == 2 - assert info['children'][0]['name'] == 'w1' - assert info['children'][0]['status'] == StateStatus.RUNNING - assert info['children'][1]['name'] == 'w2' - assert info['children'][1]['status'] == StateStatus.RUNNING + assert info["name"] == "pm" + assert len(info["children"]) == 2 + assert info["children"][0]["name"] == "w1" + assert info["children"][0]["status"] == StateStatus.RUNNING + assert info["children"][1]["name"] == "w2" + assert info["children"][1]["status"] == StateStatus.RUNNING pm.interrupt() + def test_parallel_state_flow(): class FirstState(State): def execute(self, board): @@ -248,23 +246,23 @@ class PreStateOne(State): def execute(self, board): self.flow_out = "pre-one-done" return StateStatus.SUCCESS - + class PreStateTwo(State): def execute(self, board): self.flow_out = "pre-two-done" return StateStatus.SUCCESS - + class PreStateThree(State): def execute(self, board): return StateStatus.SUCCESS - + class ReceiveState(State): def execute(self, board): assert self.flow_in[0] == "pre-one-done" assert self.flow_in[1] == "pre-two-done" assert self.flow_in[2] == None return StateStatus.SUCCESS - + ps = FirstState("ps") ps1 = PreStateOne("ps1") ps2 = PreStateTwo("ps2") @@ -273,7 +271,7 @@ def execute(self, board): rs = ReceiveState("rs") ps.add_transition_on_success(pm) pm.add_transition_on_success(rs) - exe = Machine(ps, end_state_ids=['rs'], rate=10) + exe = Machine(ps, end_state_ids=["rs"], rate=10) exe.start(None) exe.wait() assert exe.check_status(StateStatus.SUCCESS) @@ -282,4 +280,53 @@ def execute(self, board): assert pm.flow_in == "first-done" assert pm._children[0].flow_in == "first-done" assert pm._children[1].flow_in == "first-done" - assert pm._children[2].flow_in == "first-done" \ No newline at end of file + assert pm._children[2].flow_in == "first-done" + + +def test_parallel_flow_deepcopy(): + class FirstState(State): + def execute(self, board): + self.flow_out = {"msg": 10} + return StateStatus.SUCCESS + + class PreStateOne(State): + def execute(self, board): + self.flow_in["msg"] += 2 + self.flow_out = self.flow_in + return StateStatus.SUCCESS + + class PreStateTwo(State): + def execute(self, board): + self.flow_in["msg"] += 5 + self.flow_out = self.flow_in + return StateStatus.SUCCESS + + class PreStateThree(State): + def execute(self, board): + return StateStatus.SUCCESS + + class ReceiveState(State): + def execute(self, board): + assert self.flow_in[0]["msg"] == 12 + assert self.flow_in[1]["msg"] == 15 + assert self.flow_in[2] == None + return StateStatus.SUCCESS + + ps = FirstState("ps") + ps1 = PreStateOne("ps1") + ps2 = PreStateTwo("ps2") + ps3 = PreStateThree("ps3") + pm = ParallelState(children=[ps1, ps2, ps3], name="pm") + rs = ReceiveState("rs") + ps.add_transition_on_success(pm) + pm.add_transition_on_success(rs) + exe = Machine(ps, end_state_ids=["rs"], rate=10) + exe.start(None) + exe.wait() + assert exe.check_status(StateStatus.SUCCESS) + assert exe._curr_state == rs + assert exe.flow_out == None + assert pm.flow_in["msg"] == 10 + assert pm._children[0].flow_in["msg"] == 12 + assert pm._children[1].flow_in["msg"] == 15 + assert pm._children[2].flow_in["msg"] == 10 From 643fd84153519669b801f1454f4be989a35e4379 Mon Sep 17 00:00:00 2001 From: Arushi55 Date: Thu, 11 Dec 2025 07:59:53 -0800 Subject: [PATCH 7/7] fixed visualize recursion error --- behavior_machine/visualization/visualize.py | 51 +++++++++++---------- 1 file changed, 27 insertions(+), 24 deletions(-) diff --git a/behavior_machine/visualization/visualize.py b/behavior_machine/visualization/visualize.py index 3e88c6f..23344f3 100644 --- a/behavior_machine/visualization/visualize.py +++ b/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