jpayne@69: from typing import (
jpayne@69:     Callable,
jpayne@69:     List,
jpayne@69:     Optional,
jpayne@69:     Tuple,
jpayne@69:     Iterable,
jpayne@69:     Union,
jpayne@69: )
jpayne@69: 
jpayne@69: from pysam.libcutils import _pysam_dispatch
jpayne@69: 
jpayne@69: 
jpayne@69: class SamtoolsError(Exception):
jpayne@69:     '''exception raised in case of an error incurred in the samtools
jpayne@69:     library.'''
jpayne@69: 
jpayne@69:     def __init__(self, value):
jpayne@69:         self.value = value
jpayne@69: 
jpayne@69:     def __str__(self):
jpayne@69:         return repr(self.value)
jpayne@69: 
jpayne@69: 
jpayne@69: class PysamDispatcher(object):
jpayne@69:     '''The dispatcher emulates the samtools/bctools command line.
jpayne@69: 
jpayne@69:     Captures stdout and stderr.
jpayne@69: 
jpayne@69:     Raises a :class:`pysam.SamtoolsError` exception in case samtools
jpayne@69:     exits with an error code other than 0.
jpayne@69: 
jpayne@69:     Some command line options are associated with parsers.  For
jpayne@69:     example, the samtools command "pileup -c" creates a tab-separated
jpayne@69:     table on standard output. In order to associate parsers with
jpayne@69:     options, an optional list of parsers can be supplied. The list
jpayne@69:     will be processed in order checking for the presence of each
jpayne@69:     option.
jpayne@69: 
jpayne@69:     If no parser is given or no appropriate parser is found, the
jpayne@69:     stdout output of samtools/bcftools commands will be returned.
jpayne@69: 
jpayne@69:     '''
jpayne@69: 
jpayne@69:     dispatch = None
jpayne@69:     parsers = None
jpayne@69:     collection = None
jpayne@69: 
jpayne@69:     def __init__(
jpayne@69:         self,
jpayne@69:         collection: str,
jpayne@69:         dispatch: str,
jpayne@69:         parsers: Optional[Iterable[Tuple[str, Callable[[Union[str, List[str]]], Union[str, List[str]]]]]] = None,
jpayne@69:     ):
jpayne@69:         self.collection = collection
jpayne@69:         self.dispatch = dispatch
jpayne@69:         self.parsers = parsers
jpayne@69:         self.stderr = []
jpayne@69: 
jpayne@69:     def __call__(self, *args: str, **kwargs) -> Union[str, List[str]]:
jpayne@69:         '''
jpayne@69:         execute a samtools command.
jpayne@69: 
jpayne@69:         Keyword arguments:
jpayne@69:         catch_stdout -- redirect stdout from the samtools command and
jpayne@69:             return as variable (default True)
jpayne@69:         save_stdout -- redirect stdout to a filename.
jpayne@69:         raw -- ignore any parsers associated with this samtools command.
jpayne@69:         split_lines -- return stdout (if catch_stdout is True and stderr
jpayne@69:                        as a list of strings.
jpayne@69:         '''
jpayne@69:         retval, stderr, stdout = _pysam_dispatch(
jpayne@69:             self.collection,
jpayne@69:             self.dispatch,
jpayne@69:             args,
jpayne@69:             catch_stdout=kwargs.get("catch_stdout", True),
jpayne@69:             save_stdout=kwargs.get("save_stdout", None))
jpayne@69: 
jpayne@69:         if kwargs.get("split_lines", False):
jpayne@69:             stdout = stdout.splitlines()
jpayne@69:             if stderr:
jpayne@69:                 stderr = stderr.splitlines()
jpayne@69: 
jpayne@69:         if retval:
jpayne@69:             raise SamtoolsError(
jpayne@69:                 "%s returned with error %i: "
jpayne@69:                 "stdout=%s, stderr=%s" %
jpayne@69:                 (self.collection,
jpayne@69:                  retval,
jpayne@69:                  stdout,
jpayne@69:                  stderr))
jpayne@69: 
jpayne@69:         self.stderr = stderr
jpayne@69: 
jpayne@69:         # call parser for stdout:
jpayne@69:         if not kwargs.get("raw") and stdout and self.parsers:
jpayne@69:             for options, parser in self.parsers:
jpayne@69:                 for option in options:
jpayne@69:                     if option not in args:
jpayne@69:                         break
jpayne@69:                 else:
jpayne@69:                     return parser(stdout)
jpayne@69: 
jpayne@69:         return stdout
jpayne@69: 
jpayne@69:     def get_messages(self):
jpayne@69:         return self.stderr
jpayne@69: 
jpayne@69:     def usage(self):
jpayne@69:         '''return the samtools usage information for this command'''
jpayne@69:         retval, stderr, stdout = _pysam_dispatch(
jpayne@69:             self.collection,
jpayne@69:             self.dispatch,
jpayne@69:             is_usage=True,
jpayne@69:             catch_stdout=True)
jpayne@69:         # some tools write usage to stderr, such as mpileup
jpayne@69:         if stderr:
jpayne@69:             return stderr
jpayne@69:         else:
jpayne@69:             return stdout
jpayne@69: 
jpayne@69: 
jpayne@69: class unquoted_str(str):
jpayne@69:     '''Tag a value as an unquoted string. Meta-information in the VCF
jpayne@69:     header takes the form of key=value pairs. By default, pysam will
jpayne@69:     enclose the value in quotation marks. Tagging that value with
jpayne@69:     unquoted_str will prevent this quoting.'''