MULTI-evolve / model /utils /cloning_utils.py
Upload folder using huggingface_hub
6f1e670 verified 2 months ago
31.7 kB
# updated 3/25/2025
from concurrent.futures import ProcessPoolExecutor
import copy
import re

import pandas as pd
import numpy as np
import os

from Bio import Align, SeqIO
from Bio.Seq import Seq
from Bio.SeqUtils import MeltingTemp as mt
from Bio.SeqRecord import SeqRecord
from typing import Optional, Tuple

codon_dicts = {
    'human': {
    'F': 'TTT', 'L': 'CTG', 'Y': 'TAT', 'H': 'CAT', 'Q': 'CAG',
    'I': 'ATT', 'M': 'ATG', 'N': 'AAT', 'K': 'AAG', 'V': 'GTG',
    'D': 'GAT', 'E': 'GAG', 'S': 'TCT', 'C': 'TGT', 'W': 'TGG',
    'P': 'CCT', 'R': 'CGG', 'T': 'ACT', 'A': 'GCT', 'G': 'GGG',
    },
    'ecoli': {
        'F': 'TTT', 'L': 'CTG', 'Y': 'TAT', 'H': 'CAT', 'Q': 'CAG',
        'I': 'ATT', 'M': 'ATG', 'N': 'AAC', 'K': 'AAA', 'V': 'GTG',
        'D': 'GAT', 'E': 'GAA', 'S': 'TCT', 'C': 'TGC', 'W': 'TGG',
        'P': 'CCG', 'R': 'CGT', 'T': 'ACC', 'A': 'GCG', 'G': 'GGC',
    },
    'yeast': {
        'F': 'TTT', 'L': 'CTA', 'Y': 'TAT', 'H': 'CAT', 'Q': 'CAA',
        'I': 'ATT', 'M': 'ATG', 'N': 'AAT', 'K': 'AAA', 'V': 'GTT',
        'D': 'GAT', 'E': 'GAA', 'S': 'TCT', 'C': 'TGT', 'W': 'TGG',
        'P': 'CCA', 'R': 'AGA', 'T': 'ACT', 'A': 'GCT', 'G': 'GGT',
    }
}

