import os, time, posixpath import threading from threading import Thread import logging import traceback import subprocess import scitbx import re import BLASTMRutils from phaser import CalcCCFromMRsolutions class ThreadLogger(): def __init__(self, logger, mylock=None): self.logger = logger self.mylock = mylock def log(self,loglevel, mstr): # protect stdout from multiple output if we're multiprocessing if self.mylock != None: self.mylock.acquire() self.logger.log(loglevel, mstr) self.logger.handlers[0].flush() self.logger.handlers[1].flush() if self.mylock != None: self.mylock.release() class PhaserRunner(Thread): def __init__ (self, baseinputfname, modelinputfnameprefix, calclabel, rootstr, parent, searches=None, fmaxhours = 48, nmaxsols = 0, mylock = None, threadnumber = 0, maxthreads = 1, minmodelorder = 0): Thread.__init__(self) self.searches = searches self.baseinputfname = baseinputfname self.calclabel = calclabel self.mylock = mylock self.tlogger = ThreadLogger(parent.logger, mylock) self.modelinputfnameprefix = modelinputfnameprefix self.modelinputfname = self.modelinputfnameprefix + ".txt" self.osdepbool = False self.rootstr = rootstr self.nmaxsols = nmaxsols self.fmaxhours = fmaxhours self.nmaxminutes = self.fmaxhours * 60 self.errorcodesum = 0 self.minmodelorder = minmodelorder self.prefix = "The " self.maxthreads = maxthreads self.parent = parent # label thread with our tag so we can identify it as ours in GetActivethreads() self.name = self.prefix + str(threadnumber) + ". thread" if os.sys.platform == "win32": self.osdepbool = True def PhaserRun(self,path): # concatenate the two or three input scripts into MRinput.txt that is used by phaser baseinputfile = open(self.baseinputfname,"r") baseinput = baseinputfile.read() baseinputfile.close() modelinputfile = open(os.path.join(path, self.modelinputfname),"r") mstr = open(os.path.join(path, self.modelinputfname),"r").read() modelinputfile.close() # ignore VRMS in solution for first fixed component and template solutions so get rid of any SOLU ENSE lines modelinput_no_vrms = re.sub("SOLU ENSE.*\n", "", mstr) self.tlogger.log(logging.DEBUG, "Concatenating: " + self.calclabel + "MRinput.txt = " \ + self.baseinputfname + " + rootstr + " + self.modelinputfname) # Avoid race condition so not just the very first thread continues with self.name being assigned # but waits a little until all other threads have also had their self.name assigned time.sleep(1) # number of our threads is the count of self.name assigned to "phaser-" nactivethreads = len(self.GetActivethreads()) # crank up the throttle by using more CPUs if they are idle nopenmpthreads = max(1,self.maxthreads/(nactivethreads+1)) jobstr = "JOBS %d\n" %nopenmpthreads myinput = baseinput + self.rootstr + jobstr + modelinput_no_vrms inputfilename = posixpath.join(path, self.calclabel + "MRinput.txt") inputfile = open(inputfilename, "w") inputfile.write(myinput) inputfile.close() command = self.parent.phaserexe output = subprocess retval = 0 try: self.tlogger.log(logging.INFO, self.name + " is executing Phaser in folder, "\ + path + "\n\n" + myinput) myproc = subprocess.Popen(command, shell=self.osdepbool, stdin = subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE) (stdoutdata, stderrdata) = myproc.communicate( input = myinput ) retval = myproc.returncode except Exception, m: self.failure = True self.tlogger.log(logging.ERROR, m) finally: if (retval != 0): self.tlogger.log(logging.ERROR, "Something went wrong with the "\ "phaser calculation using input file: \n" + os.path.join(self.parent.topfolder, inputfilename) + \ "\nReturn code is: " + str(retval) + "\n" + stderrdata) self.errorcodesum += retval return stdoutdata def AbortRequested(self, folder): # stop further phaser jobs if this file is present stop1 = os.path.exists(posixpath.join(self.parent.topfolder,"StopThreadsGracefully")) # or present in the directory above stop2 = os.path.exists(posixpath.join(self.parent.topfolder,"..","StopThreadsGracefully")) if stop1 or stop2: self.tlogger.log(logging.INFO,"Aborting calculation on %s.\n" %folder) return True else: return False def GetActivethreads(self): # list names of still active threads active = [t.name for t in threading.enumerate()] setofthreadnames = set(active) if self.name in setofthreadnames: active.remove(self.name) namesofproperthreads = [] # avoid listing names of spurious threads not created explicitly by us for t in active: if t.find(self.prefix) > -1: # filter them out by if they are not tagged with our prefix namesofproperthreads.append(t) return namesofproperthreads def run(self): try: completedrunfname = posixpath.join(self.parent.topfolder, self.calclabel + self.parent.targetpdbid + "_DoneMRJobs.txt") testsol = BLASTMRutils.BLASTMRutils(self.parent.topfolder, self.parent, completedrunfname, self.mylock) for i, models in enumerate( self.searches ): # create modelsdir as concatenated PDB codes separated by commas modelsdir = [ [e[0] for e in e1] for e1 in models[:-1]][0] workdir = ",".join( modelsdir ) MRfolder = posixpath.join(self.parent.topfolder, "MR", workdir) MRfolder = posixpath.relpath(MRfolder, self.parent.topfolder) if self.AbortRequested(MRfolder): break self.tlogger.log(logging.INFO,"\n%d. MR calculation out of %s by %s" %((i+1), len(self.searches), self.name)) self.parent.PrepareInputFiles(models[:-1][0]) #import code; code.interact(local=locals()) for partmodels in models[1]: self.tlogger.log(logging.INFO,"%s: Adding one more partial model"\ %self.name) modelorder = len(partmodels.split(",")) if self.minmodelorder > modelorder: continue self.tlogger.log(logging.INFO, "%s: modelorder %s" %(self.name, modelorder)) if self.AbortRequested(MRfolder): break bsolved = [False] founddir = [""] brunphaser = True if testsol.PhaserRanPartModelsBefore(founddir,partmodels,bsolved) == True: if bsolved[0] == True: # and solved it self.tlogger.log(logging.INFO, "%s: A previous job found the partial structure %s in %s" %(self.name,partmodels, founddir[0])) else: self.tlogger.log(logging.INFO, "%s: A previous job didn't find the partial structure %s in %s" %(self.name,partmodels, workdir)) brunphaser = False # Run phaser with the model created by PrepareInputfiles solfnameprefix = self.calclabel + "{" + workdir + "}"+ partmodels solfname = solfnameprefix + ".sol" rootname = posixpath.relpath(MRfolder) + posixpath.sep \ + self.calclabel + "{" + workdir + "}" + partmodels # stop phaser after some time and limit number of solutions from an MR run self.rootstr = "ROOT " + rootname + \ "\nKILL TIME %d\nPURGE RNP NUMB %d\n" %(self.nmaxminutes,self.nmaxsols) self.modelinputfname = self.modelinputfnameprefix + "{" + partmodels + "}.txt" if brunphaser: stdoutput = self.PhaserRun(MRfolder) phaserlogfile = open(rootname + ".log","w") phaserlogfile.write(stdoutput) phaserlogfile.close() # see if phaser found a solution for these models # "models" is the search order for this MR calculation: pdb_a, pdb_b, pdb_c,...pdb_m, pdb_n nPartSolWithThisStrategy = testsol.IsthereaPartialSolution( workdir, partmodels, solfname) testsol.NoticeCompletedPhaserRun(workdir, partmodels, nPartSolWithThisStrategy) if nPartSolWithThisStrategy < 0: self.tlogger.log(logging.INFO, "%s: Phaser didn't even write a solution file, %s, for the %s job"\ %(self.name, partmodels, modelsdir)) self.tlogger.log(logging.INFO, "%s: Done placing partial model, %s, for the %s job." \ "\nNow calculating correlation coefficients" \ %(self.name, partmodels, modelsdir)) # crank up the throttle by using more CPUs if they are idle nthreads = max(1, int(self.maxthreads/(len(self.GetActivethreads())+1))) MRfolder = posixpath.join(self.parent.topfolder, "MR", workdir) logfname = posixpath.join(MRfolder, self.calclabel + "CalcCC_" + workdir + partmodels + ".log") CCfname = posixpath.join(MRfolder, self.calclabel + "CalcCC_{" + workdir + "}" + partmodels + ".txt") if os.path.exists(CCfname) == False: CCoutput = CalcCCFromMRsolutions.CalcCCFromMRsolution(self.parent.targetpdbid, MRfolder, self.calclabel, workdir, partmodels, bCalcCC = True, mylock= self.mylock, nthreads = nthreads, debug = False) mfile = open(logfname,"w") mfile.write(CCoutput) mfile.close() else: self.tlogger.log(logging.INFO,"%s: %s already done" %(self.name, CCfname)) templfolder = posixpath.join(self.parent.topfolder, "TemplateSolutions", workdir) templworkdir = workdir if os.path.exists(templfolder) == False: # if 1QZ7\TemplateSolutions\1IAL_A0,3B8E_A0 doesn't exists try 1QZ7\TemplateSolutions\3B8E_A0,1IAL_A0 revmodelsdir = [] revmodelsdir.extend(modelsdir) # deep copy rather than assignment revmodelsdir.reverse() templworkdir = ",".join( revmodelsdir ) templfolder = posixpath.join(self.parent.topfolder, "TemplateSolutions", templworkdir) logfname = posixpath.join(templfolder, self.calclabel + "RNPCalcCC_" + templworkdir + ".superposed.log") CCfname = posixpath.join(templfolder, self.calclabel + "RNPCalcCC_{" + templworkdir + "}.superposed.txt") if (os.path.exists(templfolder) and os.path.exists(CCfname) )== False: CCoutput = CalcCCFromMRsolutions.CalcCCFromMRsolution(self.parent.targetpdbid, templfolder, self.calclabel + "RNP", templworkdir, ".superposed", bCalcCC = True, mylock= self.mylock, nthreads = nthreads, debug = False) mfile = open(logfname,"w") mfile.write(CCoutput) mfile.close() else: self.tlogger.log(logging.INFO,"%s: %s already done" %(self.name, CCfname)) # also calculate CCs for for the template solutions for the 1st component if modelsdir[0] != workdir: templfolder = posixpath.join(self.parent.topfolder, "TemplateSolutions", modelsdir[0]) templworkdir = modelsdir[0] logfname = posixpath.join(templfolder, self.calclabel + "RNPCalcCC_" + templworkdir + ".superposed.log") CCfname = posixpath.join(templfolder, self.calclabel + "RNPCalcCC_{" + templworkdir + "}.superposed.txt") if (os.path.exists(templfolder) and os.path.exists(CCfname) )== False: CCoutput = CalcCCFromMRsolutions.CalcCCFromMRsolution(self.parent.targetpdbid, templfolder, self.calclabel + "RNP", templworkdir, ".superposed", bCalcCC = True, mylock= self.mylock, nthreads = nthreads, debug = False) mfile = open(logfname,"w") mfile.write(CCoutput) mfile.close() else: self.tlogger.log(logging.INFO,"%s: %s already done" %(self.name, CCfname)) # list names of still active threads namesofproperthread = self.GetActivethreads() threadnames = ", ".join(namesofproperthread) if threadnames == "": threadnames = "no others" self.tlogger.log(logging.INFO,"%s is done and waits for %s to finish.\n" %(self.name,threadnames)) except Exception, m: self.tlogger.log(logging.ERROR, self.name + ": " + str(m) + "\n" \ + traceback.format_exc()) self.errorcodesum += 1 raise Exception, m