Changeset 33415 for branches/meh_branches/ppstack_test/ippToPsps/jython/batch.py

Timestamp:

Mar 5, 2012, 5:19:48 PM (14 years ago)

Author:

mhuber

Message:

merging latest r33407 trunk changes to meh_branches/ppstack_test

Location:

branches/meh_branches/ppstack_test

Files:

• . (modified) (1 prop)
• ippToPsps/jython/batch.py (modified) (19 diffs)

branches/meh_branches/ppstack_test/ippToPsps/jython/batch.py

@@ -21 +21 @@
 
 '''
-Base class of all batch types.
+Abstract base class of all batch types.
 '''
 class Batch(object):
@@ -27 +27 @@
     '''
     Constructor
-
-    >>> batch = Batch(1,2,3,4,5,6,7)
-    >>> print batch.pspsVoTableFilePath
-    "../config/2/tables.vot"
     '''
     def __init__(self, 
                  logger, 
-                 configPath,
-                 doc,
+                 config,
                  gpc1Db,
                  ippToPspsDb,
+                 scratchDb,
                  id,
+                 batchID,
                  batchType, 
-                 fits): 
+                 fits,
+                 useFullTables): 
 
         self.everythingOK = False
         self.readHeader = False
-        self.configPath = configPath
-        self.doc = doc
+        self.config = config
         self.fits = fits
-        if self.fits: self.header = self.fits.getPrimaryHeader()
+        self.useFullTables = useFullTables
 
         # set up logging
@@ -56 +53 @@
         # set up class variables
         self.id = id
+        self.batchID = batchID
         self.gpc1Db = gpc1Db
         self.ippToPspsDb = ippToPspsDb
+        self.scratchDb = scratchDb
         self.batchType = batchType;
         self.pspsVoTableFilePath = "../config/" + batchType + "/tables.vot"
 
-        # get info from config
-        self.survey = self.doc.find("options/survey").text
-        self.pspsSurvey = self.doc.find("options/pspsSurvey").text
-        dvoGpc1Label = self.doc.find("dvo/gpc1Label").text
-        self.dvoLocation = self.doc.find("dvo/location").text
-        self.useFullTables = int(self.doc.find("dvo/useFullTables").text)
-        self.scratchDb = ScratchDb(logger, self.doc, self.useFullTables)
-
-        if not self.scratchDb.everythingOK: return
+        # check FITS file is ok
+        if self.fits: 
+            self.header = self.fits.getPrimaryHeader()
+
+            # now check that the fits header is readable
+            if not self.header: 
+                logger.errorPair("Could not read FITS for id", "%d" % id)
+                return
+
+        # if no fits file, and this is not an IN batch, then get out of here
+        else:
+           if batchType != "IN": 
+               logger.errorPair("Could not read FITS for id", "%d" % id)
+               return
+
+        self.scratchDb.setUseFullTables(self.useFullTables)
 
         # TODO
         self.tablesToExport = []
 
-        if self.survey != "":
-            self.surveyID = self.scratchDb.getSurveyID(self.survey)
+        if self.config.survey != "":
+            self.surveyID = self.scratchDb.getSurveyID(self.config.survey)
         else:
             self.surveyID = -1;
-
        
-        # get some options from the config
-        self.testMode = int(self.doc.find("options/testMode").text)
-        self.dataRelease = int(self.doc.find("metadata/dataRelease").text)
-
         # create datastore object
-        self.datastore = Datastore(self.logger, self.doc)
-
-        # create a new batch
-        self.batchID = self.ippToPspsDb.createNewBatch( 
-                self.batchType, 
-                self.id,
-                self.survey,
-                dvoGpc1Label, 
-                self.datastore.product)
+        self.datastore = Datastore(self.logger, self.config, self.ippToPspsDb)
 
         # get local storage location from config
         self.batchName = Batch.getNameFromID(self.batchID)
-        basePath = self.doc.find("localOutPath").text
         self.subDir = Batch.getSubDir(
-                basePath,
+                self.config.basePath,
                 self.batchType, 
-                dvoGpc1Label)
+                self.config.dvoLabel)
 
         self.localOutPath = Batch.getOutputPath(
-                basePath,
+                self.config.basePath,
                 self.batchType, 
-                dvoGpc1Label,
+                self.config.dvoLabel,
                 self.batchID)
 
@@ -121 +113 @@
         self.logger.infoTitle("New " + self.batchType + " batch")
         self.logger.infoPair("Batch name", self.batchName)
-        self.logger.infoPair("Survey", self.survey)
         self.logger.infoPair("Survey ID", "%d" % self.surveyID)
-        self.logger.infoPair("Publishing to PSPS as survey", self.pspsSurvey)
-        self.logger.infoPair("DVO location", self.dvoLocation)
         self.logger.infoBool("Use full DVO tables?", self.useFullTables)
 
