from multiprocessing import Process import numpy as np import edlib as ed import progressbar import time from skbio.alignment import local_pairwise_align_ssw from skbio import DNA import copy from save_data import write_data_synteny class Thread_objects(): def __init__(self, df_temp, gene_sequences, lsy, i, name): self.gene_sequences = copy.deepcopy(gene_sequences) self.lsy = copy.deepcopy(lsy) self.df = copy.deepcopy(df_temp) self.smg = [] self.sml = [] self.indexes = [] self.i = i self.start = 0 self.end = 0 self.name = name def create_synteny_matrix_mul(self, gene_seq, g1, g2, n): for gene in g1: if gene == "NULL_GENE": continue try: _ = gene_seq[gene] except BaseException: return np.zeros((n, n, 2)), np.zeros((n, n, 2)) for gene in g2: if gene == "NULL_GENE": continue try: _ = gene_seq[gene] except BaseException: return np.zeros((n, n, 2)), np.zeros((n, n, 2)) sm = np.zeros((n, n, 2)) sml = np.zeros((n, n, 2)) for i in range(n): if g1[i] == "NULL_GENE": continue if gene_seq[g1[i]] == "": return np.zeros((n, n, 2)), np.zeros((n, n, 2)) for j in range(n): if g2[j] == "NULL_GENE": continue if gene_seq[g2[j]] == "": return np.zeros((n, n, 2)), np.zeros((n, n, 2)) norm_len = max(len(gene_seq[g1[i]]), len(gene_seq[g2[j]])) try: result = ed.align( gene_seq[g1[i]], gene_seq[g2[j]], mode="NW", task="distance") sm[i][j][0] = result["editDistance"] / (norm_len) result = ed.align( gene_seq[g1[i]], gene_seq[g2[j]][::-1], mode="NW", task="distance") sm[i][j][1] = result["editDistance"] / (norm_len) _, result, _ = local_pairwise_align_ssw( DNA(gene_seq[g1[i]]), DNA(gene_seq[g2[j]])) sml[i][j][0] = result / (norm_len) _, result, _ = local_pairwise_align_ssw( DNA(gene_seq[g1[i]]), DNA(gene_seq[g2[j]][::-1])) sml[i][j][1] = result / (norm_len) except BaseException: return np.zeros((n, n, 2)), np.zeros((n, n, 2)) return sm, sml def synteny_matrix(self, gene_seq, hdf, lsy, n): t = 0 self.start = time.time() for index, row in progressbar.progressbar(hdf.iterrows()): g1 = str(row["gene_stable_id"]) g2 = str(row["homology_gene_stable_id"]) x = [] y = [] t += 1 try: _ = lsy[g1] except BaseException: continue try: _ = lsy[g2] except BaseException: continue for i in range(len(lsy[g1]['b']) - 1, -1, -1): x.append(lsy[g1]['b'][i]) x.append(g1) for k in lsy[g1]['f']: x.append(k) for i in range(len(lsy[g2]['b']) - 1, -1, -1): y.append(lsy[g2]['b'][i]) y.append(g2) for k in lsy[g2]['f']: y.append(k) assert(len(x) == len(y)) assert(len(x) == (2 * n + 1)) smgtemp, smltemp = self.create_synteny_matrix_mul( gene_seq, x, y, 2 * n + 1) if np.all(smgtemp == 0): continue self.smg.append(smgtemp) self.sml.append(smltemp) self.indexes.append(index) self.end = time.time() print( "Thread {} finished in {}s.".format( self.i + 1, self.end - self.start)) write_data_synteny(self.smg, self.sml, self.indexes, self.i, self.name) class Procerssrunner(): def __init__(self): self.thread_alive = [] self.obj_list = [] def start_thread(self, obj, i, thread_alive, n, name): t = Process(target=obj.synteny_matrix, args=( obj.gene_sequences, obj.df, obj.lsy, n), name="Thread_" + str(i + 1)) print("Thread ", (i + 1), " started for ", name, ".") thread_alive.append(t) def start_processes(self, nop, df, gene_sequences, lsy, part, n, name): for i in range(nop): df_temp = df.loc[df.index.values[i * part:(i + 1) * part]] obj = Thread_objects(df_temp, gene_sequences, lsy, i, name) print("Object Created") self.start_thread(obj, i, self.thread_alive, n, name) self.obj_list.append(obj) st = time.time() for t in self.thread_alive: t.start() # for t in self.thread_alive: # t.join() while(len(self.thread_alive) != 0): time.sleep(60) self.thread_alive = [t for t in self.thread_alive if t.is_alive()] end = time.time() print("Ending Processes") print("Time taken:{}s".format(end - st))