Changeset 37801 for branches/eam_branches/ipp-20140904/ippToPsps/jython/diffbatch.py

Timestamp:

Jan 11, 2015, 1:31:50 PM (12 years ago)

Author:

eugene

Message:

merge changes from trunk

File:

• branches/eam_branches/ipp-20140904/ippToPsps/jython/diffbatch.py (modified) (1 diff)

branches/eam_branches/ipp-20140904/ippToPsps/jython/diffbatch.py

@@ +1 @@
+#!/usr/bin/env jython
+
+import os.path
+import sys
+import glob
+import time
+import stilts
+import re
+from java.lang import *
+from java.sql import *
+
+from java.lang import Math
+from org.apache.commons.math.special import Erf
+
+from xml.etree.ElementTree import ElementTree, Element, tostring
+
+from batch import Batch
+from gpc1db import Gpc1Db
+from ipptopspsdb import IppToPspsDb
+from scratchdb import ScratchDb
+from sqlUtility import sqlUtility
+
+import logging.config
+
+'''
+DiffBatch class
+
+This class, a sub-class of Batch, processes single exposures in the form of IPP smf files
+
+'''
+class DiffBatch(Batch):
+
+    '''
+    Constructor
+    '''
+    def __init__(self, 
+                 logger,
+                 config,
+                 skychunk,
+                 gpc1Db,
+                 ippToPspsDb,
+                 scratchDb,
+                 diffSkyFileID,
+                 batchID):
+
+       super(DiffBatch, self).__init__(
+               logger,
+               config,
+               skychunk,
+               gpc1Db,
+               ippToPspsDb,
+               scratchDb,
+               diffSkyFileID,
+               batchID,
+               "DF", 
+               None)
+    #           gpc1Db.getDiffStageCmf(skychunk.dvoLabel,diffSkyFileID))
+
+       # get diff meta data
+       # if error about catID you are usings an old ipptopsps_scratch db
+       self.logger.infoPair("getting diff" ," metadata")
+       meta = self.gpc1Db.getDiffStageMeta(diffSkyFileID)
+       self.logger.infoPair("ok, I guess that worked","now doing this")
+       self.diffSkyFileID = diffSkyFileID
+       if not meta:
+           self.logger.errorPair("Could not get", "diff metadata")
+           raise
+       self.logger.infoPair("so meta", "much amaze")
+       self.filterName = meta[0];
+       self.difftypeID = meta[1];
+       self.posImageID = meta[2];
+       self.negImageID = meta[3];
+       self.skycellName = meta[4];
+       self.analVer =  meta[5];
+       self.expTime = meta[6];
+       self.obsTime = meta[7];
+       self.fits = gpc1Db.getDiffStageCmf(skychunk.dvoLabel, diffSkyFileID)
+       self.header = self.fits.getPrimaryHeader()
+       tessName = self.getKeyValue(self.header, 'TESS_ID')
+       self.tessID = self.scratchDb.getTessID(tessName)
+
+
+       # XXX EAM 20140812 : I am hardwiring the parsing logic for RINGS vs LOCAL
+       if tessName == 'RINGS.V3':
+           # skycell is, eg "skycell.1133.081"
+           #                 0123456789012345
+           self.projectionID = self.skycellName[8:12]
+           self.skycellID = self.skycellName[13:]
+           self.skycellID = self.skycellName[8:] 
+       else:
+           # skycell is, eg "skycell.081"
+           #                 01234567890
+           self.projectionID = 0
+           self.skycellID = self.skycellName[8:11]
+           self.skycellID=self.skycellName[8:]
+       if self.skycellID == "":
+           self.skycellID = -1
+
+
+       # drop the existing tables    
+
+       self.dropTableVerbose("DiffMeta")
+       self.dropTableVerbose("DiffToImage")
+       self.dropTableVerbose("DiffDetection")
+
+       
+
+
+       self.logger.infoPair("I can haz",diffSkyFileID)
+
+
+
+       if not self.analVer: self.analVer = -999
+
+       # create an output filename, which is {expID}.FITS
+       self.outputFitsFile = "%08d.FITS" % self.diffSkyFileID
+       self.outputFitsPath = "%s/%s" % (self.localOutPath, self.outputFitsFile)
+       print "got here"
+       self.photoCalID = str(self.scratchDb.getPhotoCalID(self.diffSkyFileID))
+       print "got photocalid"
+       self.calibModNum = 0;
+       self.dataRelease = skychunk.dataRelease;
+        
+       self.filterID = self.scratchDb.getFilterID(self.filterName[0]) 
+       
+       print self.filterName
+       print self.filterName[0]
+       print self.filterID
+        # insert what we know about this stack batch into the stack table
+       self.ippToPspsDb.insertDiffMeta(self.batchID, self.filterName[0], self.difftypeID)
+
+       # dump stuff to log
+       self.logger.infoPair("diff_skyfile_ID", "%d" % self.diffSkyFileID)
+
+
+    '''
+    Populates the FrameMeta table, mainly from dictionary values found in IPP FITS header
+    '''
+    def populateDiffMeta(self):
+        self.logger.infoPair("Proccesing table", "DiffMeta")
+
+        sqlLine = sqlUtility("INSERT INTO DiffMeta (")
+
+        sqlLine.group("diffMetaID",            str(self.diffSkyFileID));
+        sqlLine.group("difftypeID",            self.difftypeID);
+        sqlLine.group("posImageID",            self.posImageID);
+        sqlLine.group("negImageID",            self.negImageID);
+        sqlLine.group("skycellID",             self.skycellID);
+        sqlLine.group("photoCalID",            self.photoCalID);
+        sqlLine.group("magSat",                self.getKeyFloat(self.header, "%.8f", 'FSATUR'));
+        sqlLine.group("analVer",               str(self.analVer));
+        sqlLine.group("expTime",               self.expTime);
+        sqlLine.group("ctype1",                self.getKeyValue(self.header, 'CTYPE1'));
+        sqlLine.group("ctype2",                self.getKeyValue(self.header, 'CTYPE2'));
+        sqlLine.group("crval1",                self.getKeyFloat(self.header, "%.8f", 'CRVAL1'));
+        sqlLine.group("crval2",                self.getKeyFloat(self.header, "%.8f", 'CRVAL2'));
+        sqlLine.group("crpix1",                self.getKeyFloat(self.header, "%.8f", 'CRPIX1'));
+        sqlLine.group("crpix2",                self.getKeyFloat(self.header, "%.8f", 'CRPIX2'));
+        sqlLine.group("cdelt1",                self.getKeyFloat(self.header, "%.8e", 'CDELT1'));
+        sqlLine.group("cdelt2",                self.getKeyFloat(self.header, "%.8e", 'CDELT2'));
+        sqlLine.group("pc001001",              self.getKeyFloat(self.header, "%.8e", 'PC001001'));
+        sqlLine.group("pc001002",              self.getKeyFloat(self.header, "%.8e", 'PC001002'));
+        sqlLine.group("pc002001",              self.getKeyFloat(self.header, "%.8e", 'PC002001'));
+        sqlLine.group("pc002002",              self.getKeyFloat(self.header, "%.8e", 'PC002002'));
+
+        sql = sqlLine.make(") VALUES ( ", ")")
+
+        try: self.scratchDb.execute(sql)
+        except:
+            self.logger.errorPair('failed sql: ', sql)
+            raise
+        self.logger.infoPair("DiffMeta","final stuff")
+        self.scratchDb.updateAllRows("DiffMeta", "batchID", str(self.batchID))
+        self.scratchDb.updateAllRows("DiffMeta", "surveyID", str(self.surveyID))
+        self.scratchDb.updateFilterID("DiffMeta", self.filterName[0])
+        self.scratchDb.updateAllRows("DiffMeta", "calibModNum", str(self.calibModNum))
+        self.scratchDb.updateAllRows("DiffMeta", "dataRelease", str(self.skychunk.dataRelease))
+        self.logger.infoPair("DiffMeta","done")
+
+    '''
+    Populates the Diff table for this OTA
+    '''
+    def populateDiffDetectionTable(self):
+
+        pspsTableName = "DiffDetection"
+        ippTableName =  "SkyChip_psf"
+       
+        #results['ORIGINALTOTAL'] = self.scratchDb.getRowCount(ippTableName)
+
+        # drop then re-create table
+        #self.scratchDb.dropTable(pspsTableName)
+        #sql = "CREATE TABLE " + pspsTableName + " LIKE DiffDetection"
+        #try: self.scratchDb.execute(sql)
+        #except: pass
+       
+        # XXX we no longer delete detections with funny mags (eg PSF_INST_MAG < -17.5)
+        # for examples of code which do this, go to r35097 or earlier
+
+        BEFORE = self.scratchDb.getRowCount(ippTableName)
+        #results['SATDET'] = 0 
+
+        extTimeString = str(self.expTime)
+
+        # insert all detections into table : do NOT include the IGNORE unless we have to
+        # sqlLine = sqlUtility("INSERT IGNORE INTO " + pspsTableName + " (")
+        sqlLine = sqlUtility("INSERT INTO " + pspsTableName + " (")
+
+        # XXX WARNING: need to apply platescale to convert sizes to arcsec
+        # XXX apFluxF or apFluxNpix + apFluxRadius?
+
+        sqlLine.group("ippDetectID",     "IPP_IDET")
+        sqlLine.group("diffDetMetaID",            str(self.diffSkyFileID));                                                
+        sqlLine.group("randomDiffID",     "FLOOR(RAND("+str(self.batchID)+")*9223372036854775807)")                       
+        sqlLine.group("filterID",        str(self.filterID))                                        
+        sqlLine.group("surveyID",        str(self.surveyID))         
+        sqlLine.group("skycellID",       str(self.skycellID))
+##        sqlLine.group("obsTime",         str(self.obsTime))     
+        sqlLine.group("expTime",         str(self.expTime))
+        sqlLine.group("obsTime", str(self.obsTime))
+        sqlLine.group("xPos",            "X_PSF")                                                   
+        sqlLine.group("yPos",            "Y_PSF")                                                   
+        sqlLine.group("xPosErr",         "X_PSF_SIG")                                               
+        sqlLine.group("yPosErr",         "Y_PSF_SIG")                                               
+        if (self.id >= 982483):
+            sqlLine.group("pltScale",        "PLTSCALE")                                      
+            sqlLine.group("posAngle",        "POSANGLE")                                      
+#            sqlLine.group("raErr",           "X_PSF_SIG * abs(PLTSCALE)") 
+#            sqlLine.group("decErr",          "Y_PSF_SIG * abs(PLTSCALE)") 
+        else:
+            sqlLine.group("pltScale",        "0.257")                                      
+            sqlLine.group("posAngle",        "-999")                                      
+#            sqlLine.group("raErr",           "X_PSF_SIG * 0.257") 
+#            sqlLine.group("decErr",          "Y_PSF_SIG * 0.257") 
+        sqlLine.group("psfFlux",         "PSF_INST_FLUX / " + extTimeString)
+        sqlLine.group("psfFluxErr",      "PSF_INST_FLUX_SIG / " + extTimeString)
+#        if (self.id >= 982483):   #we don't have PSF_FWHM_MAJ or MIN or PSF_CORE
+#            sqlLine.group("psfMajorFWHM",    "PSF_FWHM_MAJ")                                               
+#            sqlLine.group("psfMinorFWHM",    "PSF_FWHM_MIN")                                               
+#            sqlLine.group("psfCore",         "PSF_CORE")                                                    
+        sqlLine.group("psfTheta",        "PSF_THETA")                                               
+        sqlLine.group("psfQf",           "PSF_QF")                                                  
+        sqlLine.group("psfQfPerfect",    "PSF_QF_PERFECT")                                
+        sqlLine.group("psfChiSq",        "PSF_CHISQ")                                     
+        sqlLine.group("psfLikelihood",   "psfLikelihood(EXT_NSIGMA)")                               
+        sqlLine.group("momentXX",        "MOMENTS_XX")                                              
+        sqlLine.group("momentXY",        "MOMENTS_XY")                                              
+        sqlLine.group("momentYY",        "MOMENTS_YY")                                              
+        sqlLine.group("momentR1",        "MOMENTS_R1")                                              
+        sqlLine.group("momentRH",        "MOMENTS_RH")                                              
+#        sqlLine.group("momentM3C",       "MOMENTS_M3C")                                             
+#        sqlLine.group("momentM3S",       "MOMENTS_M3S")                                             
+#        sqlLine.group("momentM4C",       "MOMENTS_M4C")                                             
+#        sqlLine.group("momentM4S",       "MOMENTS_M4S")                                             
+        if (self.id >= 982483):
+            sqlLine.group("apFlux",          "AP_FLUX / " + extTimeString)
+            sqlLine.group("apFluxErr",       "AP_FLUX_SIG / " + extTimeString)
+ #           sqlLine.group("apFillF",         "AP_NPIX / (3.14159265359 * POW(AP_MAG_RADIUS - 0.5, 2))")
+ #       sqlLine.group("apRadius",        "AP_MAG_RADIUS")
+        sqlLine.group("kronFlux",        "KRON_FLUX / " + extTimeString)
+        sqlLine.group("kronFluxErr",     "KRON_FLUX_ERR / " + extTimeString)
+        sqlLine.group("kronRad",         "MOMENTS_R1 * 2.5")
+        sqlLine.group("diffNPos",         "DIFF_NPOS")
+        sqlLine.group("diffFRatio",         "DIFF_FRATIO")
+        sqlLine.group("diffNBad",         "DIFF_NRATIO_BAD")
+        sqlLine.group("diffNMask",         "DIFF_NRATIO_MASK")
+        sqlLine.group("diffNAll",         "DIFF_NRATIO_ALL")
+        sqlLine.group("diffPdist",         "DIFF_R_P")
+        sqlLine.group("diffNdist",         "DIFF_R_M")
+        sqlLine.group("diffPSN",         "DIFF_SN_P")
+        sqlLine.group("diffNSN",         "DIFF_SN_M")
+        sqlLine.group("sky",             "SKY  / " + extTimeString)
+        sqlLine.group("skyErr",          "SKY_SIGMA  / " + extTimeString)
+        sqlLine.group("infoFlag",        "FLAGS")                          
+        sqlLine.group("infoFlag2",       "FLAGS2")                          
+        sqlLine.group("dataRelease",     str(self.skychunk.dataRelease))                  
+
+        sql = sqlLine.makeRaw(") SELECT ", " FROM " + ippTableName)
+
+        try: self.scratchDb.execute(sql)
+        except:
+            self.logger.errorPair("failed sql: ", sql)
+            raise
+
+        # NOTE : Flux limits : in the current PSPS schema, negative fluxes 
+        # cause problems for sql queries which work in mags as SQL cannot do
+        # something like (f < 0.0) ? -999 : -2.5*log10(f)
+        # as a result, the negative fluxes here result in floating point errors
+        # XXX EAM 2014072 : Is this still a problem?
+ #       sql = "UPDATE " + pspsTableName + " SET psfFlux = 1e20 WHERE psfFlux <= 0.0"
+ #       self.scratchDb.execute(sql)
+        
+ #       sql = "UPDATE " + pspsTableName + " SET apFlux = 1e20 WHERE apFlux <= 0.0"
+ #       self.scratchDb.execute(sql)
+
+ #       sql = "UPDATE " + pspsTableName + " SET kronFlux = 1e20 WHERE kronFlux <= 0.0"
+ #       self.scratchDb.execute(sql)
+
+        # we don't delete these anymore
+        #results['NULLINSTFLUX'] = 0;
+
+
+# insert stuff from dvo
+        self.logger.infoPair("inserting dvo info:find imageID"," from externID")
+        imageID = self.scratchDb.getImageIDFromExternID(self.diffSkyFileID)
+        self.logger.infoPair("updating","diffdetection")
+        sqlLine  = sqlUtility("UPDATE DiffDetection as a, " + self.scratchDb.dvoDetectionTable + " as b SET ")
+        sqlLine.group("a.diffObjID", "b.objId")
+        sqlLine.group("a.diffDetID", "b.detectID")
+        sqlLine.group("a.ippObjID","b.ippObjID")
+        sqlLine.group("a.dvoRegionID",             "b.catID")
+        sqlLine.group("a.ra", "b.ra")
+        sqlLine.group("a.dec", "b.dec_")
+        sqlLine.group("a.raErr", "b.raErr")
+        sqlLine.group("a.decErr", "b.decErr")
+        
+        sqlLine.group("a.zp","b.zp")
+        sqlLine.group("a.telluricExt", "b.telluricExt")
+        sqlLine.group("a.airmass","b.airmass")
+
+        
+        sql = sqlLine.makeEquals("WHERE a.ippDetectID = b.ippDetectID AND b.imageID = " +str( imageID))   
+        
+        try: self.scratchDb.execute(sql)
+        except:
+            self.logger.errorPair("failed sql:",sql)
+            raise
+        self.logger.infoPair("need to cull"," nulls objid")
+        results= self.scratchDb.reportAndDeleteRowsWithNULLS("DiffDetection", "diffObjID")
+        self.logger.infoPair("adding","uniquePSPSDFid")
+        self.updatePspsUniqueIDs("DiffDetection")
+
+    '''
+    Populates the DiffToImage table
+    '''
+    def populateDiffToImage(self):
+
+        self.logger.infoPair("Procesing table", "DiffToImage")
+
+        imageIDs = self.gpc1Db.getImageIDsForThisDiffID(self.diffSkyFileID)
+
+        for imageID in imageIDs:
+
+            sql = "INSERT INTO DiffToImage (diffMetaID, imageID) \
+            VALUES (\
+            " + str(self.diffSkyFileID) + ", " + imageID + ")"
+            self.scratchDb.execute(sql)
+
+            # now update StackMeta with correct number of inputs
+            #sql = "UPDATE "+filter+"StackMeta SET nP2Images = (SELECT COUNT(distinct(floor(imageID/100))) FROM StackToImage where stackMetaID = "+str(stackmetaID)+")"
+            #self.scratchDb.execute(sql)
+
+        #self.tablesToExport.append("DiffToImage")
+
+
+
+
+
+
+
+    '''
+    Applies indexes and other constraints to the PSPS tables
+    '''
+    def alterPspsTables(self):
+
+        self.logger.debug("Altering PSPS tables")
+        self.logger.debug("Creating indexes on PSPS tables")
+        self.scratchDb.makeColumnUnique("DiffDetection", "objID")
+        self.scratchDb.makeColumnPrimaryKey("DiffDetection", "ippDetectID")
+
+        #self.populateDiffMeta()
+     
+        # dictionary objects to hold imageIDs for later
+        self.imageIDs = {}
+        self.sourceIDs = {}
+
+        # loop through all OTAs and populate ImageMeta extensions
+        # the column in PSPS
+        self.scratchDb.execute("ALTER TABLE DiffDetection CHANGE dec_ `dec` double")
+
+        return True
+
+
+    '''
+    Applies indexes to the IPP tables
+    '''
+    def indexIppTables(self):
+
+        self.logger.infoPair("Creating indexes on", "IPP tables")
+
+        #self.scratchDb.createIndex(extension, "IPP_IDET")
+        # try the test Chip
+        self.scratchDb.createIndex("SkyChip_psf", "IPP_IDET")
+
+        self.logger.infoPair("created indexes on", "IPP tables")      
+
+    '''
+    Updates provided table with DVO IDs from DVO table
+    '''
+    def updateDvoIDs(self, table, externID):
+
+        imageID = self.scratchDb.getImageIDFromExternID(externID)
+        self.logger.debug("Updating table '" + table + "' with DVO IDs using imageID = %d" % imageID)
+
+        sql = "UPDATE IGNORE " + table + " AS a, " + self.scratchDb.dvoDetectionTable + " AS b SET \
+               a.objID        = b.objID, \
+               a.detectID     = b.detectID, \
+               a.ippObjID     = b.ippObjID, \
+               a.dvoRegionID  = b.catID, \
+               a.ra           = b.ra, \
+               a.dec          = b.dec_, \
+               a.zp           = b.zp, \
+               a.telluricExt  = b.telluricExt, \
+               a.airmass      = b.airmass, \
+               a.expTime      = b.expTime, \
+               a.infoFlag2    = (b.flags << 13) | a.infoFlag2 \
+               WHERE a.ippDetectID = b.ippDetectID \
+               AND b.imageID = " + str(imageID)
+
+        self.scratchDb.execute(sql)
+
+    '''
+    Updates table and generates pspsuniqueids
+    '''
+
+    def updatePspsUniqueIDs(self,table):
+        sql = "UPDATE "+table+" join (select @r:=@r+1 rownum, diffobjID from \
+        (select @r:=0) r, "+table+" t) as foo using (diffobjID) set \
+        uniquePspsDFid = ((" +str(self.batchID)+ "*1000000000 ) + rownum)"
+        try: self.scratchDb.execute(sql)
+        except:
+            self.logger.errorPair('failed sql',sql)
+            raise
+        
+    '''
+    Does the processing, i.e. pulling stuff from IPP tables into PSPS tables
+    '''
+
+    '''
+    Does the processing, i.e. pulling stuff from IPP tables into PSPS tables
+    '''
+    def populatePspsTables(self):
+        self.logger.infoPair("starting","populatePspsTables for diffBatch")
+        # loop through all OTAs again to update with DVO IDs
+        self.skipBatch = False
+        self.tablesToExport = []    
+        self.tablesToExport.append("DiffMeta")
+        self.populateDiffMeta()
+        self.logger.infoPair("populated","DiffMeta")
+        self.tablesToExport.append("DiffToImage")
+        self.populateDiffToImage()
+        self.logger.infoPair("populated","DiffToImage")
+        self.tablesToExport.append("DiffDetection")
+        self.populateDiffDetectionTable()
+        self.logger.infoPair("populated","DiffDetection")
+
+
+
+        
+        self.setMinMaxDiffObjID(["DiffDetection"])
+
+
+        #test = self.scratchDb.getRowCount("Object")
+        #return False
+        
+        return True
+
+
+    '''
+    Imports IPP tables from FITS file
+    Accepts a regular expression to match the tables so not all tables need to be imported
+    (This regex feature is not currently used...)
+    '''
+    def importIppTables(self, columns="*", tableRE=""):
+        
+        if self.config.retry: return True
+
+        self.logger.infoPair("Importing DF tables with table match expression: ", tableRE)
+
+        count = 0
+        fileName = self.fits.getPath()
+            
+        self.logger.infoPair("using filename:",fileName)
+        
+        try:
+            tables = stilts.treads(fileName)
+        except:
+            self.logger.errorPair("STILTS could not import from", fileName)
+            return False
+        
+            # count = 0
+        for table in tables:
+              
+            match = re.match(tableRE, table.name)
+            if not match: continue
+
+            self.logger.infoPair("Reading IPP table", table.name)
+            table = stilts.tpipe(table, cmd='addcol table_index $0')
+            table = stilts.tpipe(table, cmd='explodeall')
+                  
+            # drop any previous tables before import
+            self.scratchDb.dropTable(table.name)
+                  
+                  
+            # IPP FITS files are littered with infinities, so remove these
+            self.logger.info("Removing Infinity values from all columns")
+            table = stilts.tpipe(table, cmd='keepcols "' + columns + '"')
+            table = stilts.tpipe(table, cmd='replaceval -Infinity null *')
+            table = stilts.tpipe(table, cmd='replaceval Infinity null *')
+                  
+            try:
+                table.write(self.scratchDb.url + '#' +  table.name)
+                count = count + 1
+            except:
+                self.logger.exception("Problem writing table '" +  table.name + "' to the database")
+                      
+        self.logger.infoPair("Done. Imported", "%d tables" % count)
+        self.indexIppTables()
+                      
+        return True
+
+
+
+
+
+
+    '''
+    Overriding this method. Use regex to trim off, eg _XY33 extension
+    '''
+    def exportPspsTablesToFits(self, regex="(.*)"):
+       return super(DiffBatch, self).exportPspsTablesToFits("([a-zA-Z]+)")
+
+
+    '''
+    Drop a table and report hte drop
+    '''
+    def dropTableVerbose(self, table):
+        self.logger.infoPair("dropping table:",table)
+        self.scratchDb.dropTable(table)
+