initial working commit for graphml location in workflow archive

beeflow/client/bee_client.py

@@ -641,7 +641,9 @@ def dag(wf_id: str = typer.Argument(..., callback=match_short_id),
 
     # output_dir must be a string
     output_dir = str(output_dir)
-    wf_utils.export_dag(wf_id, output_dir, no_dag_dir)
+    wf_dir = wf_utils.get_workflow_dir(wf_id)
+    graphmls_dir = wf_dir + "/graphmls"
+    wf_utils.export_dag(wf_id, output_dir, wf_dir, no_dag_dir)
     typer.secho(f"DAG for workflow {_short_id(wf_id)} has been exported successfully.",
                 fg=typer.colors.GREEN)
 

beeflow/common/deps/neo4j_manager.py

@@ -23,7 +23,7 @@
 run_dir = mount_dir + '/run'
 certs_dir = mount_dir + '/certificates'
 confs_dir = mount_dir + "/conf"
-graphmls_dir = os.path.join(bee_workdir, 'graphmls')
+graphmls_dir = mount_dir + '/graphmls'
 container_path = container_manager.get_container_dir('neo4j')
 log = bee_logging.setup('neo4j')
 

beeflow/common/gdb/generate_graph.py

@@ -5,13 +5,8 @@
 import networkx as nx
 import graphviz
 
-from beeflow.common import paths
 
-bee_workdir = paths.workdir()
-graphmls_dir = os.path.join(bee_workdir, 'graphmls')
-
-
-def generate_viz(wf_id, output_dir, no_dag_dir):
+def generate_viz(wf_id, output_dir, graphmls_dir, no_dag_dir):
     """Generate a PNG of a workflow graph from a GraphML file."""
     short_id = wf_id[:6]
     graphml_path = graphmls_dir + "/" + short_id + ".graphml"

beeflow/common/gdb/graphml_key_updater.py

@@ -6,8 +6,8 @@
 from beeflow.common import paths
 
 bee_workdir = paths.workdir()
-graphmls_dir = os.path.join(bee_workdir, 'graphmls')
-
+mount_dir = os.path.join(bee_workdir, 'gdb_mount')
+gdb_graphmls_dir = mount_dir + '/graphmls'
 
 expected_keys = {"id", "name", "state", "class", "type", "value", "source",
                  "workflow_id", "base_command", "stdout", "stderr", "default",
@@ -33,12 +33,13 @@
 }
 
 
-def update_graphml(wf_id):
+def update_graphml(wf_id, graphmls_dir):
     """Update GraphML file by ensuring required keys are present and updating its structure."""
     short_id = wf_id[:6]
-    graphml_path = graphmls_dir + "/" + short_id + ".graphml"
+    gdb_graphml_path = gdb_graphmls_dir + "/" + short_id + ".graphml"
+    output_graphml_path = graphmls_dir + "/" + short_id + ".graphml"
     # Parse the GraphML file and preserve namespaces
-    tree = ET.parse(graphml_path)
+    tree = ET.parse(gdb_graphml_path)
     root = tree.getroot()
 
     name_space = {'graphml': 'http://graphml.graphdrawing.org/xmlns'}
@@ -56,5 +57,5 @@ def update_graphml(wf_id):
                                      **default_def)
             root.insert(0, key_element)
 
-    # Save the updated GraphML file by overwriting the original one
-    tree.write(graphml_path, encoding='UTF-8', xml_declaration=True)
+    # Save the updated GraphML file
+    tree.write(output_graphml_path, encoding='UTF-8', xml_declaration=True)

beeflow/wf_manager/resources/wf_update.py

@@ -37,7 +37,7 @@ def archive_workflow(db, wf_id, final_state=None):
     archive_path = f'../archives/{wf_id}.tgz'
     # We use tar directly since tarfile is apparently very slow
     workflows_dir = wf_utils.get_workflows_dir()
-    wf_utils.export_dag(wf_id, workflow_dir, no_dag_dir=True)
+    wf_utils.export_dag(wf_id, workflow_dir, workflow_dir, no_dag_dir=True)
     subprocess.call(['tar', '-czf', archive_path, wf_id], cwd=workflows_dir)
     remove_wf_dir = bc.get('DEFAULT', 'delete_completed_workflow_dirs')
     if remove_wf_dir:

beeflow/wf_manager/resources/wf_utils.py

@@ -294,12 +294,14 @@ def setup_workflow(wf_id, wf_name, wf_dir, wf_workdir, no_start, workflow=None,
         start_workflow(wf_id)
 
 
-def export_dag(wf_id, output_dir, no_dag_dir):
+def export_dag(wf_id, output_dir, wf_dir, no_dag_dir):
     """Export the DAG of the workflow."""
     wfi = get_workflow_interface(wf_id)
     wfi.export_graphml()
-    update_graphml(wf_id)
-    generate_viz(wf_id, output_dir, no_dag_dir)
+    graphmls_dir = wf_dir + "/graphmls"
+    os.makedirs(graphmls_dir, exist_ok=True)
+    update_graphml(wf_id, graphmls_dir)
+    generate_viz(wf_id, output_dir, graphmls_dir, no_dag_dir)
 
 
 def start_workflow(wf_id):