_RANGE_DEFAULT_START = float('-inf')
_RANGE_DEFAULT_END = float('inf')
_RANGE_DEFAULT_INCLUSIVE_START = True
_RANGE_DEFAULT_INCLUSIVE_END = False
class _PrimitivePredicateBase(PredicateTemplate):
class Range(_PrimitivePredicateBase):
class Choices(_PrimitivePredicateBase):
class _PrimitiveTemplateBase(TypeTemplate):
class _Int(_PrimitiveTemplateBase):
class _Str(_PrimitiveTemplateBase):
class _Float(_PrimitiveTemplateBase):
class _Bool(_PrimitiveTemplateBase):
class _Metadata(_PrimitiveTemplateBase):
class _MetadataColumn(_PrimitiveTemplateBase):
class _Categorical(_PrimitiveTemplateBase):
class _Numeric(_PrimitiveTemplateBase):
    def get_kind(self):
    def get_name(self):
    def __init__(self, *args, inclusive_start=_RANGE_DEFAULT_INCLUSIVE_START,
                 inclusive_end=_RANGE_DEFAULT_INCLUSIVE_END):
    def __hash__(self):
    def __eq__(self, other):
    def __repr__(self):
    def is_element(self, value):
    def is_symbol_subtype(self, other):
    def is_symbol_supertype(self, other):
    def collapse_intersection(self, other):
    def iter_boundaries(self):
    def update_ast(self, ast):
def Start(start, inclusive=_RANGE_DEFAULT_INCLUSIVE_START):
def End(end, inclusive=_RANGE_DEFAULT_INCLUSIVE_END):
    def __init__(self, *choices):
    def __hash__(self):
    def __eq__(self, other):
    def __repr__(self):
    def is_element(self, value):
    def is_symbol_subtype(self, other):
    def is_symbol_supertype(self, other):
    def collapse_intersection(self, other):
    def iter_boundaries(self):
    def update_ast(self, ast):
    def unpack_union(self):
    def __eq__(self, other):
    def __hash__(self):
    def get_name(self):
    def get_kind(self):
    def get_field_names(self):
    def validate_field(self, name, field):
    def validate_predicate_expr(self, self_expr, predicate_expr):
    def validate_predicate(self, predicate):
    def is_element(self, value):
    def is_symbol_subtype(self, other):
    def decode(self, string):
    def encode(self, value):
    def is_element(self, value):
    def decode(self, string):
    def encode(self, value):
    def is_symbol_supertype(self, other):
    def is_element(self, value):
    def decode(self, string):
    def encode(self, value):
    def is_element(self, value):
    def validate_predicate(self, predicate):
    def decode(self, string):
    def encode(self, value):
    def is_element(self, value):
    def decode(self, metadata):
    def encode(self, value):
    def is_element_expr(self, self_expr, value):
    def is_element(self, value):
    def get_field_names(self):
    def validate_field(self, name, field):
    def decode(self, value):
    def encode(self, value):
    def get_union_membership_expr(self, self_expr):
    def is_element(self, value):
    def get_union_membership_expr(self, self_expr):
    def is_element(self, value):
