cas9on.py

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(gRNAs, model_path):
    dcModel = DCModelOntar(model_path)
    formatted_data = []

    for gRNA in gRNAs:
        # Encode the gRNA sequence
        encoded_seq = get_seqcode(gRNA[0]).reshape(-1,4,1,23)

        # Predict on-target efficiency using the model
        prediction = dcModel.ontar_predict(encoded_seq)

        # Format output
        chr = gRNA[1]
        start = gRNA[2]
        end = gRNA[3]
        strand = gRNA[4]
        formatted_data.append([chr, start, end, strand, gRNA[0], prediction[0]])

    return formatted_data

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="NGG", target_length=20):
    targets = []
    len_sequence = len(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
                targets.append([target_seq, chr, tar_start, tar_end, strand])

    return targets

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

    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:
                gRNA_sites = find_crispr_targets(gene_sequence, chr, start, strand)
                if gRNA_sites:
                    formatted_data = format_prediction_output(gRNA_sites, model_path)
                    all_data.extend(formatted_data)

    # Return both the data and the fetched sequence
    return all_data, gene_sequence

def create_genbank_features(gRNAs, predictions):
    features = []
    for gRNA, prediction in zip(gRNAs, predictions):
        # Assuming gRNA structure: [Target Seq, Chrom, Start Pos, End Pos, Strand]
        # And prediction is a single floating point value
        location = FeatureLocation(start=gRNA[2], end=gRNA[3], strand=gRNA[4])
        # Creating a feature with type 'CDS' just as an example, change as appropriate
        feature = SeqFeature(location=location, type="CDS", qualifiers={
            'label': gRNA[0],  # Target sequence as label
            'note': f"Prediction: {prediction}"  # Prediction score in note
        })
        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():
        location = FeatureLocation(start=int(row["Start Pos"]),
                                   end=int(row["End Pos"]),
                                   strand=int(row["Strand"]))
        feature = SeqFeature(location=location, type="gene", qualifiers={
            'locus_tag': row["Gene ID"],  # Assuming Gene ID is equivalent to Chromosome here
            'note': f"gRNA: {row['gRNA']}, Prediction: {row['Prediction']}"
        })
        features.append(feature)

    record = SeqRecord(Seq(gene_sequence), id=gene_symbol, name=gene_symbol,
                       description='CRISPR Cas9 predicted targets', features=features)

    # Add the missing molecule_type annotation
    record.annotations["molecule_type"] = "DNA"

    SeqIO.write(record, output_path, "genbank")