IPP Software Navigation Tools IPP Links Communication Pan-STARRS Links

Changeset 33599 for trunk/ippToPsps


Ignore:
Timestamp:
Mar 23, 2012, 12:45:53 PM (14 years ago)
Author:
rhenders
Message:

Dvo is now a base-class to two inherting classes for detections and objects ingestion

Location:
trunk/ippToPsps/jython
Files:
2 added
1 edited

Legend:

Unmodified
Added
Removed

  • trunk/ippToPsps/jython/dvo.py

    r33598 r33599  
    147147
    148148    '''
     149    Sets a bunch of region files from DVO for a stripe starting at this RA and with this width
     150    '''
     151    def setSkyAreaAsStripe(self, ra=0., width=4.):
     152
     153        self.setSkyArea(ra, ra+width, -91., 91.)
     154
     155    '''
    149156    Sets a bunch of region files from DVO for a RA/Dec box centered with these coords and with this side length
    150157    '''
    … …  
    155162
    156163    '''
    157     Sets a bunch of region files from DVO for this RA/Dec box
     164    Determines what DVO files need to be ingested to the defines region
    158165    '''
    159166    def setSkyArea(self, minRa=-1., maxRa=361., minDec=-91., maxDec=91):
    … …  
    168175
    169176        # update the two meta tables, Images and SkyTable
    170         #self.loadImages() dont need a new Images table at this point
    171177        self.loadSkyTable()
    172178
    … …  
    189195        for region in allRegions:
    190196
    191            cpmPath = self.config.dvoLocation + "/" + region + ".cpm"
    192            cptPath = self.config.dvoLocation + "/" + region + ".cpt"
    193 
    194            # check for existence of cpm and cpt files
    195            if not os.path.isfile(cpmPath): continue
    196            if not os.path.isfile(cptPath): continue
    197 
    198            # ok, have a real, populated file so add it to the list
     197           # get paths for this region for the file types we are interested in
     198           paths = []
     199           for fileType in self.ingestFileTypes:
     200               paths.append(self.config.dvoLocation + "/" + region + "." + fileType)
     201
     202           # check for the existence of all interested file types
     203           skipThisRegion = False
     204           for path in paths:
     205               if not os.path.isfile(path):
     206                   skipThisRegion = True
     207                   break
     208
     209           if skipThisRegion: continue
     210
     211           # ok, have real, populated files so add this region to the list
    199212           self.allPopulatedRegions.append(region)
    200213
    201            # if we have already imported up-to-date versions of both the cpm and cpt then we skip this region
    202            if self.scratchDb.alreadyImportedThisDvoTable(cpmPath) and self.scratchDb.alreadyImportedThisDvoTable(cptPath):
     214           # if we have already imported up-to-date versions of all interested files, then we skip this region
     215           alreadyIngested = True
     216           for path in paths:
     217               if not self.scratchDb.alreadyImportedThisDvoTable(path):
     218                   alreadyIngested = False
     219                   break
     220   
     221           if alreadyIngested:
    203222               self.regionsAlreadyIngested.append(region)
    204223               continue
    205224
    206            # check if we have an out-of-date version of either region file, if so, add it to the purge list
    207            if self.scratchDb.haveOutOfDateVersionOfThisDvoTable(cpmPath) or self.scratchDb.haveOutOfDateVersionOfThisDvoTable(cptPath):
    208                self.regionsIngestedButOutOfDate.append(region)
     225           # check if we have out-of-date versions of any interested files, if so, add region to the purge list
     226           outOfDate = False
     227           for path in paths:
     228               if self.scratchDb.haveOutOfDateVersionOfThisDvoTable(path):
     229                   outOfDate = True
     230                   break
     231
     232           if outOfDate: self.regionsIngestedButOutOfDate.append(region)
    209233
    210234           self.regionsToIngest.append(region)
    … …  
    229253
    230254        self.logger.infoTitle("Summary for currently set DVO region")
     255        self.logger.infoPair("DVO ingest type", self.__class__.__name__)
    231256
    232257        # go no further if we've already partly ingested a different DVO
    … …  
    235260            return
    236261
     262        for fileType in self.ingestFileTypes:
     263            self.logger.infoPair("Ingesting files of type", fileType)
     264
    237265        self.logger.infoPair("RA range", "%.2f -> %.2f" % (self.minRa, self.maxRa))
    238266        self.logger.infoPair("Dec range", "%.2f -> %.2f" % (self.minDec, self.maxDec))
    239         self.logger.infoPair("Total regions", "%d (%s)" % (len(self.allPopulatedRegions), self.getFileSizeStr(self.sizeOfAllPopulatedRegions)))
    240         self.logger.infoPair("Up-to-date regions already ingested", "%d (%s)" % (len(self.regionsAlreadyIngested), self.getFileSizeStr(self.sizeOfRegionsAlreadyIngested)))
    241         self.logger.infoPair("Out-of-date regions ingested", "%d (%s)" % (len(self.regionsIngestedButOutOfDate), self.getFileSizeStr(self.sizeOfRegionsIngestedButOutOfDate)))
    242         self.logger.infoPair("Total regions to purge", "%d (%s)" % (len(self.regionsToPurge), self.getFileSizeStr(self.sizeOfRegionsToPurge)))
    243         self.logger.infoPair("Total regions to ingest", "%d (%s)" % (len(self.regionsToIngest), self.getFileSizeStr(self.sizeOfRegionsToIngest)))
    244 
     267
     268        self.logger.info("+----------------+----------------+----------------+----------------+----------------+")
     269        self.logger.info("|   populated    |   ingested     |  out-of-date   |    to-purge    |   to ingest    |")
     270        self.logger.info("+----------------+----------------+----------------+----------------+----------------+")
     271        self.logger.info("| %5d (%6s) | %5d (%6s) | %5d (%6s) | %5d (%6s) | %5d (%6s) |" % (
     272                    len(self.allPopulatedRegions), self.getFileSizeStr(self.sizeOfAllPopulatedRegions),
     273                    len(self.regionsAlreadyIngested), self.getFileSizeStr(self.sizeOfRegionsAlreadyIngested),
     274                    len(self.regionsIngestedButOutOfDate), self.getFileSizeStr(self.sizeOfRegionsIngestedButOutOfDate),
     275                    len(self.regionsToPurge), self.getFileSizeStr(self.sizeOfRegionsToPurge),
     276                    len(self.regionsToIngest), self.getFileSizeStr(self.sizeOfRegionsToIngest)))
     277        self.logger.info("+----------------+----------------+----------------+----------------+----------------+")
    245278
    246279    '''
    … …  
    260293
    261294        # go no further if we've already partly ingested a different DVO
    262         if not self.correctDvo: return
    263 
    264         if len(self.regionsToPurge) < 1: return
     295        if not self.correctDvo: return False
     296
     297        if len(self.regionsToPurge) < 1: return False
    265298
    266299        self.logger.infoPair("Purging", "%d regions" % len(self.regionsToPurge))
    267         deleteCount = self.scratchDb.purgeTheseDvoRegions(self.regionsToPurge)
    268         self.logger.infoPair("Deleted", "%d detections" % deleteCount)
     300
     301        detectionDeleteCount = self.scratchDb.purgeTheseDvoRegions(self.regionsToPurge, self.ingestFileTypes)
     302        if detectionDeleteCount > 0: self.logger.infoPair("Deleted", "%d detections" % detectionDeleteCount)
     303
     304        return True
    269305
    270306    '''
    … …  
    289325
    290326    '''
    291     Returns the disk size of these DVO regions, i.e. the size on disk of all cpm and cpt files
     327    Returns the disk size of these DVO regions, i.e. the size on disk of all relevant DVO FITS files
    292328    '''
    293329    def getDiskSizeOfDvoRegions(self, regions):
    … …  
    296332        for region in regions:
    297333
    298             # get combined size of cpm and cpt files
    299             size = size + self.getDiskSize(self.config.dvoLocation + "/" + region + ".cpm")
    300             size = size + self.getDiskSize(self.config.dvoLocation + "/" + region + ".cpt")
     334             # get combined size of all interested files
     335             for fileType in self.ingestFileTypes:
     336                 size = size + self.getDiskSize(self.config.dvoLocation + "/" + region + "." + fileType)
    301337
    302338        return size
     339
     340    '''
     341    Abstract method for ingesting a particular reegion. Must be implemented in all subclasses
     342    '''
     343    def ingestRegion(self, region):
     344        self.logger.errorPair("ingestRegion()", "not implemented for this class")
     345
    303346
    304347    '''
    … …  
    327370
    328371        # now loop through all regions and ingest to MySQL
    329         regionCount = 0
     372        ingestedRegionsCount = 0
     373        regionCount = 1
    330374        for region in self.regionsToIngest:
    331375
    332            cpmPath = self.config.dvoLocation + "/" + region + ".cpm"
    333            cptPath = self.config.dvoLocation + "/" + region + ".cpt"
    334 
    335            cpmTableName = self.scratchDb.getDbFriendlyTableName(region + ".cpm")
    336            cptTableName = self.scratchDb.getDbFriendlyTableName(region + ".cpt")
    337 
    338            self.logger.infoTitle("Region %d of %d: %s" % (regionCount+1, totalRegions, region))
    339 
    340            # if we have not already ingested the cpm table, then do it now
    341            if not self.scratchDb.alreadyImportedThisDvoTable(cpmPath):
    342    
    343                # import cpm table and index
    344                self.importFits(
    345                        cpmPath,
    346                        "IMAGE_ID DET_ID OBJ_ID CAT_ID EXT_ID DB_FLAGS",
    347                        cpmTableName)
    348 
    349                self.scratchDb.createIndex(cpmTableName, "IMAGE_ID")
    350                self.scratchDb.createIndex(cpmTableName, "CAT_ID")
    351                self.scratchDb.createIndex(cpmTableName, "OBJ_ID")
    352 
    353                # shove SOURCE_IDs into measurement table
    354                #self.logger.infoPair("Adding", "SOURCE_IDs")
    355                #sql = "ALTER TABLE " + cpmTableName + " ADD COLUMN (SOURCE_ID SMALLINT)"
    356                #self.scratchDb.execute(sql)
    357                #sql = "UPDATE " + cpmTableName + " AS a, " + self.scratchDb.dvoImagesTable + " AS b \
    358                #       SET a.SOURCE_ID = b.SOURCE_ID \
    359                #       WHERE a.IMAGE_ID = b.IMAGE_ID"
    360                #self.scratchDb.execute(sql)
    361 
    362                # we can now report that we have imported this table
    363                self.scratchDb.setImportedThisDvoTable(cpmPath, region)
    364 
    365            # get the id for this cpm file so we can insert it in the dvoDetection table
    366            fileId = self.scratchDb.getDvoFileId(cpmPath)
    367 
    368            # import cpt table and index
    369            self.importFits(
    370                    cptPath,
    371                    "OBJ_ID CAT_ID EXT_ID",
    372                    cptTableName)
    373 
    374            self.scratchDb.createIndex(cptTableName, "IMAGE_ID")
    375            self.scratchDb.createIndex(cptTableName, "CAT_ID")
    376            self.scratchDb.createIndex(cptTableName, "OBJ_ID")
    377 
    378            self.logger.infoPair("Adding","PSPS objIDs")
    379            # first try to add a column for PSPS_OBJ_ID. catch failure and continue
    380            try:
    381                sql = "ALTER TABLE " + cpmTableName + " ADD COLUMN (PSPS_OBJ_ID BIGINT)"
    382                self.scratchDb.execute(sql)
    383            except:
    384                self.logger.infoPair("Already added PSPS_OBJ_ID column to", cpmTableName)
    385                
    386            # shove PSPS objIDs from cpt table into measurement table
    387            sql = "UPDATE " + cpmTableName + " AS a, " + cptTableName + " AS b \
    388                   SET a.PSPS_OBJ_ID = b.EXT_ID \
    389                   WHERE a.CAT_ID = b.CAT_ID \
    390                   AND a.OBJ_ID = b.OBJ_ID"
    391            self.scratchDb.execute(sql)
    392 
    393 
    394            # NB we use an INSERT IGNORE here. This is because of a known issue where multiple DVO cpm
    395            # files can include the same detection, with the same IMAGE_ID/DET_ID pairing, but different
    396            # PSPS object IDs assigned in the corresponding cpt file. This is believed to be a chip-boundary
    397            # issue within DVO. So, for now, we take the first IMAGE_ID/DET_ID detection we find, and ignore the rest
    398            self.logger.infoPair("Merging into", self.scratchDb.dvoDetectionTable)
    399            sql = "INSERT IGNORE INTO " + self.scratchDb.dvoDetectionTable + " (\
    400                   imageID \
    401                   ,ippDetectID \
    402                   ,detectID \
    403                   ,ippObjID \
    404                   ,objID \
    405                   ,flags \
    406                   ,fileID \
    407                   ) SELECT \
    408                   IMAGE_ID \
    409                   ,DET_ID \
    410                   ,EXT_ID \
    411                   ,CAT_ID * 1000000000 + OBJ_ID \
    412                   ,PSPS_OBJ_ID \
    413                   ,DB_FLAGS \
    414                   , " + str(fileId) + " \
    415                   FROM " + cpmTableName
    416            try:
    417                self.scratchDb.execute(sql)
    418            except:
    419                self.logger.error("FAILED: " + sql)
    420                continue
    421 
    422            # now drop what we don't need
    423            self.logger.infoPair("Dropping tables", "%s and %s" % (cpmTableName, cptTableName))
    424            self.scratchDb.dropTable(cpmTableName)
    425            self.scratchDb.dropTable(cptTableName)
    426 
    427            self.scratchDb.setImportedThisDvoTable(cptPath, region)
    428            regionCount = regionCount + 1
     376           self.logger.infoTitle("Region %d of %d: %s" % (regionCount, totalRegions, region))
     377           if self.ingestRegion(region): ingestedRegionsCount += 1
     378           regionCount += 1
    429379
    430380        self.logger.infoSeparator()
    431 
    432         self.logger.infoPair("Ingested", "%d DVO regions" % regionCount)
     381        self.logger.infoPair("Ingested", "%d DVO regions (out of %d)" % (ingestedRegionsCount, len(self.regionsToIngest)))
    433382
    434383        # update lists after attempted sync
Note: See TracChangeset for help on using the changeset viewer.