<commit_before>"""
Check if a mutation is reverted
"""
import argparse
import sys
import revmut
import logging
from revmut import oncotator
from revmut.utils import deprecated

import pyhgvs as hgvs
from Bio import SeqIO


class RevertantMutationsInfo(object):
    def __init__(self, transcript_record, hgvs_mut):
        self.record = transcript_record
        self.mut = hgvs_mut
        self.normal_p = self.record.seq.translate(to_stop=True)
        self.mut_p = apply_hgvs(self.record.seq, hgvs_mut).translate(to_stop=True)
        self.revmuts = []
        self.revmuts_pos_adj = []
        self.revmuts_p = []

    def add_revmut(self, hgvs_revmut):
        assert(len(self.revmuts) == len(self.revmuts_p))
        self.revmuts += [hgvs_revmut]
        hgvs_revmut_pos_adj = alter_coords_hgvs_sequential(self.mut, hgvs_revmut)
        self.revmuts_pos_adj += [hgvs_revmut_pos_adj]
        self.revmuts_p += [apply_hgvs(apply_hgvs(self.record.seq, self.mut), hgvs_revmut_pos_adj).translate(to_stop=True)]

    def to_tsv(self, header=True):
        assert(len(self.revmuts) == len(self.revmuts_p))
        rv = ""
        if header:
            rv += "mut\trevmut\trevmut_pos_adj\ttranscript\tnormal_protein_length\tmut_protein_length\trevmut_protein_length\n"
        rv += "\n".join(["\t".join(["{}"]*7).format(self.mut, self.revmuts[i], self.revmuts_pos_adj[i], self.record.id,
                         len(self.normal_p), len(self.mut_p),
                         len(self.revmuts_p[i])) for i in
                         range(len(self.revmuts))]) + "\n"
        return rv

    def __len__(self):
        return len(self.revmuts)


def alter_coords_hgvs_sequential(h1, h2):
    """Change HGVS coords of h2 after applying h1"""
    if h1.kind == "c" and h2.kind == "c":
        if h1.mutation_type == ">":
            h3 = h2
        elif h1.mutation_type == "del":
            if h1.cdna_start.coord > h2.cdna_end.coord:
                h3 = h2
            elif h1.cdna_end.coord < h2.cdna_start.coord:
                h3 = h2
                h3.cdna_start = hgvs.CDNACoord(coord=h3.cdna_start.coord-len(h1.ref_allele))
                h3.cdna_end = hgvs.CDNACoord(coord=h3.cdna_end.coord-len(h1.ref_allele))
            else:
                raise(Exception("Overlapping del not implemented"))
        elif h1.mutation_type == "ins":
            if h1.cdna_start.coord > h2.cdna_end.coord:
                h3 = h2
            elif h1.cdna_end.coord < h2.cdna_start.coord:
                h3 = h2
                h3.cdna_start = hgvs.CDNACoord(coord=h3.cdna_start.coord+len(h1.alt_allele))
                h3.cdna_end = hgvs.CDNACoord(coord=h3.cdna_end.coord+len(h1.alt_allele))
            else:
                raise(Exception("Overlapping ins not implemented"))
        elif h1.mutation_type == "dup":
            if h1.cdna_start.coord > h2.cdna_end.coord:
                h3 = h2
            elif h1.cdna_end.coord < h2.cdna_start.coord:
                h3 = h2
                h3.cdna_start = hgvs.CDNACoord(coord=h3.cdna_start.coord+len(h1.alt_allele))
                h3.cdna_end = hgvs.CDNACoord(coord=h3.cdna_end.coord+len(h1.alt_allele))
            else:
                raise(Exception("Overlapping dup not implemented"))
        elif h1.mutation_type == "delins":
            if h1.cdna_start.coord > h2.cdna_end.coord:
                h3 = h2
            elif h1.cdna_end.coord < h2.cdna_start.coord:
                h3 = h2
                h3.cdna_start = hgvs.CDNACoord(coord=h3.cdna_start.coord-len(h1.ref_allele)+len(h1.alt_allele))
                h3.cdna_end = hgvs.CDNACoord(coord=h3.cdna_end.coord-len(h1.ref_allele)+len(h1.alt_allele))
            else:
                raise(Exception("Overlapping delins not implemented"))
        else:
            raise(Exception("Unexpected mutation_type {}".format(h1.mutation_type)))
        return h3
    else:
        raise(Exception("Only cDNA mutations have been implemented"))


