Index: trunk/ippToPsps/jython/forcedobjectbatch.py =================================================================== --- trunk/ippToPsps/jython/forcedobjectbatch.py (revision 36697) +++ trunk/ippToPsps/jython/forcedobjectbatch.py (revision 37551) @@ -0,0 +1,360 @@ +#!/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