Repository: nipype/pydra

Issue:
TODO for Shell Task and ContainerTask
There are some issues tak don't have to be fixed right away, but should be done soon:
(will be updated)

- [x] cmdline for Shell: doesn't make sense for tasks with splitter

- [x] inputs for Containers: 
      - checking if has image and it's proper image, raising exceptions; 
      - bindings: allowing for tuple (not only list), tuple should be 3 or 2 elements (without `ro`)
      - checking if the local directory exists

- [x] adding an option to `Task` that converts it to the `DockerTask`: see @satra comments in #144 

- [x] provide option to have position for arguments

Discussion:

Patch:
diff --git a/pydra/engine/core.py b/pydra/engine/core.py
index 11111396..7cd8b7b8 100644
--- a/pydra/engine/core.py
+++ b/pydra/engine/core.py
@@ -247,9 +247,9 @@ class TaskBase:
             TODO
 
         """
+        self.state.prepare_states(self.inputs)
+        self.state.prepare_inputs()
         if state_index is not None:
-            if self.state is None:
-                raise Exception("can't use state_index if no splitter is used")
             inputs_copy = deepcopy(self.inputs)
             for key, ind in self.state.inputs_ind[state_index].items():
                 setattr(
@@ -510,6 +510,9 @@ class TaskBase:
     @property
     def done(self):
         """Check whether the tasks has been finalized and all outputs are stored."""
+        # if any of the field is lazy, there is no need to check results
+        if is_lazy(self.inputs):
+            return False
         if self.state:
             # TODO: only check for needed state result
             if self.result() and all(self.result()):
@@ -866,3 +869,11 @@ def is_task(obj):
 def is_workflow(obj):
     """Check whether an object is a :class:`Workflow` instance."""
     return isinstance(obj, Workflow)
+
+
+def is_lazy(obj):
+    """Check whether an object has any field that is a Lazy Field"""
+    for f in attr_fields(obj):
+        if isinstance(getattr(obj, f.name), LazyField):
+            return True
+    return False
diff --git a/pydra/engine/task.py b/pydra/engine/task.py
index 6f1d2ebe..0519b78e 100644
--- a/pydra/engine/task.py
+++ b/pydra/engine/task.py
@@ -44,7 +44,7 @@ import inspect
 import typing as ty
 from pathlib import Path
 
-from .core import TaskBase
+from .core import TaskBase, is_lazy
 from ..utils.messenger import AuditFlag
 from .specs import (
     File,
@@ -215,17 +215,17 @@ class ShellCommandTask(TaskBase):
             return super(ShellCommandTask, cls).__new__(cls)
 
         if len(container_info) == 3:
-            type_cont, image, bind = container_info
+            type, image, bind = container_info
         elif len(container_info) == 2:
-            type_cont, image, bind = container_info + (None,)
+            type, image, bind = container_info + (None,)
         else:
             raise Exception(
                 f"container_info has to have 2 or 3 elements, but {container_info} provided"
             )
 
-        if type_cont == "docker":
+        if type == "docker":
             return DockerTask(image=image, bindings=bind, *args, **kwargs)
-        elif type_cont == "singularity":
+        elif type == "singularity":
             return SingularityTask(image=image, bindings=bind, *args, **kwargs)
         else:
             raise Exception(
@@ -288,7 +288,20 @@ class ShellCommandTask(TaskBase):
 
     @property
     def command_args(self):
-        """Get command line arguments."""
+        """Get command line arguments, returns a list if task has a state"""
+        if is_lazy(self.inputs):
+            raise Exception("can't return cmdline, self.inputs has LazyFields")
+        if self.state:
+            command_args_list = []
+            self.state.prepare_states(self.inputs)
+            for ii, el in enumerate(self.state.states_ind):
+                command_args_list.append(self._command_args_single(el, ind=ii))
+            return command_args_list
+        else:
+            return self._command_args_single(self.inputs)
+
+    def _command_args_single(self, state_ind, ind=None):
+        """Get command line arguments for a single state"""
         pos_args = []  # list for (position, command arg)
         for f in attr_fields(self.inputs):
             if f.name == "executable":
@@ -311,13 +324,16 @@ class ShellCommandTask(TaskBase):
             # if f.metadata.get("copyfile") in [True, False]:
             #    value = str(self.inputs.map_copyfiles[f.name])
             # else:
-            value = getattr(self.inputs, f.name)
+            if self.state and f"{self.name}.{f.name}" in state_ind:
+                value = getattr(self.inputs, f.name)[state_ind[f"{self.name}.{f.name}"]]
+            else:
+                value = getattr(self.inputs, f.name)
             if is_local_file(f):
                 value = str(value)
             # changing path to the cpath (the directory should be mounted)
             if getattr(self, "bind_paths", None) and is_local_file(f):
                 lpath = Path(value)
-                cdir = self.bind_paths[lpath.parent][0]
+                cdir = self.bind_paths(ind=ind)[lpath.parent][0]
                 cpath = cdir.joinpath(lpath.name)
                 value = str(cpath)
             if f.type is bool:
@@ -338,21 +354,32 @@ class ShellCommandTask(TaskBase):
             cmd_args += el[1]
         return cmd_args
 
-    @command_args.setter
-    def command_args(self, args: ty.Dict):
-        self.inputs = attr.evolve(self.inputs, **args)
-
     @property
     def cmdline(self):
-        """Get the actual command line that will be submitted."""
+        """ Get the actual command line that will be submitted
+            Returns a list if the task has a state.
+        """
+        if is_lazy(self.inputs):
+            raise Exception("can't return cmdline, self.inputs has LazyFields")
         orig_inputs = attr.asdict(self.inputs)
         modified_inputs = template_update(self.inputs)
         if modified_inputs is not None:
             self.inputs = attr.evolve(self.inputs, **modified_inputs)
         if isinstance(self, ContainerTask):
-            cmdline = " ".join(self.container_args + self.command_args)
+            if self.state:
+                cmdline = []
+                for con, com in zip(self.container_args, self.command_args):
+                    cmdline.append(" ".join(con + com))
+            else:
+                cmdline = " ".join(self.container_args + self.command_args)
         else:
-            cmdline = " ".join(self.command_args)
+            if self.state:
+                cmdline = []
+                for el in self.command_args:
+                    cmdline.append(" ".join(el))
+            else:
+                cmdline = " ".join(self.command_args)
+
         self.inputs = attr.evolve(self.inputs, **orig_inputs)
         return cmdline
 
@@ -428,23 +455,27 @@ class ContainerTask(ShellCommandTask):
             **kwargs,
         )
 
-    @property
-    def container_args(self):
+    def container_check(self, container_type):
         """Get container-specific CLI arguments."""
         if self.inputs.container is None:
             raise AttributeError("Container software is not specified")
-        cargs = [self.inputs.container]
+        elif self.inputs.container != container_type:
+            raise AttributeError(
+                f"Container type should be {container_type}, but {self.inputs.container} given"
+            )
         if self.inputs.image is None:
             raise AttributeError("Container image is not specified")
-        return cargs
 
-    @property
-    def bind_paths(self):
+    def bind_paths(self, ind=None):
         """Return bound mount points: ``dict(lpath: (cpath, mode))``."""
         bind_paths = {}
         output_dir_cpath = None
         if self.inputs.bindings is None:
             self.inputs.bindings = []
+        if ind is None:
+            output_dir = self.output_dir
+        else:
+            output_dir = self.output_dir[ind]
         for binding in self.inputs.bindings:
             if len(binding) == 3:
                 lpath, cpath, mode = binding
@@ -454,17 +485,17 @@ class ContainerTask(ShellCommandTask):
                 raise Exception(
                     f"binding should have length 2, 3, or 4, it has {len(binding)}"
                 )
-            if Path(lpath) == self.output_dir:
+            if Path(lpath) == output_dir:
                 output_dir_cpath = cpath
             if mode is None:
                 mode = "rw"  # default
             bind_paths[Path(lpath)] = (Path(cpath), mode)
         # output_dir is added to the bindings if not part of self.inputs.bindings
         if not output_dir_cpath:
-            bind_paths[self.output_dir] = (self.output_cpath, "rw")
+            bind_paths[output_dir] = (self.output_cpath, "rw")
         return bind_paths
 
-    def binds(self, opt):
+    def binds(self, opt, ind=None):
         """
         Specify mounts to bind from local filesystems to container and working directory.
 
