Mercurial > repos > shellac > guppy_basecaller

comparison env/lib/python3.7/site-packages/planemo/cli.py @ 5:9b1c78e6ba9c draft default tip

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload commit 6c0a8142489327ece472c84e558c47da711a9142"
author shellac
date Mon, 01 Jun 2020 08:59:25 -0400
parents 79f47841a781
children
comparison
equal deleted inserted replaced
4:79f47841a781 5:9b1c78e6ba9c
1 """The module describes a CLI framework extending ``click``."""
2 import functools
3 import logging.config
4 import os
5 import shutil
6 import sys
7 import traceback
8
9 import click
10 from six.moves.urllib.request import urlopen
11
12 from planemo import __version__
13 from planemo.exit_codes import ExitCodeException
14 from planemo.galaxy import profiles
15 from .config import (
16 OptionSource,
17 read_global_config,
18 )
19 from .io import error
20
21
22 CONTEXT_SETTINGS = dict(auto_envvar_prefix='PLANEMO')
23 COMMAND_ALIASES = {
24 "l": "lint",
25 "o": "open",
26 "t": "test",
27 "s": "serve",
28 }
29
30
31 class Context(object):
32 """Describe context of Planemo computation.
33
34 Handles cross cutting concerns for Planemo such as verbose log
35 tracking, the definition of the Planemo workspace (``~/.planemo``),
36 and the global configuraton defined in ``~/.planemo.yml``.
37 """
38
39 def __init__(self):
40 """Construct a Context object using execution environment."""
41 self.home = os.getcwd()
42 self._global_config = None
43 # Will be set by planemo CLI driver
44 self.verbose = False
45 self.planemo_config = None
46 self.planemo_directory = None
47 self.option_source = {}
48
49 def set_option_source(self, param_name, option_source, force=False):
50 """Specify how an option was set."""
51 if not force:
52 assert param_name not in self.option_source, "No option source for [%s]" % param_name
53 self.option_source[param_name] = option_source
54
55 def get_option_source(self, param_name):
56 """Return OptionSource value indicating how the option was set."""
57 assert param_name in self.option_source, "No option source for [%s]" % param_name
58 return self.option_source[param_name]
59
60 @property
61 def global_config(self):
62 """Read Planemo's global configuration.
63
64 As defined most simply by ~/.planemo.yml.
65 """
66 if self._global_config is None:
67 self._global_config = read_global_config(self.planemo_config)
68 return self._global_config or {}
69
70 def log(self, msg, *args):
71 """Log a message to stderr."""
72 if args:
73 msg %= args
74 click.echo(msg, file=sys.stderr)
75
76 def vlog(self, msg, *args, **kwds):
77 """Log a message to stderr only if verbose is enabled."""
78 if self.verbose:
79 self.log(msg, *args)
80 if kwds.get("exception", False):
81 traceback.print_exc(file=sys.stderr)
82
83 @property
84 def workspace(self):
85 """Create and return Planemo's workspace.
86
87 By default this will be ``~/.planemo``.
88 """
89 if not self.planemo_directory:
90 raise Exception("No planemo workspace defined.")
91 workspace = self.planemo_directory
92 return self._ensure_directory(workspace, "workspace")
93
94 @property
95 def galaxy_profiles_directory(self):
96 """Create a return a directory for storing Galaxy profiles."""
97 path = os.path.join(self.workspace, "profiles")
98 return self._ensure_directory(path, "Galaxy profiles")
99
100 def _ensure_directory(self, path, name):
101 if not os.path.exists(path):
102 os.makedirs(path)
103 if not os.path.isdir(path):
104 template = "Planemo %s directory [%s] unavailable."
105 message = template % (name, path)
106 raise Exception(message)
107 return path
108
109 def exit(self, exit_code):
110 """Exit planemo with the supplied exit code."""
111 self.vlog("Exiting planemo with exit code [%d]" % exit_code)
112 raise ExitCodeException(exit_code)
113
114 def cache_download(self, url, destination):
115 cache = os.path.join(self.workspace, "cache")
116 if not os.path.exists(cache):
117 os.makedirs(cache)
118 filename = os.path.basename(url)
119 cache_destination = os.path.join(cache, filename)
120 if not os.path.exists(cache_destination):
121 with urlopen(url) as fh:
122 content = fh.read()
123 if len(content) == 0:
124 raise Exception("Failed to download [%s]." % url)
125 with open(cache_destination, "wb") as f:
126 f.write(content)
127
128 shutil.copy(cache_destination, destination)
129
130
131 pass_context = click.make_pass_decorator(Context, ensure=True)
132 cmd_folder = os.path.abspath(os.path.join(os.path.dirname(__file__),
133 'commands'))
134
135
136 def list_cmds():
137 """List planemo commands from commands folder."""
138 rv = []
139 for filename in os.listdir(cmd_folder):
140 if filename.endswith('.py') and \
141 filename.startswith('cmd_'):
142 rv.append(filename[len("cmd_"):-len(".py")])
143 rv.sort()
144 return rv
145
146
147 def name_to_command(name):
148 """Convert a subcommand name to the cli function for that command.
149
150 Command <X> is defined by the method 'planemo.commands.cmd_<x>:cli',
151 this method uses `__import__` to load and return that method.
152 """
153 try:
154 if sys.version_info[0] == 2:
155 name = name.encode('ascii', 'replace')
156 mod_name = 'planemo.commands.cmd_' + name
157 mod = __import__(mod_name, None, None, ['cli'])
158 except ImportError as e:
159 error("Problem loading command %s, exception %s" % (name, e))
160 return
161 return mod.cli
162
163
164 class PlanemoCLI(click.MultiCommand):
165
166 def list_commands(self, ctx):
167 return list_cmds()
168
169 def get_command(self, ctx, name):
170 if name in COMMAND_ALIASES:
171 name = COMMAND_ALIASES[name]
172 return name_to_command(name)
173
174
175 def command_function(f):
176 """Extension point for processing kwds after click callbacks."""
177 @functools.wraps(f)
178 def handle_blended_options(*args, **kwds):
179 profile = kwds.get("profile", None)
180 if profile:
181 ctx = args[0]
182 profile_defaults = profiles.ensure_profile(
183 ctx, profile, **kwds
184 )
185 _setup_profile_options(ctx, profile_defaults, kwds)
186
187 _setup_galaxy_source_options(args[0], kwds)
188
189 try:
190 return f(*args, **kwds)
191 except ExitCodeException as e:
192 sys.exit(e.exit_code)
193
194 return pass_context(handle_blended_options)
195
196
197 EXCLUSIVE_OPTIONS_LIST = [
198 ]
199
200
201 def _setup_galaxy_source_options(ctx, kwds):
202 for exclusive_options in EXCLUSIVE_OPTIONS_LIST:
203 option_source = {}
204 for option in exclusive_options:
205 if option in kwds:
206 option_source[option] = ctx.get_option_source(option)
207 else:
208 option_source[option] = None
209
210 most_authoratative_source = None
211 most_authoratative_source_options = []
212 for key, value in option_source.items():
213 if value is None:
214 continue
215 if most_authoratative_source is None or value.value < most_authoratative_source.value:
216 most_authoratative_source = value
217 most_authoratative_source_options = [key]
218 elif value == most_authoratative_source:
219 most_authoratative_source_options.append(key)
220
221 if most_authoratative_source != OptionSource.default and len(most_authoratative_source_options) > 1:
222 raise click.UsageError("Cannot specify multiple of %s" % most_authoratative_source_options)
223
224 for option in exclusive_options:
225 if option in kwds and option not in most_authoratative_source_options:
226 del kwds[option]
227
228
229 def _setup_profile_options(ctx, profile_defaults, kwds):
230 for key, value in profile_defaults.items():
231 option_present = key in kwds
232 option_cli_specified = option_present and (ctx.get_option_source(key) == OptionSource.cli)
233 use_profile_option = not option_present or not option_cli_specified
234 if use_profile_option:
235 kwds[key] = value
236 ctx.set_option_source(
237 key, OptionSource.profile, force=True
238 )
239
240
241 @click.command(cls=PlanemoCLI, context_settings=CONTEXT_SETTINGS)
242 @click.version_option(__version__)
243 @click.option('-v', '--verbose', is_flag=True,
244 help='Enables verbose mode.')
245 @click.option('--config',
246 default="~/.planemo.yml",
247 envvar="PLANEMO_GLOBAL_CONFIG_PATH",
248 help="Planemo configuration YAML file.")
249 @click.option('--directory',
250 default="~/.planemo",
251 envvar="PLANEMO_GLOBAL_WORKSPACE",
252 help="Workspace for planemo.")
253 @pass_context
254 def planemo(ctx, config, directory, verbose, configure_logging=True):
255 """A command-line toolkit for building tools and workflows for Galaxy.
256
257 Check out the full documentation for Planemo online
258 http://planemo.readthedocs.org or open with ``planemo docs``.
259
260 All the individual planemo commands support the ``--help`` option, for
261 example use ``planemo lint --help`` for more details on checking tools.
262 """
263 ctx.verbose = verbose
264 if configure_logging:
265 logging_config = {
266 'version': 1,
267 'disable_existing_loggers': False,
268 'formatters': {
269 'verbose': {
270 'format': '%(name)s %(levelname)s %(asctime)s: %(message)s'
271 },
272 'simple': {
273 'format': '%(name)s %(levelname)s: %(message)s'
274 },
275 },
276 'handlers': {
277 'console': {
278 'level': 'DEBUG',
279 'class': 'logging.StreamHandler',
280 'formatter': 'simple' if not verbose else 'verbose'
281 },
282 },
283 'loggers': {
284 # Suppress CWL is beta warning, for Planemo purposes - it is absolutely not.
285 'galaxy.tools.parser.factory': {
286 'handlers': ['console'],
287 'propagate': False,
288 'level': 'ERROR' if not verbose else "DEBUG",
289 },
290 'galaxy.tools.deps.commands': {
291 'handlers': ['console'],
292 'propagate': False,
293 'level': 'ERROR' if not verbose else "DEBUG",
294 },
295 'galaxy': {
296 'handlers': ['console'],
297 'propagate': False,
298 'level': 'INFO' if not verbose else "DEBUG",
299 },
300 # @jmchilton
301 # I'm fixing up Planemo's lint functionality for CWL and I keep seeing this for the
302 # schema metadata stuff (e.g. in the workflows repo). "rdflib.term WARNING:
303 # http://schema.org/docs/!DOCTYPE html does not look like a valid URI, trying to
304 # serialize this will break.". I'm going to suppress this warning I think, or are the
305 # examples wrong and should declare their namespaces differently in some way?
306 # @mr-c
307 # That particular warning is worth suppressing. A PR to silence it permanently would be very welcome!
308 # https://github.com/RDFLib/rdflib/blob/master/rdflib/term.py#L225
309 'rdflib.term': {
310 'handlers': ['console'],
311 'propagate': False,
312 'level': 'ERROR' if not verbose else "DEBUG",
313 }
314 },
315 'root': {
316 'handlers': ['console'],
317 'propagate': False,
318 'level': 'WARNING' if not verbose else "DEBUG",
319 }
320 }
321 logging.config.dictConfig(logging_config)
322 ctx.planemo_config = os.path.expanduser(config)
323 ctx.planemo_directory = os.path.expanduser(directory)
324
325
326 __all__ = (
327 "command_function",
328 "Context",
329 "list_cmds",
330 "name_to_command",
331 "planemo",
332 )