Mercurial > repos > shellac > guppy_basecaller

comparison env/lib/python3.7/site-packages/planemo/cwl/script.py @ 2:6af9afd405e9 draft

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload commit 0a63dd5f4d38a1f6944587f52a8cd79874177fc1"
author shellac
date Thu, 14 May 2020 14:56:58 -0400
parents 26e78fe6e8c4
children
comparison
equal deleted inserted replaced
1:75ca89e9b81c 2:6af9afd405e9
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