Changeset 33098 for branches/meh_branches/ppsub_test/ippToPsps/jython/batch.py

Timestamp:

Jan 11, 2012, 11:19:21 PM (15 years ago)

Author:

mhuber

Message:

merged latest r33093 trunk changes to meh_branches/ppsub_test

Location:

branches/meh_branches/ppsub_test

Files:

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

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

@@ -39 +39 @@
                  ippToPspsDb,
                  id,
+                 batchID,
                  batchType, 
                  fits): 
@@ -47 +48 @@
         self.doc = doc
         self.fits = fits
-        if self.fits: self.header = self.fits.getPrimaryHeader()
 
         # set up logging
@@ -56 +56 @@
         # set up class variables
         self.id = id
+        self.batchID = batchID
         self.gpc1Db = gpc1Db
         self.ippToPspsDb = ippToPspsDb
@@ -61 +62 @@
         self.pspsVoTableFilePath = "../config/" + batchType + "/tables.vot"
 
+        # 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
+
         # 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.dvoGpc1Label = self.doc.find("dvo/gpc1Label").text
         self.dvoLocation = self.doc.find("dvo/location").text
         self.useFullTables = int(self.doc.find("dvo/useFullTables").text)
@@ -78 +94 @@
         else:
             self.surveyID = -1;
-
        
         # get some options from the config
@@ -85 +100 @@
 
         # 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.doc, self.ippToPspsDb)
 
         # get local storage location from config
         self.batchName = Batch.getNameFromID(self.batchID)
-        basePath = self.doc.find("localOutPath").text
+        self.basePath = self.doc.find("localOutPath").text
         self.subDir = Batch.getSubDir(
-                basePath,
+                self.basePath,
                 self.batchType, 
-                dvoGpc1Label)
+                self.dvoGpc1Label)
 
         self.localOutPath = Batch.getOutputPath(
-                basePath,
+                self.basePath,
                 self.batchType, 
-                dvoGpc1Label,
+                self.dvoGpc1Label,
                 self.batchID)
 
@@ -157 +164 @@
 
     '''
+    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
     '''
@@ -194 +251 @@
 
         outPath = self.localOutPath + "/BatchManifest.xml"
-        tmpPath = "./tmp.xml"
+        tmpPath = self.localOutPath + "/tmp.xml"
         self.logger.infoPair("Creating manifest", outPath)
         root = Element('manifest')
@@ -210 +267 @@
         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 +286 @@
     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.basePath, self.batchType, self.dvoGpc1Label, self.batchID)
+        tarballFile = Batch.getTarballFile(self.batchID)
+        tarballPath = Batch.getTarballPath(self.basePath, self.batchType, self.dvoGpc1Label, 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 +321 @@
         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 +356 @@
             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 +372 @@
              self.tablesToExport.append(table.name)
 
-         self.alterPspsTables();
+         return self.alterPspsTables();
 
     '''
@@ -329 +392 @@
     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)
@@ -347 +410 @@
           # 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 +460 @@
         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
 
@@ -441 +505 @@
         p = Popen(cmd, shell=True, stdout=PIPE)
         p.wait()
-        #        out = p.stdout.read()
 
         rowCount = self.scratchDb.getRowCount("dvoDetection")
@@ -453 +516 @@
 
     '''
-    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 +523 @@
         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/publishToDatastore").text): 
+                    # 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)
+
                 if int(self.doc.find("options/reportNulls").text): self.reportNullsInAllPspsTables(False)
 
-        self.logger.infoPair("Batch......", "complete")
-
 from datastore import Datastore