From 3741c2268009f6efe6d9aa196f0713f58cdc9ad5 Mon Sep 17 00:00:00 2001 From: Joe Zuntz Date: Wed, 16 Sep 2026 16:41:36 +0100 Subject: [PATCH 1/4] add new approach to tighten flow charts more --- ceci/pipeline/graph.py | 37 +++++++++++++++++++++++++++++++++++++ ceci/pipeline/pipeline.py | 3 ++- tests/test_graph.py | 18 +++++++++++++++++- 3 files changed, 56 insertions(+), 2 deletions(-) diff --git a/ceci/pipeline/graph.py b/ceci/pipeline/graph.py index 0aa2dce..b6fbe07 100644 --- a/ceci/pipeline/graph.py +++ b/ceci/pipeline/graph.py @@ -238,3 +238,40 @@ def combine_output_nodes(graph): graph.add_edge(parent, new_node_name) # remove the old nodes graph.remove_nodes_from(group) + + +def combine_input_nodes(graph): + """ + Combine input nodes that share the same successor and have no + predecessors into a single node. + + This is useful for plotting flow charts. + + Parameters + ---------- + graph : networkx.DiGraph + The pipeline graph to modify + """ + groups = collections.defaultdict(list) + for node in graph.nodes_iter(): + node_data = graph.get_node(node).attr + if node_data["type"] != "input" or len(graph.predecessors(node)) != 0: + continue + successors = list(graph.successors(node)) + + # This shouldn't happen in the default case, + # but might if things change later, so let's deal with it + # just in case. + if len(successors) != 1: + continue + + successor = successors[0] + groups[successor].append(node) + + for successor, group in groups.items(): + if len(group) < 2: + continue + new_node_name = "\n".join(group) + graph.add_node(new_node_name, type="input") + graph.add_edge(new_node_name, successor) + graph.remove_nodes_from(group) \ No newline at end of file diff --git a/ceci/pipeline/pipeline.py b/ceci/pipeline/pipeline.py index b3a66cc..46cea00 100644 --- a/ceci/pipeline/pipeline.py +++ b/ceci/pipeline/pipeline.py @@ -9,7 +9,7 @@ from ..stage import PipelineStage from ..sites import load, set_default_site, get_default_site from ..utils import extra_paths -from .graph import build_graph, get_static_ordering, trim_pipeline_graph, combine_output_nodes +from .graph import build_graph, get_static_ordering, trim_pipeline_graph, combine_output_nodes, combine_input_nodes from .file_manager import FileManager from .sec import StageExecutionConfig from .templates import read_and_apply_template @@ -910,6 +910,7 @@ def make_flow_chart(self, filename=None): # are final outputs not used in the rest of the pipeline. We # can combine these to make the graph more readable. combine_output_nodes(graph) + combine_input_nodes(graph) # set the colours and styles for the boxes for node in graph.nodes_iter(): diff --git a/tests/test_graph.py b/tests/test_graph.py index 2e345d0..f9c54a8 100644 --- a/tests/test_graph.py +++ b/tests/test_graph.py @@ -1,4 +1,4 @@ -from ceci.pipeline.graph import trim_pipeline_graph +from ceci.pipeline.graph import trim_pipeline_graph, combine_input_nodes import networkx @@ -97,3 +97,19 @@ def test_trim_pipeline_graph(): } check_membership(sub4, everything, expected4) + + +def test_combine_input_nodes(): + graph = networkx.DiGraph() + graph.add_node("a", type="input") + graph.add_node("b", type="input") + graph.add_node("stage", type="stage") + graph.add_edge("a", "stage") + graph.add_edge("b", "stage") + + combine_input_nodes(graph) + + input_nodes = [n for n in graph.nodes() if graph.nodes[n]["type"] == "input"] + assert len(input_nodes) == 1 + assert input_nodes[0] == "a\nb" + assert list(graph.predecessors("stage")) == ["a\nb"] \ No newline at end of file From fa2330c1a4f5ec5d5947393e95fda56dd60d5d39 Mon Sep 17 00:00:00 2001 From: Joe Zuntz Date: Tue, 6 Oct 2026 10:57:28 +0100 Subject: [PATCH 2/4] Replace nodes_iter with nodes --- ceci/pipeline/graph.py | 4 ++-- ceci/pipeline/pipeline.py | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/ceci/pipeline/graph.py b/ceci/pipeline/graph.py index b6fbe07..7d81068 100644 --- a/ceci/pipeline/graph.py +++ b/ceci/pipeline/graph.py @@ -213,7 +213,7 @@ def combine_output_nodes(graph): groups = collections.defaultdict(list) # find groups of nodes that have no outgoing edges and all # share the same incoming edge. - for node in graph.nodes_iter(): + for node in graph.nodes(): node_data = graph.get_node(node).attr if node_data["type"] != "output" or len(graph.successors(node)) != 0: continue @@ -253,7 +253,7 @@ def combine_input_nodes(graph): The pipeline graph to modify """ groups = collections.defaultdict(list) - for node in graph.nodes_iter(): + for node in graph.nodes(): node_data = graph.get_node(node).attr if node_data["type"] != "input" or len(graph.predecessors(node)) != 0: continue diff --git a/ceci/pipeline/pipeline.py b/ceci/pipeline/pipeline.py index 46cea00..73ec850 100644 --- a/ceci/pipeline/pipeline.py +++ b/ceci/pipeline/pipeline.py @@ -913,7 +913,7 @@ def make_flow_chart(self, filename=None): combine_input_nodes(graph) # set the colours and styles for the boxes - for node in graph.nodes_iter(): + for node in graph.nodes(): node_type = node.attr['type'] if node_type == "input": node.attr.update(shape="box", color="gold", style="filled") From 93627c6b090b3c210eefba26e89488be282af45e Mon Sep 17 00:00:00 2001 From: Joe Zuntz Date: Tue, 6 Oct 2026 11:08:09 +0100 Subject: [PATCH 3/4] update test to use correct object --- tests/test_graph.py | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/tests/test_graph.py b/tests/test_graph.py index f9c54a8..d043b09 100644 --- a/tests/test_graph.py +++ b/tests/test_graph.py @@ -1,5 +1,6 @@ from ceci.pipeline.graph import trim_pipeline_graph, combine_input_nodes import networkx +import pygraphviz def check_membership(graph, everything, expected): @@ -100,7 +101,7 @@ def test_trim_pipeline_graph(): def test_combine_input_nodes(): - graph = networkx.DiGraph() + graph = pygraphviz.AGraph(directed=True) graph.add_node("a", type="input") graph.add_node("b", type="input") graph.add_node("stage", type="stage") @@ -109,7 +110,11 @@ def test_combine_input_nodes(): combine_input_nodes(graph) - input_nodes = [n for n in graph.nodes() if graph.nodes[n]["type"] == "input"] + input_nodes = [ + str(node) + for node in graph.nodes() + if graph.get_node(node).attr["type"] == "input" + ] assert len(input_nodes) == 1 assert input_nodes[0] == "a\nb" - assert list(graph.predecessors("stage")) == ["a\nb"] \ No newline at end of file + assert [str(node) for node in graph.predecessors("stage")] == ["a\nb"] \ No newline at end of file From 1bcbf61387b5648e3040e77169ff8bd6c54aac3c Mon Sep 17 00:00:00 2001 From: Joe Zuntz Date: Tue, 6 Oct 2026 11:10:51 +0100 Subject: [PATCH 4/4] Correct docstrings --- ceci/pipeline/graph.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/ceci/pipeline/graph.py b/ceci/pipeline/graph.py index 7d81068..0b1fb2b 100644 --- a/ceci/pipeline/graph.py +++ b/ceci/pipeline/graph.py @@ -207,7 +207,7 @@ def combine_output_nodes(graph): Parameters ---------- - graph : networkx.DiGraph + graph : pygraphviz.AGraph The pipeline graph to modify """ groups = collections.defaultdict(list) @@ -249,7 +249,7 @@ def combine_input_nodes(graph): Parameters ---------- - graph : networkx.DiGraph + graph : pygraphviz.AGraph The pipeline graph to modify """ groups = collections.defaultdict(list)