Created a directory to collect the workflow output data

streamflow/cwl/main.py

@@ -74,6 +74,7 @@ async def main(
         context=context,
         name=args.name,
         output_directory=args.outdir,
+        hierarchical_output=args.hierarchical,
         cwl_definition=cwl_definition,
         cwl_inputs=cwl_inputs,
         cwl_inputs_path=cwl_args[1] if len(cwl_args) == 2 else None,

streamflow/cwl/runner.py

@@ -87,6 +87,7 @@ async def _async_main(args: argparse.Namespace) -> None:
     streamflow_config["path"] = (
         args.streamflow_file if args.streamflow_file is not None else os.getcwd()
     )
+    args.hierarchical = False
     workflow_config = WorkflowConfig(workflow_name, streamflow_config)
     context = build_context(streamflow_config)
     try:

streamflow/cwl/translator.py

@@ -1589,6 +1589,7 @@ def __init__(
         context: StreamFlowContext,
         name: str,
         output_directory: str,
+        hierarchical_output: bool,
         cwl_definition: (
             cwl_utils.parser.CommandLineTool
             | cwl_utils.parser.ExpressionTool
@@ -1601,6 +1602,7 @@ def __init__(
         self.context: StreamFlowContext = context
         self.name: str = name
         self.output_directory: str = output_directory
+        self.hierarchical_output: bool = hierarchical_output
         self.cwl_definition: (
             cwl_utils.parser.CommandLineTool
             | cwl_utils.parser.ExpressionTool
@@ -3143,7 +3145,14 @@ def translate(self) -> Workflow:
                         connector_ports={
                             target.deployment.name: deploy_step.get_output_port()
                         },
-                        input_directory=self.output_directory,
+                        input_directory=(
+                            os.path.join(
+                                self.output_directory,
+                                output_name.lstrip(posixpath.sep),
+                            )
+                            if self.hierarchical_output
+                            else self.output_directory
+                        ),
                         binding_config=BindingConfig(targets=[target]),
                     )
                     # Add the port as an input of the schedule step

streamflow/parser.py

@@ -248,6 +248,11 @@ def __call__(
     type=str,
     help="Output directory in which to store final results of the workflow (default: current directory)",
 )
+run_parser.add_argument(
+    "--hierarchical",
+    action="store_true",
+    help="Preserve the output port structure (default: collapse all files into outdir)",
+)
 run_parser.add_argument(
     "--quiet", action="store_true", help="Only prints results, warnings and errors"
 )

tests/test_translator.py

@@ -194,6 +194,7 @@ async def test_inject_remote_input(
         context=context,
         name=utils.random_name(),
         output_directory=tempfile.gettempdir(),
+        hierarchical_output=True,
         cwl_definition=None,  # CWL object
         cwl_inputs=cwl_inputs,
         cwl_inputs_path=None,