- Timestamp:
- May 29, 2011, 11:26:31 AM (15 years ago)
- File:
-
- 1 edited
Legend:
- Unmodified
- Added
- Removed
-
branches/eam_branches/ipp-20110505/ippToPsps/jython/stackbatch.py
r31402 r31587 8 8 from java.sql import * 9 9 10 from batch import Batch 10 11 from gpc1db import Gpc1Db 11 from batch import Batch 12 from ipptopspsdb import IppToPspsDb 13 from scratchdb import ScratchDb 14 12 15 import logging.config 13 16 … … 20 23 Constructor 21 24 ''' 22 def __init__(self, logger, skyID, inputFile, stackType, useFullTables=False): 25 def __init__(self, 26 logger, 27 gpc1Db, 28 ippToPspsDb, 29 scratchDb, 30 skyID, 31 inputFile, 32 stackType, 33 useFullTables=False): 34 23 35 super(StackBatch, self).__init__( 24 36 logger, 37 gpc1Db, 38 ippToPspsDb, 39 scratchDb, 40 skyID, 25 41 "stack", 26 42 inputFile, … … 28 44 useFullTables) # TODO 29 45 30 self.logger.info("StackBatch constructor. Creating batch from: '" + inputFile + "'") 31 32 self.skyID = skyID 33 34 # get filterID using init table 35 self.filter = self.header['FPA.FILTER'] 36 self.filter = self.filter[0:1] 37 38 self.stackType = stackType 39 meta = self.gpc1Db.getStackStageMeta(self.skyID, self.header['FPA.FILTER']) 40 if len(meta) < 1: return 41 self.stackID = meta[0]; 42 self.skycell = meta[1]; 43 44 # determine skycell from header value 45 #self.skycell = "skycell.34" #= self.header['SKYCELL'] 46 self.skycell = self.skycell[8:] 47 48 self.logger.info("Processing stack with ID: %d, type: %s and skycell: %s filter: %s" % (self.stackID, self.stackType, self.skycell, self.filter)) 49 46 if not self.everythingOK: return 47 48 self.expTime = gpc1Db.getStackExpTime(self.stackID) 49 50 self.logger.info("got exp time of %d" % self.expTime) 51 52 # meta data to the log 53 self.logger.info("New Stack Batch:") 54 self.logger.info("Sky ID: %d" % self.id) 55 self.logger.info("File: %s" % inputFile) 56 self.logger.info("Stack ID: %d" % self.stackID) 57 self.logger.info("Stack type: %s" % self.stackType) 58 self.logger.info("Skycell: %s" % self.skycell) 59 self.logger.info("Filter: %s" % self.filter) 50 60 51 61 # delete PSPS tables … … 58 68 self.scratchDb.dropTable("ObjectCalColor") 59 69 60 # delete IPP tables61 #self.scratchDb.dropTable("SkyChip_psf")62 #self.scratchDb.dropTable("SkyChip_xsrc")63 #self.scratchDb.dropTable("SkyChip_xfit")64 #self.scratchDb.dropTable("SkyChip_xrad")65 66 self.logger.info("Stack type: " + self.safeDictionaryAccess(self.header, self.stackType))67 # obs time makes no sense except for nightly stacks68 #if self.header['STK_TYPE'] != "NIGHTLY_STACK": self.header['MJD-OBS'] = "-999"69 70 70 # create an output filename, which is {filterID}{skycellID}.FITS 71 self.outputFitsFile = "% s%07d.FITS" % (self.filter, int(self.skycell))71 self.outputFitsFile = "%08d.FITS" % self.stackID 72 72 self.outputFitsPath = "%s/%s" % (self.localOutPath, self.outputFitsFile) 73 73 … … 77 77 78 78 # insert what we know about this stack batch into the stack table 79 self.ippToPspsDb.insertStackMeta(self.batchID, self. skyID, self.stackID, self.filter, self.stackType)79 self.ippToPspsDb.insertStackMeta(self.batchID, self.id, self.stackID, self.filter, self.stackType) 80 80 81 81 # insert sourceID/imageID combo so DVO can look it up … … 89 89 90 90 sql = "UPDATE " + table + " SET stackMetaID=" + str(self.stackID) 91 self.scratchDb. stmt.execute(sql)91 self.scratchDb.execute(sql) 92 92 93 93 ''' … … 97 97 98 98 sql = "UPDATE "+table+" AS a, StackType AS b SET a.stackTypeID=b.stackTypeID WHERE b.name = '" + self.stackType + "'" 99 self.scratchDb. stmt.execute(sql)99 self.scratchDb.execute(sql) 100 100 101 101 … … 148 148 WHERE a.ippDetectID=b.IPP_IDET AND b.PSF_FWHM "+psfCondition 149 149 150 self.scratchDb. stmt.execute(sql)150 self.scratchDb.execute(sql) 151 151 152 152 ''' … … 196 196 WHERE a.ippDetectID=b.IPP_IDET AND b.MODEL_TYPE = '"+ippModelType+"'" 197 197 198 self.scratchDb. stmt.execute(sql)198 self.scratchDb.execute(sql) 199 199 200 200 # sersic fit has an extra parameter … … 213 213 WHERE a.ippDetectID=b.IPP_IDET AND b.MODEL_TYPE = '"+ippModelType+"'" 214 214 215 self.scratchDb. stmt.execute(sql)215 self.scratchDb.execute(sql) 216 216 217 217 … … 220 220 ''' 221 221 def populateStackMeta(self): 222 222 223 self.logger.info("Procesing StackMeta table") 223 224 … … 246 247 ," + str(self.scratchDb.getPhotoCalID(self.header['SOURCEID'], self.header['IMAGEID'])) + " \ 247 248 ," + self.header['FPA.ZP'] + " \ 248 ," + s elf.header['EXPTIME']+ " \249 ," + str(self.expTime) + " \ 249 250 ,'" + self.safeDictionaryAccess(self.header, 'PSFMODEL') + "' \ 250 251 ,'" + self.header['CTYPE1'] + "' \ … … 261 262 ," + self.header['PC002002'] + " \ 262 263 )" 263 self.scratchDb. stmt.execute(sql)264 self.scratchDb.execute(sql) 264 265 265 266 self.scratchDb.updateAllRows("StackMeta", "surveyID", str(self.surveyID)) … … 272 273 ''' 273 274 def populateStackDetection(self): 275 274 276 self.logger.info("Procesing StackDetection table") 275 277 … … 321 323 ,X_PSF_SIG \ 322 324 ,Y_PSF_SIG \ 323 ,POW(10.0, (-0.4*PSF_INST_MAG)) / "+s elf.header['EXPTIME']+" \324 ,ABS((PSF_INST_MAG_SIG*(POW(10.0, (-0.4*PSF_INST_MAG)) / "+s elf.header['EXPTIME']+")) / 1.085736) \325 ,POW(10.0, (-0.4*PEAK_FLUX_AS_MAG)) / "+s elf.header['EXPTIME']+" \325 ,POW(10.0, (-0.4*PSF_INST_MAG)) / "+str(self.expTime)+" \ 326 ,ABS((PSF_INST_MAG_SIG*(POW(10.0, (-0.4*PSF_INST_MAG)) / "+str(self.expTime)+")) / 1.085736) \ 327 ,POW(10.0, (-0.4*PEAK_FLUX_AS_MAG)) / "+str(self.expTime)+" \ 326 328 ,SKY \ 327 329 ,SKY_SIGMA \ … … 352 354 FROM SkyChip_psf" 353 355 354 self.scratchDb. stmt.execute(sql)356 self.scratchDb.execute(sql) 355 357 356 358 self.scratchDb.updateAllRows("StackDetection", "surveyID", str(self.surveyID)) … … 362 364 self.updateStackTypeID("StackDetection") 363 365 self.updateDvoIDs("StackDetection") 364 365 # now delete bad flux 366 sql = "ALTER TABLE StackDetection ADD PRIMARY KEY (objID, stackDetectID)" 367 self.scratchDb.execute(sql) 368 369 if self.stackType == "DEEP_STACK": 370 371 #if deep stack and instFlux = null and err not null 372 sql = "UPDATE StackDetection AS a, SkyChip_psf AS b \ 373 SET instFlux = 2*b.PSF_INST_FLUX_SIG \ 374 WHERE instFlux IS NULL \ 375 AND b.PSF_INST_FLUX_SIG IS NOT NULL" 376 self.scratchDb.execute(sql) 377 # instFlux = 2*PSF_INST_FLUX_SIG 378 366 379 self.scratchDb.reportAndDeleteRowsWithNULLS("StackDetection", "instFlux") 367 380 self.scratchDb.reportAndDeleteRowsWithNULLS("StackDetection", "objID") … … 372 385 ''' 373 386 def populateStackApFlx(self): 387 374 388 self.logger.info("Procesing StackApFlx table") 375 389 … … 381 395 382 396 try: 383 self.scratchDb. stmt.execute(sql)397 self.scratchDb.execute(sql) 384 398 except: return 385 399 386 400 # TODO temporarily loading 1st convolved fluxes into unconvolved fields 387 self.logger.info(" 401 self.logger.info("Adding un-convolved fluxes") 388 402 self.updateApFlxs("", "< 7.0") 389 self.logger.info(" 403 self.logger.info("Adding 1st convolved fluxes") 390 404 self.updateApFlxs("c1", "< 7.0") 391 self.logger.info(" 405 self.logger.info("Adding 2nd convolved fluxes") 392 406 self.updateApFlxs("c2", "> 7.0") 393 407 394 self.logger.info(" 408 self.logger.info("Adding petrosians for extended sources") 395 409 sql = "UPDATE StackApFlx AS a, SkyChip_xsrc AS b SET \ 396 410 petRadius=b.PETRO_RADIUS \ … … 403 417 ,petR90Err=b.PETRO_RADIUS_90_ERR \ 404 418 WHERE a.ippDetectID=b.IPP_IDET" 405 self.scratchDb. stmt.execute(sql)419 self.scratchDb.execute(sql) 406 420 407 421 self.scratchDb.updateAllRows("StackApFlx", "surveyID", str(self.surveyID)) … … 413 427 self.updateStackTypeID("StackApFlx") 414 428 self.updateDvoIDs("StackApFlx") 429 self.scratchDb.reportAndDeleteRowsWithNULLS("StackApFlx", "objID") 430 self.deleteDetectionsNotInStackDetection("StackApFlx") 431 415 432 416 433 ''' … … 418 435 ''' 419 436 def populateStackModelFit(self): 437 420 438 self.logger.info("Procesing StackModelFit table") 421 439 … … 423 441 sql = "INSERT INTO StackModelFit (ippDetectID) SELECT DISTINCT IPP_IDET from SkyChip_xfit" 424 442 try: 425 self.scratchDb. stmt.execute(sql)443 self.scratchDb.execute(sql) 426 444 except: 427 445 return 428 446 429 430 447 # populate model parameters 431 self.logger.info(" 448 self.logger.info("Adding deVaucouleurs fit") 432 449 self.updateModelFit("deV", "PS_MODEL_DEV") 433 self.logger.info(" 450 self.logger.info("Adding exponential fit") 434 451 self.updateModelFit("exp", "PS_MODEL_EXP") 435 self.logger.info(" 452 self.logger.info("Adding sersic fit") 436 453 self.updateModelFit("ser", "PS_MODEL_SERSIC") 437 454 … … 444 461 self.updateStackTypeID("StackModelFit") 445 462 self.updateDvoIDs("StackModelFit") 463 self.scratchDb.reportAndDeleteRowsWithNULLS("StackModelFit", "objID") 464 self.deleteDetectionsNotInStackDetection("StackModelFit") 465 466 ''' 467 Reports and deletes detections in this table that are not in StackDetection 468 ''' 469 def deleteDetectionsNotInStackDetection(self, table): 470 471 sql = "SELECT COUNT(*) FROM " + table + " WHERE ippDetectID NOT IN (SELECT ippDetectID FROM StackDetection)" 472 rs = self.scratchDb.executeQuery(sql) 473 rs.first() 474 nMissing = rs.getInt(1) 475 self.logger.info("%5d detections in %s table that are not in StackDetection. Deleting" % (nMissing, table)) 476 477 if nMissing < 1: return 478 479 sql = "DELETE FROM " + table + " WHERE ippDetectID NOT IN (SELECT ippDetectID FROM StackDetection)" 480 self.scratchDb.execute(sql) 481 446 482 447 483 ''' … … 449 485 ''' 450 486 def populateStackToImage(self): 487 451 488 self.logger.info("Procesing StackToImage table") 452 489 … … 457 494 VALUES (\ 458 495 " + str(self.stackID) + ", " + imageID + ")" 459 self.scratchDb. stmt.execute(sql)496 self.scratchDb.execute(sql) 460 497 461 498 # now update StackMeta with correct number of inputs 462 499 sql = "UPDATE StackMeta SET nP2Images = (SELECT COUNT(*) FROM StackToImage)" 463 self.scratchDb. stmt.execute(sql)500 self.scratchDb.execute(sql) 464 501 465 502 ''' … … 467 504 ''' 468 505 def populateSkinnyObject(self): 506 469 507 self.logger.info("Procesing SkinnyObject table") 470 508 … … 474 512 ) \ 475 513 SELECT \ 476 objID \514 DISTINCT objID \ 477 515 ,ippObjID \ 478 516 FROM StackDetection" 479 self.scratchDb. stmt.execute(sql)517 self.scratchDb.execute(sql) 480 518 481 519 self.scratchDb.updateAllRows("SkinnyObject", "surveyID", str(self.surveyID)) … … 486 524 ''' 487 525 def populateObjectCalColor(self): 526 488 527 self.logger.info("Procesing ObjectCalColor table") 489 528 … … 493 532 ) \ 494 533 SELECT \ 495 objID \534 DISTINCT objID \ 496 535 ,ippObjID \ 497 536 FROM StackDetection" 498 self.scratchDb. stmt.execute(sql)537 self.scratchDb.execute(sql) 499 538 500 539 self.scratchDb.updateFilterID("ObjectCalColor", self.filter) … … 508 547 509 548 self.logger.info("Altering PSPS tables") 510 self.scratchDb.makeColumnUnique("StackDetection", "objID")549 #self.scratchDb.makeColumnUnique("StackDetection", "objID") 511 550 self.scratchDb.createIndex("StackDetection", "ippDetectID") 512 551 self.scratchDb.createIndex("StackApFlx", "ippDetectID") … … 531 570 imageID = self.scratchDb.getImageIDFromExternID(self.header['SOURCEID'], self.header['IMAGEID']) 532 571 533 self.logger. info("Updating table '" + table + "' with DVO IDs...")572 self.logger.debug("Updating table '" + table + "' with DVO IDs...") 534 573 sql = "UPDATE IGNORE " + table + " AS a, dvoDetectionFull AS b SET \ 535 574 a.ippObjID = b.ippObjID, \ … … 539 578 AND b.sourceID = " + self.header['SOURCEID'] + "\ 540 579 AND b.imageID = " + str(imageID) 541 self.scratchDb.stmt.execute(sql) 542 580 self.scratchDb.execute(sql) 543 581 544 582 ''' … … 553 591 self.populateStackMeta() 554 592 self.populateStackDetection() 555 self.populateStackModelFit() 556 self.populateStackApFlx() 593 594 if self.stackType != "NIGHTLY_STACK": 595 self.populateStackModelFit() 596 self.populateStackApFlx() 597 557 598 self.populateStackToImage() 558 599 self.populateSkinnyObject() … … 560 601 561 602 self.setMinMaxObjID(["StackDetection"]) 562 603 604 if self.totalDetections < 1: 605 606 self.logger.error("No detections to publish") 607 return False 608 563 609 return True 564 610 … … 568 614 def alreadyProcessed(self): 569 615 570 return self.ippToPspsDb.alreadyProcessed("stack", "stack_id", self.stackID) 616 # sadly, we have to read the FITS primary header first 617 if not self.readPrimaryHeader(): return False 618 619 # get filterID using init table 620 self.filter = self.header['FPA.FILTER'] 621 self.filter = self.filter[0:1] 622 623 self.stackType = stackType 624 meta = self.gpc1Db.getStackStageMeta(self.id, self.header['FPA.FILTER']) 625 if len(meta) < 1: return False 626 self.stackID = meta[0]; 627 self.skycell = meta[1]; 628 self.skycell = self.skycell[8:] 629 630 #return self.ippToPspsDb.alreadyProcessed("stack", "stack_id", self.stackID) 631 return False # TODOI 632 633 634 useFullTables=True 571 635 572 636 logging.config.fileConfig("logging.conf") 573 637 logger = logging.getLogger("stackbatch") 638 logger.setLevel(logging.INFO) 574 639 logger.info("Starting") 640 575 641 gpc1Db = Gpc1Db(logger) 576 stackType = "NIGHTLY_STACK" 577 skyIDs = gpc1Db.getIDsInThisDVODbForThisStage("MD04.Staticsky", "staticsky") 578 #skyIDs = gpc1Db.getIDsInThisDVODbForThisStage("MD04.GENE.PSPSDEEP", "staticsky") 579 #stackType = "DEEP_STACK" 580 #skyIDs = [689] 642 ippToPspsDb = IppToPspsDb(logger) 643 scratchDb = ScratchDb(logger, useFullTables) 644 645 #stackType = "NIGHTLY_STACK" 646 #skyIDs = gpc1Db.getIDsInThisDVODbForThisStageFudge() 647 #skyIDs = gpc1Db.getIDsInThisDVODbForThisStage("MD04.Staticsky", "staticsky") 648 649 stackType = "DEEP_STACK" 650 skyIDs = gpc1Db.getIDsInThisDVODbForThisStage("MD04.GENE.PSPSDEEP", "staticsky") 651 652 #skyIDs = [942] 581 653 #skyIDs = [299] 582 654 #skyIDs = [302] 583 655 #skyIDs = [8508] 584 i = 0656 #i = 0 585 657 for skyID in skyIDs: 586 587 logger.info("-------------------------------------------------- sky ID: %d" % skyID) 658 659 #if skyID < 1340: continue # nightly 660 #if skyID < 238: continue # deep 588 661 589 662 cmfFiles = gpc1Db.getStackStageCmfs(skyID) … … 591 664 for file in cmfFiles: 592 665 593 if not os.path.isfile(file): 594 logger.error("Cannot read file at '" + file) 595 continue 596 597 stackBatch = StackBatch(logger, skyID, file, stackType, True) 598 599 if not stackBatch.alreadyProcessed(): 600 601 stackBatch.createEmptyPspsTables() 602 stackBatch.importIppTables("") 603 if stackBatch.populatePspsTables(): 604 605 #stackBatch.reportNullsInAllPspsTables(False) 606 stackBatch.exportPspsTablesToFits() 607 stackBatch.writeBatchManifest() 608 #stackBatch.createTarball() 609 #stackBatch.publishToDatastore() 610 611 i = i + 1 612 #if i > 0: sys.exit() 613 614 logger.info("Finished") 666 stackBatch = StackBatch(logger, 667 gpc1Db, 668 ippToPspsDb, 669 scratchDb, 670 skyID, 671 file, 672 stackType, 673 useFullTables) 674 675 stackBatch.run() 676
Note:
See TracChangeset
for help on using the changeset viewer.
