Index: trunk/ippToPsps/jython/dvo.py =================================================================== --- trunk/ippToPsps/jython/dvo.py (revision 33598) +++ trunk/ippToPsps/jython/dvo.py (revision 33599) @@ -147,4 +147,11 @@ ''' + Sets a bunch of region files from DVO for a stripe starting at this RA and with this width + ''' + def setSkyAreaAsStripe(self, ra=0., width=4.): + + self.setSkyArea(ra, ra+width, -91., 91.) + + ''' Sets a bunch of region files from DVO for a RA/Dec box centered with these coords and with this side length ''' @@ -155,5 +162,5 @@ ''' - Sets a bunch of region files from DVO for this RA/Dec box + Determines what DVO files need to be ingested to the defines region ''' def setSkyArea(self, minRa=-1., maxRa=361., minDec=-91., maxDec=91): @@ -168,5 +175,4 @@ # update the two meta tables, Images and SkyTable - #self.loadImages() dont need a new Images table at this point self.loadSkyTable() @@ -189,22 +195,40 @@ for region in allRegions: - cpmPath = self.config.dvoLocation + "/" + region + ".cpm" - cptPath = self.config.dvoLocation + "/" + region + ".cpt" - - # check for existence of cpm and cpt files - if not os.path.isfile(cpmPath): continue - if not os.path.isfile(cptPath): continue - - # ok, have a real, populated file so add it to the list + # get paths for this region for the file types we are interested in + paths = [] + for fileType in self.ingestFileTypes: + paths.append(self.config.dvoLocation + "/" + region + "." + fileType) + + # check for the existence of all interested file types + skipThisRegion = False + for path in paths: + if not os.path.isfile(path): + skipThisRegion = True + break + + if skipThisRegion: continue + + # ok, have real, populated files so add this region to the list self.allPopulatedRegions.append(region) - # if we have already imported up-to-date versions of both the cpm and cpt then we skip this region - if self.scratchDb.alreadyImportedThisDvoTable(cpmPath) and self.scratchDb.alreadyImportedThisDvoTable(cptPath): + # if we have already imported up-to-date versions of all interested files, then we skip this region + alreadyIngested = True + for path in paths: + if not self.scratchDb.alreadyImportedThisDvoTable(path): + alreadyIngested = False + break + + if alreadyIngested: self.regionsAlreadyIngested.append(region) continue - # check if we have an out-of-date version of either region file, if so, add it to the purge list - if self.scratchDb.haveOutOfDateVersionOfThisDvoTable(cpmPath) or self.scratchDb.haveOutOfDateVersionOfThisDvoTable(cptPath): - self.regionsIngestedButOutOfDate.append(region) + # check if we have out-of-date versions of any interested files, if so, add region to the purge list + outOfDate = False + for path in paths: + if self.scratchDb.haveOutOfDateVersionOfThisDvoTable(path): + outOfDate = True + break + + if outOfDate: self.regionsIngestedButOutOfDate.append(region) self.regionsToIngest.append(region) @@ -229,4 +253,5 @@ self.logger.infoTitle("Summary for currently set DVO region") + self.logger.infoPair("DVO ingest type", self.__class__.__name__) # go no further if we've already partly ingested a different DVO @@ -235,12 +260,20 @@ return + for fileType in self.ingestFileTypes: + self.logger.infoPair("Ingesting files of type", fileType) + self.logger.infoPair("RA range", "%.2f -> %.2f" % (self.minRa, self.maxRa)) self.logger.infoPair("Dec range", "%.2f -> %.2f" % (self.minDec, self.maxDec)) - self.logger.infoPair("Total regions", "%d (%s)" % (len(self.allPopulatedRegions), self.getFileSizeStr(self.sizeOfAllPopulatedRegions))) - self.logger.infoPair("Up-to-date regions already ingested", "%d (%s)" % (len(self.regionsAlreadyIngested), self.getFileSizeStr(self.sizeOfRegionsAlreadyIngested))) - self.logger.infoPair("Out-of-date regions ingested", "%d (%s)" % (len(self.regionsIngestedButOutOfDate), self.getFileSizeStr(self.sizeOfRegionsIngestedButOutOfDate))) - self.logger.infoPair("Total regions to purge", "%d (%s)" % (len(self.regionsToPurge), self.getFileSizeStr(self.sizeOfRegionsToPurge))) - self.logger.infoPair("Total regions to ingest", "%d (%s)" % (len(self.regionsToIngest), self.getFileSizeStr(self.sizeOfRegionsToIngest))) - + + self.logger.info("+----------------+----------------+----------------+----------------+----------------+") + self.logger.info("| populated | ingested | out-of-date | to-purge | to ingest |") + self.logger.info("+----------------+----------------+----------------+----------------+----------------+") + self.logger.info("| %5d (%6s) | %5d (%6s) | %5d (%6s) | %5d (%6s) | %5d (%6s) |" % ( + len(self.allPopulatedRegions), self.getFileSizeStr(self.sizeOfAllPopulatedRegions), + len(self.regionsAlreadyIngested), self.getFileSizeStr(self.sizeOfRegionsAlreadyIngested), + len(self.regionsIngestedButOutOfDate), self.getFileSizeStr(self.sizeOfRegionsIngestedButOutOfDate), + len(self.regionsToPurge), self.getFileSizeStr(self.sizeOfRegionsToPurge), + len(self.regionsToIngest), self.getFileSizeStr(self.sizeOfRegionsToIngest))) + self.logger.info("+----------------+----------------+----------------+----------------+----------------+") ''' @@ -260,11 +293,14 @@ # go no further if we've already partly ingested a different DVO - if not self.correctDvo: return - - if len(self.regionsToPurge) < 1: return + if not self.correctDvo: return False + + if len(self.regionsToPurge) < 1: return False self.logger.infoPair("Purging", "%d regions" % len(self.regionsToPurge)) - deleteCount = self.scratchDb.purgeTheseDvoRegions(self.regionsToPurge) - self.logger.infoPair("Deleted", "%d detections" % deleteCount) + + detectionDeleteCount = self.scratchDb.purgeTheseDvoRegions(self.regionsToPurge, self.ingestFileTypes) + if detectionDeleteCount > 0: self.logger.infoPair("Deleted", "%d detections" % detectionDeleteCount) + + return True ''' @@ -289,5 +325,5 @@ ''' - Returns the disk size of these DVO regions, i.e. the size on disk of all cpm and cpt files + Returns the disk size of these DVO regions, i.e. the size on disk of all relevant DVO FITS files ''' def getDiskSizeOfDvoRegions(self, regions): @@ -296,9 +332,16 @@ for region in regions: - # get combined size of cpm and cpt files - size = size + self.getDiskSize(self.config.dvoLocation + "/" + region + ".cpm") - size = size + self.getDiskSize(self.config.dvoLocation + "/" + region + ".cpt") + # get combined size of all interested files + for fileType in self.ingestFileTypes: + size = size + self.getDiskSize(self.config.dvoLocation + "/" + region + "." + fileType) return size + + ''' + Abstract method for ingesting a particular reegion. Must be implemented in all subclasses + ''' + def ingestRegion(self, region): + self.logger.errorPair("ingestRegion()", "not implemented for this class") + ''' @@ -327,108 +370,14 @@ # now loop through all regions and ingest to MySQL - regionCount = 0 + ingestedRegionsCount = 0 + regionCount = 1 for region in self.regionsToIngest: - cpmPath = self.config.dvoLocation + "/" + region + ".cpm" - cptPath = self.config.dvoLocation + "/" + region + ".cpt" - - cpmTableName = self.scratchDb.getDbFriendlyTableName(region + ".cpm") - cptTableName = self.scratchDb.getDbFriendlyTableName(region + ".cpt") - - self.logger.infoTitle("Region %d of %d: %s" % (regionCount+1, totalRegions, region)) - - # if we have not already ingested the cpm table, then do it now - if not self.scratchDb.alreadyImportedThisDvoTable(cpmPath): - - # import cpm table and index - self.importFits( - cpmPath, - "IMAGE_ID DET_ID OBJ_ID CAT_ID EXT_ID DB_FLAGS", - cpmTableName) - - self.scratchDb.createIndex(cpmTableName, "IMAGE_ID") - self.scratchDb.createIndex(cpmTableName, "CAT_ID") - self.scratchDb.createIndex(cpmTableName, "OBJ_ID") - - # shove SOURCE_IDs into measurement table - #self.logger.infoPair("Adding", "SOURCE_IDs") - #sql = "ALTER TABLE " + cpmTableName + " ADD COLUMN (SOURCE_ID SMALLINT)" - #self.scratchDb.execute(sql) - #sql = "UPDATE " + cpmTableName + " AS a, " + self.scratchDb.dvoImagesTable + " AS b \ - # SET a.SOURCE_ID = b.SOURCE_ID \ - # WHERE a.IMAGE_ID = b.IMAGE_ID" - #self.scratchDb.execute(sql) - - # we can now report that we have imported this table - self.scratchDb.setImportedThisDvoTable(cpmPath, region) - - # get the id for this cpm file so we can insert it in the dvoDetection table - fileId = self.scratchDb.getDvoFileId(cpmPath) - - # import cpt table and index - self.importFits( - cptPath, - "OBJ_ID CAT_ID EXT_ID", - cptTableName) - - self.scratchDb.createIndex(cptTableName, "IMAGE_ID") - self.scratchDb.createIndex(cptTableName, "CAT_ID") - self.scratchDb.createIndex(cptTableName, "OBJ_ID") - - self.logger.infoPair("Adding","PSPS objIDs") - # first try to add a column for PSPS_OBJ_ID. catch failure and continue - try: - sql = "ALTER TABLE " + cpmTableName + " ADD COLUMN (PSPS_OBJ_ID BIGINT)" - self.scratchDb.execute(sql) - except: - self.logger.infoPair("Already added PSPS_OBJ_ID column to", cpmTableName) - - # shove PSPS objIDs from cpt table into measurement table - sql = "UPDATE " + cpmTableName + " AS a, " + cptTableName + " AS b \ - SET a.PSPS_OBJ_ID = b.EXT_ID \ - WHERE a.CAT_ID = b.CAT_ID \ - AND a.OBJ_ID = b.OBJ_ID" - self.scratchDb.execute(sql) - - - # NB we use an INSERT IGNORE here. This is because of a known issue where multiple DVO cpm - # files can include the same detection, with the same IMAGE_ID/DET_ID pairing, but different - # PSPS object IDs assigned in the corresponding cpt file. This is believed to be a chip-boundary - # issue within DVO. So, for now, we take the first IMAGE_ID/DET_ID detection we find, and ignore the rest - self.logger.infoPair("Merging into", self.scratchDb.dvoDetectionTable) - sql = "INSERT IGNORE INTO " + self.scratchDb.dvoDetectionTable + " (\ - imageID \ - ,ippDetectID \ - ,detectID \ - ,ippObjID \ - ,objID \ - ,flags \ - ,fileID \ - ) SELECT \ - IMAGE_ID \ - ,DET_ID \ - ,EXT_ID \ - ,CAT_ID * 1000000000 + OBJ_ID \ - ,PSPS_OBJ_ID \ - ,DB_FLAGS \ - , " + str(fileId) + " \ - FROM " + cpmTableName - try: - self.scratchDb.execute(sql) - except: - self.logger.error("FAILED: " + sql) - continue - - # now drop what we don't need - self.logger.infoPair("Dropping tables", "%s and %s" % (cpmTableName, cptTableName)) - self.scratchDb.dropTable(cpmTableName) - self.scratchDb.dropTable(cptTableName) - - self.scratchDb.setImportedThisDvoTable(cptPath, region) - regionCount = regionCount + 1 + self.logger.infoTitle("Region %d of %d: %s" % (regionCount, totalRegions, region)) + if self.ingestRegion(region): ingestedRegionsCount += 1 + regionCount += 1 self.logger.infoSeparator() - - self.logger.infoPair("Ingested", "%d DVO regions" % regionCount) + self.logger.infoPair("Ingested", "%d DVO regions (out of %d)" % (ingestedRegionsCount, len(self.regionsToIngest))) # update lists after attempted sync Index: trunk/ippToPsps/jython/dvodetections.py =================================================================== --- trunk/ippToPsps/jython/dvodetections.py (revision 33599) +++ trunk/ippToPsps/jython/dvodetections.py (revision 33599) @@ -0,0 +1,118 @@ +#!/usr/bin/env jython + +import stilts +import datetime +import re +import sys +import os +import logging +import glob + +from dvo import Dvo + + +''' +A class for ingesting DVO detections into MySQL +''' +class DvoDetections(Dvo): + + ''' + Constructor + ''' + def __init__(self, logger, config, scratchDbName): + + super(DvoDetections, self).__init__(logger, config, scratchDbName) + self.ingestFileTypes = ['cpm','cpt'] + + ''' + Ingests detections for this region + ''' + def ingestRegion(self, region): + + cpmPath = self.config.dvoLocation + "/" + region + ".cpm" + cptPath = self.config.dvoLocation + "/" + region + ".cpt" + + cpmTableName = self.scratchDb.getDbFriendlyTableName(region + ".cpm") + cptTableName = self.scratchDb.getDbFriendlyTableName(region + ".cpt") + + # if we have not already ingested the cpm table, then do it now + if not self.scratchDb.alreadyImportedThisDvoTable(cpmPath): + + # import cpm table and index + self.importFits( + cpmPath, + "IMAGE_ID DET_ID OBJ_ID CAT_ID EXT_ID DB_FLAGS", + cpmTableName) + + self.scratchDb.createIndex(cpmTableName, "IMAGE_ID") + self.scratchDb.createIndex(cpmTableName, "CAT_ID") + self.scratchDb.createIndex(cpmTableName, "OBJ_ID") + + # we can now report that we have imported this table + self.scratchDb.setImportedThisDvoTable(cpmPath, region) + + # get the id for this cpm file so we can insert it in the dvoDetection table + fileId = self.scratchDb.getDvoFileId(cpmPath) + + # import cpt table and index + self.importFits( + cptPath, + "OBJ_ID CAT_ID EXT_ID", + cptTableName) + + self.scratchDb.createIndex(cptTableName, "IMAGE_ID") + self.scratchDb.createIndex(cptTableName, "CAT_ID") + self.scratchDb.createIndex(cptTableName, "OBJ_ID") + + self.logger.infoPair("Adding","PSPS objIDs") + # first try to add a column for PSPS_OBJ_ID. catch failure and continue + try: + sql = "ALTER TABLE " + cpmTableName + " ADD COLUMN (PSPS_OBJ_ID BIGINT)" + self.scratchDb.execute(sql) + except: + self.logger.infoPair("Already added PSPS_OBJ_ID column to", cpmTableName) + + # shove PSPS objIDs from cpt table into measurement table + sql = "UPDATE " + cpmTableName + " AS a, " + cptTableName + " AS b \ + SET a.PSPS_OBJ_ID = b.EXT_ID \ + WHERE a.CAT_ID = b.CAT_ID \ + AND a.OBJ_ID = b.OBJ_ID" + self.scratchDb.execute(sql) + + # NB we use an INSERT IGNORE here. This is because of a known issue where multiple DVO cpm + # files can include the same detection, with the same IMAGE_ID/DET_ID pairing, but different + # PSPS object IDs assigned in the corresponding cpt file. This is believed to be a chip-boundary + # issue within DVO. So, for now, we take the first IMAGE_ID/DET_ID detection we find, and ignore the rest + self.logger.infoPair("Merging into", self.scratchDb.dvoDetectionTable) + sql = "INSERT IGNORE INTO " + self.scratchDb.dvoDetectionTable + " (\ + imageID \ + ,ippDetectID \ + ,detectID \ + ,ippObjID \ + ,objID \ + ,flags \ + ,fileID \ + ) SELECT \ + IMAGE_ID \ + ,DET_ID \ + ,EXT_ID \ + ,CAT_ID * 1000000000 + OBJ_ID \ + ,PSPS_OBJ_ID \ + ,DB_FLAGS \ + , " + str(fileId) + " \ + FROM " + cpmTableName + try: + self.scratchDb.execute(sql) + except: + self.logger.error("FAILED: " + sql) + return False + finally: + self.logger.infoPair("Dropping tables", "%s and %s" % (cpmTableName, cptTableName)) + self.scratchDb.dropTable(cpmTableName) + self.scratchDb.dropTable(cptTableName) + + self.scratchDb.setImportedThisDvoTable(cptPath, region) + + return True + + Index: trunk/ippToPsps/jython/dvoobjects.py =================================================================== --- trunk/ippToPsps/jython/dvoobjects.py (revision 33599) +++ trunk/ippToPsps/jython/dvoobjects.py (revision 33599) @@ -0,0 +1,51 @@ +#!/usr/bin/env jython + +import stilts +import datetime +import re +import sys +import os +import logging +import glob + +from dvo import Dvo + + +''' +A class for ingesting DVO detections into MySQL +''' +class DvoObjects(Dvo): + + ''' + Constructor + ''' + def __init__(self, logger, config, scratchDbName): + + super(DvoObjects, self).__init__(logger, config, scratchDbName) + self.ingestFileTypes = ['cps','cpt'] + + ''' + Ingests detections for this region + ''' + def ingestRegion(self, region): + + cpsPath = self.config.dvoLocation + "/" + region + ".cps" + cptPath = self.config.dvoLocation + "/" + region + ".cpt" + + cpsTableName = self.scratchDb.getDbFriendlyTableName(region + ".cps") + cptTableName = self.scratchDb.getDbFriendlyTableName(region + ".cpt") + + self.importFits(cpsPath, "*", cpsTableName) + self.importFits(cptPath, "*", cptTableName) + + # we can now report that we have imported this table + self.scratchDb.setImportedThisDvoTable(cpsPath, region) + self.scratchDb.setImportedThisDvoTable(cptPath, region) + + #self.logger.infoPair("Dropping tables", "%s and %s" % (cpsTableName, cptTableName)) + #self.scratchDb.dropTable(cpsTableName) + #self.scratchDb.dropTable(cptTableName) + + return True + +