diff --git a/pyproject.toml b/pyproject.toml index aa52b71..c5c757f 100644 --- a/pyproject.toml +++ b/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" diff --git a/src/ewoksgalaxy/engine.py b/src/ewoksgalaxy/engine.py new file mode 100644 index 0000000..fb353af --- /dev/null +++ b/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 diff --git a/src/ewoksgalaxy/io.py b/src/ewoksgalaxy/io.py new file mode 100644 index 0000000..c501a88 --- /dev/null +++ b/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 diff --git a/src/ewoksgalaxy/tests/conftest.py b/src/ewoksgalaxy/tests/conftest.py new file mode 100644 index 0000000..fb3901d --- /dev/null +++ b/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} diff --git a/src/ewoksgalaxy/tests/test_engine.py b/src/ewoksgalaxy/tests/test_engine.py new file mode 100644 index 0000000..bc35f28 --- /dev/null +++ b/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, + ) diff --git a/src/ewoksgalaxy/tests/test_io.py b/src/ewoksgalaxy/tests/test_io.py new file mode 100644 index 0000000..8fb36d3 --- /dev/null +++ b/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": [], + }, + ], + }