Fill in args with defaults when using custom argument parser. Pass basedir

through kwargs instead of separate option.

Author: Peter Amstutz

cwltool/draft2tool.py

@@ -52,9 +52,9 @@ def run(self, **kwargs):  # type: (**Any) -> None
                 _logger.warn(u"Failed to evaluate expression:\n%s", e, exc_info=(e if kwargs.get('debug') else False))
                 self.output_callback({}, "permanentFail")
 
-    def job(self, joborder, input_basedir, output_callback, **kwargs):
+    def job(self, joborder, output_callback, **kwargs):
         # type: (Dict[str,str], str, Callable[[Any, Any], Any], **Any) -> Generator[ExpressionTool.ExpressionJob, None, None]
-        builder = self._init_job(joborder, input_basedir, **kwargs)
+        builder = self._init_job(joborder, **kwargs)
 
         j = ExpressionTool.ExpressionJob()
         j.builder = builder
@@ -119,19 +119,19 @@ def __init__(self, toolpath_object, **kwargs):
     def makeJobRunner(self):  # type: () -> CommandLineJob
         return CommandLineJob()
 
-    def makePathMapper(self, reffiles, input_basedir, **kwargs):
+    def makePathMapper(self, reffiles, **kwargs):
         # type: (Set[str], str, **Any) -> PathMapper
         dockerReq, _ = self.get_requirement("DockerRequirement")
         try:
             if dockerReq and kwargs.get("use_container"):
-                return DockerPathMapper(reffiles, input_basedir)
+                return DockerPathMapper(reffiles, kwargs["basedir"])
             else:
-                return PathMapper(reffiles, input_basedir)
+                return PathMapper(reffiles, kwargs["basedir"])
         except OSError as e:
             if e.errno == errno.ENOENT:
                 raise WorkflowException(u"Missing input file %s" % e)
 
-    def job(self, joborder, input_basedir, output_callback, **kwargs):
+    def job(self, joborder, output_callback, **kwargs):
         # type: (Dict[str,str], str, Callable[..., Any], **Any) -> Generator[Union[CommandLineJob, CallbackJob], None, None]
 
         jobname = uniquename(kwargs.get("name", shortname(self.tool.get("id", "job"))))
@@ -140,9 +140,9 @@ def job(self, joborder, input_basedir, output_callback, **kwargs):
             cacheargs = kwargs.copy()
             cacheargs["outdir"] = "/out"
             cacheargs["tmpdir"] = "/tmp"
-            cachebuilder = self._init_job(joborder, input_basedir, **cacheargs)
+            cachebuilder = self._init_job(joborder, **cacheargs)
             cachebuilder.pathmapper = PathMapper(set((f["path"] for f in cachebuilder.files)),
-                                                 input_basedir)
+                                                 kwargs["basedir"])
 
             cmdline = flatten(map(cachebuilder.generate_arg, cachebuilder.bindings))
             (docker_req, docker_is_req) = self.get_requirement("DockerRequirement")
@@ -198,7 +198,7 @@ def rm_pending_output_callback(output_callback, jobcachepending,
                         partial(rm_pending_output_callback, output_callback,
                             jobcachepending))
 
-        builder = self._init_job(joborder, input_basedir, **kwargs)
+        builder = self._init_job(joborder, **kwargs)
 
         reffiles = set((f["path"] for f in builder.files))
 
