Implement export to cwl

Author: jan-janssen

.github/workflows/pipeline.yml

@@ -128,6 +128,7 @@ jobs:
       run: |
         cd example_workflows/arithmetic
         papermill aiida.ipynb aiida_out.ipynb -k "python3"
+        papermill cwl.ipynb cwl_out.ipynb -k "python3"
         papermill jobflow.ipynb jobflow_out.ipynb -k "python3"
         papermill pyiron_base.ipynb pyiron_base_out.ipynb -k "python3"
         papermill universal_workflow.ipynb universal_workflow_out.ipynb -k "python3"

binder/environment.yml

@@ -12,3 +12,4 @@ dependencies:
 - aiida-workgraph =0.5.2
 - conda_subprocess =0.0.6
 - networkx =3.4.2
+- cwltool =3.1.20250110105449

example_workflows/arithmetic/cwl.ipynb

@@ -0,0 +1 @@
+{"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.10"}},"nbformat_minor":5,"nbformat":4,"cells":[{"id":"377fef56-484d-491c-b19e-1be6931e44eb","cell_type":"code","source":"import pickle","metadata":{"trusted":true},"outputs":[],"execution_count":4},{"id":"92e3921b-2bb8-4333-8cfe-4bd27f785d24","cell_type":"code","source":"from python_workflow_definition.cwl.export import load_workflow_json","metadata":{"trusted":true},"outputs":[],"execution_count":17},{"id":"5303c059-8ae4-4557-858e-b4bd64eac711","cell_type":"code","source":"load_workflow_json(file_name=\"workflow.json\")","metadata":{"trusted":true},"outputs":[],"execution_count":18},{"id":"df302bd2-e9b6-4595-979c-67c46414d986","cell_type":"code","source":"! cwltool workflow.cwl workflow.yml","metadata":{"trusted":true},"outputs":[{"name":"stdout","output_type":"stream","text":"/srv/conda/envs/notebook/bin/cwltool:11: DeprecationWarning: Nesting argument groups is deprecated.\n  sys.exit(run())\n\u001b[1;30mINFO\u001b[0m /srv/conda/envs/notebook/bin/cwltool 3.1.20250110105449\n\u001b[1;30mINFO\u001b[0m Resolved 'workflow.cwl' to 'file:///home/jovyan/workflow.cwl'\n\u001b[1;30mINFO\u001b[0m [workflow ] start\n\u001b[1;30mINFO\u001b[0m [workflow ] starting step get_prod_and_div\n\u001b[1;30mINFO\u001b[0m [step get_prod_and_div] start\n\u001b[1;30mINFO\u001b[0m [job get_prod_and_div] /tmp/abngt5eq$ python \\\n    /tmp/7pxhgcet/stg6ba69a25-7dc0-46be-9bbe-5c08e6ce1925/wrapper.py \\\n    --function=workflow.get_prod_and_div \\\n    --arg_x=/tmp/7pxhgcet/stg007c65c6-d1cd-44cd-b761-e181c0cc37d8/x.pickle \\\n    --arg_y=/tmp/7pxhgcet/stg23ac56f7-9c1d-43d9-a90c-86036b3cbafc/y.pickle\n\u001b[1;30mINFO\u001b[0m [job get_prod_and_div] completed success\n\u001b[1;30mINFO\u001b[0m [step get_prod_and_div] completed success\n\u001b[1;30mINFO\u001b[0m [workflow ] starting step get_sum\n\u001b[1;30mINFO\u001b[0m [step get_sum] start\n\u001b[1;30mINFO\u001b[0m [job get_sum] /tmp/wzuzdwec$ python \\\n    /tmp/lz9a89rn/stg44b61930-48fe-423d-a68a-593e237e0477/wrapper.py \\\n    --function=workflow.get_sum \\\n    --arg_x=/tmp/lz9a89rn/stgb7917545-93cd-45b4-833a-01559a22e620/prod.pickle \\\n    --arg_y=/tmp/lz9a89rn/stg0c04cf0f-bc08-494d-ae4b-b12bde14f87a/div.pickle\n\u001b[1;30mINFO\u001b[0m [job get_sum] completed success\n\u001b[1;30mINFO\u001b[0m [step get_sum] completed success\n\u001b[1;30mINFO\u001b[0m [workflow ] completed success\n{\n    \"result_file\": {\n        \"location\": \"file:///home/jovyan/result.pickle\",\n        \"basename\": \"result.pickle\",\n        \"class\": \"File\",\n        \"checksum\": \"sha1$3dfd802cefb317cc7138af1e3a299f565c74ddec\",\n        \"size\": 21,\n        \"path\": \"/home/jovyan/result.pickle\"\n    }\n}\u001b[1;30mINFO\u001b[0m Final process status is success\n"}],"execution_count":19},{"id":"2942dbba-ea0a-4d20-be5c-ed9992d09ff8","cell_type":"code","source":"with open(\"result.pickle\", \"rb\") as f:\n    print(pickle.load(f))","metadata":{"trusted":true},"outputs":[{"name":"stdout","output_type":"stream","text":"2.5\n"}],"execution_count":20},{"id":"60e909ee-d0d0-4bd1-81c8-dd5274ae5834","cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}

python_workflow_definition/src/python_workflow_definition/cwl/__init__.py

@@ -0,0 +1,42 @@
+import sys
+import pickle
+from ast import literal_eval
+from importlib import import_module
+
+
+def load_function(funct):
+    p, m = funct.rsplit('.', 1)
+    return getattr(import_module(p), m)
+
+
+def convert_argument(arg):
+    if ".pickle" in arg:
+        with open(arg, 'rb') as f:
+            return pickle.load(f)
+    else:
+        return literal_eval(arg)
+
+
+if __name__ == "__main__":
+    # load input
+    argument_lst = sys.argv[1:]
+    funct = [
+        load_function(funct=arg.split("=")[-1])
+        for arg in argument_lst if "--function=" in arg
+    ][0]
+    kwargs = {
+        arg.split("=")[0][6:]: convert_argument(arg=arg.split("=")[-1])
+        for arg in argument_lst if "--arg_" in arg
+    }
+
+    # evaluate function
+    result = funct(**kwargs)
+
+    # store output
+    if isinstance(result, dict):
+        for k, v in result.items():
+            with open(k + ".pickle", 'wb') as f:
+                pickle.dump(v, f)
+    else:
+        with open("result.pickle", 'wb') as f:
+            pickle.dump(result, f)

python_workflow_definition/src/python_workflow_definition/cwl/__main__.py

@@ -0,0 +1,174 @@
+import json
+import pickle
+from yaml import CDumper as Dumper, dump
+
+
+from python_workflow_definition.purepython import (
+    group_edges,
+    resort_total_lst,
+)
+from python_workflow_definition.shared import (
+    convert_nodes_list_to_dict,
+    remove_result,
+    EDGES_LABEL,
+    NODES_LABEL,
+    TARGET_LABEL,
+    TARGET_PORT_LABEL,
+    SOURCE_LABEL,
+    SOURCE_PORT_LABEL,
+)
+
+
+def get_function_argument(argument: str, position: int = 3) -> dict:
+    return {
+        argument + '_file': {
+            'type': 'File',
+            'inputBinding': {'prefix': '--arg_' + argument + '=', 'separate': False, 'position': position},
+        },
+    }
+
+
+def get_function_template(function_name: str) -> dict:
+    return {
+        'function': {
+            'default': function_name,
+            'inputBinding': {'position': 2, 'prefix': '--function=', 'separate': False},
+            'type': 'string',
+        },
+    }
+
+
+def get_output_name(output_name: str) -> dict:
+    return {
+        output_name + '_file': {
+            'type': 'File',
+            'outputBinding': {
+                'glob': output_name + '.pickle'
+            }
+        }
+    }
+
+
+def get_function(workflow):
+    function_nodes_dict = {
+        n['id']: n['value']
+        for n in workflow[NODES_LABEL] if n["type"] == "function"
+    }
+    funct_dict = {}
+    for funct_id in function_nodes_dict.keys():
+        target_ports = list(set([e[TARGET_PORT_LABEL] for e in workflow[EDGES_LABEL] if e["target"] == funct_id]))
+        source_ports = list(set([e[SOURCE_PORT_LABEL] for e in workflow[EDGES_LABEL] if e["source"] == funct_id]))
+        funct_dict[funct_id] = {"targetPorts": target_ports, "sourcePorts": source_ports}
+    return function_nodes_dict, funct_dict
+
+
+def write_function_cwl(workflow):
+    function_nodes_dict, funct_dict = get_function(workflow)
+    file_lst = []
+
+    for i in range(len(function_nodes_dict)):
+        template = {
+            'cwlVersion': 'v1.2',
+            'class': 'CommandLineTool',
+            'baseCommand': 'python',
+            'inputs': {
+                'wrapper': {
+                    'type': 'File',
+                    'inputBinding': {'position': 1},
+                    'default': {'class': 'File', 'location': 'wrapper.py'}
+                },
+            },
+            'outputs': {
+            }
+        }
+        file_name = function_nodes_dict[i].split(".")[-1] + ".cwl"
+        if file_name not in file_lst:
+            file_lst.append(file_name)
+            template["inputs"].update(get_function_template(function_name=function_nodes_dict[i]))
+            for j, arg in enumerate(funct_dict[i]['targetPorts']):
+                template["inputs"].update(get_function_argument(argument=arg, position=3+j))
+            for out in funct_dict[i]['sourcePorts']:
+                if out is None:
+                    template["outputs"].update(get_output_name(output_name="result"))
+                else:
+                    template["outputs"].update(get_output_name(output_name=out))
+            with open(file_name, "w") as f:
+                dump(template, f, Dumper=Dumper)
+
+
+def write_workflow_config(workflow):
+    input_dict = {
+        n["name"]: n["value"]
+        for n in workflow[NODES_LABEL] if n["type"] == "input"
+    }
+    with open("workflow.yml", "w") as f:
+        dump(
+            {
+                k + "_file": {"class": "File", "path": k + ".pickle"}
+                for k in input_dict.keys()
+            },
+            f,
+            Dumper=Dumper,
+        )
+    for k, v in input_dict.items():
+        with open(k + ".pickle", "wb") as f:
+            pickle.dump(v, f)
+
+
+def write_workflow(workflow):
+    workflow_template = {
+        'cwlVersion': 'v1.2',
+        'class': 'Workflow',
+        'inputs': {},
+        'steps': {},
+        'outputs': {},
+    }
+    input_dict = {
+        n["name"]: n["value"]
+        for n in workflow[NODES_LABEL] if n["type"] == "input"
+    }
+    function_nodes_dict, funct_dict = get_function(workflow)
+    result_id = [n["id"] for n in workflow[NODES_LABEL] if n["type"] == "output"][0]
+    last_compute_id = [e[SOURCE_LABEL] for e in workflow[EDGES_LABEL] if e[TARGET_LABEL] == result_id][0]
+    workflow_template["inputs"].update({k + "_file": "File" for k in input_dict.keys()})
+    if funct_dict[last_compute_id]["sourcePorts"] == [None]:
+        workflow_template["outputs"] = {
+            "result_file": {
+                "type": "File",
+                "outputSource": function_nodes_dict[last_compute_id].split(".")[-1] + "/result_file"
+            },
+        }
+    else:
+        raise ValueError()
+
+    content = remove_result(workflow_dict=workflow)
+    edges_new_lst = content[EDGES_LABEL]
+    total_lst = group_edges(edges_new_lst)
+    nodes_new_dict = {
+        int(k): v for k, v in convert_nodes_list_to_dict(nodes_list=content[NODES_LABEL]).items()
+    }
+    total_new_lst = resort_total_lst(total_lst=total_lst, nodes_dict=nodes_new_dict)
+    step_name_lst = {t[0]: function_nodes_dict[t[0]].split(".")[-1] for t in total_new_lst}
+    input_id_dict = {n["id"]: n["name"] for n in workflow[NODES_LABEL] if n["type"] == "input"}
+    for t in total_new_lst:
+        ind = t[0]
+        node_script = step_name_lst[ind] + ".cwl"
+        output = [o + "_file" if o is not None else "result_file" for o in funct_dict[ind]['sourcePorts']]
+        in_dict = {}
+        for k, v in t[1].items():
+            if v[SOURCE_LABEL] in input_id_dict:
+                in_dict[k + "_file"] = input_id_dict[v[SOURCE_LABEL]] + "_file"
+            else:
+                in_dict[k + "_file"] = step_name_lst[v[SOURCE_LABEL]] + "/" + v[SOURCE_PORT_LABEL] + "_file"
+        workflow_template["steps"].update({step_name_lst[ind]: {"run": node_script, "in": in_dict, "out": output}})
+    with open("workflow.cwl", "w") as f:
+        dump(workflow_template, f, Dumper=Dumper)
+
+
+def load_workflow_json(file_name: str):
+    with open(file_name, "r") as f:
+        workflow = json.load(f)
+
+    write_function_cwl(workflow=workflow)
+    write_workflow_config(workflow=workflow)
+    write_workflow(workflow=workflow)