cwl_utils.parser.cwl_v1_0_utils

Attributes

CONTENT_LIMIT

Classes

SrcSink

Functions

`can_assign_src_to_sink`(src, sink[, strict])	Check for identical type specifications, ignoring extra keys like inputBinding.
`check_all_types`(src_dict, sinks, type_dict)	Given a list of sinks, check if their types match with the types of their sources.
`check_types`(srctype, sinktype, linkMerge[, valueFrom])	Check if the source and sink types are correct.
`content_limit_respected_read_bytes`(f)	Read file content up to 64 kB as a byte array.
`content_limit_respected_read`(f)	Read file content up to 64 kB as an utf-8 encoded string.
`convert_stdstreams_to_files`(clt)	Convert stdout and stderr type shortcuts to files.
`load_inputfile`(doc[, baseuri, loadingOptions])	Load a CWL v1.0 input file from a serialized YAML string or a YAML object.
`load_inputfile_by_string`(string, uri[, loadingOptions])	Load a CWL v1.0 input file from a serialized YAML string.
`load_inputfile_by_yaml`(yaml, uri[, loadingOptions])	Load a CWL v1.0 input file from a YAML object.
`merge_flatten_type`(src)	Return the merge flattened type of the source type.
`type_for_step_input`(step, in_)	Determine the type for the given step input.
`type_for_step_output`(step, sourcename)	Determine the type for the given step output.
`type_for_source`(process, sourcenames[, parent, linkMerge])	Determine the type for the given sourcenames.
`param_for_source_id`(process, sourcenames[, parent, ...])	Find the process input parameter that matches one of the given sourcenames.

Module Contents

cwl_utils.parser.cwl_v1_0_utils.CONTENT_LIMIT: int = 65536

class cwl_utils.parser.cwl_v1_0_utils.SrcSink

Bases: tuple

src

sink

linkMerge

message

cwl_utils.parser.cwl_v1_0_utils.can_assign_src_to_sink(src, sink, strict=False)

Check for identical type specifications, ignoring extra keys like inputBinding.

src: admissible source types sink: admissible sink types

In non-strict comparison, at least one source type must match one sink type,: except for ‘null’.

In strict comparison, all source types must match at least one sink type.

Parameters:

src (Any)
sink (Any)
strict (bool)

Return type:

bool

cwl_utils.parser.cwl_v1_0_utils.check_all_types(src_dict, sinks, type_dict)

Given a list of sinks, check if their types match with the types of their sources.

Parameters:

src_dict (dict[str, Any])
sinks (collections.abc.MutableSequence[Union[cwl_utils.parser.cwl_v1_0.WorkflowStepInput, cwl_utils.parser.cwl_v1_0.WorkflowOutputParameter]])
type_dict (dict[str, Any])

Return type:

dict[str, list[SrcSink]]

cwl_utils.parser.cwl_v1_0_utils.check_types(srctype, sinktype, linkMerge, valueFrom=None)

Check if the source and sink types are correct.

Acceptable types are “pass”, “warning”, or “exception”.

Parameters:

srctype (Any)
sinktype (Any)
linkMerge (Optional[str])
valueFrom (Optional[str])

Return type:

str

cwl_utils.parser.cwl_v1_0_utils.content_limit_respected_read_bytes(f)

Read file content up to 64 kB as a byte array.

Truncate content for larger files.

Parameters:: f (IO[bytes])
Return type:: bytes

cwl_utils.parser.cwl_v1_0_utils.content_limit_respected_read(f)

Read file content up to 64 kB as an utf-8 encoded string.

Truncate content for larger files.

Parameters:: f (IO[bytes])
Return type:: str

cwl_utils.parser.cwl_v1_0_utils.convert_stdstreams_to_files(clt)

Convert stdout and stderr type shortcuts to files.

Parameters:: clt (cwl_utils.parser.cwl_v1_0.CommandLineTool)
Return type:: None

cwl_utils.parser.cwl_v1_0_utils.load_inputfile(doc, baseuri=None, loadingOptions=None)

Load a CWL v1.0 input file from a serialized YAML string or a YAML object.

Parameters:

doc (Any)
baseuri (Optional[str])
loadingOptions (Optional[cwl_utils.parser.cwl_v1_0.LoadingOptions])

Return type:

Any

cwl_utils.parser.cwl_v1_0_utils.load_inputfile_by_string(string, uri, loadingOptions=None)

Load a CWL v1.0 input file from a serialized YAML string.

Parameters:

string (Any)
uri (str)
loadingOptions (Optional[cwl_utils.parser.cwl_v1_0.LoadingOptions])

Return type:

Any

cwl_utils.parser.cwl_v1_0_utils.load_inputfile_by_yaml(yaml, uri, loadingOptions=None)

Load a CWL v1.0 input file from a YAML object.

Parameters:

yaml (Any)
uri (str)
loadingOptions (Optional[cwl_utils.parser.cwl_v1_0.LoadingOptions])

Return type:

Any

cwl_utils.parser.cwl_v1_0_utils.merge_flatten_type(src)

Return the merge flattened type of the source type.

Parameters:: src (Any)
Return type:: Any

cwl_utils.parser.cwl_v1_0_utils.type_for_step_input(step, in_)

Determine the type for the given step input.

Parameters:

step (cwl_utils.parser.cwl_v1_0.WorkflowStep)
in_ (cwl_utils.parser.cwl_v1_0.WorkflowStepInput)

Return type:

Any

cwl_utils.parser.cwl_v1_0_utils.type_for_step_output(step, sourcename)

Determine the type for the given step output.

Parameters:

step (cwl_utils.parser.cwl_v1_0.WorkflowStep)
sourcename (str)

Return type:

Any

cwl_utils.parser.cwl_v1_0_utils.type_for_source(process, sourcenames, parent=None, linkMerge=None)

Determine the type for the given sourcenames.

Parameters:

process (Union[cwl_utils.parser.cwl_v1_0.CommandLineTool, cwl_utils.parser.cwl_v1_0.Workflow, cwl_utils.parser.cwl_v1_0.ExpressionTool])
sourcenames (Union[str, list[str]])
parent (Optional[cwl_utils.parser.cwl_v1_0.Workflow])
linkMerge (Optional[str])

Return type:

Any

cwl_utils.parser.cwl_v1_0_utils.param_for_source_id(process, sourcenames, parent=None, scatter_context=None)

Find the process input parameter that matches one of the given sourcenames.

Parameters:

process (Union[cwl_utils.parser.cwl_v1_0.CommandLineTool, cwl_utils.parser.cwl_v1_0.Workflow, cwl_utils.parser.cwl_v1_0.ExpressionTool])
sourcenames (Union[str, list[str]])
parent (Optional[cwl_utils.parser.cwl_v1_0.Workflow])
scatter_context (Optional[list[Optional[tuple[int, str]]]])

Return type:

Union[list[cwl_utils.parser.cwl_v1_0.InputParameter], cwl_utils.parser.cwl_v1_0.InputParameter]