python-pbcommand 0.2.17-1 / usr / lib / python2.7 / dist-packages / pbcommand / models / parser.py

""" Commandline Parser for Tools. Supports Tool Contracts

# Author: Michael Kocher
"""
import abc
import os
import logging
import argparse
import functools
import re

# there's a problem with functools32 and jsonschema. This import raise an
# import error.
#import jsonschema

from pbcommand.common_options import (add_base_options_with_emit_tool_contract,
                                      add_subcomponent_versions_option)
from pbcommand.models import SymbolTypes
from .tool_contract import (ToolDriver,
                            InputFileType, OutputFileType,
                            ToolContract, ToolContractTask,
                            ScatterToolContractTask, GatherToolContractTask)

log = logging.getLogger(__name__)

__version__ = "0.1.1"

__all__ = ["PbParser",
           "PyParser",
           "ToolContractParser",
           "get_pbparser",
           "get_scatter_pbparser",
           "get_gather_pbparser"]

RX_TASK_ID = re.compile(r'^([A-z0-9_]*)\.tasks\.([A-z0-9_]*)$')
RX_TASK_OPTION_ID = re.compile(r'^([A-z0-9_]*)\.task_options\.([A-z0-9_\.]*)')


def _to_file_type(format_):
    return "pacbio.file_types.{x}".format(x=format_)


class JsonSchemaTypes(object):
    # array is a native type, but not supported
    BOOL = "boolean"
    INT = "integer"
    NUM = "number"
    STR = "string"
    NULL = "null"
    OBJ = "object"

    # Optional values e.g., Option[String]
    OPT_BOOL = [BOOL, NULL]
    OPT_INT = [INT, NULL]
    OPT_STR = [STR, NULL]
    OPT_NUM = [NUM, NULL]


def _validate_file(label, path):
    if os.path.exists(path):
        return os.path.abspath(path)
    else:
        raise IOError("Unable to find '{x}' file '{p}'".format(x=label, p=path))


def _validate_option_or_cast(dtype, dvalue):
    if isinstance(dvalue, dtype):
        return dvalue
    else:
        # XXX this is almost always going to be the case...
        if isinstance(dvalue, basestring):
            try:
                return dtype(dvalue)
            except ValueError as e:
                pass
        raise TypeError("Invalid option type: '{a}' provided, '{e}' "
                        "expected".format(a=dvalue, e=dtype))


def _validate_option(dtype, dvalue):
    if isinstance(dvalue, dtype):
        return dvalue
    else:
        raise TypeError("Invalid option type: '{a}' provided, '{e}' "
                        "expected".format(a=dvalue, e=dtype))


def _validate_id(prog, idtype, tid):
    if prog.match(tid):
        return tid
    else:
        raise ValueError("Invalid format {t}: '{i}' {p}".format(t=idtype, i=tid, p=repr(prog.pattern)))

_validate_task_id = functools.partial(_validate_id, RX_TASK_ID, 'task id')
_validate_task_option_id = functools.partial(_validate_id, RX_TASK_OPTION_ID,
                                             'task option id')


def to_opt_id(namespace, s):
    return ".".join([namespace, "options", s])


def validate_value(schema, v):
    import jsonschema
    return jsonschema.validate(v, schema)


def is_valid(schema, v):
    """Returns a bool if the schema is valid"""
    import jsonschema
    try:
        validate_value(schema, v)
        return True
    except jsonschema.ValidationError:
        pass
    return False


def validate_schema(f):
    """Deco for validate the returned jsonschema against Draft 4 of the spec"""
    def w(*args, **kwargs):
        schema = f(*args, **kwargs)
        import jsonschema
        _ = jsonschema.Draft4Validator(schema)
        return schema
    return w


def to_option_schema(option_id, dtype_or_dtypes, display_name, description, default_value):
    """
    Simple util factory method
    :param dtype_or_dtypes: single data type or list of data types
    :param option_id: globally unique task option id. Must begin with
    'pbsmrtpipe.task_options.'
    :param display_name: display name of task options
    :param description: Short description of the task options
    :param required: Is the option required.
    """
    # annoying that you can't specify a tuple
    if isinstance(dtype_or_dtypes, tuple):
        dtype_or_dtypes = list(dtype_or_dtypes)

    _validate_task_option_id(option_id)

    # Steps toward moving away from JSON schema as the format, but reuse
    # the jsonschema defined types. Only non-union types are supported.
    pbd = {"option_id": option_id,
           "type": dtype_or_dtypes,
           "default": default_value,
           "name": display_name,
           "description": description}

    d = {'$schema': "http://json-schema.org/draft-04/schema#",
         'type': 'object',
         'title': "JSON Schema for {o}".format(o=option_id),
         'properties': {option_id: {'description': description,
                                    'title': display_name,
                                    'type': dtype_or_dtypes},
                        },
         "pb_option": pbd
         }

    d['required'] = [option_id]
    d['properties'][option_id]['default'] = default_value
    return d


