""" aimless_pipe.py: CCP4 GUI Project Copyright (C) 2012 STFC """ import os from CCP4PluginScript import CPluginScript from lxml import etree class aimless_pipe(CPluginScript): TASKMODULE = 'data_reduction' # Where this plugin will appear on the gui TASKTITLE = 'Scale and merge data' # A short title for gui menu TASKNAME = 'aimless_pipe' # Task name - should be same as class name TASKVERSION= 0.0 # Version of this plugin PERFORMANCECLASS = 'CDataReductionPerformance' MAINTAINER = 'pre@mrc-lmb.cam.ac.uk' ERROR_CODES = { 201 :{'description':'Pointless failed'}, 202 :{'description':'Aimless failed'}, 203 :{'description':'Ctruncate failed'}, 204 :{'description':'FreeR failed'} } PURGESEARCHLIST = [[ 'ctruncate%*/hklout.mtz', 0, None ] ] # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # def process(self): self.rootXML = etree.Element('AIMLESS_PIPE') # Start up print "### Starting aimless pipeline ###" unsetData = self.checkInputData() if len(unsetData)>0: self.reportStatus(CPluginScript.FAILED) return self.fatalError = None self.process_pointless() # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # def process_pointless(self): print "### Running pointless ###" #print self.container.inputData #nfiles = len(self.container.inputData.UNMERGEDFILES) #for i in range(nfiles): # if not self.container.inputData.UNMERGEDFILES[i].file.fileContent.knowncell: # self.container.controlParameters.CELL = self.container.inputData.UNMERGEDFILES[i].cell #print "SCALING_PROTOCOL",self.container.controlParameters.SCALING_PROTOCOL # Run pointless self.pointless = self.makePluginObject('pointless') self.pointless.container.inputData.copyData(self.container.inputData,['UNMERGEDFILES']) if self.container.inputData.HKLIN_REF: self.pointless.container.inputData.copyData(self.container.inputData,['HKLIN_REF']) if self.container.inputData.HKLIN_REF: self.pointless.container.inputData.copyData(self.container.inputData,['XYZIN_REF']) self.pointless.container.controlParameters.copyData \ (self.container.controlParameters, ['EXCLUDE_BATCH','EXCLUDED_BATCHES', 'POINTLESS_USE_RESOLUTION_RANGE', 'RESOLUTION_RANGE', 'ISIGLIMIT','CCHALFLIMIT','TOLERANCE','MODE','REFERENCE_DATASET','SET_SETTING', 'CHOOSE_MODE','CHOOSE_SOLUTION_NO','CHOOSE_LAUEGROUP','CHOOSE_SPACEGROUP', 'REINDEX_OPERATOR','CELL','WAVELENGTH','RUN_MODE','RUN_BATCHLIST']) self.connectSignal(self.pointless,'finished',self.process_aimless) self.pointless.process() # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # def process_aimless(self,status): print "process_aimless" if status.get('finishStatus') == CPluginScript.FAILED: print "process_aimless failed" self.fatalError = [201, 'Pointless failed', status] # self.appendErrorReport(201, details='Pointless failed') # self.reportStatus(status) self.process_finish(CPluginScript.FAILED) return try: pointlessXMLPath = self.pointless.makeFileName('PROGRAMXML') pointlessEtree = etree.parse(pointlessXMLPath) self.rootXML.append(pointlessEtree.getroot()) with open (self.makeFileName('PROGRAMXML'),"w") as outputXML: outputXML.write(etree.tostring(self.rootXML,pretty_print=True)) print "### Running aimless ###" # Run aimless to edit model self.aimless = self.makePluginObject('aimless') # Copy all relevant parameters from GUI to Aimless script self.aimless.container.controlParameters.copyData \ (self.container.controlParameters, ['REFERENCE_DATASET', 'REFERENCE_FOR_AIMLESS', 'EXCLUDE_BATCH','EXCLUDED_BATCHES','RESOLUTION_RANGE', 'ACCEPT_OVERLOADS','ACCEPT_EDGES','ACCEPT_XDS_MISFITS', 'SCALING_PROTOCOL','BFACTOR_SCALE','SCALES_ROTATION_TYPE', 'SCALES_ROTATION_SPACING','SCALES_ROTATION_NBINS', 'SCALES_BROTATION_TYPE','SCALES_BROTATION_SPACING', 'SCALES_BROTATION_NBINS', 'SCALES_SECONDARY_NSPHHARMONICS', 'SCALES_TILETYPE','SCALES_NTILEX','SCALES_NTILEY', 'OUTLIER_OVERRIDE','OUTLIER_EMAX', 'OUTLIER_SDMAX','OUTLIER_SDMAX2', 'OUTLIER_SDMAXALL','OUTLIER_SDMAXALL_ADJUST','OUTLIER_COMBINE', 'RUN_MODE', 'RUN_BATCHLIST', 'SDCORRECTION_OVERRIDE','SDCORRECTION_REFINE','SDCORRECTION_FIXSDB', 'SDCORRECTION_OPTIONS','SDCORRECTION_DAMP','SDCORRECTION_SIMILARITY_SDFAC', 'SDCORRECTION_SIMILARITY_SDB','SDCORRECTION_SIMILARITY_SDADD', 'SDCORRECTION_SET', 'SDCORRECTION_SDFAC','SDCORRECTION_SDB','SDCORRECTION_SDADD', 'SDCORRECTION_TIESDB_SD', 'INTENSITIES_OVERRIDE','INTENSITIES_OPTIONS', 'PARTIALS_TEST','PARTIALS_FRACLOW','PARTIALS_FRACHIGH', 'PARTIALS_CHECK','PARTIALS_SCALE','PARTIALS_SCALE_MIN', 'SCALING_DETAILS','CYCLES_FLAG','CYCLES_N', 'SELECT1','SELECT_IOVSDMIN','SELECT2','SELECT_EMIN', 'TIE_ROTATION','TIE_ROTATION_SD', 'TIE_BFACTOR','TIE_BFACTOR_SD', 'TIE_SURFACE','TIE_SURFACE_SD', 'TIE_BZERO','TIE_BZERO_SD','ONLYMERGE' ]) self.aimless.container.inputData.copyData \ (self.container.inputData, ['HKLIN_REF', 'XYZIN_REF']) self.aimless.container.inputData.UNMERGEDFILE = \ self.pointless.container.outputData.MTZUNMERGEDOUT self.aimless.container.controlParameters.OUTPUT_MODE = 'MERGED' self.connectSignal(self.aimless,'finished',self.process_cycle_ctruncate) self.aimless.process() except Exception as e: # failed in Aimless print "Aimless error" self.appendErrorReport(CPluginScript,39,str(e)) self.fatalError = [202, 'Aimless failed', status] self.process_finish(CPluginScript.FAILED) return # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # def process_cycle_ctruncate(self,status): print 'process_cycle_ctruncate', status if status.get('finishStatus') == CPluginScript.FAILED: self.fatalError = [202, 'Aimless failed', status] self.process_finish(CPluginScript.FAILED) self.reportStatus(status) return print 'process_cycle_ctruncate MTZMERGEDOUT',self.aimless.container.outputData.MTZMERGEDOUT # aimless has finished, and produced an output file for each dataset self.ndatasets = len(self.aimless.container.outputData.MTZMERGEDOUT) self.ndatasets_processed = 0 self.ndatasets_failed = 0 self.collectedCtruncateEtree = etree.Element('CTRUNCATES') for file in self.aimless.container.outputData.MTZMERGEDOUT: self.process_ctruncate(file) # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # def process_ctruncate(self,infile): print "### Running ctruncate on file %s ###" % infile if not os.path.exists(infile.fullPath.get()): self.process_post_ctruncate(CPluginScript.FAILED) else: hkloutList = self.container.outputData.HKLOUT hkloutList.append(hkloutList.makeItem()) filePath = os.path.join(self.workDirectory,'HKLOUT_'+str(self.ndatasets_processed)+'-observed_data.mtz') self.ctruncateOutputDataObject = hkloutList[-1] hkloutList[-1].setFullPath(filePath) import CCP4XtalData hkloutList[-1].contentFlag = CCP4XtalData.CObsDataFile.CONTENT_FLAG_IPAIR try: name = os.path.splitext(os.path.split(self.container.inputData.UNMERGEDFILES[len(hkloutList)-1].file.__str__())[1])[0] hkloutList[-1].annotation = 'Observed dataset '+str(len(hkloutList))+' from '+name except: hkloutList[-1].annotation = 'Observed dataset '+str(len(hkloutList)) # Run ctruncate to generate amplitudes self.ctruncate = self.makePluginObject('ctruncate') self.ctruncate.container.inputData.HKLIN = infile # We assume standard labels from aimless self.ctruncate.container.inputData.ISIGI.I = "IMEAN" self.ctruncate.container.inputData.ISIGI.SIGI = "SIGIMEAN" self.ctruncate.container.inputData.ISIGIanom.Ip = "I(+)" self.ctruncate.container.inputData.ISIGIanom.SIGIp = "SIGI(+)" self.ctruncate.container.inputData.ISIGIanom.Im = "I(-)" self.ctruncate.container.inputData.ISIGIanom.SIGIm = "SIGI(-)" self.ctruncate.container.controlParameters.copyData(self.container.controlParameters,['NRES']) self.ctruncate.container.controlParameters.OUTPUTMINIMTZ = True self.ctruncate.container.controlParameters.OUTPUT_INTENSITIES = True self.ctruncate.container.outputData.OBSOUT.setFullPath(filePath) self.connectSignal(self.ctruncate,'finished',self.process_post_ctruncate) self.ctruncate.process() print "DONE Running ctruncate on file %s ###" % infile # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # def process_post_ctruncate(self,status): self.ndatasets_processed += 1 if status.get('finishStatus') == CPluginScript.FAILED: self.ndatasets_failed += 1 print "Datasets processed so far: ",self.ndatasets_processed print "Datasets failed so far: ",self.ndatasets_failed if self.ndatasets_failed > 0: self.fatalError = [203, 'Ctruncate failed', status] self.process_finish(CPluginScript.FAILED) return #Catenate Ctruncate output import CCP4Utils latestCTruncateEtree = CCP4Utils.openFileToEtree(self.ctruncate.makeFileName('PROGRAMXML')) self.collectedCtruncateEtree.append(latestCTruncateEtree) # the last dataset so far crystalDatasedId = latestCTruncateEtree.xpath('//CrystalDatasetId')[-1].text self.ctruncateOutputDataObject.annotation = crystalDatasedId #print "crystalDatasedId",crystalDatasedId # # this is called by each ctruncate process # on the last time, proceed to cad or terminate ## todo: cad step not implemented yet if self.ndatasets_processed == self.ndatasets: #for indx in range(len(self.container.outputData.HKLOUT)): # self.container.outputData.HKLOUT[indx].annotation = 'Observed data set '+str(indx+1) self.process_finish(CPluginScript.SUCCEEDED) # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # def process_finish(self,status): import os,shutil import CCP4Utils print "process_finish" xmlout = str( self.makeFileName( 'PROGRAMXML' ) ) xmlroot = etree.Element("AIMLESS_PIPE") # fatalError may contain a list of [errornumber, message, status] # for failures in Pointless or Aimless if self.fatalError is not None: errorxml = etree.Element('PIPELINE_ERROR') errorxml.text = self.fatalError[1] xmlroot.append(errorxml) try: pointlessxml = CCP4Utils.openFileToEtree(self.pointless.makeFileName( 'PROGRAMXML' ) ) except: pointlessxml = etree.Element('POINTLESS') xmlroot.append (pointlessxml) try: aimlessxml = CCP4Utils.openFileToEtree(self.aimless.makeFileName( 'PROGRAMXML' ) ) except: aimlessxml = etree.Element('AIMLESS') xmlroot.append (aimlessxml) try: xmlroot.append (self.collectedCtruncateEtree) except: pass if (self.fatalError is None) and not self.container.controlParameters.ANALYSIS_MODE: freeStatus, freerReportXML = self.runFreerflag() #Add freerflag status to the xmlout - this can be the usual SUCCEEDED,FAILED or #is 100 if failure in the mtzjoin stage (likely due to space group mismatch) e = etree.Element('FREERFLAG') e1= etree.Element('status') e1.text=str(freeStatus) e.append(e1) if freerReportXML is not None: e.append(freerReportXML) xmlroot.append(e) newXml = etree.tostring(xmlroot,pretty_print=True) xmlfile = open( xmlout, "w" ) xmlfile.write (newXml) xmlfile.close() # Populate the performance indicator try: eList = xmlroot.xpath('AIMLESS/ReflectionFile/SpacegroupName') print 'aimless_pipe.process_finish spacegroup',str(eList[0].text).strip() spGp = self.container.outputData.PERFORMANCE.spaceGroup if len(eList)>0: spGp.set(spGp.fix(str(eList[0].text).strip())) except: pass eList = xmlroot.xpath('AIMLESS/Result/Dataset/RmeasOverall/Overall') if len(eList)>0: self.container.outputData.PERFORMANCE.rMeas.set(float(str(eList[0].text).strip())) eList = xmlroot.xpath('AIMLESS/Result/Dataset/ResolutionHigh/Overall') if len(eList)>0: self.container.outputData.PERFORMANCE.highResLimit.set(float(str(eList[0].text).strip())) # Emit finished signal with whatever status has been emitted # by the last wrapper print 'aimless_pip.process_finish' if status == CPluginScript.FAILED: self.reportStatus(CPluginScript.UNSATISFACTORY) else: self.reportStatus(status) # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # def formatCellLength(self, p): return "%7.1f" % float(p) #return "%7.2f" % float(p) # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # def formatCellAngle(self, p): if float(p) < 10.0: return "%7.1f" % (float(p) * 57.29577951308233) else: return "%7.1f" % float(p) #return "%7.1f" % (float(p) * 57.29577951308233) #return "%7.2f" % (float(p) * 57.29577951308233) # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # def shortformatCell(self, cell): #print 'shortformatCell',cell s = "" s += self.formatCellLength(cell.a).strip()+', ' s += self.formatCellLength(cell.b).strip()+', ' s += self.formatCellLength(cell.c).strip()+', ' s += self.formatCellAngle(cell.alpha).strip()+', ' s += self.formatCellAngle(cell.beta).strip()+', ' s += self.formatCellAngle(cell.gamma).strip() return s # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # def addElement(self, containerXML, elementname, elementtext): e2 = etree.Element(elementname) e2.text = elementtext containerXML.append(e2) # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # def cellCompatibilityXML(self,mtzContent1 ,mtzContent2, cellsAreTheSame): "Make XML report of cells and their compatibility" sgnameObs = 'Unk' sgnameFree = 'Unk' if mtzContent1.spaceGroup.isSet(): sgnameObs = mtzContent1.spaceGroup.__str__() if mtzContent2.spaceGroup.isSet(): sgnameFree = mtzContent2.spaceGroup.__str__() """ cellFsigF, cellFree cells highres, highresFree high resolution in A for data and free set cellsAreTheSame result dictionary from SameCell 'validity' True if cells are simlar within resolution of tolerance 'maximumResolution1' maximum allowed resolution in cell1 'maximumResolution2' maximum allowed resolution in cell2 'difference' average cell difference in A 'tolerance' in A """ #print "freeCellCompatibilityXML" #print "cellsAreTheSame", cellsAreTheSame #print "tolerance", cellsAreTheSame['tolerance'] #print "sgnames", sgnameObs, sgnameFree cellReportXML= etree.Element('ObsFreeCellComparison') self.addElement(cellReportXML, 'cellObserved',self.shortformatCell(mtzContent1.cell)) self.addElement(cellReportXML, 'cellFreeR', self.shortformatCell(mtzContent2.cell)) self.addElement(cellReportXML,'tolerance', ("%7.2f" % cellsAreTheSame['tolerance']).strip()) # Validity is true if cells are within tolerance self.addElement(cellReportXML, 'validity', str(cellsAreTheSame['validity'])) # Cell difference self.addElement(cellReportXML, 'CellDifference', ("%7.2f" % cellsAreTheSame['difference']).strip()) # Maximum resolution that extension from FreeR set would be valid # calculated from FreeR cell to data cell self.addElement(cellReportXML, 'MaxAcceptableResolution', ("%7.2f" % cellsAreTheSame['maximumResolution2']).strip()) self.addElement(cellReportXML, 'SGnameObserved', sgnameObs) self.addElement(cellReportXML, 'SGnameFreeR', sgnameFree) print "cellReportXML",etree.tostring(cellReportXML,pretty_print=True) return cellReportXML # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # def runFreerflag(self): import freerflag freerReportXML = None self.freerflag = self.makePluginObject('freerflag') self.freerflag.container.inputData.F_SIGF = self.container.outputData.HKLOUT[0] self.freerflag.container.inputData.FREERFLAG = self.container.inputData.FREERFLAG self.freerflag.container.controlParameters.COMPLETE = self.container.controlParameters.COMPLETE self.freerflag.container.controlParameters.FRAC = \ self.container.controlParameters.FREER_FRACTION filePath = os.path.join(self.workDirectory,'FREERFLAG.mtz') self.freerflag.container.outputData.FREEROUT.setFullPath(filePath) # Set optional COMPLETE flag: GEN_MODE = 'GEN_NEW' or 'COMPLETE' self.freerflag.container.controlParameters.GEN_MODE = 'GEN_NEW' if self.freerflag.container.inputData.FREERFLAG.isSet(): # Extending existing FreeR self.freerflag.container.controlParameters.GEN_MODE = 'COMPLETE' print 'calling cellsAreTheSame' # # TEMPORARY FIX UNTIL SameCell is FIXED ## cellsAreTheSame = True cellsAreTheSame = self.freerflag.container.inputData.F_SIGF.fileContent.clipperSameCell(self.freerflag.container.inputData.FREERFLAG.fileContent) print 'cellsAreTheSame', cellsAreTheSame # Make XML block to report cells and their compatibility freerReportXML = self.cellCompatibilityXML(self.freerflag.container.inputData.F_SIGF.fileContent, self.freerflag.container.inputData.FREERFLAG.fileContent, cellsAreTheSame) if (not self.container.controlParameters.OVERRIDE_CELL_DIFFERENCE) and \ (not cellsAreTheSame['validity']): # Not compatible status = CPluginScript.FAILED return status, freerReportXML else: # Generating new FreeR # Fraction for FreeR if specified if self.container.controlParameters.FREER_FRACTION.isSet(): frac = self.container.controlParameters.FREER_FRACTION print "FreeR fraction", type(frac), frac freerReportXML = etree.Element('FreeRfraction') freerReportXML.text = frac.__str__() status = self.freerflag.process() if status == CPluginScript.SUCCEEDED: self.container.outputData.FREEROUT.setFullPath(filePath) print "FREEROUT",self.container.outputData.FREEROUT.fileContent if self.container.outputData.FREEROUT.fileContent.spaceGroup.isSet(): sgname = self.container.outputData.FREEROUT.fileContent.spaceGroup.__str__() else: sgname = 'Unk' highresFRformatted = "%7.2f" % float(self.container.outputData.FREEROUT.fileContent.resolutionRange.high) title ='FreeR - Spg:'+str(sgname).strip()+';Resln:'+highresFRformatted.strip() + "A;" try: title = title + "Cell:"+self.container.outputData.FREEROUT.fileContent.cell.guiLabel() except Exception as e: print 'Error writing cell parameters',e print "Title:",title self.container.outputData.FREEROUT.annotation = title else: nMergeFail = self.freerflag.getErrorReport().count(cls=CPluginScript,code=32) print "nMergeFail", nMergeFail if nMergeFail>0: return 100, freerReportXML return status, freerReportXML # ---------------------------------------------------------------------- # Function to return list of names of exportable MTZ(s) def exportJobFile(jobId=None,mode=None): import os import CCP4Modules import CCP4XtalData jobDir = CCP4Modules.PROJECTSMANAGER().jobDirectory(jobId=jobId,create=False) exportFile = os.path.join(jobDir,'exportMtz.mtz') if os.path.exists(exportFile): return exportFile childJobs = CCP4Modules.PROJECTSMANAGER().db().getChildJobs(jobId=jobId,details=True) #print 'aimless.exportMtz',childJobs truncateOut = None freerflagOut = None for jobNo,subJobId,taskName in childJobs: if taskName == 'ctruncate': truncateOut = os.path.join( CCP4Modules.PROJECTSMANAGER().jobDirectory(jobId=subJobId,create=False),'HKLOUT.mtz') if not os.path.exists(truncateOut): truncateOut = None elif taskName == 'freerflag': freerflagOut = os.path.join( CCP4Modules.PROJECTSMANAGER().jobDirectory(jobId=jobId,create=False),'FREERFLAG.mtz') if not os.path.exists(freerflagOut): freerflagOut = None if truncateOut is None: return None if freerflagOut is None: return truncateOut print 'aimless_pipe.exportJobFile runCad:',exportFile,[ freerflagOut ] m = CCP4XtalData.CMtzDataFile(truncateOut) #print m.runCad.__doc__ #Print out docs for the function outfile,err = m.runCad(exportFile,[ freerflagOut ] ) print 'aimless_pipe.exportJobFile',outfile,err.report() return outfile def exportJobFileMenu(jobId=None): # Return a list of items to appear on the 'Export' menu - each has three subitems: # [ unique identifier - will be mode argument to exportJobFile() , menu item , mime type (see CCP4CustomMimeTypes module) ] return [ [ 'complete_mtz' ,'MTZ file' , 'application/CCP4-mtz' ] ] # --------------------------------------------------------------------- import clipper import math # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # class SameCell: "Check if two cells are the same" def __init__(self): self.ccell1 = None # clipper cells self.ccell2 = None self.tolerance = None # . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . def compareCellsinFileContent(self, filecontent1cell, filecontent2cell, tolerance=None): "Compare cell from two fileContent cell objects" """ Return dictionary of: 'validity' True if cells are simlar within resolution of tolerance 'maximumResolution1' maximum allowed resolution in cell1 'maximumResolution2' maximum allowed resolution in cell2 'difference' average cell difference in A 'tolerance' in A """ if tolerance is not None: self.tolerance = tolerance cell1 = self.extractCell(filecontent1cell.get()) cell2 = self.extractCell(filecontent2cell.get()) return self.computeDiffs(cell1, cell2) # . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . def compareCells(self, gcell1, gcell2, tolerance=None): "Compare two cell dictionaries (entry for testing)" if tolerance is not None: self.tolerance = tolerance cell1 = self.extractCell(gcell1) cell2 = self.extractCell(gcell2) return self.computeDiffs(cell1, cell2) # . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . def computeDiffs(self, cell1, cell2): "cell1, cell2 are arrays of cell dimensions" # store clipper cells self.ccell1 = self.makeCell(cell1[0], cell1[1], cell1[2], cell1[3], cell1[4], cell1[5]) self.ccell2 = self.makeCell(cell2[0], cell2[1], cell2[2], cell2[3], cell2[4], cell2[5]) return self.computeResults() # . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . def makeCell(self, a, b, c, alpha, beta, gamma): "Return clipper::Cell from cell dimensions, angles in degress or radians" return clipper.Cell(clipper.Cell_descr(a, b, c, alpha, beta, gamma)) # . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . def computeResults(self): "Compute results, return as dictionary" #print "computeResults" recipdifference = self.reciprocalcellDifference() difference = self.cellDifference() if self.tolerance is None: self.tolerance = 1.0 #print "tolerance", self.tolerance equals = self.ccell1.equals(self.ccell2, self.tolerance) # Boolean #print "Result:",equals, difference, recipdifference """ Return dictionary of: 'validity' True if cells are simlar within resolution of tolerance 'maximumResolution1' maximum allowed resolution in cell1 'maximumResolution2' maximum allowed resolution in cell2 'difference' average cell difference in A 'tolerance' in A """ result = {'validity': equals, 'maximumResolution1': recipdifference[0], 'maximumResolution2': recipdifference[1], 'difference': difference, 'tolerance': self.tolerance} return result # . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . def extractCell(self, celldict): "extract cell dimensions, return array" cell = [] for item in ['a','b','c','alpha','beta', 'gamma']: cell.append(float(celldict[item])) return cell # . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . def reciprocalcellDifference(self): "Get resolution at which ccell2 is different from ccell1, two values" fracmat1 = self.ccell1.matrix_frac() fracmat2 = self.ccell2.matrix_frac() s = 0.0 for j in range(3): for i in range(3): s += (fracmat1[i][j] - fracmat2[i][j])**2 s = math.sqrt(s) volume1 = self.ccell1.volume() volume2 = self.ccell2.volume() return s * math.pow(volume1, 0.66666666667), \ s * math.pow(volume2, 0.66666666667) # . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . def cellDifference(self): "Get resolution at which self.ccell2 is different from self.ccell1" orthmat1 = self.ccell1.matrix_orth() orthmat2 = self.ccell2.matrix_orth() #print "orth1\n", orthmat1.format() #print "orth2\n", orthmat2.format() s = 0.0 for j in range(3): for i in range(3): s += (orthmat1[i][j] - orthmat2[i][j])**2 return math.sqrt(s) # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #====================================================== # PLUGIN TESTS # See Python documentation on unittest module import unittest class testaimless_pipe(unittest.TestCase): def setUp(self): import CCP4Modules self.app = CCP4Modules.QTAPPLICATION() # make all background jobs wait for completion # this is essential for unittest to work CCP4Modules.PROCESSMANAGER().setWaitForFinished(10000) def tearDown(self): import CCP4Modules CCP4Modules.PROCESSMANAGER().setWaitForFinished(-1) def test_1(self): import CCP4Modules, CCP4Utils, os workDirectory = os.path.join(CCP4Utils.getTestTmpDir(),'test1') if not os.path.exists(workDirectory): os.mkdir(workDirectory) self.wrapper = aimless_pipe(parent=CCP4Modules.QTAPPLICATION(),name='test1',workDirectory=workDirectory) self.wrapper.container.loadDataFromXml(os.path.join(CCP4Utils.getCCP4I2Dir(),'pipelines','aimless_pipe','test_data','test1.data.xml')) self.wrapper.setWaitForFinished(1000000) pid = self.wrapper.process() self.wrapper.setWaitForFinished(-1) if len(self.wrapper.errorReport)>0: print self.wrapper.errorReport.report() def TESTSUITE(): suite = unittest.TestLoader().loadTestsFromTestCase(testaimless_pipe) return suite def testModule(): suite = TESTSUITE() unittest.TextTestRunner(verbosity=2).run(suite)