<commit_before>"""
Align multiple sequences from FASTA.
"""

import os
from shutil import copyfile
import numpy as np
from Bio import AlignIO, SeqIO, Seq, Align
from .utils import run_shell_command, nthreads_value, shquote

class AlignmentError(Exception):
    # TODO: this exception should potentially be renamed and made augur-wide
    # thus allowing any module to raise it and have the message printed & augur
    # exit with code 1
    pass

def register_arguments(parser):
    parser.add_argument('--sequences', '-s', required=True, nargs="+", metavar="FASTA", help="sequences to align")
    parser.add_argument('--output', '-o', default="alignment.fasta", help="output file (default: %(default)s)")
    parser.add_argument('--nthreads', type=nthreads_value, default=1,
                                help="number of threads to use; specifying the value 'auto' will cause the number of available CPU cores on your system, if determinable, to be used")
    parser.add_argument('--method', default='mafft', choices=["mafft"], help="alignment program to use")
    parser.add_argument('--reference-name', metavar="NAME", type=str, help="strip insertions relative to reference sequence; use if the reference is already in the input sequences")
    parser.add_argument('--reference-sequence', metavar="PATH", type=str, help="Add this reference sequence to the dataset & strip insertions relative to this. Use if the reference is NOT already in the input sequences")
    parser.add_argument('--remove-reference', action="store_true", default=False, help="remove reference sequence from the alignment")
    parser.add_argument('--fill-gaps', action="store_true", default=False, help="If gaps represent missing data rather than true indels, replace by N after aligning.")
    parser.add_argument('--existing-alignment', metavar="FASTA", default=False, help="An existing alignment to which the sequences will be added. The ouput alignment will be the same length as this existing alignment.")
    parser.add_argument('--debug', action="store_true", default=False, help="Produce extra files (e.g. pre- and post-aligner files) which can help with debugging poor alignments.")

def prepare(sequences, existing_alignment, output, ref_name, ref_seq_fname):
    seqs = read_sequences(*sequences)
    seqs_to_align_fname = output + "to_align.fasta"

    # Load existing alignment
    existing_aln = None
    existing_aln_fname = None
    if existing_alignment:
        existing_aln_fname = existing_alignment
        existing_aln = read_alignment(existing_alignment)

    # Load reference alignment
    if ref_name:
        ensure_reference_strain_present(ref_name, existing_aln, seqs)
    elif ref_seq_fname:
        ref_seq = read_reference(ref_seq_fname)
        ref_name = ref_seq.id

    if existing_aln:
        # Strip the existing sequences from the new sequences, add the reference to the alignment
        seqs_to_align_fname = output + "new_seqs_to_align.fasta"
        seqs = prune_seqs_matching_alignment(seqs, existing_aln)
        write_seqs(list(seqs.values()), seqs_to_align_fname)
        if ref_seq:
            if len(ref_seq) != existing_aln.get_alignment_length():
                raise AlignmentError("ERROR: Provided existing alignment ({}bp) is not the same length as the reference sequence ({}bp)".format(existing_aln.get_alignment_length(), len(ref_seq)))
            existing_aln_fname = existing_aln_fname + ".ref.fasta"
            existing_aln.append(ref_seq)
            write_seqs(existing_aln, existing_aln_fname)
    elif ref_seq: # Got a reference sequence but no existing alignment
        # reference sequence needs to be the first one for auto direction
        # adjustment (auto reverse-complement)
        write_seqs([ref_seq] + list(seqs.values()), seqs_to_align_fname)
        seqs[ref_name] = ref_seq
    else:
        write_seqs(list(seqs.values()), seqs_to_align_fname)

    check_duplicates(existing_aln, seqs)
    return existing_aln_fname, seqs_to_align_fname, ref_name



def run(args):
    '''
    Parameters
    ----------
    args : namespace
        arguments passed in via the command-line from augur

    Returns
    -------
    int
        returns 0 for success, 1 for general error
    '''
    temp_files_to_remove = []

    try:
        check_arguments(args)
        existing_aln_fname, seqs_to_align_fname, ref_name = prepare(args.sequences, args.existing_alignment, args.output, args.reference_name, args.reference_sequence)
        # -- existing_aln_fname, seqs_to_align_fname, ref_name --

        # before aligning, make a copy of the data that the aligner receives as input (very useful for debugging purposes)<commit_msg>clean up existing_alignment in prepare()<commit_after>"""
