Mercurial > repos > shellac > sam_consensus_v3
comparison env/lib/python3.9/site-packages/ephemeris/get_tool_list_from_galaxy.py @ 0:4f3585e2f14b draft default tip
"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"
author | shellac |
---|---|
date | Mon, 22 Mar 2021 18:12:50 +0000 |
parents | |
children |
comparison
equal
deleted
inserted
replaced
-1:000000000000 | 0:4f3585e2f14b |
---|---|
1 #!/usr/bin/env python | |
2 """Tool to extract a tool list from galaxy.""" | |
3 | |
4 from argparse import ArgumentDefaultsHelpFormatter | |
5 from argparse import ArgumentParser | |
6 from distutils.version import StrictVersion | |
7 | |
8 import yaml | |
9 from bioblend.galaxy.tools import ToolClient | |
10 from bioblend.galaxy.toolshed import ToolShedClient | |
11 | |
12 from . import get_galaxy_connection | |
13 from .common_parser import get_common_args | |
14 from .shed_tools_methods import format_tool_shed_url | |
15 | |
16 | |
17 def get_tool_panel(gi): | |
18 tool_client = ToolClient(gi) | |
19 return tool_client.get_tool_panel() | |
20 | |
21 | |
22 def tools_for_repository(gi, repository): | |
23 tool_shed_url = repository.get('tool_shed_url') | |
24 name = repository['name'] | |
25 owner = repository['owner'] | |
26 changeset_revision = repository.get('changeset_revision') | |
27 | |
28 tools = [] | |
29 | |
30 def handle_tool(tool_elem): | |
31 if not tool_elem.get('tool_shed_repository', None): | |
32 return | |
33 tsr = tool_elem['tool_shed_repository'] | |
34 if tsr['name'] != name or tsr['owner'] != owner: | |
35 return | |
36 | |
37 if tool_shed_url and format_tool_shed_url(tsr['tool_shed']) != format_tool_shed_url(tool_shed_url): | |
38 return | |
39 | |
40 if changeset_revision and changeset_revision != tsr["changeset_revision"]: | |
41 return | |
42 | |
43 tools.append(tool_elem) | |
44 | |
45 walk_tools(get_tool_panel(gi), handle_tool) | |
46 | |
47 return tools | |
48 | |
49 | |
50 def walk_tools(tool_panel, f): | |
51 for elem in tool_panel: | |
52 if elem['model_class'] == 'Tool': | |
53 f(elem) | |
54 elif elem['model_class'] == 'ToolSection': | |
55 walk_tools(elem.get("elems", []), f) | |
56 | |
57 | |
58 class GiToToolYaml: | |
59 def __init__(self, gi, | |
60 include_tool_panel_section_id=False, | |
61 skip_tool_panel_section_name=True, | |
62 skip_changeset_revision=False, | |
63 get_data_managers=False, | |
64 get_all_tools=False): | |
65 self.gi = gi | |
66 | |
67 self.include_tool_panel_section_id = include_tool_panel_section_id | |
68 self.skip_tool_panel_section_name = skip_tool_panel_section_name | |
69 self.skip_changeset_revision = skip_changeset_revision | |
70 self.get_data_managers = get_data_managers | |
71 self.get_all_tools = get_all_tools | |
72 | |
73 @property | |
74 def toolbox(self): | |
75 """ | |
76 Gets the toolbox elements from <galaxy_url>/api/tools | |
77 """ | |
78 return get_tool_panel(self.gi) | |
79 | |
80 @property | |
81 def installed_tool_list(self): | |
82 """ | |
83 gets a tool list from the toolclient | |
84 :return: | |
85 """ | |
86 tool_client = ToolClient(self.gi) | |
87 return tool_client.get_tools() | |
88 | |
89 @property | |
90 def repository_list(self): | |
91 """ | |
92 Toolbox elements returned by api/tools may be of class ToolSection or Tool. | |
93 Parse these accordingly to get a list of repositories. | |
94 """ | |
95 repositories = [] | |
96 | |
97 def record_repo(tool_elem): | |
98 repo = get_repo_from_tool(tool_elem) | |
99 if repo: | |
100 repositories.append(repo) | |
101 | |
102 walk_tools(self.toolbox, record_repo) | |
103 | |
104 if self.get_data_managers: | |
105 for tool in self.installed_tool_list: | |
106 if tool.get("model_class") == 'DataManagerTool': | |
107 repo = get_repo_from_tool(tool) | |
108 if repo: | |
109 repositories.append(repo) | |
110 | |
111 if self.get_all_tools: | |
112 tools_with_panel = repositories[:] | |
113 tsc = ToolShedClient(self.gi) | |
114 repos = tsc.get_repositories() | |
115 # Hereafter follows a gruesomely ineffecient algorithm. | |
116 # The for loop and if statement are needed to retrieve tool_panel | |
117 # section labels and ids. | |
118 # If someone knows a more effecient way around this problem it | |
119 # will be greatly appreciated. | |
120 for repo in repos: | |
121 if not repo['deleted']: | |
122 tool_panel_section_id = None | |
123 tool_panel_section_label = None | |
124 for repo_with_panel in tools_with_panel: | |
125 if the_same_repository(repo_with_panel, repo, check_revision=False): | |
126 tool_panel_section_id = repo_with_panel.get('tool_panel_section_id') | |
127 tool_panel_section_label = repo_with_panel.get('tool_panel_section_label') | |
128 break | |
129 repositories.append( | |
130 dict(name=repo.get('name'), | |
131 owner=repo.get('owner'), | |
132 tool_shed_url=repo.get('tool_shed'), | |
133 revisions=[repo.get('changeset_revision')], | |
134 tool_panel_section_label=tool_panel_section_label, | |
135 tool_panel_section_id=tool_panel_section_id) | |
136 ) | |
137 return repositories | |
138 | |
139 @property | |
140 def tool_list(self): | |
141 repo_list = self.repository_list | |
142 repo_list = merge_repository_changeset_revisions(repo_list) | |
143 repo_list = self.filter_section_name_or_id_or_changeset(repo_list) | |
144 return {"tools": repo_list} | |
145 | |
146 def filter_section_name_or_id_or_changeset(self, repository_list): | |
147 new_repo_list = [] | |
148 for repo in repository_list: | |
149 if self.skip_tool_panel_section_name: | |
150 del repo['tool_panel_section_label'] | |
151 if not self.include_tool_panel_section_id: | |
152 del repo['tool_panel_section_id'] | |
153 if self.skip_changeset_revision: | |
154 del repo['revisions'] | |
155 new_repo_list.append(repo) | |
156 return new_repo_list | |
157 | |
158 def write_to_yaml(self, output_file): | |
159 with open(output_file, "w") as output: | |
160 output.write(yaml.safe_dump(self.tool_list, default_flow_style=False)) | |
161 | |
162 | |
163 def the_same_repository(repo_1_info, repo_2_info, check_revision=True): | |
164 """ | |
165 Given two dicts containing info about repositories, determine if they are the same | |
166 repository. | |
167 Each of the dicts must have the following keys: `changeset_revisions`( if check revisions is true), `name`, `owner`, and | |
168 (either `tool_shed` or `tool_shed_url`). | |
169 """ | |
170 # Sort from most unique to least unique for fast comparison. | |
171 if not check_revision or repo_1_info.get('changeset_revision') == repo_2_info.get('changeset_revision'): | |
172 if repo_1_info.get('name') == repo_2_info.get('name'): | |
173 if repo_1_info.get('owner') == repo_2_info.get('owner'): | |
174 t1ts = repo_1_info.get('tool_shed', repo_1_info.get('tool_shed_url', None)) | |
175 t2ts = repo_2_info.get('tool_shed', repo_2_info.get('tool_shed_url', None)) | |
176 if t1ts in t2ts or t2ts in t1ts: | |
177 return True | |
178 return False | |
179 | |
180 | |
181 def merge_repository_changeset_revisions(repository_list): | |
182 """ | |
183 Each installed changeset revision of a tool is listed individually. | |
184 Merge revisions of the same tool into a list. | |
185 """ | |
186 repositories = {} | |
187 repo_key_template = "{tool_shed_url}|{name}|{owner}|{tool_panel_section_id}|{tool_panel_section_label}" | |
188 for repo in repository_list: | |
189 repo_key = repo_key_template.format(**repo) | |
190 if repo_key in repositories: | |
191 repositories[repo_key].extend(repo['revisions']) | |
192 else: | |
193 repositories[repo_key] = repo['revisions'] | |
194 new_repository_list = [] | |
195 for repo_key, changeset_revisions in repositories.items(): | |
196 changeset_revisions = list(set(changeset_revisions)) | |
197 tool_shed_url, name, owner, tool_panel_section_id, tool_panel_section_label = repo_key.split('|') | |
198 new_repository_list.append( | |
199 {'tool_shed_url': tool_shed_url, | |
200 'name': name, | |
201 'owner': owner, | |
202 'tool_panel_section_id': tool_panel_section_id, | |
203 'tool_panel_section_label': tool_panel_section_label, | |
204 'revisions': changeset_revisions} | |
205 ) | |
206 return new_repository_list | |
207 | |
208 | |
209 def _parser(): | |
210 """Creates the parser object.""" | |
211 parent = get_common_args(login_required=True) | |
212 parser = ArgumentParser(parents=[parent], | |
213 formatter_class=ArgumentDefaultsHelpFormatter) | |
214 parser.add_argument("-o", "--output-file", | |
215 required=True, | |
216 dest="output", | |
217 help="tool_list.yml output file") | |
218 parser.add_argument("--include_tool_panel_id", | |
219 action="store_true", | |
220 help="Include tool_panel_id in tool_list.yml ? " | |
221 "Use this only if the tool panel id already exists. See " | |
222 "https://github.com/galaxyproject/ansible-galaxy-tools/blob/master/files/tool_list.yaml.sample") | |
223 parser.add_argument("--skip_tool_panel_name", | |
224 action="store_true", | |
225 help="Do not include tool_panel_name in tool_list.yml ?") | |
226 parser.add_argument("--skip_changeset_revision", | |
227 action="store_true", | |
228 help="Do not include the changeset revision when generating the tool list." | |
229 "Use this if you would like to use the list to update all the tools in" | |
230 "your galaxy instance using shed-install." | |
231 ) | |
232 parser.add_argument("--get_data_managers", | |
233 action="store_true", | |
234 help="Include the data managers in the tool list. Requires admin login details") | |
235 parser.add_argument("--get_all_tools", | |
236 action="store_true", | |
237 help="Get all tools and revisions, not just those which are present on the web ui." | |
238 "Requires login details.") | |
239 return parser | |
240 | |
241 | |
242 def get_repo_from_tool(tool): | |
243 """ | |
244 Get the minimum items required for re-installing a (list of) tools | |
245 """ | |
246 if not tool.get('tool_shed_repository', None): | |
247 # Tool or Data Manager not installed from a tool shed | |
248 return {} | |
249 tsr = tool['tool_shed_repository'] | |
250 repo = {'name': tsr['name'], | |
251 'owner': tsr['owner'], | |
252 'tool_shed_url': tsr['tool_shed'], | |
253 'revisions': [tsr['changeset_revision']], | |
254 'tool_panel_section_id': tool['panel_section_id'], | |
255 'tool_panel_section_label': tool['panel_section_name']} | |
256 return repo | |
257 | |
258 | |
259 def _parse_cli_options(): | |
260 """ | |
261 Parse command line options, returning `parse_args` from `ArgumentParser`. | |
262 """ | |
263 parser = _parser() | |
264 return parser.parse_args() | |
265 | |
266 | |
267 def check_galaxy_version(gi): | |
268 version = gi.config.get_version() | |
269 if StrictVersion(version['version_major']) < StrictVersion('16.04'): | |
270 raise Exception('This script needs galaxy version 16.04 or newer') | |
271 | |
272 | |
273 def main(): | |
274 options = _parse_cli_options() | |
275 gi = get_galaxy_connection(options, login_required=False) | |
276 check_galaxy_version(gi) | |
277 gi_to_tool_yaml = GiToToolYaml( | |
278 gi=gi, | |
279 include_tool_panel_section_id=options.include_tool_panel_id, | |
280 skip_tool_panel_section_name=options.skip_tool_panel_name, | |
281 skip_changeset_revision=options.skip_changeset_revision, | |
282 get_data_managers=options.get_data_managers, | |
283 get_all_tools=options.get_all_tools) | |
284 gi_to_tool_yaml.write_to_yaml(options.output) | |
285 | |
286 | |
287 if __name__ == "__main__": | |
288 main() |