def infer_primitive_type(value): class MethodSignature(PipelineSignature):
    builtin_args = ()

    def _assert_valid_outputs(self, outputs):
        super()._assert_valid_outputs(outputs)
        # Assert all output types are semantic types. The parent class is less
        # strict in its output type requirements.
        for output_name, spec in outputs.items():
            if not is_semantic_type(spec.qiime_type):
                raise TypeError(
                    "Output %r must be a semantic QIIME type, not %r" %
                    (output_name, spec.qiime_type))

    def _assert_valid_views(self, inputs, parameters, outputs):
        for name, spec in itertools.chain(inputs.items(),
                                          parameters.items(),
                                          outputs.items()):
            if not spec.has_view_type():
                raise TypeError("Method is missing a function annotation for"
                                " parameter: %r" % name) class Method(Action):
    """QIIME 2 Method"""

    type = 'method'

    # Abstract method implementations:

    def _callable_sig_converter_(self, callable):
        # No conversion necessary.
        return callable

    def _callable_executor_(self, scope, view_args, output_types, provenance):
        output_views = self._callable(**view_args)
        output_views = tuplize(output_views)

        # TODO this won't work if the user has annotated their "view API" to
        # return a `typing.Tuple` with some number of components. Python will
        # return a tuple when there are multiple return values, and this length
        # check will fail because the tuple as a whole should be matched up to
        # a single output type instead of its components. This is an edgecase
        # due to how Python handles multiple returns, and can be worked around
        # by using something like `typing.List` instead.
        if len(output_views) != len(output_types):
            raise TypeError(
                "Number of output views must match number of output "
                "semantic types: %d != %d"
                % (len(output_views), len(output_types)))

        output_artifacts = []
        for output_view, (name, spec) in zip(output_views,
                                             output_types.items()):
            if type(output_view) is not spec.view_type:
                raise TypeError(
                    "Expected output view type %r, received %r" %
                    (spec.view_type.__name__, type(output_view).__name__))

            prov = provenance.fork(name)
            scope.add_reference(prov)

            artifact = qiime2.sdk.Artifact._from_view(
                spec.qiime_type, output_view, spec.view_type, prov)
            scope.add_parent_reference(artifact)

            output_artifacts.append(artifact)

        return tuple(output_artifacts)

    @classmethod
    def _init(cls, callable, inputs, parameters, outputs, plugin_id, name,
              description, input_descriptions, parameter_descriptions,
              output_descriptions, citations, deprecated, examples):
        signature = qtype.MethodSignature(callable, inputs, parameters,
                                          outputs, input_descriptions,
                                          parameter_descriptions,
                                          output_descriptions)
        return super()._init(callable, signature, plugin_id, name, description,
                             citations, deprecated, examples) class Artifact(Result):
    extension = '.qza'

    @classmethod
    def _is_valid_type(cls, type_):
        if qiime2.core.type.is_semantic_type(type_) and type_.is_concrete():
            return True
        else:
            return False

    @classmethod
    def import_data(cls, type, view, view_type=None):
        type_, type = type, __builtins__['type']

        is_format = False
        if isinstance(type_, str):
            type_ = qiime2.sdk.parse_type(type_)

        if isinstance(view_type, str):
            view_type = qiime2.sdk.parse_format(view_type)
            is_format = True

        if view_type is None:
            if type(view) is str or isinstance(view, pathlib.PurePath):
                is_format = True
                pm = qiime2.sdk.PluginManager()
                output_dir_fmt = pm.get_directory_format(type_)
                if pathlib.Path(view).is_file():
                    if not issubclass(output_dir_fmt,
                                      model.SingleFileDirectoryFormatBase):
                        raise qiime2.plugin.ValidationError(
                            "Importing %r requires a directory, not %s"
                            % (output_dir_fmt.__name__, view))
                    view_type = output_dir_fmt.file.format
                else:
                    view_type = output_dir_fmt
            else:
                view_type = type(view)

        format_ = None
        md5sums = None
        if is_format:
            path = pathlib.Path(view)
            if path.is_file():
                md5sums = {path.name: util.md5sum(path)}
            elif path.is_dir():
                md5sums = util.md5sum_directory(path)
            else:
                raise qiime2.plugin.ValidationError(
                    "Path '%s' does not exist." % path)
            format_ = view_type

        provenance_capture = archive.ImportProvenanceCapture(format_, md5sums)
        return cls._from_view(type_, view, view_type, provenance_capture,
                              validate_level='max')

    @classmethod
    def _from_view(cls, type, view, view_type, provenance_capture,
                   validate_level='min'):
        type_raw = type
        if isinstance(type, str):
            type = qiime2.sdk.parse_type(type)

        if not cls._is_valid_type(type):
            raise TypeError(
                "An artifact requires a concrete semantic type, not type %r."
                % type)

        pm = qiime2.sdk.PluginManager()
        output_dir_fmt = pm.get_directory_format(type)

        if view_type is None:
            # lookup default format for the type
            view_type = output_dir_fmt

        from_type = transform.ModelType.from_view_type(view_type)
        to_type = transform.ModelType.from_view_type(output_dir_fmt)

        recorder = provenance_capture.transformation_recorder('return')
        transformation = from_type.make_transformation(to_type,
                                                       recorder=recorder)
        result = transformation(view, validate_level)

        if type_raw in pm.validators:
            validation_object = pm.validators[type]
            validation_object(data=result, level=validate_level)

        artifact = cls.__new__(cls)
        artifact._archiver = archive.Archiver.from_data(
            type, output_dir_fmt,
            data_initializer=result.path._move_or_copy,
            provenance_capture=provenance_capture)
        return artifact

    def view(self, view_type):
        return self._view(view_type)

    def _view(self, view_type, recorder=None):
        if view_type is qiime2.Metadata and not self.has_metadata():
            raise TypeError(
                "Artifact %r cannot be viewed as QIIME 2 Metadata." % self)

        from_type = transform.ModelType.from_view_type(self.format)
        to_type = transform.ModelType.from_view_type(view_type)

        transformation = from_type.make_transformation(to_type,
                                                       recorder=recorder)
        result = transformation(self._archiver.data_dir)

        if view_type is qiime2.Metadata:
            result._add_artifacts([self])

        to_type.set_user_owned(result, True)
        return result

    def has_metadata(self):
        """ Checks for metadata within an artifact

        Returns
        -------
        bool
           True if the artifact has metadata (i.e. can be viewed as
           ``qiime2.Metadata``), False otherwise.

        """
        from_type = transform.ModelType.from_view_type(self.format)
        to_type = transform.ModelType.from_view_type(qiime2.Metadata)
        return from_type.has_transformation(to_type)

    def validate(self, level='max'):
        """ Validates the data contents of an artifact

        Raises
        ------
        ValidationError
            If the artifact is invalid at the specified level of validation.
        """
        super().validate()

        self.format.validate(self.view(self.format), level) class Results(tuple):
    """Tuple class representing the named results of an ``Action``.

    Provides an interface similar to a ``namedtuple`` type (e.g. fields are
    accessible as attributes).

    Users should not need to instantiate this class directly.

    """

    # Subclassing `tuple` requires `__new__` override.
    def __new__(cls, fields, values):
        fields = tuple(fields)
        values = tuple(values)

        if len(fields) != len(values):
            raise ValueError(
                "`fields` and `values` must have matching length: %d != %d" %
                (len(fields), len(values)))

        # Create tuple instance, store fields, and create read-only attributes
        # for each field name. Fields must be stored for pickling/copying (see
        # `__getnewargs__`).
        #
        # Note: setting field names as attributes allows for tab-completion in
        # interactive contexts! Using `__getattr__` does not support this.
        self = super().__new__(cls, values)

        # Must set attributes this way because `__setattr__` prevents
        # setting directly (necessary for immutability).
        object.__setattr__(self, '_fields', fields)

        # Attach field names as instance attributes.
        for field, value in zip(fields, values):
            object.__setattr__(self, field, value)

        return self

    def __getnewargs__(self):
        """Arguments to pass to `__new__`. Used by copy and pickle."""
        # `tuple(self)` returns `values`.
        return self._fields, tuple(self)

    # `__setattr__` and `__delattr__` must be defined to prevent users from
    # creating or deleting attributes after this class has been instantiated.
    # `tuple` and `namedtuple` do not have this problem because they are
    # immutable (`__slots__ = ()`). We cannot make this class immutable because
    # we cannot define nonempty `__slots__` when subclassing `tuple`, and we
    # need the `_fields` attribute. We work around this issue by disallowing
    # setting and deleting attributes. The error messages here match those
    # raised by `namedtuple` in Python 3.5.1.
    def __setattr__(self, name, value):
        raise AttributeError("can't set attribute")

    def __delattr__(self, name):
        raise AttributeError("can't delete attribute")

    def __eq__(self, other):
        # Results with different field names should not compare equal, even if
        # their values are equal.
        return (
            isinstance(other, Results) and
            self._fields == other._fields and
            tuple(self) == tuple(other)
        )

    def __ne__(self, other):
        return not (self == other)

    def __repr__(self):
        # It is possible to provide an evalable repr but this type of repr does
        # not make the field/value pairs apparent. If the constructor accepted
        # **kwargs, the order of field/value pairs would be lost.
        lines = []
        lines.append('%s (name = value)' % self.__class__.__name__)
        lines.append('')

        max_len = -1
        for field in self._fields:
            if len(field) > max_len:
                max_len = len(field)

        for field, value in zip(self._fields, self):
            field_padding = ' ' * (max_len - len(field))
            lines.append('%s%s = %r' % (field, field_padding, value))

        max_len = -1
        for line in lines:
            if len(line) > max_len:
                max_len = len(line)
        lines[1] = '-' * max_len

        return '\n'.join(lines)

    def _asdict(self):
        return dict(zip(self._fields, self)) def concatenate_ints(ints1: list, ints2: list, ints3: list, int1: int,
                     int2: int) -> list:
    return ints1 + ints2 + ints3 + [int1] + [int2] def merge_mappings(mapping1: dict, mapping2: dict) -> dict:
    merged = mapping1.copy()
    for key, value in mapping2.items():
        if key in merged and merged[key] != value:
            raise ValueError(
                "Key %r exists in `mapping1` and `mapping2` with conflicting "
                "values: %r != %r" % (key, merged[key], value))
        merged[key] = value
    return merged def params_only_method(name: str, age: int) -> dict:
    return {name: age} def no_input_method() -> dict:
    return {'foo': 42} C1 = plugin.SemanticType('C1', field_names='first')
