IPP Software Navigation Tools IPP Links Communication Pan-STARRS Links

Changeset 33234


Ignore:
Timestamp:
Feb 10, 2012, 1:37:43 PM (14 years ago)
Author:
rhenders
Message:

lots of changes: now, a region is first defined using setSkyArea() then MySQL and DVO are synced using the sync method()

File:
1 edited

Legend:

Unmodified
Added
Removed

  • trunk/ippToPsps/jython/dvo.py

    r33193 r33234  
    4949        self.correctDvo = self.scratchDb.isCorrectDvo(self.dvoLocation)
    5050
     51        # set up empty lists
     52        self.regionsToIngest = []
     53        self.regionsToPurge = []
     54        self.allPopulatedRegions = []
     55        self.regionsAlreadyIngested = []
     56        self.regionsIngestedButOutOfDate = []
     57
     58        if not self.correctDvo:
     59            response = raw_input("* Wrong DVO is use. Do you want to reset and use '" + self.dvoLocation + "' instead (y/n)? ")
     60            if response == "y":
     61                response = raw_input("* Are you ABSOLUTELY sure you want to do this? (y/n)? ")
     62                if response == "y":
     63                    self.resetAllTables()
     64
    5165    '''
    5266    Destructor
    … …  
    6781
    6882        self.logger.infoPair("Resetting", "just detections")
    69         self.scratchDb.resetOnlyDetectionDvoTables()
     83        self.scratchDb.resetDvoDetectionTables()
    7084
    7185    '''
    … …  
    7589
    7690        self.logger.infoPair("Resetting", "all tables")
    77         self.scratchDb.resetDvoToMysqlTables()
     91        self.scratchDb.resetAllDvoTables()
    7892        self.correctDvo = True
    7993
    8094    '''
    81     Loads the DVO Iamges.dat FITS file
     95    Loads the DVO Images.dat FITS file
    8296    '''
    8397    def loadImages(self):
    … …  
    96110
    97111        # first, delete old dvoMeta table
    98         self.logger.infoPair("Deleting from table", self.scratchDb.dvoMetaTable)
    99         sql = "DELETE FROM " + self.scratchDb.dvoMetaTable
    100         self.scratchDb.execute(sql)
    101         self.scratchDb.optimizeTable(self.scratchDb.dvoMetaTable)
     112        #self.logger.infoPair("Deleting from table", self.scratchDb.dvoMetaTable)
     113        #sql = "DELETE FROM " + self.scratchDb.dvoMetaTable
     114        #self.scratchDb.execute(sql)
     115        #self.scratchDb.optimizeTable(self.scratchDb.dvoMetaTable)
    102116   
    103117        self.importFits(
    104118                path,
    105                 "SOURCE_ID IMAGE_ID CCDNUM EXTERN_ID FLAGS PHOTCODE NSTAR",
    106                 self.scratchDb.dvoImageTable)
    107         self.scratchDb.createIndex(self.scratchDb.dvoImageTable, "EXTERN_ID")
     119                "SOURCE_ID IMAGE_ID EXTERN_ID FLAGS PHOTCODE",
     120                self.scratchDb.dvoImagesTable)
     121        self.logger.infoPair("Adding primary key to", self.scratchDb.dvoImagesTable)
     122        self.scratchDb.execute("ALTER TABLE " + self.scratchDb.dvoImagesTable + " ADD PRIMARY KEY (IMAGE_ID, EXTERN_ID)")
    108123
    109124        # insert into dvoMetaFull
    110125        # NB what we and smf files call IMAGE_ID, DVO calls EXTERN_ID. We are sticking
    111126        # with the smf name
     127        '''
    112128        self.logger.infoPair("Populating", "all image meta data")
    113129        sql = "INSERT INTO " + self.scratchDb.dvoMetaTable + " ( \
    114                sourceID, \
    115130               imageID, \
    116131               externID, \
    … …  
    118133               photcode \
    119134               ) SELECT \
    120                SOURCE_ID, \
    121135               IMAGE_ID, \
    122136               EXTERN_ID, \
    123137               FLAGS, \
    124138               PHOTCODE \
    125                FROM " + self.scratchDb.dvoImageTable
     139               FROM " + self.scratchDb.dvoImagesTable
    126140
    127141        try:
    … …  
    130144            self.logger.errorPair("Could not insert into", self.scratchDb.dvoMetaTable)
    131145            return
     146        '''
    132147
    133148        self.scratchDb.setImportedThisDvoTable(path)
    … …  
    160175        self.logger.infoPair("Finished importing SkyTable at", self.dvoLocation)
    161176
    162 
    163     '''
    164     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
    165     '''
    166     def getRegionsInThisBoxWithBorder(self, regionsToIngest, regionsToPurge, ra=180., dec=0., side=181.0, border=0.):
    167 
    168         halfSide = (side + (border*2.0)) /2.0
    169         self.getRegions(regionsToIngest, regionsToPurge, ra-halfSide, ra+halfSide, dec-halfSide, dec+halfSide)
     177    '''
     178    Sets a bunch of region files from DVO for a RA/Dec box centered with these coords and with this side length
     179    '''
     180    def setSkyAreaAsBox(self, ra=180., dec=0., side=181.0):
     181
     182        halfSide = side/2.0
     183        self.setSkyArea(ra-halfSide, ra+halfSide, dec-halfSide, dec+halfSide)
    170184   
    171     '''
    172     Gets a bunch of region files from DVO for a RA/Dec box centered with these coords and with this side length
    173     '''
    174     def getRegionsInThisBox(self, regionsToIngest, regionsToPurge, ra=180., dec=0., side=181.0):
    175 
    176         halfSide = side/2.0
    177         self.getRegions(regionsToIngest, regionsToPurge, ra-halfSide, ra+halfSide, dec-halfSide, dec+halfSide)
    178    
    179 
    180     '''
    181     Gets a bunch of region files from DVO for this RA/Dec box
    182     '''
    183     def getRegions(self, regionsToIngest, regionsToPurge, minRa=-1., maxRa=361., minDec=-91., maxDec=91):
     185
     186    '''
     187    Sets a bunch of region files from DVO for this RA/Dec box
     188    '''
     189    def setSkyArea(self, minRa=-1., maxRa=361., minDec=-91., maxDec=91):
    184190   
    185191        # go no further if we've already partly ingested a different DVO
    … …  
    191197            return False       
    192198
     199        # update the two meta tables, Images and SkyTable
     200        self.loadImages()
     201        self.loadSkyTable()
     202
     203        # set bounding-box coords
     204        self.minRa = minRa
     205        self.maxRa = maxRa
     206        self.minDec = minDec
     207        self.maxDec = maxDec
     208
     209        # reset all lists
    193210        allRegions = self.scratchDb.getDvoRegionsForThisBox(minRa, maxRa, minDec, maxDec)
    194211        allIngestedRegions = self.scratchDb.getIngestedDvoRegions()
    195 
    196         allPopulatedRegions = []
    197         regionsAlreadyIngested = []
    198         regionsIngestedButOutOfDate = []
    199 
     212        self.regionsToIngest = []
     213        self.regionsToPurge = []
     214        self.allPopulatedRegions = []
     215        self.regionsAlreadyIngested = []
     216        self.regionsIngestedButOutOfDate = []
     217
     218        # loop through all regions in DVO inside this bounding-box
    200219        for region in allRegions:
    201220
    … …  
    207226           if not os.path.isfile(cptPath): continue
    208227
    209            allPopulatedRegions.append(region)
     228           # ok, have a real, populated file so add it to the list
     229           self.allPopulatedRegions.append(region)
    210230
    211231           # if we have already imported up-to-date versions of both the cpm and cpt then we skip this region
    212232           if self.scratchDb.alreadyImportedThisDvoTable(cpmPath) and self.scratchDb.alreadyImportedThisDvoTable(cptPath):
    213                regionsAlreadyIngested.append(region)
     233               self.regionsAlreadyIngested.append(region)
    214234               continue
    215235
    216236           # check if we have an out-of-date version of either region file, if so, add it to the purge list
    217237           if self.scratchDb.haveOutOfDateVersionOfThisDvoTable(cpmPath) or self.scratchDb.haveOutOfDateVersionOfThisDvoTable(cptPath):
    218                regionsIngestedButOutOfDate.append(region)
    219 
    220            regionsToIngest.append(region)
     238               self.regionsIngestedButOutOfDate.append(region)
     239
     240           self.regionsToIngest.append(region)
    221241
    222242        # add out-of-range regions for purge list
    223         items = list(set(allIngestedRegions) - set(allPopulatedRegions))
    224         for item in items: regionsToPurge.append(item)
     243        items = list(set(allIngestedRegions) - set(self.allPopulatedRegions))
     244        for item in items: self.regionsToPurge.append(item)
    225245
    226246        # now add out-of-date items to purge list
    227         for item in regionsIngestedButOutOfDate: regionsToPurge.append(item)
    228 
    229         sizeOfAllPopulatedRegions = self.getFileSizeStr(self.getDiskSizeOfDvoRegions(allPopulatedRegions))
    230         sizeOfRegionsAlreadyIngested = self.getFileSizeStr(self.getDiskSizeOfDvoRegions(regionsAlreadyIngested))
    231         sizeOfRegionsIngestedButOutOfDate = self.getFileSizeStr(self.getDiskSizeOfDvoRegions(regionsIngestedButOutOfDate))
    232         sizeOfRegionsToPurge = self.getFileSizeStr(self.getDiskSizeOfDvoRegions(regionsToPurge))
    233         sizeOfRegionsToIngest = self.getFileSizeStr(self.getDiskSizeOfDvoRegions(regionsToIngest))
    234 
    235         self.logger.infoTitle("Summary for this region")
    236         self.logger.infoPair("RA range", "%.2f -> %.2f" % (minRa, maxRa))
    237         self.logger.infoPair("Dec range", "%.2f -> %.2f" % (minDec, maxDec))
    238         self.logger.infoPair("Total regions", "%d (%s)" % (len(allPopulatedRegions), sizeOfAllPopulatedRegions))
    239         self.logger.infoPair("Up-to-date regions already ingested", "%d (%s)" % (len(regionsAlreadyIngested), sizeOfRegionsAlreadyIngested))
    240         self.logger.infoPair("Out-of-date regions ingested", "%d (%s)" % (len(regionsIngestedButOutOfDate), sizeOfRegionsIngestedButOutOfDate))
    241         self.logger.infoPair("Total regions to purge", "%d (%s)" % (len(regionsToPurge), sizeOfRegionsToPurge))
    242         self.logger.infoPair("Total regions to ingest", "%d (%s)" % (len(regionsToIngest), sizeOfRegionsToIngest))
    243 
     247        for item in self.regionsIngestedButOutOfDate: self.regionsToPurge.append(item)
     248
     249        self.sizeOfAllPopulatedRegions = self.getDiskSizeOfDvoRegions(self.allPopulatedRegions)
     250        self.sizeOfRegionsAlreadyIngested = self.getDiskSizeOfDvoRegions(self.regionsAlreadyIngested)
     251        self.sizeOfRegionsIngestedButOutOfDate = self.getDiskSizeOfDvoRegions(self.regionsIngestedButOutOfDate)
     252        self.sizeOfRegionsToPurge = self.getDiskSizeOfDvoRegions(self.regionsToPurge)
     253        self.sizeOfRegionsToIngest = self.getDiskSizeOfDvoRegions(self.regionsToIngest)
     254
     255    '''
     256    Prints out a summary of the currently set regions
     257    '''
     258    def printSummary(self):
     259
     260        self.logger.infoTitle("Summary for currently set DVO region")
     261
     262        # go no further if we've already partly ingested a different DVO
     263        if not self.correctDvo:
     264            self.logger.infoPair("Wrong DVO in use", self.dvoLocation)
     265            return
     266
     267        self.logger.infoPair("RA range", "%.2f -> %.2f" % (self.minRa, self.maxRa))
     268        self.logger.infoPair("Dec range", "%.2f -> %.2f" % (self.minDec, self.maxDec))
     269        self.logger.infoPair("Total regions", "%d (%s)" % (len(self.allPopulatedRegions), self.getFileSizeStr(self.sizeOfAllPopulatedRegions)))
     270        self.logger.infoPair("Up-to-date regions already ingested", "%d (%s)" % (len(self.regionsAlreadyIngested), self.getFileSizeStr(self.sizeOfRegionsAlreadyIngested)))
     271        self.logger.infoPair("Out-of-date regions ingested", "%d (%s)" % (len(self.regionsIngestedButOutOfDate), self.getFileSizeStr(self.sizeOfRegionsIngestedButOutOfDate)))
     272        self.logger.infoPair("Total regions to purge", "%d (%s)" % (len(self.regionsToPurge), self.getFileSizeStr(self.sizeOfRegionsToPurge)))
     273        self.logger.infoPair("Total regions to ingest", "%d (%s)" % (len(self.regionsToIngest), self.getFileSizeStr(self.sizeOfRegionsToIngest)))
     274
     275
     276    '''
     277    Returns the size on disk of the regions that need to be ingested in GB
     278    '''
     279    def getDiskSizeOfRegionsToBeIngested(self):
     280
     281        try:
     282            return self.sizeOfRegionsToIngest/1073741824.0
     283        except:
     284            return 0.0
    244285
    245286    '''
    246287    Purges the provided regions from the database
    247288    '''
    248     def purgeRegions(self, regions):
     289    def purgeRegions(self):
    249290
    250291        # go no further if we've already partly ingested a different DVO
    251292        if not self.correctDvo: return
    252293
    253         if len(regions) < 1: return
    254 
    255         self.logger.infoPair("Purging", "%d regions" % len(regions))
    256         deleteCount = self.scratchDb.purgeTheseDvoRegions(regions)
     294        if len(self.regionsToPurge) < 1: return
     295
     296        self.logger.infoPair("Purging", "%d regions" % len(self.regionsToPurge))
     297        deleteCount = self.scratchDb.purgeTheseDvoRegions(self.regionsToPurge)
    257298        self.logger.infoPair("Deleted", "%d detections" % deleteCount)
    258299
    … …  
    271312
    272313    '''
     314    Returns the disk size of this file
     315    '''
     316    def getDiskSize(self, path):
     317
     318        return os.stat(path).st_size
     319
     320    '''
    273321    Returns the disk size of these DVO regions, i.e. the size on disk of all cpm and cpt files
    274322    '''
    … …  
    279327
    280328            # get combined size of cpm and cpt files
    281             size = size + os.stat(self.dvoLocation + "/" + region + ".cpm").st_size
    282             size = size + os.stat(self.dvoLocation + "/" + region + ".cpt").st_size
     329            size = size + self.getDiskSize(self.dvoLocation + "/" + region + ".cpm")
     330            size = size + self.getDiskSize(self.dvoLocation + "/" + region + ".cpt")
    283331
    284332        return size
    285333
    286334    '''
    287     Populates dvoDetections and dvoMeta with FITS tables from DVO for
    288     a RA/Dec box centered with these coords and with this side length PLUS a border
    289     '''
    290     def loadRegionInThisBoxWithBorder(self, ra=180., dec=0., side=181.0, border=0.):
    291 
    292         halfSide = (side + (border*2.0)) /2.0
    293         self.loadRegion(ra-halfSide, ra+halfSide, dec-halfSide, dec+halfSide)
    294    
    295     '''
    296     Populates dvoDetections and dvoMeta with FITS tables from DVO for
    297     a RA/Dec box centered with these coords and with this side length
    298     '''
    299     def loadRegionInThisBox(self, regionsToIngest, regionsToPurge, ra=180., dec=0., side=181.0):
    300 
    301         halfSide = side/2.0
    302         self.loadRegion(ra-halfSide, ra+halfSide, dec-halfSide, dec+halfSide)
    303    
    304     '''
    305     Populates dvoDetections and dvoMeta with FITS tables from DVO for the defined region
    306     Defaults to whole sky if limits are missing
    307     '''
    308     def loadRegion(self, minRa=-1., maxRa=361., minDec=-91., maxDec=91):
     335    Synchronises MySQL database with the sky area currently set from DVO
     336    Populates dvoDetections and dvoDone with FITS tables from DVO for the defined sky area, this
     337    includes purging detections outside sky area
     338    '''
     339    def sync(self):
    309340
    310341        # go no further if we've already partly ingested a different DVO
    311         if not self.correctDvo: return
    312 
    313         # update the two meta tables, Images and SkyTable
    314         self.loadImages()
    315         self.loadSkyTable()
    316 
    317         regionsToIngest = []
    318         regionsToPurge = []
    319         self.getRegions(regionsToIngest, regionsToPurge, minRa, maxRa, minDec, maxDec)
    320 
    321         self.purgeRegions(regionsToPurge)
    322 
    323         totalRegions = len(regionsToIngest)
     342        if not self.correctDvo: return False
     343
     344        # check we have something to do
     345        if len(self.regionsToPurge) < 1 and len(self.regionsToIngest) < 1:
     346            self.logger.infoPair("Nothing to ingest or purge", "finished ingest")
     347            return True
     348
     349        # purge stuff
     350        self.purgeRegions()
     351
     352        totalRegions = len(self.regionsToIngest)
    324353        if totalRegions > 0: self.logger.infoSeparator()
    325354
    326355        # now loop through all regions and ingest to MySQL
    327356        regionCount = 0
    328         for region in regionsToIngest:
     357        for region in self.regionsToIngest:
    329358
    330359           cpmPath = self.dvoLocation + "/" + region + ".cpm"
    … …  
    334363           cptTableName = self.scratchDb.getDbFriendlyTableName(region + ".cpt")
    335364
    336            self.logger.infoTitle("Region %d of %d: %s" % (regionCount, totalRegions, region))
     365           self.logger.infoTitle("Region %d of %d: %s" % (regionCount+1, totalRegions, region))
    337366
    338367           # if we have not already ingested the cpm table, then do it now
    … …  
    350379
    351380               # shove SOURCE_IDs into measurement table
    352                self.logger.infoPair("Adding", "SOURCE_IDs")
    353                sql = "ALTER TABLE " + cpmTableName + " ADD COLUMN (SOURCE_ID SMALLINT)"
    354                self.scratchDb.execute(sql)
    355                sql = "UPDATE " + cpmTableName + " AS a, " + self.scratchDb.dvoImageTable + " AS b \
    356                       SET a.SOURCE_ID = b.SOURCE_ID \
    357                       WHERE a.IMAGE_ID = b.IMAGE_ID"
    358                self.scratchDb.execute(sql)
     381               #self.logger.infoPair("Adding", "SOURCE_IDs")
     382               #sql = "ALTER TABLE " + cpmTableName + " ADD COLUMN (SOURCE_ID SMALLINT)"
     383               #self.scratchDb.execute(sql)
     384               #sql = "UPDATE " + cpmTableName + " AS a, " + self.scratchDb.dvoImagesTable + " AS b \
     385               #       SET a.SOURCE_ID = b.SOURCE_ID \
     386               #       WHERE a.IMAGE_ID = b.IMAGE_ID"
     387               #self.scratchDb.execute(sql)
    359388
    360389               # we can now report that we have imported this table
    … …  
    389418           # PSPS object IDs assigned in the corresponding cpt file. This is believed to be a chip-boundary
    390419           # issue within DVO. So, for now, we take the first IMAGE_ID/DET_ID detection we find, and ignore the rest
    391            self.logger.infoPair("Populating", self.scratchDb.dvoDetectionTable)
     420           self.logger.infoPair("Merging into", self.scratchDb.dvoDetectionTable)
    392421           sql = "INSERT IGNORE INTO " + self.scratchDb.dvoDetectionTable + " (\
    393                   sourceID \
    394                   ,imageID \
     422                  imageID \
    395423                  ,ippDetectID \
    396424                  ,detectID \
    … …  
    400428                  ,fileID \
    401429                  ) SELECT \
    402                   SOURCE_ID \
    403                   ,IMAGE_ID \
     430                  IMAGE_ID \
    404431                  ,DET_ID \
    405432                  ,EXT_ID \
    … …  
    413440           except:
    414441               self.logger.error("FAILED: " + sql)
    415                return
     442               continue
    416443
    417444           # now drop what we don't need
    418            self.logger.infoPair("Dropping table", cpmTableName)
     445           self.logger.infoPair("Dropping tables", "%s and %s" % (cpmTableName, cptTableName))
    419446           self.scratchDb.dropTable(cpmTableName)
    420            self.logger.infoPair("Dropping table", cptTableName)
    421447           self.scratchDb.dropTable(cptTableName)
    422448
    … …  
    428454        self.logger.infoPair("Ingested", "%d DVO regions" % regionCount)
    429455
     456        # update lists after attempted sync
     457        self.setSkyArea(self.minRa, self.maxRa, self.minDec, self.maxDec)
     458        self.printSummary()
     459
    430460        return True
    431461
    … …  
    436466    def importFits(self, path, columns, tableName):
    437467
    438       self.logger.infoPair("Importing file", path)
     468      self.logger.infoPair("Importing file", "%s (%s)" % (path, self.getFileSizeStr(self.getDiskSize(path))))
    439469      self.logger.debugPair("Writing to database table", tableName)
    440470
    … …  
    444474      for table in tables:
    445475
    446           self.logger.infoPair("Reading IPP table", table.name)
     476          self.logger.debugPair("Reading IPP table", table.name)
    447477          table = stilts.tpipe(table, cmd='explodeall')
    448478     
    … …  
    453483
    454484          attempts = 0
    455           self.logger.infoPair("Writing table to database",  table.name)
     485          self.logger.debugPair("Writing table to database",  table.name)
    456486          while True:
    457487   
    458488              try:
    459                   if attempts > 0: self.logger.infoPair("Attempt", "%d" % attempts)
     489                  if attempts > 0: self.logger.infoPair("Attempt %d to write" % attempts, table.name)
    460490                  table.cmd_keepcols(columns).write(self.scratchDb.url + '#' + tableName)
    461491                  break
Note: See TracChangeset for help on using the changeset viewer.