CRISPRTool / cas12lstmvcf.py
supercat666's picture
change cas9
4a303ce
raw
history blame
10.5 kB
import tensorflow as tf
from keras import regularizers
from keras.layers import Input, Dense, Dropout, Activation, Conv1D
from keras.layers import GlobalAveragePooling1D, AveragePooling1D
from keras.layers import Bidirectional, LSTM
from keras import Model
from keras.metrics import MeanSquaredError
import pandas as pd
import numpy as np
import requests
from functools import reduce
from operator import add
import tabulate
from difflib import SequenceMatcher
import cyvcf2
import parasail
import re
ntmap = {'A': (1, 0, 0, 0),
'C': (0, 1, 0, 0),
'G': (0, 0, 1, 0),
'T': (0, 0, 0, 1)
}
def get_seqcode(seq):
return np.array(reduce(add, map(lambda c: ntmap[c], seq.upper()))).reshape((1, len(seq), -1))
def BiLSTM_model(input_shape):
input = Input(shape=input_shape)
conv1 = Conv1D(128, 5, activation="relu")(input)
pool1 = AveragePooling1D(2)(conv1)
drop1 = Dropout(0.1)(pool1)
conv2 = Conv1D(128, 5, activation="relu")(drop1)
pool2 = AveragePooling1D(2)(conv2)
drop2 = Dropout(0.1)(pool2)
lstm1 = Bidirectional(LSTM(128,
dropout=0.1,
activation='tanh',
return_sequences=True,
kernel_regularizer=regularizers.l2(1e-4)))(drop2)
avgpool = GlobalAveragePooling1D()(lstm1)
dense1 = Dense(128,
kernel_regularizer=regularizers.l2(1e-4),
bias_regularizer=regularizers.l2(1e-4),
activation="relu")(avgpool)
drop3 = Dropout(0.1)(dense1)
dense2 = Dense(32,
kernel_regularizer=regularizers.l2(1e-4),
bias_regularizer=regularizers.l2(1e-4),
activation="relu")(drop3)
drop4 = Dropout(0.1)(dense2)
dense3 = Dense(32,
kernel_regularizer=regularizers.l2(1e-4),
bias_regularizer=regularizers.l2(1e-4),
activation="relu")(drop4)
drop5 = Dropout(0.1)(dense3)
output = Dense(1, activation="linear")(drop5)
model = Model(inputs=[input], outputs=[output])
return model
def fetch_ensembl_transcripts(gene_symbol):
url = f"https://rest.ensembl.org/lookup/symbol/homo_sapiens/{gene_symbol}?expand=1;content-type=application/json"
response = requests.get(url)
if response.status_code == 200:
gene_data = response.json()
if 'Transcript' in gene_data:
return gene_data['Transcript']
else:
print("No transcripts found for gene:", gene_symbol)
return None
else:
print(f"Error fetching gene data from Ensembl: {response.text}")
return None
def fetch_ensembl_sequence(transcript_id):
url = f"https://rest.ensembl.org/sequence/id/{transcript_id}?content-type=application/json"
response = requests.get(url)
if response.status_code == 200:
sequence_data = response.json()
if 'seq' in sequence_data:
return sequence_data['seq']
else:
print("No sequence found for transcript:", transcript_id)
return None
else:
print(f"Error fetching sequence data from Ensembl: {response.text}")
return None
def apply_mutation(ref_sequence, offset, ref, alt):
"""
Apply a single mutation to the sequence.
"""
if len(ref) == len(alt) and alt != "*": # SNP
mutated_seq = ref_sequence[:offset] + alt + ref_sequence[offset+len(alt):]
elif len(ref) < len(alt): # Insertion
mutated_seq = ref_sequence[:offset] + alt + ref_sequence[offset+1:]
elif len(ref) == len(alt) and alt == "*": # Deletion
mutated_seq = ref_sequence[:offset] + ref_sequence[offset+1:]
elif len(ref) > len(alt) and alt != "*": # Deletion
mutated_seq = ref_sequence[:offset] + alt + ref_sequence[offset+len(ref):]
elif len(ref) > len(alt) and alt == "*": # Deletion
mutated_seq = ref_sequence[:offset] + ref_sequence[offset+len(ref):]
return mutated_seq
def construct_combinations(sequence, mutations):
"""
Construct all combinations of mutations.
mutations is a list of tuples (position, ref, [alts])
"""
if not mutations:
return [sequence]
# Take the first mutation and recursively construct combinations for the rest
first_mutation = mutations[0]
rest_mutations = mutations[1:]
offset, ref, alts = first_mutation
sequences = []
for alt in alts:
mutated_sequence = apply_mutation(sequence, offset, ref, alt)
sequences.extend(construct_combinations(mutated_sequence, rest_mutations))
return sequences
def needleman_wunsch_alignment(query_seq, ref_seq):
"""
Use Needleman-Wunsch alignment to find the maximum alignment position in ref_seq
Use this position to represent the position of target sequence with mutations
"""
# Needleman-Wunsch alignment
alignment = parasail.nw_trace(query_seq, ref_seq, 10, 1, parasail.blosum62)
# extract CIGAR object
cigar = alignment.cigar
cigar_string = cigar.decode.decode("utf-8")
# record ref_pos
ref_pos = 0
matches = re.findall(r'(\d+)([MIDNSHP=X])', cigar_string)
max_num_before_equal = 0
max_equal_index = -1
total_before_max_equal = 0
for i, (num_str, op) in enumerate(matches):
num = int(num_str)
if op == '=':
if num > max_num_before_equal:
max_num_before_equal = num
max_equal_index = i
total_before_max_equal = sum(int(matches[j][0]) for j in range(max_equal_index))
ref_pos = total_before_max_equal
return ref_pos
def find_gRNA_with_mutation(ref_sequence, exon_chr, start, end, strand, transcript_id,
exon_id, gene_symbol, vcf_reader, pam="TTTN", target_length=34):
# initialization
mutated_sequences = [ref_sequence]
# find mutations within interested region
mutations = vcf_reader(f"{exon_chr}:{start}-{end}")
if mutations:
# find mutations
mutation_list = []
for mutation in mutations:
offset = mutation.POS - start
ref = mutation.REF
alts = mutation.ALT[:-1]
mutation_list.append((offset, ref, alts))
# replace reference sequence of mutation
mutated_sequences = construct_combinations(ref_sequence, mutation_list)
# find gRNA in ref_sequence or all mutated_sequences
targets = []
for seq in mutated_sequences:
len_sequence = len(seq)
dnatorna = {'A': 'A', 'T': 'U', 'C': 'C', 'G': 'G'}
for i in range(len_sequence - target_length + 1):
target_seq = seq[i:i + target_length]
if target_seq[4:7] == 'TTT':
pos = ref_sequence.find(target_seq)
if pos != -1:
is_mut = False
if strand == -1:
tar_start = end - pos - target_length + 1
else:
tar_start = start + pos
else:
is_mut = True
nw_pos = needleman_wunsch_alignment(target_seq, ref_sequence)
if strand == -1:
tar_start = str(end - nw_pos - target_length + 1) + '*'
else:
tar_start = str(start + nw_pos) + '*'
gRNA = ''.join([dnatorna[base] for base in target_seq[8:28]])
targets.append([target_seq, gRNA, exon_chr, str(strand), str(tar_start), transcript_id, exon_id, gene_symbol, is_mut])
# filter duplicated targets
unique_targets_set = set(tuple(element) for element in targets)
unique_targets = [list(element) for element in unique_targets_set]
return unique_targets
def format_prediction_output_with_mutation(targets, model_path):
Crispr_BiLSTM = BiLSTM_model(input_shape=(34, 4))
Crispr_BiLSTM.load_weights(model_path)
formatted_data = []
for target in targets:
# Predict
encoded_seq = get_seqcode(target[0])
prediction = float(list(Crispr_BiLSTM.predict(encoded_seq, verbose=0)[0])[0])
if prediction > 100:
prediction = 100
# Format output
gRNA = target[1]
exon_chr = target[2]
strand = target[3]
tar_start = target[4]
transcript_id = target[5]
exon_id = target[6]
gene_symbol = target[7]
is_mut = target[8]
formatted_data.append([gene_symbol, exon_chr, strand, tar_start, transcript_id, exon_id, target[0], gRNA, prediction, is_mut])
return formatted_data
def process_gene(gene_symbol, vcf_reader, model_path):
transcripts = fetch_ensembl_transcripts(gene_symbol)
results = []
all_exons = [] # To accumulate all exons
all_gene_sequences = [] # To accumulate all gene sequences
if transcripts:
for transcript in transcripts:
Exons = transcript['Exon']
all_exons.extend(Exons) # Add all exons from this transcript to the list
transcript_id = transcript['id']
for Exon in Exons:
exon_id = Exon['id']
gene_sequence = fetch_ensembl_sequence(exon_id) # Reference exon sequence
if gene_sequence:
all_gene_sequences.append(gene_sequence) # Add this gene sequence to the list
exon_chr = Exon['seq_region_name']
start = Exon['start']
end = Exon['end']
strand = Exon['strand']
targets = find_gRNA_with_mutation(gene_sequence, exon_chr, start, end, strand, transcript_id, exon_id, gene_symbol, vcf_reader)
if targets:
# Predict on-target efficiency for each gRNA site
formatted_data = format_prediction_output_with_mutation(targets, model_path)
results.extend(formatted_data) # Flatten the results
else:
print(f"Failed to retrieve gene sequence for exon {exon_id}.")
else:
print("Failed to retrieve transcripts.")
# Sort results based on prediction score (assuming score is at the 8th index)
sorted_results = sorted(results, key=lambda x: x[8], reverse=True)
# Return the sorted output, combined gene sequences, and all exons
return sorted_results, all_gene_sequences, all_exons