X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/f4aa4dbbefe8b6dd65e3a112642da288774cf951..cb97316068f201ffd03f54d67074a83601c2bb45:/sdk/cwl/arvados_cwl/arvworkflow.py diff --git a/sdk/cwl/arvados_cwl/arvworkflow.py b/sdk/cwl/arvados_cwl/arvworkflow.py index 6f731fd687..5939b0413e 100644 --- a/sdk/cwl/arvados_cwl/arvworkflow.py +++ b/sdk/cwl/arvados_cwl/arvworkflow.py @@ -15,6 +15,7 @@ from cwltool.process import shortname from cwltool.workflow import Workflow, WorkflowException from cwltool.pathmapper import adjustFileObjs, adjustDirObjs, visit_class from cwltool.builder import Builder +from cwltool.context import LoadingContext import ruamel.yaml as yaml @@ -109,17 +110,16 @@ def get_overall_res_req(res_reqs): class ArvadosWorkflow(Workflow): """Wrap cwltool Workflow to override selected methods.""" - def __init__(self, arvrunner, toolpath_object, **kwargs): - super(ArvadosWorkflow, self).__init__(toolpath_object, **kwargs) + def __init__(self, arvrunner, toolpath_object, loadingContext): + super(ArvadosWorkflow, self).__init__(toolpath_object, loadingContext) self.arvrunner = arvrunner - self.work_api = kwargs["work_api"] self.wf_pdh = None self.dynamic_resource_req = [] self.static_resource_req = [] self.wf_reffiles = [] + self.loadingContext = loadingContext - def job(self, joborder, output_callback, **kwargs): - kwargs["work_api"] = self.work_api + def job(self, joborder, output_callback, runtimeContext): req, _ = self.get_requirement("http://arvados.org/cwl#RunInSingleContainer") if req: with SourceLine(self.tool, None, WorkflowException, logger.isEnabledFor(logging.DEBUG)): @@ -143,11 +143,10 @@ class ArvadosWorkflow(Workflow): packed = pack(document_loader, workflowobj, uri, self.metadata) - builder = Builder() - builder.job = joborder - builder.requirements = workflowobj["requirements"] - builder.hints = workflowobj["hints"] - builder.resources = {} + builder = Builder(joborder, + requirements=workflowobj["requirements"], + hints=workflowobj["hints"], + resources={}) def visit(item): for t in ("hints", "requirements"): @@ -176,7 +175,7 @@ class ArvadosWorkflow(Workflow): self.static_resource_req = [get_overall_res_req(self.static_resource_req)] upload_dependencies(self.arvrunner, - kwargs.get("name", ""), + runtimeContext.name, document_loader, packed, uri, @@ -213,15 +212,16 @@ class ArvadosWorkflow(Workflow): reffiles = [] visit_class(joborder_keepmount, ("File", "Directory"), reffiles.append) - mapper = ArvPathMapper(self.arvrunner, reffiles+self.wf_reffiles, kwargs["basedir"], - "/keep/%s", - "/keep/%s/%s", - **kwargs) + mapper = ArvPathMapper(self.arvrunner, reffiles+self.wf_reffiles, runtimeContext.basedir, + "/keep/%s", + "/keep/%s/%s") # For containers API, we need to make sure any extra # referenced files (ie referenced by the workflow but # not in the inputs) are included in the mounts. - kwargs["extra_reffiles"] = copy.deepcopy(self.wf_reffiles) + if self.wf_reffiles: + runtimeContext = runtimeContext.copy() + runtimeContext.extra_reffiles = copy.deepcopy(self.wf_reffiles) def keepmount(obj): remove_redundant_fields(obj) @@ -258,14 +258,12 @@ class ArvadosWorkflow(Workflow): "outputs": self.tool["outputs"], "stdout": "cwl.output.json", "requirements": self.requirements+job_res_reqs+[ + {"class": "InlineJavascriptRequirement"}, { "class": "InitialWorkDirRequirement", "listing": [{ "entryname": "workflow.cwl", - "entry": { - "class": "File", - "location": "keep:%s/workflow.cwl" % self.wf_pdh - } + "entry": '$({"class": "File", "location": "keep:%s/workflow.cwl"})' % self.wf_pdh }, { "entryname": "cwl.input.yml", "entry": json.dumps(joborder_keepmount, indent=2, sort_keys=True, separators=(',',': ')).replace("\\", "\\\\").replace('$(', '\$(').replace('${', '\${') @@ -275,8 +273,6 @@ class ArvadosWorkflow(Workflow): "arguments": ["--no-container", "--move-outputs", "--preserve-entire-environment", "workflow.cwl#main", "cwl.input.yml"], "id": "#" }) - kwargs["loader"] = self.doc_loader - kwargs["avsc_names"] = self.doc_schema - return ArvadosCommandTool(self.arvrunner, wf_runner, **kwargs).job(joborder_resolved, output_callback, **kwargs) + return ArvadosCommandTool(self.arvrunner, wf_runner, self.loadingContext).job(joborder_resolved, output_callback, runtimeContext) else: - return super(ArvadosWorkflow, self).job(joborder, output_callback, **kwargs) + return super(ArvadosWorkflow, self).job(joborder, output_callback, runtimeContext)