Add jobflow test

Author: jan-janssen

.github/workflows/jobflow.yml

@@ -0,0 +1,28 @@
+name: jobflow
+
+on:
+  push:
+    branches: [ main ]
+  pull_request:
+
+jobs:
+  build:
+
+    runs-on: ubuntu-latest
+
+    steps:
+    - uses: actions/checkout@v4
+    - uses: conda-incubator/setup-miniconda@v3
+      with:
+        auto-update-conda: true
+        python-version: "3.12"
+        environment-file: environment.yml
+        auto-activate-base: false
+    - name: Tests
+      shell: bash -l {0}
+      run: |
+        pip install -e aiida_qe_basic
+        pip install -e adis_tools
+        conda install -c conda-forge jupyter papermill
+        export ESPRESSO_PSEUDO=$(pwd)/espresso/pseudo
+        papermill universal_qe_to_jobflow.ipynb universal_qe_to_jobflow_out.ipynb -k "python3"

python_workflow_definition/src/python_workflow_definition/jobflow.py

@@ -1,3 +1,5 @@
+import json
+from importlib import import_module
 from inspect import isfunction
 
 from jobflow import job, Flow
@@ -186,6 +188,28 @@ def get_source_handles(edges_lst):
     }
 
 
-def find_root_node(nodes_dict, edges_lst):
-    source_count_dict = {k: sum([ed["source"] == k for ed in edges_lst]) for k in nodes_dict.keys()}
-    return min(source_count_dict, key=source_count_dict.get)
+def load_workflow_json(file_name):
+    with open(file_name, "r") as f:
+        content = json.load(f)
+
+    edges_new_lst = content["edges"]
+    nodes_new_dict = {}
+    for k, v in content["nodes"].items():
+        if isinstance(v, str) and "." in v:
+            p, m = v.rsplit('.', 1)
+            mod = import_module(p)
+            nodes_new_dict[int(k)] = getattr(mod, m)
+        else:
+            nodes_new_dict[int(k)] = v
+
+    source_handles_dict = get_source_handles(edges_lst=edges_new_lst)
+    total_dict = group_edges(edges_lst=edges_new_lst)
+    input_dict = get_input_dict(nodes_dict=nodes_new_dict)
+    new_total_dict = resort_total_lst(total_dict=total_dict, nodes_dict=nodes_new_dict)
+    task_lst = get_workflow(
+        nodes_dict=nodes_new_dict,
+        input_dict=input_dict,
+        total_dict=new_total_dict,
+        source_handles_dict=source_handles_dict,
+    )
+    return Flow(task_lst)

universal_qe_to_jobflow.ipynb

@@ -0,0 +1,66 @@
+{
+ "cells": [
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "eab942d5-f3c5-47e9-8b3b-a7a5a8481aed",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from jobflow.managers.local import run_locally"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "f6f83a43-4d91-4028-9661-a4700509be1b",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from python_workflow_definition.jobflow import load_workflow_json"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "285ca46b-19e3-4870-9fd1-0c8ddbcf819c",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "flow = load_workflow_json(file_name=\"workflow_qe.json\")"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "663ac4b3-dee8-474f-a9bc-089a89bde011",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "result = run_locally(flow)\n",
+    "result"
+   ]
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "Python 3 (ipykernel)",
+   "language": "python",
+   "name": "python3"
+  },
+  "language_info": {
+   "codemirror_mode": {
+    "name": "ipython",
+    "version": 3
+   },
+   "file_extension": ".py",
+   "mimetype": "text/x-python",
+   "name": "python",
+   "nbconvert_exporter": "python",
+   "pygments_lexer": "ipython3",
+   "version": "3.12.5"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 5
+}