weaver.processes.wps_workflow
Module Contents
- weaver.processes.wps_workflow.default_make_tool(toolpath_object: weaver.typedefs.CWL_ToolPathObject, loading_context: cwltool.context.LoadingContext, get_job_process_definition: weaver.typedefs.JobProcessDefinitionCallback) cwltool.process.Process [source]
Generate the tool class object from the CWL definition to handle its execution.
Warning
Package
cwltool
introduces explicit typing definitions withmypy_extensions
. This can causeTypeError("interpreted classes cannot inherit from compiled")
when usingcwltool.process.Process
as base class for our custom definitions below. To avoid the error, we must enforce the type usingcast()
.
- class weaver.processes.wps_workflow.WpsWorkflow(toolpath_object: weaver.typedefs.CWL_ToolPathObject, loading_context: cwltool.context.LoadingContext, get_job_process_definition: weaver.typedefs.JobProcessDefinitionCallback)[source]
Definition of a CWL
workflow
that can executeWPS
application packages as intermediate job steps.Steps are expected to be defined as individual
weaver.processes.wps_package.WpsPackage
references.Initialize this CommandLineTool.
- job(job_order: cwltool.utils.CWLObjectType, output_callbacks: Callable[[Any, Any], Any], runtime_context: cwltool.context.RuntimeContext) cwltool.utils.JobsGeneratorType [source]
Workflow job generator.
- Parameters:
job_order – inputs of the job submission
output_callbacks – method to fetch step outputs and corresponding step details
runtime_context – configs about execution environment
- Returns:
- collect_output(schema: cwltool.utils.CWLObjectType, builder: cwltool.builder.Builder, outdir: str, fs_access: cwltool.stdfsaccess.StdFsAccess, compute_checksum: bool = True) cwltool.utils.CWLOutputType | None [source]
Collect outputs from the step Process following its execution.
Note
Because the staging operation following remote Process execution nests each output under a directory name matching respective output IDs, globs must be update with that modified nested directory as well.
- class weaver.processes.wps_workflow.WpsWorkflowJob(builder: cwltool.builder.Builder, job_order: cwltool.utils.CWLObjectType, make_path_mapper: Callable[Ellipsis, cwltool.pathmapper.PathMapper], requirements: weaver.typedefs.CWL_RequirementsList, hints: weaver.typedefs.CWL_RequirementsList, name: str, wps_process: weaver.processes.wps_process_base.WpsProcessInterface, expected_outputs: List[weaver.typedefs.CWL_Output_Type])[source]
Base class for get_requirement().
Initialize the job object.