Mercurial > repos > guerler > springsuite

comparison planemo/lib/python3.7/site-packages/galaxy/tool_util/parser/yaml.py @ 1:56ad4e20f292 draft

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload commit 6eee67778febed82ddd413c3ca40b3183a3898f1"
author guerler
date Fri, 31 Jul 2020 00:32:28 -0400
parents
children
comparison
equal deleted inserted replaced
0:d30785e31577 1:56ad4e20f292
1 from collections import OrderedDict
2
3 import packaging.version
4
5 from galaxy.tool_util.deps import requirements
6 from galaxy.tool_util.parser.util import (
7 DEFAULT_DELTA,
8 DEFAULT_DELTA_FRAC
9 )
10 from .interface import (
11 InputSource,
12 PageSource,
13 PagesSource,
14 ToolSource,
15 )
16 from .output_collection_def import dataset_collector_descriptions_from_output_dict
17 from .output_objects import (
18 ToolOutput,
19 ToolOutputCollection,
20 ToolOutputCollectionStructure,
21 )
22 from .stdio import error_on_exit_code
23 from .util import is_dict
24
25
26 class YamlToolSource(ToolSource):
27
28 def __init__(self, root_dict, source_path=None):
29 self.root_dict = root_dict
30 self._source_path = source_path
31 self._macro_paths = []
32
33 @property
34 def source_path(self):
35 return self._source_path
36
37 def parse_id(self):
38 return self.root_dict.get("id")
39
40 def parse_version(self):
41 return self.root_dict.get("version")
42
43 def parse_name(self):
44 return self.root_dict.get("name")
45
46 def parse_description(self):
47 return self.root_dict.get("description", "")
48
49 def parse_edam_operations(self):
50 return self.root_dict.get("edam_operations", [])
51
52 def parse_edam_topics(self):
53 return self.root_dict.get("edam_topics", [])
54
55 def parse_xrefs(self):
56 xrefs = self.root_dict.get("xrefs", [])
57 return [dict(value=xref["value"], reftype=xref["type"]) for xref in xrefs if xref["type"]]
58
59 def parse_is_multi_byte(self):
60 return self.root_dict.get("is_multi_byte", self.default_is_multi_byte)
61
62 def parse_sanitize(self):
63 return self.root_dict.get("sanitize", True)
64
65 def parse_display_interface(self, default):
66 return self.root_dict.get('display_interface', default)
67
68 def parse_require_login(self, default):
69 return self.root_dict.get('require_login', default)
70
71 def parse_command(self):
72 return self.root_dict.get("command")
73
74 def parse_expression(self):
75 return self.root_dict.get("expression")
76
77 def parse_environment_variables(self):
78 return []
79
80 def parse_interpreter(self):
81 return self.root_dict.get("interpreter")
82
83 def parse_version_command(self):
84 return self.root_dict.get("runtime_version", {}).get("command", None)
85
86 def parse_version_command_interpreter(self):
87 return self.root_dict.get("runtime_version", {}).get("interpreter", None)
88
89 def parse_requirements_and_containers(self):
90 return requirements.parse_requirements_from_dict(self.root_dict)
91
92 def parse_input_pages(self):
93 # All YAML tools have only one page (feature is deprecated)
94 page_source = YamlPageSource(self.root_dict.get("inputs", {}))
95 return PagesSource([page_source])
96
97 def parse_strict_shell(self):
98 # TODO: Add ability to disable this.
99 return True
100
101 def parse_stdio(self):
102 return error_on_exit_code()
103
104 def parse_help(self):
105 return self.root_dict.get("help", None)
106
107 def parse_outputs(self, tool):
108 outputs = self.root_dict.get("outputs", {})
109 output_defs = []
110 output_collection_defs = []
111 for name, output_dict in outputs.items():
112 output_type = output_dict.get("type", "data")
113 if output_type == "data":
114 output_defs.append(self._parse_output(tool, name, output_dict))
115 elif output_type == "collection":
116 output_collection_defs.append(self._parse_output_collection(tool, name, output_dict))
117 else:
118 message = "Unknown output_type [%s] encountered." % output_type
119 raise Exception(message)
120 outputs = OrderedDict()
121 for output in output_defs:
122 outputs[output.name] = output
123 output_collections = OrderedDict()
124 for output in output_collection_defs:
125 output_collections[output.name] = output
126
127 return outputs, output_collections
128
129 def _parse_output(self, tool, name, output_dict):
130 output = ToolOutput.from_dict(name, output_dict, tool=tool)
131 return output
132
133 def _parse_output_collection(self, tool, name, output_dict):
134 name = output_dict.get("name")
135 label = output_dict.get("label")
136 default_format = output_dict.get("format", "data")
137 collection_type = output_dict.get("type", None)
138 collection_type_source = output_dict.get("type_source", None)
139 structured_like = output_dict.get("structured_like", None)
140 inherit_format = False
141 inherit_metadata = False
142 if structured_like:
143 inherit_format = output_dict.get("inherit_format", None)
144 inherit_metadata = output_dict.get("inherit_metadata", None)
145 default_format_source = output_dict.get("format_source", None)
146 default_metadata_source = output_dict.get("metadata_source", "")
147 filters = []
148 dataset_collector_descriptions = dataset_collector_descriptions_from_output_dict(output_dict)
149
150 structure = ToolOutputCollectionStructure(
151 collection_type=collection_type,
152 collection_type_source=collection_type_source,
153 structured_like=structured_like,
154 dataset_collector_descriptions=dataset_collector_descriptions,
155 )
156 output_collection = ToolOutputCollection(
157 name,
158 structure,
159 label=label,
160 filters=filters,
161 default_format=default_format,
162 inherit_format=inherit_format,
163 inherit_metadata=inherit_metadata,
164 default_format_source=default_format_source,
165 default_metadata_source=default_metadata_source,
166 )
167 return output_collection
168
169 def parse_tests_to_dict(self):
170 tests = []
171 rval = dict(
172 tests=tests
173 )
174
175 for i, test_dict in enumerate(self.root_dict.get("tests", [])):
176 tests.append(_parse_test(i, test_dict))
177
178 return rval
179
180 def parse_profile(self):
181 return self.root_dict.get("profile", "16.04")
182
183 def parse_interactivetool(self):
184 return self.root_dict.get("entry_points", [])
185
186 def parse_python_template_version(self):
187 python_template_version = self.root_dict.get("python_template_version", None)
188 if python_template_version is not None:
189 python_template_version = packaging.version.parse(python_template_version)
190 return python_template_version
191
192
193 def _parse_test(i, test_dict):
194 inputs = test_dict["inputs"]
195 if is_dict(inputs):
196 new_inputs = []
197 for key, value in inputs.items():
198 new_inputs.append({"name": key, "value": value, "attributes": {}})
199 test_dict["inputs"] = new_inputs
200
201 outputs = test_dict["outputs"]
202
203 new_outputs = []
204 if is_dict(outputs):
205 for key, value in outputs.items():
206 if is_dict(value):
207 attributes = value
208 file = attributes.get("file")
209 else:
210 file = value
211 attributes = {}
212 new_outputs.append({
213 "name": key,
214 "value": file,
215 "attributes": attributes
216 })
217 else:
218 for output in outputs:
219 name = output["name"]
220 value = output.get("file", None)
221 attributes = output
222 new_outputs.append((name, value, attributes))
223
224 for output in new_outputs:
225 attributes = output["attributes"]
226 defaults = {
227 'compare': 'diff',
228 'lines_diff': 0,
229 'delta': DEFAULT_DELTA,
230 'delta_frac': DEFAULT_DELTA_FRAC,
231 'sort': False,
232 }
233 # TODO
234 attributes["extra_files"] = []
235 # TODO
236 attributes["metadata"] = {}
237 # TODO
238 assert_list = []
239 assert_list = __to_test_assert_list(attributes.get("asserts", []))
240 attributes["assert_list"] = assert_list
241 _ensure_has(attributes, defaults)
242
243 test_dict["outputs"] = new_outputs
244 # TODO: implement output collections for YAML tools.
245 test_dict["output_collections"] = []
246 test_dict["command"] = __to_test_assert_list(test_dict.get("command", []))
247 test_dict["stdout"] = __to_test_assert_list(test_dict.get("stdout", []))
248 test_dict["stderr"] = __to_test_assert_list(test_dict.get("stderr", []))
249 test_dict["expect_exit_code"] = test_dict.get("expect_exit_code", None)
250 test_dict["expect_failure"] = test_dict.get("expect_exit_code", False)
251 return test_dict
252
253
254 def __to_test_assert_list(assertions):
255 def expand_dict_form(item):
256 key, value = item
257 new_value = value.copy()
258 new_value["that"] = key
259 return new_value
260
261 if is_dict(assertions):
262 assertions = map(expand_dict_form, assertions.items())
263
264 assert_list = []
265 for assertion in assertions:
266 # TODO: not handling nested assertions correctly,
267 # not sure these are used though.
268 children = []
269 if "children" in assertion:
270 children = assertion["children"]
271 del assertion["children"]
272 assert_dict = dict(
273 tag=assertion["that"],
274 attributes=assertion,
275 children=children,
276 )
277 assert_list.append(assert_dict)
278
279 return assert_list or None # XML variant is None if no assertions made
280
281
282 class YamlPageSource(PageSource):
283
284 def __init__(self, inputs_list):
285 self.inputs_list = inputs_list
286
287 def parse_input_sources(self):
288 return map(YamlInputSource, self.inputs_list)
289
290
291 class YamlInputSource(InputSource):
292
293 def __init__(self, input_dict):
294 self.input_dict = input_dict
295
296 def get(self, key, default=None):
297 return self.input_dict.get(key, default)
298
299 def get_bool(self, key, default):
300 return self.input_dict.get(key, default)
301
302 def parse_input_type(self):
303 input_type = self.input_dict["type"]
304 if input_type == "repeat":
305 return "repeat"
306 elif input_type == "conditional":
307 return "conditional"
308 else:
309 return "param"
310
311 def parse_nested_inputs_source(self):
312 assert self.parse_input_type() == "repeat"
313 return YamlPageSource(self.input_dict["blocks"])
314
315 def parse_test_input_source(self):
316 test_dict = self.input_dict.get("test", None)
317 assert test_dict is not None, "conditional must contain a `test` definition"
318 return YamlInputSource(test_dict)
319
320 def parse_when_input_sources(self):
321 input_dict = self.input_dict
322
323 sources = []
324 for value, block in input_dict.get("when", {}).items():
325 if value is True:
326 value = "true"
327 elif value is False:
328 value = "false"
329 else:
330 value = str(value)
331
332 # str here to lose type information like XML, needed?
333 if not isinstance(block, list):
334 block = [block]
335 case_page_source = YamlPageSource(block)
336 sources.append((value, case_page_source))
337 return sources
338
339 def parse_static_options(self):
340 static_options = list()
341 input_dict = self.input_dict
342 for index, option in enumerate(input_dict.get("options", {})):
343 value = option.get("value")
344 label = option.get("label", value)
345 selected = option.get("selected", False)
346 static_options.append((label, value, selected))
347 return static_options
348
349
350 def _ensure_has(dict, defaults):
351 for key, value in defaults.items():
352 if key not in dict:
353 dict[key] = value