class MultiAssemblyDesigner:

    """
    Designs oligos for protein mutations.
    Args:
        data (pd.DataFrame): DataFrame containing mutation data.
        start_seq_fasta (str): Path to FASTA file with starting sequence.
        overhang (int): Overhang length.
        species (str): Species, 'human', 'ecoli', or 'yeast'.
        oligo_direction (str): Direction of oligo, 'bottom' or 'top'.
        tm (float): Target melting temperature.
        output (str): Type of output, 'design' or 'update'.
    """

    def __init__(self, data, start_seq_fasta, overhang, species='human', oligo_direction='bottom', tm=80, output='design'):
        
        print("Initializing MultiAssemblyDesigner...")
        self.data = data.rename(columns={data.columns[0]:'aa_mut'})
        self.data['aa_mut'] = self.data['aa_mut'].apply(lambda x: self._sort_mutations(x))
        self.fasta_dir = os.path.dirname(start_seq_fasta)

        print(f'The melting temperature is {tm}')
        self.tm = tm
        self.start_seq = SeqIO.read(start_seq_fasta, "fasta").seq.upper()
        self.overhang = overhang
        self.oligo_direction = oligo_direction
        self.codon_dict = codon_dicts[species]
        # print('Processing mutations...')
        self._process_mutations()

        # print('Designing oligos...')
        self._design_oligos()
        self._find_unique_mutant_oligos()

        if output == 'design':
            print('Exporting design...')
            self._export_design()

        elif output == 'update':
            print('Updating oligo IDs...')
            self._modify_oligo_id()

    def _sort_mutations(self, mutation_string):
        """
        Sort mutations within a string based on their position numbers.
        
        Args:
            mutation_string (str): String containing mutations (e.g., 'A167R/T192V')
        
        Returns:
            str: Sorted mutation string
        """
        mutations = mutation_string.split('/')
        sorted_mutations = sorted(mutations, key=lambda x: int(''.join(filter(str.isdigit, x))))
        return sorted_mutations

    def _process_mutations(self):
        """Processes mutations to extract positions and bases."""
        self.data[['Positions','Reference_bases','Alternative_bases']] = self.data.apply(
            lambda x: pd.Series(self._get_codon_mutation_list(x['aa_mut'], self.codon_dict, self.overhang, str(self.start_seq))),
            axis=1
        )
        
        self.data['mut_seq'] = self.data.apply(
            lambda x: self._get_mut_seq(x['Positions'], x['Alternative_bases'], x['aa_mut']),
            axis=1
        )

    def _design_oligos(self):
        """Designs oligos for each mutation in the dataset."""
        self.data[['oligos','oligo_mut']] = self.data.apply(
            lambda x: pd.Series(self._design_oligo_pipeline(x)),
            axis=1
        )

    def _get_codon_mutation_list(self, mut_ls, codon_dict, overhang, start_seq):
        """
        Retrieves list of codon mutations.
        Args:
            mut_ls (list): List of mutations.
            codon_dict (dict): Codon dictionary.
            overhang (int): Overhang length.
            start_seq (str): Starting sequence.
        Returns:
            tuple: Lists of positions, old codons, and new codons.
        """
        pos_ls, old_codon_ls, new_codon_ls = [], [], []
        for mut in mut_ls:
            pos, new_codon = self._get_codon_mutation(mut, codon_dict)
            pos_ls.append(int(pos)+overhang)
            old_codon_ls.append(start_seq[(int(pos)+overhang)-1:(int(pos)+overhang+2)])
            new_codon_ls.append(new_codon)
        return pos_ls, old_codon_ls, new_codon_ls
    
    def _get_codon_mutation(self, mut, codon_dict):
        """
        Retrieves codon mutation details.
        Args:
            mut (str): Mutation string.
            codon_dict (dict): Codon dictionary.
        Returns:
            tuple: Position and new codon.
        """
        new_codon = codon_dict[mut[-1]]
        pos = str(int(mut[1:-1])*3 - 2)
        return pos, new_codon
    
    def _design_oligo_pipeline(self, row):
        """
        Designs oligos for a row of mutations.
        Args:
            row (pd.Series): Row of mutation data.
        Returns:
            tuple: Lists of oligos and oligo mutations.
        """
        pos_start_ls, pos_end_ls = [], []
        for i, pos in enumerate(row['Positions']):
            pos_start, pos_end = self._design_mutant_oligo(self.start_seq, pos, row['Alternative_bases'][i], row['Reference_bases'][i], result='positions')
            pos_start_ls.append(pos_start)
            pos_end_ls.append(pos_end)

        oligos, oligo_mt_mapping = [], []
        i = 0
        while i < len(row['Positions']):
            mut = [row['aa_mut'][i]]
            start_index = pos_start_ls[i]
            index_i = i

            if i < len(row['Positions'])-1:
                n = 0
                while pos_end_ls[i+n] >= pos_start_ls[i+n+1]:
                    n += 1
                    mut.append(row['aa_mut'][i+n])
                    if i+n+1 == len(row['Positions']):
                        break
                index_f = i + n
                i = i + n
                end_index = pos_end_ls[index_f]
            else:
                index_f = i
                end_index = pos_end_ls[i]
            
            oligos.append(str(self._get_mutant_oligo_by_pos(self.start_seq, row['Positions'], row['Alternative_bases'], row['Reference_bases'], start_index, end_index, index_i, index_f)))
            i += 1
            oligo_mt_mapping.append("-".join(mut))

        return oligos, oligo_mt_mapping
    
    def _get_mut_seq(self, pos_ls, new_codon_ls, mut_ls):
        """
        Generates mutated sequence.
        Args:
            pos_ls (list): List of positions.
            new_codon_ls (list): List of new codons.
            mut_ls (list): List of mutations.
        Returns:
            str: Mutated sequence.
        """
        mut_seq = copy.deepcopy(self.start_seq)
        for i, pos in enumerate(pos_ls):
            mod_pos = int(pos) - 1
            wt_aa = mut_ls[i][0]
            wt_aa_retrieved = Seq(self.start_seq[mod_pos:mod_pos+3]).translate()
            assert wt_aa == wt_aa_retrieved, f"{mut_ls[i]} is not a true mutation from {wt_aa_retrieved}{mut_ls[i][1:-1]}"
            mut_seq = mut_seq[:mod_pos] + new_codon_ls[i].lower() + mut_seq[mod_pos+3:]
        return str(mut_seq)
    
    def _design_mutant_oligo(self, seq, pos, new_codon, old_codon, result='oligo'):
        """
        Designs mutant oligo.
        Args:
            seq (str): Sequence.
            pos (int): Position.
            new_codon (str): New codon.
            old_codon (str): Old codon.
            result (str): Type of result to return.
        Returns:
            tuple: Oligo sequence and wild-type oligo sequence, or start and end positions.
        """
        mod_pos = int(pos) - 1
        mut_seq = seq[:mod_pos] + new_codon.lower() + seq[mod_pos+3:]
        wt_seq = seq[:mod_pos] + old_codon.lower() + seq[mod_pos+3:]
        
        start_index = mod_pos - 11
        end_index = mod_pos + 14

        if self.oligo_direction == 'bottom':
            oligo = mut_seq[start_index:end_index].reverse_complement()
            wt_oligo = wt_seq[start_index:end_index].reverse_complement()
        else:
            oligo = mut_seq[start_index:end_index]
            wt_oligo = wt_seq[start_index:end_index]
        
        while mt.Tm_NN(oligo, Na=50, K=25, Tris=35, Mg=10) <= self.tm:
            if len(oligo) % 2 == 0:
                start_index -= 1
            else: 
                end_index += 1

            if self.oligo_direction == 'bottom': 
                oligo = mut_seq[start_index:end_index].reverse_complement()
                wt_oligo = wt_seq[start_index:end_index].reverse_complement()
            else:
                oligo = mut_seq[start_index:end_index]
                wt_oligo = wt_seq[start_index:end_index]

        if result == 'oligo':
            return str(oligo), str(wt_oligo), round(mt.Tm_NN(oligo, Na=50, K=25, Tris=35, Mg=10), 2)
        else:
            return start_index+1, end_index+1
    
    def _get_mutant_oligo_by_pos(self, seq, pos_ls, new_codon_ls, old_codon_ls, start, end, index_i, index_f):
        """
        Retrieves mutant oligo by position.
        Args:
            seq (str): Sequence.
            pos_ls (list): List of positions.
            new_codon_ls (list): List of new codons.
            old_codon_ls (list): List of old codons.
            start (int): Start position.
            end (int): End position.
            index_i (int): Start index.
            index_f (int): End index.
        Returns:
            str: Mutant oligo sequence.
        """
        mod_pos_ls = pos_ls[index_i:index_f+1]
        mod_new_codon_ls = new_codon_ls[index_i:index_f+1]
        mod_old_codon_ls = old_codon_ls[index_i:index_f+1]

        for i, mod_pos in enumerate(mod_pos_ls):
            mod_pos = int(mod_pos) - 1
            old_codon = seq[mod_pos:mod_pos+3]
            assert old_codon.upper() == mod_old_codon_ls[i].upper()
            seq = seq[:mod_pos] + mod_new_codon_ls[i].lower() + seq[mod_pos+3:]

        mod_start, mod_end = int(start) - 1, int(end) - 1
        return seq[mod_start:mod_end].reverse_complement() if self.oligo_direction == 'bottom' else seq[mod_start:mod_end]
    
    def _find_unique_mutant_oligos(self):
        """Identifies unique mutant oligos in the dataset."""
        oligos = [item for sublist in self.data['oligos'].tolist() for item in sublist]
        oligo_mutation = [item for sublist in self.data['oligo_mut'].tolist() for item in sublist]

        df = pd.DataFrame({'oligos': oligos, 'mutation': oligo_mutation}).drop_duplicates(subset=['mutation'], keep='first')
        df['oligo_id'] = range(len(df))

        oligo_dict = {oligo: i for i, oligo in enumerate(df['oligos'])}
        self.data['oligo_id'] = self.data['oligos'].apply(lambda x: [oligo_dict[oligo] for oligo in x])

        self.oligos = df

        # Apply the sorting function to each row
        self.data[['oligo_id', 'oligo_mut']] = self.data.apply(self._sort_oligos, axis=1)

    def _sort_oligos(self, row):
            """Sort oligo_id and corresponding oligo_mut values in sync."""

            # Convert oligo_id string to list of integers
            oligo_ids = row['oligo_id']
            # Convert oligo_mut string to list
            oligo_muts = row['oligo_mut']
            
            # Zip together for sorting
            paired_data = list(zip(oligo_ids, oligo_muts))
            # Sort by oligo_id
            paired_data.sort(key=lambda x: x[0])
            
            # Unzip the sorted data
            sorted_ids, sorted_muts = map(list, zip(*paired_data))
            
            # Return new row with sorted, comma-joined values
            return pd.Series({
                'oligo_id': sorted_ids,
                'oligo_mut': sorted_muts
            })
    
    def _export_df_with_lists(self, df, filepath, delimiter=','):
        """
        Export DataFrame with list columns to CSV, converting lists to delimiter-separated strings
        without brackets for better readability.
        
        Parameters:
        df (pandas.DataFrame): DataFrame containing list columns
        filepath (str): Path where CSV will be saved
        delimiter (str): Delimiter to separate list items (default ';')
        """
        # Create a copy to avoid modifying the original
        df_to_save = df.copy()
        
        # Convert list columns to delimited strings
        for column in df_to_save.columns:
            if df_to_save[column].apply(lambda x: isinstance(x, list)).any():
                df_to_save[column] = df_to_save[column].apply(
                    lambda x: delimiter.join(str(item) for item in x) if isinstance(x, list) else x
                )
        
        # Save to CSV
        df_to_save.to_csv(filepath, index=False)

    def _import_df_with_lists(self, filepath, delimiter=','):
        """
        Import CSV file and convert delimiter-separated strings back to lists.
        
        Parameters:
        filepath (str): Path to the CSV file
        delimiter (str): Delimiter used to separate list items (default ';')
        
        Returns:
        pandas.DataFrame: DataFrame with list columns properly restored
        """
        # Read the CSV
        df = pd.read_csv(filepath)
        
        # Try to convert delimited strings back to lists
        for column in df.columns:
            try:
                # Check if the column contains delimiter-separated values
                if df[column].dtype == 'object':
                    first_value = str(df[column].iloc[0])
                    if delimiter in first_value:
                        # Convert to list and handle type conversion
                        def convert_to_list(value):
                            if pd.isna(value):
                                return []
                            items = str(value).split(delimiter)
                            # Try to convert to numbers if possible
                            try:
                                return [float(item) if '.' in item else int(item) 
                                    for item in items]
                            except ValueError:
                                return items
                        
                        df[column] = df[column].apply(convert_to_list)
            except:
                # If conversion fails, keep the column as is
                continue
        
        return df

    def _export_design(self):
        """Exports the cloning sheet and oligos."""

        self._export_df_with_lists(self.data[['oligo_id', 'oligo_mut']].copy(), os.path.join(self.fasta_dir, 'cloning_sheet.csv'))
        self.oligos.to_csv(os.path.join(self.fasta_dir, 'oligos.csv'), index=False)

    def _modify_oligo_id(self):
        """Modifies the oligo_id in the cloning sheet to match the updated oligo_id in the oligos file."""
        self.oligos = self._import_df_with_lists(os.path.join(self.fasta_dir, 'oligos.csv'))
        oligo_dict = dict(zip(self.oligos['mutation'], self.oligos['oligo_id']))
        self.data['oligo_id'] = self.data['oligo_mut'].apply(lambda x: [oligo_dict[mutation] for mutation in x])
        self.data[['oligo_id', 'oligo_mut']] = self.data.apply(self._sort_oligos, axis=1)
                
        self._export_df_with_lists(self.data[['oligo_id', 'oligo_mut']].copy(), os.path.join(self.fasta_dir, 'cloning_sheet.csv'))

