Merge pull request #397 from chasejohnson3/output_directory

Output directory from output_file_template

Author: djarecka

pydra/engine/helpers_file.py

@@ -556,7 +556,7 @@ def template_update_single(field, inputs_dict, output_dir=None, spec_type="input
     based on the value from inputs_dict
     (checking the types of the fields, that have "output_file_template)"
     """
-    from .specs import File, MultiOutputFile
+    from .specs import File, MultiOutputFile, Directory
 
     if spec_type == "input":
         if field.type not in [str, ty.Union[str, bool]]:
@@ -574,7 +574,7 @@ def template_update_single(field, inputs_dict, output_dir=None, spec_type="input
                 f"type of {field.name} is str, consider using Union[str, bool]"
             )
     elif spec_type == "output":
-        if field.type not in [File, MultiOutputFile]:
+        if field.type not in [File, MultiOutputFile, Directory]:
             raise Exception(
                 f"output {field.name} should be a File, but {field.type} set as the type"
             )

pydra/engine/tests/test_shelltask.py

@@ -13,6 +13,7 @@
     ShellSpec,
     SpecInfo,
     File,
+    Directory,
     MultiOutputFile,
     MultiInputObj,
 )
@@ -2927,6 +2928,110 @@ def test_shell_cmd_outputspec_7b_error():
     assert "has to have a callable" in str(e.value)
 
 
+@pytest.mark.parametrize("results_function", [result_no_submitter, result_submitter])
+def test_shell_cmd_outputspec_7c(tmpdir, plugin, results_function):
+    """
+        customised output_spec, adding Directory to the output named by args
+    """
+
+    def get_lowest_directory(directory_path):
+        return str(directory_path).replace(str(Path(directory_path).parents[0]), "")
+
+    cmd = "mkdir"
+    args = [f"{tmpdir}/dir1", f"{tmpdir}/dir2"]
+
+    my_output_spec = SpecInfo(
+        name="Output",
+        fields=[
+            (
+                "resultsDir",
+                attr.ib(
+                    type=Directory,
+                    metadata={
+                        "output_file_template": "{args}",
+                        "help_string": "output file",
+                    },
+                ),
+            )
+        ],
+        bases=(ShellOutSpec,),
+    )
+
+    shelly = ShellCommandTask(
+        name="shelly",
+        executable=cmd,
+        args=args,
+        output_spec=my_output_spec,
+        resultsDir="outdir",
+    ).split("args")
+
+    res = results_function(shelly, plugin)
+    for index, arg_dir in enumerate(args):
+        assert Path(Path(tmpdir) / Path(arg_dir)).exists() == True
+        assert get_lowest_directory(arg_dir) == f"/dir{index+1}"
+
+
+@pytest.mark.parametrize("results_function", [result_no_submitter, result_submitter])
+def test_shell_cmd_outputspec_7d(tmpdir, plugin, results_function):
+    """
+        customised output_spec, adding Directory to the output named by input spec
+    """
+
+    def get_lowest_directory(directory_path):
+        return str(directory_path).replace(str(Path(directory_path).parents[0]), "")
+
+    cmd = "mkdir"
+
+    my_input_spec = SpecInfo(
+        name="Input",
+        fields=[
+            (
+                "resultsDir",
+                attr.ib(
+                    type=str,
+                    metadata={
+                        "position": 1,
+                        "help_string": "new directory",
+                        "argstr": "",
+                    },
+                ),
+            )
+        ],
+        bases=(ShellSpec,),
+    )
+
+    my_output_spec = SpecInfo(
+        name="Output",
+        fields=[
+            (
+                "resultsDir",
+                attr.ib(
+                    type=Directory,
+                    metadata={
+                        "output_file_template": "{resultsDir}",
+                        "help_string": "output file",
+                    },
+                ),
+            )
+        ],
+        bases=(ShellOutSpec,),
+    )
+
+    shelly = ShellCommandTask(
+        name="shelly",
+        executable=cmd,
+        input_spec=my_input_spec,
+        output_spec=my_output_spec,
+        resultsDir=Path(tmpdir) / Path("test"),
+    )
+
+    res = results_function(shelly, plugin)
+    assert (Path(tmpdir) / Path("test")).exists() == True
+    assert get_lowest_directory(res.output.resultsDir) == get_lowest_directory(
+        Path(tmpdir) / Path("test")
+    )
+
+
 @pytest.mark.parametrize("results_function", [result_no_submitter, result_submitter])
 def test_shell_cmd_state_outputspec_1(plugin, results_function, tmpdir):
     """