Changeset 31469 for trunk/ippToPsps/jython/batch.py

Timestamp:

May 6, 2011, 12:42:21 PM (15 years ago)

Author:

rhenders

Message:

Hacks to kill last MySQL connection opened by STILTS functions; new run() method; now counting total detections as well as min/max objID; changed some constructor logic

File:

• trunk/ippToPsps/jython/batch.py (modified) (17 diffs)

trunk/ippToPsps/jython/batch.py

@@ -32 +32 @@
     "../config/2/tables.vot"
     '''
-    def __init__(self, logger, batchType, inputFitsPath="", survey="", useFullTables=False):
+    def __init__(self, 
+                 logger, 
+                 gpc1Db,
+                 ippToPspsDb,
+                 scratchDb,
+                 id,
+                 batchType, 
+                 inputFitsPath="", 
+                 survey="", 
+                 useFullTables=False):
+
+        self.everythingOK = False
+        self.readHeader = False
 
         # set up logging
@@ -40 +52 @@
 
         # set up class variables
+        self.id = id
+        self.gpc1Db = gpc1Db
+        self.ippToPspsDb = ippToPspsDb
+        self.scratchDb = scratchDb
         self.batchType = batchType;
         self.pspsVoTableFilePath = "../config/" + batchType + "/tables.vot"
@@ -46 +62 @@
         self.useFullTables = useFullTables
 
+        if self.alreadyProcessed(): return
+
+        # do we have an input file?
+        if self.inputFitsPath != "":
+
+            if not self.readPrimaryHeader(): return
+
         # TODO
         self.tablesToExport = []
@@ -51 +74 @@
         # open config
         doc = ElementTree(file="config.xml")
-
-        # create Gpc1Db object
-        self.gpc1Db = Gpc1Db(self.logger)
-        self.ippToPspsDb = IppToPspsDb(logger)
-        self.scratchDb = ScratchDb(logger, self.useFullTables)
 
         if self.survey != "":
@@ -88 +106 @@
         self.dateStr = now.strftime("%Y-%m-%d")
 
-        if self.inputFitsPath != "": 
-            file = open(self.inputFitsPath)
-            self.header = self.parseFitsHeader(file)
-            self.logger.info("Read primary and found " + str(len(self.header)) + " header cards") 
-            # TODO close file?
-
         # create DVO tables if accessing DVO directly
         if not self.useFullTables: self.scratchDb.createDvoTables()
 
+        self.everythingOK = True
+
     '''
     Destructor
@@ -103 +117 @@
 
         self.logger.debug("Batch destructor")
+
+
+    '''
+    Reads the primary header of the FITS file
+    '''
+    def readPrimaryHeader(self):
+
+        if self.readHeader: return True
+
+        # does it exist?
+        if not os.path.isfile(self.inputFitsPath):
+
+            self.logger.error("Cannot read file at '" + self.inputFitsPath + "'")
+            return False
+
+        file = open(self.inputFitsPath)
+        self.header = self.parseFitsHeader(file)
+        self.logger.info("Read primary header and found " + str(len(self.header)) + " header cards")
+        # TODO close file?
+
+        self.readHeader = True
+
+        return True
 
 
@@ -138 +175 @@
             file.seek(index + 2880, 0)
             
-        if found != True: self.logger.error("...could not find extension '" + name + "'")
-        else: self.logger.info("...read header at '" + name + "' and found " + str(len(header)) + " header cards") 
+        if found != True: 
+            self.logger.error("...could not read header in extension '" + name + "'")
+            return
+        #else: self.logger.info("...read header at '" + name + "' and found " + str(len(header)) + " header cards") 
 
         return header
@@ -262 +301 @@
 
         first = True
+
+        self.totalDetections = 0
         for table in tables:
 
-            sql = "SELECT MIN(objID), MAX(objID) FROM " + table
-            rs = self.scratchDb.stmt.executeQuery(sql)
+            sql = "SELECT MIN(objID), MAX(objID), COUNT(objID) FROM " + table
+            rs = self.scratchDb.executeQuery(sql)
             rs.first()
+
+            self.totalDetections = self.totalDetections + rs.getLong(3)
 
             if first:
@@ -276 +319 @@
 
             first = False
+            rs.close()
 
         self.ippToPspsDb.updateMinMaxObjID(self.batchID, self.minObjID, self.maxObjID)