Align multiple sequences from FASTA.
"""

import os
from shutil import copyfile
import numpy as np
from Bio import AlignIO, SeqIO, Seq, Align
from .utils import run_shell_command, nthreads_value, shquote

class AlignmentError(Exception):
    # TODO: this exception should potentially be renamed and made augur-wide
    # thus allowing any module to raise it and have the message printed & augur
    # exit with code 1
    pass

def register_arguments(parser):
    parser.add_argument('--sequences', '-s', required=True, nargs="+", metavar="FASTA", help="sequences to align")
    parser.add_argument('--output', '-o', default="alignment.fasta", help="output file (default: %(default)s)")
    parser.add_argument('--nthreads', type=nthreads_value, default=1,
                                help="number of threads to use; specifying the value 'auto' will cause the number of available CPU cores on your system, if determinable, to be used")
    parser.add_argument('--method', default='mafft', choices=["mafft"], help="alignment program to use")
    parser.add_argument('--reference-name', metavar="NAME", type=str, help="strip insertions relative to reference sequence; use if the reference is already in the input sequences")
    parser.add_argument('--reference-sequence', metavar="PATH", type=str, help="Add this reference sequence to the dataset & strip insertions relative to this. Use if the reference is NOT already in the input sequences")
    parser.add_argument('--remove-reference', action="store_true", default=False, help="remove reference sequence from the alignment")
    parser.add_argument('--fill-gaps', action="store_true", default=False, help="If gaps represent missing data rather than true indels, replace by N after aligning.")
    parser.add_argument('--existing-alignment', metavar="FASTA", default=False, help="An existing alignment to which the sequences will be added. The ouput alignment will be the same length as this existing alignment.")
    parser.add_argument('--debug', action="store_true", default=False, help="Produce extra files (e.g. pre- and post-aligner files) which can help with debugging poor alignments.")

def prepare(sequences, existing_aln_fname, output, ref_name, ref_seq_fname):
    seqs = read_sequences(*sequences)
    seqs_to_align_fname = output + "to_align.fasta"

    # Load existing alignment
    existing_aln = None
    if existing_aln_fname:
        existing_aln = read_alignment(existing_aln_fname)

    # Load reference alignment
    if ref_name:
        ensure_reference_strain_present(ref_name, existing_aln, seqs)
    elif ref_seq_fname:
        ref_seq = read_reference(ref_seq_fname)
        ref_name = ref_seq.id

    if existing_aln:
        # Strip the existing sequences from the new sequences, add the reference to the alignment
        seqs_to_align_fname = output + "new_seqs_to_align.fasta"
        seqs = prune_seqs_matching_alignment(seqs, existing_aln)
        write_seqs(list(seqs.values()), seqs_to_align_fname)
        if ref_seq:
            if len(ref_seq) != existing_aln.get_alignment_length():
                raise AlignmentError("ERROR: Provided existing alignment ({}bp) is not the same length as the reference sequence ({}bp)".format(existing_aln.get_alignment_length(), len(ref_seq)))
            existing_aln_fname = existing_aln_fname + ".ref.fasta"
            existing_aln.append(ref_seq)
            write_seqs(existing_aln, existing_aln_fname)
    elif ref_seq: # Got a reference sequence but no existing alignment
        # reference sequence needs to be the first one for auto direction
        # adjustment (auto reverse-complement)
        write_seqs([ref_seq] + list(seqs.values()), seqs_to_align_fname)
        seqs[ref_name] = ref_seq
    else:
        write_seqs(list(seqs.values()), seqs_to_align_fname)

    check_duplicates(existing_aln, seqs)
    return existing_aln_fname, seqs_to_align_fname, ref_name

def run(args):
    '''
    Parameters
    ----------
    args : namespace
        arguments passed in via the command-line from augur

    Returns
    -------
    int
        returns 0 for success, 1 for general error
    '''
    temp_files_to_remove = []

    try:
        check_arguments(args)
        existing_aln_fname, seqs_to_align_fname, ref_name = prepare(args.sequences, args.existing_alignment, args.output, args.reference_name, args.reference_sequence)
        # -- existing_aln_fname, seqs_to_align_fname, ref_name --

        # before aligning, make a copy of the data that the aligner receives as input (very useful for debugging purposes)
{'fraction_non_alphanumeric': 0.09373320434268516, 'fraction_numerical': 0.0008599376545200473, 'mean_word_length': 3.8382735309412377, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '1622985', 'n_tokens_mistral': 2733, 'n_tokens_neox': 2650, 'n_words': 817}