@@ -157 +146 @@
 
     '''
+    Static method to get name of tar file
+    '''
+    @staticmethod
+    def getTarFile(id):
+        return Batch.getNameFromID(id) + ".tar"
+
+    '''
+    Static method to get path of tar file
+    '''
+    @staticmethod
+    def getTarPath(basePath, batchType, dvoLabel, id):
+        return Batch.getSubDir(basePath, batchType, dvoLabel) + "/" + Batch.getTarFile(id)
+
+    '''
+    Static method to get name of tarball file
+    '''
+    @staticmethod
+    def getTarballFile(id):
+        return Batch.getTarFile(id) + ".gz"
+
+    '''
+    Static method to get path of tarball file
+    '''
+    @staticmethod
+    def getTarballPath(basePath, batchType, dvoLabel, id):
+        return Batch.getSubDir(basePath, batchType, dvoLabel) + "/" + Batch.getTarballFile(id)
+
+    '''  
+    Static method to delete this batch from local disk
+    '''
+    @staticmethod
+    def deleteFromDisk(logger, basePath, batchType, dvoLabel, id):
+       
+        tarballPath = Batch.getTarballPath(basePath, batchType, dvoLabel, id)
+        dirPath = Batch.getOutputPath(basePath, batchType, dvoLabel, id)
+
+        if not os.path.exists(tarballPath): 
+            logger.errorPair("Could not find", tarballPath)
+            return 1
+
+        try:
+            os.remove(tarballPath)
+            logger.infoPair("Deleted", tarballPath)
+        except:
+            logger.errorPair("Could not delete", tarballPath)
+            return 0
+   
+        return 1
+
+    '''
     Destructor
     '''
@@ -173 +212 @@
          if value != "NULL": return value
          else:
-             if not self.testMode: return "NULL"
+             if not self.config.test: return "NULL"
              header[key] = default
              self.logger.infoPair("Hardcoding " + key + " to", header[key])
@@ -194 +233 @@
 
         outPath = self.localOutPath + "/BatchManifest.xml"
-        tmpPath = "./tmp.xml"
+        tmpPath = self.localOutPath + "/tmp.xml"
         self.logger.infoPair("Creating manifest", outPath)
         root = Element('manifest')
@@ -202 +241 @@
         root.attrib['type'] = self.batchType
         root.attrib['timestamp'] = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") 
-        if self.batchType != "IN": root.attrib['survey'] = self.pspsSurvey
+        if self.batchType != "IN": root.attrib['survey'] = self.config.pspsSurvey
         try: self.minObjID
         except: pass
@@ -210 +249 @@
         else: root.attrib['maxObjId'] = str(self.maxObjID)
 
-        # get md5sum
-        p = Popen("md5sum " + self.outputFitsPath, shell=True, stdout=PIPE)
-        p.wait()
-        out = p.stdout.read()
-        md5sum = out[0:out.rfind(" ")]
-
-        # get file size
-        fileSize = os.path.getsize(self.outputFitsPath)
-
         # file information
         child = Element('file')
         root.append(child)
         child.attrib['name'] = self.outputFitsFile
-        child.attrib['bytes'] = str(fileSize)
-        child.attrib['md5'] = md5sum
 
         # now create doc and write to file
@@ -240 +268 @@
     Publishes this batch to the datastore
     '''
-    def publishToDatastore(self):
+    def tarAndZip(self):
       
         # set up filenams and paths
-        tarFile = self.batchName + ".tar"
-        tarPath = self.subDir + "/" + tarFile
-
-        tarballFile = tarFile + ".gz"
-        tarballPath = self.subDir + "/" + tarballFile
+        tarFile = Batch.getTarFile(self.batchID)
+        tarPath = Batch.getTarPath(self.config.basePath, self.batchType, self.config.dvoLabel, self.batchID)
+        tarballFile = Batch.getTarballFile(self.batchID)
+        tarballPath = Batch.getTarballPath(self.config.basePath, self.batchType, self.config.dvoLabel, self.batchID)
 
         # tar directory
-        p = Popen("tar -cvf " + tarPath + "\
-                -C " + self.subDir + " \
-                " + self.batchName, shell=True, stdout=PIPE)
+        cmd = "tar -cvf " + tarPath + " -C " + self.subDir + " " + self.batchName
+        self.logger.infoPair("Creating tar archive", cmd)
+        p = Popen(cmd, shell=True, stdout=PIPE)
         p.wait()
 
+        if p.returncode != 0:
+            self.logger.errorPair("tar command", "failed")
+            return False
+
         # zip tar archive
-        p = Popen("gzip -c " + tarPath + " > " + tarballPath, shell=True, stdout=PIPE)
+        cmd = "gzip -c " + tarPath + " > " + tarballPath
+        self.logger.infoPair("Compressing tar archive", cmd)
+        p = Popen(cmd, shell=True, stdout=PIPE)
         p.wait()
+
+        if p.returncode != 0:
+            self.logger.errorPair("gzip command", "failed")
+            return False
+
+        # only now can we report that the batch has successfully processed
+        self.ippToPspsDb.updateProcessed(self.batchID, 1)
 
         # delete tar file and original directory
@@ -263 +303 @@
         shutil.rmtree(self.localOutPath)
 
-        # now publish to the datastore
-        if self.datastore.publish(self.batchName, self.subDir, tarballFile, "tgz"):
-            self.ippToPspsDb.updateLoadedToDatastore(self.batchID, 1)
+        return True
+
+    '''
+    Static method to publish a batch to the datastore
+    '''
+    @staticmethod
+    def publishToDatastore(datastore, batchID, batchName, subDir, tarballFile):
+        datastore.publish(batchName, subDir, tarballFile, "tgz")
 
     '''
