Mercurial > repos > shellac > guppy_basecaller
diff env/lib/python3.7/site-packages/planemo/cli.py @ 5:9b1c78e6ba9c draft default tip
"planemo upload commit 6c0a8142489327ece472c84e558c47da711a9142"
author | shellac |
---|---|
date | Mon, 01 Jun 2020 08:59:25 -0400 |
parents | 79f47841a781 |
children |
line wrap: on
line diff
--- a/env/lib/python3.7/site-packages/planemo/cli.py Thu May 14 16:47:39 2020 -0400 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,332 +0,0 @@ -"""The module describes a CLI framework extending ``click``.""" -import functools -import logging.config -import os -import shutil -import sys -import traceback - -import click -from six.moves.urllib.request import urlopen - -from planemo import __version__ -from planemo.exit_codes import ExitCodeException -from planemo.galaxy import profiles -from .config import ( - OptionSource, - read_global_config, -) -from .io import error - - -CONTEXT_SETTINGS = dict(auto_envvar_prefix='PLANEMO') -COMMAND_ALIASES = { - "l": "lint", - "o": "open", - "t": "test", - "s": "serve", -} - - -class Context(object): - """Describe context of Planemo computation. - - Handles cross cutting concerns for Planemo such as verbose log - tracking, the definition of the Planemo workspace (``~/.planemo``), - and the global configuraton defined in ``~/.planemo.yml``. - """ - - def __init__(self): - """Construct a Context object using execution environment.""" - self.home = os.getcwd() - self._global_config = None - # Will be set by planemo CLI driver - self.verbose = False - self.planemo_config = None - self.planemo_directory = None - self.option_source = {} - - def set_option_source(self, param_name, option_source, force=False): - """Specify how an option was set.""" - if not force: - assert param_name not in self.option_source, "No option source for [%s]" % param_name - self.option_source[param_name] = option_source - - def get_option_source(self, param_name): - """Return OptionSource value indicating how the option was set.""" - assert param_name in self.option_source, "No option source for [%s]" % param_name - return self.option_source[param_name] - - @property - def global_config(self): - """Read Planemo's global configuration. - - As defined most simply by ~/.planemo.yml. - """ - if self._global_config is None: - self._global_config = read_global_config(self.planemo_config) - return self._global_config or {} - - def log(self, msg, *args): - """Log a message to stderr.""" - if args: - msg %= args - click.echo(msg, file=sys.stderr) - - def vlog(self, msg, *args, **kwds): - """Log a message to stderr only if verbose is enabled.""" - if self.verbose: - self.log(msg, *args) - if kwds.get("exception", False): - traceback.print_exc(file=sys.stderr) - - @property - def workspace(self): - """Create and return Planemo's workspace. - - By default this will be ``~/.planemo``. - """ - if not self.planemo_directory: - raise Exception("No planemo workspace defined.") - workspace = self.planemo_directory - return self._ensure_directory(workspace, "workspace") - - @property - def galaxy_profiles_directory(self): - """Create a return a directory for storing Galaxy profiles.""" - path = os.path.join(self.workspace, "profiles") - return self._ensure_directory(path, "Galaxy profiles") - - def _ensure_directory(self, path, name): - if not os.path.exists(path): - os.makedirs(path) - if not os.path.isdir(path): - template = "Planemo %s directory [%s] unavailable." - message = template % (name, path) - raise Exception(message) - return path - - def exit(self, exit_code): - """Exit planemo with the supplied exit code.""" - self.vlog("Exiting planemo with exit code [%d]" % exit_code) - raise ExitCodeException(exit_code) - - def cache_download(self, url, destination): - cache = os.path.join(self.workspace, "cache") - if not os.path.exists(cache): - os.makedirs(cache) - filename = os.path.basename(url) - cache_destination = os.path.join(cache, filename) - if not os.path.exists(cache_destination): - with urlopen(url) as fh: - content = fh.read() - if len(content) == 0: - raise Exception("Failed to download [%s]." % url) - with open(cache_destination, "wb") as f: - f.write(content) - - shutil.copy(cache_destination, destination) - - -pass_context = click.make_pass_decorator(Context, ensure=True) -cmd_folder = os.path.abspath(os.path.join(os.path.dirname(__file__), - 'commands')) - - -def list_cmds(): - """List planemo commands from commands folder.""" - rv = [] - for filename in os.listdir(cmd_folder): - if filename.endswith('.py') and \ - filename.startswith('cmd_'): - rv.append(filename[len("cmd_"):-len(".py")]) - rv.sort() - return rv - - -def name_to_command(name): - """Convert a subcommand name to the cli function for that command. - - Command <X> is defined by the method 'planemo.commands.cmd_<x>:cli', - this method uses `__import__` to load and return that method. - """ - try: - if sys.version_info[0] == 2: - name = name.encode('ascii', 'replace') - mod_name = 'planemo.commands.cmd_' + name - mod = __import__(mod_name, None, None, ['cli']) - except ImportError as e: - error("Problem loading command %s, exception %s" % (name, e)) - return - return mod.cli - - -class PlanemoCLI(click.MultiCommand): - - def list_commands(self, ctx): - return list_cmds() - - def get_command(self, ctx, name): - if name in COMMAND_ALIASES: - name = COMMAND_ALIASES[name] - return name_to_command(name) - - -def command_function(f): - """Extension point for processing kwds after click callbacks.""" - @functools.wraps(f) - def handle_blended_options(*args, **kwds): - profile = kwds.get("profile", None) - if profile: - ctx = args[0] - profile_defaults = profiles.ensure_profile( - ctx, profile, **kwds - ) - _setup_profile_options(ctx, profile_defaults, kwds) - - _setup_galaxy_source_options(args[0], kwds) - - try: - return f(*args, **kwds) - except ExitCodeException as e: - sys.exit(e.exit_code) - - return pass_context(handle_blended_options) - - -EXCLUSIVE_OPTIONS_LIST = [ -] - - -def _setup_galaxy_source_options(ctx, kwds): - for exclusive_options in EXCLUSIVE_OPTIONS_LIST: - option_source = {} - for option in exclusive_options: - if option in kwds: - option_source[option] = ctx.get_option_source(option) - else: - option_source[option] = None - - most_authoratative_source = None - most_authoratative_source_options = [] - for key, value in option_source.items(): - if value is None: - continue - if most_authoratative_source is None or value.value < most_authoratative_source.value: - most_authoratative_source = value - most_authoratative_source_options = [key] - elif value == most_authoratative_source: - most_authoratative_source_options.append(key) - - if most_authoratative_source != OptionSource.default and len(most_authoratative_source_options) > 1: - raise click.UsageError("Cannot specify multiple of %s" % most_authoratative_source_options) - - for option in exclusive_options: - if option in kwds and option not in most_authoratative_source_options: - del kwds[option] - - -def _setup_profile_options(ctx, profile_defaults, kwds): - for key, value in profile_defaults.items(): - option_present = key in kwds - option_cli_specified = option_present and (ctx.get_option_source(key) == OptionSource.cli) - use_profile_option = not option_present or not option_cli_specified - if use_profile_option: - kwds[key] = value - ctx.set_option_source( - key, OptionSource.profile, force=True - ) - - -@click.command(cls=PlanemoCLI, context_settings=CONTEXT_SETTINGS) -@click.version_option(__version__) -@click.option('-v', '--verbose', is_flag=True, - help='Enables verbose mode.') -@click.option('--config', - default="~/.planemo.yml", - envvar="PLANEMO_GLOBAL_CONFIG_PATH", - help="Planemo configuration YAML file.") -@click.option('--directory', - default="~/.planemo", - envvar="PLANEMO_GLOBAL_WORKSPACE", - help="Workspace for planemo.") -@pass_context -def planemo(ctx, config, directory, verbose, configure_logging=True): - """A command-line toolkit for building tools and workflows for Galaxy. - - Check out the full documentation for Planemo online - http://planemo.readthedocs.org or open with ``planemo docs``. - - All the individual planemo commands support the ``--help`` option, for - example use ``planemo lint --help`` for more details on checking tools. - """ - ctx.verbose = verbose - if configure_logging: - logging_config = { - 'version': 1, - 'disable_existing_loggers': False, - 'formatters': { - 'verbose': { - 'format': '%(name)s %(levelname)s %(asctime)s: %(message)s' - }, - 'simple': { - 'format': '%(name)s %(levelname)s: %(message)s' - }, - }, - 'handlers': { - 'console': { - 'level': 'DEBUG', - 'class': 'logging.StreamHandler', - 'formatter': 'simple' if not verbose else 'verbose' - }, - }, - 'loggers': { - # Suppress CWL is beta warning, for Planemo purposes - it is absolutely not. - 'galaxy.tools.parser.factory': { - 'handlers': ['console'], - 'propagate': False, - 'level': 'ERROR' if not verbose else "DEBUG", - }, - 'galaxy.tools.deps.commands': { - 'handlers': ['console'], - 'propagate': False, - 'level': 'ERROR' if not verbose else "DEBUG", - }, - 'galaxy': { - 'handlers': ['console'], - 'propagate': False, - 'level': 'INFO' if not verbose else "DEBUG", - }, - # @jmchilton - # I'm fixing up Planemo's lint functionality for CWL and I keep seeing this for the - # schema metadata stuff (e.g. in the workflows repo). "rdflib.term WARNING: - # http://schema.org/docs/!DOCTYPE html does not look like a valid URI, trying to - # serialize this will break.". I'm going to suppress this warning I think, or are the - # examples wrong and should declare their namespaces differently in some way? - # @mr-c - # That particular warning is worth suppressing. A PR to silence it permanently would be very welcome! - # https://github.com/RDFLib/rdflib/blob/master/rdflib/term.py#L225 - 'rdflib.term': { - 'handlers': ['console'], - 'propagate': False, - 'level': 'ERROR' if not verbose else "DEBUG", - } - }, - 'root': { - 'handlers': ['console'], - 'propagate': False, - 'level': 'WARNING' if not verbose else "DEBUG", - } - } - logging.config.dictConfig(logging_config) - ctx.planemo_config = os.path.expanduser(config) - ctx.planemo_directory = os.path.expanduser(directory) - - -__all__ = ( - "command_function", - "Context", - "list_cmds", - "name_to_command", - "planemo", -)