class PbParserBase(object):

    __metaclass__ = abc.ABCMeta

    def __init__(self, tool_id, version, name, description):
        self.tool_id = _validate_task_id(tool_id)
        self.version = version
        self.description = description
        self.name = name

    def __repr__(self):
        _d = dict(k=self.__class__.__name__, i=self.tool_id, v=self.version)
        return "<{k} id:{i} {v} >".format(**_d)

    @abc.abstractmethod
    def add_input_file_type(self, file_type, file_id, name, description):
        """
        Add a mandatory input file parameter.  On the Python argparse side,
        this will be a positional argument.

        :param file_type: file type ID from pbcommand.models.common, e.g.
                          FileTypes.DS_REF
        :param file_id: parameter name, mainly used on argparse side
        :param name: plain-English name
        :param description: help string
        """
        raise NotImplementedError

    @abc.abstractmethod
    def add_output_file_type(self, file_type, file_id, name, description, default_name):
        """
        Add a mandatory output file parameter.  On the Python argparse side,
        this will be a positional argument.

        :param file_type: file type ID from pbcommand.models.common, e.g.
                          FileTypes.DS_REF
        :param file_id: parameter name, mainly used on argparse side
        :param name: plain-English name
        :param description: help string
        :param default_name: tuple of form (base_name, extension) specifying
                             the default output file name
        """
        raise NotImplementedError

    @abc.abstractmethod
    def add_int(self, option_id, option_str, default, name, description):
        """
        Add an optional integer keyword argument (e.g. "--n=10" or "--n 10" on
        the command line).

        :param option_id: fully-qualified option name used in tool contract
                          layer, of form "pbcommand.task_options.my_option"
        :param option_str: shorter parameter name, mainly used in Python
                           argparse layer, but *without* leading dashes
        :param default: default value (must be an actual integer, not None)
        :param name: plain-English name
        :param description: help string
        """
        raise NotImplementedError

    @abc.abstractmethod
    def add_float(self, option_id, option_str, default, name, description):
        """
        Add an optional float keyword argument (e.g. "--n=10" or "--n 10" on
        the command line).

        :param option_id: fully-qualified option name used in tool contract
                          layer, of form "pbcommand.task_options.my_option"
        :param option_str: shorter parameter name, mainly used in Python
                           argparse layer, but *without* leading dashes
        :param default: default value (must be an actual number, not None)
        :param name: plain-English name
        :param description: help string
        """
        raise NotImplementedError

    @abc.abstractmethod
    def add_str(self, option_id, option_str, default, name, description):
        """
        Add a generic keyword argument whose type is a string.

        :param option_id: fully-qualified option name used in tool contract
                          layer, of form "pbcommand.task_options.my_option"
        :param option_str: shorter parameter name, mainly used in Python
                           argparse layer, but *without* leading dashes
        :param default: default value (can be blank, but not None)
        :param name: plain-English name
        :param description: help string
        """
        raise NotImplementedError

    @abc.abstractmethod
    def add_boolean(self, option_id, option_str, default, name, description):
        """
        Add a boolean option.

        :param option_id: fully-qualified option name used in tool contract
                          layer, of form "pbcommand.task_options.my_option"
        :param option_str: shorter parameter name, mainly used in Python
                           argparse layer, but *without* leading dashes
        :param default: specifies the boolean value of this option **if the
                        argument was supplied**, i.e. on the argparse layer,
                        default=True is equivalent to action="store_true"
        :param name: plain-English name
        :param description: help string
        """
        raise NotImplementedError

_validate_argparse_int = functools.partial(_validate_option_or_cast, int)
_validate_argparse_float = functools.partial(_validate_option_or_cast, float)
_validate_argparse_bool = functools.partial(_validate_option_or_cast, bool)
_validate_argparse_str = functools.partial(_validate_option_or_cast, str)


