IPP Software Navigation Tools IPP Links Communication Pan-STARRS Links

Ignore:
Timestamp:
Sep 6, 2011, 11:00:22 AM (15 years ago)
Author:
eugene
Message:

merge changes from trunk

Location:
branches/eam_branches/ipp-20110710/ippToPsps/jython
Files:
2 edited

Legend:

Unmodified
Added
Removed

  • branches/eam_branches/ipp-20110710/ippToPsps/jython

    • Property svn:ignore set to
      *.class

  • branches/eam_branches/ipp-20110710/ippToPsps/jython/batch.py

    r31840 r32337  
    1111from subprocess import call, PIPE, Popen
    1212
    13 from datastore import Datastore
    1413from scratchdb import ScratchDb
    1514from gpc1db import Gpc1Db
    … …  
    3534    def __init__(self,
    3635                 logger,
     36                 configPath,
    3737                 doc,
    3838                 gpc1Db,
    … …  
    4040                 id,
    4141                 batchType,
    42                  fits,
    43                  survey=""):
     42                 fits):
    4443
    4544        self.everythingOK = False
    4645        self.readHeader = False
     46        self.configPath = configPath
    4747        self.doc = doc
    4848        self.fits = fits
    49         self.header = self.fits.getPrimaryHeader()
    5049
    5150        # set up logging
    … …  
    5352        self.logger.infoSeparator()
    5453        self.logger.debug("Batch class constructor")
     54
     55        # check FITS file is ok - TODO could be neater
     56        if self.fits:
     57            self.header = self.fits.getPrimaryHeader()
     58            if not self.header: return
    5559
    5660        # set up class variables
    … …  
    6064        self.batchType = batchType;
    6165        self.pspsVoTableFilePath = "../config/" + batchType + "/tables.vot"
    62         self.survey = survey
    63 
    64         # get dvo info from config
    65         dvoName = self.doc.find("dvo/name").text
     66
     67        # get info from config
     68        self.survey = self.doc.find("options/survey").text
     69        self.pspsSurvey = self.doc.find("options/pspsSurvey").text
     70        self.dvoGpc1Label = self.doc.find("dvo/gpc1Label").text
    6671        self.dvoLocation = self.doc.find("dvo/location").text
    6772        self.useFullTables = int(self.doc.find("dvo/useFullTables").text)
    6873        self.scratchDb = ScratchDb(logger, self.doc, self.useFullTables)
     74
     75        if not self.scratchDb.everythingOK: return
    6976
    7077        # TODO
    … …  
    7582        else:
    7683            self.surveyID = -1;
    77 
    7884       
    7985        # get some options from the config
    … …  
    8187        self.dataRelease = int(self.doc.find("metadata/dataRelease").text)
    8288
    83         # get datastore info from config
    84         self.datastore = Datastore(self.logger)
     89        # create datastore object
     90        self.datastore = Datastore(self.logger, self.doc)
    8591
    8692        # create a new batch
    8793        self.batchID = self.ippToPspsDb.createNewBatch(
    88                 self.getPspsBatchType(),
     94                self.batchType,
    8995                self.id,
    90                 survey,
    91                 dvoName,
     96                self.survey,
     97                self.dvoGpc1Label,
    9298                self.datastore.product)
    9399
    94100        # get local storage location from config
    95         self.batchName = "B%08d" % self.batchID
    96         self.subDir = self.doc.find("localOutPath").text + "/" + self.getPspsBatchType() + "/" + dvoName
    97         self.localOutPath = self.subDir + "/" + self.batchName
     101        self.batchName = Batch.getNameFromID(self.batchID)
     102        self.basePath = self.doc.find("localOutPath").text
     103        self.subDir = Batch.getSubDir(
     104                self.basePath,
     105                self.batchType,
     106                self.dvoGpc1Label)
     107
     108        self.localOutPath = Batch.getOutputPath(
     109                self.basePath,
     110                self.batchType,
     111                self.dvoGpc1Label,
     112                self.batchID)
     113
    98114        if not os.path.exists(self.localOutPath): os.makedirs(self.localOutPath)
    99115
    … …  
    105121        if not self.useFullTables: self.scratchDb.createDvoTables()
    106122
    107         self.everythingOK = True
    108 
    109     '''
    110     Destructor
    111     '''
    112     def __del__(self):
    113 
    114         self.logger.debug("Batch destructor")
    115 
    116     '''
    117     Prints metadata to the log
    118     '''
    119     def printMe(self):
    120 
     123        # dump stuff to the log
     124        self.logger.infoTitle("New " + self.batchType + " batch")
    121125        self.logger.infoPair("Batch name", self.batchName)
    122126        self.logger.infoPair("Survey", self.survey)
    123127        self.logger.infoPair("Survey ID", "%d" % self.surveyID)
     128        self.logger.infoPair("Publishing to PSPS as survey", self.pspsSurvey)
    124129        self.logger.infoPair("DVO location", self.dvoLocation)
    125         self.logger.infoPair("Use full DVO tables?", "%d" % self.useFullTables)
     130        self.logger.infoBool("Use full DVO tables?", self.useFullTables)
     131
     132        if self.fits:
     133            self.logger.infoPair("Input FITS file", self.fits.getOriginalPath())
     134            self.logger.infoPair("Input FITS primary header", "%s cards found" % self.fits.getPrimaryHeaderCardCount())
     135
    126136        self.logger.infoPair("Output path", self.localOutPath)
    127137
     138        self.everythingOK = True
     139   
     140    '''
     141    Static method to generated batch name from batch ID
     142    '''
     143    @staticmethod
     144    def getNameFromID(id):
     145       return "B%08d" % id
     146
     147    '''
     148    Static method to generated sub directory for batch on local disk
     149    '''
     150    @staticmethod
     151    def getSubDir(basePath, batchType, dvoLabel):
     152       return basePath + "/" + batchType + "/" + dvoLabel
     153
     154    '''
     155    Static method to generated output path for batch
     156    '''
     157    @staticmethod
     158    def getOutputPath(basePath, batchType, dvoLabel, id):
     159       return Batch.getSubDir(basePath, batchType, dvoLabel) + "/" + Batch.getNameFromID(id)
     160
     161    '''
     162    Static method to get name of tar file
     163    '''
     164    @staticmethod
     165    def getTarFile(id):
     166        return Batch.getNameFromID(id) + ".tar"
     167
     168    '''
     169    Static method to get path of tar file
     170    '''
     171    @staticmethod
     172    def getTarPath(basePath, batchType, dvoLabel, id):
     173        return Batch.getSubDir(basePath, batchType, dvoLabel) + "/" + Batch.getTarFile(id)
     174
     175    '''
     176    Static method to get name of tarball file
     177    '''
     178    @staticmethod
     179    def getTarballFile(id):
     180        return Batch.getTarFile(id) + ".gz"
     181
     182    '''
     183    Static method to get path of tarball file
     184    '''
     185    @staticmethod
     186    def getTarballPath(basePath, batchType, dvoLabel, id):
     187        return Batch.getSubDir(basePath, batchType, dvoLabel) + "/" + Batch.getTarballFile(id)
     188
     189    ''' 
     190    Static method to delete this batch from local disk
     191    '''
     192    @staticmethod
     193    def deleteFromDisk(logger, basePath, batchType, dvoLabel, id):
     194       
     195        tarballPath = Batch.getTarballPath(basePath, batchType, dvoLabel, id)
     196        dirPath = Batch.getOutputPath(basePath, batchType, dvoLabel, id)
     197
     198        if not os.path.exists(tarballPath):
     199            logger.errorPair("Could not find", tarballPath)
     200            return 1
     201
     202        try:
     203            os.remove(tarballPath)
     204            logger.infoPair("Deleted", tarballPath)
     205        except:
     206            logger.errorPair("Could not delete", tarballPath)
     207            return 0
     208   
     209        return 1
     210
     211    '''
     212    Destructor
     213    '''
     214    def __del__(self):
     215
     216        self.logger.debug("Batch destructor")
     217
     218    '''
     219    Looks for a dictionary entry. If it is not there, then replaces it with the supplied default
     220    if, and only if, in test mode
     221    '''
     222    def safeDictionaryAccessWithDefault(self, header, key, default):
     223
     224         value = self.safeDictionaryAccess(header, key)
     225
     226         if value != "NULL": return value
     227         else:
     228             if not self.testMode: return "NULL"
     229             header[key] = default
     230             self.logger.infoPair("Hardcoding " + key + " to", header[key])
     231             return header[key]
     232
    128233    '''
    129234    Returns the string value from this dictionary or else "NULL"
    … …  
    132237
    133238         if key in header: return header[key]
    134          else: return "NULL"
     239         else:
     240             self.logger.errorPair("Missing header field", key)
     241             return "NULL"
    135242
    136243    '''
    … …  
    146253        # batch information
    147254        root.attrib['name'] = self.batchName
    148         root.attrib['type'] = self.getPspsBatchType()
     255        root.attrib['type'] = self.batchType
    149256        root.attrib['timestamp'] = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    150         if self.survey != "":
    151             root.attrib['survey'] = self.getBatchFriendlySurveyType()
     257        if self.batchType != "IN": root.attrib['survey'] = self.pspsSurvey
    152258        try: self.minObjID
    153259        except: pass
    … …  
    158264
    159265        # get md5sum
    160         p = Popen("md5sum " + self.outputFitsPath, shell=True, stdout=PIPE)
    161         p.wait()
    162         out = p.stdout.read()
    163         md5sum = out[0:out.rfind(" ")]
     266        #p = Popen("md5sum " + self.outputFitsPath, shell=True, stdout=PIPE)
     267        #p.wait()
     268        #out = p.stdout.read()
     269        #md5sum = out[0:out.rfind(" ")]
    164270
    165271        # get file size
    166         fileSize = os.path.getsize(self.outputFitsPath)
     272        #fileSize = os.path.getsize(self.outputFitsPath)
    167273
    168274        # file information
    … …  
    170276        root.append(child)
    171277        child.attrib['name'] = self.outputFitsFile
    172         child.attrib['bytes'] = str(fileSize)
    173         child.attrib['md5'] = md5sum
     278        #child.attrib['bytes'] = str(fileSize)
     279        #child.attrib['md5'] = md5sum
    174280
    175281        # now create doc and write to file
    … …  
    187293    Publishes this batch to the datastore
    188294    '''
    189     def publishToDatastore(self):
     295    def tarZipAndpublishToDatastore(self):
    190296     
    191297        # set up filenams and paths
    192         tarFile = self.batchName + ".tar"
    193         tarPath = self.subDir + "/" + tarFile
    194 
    195         tarballFile = tarFile + ".gz"
    196         tarballPath = self.subDir + "/" + tarballFile
     298        tarFile = Batch.getTarFile(self.batchID)
     299        tarPath = Batch.getTarPath(self.basePath, self.batchType, self.dvoGpc1Label, self.batchID)
     300        tarballFile = Batch.getTarballFile(self.batchID)
     301        tarballPath = Batch.getTarballPath(self.basePath, self.batchType, self.dvoGpc1Label, self.batchID)
    197302
    198303        # tar directory
    … …  
    211316
    212317        # now publish to the datastore
    213         if self.datastore.publish(self.batchName, self.subDir, tarballFile, "tgz"):
    214             self.ippToPspsDb.updateLoadedToDatastore(self.batchID, 1)
    215 
    216     '''
    217     Gets PSPS-friendly survey type
    218     '''
    219     def getBatchFriendlySurveyType(self):
    220 
    221         #return "SCR" # TODO
    222         try:
    223             self.survey
    224         except:
    225             return "NA"
    226 
    227         if self.survey == "3PI": return "3PI"
    228         elif self.survey == "MD04": return "MD4"
     318        Batch.publishToDatastore(self.datastore, self.ippToPspsDb, self.batchID, self.batchName, self.subDir, tarballFile)
     319
     320    '''
     321    Static method to publish a batch to the datastore
     322    '''
     323    @staticmethod
     324    def publishToDatastore(datastore, ippToPspsDb, batchID, batchName, subDir, tarballFile):
     325
     326        if datastore.publish(batchName, subDir, tarballFile, "tgz"):
     327            ippToPspsDb.updateLoadedToDatastore(batchID, 1)
    229328        else:
    230             self.logger.error("Don't know this survey: '" + self.survey + "'")
    231             return "NA"
    232 
    233     '''
    234     Gets PSPS friendly batch type TODO only use PSPS batch names, P2, ST etc
    235     '''
    236     def getPspsBatchType(self):
    237 
    238         if self.batchType == "init": return "IN"
    239         elif self.batchType == "detection": return "P2"
    240         elif self.batchType == "stack": return "ST"
    241         else: self.logger.error("Don't know this batch type: " + self.survey)
     329            ippToPspsDb.updateLoadedToDatastore(batchID, -1)
     330           
    242331
    243332    '''
    … …  
    267356            rs.close()
    268357
    269         self.ippToPspsDb.updateMinMaxObjID(self.batchID, self.minObjID, self.maxObjID)
     358        self.ippToPspsDb.updateDetectionStats(self.batchID, self.minObjID, self.maxObjID, self.totalDetections)
    270359        self.logger.infoPair("Total detections", "%ld" % self.totalDetections)
    271360        self.logger.infoPair("Min objID", "%ld" % self.minObjID)
    … …  
    283372             self.tablesToExport.append(table.name)
    284373
    285          self.alterPspsTables();
     374         return self.alterPspsTables();
    286375
    287376    '''
    … …  
    303392    Accepts a regular expression filter so not all tables need to be imported
    304393    '''
    305     def importIppTables(self, filter=""):
     394    def importIppTables(self, columns="*", filter=""):
    306395
    307396      self.logger.infoPair("Importing tables with filter", filter)
    … …  
    313402          match = re.match(filter, table.name)
    314403          if not match: continue
    315           self.logger.infoPair("Reading IPP table", table.name)
     404          self.logger.debugPair("Reading IPP table", table.name)
    316405          table = stilts.tpipe(table, cmd='explodeall')
    317406
    … …  
    321410          # IPP FITS files are littered with infinities, so remove these
    322411          self.logger.debug("Removing Infinity values from all columns")
     412          table = stilts.tpipe(table, cmd='keepcols "' + columns + '"')
    323413          table = stilts.tpipe(table, cmd='replaceval -Infinity null *')
    324414          table = stilts.tpipe(table, cmd='replaceval Infinity null *')
    … …  
    330420          except:
    331421              self.logger.exception("Problem writing table '" + table.name + "' to the database")
    332 
    333422
    334423      self.logger.infoPair("Done. Imported", "%d tables" % count)
    … …  
    374463        except:
    375464            self.logger.exception("Could not write to FITS")
     465            self.ippToPspsDb.updateProcessed(self.batchID, -1)
    376466            return False
    377467
    … …  
    400490    '''
    401491    def populatePspsTables(self):
    402         self.logger.warn("Not implemented yet")
     492        self.logger.warn("populatePspsTables not implemented yet")
    403493
    404494    '''
    … …  
    407497    def getIDsFromDVO(self):
    408498
     499        if self.scratchDb.getRowCount("dvoMeta") < 1:
     500            self.logger.error("No DVO IDs found in dvoMeta")
     501            return False
     502
    409503        # TODO path to DVO prog hardcoded temporarily
    410         cmd = "../src/dvograbber " + self.dvoLocation
     504        cmd = "../src/dvograbber " + self.configPath + " " + self.dvoLocation
    411505        self.logger.infoPair("Running DVO", cmd)
    412506        p = Popen(cmd, shell=True, stdout=PIPE)
    413507        p.wait()
    414508        #        out = p.stdout.read()
    415         self.logger.infoPair("DVO access", "complete")
    416 
    417         if self.scratchDb.getRowCount("dvoDetection") < 1:
    418             self.logger.error("No DVO IDs found")
     509
     510        rowCount = self.scratchDb.getRowCount("dvoDetection")
     511        if rowCount < 1:
     512            self.logger.errorPair("DVO error", "No IDs found")
    419513            return False
     514
     515        self.logger.infoPair("DVO access complete. Found", "%d detections" % rowCount)
    420516           
    421517        return True
    422518
    423519    '''
    424     Checks whether this batch has already been processed and published. To be implemented by all subclasses
    425     '''
    426     def alreadyProcessed(self):
    427         self.logger.error("Not implemented")
    428 
    429 
    430     '''
    431520    Creates and publishes a batch
     521    TODO all methdds call below should throw exceptions on failure
    432522    '''
    433523    def run(self):
    434524
    435         if not self.everythingOK: return
    436 
    437         self.createEmptyPspsTables()
     525        if not self.everythingOK:
     526            self.logger.error("Aborting this batch due to (hopefully) previously reported errors")
     527            self.ippToPspsDb.updateProcessed(self.batchID, -1)
     528            return
     529
     530        if not self.createEmptyPspsTables():
     531            self.logger.error("Aborting this batch due to (hopefully) previously reported errors")
     532            self.ippToPspsDb.updateProcessed(self.batchID, -1)
     533            return
     534
    438535        self.importIppTables()
    439         if self.populatePspsTables():
    440             if self.exportPspsTablesToFits():
     536        if not self.populatePspsTables():
     537            self.logger.errorPair("Aborting this batch", "unable to populate PSPS tables")
     538            self.ippToPspsDb.updateProcessed(self.batchID, -1)
     539        else:
     540            if not self.exportPspsTablesToFits():
     541                self.logger.errorPair("Aborting this batch", "unable to export tables to FITS file")
     542            else:
    441543                self.writeBatchManifest()
    442                 if int(self.doc.find("options/publishToDatastore").text): self.publishToDatastore()
     544                if int(self.doc.find("options/publishToDatastore").text): self.tarZipAndpublishToDatastore()
    443545                if int(self.doc.find("options/reportNulls").text): self.reportNullsInAllPspsTables(False)
    444546
    445         self.logger.infoPair("Batch......", "complete")
    446         self.logger.infoSeparator()
    447 
    448 
     547from datastore import Datastore
     548
Note: See TracChangeset for help on using the changeset viewer.