class SequenceTrimmer:
    """
    Trims adapter sequences from DNA sequences, handling both forward and reverse orientations.
    
    Args:
        five_prime (str): 5' adapter sequence to find and trim before
        three_prime (str): 3' adapter sequence to find and trim after
        max_error_rate (float): Maximum mismatch rate allowed when matching adapters (default: 0.1)
        min_length (int): Minimum sequence length after trimming (default: 15)
        
    Attributes:
        five_prime (str): Uppercase 5' adapter sequence
        three_prime (str): Uppercase 3' adapter sequence
        max_error_rate (float): Maximum allowed mismatch rate
        min_length (int): Minimum allowed sequence length
    """

    def __init__(self, 
                 five_prime: str,
                 three_prime: str,
                 min_length: int,
                 max_error_rate: float = 0
                 ):
        self.five_prime = five_prime.upper() 
        self.three_prime = three_prime.upper()
        self.max_error_rate = max_error_rate
        self.min_length = min_length
    
    def _count_mismatches(self, seq1: str, seq2: str) -> int:
        """
        Count mismatches between two sequences of equal length.
        
        Args:
            seq1 (str): First sequence
            seq2 (str): Second sequence
            
        Returns:
            int: Number of mismatched positions
        """
        return sum(c1 != c2 for c1, c2 in zip(seq1, seq2))
    
    def _reverse_complement(self, seq: str) -> str:
        """
        Generate reverse complement of a DNA sequence.
        
        Args:
            seq (str): Input DNA sequence
            
        Returns:
            str: Reverse complement sequence
        """
        seq = seq.upper()
        complement = {'A':'T', 'T':'A', 'G':'C', 'C':'G'}
        return ''.join(complement.get(base, base) for base in reversed(seq))
    
    def _find_with_mismatches(self, sequence: str, pattern: str) -> Optional[Tuple[Tuple[int, int], str]]:
        """
        Find pattern in sequence and its reverse complement, allowing mismatches.
        
        Args:
            sequence (str): Input sequence to search
            pattern (str): Pattern to find
            
        Returns:
            Optional[Tuple[Tuple[int, int], str]]: Tuple of ((start, end), strand) if found, None if not found
        """
        sequence = sequence.upper()
        pattern_len = len(pattern)
        
        if len(sequence) < pattern_len:
            return None
            
        scores = {}
        rev_comp = self._reverse_complement(sequence)
        
        for i, seq in enumerate([sequence, rev_comp]):
            for start in range(len(seq) - pattern_len + 1):
                window = seq[start:start + pattern_len]
                score = self._count_mismatches(window, pattern)
                scores[(start, start + pattern_len), "fwd" if i == 0 else "rev"] = score
                
        if not scores:
            return None
            
        best_pos = min(scores.items(), key=lambda x: x[1])
        return best_pos[0] if best_pos[1] <= (1 - self.max_error_rate) * pattern_len else None

    def _trim_record(self, seq: str) -> Optional[str]:
        """
        Trim adapters from a single sequence.
        
        Args:
            seq (str): Input DNA sequence
            
        Returns:
            Optional[str]: Trimmed sequence if successful, None if discarded
        """
        if len(seq) < self.min_length:
            return None
        
        sequence = seq
        sequence_rev_comp = self._reverse_complement(sequence)
        start = 0
        end = len(sequence)
        strand = "fwd"
        
        five_prime_pos = self._find_with_mismatches(sequence, self.five_prime)
        if five_prime_pos:
            start = five_prime_pos[0][0]
            strand = five_prime_pos[1]
                
        three_prime_pos = self._find_with_mismatches(sequence, self.three_prime)
        if three_prime_pos:
            end = three_prime_pos[0][1]


        # check if start position is less than end position
        if start < end:

            if end - start < self.min_length:
                return None
                
            return sequence[start:end] if strand == "fwd" else sequence_rev_comp[start:end]
        
        # if start is greater than end, then the region of interest is wrapping around (given the sequence is circular)
        else:

            if strand == "fwd":
                trim = sequence[start:] + sequence[:end]
            else:
                trim = sequence_rev_comp[start:] + sequence_rev_comp[:end]

            return trim

    def trim_file(self, input, input_type: str = 'fasta') -> Optional[list]:
        """
        Process FASTQ file and output trimmed sequences.
        
        Args:
            input: Path to input FASTQ file or FASTA file or list of either (fasta, fastq, fasta list, fastq list)
            input_type (str): Type of input, either 'fastq' or 'fasta'
            
        Returns:
            Optional[list]: List of trimmed sequences if output='list', None otherwise
        """
        records_stored = []
        if input_type == 'fastq':
            records_stored = [record for record in SeqIO.parse(input, "fastq")]
            seqs = [str(record.seq) for record in SeqIO.parse(input, "fastq")]
        elif input_type == 'fasta':
            records_stored = [record for record in SeqIO.parse(input, "fasta")]
            seqs = [str(record.seq) for record in SeqIO.parse(input, "fasta")]
        elif input_type == 'fasta list':
            records_stored = [record for file in input for record in SeqIO.parse(file, "fasta")]
            seqs = [str(record.seq) for file in input for record in SeqIO.parse(file, "fasta")]
        elif input_type == 'fastq list':
            records_stored = [record for file in input for record in SeqIO.parse(file, "fastq")]
            seqs = [str(record.seq) for file in input for record in SeqIO.parse(file, "fastq")]

        with ProcessPoolExecutor(max_workers=10) as executor:
            trimmed_seqs = list(executor.map(self._trim_record, seqs))

        records = [] 
        for seq, record in zip(trimmed_seqs, records_stored):
            if seq is not None and len(seq) >= self.min_length:
                records.append(SeqRecord(seq=Seq(seq), id=record.id, 
                                      name=record.name, description=record.description))
                
        if input_type == 'fasta list' or input_type == 'fastq list':
            SeqIO.write(records, f"seqs_trimmed.fasta", "fasta")
        else:
            SeqIO.write(records, f"{input.split('.')[0]}_trimmed.fasta", "fasta")

