Mercurial > repos > shellac > guppy_basecaller

comparison env/lib/python3.7/site-packages/cwltool/expression.py @ 0:26e78fe6e8c4 draft

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload commit c699937486c35866861690329de38ec1a5d9f783"
author shellac
date Sat, 02 May 2020 07:14:21 -0400
parents
children
comparison
equal deleted inserted replaced
-1:000000000000 0:26e78fe6e8c4
1 """Parse CWL expressions."""
2 from __future__ import absolute_import
3
4 import copy
5 import re
6 from typing import (Any, Dict, List, Mapping, MutableMapping, MutableSequence, Optional,
7 Union)
8
9 import six
10 from six import string_types, u
11 from future.utils import raise_from
12 from typing_extensions import Text # pylint: disable=unused-import
13 # move to a regular typing import when Python 3.3-3.6 is no longer supported
14
15 from .sandboxjs import default_timeout, execjs, JavascriptException
16 from .errors import WorkflowException
17 from .utils import bytes2str_in_dicts, docker_windows_path_adjust, json_dumps
18
19
20 def jshead(engine_config, rootvars):
21 # type: (List[Text], Dict[Text, Any]) -> Text
22
23 # make sure all the byte strings are converted
24 # to str in `rootvars` dict.
25
26 return u"\n".join(
27 engine_config + [u"var {} = {};".format(k, json_dumps(v, indent=4))
28 for k, v in rootvars.items()])
29
30
31 # decode all raw strings to unicode
32 seg_symbol = r"""\w+"""
33 seg_single = r"""\['([^']|\\')+'\]"""
34 seg_double = r"""\["([^"]|\\")+"\]"""
35 seg_index = r"""\[[0-9]+\]"""
36 segments = r"(\.%s|%s|%s|%s)" % (seg_symbol, seg_single, seg_double, seg_index)
37 segment_re = re.compile(u(segments), flags=re.UNICODE)
38 param_str = r"\((%s)%s*\)$" % (seg_symbol, segments)
39 param_re = re.compile(u(param_str), flags=re.UNICODE)
40
41 JSON = Union[Dict[Any, Any], List[Any], Text, int, float, bool, None]
42
43
44 class SubstitutionError(Exception):
45 pass
46
47
48 def scanner(scan): # type: (Text) -> List[int]
49 DEFAULT = 0
50 DOLLAR = 1
51 PAREN = 2
52 BRACE = 3
53 SINGLE_QUOTE = 4
54 DOUBLE_QUOTE = 5
55 BACKSLASH = 6
56
57 i = 0
58 stack = [DEFAULT]
59 start = 0
60 while i < len(scan):
61 state = stack[-1]
62 c = scan[i]
63
64 if state == DEFAULT:
65 if c == '$':
66 stack.append(DOLLAR)
67 elif c == '\\':
68 stack.append(BACKSLASH)
69 elif state == BACKSLASH:
70 stack.pop()
71 if stack[-1] == DEFAULT:
72 return [i - 1, i + 1]
73 elif state == DOLLAR:
74 if c == '(':
75 start = i - 1
76 stack.append(PAREN)
77 elif c == '{':
78 start = i - 1
79 stack.append(BRACE)
80 else:
81 stack.pop()
82 elif state == PAREN:
83 if c == '(':
84 stack.append(PAREN)
85 elif c == ')':
86 stack.pop()
87 if stack[-1] == DOLLAR:
88 return [start, i + 1]
89 elif c == "'":
90 stack.append(SINGLE_QUOTE)
91 elif c == '"':
92 stack.append(DOUBLE_QUOTE)
93 elif state == BRACE:
94 if c == '{':
95 stack.append(BRACE)
96 elif c == '}':
97 stack.pop()
98 if stack[-1] == DOLLAR:
99 return [start, i + 1]
100 elif c == "'":
101 stack.append(SINGLE_QUOTE)
102 elif c == '"':
103 stack.append(DOUBLE_QUOTE)
104 elif state == SINGLE_QUOTE:
105 if c == "'":
106 stack.pop()
107 elif c == '\\':
108 stack.append(BACKSLASH)
109 elif state == DOUBLE_QUOTE:
110 if c == '"':
111 stack.pop()
112 elif c == '\\':
113 stack.append(BACKSLASH)
114 i += 1
115
116 if len(stack) > 1:
117 raise SubstitutionError(
118 "Substitution error, unfinished block starting at position {}: {}".format(start, scan[start:]))
119 else:
120 return []
121
122
123 def next_seg(parsed_string, remaining_string, current_value): # type: (Text, Text, JSON) -> JSON
124 if remaining_string:
125 m = segment_re.match(remaining_string)
126 if not m:
127 return current_value
128 next_segment_str = m.group(0)
129
130 key = None # type: Optional[Union[Text, int]]
131 if next_segment_str[0] == '.':
132 key = next_segment_str[1:]
133 elif next_segment_str[1] in ("'", '"'):
134 key = next_segment_str[2:-2].replace("\\'", "'").replace('\\"', '"')
135
136 if key is not None:
137 if isinstance(current_value, MutableSequence) and key == "length" and not remaining_string[m.end(0):]:
138 return len(current_value)
139 if not isinstance(current_value, MutableMapping):
140 raise WorkflowException("%s is a %s, cannot index on string '%s'" % (parsed_string, type(current_value).__name__, key))
141 if key not in current_value:
142 raise WorkflowException("%s does not contain key '%s'" % (parsed_string, key))
143 else:
144 try:
145 key = int(next_segment_str[1:-1])
146 except ValueError as v:
147 raise_from(WorkflowException(u(str(v))), v)
148 if not isinstance(current_value, MutableSequence):
149 raise WorkflowException("%s is a %s, cannot index on int '%s'" % (parsed_string, type(current_value).__name__, key))
150 if key >= len(current_value):
151 raise WorkflowException("%s list index %i out of range" % (parsed_string, key))
152
153 if isinstance(current_value, Mapping):
154 try:
155 return next_seg(parsed_string + remaining_string, remaining_string[m.end(0):], current_value[key])
156 except KeyError:
157 raise WorkflowException("%s doesn't have property %s" % (parsed_string, key))
158 elif isinstance(current_value, list) and isinstance(key, int):
159 try:
160 return next_seg(parsed_string + remaining_string, remaining_string[m.end(0):], current_value[key])
161 except KeyError:
162 raise WorkflowException("%s doesn't have property %s" % (parsed_string, key))
163 else:
164 raise WorkflowException("%s doesn't have property %s" % (parsed_string, key))
165 else:
166 return current_value
167
168
169 def evaluator(ex, # type: Text
170 jslib, # type: Text
171 obj, # type: Dict[Text, Any]
172 timeout, # type: float
173 fullJS=False, # type: bool
174 force_docker_pull=False, # type: bool
175 debug=False, # type: bool
176 js_console=False # type: bool
177 ):
178 # type: (...) -> JSON
179 match = param_re.match(ex)
180
181 expression_parse_exception = None
182 expression_parse_succeeded = False
183
184 if match is not None:
185 first_symbol = match.group(1)
186 first_symbol_end = match.end(1)
187
188 if first_symbol_end + 1 == len(ex) and first_symbol == "null":
189 return None
190 try:
191 if obj.get(first_symbol) is None:
192 raise WorkflowException("%s is not defined" % first_symbol)
193
194 return next_seg(first_symbol, ex[first_symbol_end:-1], obj[first_symbol])
195 except WorkflowException as werr:
196 expression_parse_exception = werr
197 else:
198 expression_parse_succeeded = True
199
200 if fullJS and not expression_parse_succeeded:
201 return execjs(
202 ex, jslib, timeout, force_docker_pull=force_docker_pull,
203 debug=debug, js_console=js_console)
204 else:
205 if expression_parse_exception is not None:
206 raise JavascriptException(
207 "Syntax error in parameter reference '%s': %s. This could be "
208 "due to using Javascript code without specifying "
209 "InlineJavascriptRequirement." % \
210 (ex[1:-1], expression_parse_exception))
211 else:
212 raise JavascriptException(
213 "Syntax error in parameter reference '%s'. This could be due "
214 "to using Javascript code without specifying "
215 "InlineJavascriptRequirement." % ex)
216
217
218 def interpolate(scan, # type: Text
219 rootvars, # type: Dict[Text, Any]
220 timeout=default_timeout, # type: float
221 fullJS=False, # type: bool
222 jslib="", # type: Text
223 force_docker_pull=False, # type: bool
224 debug=False, # type: bool
225 js_console=False, # type: bool
226 strip_whitespace=True # type: bool
227 ): # type: (...) -> JSON
228 if strip_whitespace:
229 scan = scan.strip()
230 parts = []
231 w = scanner(scan)
232 while w:
233 parts.append(scan[0:w[0]])
234
235 if scan[w[0]] == '$':
236 e = evaluator(scan[w[0] + 1:w[1]], jslib, rootvars, timeout,
237 fullJS=fullJS, force_docker_pull=force_docker_pull,
238 debug=debug, js_console=js_console)
239 if w[0] == 0 and w[1] == len(scan) and len(parts) <= 1:
240 return e
241 leaf = json_dumps(e, sort_keys=True)
242 if leaf[0] == '"':
243 leaf = leaf[1:-1]
244 parts.append(leaf)
245 elif scan[w[0]] == '\\':
246 e = scan[w[1] - 1]
247 parts.append(e)
248
249 scan = scan[w[1]:]
250 w = scanner(scan)
251 parts.append(scan)
252 return ''.join(parts)
253
254 def needs_parsing(snippet): # type: (Any) -> bool
255 return isinstance(snippet, string_types) \
256 and ("$(" in snippet or "${" in snippet)
257
258 def do_eval(ex, # type: Union[Text, Dict[Text, Text]]
259 jobinput, # type: Dict[Text, JSON]
260 requirements, # type: List[Dict[Text, Any]]
261 outdir, # type: Optional[Text]
262 tmpdir, # type: Optional[Text]
263 resources, # type: Dict[str, int]
264 context=None, # type: Any
265 timeout=default_timeout, # type: float
266 force_docker_pull=False, # type: bool
267 debug=False, # type: bool
268 js_console=False, # type: bool
269 strip_whitespace=True # type: bool
270 ): # type: (...) -> Any
271
272 runtime = copy.deepcopy(resources) # type: Dict[str, Any]
273 runtime["tmpdir"] = docker_windows_path_adjust(tmpdir) if tmpdir else None
274 runtime["outdir"] = docker_windows_path_adjust(outdir) if outdir else None
275
276 rootvars = {
277 u"inputs": jobinput,
278 u"self": context,
279 u"runtime": runtime}
280
281 # TODO: need to make sure the `rootvars dict`
282 # contains no bytes type in the first place.
283 if six.PY3:
284 rootvars = bytes2str_in_dicts(rootvars) # type: ignore
285
286 if isinstance(ex, string_types) and needs_parsing(ex):
287 fullJS = False
288 jslib = u""
289 for r in reversed(requirements):
290 if r["class"] == "InlineJavascriptRequirement":
291 fullJS = True
292 jslib = jshead(r.get("expressionLib", []), rootvars)
293 break
294
295 try:
296 return interpolate(ex,
297 rootvars,
298 timeout=timeout,
299 fullJS=fullJS,
300 jslib=jslib,
301 force_docker_pull=force_docker_pull,
302 debug=debug,
303 js_console=js_console,
304 strip_whitespace=strip_whitespace)
305
306 except Exception as e:
307 raise_from(WorkflowException("Expression evaluation error:\n%s" % Text(e)), e)
308 else:
309 return ex