import os import sys import copy sys.path.append(os.path.dirname(__file__)) sys.path.append(os.path.dirname(os.path.dirname(__file__))) import json import torch import numpy as np import pandas as pd from torch import nn from Bio import SeqIO from Bio.SeqRecord import SeqRecord from torch.utils.data import DataLoader, Dataset ,random_split,IterableDataset from sklearn.model_selection import KFold,train_test_split from bucket_sampler import Bucket_Sampler from utils import Seq_one_hot global script_dir global data_dir with open(os.path.join(os.path.dirname(os.path.dirname(__file__)),"machine_configure.json"),'r') as f: config = json.load(f) script_dir = config['script_dir'] data_dir = config['data_dir'] def one_hot(seq,complementary=False): """ one_hot encoding on sequence complementary: encode nucleatide into complementary one """ # setting seq = list(seq.replace("U","T")) seq_len = len(seq) complementary = -1 if complementary else 1 # compose dict keys = ['A', 'C', 'G', 'T'][::complementary] oh_dict = {keys[i]:i for i in range(4)} # array oh_array = np.zeros((seq_len,4)) for i,C in enumerate(seq): try: oh_array[i,oh_dict[C]]=1 except: continue # for nucleotide that are not in A C G T    return oh_array def read_bpseq(test_bpseq_path): """ read bpseq file, extract sequence and ptable """ with open(test_bpseq_path,'r') as f: test_bpseq = f.readlines() f.close() for i in range(8): if test_bpseq[i].startswith('1'): start_index = i bp_seq = test_bpseq[start_index:] seq = ''.join([line.strip().split(" ")[1] for line in bp_seq]) ptable = [int(line.strip().split(" ")[2]) for line in bp_seq] assert len(ptable) == int(bp_seq[-1].split(" ")[0]) return seq,ptable def pad_zeros(X,pad_to): """ zero padding at the right end of the sequence """ if pad_to == 0: # pad_to = 8*(X.shape[0]//8 + 1) + 1 try: seq_len = X.shape[0] except: seq_len = X[0].shape[0] pad_to = 3*(seq_len//3 + 1) + 1 seq_len = X.shape[0] if isinstance(X, np.ndarray) else X[0].shape[0] gap = pad_to - seq_len # here we change to padding ahead , previously nn.ZeroPad2d([0,0,0,gap]) pad_fn = nn.ZeroPad2d([0,0,gap,0]) # (padding_left , padding_right , padding_top , padding_bottom ) # gap_array = np.zeros() if isinstance(X, list): # additional input X_padded = [pad_fn(torch.tensor(X[0])), X[1]] elif isinstance(X, np.ndarray): X_padded = pad_fn(torch.tensor(X)) return X_padded def pack_seq(ds_zls:list,pad_to:int): X_ts = [X for X,Y in ds_zls] if pad_to == 0: try: max_len = np.max([X.shape[0] for X in X_ts]) except: max_len = np.max([X[0].shape[0] for X in X_ts]) pad_to = 3*(max_len//3 + 1) + 1 if isinstance(X_ts[0],np.ndarray): X_packed = torch.stack([pad_zeros(X=x,pad_to=pad_to) for x in X_ts]) else: X_packed = [torch.stack([pad_zeros(X=x[0],pad_to=pad_to) for x in X_ts]), torch.tensor([x[1] for x in X_ts])] Y_packed = torch.tensor(np.array([Y for X,Y in ds_zls])) return X_packed , Y_packed class mask_reader(Dataset): def __init__(self,npy_path): """ read the mask A549 sequence and with real sequence """ self.data_set = np.load(npy_path) self.X = self.data_set[:,0] self.Y = self.data_set[:,1] def __len__(self): return self.X.shape[0] def __getitem__(self,index): return (self.X[index,:,:],self.Y[index,:,:]) class mix_dataset(Dataset): def __init__(self): # dataset_ls = [mask_reader(os.path.join(data_dir,'mix_data',"mix_%s.npy"%set)) for set in ['train','val','test']] raise ValueError("the dataset is not defined") class mask_dataset(Dataset): def __init__(self): raise ValueError("the dataset is not defined") class ribo_dataset(Dataset): def __init__(self,DF, pad_to, trunc_len=50, seq_col='utr', aux_columns='TE_count', other_input_columns=None): """ Dataset to trancate sequence and return in one-hot encoding way `dataset(DF,pad_to,trunc_len=50,seq_col='utr')` ...DF: the dataframe contain sequence and its meta-info ...pad_to: final size of the output tensor ...trunc_len: maximum sequence to retain. number of nt preceding AUG ...seq_col : which col of the DF contain sequence to convert """ DF[seq_col] = DF[seq_col].astype(str) self.df = DF self.pad_to = pad_to self.trunc_len = 0 if trunc_len is None else trunc_len # X and Y self.seqs = self.df.loc[:,seq_col].values self.other_input_columns = other_input_columns self.Y = self.df.loc[:,aux_columns].values def __len__(self): return self.df.shape[0] def __getitem__(self,i): seq = self.seqs[i] x_padded = self.seq_chunk_N_oh(seq) input = x_padded if self.other_input_columns is not None: input = [x_padded] for col in self.other_input_columns: input.append(self.df.loc[:,col].values[i]) y = self.Y[i] return input,y def seq_chunk_N_oh(self,seq): """ truncate the sequence and encode in one hot """ if (len(seq) > self.trunc_len)&(self.trunc_len >0): seq = seq[-1* self.trunc_len:] X = one_hot(seq) X = torch.tensor(X) # X_padded = pad_zeros(X, self.pad_to) return X.float() class MTL_dataset(Dataset): def __init__(self, DF, pad_to=100, seq_col='utr', aux_columns=None, other_input_columns=None, trunc_len=None): """ the dataset for Multi-task learning, will return sequence in one-hot encoded version, together with some auxilary task label arguments: ...csv_path: abs path of csv file, column `utr` should be in the csv ...pad_to : maximum length of the sequences ...columns : list contains what axuslary task label will be """ self.pad_to = pad_to self.trunc_len = trunc_len DF[seq_col] = DF[seq_col].astype(str) self.df = DF # read Df self.seqs = self.df[seq_col].values # take out all the sequence in DF self.columns = aux_columns self.other_input_columns = other_input_columns assert trunc_len == None, "MTL dataset does not support argument trunc len" def __len__(self): return self.df.shape[0] def __getitem__(self,index): seq = self.seqs[index] # sequence: str of len 25~100 X = one_hot(seq).astype(float) # seq_oh : np array, one hot encoding sequence # X = pad_zeros(X) # X : torch.tensor , zero padded to 100 if self.columns == None: # which means no auxilary label is needed item = X ,X elif (type(self.columns) == list) & (len(self.columns)!=0): # return what's in columns aux_labels = self.df.loc[:,self.columns].values[index] # if len(self.columns) == 1: # aux_labels = aux_labels.reshape(-1,1) item = X ,aux_labels if self.other_input_columns is not None: input = [] for col in self.other_input_columns: input.append(self.df.loc[:,col].values[index]) item = (X, input),aux_labels return item class kmer_scan_dataset(Dataset): def __init__(self, DF, seq_col, kmer_size, aux_columns=None, *args): super().__init__() self.df = DF self.seqs = DF[seq_col].values self.rls = DF[aux_columns].values self.k = kmer_size self.kmer_sets = self.create_kmer() self.kmer_lookup = { kmer : i for i, kmer in enumerate(self.kmer_sets) } def __len__(self): return self.df.shape[0] def __getitem__(self, index): seq = self.seqs[index] y = self.rls[index] x = self.scan_kmer(seq) return torch.tensor(x.astype(float)), torch.tensor(y) def scan_kmer(self, seq): mat = np.zeros((len(self.kmer_lookup), len(seq)-self.k)) for i in range(0, len(seq)-self.k): seq_let = seq[i:i+self.k] mat[self.kmer_lookup[seq_let], i] = 1 return mat def create_kmer(self): all_kmer = {0:['']} k = 1 while k <= self.k: k_mer = [] # 1 ; 4 ; 2: 4**2 ... for source in all_kmer[k-1]: k_mer += [source + base for base in ['A','C','G','T']] assert len(k_mer) == 4**k, f"new kmers {len(k_mer)}, not equal to {4**k}" all_kmer[k] = k_mer k += 1 all_kmer.pop(0) return all_kmer[self.k] class FASTA_dataset(MTL_dataset): def __init__(self,DF, pad_to,trunc_len, seq_col, aux_columns, other_input_columns): super().__init__(DF=DF, pad_to=pad_to, trunc_len=trunc_len, seq_col=seq_col, aux_columns=aux_columns, other_input_columns=other_input_columns) def get_splited_dataloader(dataset_func, df_ls, ratio:list, batch_size, shuffle, pad_to, seed=42,return_dataset=False): """ split the total dataset into train val test, and return in a DataLoader (train_loader,val_loader,test_loader) dataset : the defined ratio : the ratio of train : val : test batch_size : int """ # determined set-ls set_ls = [dataset_func(df) for df in df_ls] if return_dataset: return set_ls if pad_to == 0 : # automatic padding to `8x -1` # and we will pad the sequence of similar length with bucket sampler col_fn = lambda x: pack_seq(x,0) sampler_ls = [{"batch_sampler":Bucket_Sampler(df,batch_size=batch_size,drop_last=True), "num_workers":4, "drop_last":False, "collate_fn":col_fn} for df in df_ls] # wrap dataset to dataloader loader_ls = [DataLoader(subset,**kwargs) for subset,kwargs in zip(set_ls,sampler_ls)] else: col_fn = lambda x: pack_seq(x,pad_to) loaderargs = {"batch_size": batch_size, "generator":torch.Generator().manual_seed(42), "drop_last":False , "num_workers":4, "shuffle":shuffle, "collate_fn":col_fn} # wrap dataset to dataloader loader_ls = [DataLoader(subset,**loaderargs) for subset in set_ls] if len(loader_ls) == 2: # a complement of empty test set loader_ls.append(None) return loader_ls def split_DF(data_path,split_like,ratio, kfold_cv, kfold_index=None,seed=42): class _cf_data(object): def __init__(self,data_path): self.path = data_path self.is_fasta = data_path.endswith('.fasta') self.__read__() def __read__(self): if self.is_fasta: self.data = list(SeqIO.parse(self.path,'fasta')) else: self.data = pd.read_csv(os.path.join(data_dir,self.path),low_memory=False) def __len__(self): return len(self.data) def _slice_(self, indices): if isinstance(self.data, list) & self.is_fasta: return np.array(self.data)[indices] elif isinstance(self.data, pd.DataFrame): return self.data.iloc[indices] if kfold_cv == True: full_df = pd.read_csv(os.path.join(data_dir,data_path),low_memory=False) # K-fold CV : 8:1:1 for each partition df_ls = KFold_df_split(full_df,kfold_index) if kfold_cv == 'train_val': train_val, test = [_cf_data(data_path).data for data_path in split_like] # K-fold CV : 8:1:1 for each partition train, val = KFold_1_split(train_val,kfold_index) df_ls = [train, val , test] elif type(split_like) == list: df_ls = [_cf_data(data_path).data for data_path in split_like] elif type(kfold_cv) == str: full_df = pd.read_csv(os.path.join(data_dir,data_path),low_memory=False) if kfold_cv in full_df.columns: kfold_index = str(kfold_index) test = full_df.query(f"`{kfold_cv}` == @kfold_index") train_val = full_df.query(f"`{kfold_cv}` != @kfold_index") train, val = train_test_split(train_val, test_size=0.05) df_ls = [train, val, test] else: full_df = _cf_data(data_path) # POPEN.ratio will determine train :val :test ratio total_len = len(full_df) lengths = [int(total_len*sub_ratio) for sub_ratio in ratio[:-1]] lengths.append(total_len-sum(lengths)) # make sure the sum of length is the total len set_ls = random_split(full_df,lengths,generator=torch.Generator().manual_seed(seed)) df_ls = [full_df._slice_(subset.indices) for subset in set_ls] # df.iloc [ idx ] return df_ls def split_DF_call(data_path,split_like,ratio, kfold_cv, kfold_index=None,seed=42): class _cf_data(object): def __init__(self,data_path): self.path = data_path self.is_fasta = data_path.endswith('.fasta') self.__read__() def __read__(self): if self.is_fasta: self.data = list(SeqIO.parse(self.path,'fasta')) else: self.data = pd.read_csv(os.path.join(data_dir,self.path),low_memory=False) def __len__(self): return len(self.data) def _slice_(self, indices): if isinstance(self.data, list) & self.is_fasta: return np.array(self.data)[indices] elif isinstance(self.data, pd.DataFrame): return self.data.iloc[indices] if kfold_cv == True: full_df = pd.read_csv(os.path.join(data_dir,data_path),low_memory=False) # K-fold CV : 8:1:1 for each partition df_ls = KFold_df_split(full_df,kfold_index) if kfold_cv == 'train_val': train_val, test = [_cf_data(data_path).data for data_path in split_like] # K-fold CV : 8:1:1 for each partition train, val = KFold_1_split(train_val,kfold_index) df_ls = [train, val , test] elif type(split_like) == list: df_ls = [_cf_data(data_path).data for data_path in split_like] elif type(kfold_cv) == str: full_df = pd.read_csv(os.path.join(data_dir,data_path),low_memory=False) if kfold_cv in full_df.columns: kfold_index = str(kfold_index) test = full_df.query(f"`{kfold_cv}` == @kfold_index") train_val = full_df.query(f"`{kfold_cv}` != @kfold_index") train, val = train_test_split(train_val, test_size=0.05) df_ls = [train, val, test] else: full_df = _cf_data(data_path) # POPEN.ratio will determine train :val :test ratio total_len = len(full_df) lengths = [int(total_len*sub_ratio) for sub_ratio in ratio[:-1]] lengths.append(total_len-sum(lengths)) # make sure the sum of length is the total len set_ls = random_split(full_df,lengths,generator=torch.Generator().manual_seed(seed)) df_ls = [full_df._slice_(subset.indices) for subset in set_ls] # df.iloc [ idx ] return df_ls def KFold_df_split(df,K,**kfoldargs): """ split the dataset DF in a ratio of 8:1:1 , train:val:test in the framework of K-fold CV set random seed = 43 arguments: df : the `pd.DataFrame` object containing all data info K : [0,4] , the index of subfold """ # K-fold partition : n_splits=5 fold_index = list(KFold(10,shuffle=True,random_state=42).split(df)) train_val_index, test_index = fold_index[K] # the first 4/5 part of it is train set # index the df train_val_df = df.iloc[train_val_index] test_df = df.iloc[test_index] # the remaining 1/5 data will further break into val and test train_df, val_df = train_test_split(train_val_df,test_size=0.05,random_state=42) return [train_df,val_df,test_df] def KFold_1_split(df,K,**kfoldargs): """ for dataset with standard test set, where only train_val can be splited """ fold_index = list(KFold(10,shuffle=True,random_state=42).split(df)) train_index, val_index = fold_index[K] # the first 4/5 part of it is train set # index the df train = df.iloc[train_index] val = df.iloc[val_index] return train , val def get_dataloader(POPEN): """ wrapper """ # POPEN.csv_path = "/mnt/sina/run/ml/gan/motif/MTtrans/test.csv" # print(POPEN.csv_path) # print(POPEN.split_like) # print(POPEN.train_test_ratio) # print(POPEN.kfold_cv) # print(POPEN.kfold_index) df_ls = split_DF(POPEN.csv_path,POPEN.split_like,POPEN.train_test_ratio,POPEN.kfold_cv,POPEN.kfold_index,seed=42) DS_Class = eval(POPEN.dataset+"_dataset") dataset_func = lambda x : DS_Class(x,pad_to=POPEN.pad_to,trunc_len=POPEN.trunc_len, seq_col=POPEN.seq_col, aux_columns=POPEN.aux_task_columns, other_input_columns=POPEN.other_input_columns) loader_ls = get_splited_dataloader(dataset_func, df_ls,ratio=POPEN.train_test_ratio, batch_size=POPEN.batch_size, shuffle=POPEN.shuffle, pad_to=POPEN.pad_to, seed=42) # new function return loader_ls