def apply_hgvs(seq, h):
    """Apply 1-based HGVS mutation to 0-based biopython Sequence"""
    if h.kind == "c":
        start = h.cdna_start.coord - 1
        end = h.cdna_end.coord - 1
        if h.mutation_type == ">":
            assert(seq[start] == h.ref_allele)
            new_seq = seq[:start] + h.alt_allele + seq[start+1:]
        elif h.mutation_type == "del":
            assert(seq[start:end+1] == h.ref_allele)
            new_seq = seq[:start] + seq[end+1:]
            assert(len(seq) == len(new_seq) + len(h.ref_allele))
        elif h.mutation_type == "ins":
            new_seq = seq[:start+1] + h.alt_allele + seq[end:]
            assert(len(seq) + len(h.alt_allele) == len(new_seq))
        elif h.mutation_type == "dup":
            assert(seq[start:end+1] == h.ref_allele)
            new_seq = seq[:end+1] + h.alt_allele + seq[end+1:]
        elif h.mutation_type == "delins":
            assert(seq[start:end+1] == h.ref_allele)
            new_seq = seq[:start] + h.alt_allele + seq[end+1:]
            assert(len(seq) - len(h.ref_allele) + len(h.alt_allele) == len(new_seq))
        else:
            raise(Exception("Unexpected mutation_type {}".format(h.mutation_type)))
        return new_seq
    else:
        raise(Exception("Only cDNA mutations have been implemented"))


@deprecated
def is_revertant(record, hgvs_mut, hgvs_rev_mut):
    normal_p = record.seq.translate(to_stop=True)
    mut_p = apply_hgvs(record.seq, hgvs_mut).translate(to_stop=True)
    revmut_p = apply_hgvs(apply_hgvs(record.seq, hgvs_mut), hgvs_rev_mut).translate(to_stop=True)
    logging.info("---")
    logging.info("mutation: {}".format(hgvs_mut))
    logging.info("putative revertant mutation: {}".format(hgvs_rev_mut))
    logging.info("transcript: {}".format(record.id))
    logging.info("normal length: {}".format(len(normal_p)))
    logging.info("mutation length: {}".format(len(mut_p)))
    logging.info("putative revertant length: {}".format(len(revmut_p)))
    logging.info("---")
    return len(normal_p) == len(revmut_p)


def parse_hgvs_muts_file(hgvs_muts_file, raise_exception=True):
    hgvs_muts = []
    for m in [l.rstrip('\n') for l in open(hgvs_muts_file).readlines()]:<commit_msg>Fix HGVS dup for applying to seq & adjusting coord<commit_after>"""
Check if a mutation is reverted
"""
import argparse
import sys
import revmut
import logging
from revmut import oncotator
from revmut.utils import deprecated

import pyhgvs as hgvs
from Bio import SeqIO


class RevertantMutationsInfo(object):
    def __init__(self, transcript_record, hgvs_mut):
        self.record = transcript_record
        self.mut = hgvs_mut
        self.normal_p = self.record.seq.translate(to_stop=True)
        self.mut_p = apply_hgvs(self.record.seq, hgvs_mut).translate(to_stop=True)
        self.revmuts = []
        self.revmuts_pos_adj = []
        self.revmuts_p = []

    def add_revmut(self, hgvs_revmut):
        assert(len(self.revmuts) == len(self.revmuts_p))
        self.revmuts += [hgvs_revmut]
        hgvs_revmut_pos_adj = alter_coords_hgvs_sequential(self.mut, hgvs_revmut)
        self.revmuts_pos_adj += [hgvs_revmut_pos_adj]
        self.revmuts_p += [apply_hgvs(apply_hgvs(self.record.seq, self.mut), hgvs_revmut_pos_adj).translate(to_stop=True)]

    def to_tsv(self, header=True):
        assert(len(self.revmuts) == len(self.revmuts_p))
        rv = ""
        if header:
            rv += "mut\trevmut\trevmut_pos_adj\ttranscript\tnormal_protein_length\tmut_protein_length\trevmut_protein_length\n"
        rv += "\n".join(["\t".join(["{}"]*7).format(self.mut, self.revmuts[i], self.revmuts_pos_adj[i], self.record.id,
                         len(self.normal_p), len(self.mut_p),
                         len(self.revmuts_p[i])) for i in
                         range(len(self.revmuts))]) + "\n"
        return rv

    def __len__(self):
        return len(self.revmuts)


