<commit_before># ----------------------------------------------------------------------------
# Copyright (c) 2015--, micronota development team.
#
# Distributed under the terms of the Modified BSD License.
#
# The full license is in the file COPYING.txt, distributed with this software.
# ----------------------------------------------------------------------------

from os import makedirs
from os.path import join, basename, splitext
import re

from skbio import Sequence
from skbio.io.format.genbank import _parse_features
from burrito.parameters import FlagParameter, ValuedParameter
from burrito.util import CommandLineApplication, ResultPath

from ..parsers.embl import _parse_records
from ..util import _tmp_file


class Prodigal(CommandLineApplication):
    '''Prodigal (version 2.6.2) application controller.'''
    _command = 'prodigal'
    _valued_path_options = [
        # Specify FASTA/Genbank input file (default reads from stdin).
        '-i',
        # Write protein translations to the selected file.
        '-a',
        # Write nucleotide sequences of genes to the selected file.
        '-d',
        # Write all potential genes (with scores) to the selected file.
        '-s',
        # Write a training file (if none exists);
        # otherwise, read and use the specified training file.
        '-t',
        # Specify output file (default writes to stdout).
        '-o',
    ]
    _valued_nonpath_options = [
        # Select output format (gbk, gff, or sco).  Default is gbk.
        '-f',
        # Specify a translation table to use (default 11).
        '-g',
        # Treat runs of N as masked sequence; don't build genes across them.
        '-m',
        # Select procedure (single or meta).  Default is single.
        '-p',
    ]
    _flag_options = [
        # Closed ends.  Do not allow genes to run off edges.
        '-c',
        # Print version number and exit.
        '-v',
        # Run quietly (suppress normal stderr output).
        '-q',
        # Print help menu and exit.
        '-h',
        # Bypass Shine-Dalgarno trainer and force a full motif scan.
        '-n',
    ]

    _parameters = {}
    _parameters.update({
        i: ValuedParameter(
            Prefix=i[0], Name=i[1:], Delimiter=' ',
            IsPath=True)
        for i in _valued_path_options})
    _parameters.update({
        i: ValuedParameter(
            Prefix=i[0], Name=i[1:], Delimiter=' ')
        for i in _valued_nonpath_options})
    _parameters.update({
        i: FlagParameter(
            Prefix=i[0], Name=i[1:])
        for i in _flag_options})
    _suppress_stderr = False

    def _accept_exit_status(self, exit_status):
        return exit_status == 0

    def _get_result_paths(self, data):
        result = {}
        for i in self._valued_path_options:
            if i != '-i':
                o = self.Parameters[i]
                if o.isOn():
                    out_fp = self._absolute(o.Value)
                    result[i] = ResultPath(Path=out_fp, IsWritten=True)
        return result


def identify_features(in_fp, out_dir, prefix='prodigal', params=None):
    '''Predict genes for the input file.

    Notes
    -----
    It will create 3 output files:
      1. the annotation file in format of GFF3 or
         GenBank feature table with .gbk suffix.
      2. the nucleotide sequences for each predicted gene
         with file suffix of .fna.
      3. the protein sequence translated from each gene
         with file suffix of .faa.

    Parameters
    ----------
    in_fp : str
        input file path
    out_dir : str
        output directory
    prefix : str
        prefix of output file name (without suffix)
    params : dict
        Other command line parameters for Prodigal. key is the option
        (e.g. "-p") and value is the value for the option (e.g. "single").
        If the option is a flag, set the value to None.

    Returns
    -------
    burrito.util.CommandLineAppResult
        It contains opened file handlers of stdout, stderr, and the 3
        output files, which can be accessed in a dict style with the
        keys of "StdOut", "StdErr", "-o", "-d", "-a". The exit status
        of the run can be similarly fetched with the key of "ExitStatus".
    '''
    # create dir if not exist
    makedirs(out_dir, exist_ok=True)

    if prefix is None:
        prefix = splitext(basename(in_fp))[0]
    if params is None:
        params = {}

    # default is genbank
    f_param = params.get('-f', 'gbk')

    out_suffices = {
        '-a': 'faa',
        # Write nucleotide sequences of genes to the selected file.
        '-d': 'fna',
        '-o': f_param}
    for i in out_suffices:
        out_fp = join(out_dir, '.'.join([prefix, out_suffices[i]]))
        params[i] = out_fp

    params['-i'] = in_fp

    app = Prodigal(params=params)
    return app()


def predict_cds(seq, out_dir, prefix=None, params=None):
    '''Predict genes for the input sequence with ``Prodigal``.

    Parameters
    ----------
    seq : str or ``skbio.Sequence`` or its child classes.
        input fasta
    out_dir : str
        output directory

    Returns
    -------
    ``skbio.Sequence`` or its child classes.
    '''
    if isinstance(seq, str):
        seq = Sequence(seq, metadata={'id': id(seq)})
    if not isinstance(seq, Sequence):
        raise
    with _tmp_file() as f:
        _, fp = f
        seq.write(fp)
        interval_metadata = {}
        if prefix is None:
            prefix = seq.metadata['id']
        res = identify_features(fp, out_dir, prefix, params)
        if res['Exitstatus'] != 0:
            raise RuntimeError(
                'The Prodigal prediction is finished with an error:\n%s' %
                res['StdErr'])
        interval_metadata = next(parse_output(res))
        seq.interval_metadata = interval_metadata
        return seq


