import torch import torch.nn as nn import torch.nn.functional as F import torch # import os import pandas as pd import numpy as np # import json from fastNLP.core.metrics import MetricBase,seq_len_to_mask from fastNLP.core.losses import LossBase from preprocess import NPPeptidePipe,PPeptidePipe from torch.nn import CosineSimilarity import ipdb from utils import * from sklearn.metrics import r2_score def set_seed(seed): np.random.seed(seed) torch.manual_seed(seed) torch.cuda.manual_seed(seed) torch.cuda.manual_seed_all(seed) def attentionmask(seq_len, max_len=None): r""" 将一个表示sequence length的一维数组转换为二维的mask,不包含的位置为0。 转变 1-d seq_len到2-d mask. .. code-block:: # # >>> seq_len = torch.arange(2, 16) # >>> mask = seq_len_to_mask(seq_len) # >>> print(mask.size()) # torch.Size([14, 15]) # >>> seq_len = np.arange(2, 16) # >>> mask = seq_len_to_mask(seq_len) # >>> print(mask.shape) # (14, 15) # >>> seq_len = torch.arange(2, 16) # >>> mask = seq_len_to_mask(seq_len, max_len=100) # >>>print(mask.size()) torch.Size([14, 100]) :param np.ndarray,torch.LongTensor seq_len: shape将是(B,) :param int max_len: 将长度pad到这个长度。默认(None)使用的是seq_len中最长的长度。但在nn.DataParallel的场景下可能不同卡的seq_len会有 区别，所以需要传入一个max_len使得mask的长度是pad到该长度。 :return: np.ndarray, torch.Tensor 。shape将是(B, max_length)， 元素类似为bool或torch.uint8 """ if isinstance(seq_len, np.ndarray): assert len(np.shape(seq_len)) == 1, f"seq_len can only have one dimension, got {len(np.shape(seq_len))}." max_len = int(max_len) if max_len else int(seq_len.max()) broad_cast_seq_len = np.tile(np.arange(max_len), (len(seq_len), 1)) mask = broad_cast_seq_len < seq_len.reshape(-1, 1) elif isinstance(seq_len, torch.Tensor): assert seq_len.dim() == 1, f"seq_len can only have one dimension, got {seq_len.dim() == 1}." batch_size = seq_len.size(0) max_len = int(max_len) if max_len else seq_len.max().long() broad_cast_seq_len = torch.arange(max_len).expand(batch_size, -1).to(seq_len) mask = broad_cast_seq_len.ge(seq_len.unsqueeze(1)) else: raise TypeError("Only support 1-d numpy.ndarray or 1-d torch.Tensor.") return mask class CossimilarityMetric(MetricBase): r""" 准确率Metric（其它的Metric参见 :mod:`fastNLP.core.metrics` ） """ def __init__(self, pred=None, target=None, seq_len=None,num_col=None): r""" :param pred: 参数映射表中 `pred` 的映射关系，None表示映射关系为 `pred` -> `pred` :param target: 参数映射表中 `target` 的映射关系，None表示映射关系为 `target` -> `target` :param seq_len: 参数映射表中 `seq_len` 的映射关系，None表示映射关系为 `seq_len` -> `seq_len` """ super().__init__() self._init_param_map(pred=pred, target=target, seq_len=seq_len) self.bestcos=0 self.total = 0 self.cos = 0 self.listcos=[] self.num_col=num_col def evaluate(self, pred, target, seq_len=None): r""" evaluate函数将针对一个批次的预测结果做评价指标的累计 :param torch.Tensor pred: 预测的tensor, tensor的形状可以是torch.Size([B,]), torch.Size([B, n_classes]), torch.Size([B, max_len]), 或者torch.Size([B, max_len, n_classes]) :param torch.Tensor target: 真实值的tensor, tensor的形状可以是Element's can be: torch.Size([B,]), torch.Size([B,]), torch.Size([B, max_len]), 或者torch.Size([B, max_len]) :param torch.Tensor seq_len: 序列长度标记, 标记的形状可以是None, None, torch.Size([B]), 或者torch.Size([B]). 如果mask也被传进来的话seq_len会被忽略. """ N=pred.size(0) L_1=pred.size(1) if seq_len is not None and target.dim() > 1: max_len = target.size(1) masks = seq_len_to_mask(seq_len=(seq_len-1)*int(self.num_col)) else: masks = None cos=CosineSimilarity(dim=1) if masks is not None: pred=pred.masked_fill(masks.eq(False), 0) self.cos += torch.sum(cos(pred,target)).item() self.total += pred.size(0) self.bestcos=max(self.bestcos,torch.max(cos(pred,target)).item()) self.listcos += cos(pred, target).reshape(N, ).cpu().numpy().tolist() else: self.cos += torch.sum(cos(pred,target)).item() self.total += pred.size(0) self.bestcos = max(self.bestcos, torch.max(cos(pred,target)).item()) self.listcos +=cos(pred, target).reshape(N,).cpu().numpy().tolist() def get_metric(self, reset=True): r""" get_metric函数将根据evaluate函数累计的评价指标统计量来计算最终的评价结果. :param bool reset: 在调用完get_metric后是否清空评价指标统计量. :return dict evaluate_result: {"acc": float} """ evaluate_result = {'mediancos':round(np.median(self.listcos),6), 'coss': round(float(self.cos) / (self.total + 1e-12), 6), 'bestcoss':round(self.bestcos, 6), } if reset: self.cos = 0 self.total = 0 self.bestcos=0 self.listcos=[] return evaluate_result class CossimilarityMetricfortest(MetricBase): r""" 准确率Metric（其它的Metric参见 :mod:`fastNLP.core.metrics` ） """ def __init__(self,savename, pred=None, target=None, seq_len=None,num_col=None,sequence=None,charge=None,decoration=None): r""" :param pred: 参数映射表中 `pred` 的映射关系，None表示映射关系为 `pred` -> `pred` :param target: 参数映射表中 `target` 的映射关系，None表示映射关系为 `target` -> `target` :param seq_len: 参数映射表中 `seq_len` 的映射关系，None表示映射关系为 `seq_len` -> `seq_len` """ super().__init__() self._init_param_map(pred=pred, target=target, seq_len=seq_len,sequence=sequence,charge=charge,decoration=decoration,_id="_id") self.bestcos=0 self.total = 0 self.cos = 0 self.listcos=[] self.nan=0 self.bestanswer=0 self.nansequence=pd.DataFrame(columns=['nansequence','charge','decoration']) self.num_col=num_col self.savename=savename if savename else "" self.id_list=[] def evaluate(self, pred, target, seq_len=None,sequence=None, charge=None,decoration=None,_id=None): r""" evaluate函数将针对一个批次的预测结果做评价指标的累计 :param torch.Tensor pred: 预测的tensor, tensor的形状可以是torch.Size([B,]), torch.Size([B, n_classes]), torch.Size([B, max_len]), 或者torch.Size([B, max_len, n_classes]) :param torch.Tensor target: 真实值的tensor, tensor的形状可以是Element's can be: torch.Size([B,]), torch.Size([B,]), torch.Size([B, max_len]), 或者torch.Size([B, max_len]) :param torch.Tensor seq_len: 序列长度标记, 标记的形状可以是None, None, torch.Size([B]), 或者torch.Size([B]). 如果mask也被传进来的话seq_len会被忽略. """ N=pred.size(0) L_1=pred.size(1) if seq_len is not None and target.dim() > 1: max_len = target.size(1) masks = seq_len_to_mask(seq_len=(seq_len-1)*int(self.num_col)) else: masks = None cos=CosineSimilarity(dim=1) #可以开根号，或者correlation coefficient # self.id_list+=_id.cpu().numpy().tolist() if masks is not None: s=torch.sum(cos(pred, target)).item() pred=pred.masked_fill(masks.eq(False), 0) if math.isnan(s): ipdb.set_trace() self.nansequence.loc[self.nan] = [sequence.cpu().numpy().tolist(),charge.cpu().numpy().tolist(),decoration.cpu().numpy().tolist()] self.nan+=1 else: self.cos += torch.sum(cos(pred,target)).item() self.total += pred.size(0) self.bestcos=max(self.bestcos,torch.max(cos(pred,target)).item()) self.listcos += cos(pred, target).reshape(N, ).cpu().numpy().tolist() else: s=torch.sum(cos(pred, target)).item() print(s) if math.isnan(s): # ipdb.set_trace() print("getnan:{}".format(_id.cpu().numpy().tolist()[0])) self.nansequence.loc[self.nan] = [sequence.cpu().numpy().tolist(),charge.cpu().numpy().tolist(),decoration.cpu().numpy().tolist()] self.nan+=1 else: self.cos += s self.total += pred.size(0) self.bestcos = max(self.bestcos, torch.max(cos(pred,target)).item()) self.listcos +=cos(pred, target).reshape(N,).cpu().numpy().tolist() def get_metric(self, reset=True): r""" get_metric函数将根据evaluate函数累计的评价指标统计量来计算最终的评价结果. :param bool reset: 在调用完get_metric后是否清空评价指标统计量. :return dict evaluate_result: {"acc": float} """ data=pd.Series(self.listcos) mediancos=np.median(self.listcos) # if mediancos>self.bestanswer: # data.to_csv(self.savename+"Cossimilaritylist.csv",index=False) if self.nan>0: self.nansequence.to_json(self.savename+"nansequence.json") evaluate_result = {'mediancos':round(mediancos,6), 'nan number':self.nan, 'coss': round(float(self.cos) / (self.total + 1e-12), 6), 'bestcoss':round(self.bestcos, 6), } if reset: self.cos = 0 self.total = 0 self.bestcos=0 self.listcos=[] return evaluate_result class CossimilarityMetricfortest_outputmsms(MetricBase): r""" 准确率Metric（其它的Metric参见 :mod:`fastNLP.core.metrics` ） """ def __init__(self, savename,pred=None, target=None, seq_len=None,num_col=None,sequence=None,charge=None,decoration=None,_id=None): r""" :param pred: 参数映射表中 `pred` 的映射关系，None表示映射关系为 `pred` -> `pred` :param target: 参数映射表中 `target` 的映射关系，None表示映射关系为 `target` -> `target` :param seq_len: 参数映射表中 `seq_len` 的映射关系，None表示映射关系为 `seq_len` -> `seq_len` """ super().__init__() self._init_param_map(pred=pred, target=target, seq_len=seq_len,sequence=sequence,charge=charge,decoration=decoration,_id=_id) self.bestcos=0 self.total = 0 self.cos = 0 self.listcos=[] self.nj=0 self.savename=savename if savename else "" self.repsequence=pd.DataFrame(columns=['repsequence','charge','decoration','ms2',"cos","id"]) self.numcol=num_col self.id_list=[] def evaluate(self, pred, target, seq_len=None,sequence=None,charge=None,decoration=None,_id=None): r""" evaluate函数将针对一个批次的预测结果做评价指标的累计 :param torch.Tensor pred: 预测的tensor, tensor的形状可以是torch.Size([B,]), torch.Size([B, n_classes]), torch.Size([B, max_len]), 或者torch.Size([B, max_len, n_classes]) :param torch.Tensor target: 真实值的tensor, tensor的形状可以是Element's can be: torch.Size([B,]), torch.Size([B,]), torch.Size([B, max_len]), 或者torch.Size([B, max_len]) :param torch.Tensor seq_len: 序列长度标记, 标记的形状可以是None, None, torch.Size([B]), 或者torch.Size([B]). 如果mask也被传进来的话seq_len会被忽略. """ N=pred.size(0) #batch size L=pred.size(1) # if seq_len is not None and target.dim() > 1: # max_len = target.size(1) # masks = seq_len_to_mask(seq_len=(seq_len-1)*int(self.num_col)) # else: cos=CosineSimilarity(dim=1) s = torch.sum(cos(pred, target)).item() self.cos += s self.total += pred.size(0) self.bestcos = max(self.bestcos, torch.max(cos(pred, target)).item()) self.listcos += cos(pred, target).reshape(N, ).cpu().numpy().tolist() # print(_id) # import ipdb # ipdb.set_trace() self.id_list+=_id.cpu().numpy().tolist() for i in range(N): il = seq_len[i] isequence=sequence[i][:il] icharge=charge[i] # ipdb.set_trace() idecoration=decoration[i][:il] ims2=pred[i].reshape((-1,self.numcol))[:il-1,:self.numcol] icos=self.listcos[self.nj] # ipdb.set_trace() self.repsequence.loc[self.nj] = [isequence.cpu().numpy().tolist(), icharge.cpu().numpy().tolist(), idecoration.cpu().numpy().tolist(), ims2.cpu().numpy().tolist(), icos, self.id_list[self.nj]] self.nj+=1 def get_metric(self, reset=True): r""" get_metric函数将根据evaluate函数累计的评价指标统计量来计算最终的评价结果. :param bool reset: 在调用完get_metric后是否清空评价指标统计量. :return dict evaluate_result: {"acc": float} """ # data=pd.Series(self.listcos) # data.to_csv("Cossimilaritylist.csv",index=False) # self.repsequence.to_json(self.savename+"result.json") self.repsequence.to_csv(self.savename+"_by_result.csv",index=False) evaluate_result = {'mediancos':round(np.median(self.listcos),6), 'total number':self.nj, 'coss': round(float(self.cos) / (self.total + 1e-12), 6), 'bestcoss':round(self.bestcos, 6), } if reset: self.cos = 0 self.total = 0 self.bestcos=0 self.listcos=[] self.id_list=[] return evaluate_result class CossimilarityMetricfortest_BY(MetricBase): r""" 准确率Metric（其它的Metric参见 :mod:`fastNLP.core.metrics` ） """ def __init__(self,savename, pred=None, target=None, seq_len=None, num_col=None,sequence=None,charge=None,decoration=None): r""" :param pred: 参数映射表中 `pred` 的映射关系，None表示映射关系为 `pred` -> `pred` :param target: 参数映射表中 `target` 的映射关系，None表示映射关系为 `target` -> `target` :param seq_len: 参数映射表中 `seq_len` 的映射关系，None表示映射关系为 `seq_len` -> `seq_len` """ super().__init__() self._init_param_map(pred=pred, target=target, seq_len=seq_len,sequence=sequence,charge=charge,decoration=decoration,_id="_id",graph_edges="graph_edges") self.bestcos=0 self.total = 0 self.cos = 0 self.listcos=[] self.nan=0 self.bestanswer=0 self.nansequence=pd.DataFrame(columns=['nansequence','charge','decoration']) self.num_col=num_col self.savename=savename if savename else "" self.id_list=[] def evaluate(self, pred, target, seq_len=None,sequence=None,charge=None, decoration=None,_id=None,graph_edges=None): r""" evaluate函数将针对一个批次的预测结果做评价指标的累计 :param torch.Tensor pred: 预测的tensor, tensor的形状可以是torch.Size([B,]), torch.Size([B, n_classes]), torch.Size([B, max_len]), 或者torch.Size([B, max_len, n_classes]) :param torch.Tensor target: 真实值的tensor, tensor的形状可以是Element's can be: torch.Size([B,]), torch.Size([B,]), torch.Size([B, max_len]), 或者torch.Size([B, max_len]) :param torch.Tensor seq_len: 序列长度标记, 标记的形状可以是None, None, torch.Size([B]), 或者torch.Size([B]). 如果mask也被传进来的话seq_len会被忽略. """ N=pred.size(0) #改输入格式 cos=CosineSimilarity(dim=0) #现在的cos就是这样的，也没有开根号，因为target和pred都拍平了，这里不用dim=1 #可以开根号，或者correlation coefficient # self.id_list+=_id.cpu().numpy().tolist() pred[pred<0]=0 pred = torch.split(pred, graph_edges.tolist()) target= torch.split(target, graph_edges.tolist()) for c in range(len(graph_edges)): self.listcos+=[cos(pred[c].flatten(), target[c].flatten()).cpu().numpy()] s=sum(self.listcos) if math.isnan(s): print("getnan:{}".format(_id.cpu().numpy().tolist()[0])) self.nansequence.loc[self.nan] = [sequence.cpu().numpy().tolist(),charge.cpu().numpy().tolist(),decoration.cpu().numpy().tolist()] self.nan+=1 else: self.cos = s self.total += len(graph_edges) self.bestcos = max(self.listcos).item() assert len(self.listcos)==self.total,"the length of cos list is different from the number of graphs" def get_metric(self, reset=True): r""" get_metric函数将根据evaluate函数累计的评价指标统计量来计算最终的评价结果. :param bool reset: 在调用完get_metric后是否清空评价指标统计量. :return dict evaluate_result: {"acc": float} """ # data=pd.DataFrame(self.listcos) # data.columns=["cs"] mediancos=np.median(self.listcos) # if mediancos>self.bestanswer: # data.to_csv(self.savename+"Cossimilaritylist.csv",index=False) if self.nan>0: self.nansequence.to_json(self.savename+"nansequence.json") evaluate_result = {'mediancos':round(mediancos,6), 'nan number':self.nan, 'coss': round(float(self.cos) / (self.total + 1e-12), 6), 'bestcoss':round(self.bestcos, 6), } if reset: self.cos = 0 self.total = 0 self.bestcos=0 self.listcos=[] return evaluate_result class CossimilarityMetricfortest_outputmsmsBY(MetricBase): r""" 准确率Metric（其它的Metric参见 :mod:`fastNLP.core.metrics` ） """ def __init__(self, savename,pred=None, target=None, seq_len=None, num_col=None,sequence=None,charge=None,decoration=None,_id=None, peptide=None,PlausibleStruct=None): r""" :param pred: 参数映射表中 `pred` 的映射关系，None表示映射关系为 `pred` -> `pred` :param target: 参数映射表中 `target` 的映射关系，None表示映射关系为 `target` -> `target` :param seq_len: 参数映射表中 `seq_len` 的映射关系，None表示映射关系为 `seq_len` -> `seq_len` """ super().__init__() self._init_param_map(pred=pred, target=target, seq_len=seq_len, sequence=sequence,charge=charge,decoration=decoration, _id="_id",graph_edges="graph_edges", peptide=peptide,PlausibleStruct=PlausibleStruct) #增加了“graph_edges” self.bestcos=0 self.total = 0 self.cos = 0 self.listcos=[] self.nan=0 self.nj=0 self.bestanswer=0 self.nansequence=pd.DataFrame(columns=['nansequence','charge','decoration']) self.num_col=num_col self.savename=savename if savename else "" self.id_list=[] self.repsequence=pd.DataFrame(columns=['repsequence','charge', "ipeptide","iPlausibleStruct", 'ms2',"cos","id"]) def evaluate(self, pred, target, seq_len=None,sequence=None,charge=None, decoration=None,_id=None,graph_edges=None, peptide=None,PlausibleStruct=None): r""" evaluate函数将针对一个批次的预测结果做评价指标的累计 :param torch.Tensor pred: 预测的tensor, tensor的形状可以是torch.Size([B,]), torch.Size([B, n_classes]), torch.Size([B, max_len]), 或者torch.Size([B, max_len, n_classes]) :param torch.Tensor target: 真实值的tensor, tensor的形状可以是Element's can be: torch.Size([B,]), torch.Size([B,]), torch.Size([B, max_len]), 或者torch.Size([B, max_len]) :param torch.Tensor seq_len: 序列长度标记, 标记的形状可以是None, None, torch.Size([B]), 或者torch.Size([B]). 如果mask也被传进来的话seq_len会被忽略. """ N=pred.size(0) #改输入格式 cos=CosineSimilarity(dim=0) pred[pred<0]=0 pred = torch.split(pred, graph_edges.tolist()) target= torch.split(target, graph_edges.tolist()) for c in range(len(graph_edges)): self.listcos.append(cos(pred[c].flatten(), target[c].flatten()).cpu().numpy()) s=sum(self.listcos) if math.isnan(s): print("getnan:{}".format(_id.cpu().numpy().tolist()[0])) self.nansequence.loc[self.nan] = [sequence.cpu().numpy().tolist(),charge.cpu().numpy().tolist(),decoration.cpu().numpy().tolist()] self.nan+=1 else: self.cos = s self.total += len(graph_edges) self.bestcos = max(self.listcos).item() assert len(self.listcos)==self.total,"the length of cos list is different from the number of graphs" self.id_list+=_id.cpu().numpy().tolist() for i in range(len(graph_edges)): il = seq_len[i] isequence=sequence[i][:il] icharge=charge[i] ipeptide=peptide[i] iPlausibleStruct=PlausibleStruct[i] ims2=pred[i] icos=self.listcos[self.nj] # import ipdb # ipdb.set_trace() self.repsequence.loc[self.nj] = [isequence.cpu().numpy().tolist(), icharge.cpu().numpy().tolist(), ipeptide.tolist(), iPlausibleStruct.tolist(), ims2.cpu().numpy().tolist(), icos.tolist(), self.id_list[self.nj]] self.nj+=1 def get_metric(self, reset=True): r""" get_metric函数将根据evaluate函数累计的评价指标统计量来计算最终的评价结果. :param bool reset: 在调用完get_metric后是否清空评价指标统计量. :return dict evaluate_result: {"acc": float} """ data=pd.DataFrame(self.listcos) data.columns=["cs"] mediancos=np.median(self.listcos) if mediancos>self.bestanswer: pass # data.to_csv(self.savename+"Cossimilaritylist.csv",index=False) if self.nan>0: self.nansequence.to_json(self.savename+"nansequence.json") evaluate_result = {'mediancos':round(mediancos,6), 'nan number':self.nan, 'coss': round(float(self.cos) / (self.total + 1e-12), 6), 'bestcoss':round(self.bestcos, 6), } self.repsequence.to_csv(self.savename+"_BY_result.csv",index=False) if reset: self.cos = 0 self.total = 0 self.bestcos=0 self.listcos=[] return evaluate_result # --------------------------- byBY ---------------------# def pearsonr( x, y, batch_first=True, ): r"""Computes Pearson Correlation Coefficient across rows. Pearson Correlation Coefficient (also known as Linear Correlation Coefficient or Pearson's :math:`\rho`) is computed as: .. math:: \rho = \frac {E[(X-\mu_X)(Y-\mu_Y)]} {\sigma_X\sigma_Y} If inputs are matrices, then then we assume that we are given a mini-batch of sequences, and the correlation coefficient is computed for each sequence independently and returned as a vector. If `batch_fist` is `True`, then we assume that every row represents a sequence in the mini-batch, otherwise we assume that batch information is in the columns. Warning: We do not account for the multi-dimensional case. This function has been tested only for the 2D case, either in `batch_first==True` or in `batch_first==False` mode. In the multi-dimensional case, it is possible that the values returned will be meaningless. Args: x (torch.Tensor): input tensor y (torch.Tensor): target tensor batch_first (bool, optional): controls if batch dimension is first. Default: `True` Returns: torch.Tensor: correlation coefficient between `x` and `y` Note: :math:`\sigma_X` is computed using **PyTorch** builtin **Tensor.std()**, which by default uses Bessel correction: .. math:: \sigma_X=\displaystyle\frac{1}{N-1}\sum_{i=1}^N({x_i}-\bar{x})^2 We therefore account for this correction in the computation of the covariance by multiplying it with :math:`\frac{1}{N-1}`. Shape: - Input: :math:`(N, M)` for correlation between matrices, or :math:`(M)` for correlation between vectors - Target: :math:`(N, M)` or :math:`(M)`. Must be identical to input - Output: :math:`(N, 1)` for correlation between matrices, or :math:`(1)` for correlation between vectors Examples: >>> import torch >>> _ = torch.manual_seed(0) >>> input = torch.rand(3, 5) >>> target = torch.rand(3, 5) >>> output = pearsonr(input, target) >>> print('Pearson Correlation between input and target is {0}'.format(output[:, 0])) Pearson Correlation between input and target is tensor([ 0.2991, -0.8471, 0.9138]) """ # noqa: E501 assert x.shape == y.shape if batch_first: dim = -1 else: dim = 0 centered_x = x - x.mean(dim=dim, keepdim=True) centered_y = y - y.mean(dim=dim, keepdim=True) covariance = (centered_x * centered_y).sum(dim=dim, keepdim=True) bessel_corrected_covariance = covariance / (x.shape[dim] - 1) x_std = x.std(dim=dim, keepdim=True) y_std = y.std(dim=dim, keepdim=True) corr = bessel_corrected_covariance / (x_std * y_std) return corr def simlarcalc(spectrum_1_intensity,spectrum_2_intensity,type): #提供两种方法，开根号的cosine similarity与correlation coefficient "cos" or "corre" # 开根号也可以用poisson GL代替 if type =="cos": cos = torch.nn.CosineSimilarity(dim=1) sim = cos(spectrum_1_intensity, spectrum_2_intensity) if type=="pcc": sim = pearsonr(spectrum_1_intensity, spectrum_2_intensity).squeeze() if type=="cos_sqrt": cos = torch.nn.CosineSimilarity(dim=1) sim = cos(torch.sqrt(spectrum_1_intensity), torch.sqrt(spectrum_2_intensity)) if type=="corre_sqrt": spectrum_1_intensity=spectrum_1_intensity.sqrt() spectrum_2_intensity=spectrum_2_intensity.sqrt() sim = pearsonr(spectrum_1_intensity, spectrum_2_intensity).squeeze() return sim class CossimilarityMetricfortest_byBY(MetricBase): def __init__(self,savename, pred=None, target=None, seq_len=None, num_col=None,sequence=None,charge=None,decoration=None, args=None): r""" :param pred: 参数映射表中 `pred` 的映射关系，None表示映射关系为 `pred` -> `pred` :param target: 参数映射表中 `target` 的映射关系，None表示映射关系为 `target` -> `target` :param seq_len: 参数映射表中 `seq_len` 的映射关系，None表示映射关系为 `seq_len` -> `seq_len` """ super().__init__() self._init_param_map(pred=pred, target=target, seq_len=seq_len,sequence=sequence, charge=charge,decoration=decoration,_id="_id",graph_edges="graph_edges") self.bestcos=0 self.total = 0 self.cos = 0 self.listcos=[] self.nan=0 self.bestanswer=0 self.nansequence=pd.DataFrame(columns=['nansequence','charge','decoration']) self.num_col=num_col self.savename=savename if savename else "" self.id_list=[] self.args=args def evaluate(self, pred, target, seq_len=None,sequence=None,charge=None, decoration=None,_id=None,graph_edges=None): r""" evaluate函数将针对一个批次的预测结果做评价指标的累计 :param torch.Tensor pred: 预测的tensor, tensor的形状可以是torch.Size([B,]), torch.Size([B, n_classes]), torch.Size([B, max_len]), 或者torch.Size([B, max_len, n_classes]) :param torch.Tensor target: 真实值的tensor, tensor的形状可以是Element's can be: torch.Size([B,]), torch.Size([B,]), torch.Size([B, max_len]), 或者torch.Size([B, max_len]) :param torch.Tensor seq_len: 序列长度标记, 标记的形状可以是None, None, torch.Size([B]), 或者torch.Size([B]). 如果mask也被传进来的话seq_len会被忽略. """ N=pred.size(0) #改输入格式 #simlarcalc:cos,corre,cos_sqrt,corre_sqrt # cos=CosineSimilarity(dim=0) #现在的cos就是这样的，也没有开根号，因为target和pred都拍平了，这里不用dim=1 #可以开根号，或者correlation coefficient # self.id_list+=_id.cpu().numpy().tolist() # pred = torch.split(pred, graph_edges.tolist()) # target= torch.split(target, graph_edges.tolist()) # L_1=pred.size(1) pred[pred<0]=0 self.listcos+=simlarcalc(pred, target,self.args.ms2_method).cpu().numpy().tolist() # import ipdb # ipdb.set_trace() s=sum(self.listcos) #这里mask要改掉，因为超过序列部分的数据不能直接被mask掉，合并了 self.id_list+=_id.cpu().numpy().tolist() if math.isnan(s): ipdb.set_trace() print("getnan:{}".format(_id.cpu().numpy().tolist()[0])) self.nansequence.loc[self.nan] = [sequence.cpu().numpy().tolist(),charge.cpu().numpy().tolist(),decoration.cpu().numpy().tolist()] self.nan+=1 else: self.cos = s self.total += N self.bestcos = max(self.bestcos,max(self.listcos)) assert len(self.listcos)==self.total,"the length of cos list is different from the number of graphs" def get_metric(self, reset=True): r""" get_metric函数将根据evaluate函数累计的评价指标统计量来计算最终的评价结果. :param bool reset: 在调用完get_metric后是否清空评价指标统计量. :return dict evaluate_result: {"acc": float} """ # data=pd.DataFrame(self.listcos) # data.columns=["cs"] # import ipdb # ipdb.set_trace() mediancos=np.median(self.listcos) # if mediancos>self.bestanswer: # data.to_csv(self.savename+"Cossimilaritylist.csv",index=False) if self.nan>0: self.nansequence.to_json(self.savename+"nansequence.json") evaluate_result = {'mediancos':round(mediancos,6), 'nan number':self.nan, 'coss': round(float(self.cos) / (self.total + 1e-12), 6), 'bestcoss':round(self.bestcos, 6), "metric":self.args.ms2_method } if reset: self.cos = 0 self.total = 0 self.bestcos=0 self.listcos=[] return evaluate_result class Metric_byBY_outputmsms(MetricBase): def __init__(self,savename, pred=None, target=None, pred_by=None,pred_BY=None, target_by=None,target_BY=None,seq_len=None, num_col=None,sequence=None,charge=None,decoration=None, _id=None,peptide=None,PlausibleStruct=None, args=None): r""" :param pred: 参数映射表中 `pred` 的映射关系，None表示映射关系为 `pred` -> `pred` :param target: 参数映射表中 `target` 的映射关系，None表示映射关系为 `target` -> `target` :param seq_len: 参数映射表中 `seq_len` 的映射关系，None表示映射关系为 `seq_len` -> `seq_len` """ super().__init__() self._init_param_map(pred=pred, target=target, pred_by=pred_by,pred_BY=pred_BY,target_by=target_by,target_BY=target_BY, seq_len=seq_len,sequence=sequence, charge=charge,decoration=decoration,_id="_id",graph_edges="graph_edges", peptide=peptide,PlausibleStruct=PlausibleStruct) self.bestcos=0 self.total = 0 self.cos = 0 self.listcos=[] self.listcosBY=[] self.nan=0 self.bestanswer=0 self.nansequence=pd.DataFrame(columns=['nansequence','charge','decoration']) self.num_col=num_col self.savename=savename if savename else "" self.id_list=[] self.nj=0 self.repsequence=pd.DataFrame(columns=['repsequence',"idecoration",'charge', "ipeptide","iPlausibleStruct", 'ms2by',"ms2BY","metric","metricBY_cos","id"]) # self.repsequence=pd.DataFrame(columns=['repsequence','charge', # "ipeptide","iPlausibleStruct", # 'ms2by',"ms2BY","metric","id"]) self.args=args def evaluate(self, pred, target, pred_by=None,pred_BY=None, target_by=None,target_BY=None, seq_len=None,sequence=None,charge=None, decoration=None,_id=None,graph_edges=None, peptide=None,PlausibleStruct=None): r""" evaluate函数将针对一个批次的预测结果做评价指标的累计 :param torch.Tensor pred: 预测的tensor, tensor的形状可以是torch.Size([B,]), torch.Size([B, n_classes]), torch.Size([B, max_len]), 或者torch.Size([B, max_len, n_classes]) :param torch.Tensor target: 真实值的tensor, tensor的形状可以是Element's can be: torch.Size([B,]), torch.Size([B,]), torch.Size([B, max_len]), 或者torch.Size([B, max_len]) :param torch.Tensor seq_len: 序列长度标记, 标记的形状可以是None, None, torch.Size([B]), 或者torch.Size([B]). 如果mask也被传进来的话seq_len会被忽略. """ N=pred.size(0) #改输入格式 #simlarcalc:cos,corre,cos_sqrt,corre_sqrt # cos=CosineSimilarity(dim=0) #现在的cos就是这样的，也没有开根号，因为target和pred都拍平了，这里不用dim=1 #可以开根号，或者correlation coefficient # self.id_list+=_id.cpu().numpy().tolist() # pred = torch.split(pred, graph_edges.tolist()) # target= torch.split(target, graph_edges.tolist()) # L_1=pred.size(1) pred[pred<0]=0 pred_BY[pred_BY<0]=0 self.listcos+=simlarcalc(pred, target,self.args.ms2_method).cpu().numpy().tolist() # ipdb.set_trace() pred_BY = torch.split(pred_BY, graph_edges.tolist()) target_BY= torch.split(target_BY, graph_edges.tolist()) for c in range(len(graph_edges)): # self.listcos+=[cos(pred[c].flatten(), target[c].flatten()).cpu().numpy()] cos=CosineSimilarity(dim=0) self.listcosBY+=[cos(pred_BY[c].flatten(), target_BY[c].flatten()).cpu().numpy()] # print(self.listcosBY) # ipdb.set_trace() # ipdb.set_trace() # print("self.listcos",len(self.listcos)) # import ipdb # ipdb.set_trace() if any(math.isnan(x) for x in self.listcos): print('The list contains nan values.') # ipdb.set_trace() self.listcos = [0 if math.isnan(x) else x for x in self.listcos] print(self.listcos.count(0)) s=sum(self.listcos) #这里mask要改掉，因为超过序列部分的数据不能直接被mask掉，合并了 self.id_list+=_id.cpu().numpy().tolist() if math.isnan(s): # ipdb.set_trace() print("getnan:{}".format(_id.cpu().numpy().tolist()[0])) self.nansequence.loc[self.nan] = [sequence.cpu().numpy().tolist(),charge.cpu().numpy().tolist(),decoration.cpu().numpy().tolist()] self.nan+=1 else: self.cos = s self.total += N self.bestcos = max(self.bestcos,max(self.listcos)) assert len(self.listcos)==self.total,"the length of cos list is different from the number of graphs" masks = seq_len_to_mask(seq_len=(seq_len-1)*int(self.num_col)) pred_by=pred_by.masked_fill(masks.eq(False), 0) for i in range(len(graph_edges)): # import ipdb # ipdb.set_trace() il = seq_len[i] isequence=sequence[i][:il] icharge=charge[i] ipeptide=peptide[i] idecoration=decoration[i] iPlausibleStruct=PlausibleStruct[i] # import ipdb # ipdb.set_trace() ims2by=pred_by[i].reshape((-1,(self.num_col)))[:il-1,:(self.num_col)] ims2BY=pred_BY[i] icos=self.listcos[self.nj] icosBY=self.listcosBY[self.nj] self.repsequence.loc[self.nj] = [isequence.cpu().numpy().tolist(), idecoration.cpu().numpy().tolist(), icharge.cpu().numpy().tolist(), ipeptide.tolist(), iPlausibleStruct.tolist(), ims2by.cpu().numpy().tolist(), ims2BY.cpu().numpy().tolist(), icos, icosBY, self.id_list[self.nj]] # self.repsequence.loc[self.nj] = [isequence.cpu().numpy().tolist(), # icharge.cpu().numpy().tolist(), # ipeptide.tolist(), # iPlausibleStruct.tolist(), # ims2by.cpu().numpy().tolist(), # ims2BY.cpu().numpy().tolist(), # icos, # self.id_list[self.nj]] # import ipdb # ipdb.set_trace() # print(self.repsequence) self.nj+=1 def get_metric(self, reset=True): r""" get_metric函数将根据evaluate函数累计的评价指标统计量来计算最终的评价结果. :param bool reset: 在调用完get_metric后是否清空评价指标统计量. :return dict evaluate_result: {"acc": float} """ data=pd.DataFrame(self.listcos) data.columns=[self.args.ms2_method] # import ipdb # ipdb.set_trace() mediancos=np.median(self.listcos) if mediancos>self.bestanswer: data.to_csv(self.savename+"similaritylist.csv",index=False) print(f"file saved {self.savename} similaritylist.csv") if self.nan>0: self.nansequence.to_json(self.savename+"nansequence.json") evaluate_result = {'medianmetric':round(mediancos,6), 'nan number':self.nan, 'averagemetric': round(float(self.cos) / (self.total + 1e-12), 6), 'bestmetric':round(self.bestcos, 6), "metric":self.args.ms2_method } self.repsequence.to_csv(self.savename+"_byBY_result.csv",index=False) if reset: self.cos = 0 self.total = 0 self.bestcos=0 self.listcos=[] return evaluate_result # --------------------------- rt ---------------------# #rt的metric和loss是不是都要改一下，CS感觉不行 class CossimilarityMetricfortest_outputrt(MetricBase): r""" 准确率Metric（其它的Metric参见 :mod:`fastNLP.core.metrics` ） """ def __init__(self, savename,pred=None, target=None, seq_len=None,num_col=None,sequence=None,charge=None,decoration=None,_id=None): r""" :param pred: 参数映射表中 `pred` 的映射关系，None表示映射关系为 `pred` -> `pred` :param target: 参数映射表中 `target` 的映射关系，None表示映射关系为 `target` -> `target` :param seq_len: 参数映射表中 `seq_len` 的映射关系，None表示映射关系为 `seq_len` -> `seq_len` """ super().__init__() self._init_param_map(pred="predirt", target="irt", seq_len=seq_len,sequence=sequence,charge=charge, decoration=decoration,_id=_id) self.bestcos=0 self.total = 0 self.cos = 0 self.listcos=[] self.savename=savename if savename else "" self.repsequence=pd.DataFrame(columns=['repsequence','charge','decoration','rt',"rt_target","id"]) self.numcol=num_col self.id_list=[] def evaluate(self, pred, target, seq_len=None,sequence=None,charge=None,decoration=None,_id=None): r""" evaluate函数将针对一个批次的预测结果做评价指标的累计 :param torch.Tensor pred: 预测的tensor, tensor的形状可以是torch.Size([B,]), torch.Size([B, n_classes]), torch.Size([B, max_len]), 或者torch.Size([B, max_len, n_classes]) :param torch.Tensor target: 真实值的tensor, tensor的形状可以是Element's can be: torch.Size([B,]), torch.Size([B,]), torch.Size([B, max_len]), 或者torch.Size([B, max_len]) :param torch.Tensor seq_len: 序列长度标记, 标记的形状可以是None, None, torch.Size([B]), 或者torch.Size([B]). 如果mask也被传进来的话seq_len会被忽略. """ # ipdb.set_trace() if pred.dim() == 0: pred=pred.unsqueeze(0) N=pred.size(0) #batch size # N=pred.numel() # ipdb.set_trace() # if N!=128: # ipdb.set_trace() # import ipdb # pred L=sequence.shape[1] # if seq_len is not None and target.dim() > 1: # max_len = target.size(1) # masks = seq_len_to_mask(seq_len=(seq_len-1)*int(self.num_col)) # else: if rt_method=="cos": cos=CosineSimilarity(dim=0) s = torch.sum(cos(pred, target)).item() if rt_method=="delta": def delta(pred,target): return 1-torch.abs(pred-target) s = torch.sum(delta(pred,target)).item() if rt_method !="R2": self.cos += s self.total += N if rt_method =="cos": self.bestcos = max(self.bestcos, torch.max(cos(pred, target)).item()) if rt_method =="delta": self.bestcos = max(self.bestcos, torch.max(delta(pred,target)).item()) self.listcos += delta(pred,target).reshape(N, ).cpu().numpy().tolist() # ipdb.set_trace() # print(_id) # import ipdb # ipdb.set_trace() self.id_list+=_id.cpu().numpy().tolist() for i in range(N): il = seq_len[i]#il是序列真实长度 isequence=sequence[i][:il] icharge=charge[i] idecoration=decoration[i][:il] iirt=pred[i] iirt_target=target[i] self.repsequence.loc[len(self.repsequence)] = [isequence.cpu().numpy().tolist(), icharge.cpu().numpy().tolist(), idecoration.cpu().numpy().tolist(), iirt.cpu().numpy().tolist(), iirt_target.cpu().numpy().tolist(), self.id_list[i]] #不是seld.nj，是i def get_metric(self, reset=True): r""" get_metric函数将根据evaluate函数累计的评价指标统计量来计算最终的评价结果. :param bool reset: 在调用完get_metric后是否清空评价指标统计量. :return dict evaluate_result: {"acc": float} """ # data=pd.Series(self.listcos) # data.to_csv("Cossimilaritylist.csv",index=False) # self.repsequence.to_json(self.savename+"result.json") if rt_method =="cos": evaluate_result = {'mediancos':round(np.median(self.listcos),6), 'total number':len(self.repsequence), 'coss': round(float(self.cos) / (self.total + 1e-12), 6), 'bestcoss':round(self.bestcos, 6), } if rt_method =="delta": # ipdb.set_trace() evaluate_result = {'mediandelta':round(np.median(self.listcos),6), 'total number':len(self.repsequence), 'coss': round(float(self.cos) / (self.total + 1e-12), 6), 'bestcoss':round(self.bestcos, 6), } if rt_method=="R2": r2 = r2_score(self.repsequence['rt'], self.repsequence['rt_target']) evaluate_result = {'r2':round(r2,6), 'total number':len(self.repsequence) } if reset: r2=0 self.cos = 0 self.total = 0 self.cos = 0 self.listcos=[] self.id_list=[] self.repsequence=pd.DataFrame(columns=['repsequence','charge','decoration','rt',"rt_target","id"]) return evaluate_result class CossimilarityMetricfortest_predrt(MetricBase): r""" 准确率Metric（其它的Metric参见 :mod:`fastNLP.core.metrics` ） """ def __init__(self, savename,pred=None, target=None, seq_len=None,num_col=None,sequence=None,charge=None,decoration=None,_id=None): r""" :param pred: 参数映射表中 `pred` 的映射关系，None表示映射关系为 `pred` -> `pred` :param target: 参数映射表中 `target` 的映射关系，None表示映射关系为 `target` -> `target` :param seq_len: 参数映射表中 `seq_len` 的映射关系，None表示映射关系为 `seq_len` -> `seq_len` """ super().__init__() self._init_param_map(pred="predirt", target="irt", seq_len=seq_len,sequence=sequence,charge=charge, decoration=decoration,_id=_id) self.bestcos=0 self.total = 0 self.cos = 0 self.listcos=[] self.nj=0 self.savename=savename if savename else "" self.repsequence=pd.DataFrame(columns=['repsequence','charge','decoration','rt',"rt_target","id"]) self.numcol=num_col self.id_list=[] def evaluate(self, pred, target, seq_len=None,sequence=None,charge=None,decoration=None,_id=None): r""" evaluate函数将针对一个批次的预测结果做评价指标的累计 :param torch.Tensor pred: 预测的tensor, tensor的形状可以是torch.Size([B,]), torch.Size([B, n_classes]), torch.Size([B, max_len]), 或者torch.Size([B, max_len, n_classes]) :param torch.Tensor target: 真实值的tensor, tensor的形状可以是Element's can be: torch.Size([B,]), torch.Size([B,]), torch.Size([B, max_len]), 或者torch.Size([B, max_len]) :param torch.Tensor seq_len: 序列长度标记, 标记的形状可以是None, None, torch.Size([B]), 或者torch.Size([B]). 如果mask也被传进来的话seq_len会被忽略. """ N=pred.size(0) #batch size # import ipdb # ipdb.set_trace() L=sequence.shape[1] # if seq_len is not None and target.dim() > 1: # max_len = target.size(1) # masks = seq_len_to_mask(seq_len=(seq_len-1)*int(self.num_col)) # else: if rt_method=="cos": cos=CosineSimilarity(dim=0) s = torch.sum(cos(pred, target)).item() self.cos += s self.bestcos = max(self.bestcos, torch.max(cos(pred, target)).item()) if rt_method=="delta": def delta(pred,target): return 1-torch.abs(pred-target) s = torch.sum(delta(pred,target)).item() self.cos += s self.bestcos = max(self.bestcos, torch.max(delta(pred,target)).item()) self.listcos += delta(pred,target).reshape(N, ).cpu().numpy().tolist() self.total += pred.size(0) self.id_list+=_id.cpu().numpy().tolist() for i in range(N): il = seq_len[i]#il是序列真实长度 isequence=sequence[i][:il] icharge=charge[i] idecoration=decoration[i][:il] iirt=pred[i] iirt_target=target[i] self.repsequence.loc[len(self.repsequence)] = [isequence.cpu().numpy().tolist(), icharge.cpu().numpy().tolist(), idecoration.cpu().numpy().tolist(), iirt.cpu().numpy().tolist(), iirt_target.cpu().numpy().tolist(), self.id_list[len(self.repsequence)]] #不是seld.nj，是i def get_metric(self, reset=True): r""" get_metric函数将根据evaluate函数累计的评价指标统计量来计算最终的评价结果. :param bool reset: 在调用完get_metric后是否清空评价指标统计量. :return dict evaluate_result: {"acc": float} """ # data=pd.Series(self.listcos) # data.to_csv("Cossimilaritylist.csv",index=False) # import ipdb # ipdb.set_trace() repsequence_output=pd.DataFrame(self.repsequence) repsequence_output.to_csv(self.savename+"rtresult.csv") print(f"file saved {self.savename} rtresult.csv") if rt_method =="cos": evaluate_result = {'mediancos':round(np.median(self.listcos),6), 'total number':len(self.repsequence), 'coss': round(float(self.cos) / (self.total + 1e-12), 6), 'bestcoss':round(self.bestcos, 6), } if rt_method =="delta": # ipdb.set_trace() evaluate_result = {'mediandelta':round(np.median(self.listcos),6), 'total number':len(self.repsequence), 'coss': round(float(self.cos) / (self.total + 1e-12), 6), 'bestcoss':round(self.bestcos, 6), } if rt_method=="R2": r2 = r2_score(self.repsequence['rt'], self.repsequence['rt_target']) evaluate_result = {'r2':round(r2,6), 'total number':len(self.repsequence) } if reset: r2=0 self.cos = 0 self.total = 0 self.cos = 0 self.listcos=[] self.id_list=[] self.repsequence=pd.DataFrame(columns=['repsequence','charge','decoration','rt',"rt_target","id"]) return evaluate_result # --------------------------- not in use ---------------------# class PearsonCCMetric(MetricBase): r""" 准确率Metric（其它的Metric参见 :mod:`fastNLP.core.metrics` ） """ def __init__(self, pred=None, target=None): r""" :param pred: 参数映射表中 `pred` 的映射关系，None表示映射关系为 `pred` -> `pred` :param target: 参数映射表中 `target` 的映射关系，None表示映射关系为 `target` -> `target` :param seq_len: 参数映射表中 `seq_len` 的映射关系，None表示映射关系为 `seq_len` -> `seq_len` """ super().__init__() self._init_param_map(pred=pred, target=target) self.prelist=[] self.targetlist=[] def evaluate(self, pred, target, seq_len=None): r""" evaluate函数将针对一个批次的预测结果做评价指标的累计 :param torch.Tensor pred: 预测的tensor, tensor的形状可以是torch.Size([B,]), torch.Size([B, n_classes]), torch.Size([B, max_len]), 或者torch.Size([B, max_len, n_classes]) :param torch.Tensor target: 真实值的tensor, tensor的形状可以是Element's can be: torch.Size([B,]), torch.Size([B,]), torch.Size([B, max_len]), 或者torch.Size([B, max_len]) :param torch.Tensor seq_len: 序列长度标记, 标记的形状可以是None, None, torch.Size([B]), 或者torch.Size([B]). 如果mask也被传进来的话seq_len会被忽略. """ self.prelist+=pred.cpu().numpy().tolist() self.targetlist+=target.cpu().numpy().tolist() def get_metric(self, reset=True): r""" get_metric函数将根据evaluate函数累计的评价指标统计量来计算最终的评价结果. :param bool reset: 在调用完get_metric后是否清空评价指标统计量. :return dict evaluate_result: {"acc": float} """ cos=CosineSimilarity(dim=0) MAE=-np.mean(np.abs(np.array(self.prelist)-np.array(self.targetlist))) Pprelist=self.prelist-np.mean(self.prelist) Ptargetlist=self.targetlist-np.mean(self.targetlist) PCC=cos(torch.Tensor(Pprelist),torch.Tensor(Ptargetlist)) PCC=PCC.item() evaluate_result = {"meanl1loss":round(MAE,6), 'PCC':round(PCC,6), } if reset: self.prelist = [] self.targetlist=[] return evaluate_result class PearsonCCMetricfortest(MetricBase): r""" 准确率Metric（其它的Metric参见 :mod:`fastNLP.core.metrics` ） """ def __init__(self, pred=None, target=None): r""" :param pred: 参数映射表中 `pred` 的映射关系，None表示映射关系为 `pred` -> `pred` :param target: 参数映射表中 `target` 的映射关系，None表示映射关系为 `target` -> `target` :param seq_len: 参数映射表中 `seq_len` 的映射关系，None表示映射关系为 `seq_len` -> `seq_len` """ super().__init__() self._init_param_map(pred=pred, target=target) self.prelist=[] self.targetlist=[] def evaluate(self, pred, target, seq_len=None): r""" evaluate函数将针对一个批次的预测结果做评价指标的累计 :param torch.Tensor pred: 预测的tensor, tensor的形状可以是torch.Size([B,]), torch.Size([B, n_classes]), torch.Size([B, max_len]), 或者torch.Size([B, max_len, n_classes]) :param torch.Tensor target: 真实值的tensor, tensor的形状可以是Element's can be: torch.Size([B,]), torch.Size([B,]), torch.Size([B, max_len]), 或者torch.Size([B, max_len]) :param torch.Tensor seq_len: 序列长度标记, 标记的形状可以是None, None, torch.Size([B]), 或者torch.Size([B]). 如果mask也被传进来的话seq_len会被忽略. """ self.prelist+=pred.cpu().numpy().tolist() self.targetlist+=target.cpu().numpy().tolist() def get_metric(self, reset=True): r""" get_metric函数将根据evaluate函数累计的评价指标统计量来计算最终的评价结果. :param bool reset: 在调用完get_metric后是否清空评价指标统计量. :return dict evaluate_result: {"acc": float} """ cos=CosineSimilarity(dim=0) MAE=np.mean(np.abs(np.array(self.prelist)-np.array(self.targetlist))) Pprelist=self.prelist-np.mean(self.prelist) Ptargetlist=self.targetlist-np.mean(self.targetlist) PCC=cos(torch.Tensor(Pprelist),torch.Tensor(Ptargetlist)) PCC=PCC.item() outdata=pd.DataFrame(columns=["pred_irt","exp_irt"]) outdata["pred_irt"]=self.prelist outdata["exp_irt"]=self.targetlist outdata.to_csv("irt_pred_experiment.csv",index=False) evaluate_result = {"meanl1loss":round(MAE,6), 'PCC':round(PCC,6), } if reset: self.prelist = [] self.targetlist=[] return evaluate_result # --------------------------- position embedding---------------------# class PositionEmbedding(nn.Module):#input:sequence:N*L(N:batch) def __init__(self,emb_size,maxlength): super().__init__() pe=torch.arange(0,maxlength) pe.requires_grad=False pe = pe.unsqueeze(0) self.embedding=nn.Embedding(maxlength,emb_size) self.register_buffer('pe', pe)#1LE def forward(self,x,device): pe=self.embedding(self.pe[:,:x.size(1)]) return pe import math class PositionalEmbedding(nn.Module): def __init__(self, d_model, max_len=512): super().__init__() # Compute the positional encodings once in log space. pe = torch.zeros(max_len, d_model).float() pe.require_grad = False position = torch.arange(0, max_len).float().unsqueeze(1) div_term = (torch.arange(0, d_model, 2).float() * -(math.log(10000.0) / d_model)).exp() pe[:, 0::2] = torch.sin(position * div_term) pe[:, 1::2] = torch.cos(position * div_term) pe = pe.unsqueeze(0) self.register_buffer('pe', pe) def forward(self, x): return self.pe[:, :x.size(1)] # --------------------------- model--------------------# class deepdiaModelms2(nn.Module):#input:sequence:N*L(N:batch) def __init__(self,maxlength,acid_size,embed_dim,nhead,num_layers,dropout=0.2,num_col=12): super().__init__() self.num_col=int(num_col) self.edim=embed_dim self.conv=nn.Conv1d(embed_dim,embed_dim,2) self.pos_embedding=nn.Embedding(maxlength,embed_dim) self.a_embedding=nn.Embedding(acid_size,embed_dim,padding_idx=0) self.phos_embedding=nn.Embedding(2,embed_dim)#只有两种磷酸化情况 encoder_layer=nn.TransformerEncoderLayer(embed_dim,nhead,dropout=dropout) self.transformer=nn.TransformerEncoder(encoder_layer,num_layers) self.rtlinear=nn.Linear(embed_dim,1) self.activation=nn.ReLU() self.mslinear=nn.Linear(embed_dim,num_col) self.dropout=nn.Dropout(p=dropout) def forward(self,peptide_tokens,peptide_length,phos=None): #input:sequence:N*L(N:batch) # lengths:N*L(N:batch) # phos:N*L(N:batch) N=peptide_tokens.size(0) L=peptide_tokens.size(1) device=peptide_tokens.device ll=torch.LongTensor(range(L)) slengths=ll.expand(N,L) slengths=slengths.to(device) sequence=peptide_tokens assert N==peptide_length.size(0) a_embed=self.a_embedding(sequence)#NLE pos_embed=self.pos_embedding(slengths)#NLE if phos: assert sequence.size(0) == phos.size(0) phos_embed = self.phos_embedding(phos) ninput=pos_embed+a_embed+phos_embed#NLE else: ninput = pos_embed + a_embed #NLE key_padding_mask=attentionmask(peptide_length-1) ninput=self.activation(self.conv(ninput.permute(0,2,1)))#NE(L-1) output =self.transformer(ninput.permute(2,0,1),src_key_padding_mask=key_padding_mask)#(L-1)NE outputmean=output.mean(dim=0)#N*E # outputrt=self.activation(self.rtlinear(outputmean))#N*1 outputms=self.dropout(self.mslinear(output))#(L-1)*N*12 outputms=self.activation(outputms) outputms=outputms.permute(1,0,2).reshape(N,-1)#N*((L-1)*12) masks = seq_len_to_mask(seq_len=(peptide_length - 1) * self.num_col)##加上mask outputms=outputms.masked_fill(masks.eq(False), 0) # print(torch.sum(outputms)) return {'pred':outputms} class _2deepdiaModelms2(nn.Module):#input:sequence:N*L(N:batch) def __init__(self,maxlength,acid_size,embed_dim,nhead,num_layers,dropout=0.2,num_col=12): super().__init__() self.edim=embed_dim self.num_col=int(num_col) self.conv=nn.Conv1d(embed_dim,embed_dim,2) self.pos_embedding=PositionEmbedding(embed_dim,maxlength) self.a_embedding=nn.Embedding(acid_size,embed_dim,padding_idx=0) self.phos_embedding=nn.Embedding(5,embed_dim)#修饰三种加上padding###完整版4种修饰 encoder_layer=nn.TransformerEncoderLayer(embed_dim,nhead,dropout=dropout) self.transformer=nn.TransformerEncoder(encoder_layer,num_layers) self.rtlinear=nn.Linear(embed_dim,1) self.activation=nn.ReLU() self.mslinear=nn.Linear(embed_dim,num_col) self.dropout=nn.Dropout(p=dropout) def forward(self,peptide_tokens,peptide_length,decoration=None): #input:sequence:N*L(N:batch) # lengths:N*L(N:batch) # phos:N*L(N:batch) N=peptide_tokens.size(0) L=peptide_tokens.size(1) sequence=peptide_tokens device=peptide_tokens.device assert N==peptide_length.size(0) a_embed=self.a_embedding(sequence)#NLE pos_embed=self.pos_embedding(peptide_tokens,device)#NLE assert sequence.size(0) == decoration.size(0) phos_embed = self.phos_embedding(decoration) ninput=pos_embed+a_embed+phos_embed#NLE key_padding_mask=attentionmask(peptide_length) output =self.transformer(ninput.permute(1,0,2),src_key_padding_mask=key_padding_mask)#(L)NE outputmean=output.mean(dim=0)#N*E # outputrt=self.activation(self.rtlinear(outputmean))#N*1 output=self.activation(self.conv(output.permute(1,2,0)))# NE(L-1) output=output.permute(0,2,1)#N*(L-1)*E outputms=self.dropout(self.mslinear(output))#N*(L-1)*24 outputms=self.activation(outputms) outputms=outputms.reshape(N,-1)#N*((L-1)*24) masks = seq_len_to_mask(seq_len=(peptide_length - 1) * self.num_col)##加上mask outputms=outputms.masked_fill(masks.eq(False), 0) # print(torch.sum(outputms)) return {'pred':outputms} # --------------------------- charge embedding --------------------# class _2deepchargeModelms2(nn.Module):#input:sequence:N*L(N:batch) def __init__(self,maxlength,acid_size,embed_dim,nhead,num_layers,dropout=0.2,num_col=12): super().__init__() self.edim=embed_dim self.num_col=int(num_col) self.conv=nn.Conv1d(embed_dim,embed_dim,2) self.pos_embedding=nn.Embedding(maxlength,embed_dim) self.charge_embedding=nn.Embedding(10,embed_dim,padding_idx=0) self.a_embedding=nn.Embedding(acid_size,embed_dim,padding_idx=0) self.phos_embedding=nn.Embedding(4,embed_dim)#修饰三种加上padding###完整版4种修饰 encoder_layer=nn.TransformerEncoderLayer(embed_dim,nhead,dropout=dropout) self.transformer=nn.TransformerEncoder(encoder_layer,num_layers) self.rtlinear=nn.Linear(embed_dim,1) self.activation=nn.ReLU() self.mslinear=nn.Linear(embed_dim,num_col) self.dropout=nn.Dropout(p=dropout) def forward(self,peptide_tokens,peptide_length,charge,decoration=None): #input:sequence:N*L(N:batch) # lengths:N*L(N:batch) # phos:N*L(N:batch) #charge:N*1 N=peptide_tokens.size(0) L=peptide_tokens.size(1) sequence=peptide_tokens device=peptide_length.device assert N==peptide_length.size(0) ll = torch.arange(0, L, device=device).unsqueeze(0)#1*L a_embed=self.a_embedding(sequence)#NLE pos_embed=self.pos_embedding(ll)#1LE charge_embed=self.charge_embedding(charge.unsqueeze(1).expand(N,L)) assert sequence.size(0) == decoration.size(0) phos_embed = self.phos_embedding(decoration) ninput=pos_embed+a_embed+phos_embed+charge_embed#NLE # ninput=self.dropout(ninput) key_padding_mask=attentionmask(peptide_length) output =self.transformer(ninput.permute(1,0,2),src_key_padding_mask=key_padding_mask)#(L)NE # outputmean=output.mean(dim=0)#N*E # outputrt=self.activation(self.rtlinear(outputmean))#N*1 output=self.activation(self.conv(output.permute(1,2,0)))# NE(L-1) output=output.permute(0,2,1)#N*(L-1)*E outputms=self.dropout(self.mslinear(output))#N*(L-1)*24 outputms=self.activation(outputms) outputms=outputms.reshape(N,-1)#N*((L-1)*24) masks = seq_len_to_mask(seq_len=(peptide_length - 1) * self.num_col)##加上mask outputms=outputms.masked_fill(masks.eq(False), 0) # print(outputms) # print(torch.sum(outputms)) return {'pred':outputms,'sequence':sequence,'charge':charge,"decoration":decoration,"seq_len":peptide_length} # --------------------------- irt --------------------# class _2deepchargeModelirt_ll(nn.Module):#input:sequence:N*L(N:batch)###不使用charge def __init__(self,maxlength,acid_size,embed_dim,nhead,num_layers,dropout=0.2,num_col=12): super().__init__() self.edim=embed_dim self.num_col=int(num_col) self.conv=nn.Conv1d(embed_dim,embed_dim,2) self.pos_embedding=nn.Embedding(maxlength,embed_dim) self.charge_embedding=nn.Embedding(10,embed_dim,padding_idx=0) self.a_embedding=nn.Embedding(acid_size,embed_dim,padding_idx=0) self.phos_embedding=nn.Embedding(5,embed_dim)#修饰三种加上padding###完整版4种修饰 encoder_layer=nn.TransformerEncoderLayer(embed_dim,nhead,dropout=dropout) self.transformer=nn.TransformerEncoder(encoder_layer,num_layers) self.rtlinear1=nn.Linear(embed_dim,256) self.rtlinear2=nn.Linear(256,1) self.activation=nn.ReLU() self.mslinear=nn.Linear(embed_dim,num_col) self.dropout=nn.Dropout(p=dropout) def forward(self,peptide_tokens,peptide_length,charge,decoration=None): #input:sequence:N*L(N:batch) # lengths:N*L(N:batch) # phos:N*L(N:batch) #charge:N*1 #sequence前面加入[CLS]的embedding N=peptide_tokens.size(0) L=peptide_tokens.size(1) device=peptide_tokens.device ll=torch.arange(0,L,device=device).unsqueeze(0) sequence=peptide_tokens assert N==peptide_length.size(0) a_embed=self.a_embedding(sequence)#NLE pos_embed=self.pos_embedding(ll)#1LE assert sequence.size(0) == decoration.size(0) phos_embed = self.phos_embedding(decoration) ninput=pos_embed+a_embed+phos_embed#NLE # ninput=self.dropout(ninput) key_padding_mask=attentionmask(peptide_length)# output =self.transformer(ninput.permute(1,0,2),src_key_padding_mask=key_padding_mask)#(L)NE output=torch.max(output,dim=0).values#maxpooling #N*E output = self.activation(self.rtlinear1(output)) outputrt=self.activation(self.rtlinear2(output).squeeze())#N*1 # outputrt=outputrt # print(torch.sum(outputms)) return {'pred':outputrt} # --------------------------- sigmoid --------------------# class _2deepchargeModelirt_ll_sigmoid(nn.Module):#input:sequence:N*L(N:batch)###不使用charge def __init__(self,maxlength,acid_size,embed_dim,nhead,num_layers,dropout=0.2,num_col=12): super().__init__() self.edim=embed_dim self.num_col=int(num_col) self.conv=nn.Conv1d(embed_dim,embed_dim,2) self.pos_embedding=nn.Embedding(maxlength,embed_dim) self.charge_embedding=nn.Embedding(10,embed_dim,padding_idx=0) self.a_embedding=nn.Embedding(acid_size,embed_dim,padding_idx=0) self.phos_embedding=nn.Embedding(5,embed_dim)#修饰三种加上padding###完整版4种修饰 encoder_layer=nn.TransformerEncoderLayer(embed_dim,nhead,dropout=dropout) self.transformer=nn.TransformerEncoder(encoder_layer,num_layers) self.rtlinear1=nn.Linear(embed_dim,256) self.rtlinear2=nn.Linear(256,1) self.activation=nn.ReLU() self.mslinear=nn.Linear(embed_dim,num_col) self.dropout=nn.Dropout(p=dropout) def forward(self,peptide_tokens,peptide_length,charge,decoration=None): #input:sequence:N*L(N:batch) # lengths:N*L(N:batch) # phos:N*L(N:batch) #charge:N*1 #sequence前面加入[CLS]的embedding N=peptide_tokens.size(0) L=peptide_tokens.size(1) device=peptide_tokens.device ll=torch.arange(0,L,device=device).unsqueeze(0) sequence=peptide_tokens assert N==peptide_length.size(0) a_embed=self.a_embedding(sequence)#NLE pos_embed=self.pos_embedding(ll)#1LE assert sequence.size(0) == decoration.size(0) phos_embed = self.phos_embedding(decoration) ninput=pos_embed+a_embed+phos_embed#NLE # ninput=self.dropout(ninput) key_padding_mask=attentionmask(peptide_length)# output =self.transformer(ninput.permute(1,0,2),src_key_padding_mask=key_padding_mask)#(L)NE output=torch.max(output,dim=0).values#maxpooling #N*E output = self.activation(self.rtlinear1(output)) outputrt=torch.sigmoid(self.rtlinear2(output).squeeze())#N*1 # outputrt=outputrt # print(torch.sum(outputms)) return {'pred':outputrt} # --------------------------- clsembedding --------------------# class _2deepchargeModelirt_cls(nn.Module):#input:sequence:N*L(N:batch) def __init__(self,maxlength,acid_size,embed_dim,nhead,num_layers,dropout=0.2,num_col=12): super().__init__() self.edim=embed_dim self.num_col=int(num_col) self.conv=nn.Conv1d(embed_dim,embed_dim,2) self.pos_embedding=nn.Embedding(maxlength,embed_dim) self.cls_embedding=nn.Embedding(2,embed_dim)## self.charge_embedding=nn.Embedding(10,embed_dim,padding_idx=0) self.a_embedding=nn.Embedding(acid_size,embed_dim,padding_idx=0) self.phos_embedding=nn.Embedding(5,embed_dim)#修饰三种加上padding###完整版4种修饰 encoder_layer=nn.TransformerEncoderLayer(embed_dim,nhead,dropout=dropout) self.transformer=nn.TransformerEncoder(encoder_layer,num_layers) self.rtlinear=nn.Linear(embed_dim,1) self.activation=nn.ReLU() self.mslinear=nn.Linear(embed_dim,num_col) self.dropout=nn.Dropout(p=dropout) def forward(self,peptide_tokens,peptide_length,charge,decoration=None): #input:sequence:N*L(N:batch) # lengths:N*L(N:batch) # phos:N*L(N:batch) #charge:N*1 #sequence前面加入[CLS]的embedding N=peptide_tokens.size(0) L=peptide_tokens.size(1) device=peptide_tokens.device ll=torch.arange(0,L+1,device=device).unsqueeze(0)##L+1 sequence=peptide_tokens assert N==peptide_length.size(0) a_embed=self.a_embedding(sequence)#NLE cls_embed=self.cls_embedding(torch.ones(N,device=device,dtype=int)).unsqueeze(1)#N*1*E pos_embed=self.pos_embedding(ll)#1(L+1)E assert sequence.size(0) == decoration.size(0) phos_embed = self.phos_embedding(decoration)#NLE ninput=a_embed+phos_embed#NLE ninput=torch.cat([cls_embed,ninput],dim=1)+pos_embed#N(L+1)E # ninput=self.dropout(ninput) key_padding_mask=attentionmask(peptide_length+1)# output =self.transformer(ninput.permute(1,0,2),src_key_padding_mask=key_padding_mask)#(L+1)NE output=output[0].squeeze()#maxpooling #N*E outputrt=self.rtlinear(output).squeeze()#N*1 # outputrt=outputrt # print(torch.sum(outputms)) return {'pred':outputrt} ####################### class _2deepchargeModelirt(nn.Module):#input:sequence:N*L(N:batch)###不使用charge def __init__(self,maxlength,acid_size,embed_dim,nhead,num_layers,dropout=0.2,num_col=12): super().__init__() self.edim=embed_dim self.num_col=int(num_col) self.conv=nn.Conv1d(embed_dim,embed_dim,2) self.pos_embedding=nn.Embedding(maxlength,embed_dim) self.charge_embedding=nn.Embedding(10,embed_dim,padding_idx=0) self.a_embedding=nn.Embedding(acid_size,embed_dim,padding_idx=0) self.phos_embedding=nn.Embedding(5,embed_dim)#修饰三种加上padding###完整版4种修饰 encoder_layer=nn.TransformerEncoderLayer(embed_dim,nhead,dropout=dropout) self.transformer=nn.TransformerEncoder(encoder_layer,num_layers) self.rtlinear=nn.Linear(embed_dim,1) self.activation=nn.ReLU() self.mslinear=nn.Linear(embed_dim,num_col) self.dropout=nn.Dropout(p=dropout) def forward(self,peptide_tokens,peptide_length,charge,decoration=None): #input:sequence:N*L(N:batch) # lengths:N*L(N:batch) # phos:N*L(N:batch) #charge:N*1 #sequence前面加入[CLS]的embedding N=peptide_tokens.size(0) L=peptide_tokens.size(1) device=peptide_tokens.device ll=torch.arange(0,L,device=device).unsqueeze(0) sequence=peptide_tokens assert N==peptide_length.size(0) a_embed=self.a_embedding(sequence)#NLE pos_embed=self.pos_embedding(ll)#1LE assert sequence.size(0) == decoration.size(0) phos_embed = self.phos_embedding(decoration) ninput=pos_embed+a_embed+phos_embed#NLE # ninput=self.dropout(ninput) key_padding_mask=attentionmask(peptide_length)# output =self.transformer(ninput.permute(1,0,2),src_key_padding_mask=key_padding_mask)#(L)NE output=torch.max(output,dim=0).values#maxpooling #N*E outputrt=self.rtlinear(output).squeeze()#N*1 # outputrt=outputrt # print(torch.sum(outputms)) return {'pred':outputrt} ##################################################################取0位置linear class _2deepchargeModelirt_zero(nn.Module):#input:sequence:N*L(N:batch)###不使用charge def __init__(self,maxlength,acid_size,embed_dim,nhead,num_layers,dropout=0.2,num_col=12): super().__init__() self.edim=embed_dim self.num_col=int(num_col) self.conv=nn.Conv1d(embed_dim,embed_dim,2) self.pos_embedding=nn.Embedding(maxlength,embed_dim) self.charge_embedding=nn.Embedding(10,embed_dim,padding_idx=0) self.a_embedding=nn.Embedding(acid_size,embed_dim,padding_idx=0) self.phos_embedding=nn.Embedding(4,embed_dim)#修饰三种加上padding###完整版4种修饰 encoder_layer=nn.TransformerEncoderLayer(embed_dim,nhead,dropout=dropout) self.transformer=nn.TransformerEncoder(encoder_layer,num_layers) self.rtlinear=nn.Linear(embed_dim,1) self.activation=nn.ReLU() self.mslinear=nn.Linear(embed_dim,num_col) self.dropout=nn.Dropout(p=dropout) def forward(self,peptide_tokens,peptide_length,charge,decoration=None): #input:sequence:N*L(N:batch) # lengths:N*L(N:batch) # phos:N*L(N:batch) #charge:N*1 #sequence前面加入[CLS]的embedding N=peptide_tokens.size(0) L=peptide_tokens.size(1) device=peptide_tokens.device ll=torch.arange(0,L,device=device).unsqueeze(0) sequence=peptide_tokens assert N==peptide_length.size(0) a_embed=self.a_embedding(sequence)#NLE pos_embed=self.pos_embedding(ll)#1LE assert sequence.size(0) == decoration.size(0) phos_embed = self.phos_embedding(decoration) ninput=pos_embed+a_embed+phos_embed#NLE # ninput=self.dropout(ninput) key_padding_mask=attentionmask(peptide_length)# output =self.transformer(ninput.permute(1,0,2),src_key_padding_mask=key_padding_mask)#(L)NE output=output[0].squeeze()#maxpooling #N*E outputrt=self.rtlinear(output).squeeze()#N*1 # outputrt=outputrt # print(torch.sum(outputms)) return {'pred':outputrt,'sequence':sequence,'charge':charge,"decoration":decoration,"seq_len":peptide_length} ######完整版模型############################# class _2deepchargeModelirt_zero_all(nn.Module):#input:sequence:N*L(N:batch)###不使用charge def __init__(self,maxlength,acid_size,embed_dim,nhead,num_layers,dropout=0.2,num_col=12): super().__init__() self.edim=embed_dim self.num_col=int(num_col) self.conv=nn.Conv1d(embed_dim,embed_dim,2) self.pos_embedding=nn.Embedding(maxlength,embed_dim) self.charge_embedding=nn.Embedding(10,embed_dim,padding_idx=0) self.a_embedding=nn.Embedding(acid_size,embed_dim,padding_idx=0) self.phos_embedding=nn.Embedding(5,embed_dim)#修饰三种加上padding###完整版4种修饰 encoder_layer=nn.TransformerEncoderLayer(embed_dim,nhead,dropout=dropout) self.transformer=nn.TransformerEncoder(encoder_layer,num_layers) self.rtlinear=nn.Linear(embed_dim,1) self.activation=nn.ReLU() self.mslinear=nn.Linear(embed_dim,num_col) self.dropout=nn.Dropout(p=dropout) def forward(self,peptide_tokens,peptide_length,charge,decoration=None): #input:sequence:N*L(N:batch) # lengths:N*L(N:batch) # phos:N*L(N:batch) #charge:N*1 #sequence前面加入[CLS]的embedding N=peptide_tokens.size(0) L=peptide_tokens.size(1) device=peptide_tokens.device ll=torch.arange(0,L,device=device).unsqueeze(0) sequence=peptide_tokens assert N==peptide_length.size(0) a_embed=self.a_embedding(sequence)#NLE pos_embed=self.pos_embedding(ll)#1LE assert sequence.size(0) == decoration.size(0) phos_embed = self.phos_embedding(decoration) ninput=pos_embed+a_embed+phos_embed#NLE # ninput=self.dropout(ninput) key_padding_mask=attentionmask(peptide_length)# output =self.transformer(ninput.permute(1,0,2),src_key_padding_mask=key_padding_mask)#(L)NE output=output[0].squeeze()#maxpooling #N*E outputrt=self.rtlinear(output).squeeze()#N*1 # outputrt=outputrt # print(torch.sum(outputms)) return {'pred':outputrt,'sequence':sequence,'charge':charge,"decoration":decoration,"seq_len":peptide_length} class _2deepchargeModelms2_all(nn.Module):#input:sequence:N*L(N:batch) def __init__(self,maxlength,acid_size,embed_dim,nhead,num_layers,dropout=0.2,num_col=12): super().__init__() self.edim=embed_dim self.num_col=int(num_col) self.conv=nn.Conv1d(embed_dim,embed_dim,2) self.pos_embedding=nn.Embedding(maxlength,embed_dim) self.charge_embedding=nn.Embedding(10,embed_dim,padding_idx=0) self.a_embedding=nn.Embedding(acid_size,embed_dim,padding_idx=0) self.phos_embedding=nn.Embedding(5,embed_dim)#修饰三种加上padding###完整版4种修饰 encoder_layer=nn.TransformerEncoderLayer(embed_dim,nhead,dropout=dropout) self.transformer=nn.TransformerEncoder(encoder_layer,num_layers) self.rtlinear=nn.Linear(embed_dim,1) self.activation=nn.ReLU() self.mslinear=nn.Linear(embed_dim,num_col) self.dropout=nn.Dropout(p=dropout) def forward(self,peptide_tokens,peptide_length,charge,decoration,pnumber): #input:sequence:N*L(N:batch) # lengths:N*L(N:batch) # phos:N*L(N:batch) #charge:N*1 N=peptide_tokens.size(0) L=peptide_tokens.size(1) sequence=peptide_tokens device=peptide_length.device assert N==peptide_length.size(0) ll = torch.arange(0, L, device=device).unsqueeze(0)#1*L a_embed=self.a_embedding(sequence)#NLE pos_embed=self.pos_embedding(ll)#1LE charge_embed=self.charge_embedding(charge.unsqueeze(1).expand(N,L)) assert sequence.size(0) == decoration.size(0) phos_embed = self.phos_embedding(decoration) ninput=pos_embed+a_embed+phos_embed+charge_embed#NLE # ninput=self.dropout(ninput) key_padding_mask=attentionmask(peptide_length) output =self.transformer(ninput.permute(1,0,2),src_key_padding_mask=key_padding_mask)#(L)NE # outputmean=output.mean(dim=0)#N*E # outputrt=self.activation(self.rtlinear(outputmean))#N*1 output=self.activation(self.conv(output.permute(1,2,0)))# NE(L-1) output=output.permute(0,2,1)#N*(L-1)*E outputms=self.dropout(self.mslinear(output))#N*(L-1)*24 outputms=self.activation(outputms) outputms=outputms.reshape(N,-1)#N*((L-1)*num_col) masks = seq_len_to_mask(seq_len=(peptide_length - 1) * self.num_col)##加上mask outputms=outputms.masked_fill(masks.eq(False), 0) # print(outputms) # print(torch.sum(outputms)) return {'pred':outputms,'sequence':sequence,'charge':charge,"decoration":decoration,"seq_len":peptide_length, 'pnumber':pnumber} ########################contrast training for decoy class _2deepchargeModelms2_all_contrast(nn.Module):#input:sequence:N*L(N:batch) def __init__(self,maxlength,acid_size,embed_dim,nhead,num_layers,dropout=0.2,num_col=12): super().__init__() self.edim=embed_dim self.num_col=int(num_col) self.conv=nn.Conv1d(embed_dim,embed_dim,2) self.pos_embedding=nn.Embedding(maxlength,embed_dim) self.charge_embedding=nn.Embedding(10,embed_dim,padding_idx=0) self.a_embedding=nn.Embedding(acid_size,embed_dim,padding_idx=0) self.phos_embedding=nn.Embedding(5,embed_dim)#修饰三种加上padding###完整版4种修饰 encoder_layer=nn.TransformerEncoderLayer(embed_dim,nhead,dropout=dropout) self.transformer=nn.TransformerEncoder(encoder_layer,num_layers) self.rtlinear=nn.Linear(embed_dim,1) self.activation=nn.ReLU() self.mslinear=nn.Linear(embed_dim,num_col) self.dropout=nn.Dropout(p=dropout) def forward(self,peptide_tokens,peptide_length,charge,decoration,pnumber,false_samples=None): #input:sequence:N*L(N:batch) # lengths:N*L(N:batch) # phos:N*L(N:batch) #charge:N*1 #false_samples:N*F(number of decoy of every sample)*L 这里面只存了phos的数据. N=peptide_tokens.size(0) L=peptide_tokens.size(1) sequence=peptide_tokens device=peptide_length.device assert N==peptide_length.size(0) ll = torch.arange(0, L, device=device).unsqueeze(0)#1*L a_embed=self.a_embedding(sequence)#NLE pos_embed=self.pos_embedding(ll)#1LE charge_embed=self.charge_embedding(charge.unsqueeze(1).expand(N,L)) assert sequence.size(0) == decoration.size(0) phos_embed = self.phos_embedding(decoration) ninput=pos_embed+a_embed+phos_embed+charge_embed#NLE # ninput=self.dropout(ninput) key_padding_mask=attentionmask(peptide_length) output =self.transformer(ninput.permute(1,0,2),src_key_padding_mask=key_padding_mask)#(L)NE # outputmean=output.mean(dim=0)#N*E # outputrt=self.activation(self.rtlinear(outputmean))#N*1 output=self.activation(self.conv(output.permute(1,2,0)))# NE(L-1) output=output.permute(0,2,1)#N*(L-1)*E outputms=self.dropout(self.mslinear(output))#N*(L-1)*24 outputms=self.activation(outputms) outputms=outputms.reshape(N,-1)#N*((L-1)*num_col) masks = seq_len_to_mask(seq_len=(peptide_length - 1) * self.num_col)##加上mask outputms=outputms.masked_fill(masks.eq(False), 0) if false_samples is None: return {'pred': outputms, 'sequence': sequence, 'charge': charge, "decoration": decoration, "seq_len": peptide_length, 'pnumber': pnumber} # print(outputms) # print(torch.sum(outputms)) else: false_phos_embed = self.phos_embedding(false_samples) false_ninput = pos_embed + a_embed + charge_embed ##NLE false_ninput=false_ninput.unsqueeze(1) # import ipdb # ipdb.set_trace() false_ninput = false_ninput + false_phos_embed F = false_ninput.size(1) false_ninput = false_ninput.reshape(N * F, L, -1) false_peplen = peptide_length.expand(F, N).T.reshape(N * F) false_key_padding_mask = attentionmask(false_peplen) false_output = self.transformer(false_ninput.permute(1, 0, 2), src_key_padding_mask=false_key_padding_mask) # (L)(N*F)E false_output = self.activation(self.conv(false_output.permute(1, 2, 0))) # (N*F)E(L-1) false_output = false_output.permute(0, 2, 1) # (N*F)*(L-1)*E false_outputms = self.dropout(self.mslinear(false_output)) # (N*F)*(L-1)*24 false_outputms = self.activation(false_outputms) false_outputms=false_outputms.reshape(N*F,-1) # (N*F)*((L-1)*num_col) false_seq_len = ((peptide_length - 1) * self.num_col).expand(F, N).T.reshape(N * F) false_masks = seq_len_to_mask(seq_len=false_seq_len) ##加上mask false_outputms = false_outputms.masked_fill(false_masks.eq(False), 0) false_outputms = false_outputms.reshape(N, F, -1) # N*F*((L-1)*num_col) return {'pred': outputms, 'sequence': sequence, 'charge': charge, "decoration": decoration, "seq_len": peptide_length, 'pnumber': pnumber, "false_outputms": false_outputms}