| import _bootstrap
|
| import torch
|
| from model.build_vocab import WordVocab |
| from model.pretrain_trfm import TrfmSeq2seq |
| from model.utils import split |
| import json
|
| from transformers import T5EncoderModel, T5Tokenizer
|
| import re
|
| import gc
|
| from sklearn import metrics
|
| from sklearn.ensemble import ExtraTreesRegressor
|
| import numpy as np
|
| import pandas as pd
|
| from sklearn.model_selection import KFold
|
| from sklearn.metrics import mean_squared_error, mean_absolute_error
|
| from sklearn.metrics import r2_score
|
| from sklearn.model_selection import train_test_split
|
| import random
|
| import pickle
|
| import math
|
| from project_paths import (
|
|
|
| DEGREE_SMILES_PATH, KCAT_DATASET_PATH, KCAT_KM_SAMPLES_PATH, KM_TEST_PATH,
|
|
|
| PH_SMILES_PATH, PROT_T5_MODEL, TRFM_PATH, UNIKP_MODEL_DIR, VOCAB_PATH,
|
|
|
| )
|
|
|
|
|
|
|
| def smiles_to_vec(Smiles):
|
| pad_index = 0
|
| unk_index = 1
|
| eos_index = 2
|
| sos_index = 3
|
| mask_index = 4
|
| vocab = WordVocab.load_vocab(VOCAB_PATH)
|
| def get_inputs(sm):
|
| seq_len = 220
|
| sm = sm.split()
|
| if len(sm)>218:
|
|
|
| sm = sm[:109]+sm[-109:]
|
| ids = [vocab.stoi.get(token, unk_index) for token in sm]
|
| ids = [sos_index] + ids + [eos_index]
|
| seg = [1]*len(ids)
|
| padding = [pad_index]*(seq_len - len(ids))
|
| ids.extend(padding), seg.extend(padding)
|
| return ids, seg
|
| def get_array(smiles):
|
| x_id, x_seg = [], []
|
| for sm in smiles:
|
| a,b = get_inputs(sm)
|
| x_id.append(a)
|
| x_seg.append(b)
|
| return torch.tensor(x_id), torch.tensor(x_seg)
|
| trfm = TrfmSeq2seq(len(vocab), 256, len(vocab), 4)
|
| trfm.load_state_dict(torch.load(TRFM_PATH))
|
| trfm.eval()
|
| x_split = [split(sm) for sm in Smiles]
|
| xid, xseg = get_array(x_split)
|
| X = trfm.encode(torch.t(xid))
|
| return X
|
|
|
|
|
| def Seq_to_vec(Sequence):
|
| for i in range(len(Sequence)):
|
| if len(Sequence[i]) > 1000:
|
| Sequence[i] = Sequence[i][:500] + Sequence[i][-500:]
|
| sequences_Example = []
|
| for i in range(len(Sequence)):
|
| zj = ''
|
| for j in range(len(Sequence[i]) - 1):
|
| zj += Sequence[i][j] + ' '
|
| zj += Sequence[i][-1]
|
| sequences_Example.append(zj)
|
| tokenizer = T5Tokenizer.from_pretrained(PROT_T5_MODEL, do_lower_case=False)
|
| model = T5EncoderModel.from_pretrained(PROT_T5_MODEL)
|
| gc.collect()
|
| print(torch.cuda.is_available())
|
|
|
| device = torch.device('cuda:0' if torch.cuda.is_available() else 'cpu')
|
| model = model.to(device)
|
| model = model.eval()
|
| features = []
|
| for i in range(len(sequences_Example)):
|
| print('For sequence ', str(i+1))
|
| sequences_Example_i = sequences_Example[i]
|
| sequences_Example_i = [re.sub(r"[UZOB]", "X", sequences_Example_i)]
|
| ids = tokenizer.batch_encode_plus(sequences_Example_i, add_special_tokens=True, padding=True)
|
| input_ids = torch.tensor(ids['input_ids']).to(device)
|
| attention_mask = torch.tensor(ids['attention_mask']).to(device)
|
| with torch.no_grad():
|
| embedding = model(input_ids=input_ids, attention_mask=attention_mask)
|
| embedding = embedding.last_hidden_state.cpu().numpy()
|
| for seq_num in range(len(embedding)):
|
| seq_len = (attention_mask[seq_num] == 1).sum()
|
| seq_emd = embedding[seq_num][:seq_len - 1]
|
| features.append(seq_emd)
|
| features_normalize = np.zeros([len(features), len(features[0][0])], dtype=float)
|
| for i in range(len(features)):
|
| for k in range(len(features[0][0])):
|
| for j in range(len(features[i])):
|
| features_normalize[i][k] += features[i][j][k]
|
| features_normalize[i][k] /= len(features[i])
|
| return features_normalize
|
|
|
|
|
| def Kcat_predict(Ifeature, Label, sequence_new, Smiles_new, ECNumber_new, Organism_new, Substrate_new, Type_new):
|
| for i in range(10):
|
|
|
| ALL_index = [j for j in range(len(Ifeature))]
|
| train_index = np.array(random.sample(ALL_index, int(len(ALL_index)*0.9)))
|
| Training_or_test = []
|
| for j in range(len(ALL_index)):
|
| if ALL_index[j] in train_index:
|
| Training_or_test.append(0)
|
| else:
|
| Training_or_test.append(1)
|
| Training_or_test = np.array(Training_or_test)
|
| Train_data, Train_label = Ifeature[train_index], Label[train_index]
|
| model = ExtraTreesRegressor()
|
| model.fit(Train_data, Train_label)
|
| Pre_all_label = model.predict(Ifeature)
|
| res = pd.DataFrame({'sequence': sequence_new, 'smiles': Smiles_new, 'ECNumber': ECNumber_new,
|
| 'Organism': Organism_new, 'Substrate': Substrate_new, 'Type': Type_new,
|
| 'Label': Label, 'Predict_Label': Pre_all_label, 'Training or test': Training_or_test})
|
| res.to_excel('PreKcat_new/'+str(i+1)+'_all_samples_metrics.xlsx')
|
|
|
|
|
| if __name__ == '__main__':
|
|
|
| with open(KCAT_DATASET_PATH, 'r') as file:
|
| datasets = json.load(file)
|
|
|
| sequence = [data['Sequence'] for data in datasets]
|
| Smiles = [data['Smiles'] for data in datasets]
|
| Label = [float(data['Value']) for data in datasets]
|
| ECNumber = [data['ECNumber'] for data in datasets]
|
| Organism = [data['Organism'] for data in datasets]
|
| Substrate = [data['Substrate'] for data in datasets]
|
| Type = [data['Type'] for data in datasets]
|
| for i in range(len(Label)):
|
| if Label[i] == 0:
|
| Label[i] = -10000000000
|
| else:
|
| Label[i] = math.log(Label[i], 10)
|
| print(max(Label), min(Label))
|
|
|
| smiles_input = smiles_to_vec(Smiles)
|
| sequence_input = Seq_to_vec(sequence)
|
| feature = np.concatenate((smiles_input, sequence_input), axis=1)
|
| with open("PreKcat_new/features_16838_PreKcat.pkl", "wb") as f:
|
| pickle.dump(feature, f)
|
| Label = np.array(Label)
|
|
|
| feature_new = []
|
| Label_new = []
|
| sequence_new = []
|
| Smiles_new = []
|
| ECNumber_new = []
|
| Organism_new = []
|
| Substrate_new = []
|
| Type_new = []
|
| for i in range(len(Label)):
|
| if -10000000000 < Label[i] and '.' not in Smiles[i]:
|
| feature_new.append(feature[i])
|
| Label_new.append(Label[i])
|
| sequence_new.append(sequence[i])
|
| Smiles_new.append(Smiles[i])
|
| ECNumber_new.append(ECNumber[i])
|
| Organism_new.append(Organism[i])
|
| Substrate_new.append(Substrate[i])
|
| Type_new.append(Type[i])
|
| print(len(Label_new), min(Label_new), max(Label_new))
|
| Label_new = np.array(Label_new)
|
| feature_new = np.array(feature_new)
|
|
|
| Kcat_predict(feature_new, Label_new, sequence_new, Smiles_new, ECNumber_new,
|
| Organism_new, Substrate_new, Type_new)
|
|
|
