diff --git a/src/cwl_utils/cwl_v1_0_expression_refactor.py b/src/cwl_utils/cwl_v1_0_expression_refactor.py index d584224e..e02bf52c 100755 --- a/src/cwl_utils/cwl_v1_0_expression_refactor.py +++ b/src/cwl_utils/cwl_v1_0_expression_refactor.py @@ -7,7 +7,6 @@ import hashlib import uuid from collections.abc import MutableSequence, Sequence -from contextlib import suppress from typing import Any, cast, Final from ruamel import yaml @@ -24,7 +23,6 @@ from cwl_utils.expression import do_eval, interpolate from cwl_utils.parser.utils import param_for_source_id from cwl_utils.types import ( - CWLDirectoryType, CWLFileType, CWLObjectType, CWLOutputType, @@ -32,6 +30,7 @@ CWLRuntimeParameterContext, is_file_or_directory, ) +from cwl_utils.utils import get_step_uri _DEFAULT_CWL_VERSION: Final = "v1.0" @@ -235,8 +234,11 @@ def traverse( inside: bool, skip_command_line1: bool, skip_command_line2: bool, + context: dict[str, tuple[cwl_utils.parser.Process, bool]] | None = None, ) -> tuple[cwl.CommandLineTool | cwl.ExpressionTool | cwl.Workflow, bool]: """Convert the given process and any subprocesses.""" + if context is None: + context = {} match process: case cwl.CommandLineTool() if not inside: process = expand_stream_shortcuts(process) @@ -300,14 +302,18 @@ def traverse( cwlVersion=process.cwlVersion, ) result, modified = traverse_workflow( - workflow, replace_etool, skip_command_line1, skip_command_line2 + workflow, replace_etool, skip_command_line1, skip_command_line2, context ) if modified: return result, True else: return process, False case cwl.ExpressionTool() if replace_etool: - expression = get_expression(process.expression, empty_inputs(process), None) + expression = get_expression( + process.expression, + cwl_utils.expression_refactor.empty_inputs(process), + None, + ) # Why call get_expression on an ExpressionTool? # It normalizes the form of $() CWL expressions into the ${} style if expression: @@ -318,7 +324,7 @@ def traverse( return etool_to_cltool(process2), True case cwl.Workflow(): return traverse_workflow( - process, replace_etool, skip_command_line1, skip_command_line2 + process, replace_etool, skip_command_line1, skip_command_line2, context ) case _: return process, False @@ -489,76 +495,6 @@ def replace_wf_input_ref_with_step_output( outp.outputSource[index] = target -def empty_inputs( - process_or_step: ( - cwl.CommandLineTool | cwl.WorkflowStep | cwl.ExpressionTool | cwl.Workflow - ), - parent: cwl.Workflow | None = None, -) -> dict[str, Any]: - """Produce a mock input object for the given inputs.""" - result = {} - if isinstance(process_or_step, cwl.Process): - for param in process_or_step.inputs: - result[param.id.split("#")[-1]] = example_input(param.type_) - else: - for param in process_or_step.in_: - param_id = param.id.split("/")[-1] - if param.source is None and param.valueFrom: - result[param_id] = example_input("string") - elif param.source is None and param.default: - result[param_id] = param.default - else: - with suppress(WorkflowException): - result[param_id] = example_input( - utils.type_for_source(process_or_step.run, param.source, parent) - ) - return result - - -def example_input(some_type: Any) -> Any: - """Produce a fake input for the given type.""" - # TODO: accept some sort of context object with local custom type definitions - if some_type == "Directory": - return CWLDirectoryType( - **{ - "class": "Directory", - "location": "https://www.example.com/example", - "basename": "example", - "listing": [ - CWLFileType( - **{ - "class": "File", - "basename": "example.txt", - "size": 23, - "contents": "hoopla", - "nameroot": "example", - "nameext": "txt", - } - ) - ], - } - ) - if some_type == "File": - return CWLFileType( - **{ - "class": "File", - "location": "https://www.example.com/example.txt", - "basename": "example.txt", - "size": 23, - "contents": "hoopla", - "nameroot": "example", - "nameext": "txt", - } - ) - if some_type == "int": - return 23 - if some_type == "string": - return "hoopla!" - if some_type == "boolean": - return True - return None - - EMPTY_FILE = CWLFileType( **{ "class": "File", @@ -613,7 +549,7 @@ def process_workflow_inputs_and_outputs( ) -> bool: """Do any needed conversions on the given Workflow's inputs and outputs.""" modified = False - inputs = empty_inputs(workflow) + inputs = cwl_utils.expression_refactor.empty_inputs(workflow) for index, param in enumerate(workflow.inputs): with SourceLine(workflow.inputs, index, WorkflowException): if param.format and get_expression(param.format, inputs, None): @@ -659,7 +595,7 @@ def process_workflow_reqs_and_hints( # ^ By refactoring replace_expr_etool to allow multiple inputs, # and connecting all workflow inputs to the generated step modified = False - inputs = empty_inputs(workflow) + inputs = cwl_utils.expression_refactor.empty_inputs(workflow) generated_res_reqs: list[tuple[str, int | str]] = [] generated_iwdr_reqs: list[tuple[str, int | str]] = [] generated_envVar_reqs: list[tuple[str, int | str]] = [] @@ -966,6 +902,7 @@ def process_level_reqs( replace_etool: bool, skip_command_line1: bool, skip_command_line2: bool, + context: dict[str, tuple[cwl_utils.parser.Process, bool]], ) -> bool: """Convert expressions inside a process into new adjacent steps.""" # This is for reqs inside a Process (CommandLineTool, ExpressionTool) @@ -981,7 +918,8 @@ def process_level_reqs( return False modified = False target_process = step.run - inputs = cwl_utils.expression_refactor.empty_inputs(process, _DEFAULT_CWL_VERSION) + inputs = cwl_utils.expression_refactor.empty_inputs(process) + generated_res_reqs: list[tuple[str, str]] = [] generated_iwdr_reqs: list[tuple[str, int | str, Any]] = [] generated_envVar_reqs: list[tuple[str, int | str]] = [] @@ -1030,6 +968,7 @@ def process_level_reqs( target, step, replace_etool, + context, ) setattr( target_process.requirements[req_index], @@ -1157,6 +1096,7 @@ def process_level_reqs( target, step, replace_etool, + context, ) target_process.requirements[req_index].listing[ listing_index @@ -1233,15 +1173,15 @@ def traverse_CommandLineTool( clt: cwl_utils.parser.CommandLineTool, parent: cwl.Workflow, step: cwl.WorkflowStep, + target_clt: cwl_utils.parser.CommandLineTool, replace_etool: bool, skip_command_line1: bool, skip_command_line2: bool, + context: dict[str, tuple[cwl_utils.parser.Process, bool]], ) -> bool: """Extract any CWL Expressions within the given CommandLineTool into sibling steps.""" modified = False - # don't modify clt, modify step.run - target_clt = step.run - inputs = cwl_utils.expression_refactor.empty_inputs(clt, _DEFAULT_CWL_VERSION) + inputs = cwl_utils.expression_refactor.empty_inputs(clt) if not step.id: return False step_id = step.id.split("#")[-1] @@ -1256,9 +1196,15 @@ def traverse_CommandLineTool( target_type = "Any" target = cwl.InputParameter(id=None, type_=target_type) replace_step_clt_expr_with_etool( - expression, etool_id, parent, target, step, replace_etool + expression, + etool_id, + parent, + target, + step, + replace_etool, + context, ) - target_clt.arguments[index] = ( + cast(list[Any], target_clt.arguments)[index] = ( cwl_utils.expression_refactor.get_command_line_binding( target_clt.cwlVersion or _DEFAULT_CWL_VERSION, valueFrom=f"$(inputs.{inp_id})", @@ -1288,10 +1234,16 @@ def traverse_CommandLineTool( target_type = "Any" target = cwl.InputParameter(id=None, type_=target_type) replace_step_clt_expr_with_etool( - expression, etool_id, parent, target, step, replace_etool + expression, + etool_id, + parent, + target, + step, + replace_etool, + context, ) - target_clt.arguments[index].valueFrom = "$(inputs.{})".format( - inp_id + cast(list[Any], target_clt.arguments)[index].valueFrom = ( + "$(inputs.{})".format(inp_id) ) target_clt.inputs.append( cwl_utils.expression_refactor.get_command_input_parameter( @@ -1317,7 +1269,7 @@ def traverse_CommandLineTool( target_type = "string" target = cwl.InputParameter(id=None, type_=target_type) replace_step_clt_expr_with_etool( - expression, etool_id, parent, target, step, replace_etool + expression, etool_id, parent, target, step, replace_etool, context ) setattr(target_clt, streamtype, f"$(inputs.{inp_id})") target_clt.inputs.append( @@ -1333,7 +1285,9 @@ def traverse_CommandLineTool( for inp in clt.inputs: if not skip_command_line1 and inp.inputBinding and inp.inputBinding.valueFrom: expression = get_expression( - inp.inputBinding.valueFrom, inputs, example_input(inp.type_) + inp.inputBinding.valueFrom, + inputs, + cwl_utils.expression_refactor.example_input(inp.type_), ) if expression: modified = True @@ -1363,7 +1317,13 @@ def traverse_CommandLineTool( glob_target_type = ["string", ArraySchema("string", "array")] target = cwl.InputParameter(id=None, type_=glob_target_type) replace_step_clt_expr_with_etool( - expression, etool_id, parent, target, step, replace_etool + expression, + etool_id, + parent, + target, + step, + replace_etool, + context, ) outp.outputBinding.glob = f"$(inputs.{inp_id})" target_clt.inputs.append( @@ -1396,7 +1356,7 @@ def traverse_CommandLineTool( inp_id = f"_{outp_id}_outputEval" etool_id = f"expression{inp_id}" sub_wf_outputs = cltool_step_outputs_to_workflow_outputs( - step, etool_id, outp_id + step, target_clt, etool_id, outp_id ) self_type = cwl.InputParameter( id=None, @@ -1450,24 +1410,29 @@ def traverse_CommandLineTool( ) else: final_etool = etool + if isinstance(final_etool, cwl.Process): + step_run: cwl.Process | str = etool + else: + context[f"{etool_id}.cwl"] = (final_etool, True) + step_run = f"{etool_id}.cwl" etool_step = cwl.WorkflowStep( id=etool_id, in_=orig_step_inputs, out=[cwl.WorkflowStepOutput("result")], - run=final_etool, + run=step_run, scatterMethod=step.scatterMethod, ) new_clt_step = copy.copy( step ) # a deepcopy would be convenient, but params2.cwl gives it problems new_clt_step.id = new_clt_step.id.split("#")[-1] - new_clt_step.run = copy.copy(step.run) - new_clt_step.run.id = None + new_clt = copy.copy(target_clt) + new_clt.id = "" cwl_utils.expression_refactor.remove_JSReq( - new_clt_step.run, skip_command_line1 + new_clt, skip_command_line1 ) cwl_utils.expression_refactor.process_CommandLineTool_output( - new_clt_step.run, _DEFAULT_CWL_VERSION, outp_id + new_clt, _DEFAULT_CWL_VERSION, outp_id ) new_clt_step.in_ = copy.deepcopy(step.in_) for inp in new_clt_step.in_: @@ -1476,10 +1441,14 @@ def traverse_CommandLineTool( inp.linkMerge = None for index, out in enumerate(new_clt_step.out): new_clt_step.out[index] = out.split("/")[-1] - for tool_inp in new_clt_step.run.inputs: + for tool_inp in new_clt.inputs: tool_inp.id = tool_inp.id.split("#")[-1] - for tool_out in new_clt_step.run.outputs: + for tool_out in new_clt.outputs: tool_out.id = tool_out.id.split("#")[-1] + if isinstance(new_clt_step.run, str): + context[get_step_uri(new_clt_step)] = (new_clt, True) + else: + new_clt_step.run = new_clt sub_wf_steps = [new_clt_step, etool_step] sub_workflow = cwl.Workflow( inputs=sub_wf_inputs, @@ -1549,6 +1518,7 @@ def replace_step_clt_expr_with_etool( target: cwl.InputParameter, step: cwl.WorkflowStep, replace_etool: bool, + context: dict[str, tuple[cwl_utils.parser.Process, bool]], self_name: str | None = None, ) -> None: """Convert a step level CWL Expression to a sibling expression step.""" @@ -1578,12 +1548,17 @@ def replace_step_clt_expr_with_etool( for wf_step_input in wf_step_inputs: wf_step_input.id = wf_step_input.id.split("/")[-1] wf_step_inputs[:] = [x for x in wf_step_inputs if not x.id.startswith("_")] + if isinstance(etool, cwl.Process): + step_run: cwl.Process | str = etool + else: + context[f"{name}.cwl"] = (etool, True) + step_run = f"{name}.cwl" workflow.steps.append( cwl.WorkflowStep( id=name, in_=wf_step_inputs, out=[cwl.WorkflowStepOutput("result")], - run=etool, + run=step_run, ) ) @@ -1595,6 +1570,7 @@ def replace_clt_hintreq_expr_with_etool( target: cwl.InputParameter, step: cwl.WorkflowStep, replace_etool: bool, + context: dict[str, tuple[cwl_utils.parser.Process, bool]], self_name: str | None = None, ) -> None: """Factor out an expression inside a CommandLineTool req or hint into a sibling step.""" @@ -1625,12 +1601,17 @@ def replace_clt_hintreq_expr_with_etool( for wf_step_input in wf_step_inputs: wf_step_input.id = wf_step_input.id.split("/")[-1] wf_step_inputs[:] = [x for x in wf_step_inputs if not x.id.startswith("_")] + if isinstance(etool, cwl.Process): + step_run: cwl.Process | str = etool + else: + context[f"{name}.cwl"] = (etool, True) + step_run = f"{name}.cwl" workflow.steps.append( cwl.WorkflowStep( id=name, in_=wf_step_inputs, out=[cwl.WorkflowStepOutput("result")], - run=etool, + run=step_run, ) ) @@ -1662,7 +1643,10 @@ def cltool_inputs_to_etool_inputs( def cltool_step_outputs_to_workflow_outputs( - cltool_step: cwl.WorkflowStep, etool_step_id: str, etool_out_id: str + cltool_step: cwl.WorkflowStep, + clt: cwl_utils.parser.CommandLineTool, + etool_step_id: str, + etool_out_id: str, ) -> list[cwl.OutputParameter]: """ Copy CommandLineTool outputs into the equivalent Workflow output parameters. @@ -1674,8 +1658,8 @@ def cltool_step_outputs_to_workflow_outputs( if not cltool_step.id: raise WorkflowException(f"Missing step id from {cltool_step}.") default_step_id = cltool_step.id.split("#")[-1] - if cltool_step.run.outputs: - for clt_out in cltool_step.run.outputs: + if clt.outputs: + for clt_out in clt.outputs: clt_out_id = clt_out.id.split("#")[-1].split("/")[-1] if clt_out_id == etool_out_id: outputSource = f"{etool_step_id}/result" @@ -1745,17 +1729,19 @@ def generate_etool_from_expr2( def traverse_step( step: cwl.WorkflowStep, parent: cwl.Workflow, + process: cwl_utils.parser.Process, replace_etool: bool, skip_command_line1: bool, skip_command_line2: bool, + context: dict[str, tuple[cwl_utils.parser.Process, bool]], ) -> bool: """Process the given WorkflowStep.""" modified = False - inputs = empty_inputs(step, parent) + inputs = cwl_utils.expression_refactor.empty_inputs(step, context, parent) if not step.id: return False step_id = step.id.split("#")[-1] - original_process = copy.deepcopy(step.run) + original_process = copy.deepcopy(process) original_step_ins = copy.deepcopy(step.in_) for inp in step.in_: if inp.valueFrom: @@ -1767,7 +1753,7 @@ def traverse_step( for source in inp.source: if not step.scatter: self.append( - example_input( + cwl_utils.expression_refactor.example_input( utils.type_for_source(parent, source.split("#")[-1]) ) ) @@ -1777,12 +1763,20 @@ def traverse_step( ) if isinstance(scattered_source_type, list): for stype in scattered_source_type: - self.append(example_input(stype.type_)) + self.append( + cwl_utils.expression_refactor.example_input( + stype.type_ + ) + ) else: - self.append(example_input(scattered_source_type.type_)) + self.append( + cwl_utils.expression_refactor.example_input( + scattered_source_type.type_ + ) + ) else: if not step.scatter: - self = example_input( + self = cwl_utils.expression_refactor.example_input( utils.type_for_source(parent, inp.source.split("#")[-1]) ) else: @@ -1790,9 +1784,13 @@ def traverse_step( parent, inp.source ) if isinstance(scattered_source_type2, list): - self = example_input(scattered_source_type2[0].type_) + self = cwl_utils.expression_refactor.example_input( + scattered_source_type2[0].type_ + ) else: - self = example_input(scattered_source_type2.type_) + self = cwl_utils.expression_refactor.example_input( + scattered_source_type2.type_ + ) expression = get_expression(inp.valueFrom, inputs, self) if expression: modified = True @@ -1810,8 +1808,11 @@ def traverse_step( for source in inp.source: source_id = source.split("#")[-1] input_source_id.append(source_id) - temp_type = utils.type_for_source( - step.run, source_id, parent + temp_type = cwl_utils.parser.utils.type_for_source( + process, + process.cwlVersion or _DEFAULT_CWL_VERSION, + source_id, + parent, ) if isinstance(temp_type, list): for ttype in temp_type: @@ -1824,7 +1825,7 @@ def traverse_step( input_source_id = inp.source.split("#")[-1] # target.id = target.id.split('#')[-1] if isinstance(original_process, cwl_utils.parser.ExpressionTool): - reqs: list[cwl.ProcessRequirement] = [] + reqs: list[cwl_utils.parser.ProcessRequirement] = [] if original_process.hints: reqs.extend(original_process.hints) if original_process.requirements: @@ -1835,14 +1836,14 @@ def traverse_step( ): break else: - if not step.run.requirements: - step.run.requirements = [] + if not process.requirements: + process.requirements = [] expr_lib = cwl_utils.expression_refactor.find_expressionLib( [parent] ) - step.run.requirements.append( + process.requirements.append( cwl_utils.expression_refactor.get_inline_javascript_requirement( - step.run, _DEFAULT_CWL_VERSION, expr_lib + original_process, _DEFAULT_CWL_VERSION, expr_lib ) ) replace_step_valueFrom_expr_with_etool( @@ -1856,6 +1857,7 @@ def traverse_step( original_step_ins, input_source_id, replace_etool, + context, ) inp.valueFrom = None inp.source = f"{etool_id}/result" @@ -1867,6 +1869,7 @@ def traverse_step( replace_etool, skip_command_line1, skip_command_line2, + context, ) if process_modified: modified = True @@ -1875,9 +1878,11 @@ def traverse_step( original_process, parent, step, + cast(cwl_utils.parser.CommandLineTool, process), replace_etool, skip_command_line1, skip_command_line2, + context, ) if clt_modified: modified = True @@ -1934,6 +1939,7 @@ def replace_step_valueFrom_expr_with_etool( original_step_ins: list[cwl.WorkflowStepInput], source: str | list[str] | None, replace_etool: bool, + context: dict[str, tuple[cwl_utils.parser.Process, bool]], ) -> None: """Replace a WorkflowStep level 'valueFrom' expression with a sibling ExpressionTool step.""" if not step_inp.id: @@ -1999,12 +2005,17 @@ def replace_step_valueFrom_expr_with_etool( # do we still need to scatter? else: scatter = None + if isinstance(etool, cwl.Process): + step_run: cwl.Process | str = etool + else: + context[f"{name}.cwl"] = (etool, True) + step_run = f"{name}.cwl" workflow.steps.append( cwl.WorkflowStep( id=name, in_=wf_step_inputs, out=[cwl.WorkflowStepOutput("result")], - run=etool, + run=step_run, scatter=scatter, scatterMethod=step.scatterMethod, ) @@ -2016,6 +2027,7 @@ def traverse_workflow( replace_etool: bool, skip_command_line1: bool, skip_command_line2: bool, + context: dict[str, tuple[cwl_utils.parser.Process, bool]], ) -> tuple[cwl.Workflow, bool]: """Traverse a workflow, processing each step.""" modified = False @@ -2025,14 +2037,24 @@ def traverse_workflow( modified = True else: step_modified = cwl_utils.expression_refactor.load_step( - step, replace_etool, skip_command_line1, skip_command_line2 + step, replace_etool, skip_command_line1, skip_command_line2, context ) if step_modified: modified = True for step in workflow.steps: if not step.id.startswith("_expression"): step_modified = traverse_step( - step, workflow, replace_etool, skip_command_line1, skip_command_line2 + step, + workflow, + ( + context[get_step_uri(step)][0] + if isinstance(step.run, str) + else cast(cwl_utils.parser.Process, step.run) + ), + replace_etool, + skip_command_line1, + skip_command_line2, + context, ) if step_modified: modified = True diff --git a/src/cwl_utils/cwl_v1_1_expression_refactor.py b/src/cwl_utils/cwl_v1_1_expression_refactor.py index 685520e7..18e0675f 100755 --- a/src/cwl_utils/cwl_v1_1_expression_refactor.py +++ b/src/cwl_utils/cwl_v1_1_expression_refactor.py @@ -7,7 +7,6 @@ import hashlib import uuid from collections.abc import MutableSequence, Sequence -from contextlib import suppress from typing import Any, cast, Final from ruamel import yaml @@ -24,7 +23,6 @@ from cwl_utils.expression import do_eval, interpolate from cwl_utils.parser.utils import param_for_source_id from cwl_utils.types import ( - CWLDirectoryType, CWLFileType, CWLObjectType, CWLOutputType, @@ -32,6 +30,7 @@ CWLRuntimeParameterContext, is_file_or_directory, ) +from cwl_utils.utils import get_step_uri _DEFAULT_CWL_VERSION: Final = "v1.1" @@ -235,8 +234,11 @@ def traverse( inside: bool, skip_command_line1: bool, skip_command_line2: bool, + context: dict[str, tuple[cwl_utils.parser.Process, bool]] | None = None, ) -> tuple[cwl.CommandLineTool | cwl.ExpressionTool | cwl.Workflow, bool]: """Convert the given process and any subprocesses.""" + if context is None: + context = {} match process: case cwl.CommandLineTool() if not inside: process = expand_stream_shortcuts(process) @@ -300,14 +302,18 @@ def traverse( cwlVersion=process.cwlVersion, ) result, modified = traverse_workflow( - workflow, replace_etool, skip_command_line1, skip_command_line2 + workflow, replace_etool, skip_command_line1, skip_command_line2, context ) if modified: return result, True else: return process, False case cwl.ExpressionTool() if replace_etool: - expression = get_expression(process.expression, empty_inputs(process), None) + expression = get_expression( + process.expression, + cwl_utils.expression_refactor.empty_inputs(process), + None, + ) # Why call get_expression on an ExpressionTool? # It normalizes the form of $() CWL expressions into the ${} style if expression: @@ -318,7 +324,7 @@ def traverse( return etool_to_cltool(process2), True case cwl.Workflow(): return traverse_workflow( - process, replace_etool, skip_command_line1, skip_command_line2 + process, replace_etool, skip_command_line1, skip_command_line2, context ) case _: return process, False @@ -487,76 +493,6 @@ def replace_wf_input_ref_with_step_output( outp.outputSource[index] = target -def empty_inputs( - process_or_step: ( - cwl.CommandLineTool | cwl.WorkflowStep | cwl.ExpressionTool | cwl.Workflow - ), - parent: cwl.Workflow | None = None, -) -> dict[str, Any]: - """Produce a mock input object for the given inputs.""" - result = {} - if isinstance(process_or_step, cwl.Process): - for param in process_or_step.inputs: - result[param.id.split("#")[-1]] = example_input(param.type_) - else: - for param in process_or_step.in_: - param_id = param.id.split("/")[-1] - if param.source is None and param.valueFrom: - result[param_id] = example_input("string") - elif param.source is None and param.default: - result[param_id] = param.default - else: - with suppress(WorkflowException): - result[param_id] = example_input( - utils.type_for_source(process_or_step.run, param.source, parent) - ) - return result - - -def example_input(some_type: Any) -> Any: - """Produce a fake input for the given type.""" - # TODO: accept some sort of context object with local custom type definitions - if some_type == "Directory": - return CWLDirectoryType( - **{ - "class": "Directory", - "location": "https://www.example.com/example", - "basename": "example", - "listing": [ - CWLFileType( - **{ - "class": "File", - "basename": "example.txt", - "size": 23, - "contents": "hoopla", - "nameroot": "example", - "nameext": "txt", - } - ) - ], - } - ) - if some_type == "File": - return CWLFileType( - **{ - "class": "File", - "location": "https://www.example.com/example.txt", - "basename": "example.txt", - "size": 23, - "contents": "hoopla", - "nameroot": "example", - "nameext": "txt", - } - ) - if some_type == "int": - return 23 - if some_type == "string": - return "hoopla!" - if some_type == "boolean": - return True - return None - - EMPTY_FILE = CWLFileType( **{ "class": "File", @@ -611,7 +547,7 @@ def process_workflow_inputs_and_outputs( ) -> bool: """Do any needed conversions on the given Workflow's inputs and outputs.""" modified = False - inputs = empty_inputs(workflow) + inputs = cwl_utils.expression_refactor.empty_inputs(workflow) for index, param in enumerate(workflow.inputs): with SourceLine(workflow.inputs, index, WorkflowException): if param.format and get_expression(param.format, inputs, None): @@ -659,7 +595,7 @@ def process_workflow_reqs_and_hints( # ^ By refactoring replace_expr_etool to allow multiple inputs, # and connecting all workflow inputs to the generated step modified = False - inputs = empty_inputs(workflow) + inputs = cwl_utils.expression_refactor.empty_inputs(workflow) generated_res_reqs: list[tuple[str, int | str]] = [] generated_iwdr_reqs: list[tuple[str, int | str]] = [] generated_envVar_reqs: list[tuple[str, int | str]] = [] @@ -962,12 +898,13 @@ def process_workflow_reqs_and_hints( def process_level_reqs( - process: cwl.CommandLineTool, + process: cwl_utils.parser.Process, step: cwl.WorkflowStep, parent: cwl.Workflow, replace_etool: bool, skip_command_line1: bool, skip_command_line2: bool, + context: dict[str, tuple[cwl_utils.parser.Process, bool]], ) -> bool: """Convert expressions inside a process into new adjacent steps.""" # This is for reqs inside a Process (CommandLineTool, ExpressionTool) @@ -983,7 +920,7 @@ def process_level_reqs( return False modified = False target_process = step.run - inputs = cwl_utils.expression_refactor.empty_inputs(process, "v1.1") + inputs = cwl_utils.expression_refactor.empty_inputs(process) generated_res_reqs: list[tuple[str, str]] = [] generated_iwdr_reqs: list[tuple[str, int | str, Any]] = [] generated_envVar_reqs: list[tuple[str, int | str]] = [] @@ -1032,6 +969,7 @@ def process_level_reqs( target, step, replace_etool, + context, ) setattr( target_process.requirements[req_index], @@ -1159,6 +1097,7 @@ def process_level_reqs( target, step, replace_etool, + context, ) target_process.requirements[req_index].listing[ listing_index @@ -1235,15 +1174,15 @@ def traverse_CommandLineTool( clt: cwl_utils.parser.CommandLineTool, parent: cwl.Workflow, step: cwl.WorkflowStep, + target_clt: cwl_utils.parser.CommandLineTool, replace_etool: bool, skip_command_line1: bool, skip_command_line2: bool, + context: dict[str, tuple[cwl_utils.parser.Process, bool]], ) -> bool: """Extract any CWL Expressions within the given CommandLineTool into sibling steps.""" modified = False - # don't modify clt, modify step.run - target_clt = step.run - inputs = cwl_utils.expression_refactor.empty_inputs(clt, _DEFAULT_CWL_VERSION) + inputs = cwl_utils.expression_refactor.empty_inputs(clt) if not step.id: return False step_id = step.id.split("#")[-1] @@ -1258,9 +1197,15 @@ def traverse_CommandLineTool( target_type = "Any" target = cwl.WorkflowInputParameter(id=None, type_=target_type) replace_step_clt_expr_with_etool( - expression, etool_id, parent, target, step, replace_etool + expression, + etool_id, + parent, + target, + step, + replace_etool, + context, ) - target_clt.arguments[index] = ( + cast(list[Any], target_clt.arguments)[index] = ( cwl_utils.expression_refactor.get_command_line_binding( target_clt.cwlVersion or _DEFAULT_CWL_VERSION, valueFrom=f"$(inputs.{inp_id})", @@ -1290,10 +1235,16 @@ def traverse_CommandLineTool( target_type = "Any" target = cwl.WorkflowInputParameter(id=None, type_=target_type) replace_step_clt_expr_with_etool( - expression, etool_id, parent, target, step, replace_etool + expression, + etool_id, + parent, + target, + step, + replace_etool, + context, ) - target_clt.arguments[index].valueFrom = "$(inputs.{})".format( - inp_id + cast(list[Any], target_clt.arguments)[index].valueFrom = ( + "$(inputs.{})".format(inp_id) ) target_clt.inputs.append( cwl_utils.expression_refactor.get_command_input_parameter( @@ -1319,7 +1270,7 @@ def traverse_CommandLineTool( target_type = "string" target = cwl.WorkflowInputParameter(id=None, type_=target_type) replace_step_clt_expr_with_etool( - expression, etool_id, parent, target, step, replace_etool + expression, etool_id, parent, target, step, replace_etool, context ) setattr(target_clt, streamtype, f"$(inputs.{inp_id})") target_clt.inputs.append( @@ -1335,7 +1286,9 @@ def traverse_CommandLineTool( for inp in clt.inputs: if not skip_command_line1 and inp.inputBinding and inp.inputBinding.valueFrom: expression = get_expression( - inp.inputBinding.valueFrom, inputs, example_input(inp.type_) + inp.inputBinding.valueFrom, + inputs, + cwl_utils.expression_refactor.example_input(inp.type_), ) if expression: modified = True @@ -1365,7 +1318,13 @@ def traverse_CommandLineTool( glob_target_type = ["string", ArraySchema("string", "array")] target = cwl.WorkflowInputParameter(id=None, type_=glob_target_type) replace_step_clt_expr_with_etool( - expression, etool_id, parent, target, step, replace_etool + expression, + etool_id, + parent, + target, + step, + replace_etool, + context, ) outp.outputBinding.glob = f"$(inputs.{inp_id})" target_clt.inputs.append( @@ -1398,7 +1357,7 @@ def traverse_CommandLineTool( inp_id = f"_{outp_id}_outputEval" etool_id = f"expression{inp_id}" sub_wf_outputs = cltool_step_outputs_to_workflow_outputs( - step, etool_id, outp_id + step, target_clt, etool_id, outp_id ) self_type = cwl.WorkflowInputParameter( id=None, @@ -1452,24 +1411,29 @@ def traverse_CommandLineTool( ) else: final_etool = etool + if isinstance(final_etool, cwl.Process): + step_run: cwl.Process | str = etool + else: + context[f"{etool_id}.cwl"] = (final_etool, True) + step_run = f"{etool_id}.cwl" etool_step = cwl.WorkflowStep( id=etool_id, in_=orig_step_inputs, out=[cwl.WorkflowStepOutput("result")], - run=final_etool, + run=step_run, scatterMethod=step.scatterMethod, ) new_clt_step = copy.copy( step ) # a deepcopy would be convenient, but params2.cwl gives it problems new_clt_step.id = new_clt_step.id.split("#")[-1] - new_clt_step.run = copy.copy(step.run) - new_clt_step.run.id = None + new_clt = copy.copy(target_clt) + new_clt.id = "" cwl_utils.expression_refactor.remove_JSReq( - new_clt_step.run, skip_command_line1 + new_clt, skip_command_line1 ) cwl_utils.expression_refactor.process_CommandLineTool_output( - new_clt_step.run, _DEFAULT_CWL_VERSION, outp_id + new_clt, _DEFAULT_CWL_VERSION, outp_id ) new_clt_step.in_ = copy.deepcopy(step.in_) for inp in new_clt_step.in_: @@ -1478,10 +1442,14 @@ def traverse_CommandLineTool( inp.linkMerge = None for index, out in enumerate(new_clt_step.out): new_clt_step.out[index] = out.split("/")[-1] - for tool_inp in new_clt_step.run.inputs: + for tool_inp in new_clt.inputs: tool_inp.id = tool_inp.id.split("#")[-1] - for tool_out in new_clt_step.run.outputs: + for tool_out in new_clt.outputs: tool_out.id = tool_out.id.split("#")[-1] + if isinstance(new_clt_step.run, str): + context[get_step_uri(new_clt_step)] = (new_clt, True) + else: + new_clt_step.run = new_clt sub_wf_steps = [new_clt_step, etool_step] sub_workflow = cwl.Workflow( inputs=sub_wf_inputs, @@ -1551,6 +1519,7 @@ def replace_step_clt_expr_with_etool( target: cwl.WorkflowInputParameter, step: cwl.WorkflowStep, replace_etool: bool, + context: dict[str, tuple[cwl_utils.parser.Process, bool]], self_name: str | None = None, ) -> None: """Convert a step level CWL Expression to a sibling expression step.""" @@ -1580,12 +1549,17 @@ def replace_step_clt_expr_with_etool( for wf_step_input in wf_step_inputs: wf_step_input.id = wf_step_input.id.split("/")[-1] wf_step_inputs[:] = [x for x in wf_step_inputs if not x.id.startswith("_")] + if isinstance(etool, cwl.Process): + step_run: cwl.Process | str = etool + else: + context[f"{name}.cwl"] = (etool, True) + step_run = f"{name}.cwl" workflow.steps.append( cwl.WorkflowStep( id=name, in_=wf_step_inputs, out=[cwl.WorkflowStepOutput("result")], - run=etool, + run=step_run, ) ) @@ -1597,6 +1571,7 @@ def replace_clt_hintreq_expr_with_etool( target: cwl.WorkflowInputParameter, step: cwl.WorkflowStep, replace_etool: bool, + context: dict[str, tuple[cwl_utils.parser.Process, bool]], self_name: str | None = None, ) -> None: """Factor out an expression inside a CommandLineTool req or hint into a sibling step.""" @@ -1627,12 +1602,17 @@ def replace_clt_hintreq_expr_with_etool( for wf_step_input in wf_step_inputs: wf_step_input.id = wf_step_input.id.split("/")[-1] wf_step_inputs[:] = [x for x in wf_step_inputs if not x.id.startswith("_")] + if isinstance(etool, cwl.Process): + step_run: cwl.Process | str = etool + else: + context[f"{name}.cwl"] = (etool, True) + step_run = f"{name}.cwl" workflow.steps.append( cwl.WorkflowStep( id=name, in_=wf_step_inputs, out=[cwl.WorkflowStepOutput("result")], - run=etool, + run=step_run, ) ) @@ -1664,7 +1644,10 @@ def cltool_inputs_to_etool_inputs( def cltool_step_outputs_to_workflow_outputs( - cltool_step: cwl.WorkflowStep, etool_step_id: str, etool_out_id: str + cltool_step: cwl.WorkflowStep, + clt: cwl_utils.parser.CommandLineTool, + etool_step_id: str, + etool_out_id: str, ) -> list[cwl.OutputParameter]: """ Copy CommandLineTool outputs into the equivalent Workflow output parameters. @@ -1676,8 +1659,8 @@ def cltool_step_outputs_to_workflow_outputs( if not cltool_step.id: raise WorkflowException(f"Missing step id from {cltool_step}.") default_step_id = cltool_step.id.split("#")[-1] - if cltool_step.run.outputs: - for clt_out in cltool_step.run.outputs: + if clt.outputs: + for clt_out in clt.outputs: clt_out_id = clt_out.id.split("#")[-1].split("/")[-1] if clt_out_id == etool_out_id: outputSource = f"{etool_step_id}/result" @@ -1751,17 +1734,19 @@ def generate_etool_from_expr2( def traverse_step( step: cwl.WorkflowStep, parent: cwl.Workflow, + process: cwl_utils.parser.Process, replace_etool: bool, skip_command_line1: bool, skip_command_line2: bool, + context: dict[str, tuple[cwl_utils.parser.Process, bool]], ) -> bool: """Process the given WorkflowStep.""" modified = False - inputs = empty_inputs(step, parent) + inputs = cwl_utils.expression_refactor.empty_inputs(step, context, parent) if not step.id: return False step_id = step.id.split("#")[-1] - original_process = copy.deepcopy(step.run) + original_process = copy.deepcopy(process) original_step_ins = copy.deepcopy(step.in_) for inp in step.in_: if inp.valueFrom: @@ -1773,7 +1758,7 @@ def traverse_step( for source in inp.source: if not step.scatter: self.append( - example_input( + cwl_utils.expression_refactor.example_input( utils.type_for_source(parent, source.split("#")[-1]) ) ) @@ -1783,12 +1768,20 @@ def traverse_step( ) if isinstance(scattered_source_type, list): for stype in scattered_source_type: - self.append(example_input(stype.type_)) + self.append( + cwl_utils.expression_refactor.example_input( + stype.type_ + ) + ) else: - self.append(example_input(scattered_source_type.type_)) + self.append( + cwl_utils.expression_refactor.example_input( + scattered_source_type.type_ + ) + ) else: if not step.scatter: - self = example_input( + self = cwl_utils.expression_refactor.example_input( utils.type_for_source(parent, inp.source.split("#")[-1]) ) else: @@ -1796,9 +1789,13 @@ def traverse_step( parent, inp.source ) if isinstance(scattered_source_type2, list): - self = example_input(scattered_source_type2[0].type_) + self = cwl_utils.expression_refactor.example_input( + scattered_source_type2[0].type_ + ) else: - self = example_input(scattered_source_type2.type_) + self = cwl_utils.expression_refactor.example_input( + scattered_source_type2.type_ + ) expression = get_expression(inp.valueFrom, inputs, self) if expression: modified = True @@ -1816,8 +1813,11 @@ def traverse_step( for source in inp.source: source_id = source.split("#")[-1] input_source_id.append(source_id) - temp_type = utils.type_for_source( - step.run, source_id, parent + temp_type = cwl_utils.parser.utils.type_for_source( + process, + process.cwlVersion or _DEFAULT_CWL_VERSION, + source_id, + parent, ) if isinstance(temp_type, list): for ttype in temp_type: @@ -1830,7 +1830,7 @@ def traverse_step( input_source_id = inp.source.split("#")[-1] # target.id = target.id.split('#')[-1] if isinstance(original_process, cwl_utils.parser.ExpressionTool): - reqs: list[cwl.ProcessRequirement] = [] + reqs: list[cwl_utils.parser.ProcessRequirement] = [] if original_process.hints: reqs.extend(original_process.hints) if original_process.requirements: @@ -1841,14 +1841,14 @@ def traverse_step( ): break else: - if not step.run.requirements: - step.run.requirements = [] + if not process.requirements: + process.requirements = [] expr_lib = cwl_utils.expression_refactor.find_expressionLib( [parent] ) - step.run.requirements.append( + process.requirements.append( cwl_utils.expression_refactor.get_inline_javascript_requirement( - step.run, _DEFAULT_CWL_VERSION, expr_lib + original_process, _DEFAULT_CWL_VERSION, expr_lib ) ) replace_step_valueFrom_expr_with_etool( @@ -1862,6 +1862,7 @@ def traverse_step( original_step_ins, input_source_id, replace_etool, + context, ) inp.valueFrom = None inp.source = f"{etool_id}/result" @@ -1873,6 +1874,7 @@ def traverse_step( replace_etool, skip_command_line1, skip_command_line2, + context, ) if process_modified: modified = True @@ -1881,9 +1883,11 @@ def traverse_step( original_process, parent, step, + cast(cwl_utils.parser.CommandLineTool, process), replace_etool, skip_command_line1, skip_command_line2, + context, ) if clt_modified: modified = True @@ -1936,6 +1940,7 @@ def replace_step_valueFrom_expr_with_etool( original_step_ins: list[cwl.WorkflowStepInput], source: str | list[str] | None, replace_etool: bool, + context: dict[str, tuple[cwl_utils.parser.Process, bool]], ) -> None: """Replace a WorkflowStep level 'valueFrom' expression with a sibling ExpressionTool step.""" if not step_inp.id: @@ -2003,12 +2008,17 @@ def replace_step_valueFrom_expr_with_etool( # do we still need to scatter? else: scatter = None + if isinstance(etool, cwl.Process): + step_run: cwl.Process | str = etool + else: + context[f"{name}.cwl"] = (etool, True) + step_run = f"{name}.cwl" workflow.steps.append( cwl.WorkflowStep( id=name, in_=wf_step_inputs, out=[cwl.WorkflowStepOutput("result")], - run=etool, + run=step_run, scatter=scatter, scatterMethod=step.scatterMethod, ) @@ -2020,6 +2030,7 @@ def traverse_workflow( replace_etool: bool, skip_command_line1: bool, skip_command_line2: bool, + context: dict[str, tuple[cwl_utils.parser.Process, bool]], ) -> tuple[cwl.Workflow, bool]: """Traverse a workflow, processing each step.""" modified = False @@ -2029,14 +2040,24 @@ def traverse_workflow( modified = True else: step_modified = cwl_utils.expression_refactor.load_step( - step, replace_etool, skip_command_line1, skip_command_line2 + step, replace_etool, skip_command_line1, skip_command_line2, context ) if step_modified: modified = True for step in workflow.steps: if not step.id.startswith("_expression"): step_modified = traverse_step( - step, workflow, replace_etool, skip_command_line1, skip_command_line2 + step, + workflow, + ( + context[get_step_uri(step)][0] + if isinstance(step.run, str) + else cast(cwl_utils.parser.Process, step.run) + ), + replace_etool, + skip_command_line1, + skip_command_line2, + context, ) if step_modified: modified = True diff --git a/src/cwl_utils/cwl_v1_2_expression_refactor.py b/src/cwl_utils/cwl_v1_2_expression_refactor.py index eca589a6..c9ca9a80 100755 --- a/src/cwl_utils/cwl_v1_2_expression_refactor.py +++ b/src/cwl_utils/cwl_v1_2_expression_refactor.py @@ -7,7 +7,6 @@ import hashlib import uuid from collections.abc import Mapping, MutableSequence, Sequence -from contextlib import suppress from typing import Any, cast, Final from ruamel import yaml @@ -24,13 +23,13 @@ from cwl_utils.expression import do_eval, interpolate from cwl_utils.parser.utils import param_for_source_id from cwl_utils.types import ( - CWLDirectoryType, CWLFileType, CWLObjectType, CWLOutputType, CWLParameterContext, CWLRuntimeParameterContext, ) +from cwl_utils.utils import get_step_uri _DEFAULT_CWL_VERSION: Final = "v1.2" @@ -234,10 +233,13 @@ def traverse( inside: bool, skip_command_line1: bool, skip_command_line2: bool, + context: dict[str, tuple[cwl_utils.parser.Process, bool]] | None = None, ) -> tuple[ cwl.CommandLineTool | cwl.ExpressionTool | cwl.Workflow | cwl.Operation, bool ]: """Convert the given process and any subprocesses.""" + if context is None: + context = {} match process: case cwl.CommandLineTool() if not inside: process = expand_stream_shortcuts(process) @@ -301,14 +303,18 @@ def traverse( cwlVersion=process.cwlVersion, ) result, modified = traverse_workflow( - workflow, replace_etool, skip_command_line1, skip_command_line2 + workflow, replace_etool, skip_command_line1, skip_command_line2, context ) if modified: return result, True else: return process, False case cwl.ExpressionTool() if replace_etool: - expression = get_expression(process.expression, empty_inputs(process), None) + expression = get_expression( + process.expression, + cwl_utils.expression_refactor.empty_inputs(process), + None, + ) # Why call get_expression on an ExpressionTool? # It normalizes the form of $() CWL expressions into the ${} style if expression: @@ -319,7 +325,7 @@ def traverse( return etool_to_cltool(process2), True case cwl.Workflow(): return traverse_workflow( - process, replace_etool, skip_command_line1, skip_command_line2 + process, replace_etool, skip_command_line1, skip_command_line2, context ) case _: return process, False @@ -488,80 +494,6 @@ def replace_wf_input_ref_with_step_output( outp.outputSource[index] = target -def empty_inputs( - process_or_step: ( - cwl.CommandLineTool - | cwl.WorkflowStep - | cwl.ExpressionTool - | cwl.Workflow - | cwl.Operation - ), - parent: cwl.Workflow | None = None, -) -> dict[str, Any]: - """Produce a mock input object for the given inputs.""" - result = {} - if isinstance(process_or_step, cwl.Process): - for param in process_or_step.inputs: - result[param.id.split("#")[-1]] = example_input(param.type_) - else: - for param in process_or_step.in_: - param_id = param.id.split("/")[-1] - if param.source is None and param.valueFrom: - result[param_id] = example_input("string") - elif param.source is None and param.default: - result[param_id] = param.default - else: - with suppress(WorkflowException): - result[param_id] = example_input( - utils.type_for_source(process_or_step.run, param.source, parent) - ) - return result - - -def example_input(some_type: Any) -> Any: - """Produce a fake input for the given type.""" - # TODO: accept some sort of context object with local custom type definitions - if some_type == "Directory": - return CWLDirectoryType( - **{ - "class": "Directory", - "location": "https://www.example.com/example", - "basename": "example", - "listing": [ - CWLFileType( - **{ - "class": "File", - "basename": "example.txt", - "size": 23, - "contents": "hoopla", - "nameroot": "example", - "nameext": "txt", - } - ) - ], - } - ) - if some_type == "File": - return CWLFileType( - **{ - "class": "File", - "location": "https://www.example.com/example.txt", - "basename": "example.txt", - "size": 23, - "contents": "hoopla", - "nameroot": "example", - "nameext": "txt", - } - ) - if some_type == "int": - return 23 - if some_type == "string": - return "hoopla!" - if some_type == "boolean": - return True - return None - - EMPTY_FILE = CWLFileType( **{ "class": "File", @@ -656,11 +588,12 @@ def process_CommandLineTool_output(ctool: cwl.CommandLineTool, outp_id: str) -> def process_workflow_inputs_and_outputs( - workflow: cwl.Workflow, replace_etool: bool + workflow: cwl.Workflow, + replace_etool: bool, ) -> bool: """Do any needed conversions on the given Workflow's inputs and outputs.""" modified = False - inputs = empty_inputs(workflow) + inputs = cwl_utils.expression_refactor.empty_inputs(workflow) for index, param in enumerate(workflow.inputs): with SourceLine(workflow.inputs, index, WorkflowException): if param.format and get_expression(param.format, inputs, None): @@ -746,7 +679,8 @@ def process_workflow_inputs_and_outputs( def process_workflow_reqs_and_hints( - workflow: cwl.Workflow, replace_etool: bool + workflow: cwl.Workflow, + replace_etool: bool, ) -> bool: """ Convert any expressions in a workflow's reqs and hints. @@ -761,7 +695,7 @@ def process_workflow_reqs_and_hints( # ^ By refactoring replace_expr_etool to allow multiple inputs, # and connecting all workflow inputs to the generated step modified = False - inputs = empty_inputs(workflow) + inputs = cwl_utils.expression_refactor.empty_inputs(workflow) generated_res_reqs: list[tuple[str, int | str]] = [] generated_iwdr_reqs: list[tuple[str, int | str]] = [] generated_envVar_reqs: list[tuple[str, int | str]] = [] @@ -1071,12 +1005,13 @@ def process_workflow_reqs_and_hints( def process_level_reqs( - process: cwl.CommandLineTool, + process: cwl_utils.parser.Process, step: cwl.WorkflowStep, parent: cwl.Workflow, replace_etool: bool, skip_command_line1: bool, skip_command_line2: bool, + context: dict[str, tuple[cwl_utils.parser.Process, bool]], ) -> bool: """Convert expressions inside a process into new adjacent steps.""" # This is for reqs inside a Process (CommandLineTool, ExpressionTool) @@ -1092,7 +1027,7 @@ def process_level_reqs( return False modified = False target_process = step.run - inputs = cwl_utils.expression_refactor.empty_inputs(process, _DEFAULT_CWL_VERSION) + inputs = cwl_utils.expression_refactor.empty_inputs(process) generated_res_reqs: list[tuple[str, str]] = [] generated_iwdr_reqs: list[tuple[str, int | str, Any]] = [] generated_envVar_reqs: list[tuple[str, int | str]] = [] @@ -1141,6 +1076,7 @@ def process_level_reqs( target, step, replace_etool, + context, ) setattr( target_process.requirements[req_index], @@ -1268,6 +1204,7 @@ def process_level_reqs( target, step, replace_etool, + context, ) target_process.requirements[req_index].listing[ listing_index @@ -1344,15 +1281,15 @@ def traverse_CommandLineTool( clt: cwl_utils.parser.CommandLineTool, parent: cwl.Workflow, step: cwl.WorkflowStep, + target_clt: cwl_utils.parser.CommandLineTool, replace_etool: bool, skip_command_line1: bool, skip_command_line2: bool, + context: dict[str, tuple[cwl_utils.parser.Process, bool]], ) -> bool: """Extract any CWL Expressions within the given CommandLineTool into sibling steps.""" modified = False - # don't modify clt, modify step.run - target_clt = step.run - inputs = cwl_utils.expression_refactor.empty_inputs(clt, _DEFAULT_CWL_VERSION) + inputs = cwl_utils.expression_refactor.empty_inputs(clt) if not step.id: return False step_id = step.id.split("#")[-1] @@ -1367,9 +1304,15 @@ def traverse_CommandLineTool( target_type = "Any" target = cwl.WorkflowInputParameter(id=None, type_=target_type) replace_step_clt_expr_with_etool( - expression, etool_id, parent, target, step, replace_etool + expression, + etool_id, + parent, + target, + step, + replace_etool, + context, ) - target_clt.arguments[index] = ( + cast(list[Any], target_clt.arguments)[index] = ( cwl_utils.expression_refactor.get_command_line_binding( target_clt.cwlVersion or _DEFAULT_CWL_VERSION, valueFrom=f"$(inputs.{inp_id})", @@ -1399,10 +1342,16 @@ def traverse_CommandLineTool( target_type = "Any" target = cwl.WorkflowInputParameter(id=None, type_=target_type) replace_step_clt_expr_with_etool( - expression, etool_id, parent, target, step, replace_etool + expression, + etool_id, + parent, + target, + step, + replace_etool, + context, ) - target_clt.arguments[index].valueFrom = "$(inputs.{})".format( - inp_id + cast(list[Any], target_clt.arguments)[index].valueFrom = ( + "$(inputs.{})".format(inp_id) ) target_clt.inputs.append( cwl_utils.expression_refactor.get_command_input_parameter( @@ -1428,7 +1377,7 @@ def traverse_CommandLineTool( target_type = "string" target = cwl.WorkflowInputParameter(id=None, type_=target_type) replace_step_clt_expr_with_etool( - expression, etool_id, parent, target, step, replace_etool + expression, etool_id, parent, target, step, replace_etool, context ) setattr(target_clt, streamtype, f"$(inputs.{inp_id})") target_clt.inputs.append( @@ -1444,7 +1393,9 @@ def traverse_CommandLineTool( for inp in clt.inputs: if not skip_command_line1 and inp.inputBinding and inp.inputBinding.valueFrom: expression = get_expression( - inp.inputBinding.valueFrom, inputs, example_input(inp.type_) + inp.inputBinding.valueFrom, + inputs, + cwl_utils.expression_refactor.example_input(inp.type_), ) if expression: modified = True @@ -1474,7 +1425,13 @@ def traverse_CommandLineTool( glob_target_type = ["string", ArraySchema("string", "array")] target = cwl.WorkflowInputParameter(id=None, type_=glob_target_type) replace_step_clt_expr_with_etool( - expression, etool_id, parent, target, step, replace_etool + expression, + etool_id, + parent, + target, + step, + replace_etool, + context, ) outp.outputBinding.glob = f"$(inputs.{inp_id})" target_clt.inputs.append( @@ -1507,7 +1464,7 @@ def traverse_CommandLineTool( inp_id = f"_{outp_id}_outputEval" etool_id = f"expression{inp_id}" sub_wf_outputs = cltool_step_outputs_to_workflow_outputs( - step, etool_id, outp_id + step, target_clt, etool_id, outp_id ) self_type = cwl.WorkflowInputParameter( id=None, @@ -1561,24 +1518,29 @@ def traverse_CommandLineTool( ) else: final_etool = etool + if isinstance(final_etool, cwl.Process): + step_run: cwl.Process | str = etool + else: + context[f"{etool_id}.cwl"] = (final_etool, True) + step_run = f"{etool_id}.cwl" etool_step = cwl.WorkflowStep( id=etool_id, in_=orig_step_inputs, out=[cwl.WorkflowStepOutput("result")], - run=final_etool, + run=step_run, scatterMethod=step.scatterMethod, ) new_clt_step = copy.copy( step ) # a deepcopy would be convenient, but params2.cwl gives it problems new_clt_step.id = new_clt_step.id.split("#")[-1] - new_clt_step.run = copy.copy(step.run) - new_clt_step.run.id = None + new_clt = copy.copy(target_clt) + new_clt.id = "" cwl_utils.expression_refactor.remove_JSReq( - new_clt_step.run, skip_command_line1 + new_clt, skip_command_line1 ) cwl_utils.expression_refactor.process_CommandLineTool_output( - new_clt_step.run, _DEFAULT_CWL_VERSION, outp_id + new_clt, _DEFAULT_CWL_VERSION, outp_id ) new_clt_step.in_ = copy.deepcopy(step.in_) for inp in new_clt_step.in_: @@ -1587,10 +1549,14 @@ def traverse_CommandLineTool( inp.linkMerge = None for index, out in enumerate(new_clt_step.out): new_clt_step.out[index] = out.split("/")[-1] - for tool_inp in new_clt_step.run.inputs: + for tool_inp in new_clt.inputs: tool_inp.id = tool_inp.id.split("#")[-1] - for tool_out in new_clt_step.run.outputs: + for tool_out in new_clt.outputs: tool_out.id = tool_out.id.split("#")[-1] + if isinstance(new_clt_step.run, str): + context[get_step_uri(new_clt_step)] = (new_clt, True) + else: + new_clt_step.run = new_clt sub_wf_steps = [new_clt_step, etool_step] sub_workflow = cwl.Workflow( inputs=sub_wf_inputs, @@ -1660,6 +1626,7 @@ def replace_step_clt_expr_with_etool( target: cwl.WorkflowInputParameter, step: cwl.WorkflowStep, replace_etool: bool, + context: dict[str, tuple[cwl_utils.parser.Process, bool]], self_name: str | None = None, ) -> None: """Convert a step level CWL Expression to a sibling expression step.""" @@ -1689,12 +1656,17 @@ def replace_step_clt_expr_with_etool( for wf_step_input in wf_step_inputs: wf_step_input.id = wf_step_input.id.split("/")[-1] wf_step_inputs[:] = [x for x in wf_step_inputs if not x.id.startswith("_")] + if isinstance(etool, cwl.Process): + step_run: cwl.Process | str = etool + else: + context[f"{name}.cwl"] = (etool, True) + step_run = f"{name}.cwl" workflow.steps.append( cwl.WorkflowStep( id=name, in_=wf_step_inputs, out=[cwl.WorkflowStepOutput("result")], - run=etool, + run=step_run, ) ) @@ -1706,6 +1678,7 @@ def replace_clt_hintreq_expr_with_etool( target: cwl.WorkflowInputParameter, step: cwl.WorkflowStep, replace_etool: bool, + context: dict[str, tuple[cwl_utils.parser.Process, bool]], self_name: str | None = None, ) -> None: """Factor out an expression inside a CommandLineTool req or hint into a sibling step.""" @@ -1736,12 +1709,17 @@ def replace_clt_hintreq_expr_with_etool( for wf_step_input in wf_step_inputs: wf_step_input.id = wf_step_input.id.split("/")[-1] wf_step_inputs[:] = [x for x in wf_step_inputs if not x.id.startswith("_")] + if isinstance(etool, cwl.Process): + step_run: cwl.Process | str = etool + else: + context[f"{name}.cwl"] = (etool, True) + step_run = f"{name}.cwl" workflow.steps.append( cwl.WorkflowStep( id=name, in_=wf_step_inputs, out=[cwl.WorkflowStepOutput("result")], - run=etool, + run=step_run, ) ) @@ -1773,7 +1751,10 @@ def cltool_inputs_to_etool_inputs( def cltool_step_outputs_to_workflow_outputs( - cltool_step: cwl.WorkflowStep, etool_step_id: str, etool_out_id: str + cltool_step: cwl.WorkflowStep, + clt: cwl_utils.parser.CommandLineTool, + etool_step_id: str, + etool_out_id: str, ) -> list[cwl.OutputParameter]: """ Copy CommandLineTool outputs into the equivalent Workflow output parameters. @@ -1785,8 +1766,8 @@ def cltool_step_outputs_to_workflow_outputs( if not cltool_step.id: raise WorkflowException(f"Missing step id from {cltool_step}.") default_step_id = cltool_step.id.split("#")[-1] - if cltool_step.run.outputs: - for clt_out in cltool_step.run.outputs: + if clt.outputs: + for clt_out in clt.outputs: clt_out_id = clt_out.id.split("#")[-1].split("/")[-1] if clt_out_id == etool_out_id: outputSource = f"{etool_step_id}/result" @@ -1864,17 +1845,19 @@ def generate_etool_from_expr2( def traverse_step( step: cwl.WorkflowStep, parent: cwl.Workflow, + process: cwl_utils.parser.Process, replace_etool: bool, skip_command_line1: bool, skip_command_line2: bool, + context: dict[str, tuple[cwl_utils.parser.Process, bool]], ) -> bool: """Process the given WorkflowStep.""" modified = False - inputs = empty_inputs(step, parent) + inputs = cwl_utils.expression_refactor.empty_inputs(step, context, parent) if not step.id: return False step_id = step.id.split("#")[-1] - original_process = copy.deepcopy(step.run) + original_process = copy.deepcopy(process) original_step_ins = copy.deepcopy(step.in_) for inp in step.in_: if inp.valueFrom: @@ -1886,7 +1869,7 @@ def traverse_step( for source in inp.source: if not step.scatter: self.append( - example_input( + cwl_utils.expression_refactor.example_input( utils.type_for_source(parent, source.split("#")[-1]) ) ) @@ -1896,12 +1879,20 @@ def traverse_step( ) if isinstance(scattered_source_type, list): for stype in scattered_source_type: - self.append(example_input(stype.type_)) + self.append( + cwl_utils.expression_refactor.example_input( + stype.type_ + ) + ) else: - self.append(example_input(scattered_source_type.type_)) + self.append( + cwl_utils.expression_refactor.example_input( + scattered_source_type.type_ + ) + ) else: if not step.scatter: - self = example_input( + self = cwl_utils.expression_refactor.example_input( utils.type_for_source(parent, inp.source.split("#")[-1]) ) else: @@ -1909,9 +1900,13 @@ def traverse_step( parent, inp.source ) if isinstance(scattered_source_type2, list): - self = example_input(scattered_source_type2[0].type_) + self = cwl_utils.expression_refactor.example_input( + scattered_source_type2[0].type_ + ) else: - self = example_input(scattered_source_type2.type_) + self = cwl_utils.expression_refactor.example_input( + scattered_source_type2.type_ + ) expression = get_expression(inp.valueFrom, inputs, self) if expression: modified = True @@ -1929,8 +1924,11 @@ def traverse_step( for source in inp.source: source_id = source.split("#")[-1] input_source_id.append(source_id) - temp_type = utils.type_for_source( - step.run, source_id, parent + temp_type = cwl_utils.parser.utils.type_for_source( + process, + process.cwlVersion or _DEFAULT_CWL_VERSION, + source_id, + parent, ) if isinstance(temp_type, list): for ttype in temp_type: @@ -1943,7 +1941,7 @@ def traverse_step( input_source_id = inp.source.split("#")[-1] # target.id = target.id.split('#')[-1] if isinstance(original_process, cwl_utils.parser.ExpressionTool): - reqs: list[cwl.ProcessRequirement] = [] + reqs: list[cwl_utils.parser.ProcessRequirement] = [] if original_process.hints: reqs.extend(original_process.hints) if original_process.requirements: @@ -1954,14 +1952,14 @@ def traverse_step( ): break else: - if not step.run.requirements: - step.run.requirements = [] + if not process.requirements: + process.requirements = [] expr_lib = cwl_utils.expression_refactor.find_expressionLib( [parent] ) - step.run.requirements.append( + process.requirements.append( cwl_utils.expression_refactor.get_inline_javascript_requirement( - step.run, _DEFAULT_CWL_VERSION, expr_lib + original_process, _DEFAULT_CWL_VERSION, expr_lib ) ) replace_step_valueFrom_expr_with_etool( @@ -1975,6 +1973,7 @@ def traverse_step( original_step_ins, input_source_id, replace_etool, + context, ) inp.valueFrom = None inp.source = f"{etool_id}/result" @@ -1983,7 +1982,7 @@ def traverse_step( if expression: modified = True replace_step_when_expr_with_etool( - expression, parent, step, original_step_ins, replace_etool + expression, parent, step, original_step_ins, replace_etool, context ) # TODO: skip or special process for sub workflows? @@ -1994,6 +1993,7 @@ def traverse_step( replace_etool, skip_command_line1, skip_command_line2, + context, ) if process_modified: modified = True @@ -2002,9 +2002,11 @@ def traverse_step( original_process, parent, step, + cast(cwl_utils.parser.CommandLineTool, process), replace_etool, skip_command_line1, skip_command_line2, + context, ) if clt_modified: modified = True @@ -2057,6 +2059,7 @@ def replace_step_valueFrom_expr_with_etool( original_step_ins: list[cwl.WorkflowStepInput], source: str | list[str] | None, replace_etool: bool, + context: dict[str, tuple[cwl_utils.parser.Process, bool]], ) -> None: """Replace a WorkflowStep level 'valueFrom' expression with a sibling ExpressionTool step.""" if not step_inp.id: @@ -2124,12 +2127,17 @@ def replace_step_valueFrom_expr_with_etool( # do we still need to scatter? else: scatter = None + if isinstance(etool, cwl.Process): + step_run: cwl.Process | str = etool + else: + context[f"{name}.cwl"] = (etool, True) + step_run = f"{name}.cwl" workflow.steps.append( cwl.WorkflowStep( id=name, in_=wf_step_inputs, out=[cwl.WorkflowStepOutput("result")], - run=etool, + run=step_run, scatter=scatter, scatterMethod=step.scatterMethod, ) @@ -2142,6 +2150,7 @@ def replace_step_when_expr_with_etool( step: cwl.WorkflowStep, original_step_ins: list[cwl.WorkflowStepInput], replace_etool: bool, + context: dict[str, tuple[cwl_utils.parser.Process, bool]], ) -> None: """Replace a WorkflowStep level 'when' expression with a sibling ExpressionTool step.""" if not step.id: @@ -2193,12 +2202,17 @@ def replace_step_when_expr_with_etool( for index, entry in enumerate(scatter): scatter[index] = entry.split("/")[-1] scatter = step.scatter + if isinstance(etool, cwl.Process): + step_run: cwl.Process | str = etool + else: + context[f"{etool_id}.cwl"] = (etool, True) + step_run = f"{etool_id}.cwl" workflow.steps.append( cwl.WorkflowStep( id=etool_id, in_=wf_step_inputs, out=[cwl.WorkflowStepOutput("result")], - run=etool, + run=step_run, scatter=scatter, scatterMethod=step.scatterMethod, ) @@ -2212,6 +2226,7 @@ def traverse_workflow( replace_etool: bool, skip_command_line1: bool, skip_command_line2: bool, + context: dict[str, tuple[cwl_utils.parser.Process, bool]], ) -> tuple[cwl.Workflow, bool]: """Traverse a workflow, processing each step.""" modified = False @@ -2221,14 +2236,24 @@ def traverse_workflow( modified = True else: step_modified = cwl_utils.expression_refactor.load_step( - step, replace_etool, skip_command_line1, skip_command_line2 + step, replace_etool, skip_command_line1, skip_command_line2, context ) if step_modified: modified = True for step in workflow.steps: if not step.id.startswith("_expression"): step_modified = traverse_step( - step, workflow, replace_etool, skip_command_line1, skip_command_line2 + step, + workflow, + ( + context[get_step_uri(step)][0] + if isinstance(step.run, str) + else cast(cwl_utils.parser.Process, step.run) + ), + replace_etool, + skip_command_line1, + skip_command_line2, + context, ) if step_modified: modified = True diff --git a/src/cwl_utils/expression_refactor.py b/src/cwl_utils/expression_refactor.py index f4f4353f..5eee94ff 100755 --- a/src/cwl_utils/expression_refactor.py +++ b/src/cwl_utils/expression_refactor.py @@ -9,6 +9,7 @@ import shutil import sys from collections.abc import MutableMapping, MutableSequence, Sequence +from contextlib import suppress from pathlib import Path from typing import Any, Protocol, cast, Literal, overload @@ -41,7 +42,11 @@ CommandLineTool, load_document_by_uri, CommandLineBinding, + AbstractProcess, ) +from cwl_utils.parser.utils import type_for_source +from cwl_utils.types import CWLFileType, CWLDirectoryType +from cwl_utils.utils import get_step_uri _logger = logging.getLogger("cwl-expression-refactor") # pylint: disable=invalid-name defaultStreamHandler = logging.StreamHandler() # pylint: disable=invalid-name @@ -153,55 +158,66 @@ def cltool_inputs_to_etool_inputs( ) +@overload +def empty_inputs( + process_or_step: Process, + context: dict[str, tuple[Process, bool]] | None = ..., + parent: Workflow | None = ..., +) -> dict[str, Any]: ... + + +@overload +def empty_inputs( + process_or_step: WorkflowStep, + context: dict[str, tuple[Process, bool]], + parent: Workflow, +) -> dict[str, Any]: ... + + def empty_inputs( process_or_step: Process | WorkflowStep, - cwlVersion: Literal["v1.0", "v1.1", "v1.2"], + context: dict[str, tuple[Process, bool]] | None = None, parent: Workflow | None = None, ) -> dict[str, Any]: - cwlVersion = ( - process_or_step.cwlVersion - if isinstance(process_or_step, Process) and process_or_step.cwlVersion - else cwlVersion - ) - match cwlVersion: - case "v1.0": - return cwl_v1_0_expression_refactor.empty_inputs( - cast( - cwl_v1_0.CommandLineTool - | cwl_v1_0.WorkflowStep - | cwl_v1_0.ExpressionTool - | cwl_v1_0.Workflow, - process_or_step, - ), - cast(cwl_v1_0.Workflow, parent), - ) - case "v1.1": - return cwl_v1_1_expression_refactor.empty_inputs( - cast( - cwl_v1_1.CommandLineTool - | cwl_v1_1.WorkflowStep - | cwl_v1_1.ExpressionTool - | cwl_v1_1.Workflow, - process_or_step, - ), - cast(cwl_v1_1.Workflow, parent), - ) - case "v1.2": - return cwl_v1_2_expression_refactor.empty_inputs( - cast( - cwl_v1_2.CommandLineTool - | cwl_v1_2.WorkflowStep - | cwl_v1_2.ExpressionTool - | cwl_v1_2.Workflow - | cwl_v1_2.Operation, - process_or_step, - ), - cast(cwl_v1_2.Workflow, parent), - ) - case _: - raise WorkflowException( - f"Sorry, {cwlVersion} is not a supported CWL version by this tool.", - ) + """Produce a mock input object for the given inputs.""" + result = {} + if isinstance(process_or_step, Process): + for param in process_or_step.inputs: + result[param.id.split("#")[-1]] = example_input(param.type_) + else: + for param1 in process_or_step.in_: + param_id = param1.id.split("/")[-1] + if param1.source is None and param1.valueFrom: + result[param_id] = example_input("string") + elif param1.source is None and param1.default: + result[param_id] = param1.default + elif param1.source is not None: + with suppress(WorkflowException): + if isinstance(process_or_step.run, str): + process = cast(dict[str, tuple[Process, bool]], context)[ + get_step_uri(process_or_step) + ][0] + else: + process = cast(Process, process_or_step.run) + if (cwlVersion := process.cwlVersion) is not None: + result[param_id] = example_input( + type_for_source( + process, + cast(Literal["v1.0", "v1.1", "v1.2"], cwlVersion), + param1.source, + parent, + ) + ) + elif (cwlVersion := cast(Workflow, parent).cwlVersion) is not None: + result[param_id] = example_input( + type_for_source( + process, + cast(Literal["v1.0", "v1.1", "v1.2"], cwlVersion), + param1.source, + parent, + ) + ) + return result def etool_to_cltool( @@ -229,6 +245,50 @@ def etool_to_cltool( ) +def example_input(some_type: Any) -> Any: + """Produce a fake input for the given type.""" + # TODO: accept some sort of context object with local custom type definitions + if some_type == "Directory": + return CWLDirectoryType( + **{ + "class": "Directory", + "location": "https://www.example.com/example", + "basename": "example", + "listing": [ + CWLFileType( + **{ + "class": "File", + "basename": "example.txt", + "size": 23, + "contents": "hoopla", + "nameroot": "example", + "nameext": "txt", + } + ) + ], + } + ) + if some_type == "File": + return CWLFileType( + **{ + "class": "File", + "location": "https://www.example.com/example.txt", + "basename": "example.txt", + "size": 23, + "contents": "hoopla", + "nameroot": "example", + "nameext": "txt", + } + ) + if some_type == "int": + return 23 + if some_type == "string": + return "hoopla!" + if some_type == "boolean": + return True + return None + + def find_expressionLib( processes: Sequence[Process | WorkflowStep], ) -> list[str] | None: @@ -498,59 +558,76 @@ def load_step( replace_etool: bool, skip_command_line1: bool, skip_command_line2: bool, + context: dict[str, tuple[Process, bool]], ) -> bool: """If the step's Process is not inline, load and process it.""" modified = False if isinstance(step.run, str): - process = cast( - Process, load_document_by_uri(step.run, loadingOptions=step.loadingOptions) - ) - # FIXME: with strong typing, it won't be possible to directly assign to step.run - match process.cwlVersion: - case "v1.0": - step.run, modified = cwl_v1_0_expression_refactor.traverse( - cast( - cwl_v1_0.CommandLineTool - | cwl_v1_0.ExpressionTool - | cwl_v1_0.Workflow, - process, - ), - replace_etool, - True, - skip_command_line1, - skip_command_line2, - ) - case "v1.1": - step.run, modified = cwl_v1_1_expression_refactor.traverse( - cast( - cwl_v1_1.CommandLineTool - | cwl_v1_1.ExpressionTool - | cwl_v1_1.Workflow, - process, - ), - replace_etool, - True, - skip_command_line1, - skip_command_line2, - ) - case "v1.2": - step.run, modified = cwl_v1_2_expression_refactor.traverse( - cast( - cwl_v1_2.CommandLineTool - | cwl_v1_2.ExpressionTool - | cwl_v1_2.Workflow - | cwl_v1_2.Operation, - process, - ), - replace_etool, - True, - skip_command_line1, - skip_command_line2, - ) - case _: - raise WorkflowException( - f"Sorry, {process.cwlVersion} is not a supported CWL version by this tool.", + if (uri := get_step_uri(step)) not in context: + process = cast( + AbstractProcess, + load_document_by_uri( + path=uri, + loadingOptions=step.loadingOptions, + ), + ) + if not isinstance(process, Process): + raise Exception( + f"Unsupported process type: {process.__class__.__name__}" ) + match process.cwlVersion: + case "v1.0": + process, modified = cwl_v1_0_expression_refactor.traverse( + cast( + cwl_v1_0.CommandLineTool + | cwl_v1_0.ExpressionTool + | cwl_v1_0.Workflow, + process, + ), + replace_etool, + True, + skip_command_line1, + skip_command_line2, + context, + ) + case "v1.1": + process, modified = cwl_v1_1_expression_refactor.traverse( + cast( + cwl_v1_1.CommandLineTool + | cwl_v1_1.ExpressionTool + | cwl_v1_1.Workflow, + process, + ), + replace_etool, + True, + skip_command_line1, + skip_command_line2, + context, + ) + case "v1.2": + process, modified = cwl_v1_2_expression_refactor.traverse( + cast( + cwl_v1_2.CommandLineTool + | cwl_v1_2.ExpressionTool + | cwl_v1_2.Workflow + | cwl_v1_2.Operation, + process, + ), + replace_etool, + True, + skip_command_line1, + skip_command_line2, + context, + ) + case _: + raise WorkflowException( + f"Sorry, {process.cwlVersion} is not a supported CWL version by this tool.", + ) + context[uri] = (process, modified) + else: + process = step.run + if not isinstance(process, Process): + raise Exception(f"Unsupported process type: {process.__class__.__name__}") return modified @@ -621,6 +698,7 @@ def refactor(args: argparse.Namespace) -> int: with open(document) as doc_handle: result = yaml.load(doc_handle) uri = Path(document).resolve().as_uri() + context: dict[str, tuple[Process, bool]] = {} try: match result["cwlVersion"]: case "v1.0": @@ -630,6 +708,7 @@ def refactor(args: argparse.Namespace) -> int: False, args.skip_some1, args.skip_some2, + context, ) case "v1.1": result, modified = cwl_v1_1_expression_refactor.traverse( @@ -638,6 +717,7 @@ def refactor(args: argparse.Namespace) -> int: False, args.skip_some1, args.skip_some2, + context, ) case "v1.2": result, modified = cwl_v1_2_expression_refactor.traverse( @@ -646,6 +726,7 @@ def refactor(args: argparse.Namespace) -> int: False, args.skip_some1, args.skip_some2, + context, ) case _: _logger.error( @@ -653,33 +734,35 @@ def refactor(args: argparse.Namespace) -> int: result["cwlVersion"], ) return -1 - output = Path(args.dir) / Path(document).name - if not modified: - if len(args.inputs) > 1: - shutil.copyfile(document, output) - continue + if not modified and len(args.inputs) == 1: + return 7 + context[document] = (result, modified) + for path, (process, modified) in context.items(): + output = Path(args.dir) / Path(path).name + if not modified: + if len(args.inputs) > 1: + shutil.copyfile(path, output) + continue + if not isinstance(process, MutableSequence): + result_json = save( + process, + base_url=(process.loadingOptions.fileuri or ""), + ) + # ^^ Setting the base_url and keeping the default value + # for relative_uris=True means that the IDs in the generated + # JSON/YAML are kept clean of the path to the input document else: - return 7 - if not isinstance(result, MutableSequence): - result_json = save( - result, - base_url=(result.loadingOptions.fileuri or ""), - ) - # ^^ Setting the base_url and keeping the default value - # for relative_uris=True means that the IDs in the generated - # JSON/YAML are kept clean of the path to the input document - else: - result_json = [ - save(result_item, base_url=result_item.loadingOptions.fileuri) - for result_item in result - ] - walk_tree(result_json) - # ^ converts multiline strings to nice multiline YAML - with output.open("w", encoding="utf-8") as output_filehandle: - output_filehandle.write( - "#!/usr/bin/env cwl-runner\n" - ) # TODO: teach the codegen to do this? - yaml.dump(result_json, output_filehandle) + result_json = [ + save(result_item, base_url=result_item.loadingOptions.fileuri) + for result_item in process + ] + walk_tree(result_json) + # ^ converts multiline strings to nice multiline YAML + with output.open("w", encoding="utf-8") as output_filehandle: + output_filehandle.write( + "#!/usr/bin/env cwl-runner\n" + ) # TODO: teach the codegen to do this? + yaml.dump(result_json, output_filehandle) except WorkflowException as exc: return_code = 1 _logger.exception("Skipping %s due to error.", document, exc_info=exc) diff --git a/src/cwl_utils/parser/__init__.py b/src/cwl_utils/parser/__init__.py index e7fef0dd..f620156d 100644 --- a/src/cwl_utils/parser/__init__.py +++ b/src/cwl_utils/parser/__init__.py @@ -193,8 +193,10 @@ class NoType(ABC): cwl_v1_2.DockerRequirement, ) """Type union for a CWL v1.x DockerRequirement object.""" -Process: TypeAlias = Workflow | CommandLineTool | ExpressionTool | cwl_v1_2.Operation +AbstractProcess: TypeAlias = cwl_v1_0.Process | cwl_v1_1.Process | cwl_v1_2.Process """Type Union for a CWL v1.x Process object.""" +Process: TypeAlias = Workflow | CommandLineTool | ExpressionTool | cwl_v1_2.Operation +"""Type Union for a CWL v1.x Process implementations.""" ProcessRequirement: TypeAlias = ( cwl_v1_0.ProcessRequirement | cwl_v1_1.ProcessRequirement @@ -449,7 +451,7 @@ def save( def is_process(v: Any) -> bool: """Test to see if the object is a CWL v1.x Python Process object.""" - return isinstance(v, cwl_v1_0.Process | cwl_v1_1.Process | cwl_v1_2.Process) + return isinstance(v, AbstractProcess) def version_split(version: str) -> MutableSequence[int]: diff --git a/src/cwl_utils/parser/cwl_v1_0_utils.py b/src/cwl_utils/parser/cwl_v1_0_utils.py index f80c74bb..5310af5a 100644 --- a/src/cwl_utils/parser/cwl_v1_0_utils.py +++ b/src/cwl_utils/parser/cwl_v1_0_utils.py @@ -265,10 +265,9 @@ def type_for_step_input( """Determine the type for the given step input.""" if in_.valueFrom is not None: return "Any" - step_run = cwl_utils.parser.utils.load_step(step) - cwl_utils.parser.utils.convert_stdstreams_to_files(step_run) - if step_run and step_run.inputs: - for step_input in step_run.inputs: + if step_run := cwl_utils.parser.utils.load_step(step): + cwl_utils.parser.utils.convert_stdstreams_to_files(step_run) + for step_input in cast(cwl_utils.parser.Process, step_run).inputs or []: if cast(str, step_input.id).split("#")[-1] == in_.id.split("#")[-1]: input_type = step_input.type_ if step.scatter is not None and in_.id in aslist(step.scatter): @@ -282,10 +281,9 @@ def type_for_step_output( sourcename: str, ) -> Any: """Determine the type for the given step output.""" - step_run = cwl_utils.parser.utils.load_step(step) - cwl_utils.parser.utils.convert_stdstreams_to_files(step_run) - if step_run and step_run.outputs: - for step_output in step_run.outputs: + if step_run := cwl_utils.parser.utils.load_step(step): + cwl_utils.parser.utils.convert_stdstreams_to_files(step_run) + for step_output in cast(cwl_utils.parser.Process, step_run).outputs or []: if ( step_output.id.split("#")[-1].split("/")[-1] == sourcename.split("#")[-1].split("/")[-1] @@ -311,11 +309,12 @@ def type_for_source( sourcenames: str | list[str], parent: cwl.Workflow | None = None, linkMerge: str | None = None, + loaded_steps: dict[str, cwl_utils.parser.AbstractProcess] | None = None, ) -> Any: """Determine the type for the given sourcenames.""" scatter_context: list[tuple[int, str] | None] = [] params = cwl_utils.parser.utils.param_for_source_id( - process, sourcenames, parent, scatter_context + process, sourcenames, parent, scatter_context, loaded_steps ) if not isinstance(params, MutableSequence): new_type = params.type_ diff --git a/src/cwl_utils/parser/cwl_v1_1_utils.py b/src/cwl_utils/parser/cwl_v1_1_utils.py index 838922e8..3bf53008 100644 --- a/src/cwl_utils/parser/cwl_v1_1_utils.py +++ b/src/cwl_utils/parser/cwl_v1_1_utils.py @@ -347,10 +347,9 @@ def type_for_step_input( """Determine the type for the given step input.""" if in_.valueFrom is not None: return "Any" - step_run = cwl_utils.parser.utils.load_step(step) - cwl_utils.parser.utils.convert_stdstreams_to_files(step_run) - if step_run and step_run.inputs: - for step_input in step_run.inputs: + if step_run := cwl_utils.parser.utils.load_step(step): + cwl_utils.parser.utils.convert_stdstreams_to_files(step_run) + for step_input in cast(cwl_utils.parser.Process, step_run).inputs or []: if cast(str, step_input.id).split("#")[-1] == in_.id.split("#")[-1]: input_type = step_input.type_ if step.scatter is not None and in_.id in aslist(step.scatter): @@ -364,10 +363,9 @@ def type_for_step_output( sourcename: str, ) -> Any: """Determine the type for the given step output.""" - step_run = cwl_utils.parser.utils.load_step(step) - cwl_utils.parser.utils.convert_stdstreams_to_files(step_run) - if step_run and step_run.outputs: - for output in step_run.outputs: + if step_run := cwl_utils.parser.utils.load_step(step): + cwl_utils.parser.utils.convert_stdstreams_to_files(step_run) + for output in cast(cwl_utils.parser.Process, step_run).outputs or []: if ( output.id.split("#")[-1].split("/")[-1] == sourcename.split("#")[-1].split("/")[-1] @@ -393,11 +391,12 @@ def type_for_source( sourcenames: str | list[str], parent: cwl.Workflow | None = None, linkMerge: str | None = None, + loaded_steps: dict[str, cwl_utils.parser.AbstractProcess] | None = None, ) -> Any: """Determine the type for the given sourcenames.""" scatter_context: list[tuple[int, str] | None] = [] params = cwl_utils.parser.utils.param_for_source_id( - process, sourcenames, parent, scatter_context + process, sourcenames, parent, scatter_context, loaded_steps ) if not isinstance(params, MutableSequence): new_type = params.type_ diff --git a/src/cwl_utils/parser/cwl_v1_2_utils.py b/src/cwl_utils/parser/cwl_v1_2_utils.py index e0f94e25..a30b35a8 100644 --- a/src/cwl_utils/parser/cwl_v1_2_utils.py +++ b/src/cwl_utils/parser/cwl_v1_2_utils.py @@ -375,10 +375,9 @@ def type_for_step_input( """Determine the type for the given step input.""" if in_.valueFrom is not None: return "Any" - step_run = cwl_utils.parser.utils.load_step(step) - cwl_utils.parser.utils.convert_stdstreams_to_files(step_run) - if step_run and step_run.inputs: - for step_input in step_run.inputs: + if step_run := cwl_utils.parser.utils.load_step(step): + cwl_utils.parser.utils.convert_stdstreams_to_files(step_run) + for step_input in cast(cwl_utils.parser.Process, step_run).inputs or []: if cast(str, step_input.id).split("#")[-1] == in_.id.split("#")[-1]: input_type = step_input.type_ if step.scatter is not None and in_.id in aslist(step.scatter): @@ -392,10 +391,9 @@ def type_for_step_output( sourcename: str, ) -> Any: """Determine the type for the given step output.""" - step_run = cwl_utils.parser.utils.load_step(step) - cwl_utils.parser.utils.convert_stdstreams_to_files(step_run) - if step_run and step_run.outputs: - for output in step_run.outputs: + if step_run := cwl_utils.parser.utils.load_step(step): + cwl_utils.parser.utils.convert_stdstreams_to_files(step_run) + for output in cast(cwl_utils.parser.Process, step_run).outputs or []: if ( output.id.split("#")[-1].split("/")[-1] == sourcename.split("#")[-1].split("/")[-1] @@ -422,11 +420,12 @@ def type_for_source( parent: cwl.Workflow | None = None, linkMerge: str | None = None, pickValue: str | None = None, + loaded_steps: dict[str, cwl_utils.parser.AbstractProcess] | None = None, ) -> Any: """Determine the type for the given sourcenames.""" scatter_context: list[tuple[int, str] | None] = [] params = cwl_utils.parser.utils.param_for_source_id( - process, sourcenames, parent, scatter_context + process, sourcenames, parent, scatter_context, loaded_steps ) if not isinstance(params, MutableSequence): new_type = params.type_ diff --git a/src/cwl_utils/parser/utils.py b/src/cwl_utils/parser/utils.py index 26696165..87d145eb 100644 --- a/src/cwl_utils/parser/utils.py +++ b/src/cwl_utils/parser/utils.py @@ -32,9 +32,10 @@ CommandOutputParameter, WorkflowInputParameter, load_document_by_uri, + AbstractProcess, ) from cwl_utils.errors import WorkflowException -from cwl_utils.utils import yaml_dumps +from cwl_utils.utils import yaml_dumps, get_step_uri _logger = logging.getLogger("cwl_utils") @@ -144,7 +145,7 @@ def check_types( raise ValidationException(f"Invalid value {linkMerge} for linkMerge field.") -def convert_stdstreams_to_files(process: Process) -> None: +def convert_stdstreams_to_files(process: AbstractProcess) -> None: """Convert stdin, stdout and stderr type shortcuts to files.""" match process: case cwl_v1_0.CommandLineTool(): @@ -230,18 +231,28 @@ def load_inputfile_by_yaml( def load_step( - step: WorkflowStep, -) -> Process: + step: WorkflowStep, loaded_steps: dict[str, AbstractProcess] | None = None +) -> AbstractProcess: if isinstance(step.run, str): - step_run = load_document_by_uri( - path=step.loadingOptions.fetcher.urljoin( - base_url=cast(str, step.loadingOptions.fileuri), - url=step.run, - ), - loadingOptions=step.loadingOptions, - ) - return cast(Process, step_run) - return cast(Process, copy.deepcopy(step.run)) + uri = get_step_uri(step) + if loaded_steps is not None and uri in loaded_steps: + return loaded_steps[uri] + else: + step_run = cast( + AbstractProcess, + load_document_by_uri( + path=uri, + loadingOptions=step.loadingOptions, + ), + ) + if loaded_steps is not None: + loaded_steps[uri] = step_run + return step_run + else: + step_run = copy.deepcopy(step.run) + if not isinstance(step_run, cwl_utils.parser.Process): + raise Exception(f"Unsupported process type: {step_run.__class__.__name__}") + return step_run def merge_flatten_type(src: Any) -> Any: @@ -258,6 +269,7 @@ def param_for_source_id( sourcenames: str | list[str], parent: Workflow | None = None, scatter_context: list[tuple[int, str] | None] | None = None, + loaded_steps: dict[str, cwl_utils.parser.AbstractProcess] | None = None, ) -> ( CommandInputParameter | CommandOutputParameter @@ -295,7 +307,10 @@ def param_for_source_id( == step.id.split("#")[-1] and step.out ): - step_run = cwl_utils.parser.utils.load_step(step) + step_run = cast( + Process, + cwl_utils.parser.utils.load_step(step, loaded_steps), + ) cwl_utils.parser.utils.convert_stdstreams_to_files(step_run) for outp in step.out: outp_id = outp if isinstance(outp, str) else outp.id @@ -527,6 +542,7 @@ def type_for_source( parent: Workflow | None = None, linkMerge: str | None = None, pickValue: str | None = None, + loaded_steps: dict[str, cwl_utils.parser.AbstractProcess] | None = None, ) -> Any: """Determine the type for the given sourcenames.""" match process.cwlVersion or cwlVersion: @@ -541,6 +557,7 @@ def type_for_source( sourcenames, cast(cwl_v1_0.Workflow | None, parent), linkMerge, + loaded_steps, ) case "v1.1": return cwl_v1_1_utils.type_for_source( @@ -553,6 +570,7 @@ def type_for_source( sourcenames, cast(cwl_v1_1.Workflow | None, parent), linkMerge, + loaded_steps, ) case "v1.2": return cwl_v1_2_utils.type_for_source( @@ -566,6 +584,7 @@ def type_for_source( cast(cwl_v1_2.Workflow | None, parent), linkMerge, pickValue, + loaded_steps, ) case _ as cwlVersion: raise ValidationException( diff --git a/src/cwl_utils/utils.py b/src/cwl_utils/utils.py index bfae73dd..daa2b381 100644 --- a/src/cwl_utils/utils.py +++ b/src/cwl_utils/utils.py @@ -11,7 +11,7 @@ from copy import deepcopy from importlib.resources import files from io import StringIO -from typing import Any +from typing import Any, cast from urllib.parse import urlparse from ruamel.yaml.main import YAML @@ -22,7 +22,7 @@ from cwl_utils.loghandler import _logger # Type hinting -from cwl_utils.parser import cwl_v1_0, cwl_v1_1, cwl_v1_2 +from cwl_utils.parser import cwl_v1_0, cwl_v1_1, cwl_v1_2, WorkflowStep # Load as 1.2 files from cwl_utils.parser.cwl_v1_2 import InputArraySchema as InputArraySchemaV1_2 @@ -454,6 +454,17 @@ def is_local_uri(uri: str) -> bool: return False +def get_step_uri(step: WorkflowStep) -> str: + if not isinstance(step.run, str): + raise Exception( + f"Impossible to retrieve URI for step {step.id}: it embeds a process" + ) + return step.loadingOptions.fetcher.urljoin( + base_url=cast(str, step.loadingOptions.fileuri), + url=step.run, + ) + + def get_value_from_uri(uri: str) -> str: """ Given a URI, return the value after #.