Add janis converter

aclimatise/__init__.py

@@ -2,6 +2,7 @@
 
 from aclimatise.converter import WrapperGenerator
 from aclimatise.converter.cwl import CwlGenerator
+from aclimatise.converter.janis import JanisGenerator
 from aclimatise.converter.wdl import WdlGenerator
 from aclimatise.converter.yml import YmlGenerator
 from aclimatise.execution import Executor
@@ -38,6 +39,7 @@ def explore_command(
     CwlGenerator,
     WdlGenerator,
     YmlGenerator,
+    JanisGenerator,
     LocalExecutor,
     DockerExecutor,
     ManPageExecutor,

aclimatise/cli.py

@@ -126,7 +126,7 @@ def explore(
 @click.option(
     "--format",
     "-f",
-    type=click.Choice(["wdl", "cwl", "yml"]),
+    type=click.Choice(["wdl", "cwl", "yml", "janis"]),
     default="cwl",
     help="The language in which to output the CLI wrapper",
 )

aclimatise/converter/janis.py

@@ -0,0 +1,107 @@
+from typing import List
+
+import janis_core as janis
+from aclimatise import cli_types
+from aclimatise.cli_types import CliType
+from aclimatise.converter import NamedArgument, WrapperGenerator
+from aclimatise.model import CliArgument, Command, Flag, Positional
+
+
+class JanisGenerator(WrapperGenerator):
+    @classmethod
+    def format(cls) -> str:
+        return "janis"
+
+    def save_to_string(self, cmd: Command) -> str:
+
+        clt = self.command_to_tool(cmd)
+        return clt.translate("janis", to_console=False)
+
+    def command_to_tool(self, cmd: Command) -> janis.CommandToolBuilder:
+
+        inputs: List[CliArgument] = [*cmd.named] + (
+            [] if self.ignore_positionals else [*cmd.positional]
+        )
+        names = self.choose_variable_names(inputs)
+
+        tool = janis.CommandToolBuilder(
+            tool=cmd.as_filename,
+            base_command=list(cmd.command),
+            inputs=self.get_inputs(names),
+            outputs=self.get_outputs(names),
+            version="v0.1.0",
+            container=cmd.docker_image,
+        )
+
+        return tool
+
+    def type_to_janis_type(
+        self, typ: cli_types.CliType, optional: bool
+    ) -> janis.DataType:
+
+        if isinstance(typ, cli_types.CliFile):
+            return janis.File(optional=optional)
+        elif isinstance(typ, cli_types.CliDir):
+            return janis.Directory(optional=optional)
+        elif isinstance(typ, cli_types.CliString):
+            return janis.String(optional=optional)
+        elif isinstance(typ, cli_types.CliFloat):
+            return janis.Float(optional=optional)
+        elif isinstance(typ, cli_types.CliInteger):
+            return janis.Int(optional=optional)
+        elif isinstance(typ, cli_types.CliBoolean):
+            return janis.Boolean(optional=optional)
+        elif isinstance(typ, cli_types.CliEnum):
+            return janis.String(optional=optional)
+        elif isinstance(typ, cli_types.CliList):
+            # TODO: how is Array<String?> represented?
+            inner = self.type_to_janis_type(typ.value, optional=False)
+            return janis.Array(inner, optional=optional)
+
+        elif isinstance(typ, cli_types.CliTuple):
+            return self.type_to_janis_type(
+                CliType.lowest_common_type(typ.values), optional=False
+            )
+        else:
+            raise Exception(f"Invalid type {typ}!")
+
+    def arg_to_janis_type(self, arg: CliArgument) -> janis.DataType:
+        return self.type_to_janis_type(arg.get_type(), arg.optional)
+
+    def get_inputs(self, names: List[NamedArgument]) -> List[janis.ToolInput]:
+        ret = []
+        for arg in names:
+            assert arg.name != "", arg
+            ret.append(
+                janis.ToolInput(
+                    tag="in_" + arg.name,
+                    input_type=self.arg_to_janis_type(arg.arg),
+                    position=arg.arg.position
+                    if isinstance(arg.arg, Positional)
+                    else None,
+                    prefix=arg.arg.longest_synonym
+                    if isinstance(arg.arg, Flag)
+                    else None,
+                    doc=arg.arg.description,
+                )
+            )
+        return ret
+
+    def get_outputs(self, names: List[NamedArgument]) -> List[janis.ToolOutput]:
+        ret = []
+        for arg in names:
+            typ = arg.arg.get_type()
+            if isinstance(typ, cli_types.CliFileSystemType) and typ.output:
+                ret.append(
+                    janis.ToolOutput(
+                        tag="out_" + arg.name,
+                        output_type=self.arg_to_janis_type(arg.arg),
+                        doc=arg.arg.description,
+                        selector=janis.InputSelector("in_" + arg.name),
+                    )
+                )
+        return ret
+
+    @property
+    def suffix(self) -> str:
+        return ".py"

setup.py

@@ -21,6 +21,7 @@
         "psutil",
         "deprecated",
         "attrs",
+        "janis-pipelines.core >= 0.11.2",
     ],
     python_requires=">=3.6",
     entry_points={"console_scripts": ["aclimatise = aclimatise.cli:main"]},

test/test_cli.py

@@ -11,7 +11,7 @@
 from aclimatise.cli import main
 from aclimatise.yaml import yaml
 
-from .util import skip_not_installed, validate_cwl, validate_wdl
+from .util import skip_not_installed, validate_cwl, validate_janis, validate_wdl
 
 
 @pytest.fixture()
@@ -43,6 +43,14 @@ def test_pipe_cwl(runner, htseq_help):
     validate_cwl(result.output)
 
 
+def test_pipe_janis(runner, htseq_help):
+    result = runner.invoke(
+        main, ["pipe", "htseq-count", "--format", "janis"], input=htseq_help
+    )
+    cli_worked(result)
+    validate_janis(result.output)
+
+
 @skip_not_installed("htseq-count")
 def test_explore_htseq(runner, caplog):
     caplog.set_level(100000)

test/util.py

@@ -402,6 +402,20 @@ def validate_cwl(cwl: str, cmd: Command = None, explore: bool = True):
             assert len(parsed["outputs"]) == len(cmd.outputs) + 1
 
 
+def validate_janis(janis: str, cmd: Command = None):
+    fn = "temp-janis"
+    if cmd and cmd.as_filename:
+        fn = cmd.as_filename
+    with tempfile.TemporaryDirectory() as tmpdir:
+        tmpdir = Path(tmpdir)
+        tmpfile = tmpdir / (fn + ".py")
+        tmpfile.write_text(janis)
+
+        # just run the python file, and if it succeeds we're sweet
+        result = subprocess.run(["python", tmpfile])
+        assert result.returncode == 0
+
+
 def validate_wdl(wdl: str, cmd: Command = None, explore=True):
     """
     :param explore: If true, we're in explore mode, and we should ignore subcommands