Spaces:
Sleeping
Sleeping
| import requests | |
| import tensorflow as tf | |
| import pandas as pd | |
| import numpy as np | |
| from operator import add | |
| from functools import reduce | |
| from Bio import SeqIO | |
| from Bio.SeqRecord import SeqRecord | |
| from Bio.SeqFeature import SeqFeature, FeatureLocation | |
| from Bio.Seq import Seq | |
| from keras.models import load_model | |
| import random | |
| # configure GPUs | |
| for gpu in tf.config.list_physical_devices('GPU'): | |
| tf.config.experimental.set_memory_growth(gpu, enable=True) | |
| if len(tf.config.list_physical_devices('GPU')) > 0: | |
| tf.config.experimental.set_visible_devices(tf.config.list_physical_devices('GPU')[0], 'GPU') | |
| 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)) | |
| from keras.models import load_model | |
| class DCModelOntar: | |
| def __init__(self, ontar_model_dir, is_reg=False): | |
| self.model = load_model(ontar_model_dir) | |
| def ontar_predict(self, x, channel_first=True): | |
| if channel_first: | |
| x = x.transpose([0, 2, 3, 1]) | |
| yp = self.model.predict(x) | |
| return yp.ravel() | |
| # Function to predict on-target efficiency and format output | |
| def format_prediction_output(targets, model_path): | |
| dcModel = DCModelOntar(model_path) | |
| formatted_data = [] | |
| for target in targets: | |
| # Encode the gRNA sequence | |
| encoded_seq = get_seqcode(target[0]).reshape(-1,4,1,23) | |
| # Predict on-target efficiency using the model | |
| prediction = dcModel.ontar_predict(encoded_seq) | |
| # Format output | |
| sgRNA = target[1] | |
| chr = target[2] | |
| start = target[3] | |
| end = target[4] | |
| strand = target[5] | |
| transcript_id = target[6] | |
| formatted_data.append([chr, start, end, strand, transcript_id, target[0], sgRNA, prediction[0]]) | |
| return formatted_data | |
| def fetch_ensembl_transcripts(gene_symbol): | |
| headers = {"Content-Type": "application/json"} | |
| url = f"https://rest.ensembl.org/lookup/symbol/homo_sapiens/{gene_symbol}?expand=1" | |
| response = requests.get(url, headers=headers) | |
| if response.status_code == 200: | |
| gene_data = response.json() | |
| return gene_data.get('Transcript', []) | |
| else: | |
| print(f"Error fetching gene data from Ensembl: {response.text}") | |
| return None | |
| def fetch_ensembl_sequence(transcript_id): | |
| headers = {"Content-Type": "application/json"} | |
| url = f"https://rest.ensembl.org/sequence/id/{transcript_id}" | |
| response = requests.get(url, headers=headers) | |
| if response.status_code == 200: | |
| sequence_data = response.json() | |
| return sequence_data.get('seq', '') | |
| else: | |
| print(f"Error fetching sequence data from Ensembl for transcript {transcript_id}: {response.text}") | |
| return None | |
| def fetch_ensembl_exons(transcript_id): | |
| headers = {"Content-Type": "application/json"} | |
| url = f"https://rest.ensembl.org/overlap/id/{transcript_id}?feature=exon" | |
| response = requests.get(url, headers=headers) | |
| if response.status_code == 200: | |
| return response.json() | |
| else: | |
| print(f"Error fetching exon data from Ensembl for transcript {transcript_id}: {response.text}") | |
| return None | |
| def fetch_ensembl_cds(transcript_id): | |
| headers = {"Content-Type": "application/json"} | |
| url = f"https://rest.ensembl.org/overlap/id/{transcript_id}?feature=cds" | |
| response = requests.get(url, headers=headers) | |
| if response.status_code == 200: | |
| return response.json() | |
| else: | |
| print(f"Error fetching CDS data from Ensembl for transcript {transcript_id}: {response.text}") | |
| return None | |
| def find_crispr_targets(sequence, chr, start, strand, transcript_id, pam="NGG", target_length=20): | |
| targets = [] | |
| len_sequence = len(sequence) | |
| complement = {'A': 'T', 'T': 'A', 'C': 'G', 'G': 'C'} | |
| if strand == -1: | |
| sequence = ''.join([complement[base] for base in reversed(sequence)]) | |
| for i in range(len_sequence - len(pam) + 1): | |
| if sequence[i + 1:i + 3] == pam[1:]: | |
| if i >= target_length: | |
| target_seq = sequence[i - target_length:i + 3] | |
| tar_start = start + i - target_length | |
| tar_end = start + i + 3 | |
| sgRNA = sequence[i - target_length:i] | |
| targets.append([target_seq, sgRNA, chr, str(tar_start), str(tar_end), str(strand), transcript_id]) | |
| return targets | |
| def process_gene(gene_symbol, model_path): | |
| transcripts = fetch_ensembl_transcripts(gene_symbol) | |
| all_data = [] | |
| if transcripts: | |
| cdslist = fetch_ensembl_cds(transcripts[0].get('id')) | |
| for transcript in transcripts: | |
| transcript_id = transcript.get('id') | |
| chr = transcript.get('seq_region_name', 'unknown') | |
| start = transcript.get('start', 0) | |
| strand = transcript.get('strand', 'unknown') | |
| # Fetch the gene sequence for each transcript | |
| gene_sequence = fetch_ensembl_sequence(transcript_id) or '' | |
| # Fetch exon and CDS information is not directly used here but you may need it elsewhere | |
| exons = fetch_ensembl_exons(transcript_id) | |
| if gene_sequence: | |
| # Now correctly passing transcript_id as an argument | |
| gRNA_sites = find_crispr_targets(gene_sequence, chr, start, strand, transcript_id) | |
| if gRNA_sites: | |
| formatted_data = format_prediction_output(gRNA_sites, model_path) | |
| all_data.extend(formatted_data) | |
| # Return the data and potentially any other information as needed | |
| return all_data, gene_sequence, exons, cdslist | |
| def create_genbank_features(formatted_data): | |
| features = [] | |
| for data in formatted_data: | |
| # Strand conversion to Biopython's convention | |
| strand = 1 if data[3] == '+' else -1 | |
| location = FeatureLocation(start=int(data[1]), end=int(data[2]), strand=strand) | |
| feature = SeqFeature(location=location, type="misc_feature", qualifiers={ | |
| 'label': data[5], # Use gRNA as the label | |
| 'target': data[4], # Include the target sequence | |
| 'note': f"Prediction: {data[6]}" # Include the prediction score | |
| }) | |
| features.append(feature) | |
| return features | |
| def generate_genbank_file_from_df(df, gene_sequence, gene_symbol, output_path): | |
| features = [] | |
| for index, row in df.iterrows(): | |
| # Use 'Transcript ID' if it exists, otherwise use a default value like 'Unknown' | |
| transcript_id = row.get("Transcript ID", "Unknown") | |
| # Make sure to use the correct column names for Start Pos, End Pos, and Strand | |
| location = FeatureLocation(start=int(row["Start Pos"]), | |
| end=int(row["End Pos"]), | |
| strand=1 if row["Strand"] == '+' else -1) | |
| feature = SeqFeature(location=location, type="gene", qualifiers={ | |
| 'locus_tag': transcript_id, # Now using the variable that holds the safe value | |
| 'note': f"gRNA: {row['gRNA']}, Prediction: {row['Prediction']}" | |
| }) | |
| features.append(feature) | |
| # The rest of the function remains unchanged | |
| record = SeqRecord(Seq(gene_sequence), id=gene_symbol, name=gene_symbol, | |
| description=f'CRISPR Cas9 predicted targets for {gene_symbol}', features=features) | |
| record.annotations["molecule_type"] = "DNA" | |
| SeqIO.write(record, output_path, "genbank") | |
| def create_bed_file_from_df(df, output_path): | |
| with open(output_path, 'w') as bed_file: | |
| for index, row in df.iterrows(): | |
| # Adjust field names based on your actual formatted data | |
| chrom = row["Chr"] | |
| start = int(row["Start Pos"]) | |
| end = int(row["End Pos"]) | |
| strand = '+' if row["Strand"] == '+' else '-' # Ensure strand is correctly interpreted | |
| gRNA = row["gRNA"] | |
| score = str(row["Prediction"]) # Ensure score is converted to string if not already | |
| transcript_id = row["Transcript"] # Extract transcript ID | |
| bed_file.write(f"{chrom}\t{start}\t{end}\t{gRNA}\t{score}\t{strand}\t{transcript_id}\n") # Include transcript ID in BED output | |
| def create_csv_from_df(df, output_path): | |
| df.to_csv(output_path, index=False) |