def parse_output(res):
    '''Parse gene prediction result from ``Prodigal``.

    It is parsed into ``skbio.sequence.IntervalMetadata``.
    This is only for Prodigal GenBank output.

    Parameters
    ----------
    fh : file handler

    Returns
    -------
    ``IntervalMetadata``
    '''
    # make sure to move to the beginning of the file.
    return _parse_records(res['-o'], _parse_single_record)


def _parse_single_record(chunks):
    '''Parse single record of Prodigal GenBank output.

    Parameters
    ----------
    chunks : list of str
        a list of lines of the record to parse.

    Returns
    -------
    ``skbio.sequence.IntervalMetadata``
    '''
    # get the head line
    head = chunks[0]
    _, description = head.split(None, 1)
    pattern = re.compile(r'''((?:[^;"']|"[^"]*"|'[^']*')+)''')
    desc = {}
    for i in pattern.findall(description):
        k, v = i.split('=', 1)
        desc[k] = v
    return _parse_features(chunks[1:], int(desc['seqlen']))<commit_msg>add _parse_faa()<commit_after># ----------------------------------------------------------------------------
# Copyright (c) 2015--, micronota development team.
#
# Distributed under the terms of the Modified BSD License.
#
# The full license is in the file COPYING.txt, distributed with this software.
# ----------------------------------------------------------------------------

from os import makedirs
from os.path import join, basename, splitext
from tempfile import NamedTemporaryFile
import re

from skbio import Sequence, read
from skbio.metadata import Feature
from skbio.io.format.genbank import _parse_features
from burrito.parameters import FlagParameter, ValuedParameter
from burrito.util import CommandLineApplication, ResultPath


class Prodigal(CommandLineApplication):
    '''Prodigal (version 2.6.2) application controller.'''
    _command = 'prodigal'
    _valued_path_options = [
        # Specify FASTA/Genbank input file (default reads from stdin).
        '-i',
        # Write protein translations to the selected file.
        '-a',
        # Write nucleotide sequences of genes to the selected file.
        '-d',
        # Write all potential genes (with scores) to the selected file.
        '-s',
        # Write a training file (if none exists);
        # otherwise, read and use the specified training file.
        '-t',
        # Specify output file (default writes to stdout).
        '-o',
    ]
    _valued_nonpath_options = [
        # Select output format (gbk, gff, or sco).  Default is gbk.
        '-f',
        # Specify a translation table to use (default 11).
        '-g',
        # Treat runs of N as masked sequence; don't build genes across them.
        '-m',
        # Select procedure (single or meta).  Default is single.
        '-p',
    ]
    _flag_options = [
        # Closed ends.  Do not allow genes to run off edges.
        '-c',
        # Print version number and exit.
        '-v',
        # Run quietly (suppress normal stderr output).
        '-q',
        # Print help menu and exit.
        '-h',
        # Bypass Shine-Dalgarno trainer and force a full motif scan.
        '-n',
    ]

    _parameters = {}
    _parameters.update({
        i: ValuedParameter(
            Prefix=i[0], Name=i[1:], Delimiter=' ',
            IsPath=True)
        for i in _valued_path_options})
    _parameters.update({
        i: ValuedParameter(
            Prefix=i[0], Name=i[1:], Delimiter=' ')
        for i in _valued_nonpath_options})
    _parameters.update({
        i: FlagParameter(
            Prefix=i[0], Name=i[1:])
        for i in _flag_options})
    _suppress_stderr = False

    def _accept_exit_status(self, exit_status):
        return exit_status == 0

    def _get_result_paths(self, data):
        result = {}
        for i in self._valued_path_options:
            if i != '-i':
                o = self.Parameters[i]
                if o.isOn():
                    out_fp = self._absolute(o.Value)
                    result[i] = ResultPath(Path=out_fp, IsWritten=True)
        return result


def identify_features(in_fp, out_dir, prefix='prodigal', params=None):
    '''Predict genes for the input file.

    Notes
    -----
    It will create 3 output files:
      1. the annotation file in format of GFF3 or
         GenBank feature table with .gbk suffix.
      2. the nucleotide sequences for each predicted gene
         with file suffix of .fna.
      3. the protein sequence translated from each gene
         with file suffix of .faa.

    Parameters
    ----------
    in_fp : str
        input file path
    out_dir : str
        output directory
    prefix : str
        prefix of output file name (without suffix)
    params : dict
        Other command line parameters for Prodigal. key is the option
        (e.g. "-p") and value is the value for the option (e.g. "single").
        If the option is a flag, set the value to None.

    Returns
    -------
    burrito.util.CommandLineAppResult
        It contains opened file handlers of stdout, stderr, and the 3
        output files, which can be accessed in a dict style with the
        keys of "StdOut", "StdErr", "-o", "-d", "-a". The exit status
        of the run can be similarly fetched with the key of "ExitStatus".
    '''
    # create dir if not exist
    makedirs(out_dir, exist_ok=True)

    if prefix is None:
        prefix = splitext(basename(in_fp))[0]
    if params is None:
        params = {}

    # default is genbank
    f_param = params.get('-f', 'gbk')

    out_suffices = {
        '-a': 'faa',
        # Write nucleotide sequences of genes to the selected file.
        '-d': 'fna',
        '-o': f_param}
    for i in out_suffices:
        out_fp = join(out_dir, '.'.join([prefix, out_suffices[i]]))
        params[i] = out_fp

    params['-i'] = in_fp
    params['-p'] = 'meta'
    app = Prodigal(params=params)
    return app()