C2 = plugin.SemanticType('C2', field_names=['first', 'second'],
                         variant_of=C1.field['first'],
                         field_members={'first': [C1], 'second': [C1]})
C3 = plugin.SemanticType('C3', field_names=['first', 'second', 'third'],
                         variant_of=[C1.field['first'], C2.field['first'],
                                     C2.field['second']],
                         field_members={'first': [C1, C2],
                                        'second': [C1, C2],
                                        'third': [C1, C2]}) def get_dummy_plugin():
    plugin_manager = qiime2.sdk.PluginManager()
    if 'dummy-plugin' not in plugin_manager.plugins:
        raise RuntimeError(
            "When running QIIME 2 unit tests, the QIIMETEST environment "
            "variable must be defined so that plugins required by unit tests "
            "are loaded. The value of the QIIMETEST environment variable can "
            "be anything. Example command: QIIMETEST=1 nosetests")
    return plugin_manager.plugins['dummy-plugin']
import collections
import concurrent.futures
import inspect
import unittest
import uuid
import qiime2.plugin
from qiime2.core.type import MethodSignature, Int
from qiime2.sdk import Artifact, Method, Results
from qiime2.core.testing.method import (concatenate_ints, merge_mappings,
                                        params_only_method, no_input_method)
from qiime2.core.testing.type import (
    IntSequence1, IntSequence2, SingleInt, Mapping)
