Skip to content
Commits on Source (2)
q2templates/_version.py export-subst
q2cli/_version.py export-subst
......@@ -64,4 +64,5 @@ target/
# vi
.*.swp
# Ignore .DS_Store files for Mac users
.DS_Store
......@@ -14,10 +14,12 @@ install:
- wget -q https://raw.githubusercontent.com/qiime2/environment-files/master/latest/staging/qiime2-latest-py36-linux-conda.yml
- conda env create -q -n test-env --file qiime2-latest-py36-linux-conda.yml
- source activate test-env
- conda install -q pytest
- conda install -q nose
- pip install -q flake8
- pip install -q https://github.com/qiime2/q2lint/archive/master.zip
- make install
script:
- make lint
- make test
- QIIMETEST= source tab-qiime
- QIIMETEST= qiime info
include versioneer.py
include q2templates/_version.py
include q2cli/_version.py
.PHONY: all lint test install dev clean distclean
PYTHON ?= python
PREFIX ?= $(CONDA_PREFIX)
all: ;
......@@ -9,13 +10,17 @@ lint:
flake8
test: all
py.test
QIIMETEST= nosetests
install: all
$(PYTHON) setup.py install
$(PYTHON) setup.py install && \
mkdir -p $(PREFIX)/etc/conda/activate.d && \
cp hooks/50_activate_q2cli_tab_completion.sh $(PREFIX)/etc/conda/activate.d/
dev: all
pip install -e .
pip install -e . && \
mkdir -p $(PREFIX)/etc/conda/activate.d && \
cp hooks/50_activate_q2cli_tab_completion.sh $(PREFIX)/etc/conda/activate.d/
clean: distclean
......
# q2templates
Design template package for QIIME 2 Plugins
# q2cli
A [click-based](http://click.pocoo.org/) command line interface for [QIIME 2](https://github.com/qiime2/qiime2).
## Installation and getting help
Visit https://qiime2.org to learn more about q2cli and the QIIME 2 project.
## Enabling tab completion
### Bash
To enable tab completion in Bash, run the following command or add it to your `.bashrc`/`.bash_profile`:
```bash
source tab-qiime
```
### ZSH
To enable tab completion in ZSH, run the following commands or add them to your `.zshrc`:
```bash
autoload bashcompinit && bashcompinit && source tab-qiime
```
#!/usr/bin/env bash
# Bash completion script that defers to a cached completion script representing
# the state of the current QIIME 2 deployment.
#
# This script is intended to be executed on the command-line or in
# .bashrc/.bash_profile:
#
# source tab-qiime
#
_qiime_completion()
{
# Attempt to find the cached completion script. If q2cli isn't installed, or
# is an incompatible version, don't attempt completion.
local completion_path="$(python -c "import q2cli.util; print(q2cli.util.get_completion_path())" 2> /dev/null)"
if [[ $? != 0 ]]; then
unset COMPREPLY
return 0
fi
# If the completion script exists, attempt completion by invoking the script
# in a subshell, supplying COMP_WORDS and COMP_CWORD. Capture the output as
# the completion reply. If the completion script failed, don't attempt
# completion.
if [[ -f "$completion_path" ]] ; then
COMPREPLY=( $(COMP_WORDS="${COMP_WORDS[*]}" COMP_CWORD="${COMP_CWORD}" "$completion_path" 2> /dev/null) )
if [[ $? != 0 ]]; then
unset COMPREPLY
return 0
fi
else
unset COMPREPLY
return 0
fi
return 0
}
# Enable default readline and bash completion behavior when `_qiime_completion`
# doesn't have a reply.
complete -F _qiime_completion -o default -o bashdefault qiime
# Execute a `qiime` command (any command will do) so that tab-completion will
# work out-of-the-box (e.g. with a fresh installation of q2cli). Running a
# command will create or refresh the cache if necessary, which contains the
# actual completion script.
#
# Ignore stdout to avoid displaying help text to users enabling tab-completion.
# stderr displays the note about cache refreshing, as that can take a few
# moments to complete.
qiime > /dev/null
{% set data = load_setup_py_data() %}
{% set version = data.get('version') %}
{% set version = data.get('version') or 'placehold' %}
{% set release = '.'.join(version.split('.')[:2]) %}
# TODO review meta.yml spec and fill in missing fields as necessary (e.g. build number).
package:
name: q2templates
name: q2cli
version: {{ version }}
source:
......@@ -11,21 +11,26 @@ source:
build:
script: make install
entry_points:
- qiime=q2cli.__main__:qiime
requirements:
host:
- python {{ python }}
- python {{ python }}
- setuptools
run:
- python {{ python }}
- setuptools
- jinja2
- pandas
- python {{ python }}
- pip
- click
- qiime2 {{ release }}.*
test:
imports:
- q2templates
- q2cli
commands:
- QIIMETEST= qiime --help
about:
home: https://qiime2.org
......
if [ -n "${ZSH_VERSION-}" ]; then
autoload bashcompinit && bashcompinit && source tab-qiime
elif [ -n "${BASH_VERSION-}" ]; then
source tab-qiime
fi
......@@ -6,12 +6,7 @@
# The full license is in the file LICENSE, distributed with this software.
# ----------------------------------------------------------------------------
from ._templates import render
from .util import df_to_html
from ._version import get_versions
__version__ = get_versions()['version']
del get_versions
__all__ = ['render', 'df_to_html']
# ----------------------------------------------------------------------------
# Copyright (c) 2016-2019, QIIME 2 development team.
#
# Distributed under the terms of the Modified BSD License.
#
# The full license is in the file LICENSE, distributed with this software.
# ----------------------------------------------------------------------------
import click
import q2cli.commands
ROOT_COMMAND_HELP = """\
QIIME 2 command-line interface (q2cli)
--------------------------------------
To get help with QIIME 2, visit https://qiime2.org.
To enable tab completion in Bash, run the following command or add it to your \
.bashrc/.bash_profile:
source tab-qiime
To enable tab completion in ZSH, run the following commands or add them to \
your .zshrc:
autoload bashcompinit && bashcompinit && source tab-qiime
"""
# Entry point for CLI
@click.command(cls=q2cli.commands.RootCommand, invoke_without_command=True,
no_args_is_help=True, help=ROOT_COMMAND_HELP)
@click.version_option(prog_name='q2cli',
message='%(prog)s version %(version)s\nRun `qiime info` '
'for more version details.')
def qiime():
pass
if __name__ == '__main__':
qiime()
......@@ -24,7 +24,7 @@ def get_keywords():
# each be defined on a line of their own. _version.py will just call
# get_keywords().
git_refnames = " (tag: 2019.10.0)"
git_full = "2ea56bdbdb44d38e1bdbc9f9516cc42793911321"
git_full = "271df1dcf0c419046811dd20832f9c13446732f8"
git_date = "2019-11-01 01:04:25 +0000"
keywords = {"refnames": git_refnames, "full": git_full, "date": git_date}
return keywords
......@@ -42,8 +42,8 @@ def get_config():
cfg.VCS = "git"
cfg.style = "pep440"
cfg.tag_prefix = ""
cfg.parentdir_prefix = "q2templates-"
cfg.versionfile_source = "q2templates/_version.py"
cfg.parentdir_prefix = "q2cli-"
cfg.versionfile_source = "q2cli/_version.py"
cfg.verbose = False
return cfg
......
# ----------------------------------------------------------------------------
# Copyright (c) 2016-2019, QIIME 2 development team.
#
# Distributed under the terms of the Modified BSD License.
#
# The full license is in the file LICENSE, distributed with this software.
# ----------------------------------------------------------------------------
import click
from q2cli.click.command import ToolCommand, ToolGroupCommand
@click.group(help='Utilities for developers and advanced users.',
cls=ToolGroupCommand)
def dev():
pass
@dev.command(name='refresh-cache',
short_help='Refresh CLI cache.',
help="Refresh the CLI cache. Use this command if you are "
"developing a plugin, or q2cli itself, and want your "
"changes to take effect in the CLI. A refresh of the cache "
"is necessary because package versions do not typically "
"change each time an update is made to a package's code. "
"Setting the environment variable Q2CLIDEV to any value "
"will always refresh the cache when a command is run.",
cls=ToolCommand)
def refresh_cache():
import q2cli.core.cache
q2cli.core.cache.CACHE.refresh()
import_theme_help = \
("Allows for customization of q2cli's command line styling based on an "
"imported .theme (INI formatted) file. If you are unfamiliar with .ini "
"formatted files look here https://en.wikipedia.org/wiki/INI_file."
"\n"
"\n"
"The .theme file allows you to customize text on the basis of what that "
"text represents with the following supported text types: command, "
"option, type, default_arg, required, emphasis, problem, warning, error, "
"and success. These will be your headers in the '[]' brackets. "
"\n"
"\n"
"`command` refers to the name of the command you issued. `option` refers "
"to the arguments you give to the command when running it. `type` refers "
"to the QIIME 2 semantic typing of these arguments (where applicable). "
"`default_arg` refers to the label next to the argument indicating its "
"default value (where applicable), and if it is required (where "
"applicable). `required` refers to any arguments that must be passed to "
"the command for it to work and gives them special formatting on top of "
"your normal `option` formatting. `emphasis` refers to any emphasized "
"pieces of text within help text. `problem` refers to the text informing "
"you there were issues with the command. `warning` refers to the text "
"for non-fatal issues while `error` refers to the text for fatal issues."
"`success` refers to text indicating a process completed as expected."
"\n"
"\n"
"Depending on what your terminal supports, some or all of the following "
"pieces of the text's formatting may be customized: bold, dim (if true "
"the text's brightness is reduced), underline, blink, reverse (if true "
"foreground and background colors are reversed), and finally fg "
"(foreground color) and bg (background color). The first five may each "
"be either true or false, while the colors may be set to any of the "
"following: black, red, green, yellow, blue, magenta, cyan, white, "
"bright_black, bright_red, bright_green, bright_yellow, bright_blue, "
"bright_magenta, bright_cyan, or bright_white.")
@dev.command(name='import-theme',
short_help='Install new command line theme.',
help=import_theme_help,
cls=ToolCommand)
@click.option('--theme', required=True,
type=click.Path(exists=True, file_okay=True,
dir_okay=False, readable=True),
help='Path to file containing new theme info')
def import_theme(theme):
import os
import shutil
from configparser import Error
import q2cli.util
from q2cli.core.config import CONFIG
try:
CONFIG.parse_file(theme)
except Error as e:
# If they tried to change [error] in a valid manner before we hit our
# parsing error, we don't want to use their imported error settings
CONFIG.styles = CONFIG.get_default_styles()
header = 'Something went wrong while parsing your theme: '
q2cli.util.exit_with_error(e, header=header, traceback=None)
shutil.copy(theme, os.path.join(q2cli.util.get_app_dir(),
'cli-colors.theme'))
@dev.command(name='export-default-theme',
short_help='Export the default settings.',
help='Create a .theme (INI formatted) file from the default '
'settings at the specified filepath.',
cls=ToolCommand)
@click.option('--output-path', required=True,
type=click.Path(exists=False, file_okay=True,
dir_okay=False, readable=True),
help='Path to output the config to')
def export_default_theme(output_path):
import configparser
from q2cli.core.config import CONFIG
parser = configparser.ConfigParser()
parser.read_dict(CONFIG.get_default_styles())
with open(output_path, 'w') as fh:
parser.write(fh)
def abort_if_false(ctx, param, value):
if not value:
ctx.abort()
@dev.command(name='reset-theme',
short_help='Reset command line theme to default.',
help="Reset command line theme to default. Requres the '--yes' "
"parameter to be passed asserting you do want to reset.",
cls=ToolCommand)
@click.option('--yes', is_flag=True, callback=abort_if_false,
expose_value=False,
prompt='Are you sure you want to reset your theme?')
def reset_theme():
import os
import q2cli.util
path = os.path.join(q2cli.util.get_app_dir(), 'cli-colors.theme')
if os.path.exists(path):
os.unlink(path)
click.echo('Theme reset.')
else:
click.echo('Theme was already default.')
# ----------------------------------------------------------------------------
# Copyright (c) 2016-2019, QIIME 2 development team.
#
# Distributed under the terms of the Modified BSD License.
#
# The full license is in the file LICENSE, distributed with this software.
# ----------------------------------------------------------------------------
import click
from q2cli.click.command import ToolCommand
def _echo_version():
import sys
import qiime2
import q2cli
pyver = sys.version_info
click.echo('Python version: %d.%d.%d' %
(pyver.major, pyver.minor, pyver.micro))
click.echo('QIIME 2 release: %s' % qiime2.__release__)
click.echo('QIIME 2 version: %s' % qiime2.__version__)
click.echo('q2cli version: %s' % q2cli.__version__)
def _echo_plugins():
import q2cli.core.cache
plugins = q2cli.core.cache.CACHE.plugins
if plugins:
for name, plugin in sorted(plugins.items()):
click.echo('%s: %s' % (name, plugin['version']))
else:
click.secho('No plugins are currently installed.\nYou can browse '
'the official QIIME 2 plugins at https://qiime2.org')
@click.command(help='Display information about current deployment.',
cls=ToolCommand)
def info():
import q2cli.util
# This import improves performance for repeated _echo_plugins
import q2cli.core.cache
click.secho('System versions', fg='green')
_echo_version()
click.secho('\nInstalled plugins', fg='green')
_echo_plugins()
click.secho('\nApplication config directory', fg='green')
click.secho(q2cli.util.get_app_dir())
click.secho('\nGetting help', fg='green')
click.secho('To get help with QIIME 2, visit https://qiime2.org')
# ----------------------------------------------------------------------------
# Copyright (c) 2016-2019, QIIME 2 development team.
#
# Distributed under the terms of the Modified BSD License.
#
# The full license is in the file LICENSE, distributed with this software.
# ----------------------------------------------------------------------------
import os
import click
import q2cli.util
from q2cli.click.command import ToolCommand, ToolGroupCommand
_COMBO_METAVAR = 'ARTIFACT/VISUALIZATION'
@click.group(help='Tools for working with QIIME 2 files.',
cls=ToolGroupCommand)
def tools():
pass
@tools.command(name='export',
short_help='Export data from a QIIME 2 Artifact '
'or a Visualization',
help='Exporting extracts (and optionally transforms) data '
'stored inside an Artifact or Visualization. Note that '
'Visualizations cannot be transformed with --output-format',
cls=ToolCommand)
@click.option('--input-path', required=True, metavar=_COMBO_METAVAR,
type=click.Path(exists=True, file_okay=True,
dir_okay=False, readable=True),
help='Path to file that should be exported')
@click.option('--output-path', required=True,
type=click.Path(exists=False, file_okay=True, dir_okay=True,
writable=True),
help='Path to file or directory where '
'data should be exported to')
@click.option('--output-format', required=False,
help='Format which the data should be exported as. '
'This option cannot be used with Visualizations')
def export_data(input_path, output_path, output_format):
import qiime2.util
import qiime2.sdk
import distutils
from q2cli.core.config import CONFIG
result = qiime2.sdk.Result.load(input_path)
if output_format is None:
if isinstance(result, qiime2.sdk.Artifact):
output_format = result.format.__name__
else:
output_format = 'Visualization'
result.export_data(output_path)
else:
if isinstance(result, qiime2.sdk.Visualization):
error = '--output-format cannot be used with visualizations'
click.echo(CONFIG.cfg_style('error', error), err=True)
click.get_current_context().exit(1)
else:
source = result.view(qiime2.sdk.parse_format(output_format))
if os.path.isfile(str(source)):
if os.path.isfile(output_path):
os.remove(output_path)
elif os.path.dirname(output_path) == '':
# This allows the user to pass a filename as a path if they
# want their output in the current working directory
output_path = os.path.join('.', output_path)
if os.path.dirname(output_path) != '':
# create directory (recursively) if it doesn't exist yet
os.makedirs(os.path.dirname(output_path), exist_ok=True)
qiime2.util.duplicate(str(source), output_path)
else:
distutils.dir_util.copy_tree(str(source), output_path)
output_type = 'file' if os.path.isfile(output_path) else 'directory'
success = 'Exported %s as %s to %s %s' % (input_path, output_format,
output_type, output_path)
click.echo(CONFIG.cfg_style('success', success))
def show_importable_types(ctx, param, value):
if not value or ctx.resilient_parsing:
return
import qiime2.sdk
importable_types = sorted(qiime2.sdk.PluginManager().importable_types,
key=repr)
if importable_types:
for name in importable_types:
click.echo(name)
else:
click.echo('There are no importable types in the current deployment.')
ctx.exit()
def show_importable_formats(ctx, param, value):
if not value or ctx.resilient_parsing:
return
import qiime2.sdk
importable_formats = sorted(qiime2.sdk.PluginManager().importable_formats)
if importable_formats:
for format in importable_formats:
click.echo(format)
else:
click.echo('There are no importable formats '
'in the current deployment.')
ctx.exit()
@tools.command(name='import',
short_help='Import data into a new QIIME 2 Artifact.',
help="Import data to create a new QIIME 2 Artifact. See "
"https://docs.qiime2.org/ for usage examples and details "
"on the file types and associated semantic types that can "
"be imported.",
cls=ToolCommand)
@click.option('--type', required=True,
help='The semantic type of the artifact that will be created '
'upon importing. Use --show-importable-types to see what '
'importable semantic types are available in the current '
'deployment.')
@click.option('--input-path', required=True,
type=click.Path(exists=True, file_okay=True, dir_okay=True,
readable=True),
help='Path to file or directory that should be imported.')
@click.option('--output-path', required=True, metavar='ARTIFACT',
type=click.Path(exists=False, file_okay=True, dir_okay=False,
writable=True),
help='Path where output artifact should be written.')
@click.option('--input-format', required=False,
help='The format of the data to be imported. If not provided, '
'data must be in the format expected by the semantic type '
'provided via --type.')
@click.option('--show-importable-types', is_flag=True, is_eager=True,
callback=show_importable_types, expose_value=False,
help='Show the semantic types that can be supplied to --type '
'to import data into an artifact.')
@click.option('--show-importable-formats', is_flag=True, is_eager=True,
callback=show_importable_formats, expose_value=False,
help='Show formats that can be supplied to --input-format to '
'import data into an artifact.')
def import_data(type, input_path, output_path, input_format):
import qiime2.sdk
import qiime2.plugin
from q2cli.core.config import CONFIG
try:
artifact = qiime2.sdk.Artifact.import_data(type, input_path,
view_type=input_format)
except qiime2.plugin.ValidationError as e:
header = 'There was a problem importing %s:' % input_path
q2cli.util.exit_with_error(e, header=header, traceback=None)
except Exception as e:
header = 'An unexpected error has occurred:'
q2cli.util.exit_with_error(e, header=header)
artifact.save(output_path)
if input_format is None:
input_format = artifact.format.__name__
success = 'Imported %s as %s to %s' % (input_path,
input_format,
output_path)
click.echo(CONFIG.cfg_style('success', success))
@tools.command(short_help='Take a peek at a QIIME 2 Artifact or '
'Visualization.',
help="Display basic information about a QIIME 2 Artifact or "
"Visualization, including its UUID and type.",
cls=ToolCommand)
@click.argument('path', type=click.Path(exists=True, file_okay=True,
dir_okay=False, readable=True),
metavar=_COMBO_METAVAR)
def peek(path):
import qiime2.sdk
from q2cli.core.config import CONFIG
metadata = qiime2.sdk.Result.peek(path)
click.echo(CONFIG.cfg_style('type', "UUID")+": ", nl=False)
click.echo(metadata.uuid)
click.echo(CONFIG.cfg_style('type', "Type")+": ", nl=False)
click.echo(metadata.type)
if metadata.format is not None:
click.echo(CONFIG.cfg_style('type', "Data format")+": ", nl=False)
click.echo(metadata.format)
@tools.command('inspect-metadata',
short_help='Inspect columns available in metadata.',
help='Inspect metadata files or artifacts viewable as metadata.'
' Providing multiple file paths to this command will merge'
' the metadata.',
cls=ToolCommand)
@click.option('--tsv/--no-tsv', default=False,
help='Print as machine-readable TSV instead of text.')
@click.argument('paths', nargs=-1, required=True, metavar='METADATA...',
type=click.Path(exists=True, file_okay=True, dir_okay=False,
readable=True))
@q2cli.util.pretty_failure(traceback=None)
def inspect_metadata(paths, tsv, failure):
m = [_load_metadata(p) for p in paths]
metadata = m[0]
if m[1:]:
metadata = metadata.merge(*m[1:])
# we aren't expecting errors below this point, so set traceback to default
failure.traceback = 'stderr'
failure.header = "An unexpected error has occurred:"
COLUMN_NAME = "COLUMN NAME"
COLUMN_TYPE = "TYPE"
max_name_len = max([len(n) for n in metadata.columns]
+ [len(COLUMN_NAME)])
max_type_len = max([len(p.type) for p in metadata.columns.values()]
+ [len(COLUMN_TYPE)])
if tsv:
import csv
import io
def formatter(*row):
# This is gross, but less gross than robust TSV writing.
with io.StringIO() as fh:
writer = csv.writer(fh, dialect='excel-tab', lineterminator='')
writer.writerow(row)
return fh.getvalue()
else:
formatter = ("{0:>%d} {1:%d}" % (max_name_len, max_type_len)).format
click.secho(formatter(COLUMN_NAME, COLUMN_TYPE), bold=True)
if not tsv:
click.secho(formatter("=" * max_name_len, "=" * max_type_len),
bold=True)
for name, props in metadata.columns.items():
click.echo(formatter(name, props.type))
if not tsv:
click.secho(formatter("=" * max_name_len, "=" * max_type_len),
bold=True)
click.secho(("{0:>%d} " % max_name_len).format("IDS:"),
bold=True, nl=False)
click.echo(metadata.id_count)
click.secho(("{0:>%d} " % max_name_len).format("COLUMNS:"),
bold=True, nl=False)
click.echo(metadata.column_count)
def _load_metadata(path):
import qiime2
import qiime2.sdk
# TODO: clean up duplication between this and the metadata handlers.
try:
artifact = qiime2.sdk.Result.load(path)
except Exception:
metadata = qiime2.Metadata.load(path)
else:
if isinstance(artifact, qiime2.Visualization):
raise Exception("Visualizations cannot be viewed as QIIME 2"
" metadata:\n%r" % path)
elif artifact.has_metadata():
metadata = artifact.view(qiime2.Metadata)
else:
raise Exception("Artifacts with type %r cannot be viewed as"
" QIIME 2 metadata:\n%r" % (artifact.type, path))
return metadata
@tools.command(short_help='View a QIIME 2 Visualization.',
help="Displays a QIIME 2 Visualization until the command "
"exits. To open a QIIME 2 Visualization so it can be "
"used after the command exits, use 'qiime tools extract'.",
cls=ToolCommand)
@click.argument('visualization-path', metavar='VISUALIZATION',
type=click.Path(exists=True, file_okay=True, dir_okay=False,
readable=True))
@click.option('--index-extension', required=False, default='html',
help='The extension of the index file that should be opened. '
'[default: html]')
def view(visualization_path, index_extension):
# Guard headless envs from having to import anything large
import sys
from q2cli.core.config import CONFIG
if not os.getenv("DISPLAY") and sys.platform != "darwin":
raise click.UsageError(
'Visualization viewing is currently not supported in headless '
'environments. You can view Visualizations (and Artifacts) at '
'https://view.qiime2.org, or move the Visualization to an '
'environment with a display and view it with `qiime tools view`.')
import zipfile
import qiime2.sdk
if index_extension.startswith('.'):
index_extension = index_extension[1:]
try:
visualization = qiime2.sdk.Visualization.load(visualization_path)
# TODO: currently a KeyError is raised if a zipped file that is not a
# QIIME 2 result is passed. This should be handled better by the framework.
except (zipfile.BadZipFile, KeyError, TypeError):
raise click.BadParameter(
'%s is not a QIIME 2 Visualization. Only QIIME 2 Visualizations '
'can be viewed.' % visualization_path)
index_paths = visualization.get_index_paths(relative=False)
if index_extension not in index_paths:
raise click.BadParameter(
'No index %s file is present in the archive. Available index '
'extensions are: %s' % (index_extension,
', '.join(index_paths.keys())))
else:
index_path = index_paths[index_extension]
launch_status = click.launch(index_path)
if launch_status != 0:
click.echo(CONFIG.cfg_style('error', 'Viewing visualization '
'failed while attempting to open '
f'{index_path}'), err=True)
else:
while True:
click.echo(
"Press the 'q' key, Control-C, or Control-D to quit. This "
"view may no longer be accessible or work correctly after "
"quitting.", nl=False)
# There is currently a bug in click.getchar where translation
# of Control-C and Control-D into KeyboardInterrupt and
# EOFError (respectively) does not work on Python 3. The code
# here should continue to work as expected when the bug is
# fixed in Click.
#
# https://github.com/pallets/click/issues/583
try:
char = click.getchar()
click.echo()
if char in {'q', '\x03', '\x04'}:
break
except (KeyboardInterrupt, EOFError):
break
@tools.command(short_help="Extract a QIIME 2 Artifact or Visualization "
"archive.",
help="Extract all contents of a QIIME 2 Artifact or "
"Visualization's archive, including provenance, metadata, "
"and actual data. Use 'qiime tools export' to export only "
"the data stored in an Artifact or Visualization, with "
"the choice of exporting to different formats.",
cls=ToolCommand)
@click.option('--input-path', required=True, metavar=_COMBO_METAVAR,
type=click.Path(exists=True, file_okay=True, dir_okay=False,
readable=True),
help='Path to file that should be extracted')
@click.option('--output-path', required=False,
type=click.Path(exists=False, file_okay=False, dir_okay=True,
writable=True),
help='Directory where archive should be extracted to '
'[default: current working directory]',
default=os.getcwd())
def extract(input_path, output_path):
import zipfile
import qiime2.sdk
from q2cli.core.config import CONFIG
try:
extracted_dir = qiime2.sdk.Result.extract(input_path, output_path)
except (zipfile.BadZipFile, ValueError):
raise click.BadParameter(
'%s is not a valid QIIME 2 Result. Only QIIME 2 Artifacts and '
'Visualizations can be extracted.' % input_path)
else:
success = 'Extracted %s to directory %s' % (input_path, extracted_dir)
click.echo(CONFIG.cfg_style('success', success))
@tools.command(short_help='Validate data in a QIIME 2 Artifact.',
help='Validate data in a QIIME 2 Artifact. QIIME 2 '
'automatically performs some basic validation when '
'managing your data; use this command to perform explicit '
'and/or more thorough validation of your data (e.g. when '
'debugging issues with your data or analyses).\n\nNote: '
'validation can take some time to complete, depending on '
'the size and type of your data.',
cls=ToolCommand)
@click.argument('path', type=click.Path(exists=True, file_okay=True,
dir_okay=False, readable=True),
metavar=_COMBO_METAVAR)
@click.option('--level', required=False, type=click.Choice(['min', 'max']),
help='Desired level of validation. "min" will perform minimal '
'validation, and "max" will perform maximal validation (at '
'the potential cost of runtime).',
default='max', show_default=True)
def validate(path, level):
import qiime2.sdk
from q2cli.core.config import CONFIG
try:
result = qiime2.sdk.Result.load(path)
except Exception as e:
header = 'There was a problem loading %s as a QIIME 2 Result:' % path
q2cli.util.exit_with_error(e, header=header)
try:
result.validate(level)
except qiime2.plugin.ValidationError as e:
header = 'Result %s does not appear to be valid at level=%s:' % (
path, level)
q2cli.util.exit_with_error(e, header=header, traceback=None)
except Exception as e:
header = ('An unexpected error has occurred while attempting to '
'validate result %s:' % path)
q2cli.util.exit_with_error(e, header=header)
else:
click.echo(CONFIG.cfg_style('success', f'Result {path} appears to be '
f'valid at level={level}.'))
@tools.command(short_help='Print citations for a QIIME 2 result.',
help='Print citations as a BibTex file (.bib) for a QIIME 2'
' result.',
cls=ToolCommand)
@click.argument('path', type=click.Path(exists=True, file_okay=True,
dir_okay=False, readable=True),
metavar=_COMBO_METAVAR)
def citations(path):
import qiime2.sdk
import io
from q2cli.core.config import CONFIG
ctx = click.get_current_context()
try:
result = qiime2.sdk.Result.load(path)
except Exception as e:
header = 'There was a problem loading %s as a QIIME 2 result:' % path
q2cli.util.exit_with_error(e, header=header)
if result.citations:
with io.StringIO() as fh:
result.citations.save(fh)
click.echo(fh.getvalue(), nl=False)
ctx.exit(0)
else:
click.echo(CONFIG.cfg_style('problem', 'No citations found.'),
err=True)
ctx.exit(1)
# ----------------------------------------------------------------------------
# Copyright (c) 2016-2019, QIIME 2 development team.
#
# Distributed under the terms of the Modified BSD License.
#
# The full license is in the file LICENSE, distributed with this software.
# ----------------------------------------------------------------------------
# ----------------------------------------------------------------------------
# Copyright (c) 2016-2019, QIIME 2 development team.
#
# Distributed under the terms of the Modified BSD License.
#
# The full license is in the file LICENSE, distributed with this software.
# ----------------------------------------------------------------------------
# ----------------------------------------------------------------------------
# Some of the source code in this file is derived from original work:
#
# Copyright (c) 2014 by the Pallets team.
#
# To see the license for the original work, see licenses/click.LICENSE.rst
# Specific reproduction and derivation of original work is marked below.
# ----------------------------------------------------------------------------
import click
import click.core
class BaseCommandMixin:
# Modified from original:
# < https://github.com/pallets/click/blob/
# c6042bf2607c5be22b1efef2e42a94ffd281434c/click/core.py#L867 >
# Copyright (c) 2014 by the Pallets team.
def make_parser(self, ctx):
"""Creates the underlying option parser for this command."""
from .parser import Q2Parser
parser = Q2Parser(ctx)
for param in self.get_params(ctx):
param.add_to_parser(parser, ctx)
return parser
# Modified from original:
# < https://github.com/pallets/click/blob/
# c6042bf2607c5be22b1efef2e42a94ffd281434c/click/core.py#L934 >
# Copyright (c) 2014 by the Pallets team.
def parse_args(self, ctx, args):
from q2cli.core.config import CONFIG
if isinstance(self, click.MultiCommand):
return super().parse_args(ctx, args)
errors = []
parser = self.make_parser(ctx)
skip_rest = False
for _ in range(10): # surely this is enough attempts
try:
opts, args, param_order = parser.parse_args(args=args)
break
except click.ClickException as e:
errors.append(e)
skip_rest = True
if not skip_rest:
for param in click.core.iter_params_for_processing(
param_order, self.get_params(ctx)):
try:
value, args = param.handle_parse_result(ctx, opts, args)
except click.ClickException as e:
errors.append(e)
if args and not ctx.allow_extra_args and not ctx.resilient_parsing:
errors.append(click.UsageError(
'Got unexpected extra argument%s (%s)'
% (len(args) != 1 and 's' or '',
' '.join(map(click.core.make_str, args)))))
if errors:
click.echo(ctx.get_help()+"\n", err=True)
if len(errors) > 1:
problems = 'There were some problems with the command:'
else:
problems = 'There was a problem with the command:'
click.echo(CONFIG.cfg_style('problem',
problems.center(78, ' ')), err=True)
for idx, e in enumerate(errors, 1):
msg = click.formatting.wrap_text(
e.format_message(),
initial_indent=' (%d/%d%s) ' % (idx, len(errors),
'?' if skip_rest else ''),
subsequent_indent=' ')
click.echo(CONFIG.cfg_style('error', msg), err=True)
ctx.exit(1)
ctx.args = args
return args
def get_option_names(self, ctx):
if not hasattr(self, '__option_names'):
names = set()
for param in self.get_params(ctx):
if hasattr(param, 'q2_name'):
names.add(param.q2_name)
else:
names.add(param.name)
self.__option_names = names
return self.__option_names
def list_commands(self, ctx):
if not hasattr(super(), 'list_commands'):
return []
return super().list_commands(ctx)
def get_opt_groups(self, ctx):
return {'Options': list(self.get_params(ctx))}
def format_help_text(self, ctx, formatter):
super().format_help_text(ctx, formatter)
formatter.write_paragraph()
# Modified from original:
# < https://github.com/pallets/click/blob
# /c6042bf2607c5be22b1efef2e42a94ffd281434c/click/core.py#L830 >
# Copyright (c) 2014 by the Pallets team.
def format_usage(self, ctx, formatter):
from q2cli.core.config import CONFIG
"""Writes the usage line into the formatter."""
pieces = self.collect_usage_pieces(ctx)
formatter.write_usage(CONFIG.cfg_style('command', ctx.command_path),
' '.join(pieces))
def format_options(self, ctx, formatter, COL_MAX=23, COL_MIN=10):
from q2cli.core.config import CONFIG
# write options
opt_groups = {}
records = []
for group, options in self.get_opt_groups(ctx).items():
opt_records = []
for o in options:
record = o.get_help_record(ctx)
if record is None:
continue
opt_records.append((o, record))
records.append(record)
opt_groups[group] = opt_records
first_columns = (r[0] for r in records)
border = min(COL_MAX, max(COL_MIN, *(len(col) for col in first_columns
if len(col) < COL_MAX)))
for opt_group, opt_records in opt_groups.items():
if not opt_records:
continue
formatter.write_heading(click.style(opt_group, bold=True))
formatter.indent()
padded_border = border + formatter.current_indent
for opt, record in opt_records:
self.write_option(ctx, formatter, opt, record, padded_border)
formatter.dedent()
# Modified from original:
# https://github.com/pallets/click/blob
# /c6042bf2607c5be22b1efef2e42a94ffd281434c/click/core.py#L1056
# Copyright (c) 2014 by the Pallets team.
commands = []
for subcommand in self.list_commands(ctx):
cmd = self.get_command(ctx, subcommand)
# What is this, the tool lied about a command. Ignore it
if cmd is None:
continue
if cmd.hidden:
continue
commands.append((subcommand, cmd))
# allow for 3 times the default spacing
if len(commands):
limit = formatter.width - 6 - max(len(cmd[0]) for cmd in commands)
rows = []
for subcommand, cmd in commands:
help = cmd.get_short_help_str(limit)
rows.append((CONFIG.cfg_style('command', subcommand), help))
if rows:
with formatter.section(click.style('Commands', bold=True)):
formatter.write_dl(rows)
def write_option(self, ctx, formatter, opt, record, border, COL_SPACING=2):
import itertools
from q2cli.core.config import CONFIG
full_width = formatter.width - formatter.current_indent
indent_text = ' ' * formatter.current_indent
opt_text, help_text = record
opt_text_secondary = None
if type(opt_text) is tuple:
opt_text, opt_text_secondary = opt_text
help_text, requirements = self._clean_help(help_text)
type_placement = None
type_repr = None
type_indent = 2 * indent_text
if hasattr(opt.type, 'get_type_repr'):
type_repr = opt.type.get_type_repr(opt)
if type_repr is not None:
if len(type_repr) <= border - len(type_indent):
type_placement = 'under'
else:
type_placement = 'beside'
if len(opt_text) > border:
lines = simple_wrap(opt_text, full_width)
else:
lines = [opt_text.split(' ')]
if opt_text_secondary is not None:
lines.append(opt_text_secondary.split(' '))
to_write = []
for tokens in lines:
dangling_edge = formatter.current_indent
styled = []
for token in tokens:
dangling_edge += len(token) + 1
if token.startswith('--'):
token = CONFIG.cfg_style('option', token,
required=opt.required)
styled.append(token)
line = indent_text + ' '.join(styled)
to_write.append(line)
formatter.write('\n'.join(to_write))
dangling_edge -= 1
if type_placement == 'beside':
lines = simple_wrap(type_repr, formatter.width - len(type_indent),
start_col=dangling_edge - 1)
to_write = []
first_iter = True
for tokens in lines:
line = ' '.join(tokens)
if first_iter:
dangling_edge += 1 + len(line)
line = " " + CONFIG.cfg_style('type', line)
first_iter = False
else:
dangling_edge = len(type_indent) + len(line)
line = type_indent + CONFIG.cfg_style('type', line)
to_write.append(line)
formatter.write('\n'.join(to_write))
if dangling_edge + 1 > border + COL_SPACING:
formatter.write('\n')
left_col = []
else:
padding = ' ' * (border + COL_SPACING - dangling_edge)
formatter.write(padding)
dangling_edge += len(padding)
left_col = [''] # jagged start
if type_placement == 'under':
padding = ' ' * (border + COL_SPACING
- len(type_repr) - len(type_indent))
line = ''.join(
[type_indent, CONFIG.cfg_style('type', type_repr), padding])
left_col.append(line)
if hasattr(opt, 'meta_help') and opt.meta_help is not None:
meta_help = simple_wrap(opt.meta_help,
border - len(type_indent) - 1)
for idx, line in enumerate([' '.join(t) for t in meta_help]):
if idx == 0:
line = type_indent + '(' + line
else:
line = type_indent + ' ' + line
if idx == len(meta_help) - 1:
line += ')'
line += ' ' * (border - len(line) + COL_SPACING)
left_col.append(line)
right_col = simple_wrap(help_text,
formatter.width - border - COL_SPACING)
right_col = [' '.join(self._color_important(tokens, ctx))
for tokens in right_col]
to_write = []
for left, right in itertools.zip_longest(
left_col, right_col, fillvalue=' ' * (border + COL_SPACING)):
to_write.append(left)
if right.strip():
to_write[-1] += right
formatter.write('\n'.join(to_write))
if requirements is None:
formatter.write('\n')
else:
if to_write:
if len(to_write) > 1 or ((not left_col) or left_col[0] != ''):
dangling_edge = 0
dangling_edge += click.formatting.term_len(to_write[-1])
else:
pass # dangling_edge is still correct
if dangling_edge + 1 + len(requirements) > formatter.width:
formatter.write('\n')
pad = formatter.width - len(requirements)
else:
pad = formatter.width - len(requirements) - dangling_edge
formatter.write(
(' ' * pad) + CONFIG.cfg_style(
'default_arg', requirements) + '\n')
def _color_important(self, tokens, ctx):
import re
from q2cli.core.config import CONFIG
for t in tokens:
if '_' in t:
names = self.get_option_names(ctx)
if re.sub(r'[^\w]', '', t) in names:
m = re.search(r'(\w+)', t)
word = t[m.start():m.end()]
word = CONFIG.cfg_style('emphasis', word.replace('_', '-'))
token = t[:m.start()] + word + t[m.end():]
yield token
continue
yield t
def _clean_help(self, text):
reqs = ['[required]', '[optional]', '[default: ']
requirement = None
for req in reqs:
if req in text:
requirement = req
break
else:
return text, None
req_idx = text.index(requirement)
return text[:req_idx].strip(), text[req_idx:].strip()
class ToolCommand(BaseCommandMixin, click.Command):
pass
class ToolGroupCommand(BaseCommandMixin, click.Group):
pass
def simple_wrap(text, target, start_col=0):
result = [[]]
current_line = result[0]
current_width = start_col
tokens = []
for token in text.split(' '):
if len(token) <= target:
tokens.append(token)
else:
for i in range(0, len(token), target):
tokens.append(token[i:i+target])
for token in tokens:
token_len = len(token)
if current_width + 1 + token_len > target:
current_line = [token]
result.append(current_line)
current_width = token_len
else:
result[-1].append(token)
current_width += 1 + token_len
return result
Copyright © 2014 by the Pallets team.
Some rights reserved.
Redistribution and use in source and binary forms of the software as
well as documentation, with or without modification, are permitted
provided that the following conditions are met:
- Redistributions of source code must retain the above copyright
notice, this list of conditions and the following disclaimer.
- Redistributions in binary form must reproduce the above copyright
notice, this list of conditions and the following disclaimer in the
documentation and/or other materials provided with the distribution.
- Neither the name of the copyright holder nor the names of its
contributors may be used to endorse or promote products derived from
this software without specific prior written permission.
THIS SOFTWARE AND DOCUMENTATION IS PROVIDED BY THE COPYRIGHT HOLDERS AND
CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING,
BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND
FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
THIS SOFTWARE AND DOCUMENTATION, EVEN IF ADVISED OF THE POSSIBILITY OF
SUCH DAMAGE.
----
Click uses parts of optparse written by Gregory P. Ward and maintained
by the Python Software Foundation. This is limited to code in parser.py.
Copyright © 2001-2006 Gregory P. Ward. All rights reserved.
Copyright © 2002-2006 Python Software Foundation. All rights reserved.
# ----------------------------------------------------------------------------
# Copyright (c) 2016-2019, QIIME 2 development team.
#
# Distributed under the terms of the Modified BSD License.
#
# The full license is in the file LICENSE, distributed with this software.
# ----------------------------------------------------------------------------
import click
from .type import QIIME2Type
# Sentinel to avoid the situation where `None` *is* the default value.
NoDefault = {}
class GeneratedOption(click.Option):
def __init__(self, *, prefix, name, repr, ast, multiple, is_bool_flag,
metadata, metavar, default=NoDefault, description=None,
**attrs):
import q2cli.util
if metadata is not None:
prefix = 'm'
if multiple is not None:
multiple = list if multiple == 'list' else set
if is_bool_flag:
yes = q2cli.util.to_cli_name(name)
no = q2cli.util.to_cli_name('no_' + name)
opt = f'--{prefix}-{yes}/--{prefix}-{no}'
elif metadata is not None:
cli_name = q2cli.util.to_cli_name(name)
opt = f'--{prefix}-{cli_name}-file'
if metadata == 'column':
self.q2_extra_dest, self.q2_extra_opts, _ = \
self._parse_decls([f'--{prefix}-{cli_name}-column'], True)
else:
cli_name = q2cli.util.to_cli_name(name)
opt = f'--{prefix}-{cli_name}'
click_type = QIIME2Type(ast, repr, is_output=prefix == 'o')
attrs['metavar'] = metavar
attrs['multiple'] = multiple is not None
attrs['param_decls'] = [opt]
attrs['required'] = default is NoDefault
attrs['help'] = self._add_default(description, default)
if default is not NoDefault:
attrs['default'] = default
# This is to evade clicks __DEBUG__ check
if not is_bool_flag:
attrs['type'] = click_type
else:
attrs['type'] = None
super().__init__(**attrs)
# put things back the way they _should_ be after evading __DEBUG__
self.is_bool_flag = is_bool_flag
self.type = click_type
# attrs we will use elsewhere
self.q2_multiple = multiple
self.q2_prefix = prefix
self.q2_name = name
self.q2_ast = ast
self.q2_metadata = metadata
@property
def meta_help(self):
if self.q2_metadata == 'file':
return 'multiple arguments will be merged'
def _add_default(self, desc, default):
if desc is not None:
desc += ' '
else:
desc = ''
if default is not NoDefault:
if default is None:
desc += '[optional]'
else:
desc += '[default: %r]' % (default,)
return desc
def consume_value(self, ctx, opts):
if self.q2_metadata == 'column':
return self._consume_metadata(ctx, opts)
else:
return super().consume_value(ctx, opts)
def _consume_metadata(self, ctx, opts):
# double consume
md_file = super().consume_value(ctx, opts)
# consume uses self.name, so mutate but backup for after
backup, self.name = self.name, self.q2_extra_dest
md_col = super().consume_value(ctx, opts)
self.name = backup
if (md_col is None) != (md_file is None):
# missing one or the other
if md_file is None:
raise click.MissingParameter(ctx=ctx, param=self)
else:
raise click.MissingParameter(param_hint=self.q2_extra_opts,
ctx=ctx, param=self)
if md_col is None and md_file is None:
return None
else:
return (md_file, md_col)
def get_help_record(self, ctx):
record = super().get_help_record(ctx)
if self.is_bool_flag:
metavar = self.make_metavar()
if metavar:
record = (record[0] + ' ' + self.make_metavar(), record[1])
elif self.q2_metadata == 'column':
opts = (record[0], self.q2_extra_opts[0] + ' COLUMN ')
record = (opts, record[1])
return record
# Override
def add_to_parser(self, parser, ctx):
shared = dict(dest=self.name, nargs=0, obj=self)
if self.q2_metadata == 'column':
parser.add_option(self.opts, action='store', dest=self.name,
nargs=1, obj=self)
parser.add_option(self.q2_extra_opts, action='store',
dest=self.q2_extra_dest, nargs=1, obj=self)
elif self.is_bool_flag:
if self.multiple:
action = 'append_maybe'
else:
action = 'store_maybe'
parser.add_option(self.opts, action=action, const=True,
**shared)
parser.add_option(self.secondary_opts, action=action,
const=False, **shared)
elif self.multiple:
action = 'append_greedy'
parser.add_option(self.opts, action='append_greedy', **shared)
else:
super().add_to_parser(parser, ctx)
def full_process_value(self, ctx, value):
try:
return super().full_process_value(ctx, value)
except click.MissingParameter:
if not (self.q2_prefix == 'o'
and ctx.params.get('output_dir', False)):
raise
def type_cast_value(self, ctx, value):
import sys
import q2cli.util
import qiime2.sdk.util
if self.multiple:
if value == () or value is None:
return None
elif self.q2_prefix == 'i':
value = super().type_cast_value(ctx, value)
if self.q2_multiple is set:
self._check_length(value, ctx)
value = self.q2_multiple(value)
type_expr = qiime2.sdk.util.type_from_ast(self.q2_ast)
args = ', '.join(map(repr, (x.type for x in value)))
if value not in type_expr:
raise click.BadParameter(
'received <%s> as an argument, which is incompatible'
' with parameter type: %r' % (args, type_expr),
ctx=ctx, param=self)
return value
elif self.q2_metadata == 'file':
value = super().type_cast_value(ctx, value)
if len(value) == 1:
return value[0]
else:
try:
return value[0].merge(*value[1:])
except Exception as e:
header = ("There was an issue with merging "
"QIIME 2 Metadata:")
tb = 'stderr' if '--verbose' in sys.argv else None
q2cli.util.exit_with_error(
e, header=header, traceback=tb)
elif self.q2_prefix == 'p':
try:
if self.q2_multiple is set:
self._check_length(value, ctx)
value = qiime2.sdk.util.parse_primitive(self.q2_ast, value)
except ValueError:
args = ', '.join(map(repr, value))
expr = qiime2.sdk.util.type_from_ast(self.q2_ast)
raise click.BadParameter(
'received <%s> as an argument, which is incompatible'
' with parameter type: %r' % (args, expr),
ctx=ctx, param=self)
return value
return super().type_cast_value(ctx, value)
def _check_length(self, value, ctx):
import collections
counter = collections.Counter(value)
dups = ', '.join(map(repr, (v for v, n in counter.items() if n > 1)))
args = ', '.join(map(repr, value))
if dups:
raise click.BadParameter(
'received <%s> as an argument, which contains duplicates'
' of the following: <%s>' % (args, dups), ctx=ctx, param=self)