cas12.py

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, end, strand, transcript_id, exon_id, pam="TTTN", target_length=34):
    targets = []
    len_sequence = len(sequence)
    complement = {'A': 'T', 'T': 'A', 'C': 'G', 'G': 'C'}
    dnatorna = {'A': 'A', 'T': 'U', 'C': 'C', 'G': 'G'}

    for i in range(len_sequence - target_length + 1):
        target_seq = sequence[i:i + target_length]
        if target_seq[4:7] == 'TTT':
            if strand == -1:
                tar_start = end - i - target_length + 1
                tar_end = end -i
                #seq_in_ref = ''.join([complement[base] for base in target_seq])[::-1]
            else:
                tar_start = start + i
                tar_end = start + i + target_length - 1
                #seq_in_ref = target_seq
            gRNA = ''.join([dnatorna[base] for base in target_seq[8:28]])
            targets.append([target_seq, gRNA, chr, str(tar_start), str(tar_end), str(strand), transcript_id, exon_id])
    return targets

def format_prediction_output(targets, model_path):
    # Loading weights for the model
    Seq_deepCpf1 = Seq_DeepCpf1_model(input_shape=(34, 4))
    Seq_deepCpf1.load_weights(model_path)

    formatted_data = []
    for target in targets:
        # Predict
        encoded_seq = get_seqcode(target[0])
        prediction = float(list(Seq_deepCpf1.predict(encoded_seq)[0])[0])
        if prediction > 100:
            prediction = 100

        # Format output
        gRNA = target[1]
        chr = target[2]
        start = target[3]
        end = target[4]
        strand = target[5]
        transcript_id = target[6]
        exon_id = target[7]
        formatted_data.append([chr, start, end, strand, transcript_id, exon_id, target[0], gRNA, prediction])

    return formatted_data

def process_gene(gene_symbol, model_path):
    transcripts = fetch_ensembl_transcripts(gene_symbol)
    results = []
    all_exons = []
    all_gene_sequences = []
    
    if transcripts:
        for i in range(len(transcripts)):
            Exons = transcripts[i]['Exon']
            all_exons.append(Exons)
            transcript_id = transcripts[i]['id']
            for j in range(len(Exons)):
                exon_id = Exons[j]['id']
                gene_sequence = fetch_ensembl_sequence(exon_id)
                if gene_sequence:
                    all_gene_sequences.append(gene_sequence)
                    start = Exons[j]['start']
                    end = Exons[j]['end']
                    strand = Exons[j]['strand']
                    chr = Exons[j]['seq_region_name']
                    targets = find_crispr_targets(gene_sequence, chr, start, end, strand, transcript_id, exon_id)
                    if targets:
                        formatted_data = format_prediction_output(targets, model_path)
                        results.append(formatted_data)
                        # for data in formatted_data:
                        #    print(f"Chr: {data[0]}, Start: {data[1]}, End: {data[2]}, Strand: {data[3]}, target: {data[4]}, gRNA: {data[5]}, pred_Score: {data[6]}")
                else:
                    print("Failed to retrieve gene sequence.")
    else:
        print("Failed to retrieve transcripts.")

    return results, all_gene_sequences, all_exons


# def create_genbank_features(formatted_data):
#     features = []
#     for data in formatted_data:
#         try:
#             # Attempt to convert start and end positions to integers
#             start = int(data[1])
#             end = int(data[2])
#         except ValueError as e:
#             # Log the error and skip this iteration if conversion fails
#             print(f"Error converting start/end to int: {data[1]}, {data[2]} - {e}")
#             continue  # Skip this iteration
#
#         # Proceed as normal if conversion is successful
#         strand = 1 if data[3] == '+' else -1
#         location = FeatureLocation(start=start, end=end, strand=strand)
#         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 create_csv_from_df(df, output_path):
#     df.to_csv(output_path, index=False)
#
# def generate_bed_file_from_data(formatted_data, output_path):
#     with open(output_path, 'w') as bed_file:
#         for data in formatted_data:
#             try:
#                 # Ensure data has the expected number of elements
#                 if len(data) < 7:
#                     raise ValueError("Incomplete data item")
#
#                 chrom = data[0]
#                 start = data[1]
#                 end = data[2]
#                 strand = '+' if data[3] == '+' else '-'
#                 gRNA = data[5]
#                 score = data[6]  # Ensure this index exists
#
#                 bed_file.write(f"{chrom}\t{start}\t{end}\t{gRNA}\t{score}\t{strand}\n")
#             except ValueError as e:
#                 print(f"Skipping an item due to error: {e}")
#                 continue