def predict_cds(seq, out_dir, prefix=None, params=None):
    '''Predict genes for the input sequence with ``Prodigal``.

    Parameters
    ----------
    seq : str or ``skbio.Sequence`` or its child classes.
        input fasta
    out_dir : str
        output directory

    Returns
    -------
    ``skbio.Sequence`` or its child classes.
    '''
    if isinstance(seq, str):
        seq = Sequence(seq, metadata={'id': id(seq)})
    if not isinstance(seq, Sequence):
        raise
    with NamedTemporaryFile(mode='w+') as f:
        seq.write(f)
        interval_metadata = {}
        if prefix is None:
            prefix = seq.metadata['id']
        res = identify_features(f.name, out_dir, prefix, params)
        if res['Exitstatus'] != 0:
            raise RuntimeError(
                'The Prodigal prediction is finished with an error:\n%s' %
                res['StdErr'])
        interval_metadata = next(parse_output(res))
        seq.interval_metadata = interval_metadata
        return seq


def parse_output(res):
    '''Parse gene prediction result from ``Prodigal``.

    It is parsed into a dict consumable by``skbio.meta.IntervalMetadata``.

    Parameters
    ----------
    fh : file handler

    Returns
    -------
    ``IntervalMetadata``
    '''
    # make sure to move to the beginning of the file.
    # return _parse_records(res['-o'], _parse_single_record)
    return _parse_faa(res['-a'])


def _parse_single_record(chunks):
    '''Parse single record of Prodigal GenBank output.

    Parameters
    ----------
    chunks : list of str
        a list of lines of the record to parse.

    Returns
    -------
    ``skbio.metadata.IntervalMetadata``
    '''
    # get the head line
    head = chunks[0]
    _, description = head.split(None, 1)
    pattern = re.compile(r'''((?:[^;"']|"[^"]*"|'[^']*')+)''')
    desc = {}
    for i in pattern.findall(description):
        k, v = i.split('=', 1)
        desc[k] = v
    return _parse_features(chunks[1:], int(desc['seqlen']))


def _parse_faa(faa):
    '''Parse the faa output of Prodigal.

    Notice that Prodigal do not predict genes with introns.

    Returns
    -------
    The dict passable to ``skbio.metadata.IntervalMetadata``.
    '''
    im = dict()
    i = 1
    for seq in read(faa, format='fasta'):
        interval = []
        feature = dict()
        feature['translation'] = str(seq)
        feature['type_'] = 'CDS'

        desc = seq.metadata['description']
        pattern = (r'# +([0-9]+)'    # start
                   ' +# +([0-9]+)'   # end
                   ' +# +(-?1)'      # strand
                   ' +# +ID=([0-9]+_[0-9]+);'
                   'partial=([01]{2});'
                   '(.*)')
        matches = re.match(pattern, desc)
        start, end, strand, id, partial, misc = matches.groups()
        # don't forget to convert 0-based
        interval = [(int(start)-1, int(end))]
        feature['note'] = '"%s"' % misc
        feature['id'] = id
        if partial[0] == '0':
            feature['left_partial_'] = False
        else:
            feature['left_partial_'] = True
            start = '<%s' % start
        if partial[1] == '0':
            feature['right_partial_'] = False
        else:
            feature['right_partial_'] = True
            end = '>%s' % end
        location = '{s}..{e}'.format(s=start, e=end)
        if strand == '-1':
            feature['rc_'] = True
            location = 'complement(%s)' % location
        elif strand == '1':
            feature['rc_'] = False
        else:
            raise ValueError('Inappropriate value for strand: %s' % strand)
        feature['location'] = location
        im[Feature(**feature)] = interval
        # ordinal number of the parent seq
        ordinal = int(id.split('_', 1)[0])
        if ordinal > i:
            yield im
            # reset
            i += 1
            im = dict()
    if im:
        # don't forget to return the last one if it is not empty.
        yield im
{'fraction_non_alphanumeric': 0.13544376070544947, 'fraction_numerical': 0.005075176045169067, 'mean_word_length': 1.9322916666666667, 'pattern_counts': {'":': 0, '<': 4, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6321413', 'n_tokens_mistral': 4648, 'n_tokens_neox': 4471, 'n_words': 1504}