class PyParser(PbParserBase):
    """PbParser backed that supports argparse"""

    def __init__(self, tool_id, version, name, description, subcomponents=()):
        super(PyParser, self).__init__(tool_id, version, name, description)
        self.parser = argparse.ArgumentParser(version=version,
                                              description=description,
                                              formatter_class=argparse.ArgumentDefaultsHelpFormatter,
                                              add_help=True)
        if subcomponents:
            add_subcomponent_versions_option(self.parser, subcomponents)

    def add_input_file_type(self, file_type, file_id, name, description):
        # this will propagate up the label to the exception
        vfunc = functools.partial(_validate_file, file_id)
        self.parser.add_argument(file_id, type=vfunc, help=description)

    def add_output_file_type(self, file_type, file_id, name, description, default_name):
        self.parser.add_argument(file_id, type=str, help=description)

    def add_int(self, option_id, option_str, default, name, description):
        # FIXME Need to better define and validate option_str
        opt = "--" + option_str
        self.parser.add_argument(opt, type=_validate_argparse_int,
                                 help=description,
                                 default=_validate_argparse_int(default))

    def add_float(self, option_id, option_str, default, name, description):
        if isinstance(default, int):
            default = float(default)
        opt = "--" + option_str
        self.parser.add_argument(opt, type=_validate_argparse_float,
                                 help=description,
                                 default=_validate_argparse_float(default))

    def add_str(self, option_id, option_str, default, name, description):
        # Fixme
        opt = "--" + option_str
        self.parser.add_argument(opt, type=_validate_argparse_str,
                                 help=description,
                                 default=_validate_argparse_str(default))

    def add_boolean(self, option_id, option_str, default, name, description):
        """
        Note, the default value is set by NOT setting the option.

        Example, if you have option_str of --my-option with a default value of True,
        if --my-option is NOT provided, the value is True, if the --my-option
        is provided, then the value is false.

        """
        d = {True: "store_true", False: "store_false"}
        opt = '--' + option_str
        self.parser.add_argument(opt, action=d[_validate_argparse_bool(not default)],
                                 help=description)


class ToolContractParser(PbParserBase):
    """Parser to support Emitting and running ToolContracts"""

    def __init__(self, tool_id, version, name, description, task_type, driver, nproc_symbol,
                 resource_types):
        """Keeps the required elements for creating an instance of a
        ToolContract"""
        super(ToolContractParser, self).__init__(tool_id, version, name, description)
        self.input_types = []
        self.output_types = []
        self.options = []
        self.driver = driver
        self.name = name
        self.nproc_symbol = nproc_symbol
        self.resource_types = resource_types
        self.task_type = task_type

    def add_input_file_type(self, file_type, file_id, name, description):
        x = InputFileType(file_type.file_type_id, file_id, name, description)
        self.input_types.append(x)

    def add_output_file_type(self, file_type, file_id, name, description, default_name):
        x = OutputFileType(file_type.file_type_id, file_id, name, description, default_name)
        self.output_types.append(x)

    def add_int(self, option_id, option_str, default, name, description):
        self.options.append(to_option_schema(option_id,
                                             JsonSchemaTypes.INT, name, description,
                                             _validate_option(int, default)))

    def add_float(self, option_id, option_str, default, name, description):
        if isinstance(default, int):
            default = float(default)
        self.options.append(to_option_schema(option_id,
                                             JsonSchemaTypes.NUM, name, description,
                                             _validate_option(float, default)))

    def add_str(self, option_id, option_str, default, name, description):
        self.options.append(to_option_schema(option_id,
                                             JsonSchemaTypes.STR, name, description,
                                             _validate_option(str, default)))

    def add_boolean(self, option_id, option_str, default, name, description):
        self.options.append(to_option_schema(option_id,
                                             JsonSchemaTypes.BOOL, name, description,
                                             _validate_option(bool, default)))

    def to_tool_contract(self):
        # Not a well formed tool contract, must have at least one input and
        # one output
        if not self.input_types and not self.output_types:
            raise ValueError("Malformed tool contract inputs")

        task = ToolContractTask(self.tool_id,
                                self.name,
                                self.description,
                                self.version,
                                self.task_type,
                                self.input_types,
                                self.output_types,
                                self.options,
                                self.nproc_symbol,
                                self.resource_types)
        tc = ToolContract(task, self.driver)
        # this should just return TC, not tc.to_dict()
        return tc


class ScatterToolContractParser(ToolContractParser):
    def __init__(self, tool_id, version, name, description, task_type, driver, nproc_symbol,
                 resource_types, chunk_keys, nchunks):
        super(ScatterToolContractParser, self).__init__(tool_id, version, name, description, task_type, driver,
                                                        nproc_symbol, resource_types)
        self.chunk_keys = chunk_keys
        self.nchunks = nchunks

    def to_tool_contract(self):
        task = ScatterToolContractTask(self.tool_id,
                                       self.name,
                                       self.description,
                                       self.version,
                                       self.task_type,
                                       self.input_types,
                                       self.output_types,
                                       self.options,
                                       self.nproc_symbol,
                                       self.resource_types,
                                       self.chunk_keys,
                                       self.nchunks)
        tc = ToolContract(task, self.driver)
        return tc


