Changeset 37801 for branches/eam_branches/ipp-20140904/ippToPsps/jython/diffobjectbatch.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/diffobjectbatch.py (modified) (1 diff)

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

@@ +1 @@
+#!/usr/bin/env jython
+
+import os.path
+import sys
+
+import stilts
+from java.lang import *
+from java.sql import *
+
+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 dvoobjects import DvoObjects
+from sqlUtility import sqlUtility
+
+import logging.config
+
+'''
+DiffObjectBatch class
+
+This class, a sub-class of Batch, processes individual DVO region file pairs (cpt and cps files). The cpt file contains a list of all objects in that region, the cps file contains the magnitudes and so has a row per filter per object (so, generally, has rows = 5 x number of objects). The Photcode.dat table us used to figure out the number and order of filters as they appear in the cps file.
+
+'''
+class DiffObjectBatch(Batch):
+
+    '''
+    Constructor
+    '''
+    def __init__(self, 
+                 logger, 
+                 config,
+                 skychunk,
+                 gpc1Db,
+                 ippToPspsDb,
+                 scratchDb,
+                 dvoID,
+                 batchID):
+
+       super(DiffObjectBatch, self).__init__(
+               logger,
+               config,
+               skychunk,
+               gpc1Db,
+               ippToPspsDb,
+               scratchDb,
+               dvoID,
+               batchID,
+               "DO", 
+               None)
+
+       try:
+           self.dvoObjects = DvoObjects(self.logger, self.config, self.skychunk, self.ippToPspsDb, self.scratchDb)
+       except:
+           self.logger.errorPair("Unable to create instance of", "DvoObjects")
+           raise
+
+       # create an output filename, which is {dvoINDEX}.FITS
+       self.outputFitsFile = "%08d.FITS" % self.id
+       self.outputFitsPath = "%s/%s" % (self.localOutPath, self.outputFitsFile)
+
+       # dump stuff to log
+       self.logger.infoPair("DVO INDEX", "%d" % self.id)
+
+    '''
+    Overriden from batch base-class. We import from DVO directly for objects
+    '''
+    def importIppTables(self, columns="*", filter=""):
+
+        self.region = self.scratchDb.getRegionNameFromThisDvoIndex(self.id)
+        self.ippToPspsDb.insertObjectMeta(self.batchID, self.region)
+        self.dvoObjects.nativeIngestRegion(self.region)
+
+        cptTableName = self.scratchDb.getDbFriendlyTableName(self.region + ".cpt")
+        if not self.scratchDb.tableExists(cptTableName):
+            return False
+
+        return True
+
+    '''
+    Applies indexes to the PSPS tables
+    '''
+    def alterPspsTables(self):
+
+        return True
+
+    '''
+    Applies indexes to the IPP tables
+    '''
+    def indexIppTables(self):
+
+        # since dvopsps is now used, no action is needed here
+        # self.logger.infoPair("Creating indexes on", "IPP tables")
+
+        return True
+
+    '''
+    Inserts stuff for all mags
+    '''
+    def updateMeanObjectFromCps(self, cpsTable):
+
+        # list of all filters PSPS is interested in
+        # XXX EAM : 2014.07.24 : this list should probably be in a config file somewhere
+        interestedFilters = ['g', 'r', 'i', 'z', 'y']
+        
+        filters = self.scratchDb.getOrderedListOfFiltersFromPhotcodesTable(interestedFilters)
+    
+        # get a count of the available filters
+        filterCount = self.scratchDb.getCountOfFiltersFromPhotcodesTable()
+
+        self.logger.infoPair("Available filters in Photcodes", filters)
+
+        # the 'code' now defines the order in the cps file that the mags are listed for a given filter
+        self.logger.infoPair("Adding magnitudes from", "cps table")
+        for filter in filters:
+
+            filterID = self.scratchDb.getFilterID(filter[1])
+
+            # NOTE: Manipulation of FLAGS from cpsTable is to move ID_SECF_OBJ_EXT flag (0x01000000)
+            # from bit 24 to bit 13 so that it fits into the SMALLINT Object.Flags
+            # XXX EAM : 20140724 this manipulation is no longer needed : [grizy]Flags is now 4 byte (was 8 byte!)
+
+            # set the MeanObject fields based largely on dvopsps cps fields:
+
+            # XXX EAM 20140724 : filterCount is meant to match
+            # Nsecfilt, but is potentially not determined correctly.
+            # use a call to a dvo-native command which knows how to
+            # find Nsecfilt (or save in the db with dvopsps)
+
+            # the math below depends on filterCount = Nsecfilt and MeanObject.row being 1 counting but cps being 0 counting?
+            # cps.row has a count of MeanObject.row * Nsecfilt + Nfilter 
+            #  " + cpsTable + " AS cps ON (cps.row = (MeanObject.row* " + str(filterCount) + ")-(" + str(filterCount) + " - " + str(filter[0]) + ")) \
+
+            sql = "UPDATE DiffDetObject JOIN \
+                   " + cpsTable + " AS cps ON (cps.row = (DiffDetObject.row* " + str(filterCount) + ")-(" + str(filterCount) + " - " + str(filter[0]) + ")) \
+                   SET \
+                    DiffDetObject." + filter[1] + "QfPerfect      = PSF_QF_PERF_MAX "
+
+            try: self.scratchDb.execute(sql)
+            except:
+                self.logger.errorPair("failed update DiffDetObject", sql)
+                raise
+
+    '''
+    Inserts stuff for all mags
+    '''
+    def updateDiffObjectFromCps(self, cpsTable):
+
+        # list of all filters PSPS is interested in
+        interestedFilters = ['g', 'r', 'i', 'z', 'y']
+        
+        filters = self.scratchDb.getOrderedListOfFiltersFromPhotcodesTable(interestedFilters)
+    
+        # get a count of the available filters
+        filterCount = self.scratchDb.getCountOfFiltersFromPhotcodesTable()
+        # filterCount = len(filters)
+
+        self.logger.infoPair("Available filters in Photcodes", filters)
+
+        # the 'code' now defines the order in the cps file that the mags are listed for a given filter
+        self.logger.infoPair("Adding magnitudes from", "cps table")
+        for filter in filters:
+
+            filterID = self.scratchDb.getFilterID(filter[1])
+
+            # XXX EAM 20140724 : this is quite awkward, add a objRow and ncode value to mysql db table?
+            # sqlLine = sqlUtility()
+
+            sql  = "UPDATE DiffDetObject JOIN "
+            sql += cpsTable + " AS cps "
+            sql += "ON (cps.row = (DiffDetObject.row * "
+            sql += str(filterCount) + ")-(" 
+            sql += str(filterCount) + " - " 
+            sql += str(filter[0])
+            sql += ")) "
+
+            sql += "SET DiffDetObject.n" + filter[1] + " = NCODE, "
+            sql += "DiffDetObject.nDetections = DiffDetObject.nDetections + cps.NSTACK_DET"
+            self.logger.info(sql)
+            self.scratchDb.execute(sql)
+
+        # XXX this does not seem like a good thing to leave in SQL
+        self.logger.infoPair("Calculating nDetections from", "n[filters]")
+        for filter in filters:
+            # now do a sum of n[filters], but do not include the ones with -999
+            sql  = "UPDATE DiffDetObject "
+            sql += "SET nDetections = nDetections + n" + filter[1]
+            sql += " WHERE n" + filter[1] + " != -999"
+            self.scratchDb.execute(sql)
+
+    '''
+    give objectName to objectThin
+    XXX EAM 20140714 : This seems quite inefficient in SQL, move to dvopsps?
+    '''
+    def updateDiffObjName(self):
+    ##this is a 2 part update:
+
+        self.logger.infoPair("updating diffObjName", "using means")
+    ## use mean ra and dec (dec >= 0)
+        sql = "update DiffDetObject set diffObjName =  concat('PS1.2 J', \
+        lpad(floor(ra/15.),2,'0'), \
+        lpad((ra/15.-(floor(ra/15.)))*60. ,2,'0'), \
+        lpad(format(((ra/15.-(floor(ra/15.)))*60.-floor((ra/15.-(floor(ra/15.)))*60.))*60.,2),4,'0'), \
+        '+', \
+        lpad(floor(dec_),2,'0'), \
+        lpad((dec_-(floor(dec_)))*60. ,2,'0'), \
+        lpad(format(((dec_-(floor(dec_)))*60.-floor((dec_-(floor(dec_)))*60.))*60.,2),4,'0') \
+        ) where ra > -999 and dec_ > -999 and dec_ >= 0"
+        try:
+            self.scratchDb.execute(sql)
+        except:
+            self.logger.errorPair("failed sql", sql)
+            return False
+ 
+    ## use mean ra and dec (dec < 0)
+
+        sql = "update DiffDetObject set diffObjName =  concat('PS1.2 J', \
+        lpad(floor(ra/15.),2,'0'), \
+        lpad((ra/15.-(floor(ra/15.)))*60. ,2,'0'), \
+        lpad(format(((ra/15.-(floor(ra/15.)))*60.-floor((ra/15.-(floor(ra/15.)))*60.))*60.,2),4,'0'), \
+        '-', \
+        lpad(floor(dec_),2,'0'), \
+        lpad((dec_-(floor(dec_)))*60. ,2,'0'), \
+        lpad(format(((dec_-(floor(dec_)))*60.-floor((dec_-(floor(dec_)))*60.))*60.,2),4,'0') \
+        ) where ra > -999 and dec_ > -999 and dec_ < 0"
+
+        try:
+            self.scratchDb.execute(sql)
+        except:
+            self.logger.errorPair("failed sql", sql)
+            return False
+
+    '''
+    Populates the DiffObject table
+    '''
+    def populateDiffObjectTable(self):
+
+        cptTableName = self.scratchDb.getDbFriendlyTableName(self.region + ".cpt")
+        cpsTableName = self.scratchDb.getDbFriendlyTableName(self.region + ".cps")
+
+        if False:
+            # XXX EAM 20140724 : this is probably wrong : flux measurements can be 0.0 or negative: please review
+            self.logger.infoPair("setting to null  > 1e-38 and < 1e-38 in", "cps FLUX_KRON_ERR")
+            sql = "UPDATE " + cpsTableName + " set FLUX_KRON_ERR = NULL where FLUX_KRON_ERR < 1e-37 AND FLUX_KRON_ERR > -1e-37 "
+            self.exitProgram("review this code" + sql)
+      
+            try:
+                self.scratchDb.execute(sql)
+            except:
+                self.logger.errorPair("Couldn't squash out of range stuff", sql)
+                return False
+
+            self.logger.infoPair("setting to null  > 1e-38 and < 1e-38 in", "cps FLUX_KRON")
+            sql = "UPDATE " + cpsTableName + " set FLUX_KRON = NULL where FLUX_KRON < 1e-37 AND FLUX_KRON > -1e-37 "
+      
+            try:
+                self.scratchDb.execute(sql)
+            except:
+                self.logger.errorPair("Couldn't squash out of range stuff", sql)
+                return False
+
+        
+        self.logger.info("Populating DiffDetObject")
+        self.logger.info("Inserting objects from cpt file")
+
+        # note "dec" is a reserved word in MySQL
+        # XXX EAM 20140724 : do not use INGORE unless we discover unavoidable problems...
+        # INSERT IGNORE INTO ObjectThin
+
+        sqlLine = sqlUtility("INSERT INTO DiffDetObject (")
+
+        sqlLine.group("diffObjID",           "EXT_ID") 
+        sqlLine.group("ippObjID",        "OBJ_ID + (CAT_ID << 32)") # NOTE: shift by 32 bits exactly
+        sqlLine.group("surveyID",        "'" + str(self.surveyID) + "'")
+        sqlLine.group("randomID",        "FLOOR(RAND("+str(self.batchID)+")*9223372036854775807)") # XXX where does this number come from??
+        sqlLine.group("batchID",         "'" + str(self.batchID) + "'")
+        sqlLine.group("dvoRegionID",     "CAT_ID") 
+        sqlLine.group("skycellID",       "SKYCELL_ID") 
+        sqlLine.group("dataRelease",     "'" + str(self.skychunk.dataRelease) + "'")
+        sqlLine.group("objInfoFlag",     "FLAGS") 
+        sqlLine.group("qualityFlag",     "FLAGS >> 24 & 0xFF")
+        sqlLine.group("consistencyFlag", "'0'") 
+        sqlLine.group("ra",          "RA_MEAN")
+        sqlLine.group("dec_",         "DEC_MEAN")
+        sqlLine.group("raErr",       "RA_ERR")
+        sqlLine.group("decErr",      "DEC_ERR")
+        sqlLine.group("nDetections",      "'0'")
+        sql = sqlLine.makeRaw(") SELECT ", " FROM " + cptTableName)
+
+        try:
+            self.scratchDb.execute(sql)
+        except:
+            self.logger.errorPair("Couldn't populate DiffDetObject table", sql)
+            return False
+ 
+        # add row count columns so we can perform joins to get colors
+        self.logger.infoPair("Adding 'row' columns to", "DiffDetObject and cps tables")
+        self.scratchDb.addRowCountColumn("DiffDetObject", "row")
+        self.scratchDb.addRowCountColumn(cpsTableName, "row")
+
+##        self.insertMeanMags(cpsTableName)
+## this is the new version
+        self.logger.infoPair("update diffObjName for ","DiffDetObject")
+        self.updateDiffObjName()
+
+        self.logger.infoPair("update DiffDetObject from ","cps table")
+
+        self.updateDiffObjectFromCps(cpsTableName)
+
+        # XXX EAM 20140724 : is this necessary??
+        #objects can have out of range ra dec in dvo - need to find and kill them at the end
+
+        self.logger.infoPair("Determining", "ra/dec range")
+
+        raMin = self.scratchDb.getFromdvoSkyTable("R_MIN",self.region)
+        raMax = self.scratchDb.getFromdvoSkyTable("R_MAX",self.region)
+        decMin = self.scratchDb.getFromdvoSkyTable("D_MIN",self.region)
+        decMax = self.scratchDb.getFromdvoSkyTable("D_MAX",self.region)
+
+        self.logger.infoPair("R_MIN", raMin)
+        self.logger.infoPair("R_MAX", raMax)
+        self.logger.infoPair("D_MIN", decMin)
+        self.logger.infoPair("D_MAX", decMax)
+        #count out of range
+
+        sql = "SELECT count(*) FROM DiffDetObject where \
+              DiffDetObject.dec_ > (" + str(decMax) + " + .0033) \
+              or DiffDetObject.dec_ < (" + str(decMin) + " - .0033) \
+              or DiffDetObject.ra > (" + str(raMax) + " + .0033) \
+              or DiffDetObject.ra < (" + str(raMin) + " - .0033)"       
+    
+        rs = self.scratchDb.executeQuery(sql)
+        rs.first()
+        nToDelete = rs.getInt(1)
+        
+        #delete out of range
+        
+ 
+        sql = "DELETE FROM DiffDetObject where \
+              DiffDetObject.dec_ > (" + str(decMax) + " + .0033) or \
+              DiffDetObject.dec_ < (" + str(decMin) + " - .0033) or \
+              DiffDetObject.ra > (" + str(raMax) + " + .0033) or \
+              DiffDetObject.ra < (" + str(raMin) + " - .0033)" 
+        self.logger.infoPair("Deleting", str(nToDelete) + " objects outside of ra/dec range")
+
+        try:
+            self.scratchDb.execute(sql)
+        except:
+            self.logger.errorPair("Couldn't cull outsiders from DiffDetObject table", sql)
+            return False
+
+        self.logger.infoPair("Dropping row column from", "DiffDetObject table")
+        self.scratchDb.dropColumn("DiffDetObject", "row")
+        ##self.logger.infoPair("Purging from scratch Db", self.region + " region")
+
+        ##Don't do this till after MeanObject
+        ##self.dvoObjects.purgeRegion(self.region)
+
+        self.setMinMaxDiffObjID(["DiffDetObject"])
+
+        return True
+
+    '''
+    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 \
+        uniquePspsDOid = ((" +str(self.batchID)+ "*1000000000 ) + rownum)"
+        try: self.scratchDb.execute(sql)
+        except:
+            self.logger.errorPair('failed sql',sql)
+            return
+
+
+
+    '''
+    Does the processing, i.e. pulling stuff from IPP tables into PSPS tables
+    '''
+    def populatePspsTables(self):
+
+        if not self.populateDiffObjectTable(): return False
+
+        # now remove the objID duplicates. We could not do this before as cpt/cps tables relate by row number
+        self.logger.infoPair("Forcing uniqueness on", "objID in DIffObject table")
+        rowCountBefore = self.scratchDb.getRowCount("DiffDetObject")
+
+        # XXX EAM : note that in mysql versions later than 5.1, this fails
+        # unless the following is called first: 
+        # set session old_alter_table=1
+        # follow the command with 
+        # set session old_alter_table=0
+        # OF COURSE, this fails for mysql version < 5.5...
+        if self.scratchDb.version > 5.1:
+            self.scratchDb.execute("set session old_alter_table=1")
+            
+        self.scratchDb.execute("ALTER IGNORE TABLE DiffDetObject ADD UNIQUE INDEX(diffObjID)")
+        if self.scratchDb.version > 5.1:
+            self.scratchDb.execute("set session old_alter_table=0")
+
+        rowCountAfter = self.scratchDb.getRowCount("DiffDetObject")
+        self.logger.infoPair("Number of duplicated diffobjIDs removed", "%d out of %d" % ((rowCountBefore - rowCountAfter), rowCountBefore))
+
+        self.scratchDb.execute("ALTER TABLE DiffDetObject CHANGE dec_ `dec` double")
+
+        self.updatePspsUniqueIDs("DiffDetObject")
+
+        self.dvoObjects.purgeRegion(self.region)
+
+        #this is abuse of something but this is how I get the object batches to crash to further investigate them
+        
+ 
+#        rowCountAfter = self.scratchDb.getRowCount("Object")
+        return True
+#        return False