Index: trunk/ippToPsps/jython/Makefile.am =================================================================== --- trunk/ippToPsps/jython/Makefile.am (revision 35097) +++ trunk/ippToPsps/jython/Makefile.am (revision 35097) @@ -0,0 +1,47 @@ +bin_SCRIPTS = ippjython + +jython_files = \ + batch.py \ + batchRepublisher.py \ + cleanup.py \ + config.py \ + console.py \ + czardb.py \ + datastore.py \ + datastoreRemover.py \ + detectionbatch.py \ + dvodetections.py \ + dvoobjects.py \ + dvo.py \ + dxlayer.py \ + fits.py \ + gpc1db.py \ + initbatch.py \ + ipptopspsdb.py \ + ipptopsps.py \ + loader.py \ + metrics.py \ + mysql.py \ + objectbatch.py \ + odm.py \ + plot.py \ + plotter.py \ + pollOdm.py \ + pslogger.py \ + queue.py \ + scratchdb.py \ + setupScratchDb.py \ + skychunk.py \ + stackbatch.py + +jythondir = $(datadir)/jython +jython_DATA = $(jython_files) + +INSTALL = @INSTALL@ -C + +install-data-hook: + chmod 0755 $(jythondir) + +EXTRA_DIST = $(jython_files) + +ACLOCAL_AMFLAGS = -I m4 Index: trunk/ippToPsps/jython/batch.py =================================================================== --- trunk/ippToPsps/jython/batch.py (revision 34884) +++ trunk/ippToPsps/jython/batch.py (revision 35097) @@ -30,4 +30,5 @@ logger, config, + skychunk, gpc1Db, ippToPspsDb, @@ -39,6 +40,12 @@ useFullTables): + # print "starting the batch" + + # self.printline = 0 + # self.testprint() + self.readHeader = False self.config = config + self.skychunk = skychunk self.fits = fits self.useFullTables = useFullTables @@ -48,4 +55,6 @@ self.logger.infoSeparator() self.logger.debug("Batch class constructor") + + # self.testprint() # set up class variables @@ -56,5 +65,7 @@ self.scratchDb = scratchDb self.batchType = batchType; - self.pspsVoTableFilePath = "../config/" + batchType + "/tables.vot" + self.pspsVoTableFilePath = self.config.configDir + "tables." + batchType + ".vot" + + # self.testprint() # check FITS file is ok @@ -64,5 +75,5 @@ # now check that the fits header is readable if not self.header: - logger.errorPair("Could not read FITS for id", "%d" % id) + logger.errorPair("Could not read FITS PHU for id", "%d" % id) raise @@ -72,26 +83,28 @@ self.tablesToExport = [] - if self.config.survey != "": - self.surveyID = self.scratchDb.getSurveyID(self.config.survey) + if self.skychunk.survey != "": + self.surveyID = self.scratchDb.getSurveyID(self.skychunk.survey) else: self.surveyID = -1; # create datastore object - self.datastore = Datastore(self.logger, self.config, self.ippToPspsDb) - - # get local storage location from config + self.datastore = Datastore(self.logger, self.skychunk, self.ippToPspsDb) + + # get local storage location from skychunk self.batchName = Batch.getNameFromID(self.batchID) self.subDir = Batch.getSubDir( - self.config.basePath, + self.skychunk.basePath, self.batchType, - self.config.dvoLabel) + self.skychunk.dvoLabel) self.localOutPath = Batch.getOutputPath( - self.config.basePath, + self.skychunk.basePath, self.batchType, - self.config.dvoLabel, + self.skychunk.dvoLabel, self.batchID) if not os.path.exists(self.localOutPath): os.makedirs(self.localOutPath) + + # self.testprint() # store today's date @@ -112,7 +125,13 @@ self.logger.infoPair("Input FITS primary header", "%s cards found" % self.fits.getPrimaryHeaderCardCount()) + # self.testprint() + self.logger.infoPair("Output path", self.localOutPath) + def testprint(self): + print "here ", self.printline + self.printline += 1 + ''' Static method to generated batch name from batch ID @@ -232,5 +251,5 @@ root.attrib['type'] = self.batchType root.attrib['timestamp'] = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") - root.attrib['survey'] = self.config.pspsSurvey + root.attrib['survey'] = self.skychunk.pspsSurvey # min/max object IDs @@ -273,7 +292,7 @@ # set up filenams and paths tarFile = Batch.getTarFile(self.batchID) - tarPath = Batch.getTarPath(self.config.basePath, self.batchType, self.config.dvoLabel, self.batchID) + tarPath = Batch.getTarPath(self.skychunk.basePath, self.batchType, self.skychunk.dvoLabel, self.batchID) tarballFile = Batch.getTarballFile(self.batchID) - tarballPath = Batch.getTarballPath(self.config.basePath, self.batchType, self.config.dvoLabel, self.batchID) + tarballPath = Batch.getTarballPath(self.skychunk.basePath, self.batchType, self.skychunk.dvoLabel, self.batchID) # tar directory @@ -376,4 +395,6 @@ self.logger.infoPair("Importing tables with filter", filter) + # print "trying to read ", self.fits.getPath() + try: tables = stilts.treads(self.fits.getPath()) @@ -385,8 +406,11 @@ for table in tables: + # print "import smf table ", table match = re.match(filter, table.name) if not match: continue self.logger.debugPair("Reading IPP table", table.name) table = stilts.tpipe(table, cmd='explodeall') + + # print "read smf table ", table # drop any previous tables before import @@ -398,7 +422,9 @@ table = stilts.tpipe(table, cmd='replaceval -Infinity null *') table = stilts.tpipe(table, cmd='replaceval Infinity null *') + # print "cleaned up values ", table try: table.write(self.scratchDb.url + '#' + table.name) + # print "wrote to mysql ", table self.scratchDb.killLastConnectionID() count = count + 1 @@ -488,5 +514,5 @@ # TODO path to DVO prog hardcoded temporarily - cmd = "../src/dvograbber " + self.config.settingsPath + " " + self.scratchDb.dbName + " " + self.config.dvoLocation + cmd = "../src/dvograbber " + self.config.settingsPath + " " + self.scratchDb.dbName + " " + self.skychunk.dvoLocation self.logger.infoPair("Running DVO", cmd) p = Popen(cmd, shell=True, stdout=PIPE) @@ -504,8 +530,10 @@ ''' Creates and publishes a batch - TODO all methods call below should throw exceptions on failure + TODO all method calls below should throw exceptions on failure ''' def run(self): + # this is badly named : it creates the tables and then fills in + # the ImageMeta tables for each ota (alterPspsTables) if not self.createEmptyPspsTables(): self.logger.errorPair("Aborting this batch", "could not create empty PSPS tables") @@ -513,4 +541,6 @@ raise + # for P2/ST, this reads the detection tables from the CMF/SMF file(s) + # for OB, this imports object data from DVO if not self.importIppTables(): self.logger.errorPair("Aborting this batch", "could not import IPP tables") @@ -529,5 +559,5 @@ if self.writeBatchManifest(): - if self.config.datastorePublishing: + if self.skychunk.datastorePublishing: # tar and zip ready for publication to datastore Index: trunk/ippToPsps/jython/batchRepublisher.py =================================================================== --- trunk/ippToPsps/jython/batchRepublisher.py (revision 34884) +++ trunk/ippToPsps/jython/batchRepublisher.py (revision 35097) @@ -30,5 +30,5 @@ self.exitProgram("incorrect args") - self.datastore = Datastore(self.logger, self.config, self.ippToPspsDb) + self.datastore = Datastore(self.logger, self.skychunk, self.ippToPspsDb) self.BATCHID = int(sys.argv[2]) @@ -40,5 +40,5 @@ batchName = Batch.getNameFromID(self.BATCHID) batchType = self.ippToPspsDb.getBatchType(self.BATCHID) - subDir = Batch.getSubDir(self.config.basePath, batchType, self.config.dvoLabel) + subDir = Batch.getSubDir(self.skychunk.basePath, batchType, self.skychunk.dvoLabel) tarballFile = Batch.getTarballFile(self.BATCHID) @@ -50,5 +50,5 @@ self.logger.infoPair("Reseting batch in database", "%d" % self.BATCHID) self.ippToPspsDb.resetBatch(self.BATCHID) - self.logger.infoPair("publishing to", self.config.datastoreProduct) + self.logger.infoPair("publishing to", self.skychunk.datastoreProduct) Batch.publishToDatastore(self.datastore, self.BATCHID, batchName, subDir, tarballFile) Index: trunk/ippToPsps/jython/cleanup.py =================================================================== --- trunk/ippToPsps/jython/cleanup.py (revision 34884) +++ trunk/ippToPsps/jython/cleanup.py (revision 35097) @@ -26,5 +26,5 @@ super(Cleanup, self).__init__(argv) - self.datastore = Datastore(self.logger, self.config, self.ippToPspsDb) + self.datastore = Datastore(self.logger, self.skychunk, self.ippToPspsDb) self.dxlayer = DXLayer(self.logger) if len(argv) > 2: self.parsePollTimeArg(sys.argv[2]) @@ -40,6 +40,6 @@ self.logger.infoSeparator() - self.logger.infoPair("Config", self.config.name) - self.config.printDeletionPolicy() + self.logger.infoPair("Skychunk", self.skychunk.name) + self.skychunk.printDeletionPolicy() self.clean("IN") self.clean("P2") @@ -89,9 +89,9 @@ # delete stuff from local disk - if self.config.deleteLocal: + if self.skychunk.deleteLocal: count = 0 for id in deleteFromLocalIDs: - if Batch.deleteFromDisk(self.logger, self.config.basePath, batchType, self.config.dvoLabel, id): + if Batch.deleteFromDisk(self.logger, self.skychunk.basePath, batchType, self.skychunk.dvoLabel, id): self.ippToPspsDb.updateDeletedLocal(id, 1) count = count + 1 @@ -100,5 +100,5 @@ # remove stuff from datastore - if self.config.deleteDatastore: + if self.skychunk.deleteDatastore: count = 0 for id in deleteFromDatastoreIDs: @@ -112,5 +112,5 @@ # remove stuff from DXLayer - if self.config.deleteDxLayer: + if self.skychunk.deleteDxLayer: count = 0 for id in deleteFromDxLayerIDs: Index: trunk/ippToPsps/jython/config.py =================================================================== --- trunk/ippToPsps/jython/config.py (revision 34884) +++ trunk/ippToPsps/jython/config.py (revision 35097) @@ -1,4 +1,6 @@ #!/usr/bin/env jython +import sys +import os import logging from xml.etree.ElementTree import ElementTree, Element, tostring @@ -6,11 +8,9 @@ from pslogger import PSLogger - ''' -A class encapsulating a ippToPsps configuration. This information is stored in the 'config' table -of the ipptopsps database, but there are some higher level config details in the 'settings.xml' file +A class encapsulating the globald ippToPsps configuration information. +This is stored in the 'settings.xml' file in the config directory ''' class Config(object): - ''' @@ -19,51 +19,48 @@ Basically reads the entire config and stores values to class variables ''' - def __init__(self, programName, name): + def __init__(self): - self.programName = programName - self.name = name - self.settingsPath = "../config/settings.xml" # TODO + self.test = False + for arg in sys.argv: + if arg == "-test": + self.test = True + sys.argv.remove(arg) + if arg == "-t": + self.test = True + sys.argv.remove(arg) + + ## name of the top-level jython script + self.programName = os.path.basename(sys.argv[0]) + + ## this is somewhat crude: ipptopsps programs are called like this: + ## ippjython foo.py [chunk] [other options] + ## the logging code (called below by getLogger) wants to include 'name' in the + ## output file. we blindly set name == argv[1] + ## programs which are called without argv[1] will use 'none' in the log file + if len(sys.argv) >= 2: + self.name = sys.argv[1] + + # XXX this probably goes in 'config.py' + self.configDir = os.getenv("IPPTOPSPS_DATA") + if self.configDir is None: + self.configDir = "../config/" + else: + self.configDir = self.configDir + "/" + + # self.configdir is set in ipptopsps.jy on init based on env(IPPTOPSPS_DATA) + # for test purposes, an uninstalled system may use the config information from + # a relative path + self.settingsPath = self.configDir + "settings.xml" + self.logger = None self.settingsDoc = ElementTree(file=self.settingsPath) self.logPath = self.settingsDoc.find("logPath").text - self.czarPlotsPath = self.settingsDoc.find("czarPlotsPath").text - # this is the border (in degrees) that we place around any loading box of - # PS1 pointings to ensure we pull a large enough area out of DVO - self.BORDER = 1.65 - self.isLoaded = False - - ''' - Prints everything for this config - ''' - def printAll(self): - - self.logger.infoTitle("Config") - - try: - self.logger.infoSeparator() - self.logger.infoPair("Config name", self.name) - self.logger.infoPair("Survey", self.survey) - self.logger.infoPair("Publishing to PSPS as survey", self.pspsSurvey) - self.logger.infoPair("Loading epoch", self.epoch) - self.logger.infoPair("Data release", "%d" % self.dataRelease) - for batchType in self.batchTypes: self.logger.infoPair("Queuing batch type", batchType) - except: - pass - - self.printDvoInfo() - self.printDatastoreInfo() - self.printBoxCoords() - self.printDeletionPolicy() - try: self.logger.infoSeparator() - except: pass - - ''' - Queuing this batch type? - ''' - def queuingThisBatchType(self, batchType): - if batchType in self.batchTypes: return 1 - return 0 + print "config.programName: ", self.programName + print "config.configDir: ", self.configDir + print "config.settingsPath: ", self.settingsPath + print "config.logPath: ", self.logPath + print "config.test: ", self.test ''' @@ -72,58 +69,16 @@ def getLogger(self, host, pid, stdout=1, sendToFile=0): + print "get Logger: ", host, pid logging.setLoggerClass(PSLogger) + print "done Logger 1" self.logger = logging.getLogger(self.programName) - self.logger.setup(self.programName, self.logPath, self.name, host, pid, stdout, sendToFile) + print "done Logger 2" + if (self.name is None): + self.logger.setup(self.programName, self.logPath, "none", host, pid, stdout, sendToFile) + else: + self.logger.setup(self.programName, self.logPath, self.name, host, pid, stdout, sendToFile) + print "done Logger 3" return self.logger - - ''' - Prints the currently set DVO info - ''' - def printDvoInfo(self): - - try: - self.logger.infoPair("DVO label", self.dvoLabel) - self.logger.infoPair("DVO location", self.dvoLocation) - except: - pass - - ''' - Prints datastore info - ''' - def printDatastoreInfo(self): - - try: - self.logger.infoBool("Datastore publishing?", self.datastorePublishing) - if self.datastorePublishing: - self.logger.infoPair("Datastore type", self.datastoreProduct) - self.logger.infoPair("Datastore product", self.datastoreType) - except: - pass - - ''' - Prints the current deletion policy - ''' - def printDeletionPolicy(self): - - try: - self.logger.infoBool("Deleting local?", self.deleteLocal) - self.logger.infoBool("Deleting datastore?", self.deleteDatastore) - self.logger.infoBool("Deleting DXLayer?", self.deleteDxLayer) - except: - pass - - ''' - Prints the currently set RA/Dec bounding box - ''' - def printBoxCoords(self): - - try: - self.logger.infoPair("RA limits", "%.1f -> %.1f" % (self.minRa, self.maxRa)) - self.logger.infoPair("Dec limits", "%.1f -> %.1f" % (self.minDec, self.maxDec)) - self.logger.infoPair("Loading box size", "%.1f degrees" % self.boxSize) - except: - pass - ''' @@ -134,3 +89,2 @@ def getDbUser(self, dbType): return self.settingsDoc.find(dbType +"/user").text def getDbPassword(self, dbType): return self.settingsDoc.find(dbType +"/password").text - Index: trunk/ippToPsps/jython/console.py =================================================================== --- trunk/ippToPsps/jython/console.py (revision 34884) +++ trunk/ippToPsps/jython/console.py (revision 35097) @@ -37,5 +37,5 @@ self.frame = JFrame( - "ippToPsps console for config '" + self.config.name + "'", + "ippToPsps console for skychunk '" + self.skychunk.name + "'", layout=BorderLayout(), size=(1000, 500), @@ -74,5 +74,5 @@ button = JButton('Purge dead', actionPerformed=self.purgeDead) buttonPanel.add(button) - button = JButton('Change config', actionPerformed=self.changeConfig) + button = JButton('Change skychunk', actionPerformed=self.changeSkychunk) buttonPanel.add(button) @@ -136,17 +136,17 @@ self.ippToPspsDb.purgeDeadClients() self.refreshClientTable(None) - def changeConfig(self, event): + def changeSkychunk(self, event): ids = self.getSelectedIds() if len(ids) < 1: JOptionPane.showMessageDialog(None, "No clients selected", "Error", JOptionPane.ERROR_MESSAGE) return - comboBox = JComboBox(self.ippToPspsDb.getActiveConfigList()) + comboBox = JComboBox(self.ippToPspsDb.getActiveSkychunkList()) if JOptionPane.showConfirmDialog(None, comboBox, - "Choose a config", + "Choose a skychunk", JOptionPane.OK_CANCEL_OPTION, JOptionPane.PLAIN_MESSAGE) == JOptionPane.CANCEL_OPTION: return - self.ippToPspsDb.setConfigForLoaders(comboBox.getSelectedItem(), ids) + self.ippToPspsDb.setSkychunkForLoaders(comboBox.getSelectedItem(), ids) self.refreshClientTable(None) Index: trunk/ippToPsps/jython/datastore.py =================================================================== --- trunk/ippToPsps/jython/datastore.py (revision 34884) +++ trunk/ippToPsps/jython/datastore.py (revision 35097) @@ -1,3 +1,4 @@ #!/usr/bin/env jython +# EAM : config -> skychunk DONE from subprocess import call, PIPE, Popen @@ -19,9 +20,9 @@ ''' - def __init__(self, logger, config, ippToPspsDb): + def __init__(self, logger, skychunk, ippToPspsDb): # setup logger self.logger = logger - self.config = config + self.skychunk = skychunk self.ippToPspsDb = ippToPspsDb @@ -42,6 +43,6 @@ command = "dsreg --add " + name + "\ --link --datapath " + path + "\ - --type " + self.config.datastoreType + "\ - --product " + self.config.datastoreProduct + "\ + --type " + self.skychunk.datastoreType + "\ + --product " + self.skychunk.datastoreProduct + "\ --list " + tempFile.name @@ -75,5 +76,5 @@ --del " + name + " \ --rm \ - --product " + self.config.datastoreProduct + --product " + self.skychunk.datastoreProduct p = Popen(command, shell=True, stdout=PIPE, stderr=PIPE) Index: trunk/ippToPsps/jython/datastoreRemover.py =================================================================== --- trunk/ippToPsps/jython/datastoreRemover.py (revision 34884) +++ trunk/ippToPsps/jython/datastoreRemover.py (revision 35097) @@ -24,5 +24,5 @@ self.exitProgram("incorrect args") - self.datastore = Datastore(self.logger, self.config, self.ippToPspsDb) + self.datastore = Datastore(self.logger, self.skychunk, self.ippToPspsDb) ''' Index: trunk/ippToPsps/jython/detectionbatch.py =================================================================== --- trunk/ippToPsps/jython/detectionbatch.py (revision 34884) +++ trunk/ippToPsps/jython/detectionbatch.py (revision 35097) @@ -3,6 +3,8 @@ import os.path import sys +import glob import time import stilts + from java.lang import * from java.sql import * @@ -34,4 +36,5 @@ logger, config, + skychunk, gpc1Db, ippToPspsDb, @@ -44,4 +47,5 @@ logger, config, + skychunk, gpc1Db, ippToPspsDb, @@ -72,15 +76,8 @@ self.outputFitsPath = "%s/%s" % (self.localOutPath, self.outputFitsFile) - # if test mode - if self.config.test: - self.startX = 3 - self.endX = 4 - self.startY = 3 - self.endY = 4 - else: - self.startX = 0 - self.endX = 8 - self.startY = 0 - self.endY = 8 + self.startX = 0 + self.endX = 8 + self.startY = 0 + self.endY = 8 #self.startX = 1 @@ -89,5 +86,5 @@ #self.endY = 8 - # get a fre primary header values. if in test mode, then use defaults + # get a few primary header values. if in test mode, then use defaults if self.safeDictionaryAccessWithDefault(self.header, 'MJD-OBS', "1") == "NULL": self.logger.errorPair("Could not get", "MJD-OBS") @@ -214,4 +211,7 @@ ," + self.safeDictionaryAccess(self.header, 'PCA2X0Y2') + " \ )" + + print "frame meta sql: ", sql + self.scratchDb.execute(sql) @@ -219,5 +219,5 @@ self.scratchDb.updateFilterID("FrameMeta", self.filter) self.scratchDb.updateAllRows("FrameMeta", "calibModNum", str(self.calibModNum)) - self.scratchDb.updateAllRows("FrameMeta", "dataRelease", str(self.config.dataRelease)) + self.scratchDb.updateAllRows("FrameMeta", "dataRelease", str(self.skychunk.dataRelease)) ''' @@ -228,11 +228,14 @@ tableName = "ImageMeta_" + ota + # XXX we drop the table above so it is not left behind on failure # drop then re-create table - self.scratchDb.dropTable(tableName) + # self.scratchDb.dropTable(tableName) sql = "CREATE TABLE " + tableName + " LIKE ImageMeta" try: self.scratchDb.execute(sql) except: pass - - # insert all detections into table + if (ota[0:2] == "XY"): ccdID = ota[2:4] + else: ccdID = 0 + + # insert image metadata into table sql = "INSERT INTO " + tableName + " ( \ frameID \ @@ -299,5 +302,5 @@ ) VALUES ( \ " + str(self.expID) + " \ - ," + ota[2:4] + " \ + ," + str(ccdID) + " \ ," + str(self.bias) + " \ ," + str(self.biasScat) + " \ @@ -366,5 +369,5 @@ self.scratchDb.updateFilterID(tableName, self.filter) self.scratchDb.updateAllRows(tableName, "calibModNum", str(self.calibModNum)) - self.scratchDb.updateAllRows(tableName, "dataRelease", str(self.config.dataRelease)) + self.scratchDb.updateAllRows(tableName, "dataRelease", str(self.skychunk.dataRelease)) if 'NASTRO' in header: self.totalNumPhotoRef = self.totalNumPhotoRef + int(header['NASTRO']) self.scratchDb.replaceNullsInThisColumn(tableName, "polyOrder", "0") @@ -386,5 +389,5 @@ except: pass - # delete all detections with PSF_INST_MAG < 17.5, as decreed by Gene + # delete all detections with PSF_INST_MAG < -17.5, as decreed by Gene BEFORE = self.scratchDb.getRowCount(ippTableName) #don't do this anymore @@ -464,7 +467,8 @@ , '" + self.dateStr + "' \ , 0 \ - , " + str(self.config.dataRelease) + "\ + , " + str(self.skychunk.dataRelease) + "\ FROM " + ippTableName # self.logger.info(sql) + self.scratchDb.execute(sql) @@ -517,10 +521,10 @@ self.scratchDb.execute(sql) - self.scratchDb.updateAllRows(tableName, "dataRelease", str(self.config.dataRelease)) + self.scratchDb.updateAllRows(tableName, "dataRelease", str(self.skychunk.dataRelease)) ''' Populates the DetectionCalib table for this OTA ''' - def populateDetectionCalibTable(self, ota): + def populateDetectionCalibTableUpdateInsert(self, ota): tableName = "DetectionCalib_" + ota @@ -562,5 +566,69 @@ WHERE a.objID = b.objID AND a.detectID = b.detectID" self.scratchDb.execute(sql) - self.scratchDb.updateAllRows(tableName, "dataRelease", str(self.config.dataRelease)) + self.scratchDb.updateAllRows(tableName, "dataRelease", str(self.skychunk.dataRelease)) + + ''' + Populates the DetectionCalib table for this OTA + ''' + def populateDetectionCalibTable(self, ota): + + # target table name: + tableName = "DetectionCalib_" + ota + # drop then re-create table + self.scratchDb.dropTable(tableName) + sql = "CREATE TABLE " + tableName + " LIKE DetectionCalib" + try: self.scratchDb.execute(sql) + except: pass + + externID = self.imageIDs[ota] + + imageID = self.scratchDb.getImageIDFromExternID(externID) + self.logger.infoPair("obtained","imageID") + + # check for & create output directory first + datadumpDir = "/tmp/datadump" + try: + statinfo = os.stat(datadumpDir) + # check on the stat results? + except: + print "making the data dump directory ", datadumpDir + os.mkdir(datadumpDir) + os.chmod(datadumpDir, 0777) + statinfo = os.stat(datadumpDir) + + dumpFile = datadumpDir + "/genetest.xx.dat" + files = glob.glob(dumpFile) + if len(files) > 0: + os.unlink(dumpFile) + + # insert all detections into table + sql = "SELECT \ + a.objID, \ + a.detectID, \ + a.ippObjID, \ + a.ippDetectID, \ + a.filterID, \ + a.surveyID, \ + b.ra, \ + b.dec_, \ + b.raErr, \ + b.decErr, \ + b.zp, \ + b.zpErr, \ + b.expTime, \ + b.airMass, \ + " + str(self.skychunk.dataRelease) + " \ + FROM \ + Detection_" + ota + " as a \ + JOIN " + self.scratchDb.dvoDetectionTable + " as b \ + ON (a.objID = b.objID AND a.detectID = b.detectID) \ + WHERE b.imageID = " + str(imageID) + \ + " INTO OUTFILE '" + dumpFile + "'" + print "sql: ", sql + self.scratchDb.execute(sql) + + sql = "LOAD DATA INFILE '" + dumpFile + "' INTO TABLE " + tableName + print "sql: ", sql + self.scratchDb.execute(sql) ''' @@ -591,6 +659,39 @@ self.scratchDb.updateAllRows(tableName, "calibModNum", str(self.calibModNum)) - self.scratchDb.updateAllRows(tableName, "dataRelease", str(self.config.dataRelease)) - + self.scratchDb.updateAllRows(tableName, "dataRelease", str(self.skychunk.dataRelease)) + + ''' + Applies indexes and other constraints to the PSPS tables + ''' + def alterPspsTablesChip(self, chipname, extname, x, y): + + # drop the ImageMeta_ table first, or we can leave an invalid table behind + tableName = "ImageMeta_" + chipname + + # drop then re-create table + self.scratchDb.dropTable(tableName) + + # load corresponding header into memory + header = self.fits.findAndReadHeader(extname) + if not header: + self.logger.errorPair("No header found for chip", chipname) + return False + + # check we have valid sourceID/imageID pair from the header + if self.safeDictionaryAccess(header, 'SOURCEID') == "NULL": return False + if self.safeDictionaryAccess(header, 'IMAGEID') == "NULL": return False + + # store sourceID/imageID combo in Db so DVO can look up later + if not self.useFullTables: + self.scratchDb.insertNewDvoExternID(header['SOURCEID'], header['IMAGEID']) + + # store these for later + self.imageIDs[chipname] = header['IMAGEID'] + + # populate ImageMeta + self.populateImageMetaTable(chipname, header) + self.updateImageID("ImageMeta_" + chipname, x, y) + + return True ''' @@ -623,25 +724,9 @@ ota = "XY%d%d" % (x, y) - # load corresponding header into memory - header = self.fits.findAndReadHeader(ota + ".hdr") - if not header: - self.logger.errorPair("No header found for OTA", ota) - continue - - # check we have valid sourceID/imageID pair from the header - if self.safeDictionaryAccess(header, 'SOURCEID') == "NULL": continue - if self.safeDictionaryAccess(header, 'IMAGEID') == "NULL": continue - - # store sourceID/imageID combo in Db so DVO can look up later - if not self.useFullTables: - self.scratchDb.insertNewDvoExternID(header['SOURCEID'], header['IMAGEID']) - - # store these for later - self.imageIDs[ota] = header['IMAGEID'] - - # populate ImageMeta - self.populateImageMetaTable(ota, header) - self.updateImageID("ImageMeta_" + ota, x, y) - + self.alterPspsTablesChip(ota, ota + ".hdr", x, y) + + # try the test Chip + self.alterPspsTablesChip("Chip", "Chip.hdr", 0, 0) + # now run DVO code to get all IDs if not self.useFullTables: @@ -651,6 +736,4 @@ # the column in PSPS self.scratchDb.execute("ALTER TABLE DetectionCalib CHANGE dec_ `dec` double") - - return True @@ -675,5 +758,9 @@ extension = "XY%d%d_psf" % (x, y) self.scratchDb.createIndex(extension, "IPP_IDET") + # try the test Chip + self.scratchDb.createIndex("Chip_psf", "IPP_IDET") + self.logger.infoPair("created indexes on", "IPP tables") + ''' Updates provided table with DVO IDs from DVO table @@ -693,4 +780,82 @@ self.scratchDb.execute(sql) + ''' + Does the processing, i.e. pulling stuff from IPP tables into PSPS tables + ''' + def populatePspsTablesChip(self, chipname, x, y, results, tables): + # XXX EAM NOTE: drop tables (Detection_, SkinnyObject_, DetectionCalib_, Detection_) here so + # they do not polute the db? + # XXX or put an explicit drop at the end of the loop? + + #self.logger.infoTitle("Processing " + chipname) + # this is a bit crude: if the chip is not present, this test will fail and the chip + # will be (correctly) skipped. would be better to carry that information explicitly ("chip is missing") + if not self.scratchDb.astrometricSolutionOK(chipname): + self.logger.info("| %5s | ------------- Bad astrometric solution : rejecting ------------ |" % chipname) + return False + self.logger.info("populate stuff "); + # populate remainder of tables + self.populateDetectionTable(chipname, results) + self.logger.info("successful populate "); + # now add DVO IDs + self.updateDvoIDs("Detection_" + chipname, self.imageIDs[chipname]) + self.logger.info("updated dvoids") + results['NULLOBJID'] = self.scratchDb.reportAndDeleteRowsWithNULLS("Detection_" + chipname, "objID") + #self.logger.info("deleted nulls") + self.updateImageID("Detection_" + chipname, x, y) + self.logger.info("updateImageId") + rowCount = self.scratchDb.getRowCount("Detection_" + chipname) + self.logger.info("got row count") + self.logger.info("| %5s | %13d | %13d | %13d | %13d | %13d | %13d |", + chipname, + results['ORIGINALTOTAL'], + results['SATDET'], + results['NULLINSTFLUX'], + results['NULLPEAKADU'], + results['NULLOBJID'], + rowCount) + self.totalOriginal = self.totalOriginal + results['ORIGINALTOTAL'] + self.totalSatDet = self.totalSatDet + results['SATDET'] + self.totalNulIInstFlux = self.totalNulIInstFlux + results['NULLINSTFLUX'] + self.totalNullPeakFlux = self.totalNullPeakFlux + results['NULLPEAKADU'] + self.totalNullObjID = self.totalNullObjID + results['NULLOBJID'] + self.totalDetections = self.totalDetections + rowCount + #self.logger.info("updated totals") + # check we have something in this Detection table TODO add this to table above + if rowCount < 1: + self.logger.infoPair("Skipping empty table for chipname", chipname) + return False; + + # update ImageMeta with count of detections for this CHIPNAME and photoCodeID + sql = "UPDATE ImageMeta_" + chipname + " \ + SET nDetect = %d, photoCalID = %d" % (self.scratchDb.getRowCount("Detection_" + chipname), self.scratchDb.getPhotoCalID(self.imageIDs[chipname])) + self.scratchDb.execute(sql) + + self.logger.info("updated imagedata") + self.populateSkinnyObjectTable(chipname) + self.logger.info("updated skinnyobject") + #self.populateObjectCalColorTable(chipname) + #self.logger.info("updated objectcalcolor") + self.populateDetectionCalibTable(chipname) + self.logger.info("updated detectioncalibtable") + + # add these to list of tables to export later + self.tablesToExport.append("ImageMeta_" + chipname) + self.logger.info("export ImageMeta") + + self.tablesToExport.append("Detection_" + chipname) + self.logger.info("export Detection") + + self.tablesToExport.append("SkinnyObject_" + chipname) + self.logger.info("export Skinny") + + #self.tablesToExport.append("ObjectCalColor_" + chipname) + self.tablesToExport.append("DetectionCalib_" + chipname) + self.logger.info("export DetectionCalib") + + tables.append("Detection_" + chipname) + self.logger.info("updated detectioncalibtable") + + return True ''' @@ -705,5 +870,5 @@ otaCount = 0 results = {} - totalOriginal = totalSatDet = totalNulIInstFlux = totalNullPeakFlux = totalNullObjID = totalDetections = 0 + self.totalOriginal = self.totalSatDet = self.totalNulIInstFlux = self.totalNullPeakFlux = self.totalNullObjID = self.totalDetections = 0 self.logger.info("+-------+---------------+---------------+---------------+---------------+---------------+---------------+") self.logger.info("| OTA | Initial total | Sat Det | NULL instFlux | NULL peak ADU | NULL obj ID | Remainder |") @@ -721,73 +886,17 @@ if ota not in self.imageIDs: continue - #self.logger.infoTitle("Processing " + ota) - if not self.scratchDb.astrometricSolutionOK(ota): - self.logger.info("| %5s | ------------- Bad astrometric solution : rejecting ------------ |" % ota) - continue - # self.logger.info("populate stuff "); - # populate remainder of tables - self.populateDetectionTable(ota, results) - # self.logger.info("successful populate "); - # now add DVO IDs - self.updateDvoIDs("Detection_" + ota, self.imageIDs[ota]) - # self.logger.info("updated dvoids") - results['NULLOBJID'] = self.scratchDb.reportAndDeleteRowsWithNULLS("Detection_" + ota, "objID") - #self.logger.info("deleted nulls") - self.updateImageID("Detection_" + ota, x, y) - # self.logger.info("updateImageId") - rowCount = self.scratchDb.getRowCount("Detection_" + ota) - # self.logger.info("got row count") - self.logger.info("| %5s | %13d | %13d | %13d | %13d | %13d | %13d |", - ota, - results['ORIGINALTOTAL'], - results['SATDET'], - results['NULLINSTFLUX'], - results['NULLPEAKADU'], - results['NULLOBJID'], - rowCount) - totalOriginal = totalOriginal + results['ORIGINALTOTAL'] - totalSatDet = totalSatDet + results['SATDET'] - totalNulIInstFlux = totalNulIInstFlux + results['NULLINSTFLUX'] - totalNullPeakFlux = totalNullPeakFlux + results['NULLPEAKADU'] - totalNullObjID = totalNullObjID + results['NULLOBJID'] - totalDetections = totalDetections + rowCount - #self.logger.info("updated totals") - # check we have something in this Detection table TODO add this to table above - if rowCount < 1: - self.logger.debugPair("Skipping empty table for ota", ota) - continue; - - # update ImageMeta with count of detections for this OTA and photoCodeID - sql = "UPDATE ImageMeta_" + ota + " \ - SET nDetect = %d, photoCalID = %d" % (self.scratchDb.getRowCount("Detection_" + ota), self.scratchDb.getPhotoCalID(self.imageIDs[ota])) - self.scratchDb.execute(sql) - - #self.logger.info("updated imagedata") - self.populateSkinnyObjectTable(ota) - #self.logger.info("updated skinnyobject") - #self.populateObjectCalColorTable(ota) - #self.logger.info("updated objectcalcolor") - self.populateDetectionCalibTable(ota) - #self.logger.info("updated detectioncalibtable") - - # add these to list of tables to export later - self.tablesToExport.append("ImageMeta_" + ota) - self.tablesToExport.append("Detection_" + ota) - self.tablesToExport.append("SkinnyObject_" + ota) - #self.tablesToExport.append("ObjectCalColor_" + ota) - self.tablesToExport.append("DetectionCalib_" + ota) - tables.append("Detection_" + ota) - - otaCount = otaCount + 1 - #self.logger.info("end of ota") + if self.populatePspsTablesChip(ota, x, y, results, tables): otaCount = otaCount + 1 + + if self.populatePspsTablesChip("Chip", 0, 0, results, tables): otaCount = otaCount + 1 + # print totals self.logger.info("+-------+---------------+---------------+---------------+---------------+---------------+---------------+") self.logger.info("| Total | %13d | %13d | %13d | %13d | %13d | %13d |", - totalOriginal, - totalSatDet, - totalNulIInstFlux, - totalNullPeakFlux, - totalNullObjID, - totalDetections) + self.totalOriginal, + self.totalSatDet, + self.totalNulIInstFlux, + self.totalNullPeakFlux, + self.totalNullObjID, + self.totalDetections) self.logger.info("+-------+---------------+---------------+---------------+---------------+---------------+---------------+") @@ -830,7 +939,9 @@ def importIppTables(self, filter=""): - if self.config.test: regex = "XY33.psf" - else : regex = ".*.psf" + ## if self.config.test: regex = "XY33.psf" + ## else : regex = ".*.psf" + regex = ".*.psf" + # XXX EAM NOTE : this is fragile : requires PS1_V4 columns = "IPP_IDET X_PSF Y_PSF X_PSF_SIG Y_PSF_SIG PSF_INST_MAG PSF_INST_MAG_SIG PEAK_FLUX_AS_MAG PSF_MAJOR PSF_MINOR PSF_THETA EXT_NSIGMA PSF_QF MOMENTS_XX MOMENTS_XY MOMENTS_YY AP_MAG KRON_FLUX KRON_FLUX_ERR FLAGS FLAGS2 SKY SKY_SIGMA EXT_NSIGMA PSF_QF_PERFECT PSF_CHISQ" Index: trunk/ippToPsps/jython/dvo.py =================================================================== --- trunk/ippToPsps/jython/dvo.py (revision 34884) +++ trunk/ippToPsps/jython/dvo.py (revision 35097) @@ -33,9 +33,11 @@ ''' - def __init__(self, logger, config, scratchDbName="ipptopsps_scratch"): + def __init__(self, logger, config, skychunk, ippToPspsDb, scratchDbName=None): # set up logging self.logger = logger self.config = config + self.skychunk = skychunk + self.ippToPspsDb = ippToPspsDb # set up empty lists @@ -46,18 +48,25 @@ self.regionsIngestedButOutOfDate = [] - # connect to the specified scratch database + # connect to the specified scratch database, if a name is given, otherwise use the first version try: - self.scratchDb = ScratchDb(self.logger, self.config, scratchDbName) + self.scratchDb = ScratchDb(self.logger, self.config, '1', scratchDbName) except: raise + # we have (still) 3 modes of ingest from dvo cpt/cps/cpm files + # 1) dvograbber (useFullTables == FALSE), + # 2) jython/stilts (useStilts == TRUE, useFullTables == TRUE) + # 3) dvopsps (useStilts == FALSE, useFullTables == TRUE) + # my goal (EAM, 2013/01/31) is to migrate to only dvopsps + + self.useStilts = 0 self.scratchDb.setUseFullTables(True) # decide if we are using the right DVO for this scratchDb - self.correctDvo = self.scratchDb.isCorrectDvo(self.config.dvoLocation) + self.correctDvo = self.scratchDb.isCorrectDvo(self.skychunk.dvoLocation) # this scratch Db either has no DVO info ingested, or a different DVO ingested. Prompt user if not self.correctDvo: print "*******************************************************************************" - self.logger.warning("Wrong or no DVO in use. Do you want to reset and use '" + self.config.dvoLocation + "'?") + self.logger.warning("Wrong or no DVO in use. Do you want to reset and use '" + self.skychunk.dvoLocation + "'?") response = raw_input("(y/n) ") if response == "y": @@ -101,5 +110,5 @@ # check if we have up-to-date version - path = self.config.dvoLocation + "/Photcodes.dat" + path = self.skychunk.dvoLocation + "/Photcodes.dat" if self.scratchDb.alreadyImportedThisDvoTable(path): self.logger.debugPair("DVO Photcodes.dat file", "up-to-date") @@ -126,5 +135,5 @@ # check if we have up-to-date version - path = self.config.dvoLocation + "/Images.dat" + path = self.skychunk.dvoLocation + "/Images.dat" if self.scratchDb.alreadyImportedThisDvoTable(path): self.logger.debugPair("DVO Images.dat file", "up-to-date") @@ -151,5 +160,5 @@ if not self.correctDvo: return False - path = self.config.dvoLocation + "/SkyTable.fits" + path = self.skychunk.dvoLocation + "/SkyTable.fits" if self.scratchDb.alreadyImportedThisDvoTable(path): self.logger.debugPair("DVO SkyTable.fits file", "up-to-date") @@ -158,8 +167,33 @@ self.logger.infoSeparator() self.logger.infoPair("DVO SkyTable.fits file", "NOT up-to-date") - self.importFits( + + if self.useStilts: + self.importFits( path, "R_MIN R_MAX D_MIN D_MAX INDEX NAME", self.scratchDb.dvoSkyTable) + else: + + # create dvoSkyTable + sql = "drop TABLE dvoSkyTable" + try: + self.scratchDb.execute(sql) + except: pass + + # create dvoSkyTable + sql = "CREATE TABLE dvoSkyTable (R_MIN REAL, R_MAX REAL, D_MIN REAL, D_MAX REAL, INDEX_ INT, NAME CHAR(18))" + self.scratchDb.execute(sql) + + # TODO path to DVO prog hardcoded temporarily + cmd = "dvopsps skytable" + cmd += " -dbhost " + self.scratchDb.dbHost + cmd += " -dbname " + self.scratchDb.dbName + cmd += " -dbuser " + self.scratchDb.dbUser + cmd += " -dbpass " + self.scratchDb.dbPass + cmd += " -D CATDIR " + self.skychunk.dvoLocation + + self.logger.infoPair("Running dvopsps", cmd) + p = Popen(cmd, shell=True, stdout=PIPE) + p.wait() self.logger.infoPair("Adding index to", self.scratchDb.dvoSkyTable) @@ -168,5 +202,5 @@ self.scratchDb.setImportedThisDvoTable(path) - self.logger.infoPair("Finished importing SkyTable at", self.config.dvoLocation) + self.logger.infoPair("Finished importing SkyTable at", self.skychunk.dvoLocation) return True @@ -231,5 +265,5 @@ paths = [] for fileType in self.ingestFileTypes: - paths.append(self.config.dvoLocation + "/" + regionPath + "." + fileType) + paths.append(self.skychunk.dvoLocation + "/" + regionPath + "." + fileType) # check for the existence of all interested file types @@ -304,5 +338,5 @@ # go no further if we've already partly ingested a different DVO if not self.correctDvo: - self.logger.infoPair("Wrong DVO in use", self.config.dvoLocation) + self.logger.infoPair("Wrong DVO in use", self.skychunk.dvoLocation) return @@ -387,8 +421,10 @@ size = 0.0 for region in regions: - + self.logger.infoPair("sizes for region", region) # get combined size of all interested files for fileType in self.ingestFileTypes: - size = size + self.getDiskSize(self.config.dvoLocation + "/" + region + "." + fileType) + size = size + self.getDiskSize(self.skychunk.dvoLocation + "/" + region + "." + fileType) + # EAM TEST I/O + self.logger.infoPair(fileType, size) return size @@ -519,5 +555,5 @@ try: if attempts > 0: self.logger.infoPair("Attempt %d to write" % attempts, table.name) - table.cmd_keepcols("CODE NAME TYPE_AS_INT").write(self.scratchDb.url + '#' + tableName) + table.cmd_keepcols("CODE NAME TYPE_AS_INT C_LAM K").write(self.scratchDb.url + '#' + tableName) break except: @@ -532,2 +568,79 @@ return tableName + + ''' + ingest skyregion into MySQL database using the native DVO program dvopsps + Populates dvoDetections and dvoDone with FITS tables from DVO for the defined sky area, this + includes purging detections outside sky area + ''' + def nativeIngestDetections(self, boxId, raCenter, decCenter, boxSize): + + # XXX put the chunk below in a separate method + # blow away existing dvoDetection table + + # clear the 'ingested' field for all boxes + self.ippToPspsDb.clearIngestedBoxes() + + # drop detections table + self.logger.infoPair("Dropping table", self.scratchDb.dvoDetectionTable) + self.scratchDb.dropTable(self.scratchDb.dvoDetectionTable) + + # create detections table + self.logger.infoPair("Creating table", self.scratchDb.dvoDetectionTable) + sql = "CREATE TABLE " + self.scratchDb.dvoDetectionTable + " LIKE dvoDetection" + try: self.scratchDb.execute(sql) + except: + self.logger.errorPair("Unable to create table", self.scratchDb.dvoDetectionTable) + return False + self.scratchDb.changeEngineToInnoDB(self.scratchDb.dvoDetectionTable) + + # add fileID column + sql = "ALTER TABLE " + self.scratchDb.dvoDetectionTable + " ADD fileID INT NOT NULL" + try: self.scratchDb.execute(sql) + except: + self.logger.errorPair("Unable to add fileID column to ", self.scratchDb.dvoDetectionTable) + return False + + # add an index to the fileID column + self.scratchDb.createIndex(self.scratchDb.dvoDetectionTable, "fileID") + + # make sure we have an up-to-date Images table + self.loadImages() + + # dvopsps -D catdir CATDIR -region .... -dbhost xx -dbname xx -dbuser xx -dbpass xx + halfSize = boxSize / 2.0 + # -region raCenter-halfSize raCenter+halfSize decCenter-halfSize decCenter+halfSize + + # TODO path to DVO prog hardcoded temporarily + cmd = "dvopsps detections" + cmd += " -dbhost " + self.scratchDb.dbHost + cmd += " -dbname " + self.scratchDb.dbName + cmd += " -dbuser " + self.scratchDb.dbUser + cmd += " -dbpass " + self.scratchDb.dbPass + cmd += " -D CATDIR " + self.skychunk.dvoLocation + cmd += " -region " + cmd += " " + str(raCenter-halfSize) + cmd += " " + str(raCenter+halfSize) + cmd += " " + str(decCenter-halfSize) + cmd += " " + str(decCenter+halfSize) + + if self.skychunk.parallel: + cmd += " -parallel" + + self.logger.infoPair("Running dvopsps", cmd) + p = Popen(cmd, shell=True, stdout=PIPE) + p.wait() + + self.ippToPspsDb.setIngestedBox(boxId) + + # add fileID column + sql = "ANALYZE TABLE " + self.scratchDb.dvoDetectionTable + try: self.scratchDb.execute(sql) + except: + self.logger.errorPair("Unable to analyze mysql table", self.scratchDb.dvoDetectionTable) + return False + + # update lists after attempted sync + # self.printSummary() + + return True Index: trunk/ippToPsps/jython/dvodetections.py =================================================================== --- trunk/ippToPsps/jython/dvodetections.py (revision 34884) +++ trunk/ippToPsps/jython/dvodetections.py (revision 35097) @@ -20,7 +20,7 @@ Constructor ''' - def __init__(self, logger, config, scratchDbName): + def __init__(self, logger, config, skychunk, ippToPspsDb, scratchDbName): - super(DvoDetections, self).__init__(logger, config, scratchDbName) + super(DvoDetections, self).__init__(logger, config, skychunk, ippToPspsDb, scratchDbName) # declare DVO file types of interest @@ -32,6 +32,6 @@ def ingestRegion(self, region): - cpmPath = self.config.dvoLocation + "/" + region + ".cpm" - cptPath = self.config.dvoLocation + "/" + region + ".cpt" + cpmPath = self.skychunk.dvoLocation + "/" + region + ".cpm" + cptPath = self.skychunk.dvoLocation + "/" + region + ".cpt" cpmTableName = self.scratchDb.getDbFriendlyTableName(region + ".cpm") @@ -83,21 +83,28 @@ # first try to add the new columns. catch failure and continue try: - sql = "ALTER TABLE " + cpmTableName + " ADD COLUMN (PSPS_OBJ_ID BIGINT), (ZP REAL), (RA FLOAT), (DEC FLOAT)" + sql = "ALTER TABLE " + cpmTableName + " ADD COLUMN (PSPS_OBJ_ID BIGINT), ADD COLUMN (ZP REAL), ADD COLUMN (RA FLOAT), ADD COLUMN (DEC_ FLOAT)" self.scratchDb.execute(sql) except: - self.logger.infoPair("Already added PSPS_OBJ_ID column to", cpmTableName) + # failed, but may be due to mysql error + self.logger.infoPair("mysql error or already added PSPS_OBJ_ID column to", cpmTableName) + # self.logger.infoPair("sql command", sql) # shove PSPS objIDs from cpt table and the calibrated zero point and coordinates which are formed # by combining values from multiple tables into cpt table - sql = "UPDATE " + cpmTableName + " AS a, " + cptTableName + " AS b, " \ - + self.scratchDb.dvoPhotcodesTable + " AS c \ - SET a.PSPS_OBJ_ID = b.EXT_ID, \ - b.ZP = c.C_LAM * 0.001 + c.K * (b.AIRMASS - 1) - b.M_CAL, \ - b.RA = a.RA - (b.D_RA / 3600.0), \ - b.DEC = a.DEC - (b.D_DEC / 3600.0) \ - WHERE a.CAT_ID = b.CAT_ID \ - AND a.OBJ_ID = b.OBJ_ID \ - AND b.PHOTCODE = c.CODE" + sql = "UPDATE " \ + + cpmTableName + " AS meas, " \ + + cptTableName + " AS ave, " \ + + self.scratchDb.dvoPhotcodesTable + " AS phot \ + SET \ + meas.PSPS_OBJ_ID = ave.EXT_ID, \ + meas.ZP = phot.C_LAM * 0.001 + phot.K * (meas.AIRMASS - 1) - meas.M_CAL, \ + meas.RA = ave.RA - (meas.D_RA / 3600.0), \ + meas.DEC_ = ave.DEC_ - (meas.D_DEC / 3600.0) \ + WHERE meas.CAT_ID = ave.CAT_ID \ + AND meas.OBJ_ID = ave.OBJ_ID \ + AND meas.PHOTCODE = phot.CODE" + # EAM : add some debug I/O + # self.logger.infoPair("sql command", sql) self.scratchDb.execute(sql) @@ -136,9 +143,12 @@ ,POW(10.0, 0.4 * M_TIME) \ ,RA \ - ,DEC \ + ,DEC_ \ ,X_CCD_ERR*PLTSCALE \ ,Y_CCD_ERR*PLTSCALE \ , " + str(fileId) + " \ FROM " + cpmTableName + + # print "sql :", sql + # response = raw_input("prepare to insert dvoDetectionTable ") try: @@ -148,4 +158,5 @@ return False finally: + # response = raw_input("insert dvoDetectionTable ") self.logger.infoPair("Dropping tables", "%s and %s" % (cpmTableName, cptTableName)) self.scratchDb.dropTable(cpmTableName) @@ -155,4 +166,2 @@ return True - - Index: trunk/ippToPsps/jython/dvoobjects.py =================================================================== --- trunk/ippToPsps/jython/dvoobjects.py (revision 34884) +++ trunk/ippToPsps/jython/dvoobjects.py (revision 35097) @@ -8,7 +8,7 @@ import logging import glob +from subprocess import call, PIPE, Popen from dvo import Dvo - ''' @@ -20,7 +20,7 @@ Constructor ''' - def __init__(self, logger, config, scratchDbName=None): + def __init__(self, logger, config, skychunk, ippToPspsDb, scratchDbName=None): - super(DvoObjects, self).__init__(logger, config, scratchDbName) + super(DvoObjects, self).__init__(logger, config, skychunk, ippToPspsDb, scratchDbName) # declare DVO file types of interest @@ -37,5 +37,5 @@ for fileType in self.ingestFileTypes: - path = self.config.dvoLocation + "/" + region + "." + fileType + path = self.skychunk.dvoLocation + "/" + region + "." + fileType tableName = self.scratchDb.getDbFriendlyTableName(region + "." + fileType) @@ -47,2 +47,32 @@ + ''' + ingest object data into MySQL database using the native DVO program dvopsps + Populates dvoDetections and dvoDone with FITS tables from DVO for the defined sky area, this + includes purging detections outside sky area + ''' + def nativeIngestRegion(self, region): + + # drop cpt/cps/object table? + # create detections table? + + # TODO path to DVO prog hardcoded temporarily + cmd = "dvopsps objects" + cmd += " -dbhost " + self.scratchDb.dbHost + cmd += " -dbname " + self.scratchDb.dbName + cmd += " -dbuser " + self.scratchDb.dbUser + cmd += " -dbpass " + self.scratchDb.dbPass + cmd += " -D CATDIR " + self.skychunk.dvoLocation + cmd += " -cpt " + region + + if self.skychunk.parallel: + cmd += " -parallel" + + self.logger.infoPair("Running dvopsps", cmd) + p = Popen(cmd, shell=True, stdout=PIPE) + p.wait() + + # update lists after attempted sync + # self.printSummary() + + return True Index: trunk/ippToPsps/jython/fits.py =================================================================== --- trunk/ippToPsps/jython/fits.py (revision 34884) +++ trunk/ippToPsps/jython/fits.py (revision 35097) @@ -137,5 +137,5 @@ # this regex will get param/value pairs for all header cards, ignoring comments and parsing out 'HIERACH' prefixes - match = re.match('^(HIERARCH )*([a-zA-Z0-9-_\.]+)\s*=\s+\'*([a-zA-Z0-9-_\.:\s@#]+)\'*\\/*', record) + match = re.match('^(HIERARCH )*([a-zA-Z0-9-_\.]+)\s*=\s+\'*([a-zA-Z0-9-+_\.:\s@#]+)\'*\\/*', record) if match: Index: trunk/ippToPsps/jython/gpc1db.py =================================================================== --- trunk/ippToPsps/jython/gpc1db.py (revision 34884) +++ trunk/ippToPsps/jython/gpc1db.py (revision 35097) @@ -21,5 +21,11 @@ ''' def __init__(self, logger, config): - super(Gpc1Db, self).__init__(logger, config, "gpc1database") + # define database type + if (config.test): + dbType = "gpc1database_test" + else: + dbType = "gpc1database" + + super(Gpc1Db, self).__init__(logger, config, dbType) ''' @@ -93,4 +99,6 @@ try: + # XXX EAM : test output + self.logger.infoPair("sql for dvo items:", sql) rs = self.executeQuery(sql) while (rs.next()): @@ -112,4 +120,5 @@ self.logger.debug("Querying GPC1 for image IDs for stack ID: " + str(stackID)) + # JOIN stackInputSkyfile shoule use stack_id as well as warp_id, right? (maybe not -- the join to warp is first) sql = "SELECT DISTINCT CONCAT(exp_id, SUBSTR(class_id, 3, 4)) FROM ( \ SELECT DISTINCT exp_id,class_id \ @@ -226,4 +235,5 @@ if path.startswith("neb"): + # XXX EAM : this should just use the name (path.smf) instead of a wildcard, right? f=os.popen("neb-ls -p "+path+"%smf") for i in f.readlines(): @@ -234,4 +244,6 @@ files = glob.glob(path + ".smf") + # XXX EAM : test output + self.logger.infoPair("smf files:", files) if len(files) < 1: return None @@ -290,45 +302,75 @@ + # print "staring stack stage cmf" + # get single path base for the directory containing all cmf files, one of which is hopefully for our stack_id try: rs = self.executeQuery(sql) rs.first() + except: + self.logger.errorPair("Can't query for stack cmfs", "stack_id = %d" % stackID) + return None + + try: pathBase = rs.getString(1) + except: + self.logger.errorPair("No stack cmf files found for stack_id = %d" % stackID) + return None + + # find the files (are they in nebulous or in the filesystem?) + files = [] + if pathBase.startswith("neb"): + # XXX EAM : this should just use the name (path.smf) instead of a wildcard, right? + f=os.popen("neb-ls -p " + pathBase + "%cmf") + for i in f.readlines(): + files.append(i.rstrip()) + + # or not a neb path + else: + files = glob.glob(pathBase + ".cmf") + + # print "stack cmf files:", files + if len(files) < 1: return None + + # if we get here, then the cmf is readable, now check the stack_id + fits = Fits(self.logger, self.config, files[0]) + return fits + + # XXX validate the file? + # if fits.getPrimaryHeaderValue("STK_ID") == str(stackID): # now find and loop through all cmf files at this path_base - files=os.popen("neb-ls " + pathBase + "%cmf") - for i in files.readlines(): - - nebPath = i.rstrip() - - # now attempt to run 'neb-ls -p' to actually read this cmf file - try: - self.logger.debug("Trying to neb-ls -p on '" + nebPath + "'") - paths=os.popen("neb-ls -p " + nebPath) - path = paths.readline().rstrip() - if len(path) < 1: - self.logger.debug("zero length path - skipping") - raise - - # if we get here, then the cmf is readable, now check the stack_id - fits = Fits(self.logger, self.config, path) - if fits.getPrimaryHeaderValue("STK_ID") == str(stackID): - # we have the right file! - self.logger.debug("SUCCESS!: %s == %s"% (fits.getPrimaryHeaderValue("STK_ID"), str(stackID))) - return fits - else: - # this is not the correct file - self.logger.debug("STACK ID: %s != %s"% (fits.getPrimaryHeaderValue("STK_ID"), str(stackID))) - - # an exception here means that nebulous could not read this file, so we just skip it - except: - self.logger.debug("NEB FAILED SKIPPING") - pass - - except: - self.logger.errorPair("Can't query for stack cmfs", "stack_id = %d" % stackID) - - self.logger.errorPair("Couldn't find stack cmf for", "stack_id = %d" % stackID) - return None + + ## XXX the code below was needed when we ingested from staticsky (N files per run) + ## XXX ingesting from skycal solves this problem (1 file per run) + ## files=os.popen("neb-ls " + pathBase + "%cmf") + ## for i in files.readlines(): + ## nebPath = i.rstrip() + ## + ## # now attempt to run 'neb-ls -p' to actually read this cmf file + ## try: + ## self.logger.debug("Trying to neb-ls -p on '" + nebPath + "'") + ## paths=os.popen("neb-ls -p " + nebPath) + ## path = paths.readline().rstrip() + ## if len(path) < 1: + ## self.logger.debug("zero length path - skipping") + ## raise + ## + ## print "starint stack stage cmf 3" + ## + ## # if we get here, then the cmf is readable, now check the stack_id + ## fits = Fits(self.logger, self.config, path) + ## if fits.getPrimaryHeaderValue("STK_ID") == str(stackID): + ## # we have the right file! + ## self.logger.debug("SUCCESS!: %s == %s"% (fits.getPrimaryHeaderValue("STK_ID"), str(stackID))) + ## return fits + ## else: + ## # this is not the correct file + ## self.logger.debug("STACK ID: %s != %s"% (fits.getPrimaryHeaderValue("STK_ID"), str(stackID))) + ## + ## # an exception here means that nebulous could not read this file, so we just skip it + ## except: + ## self.logger.debug("NEB FAILED SKIPPING") + ## pass ''' Index: trunk/ippToPsps/jython/initbatch.py =================================================================== --- trunk/ippToPsps/jython/initbatch.py (revision 34884) +++ trunk/ippToPsps/jython/initbatch.py (revision 35097) @@ -28,4 +28,5 @@ logger, config, + skychunk, gpc1Db, ippToPspsDb, @@ -34,4 +35,5 @@ super(InitBatch, self).__init__(logger, config, + skychunk, gpc1Db, ippToPspsDb, Index: trunk/ippToPsps/jython/ippjython =================================================================== --- trunk/ippToPsps/jython/ippjython (revision 35097) +++ trunk/ippToPsps/jython/ippjython (revision 35097) @@ -0,0 +1,49 @@ +#!/bin/csh -f + +# set JAVA = ~ipp/jre1.7.0_03/bin/java +# set JYTHON = ~ipp/jython2.5.3b1/jython.jar + +# a bit crude : get the datadir path by interpollation of the PATH element for psconfig +foreach f ( `echo $PATH | tr : ' '` ) + echo $f | grep $PSCONFDIR > /dev/null + if ($status) continue + set datadir = `echo $f | sed s/bin\$/share/` + break +end + +set JARDIR = $datadir/jar +set JYDIR = $datadir/jython +setenv IPPTOPSPS_DATA $datadir/ipptopsps + +# XXX install jython in JARDIR +set JYTHON = $JARDIR/jython2.5.3b1/jython.jar +# set JYTHON = ~/jython2.5.3b1/jython.jar + +if (! -r $JYTHON) then + echo "jython jar file not found, check jython installation" + exit 2 +endif + +if ($?JAVA == 0) then + which java >& /dev/null + if ($status) then + echo "java not found, please specify location in environment or install in path" + exit 2 + endif + set JAVA = `which java` +endif + +set CLASSPATH = $JARDIR/stilts.jar\:$JARDIR/commons-math-2.2.jar\:$JARDIR/mysql-connector-java-5.1.14-bin.jar\:$JYDIR + +if ($#argv < 1) then + echo "USAGE ippjython (classfile) [options]" + exit 2 +endif + +set classfile = $1 +shift + +# XXX if you want to run a test on a local file, use this line: +# $JAVA -Xbootclasspath/a:$CLASSPATH -jar $JYTHON $classfile $* + +$JAVA -Xbootclasspath/a:$CLASSPATH -jar $JYTHON $JYDIR/$classfile $* Index: trunk/ippToPsps/jython/ipptopsps.py =================================================================== --- trunk/ippToPsps/jython/ipptopsps.py (revision 34884) +++ trunk/ippToPsps/jython/ipptopsps.py (revision 35097) @@ -1,3 +1,4 @@ #!/usr/bin/env jython +# EAM : config -> skychunk DONE import signal @@ -9,4 +10,5 @@ from config import Config +from skychunk import Skychunk from pslogger import PSLogger from ipptopspsdb import IppToPspsDb @@ -23,12 +25,13 @@ def __init__(self, argv, logToStdout=1, logToFile=0): - # are the arsg ok? all programs require a config name + # set up config object (global config information, also parse command-line options, eg -test / -t) + self.config = Config() + + # are the args ok? all programs require a config name if len(sys.argv) < 2: self.printUsage() self.exitProgram("Wrong args") - + # some class constants - self.PROGNAME = sys.argv[0] - CONFIGNAME = sys.argv[1] self.HOST = socket.gethostname() self.PID = os.getpid() @@ -38,18 +41,12 @@ self.SECONDS = None - self.createNewConfig = False - self.rotateConfigs = False - # a new config? - if CONFIGNAME == "edit": self.createNewConfig = True - # should we rotate configs for this program? - elif CONFIGNAME == "all": self.rotateConfigs = True + self.logger = self.config.getLogger(self.HOST, self.PID, logToStdout, logToFile) - # set up config object - self.config = Config(self.PROGNAME, CONFIGNAME) - self.logger = self.config.getLogger(self.HOST, self.PID, logToStdout, logToFile) + # argv[1] -> self.skychunk.name + self.skychunk = Skychunk(self.logger) # create connection to databases database try: - self.ippToPspsDb = IppToPspsDb(self.logger, self.config) + self.ippToPspsDb = IppToPspsDb(self.logger, self.config, self.skychunk) except: self.exitProgram("Could not connect to ipptopsps Db") @@ -63,8 +60,8 @@ # title for log self.logger.infoSeparator() - self.logger.infoTitle("ippToPsps '" + self.PROGNAME + "' started") + self.logger.infoTitle("ippToPsps '" + self.config.programName + "' started") self.logger.infoPair("Host", self.HOST) self.logger.infoPair("PID", "%d" % self.PID) - self.logger.infoPair("Config", self.config.name) + self.logger.infoPair("Skychunk", self.skychunk.name) ''' @@ -104,5 +101,5 @@ # write message to log or stdout if no logger set up - msg = sys.argv[0] + " " + extra + msg = sys.argv[0] + " " + extra try: self.logger.errorPair("Usage:", msg) @@ -111,46 +108,46 @@ ''' - Refreshes the config then recreates objects with new settings + Refreshes the skychunk then recreates objects with new settings ''' - def refreshConfig(self): + def refreshSkychunk(self): - # new config? - if self.createNewConfig: - self.ippToPspsDb.editConfig() - self.ippToPspsDb.setConfigForThisClient(self.config.name, self.HOST, self.PID) - self.createNewConfig = False + # new skychunk? + if self.skychunk.createNewSkychunk: + self.ippToPspsDb.editSkychunk() + self.ippToPspsDb.setSkychunkForThisClient(self.skychunk.name, self.HOST, self.PID) + self.skychunk.createNewSkychunk = False - # if we are rotating configs, then look for next active one in list - if self.rotateConfigs: - self.ippToPspsDb.getConfigForThisClient(self.HOST, self.PID) - configs = self.ippToPspsDb.getActiveConfigList() + # if we are rotating skychunks, then look for next active one in list + if self.skychunk.rotateSkychunks: + self.ippToPspsDb.getSkychunkForThisClient(self.HOST, self.PID) + skychunks = self.ippToPspsDb.getActiveSkychunkList() i = 1 - for config in configs: - if config == self.config.name: break + for skychunk in skychunks: + if skychunk == self.skychunk.name: break i += 1 - if i >= len(configs): newConfig = configs[0] - else: newConfig = configs[i] + if i >= len(skychunks): newSkychunk = skychunks[0] + else: newSkychunk = skychunks[i] - self.ippToPspsDb.setConfigForThisClient(newConfig, self.HOST, self.PID) + self.ippToPspsDb.setSkychunkForThisClient(newSkychunk, self.HOST, self.PID) - return self.ippToPspsDb.readConfig(self.HOST, self.PID) + return self.ippToPspsDb.readSkychunk(self.HOST, self.PID) ''' - All implementing subclasses (programs) should call this method once in a while to see if it should be paused or killed, otherwise this updates the clients table with current time and reloads the config data + All implementing subclasses (programs) should call this method once in a while to see if it should be paused or killed, otherwise this updates the clients table with current time and reloads the skychunk data ''' def checkClientStatus(self): - self.ippToPspsDb.updateClient(self.PROGNAME, self.HOST, self.PID) + self.ippToPspsDb.updateClient(self.config.programName, self.HOST, self.PID) if self.ippToPspsDb.isKilled(self.HOST, self.PID): self.exitProgram("killed via Db") - # this loop pauses the process if we have no config or we have set it to pause + # this loop pauses the process if we have no skychunk or we have set it to pause firstTimeIn = True - while not self.refreshConfig() or self.ippToPspsDb.isPaused(self.HOST, self.PID): + while not self.refreshSkychunk() or self.ippToPspsDb.isPaused(self.HOST, self.PID): - self.ippToPspsDb.updateClient(self.PROGNAME, self.HOST, self.PID) + self.ippToPspsDb.updateClient(self.config.programName, self.HOST, self.PID) if self.ippToPspsDb.isKilled(self.HOST, self.PID): self.exitProgram("killed while paused") @@ -175,5 +172,5 @@ # write message to log or stdout if no logger set up try: - self.logger.infoPair(self.PROGNAME + " exited", exitReason) + self.logger.infoPair(self.config.programName + " exited", exitReason) except: print "*** Program exited: " + exitReason @@ -186,4 +183,2 @@ sys.exit(0) - - Index: trunk/ippToPsps/jython/ipptopspsdb.py =================================================================== --- trunk/ippToPsps/jython/ipptopspsdb.py (revision 34884) +++ trunk/ippToPsps/jython/ipptopspsdb.py (revision 35097) @@ -17,6 +17,12 @@ Constructor ''' - def __init__(self, logger, config): - super(IppToPspsDb, self).__init__(logger, config, "ipptopspsdatabase") + def __init__(self, logger, config, skychunk): + if (config.test): + dbType = "ipptopspsdatabase_test" + else: + dbType = "ipptopspsdatabase" + super(IppToPspsDb, self).__init__(logger, config, dbType) + + self.skychunk = skychunk self.MAX_FAILS = 5 @@ -29,7 +35,7 @@ sql = "SELECT DISTINCT batch_id \ FROM batch \ - WHERE timestamp > '" + self.config.epoch + "' \ + WHERE timestamp > '" + self.skychunk.epoch + "' \ AND batch_type = '" + batchType + "' \ - AND dvo_db = '" + self.config.dvoLabel + "' \ + AND dvo_db = '" + self.skychunk.dvoLabel + "' \ AND merged = 1 \ AND " + column + " = 0" @@ -54,7 +60,7 @@ sql = "SELECT DISTINCT batch_id \ FROM batch \ - WHERE timestamp > '" + self.config.epoch + "' \ + WHERE timestamp > '" + self.skychunk.epoch + "' \ AND batch_type = '" + batchType + "' \ - AND dvo_db = '" + self.config.dvoLabel + "' \ + AND dvo_db = '" + self.skychunk.dvoLabel + "' \ AND purged = 1 \ AND " + column + " = 0" @@ -80,7 +86,7 @@ sql = "SELECT DISTINCT batch_id \ FROM batch \ - WHERE timestamp > '" + self.config.epoch + "' \ + WHERE timestamp > '" + self.skychunk.epoch + "' \ AND batch_type = '" + batchType + "' \ - AND dvo_db = '" + self.config.dvoLabel + "' \ + AND dvo_db = '" + self.skychunk.dvoLabel + "' \ AND (loaded_to_ODM = -1 OR merge_worthy = 1) \ AND " + column + " = 0" @@ -148,7 +154,7 @@ sql = "SELECT DISTINCT batch_id \ FROM batch \ - WHERE timestamp > '" + self.config.epoch + "' \ + WHERE timestamp > '" + self.skychunk.epoch + "' \ AND batch_type = '" + batchType + "' \ - AND dvo_db = '" + self.config.dvoLabel + "' \ + AND dvo_db = '" + self.skychunk.dvoLabel + "' \ AND processed = 1\ AND loaded_to_datastore != 1 \ @@ -177,6 +183,6 @@ sql = "SELECT batch_id \ FROM batch \ - WHERE timestamp > '" + self.config.epoch + "' \ - AND dvo_db = '" + self.config.dvoLabel + "' \ + WHERE timestamp > '" + self.skychunk.epoch + "' \ + AND dvo_db = '" + self.skychunk.dvoLabel + "' \ AND loaded_to_datastore = 1 \ AND batch_type = '" + batchType + "' \ @@ -206,6 +212,6 @@ sql = "SELECT batch_id \ FROM batch \ - WHERE timestamp > '" + self.config.epoch + "' \ - AND dvo_db = '" + self.config.dvoLabel + "' \ + WHERE timestamp > '" + self.skychunk.epoch + "' \ + AND dvo_db = '" + self.skychunk.dvoLabel + "' \ AND loaded_to_datastore = 1 \ AND batch_type = '" + batchType + "' \ @@ -235,6 +241,6 @@ sql = "SELECT batch_id \ FROM batch \ - WHERE timestamp > '" + self.config.epoch + "' \ - AND dvo_db = '" + self.config.dvoLabel + "' \ + WHERE timestamp > '" + self.skychunk.epoch + "' \ + AND dvo_db = '" + self.skychunk.dvoLabel + "' \ AND merge_worthy = 1 \ AND batch_type = '" + batchType + "' \ @@ -262,6 +268,6 @@ sql = "SELECT DISTINCT batch_id \ FROM batch \ - WHERE timestamp > '" + self.config.epoch + "' \ - AND dvo_db = '" + self.config.dvoLabel + "' \ + WHERE timestamp > '" + self.skychunk.epoch + "' \ + AND dvo_db = '" + self.skychunk.dvoLabel + "' \ AND " + column + " = " + str(value) @@ -288,6 +294,6 @@ FROM batch \ WHERE batch_type = '" + batchType + "' \ - AND timestamp > '" + self.config.epoch + "' \ - AND dvo_db = '" + self.config.dvoLabel + "' \ + AND timestamp > '" + self.skychunk.epoch + "' \ + AND dvo_db = '" + self.skychunk.dvoLabel + "' \ AND processed = 1 \ AND loaded_to_datastore = 1 \ @@ -315,6 +321,6 @@ FROM batch \ WHERE batch_type = '" + batchType + "' \ - AND timestamp > '" + self.config.epoch + "' \ - AND dvo_db = '" + self.config.dvoLabel + "' \ + AND timestamp > '" + self.skychunk.epoch + "' \ + AND dvo_db = '" + self.skychunk.dvoLabel + "' \ AND " + column + " = " + str(value) @@ -387,6 +393,6 @@ WHERE batch_type = '" + batchType + "' \ AND processed = 1 \ - AND timestamp > '" + self.config.epoch + "' \ - AND dvo_db = '" + self.config.dvoLabel + "' \ + AND timestamp > '" + self.skychunk.epoch + "' \ + AND dvo_db = '" + self.skychunk.dvoLabel + "' \ AND timestamp BETWEEN now() - INTERVAL " + interval + " AND now()" @@ -412,6 +418,6 @@ WHERE batch_type = '" + batchType + "' \ AND loaded_to_datastore = 1 \ - AND timestamp > '" + self.config.epoch + "' \ - AND dvo_db = '" + self.config.dvoLabel + "' \ + AND timestamp > '" + self.skychunk.epoch + "' \ + AND dvo_db = '" + self.skychunk.dvoLabel + "' \ ORDER BY timestamp DESC LIMIT 1" @@ -511,5 +517,5 @@ WHERE stage_id = " + str(stage_id) + " \ AND batch_type = '" + batchType + "' \ - AND dvo_db = '" + self.config.dvoLabel + "' \ + AND dvo_db = '" + self.skychunk.dvoLabel + "' \ AND batch_type != 'IN' \ AND timestamp BETWEEN now() - INTERVAL 12 HOUR AND now()" @@ -554,6 +560,6 @@ WHERE stage_id = " + str(stage_id) + " \ AND batch_type = '" + batchType + "' \ - AND timestamp > '" + self.config.epoch + "' \ - AND dvo_db = '" + self.config.dvoLabel + "' \ + AND timestamp > '" + self.skychunk.epoch + "' \ + AND dvo_db = '" + self.skychunk.dvoLabel + "' \ AND batch_type != 'IN' \ AND processed = 1" @@ -579,6 +585,6 @@ WHERE stage_id = " + str(stage_id) + " \ AND batch_type = '" + batchType + "' \ - AND timestamp > '" + self.config.epoch + "' \ - AND dvo_db = '" + self.config.dvoLabel + "' \ + AND timestamp > '" + self.skychunk.epoch + "' \ + AND dvo_db = '" + self.skychunk.dvoLabel + "' \ AND processed = -1" @@ -602,6 +608,6 @@ FROM batch \ WHERE batch_type = '" + batchType + "' \ - AND timestamp > '" + self.config.epoch + "' \ - AND dvo_db = '" + self.config.dvoLabel + "' \ + AND timestamp > '" + self.skychunk.epoch + "' \ + AND dvo_db = '" + self.skychunk.dvoLabel + "' \ AND processed = -1 \ GROUP BY stage_id HAVING numoccur >= " + str(self.MAX_FAILS) @@ -634,5 +640,5 @@ batchID = -1; - if self.config.force or \ + if self.skychunk.force or \ (not self.alreadyProcessed(batchType, stageID) \ and not self.processingNow(batchType, stageID) \ @@ -650,7 +656,7 @@ '" + batchType + "', \ " + str(stageID) + ", \ - '" + self.config.survey + "', \ - '" + self.config.dvoLabel + "', \ - '" + self.config.datastoreProduct + "' \ + '" + self.skychunk.survey + "', \ + '" + self.skychunk.dvoLabel + "', \ + '" + self.skychunk.datastoreProduct + "' \ )" self.logger.infoPair("heather:","sql") @@ -749,18 +755,19 @@ ''' - Sets the config field for this client - ''' - def setConfigForThisClient(self, config, host, pid): - - self.execute("UPDATE clients SET config = '" + config + "' \ + Sets the skychunk field for this client + ''' + def setSkychunkForThisClient(self, skychunk, host, pid): + + self.execute("UPDATE clients SET skychunk = '" + skychunk + "' \ WHERE host = '" + host + "' \ AND pid = " + str(pid)) - ''' - Sets the config field for a set of loader clients - ''' - def setConfigForLoaders(self, config, ids): + + ''' + Sets the skychunk field for a set of loader clients + ''' + def setSkychunkForLoaders(self, skychunk, ids): for id in ids: - self.execute("UPDATE clients SET config = '" + config + "' \ + self.execute("UPDATE clients SET skychunk = '" + skychunk + "' \ WHERE type = 'loader.py' \ AND id = " + str(id)) @@ -796,5 +803,5 @@ ''' def insertClient(self, type, host, pid): - self.execute("INSERT INTO clients (timestamp, type, host, pid, config) VALUES (now(), '" + type + "', '" + host + "', " + str(pid) + ", '" + self.config.name + "')") + self.execute("INSERT INTO clients (timestamp, type, host, pid, skychunk) VALUES (now(), '" + type + "', '" + host + "', " + str(pid) + ", '" + self.skychunk.name + "')") ''' @@ -831,5 +838,5 @@ try: rs = self.executeQuery(sql) - rs.first() + if rs.first() is False: return False if rs.getInt(1) == 1: return True else: return False @@ -840,41 +847,41 @@ ''' - Returns a list of available configs - ''' - def getActiveConfigList(self): + Returns a list of available skychunks + ''' + def getActiveSkychunkList(self): - sql = "SELECT DISTINCT name FROM config WHERE active = 1 ORDER BY name" - - configs = [] - try: - rs = self.executeQuery(sql) - while (rs.next()): configs.append(rs.getString(1)) - except: - self.logger.errorPair("Can't get config names", sql) + sql = "SELECT DISTINCT name FROM skychunk WHERE active = 1 ORDER BY name" + + skychunks = [] + try: + rs = self.executeQuery(sql) + while (rs.next()): skychunks.append(rs.getString(1)) + except: + self.logger.errorPair("Can't get skychunk names", sql) finally: rs.close() - return configs - - ''' - Prompts user for info about a new or existing config and lets them neter the various details - ''' - def editConfig(self): + return skychunks + + ''' + Prompts user for info about a new or existing skychunk and lets them neter the various details + ''' + def editSkychunk(self): print " ********************************************************************" - print " * Config editor" + print " * Skychunk editor" print " *" - response = raw_input(" * Name for new config, or existing config to edit? ") + response = raw_input(" * Name for new skychunk, or existing skychunk to edit? ") if response == "": return - self.config.name = response - - # attempt to insert new config, if it already exists then carry on with update - sql = "INSERT INTO config (name) VALUES ('" + self.config.name + "')" + self.skychunk.name = response + + # attempt to insert new skychunk, if it already exists then carry on with update + sql = "INSERT INTO skychunk (name) VALUES ('" + self.skychunk.name + "')" try: self.execute(sql) except: pass - # for each column in the config table - sql = "SHOW COLUMNS FROM config" + # for each column in the skychunk table + sql = "SHOW COLUMNS FROM skychunk" rs = self.executeQuery(sql) while (rs.next()): @@ -887,8 +894,10 @@ if field == 'timestamp': continue if field == 'name': continue + if field == 'ingested_det': continue + if field == 'ingested_obj': continue # get current value for this field, if any try: - rs2 = self.executeQuery("SELECT " + field + " FROM config WHERE name = '" + self.config.name + "'") + rs2 = self.executeQuery("SELECT " + field + " FROM skychunk WHERE name = '" + self.skychunk.name + "'") rs2.first() default = rs2.getString(1) @@ -907,5 +916,5 @@ if type.find("varchar") != -1 or type.find("timestamp") != -1: quotes = "'" else: quotes = "" - self.execute("UPDATE config SET " + field + " = " + quotes + response + quotes + " WHERE name = '" + self.config.name + "'") + self.execute("UPDATE skychunk SET " + field + " = " + quotes + response + quotes + " WHERE name = '" + self.skychunk.name + "'") print " *" @@ -913,9 +922,9 @@ ''' - Returns config for this client - ''' - def getConfigForThisClient(self, host, pid): - - sql = "SELECT config \ + Returns skychunk for this client + ''' + def getSkychunkForThisClient(self, host, pid): + + sql = "SELECT skychunk \ FROM clients \ WHERE host = '" + host + "' \ @@ -925,8 +934,8 @@ rs = self.executeQuery(sql) rs.first() - self.config.name = rs.getString(1) - except: - self.logger.errorPair("No config set for", "%s (pid=%d)" % (host, pid)) - self.config.isLoaded = False + self.skychunk.name = rs.getString(1) + except: + self.logger.errorPair("No skychunk set for", "%s (pid=%d)" % (host, pid)) + self.skychunk.isLoaded = False return False @@ -934,12 +943,12 @@ ''' - Reads config from the database and populates Config object - ''' - def readConfig(self, host, pid): - - # first get config defined for this client - if not self.getConfigForThisClient(host, pid): return False - - # now load that config + Reads skychunk from the database and populates Skychunk object + ''' + def readSkychunk(self, host, pid): + + # first get skychunk defined for this client + if not self.getSkychunkForThisClient(host, pid): return False + + # now load that skychunk sql = "SELECT \ datastore_product \ @@ -964,48 +973,59 @@ ,queue_ST \ ,queue_OB \ - FROM config \ - WHERE name = '" + self.config.name + "'" - - self.config.batchTypes = [] - try: - rs = self.executeQuery(sql) + ,parallel \ + FROM skychunk \ + WHERE name = '" + self.skychunk.name + "'" + + self.skychunk.batchTypes = [] + try: + rs = self.executeQuery(sql) + rs.first() - self.config.datastoreProduct = rs.getString(1) - self.config.datastoreType = rs.getString(2) - if rs.getInt(3) == 1: self.config.datastorePublishing = True - else: self.config.datastorePublishing = False - self.config.dvoLabel = rs.getString(4) + self.skychunk.datastoreProduct = rs.getString(1) + self.skychunk.datastoreType = rs.getString(2) + if rs.getInt(3) == 1: self.skychunk.datastorePublishing = True + else: self.skychunk.datastorePublishing = False + self.skychunk.dvoLabel = rs.getString(4) # if dvoLabel is null is can break queries, so set to something - if not self.config.dvoLabel: self.config.dvoLabel = "none" - self.config.dvoLocation = rs.getString(5) - self.config.minRa = rs.getDouble(6) - self.config.maxRa = rs.getDouble(7) - self.config.minDec = rs.getDouble(8) - self.config.maxDec = rs.getDouble(9) - - self.config.boxSize = rs.getDouble(10) - self.config.halfBox = self.config.boxSize/2.0 - self.config.boxSizeWithBorder = self.config.boxSize + (self.config.BORDER * 2) - - self.config.basePath = rs.getString(11) - self.config.dataRelease = rs.getInt(12) - self.config.deleteLocal = rs.getInt(13) - self.config.deleteDatastore = rs.getInt(14) - self.config.deleteDxLayer = rs.getInt(15) - self.config.epoch = rs.getString(16) - self.config.survey = rs.getString(17) - self.config.pspsSurvey = rs.getString(18) - if rs.getInt(19) == 1: self.config.batchTypes.append("P2") - if rs.getInt(20) == 1: self.config.batchTypes.append("ST") - if rs.getInt(21) == 1: self.config.batchTypes.append("OB") - self.config.force = False # TODO - self.config.test = False # TODO - self.config.isLoaded = True - except: - self.logger.errorPair("Could not read config with name", self.config.name) - self.config.isLoaded = False - - return self.config.isLoaded + if not self.skychunk.dvoLabel: self.skychunk.dvoLabel = "none" + self.skychunk.dvoLocation = rs.getString(5) + self.skychunk.minRa = rs.getDouble(6) + self.skychunk.maxRa = rs.getDouble(7) + self.skychunk.minDec = rs.getDouble(8) + self.skychunk.maxDec = rs.getDouble(9) + + self.skychunk.boxSize = rs.getDouble(10) + self.skychunk.halfBox = self.skychunk.boxSize/2.0 + self.skychunk.boxSizeWithBorder = self.skychunk.boxSize + (self.skychunk.BORDER * 2) + + self.skychunk.basePath = rs.getString(11) + self.skychunk.dataRelease = rs.getInt(12) + + self.skychunk.deleteLocal = rs.getInt(13) + self.skychunk.deleteDatastore = rs.getInt(14) + self.skychunk.deleteDxLayer = rs.getInt(15) + + self.skychunk.epoch = rs.getString(16) + + self.skychunk.survey = rs.getString(17) + self.skychunk.pspsSurvey = rs.getString(18) + + if rs.getInt(19) == 1: self.skychunk.batchTypes.append("P2") + if rs.getInt(20) == 1: self.skychunk.batchTypes.append("ST") + if rs.getInt(21) == 1: self.skychunk.batchTypes.append("OB") + + self.skychunk.force = True # TODO + self.skychunk.parallel = False # TODO + + if rs.getInt(22) == 1: self.skychunk.parallel = True + + if self.skychunk.parallel: print "USING parallel" + self.skychunk.isLoaded = True + except: + self.logger.errorPair("Could not read skychunk with name", self.skychunk.name) + self.skychunk.isLoaded = False + + return self.skychunk.isLoaded ''' @@ -1015,8 +1035,8 @@ sql = "SELECT id FROM box \ - WHERE config = '" + self.config.name + "' \ + WHERE skychunk = '" + self.skychunk.name + "' \ AND ra_center = " + str(ra) + " \ AND dec_center = " + str(dec) + " \ - AND box_side = " + str(self.config.boxSize) + AND box_side = " + str(self.skychunk.boxSize) id = -1 @@ -1034,5 +1054,5 @@ ''' def removeAllBoxes(self): - self.execute("DELETE FROM box WHERE config = '" + self.config.name + "'") + self.execute("DELETE FROM box WHERE skychunk = '" + self.skychunk.name + "'") ''' @@ -1047,7 +1067,7 @@ sql = "INSERT INTO box \ - (config, ra_center, dec_center, box_side) \ + (skychunk, ra_center, dec_center, box_side) \ VALUES \ - ('" + self.config.name + "', " + str(ra) + ", " + str(dec) + ", " + str(self.config.boxSize) + ")" + ('" + self.skychunk.name + "', " + str(ra) + ", " + str(dec) + ", " + str(self.skychunk.boxSize) + ")" self.execute(sql) @@ -1055,4 +1075,48 @@ return id + + ''' + Inserts new box into box table. If it already exisits, it returns existing box id + ''' + def isBoxIngested(self, id): + + # set the field 'ingested' to 1 for this box + sql = "select ingested from box where id = " + str(id) + try: + rs = self.executeQuery(sql) + if not rs.next(): + print "no matching boxes for ", str(id) + raise + status = rs.getInt(1) + except: + print "failed to set box ", str(id), "as ingested" + raise + + return status + ''' + Inserts new box into box table. If it already exisits, it returns existing box id + ''' + def setIngestedBox(self, id): + + # set the field 'ingested' to 1 for this box + sql = "update box set ingested = 1 where id = " + str(id) + try: + self.execute(sql) + except: + print "failed to set box ", str(id), "as ingested" + raise + + ''' + Inserts new box into box table. If it already exisits, it returns existing box id + ''' + def clearIngestedBoxes(self): + + # set the field 'ingested' to 1 for this box + sql = "update box set ingested = 0" + try: + self.execute(sql) + except: + print "failed to clear boxes as ingested" + raise ''' @@ -1088,11 +1152,13 @@ FROM box \ JOIN pending ON (id = box_id) \ - WHERE config = '" + self.config.name + "' \ + WHERE skychunk = '" + self.skychunk.name + "' \ AND batch_type = '" + batchType + "' \ AND ra_center NOT IN \ (SELECT ra_center \ FROM stripe \ - WHERE config = '" + self.config.name + "') \ + WHERE skychunk = '" + self.skychunk.name + "') \ GROUP BY ra_center LIMIT 1" + + # XXX print "sql: ", sql try: @@ -1106,5 +1172,5 @@ FROM box \ JOIN pending ON (id = box_id) \ - WHERE config = '" + self.config.name + "' \ + WHERE skychunk = '" + self.skychunk.name + "' \ AND batch_type = '" + batchType + "' \ GROUP BY ra_center \ @@ -1126,5 +1192,5 @@ WHERE ra_center = " + str(raCenter) + " \ AND batch_type = '" + batchType + "' \ - AND config = '" + self.config.name + "' \ + AND skychunk = '" + self.skychunk.name + "' \ ORDER BY dec_center DESC" @@ -1159,6 +1225,6 @@ # now insert new stripe entry sql = "INSERT INTO stripe \ - (client_id, config, ra_center) \ - SELECT clients.id, clients.config, " + str(raCenter) + " \ + (client_id, skychunk, ra_center) \ + SELECT clients.id, clients.skychunk, " + str(raCenter) + " \ FROM clients \ WHERE host = '" + host + "' \ @@ -1188,5 +1254,5 @@ ''' - Returns ids for pending items for this config + Returns ids for pending items for this skychunk ''' def getPendingIds(self, batchType): @@ -1195,5 +1261,5 @@ FROM pending \ JOIN box ON (box_id = id) \ - WHERE config = '" + self.config.name + "' \ + WHERE skychunk = '" + self.skychunk.name + "' \ AND batch_type = '" + batchType + "'" @@ -1204,5 +1270,5 @@ rs.close() except: - self.logger.errorPair("Can't get pending ids for this config", sql) + self.logger.errorPair("Can't get pending ids for this skychunk", sql) return ids @@ -1235,10 +1301,10 @@ WHERE stage_id = " + str(stageID) + " \ AND batch_type = '" + batchType + "' \ - AND config = '" + self.config.name + "'" + AND skychunk = '" + self.skychunk.name + "'" self.execute(sql) ''' - Writes density-plot data to file for pending stuff over sky for given config + Writes density-plot data to file for pending stuff over sky for given skychunk ''' def createPendingDensityPlotData(self, batchType, DATFILE): @@ -1246,5 +1312,5 @@ sql = "SELECT id, ra_center, dec_center \ FROM box \ - WHERE config = '" + self.config.name + "' \ + WHERE skychunk = '" + self.skychunk.name + "' \ ORDER BY ra_center, dec_center" @@ -1299,5 +1365,5 @@ def getItemsInThisThisBox(self, ra, dec): - halfSide = self.config.boxSize/2.0 + halfSide = self.skychunk.boxSize/2.0 minRa = ra-halfSide maxRa = ra+halfSide @@ -1328,5 +1394,5 @@ FROM stripe \ JOIN clients ON (client_id = id) \ - WHERE clients.config = '" + self.config.name + "' \ + WHERE clients.skychunk = '" + self.skychunk.name + "' \ ORDER BY host, started" Index: trunk/ippToPsps/jython/loader.py =================================================================== --- trunk/ippToPsps/jython/loader.py (revision 34884) +++ trunk/ippToPsps/jython/loader.py (revision 35097) @@ -34,13 +34,18 @@ super(Loader, self).__init__(argv, 1, 1) - # create gpc1 database objects + if self.skychunk.parallel: + print "PARALLEL dvo" + else: + print "SERIAL dvo" + + # connect to the gpc1 database self.gpc1Db = Gpc1Db(self.logger, self.config) # connect to scratch database - scratchDbs = ['ipptopsps_scratch', 'ipptopsps_scratch2', 'ipptopsps_scratch3'] - for dbName in scratchDbs: - self.scratchDb = ScratchDb(self.logger, self.config, dbName) - if self.scratchDb.anyOtherConnections(): - self.logger.errorPair("This scratch Db is already in use", dbName) + scratchDbs = ['1', '2', '3'] + for dbVersion in scratchDbs: + self.scratchDb = ScratchDb(self.logger, self.config, dbVersion) + if not self.config.test and self.scratchDb.anyOtherConnections(): + self.logger.errorPair("This scratch Db is already in use", self.scratchDb.dbName) self.scratchDb.disconnect() continue @@ -52,5 +57,5 @@ # create Datastore objects - self.datastore = Datastore(self.logger, self.config, self.ippToPspsDb) + self.datastore = Datastore(self.logger, self.skychunk, self.ippToPspsDb) # if an IN batch is requested, create and quit @@ -60,4 +65,5 @@ batch = InitBatch(self.logger, self.config, + self.skychunk, self.gpc1Db, self.ippToPspsDb, @@ -72,32 +78,37 @@ + # if the 'once' option is passed, we do not loop multiple times in 'run' + self.onePassOnly = 0 + if len(sys.argv) > 2 and sys.argv[2] == "once": + self.onePassOnly = 1 + # set a poll time of about 1 minute self.parsePollTimeArg("0.0166") - self.config.printAll() + self.skychunk.printAll() ''' Overrides base-class version so we can ensure our scratch Db is using the right DVO Db ''' - def refreshConfig(self): - - ret = super(Loader, self).refreshConfig() + def refreshSkychunk(self): + + ret = super(Loader, self).refreshSkychunk() try: self.scratchDb except: return ret - self.dvoDetections = DvoDetections(self.logger, self.config, self.scratchDb.dbName) + self.dvoDetections = DvoDetections(self.logger, self.config, self.skychunk, self.ippToPspsDb, self.scratchDb.dbName) return ret ''' - Overrides base_class version so we can break out of processing loops if the config changes + Overrides base_class version so we can break out of processing loops if the skychunk changes ''' def checkClientStatus(self): - oldConfigName = self.config.name + oldSkychunkName = self.skychunk.name super(Loader, self).checkClientStatus() - if oldConfigName != self.config.name: - self.logger.infoPair("Config changed", "from '" + oldConfigName + "' to '" + self.config.name + "'") - self.config.printAll() + if oldSkychunkName != self.skychunk.name: + self.logger.infoPair("Skychunk changed", "from '" + oldSkychunkName + "' to '" + self.skychunk.name + "'") + self.skychunk.printAll() return False @@ -115,5 +126,5 @@ abort = False - for batchType in self.config.batchTypes: + for batchType in self.skychunk.batchTypes: # get a stripe's worth of box IDs @@ -141,4 +152,6 @@ ids = self.ippToPspsDb.getPendingIdsForThisBox(boxId, batchType) + # ids are the stage_ids for items to be processed (eg, stack_id for a stack CMF, cam_id for P2 smf) + if len(ids) < 1: self.logger.debugPair("No " + batchType + " items found in this box", "skipping") @@ -149,25 +162,54 @@ self.logger.infoPair("Box dimensions", "%.1f / %.1f / %.1f" % (boxDim['RA'], boxDim['DEC'], boxDim['SIDE'])) self.logger.infoPair(batchType + " items found in this box", "%d" % len(ids)) - boxSizeWithBorder = boxDim['SIDE'] + (self.config.BORDER * 2) + boxSizeWithBorder = boxDim['SIDE'] + (self.skychunk.BORDER * 2) + boxSizeSansBorder = boxDim['SIDE'] self.logger.infoPair("got here", "ok") useFullTables = 0 if batchType != "OB": - # look in DVO for this box (with extra border) - self.logger.infoPair("Querying DVO for this sky area", "") - self.dvoDetections.setSkyAreaAsBox(boxDim['RA'], boxDim['DEC'], boxSizeWithBorder) - #self.dvoDetections.setSkyArea() - sizeToBeIngested = self.dvoDetections.getDiskSizeOfRegionsToBeIngested() - if sizeToBeIngested == 0.0: smfsPerGB = 999999999 - else: smfsPerGB = len(ids)/sizeToBeIngested - self.logger.infoPair("DVO to be ingested", "%.1f GB" % sizeToBeIngested) - self.logger.infoPair("smfs-per-GB", "%.1f" % smfsPerGB) - # should do we pre-ingest stuff from DVO? - if batchType == 'P2' and smfsPerGB > 30: - if not self.dvoDetections.sync(): - self.logger.errorPair("Could not sync DVO with MySQL", "skipping") - continue - + + if (self.dvoDetections.useStilts): + # look in DVO for this box (with extra border) + self.logger.infoPair("Querying DVO for this sky area", "") + self.dvoDetections.setSkyAreaAsBox(boxDim['RA'], boxDim['DEC'], boxSizeWithBorder) + #self.dvoDetections.setSkyArea() + sizeToBeIngested = self.dvoDetections.getDiskSizeOfRegionsToBeIngested() + if sizeToBeIngested == 0.0: smfsPerGB = 999999999 + else: smfsPerGB = len(ids)/sizeToBeIngested + self.logger.infoPair("DVO to be ingested", "%7.1e GB" % sizeToBeIngested) + self.logger.infoPair("smfs-per-GB", "%.1f" % smfsPerGB) + + # should do we pre-ingest stuff from DVO? + # NOTE EAM : this skychunk loads the dvo detections into the mysql db + # XXXX EAM : this should happen for both P2 and Stack detection + # XXXX EAM : in parallel model, this should happen for all P2 & Stack batches + # + if (batchType == 'P2' or batchType == 'ST') and smfsPerGB > 30: + if not self.dvoDetections.sync(): + self.logger.errorPair("Could not sync DVO with MySQL", "skipping") + continue + + useFullTables = 1 + else: + # XXX EAM : this is not currently an + # option. either remove the non-native + # ingest code or make it an option + # if we are using the native loader, always use it + # need to work out a good boundary / region strategy in coordination with + # impact of mysql insertion + if not self.ippToPspsDb.isBoxIngested(boxId): + self.dvoDetections.nativeIngestDetections(boxId, boxDim['RA'], boxDim['DEC'], boxSizeSansBorder) + useFullTables = 1 + ## else: + ## # if we are using the native loader, always use it + ## # need to work out a good boundary / region strategy in coordination with + ## # impact of mysql insertion + ## self.dvoObjects.nativeIngestObjects(boxDim['RA'], boxDim['DEC'], boxSizeSansBorder) + ## useFullTables = 1 + + ''' + NOTE EAM : "TheseItems" refers to each of the stack CMFs, camera SMFs, or object batches + ''' self.logger.infoBool("Using pre-ingested DVO data?", useFullTables) if not self.processTheseItems(batchType, ids, useFullTables): @@ -175,7 +217,9 @@ break self.logger.infoPair("processed","ok") + if abort or not self.checkClientStatus(): abort = True elif numAttempts > 1 and not self.waitForPollTime(): break + if self.onePassOnly: self.exitProgram("one pass completed") ''' @@ -210,4 +254,5 @@ batch = DetectionBatch(self.logger, self.config, + self.skychunk, self.gpc1Db, self.ippToPspsDb, @@ -219,4 +264,5 @@ batch = StackBatch(self.logger, self.config, + self.skychunk, self.gpc1Db, self.ippToPspsDb, @@ -229,4 +275,5 @@ batch = ObjectBatch(self.logger, self.config, + self.skychunk, self.gpc1Db, self.ippToPspsDb, @@ -260,5 +307,5 @@ ''' def printUsage(self): - super(Cleanup, self).printUsage("<[init]>") + super(Loader, self).printUsage("<[init]>") Index: trunk/ippToPsps/jython/objectbatch.py =================================================================== --- trunk/ippToPsps/jython/objectbatch.py (revision 34884) +++ trunk/ippToPsps/jython/objectbatch.py (revision 35097) @@ -32,4 +32,5 @@ logger, config, + skychunk, gpc1Db, ippToPspsDb, @@ -42,4 +43,5 @@ logger, config, + skychunk, gpc1Db, ippToPspsDb, @@ -52,5 +54,5 @@ try: - self.dvoObjects = DvoObjects(self.logger, self.config, self.scratchDb.dbName) + self.dvoObjects = DvoObjects(self.logger, self.config, self.skychunk, self.ippToPspsDb, self.scratchDb.dbName) except: self.logger.errorPair("Unable to create instance of", "DvoObjects") @@ -71,5 +73,8 @@ self.region = self.scratchDb.getRegionNameFromThisDvoIndex(self.id) self.ippToPspsDb.insertObjectMeta(self.batchID, self.region) - self.dvoObjects.ingestRegion(self.region) + if True: + self.dvoObjects.nativeIngestRegion(self.region) + else: + self.dvoObjects.ingestRegion(self.region) return True @@ -261,5 +266,5 @@ ,PSF_QF_PERF \ ,STARGAL_SEP \ - , " + str(self.config.dataRelease) + "\ + , " + str(self.skychunk.dataRelease) + "\ , RAND() \ FROM " + cptTableName Index: trunk/ippToPsps/jython/plot.py =================================================================== --- trunk/ippToPsps/jython/plot.py (revision 34884) +++ trunk/ippToPsps/jython/plot.py (revision 35097) @@ -15,16 +15,16 @@ Constructor ''' - def __init__(self, logger, config, ippToPspsDb): + def __init__(self, logger, skychunk, ippToPspsDb): self.logger = logger - self.config = config + self.skychunk = skychunk self.ippToPspsDb = ippToPspsDb ''' - Creates a density plot of pending exposures for this config + Creates a density plot of pending exposures for this skychunk ''' def createDensityPlot(self, batchType, forCzartool=False): - tempFilename = "./" + self.config.name + "_" + batchType + "_plotData.dat" + tempFilename = "./" + self.skychunk.name + "_" + batchType + "_plotData.dat" DATFILE = open(tempFilename,'w') max = self.ippToPspsDb.createPendingDensityPlotData(batchType, DATFILE) @@ -36,7 +36,8 @@ if forCzartool: - OUTPUTFILE = self.config.czarPlotsPath + "/ippToPsps_density_" + self.config.name + "_" + batchType + ".png" + ## XXX from config?? + OUTPUTFILE = self.skychunk.czarPlotsPath + "/ippToPsps_density_" + self.skychunk.name + "_" + batchType + ".png" else: - OUTPUTFILE = self.config.name + "_" + batchType + "_" + timestamp + ".png" + OUTPUTFILE = self.skychunk.name + "_" + batchType + "_" + timestamp + ".png" f = os.popen('/home/panstarrs/ipp/local/bin/gnuplot', 'w') @@ -49,8 +50,8 @@ print >> f, "set term " + TERM + "; \ set output \"" + OUTPUTFILE + "\"; \ - set title \"Unprocessed " + batchType + " items for '" + self.config.name + "' as of " + timestamp + "\"; \ + set title \"Unprocessed " + batchType + " items for '" + self.skychunk.name + "' as of " + timestamp + "\"; \ set grid; \ - set xrange [" + str(self.config.minRa) + ":" + str(self.config.maxRa) + "] reverse; \ - set yrange [" + str(self.config.minDec) + ":" + str(self.config.maxDec) + "]; \ + set xrange [" + str(self.skychunk.minRa) + ":" + str(self.skychunk.maxRa) + "] reverse; \ + set yrange [" + str(self.skychunk.minDec) + ":" + str(self.skychunk.maxDec) + "]; \ unset key; \ set palette rgb 22,13,10; \ Index: trunk/ippToPsps/jython/plotter.py =================================================================== --- trunk/ippToPsps/jython/plotter.py (revision 34884) +++ trunk/ippToPsps/jython/plotter.py (revision 35097) @@ -18,5 +18,5 @@ super(Plotter, self).__init__(argv) - self.plot = Plot(self.logger, self.config, self.ippToPspsDb) + self.plot = Plot(self.logger, self.skychunk, self.ippToPspsDb) if len(argv) < 3: Index: trunk/ippToPsps/jython/pollOdm.py =================================================================== --- trunk/ippToPsps/jython/pollOdm.py (revision 34884) +++ trunk/ippToPsps/jython/pollOdm.py (revision 35097) @@ -9,5 +9,5 @@ from ipptopsps import IppToPsps -from config import Config +from skychunk import Skychunk from odm import Odm from batch import Batch @@ -56,5 +56,5 @@ self.logger.infoSeparator() - self.logger.infoPair("Config", self.config.name) + self.logger.infoPair("Skychunk", self.skychunk.name) for stage in self.stages: Index: trunk/ippToPsps/jython/pslogger.py =================================================================== --- trunk/ippToPsps/jython/pslogger.py (revision 34884) +++ trunk/ippToPsps/jython/pslogger.py (revision 35097) @@ -11,5 +11,5 @@ This will default to stout and no file output ''' - def setup(self, programName, basePath, configName, host, pid, stdout=1, sendToFile=0): + def setup(self, programName, basePath, skychunkName, host, pid, stdout=1, sendToFile=0): formatter = logging.Formatter('%(asctime)s | %(levelname)7s | %(message)s', '%Y-%m-%d %H:%M:%S') @@ -21,5 +21,5 @@ PATH = basePath + "/log" if not os.path.exists(PATH): os.makedirs(PATH) - FULLPATH = PATH + "/" + programName + "_" + configName + "_" + host + "_" + str(pid) + ".log" + FULLPATH = PATH + "/" + programName + "_" + skychunkName + "_" + host + "_" + str(pid) + ".log" # opens file to be appended hdlr = logging.FileHandler(FULLPATH, "a") Index: trunk/ippToPsps/jython/queue.py =================================================================== --- trunk/ippToPsps/jython/queue.py (revision 34884) +++ trunk/ippToPsps/jython/queue.py (revision 35097) @@ -29,19 +29,19 @@ # create various objects self.gpc1Db = Gpc1Db(self.logger, self.config) - self.datastore = Datastore(self.logger, self.config, self.ippToPspsDb) + self.datastore = Datastore(self.logger, self.skychunk, self.ippToPspsDb) + try: - self.dvoObjects = DvoObjects(self.logger, self.config) + self.dvoObjects = DvoObjects(self.logger, self.config, self.skychunk, self.ippToPspsDb) except: self.exitProgram("Unable to create instance of DvoObject") raise - self.config.printAll() + self.skychunk.printAll() if len(argv) > 2: self.parsePollTimeArg(sys.argv[2]) - ''' Main processing loop. - Tiled area is defined in config and looks for available stage_ids for each tile and writes them to ippToPsps database ready for processing + Tiled area is defined in 'skychunk' and looks for available stage_ids for each tile and writes them to ippToPsps database ready for processing by one or more loading clients that may be running on any host ''' @@ -54,5 +54,5 @@ # queue up batches that are processed but not loaded to datastore - for batchType in self.config.batchTypes: + for batchType in self.skychunk.batchTypes: self.logger.infoTitle("Previous failed datastore loads") @@ -64,10 +64,11 @@ # for object batches, get full list of stuff from Dvo + # XXX : need to fix 'dvo.setSkyArea' or this may not return the full list for a parallel dvo if batchType == "OB": self.dvoObjects.setSkyArea( - self.config.minRa, - self.config.maxRa, - self.config.minDec, - self.config.maxDec) + self.skychunk.minRa, + self.skychunk.maxRa, + self.skychunk.minDec, + self.skychunk.maxDec) rows = self.dvoObjects.allPopulatedRegionInfo self.dvoObjects.printSummary() @@ -76,10 +77,13 @@ else: rows = self.gpc1Db.getItemsInThisDVODbForThisStage( - self.config.dvoLabel, + self.skychunk.dvoLabel, batchType, - self.config.minRa, - self.config.maxRa, - self.config.minDec, - self.config.maxDec) + self.skychunk.minRa, + self.skychunk.maxRa, + self.skychunk.minDec, + self.skychunk.maxDec) + + # EAM TEST I/O + self.logger.infoPair("received rows from db:", len(rows)) # first report total stuff @@ -104,5 +108,5 @@ self.ippToPspsDb.storeAllItems(pending) - # loop through full range of RA/Dec queueing stuff in boxes of size self.config.boxSize + # loop through full range of RA/Dec queueing stuff in boxes of size self.skychunk.boxSize if len(ids) > 0: @@ -112,9 +116,9 @@ # starting positions - ra = self.config.minRa + self.config.halfBox - dec = self.config.minDec + self.config.halfBox + ra = self.skychunk.minRa + self.skychunk.halfBox + dec = self.skychunk.minDec + self.skychunk.halfBox - while ra <= self.config.maxRa: - while dec <= self.config.maxDec: + while ra <= self.skychunk.maxRa: + while dec <= self.skychunk.maxDec: box_id = self.ippToPspsDb.insertBox(ra, dec) @@ -125,12 +129,12 @@ ra, dec, - self.config.boxSize, + self.skychunk.boxSize, len(ids))) if len(ids) > 0: self.ippToPspsDb.insertPending(box_id, batchType, ids) - dec = dec + self.config.boxSize - dec = self.config.minDec + self.config.halfBox - ra = ra + self.config.boxSize + dec = dec + self.skychunk.boxSize + dec = self.skychunk.minDec + self.skychunk.halfBox + ra = ra + self.skychunk.boxSize self.logger.info("+-------------+-------------+-------------+-------------+") @@ -150,5 +154,5 @@ batchName = Batch.getNameFromID(batchID) - subDir = Batch.getSubDir(self.config.basePath, batchType, self.config.dvoLabel) + subDir = Batch.getSubDir(self.skychunk.basePath, batchType, self.skychunk.dvoLabel) tarballFile = Batch.getTarballFile(batchID) self.logger.infoPair("Batch name", batchName) Index: trunk/ippToPsps/jython/scratchdb.py =================================================================== --- trunk/ippToPsps/jython/scratchdb.py (revision 34884) +++ trunk/ippToPsps/jython/scratchdb.py (revision 35097) @@ -19,6 +19,17 @@ Constructor ''' - def __init__(self, logger, config, dbName=None): - super(ScratchDb, self).__init__(logger, config, "localdatabase", dbName) + def __init__(self, logger, config, dbVersion, dbName=None): + + # create gpc1 database objects + if (config.test): + dbType = "localdatabase_test" + else: + dbType = "localdatabase" + + if not dbName: + dbName = config.getDbName(dbType) + dbName += dbVersion + + super(ScratchDb, self).__init__(logger, config, dbType, dbName) self.dvoDoneTable = "dvoDone" @@ -340,10 +351,14 @@ rs.first() if rs.getInt(1) < 50: + # print "bad astromety" self.logger.debug("Bad astrometric solution for", ota) return False else: + # print "good astromety" return True except: + # print "query failed" self.logger.debug("Unable to check astrometric solution") + return False return True @@ -355,11 +370,15 @@ def createDvoTables(self): + dvoImagesTable = "dvoImages" + dvoDetectionTable = "dvoDetections" + # drop and create Images table - self.logger.debugPair("Creating DVO table", self.dvoImagesTable) - sql = "DROP TABLE " + self.dvoImagesTable + self.logger.debugPair("Creating DVO table", dvoImagesTable) + sql = "DROP TABLE " + dvoImagesTable + try: self.execute(sql) except: pass - sql = "CREATE TABLE " + self.dvoImagesTable + " ( \ + sql = "CREATE TABLE " + dvoImagesTable + " ( \ SOURCE_ID SMALLINT, \ IMAGE_ID INT, \ @@ -375,6 +394,6 @@ # now detection table - self.logger.debugPair("Creating DVO table", self.dvoDetectionTable) - sql = "DROP TABLE " + self.dvoDetectionTable + self.logger.debugPair("Creating DVO table", dvoDetectionTable) + sql = "DROP TABLE " + dvoDetectionTable try: self.execute(sql) except: pass @@ -396,11 +415,14 @@ raErr REAL, \ decErr REAL, \ - PRIMARY KEY (imageID, ippDetectID) \ - )" + PRIMARY KEY (imageID, ippDetectID), \ + KEY (objID, detectID) \ + )" try: self.execute(sql) except: self.logger.error("Unable to create DVO detection database table") - self.makeColumnUnique("dvoDetection", "objID") + + # XXX EAM : for the parallel mode, dvoDetection requires duplicate objID values! + # self.makeColumnUnique("dvoDetection", "objID") ''' @@ -479,14 +501,16 @@ self.createIndex(self.dvoDetectionTable, "fileID") - # now add a delete cascading foreign key constraint on fileID - sql = "ALTER TABLE dvoDetectionFull \ + # XXX : this should be based on choice of useStilts (which is set in dvo.py and must be moved) + if (False): + # now add a delete cascading foreign key constraint on fileID + sql = "ALTER TABLE dvoDetectionFull \ ADD CONSTRAINT fk_fileID \ FOREIGN KEY (fileID) \ REFERENCES dvoDone(id) \ ON DELETE CASCADE" - try: self.execute(sql) - except: - self.logger.errorPair("Unable to create foreign key on", self.dvoDoneTable) - return False + try: self.execute(sql) + except: + self.logger.errorPair("Unable to create foreign key on", self.dvoDoneTable) + return False return True Index: trunk/ippToPsps/jython/setupScratchDb.py =================================================================== --- trunk/ippToPsps/jython/setupScratchDb.py (revision 34884) +++ trunk/ippToPsps/jython/setupScratchDb.py (revision 35097) @@ -2,16 +2,51 @@ import sys +import os +import socket import stilts from java.lang import * from java.sql import * - -from ipptopsps import IppToPsps +from config import Config from scratchdb import ScratchDb +from ipptopspsdb import IppToPspsDb ''' SetupScratchDb class ''' -class SetupScratchDb(IppToPsps): +class SetupScratchDb(object): + + ## XXX EAM : 2013.01.31 - i have changed the 'config.py' concept + ## to only include the global static information (from + ## settings.xml), and stripped out the dynamic information to a + ## new entity called 'skychunk' + + ## XXX note that setupScratchDb.py fails if there are no valid + ## entries in the config table (or if it is passed a non-existent + ## config) This is because ipptopsps.__init__ calls + ## checkClientStatus, which calls refreshConfig, which calls + ## readConfig, which attempts to select the matching entry from + ## the config table. if the select returns 0 rows, the code + ## crashes. even if it were not to crash (eg, if it returned + ## False), checkClientStatus would enter a waiting loop until the + ## config existed. + + ## since a scratchDb is needed to do basically anything, it seems + ## like 'setupScratchDb' should not be dependent on the details of + ## the configs, etc. + + ## what are the options? I think this is a problem of poor + ## factoring / dependencies, esp for 'config.py'. there are two + ## (or more) aspects to 'config.py' : management of the dynamic + ## 'config' information (basically, the regions / dvodbs of + ## interest) and manipulation of the static / system configuration + ## information. + + ## these concepts should probably be split and the more general + ## part (settings & logger) placed in one module and the rest elsewhere. + + ## as for setupScratchDb.py, it looks like it only uses the static + ## settings info and the logger functions, so it should not be a + ## child of the ipptopsps class ''' @@ -19,16 +54,24 @@ ''' def __init__(self, argv): - super(SetupScratchDb, self).__init__(argv) + # set up config object + self.config = Config() - if len(argv) < 3: + if len(argv) < 2: self.printUsage() self.exitProgram("incorrect args") + # some class constants + self.HOST = socket.gethostname() + self.PID = os.getpid() + + # XXX Shouldn't config set up logger? + # (not yet, since we set HOST & PID with program...) + self.logger = self.config.getLogger(self.HOST, self.PID, 1, 0) + try: - self.scratchDb = ScratchDb(self.logger, self.config, argv[2]) + self.scratchDb = ScratchDb(self.logger, self.config, argv[1]) except: self.exitProgram("Could not connect to a scratch Db") raise - def run(self): @@ -42,14 +85,24 @@ # install IN tables self.logger.infoPair("Installing", "initialization tables") - tables = stilts.treads("../config/IN/tables.vot") + tablepath = self.config.configDir + "tables.IN.vot" + tables = stilts.treads(tablepath) + for table in tables: + # EAM TEST I/O + self.logger.infoPair("Creating IN table: ", table.name) self.logger.debug("Creating IN table: " + table.name) table.write(self.scratchDb.url + '#' + table.name) - ''' - Overrides base-class version - ''' + # create basic DVO tables + self.logger.infoPair("Installing", "initialization tables") + self.scratchDb.createDvoTables() + def printUsage(self): - super(SetupScratchDb, self).printUsage("") + # write message to log or stdout if no logger set up + msg = os.path.basename(sys.argv[0]) + " " + try: + self.logger.errorPair("Usage:", msg) + except: + print "*** Usage: " + msg ''' @@ -61,4 +114,2 @@ setupScratchDb.exitProgram("completed") except: pass - - Index: trunk/ippToPsps/jython/skychunk.py =================================================================== --- trunk/ippToPsps/jython/skychunk.py (revision 35097) +++ trunk/ippToPsps/jython/skychunk.py (revision 35097) @@ -0,0 +1,133 @@ +#!/usr/bin/env jython + +import sys + +''' +A class encapsulating a ippToPsps processing skychunk. This information +is stored in the 'skychunk' table of the ipptopsps database +''' +class Skychunk(object): + + ''' + Constructor + + Basically reads the entire skychunk and stores values to class variables + ''' + def __init__(self, logger): + + # name of skychunk to work on (edit / all have special meanings) + self.name = sys.argv[1] + + self.createNewSkychunk = False + self.rotateSkychunks = False + + # a new skychunk? + if self.name == "edit": self.createNewSkychunk = True + + # should we rotate skychunks for this program? + if self.name == "all": self.rotateSkychunks = True + + print "name: ", self.name + print "new: ", self.createNewSkychunk + print "all: ", self.rotateSkychunks + + self.logger = logger + + # this is the border (in degrees) that we place around any loading box of + # PS1 pointings to ensure we pull a large enough area out of DVO + + # XXX this is a poor place to put this information -- it is completely gpc-specific. + # probably should go elsewhere + self.BORDER = 1.65 + self.isLoaded = False + + ''' + Prints everything for this skychunk + ''' + def printAll(self): + + self.logger.infoTitle("Skychunk") + + try: + self.logger.infoSeparator() + self.logger.infoPair("Skychunk name", self.name) + self.logger.infoPair("Survey", self.survey) + self.logger.infoPair("Publishing to PSPS as survey", self.pspsSurvey) + self.logger.infoPair("Loading epoch", self.epoch) + self.logger.infoPair("Data release", "%d" % self.dataRelease) + for batchType in self.batchTypes: self.logger.infoPair("Queuing batch type", batchType) + except: + pass + + self.printDvoInfo() + self.printDatastoreInfo() + self.printBoxCoords() + self.printDeletionPolicy() + try: self.logger.infoSeparator() + except: pass + + ''' + Queuing this batch type? + ''' + def queuingThisBatchType(self, batchType): + if batchType in self.batchTypes: return 1 + return 0 + + ''' + Creates a logger object and returns it + ''' + def getLogger(self, host, pid, stdout=1, sendToFile=0): + + logging.setLoggerClass(PSLogger) + self.logger = logging.getLogger(self.programName) + self.logger.setup(self.programName, self.logPath, self.name, host, pid, stdout, sendToFile) + + return self.logger + + ''' + Prints the currently set DVO info + ''' + def printDvoInfo(self): + + try: + self.logger.infoPair("DVO label", self.dvoLabel) + self.logger.infoPair("DVO location", self.dvoLocation) + except: + pass + + ''' + Prints datastore info + ''' + def printDatastoreInfo(self): + + try: + self.logger.infoBool("Datastore publishing?", self.datastorePublishing) + if self.datastorePublishing: + self.logger.infoPair("Datastore type", self.datastoreProduct) + self.logger.infoPair("Datastore product", self.datastoreType) + except: + pass + + ''' + Prints the current deletion policy + ''' + def printDeletionPolicy(self): + + try: + self.logger.infoBool("Deleting local?", self.deleteLocal) + self.logger.infoBool("Deleting datastore?", self.deleteDatastore) + self.logger.infoBool("Deleting DXLayer?", self.deleteDxLayer) + except: + pass + + ''' + Prints the currently set RA/Dec bounding box + ''' + def printBoxCoords(self): + + try: + self.logger.infoPair("RA limits", "%.1f -> %.1f" % (self.minRa, self.maxRa)) + self.logger.infoPair("Dec limits", "%.1f -> %.1f" % (self.minDec, self.maxDec)) + self.logger.infoPair("Loading box size", "%.1f degrees" % self.boxSize) + except: + pass Index: trunk/ippToPsps/jython/stackbatch.py =================================================================== --- trunk/ippToPsps/jython/stackbatch.py (revision 34884) +++ trunk/ippToPsps/jython/stackbatch.py (revision 35097) @@ -2,4 +2,5 @@ import os.path +import glob import sys @@ -32,4 +33,5 @@ logger, config, + skychunk, gpc1Db, ippToPspsDb, @@ -42,4 +44,5 @@ logger, config, + skychunk, gpc1Db, ippToPspsDb, @@ -48,6 +51,9 @@ batchID, "ST", - gpc1Db.getStackStageCmf(config.dvoLabel, stackID), + gpc1Db.getStackStageCmf(skychunk.dvoLabel, stackID), useFullTables) + + # self.printline = 0 + # # self.testprint() self.stackType = "DEEP_STACK" # TODO @@ -63,4 +69,6 @@ self.filterID = self.scratchDb.getFilterID(self.filter) self.skycell = meta[1]; + + # self.testprint() # skycell is, eg "skycell.1133.081" @@ -76,4 +84,6 @@ self.skycell = self.skycell[8:12] self.projectioncell = self.skycell + + # self.testprint() # proposed new values. Need to coordinate with the SkyCell table @@ -96,4 +106,6 @@ self.scratchDb.dropTable("StackDetectionCalib") + # self.testprint() + # delete IPP tables self.scratchDb.dropTable("SkyChip_psf") @@ -115,4 +127,6 @@ if not self.useFullTables: self.scratchDb.insertNewDvoExternID(self.header['SOURCEID'], self.header['IMAGEID']) + + # self.testprint() # dump stuff to log @@ -139,4 +153,8 @@ self.scratchDb.execute(sql) + + def testprint(self): + print "here ", self.printline + self.printline += 1 ''' @@ -348,4 +366,14 @@ self.logger.infoPair("Procesing table", "StackMeta") + self.fwhm_maj = self.safeDictionaryAccess(self.header, 'FWHM_MAJ') + self.fwhm_maj_uq = self.safeDictionaryAccess(self.header, 'FW_MJ_UQ') + self.psfmodel = self.safeDictionaryAccess(self.header, 'PSFMODEL') + if (self.fwhm_maj == "NULL"): self.fwhm_maj = -999 + if (self.fwhm_maj_uq == "NULL"): self.fwhm_maj_uq = -999 + + # print "fwhm_maj = ", self.fwhm_maj + # print "fwhm_maj_uq = ", self.fwhm_maj_uq + # print "psfmodel = ", self.psfmodel + sql = "INSERT INTO StackMeta (\ stackMetaID \ @@ -375,7 +403,7 @@ ," + self.header['FPA.ZP'] + " \ ," + self.expTime + " \ - ,'" + self.safeDictionaryAccess(self.header, 'PSFMODEL') + "' \ - ,'" + self.safeDictionaryAccess(self.header, 'FWHM_MAJ') + "' \ - ,'" + self.safeDictionaryAccess(self.header, 'FW_MJ_UQ') + "' \ + ,'" + self.psfmodel + "' \ + ," + str(self.fwhm_maj) + " \ + ," + str(self.fwhm_maj_uq) + " \ ,'" + self.header['CTYPE1'] + "' \ ,'" + self.header['CTYPE2'] + "' \ @@ -395,5 +423,5 @@ self.scratchDb.updateAllRows("StackMeta", "surveyID", str(self.surveyID)) self.scratchDb.updateFilterID("StackMeta", self.filter) - self.scratchDb.updateAllRows("StackMeta", "dataRelease", str(self.config.dataRelease)) + self.scratchDb.updateAllRows("StackMeta", "dataRelease", str(self.skychunk.dataRelease)) self.updateStackTypeID("StackMeta") @@ -486,6 +514,10 @@ ," + self.historyModNum + " \ FROM SkyChip_psf" - self.scratchDb.execute(sql) - + + # print "sql: ", sql + # response = raw_input("ready to insert stack det ") + + self.scratchDb.execute(sql) + #it is possible to drop some detections from dvo (that are present in the cmf). when that happens we get a 0 for objid #we drop those... @@ -498,5 +530,5 @@ self.scratchDb.updateFilterID("StackDetection", self.filter) - self.scratchDb.updateAllRows("StackDetection", "dataRelease", str(self.config.dataRelease)) + self.scratchDb.updateAllRows("StackDetection", "dataRelease", str(self.skychunk.dataRelease)) self.scratchDb.updateAllRows("StackDetection", "primaryF", "0") @@ -509,8 +541,10 @@ self.updateDvoIDsAndFlags("StackDetection") + # response = raw_input("updated dvo ") sql = "ALTER IGNORE TABLE StackDetection ADD PRIMARY KEY (objID)" self.scratchDb.execute(sql) + # response = raw_input("add primary key? ") if self.stackType == "DEEP_STACK": @@ -524,4 +558,5 @@ self.scratchDb.execute(sql) + # response = raw_input("add psf flux ") #leave null instflux in @@ -529,9 +564,11 @@ self.scratchDb.reportAndDeleteRowsWithNULLS("StackDetection", "objID") + # response = raw_input("delete nulls ") sql="DELETE FROM StackDetection where objID = 0" self.scratchDb.execute(sql) - self.logger.infoPair("Delleting", "entries with StackDetection.objID = 0") + self.logger.infoPair("Deleting", "entries with StackDetection.objID = 0") + # response = raw_input("deleted objID is 0 ") @@ -577,5 +614,5 @@ self.scratchDb.updateAllRows("StackApFlx", "surveyID", str(self.surveyID)) self.scratchDb.updateFilterID("StackApFlx", self.filter) - self.scratchDb.updateAllRows("StackApFlx", "dataRelease", str(self.config.dataRelease)) + self.scratchDb.updateAllRows("StackApFlx", "dataRelease", str(self.skychunk.dataRelease)) self.scratchDb.updateAllRows("StackApFlx", "primaryF", "0") self.scratchDb.updateAllRows("StackApFlx", "activeFlag", "0") @@ -611,5 +648,5 @@ self.scratchDb.updateAllRows("StackModelFit", "surveyID", str(self.surveyID)) self.scratchDb.updateFilterID("StackModelFit", self.filter) - self.scratchDb.updateAllRows("StackModelFit", "dataRelease", str(self.config.dataRelease)) + self.scratchDb.updateAllRows("StackModelFit", "dataRelease", str(self.skychunk.dataRelease)) self.scratchDb.updateAllRows("StackModelFit", "primaryF", "0") self.scratchDb.updateAllRows("StackModelFit", "activeFlag", "0") @@ -676,10 +713,12 @@ self.scratchDb.updateAllRows("SkinnyObject", "surveyID", str(self.surveyID)) - self.scratchDb.updateAllRows("SkinnyObject", "dataRelease", str(self.config.dataRelease)) + self.scratchDb.updateAllRows("SkinnyObject", "dataRelease", str(self.skychunk.dataRelease)) ''' Populates the StackDetectionCalib table - ''' - def populateStackDetectionCalib(self): + XXX this can probably get a big speed increase by using 'SELECT () INTO OUTFILE '/tmp/name' and then + calling the load data infile '/tmp/name' into table; + ''' + def populateStackDetectionCalibInsertUpdate(self): self.logger.infoPair("Processing table", "StackDetectionCalib") tableName = "StackDetectionCalib" @@ -702,4 +741,8 @@ FROM StackDetection" self.scratchDb.execute(sql) + + imageID = self.scratchDb.getImageIDFromExternID(self.header['IMAGEID']) + self.logger.infoPair("obtained","imageID") + # insert calibration information from dvoDetections into the Table sql = "UPDATE " + tableName + " AS a, " + self.scratchDb.dvoDetectionTable + " AS b \ @@ -712,10 +755,64 @@ a.expTime = b.expTime, \ a.airMass = b.airMass \ - WHERE a.stackDetectID = b.detectID" - self.scratchDb.execute(sql) - self.scratchDb.updateAllRows(tableName, "dataRelease", str(self.config.dataRelease)) - - - + WHERE a.stackDetectID = b.detectID \ + AND b.imageID = " + str(imageID) + self.scratchDb.execute(sql) + self.scratchDb.updateAllRows(tableName, "dataRelease", str(self.skychunk.dataRelease)) + + def populateStackDetectionCalib(self): + + tableName = "StackDetectionCalib" + self.logger.infoPair("Processing table", tableName) + + imageID = self.scratchDb.getImageIDFromExternID(self.header['IMAGEID']) + self.logger.infoPair("obtained","imageID") + + # check for & create output directory first + datadumpDir = "/tmp/datadump" + try: + statinfo = os.stat(datadumpDir) + # check on the stat results? + except: + print "making the data dump directory ", datadumpDir + os.mkdir(datadumpDir) + os.chmod(datadumpDir, 0777) + statinfo = os.stat(datadumpDir) + + dumpFile = datadumpDir + "/genetest.xx.dat" + files = glob.glob(dumpFile) + if len(files) > 0: + os.unlink(dumpFile) + + # insert all detections into table + sql = "SELECT \ + a.objID, \ + a.stackDetectID, \ + a.ippObjID, \ + a.ippDetectID, \ + a.filterID, \ + a.surveyID, \ + b.ra, \ + b.dec_, \ + b.raErr, \ + b.decErr, \ + b.zp, \ + b.zpErr, \ + b.expTime, \ + b.airMass, \ + " + str(self.skychunk.dataRelease) + " \ + FROM \ + StackDetection as a \ + JOIN " + self.scratchDb.dvoDetectionTable + " as b \ + ON (a.stackDetectID = b.detectID) where b.imageID = " + str(imageID) + \ + " INTO OUTFILE '" + dumpFile + "'" + print "sql: ", sql + self.scratchDb.execute(sql) + + sql = "LOAD DATA INFILE '" + dumpFile + "' INTO TABLE " + tableName + print "sql: ", sql + self.scratchDb.execute(sql) + + ## XXX write this with the select/insert + # self.scratchDb.updateAllRows(tableName, "dataRelease", str(self.skychunk.dataRelease)) ''' @@ -737,5 +834,5 @@ self.scratchDb.updateFilterID("ObjectCalColor", self.filter) - self.scratchDb.updateAllRows("ObjectCalColor", "dataRelease", str(self.config.dataRelease)) + self.scratchDb.updateAllRows("ObjectCalColor", "dataRelease", str(self.skychunk.dataRelease)) @@ -802,4 +899,7 @@ WHERE a.ippDetectID = b.ippDetectID \ AND b.imageID = " + str(imageID) + + # print "update dvo sql: ", sql + # response = raw_input("update dvo...") self.scratchDb.execute(sql)