IPP Software Navigation Tools IPP Links Communication Pan-STARRS Links

Ignore:
Timestamp:
Jun 19, 2012, 5:24:19 PM (14 years ago)
Author:
mhuber
Message:

merging latest r34040 trunk changes to branch

Location:
branches/meh_branches/ppstack_test
Files:
2 edited

Legend:

Unmodified
Added
Removed

  • branches/meh_branches/ppstack_test

  • branches/meh_branches/ppstack_test/ippToPsps/jython/dvo.py

    r33415 r34041  
    1717from java.sql import *
    1818
    19 
    2019'''
    21 A class that encapsulates a DVO database
    22 
    23 Pulls relavant bits and pieces into the scratch MySQL database and then provides access queries
     20Abstract base-class for DVO classes
     21
     22Provides generic functionality to pull relevant stuff from a DVO database into a scratch MySQL database while also providing access queries
    2423'''
    2524class Dvo(object):
    … …  
    3433
    3534    '''
    36     def __init__(self, logger, config):
     35    def __init__(self, logger, config, scratchDbName="ipptopsps_scratch"):
    3736
    3837        # set up logging
    … …  
    4645        self.regionsAlreadyIngested = []
    4746        self.regionsIngestedButOutOfDate = []
    48        
    49         #self.correctDvo = False
    50 
    51 
    52     '''
    53     Sets to scratch database
    54     '''
    55     def setScratchDb(self, dbName):
    56 
    57         self.scratchDb = ScratchDb(self.logger, self.config, dbName)
     47       
     48        # connect to the specified scratch database
     49        try:
     50            self.scratchDb = ScratchDb(self.logger, self.config, scratchDbName)
     51        except: raise
     52
    5853        self.scratchDb.setUseFullTables(True)
    5954
    60         # or decide if we are using the right DVO
     55        # decide if we are using the right DVO for this scratchDb
    6156        self.correctDvo = self.scratchDb.isCorrectDvo(self.config.dvoLocation)
    62  
     57
     58        # this scratch Db either has no DVO info ingested, or a different DVO ingested. Prompt user
    6359        if not self.correctDvo:
    6460            print "*******************************************************************************"
    65             response = raw_input("**** Wrong or no DVO in use. Do you want to reset and use '" + self.config.dvoLocation + "' (y/n)? ")
     61            self.logger.warning("Wrong or no DVO in use. Do you want to reset and use '" + self.config.dvoLocation + "'?")
     62            response = raw_input("(y/n) ")
    6663            if response == "y":
    67                 response = raw_input("**** Are you ABSOLUTELY sure you want to do this? (y/n)? ")
     64                self.logger.warning("Are you ABSOLUTELY sure you want to do this?")
     65                response = raw_input("(y/n) ")
    6866                if response == "y":
    6967                    self.resetAllTables()
    … …  
    8381
    8482    '''
    85     Resets all detections. Be careful
    86     '''
    87     def resetDetections(self):
    88 
    89         self.logger.infoPair("Resetting", "just detections")
    90         self.scratchDb.resetDvoDetectionTables()
    91 
    92     '''
    9383    Resets all dvo tables. Be very careful....
    9484    '''
    … …  
    9686
    9787        self.logger.infoPair("Resetting", "all tables")
    98         self.scratchDb.resetAllDvoTables()
     88        if not self.scratchDb.resetAllDvoTables(): return
    9989        self.correctDvo = True
     90        if not self.loadSkyTable():
     91            self.correctDvo = False
     92            return
     93
     94    '''
     95    Loads the DVO Photcodes.dat FITS file
     96    '''
     97    def loadPhotcodes(self):
     98
     99        # go no further if we've already partly ingested a different DVO
     100        if not self.correctDvo: return
     101
     102        # check if we have up-to-date version
     103        path = self.config.dvoLocation + "/Photcodes.dat"
     104        if self.scratchDb.alreadyImportedThisDvoTable(path):
     105            self.logger.debugPair("DVO Photcodes.dat file", "up-to-date")
     106            return
     107       
     108        self.logger.infoSeparator()
     109        self.logger.infoPair("DVO Photcodes.dat file", "NOT up-to-date")
     110
     111        self.importFits(
     112                path,
     113                "CODE NAME",
     114                self.scratchDb.dvoPhotcodesTable)
     115        self.logger.infoPair("Adding primary key to", self.scratchDb.dvoPhotcodesTable)
     116        self.scratchDb.makeColumnPrimaryKey(self.scratchDb.dvoPhotcodesTable, "CODE")
     117
     118        self.scratchDb.setImportedThisDvoTable(path)
    100119
    101120    '''
    … …  
    116135        self.logger.infoPair("DVO Images.dat file", "NOT up-to-date")
    117136
    118         # first, delete old dvoMeta table
    119         #self.logger.infoPair("Deleting from table", self.scratchDb.dvoMetaTable)
    120         #sql = "DELETE FROM " + self.scratchDb.dvoMetaTable
    121         #self.scratchDb.execute(sql)
    122         #self.scratchDb.optimizeTable(self.scratchDb.dvoMetaTable)
    123    
    124137        self.importFits(
    125138                path,
    … …  
    129142        self.scratchDb.execute("ALTER TABLE " + self.scratchDb.dvoImagesTable + " ADD PRIMARY KEY (IMAGE_ID, EXTERN_ID)")
    130143
    131         # insert into dvoMetaFull
    132         # NB what we and smf files call IMAGE_ID, DVO calls EXTERN_ID. We are sticking
    133         # with the smf name
    134         '''
    135         self.logger.infoPair("Populating", "all image meta data")
    136         sql = "INSERT INTO " + self.scratchDb.dvoMetaTable + " ( \
    137                imageID, \
    138                externID, \
    139                flags, \
    140                photcode \
    141                ) SELECT \
    142                IMAGE_ID, \
    143                EXTERN_ID, \
    144                FLAGS, \
    145                PHOTCODE \
    146                FROM " + self.scratchDb.dvoImagesTable
    147 
    148         try:
    149             self.scratchDb.execute(sql)
    150         except:
    151             self.logger.errorPair("Could not insert into", self.scratchDb.dvoMetaTable)
    152             return
    153         '''
    154 
    155144        self.scratchDb.setImportedThisDvoTable(path)
    156145
    … …  
    161150       
    162151        # go no further if we've already partly ingested a different DVO
    163         if not self.correctDvo: return
     152        if not self.correctDvo: return False
    164153
    165154        path =  self.config.dvoLocation + "/SkyTable.fits"
    166155        if self.scratchDb.alreadyImportedThisDvoTable(path):
    167156            self.logger.debugPair("DVO SkyTable.fits file", "up-to-date")       
    168             return
     157            return False
    169158
    170159        self.logger.infoSeparator()
    … …  
    182171        self.logger.infoPair("Finished importing SkyTable at", self.config.dvoLocation)
    183172
     173        return True
     174
     175    '''
     176    Sets a bunch of region files from DVO for a stripe starting at this RA and with this width
     177    '''
     178    def setSkyAreaAsStripe(self, ra=0., width=4.):
     179
     180        self.setSkyArea(ra, ra+width, -91., 91.)
     181
    184182    '''
    185183    Sets a bunch of region files from DVO for a RA/Dec box centered with these coords and with this side length
    … …  
    191189
    192190    '''
    193     Sets a bunch of region files from DVO for this RA/Dec box
     191    Determines what DVO files need to be ingested to the defines region
    194192    '''
    195193    def setSkyArea(self, minRa=-1., maxRa=361., minDec=-91., maxDec=91):
    … …  
    204202
    205203        # update the two meta tables, Images and SkyTable
    206         #self.loadImages() dont need a new Images table at this point
    207204        self.loadSkyTable()
    208205
    … …  
    219216        self.regionsToPurge = []
    220217        self.allPopulatedRegions = []
     218        self.allPopulatedRegionInfo = []
    221219        self.regionsAlreadyIngested = []
    222220        self.regionsIngestedButOutOfDate = []
     221        tablesToCleanup = []
    223222
    224223        # loop through all regions in DVO inside this bounding-box
    225224        for region in allRegions:
    226225
    227            cpmPath = self.config.dvoLocation + "/" + region + ".cpm"
    228            cptPath = self.config.dvoLocation + "/" + region + ".cpt"
    229 
    230            # check for existence of cpm and cpt files
    231            if not os.path.isfile(cpmPath): continue
    232            if not os.path.isfile(cptPath): continue
    233 
    234            # ok, have a real, populated file so add it to the list
    235            self.allPopulatedRegions.append(region)
    236 
    237            # if we have already imported up-to-date versions of both the cpm and cpt then we skip this region
    238            if self.scratchDb.alreadyImportedThisDvoTable(cpmPath) and self.scratchDb.alreadyImportedThisDvoTable(cptPath):
    239                self.regionsAlreadyIngested.append(region)
     226           regionId = region[0]
     227           regionRa = region[1]
     228           regionDec = region[2]
     229           regionPath = region[3]
     230
     231           # get paths for this region for the file types we are interested in
     232           paths = []
     233           for fileType in self.ingestFileTypes:
     234               paths.append(self.config.dvoLocation + "/" + regionPath + "." + fileType)
     235
     236           # check for the existence of all interested file types
     237           skipThisRegion = False
     238           for path in paths:
     239               if not os.path.isfile(path):
     240                   skipThisRegion = True
     241                   break
     242
     243           if skipThisRegion: continue
     244
     245           # ok, have real, populated files so add this region to the list
     246           self.allPopulatedRegions.append(regionPath)
     247           self.allPopulatedRegionInfo.append([regionId, regionRa, regionDec])
     248
     249           # if we have already imported up-to-date versions of all interested files, then we skip this region
     250           alreadyIngested = True
     251           for path in paths:
     252               if not self.scratchDb.alreadyImportedThisDvoTable(path):
     253                   alreadyIngested = False
     254                   break
     255   
     256           if alreadyIngested:
     257               self.regionsAlreadyIngested.append(regionPath)
    240258               continue
    241259
    242            # check if we have an out-of-date version of either region file, if so, add it to the purge list
    243            if self.scratchDb.haveOutOfDateVersionOfThisDvoTable(cpmPath) or self.scratchDb.haveOutOfDateVersionOfThisDvoTable(cptPath):
    244                self.regionsIngestedButOutOfDate.append(region)
    245 
    246            self.regionsToIngest.append(region)
     260           # we have no record of ingesting this region, but does the table exist?
     261           # May hev been left behind after a crash, and therefore needs to be cleaned-up
     262           for fileType in self.ingestFileTypes:
     263               tableName = self.scratchDb.getDbFriendlyTableName(regionPath + "." + fileType)
     264               if self.scratchDb.tableExists(tableName):
     265                   tablesToCleanup.append(tableName)
     266
     267           # check if we have out-of-date versions of any interested files, if so, add region to the purge list
     268           outOfDate = False
     269           for path in paths:
     270               if self.scratchDb.haveOutOfDateVersionOfThisDvoTable(path):
     271                   outOfDate = True
     272                   break
     273
     274           if outOfDate: self.regionsIngestedButOutOfDate.append(regionPath)
     275
     276           self.regionsToIngest.append(regionPath)
    247277
    248278        # add out-of-range regions for purge list
    … …  
    253283        for item in self.regionsIngestedButOutOfDate: self.regionsToPurge.append(item)
    254284
    255         self.sizeOfAllPopulatedRegions = self.getDiskSizeOfDvoRegions(self.allPopulatedRegions)
    256         self.sizeOfRegionsAlreadyIngested = self.getDiskSizeOfDvoRegions(self.regionsAlreadyIngested)
    257         self.sizeOfRegionsIngestedButOutOfDate = self.getDiskSizeOfDvoRegions(self.regionsIngestedButOutOfDate)
    258         self.sizeOfRegionsToPurge = self.getDiskSizeOfDvoRegions(self.regionsToPurge)
    259         self.sizeOfRegionsToIngest = self.getDiskSizeOfDvoRegions(self.regionsToIngest)
     285        self.sizeOfAllPopulatedRegions = self.getDiskSizeOfRegions(self.allPopulatedRegions)
     286        self.sizeOfRegionsAlreadyIngested = self.getDiskSizeOfRegions(self.regionsAlreadyIngested)
     287        self.sizeOfRegionsIngestedButOutOfDate = self.getDiskSizeOfRegions(self.regionsIngestedButOutOfDate)
     288        self.sizeOfRegionsToPurge = self.getDiskSizeOfRegions(self.regionsToPurge)
     289        self.sizeOfRegionsToIngest = self.getDiskSizeOfRegions(self.regionsToIngest)
     290
     291        # any tables need cleaning up?
     292        if len(tablesToCleanup) > 0:
     293            self.logger.infoPair("Found tables to be cleaned up", "%d" % len(tablesToCleanup))
     294            self.scratchDb.dropTables(tablesToCleanup)
     295           
    260296
    261297    '''
    … …  
    265301
    266302        self.logger.infoTitle("Summary for currently set DVO region")
     303        self.logger.infoPair("DVO ingest type", self.__class__.__name__)
    267304
    268305        # go no further if we've already partly ingested a different DVO
    … …  
    271308            return
    272309
     310        for fileType in self.ingestFileTypes:
     311            self.logger.infoPair("Ingesting files of type", fileType)
     312
    273313        self.logger.infoPair("RA range", "%.2f -> %.2f" % (self.minRa, self.maxRa))
    274314        self.logger.infoPair("Dec range", "%.2f -> %.2f" % (self.minDec, self.maxDec))
    275         self.logger.infoPair("Total regions", "%d (%s)" % (len(self.allPopulatedRegions), self.getFileSizeStr(self.sizeOfAllPopulatedRegions)))
    276         self.logger.infoPair("Up-to-date regions already ingested", "%d (%s)" % (len(self.regionsAlreadyIngested), self.getFileSizeStr(self.sizeOfRegionsAlreadyIngested)))
    277         self.logger.infoPair("Out-of-date regions ingested", "%d (%s)" % (len(self.regionsIngestedButOutOfDate), self.getFileSizeStr(self.sizeOfRegionsIngestedButOutOfDate)))
    278         self.logger.infoPair("Total regions to purge", "%d (%s)" % (len(self.regionsToPurge), self.getFileSizeStr(self.sizeOfRegionsToPurge)))
    279         self.logger.infoPair("Total regions to ingest", "%d (%s)" % (len(self.regionsToIngest), self.getFileSizeStr(self.sizeOfRegionsToIngest)))
    280 
     315
     316        self.logger.info("+----------------+----------------+----------------+----------------+----------------+")
     317        self.logger.info("|   populated    |   ingested     |  out-of-date   |    to-purge    |   to ingest    |")
     318        self.logger.info("+----------------+----------------+----------------+----------------+----------------+")
     319        self.logger.info("| %5d (%6s) | %5d (%6s) | %5d (%6s) | %5d (%6s) | %5d (%6s) |" % (
     320                    len(self.allPopulatedRegions), self.getFileSizeStr(self.sizeOfAllPopulatedRegions),
     321                    len(self.regionsAlreadyIngested), self.getFileSizeStr(self.sizeOfRegionsAlreadyIngested),
     322                    len(self.regionsIngestedButOutOfDate), self.getFileSizeStr(self.sizeOfRegionsIngestedButOutOfDate),
     323                    len(self.regionsToPurge), self.getFileSizeStr(self.sizeOfRegionsToPurge),
     324                    len(self.regionsToIngest), self.getFileSizeStr(self.sizeOfRegionsToIngest)))
     325        self.logger.info("+----------------+----------------+----------------+----------------+----------------+")
    281326
    282327    '''
    … …  
    291336
    292337    '''
     338    Purges a single region from the database
     339    '''
     340    def purgeRegion(self, region):
     341
     342        regions = []
     343        regions.append(region)
     344        self.purgeRegions(regions)
     345
     346    '''
    293347    Purges the provided regions from the database
    294348    '''
    295     def purgeRegions(self):
    296 
    297         # go no further if we've already partly ingested a different DVO
    298         if not self.correctDvo: return
    299 
    300         if len(self.regionsToPurge) < 1: return
    301 
    302         self.logger.infoPair("Purging", "%d regions" % len(self.regionsToPurge))
    303         deleteCount = self.scratchDb.purgeTheseDvoRegions(self.regionsToPurge)
    304         self.logger.infoPair("Deleted", "%d detections" % deleteCount)
     349    def purgeRegions(self, regions):
     350
     351        # go no further if we've already partly ingested a different DVO
     352        if not self.correctDvo: return False
     353
     354        if len(regions) < 1: return False
     355
     356        self.logger.infoPair("Purging", "%d regions" % len(regions))
     357
     358        detectionDeleteCount = self.scratchDb.purgeTheseDvoRegions(regions, self.ingestFileTypes)
     359        if detectionDeleteCount > 0: self.logger.infoPair("Deleted", "%d detections" % detectionDeleteCount)
     360
     361        return True
    305362
    306363    '''
    … …  
    325382
    326383    '''
    327     Returns the disk size of these DVO regions, i.e. the size on disk of all cpm and cpt files
    328     '''
    329     def getDiskSizeOfDvoRegions(self, regions):
     384    Returns the disk size of these DVO regions, i.e. the size on disk of all relevant DVO FITS files
     385    '''
     386    def getDiskSizeOfRegions(self, regions):
    330387
    331388        size = 0.0
    332389        for region in regions:
    333390
    334             # get combined size of cpm and cpt files
    335             size = size + self.getDiskSize(self.config.dvoLocation + "/" + region + ".cpm")
    336             size = size + self.getDiskSize(self.config.dvoLocation + "/" + region + ".cpt")
     391             # get combined size of all interested files
     392             for fileType in self.ingestFileTypes:
     393                 size = size + self.getDiskSize(self.config.dvoLocation + "/" + region + "." + fileType)
    337394
    338395        return size
     396
     397
     398    '''
     399    Ingests object files for this region, given an INDEX from the SkyTable
     400    '''
     401    def ingestRegionUsingIndex(self, index):
     402        return self.ingestRegion(self.scratchDb.getRegionNameFromThisDvoIndex(index))
     403
     404    '''
     405    Abstract method for ingesting a particular reegion. Must be implemented in all subclasses
     406    '''
     407    def ingestRegion(self, region):
     408        return self.logger.errorPair("ingestRegion()", "not implemented for this class")
    339409
    340410    '''
    … …  
    357427
    358428        # purge stuff
    359         self.purgeRegions()
     429        self.purgeRegions(self.regionsToPurge)
    360430
    361431        totalRegions = len(self.regionsToIngest)
    … …  
    363433
    364434        # now loop through all regions and ingest to MySQL
    365         regionCount = 0
     435        ingestedRegionsCount = 0
     436        regionCount = 1
    366437        for region in self.regionsToIngest:
    367438
    368            cpmPath = self.config.dvoLocation + "/" + region + ".cpm"
    369            cptPath = self.config.dvoLocation + "/" + region + ".cpt"
    370 
    371            cpmTableName = self.scratchDb.getDbFriendlyTableName(region + ".cpm")
    372            cptTableName = self.scratchDb.getDbFriendlyTableName(region + ".cpt")
    373 
    374            self.logger.infoTitle("Region %d of %d: %s" % (regionCount+1, totalRegions, region))
    375 
    376            # if we have not already ingested the cpm table, then do it now
    377            if not self.scratchDb.alreadyImportedThisDvoTable(cpmPath):
    378    
    379                # import cpm table and index
    380                self.importFits(
    381                        cpmPath,
    382                        "IMAGE_ID DET_ID OBJ_ID CAT_ID EXT_ID DB_FLAGS",
    383                        cpmTableName)
    384 
    385                self.scratchDb.createIndex(cpmTableName, "IMAGE_ID")
    386                self.scratchDb.createIndex(cpmTableName, "CAT_ID")
    387                self.scratchDb.createIndex(cpmTableName, "OBJ_ID")
    388 
    389                # shove SOURCE_IDs into measurement table
    390                #self.logger.infoPair("Adding", "SOURCE_IDs")
    391                #sql = "ALTER TABLE " + cpmTableName + " ADD COLUMN (SOURCE_ID SMALLINT)"
    392                #self.scratchDb.execute(sql)
    393                #sql = "UPDATE " + cpmTableName + " AS a, " + self.scratchDb.dvoImagesTable + " AS b \
    394                #       SET a.SOURCE_ID = b.SOURCE_ID \
    395                #       WHERE a.IMAGE_ID = b.IMAGE_ID"
    396                #self.scratchDb.execute(sql)
    397 
    398                # we can now report that we have imported this table
    399                self.scratchDb.setImportedThisDvoTable(cpmPath, region)
    400 
    401            # get the id for this cpm file so we can insert it in the dvoDetection table
    402            fileId = self.scratchDb.getDvoFileId(cpmPath)
    403 
    404            # import cpt table and index
    405            self.importFits(
    406                    cptPath,
    407                    "OBJ_ID CAT_ID EXT_ID",
    408                    cptTableName)
    409 
    410            self.scratchDb.createIndex(cptTableName, "IMAGE_ID")
    411            self.scratchDb.createIndex(cptTableName, "CAT_ID")
    412            self.scratchDb.createIndex(cptTableName, "OBJ_ID")
    413 
    414            self.logger.infoPair("Adding","PSPS objIDs")
    415            # first try to add a column for PSPS_OBJ_ID. catch failure and continue
    416            try:
    417                sql = "ALTER TABLE " + cpmTableName + " ADD COLUMN (PSPS_OBJ_ID BIGINT)"
    418                self.scratchDb.execute(sql)
    419            except:
    420                self.logger.infoPair("Already added PSPS_OBJ_ID column to", cpmTableName)
    421                
    422            # shove PSPS objIDs from cpt table into measurement table
    423            sql = "UPDATE " + cpmTableName + " AS a, " + cptTableName + " AS b \
    424                   SET a.PSPS_OBJ_ID = b.EXT_ID \
    425                   WHERE a.CAT_ID = b.CAT_ID \
    426                   AND a.OBJ_ID = b.OBJ_ID"
    427            self.scratchDb.execute(sql)
    428 
    429 
    430            # NB we use an INSERT IGNORE here. This is because of a known issue where multiple DVO cpm
    431            # files can include the same detection, with the same IMAGE_ID/DET_ID pairing, but different
    432            # PSPS object IDs assigned in the corresponding cpt file. This is believed to be a chip-boundary
    433            # issue within DVO. So, for now, we take the first IMAGE_ID/DET_ID detection we find, and ignore the rest
    434            self.logger.infoPair("Merging into", self.scratchDb.dvoDetectionTable)
    435            sql = "INSERT IGNORE INTO " + self.scratchDb.dvoDetectionTable + " (\
    436                   imageID \
    437                   ,ippDetectID \
    438                   ,detectID \
    439                   ,ippObjID \
    440                   ,objID \
    441                   ,flags \
    442                   ,fileID \
    443                   ) SELECT \
    444                   IMAGE_ID \
    445                   ,DET_ID \
    446                   ,EXT_ID \
    447                   ,CAT_ID * 1000000000 + OBJ_ID \
    448                   ,PSPS_OBJ_ID \
    449                   ,DB_FLAGS \
    450                   , " + str(fileId) + " \
    451                   FROM " + cpmTableName
    452            try:
    453                self.scratchDb.execute(sql)
    454            except:
    455                self.logger.error("FAILED: " + sql)
    456                continue
    457 
    458            # now drop what we don't need
    459            self.logger.infoPair("Dropping tables", "%s and %s" % (cpmTableName, cptTableName))
    460            self.scratchDb.dropTable(cpmTableName)
    461            self.scratchDb.dropTable(cptTableName)
    462 
    463            self.scratchDb.setImportedThisDvoTable(cptPath, region)
    464            regionCount = regionCount + 1
     439           self.logger.infoTitle("Region %d of %d: %s" % (regionCount, totalRegions, region))
     440           if self.ingestRegion(region): ingestedRegionsCount += 1
     441           regionCount += 1
    465442
    466443        self.logger.infoSeparator()
    467 
    468         self.logger.infoPair("Ingested", "%d DVO regions" % regionCount)
     444        self.logger.infoPair("Ingested", "%d DVO regions (out of %d)" % (ingestedRegionsCount, len(self.regionsToIngest)))
    469445
    470446        # update lists after attempted sync
    … …  
    476452    '''
    477453    Imports a FITS file. A regex filter lets you choose which tables to pull from the file
    478     An optional final argument lets you name the resulatant database table
     454    An optional final argument lets you name the resultant database table
    479455    '''
    480456    def importFits(self, path, columns, tableName):
    … …  
    491467          table = stilts.tpipe(table, cmd='explodeall')
    492468     
    493           # IPP FITS files are littered with infinities, so remove these
     469          # IPP FITS files are littered with infinity values. Remove them
    494470          self.logger.debugPair("Removing", "infinity values")
    495471          table = stilts.tpipe(table, cmd='replaceval -Infinity null *')
Note: See TracChangeset for help on using the changeset viewer.