from qiime2.core.testing.util import get_dummy_plugin

        # Test properties of the `Results` object.
        self.assertIsInstance(result, tuple)
        self.assertIsInstance(result, Results)
        self.assertEqual(len(result), 1)
        self.assertEqual(result.merged_mapping.view(dict),
                         {'foo': 'abc', 'bar': 'def', 'bazz': 'abc'})

        result = result[0]

        self.assertIsInstance(result, Artifact)
        self.assertEqual(result.type, Mapping)

        self.assertIsInstance(result.uuid, uuid.UUID)

        self.assertEqual(result.view(dict),
                         {'foo': 'abc', 'bar': 'def', 'bazz': 'abc'})

    def test_call_with_parameters_only(self):
        params_only_method = self.plugin.methods['params_only_method']

        result, = params_only_method("Someone's Name", 999)

        self.assertIsInstance(result, Artifact)
        self.assertEqual(result.type, Mapping)
        self.assertIsInstance(result.uuid, uuid.UUID)
        self.assertEqual(result.view(dict), {"Someone's Name": '999'})

    def test_call_without_inputs_or_parameters(self):
        no_input_method = self.plugin.methods['no_input_method']

        result, = no_input_method()

        self.assertIsInstance(result, Artifact)
        self.assertEqual(result.type, Mapping)
        self.assertIsInstance(result.uuid, uuid.UUID)
        self.assertEqual(result.view(dict), {'foo': '42'})

    def test_call_with_optional_artifacts(self):
        method = self.plugin.methods['optional_artifacts_method']

        ints1 = Artifact.import_data(IntSequence1, [0, 42, 43])
        ints2 = Artifact.import_data(IntSequence1, [99, -22])
        ints3 = Artifact.import_data(IntSequence2, [43, 43])

        # No optional artifacts provided.
        obs = method(ints1, 42).output

        self.assertEqual(obs.view(list), [0, 42, 43, 42])

        # One optional artifact provided.
        obs = method(ints1, 42, optional1=ints2).output

        self.assertEqual(obs.view(list), [0, 42, 43, 42, 99, -22])

        # All optional artifacts provided.
        obs = method(
            ints1, 42, optional1=ints2, optional2=ints3, num2=111).output

        self.assertEqual(obs.view(list), [0, 42, 43, 42, 99, -22, 43, 43, 111])

        # Invalid type provided as optional artifact.
        with self.assertRaisesRegex(TypeError,
                                    'type IntSequence1.*type IntSequence2'):
            method(ints1, 42, optional1=ints3)

    def test_call_with_variadic_inputs(self):
        method = self.plugin.methods['variadic_input_method']

        ints = [Artifact.import_data(IntSequence1, [1, 2, 3]),
                Artifact.import_data(IntSequence2, [4, 5, 6])]
        int_set = {Artifact.import_data(SingleInt, 7),
                   Artifact.import_data(SingleInt, 8)}
        nums = {9, 10}
        opt_nums = [11, 12, 13]

        result, = method(ints, int_set, nums, opt_nums)

        self.assertEqual(result.view(list), list(range(1, 14)))

    def test_asynchronous(self):
        concatenate_ints = self.plugin.methods['concatenate_ints']

        artifact1 = Artifact.import_data(IntSequence1, [0, 42, 43])
        artifact2 = Artifact.import_data(IntSequence2, [99, -22])

        future = concatenate_ints.asynchronous(
            artifact1, artifact1, artifact2, 55, 1)

        self.assertIsInstance(future, concurrent.futures.Future)
        result = future.result()

        # Test properties of the `Results` object.
        self.assertIsInstance(result, tuple)
        self.assertIsInstance(result, Results)
        self.assertEqual(len(result), 1)
        self.assertEqual(result.concatenated_ints.view(list),
                         [0, 42, 43, 0, 42, 43, 99, -22, 55, 1])

        result = result[0]

        self.assertIsInstance(result, Artifact)
        self.assertEqual(result.type, IntSequence1)

        self.assertIsInstance(result.uuid, uuid.UUID)

        # Can retrieve multiple views of different type.
        exp_list_view = [0, 42, 43, 0, 42, 43, 99, -22, 55, 1]
        self.assertEqual(result.view(list), exp_list_view)
        self.assertEqual(result.view(list), exp_list_view)

        exp_counter_view = collections.Counter(
            {0: 2, 42: 2, 43: 2, 99: 1, -22: 1, 55: 1, 1: 1})
        self.assertEqual(result.view(collections.Counter),
                         exp_counter_view)
        self.assertEqual(result.view(collections.Counter),
                         exp_counter_view)

        # Accepts IntSequence1 | IntSequence2
        artifact3 = Artifact.import_data(IntSequence2, [10, 20])

        future = concatenate_ints.asynchronous(artifact3, artifact1, artifact2,