@@ -293 +338 @@
             rs.close()
 
-        self.ippToPspsDb.updateMinMaxObjID(self.batchID, self.minObjID, self.maxObjID)
+        self.ippToPspsDb.updateDetectionStats(self.batchID, self.minObjID, self.maxObjID, self.totalDetections)
         self.logger.infoPair("Total detections", "%ld" % self.totalDetections)
         self.logger.infoPair("Min objID", "%ld" % self.minObjID)
@@ -309 +354 @@
              self.tablesToExport.append(table.name)
 
-         self.alterPspsTables();
+         return self.alterPspsTables();
 
     '''
@@ -329 +374 @@
     Accepts a regular expression filter so not all tables need to be imported
     '''
-    def importIppTables(self, filter=""):
+    def importIppTables(self, columns="*", filter=""):
 
       self.logger.infoPair("Importing tables with filter", filter)
-      tables = stilts.treads(self.fits.getPath())
+
+      try:
+          tables = stilts.treads(self.fits.getPath())
+      except:
+          self.logger.errorPair("STILTS could not import from", self.fits.getPath())
+          return False
 
       count = 0
@@ -347 +397 @@
           # IPP FITS files are littered with infinities, so remove these
           self.logger.debug("Removing Infinity values from all columns")
+          table = stilts.tpipe(table, cmd='keepcols "' + columns + '"')
           table = stilts.tpipe(table, cmd='replaceval -Infinity null *')
           table = stilts.tpipe(table, cmd='replaceval Infinity null *')
@@ -396 +447 @@
         try:
             stilts.twrites(_tables, self.outputFitsPath, fmt='fits')
-            self.ippToPspsDb.updateProcessed(self.batchID, 1)
         except:
             self.logger.exception("Could not write to FITS")
+            self.ippToPspsDb.updateProcessed(self.batchID, -1)
             return False
 
@@ -432 +483 @@
     def getIDsFromDVO(self):
 
-        if self.scratchDb.getRowCount("dvoMeta") < 1:
-            self.logger.error("No DVO IDs found in dvoMeta")
+        if self.scratchDb.getRowCount(self.scratchDb.dvoImagesTable) < 1:
+            self.logger.errorInfo("No DVO images found in", self.scratchDb.dvoImagesTable)
             return False
 
         # TODO path to DVO prog hardcoded temporarily
-        cmd = "../src/dvograbber " + self.configPath + " " + self.dvoLocation
+        cmd = "../src/dvograbber " + self.config.settingsPath + " " + self.scratchDb.dbName + " " + self.config.dvoLocation
         self.logger.infoPair("Running DVO", cmd)
         p = Popen(cmd, shell=True, stdout=PIPE)
         p.wait()
-        #        out = p.stdout.read()
 
         rowCount = self.scratchDb.getRowCount("dvoDetection")
@@ -453 +503 @@
 
     '''
-    Checks whether this batch has already been processed and published. To be implemented by all subclasses
-    '''
-    def alreadyProcessed(self):
-        self.logger.warn("alreadyProcessed not implemented")
-
-
-    '''
     Creates and publishes a batch
+    TODO all methods call below should throw exceptions on failure
     '''
     def run(self):
@@ -466 +510 @@
         if not self.everythingOK:
             self.logger.error("Aborting this batch due to (hopefully) previously reported errors")
+            self.ippToPspsDb.updateProcessed(self.batchID, -1)
             return
 
-        self.createEmptyPspsTables()
+        if not self.createEmptyPspsTables():
+            self.logger.error("Aborting this batch due to (hopefully) previously reported errors")
+            self.ippToPspsDb.updateProcessed(self.batchID, -1)
+            return
+
         self.importIppTables()
-        if self.populatePspsTables():
-            if self.exportPspsTablesToFits():
+        if not self.populatePspsTables():
+            self.logger.errorPair("Aborting this batch", "unable to populate PSPS tables")
+            self.ippToPspsDb.updateProcessed(self.batchID, -1)
+        else:
+            if not self.exportPspsTablesToFits():
+                self.logger.errorPair("Aborting this batch", "unable to export tables to FITS file")
+            else:
                 self.writeBatchManifest()
-                if int(self.doc.find("options/publishToDatastore").text): self.publishToDatastore()
-                if int(self.doc.find("options/reportNulls").text): self.reportNullsInAllPspsTables(False)
-
-        self.logger.infoPair("Batch......", "complete")
+                if self.config.datastorePublishing: 
+                    # tar and zip ready for publication to datastore
+                    if self.tarAndZip():
+                        # now publish to the datastore
+                        tarballFile = Batch.getTarballFile(self.batchID)
+                        Batch.publishToDatastore(self.datastore, self.batchID, self.batchName, self.subDir, tarballFile)
 
 from datastore import Datastore