import scarf2 as scarf from pac import pac_api from CCP4ErrorHandling import * from CCP4Modules import JOBCONTROLLER import logging # VU: change to logging.DEBUG for development and logging.WARNING for production logging.basicConfig(level=logging.DEBUG) #Beware the i2 custom handling mechanism expects the Interface class to be first in the file class CCP4I2_SCARF_Interface: SHOW_USER_PARAMS = [ 'username' , 'password' ] POLLPERIOD = 60 ERROR_CODES = { 101 : { 'description' : 'CCP4i2 currently not available on Scarf cluster' } } def __init__(self,serverParams=None): print 'CCP4I2_SCARF_Interface' # self.serverParams is method to return an instance of CServerParams which holds meta data for server job self.serverParams = serverParams self.scarfProgramInsts = {} def getScarfInsts(self,jobId): import copy,os import CCP4Utils sP = self.serverParams(jobId) if sP is None: return None #print 'CCP4I2_SCARF_Interface.getScarfInsts 2',sP,type(sP) if not self.scarfProgramInsts.has_key(jobId): try: downloaddir = os.path.join(sP.projectDirectory,'CCP4_TMP','job_'+sP.jobNumber) self.scarfProgramInsts[jobId] = I2SCARFProgram(joball=sP.local_tarball, jobName=CCP4Utils.safeOneWord(sP.projectName+'_'+sP.jobNumber), downloaddir=downloaddir) if sP.serverProcessId is not None: self.scarfProgramInsts[jobId].jobid = copy.deepcopy(sP.serverProcessId) except Exception as e: print 'ERROR instantiating I2SCARFProgram',e return None #print 'CCP4I2_SCARF_Interface.getScarfInsts 3',self.scarfProgramInsts[jobId].jobid return self.scarfProgramInsts[jobId] def openConnection(self,jobId=None,**args): # Open connection and get server token status, msg = scarf.check(application='CCP4i2') print 'CCP4I2_SCARF_Interface.openConnection check',status, msg if not status: if msg.count('not available'): raise CErrorReport(self.__class__,101,msg) sP = self.serverParams(jobId) print 'CCP4I2_SCARF_Interface.openConnection logon',sP.username,sP.password status = scarf.logon(username=sP.username,password=sP.password,application='CCP4i2') print 'CCP4I2_SCARF_Interface.openConnection logon',status def setup(self,jobId): # Setup the job by: # 1) transporting the tarball self.serverParams(jobId).local_tarball to remote machine # 2) creating/transporting scripts # 3) The contents of self.serverParams(jobId) should be updated appropriately # If the tarball path on remote machine is remotePath/projectname_jobnumber_setup.ccp4db.zip # then need to set in self.serverParams(jobId) path to files that are going to be used later: # sP.remote_finished_tarball=remotePath/projectname_jobnumber_finished.ccp4db.zip # sP.remote_report_file = remotePath/projectname_jobnumber_work/project/CCP4_JOBS/job_jobNumber/program.xml' pass def submit(self,jobId): # Actually start the remote job status,message = self.getScarfInsts(jobId).submit() print 'CCP4I2_SCARF_Interface.submit scarfid',self.getScarfInsts(jobId).jobid,self.getScarfInsts(jobId).status,message if status: JOBCONTROLLER().setServerParam(jobId,'serverProcessId',self.getScarfInsts(jobId).jobid) return status,message def transportFiles(self,jobId,copyList=[],mode='put'): # Each item of copyList is list of: local_file,remote_file # Mode is 'put'/'get' # Beware the scarf mechanism assumes file has same local and remote path! print 'CCP4I2_SCARF_Interface.transportFiles',copyList import os if mode == 'get': scarfId = self.getScarfInsts(jobId).jobid for local,remote in copyList: #ok,msg = scarf.download(scarfId,'job_'+str(self.serverParams(jobId).jobNumber)+'_'+remote, # os.path.split(local)[0],dstName=local) print 'pac_api.downloadJobFiles',scarfId,'job_'+str(self.serverParams(jobId).jobNumber)+'_'+remote,local pac_api.downloadJobFiles(str(scarfId),local,'job_'+str(self.serverParams(jobId).jobNumber)+'_'+remote,'') #print 'CCP4I2_SCARF_Interface.transportFiles',ok,msg print 'CCP4I2_SCARF_Interface.transportFiles' def pollForFinishedJobs(self): # For each job with pollFinish code 3 or 4 if it has finished # Remove any finished job from serverParams # Return a list of jobId for jobs that have finished finished = [] for jobId in self.getJobsToPollFinish([3,4]): flag,status = self.getScarfInsts(jobId).poll() # possible status values: Pending Suspended Running Exit Done print 'CCP4I2_SCARF_Interface.pollForFinishedJob 2',flag,status if flag: return status else: return 'Poll fail' def pollForFinishedJob(self,jobId): # Return status of job 0=Pending 1=Failed 2=Finished 3=Job Unknown sP = self.serverParams(jobId) scarfObj = self.getScarfInsts(jobId) flag,status = scarfObj.poll() # possible status values: Pending Suspended Running Exit Done # (have also observed Unknown when job probably never existed) print 'CCP4I2_SCARF_Interface.pollForFinishedJob',jobId,flag,status if flag: if status in [ 'Exit' , 'Done', 'Unknown' ]: return [ 'Exit' , 'Done', 'Unknown' ].index(status)+1 return 0 def handleFinishedJob(self,jobId,status=None): # Called after remote job finished needs to: # 1) Retrieve remote tarball to self.serverParams(jobId).local_finished_tarball import os sP = self.serverParams(jobId) scarfObj = self.getScarfInsts(jobId) dstdir,finfile = os.path.split(sP.local_finished_tarball) logging.info('CCP4I2_SCARF_Interface.handleFinishedJob '+jobId+' '+finfile+' '+dstdir) #stat,msg = scarf.download(scarfObj.jobid,finfile,dstdir) logging.debug('pac_api.downloadJobFiles '+str(scarfObj.jobid)) pac_api.downloadJobFiles(str(scarfObj.jobid),dstdir+'/',finfile,'') #print 'handleFinishedJob from scarf.download',stat,msg #del self.scarfProgramInsts[jobId] def guiFrame(self,parentFrame): print 'CCP4I2_SCARF_Interface.guiFrame' from PyQt4 import QtGui,QtCore but = QtGui.QPushButton('Open SCARF portal in web viewer',parentFrame) parentFrame.connect(but,QtCore.SIGNAL('clicked()'),self.openScarfPortal) parentFrame.layout().addWidget(but) def openScarfPortal(self): import CCP4Modules from PyQt4 import QtCore CCP4Modules.WEBBROWSER().loadPage(url=QtCore.QUrl('http://portal.scarf.rl.ac.uk')) class I2SCARFProgram(scarf.SCARFProgram): # Subclass the generic SCARFProgram for i2 def __init__(self,joball=None,jobName=None,downloaddir=None): scarf.SCARFProgram.__init__(self,downloaddir=downloaddir) self.JobDict['APPLICATION_NAME'] = 'CCP4i2' self.JobDict['INPUT_FILES'] = { 'SETUP': joball + ',upload' } self.JobDict['PARAMS'] = { 'JOB_NAME': jobName, 'STDOUT': 'stdout.txt', 'STDERR': 'stderr.txt'} def download(self,jobball=''): dir0 = self.downloaddir print 'CCP4I2_SCARF_Interface.I2SCARFProgram.download',str(self.jobid), dir0,jobball pac_api.downloadJobFiles(str(self.jobid), dir0,jobball )