Mercurial > repos > shellac > guppy_basecaller

comparison env/lib/python3.7/site-packages/planemo/cwl/script.py @ 5:9b1c78e6ba9c draft default tip

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload commit 6c0a8142489327ece472c84e558c47da711a9142"
author shellac
date Mon, 01 Jun 2020 08:59:25 -0400
parents 79f47841a781
children
comparison
equal deleted inserted replaced
4:79f47841a781 5:9b1c78e6ba9c
1 import copy
2 import os
3
4 try:
5 import schema_salad
6 except (ImportError, SyntaxError):
7 # TODO: implement support for Python 2.6 and 3.X
8 schema_salad = None
9
10 try:
11 import cwltool
12 except (ImportError, SyntaxError):
13 cwltool = None
14
15 try:
16 from cwltool.main import load_tool
17 except (ImportError, SyntaxError):
18 load_tool = None
19
20 try:
21 from cwltool import process
22 except (ImportError, SyntaxError):
23 process = None
24
25 try:
26 from .cwl2script import cwl2script
27 except (ImportError, SyntaxError):
28 cwl2script = None
29
30
31 def to_script(ctx, path, job_path, **kwds):
32 if schema_salad is None:
33 raise Exception("This functionality requires schema_salad and Python 2.7.")
34
35 if cwltool is None:
36 raise Exception("This functionality requires cwltool and Python 2.7.")
37
38 uri = "file://" + os.path.abspath(job_path)
39
40 loader = schema_salad.ref_resolver.Loader({
41 "@base": uri,
42 "path": {
43 "@type": "@id"
44 }
45 })
46
47 job, _ = loader.resolve_ref(uri)
48
49 t = load_tool(path, False, False, cwltool.workflow.defaultMakeTool, True)
50
51 if type(t) == int:
52 return t
53
54 process.checkRequirements(t.tool, cwl2script.supportedProcessRequirements)
55
56 for inp in t.tool["inputs"]:
57 if process.shortname(inp["id"]) in job:
58 pass
59 elif process.shortname(inp["id"]) not in job and "default" in inp:
60 job[process.shortname(inp["id"])] = copy.copy(inp["default"])
61 elif process.shortname(inp["id"]) not in job and inp["type"][0] == "null":
62 pass
63 else:
64 raise Exception("Missing inputs `%s`" % process.shortname(inp["id"]))
65
66 if not kwds.get("basedir", None):
67 kwds["basedir"] = os.path.dirname(os.path.abspath(job_path))
68
69 outdir = kwds.get("outdir")
70
71 if t.tool["class"] == "Workflow":
72 print(cwl2script.generateScriptForWorkflow(t, job, outdir))
73 elif t.tool["class"] == "CommandLineTool":
74 print(cwl2script.generateScriptForTool(t, job, outdir))
75
76 return 0