Index: /trunk/ippToPsps/jython/dvo.py =================================================================== --- /trunk/ippToPsps/jython/dvo.py (revision 33233) +++ /trunk/ippToPsps/jython/dvo.py (revision 33234) @@ -49,4 +49,18 @@ self.correctDvo = self.scratchDb.isCorrectDvo(self.dvoLocation) + # set up empty lists + self.regionsToIngest = [] + self.regionsToPurge = [] + self.allPopulatedRegions = [] + self.regionsAlreadyIngested = [] + self.regionsIngestedButOutOfDate = [] + + if not self.correctDvo: + response = raw_input("* Wrong DVO is use. Do you want to reset and use '" + self.dvoLocation + "' instead (y/n)? ") + if response == "y": + response = raw_input("* Are you ABSOLUTELY sure you want to do this? (y/n)? ") + if response == "y": + self.resetAllTables() + ''' Destructor @@ -67,5 +81,5 @@ self.logger.infoPair("Resetting", "just detections") - self.scratchDb.resetOnlyDetectionDvoTables() + self.scratchDb.resetDvoDetectionTables() ''' @@ -75,9 +89,9 @@ self.logger.infoPair("Resetting", "all tables") - self.scratchDb.resetDvoToMysqlTables() + self.scratchDb.resetAllDvoTables() self.correctDvo = True ''' - Loads the DVO Iamges.dat FITS file + Loads the DVO Images.dat FITS file ''' def loadImages(self): @@ -96,21 +110,22 @@ # first, delete old dvoMeta table - self.logger.infoPair("Deleting from table", self.scratchDb.dvoMetaTable) - sql = "DELETE FROM " + self.scratchDb.dvoMetaTable - self.scratchDb.execute(sql) - self.scratchDb.optimizeTable(self.scratchDb.dvoMetaTable) + #self.logger.infoPair("Deleting from table", self.scratchDb.dvoMetaTable) + #sql = "DELETE FROM " + self.scratchDb.dvoMetaTable + #self.scratchDb.execute(sql) + #self.scratchDb.optimizeTable(self.scratchDb.dvoMetaTable) self.importFits( path, - "SOURCE_ID IMAGE_ID CCDNUM EXTERN_ID FLAGS PHOTCODE NSTAR", - self.scratchDb.dvoImageTable) - self.scratchDb.createIndex(self.scratchDb.dvoImageTable, "EXTERN_ID") + "SOURCE_ID IMAGE_ID EXTERN_ID FLAGS PHOTCODE", + self.scratchDb.dvoImagesTable) + self.logger.infoPair("Adding primary key to", self.scratchDb.dvoImagesTable) + self.scratchDb.execute("ALTER TABLE " + self.scratchDb.dvoImagesTable + " ADD PRIMARY KEY (IMAGE_ID, EXTERN_ID)") # insert into dvoMetaFull # NB what we and smf files call IMAGE_ID, DVO calls EXTERN_ID. We are sticking # with the smf name + ''' self.logger.infoPair("Populating", "all image meta data") sql = "INSERT INTO " + self.scratchDb.dvoMetaTable + " ( \ - sourceID, \ imageID, \ externID, \ @@ -118,10 +133,9 @@ photcode \ ) SELECT \ - SOURCE_ID, \ IMAGE_ID, \ EXTERN_ID, \ FLAGS, \ PHOTCODE \ - FROM " + self.scratchDb.dvoImageTable + FROM " + self.scratchDb.dvoImagesTable try: @@ -130,4 +144,5 @@ self.logger.errorPair("Could not insert into", self.scratchDb.dvoMetaTable) return + ''' self.scratchDb.setImportedThisDvoTable(path) @@ -160,26 +175,17 @@ self.logger.infoPair("Finished importing SkyTable at", self.dvoLocation) - - ''' - Gets a bunch of region files from DVO for a RA/Dec box centered with these coords and with this side length PLUS a border - ''' - def getRegionsInThisBoxWithBorder(self, regionsToIngest, regionsToPurge, ra=180., dec=0., side=181.0, border=0.): - - halfSide = (side + (border*2.0)) /2.0 - self.getRegions(regionsToIngest, regionsToPurge, ra-halfSide, ra+halfSide, dec-halfSide, dec+halfSide) + ''' + Sets a bunch of region files from DVO for a RA/Dec box centered with these coords and with this side length + ''' + def setSkyAreaAsBox(self, ra=180., dec=0., side=181.0): + + halfSide = side/2.0 + self.setSkyArea(ra-halfSide, ra+halfSide, dec-halfSide, dec+halfSide) - ''' - Gets a bunch of region files from DVO for a RA/Dec box centered with these coords and with this side length - ''' - def getRegionsInThisBox(self, regionsToIngest, regionsToPurge, ra=180., dec=0., side=181.0): - - halfSide = side/2.0 - self.getRegions(regionsToIngest, regionsToPurge, ra-halfSide, ra+halfSide, dec-halfSide, dec+halfSide) - - - ''' - Gets a bunch of region files from DVO for this RA/Dec box - ''' - def getRegions(self, regionsToIngest, regionsToPurge, minRa=-1., maxRa=361., minDec=-91., maxDec=91): + + ''' + Sets a bunch of region files from DVO for this RA/Dec box + ''' + def setSkyArea(self, minRa=-1., maxRa=361., minDec=-91., maxDec=91): # go no further if we've already partly ingested a different DVO @@ -191,11 +197,24 @@ return False + # update the two meta tables, Images and SkyTable + self.loadImages() + self.loadSkyTable() + + # set bounding-box coords + self.minRa = minRa + self.maxRa = maxRa + self.minDec = minDec + self.maxDec = maxDec + + # reset all lists allRegions = self.scratchDb.getDvoRegionsForThisBox(minRa, maxRa, minDec, maxDec) allIngestedRegions = self.scratchDb.getIngestedDvoRegions() - - allPopulatedRegions = [] - regionsAlreadyIngested = [] - regionsIngestedButOutOfDate = [] - + self.regionsToIngest = [] + self.regionsToPurge = [] + self.allPopulatedRegions = [] + self.regionsAlreadyIngested = [] + self.regionsIngestedButOutOfDate = [] + + # loop through all regions in DVO inside this bounding-box for region in allRegions: @@ -207,52 +226,74 @@ if not os.path.isfile(cptPath): continue - allPopulatedRegions.append(region) + # ok, have a real, populated file so add it 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): - regionsAlreadyIngested.append(region) + 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): - regionsIngestedButOutOfDate.append(region) - - regionsToIngest.append(region) + self.regionsIngestedButOutOfDate.append(region) + + self.regionsToIngest.append(region) # add out-of-range regions for purge list - items = list(set(allIngestedRegions) - set(allPopulatedRegions)) - for item in items: regionsToPurge.append(item) + items = list(set(allIngestedRegions) - set(self.allPopulatedRegions)) + for item in items: self.regionsToPurge.append(item) # now add out-of-date items to purge list - for item in regionsIngestedButOutOfDate: regionsToPurge.append(item) - - sizeOfAllPopulatedRegions = self.getFileSizeStr(self.getDiskSizeOfDvoRegions(allPopulatedRegions)) - sizeOfRegionsAlreadyIngested = self.getFileSizeStr(self.getDiskSizeOfDvoRegions(regionsAlreadyIngested)) - sizeOfRegionsIngestedButOutOfDate = self.getFileSizeStr(self.getDiskSizeOfDvoRegions(regionsIngestedButOutOfDate)) - sizeOfRegionsToPurge = self.getFileSizeStr(self.getDiskSizeOfDvoRegions(regionsToPurge)) - sizeOfRegionsToIngest = self.getFileSizeStr(self.getDiskSizeOfDvoRegions(regionsToIngest)) - - self.logger.infoTitle("Summary for this region") - self.logger.infoPair("RA range", "%.2f -> %.2f" % (minRa, maxRa)) - self.logger.infoPair("Dec range", "%.2f -> %.2f" % (minDec, maxDec)) - self.logger.infoPair("Total regions", "%d (%s)" % (len(allPopulatedRegions), sizeOfAllPopulatedRegions)) - self.logger.infoPair("Up-to-date regions already ingested", "%d (%s)" % (len(regionsAlreadyIngested), sizeOfRegionsAlreadyIngested)) - self.logger.infoPair("Out-of-date regions ingested", "%d (%s)" % (len(regionsIngestedButOutOfDate), sizeOfRegionsIngestedButOutOfDate)) - self.logger.infoPair("Total regions to purge", "%d (%s)" % (len(regionsToPurge), sizeOfRegionsToPurge)) - self.logger.infoPair("Total regions to ingest", "%d (%s)" % (len(regionsToIngest), sizeOfRegionsToIngest)) - + for item in self.regionsIngestedButOutOfDate: self.regionsToPurge.append(item) + + self.sizeOfAllPopulatedRegions = self.getDiskSizeOfDvoRegions(self.allPopulatedRegions) + self.sizeOfRegionsAlreadyIngested = self.getDiskSizeOfDvoRegions(self.regionsAlreadyIngested) + self.sizeOfRegionsIngestedButOutOfDate = self.getDiskSizeOfDvoRegions(self.regionsIngestedButOutOfDate) + self.sizeOfRegionsToPurge = self.getDiskSizeOfDvoRegions(self.regionsToPurge) + self.sizeOfRegionsToIngest = self.getDiskSizeOfDvoRegions(self.regionsToIngest) + + ''' + Prints out a summary of the currently set regions + ''' + def printSummary(self): + + self.logger.infoTitle("Summary for currently set DVO region") + + # go no further if we've already partly ingested a different DVO + if not self.correctDvo: + self.logger.infoPair("Wrong DVO in use", self.dvoLocation) + return + + 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))) + + + ''' + Returns the size on disk of the regions that need to be ingested in GB + ''' + def getDiskSizeOfRegionsToBeIngested(self): + + try: + return self.sizeOfRegionsToIngest/1073741824.0 + except: + return 0.0 ''' Purges the provided regions from the database ''' - def purgeRegions(self, regions): + def purgeRegions(self): # go no further if we've already partly ingested a different DVO if not self.correctDvo: return - if len(regions) < 1: return - - self.logger.infoPair("Purging", "%d regions" % len(regions)) - deleteCount = self.scratchDb.purgeTheseDvoRegions(regions) + if len(self.regionsToPurge) < 1: return + + self.logger.infoPair("Purging", "%d regions" % len(self.regionsToPurge)) + deleteCount = self.scratchDb.purgeTheseDvoRegions(self.regionsToPurge) self.logger.infoPair("Deleted", "%d detections" % deleteCount) @@ -271,4 +312,11 @@ ''' + Returns the disk size of this file + ''' + def getDiskSize(self, path): + + return os.stat(path).st_size + + ''' Returns the disk size of these DVO regions, i.e. the size on disk of all cpm and cpt files ''' @@ -279,52 +327,33 @@ # get combined size of cpm and cpt files - size = size + os.stat(self.dvoLocation + "/" + region + ".cpm").st_size - size = size + os.stat(self.dvoLocation + "/" + region + ".cpt").st_size + size = size + self.getDiskSize(self.dvoLocation + "/" + region + ".cpm") + size = size + self.getDiskSize(self.dvoLocation + "/" + region + ".cpt") return size ''' - Populates dvoDetections and dvoMeta with FITS tables from DVO for - a RA/Dec box centered with these coords and with this side length PLUS a border - ''' - def loadRegionInThisBoxWithBorder(self, ra=180., dec=0., side=181.0, border=0.): - - halfSide = (side + (border*2.0)) /2.0 - self.loadRegion(ra-halfSide, ra+halfSide, dec-halfSide, dec+halfSide) - - ''' - Populates dvoDetections and dvoMeta with FITS tables from DVO for - a RA/Dec box centered with these coords and with this side length - ''' - def loadRegionInThisBox(self, regionsToIngest, regionsToPurge, ra=180., dec=0., side=181.0): - - halfSide = side/2.0 - self.loadRegion(ra-halfSide, ra+halfSide, dec-halfSide, dec+halfSide) - - ''' - Populates dvoDetections and dvoMeta with FITS tables from DVO for the defined region - Defaults to whole sky if limits are missing - ''' - def loadRegion(self, minRa=-1., maxRa=361., minDec=-91., maxDec=91): + Synchronises MySQL database with the sky area currently set from DVO + Populates dvoDetections and dvoDone with FITS tables from DVO for the defined sky area, this + includes purging detections outside sky area + ''' + def sync(self): # go no further if we've already partly ingested a different DVO - if not self.correctDvo: return - - # update the two meta tables, Images and SkyTable - self.loadImages() - self.loadSkyTable() - - regionsToIngest = [] - regionsToPurge = [] - self.getRegions(regionsToIngest, regionsToPurge, minRa, maxRa, minDec, maxDec) - - self.purgeRegions(regionsToPurge) - - totalRegions = len(regionsToIngest) + if not self.correctDvo: return False + + # check we have something to do + if len(self.regionsToPurge) < 1 and len(self.regionsToIngest) < 1: + self.logger.infoPair("Nothing to ingest or purge", "finished ingest") + return True + + # purge stuff + self.purgeRegions() + + totalRegions = len(self.regionsToIngest) if totalRegions > 0: self.logger.infoSeparator() # now loop through all regions and ingest to MySQL regionCount = 0 - for region in regionsToIngest: + for region in self.regionsToIngest: cpmPath = self.dvoLocation + "/" + region + ".cpm" @@ -334,5 +363,5 @@ cptTableName = self.scratchDb.getDbFriendlyTableName(region + ".cpt") - self.logger.infoTitle("Region %d of %d: %s" % (regionCount, totalRegions, region)) + 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 @@ -350,11 +379,11 @@ # 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.dvoImageTable + " AS b \ - SET a.SOURCE_ID = b.SOURCE_ID \ - WHERE a.IMAGE_ID = b.IMAGE_ID" - self.scratchDb.execute(sql) + #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 @@ -389,8 +418,7 @@ # 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("Populating", self.scratchDb.dvoDetectionTable) + self.logger.infoPair("Merging into", self.scratchDb.dvoDetectionTable) sql = "INSERT IGNORE INTO " + self.scratchDb.dvoDetectionTable + " (\ - sourceID \ - ,imageID \ + imageID \ ,ippDetectID \ ,detectID \ @@ -400,6 +428,5 @@ ,fileID \ ) SELECT \ - SOURCE_ID \ - ,IMAGE_ID \ + IMAGE_ID \ ,DET_ID \ ,EXT_ID \ @@ -413,10 +440,9 @@ except: self.logger.error("FAILED: " + sql) - return + continue # now drop what we don't need - self.logger.infoPair("Dropping table", cpmTableName) + self.logger.infoPair("Dropping tables", "%s and %s" % (cpmTableName, cptTableName)) self.scratchDb.dropTable(cpmTableName) - self.logger.infoPair("Dropping table", cptTableName) self.scratchDb.dropTable(cptTableName) @@ -428,4 +454,8 @@ self.logger.infoPair("Ingested", "%d DVO regions" % regionCount) + # update lists after attempted sync + self.setSkyArea(self.minRa, self.maxRa, self.minDec, self.maxDec) + self.printSummary() + return True @@ -436,5 +466,5 @@ def importFits(self, path, columns, tableName): - self.logger.infoPair("Importing file", path) + self.logger.infoPair("Importing file", "%s (%s)" % (path, self.getFileSizeStr(self.getDiskSize(path)))) self.logger.debugPair("Writing to database table", tableName) @@ -444,5 +474,5 @@ for table in tables: - self.logger.infoPair("Reading IPP table", table.name) + self.logger.debugPair("Reading IPP table", table.name) table = stilts.tpipe(table, cmd='explodeall') @@ -453,9 +483,9 @@ attempts = 0 - self.logger.infoPair("Writing table to database", table.name) + self.logger.debugPair("Writing table to database", table.name) while True: try: - if attempts > 0: self.logger.infoPair("Attempt", "%d" % attempts) + if attempts > 0: self.logger.infoPair("Attempt %d to write" % attempts, table.name) table.cmd_keepcols(columns).write(self.scratchDb.url + '#' + tableName) break