Changeset 37551 for trunk/ippToPsps/jython/forcedobjectbatch.py

Timestamp:

Nov 3, 2014, 2:41:20 PM (12 years ago)

Author:

heather

Message:

added diff / ff stages to addstar, partially to ipptopsps

File:

• trunk/ippToPsps/jython/forcedobjectbatch.py (modified) (1 diff)

trunk/ippToPsps/jython/forcedobjectbatch.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 dvoforcedobjects import DvoForcedObjects # need one specific for cpy ?
+from sqlUtility import sqlUtility
+
+import logging.config
+
+'''
+ForcedObjectBatch class
+
+This class, a sub-class of Batch, processes individual DVO region file pairs (cpt, cpy, 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 ForcedMeanObjectBatch(Batch):
+
+    '''
+    Constructor
+    '''
+    def __init__(self, 
+                 logger, 
+                 config,
+                 skychunk,
+                 gpc1Db,
+                 ippToPspsDb,
+                 scratchDb,
+                 dvoID,
+                 batchID):
+
+       super(ForcedMeanObjectBatch, self).__init__(
+               logger,
+               config,
+               skychunk,
+               gpc1Db,
+               ippToPspsDb,
+               scratchDb,
+               dvoID,
+               batchID,
+               "FO", 
+               None)
+
+       try:
+           self.dvoForcedObjects = DvoForcedObjects(self.logger, self.config, self.skychunk, self.ippToPspsDb, self.scratchDb)
+       except:
+           self.logger.errorPair("Unable to create instance of", "DvoForcedObjects")
+           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 forced mean objects
+    '''
+    def importIppTables(self, columns="*", filter=""):
+
+        self.region = self.scratchDb.getRegionNameFromThisDvoIndex(self.id)
+        self.ippToPspsDb.insertForcedMeanObjectMeta(self.batchID, self.region)
+        self.dvoForcedObjects.nativeIngestRegion(self.region)
+
+        cptTableName = self.scratchDb.getDbFriendlyTableName(self.region + ".cpt")
+
+        if not self.scratchDb.tableExists(cptTableName):
+            self.logger.infoPair("can't find in scratch db:","cpt table")
+            return False
+
+        cpyTableName = self.scratchDb.getDbFriendlyTableName(self.region + ".cpy")
+
+        if not self.scratchDb.tableExists(cpyTableName):
+            self.logger.infoPair("can't find in scratch db:","cpy table")
+            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 updateForcedMeanObjectFromCps(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 ForcedMeanObject JOIN \
+                   " + cpsTable + " AS cps ON (cps.row = (ForcedMeanObject.row* " + str(filterCount) + ")-(" + str(filterCount) + " - " + str(filter[0]) + ")) \
+                   SET \
+                    ForcedMeanObject." + filter[1] + "nTotal         = NWARP \
+                   ,ForcedMeanObject." + filter[1] + "nIncPSFMag     = NUSED_WRP \
+                   ,ForcedMeanObject." + filter[1] + "nIncKronMag    = NUSED_KRON_WRP \
+                   ,ForcedMeanObject." + filter[1] + "nIncApMag      = NUSED_AP_WRP \
+                   ,ForcedMeanObject." + filter[1] + "PSFMag         = MAG_PSF_WRP \
+                   ,ForcedMeanObject." + filter[1] + "PSFMagErr      = MAG_PSF_WRP_ERR \
+                   ,ForcedMeanObject." + filter[1] + "PSFMagStd      = MAG_PSF_WRP_STD \
+                   ,ForcedMeanObject." + filter[1] + "KronMag        = MAG_KRON_WRP \
+                   ,ForcedMeanObject." + filter[1] + "KronMagErr     = MAG_KRON_WRP_ERR \
+                   ,ForcedMeanObject." + filter[1] + "KronMagStd     = MAG_KRON_WRP_STD \
+                   ,ForcedMeanObject." + filter[1] + "ApMag          = MAG_AP_WRP \
+                   ,ForcedMeanObject." + filter[1] + "ApMagErr       = MAG_AP_WRP_ERR \
+                   ,ForcedMeanObject." + filter[1] + "ApMagStd       = MAG_AP_WRP_STD \
+                   ,ForcedMeanObject." + filter[1] + "Flags          = FLAGS "
+
+
+            try: self.scratchDb.execute(sql)
+            except:
+                self.logger.errorPair("failed update MeanObject", sql)
+                raise
+
+        # now set to null all MeanMagErr values > 0.5 (cut set by Gene, 2012-04-12)
+        # XXX EAM 20140724 : keep this cut?
+        cut = 0.5
+        self.logger.infoPair("Setting to NULL all MeanMagErr value >", "%f" % cut)
+        for filter in filters:
+
+            sql = "UPDATE ForcedMeanObject \
+                   SET " + filter[1] + "PSFMagErr = null \
+                   WHERE " + filter[1] + "PSFMagErr > " + str(cut)
+            self.scratchDb.execute(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 ObjectThin "
+            sql += "SET nDetections = nDetections + n" + filter[1]
+            sql += " WHERE n" + filter[1] + " != -999"
+            self.scratchDb.execute(sql)
+
+    '''
+    Inserts stuff for all mags from the cpy files
+    '''
+    def updateForcedMeanObjectFromCpy(self, cpyTable):
+
+        # 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", "cpy 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 ForcedMeanObject JOIN \
+                   " + cpyTable + " AS cpy ON (cpy.row = (ForcedMeanObject.row* " + str(filterCount) + ")-(" + str(filterCount) + " - " + str(filter[0]) + ")) \
+                   SET \
+                    ForcedMeanObject." + filter[1] + "nIncR5          =  NMEAS \
+                   ,ForcedMeanObject." + filter[1] + "nIncR6          =  NMEAS \
+                   ,ForcedMeanObject." + filter[1] + "FmeanflxR5      =  FLUX_AP_R5 \
+                   ,ForcedMeanObject." + filter[1] + "FmeanflxR5Err   =  FLUX_ERR_AP_R5 \
+                   ,ForcedMeanObject." + filter[1] + "FmeanflxR5Std   =  FLUX_STD_AP_R5 \
+                   ,ForcedMeanObject." + filter[1] + "FmeanflxR5Fill  =  FLUX_FIL_AP_R5 \
+                   ,ForcedMeanObject." + filter[1] + "FmeanflxR6      =  FLUX_AP_R6 \
+                   ,ForcedMeanObject." + filter[1] + "FmeanflxR6Err   =  FLUX_ERR_AP_R6 \
+                   ,ForcedMeanObject." + filter[1] + "FmeanflxR6Std   =  FLUX_STD_AP_R6 \
+                   ,ForcedMeanObject." + filter[1] + "FmeanflxR6Fill  =  FLUX_FIL_AP_R6 \
+                   ,ForcedMeanObject." + filter[1] + "LensObjSmearX11 =  X11_SM_OBJ \
+                   ,ForcedMeanObject." + filter[1] + "LensObjSmearX12 =  X12_SM_OBJ \
+                   ,ForcedMeanObject." + filter[1] + "LensObjSmearX22 =  X22_SM_OBJ \
+                   ,ForcedMeanObject." + filter[1] + "LensObjSmearE1  =  E1_SM_OBJ \
+                   ,ForcedMeanObject." + filter[1] + "LensObjSmearE2  =  E2_SM_OBJ \
+                   ,ForcedMeanObject." + filter[1] + "LensObjShearX11 =  X11_SH_OBJ \
+                   ,ForcedMeanObject." + filter[1] + "LensObjShearX12 =  X12_SH_OBJ \
+                   ,ForcedMeanObject." + filter[1] + "LensObjShearX22 =  X22_SH_OBJ \
+                   ,ForcedMeanObject." + filter[1] + "LensObjShearE1  =  E1_SH_OBJ \
+                   ,ForcedMeanObject." + filter[1] + "LensObjShearE2  =  E2_SH_OBJ \
+                   ,ForcedMeanObject." + filter[1] + "LensPSFSmearX11 =  X11_SM_PSF \
+                   ,ForcedMeanObject." + filter[1] + "LensPSFSmearX12 =  X12_SM_PSF \
+                   ,ForcedMeanObject." + filter[1] + "LensPSFSmearX22 =  X22_SM_PSF \
+                   ,ForcedMeanObject." + filter[1] + "LensPSFSmearE1  =  E1_SM_PSF \
+                   ,ForcedMeanObject." + filter[1] + "LensPSFSmearE2  =  E2_SM_PSF \
+                   ,ForcedMeanObject." + filter[1] + "LensPSFShearX11 =  X11_SH_PSF \
+                   ,ForcedMeanObject." + filter[1] + "LensPSFShearX12 =  X12_SH_PSF \
+                   ,ForcedMeanObject." + filter[1] + "LensPSFShearX22 =  X22_SH_PSF \
+                   ,ForcedMeanObject." + filter[1] + "LensPSFShearE1  =  E1_SH_PSF \
+                   ,ForcedMeanObject." + filter[1] + "LensPSFShearE2  =  E2_SH_PSF \
+                   ,ForcedMeanObject." + filter[1] + "Gamma           =  GAMMA \
+                   ,ForcedMeanObject." + filter[1] + "E1              =  E1 \
+                   ,ForcedMeanObject." + filter[1] + "E2              =  E2 "
+
+
+            try: self.scratchDb.execute(sql)
+            except:
+                self.logger.errorPair("failed update on ForcedMeanObject using cpy table", sql)
+                raise
+
+
+    '''
+    Populates the ForcedMeanObject table
+    '''
+    def populateForcedMeanObjectTable(self):
+
+        cptTableName = self.scratchDb.getDbFriendlyTableName(self.region + ".cpt")
+        cpsTableName = self.scratchDb.getDbFriendlyTableName(self.region + ".cps")
+        cpyTableName = self.scratchDb.getDbFriendlyTableName(self.region + ".cpy")
+
+                
+        self.logger.info("Populating ForcedMeanObject")
+        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 ForcedMeanObject (")
+
+        sqlLine.group("objID",           "EXT_ID") 
+        sqlLine.group("ippObjID",        "OBJ_ID + (CAT_ID << 32)") # NOTE: shift by 32 bits exactly
+        sqlLine.group("batchID",         "'" + str(self.batchID) + "'")
+        sqlLine.group("nDetections",      "'0'")
+        sql = sqlLine.makeRaw(") SELECT ", " FROM " + cptTableName)
+
+        try:
+            self.scratchDb.execute(sql)
+        except:
+            self.logger.errorPair("Couldn't populate Object table from cpt", sql)
+            return False
+ 
+        # add row count columns so we can perform joins to get colors
+        self.logger.infoPair("Adding 'row' columns to", "Object and cps tables")
+        self.scratchDb.addRowCountColumn("ForcedMeanObject", "row")
+        self.scratchDb.addRowCountColumn(cpsTableName, "row")
+
+
+        self.logger.infoPair("update ForcedMeanObject from ","cps table")
+
+        self.updateForcedMeanObjectFromCps(cpsTableName)
+
+        self.logger.infoPair("update ForcedMeanObject from ","cpy table")
+
+        self.updateForcedMeanObjectFromCpy(cpyTableName)
+
+
+        self.logger.infoPair("Dropping row column from", "ForcedMeanObject table")
+        self.scratchDb.dropColumn("ForcedMeanObject", "row")
+
+        ##self.logger.infoPair("Purging from scratch Db", self.region + " region")
+
+        ##Don't do this till after MeanObject
+        ##self.dvoForcedObjects.purgeRegion(self.region)
+
+        self.setMinMaxObjID(["ForcedMeanObject"])
+
+        return True
+
+ 
+    '''
+    Does the processing, i.e. pulling stuff from IPP tables into PSPS tables
+    '''
+    def populatePspsTables(self):
+
+        if not self.populateForcedMeanObjectTable(): 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 ForcedMeanObject table")
+        rowCountBefore = self.scratchDb.getRowCount("ForcedMeanObject")
+
+        # 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 ForcedMeanObject ADD UNIQUE INDEX(objID)")
+        if self.scratchDb.version > 5.1:
+            self.scratchDb.execute("set session old_alter_table=0")
+
+        rowCountAfter = self.scratchDb.getRowCount("ForcedMeanObject")
+        self.logger.infoPair("Number of duplicated objIDs removed", "%d out of %d" % ((rowCountBefore - rowCountAfter), rowCountBefore))
+
+        self.dvoForcedObjects.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