+        self.logger.info("Total detections = %ld min objID = %ld max objID = %ld" % (self.totalDetections, self.minObjID, self.maxObjID))
+
 
     '''
@@ -313 +359 @@
          self.pspsTables = stilts.treads(self.pspsVoTableFilePath)
          for table in self.pspsTables:
-             self.logger.info("Creating PSPS table: " + table.name)
+             self.logger.debug("Creating PSPS table: " + table.name)
              table.write(self.scratchDb.url + '#' + table.name)
              self.tablesToExport.append(table.name)
@@ -337 +383 @@
     Accepts a regular expression filter so not all tables need to be imported
     '''
-    def importIppTables(self, filter):
+    def importIppTables(self, filter=""):
 
       self.logger.info("Attempting to import tables from input FITS file")
@@ -347 +393 @@
           match = re.match(filter, table.name)
           if not match: continue
-          self.logger.info("   Reading IPP table " + table.name + " from FITS file")
+          self.logger.info("Reading IPP table " + table.name + " from FITS file")
           table = stilts.tpipe(table, cmd='explodeall')
 
           # drop any previous tables before import
-          self.scratchDb.dropTable(table.name)
+          #self.scratchDb.dropTable(table.name)
 
           # IPP FITS files are littered with infinities, so remove these
-          self.logger.info("   Removing Infinity values from all columns")
+          self.logger.debug("Removing Infinity values from all columns")
           table = stilts.tpipe(table, cmd='replaceval -Infinity null *')
           table = stilts.tpipe(table, cmd='replaceval Infinity null *')
@@ -360 +406 @@
           try:
               table.write(self.scratchDb.url + '#' + table.name)
+              self.scratchDb.killLastConnectionID()
+              count = count + 1
           except:
-              self.logger.exception("   Problem writing table '" + table.name + "' to the database")
-          count = count + 1
+              self.logger.exception("Problem writing table '" + table.name + "' to the database")
+
 
       self.logger.info("Done. Imported %d tables" % count)
-
       self.indexIppTables()
 
@@ -373 +420 @@
     def exportPspsTablesToFits(self, regex="(.*)"):
 
-        self.logger.info("Replacing NULLs with -999 then exporting all PSPS tables to FITS")
+        self.logger.info("Replacing NULLs with -999, changing tables names using regex: " + regex)
         _tables = []
 
-        self.logger.info("    Selecting database tables")
+        self.logger.info("Selecting database tables")
         for table in self.tablesToExport:
 
@@ -383 +430 @@
 
            # get everything from table
-           _table = stilts.tread(self.scratchDb.url + '#SELECT * FROM ' + table)
-
+           try:
+               _table = stilts.tread(self.scratchDb.url + '#SELECT * FROM ' + table)
+               self.scratchDb.killLastConnectionID()
+           except:
+               self.logger.exception("Could not read from DB table: " + table)
+               return False
+               
            # replace nulls and empty fields with weird PSPS -999 pseudo-null
            _table = stilts.tpipe(_table, cmd='replaceval "" -999 *')
@@ -395 +447 @@
            _tables.append(_table)
 
-        self.logger.info("    Writing to FITS file '" + self.outputFitsPath + "'...")
-        stilts.twrites(_tables, self.outputFitsPath, fmt='fits')
-        self.logger.info("    ...done")
-        self.ippToPspsDb.updateProcessed(self.batchID, 1)
+        self.logger.info("Writing to FITS file '" + self.outputFitsPath + "'...")
+        try:
+            stilts.twrites(_tables, self.outputFitsPath, fmt='fits')
+            self.ippToPspsDb.updateProcessed(self.batchID, 1)
+        except:
+            self.logger.exception("Could not write to FITS")
+            return False
+
+        return True
 
     '''
@@ -447 +504 @@
     '''
     def alreadyProcessed(self):
-           self.logger.info("Not implemented")
-
-
-
+        self.logger.info("Not implemented")
+
+
+    '''
+    Creates and publishes a batch
+    '''
+    def run(self):
+
+        if not self.everythingOK: return
+
+        self.createEmptyPspsTables()
+        self.importIppTables()
+        if self.populatePspsTables():
+            if self.exportPspsTablesToFits():
+                self.writeBatchManifest()
+                self.createTarball()
+                self.publishToDatastore()
+                #self.reportNullsInAllPspsTables(False)
+                #sys.exit()
+        self.logger.info("Finished.")
+
+