Changeset 33161 for trunk

Timestamp:

Jan 26, 2012, 12:25:36 PM (14 years ago)

Author:

rhenders

Message:

now doesn't re-ingest cpm file if previous attempt was aborted after cpm but before cpt; now checking we are using correct DVO; counting stuff already ingested, stuff we need to ingest (including bytes on disk)

File:

• trunk/ippToPsps/jython/dvo.py (modified) (10 diffs)

trunk/ippToPsps/jython/dvo.py

@@ -43 +43 @@
         self.logger.infoPair("Started", "dvo")
 
+        # decide if we are using the right DVO
+        self.correctDvo = self.scratchDb.isThisDvoCurrentlyIngested(self.dvoLocation)
+
     '''
     Destructor
@@ -49 +52 @@
         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 +64 @@
     def resetAllTables(self):
         self.scratchDb.resetDvoToMysqlTables()
+        self.correctDvo = True
 
     '''
@@ -59 +70 @@
     '''
     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 +120 @@
     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 +139 @@
 
     '''
-    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 +150 @@
         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 +325 @@
            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 +350 @@
     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 +398 @@
 
 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")