class BaseProteinCDSAnalyzer:
    """
    Analyzes coding sequences (CDS) of proteins.
    Args:
        seqs (str or list): Path to FASTA file or list of sequences.
        ref_seqs (str or list): Path to reference FASTA file or list of reference sequences.
        input_type (str): Type of input, either 'fasta' or 'list'.
    """
    
    def __init__(self, seqs, ref_seqs, input_type='fasta'):
        self._load_sequences(seqs, ref_seqs, input_type)
        self._run_pipeline()
       
    def _load_sequences(self, seqs, ref_seqs, input_type):
        """
        Loads sequences from input file or list.
        Args:
            seqs (str or list): Path to FASTA file or list of sequences.
            ref_seqs (str or list): Path to reference FASTA file or list of reference sequences.
            input_type (str): Type of input, either 'fasta' or 'list'.
        """
        if input_type == 'fasta':
            self.data = pd.DataFrame([str(record.seq).upper() for record in SeqIO.parse(seqs, "fasta")], columns=['seqs'])
            self.ref_seq = str(next(SeqIO.parse(ref_seqs, "fasta")).seq).upper()
        elif input_type == 'list':
            self.data = pd.DataFrame(seqs, columns=['seqs'])
            self.ref_seq = ref_seqs[0]

    def _align_sequences(self, query_sequence):
        """
        Aligns a query sequence to the reference sequence.
        Args:
            query_sequence (str): The sequence to align.
        Returns:
            list: Aligned sequence and its length.
        """
        aligner = Align.PairwiseAligner()
        aligner.mode = 'global'
        aligner.match_score = 2
        aligner.mismatch_score = 0
        aligner.open_gap_score = -4
        aligner.extend_gap_score = -2
        alignment = next(aligner.align(self.ref_seq, query_sequence))
        return [alignment[1], len(alignment[1])]

    def _align_sequences_multithreaded(self):
        """Aligns sequences using multiple threads for improved performance."""
        with ProcessPoolExecutor() as executor:
            results = executor.map(self._align_sequences, self.data['seqs'])
        self.data[['aligned_seqs', 'aligned_seqs_length']] = pd.DataFrame(list(results))

    def _generate_mutation_name(self, input_list):
        """
        Generates a mutation name from a list of mutations.
        Args:
            input_list (list): List of mutations.
        Returns:
            str: Generated mutation name.
        """
        if not input_list:
            return 'WT'
        if input_list[0] in ['indel', 'deletion', 'contains_N']:
            return input_list[0]
        return '/'.join(sorted(input_list, key=lambda s: int(''.join(filter(str.isdigit, s)))))

    def _compare_codon_to_ref(self, sequence):
        """
        Compares codons in a sequence to the reference sequence.
        Args:
            sequence (str): The sequence to compare.
        Returns:
            tuple: Dictionary of mutation counts and dictionary of mutation details.
        """
        ref_codon_seq = [self.ref_seq[i:i+3] for i in range(0, len(self.ref_seq), 3)]
        codon_seq = [sequence[i:i+3] for i in range(0, len(sequence), 3)]
        
        if 'N' in sequence:
            return [0, 0, 0, 0, [], ['contains_N'], [], [], 'contains_N']
        
        if "-" in sequence:
            return [0, 0, 0, 0, [], ['deletion'], [], [], 'deletion']
        
        if len(sequence) > len(self.ref_seq):
            return [0, 0, 0, 0, [], ['indel'], [], [], 'indel']
        
        if len(sequence) == len(self.ref_seq):
            muts = [0, 0, 0, 0]
            seq_mutations = [[], [], [], [], '']
            for pos, (codon, ref_codon) in enumerate(zip(codon_seq, ref_codon_seq), 1):
                mismatches = sum(c1 != c2 for c1, c2 in zip(codon, ref_codon))
                if mismatches:
                    seq_mutations[mismatches].append(ref_codon + str(pos) + codon)
                muts[mismatches] += 1
            return muts + seq_mutations

    def _compare_codon_to_ref_multithreaded(self):
        """Compares codons to reference using multiple threads for improved efficiency."""
        with ProcessPoolExecutor() as executor:
            results = executor.map(self._compare_codon_to_ref, self.data['aligned_seqs'])
        self.data[['Num_Changes_0', 'Num_Changes_1', 'Num_Changes_2', 'Num_Changes_3',
                   'nt_0_mut', 'nt_1_mut', 'nt_2_mut', 'nt_3_mut', 'error']] = pd.DataFrame(list(results))

    def _convert_codon_mut_to_aa_mut(self, codon_mut_ls):
        """
        Converts codon mutations to amino acid mutations.
        Args:
            codon_mut_ls (list): List of codon mutations.
        Returns:
            list: List of amino acid mutations.
        """
        aa_mut_ls = []
        for mut in codon_mut_ls:
            if mut in ['indel', 'deletion']:
                aa_mut_ls.append(mut)
                continue
            match = re.match(r'([a-zA-Z]+)(\d+)([a-zA-Z]+)', mut)
            if match:
                part1, part2, part3 = match.groups()
                aa_i = str(Seq(part1).translate())
                aa_f = str(Seq(part3).translate())
                aa_mut_ls.append(aa_i + part2 + aa_f)
        return [aa_mut_ls]
    
    def _convert_codon_mut_to_aa_mut_multithreaded(self):
        """Converts codon mutations to amino acid mutations using multiple threads for better performance."""
        with ProcessPoolExecutor() as executor:
            results = executor.map(self._convert_codon_mut_to_aa_mut, self.data['codon_mut_ls'])
        self.data['aa_mut_ls'] = pd.DataFrame(list(results))
        self.data['aa_mutation'] = self.data['aa_mut_ls'].apply(self._generate_mutation_name)
    
    def _generate_mutation_names_all(self):
        """Generates mutation names for all sequences in the dataset."""
        self.data['codon_mut_ls'] = self.data['nt_1_mut'] + self.data['nt_2_mut'] + self.data['nt_3_mut']
        self.data['codon_mutation'] = self.data['codon_mut_ls'].apply(self._generate_mutation_name)

    def _run_pipeline(self):
        """Executes the full analysis pipeline."""
        self._align_sequences_multithreaded()
        self._compare_codon_to_ref_multithreaded()
        self._generate_mutation_names_all()
        self._convert_codon_mut_to_aa_mut_multithreaded()
        self.mutants = self.data[['aa_mut_ls','aa_mutation']]