@@ -472,9 +503,8 @@ class ContainerTask(ShellCommandTask):
 
         """
         bargs = []
-        for (key, val) in self.bind_paths.items():
+        for (key, val) in self.bind_paths(ind).items():
             bargs.extend([opt, "{0}:{1}:{2}".format(key, val[0], val[1])])
-        # TODO: would need changes for singularity
         return bargs
 
 
@@ -542,16 +572,36 @@ class DockerTask(ContainerTask):
 
     @property
     def container_args(self):
-        """Get container-specific CLI arguments."""
-        cargs = super().container_args
-        assert self.inputs.container == "docker"
-        cargs.append("run")
+        """Get container-specific CLI arguments, returns a list if the task has a state"""
+        if is_lazy(self.inputs):
+            raise Exception("can't return container_args, self.inputs has LazyFields")
+        self.container_check("docker")
+        if self.state:
+            self.state.prepare_states(self.inputs)
+            cargs_list = []
+            for ii, el in enumerate(self.state.states_ind):
+                if f"{self.name}.image" in el:
+                    cargs_list.append(
+                        self._container_args_single(
+                            self.inputs.image[el[f"{self.name}.image"]], ind=ii
+                        )
+                    )
+                else:
+                    cargs_list.append(
+                        self._container_args_single(self.inputs.image, ind=ii)
+                    )
+            return cargs_list
+        else:
+            return self._container_args_single(self.inputs.image)
+
+    def _container_args_single(self, image, ind=None):
+        cargs = ["docker", "run"]
         if self.inputs.container_xargs is not None:
             cargs.extend(self.inputs.container_xargs)
 
-        cargs.extend(self.binds("-v"))
+        cargs.extend(self.binds("-v", ind))
         cargs.extend(["-w", str(self.output_cpath)])
-        cargs.append(self.inputs.image)
+        cargs.append(image)
 
         return cargs
 
@@ -615,14 +665,35 @@ class SingularityTask(ContainerTask):
     @property
     def container_args(self):
         """Get container-specific CLI arguments."""
-        cargs = super().container_args
-        assert self.inputs.container == "singularity"
-        cargs.append("exec")
+        if is_lazy(self.inputs):
+            raise Exception("can't return container_args, self.inputs has LazyFields")
+        self.container_check("singularity")
+        if self.state:
+            self.state.prepare_states(self.inputs)
+            cargs_list = []
+            for ii, el in enumerate(self.state.states_ind):
+                if f"{self.name}.image" in el:
+                    cargs_list.append(
+                        self._container_args_single(
+                            self.inputs.image[el[f"{self.name}.image"]], ind=ii
+                        )
+                    )
+                else:
+                    cargs_list.append(
+                        self._container_args_single(self.inputs.image, ind=ii)
+                    )
+            return cargs_list
+        else:
+            return self._container_args_single(self.inputs.image)
+
+    def _container_args_single(self, image, ind=None):
+        cargs = ["singularity", "exec"]
+
         if self.inputs.container_xargs is not None:
             cargs.extend(self.inputs.container_xargs)
-        cargs.append(self.inputs.image)
+        cargs.append(image)
 
         # insert bindings before image
         idx = len(cargs) - 1
-        cargs[idx:-1] = self.binds("-B")
+        cargs[idx:-1] = self.binds("-B", ind)
         return cargs


Test patch:
diff --git a/pydra/engine/tests/test_dockertask.py b/pydra/engine/tests/test_dockertask.py
index 6a823007..4841cf2a 100644
--- a/pydra/engine/tests/test_dockertask.py
+++ b/pydra/engine/tests/test_dockertask.py
@@ -352,6 +352,12 @@ def test_docker_st_1(plugin):
     )
     assert docky.state.splitter == "docky.executable"
 
+    for ii, el in enumerate(docky.cmdline):
+        assert (
+            el
+            == f"docker run --rm -v {docky.output_dir[ii]}:/output_pydra:rw -w /output_pydra {docky.inputs.image} {cmd[ii]}"
+        )
+
     res = docky(plugin=plugin)
     assert res[0].output.stdout == "/output_pydra\n"
     assert res[1].output.stdout == "root\n"
@@ -370,6 +376,12 @@ def test_docker_st_2(plugin):
     )
     assert docky.state.splitter == "docky.image"
 
+    for ii, el in enumerate(docky.cmdline):
+        assert (
+            el
+            == f"docker run --rm -v {docky.output_dir[ii]}:/output_pydra:rw -w /output_pydra {docky.inputs.image[ii]} {' '.join(cmd)}"
+        )
+
     res = docky(plugin=plugin)
     assert "Debian" in res[0].output.stdout
     assert "Ubuntu" in res[1].output.stdout
@@ -409,6 +421,17 @@ def test_docker_st_4(plugin):
     assert docky.state.combiner == ["docky.image"]
     assert docky.state.splitter_final == "docky.executable"
 
+    for ii, el in enumerate(docky.cmdline):
+        i, j = ii // 2, ii % 2
+        if j == 0:
+            cmd_str = "whoami"
+        else:
+            cmd_str = " ".join(["cat", "/etc/issue"])
+        assert (
+            el
+            == f"docker run --rm -v {docky.output_dir[ii]}:/output_pydra:rw -w /output_pydra {docky.inputs.image[i]} {cmd_str}"
+        )
+
     res = docky(plugin=plugin)
 
     # checking the first command
@@ -458,6 +481,10 @@ def test_wf_docker_1(plugin, tmpdir):
     )
     wf.set_output([("out", wf.docky_echo.lzout.stdout)])
 
+    with pytest.raises(Exception) as excinfo:
+        wf.docky_echo.cmdline
+    assert "can't return cmdline" in str(excinfo.value)
+
     with Submitter(plugin=plugin) as sub:
         wf(submitter=sub)
 
diff --git a/pydra/engine/tests/test_shelltask.py b/pydra/engine/tests/test_shelltask.py
index 0a5d6541..7284a4cd 100644
--- a/pydra/engine/tests/test_shelltask.py
+++ b/pydra/engine/tests/test_shelltask.py
@@ -122,8 +122,7 @@ def test_shell_cmd_3(plugin):
 
     # all args given as executable
     shelly = ShellCommandTask(name="shelly", executable=cmd).split("executable")
-    # TODO: doesnt make sense for tasks with splitter
-    #    assert shelly.cmdline == " ".join(cmd)
+    assert shelly.cmdline == ["pwd", "whoami"]
     res = shelly(plugin=plugin)
     assert res[0].output.stdout == f"{str(shelly.output_dir[0])}\n"
     if "USER" in os.environ:
@@ -147,8 +146,7 @@ def test_shell_cmd_4(plugin):
     )
     assert shelly.inputs.executable == "echo"
     assert shelly.inputs.args == ["nipype", "pydra"]
-    # this doesnt work, cmdline gives echo nipype pydra
-    # assert shelly.cmdline == "echo pydra"
+    assert shelly.cmdline == ["echo nipype", "echo pydra"]
     res = shelly(plugin=plugin)
 
     assert res[0].output.stdout == "nipype\n"
@@ -173,8 +171,7 @@ def test_shell_cmd_5(plugin):
     )
     assert shelly.inputs.executable == "echo"
     assert shelly.inputs.args == ["nipype", "pydra"]
-    # this doesnt work, cmdline gives echo nipype pydra
-    # assert shelly.cmdline == "echo pydra"
+    assert shelly.cmdline == ["echo nipype", "echo pydra"]
     res = shelly(plugin=plugin)
 
     assert res[0][0].output.stdout == "nipype\n"
@@ -194,8 +191,12 @@ def test_shell_cmd_6(plugin):
     )
     assert shelly.inputs.executable == ["echo", ["echo", "-n"]]
     assert shelly.inputs.args == ["nipype", "pydra"]
-    # this doesnt work, cmdline gives echo nipype pydra
-    # assert shelly.cmdline == "echo pydra"
+    assert shelly.cmdline == [
+        "echo nipype",
+        "echo pydra",
+        "echo -n nipype",
+        "echo -n pydra",
+    ]
     res = shelly(plugin=plugin)
 
     assert res[0].output.stdout == "nipype\n"
@@ -234,8 +235,7 @@ def test_shell_cmd_7(plugin):
     )
     assert shelly.inputs.executable == ["echo", ["echo", "-n"]]
     assert shelly.inputs.args == ["nipype", "pydra"]
-    # this doesnt work, cmdline gives echo nipype pydra
-    # assert shelly.cmdline == "echo pydra"
+
     res = shelly(plugin=plugin)
 
     assert res[0][0].output.stdout == "nipype\n"
@@ -2251,9 +2251,7 @@ def test_fsl():
 
     # TODO: not sure why this has to be string
     in_file = Path(os.path.dirname(os.path.abspath(__file__))) / "data" / "foo.nii"
-    out_file = (
-        Path(os.path.dirname(os.path.abspath(__file__))) / "data" / "foo_brain.nii"
-    )
+
     # separate command into exec + args
     shelly = ShellCommandTask(
         name="bet_task", executable="bet", in_file=in_file, input_spec=bet_input_spec
diff --git a/pydra/engine/tests/test_task.py b/pydra/engine/tests/test_task.py
index ce78efe6..45b47cd5 100644
--- a/pydra/engine/tests/test_task.py
+++ b/pydra/engine/tests/test_task.py
@@ -351,7 +351,7 @@ def test_shell_cmd(tmpdir):
 
 
 def test_container_cmds(tmpdir):
-    containy = ContainerTask(name="containy", executable="pwd")
+    containy = DockerTask(name="containy", executable="pwd")
     with pytest.raises(AttributeError):
         containy.cmdline
     containy.inputs.container = "docker"