Source code for

"""Module defines a planemo abstraction around running cwltool.

cwltool is an executable Python script and library mostly maintained by
Peter Amstutz and serves the reference implementation for the CWL.
It can be found at,
import json
import tempfile

from galaxy.tool_util.cwl.cwltool_deps import (

from planemo.deps import ensure_dependency_resolvers_conf_configured
from import (
from planemo.runnable import (

JSON_PARSE_ERROR_MESSAGE = "Failed to parse JSON from cwltool output [%s] " "in file [%s]. cwltool logs [%s]."

class CwlToolRunResponse(SuccessfulRunResponse):
    """Describe the resut of a cwltool invocation."""

    def __init__(self, runnable, log, outputs=None):
        self._runnable = runnable
        self._log = log
        self._outputs = outputs

    def log(self):
        return self._log

    def job_info(self):
        return None

    def invocation_details(self):
        return None

    def outputs_dict(self):
        return self._outputs

[docs]def run_cwltool(ctx, runnable, job_path, **kwds): """Translate planemo kwds to cwltool kwds and run cwltool main function.""" ensure_cwltool_available() args = [] if ctx.verbose: args.append("--verbose") output_directory = kwds.get("output_directory", None) if output_directory: args.append("--outdir") args.append(output_directory) if kwds.get("no_container", False): args.append("--no-container") ensure_dependency_resolvers_conf_configured(ctx, kwds) args.append("--beta-dependency-resolvers-configuration") args.append(kwds["dependency_resolvers_config_file"]) if kwds.get("mulled_containers"): args.append("--beta-use-biocontainers") if kwds.get("non_strict_cwl", False): args.append("--non-strict") args.extend([runnable.path, job_path]) ctx.vlog("Calling cwltool with arguments %s" % args) with tempfile.NamedTemporaryFile("w") as tmp_stdout, tempfile.NamedTemporaryFile("w") as tmp_stderr: # cwltool passes sys.stderr to subprocess.Popen - ensure it has # and actual fileno. with real_io(): ret_code = main.main( args, stdout=tmp_stdout, stderr=tmp_stderr, ) tmp_stdout.flush() tmp_stderr.flush() with open( as stderr_f: log = ctx.vlog("cwltool log output [%s]" % log) with open( as stdout_f: try: result = json.load(stdout_f) except ValueError: message = JSON_PARSE_ERROR_MESSAGE % (open(,, log) error(message) raise Exception(message) if ret_code != 0: return ErrorRunResponse("Error running cwltool", log=log) outputs = result return CwlToolRunResponse( runnable, log, outputs=outputs, )
__all__ = ("run_cwltool",)