class GatherToolContractParser(ToolContractParser):

    def to_tool_contract(self):
        task = GatherToolContractTask(self.tool_id,
                                      self.name,
                                      self.description,
                                      self.version,
                                      self.task_type,
                                      self.input_types,
                                      self.output_types,
                                      self.options,
                                      self.nproc_symbol,
                                      self.resource_types)
        tc = ToolContract(task, self.driver)
        return tc


class PbParser(PbParserBase):
    """
    Wrapper class for managing separate tool contract and argument parsers
    (stored as tool_contract_parser and arg_parser attributes respectively).
    """

    def __init__(self, tool_contract_parser, arg_parser, *parsers):
        """

        :param tool_contract_parser:
        :type tool_contract_parser: ToolContractParser
        :param arg_parser:
        :type arg_parser: PyParser
        :param parsers:
        :return:
        """

        # Tool Contract Parser
        self.tool_contract_parser = tool_contract_parser

        # python wrapper parser.
        self.arg_parser = arg_parser
        # add options, so it will show up via --help
        add_base_options_with_emit_tool_contract(self.arg_parser.parser)

        # a list of other parsers that adhere to the PbParserBase interface
        # can be used.
        self.other_parsers = parsers

        # for now assume parsers have the same version, id, ...
        tool_id = tool_contract_parser.tool_id
        version = tool_contract_parser.version
        name = tool_contract_parser.name
        description = tool_contract_parser.description

        super(PbParser, self).__init__(tool_id, version, name, description)

    @property
    def parsers(self):
        return [self.tool_contract_parser, self.arg_parser]

    def _dispatch(self, f_name, args, kwds):
        for parser in self.parsers:
            f = getattr(parser, f_name)
            f(*args, **kwds)

    def add_input_file_type(self, file_type, file_id, name, description):
        args = file_type, file_id, name, description
        self._dispatch("add_input_file_type", args, {})

    def add_output_file_type(self, file_type, file_id, name, description, default_name):
        args = file_type, file_id, name, description, default_name
        self._dispatch("add_output_file_type", args, {})

    def add_int(self, option_id, option_str, default, name, description):
        args = option_id, option_str, default, name, description
        self._dispatch("add_int", args, {})

    def add_float(self, option_id, option_str, default, name, description):
        args = option_id, option_str, default, name, description
        self._dispatch("add_float", args, {})

    def add_str(self, option_id, option_str, default, name, description):
        args = option_id, option_str, default, name, description
        self._dispatch("add_str", args, {})

    def add_boolean(self, option_id, option_str, default, name, description):
        args = option_id, option_str, default, name, description
        self._dispatch("add_boolean", args, {})

    def to_contract(self):
        return self.tool_contract_parser.to_tool_contract()


def _factory(tool_id, version, name, description, subcomponents):
    def _f(tc_parser):
        arg_parser = PyParser(tool_id, version, name, description, subcomponents=subcomponents)
        return PbParser(tc_parser, arg_parser)
    return _f


def get_pbparser(tool_id, version, name, description, driver_exe, is_distributed=True, nproc=1,
                 resource_types=(), subcomponents=(), serialization='json'):
    """
    Central point of creating a Tool contract that can emit and run tool
    contracts.

    :returns: PbParser object
    """
    driver = ToolDriver(driver_exe, serialization=serialization)
    tc_parser = ToolContractParser(tool_id, version, name, description, is_distributed, driver,
                                   nproc, resource_types)
    return _factory(tool_id, version, name, description, subcomponents)(tc_parser)


def get_scatter_pbparser(tool_id, version, name, description, driver_exe, chunk_keys,
                         is_distributed=True, nproc=1, nchunks=SymbolTypes.MAX_NCHUNKS, resource_types=(),
                         subcomponents=(), serialization='json'):
    """Create a Scatter Tool"""
    driver = ToolDriver(driver_exe, serialization=serialization)
    tc_parser = ScatterToolContractParser(tool_id, version, name, description, is_distributed,
                                          driver, nproc, resource_types, chunk_keys,
                                          nchunks)
    return _factory(tool_id, version, name, description, subcomponents)(tc_parser)


def get_gather_pbparser(tool_id, version, name, description, driver_exe,
                        is_distributed=True, nproc=1, resource_types=(), subcomponents=(), serialization='json'):
    """Create a Gather tool"""
    driver = ToolDriver(driver_exe, serialization=serialization)
    tc_parser = GatherToolContractParser(tool_id, version, name, description,
                                         is_distributed, driver, nproc, resource_types)
    return _factory(tool_id, version, name, description, subcomponents)(tc_parser)