def alter_coords_hgvs_sequential(h1, h2):
    """Change HGVS coords of h2 after applying h1"""
    if h1.kind == "c" and h2.kind == "c":
        if h1.mutation_type == ">":
            h3 = hgvs.HGVSName(h2.name)
        elif h1.mutation_type == "del":
            if h1.cdna_start.coord > h2.cdna_end.coord:
                h3 = hgvs.HGVSName(h2.name)
            elif h1.cdna_end.coord < h2.cdna_start.coord:
                h3 = hgvs.HGVSName(h2.name)
                h3.cdna_start = hgvs.CDNACoord(coord=h3.cdna_start.coord-len(h1.ref_allele))
                h3.cdna_end = hgvs.CDNACoord(coord=h3.cdna_end.coord-len(h1.ref_allele))
            else:
                raise(Exception("Overlapping del not implemented"))
        elif h1.mutation_type == "ins":
            if h1.cdna_start.coord > h2.cdna_end.coord:
                h3 = hgvs.HGVSName(h2.name)
            elif h1.cdna_end.coord < h2.cdna_start.coord:
                h3 = hgvs.HGVSName(h2.name)
                h3.cdna_start = hgvs.CDNACoord(coord=h3.cdna_start.coord+len(h1.alt_allele))
                h3.cdna_end = hgvs.CDNACoord(coord=h3.cdna_end.coord+len(h1.alt_allele))
            else:
                raise(Exception("Overlapping ins not implemented"))
        elif h1.mutation_type == "dup":
            if h1.cdna_start.coord > h2.cdna_end.coord:
                h3 = hgvs.HGVSName(h2.name)
            elif h1.cdna_end.coord < h2.cdna_start.coord:
                h3 = hgvs.HGVSName(h2.name)
                h3.cdna_start = hgvs.CDNACoord(coord=h3.cdna_start.coord+len(h1.alt_allele)-len(h1.ref_allele))
                h3.cdna_end = hgvs.CDNACoord(coord=h3.cdna_end.coord+len(h1.alt_allele)-len(h1.ref_allele))
            else:
                raise(Exception("Overlapping dup not implemented"))
        elif h1.mutation_type == "delins":
            if h1.cdna_start.coord > h2.cdna_end.coord:
                h3 = hgvs.HGVSName(h2.name)
            elif h1.cdna_end.coord < h2.cdna_start.coord:
                h3 = hgvs.HGVSName(h2.name)
                h3.cdna_start = hgvs.CDNACoord(coord=h3.cdna_start.coord-len(h1.ref_allele)+len(h1.alt_allele))
                h3.cdna_end = hgvs.CDNACoord(coord=h3.cdna_end.coord-len(h1.ref_allele)+len(h1.alt_allele))
            else:
                raise(Exception("Overlapping delins not implemented"))
        else:
            raise(Exception("Unexpected mutation_type {}".format(h1.mutation_type)))
        return h3
    else:
        raise(Exception("Only cDNA mutations have been implemented"))


def apply_hgvs(seq, h):
    """Apply 1-based HGVS mutation to 0-based biopython Sequence"""
    if h.kind == "c":
        start = h.cdna_start.coord - 1
        end = h.cdna_end.coord - 1
        if h.mutation_type == ">":
            assert(seq[start] == h.ref_allele)
            new_seq = seq[:start] + h.alt_allele + seq[start+1:]
        elif h.mutation_type == "del":
            assert(seq[start:end+1] == h.ref_allele)
            new_seq = seq[:start] + seq[end+1:]
            assert(len(seq) == len(new_seq) + len(h.ref_allele))
        elif h.mutation_type == "ins":
            new_seq = seq[:start+1] + h.alt_allele + seq[end:]
            assert(len(seq) + len(h.alt_allele) == len(new_seq))
        elif h.mutation_type == "dup":
            assert(seq[start:end+1] == h.ref_allele)
            new_seq = seq[:start] + h.alt_allele + seq[end+1:]
            assert(len(seq) - len(h.ref_allele) + len(h.alt_allele) == len(new_seq))
        elif h.mutation_type == "delins":
            assert(seq[start:end+1] == h.ref_allele)
            new_seq = seq[:start] + h.alt_allele + seq[end+1:]
            assert(len(seq) - len(h.ref_allele) + len(h.alt_allele) == len(new_seq))
        else:
            raise(Exception("Unexpected mutation_type {}".format(h.mutation_type)))
        return new_seq
    else:
        raise(Exception("Only cDNA mutations have been implemented"))


@deprecated
def is_revertant(record, hgvs_mut, hgvs_rev_mut):
    normal_p = record.seq.translate(to_stop=True)
    mut_p = apply_hgvs(record.seq, hgvs_mut).translate(to_stop=True)
    revmut_p = apply_hgvs(apply_hgvs(record.seq, hgvs_mut), hgvs_rev_mut).translate(to_stop=True)
    logging.info("---")
    logging.info("mutation: {}".format(hgvs_mut))
    logging.info("putative revertant mutation: {}".format(hgvs_rev_mut))
    logging.info("transcript: {}".format(record.id))
    logging.info("normal length: {}".format(len(normal_p)))
    logging.info("mutation length: {}".format(len(mut_p)))
    logging.info("putative revertant length: {}".format(len(revmut_p)))
    logging.info("---")
    return len(normal_p) == len(revmut_p)


def parse_hgvs_muts_file(hgvs_muts_file, raise_exception=True):
    hgvs_muts = []
    for m in [l.rstrip('\n') for l in open(hgvs_muts_file).readlines()]:
{'fraction_non_alphanumeric': 0.13292336802270577, 'fraction_numerical': 0.01379690949227373, 'mean_word_length': 2.531458797327394, 'pattern_counts': {'":': 24, '<': 11, '<?xml version=': 0, '>': 15, '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': '4652397', 'n_tokens_mistral': 4800, 'n_tokens_neox': 4588, 'n_words': 709}