Implement serialization to Galaxy Format 2

pyproject.toml

@@ -17,7 +17,10 @@ classifiers = [
     "Programming Language :: Python :: 3",
 ]
 requires-python = ">=3.10"
-dependencies = ["ewokscore"]
+dependencies = [
+    "ewokscore",
+    "gxformat2@ git+https://github.com/galaxyproject/gxformat2", # To have https://github.com/galaxyproject/gxformat2/pull/134
+]
 
 [project.urls]
 Homepage = "https://github.com/ewoks-kit/ewoksgalaxy/"
@@ -27,7 +30,7 @@ Issues = "https://github.com/ewoks-kit/ewoksgalaxy/issues"
 Changelog = "https://github.com/ewoks-kit/ewoksgalaxy/blob/main/CHANGELOG.md"
 
 [project.optional-dependencies]
-test = ["pytest >=7"]
+test = ["pytest >=7", "ewoks"]
 dev = ["ewoksgalaxy[test]", "ruff"]
 doc = [
     "ewoksgalaxy[test]",
@@ -58,7 +61,13 @@ ignore = ["E203", "E501", "E701"]
 
 [tool.ruff.lint.per-file-ignores]
 # Ignore `S101` (assert used violations) in all test files
-"*/test_*.py" = ["S101"]
+"*/test_*.py" = ["S101", "S603"]
 
 [tool.ruff.lint.isort]
 force-single-line = true
+
+[project.entry-points."ewoks.engines"]
+"gxwf" = "ewoksgalaxy.engine:GalaxyWorkflowEngine"
+
+[project.entry-points."ewoks.engines.serialization.representations"]
+"gxwf" = "ewoksgalaxy.engine:GalaxyWorkflowEngine"

src/ewoksgalaxy/engine.py

@@ -0,0 +1,78 @@
+from pathlib import Path
+from typing import Any
+from typing import List
+from typing import Optional
+from typing import Union
+
+import yaml
+from ewokscore.engine_interface import RawExecInfoType
+from ewokscore.engine_interface import TaskGraph
+from ewokscore.engine_interface import WorkflowEngineWithSerialization
+
+from . import io
+
+
+class GalaxyWorkflowEngine(WorkflowEngineWithSerialization):
+    _GALAXY_REPR = "gxwf"
+
+    def execute_graph(
+        self,
+        graph: Any,
+        *,
+        inputs: Optional[List[dict]] = None,
+        load_options: Optional[dict] = None,
+        varinfo: Optional[dict] = None,
+        execinfo: RawExecInfoType = None,
+        task_options: Optional[dict] = None,
+        outputs: Optional[List[dict]] = None,
+        merge_outputs: Optional[bool] = True,
+        **execute_options,
+    ) -> None:
+        raise NotImplementedError("TODO")
+
+    def deserialize_graph(
+        self,
+        graph: Any,
+        *,
+        inputs: Optional[List[dict]] = None,
+        representation: Optional[str] = None,
+        root_dir: Optional[Union[str, Path]] = None,
+        root_module: Optional[str] = None,
+        **deserialize_options,
+    ) -> TaskGraph:
+        raise NotImplementedError("TODO")
+
+    def serialize_graph(
+        self,
+        graph: TaskGraph,
+        destination: str | Path,
+        *,
+        representation: Optional[str] = None,
+        **serialize_options,
+    ) -> Union[str, Path, dict]:
+        if representation is None:
+            representation = self.get_graph_representation(destination)
+
+        if representation == self._GALAXY_REPR:
+            galaxy_graph = io.ewoks_to_galaxy(graph)
+            with open(destination, "w") as dest_file:
+                yaml.dump(galaxy_graph, dest_file)
+            return destination
+
+        return graph.dump(
+            destination, representation=representation, **serialize_options
+        )
+
+    def get_graph_representation(self, graph: Any) -> Optional[str]:
+        if not isinstance(graph, (Path, str)):
+            return None
+
+        graph_path = str(graph)
+        if graph_path.endswith(".ga"):
+            return self._GALAXY_REPR
+
+        if graph_path.endswith(".yml") or graph_path.endswith(".yaml"):
+            graph_dict = yaml.safe_load(graph_path)
+            if graph_dict.get("class") == "GalaxyWorkflow":
+                return self._GALAXY_REPR
+        return None

src/ewoksgalaxy/io.py

@@ -0,0 +1,74 @@
+from typing import Any
+
+from ewokscore.engine_interface import TaskGraph
+
+
+def _ewoks_node_to_galaxy_step(node: dict) -> dict[str, Any]:
+    step: dict[str, Any] = {
+        "tool_id": node["task_identifier"],
+        "type": "tool",
+        "out": [],  # required according to specs but unused in real Galaxy workflows?
+    }
+    label = node.get("label")
+    if label is not None:
+        step["label"] = label
+
+    default_inputs = node.get("default_inputs")
+    if not default_inputs:
+        return step
+
+    state: dict[str, Any] = {}
+    for default_input in default_inputs:
+        name = default_input["name"]
+        value = default_input["value"]
+        state[name] = value
+
+    return {"state": state, **step}
+
+
+def _ewoks_link_to_galaxy_input(link: dict, galaxy_step_indices: dict[str, str]):
+    galaxy_input: dict[str, dict] = {}
+
+    for mapping in link["data_mapping"]:
+        galaxy_index = galaxy_step_indices[link["source"]]
+        galaxy_input[mapping["target_input"]] = {
+            "source": f"{galaxy_index}/{mapping['source_output']}"
+        }
+
+    return galaxy_input
+
+
+def ewoks_to_galaxy(raw_graph: TaskGraph) -> dict[str, Any]:
+    graph_dict = raw_graph.dump()
+
+    galaxy_graph: dict[str, Any] = {
+        "class": "GalaxyWorkflow",
+        "inputs": [],  # TODO
+        "outputs": [],  # TODO
+    }
+
+    graph_metadata: dict[str, str] = graph_dict["graph"]
+    graph_id = graph_metadata.get("id")
+    if graph_id is not None:
+        galaxy_graph["id"] = graph_id
+    label = graph_metadata.get("label")
+    if label is not None:
+        galaxy_graph["label"] = label
+
+    steps: list[dict] = []
+    # Keep a mapping between Ewoks ids and Galaxy step indices for parsing the data mapping
+    galaxy_step_indices: dict[str, int] = {}
+    for i, node in enumerate(graph_dict["nodes"]):
+        steps.append(_ewoks_node_to_galaxy_step(node))
+        galaxy_step_indices[node["id"]] = i
+
+    for link in graph_dict["links"]:
+        target_id = galaxy_step_indices[link["target"]]
+        if "in" not in steps[target_id]:
+            steps[target_id]["in"] = {}
+        steps[target_id]["in"].update(
+            _ewoks_link_to_galaxy_input(link, galaxy_step_indices)
+        )
+    galaxy_graph["steps"] = steps
+
+    return galaxy_graph

src/ewoksgalaxy/tests/conftest.py

@@ -0,0 +1,49 @@
+import pytest
+
+
+@pytest.fixture
+def ewoks_workflow():
+    graph = {
+        "id": "galaxy_test",
+        "label": "An Ewoks workflow to test Galaxy",
+        "schema_version": "1.1",
+    }
+
+    task = "ewokscore.tests.examples.tasks.sumtask.SumTask"
+    nodes = [
+        {
+            "id": "task1",
+            "default_inputs": [{"name": "a", "value": 1}],
+            "task_type": "class",
+            "task_identifier": task,
+        },
+        {
+            "id": "task2",
+            "default_inputs": [{"name": "b", "value": 2}],
+            "task_type": "class",
+            "task_identifier": task,
+        },
+        {
+            "id": "task3",
+            "default_inputs": [{"name": "b", "value": 1}],
+            "task_type": "class",
+            "task_identifier": task,
+        },
+    ]
+    links = [
+        {
+            "source": "task1",
+            "target": "task3",
+            "data_mapping": [{"source_output": "result", "target_input": "a"}],
+        },
+        {
+            "source": "task2",
+            "target": "task3",
+            "data_mapping": [
+                {"source_output": "result", "target_input": "b"},
+                {"source_output": "result", "target_input": "delay"},
+            ],
+        },
+    ]
+
+    return {"graph": graph, "links": links, "nodes": nodes}

src/ewoksgalaxy/tests/test_engine.py

@@ -0,0 +1,31 @@
+import json
+import subprocess
+import sys
+from pathlib import Path
+
+
+def test_convert(ewoks_workflow: dict, tmpdir: Path):
+    input_path = tmpdir / "workflow.json"
+    with open(input_path, "w") as f:
+        json.dump(ewoks_workflow, f)
+
+    output_path = tmpdir / "workflow.yaml"
+    subprocess.run(
+        [
+            sys.executable,
+            "-m",
+            "ewoks",
+            "convert",
+            str(input_path),
+            str(output_path),
+            "--dst-format",
+            "gxwf",
+        ],
+        check=True,
+    )
+    assert output_path.exists()
+
+    subprocess.run(
+        [sys.executable, "-m", "gxformat2.lint", str(output_path)],
+        check=True,
+    )

src/ewoksgalaxy/tests/test_io.py

@@ -0,0 +1,38 @@
+from ewokscore.engine_interface import TaskGraph
+
+from ewoksgalaxy.io import ewoks_to_galaxy
+
+
+def test_ewoks_to_galaxy(ewoks_workflow):
+    assert ewoks_to_galaxy(TaskGraph(ewoks_workflow)) == {
+        "class": "GalaxyWorkflow",
+        "inputs": [],
+        "outputs": [],
+        "id": "galaxy_test",
+        "label": "An Ewoks workflow to test Galaxy",
+        "steps": [
+            {
+                "state": {"a": 1},
+                "tool_id": "ewokscore.tests.examples.tasks.sumtask.SumTask",
+                "type": "tool",
+                "out": [],
+            },
+            {
+                "state": {"b": 2},
+                "tool_id": "ewokscore.tests.examples.tasks.sumtask.SumTask",
+                "type": "tool",
+                "out": [],
+            },
+            {
+                "state": {"b": 1},
+                "tool_id": "ewokscore.tests.examples.tasks.sumtask.SumTask",
+                "type": "tool",
+                "in": {
+                    "a": {"source": "0/result"},
+                    "b": {"source": "1/result"},
+                    "delay": {"source": "1/result"},
+                },
+                "out": [],
+            },
+        ],
+    }