Index: trunk/ippToPsps/jython/dvo.py =================================================================== --- trunk/ippToPsps/jython/dvo.py (revision 33160) +++ trunk/ippToPsps/jython/dvo.py (revision 33161) @@ -43,4 +43,7 @@ self.logger.infoPair("Started", "dvo") + # decide if we are using the right DVO + self.correctDvo = self.scratchDb.isThisDvoCurrentlyIngested(self.dvoLocation) + ''' Destructor @@ -49,4 +52,11 @@ self.logger.debug("Dvo destructor") + + ''' + Check we are importing the DVO currenly ingested + ''' + def isThisDvoCurrentlyIngested(self): + return self.correctDvo + ''' Resets all dvo tables. Be very careful.... @@ -54,4 +64,5 @@ def resetAllTables(self): self.scratchDb.resetDvoToMysqlTables() + self.correctDvo = True ''' @@ -59,4 +70,7 @@ ''' def loadImages(self): + + # go no further if we've already partly ingested a different DVO + if not self.correctDvo: return # check if we have up-to-date version @@ -106,4 +120,7 @@ def loadSkyTable(self): + # go no further if we've already partly ingested a different DVO + if not self.correctDvo: return + path = self.dvoLocation + "/SkyTable.fits" if self.scratchDb.alreadyImportedThisDvoTable(path): return @@ -122,9 +139,8 @@ ''' - 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): - + Gets a bunch of region files from DVO for this RA/Dec box + ''' + def getRegionFiles(self, minRa=-1., maxRa=361., minDec=-91., maxDec=91): + if (maxRa - minRa <= 0) or (maxDec - minDec) <= 0: self.logger.errorPair("Zero range in either RA or Dec", @@ -134,51 +150,139 @@ self.logger.infoPair("Finding DVO files for range", "RA: %.2f -> %.2f, Dec: %.2f -> %.2f" % (minRa, maxRa, minDec, maxDec)) - files = self.scratchDb.getDvoFilesCoveringThisRegion(minRa, maxRa, minDec, maxDec) - - count = 0 - - self.logger.infoSeparator() - for file in files: + + allFiles = self.scratchDb.getDvoFilesCoveringThisRegion(minRa, maxRa, minDec, maxDec) + + files = [] + + total = 0 + alreadyIngested = 0 + toIngest = 0 + for file in allFiles: cpmPath = self.dvoLocation + "/" + file + ".cpm" cptPath = self.dvoLocation + "/" + file + ".cpt" + # check for existence of cpm and cpt files if not os.path.isfile(cpmPath): continue - count = count + 1 - - # only if we have already imported up-to-date versions of both the cpm and ctp - # will we skip this one - if self.scratchDb.alreadyImportedThisDvoTable(cpmPath) and self.scratchDb.alreadyImportedThisDvoTable(cptPath): continue - - # import cpm table and index - cpmTableName = self.importFits( - cpmPath, - "IMAGE_ID DET_ID OBJ_ID CAT_ID EXT_ID DB_FLAGS") - self.scratchDb.createIndex(cpmTableName, "IMAGE_ID") - self.scratchDb.createIndex(cpmTableName, "CAT_ID") - self.scratchDb.createIndex(cpmTableName, "OBJ_ID") + if not os.path.isfile(cptPath): continue + + total = total + 1 + + # 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): + alreadyIngested = alreadyIngested + 1 + continue + + files.append(file) + + toIngest = toIngest + 1 + + self.logger.infoPair("Total DVO regions to ingest", "%d" % total) + self.logger.infoPair("DVO regions already ingested", "%d" % alreadyIngested) + self.logger.infoPair("DVO regions to ingest", "%d" % toIngest) + + return files + + ''' + Formats a nice file size string + Accepts a size in bytes + ''' + def getFileSizeStr(self, byteSize): + + # format a string for the user -smaller than one MB + if byteSize < 1048576: return "%.1f Kb" % (byteSize/1024.0) + # smaller than one GB + elif byteSize < 1073741824: return "%.1f Mb" % (byteSize/1048576.0) + # more than a GB + else: return "%.1f Gb" % (byteSize/1073741824.0) + + ''' + Returns the disk size of this region of data in DVO + ''' + def getSizeOfDvoRegion(self, files): + + size = 0.0 + for file in files: + + # get combined size of cpm and cpt files + size = size + os.stat(self.dvoLocation + "/" + file + ".cpm").st_size + size = size + os.stat(self.dvoLocation + "/" + file + ".cpt").st_size + + return size + + ''' + Returns (and reports to log) total size of DVO files ingested to the scratch Db + ''' + def getSizeAlreadyIngested(self): + + size = self.scratchDb.getTotalSizeOfIngestedDvoFiles() + self.logger.infoPair("Total size of DVO already ingested", self.getFileSizeStr(size)) + + + ''' + 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): + + # go no further if we've already partly ingested a different DVO + if not self.correctDvo: return + + files = self.getRegionFiles(minRa, maxRa, minDec, maxDec) + totalSize = self.getSizeOfDvoRegion(files) + self.logger.infoPair("Total size of cpm/cpt files to ingest", self.getFileSizeStr(totalSize)) + count = 0 + + self.logger.infoSeparator() + + # now loop through all files + for file in files: + + cpmPath = self.dvoLocation + "/" + file + ".cpm" + cptPath = self.dvoLocation + "/" + file + ".cpt" + + cpmTableName = self.getDatabaseFriendlyTableName(file + ".cpm") + cptTableName = self.getDatabaseFriendlyTableName(file + ".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") + + # 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) + + # we can report that we have imported this table + self.scratchDb.setImportedThisDvoTable(cpmPath) # import cpt table and index - cptTableName = self.importFits( + self.importFits( cptPath, - "OBJ_ID CAT_ID EXT_ID") + "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") - # 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) - - # shove PSPS objID in measurement table + # shove PSPS objIDs from cpt table into measurement table self.logger.infoPair("Adding","PSPS objIDs") sql = "ALTER TABLE "+cpmTableName+" ADD COLUMN (PSPS_OBJ_ID BIGINT)" self.scratchDb.execute(sql) - sql = "UPDATE "+cpmTableName+" AS a, "+cptTableName+" AS b \ + sql = "UPDATE " + cpmTableName + " AS a, " + cptTableName + " AS b \ SET a.PSPS_OBJ_ID = b.EXT_ID \ WHERE a.CAT_ID = b.CAT_ID \ @@ -221,10 +325,23 @@ self.scratchDb.dropTable(cptTableName) - self.scratchDb.setImportedThisDvoTable(cpmPath) self.scratchDb.setImportedThisDvoTable(cptPath) - - self.logger.infoPair("Found", "%d region files" % count) + count = count + 1 + + self.logger.infoSeparator() + + self.logger.infoPair("Ingested", "%d DVO region files" % count) return True + + ''' + Create Db-friendly table name from file name + ''' + def getDatabaseFriendlyTableName(self, file): + + name = file + name = name.replace('.', '_') + name = name.replace('/', '_') + + return name @@ -233,14 +350,7 @@ An optional final argument lets you name the resulatant database table ''' - def importFits(self, path, columns, tableName=None): + def importFits(self, path, columns, tableName): self.logger.infoPair("Importing file", path) - - if not tableName: - head,tail = os.path.split(path) - tableName = tail - tableName = tableName.replace('.', '_') - tableName = tableName.replace('/', '_') - self.logger.infoPair("Writing to database table", tableName) @@ -288,8 +398,11 @@ dvo = Dvo(logger, configDoc) +dvo.isThisDvoCurrentlyIngested() #dvo.resetAllTables() +dvo.getSizeAlreadyIngested() dvo.loadImages() dvo.loadSkyTable() -dvo.loadRegion(330, 330.1, 0.2, 0.7) +dvo.loadRegion(330, 331, 0.2, 3) +#dvo.loadRegion() logger.infoPair("Program...", "complete")