class RawNanoporeProteinCDSAnalyzer(BaseProteinCDSAnalyzer):
    """
    Manages raw nanopore sequencing data with high error rate.
    Inherits from BaseProteinCDSAnalyzer.
    """

    def _remove_insertions(self, reference_aligned, query_aligned):
        """
        Removes insertions from aligned query sequence.
        Args:
            reference_aligned (str): Aligned reference sequence.
            query_aligned (str): Aligned query sequence.
        Returns:
            str: Query sequence with insertions removed.
        """
        return ''.join(char for i, char in enumerate(query_aligned) if reference_aligned[i] != '-')

    def _align_sequences(self, query_sequence):
        """
        Aligns a query sequence to the reference sequence, removing insertions.
        Args:
            query_sequence (str): The sequence to align.
        Returns:
            list: Aligned sequence without insertions and its length.
        """
        aligner = Align.PairwiseAligner()
        aligner.mode = 'global'
        aligner.match_score = 2
        aligner.mismatch_score = 0
        aligner.open_gap_score = aligner.extend_gap_score = -2
        alignment = next(aligner.align(self.ref_seq, query_sequence))
        query_aligned_no_ins = self._remove_insertions(*alignment)
        return [query_aligned_no_ins, len(query_aligned_no_ins)]

    # def _generate_mutation_names_all(self):
    #     """Generates mutation names for all sequences, considering only 2 and 3 nucleotide changes."""
    #     self.data['codon_mut_ls'] = self.data['nt_2_mut'] + self.data['nt_3_mut']
    #     self.data['codon_mutation'] = self.data['codon_mut_ls'].apply(self._generate_mutation_name)
    
    def _compare_codon_to_ref(self, sequence):
        """
        Compares codons in a sequence to the reference sequence, ignoring deletions within codons.
        Args:
            sequence (str): The sequence to compare.
        Returns:
            tuple: Dictionary of mutation counts and dictionary of mutation details.
        """
        ref_codon_seq = [self.ref_seq[i:i+3] for i in range(0, len(self.ref_seq), 3)]
        codon_seq = [sequence[i:i+3] for i in range(0, len(sequence), 3)]
        
        muts = [0, 0, 0, 0]
        seq_mutations = [[], [], [], [], '']
        for pos, (codon, ref_codon) in enumerate(zip(codon_seq, ref_codon_seq), 1):
            mismatches = sum(c1 != c2 for c1, c2 in zip(codon, ref_codon))
            if mismatches:
                seq_mutations[mismatches].append(ref_codon + str(pos) + codon)
            muts[mismatches] += 1
        return muts + seq_mutations