Index: trunk/ippToPsps/jython/batch.py =================================================================== --- trunk/ippToPsps/jython/batch.py (revision 31405) +++ trunk/ippToPsps/jython/batch.py (revision 31469) @@ -32,5 +32,17 @@ "../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,4 +52,8 @@ # 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,4 +62,11 @@ 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,9 +74,4 @@ # 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,13 +106,9 @@ 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,4 +117,27 @@ 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,6 +175,8 @@ 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,9 +301,13 @@ 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,6 +319,9 @@ 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,5 +359,5 @@ 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,5 +383,5 @@ 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,12 +393,12 @@ 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,10 +406,11 @@ 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,8 +420,8 @@ 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,6 +430,11 @@ # 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,8 +447,13 @@ _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,6 +504,24 @@ ''' 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.") + +