@@ -232,7 +232,7 @@ def rm_pending_output_callback(output_callback, jobcachepending,
             if os.path.isabs(j.stdout) or ".." in j.stdout:
                 raise validate.ValidationException("stdout must be a relative path")
 
-        builder.pathmapper = self.makePathMapper(reffiles, input_basedir, **kwargs)
+        builder.pathmapper = self.makePathMapper(reffiles, **kwargs)
         builder.requirements = j.requirements
 
         # map files to assigned path inside a container. We need to also explicitly

cwltool/factory.py

@@ -13,7 +13,9 @@ def __init__(self, t, factory):  # type: (Process, Factory) -> None
         self.factory = factory
 
     def __call__(self, **kwargs):  # type: (**Any) -> Union[str,Dict[str,str]]
-        return self.factory.executor(self.t, kwargs, os.getcwd(), None, **self.factory.execkwargs)
+        execkwargs = self.factory.execkwargs.copy()
+        execkwargs["basedir"] = os.getcwd()
+        return self.factory.executor(self.t, kwargs, **execkwargs)
 
 class Factory(object):
     def __init__(self, makeTool=workflow.defaultMakeTool,

cwltool/job.py

@@ -149,7 +149,7 @@ def run(self, dry_run=False, pull_image=True, rm_container=True,
         _logger.info(u"[job %s] %s$ %s%s%s",
                      self.name,
                      self.outdir,
-                     " ".join([shellescape.quote(str(arg)) if shouldquote(str(arg)) else str(arg) for arg in (runtime + self.command_line)]),
+                     " \\\n    ".join([shellescape.quote(str(arg)) if shouldquote(str(arg)) else str(arg) for arg in (runtime + self.command_line)]),
                      u' < %s' % (self.stdin) if self.stdin else '',
                      u' > %s' % os.path.join(self.outdir, self.stdout) if self.stdout else '')
 

cwltool/main.py

@@ -148,7 +148,7 @@ def arg_parser():  # type: () -> argparse.ArgumentParser
     return parser
 
 
-def single_job_executor(t, job_order, input_basedir, args, **kwargs):
+def single_job_executor(t, job_order_object, **kwargs):
     # type: (Process, Dict[str,Any], str, argparse.Namespace,**Any) -> Union[str,Dict[str,str]]
     final_output = []
     final_status = []
@@ -161,15 +161,17 @@ def output_callback(out, processStatus):
             _logger.warn(u"Final process status is %s", processStatus)
         final_output.append(out)
 
+    if "basedir" not in kwargs:
+        raise WorkflowException("Must provide 'basedir' in kwargs")
+
     if kwargs.get("outdir"):
         pass
     elif kwargs.get("dry_run"):
         kwargs["outdir"] = "/tmp"
     else:
         kwargs["outdir"] = tempfile.mkdtemp()
 
-    jobiter = t.job(job_order,
-                    input_basedir,
+    jobiter = t.job(job_order_object,
                     output_callback,
                     **kwargs)
 
@@ -447,6 +449,31 @@ def main(argsl=None,
 
         args = parser.parse_args(argsl)
 
+        # If caller provided a custom parser, it may be not every option is
+        # set, so fill in no-op defaults to avoid crashing when dereferencing
+        # them in args.
+        for k,v in {'print_deps': False,
+                    'print_pre': False,
+                    'print_rdf': False,
+                    'print_dot': False,
+                    'relative_deps': False,
+                    'tmp_outdir_prefix': 'tmp',
+                    'tmpdir_prefix': 'tmp',
+                    'print_input_deps': False,
+                    'cachedir': None,
+                    'quiet': False,
+                    'debug': False,
+                    'version': False,
+                    'enable_dev': False,
+                    'strict': True,
+                    'rdf_serializer': None,
+                    'basedir': None,
+                    'tool_help': False,
+                    'workflow': None,
+                    'job_order': None}.iteritems():
+            if not hasattr(args, k):
+                setattr(args, k, v)
+
         if args.quiet:
             _logger.setLevel(logging.WARN)
         if args.debug:
@@ -513,7 +540,7 @@ def main(argsl=None,
             # Use user defined temp directory (if it exists)
             args.tmp_outdir_prefix = os.path.abspath(args.tmp_outdir_prefix)
             if not os.path.exists(args.tmp_outdir_prefix):
-                _logger.error("Intermediate output directory prefix doesn't exist, reverting to default")
+                _logger.error("Intermediate output directory prefix doesn't exist.")
                 return 1
 
         if args.tmpdir_prefix != 'tmp':
@@ -536,25 +563,14 @@ def main(argsl=None,
             args.move_outputs = False
 
         try:
+            args.tmp_outdir_prefix = args.cachedir if args.cachedir else args.tmp_outdir_prefix
+            args.basedir = job_order_object[1]
+            del args.workflow
+            del args.job_order
             out = executor(tool, job_order_object[0],
-                           job_order_object[1], args,
-                           conformance_test=args.conformance_test,
-                           dry_run=args.dry_run,
-                           outdir=args.outdir,
-                           tmp_outdir_prefix=args.cachedir if args.cachedir else args.tmp_outdir_prefix,
-                           use_container=args.use_container,
-                           preserve_environment=args.preserve_environment,
-                           pull_image=args.enable_pull,
-                           rm_container=args.rm_container,
-                           tmpdir_prefix=args.tmpdir_prefix,
-                           enable_net=args.enable_net,
-                           rm_tmpdir=args.rm_tmpdir,
                            makeTool=makeTool,
-                           move_outputs=args.move_outputs,
                            select_resources=selectResources,
-                           eval_timeout=args.eval_timeout,
-                           cachedir=args.cachedir
-                           )
+                           **vars(args))
             # This is the workflow output, it needs to be written
             if out is not None:
                 if isinstance(out, basestring):

cwltool/process.py

@@ -297,7 +297,7 @@ def __init__(self, toolpath_object, **kwargs):
             raise validate.ValidationException(u"Got error `%s` while prcoessing outputs of %s:\n%s" % (str(e), self.tool["id"], json.dumps(self.outputs_record_schema, indent=4)))
 
 
-    def _init_job(self, joborder, input_basedir, **kwargs):
+    def _init_job(self, joborder, **kwargs):
         # type: (Dict[str, str], str, **Any) -> Builder
         builder = Builder()
         builder.job = copy.deepcopy(joborder)
@@ -326,7 +326,7 @@ def _init_job(self, joborder, input_basedir, **kwargs):
             builder.outdir = kwargs.get("outdir") or tempfile.mkdtemp()
             builder.tmpdir = kwargs.get("tmpdir") or tempfile.mkdtemp()
 
-        builder.fs_access = kwargs.get("fs_access") or StdFsAccess(input_basedir)
+        builder.fs_access = kwargs.get("fs_access") or StdFsAccess(kwargs["basedir"])
 
         if self.formatgraph:
             for i in self.tool["inputs"]:
@@ -426,7 +426,7 @@ def visit(self, op):
         op(self.tool)
 
     @abc.abstractmethod
-    def job(self, job_order, input_basedir, output_callbacks, **kwargs):
+    def job(self, job_order, output_callbacks, **kwargs):
         # type: (Dict[str, str], str, Callable[[Any, Any], Any], **Any) -> Generator[Any, None, None]
         return None
 

cwltool/workflow.py

@@ -145,11 +145,11 @@ def __init__(self, step):  # type: (Any) -> None
         self.iterable = None  # type: Iterable
         self.name = uniquename(u"step %s" % shortname(self.id))
 
-    def job(self, joborder, basedir, output_callback, **kwargs):
+    def job(self, joborder, output_callback, **kwargs):
         # type: (Dict[str,str], str, functools.partial[None], **Any) -> Generator
         kwargs["part_of"] = self.name
         kwargs["name"] = shortname(self.id)
-        for j in self.step.job(joborder, basedir, output_callback, **kwargs):
+        for j in self.step.job(joborder, output_callback, **kwargs):
             yield j
 
 
@@ -197,7 +197,7 @@ def receive_output(self, step, outputparms, jobout, processStatus):
 
         step.completed = True
 
-    def try_make_job(self, step, basedir, **kwargs):
+    def try_make_job(self, step, **kwargs):
         # type: (WorkflowJobStep, str, **Any) -> Generator
         inputparms = step.tool["inputs"]
         outputparms = step.tool["outputs"]
@@ -241,18 +241,18 @@ def valueFromFunc(k, v):  # type: (Any, Any) -> Any
                             for k,v in inputobj.items()}
 
                 if method == "dotproduct" or method is None:
-                    jobs = dotproduct_scatter(step, inputobj, basedir, scatter,
+                    jobs = dotproduct_scatter(step, inputobj, scatter,
                                               cast(  # known bug with mypy
                                                   # https://github.com/python/mypy/issues/797
                                                   Callable[[Any], Any],callback), **kwargs)
                 elif method == "nested_crossproduct":
-                    jobs = nested_crossproduct_scatter(step, inputobj, basedir,
+                    jobs = nested_crossproduct_scatter(step, inputobj,
                             scatter, cast(Callable[[Any], Any], callback),
                             # known bug in mypy
                             # https://github.com/python/mypy/issues/797
                             **kwargs)
                 elif method == "flat_crossproduct":
-                    jobs = flat_crossproduct_scatter(step, inputobj, basedir,
+                    jobs = flat_crossproduct_scatter(step, inputobj,
                                                      scatter,
                                                      cast(Callable[[Any], Any],
                                                          # known bug in mypy
@@ -262,7 +262,7 @@ def valueFromFunc(k, v):  # type: (Any, Any) -> Any
                 _logger.debug(u"[job %s] job input %s", step.name, json.dumps(inputobj, indent=4))
                 inputobj = {k: valueFromFunc(k, v) for k,v in inputobj.items()}
                 _logger.debug(u"[job %s] evaluated job input to %s", step.name, json.dumps(inputobj, indent=4))
-                jobs = step.job(inputobj, basedir, callback, **kwargs)
+                jobs = step.job(inputobj, callback, **kwargs)
 
             step.submitted = True
 
@@ -278,7 +278,7 @@ def valueFromFunc(k, v):  # type: (Any, Any) -> Any
     def run(self, **kwargs):
         _logger.debug(u"[%s] workflow starting", self.name)
 
-    def job(self, joborder, basedir, output_callback, move_outputs=True, **kwargs):
+    def job(self, joborder, output_callback, move_outputs=True, **kwargs):
         # type: (Dict[str,str], str, Callable[[Any, Any], Any], bool, **Any) -> Generator[WorkflowJob, None, None]
         self.state = {}
         self.processStatus = "success"
@@ -307,7 +307,7 @@ def job(self, joborder, basedir, output_callback, move_outputs=True, **kwargs):
 
             for step in self.steps:
                 if not step.submitted:
-                    step.iterable = self.try_make_job(step, basedir, **kwargs)
+                    step.iterable = self.try_make_job(step, **kwargs)
 
                 if step.iterable:
                     for newjob in step.iterable:
@@ -387,15 +387,15 @@ def __init__(self, toolpath_object, **kwargs):
 
         # TODO: statically validate data links instead of doing it at runtime.
 
-    def job(self, joborder, basedir, output_callback, **kwargs):
+    def job(self, joborder, output_callback, **kwargs):
         # type: (Dict[str,str], str, Callable[[Any, Any], Any], **Any) -> Generator[WorkflowJob, None, None]
-        builder = self._init_job(joborder, basedir, **kwargs)
+        builder = self._init_job(joborder, **kwargs)
         wj = WorkflowJob(self, **kwargs)
         yield wj
 
         kwargs["part_of"] = u"workflow %s" % wj.name
 
-        for w in wj.job(builder.job, basedir, output_callback, **kwargs):
+        for w in wj.job(builder.job, output_callback, **kwargs):
             yield w
 
     def visit(self, op):
@@ -497,7 +497,7 @@ def receive_output(self, output_callback, jobout, processStatus):
                 processStatus = "permanentFail"
         output_callback(output, processStatus)
 
-    def job(self, joborder, basedir, output_callback, **kwargs):
+    def job(self, joborder, output_callback, **kwargs):
         # type: (Dict[str, Any], str, Callable[...,Any], **Any) -> Generator
         for i in self.tool["inputs"]:
             p = i["id"]
@@ -509,7 +509,7 @@ def job(self, joborder, basedir, output_callback, **kwargs):
         kwargs["hints"] = kwargs.get("hints", []) + self.tool.get("hints", [])
 
         try:
-            for t in self.embedded_tool.job(joborder, basedir,
+            for t in self.embedded_tool.job(joborder,
                                             functools.partial(self.receive_output, output_callback),
                                             **kwargs):
                 yield t
@@ -554,7 +554,7 @@ def setTotal(self, total):  # type: (int) -> None
             self.output_callback(self.dest, self.processStatus)
 
 
-def dotproduct_scatter(process, joborder, basedir, scatter_keys, output_callback, **kwargs):
+def dotproduct_scatter(process, joborder, scatter_keys, output_callback, **kwargs):
     # type: (WorkflowJobStep, Dict[str, Any], str, List[str], Callable[..., Any], **Any) -> Generator[WorkflowJob, None, None]
     l = None
     for s in scatter_keys:
@@ -574,13 +574,13 @@ def dotproduct_scatter(process, joborder, basedir, scatter_keys, output_callback
         for s in scatter_keys:
             jo[s] = kwargs["valueFrom"](s, joborder[s][n])
 
-        for j in process.job(jo, basedir, functools.partial(rc.receive_scatter_output, n), **kwargs):
+        for j in process.job(jo, functools.partial(rc.receive_scatter_output, n), **kwargs):
             yield j
 
     rc.setTotal(l)
 
 
-def nested_crossproduct_scatter(process, joborder, basedir, scatter_keys, output_callback, **kwargs):
+def nested_crossproduct_scatter(process, joborder, scatter_keys, output_callback, **kwargs):
     # type: (WorkflowJobStep, Dict[str, Any], str, List[str], Callable[..., Any], **Any) -> Generator[WorkflowJob, None, None]
     scatter_key = scatter_keys[0]
     l = len(joborder[scatter_key])
@@ -595,10 +595,10 @@ def nested_crossproduct_scatter(process, joborder, basedir, scatter_keys, output
         jo[scatter_key] = kwargs["valueFrom"](scatter_key, joborder[scatter_key][n])
 
         if len(scatter_keys) == 1:
-            for j in process.job(jo, basedir, functools.partial(rc.receive_scatter_output, n), **kwargs):
+            for j in process.job(jo, functools.partial(rc.receive_scatter_output, n), **kwargs):
                 yield j
         else:
-            for j in nested_crossproduct_scatter(process, jo, basedir,
+            for j in nested_crossproduct_scatter(process, jo,
                     scatter_keys[1:], cast(  # known bug with mypy
                         # https://github.com/python/mypy/issues/797
                         Callable[[Any], Any],
@@ -622,7 +622,7 @@ def crossproduct_size(joborder, scatter_keys):
             sum += crossproduct_size(joborder, scatter_keys[1:])
     return sum
 
-def flat_crossproduct_scatter(process, joborder, basedir, scatter_keys, output_callback, startindex, **kwargs):
+def flat_crossproduct_scatter(process, joborder, scatter_keys, output_callback, startindex, **kwargs):
     # type: (WorkflowJobStep, Dict[str, Any], str, List[str], Union[ReceiveScatterOutput,Callable[..., Any]], int, **Any) -> Generator[WorkflowJob, None, None]
     scatter_key = scatter_keys[0]
     l = len(joborder[scatter_key])
@@ -644,11 +644,11 @@ def flat_crossproduct_scatter(process, joborder, basedir, scatter_keys, output_c
         jo[scatter_key] = kwargs["valueFrom"](scatter_key, joborder[scatter_key][n])
 
         if len(scatter_keys) == 1:
-            for j in process.job(jo, basedir, functools.partial(rc.receive_scatter_output, put), **kwargs):
+            for j in process.job(jo, functools.partial(rc.receive_scatter_output, put), **kwargs):
                 yield j
             put += 1
         else:
-            for j in flat_crossproduct_scatter(process, jo, basedir, scatter_keys[1:], rc, put, **kwargs):
+            for j in flat_crossproduct_scatter(process, jo, scatter_keys[1:], rc, put, **kwargs):
                 if j:
                     put += 1
                 yield j

setup.cfg

@@ -1,5 +1,5 @@
 [flake8]
-ignore=E124,E128,E129,E201,E202,E225,E226,E231,E265,E271,E302,E303,F401,E402,E501,W503,E731,F811,F821,F841
+ignore = E124,E128,E129,E201,E202,E225,E226,E231,E265,E271,E302,E303,F401,E402,E501,W503,E731,F811,F821,F841
+
+[easy_install]
 
-#[bdist_wheel]
-#universal = 1