Mercurial > repos > shellac > sam_consensus_v3

comparison env/lib/python3.9/site-packages/planemo/conda.py @ 0:4f3585e2f14b draft default tip

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"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 """Planemo specific utilities for dealing with conda.
2
3 The extend galaxy-tool-util's features with planemo specific idioms.
4 """
5
6 from __future__ import absolute_import
7
8 import collections
9 import os
10 import threading
11
12 from galaxy.tool_util.deps import conda_util
13 from galaxy.util import unicodify
14
15 from planemo.exit_codes import EXIT_CODE_FAILED_DEPENDENCIES, ExitCodeException
16 from planemo.io import error, shell
17 from planemo.tools import yield_tool_sources_on_paths
18
19 MESSAGE_ERROR_FAILED_INSTALL = "Attempted to install conda and failed."
20 MESSAGE_ERROR_CANNOT_INSTALL = "Cannot install Conda - perhaps due to a failed installation or permission problems."
21 MESSAGE_ERROR_NOT_INSTALLING = "Conda not configured - run ``planemo conda_init`` or pass ``--conda_auto_init`` to continue."
22
23 BEST_PRACTICE_CHANNELS = ["conda-forge", "bioconda", "defaults"]
24
25
26 def build_conda_context(ctx, **kwds):
27 """Build a galaxy-tool-util CondaContext tailored to planemo use.
28
29 Using planemo's common command-line/global config options.
30 """
31 condarc_override_default = os.path.join(ctx.workspace, "condarc")
32 conda_prefix = kwds.get("conda_prefix", None)
33 use_planemo_shell = kwds.get("use_planemo_shell_exec", True)
34 ensure_channels = kwds.get("conda_ensure_channels", "")
35 condarc_override = kwds.get("condarc", condarc_override_default)
36 use_local = kwds.get("conda_use_local", False)
37 shell_exec = shell if use_planemo_shell else None
38 conda_context = conda_util.CondaContext(conda_prefix=conda_prefix,
39 ensure_channels=ensure_channels,
40 condarc_override=condarc_override,
41 use_local=use_local,
42 shell_exec=shell_exec)
43 handle_auto_init = kwds.get("handle_auto_init", False)
44 if handle_auto_init and not conda_context.is_installed():
45 auto_init = kwds.get("conda_auto_init", True)
46 failed = True
47 if auto_init:
48 if conda_context.can_install_conda():
49 if conda_util.install_conda(conda_context):
50 error(MESSAGE_ERROR_FAILED_INSTALL)
51 else:
52 failed = False
53 else:
54 error(MESSAGE_ERROR_CANNOT_INSTALL)
55 else:
56 error(MESSAGE_ERROR_NOT_INSTALLING)
57
58 if failed:
59 raise ExitCodeException(EXIT_CODE_FAILED_DEPENDENCIES)
60 if handle_auto_init:
61 conda_context.ensure_conda_build_installed_if_needed()
62 return conda_context
63
64
65 def collect_conda_targets(ctx, paths, recursive=False, found_tool_callback=None):
66 """Load CondaTarget objects from supplied artifact sources.
67
68 If a tool contains more than one requirement, the requirements will each
69 appear once in the output.
70 """
71 conda_targets = set([])
72 real_paths = []
73 for path in paths:
74 if not os.path.exists(path):
75 targets = target_str_to_targets(path)
76 [conda_targets.add(_) for _ in targets]
77 else:
78 real_paths.append(path)
79
80 for (tool_path, tool_source) in yield_tool_sources_on_paths(ctx, real_paths, recursive=recursive, exclude_deprecated=True):
81 if found_tool_callback:
82 found_tool_callback(tool_path)
83 for target in tool_source_conda_targets(tool_source):
84 conda_targets.add(target)
85 return conda_targets
86
87
88 # Copied and modified from mulled stuff - need to syncronize these concepts.
89 def target_str_to_targets(targets_raw):
90 def parse_target(target_str):
91 if "=" in target_str:
92 package_name, version = target_str.split("=", 1)
93 else:
94 package_name = target_str
95 version = None
96 target = conda_util.CondaTarget(package_name, version)
97 return target
98
99 targets = [parse_target(_) for _ in targets_raw.split(",")]
100 return targets
101
102
103 def collect_conda_target_lists(ctx, paths, recursive=False, found_tool_callback=None):
104 """Load CondaTarget lists from supplied artifact sources.
105
106 If a tool contains more than one requirement, the requirements will all
107 appear together as one list element of the output list.
108 """
109 conda_target_lists, _ = collect_conda_target_lists_and_tool_paths(ctx, paths, recursive=recursive, found_tool_callback=found_tool_callback)
110 return conda_target_lists
111
112
113 def collect_conda_target_lists_and_tool_paths(ctx, paths, recursive=False, found_tool_callback=None):
114 """Load CondaTarget lists from supplied artifact sources.
115
116 If a tool contains more than one requirement, the requirements will all
117 appear together as one list element of the output list.
118 """
119 conda_target_lists = set([])
120 tool_paths = collections.defaultdict(list)
121 for (tool_path, tool_source) in yield_tool_sources_on_paths(ctx, paths, recursive=recursive, yield_load_errors=False):
122 try:
123 if found_tool_callback:
124 found_tool_callback(tool_path)
125 targets = frozenset(tool_source_conda_targets(tool_source))
126 conda_target_lists.add(targets)
127 tool_paths[targets].append(tool_path)
128 except Exception as e:
129 ctx.log(f"Error while collecting list of conda targets for '{tool_path}': {unicodify(e)}")
130
131 # Turn them into lists so the order matches before returning...
132 conda_target_lists = list(conda_target_lists)
133 conda_target_tool_paths = [tool_paths[c] for c in conda_target_lists]
134
135 return conda_target_lists, conda_target_tool_paths
136
137
138 def tool_source_conda_targets(tool_source):
139 """Load CondaTarget object from supplied abstract tool source."""
140 requirements, _ = tool_source.parse_requirements_and_containers()
141 return conda_util.requirements_to_conda_targets(requirements)
142
143
144 best_practice_search_first = threading.local()
145
146
147 def best_practice_search(conda_target, conda_context=None, platform=None):
148 # Call it in offline mode after the first time.
149 try:
150 best_practice_search_first.previously_called
151 # TODO: Undo this...
152 offline = False
153 except AttributeError:
154 best_practice_search_first.previously_called = True
155 offline = False
156
157 if not conda_context:
158 conda_context = conda_util.CondaContext()
159 return conda_util.best_search_result(
160 conda_target,
161 conda_context=conda_context,
162 channels_override=BEST_PRACTICE_CHANNELS,
163 offline=offline,
164 platform=platform,
165 )
166
167
168 __all__ = (
169 "BEST_PRACTICE_CHANNELS",
170 "best_practice_search",
171 "build_conda_context",
172 "collect_conda_targets",
173 "collect_conda_target_lists",
174 "collect_conda_target_lists_and_tool_paths",
175 "tool_source_conda_targets",
176 )