Spaces:
Sleeping
Sleeping
| from keras import Model | |
| from keras.layers import Input | |
| from keras.layers import Multiply | |
| from keras.layers import Dense, Dropout, Activation, Flatten | |
| from keras.layers import Convolution1D, AveragePooling1D | |
| import pandas as pd | |
| import numpy as np | |
| import keras | |
| import requests | |
| from functools import reduce | |
| from operator import add | |
| from Bio.SeqRecord import SeqRecord | |
| from Bio.SeqFeature import SeqFeature, FeatureLocation | |
| from Bio.Seq import Seq | |
| from Bio import SeqIO | |
| 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 Seq_DeepCpf1_model(input_shape): | |
| Seq_deepCpf1_Input_SEQ = Input(shape=input_shape) | |
| Seq_deepCpf1_C1 = Convolution1D(80, 5, activation='relu')(Seq_deepCpf1_Input_SEQ) | |
| Seq_deepCpf1_P1 = AveragePooling1D(2)(Seq_deepCpf1_C1) | |
| Seq_deepCpf1_F = Flatten()(Seq_deepCpf1_P1) | |
| Seq_deepCpf1_DO1 = Dropout(0.3)(Seq_deepCpf1_F) | |
| Seq_deepCpf1_D1 = Dense(80, activation='relu')(Seq_deepCpf1_DO1) | |
| Seq_deepCpf1_DO2 = Dropout(0.3)(Seq_deepCpf1_D1) | |
| Seq_deepCpf1_D2 = Dense(40, activation='relu')(Seq_deepCpf1_DO2) | |
| Seq_deepCpf1_DO3 = Dropout(0.3)(Seq_deepCpf1_D2) | |
| Seq_deepCpf1_D3 = Dense(40, activation='relu')(Seq_deepCpf1_DO3) | |
| Seq_deepCpf1_DO4 = Dropout(0.3)(Seq_deepCpf1_D3) | |
| Seq_deepCpf1_Output = Dense(1, activation='linear')(Seq_deepCpf1_DO4) | |
| Seq_deepCpf1 = Model(inputs=[Seq_deepCpf1_Input_SEQ], outputs=[Seq_deepCpf1_Output]) | |
| return Seq_deepCpf1 | |
| # seq-ca model (DeepCpf1) | |
| def DeepCpf1_model(input_shape): | |
| DeepCpf1_Input_SEQ = Input(shape=input_shape) | |
| DeepCpf1_C1 = Convolution1D(80, 5, activation='relu')(DeepCpf1_Input_SEQ) | |
| DeepCpf1_P1 = AveragePooling1D(2)(DeepCpf1_C1) | |
| DeepCpf1_F = Flatten()(DeepCpf1_P1) | |
| DeepCpf1_DO1 = Dropout(0.3)(DeepCpf1_F) | |
| DeepCpf1_D1 = Dense(80, activation='relu')(DeepCpf1_DO1) | |
| DeepCpf1_DO2 = Dropout(0.3)(DeepCpf1_D1) | |
| DeepCpf1_D2 = Dense(40, activation='relu')(DeepCpf1_DO2) | |
| DeepCpf1_DO3 = Dropout(0.3)(DeepCpf1_D2) | |
| DeepCpf1_D3_SEQ = Dense(40, activation='relu')(DeepCpf1_DO3) | |
| DeepCpf1_Input_CA = Input(shape=(1,)) | |
| DeepCpf1_D3_CA = Dense(40, activation='relu')(DeepCpf1_Input_CA) | |
| DeepCpf1_M = Multiply()([DeepCpf1_D3_SEQ, DeepCpf1_D3_CA]) | |
| DeepCpf1_DO4 = Dropout(0.3)(DeepCpf1_M) | |
| DeepCpf1_Output = Dense(1, activation='linear')(DeepCpf1_DO4) | |
| DeepCpf1 = Model(inputs=[DeepCpf1_Input_SEQ, DeepCpf1_Input_CA], outputs=[DeepCpf1_Output]) | |
| return DeepCpf1 | |
| 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 find_crispr_targets(sequence, chr, start, strand, pam="TTTN", target_length=34): | |
| targets = [] | |
| len_sequence = len(sequence) | |
| for i in range(len_sequence - target_length + 1): | |
| target_seq = sequence[i:i + target_length] | |
| if target_seq[4:7] == 'TTT': | |
| tar_start = start + i | |
| tar_end = start + i + target_length | |
| gRNA = target_seq[8:28] | |
| targets.append([target_seq, gRNA, chr, str(tar_start), str(tar_end), str(strand)]) | |
| return targets | |
| def format_prediction_output(targets, seq_deepCpf1): | |
| formatted_data = [] | |
| for target in targets: | |
| # Predict | |
| encoded_seq = get_seqcode(target[0]) # 'target' seems to be the full sequence including PAM | |
| prediction = seq_deepCpf1.predict(encoded_seq) | |
| # Format output | |
| gRNA = target[1] # gRNA is presumably the guide RNA sequence | |
| chr = target[2] # Chromosome | |
| start = target[3] # Start position | |
| end = target[4] # End position | |
| strand = target[5] # Strand | |
| target_seq = target[0] # Full target sequence including PAM | |
| formatted_data.append([chr, start, end, strand, target_seq, gRNA, prediction[0][0]]) | |
| return formatted_data | |
| def process_gene(gene_symbol, model_path): | |
| transcripts = fetch_ensembl_transcripts(gene_symbol) | |
| all_data = [] | |
| gene_sequence = '' # Initialize an empty string for the gene sequence | |
| # Load the model | |
| seq_deepCpf1 = Seq_DeepCpf1_model(input_shape=(34, 4)) | |
| seq_deepCpf1.load_weights(model_path) | |
| if transcripts: | |
| for transcript in transcripts: | |
| transcript_id = transcript['id'] | |
| chr = transcript.get('seq_region_name', 'unknown') | |
| start = transcript.get('start', 0) | |
| strand = transcript.get('strand', 'unknown') | |
| # Fetch the sequence here and concatenate if multiple transcripts | |
| gene_sequence += fetch_ensembl_sequence(transcript_id) or '' | |
| if gene_sequence: | |
| targets = find_crispr_targets(gene_sequence, chr, start, strand) | |
| if targets: | |
| formatted_data = format_prediction_output(targets, seq_deepCpf1) | |
| all_data.extend(formatted_data) | |
| else: | |
| print("Failed to retrieve transcripts.") | |
| return all_data, gene_sequence | |
| def create_genbank_features(formatted_data): | |
| features = [] | |
| for data in formatted_data: | |
| location = FeatureLocation(start=int(data[1]), end=int(data[2]), strand=(1 if data[3] == '+' else -1)) | |
| feature = SeqFeature(location=location, type="misc_feature", qualifiers={ | |
| 'label': data[5], # gRNA as label | |
| 'note': f"Prediction: {data[6]}" # Prediction score in note | |
| }) | |
| features.append(feature) | |
| return features | |
| def generate_genbank_file_from_data(formatted_data, gene_sequence, gene_symbol, output_path): | |
| features = create_genbank_features(formatted_data) | |
| record = SeqRecord(Seq(gene_sequence), id=gene_symbol, name=gene_symbol, | |
| description='CRISPR Cas12 predicted targets', features=features) | |
| record.annotations["molecule_type"] = "DNA" | |
| SeqIO.write(record, output_path, "genbank") | |
| def generate_bed_file_from_data(formatted_data, output_path): | |
| with open(output_path, 'w') as bed_file: | |
| for data in formatted_data: | |
| chrom = data[0] | |
| start = data[1] | |
| end = data[2] | |
| strand = data[3] | |
| gRNA = data[5] | |
| score = data[6] | |
| bed_file.write(f"{chrom}\t{start}\t{end}\t{gRNA}\t{score}\t{strand}\n") |