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

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

@@ +1 @@
+#!/usr/bin/env jython
+
+import os.path
+import sys
+import glob
+import time
+import stilts
+
+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
+
+
+'''
+ForcedWarpBatch class
+
+This class, a sub-class of Batch, processes single exposures in the form of IPP smf files
+
+'''
+class ForcedWarpBatch(Batch):
+
+    '''
+    Constructor
+    '''
+    def __init__(self, 
+                 logger,
+                 config,
+                 skychunk,
+                 gpc1Db,
+                 ippToPspsDb,
+                 scratchDb,
+                 forcedWarpID,
+                 batchID):
+
+       super(ForcedWarpBatch, self).__init__(
+               logger,
+               config,
+               skychunk,
+               gpc1Db,
+               ippToPspsDb,
+               scratchDb,
+               forcedWarpID,
+               batchID,
+               "P2", 
+               gpc1Db.getForcedWarpStageSmf(forcedWarpID)
+
+       # get camera meta data
+       meta = self.gpc1Db.getForcedWarpStageMeta(self.id)
+       if not meta:
+           self.logger.errorPair("Could not get", "forced warp metadata")
+           raise
+
+       self.expID = meta[0];
+       self.expName = meta[1];
+       self.distGroup = meta[2];
+       self.analysisVer = meta[3];
+       self.bias = meta[4];
+       self.biasScat = meta[5];
+
+       if not self.analysisVer: self.analysisVer = -999
+
+       # create an output filename, which is {expID}.FITS
+       self.outputFitsFile = "%08d.FITS" % self.expID
+       self.outputFitsPath = "%s/%s" % (self.localOutPath, self.outputFitsFile)
+       self.dropTableVerbose("ForcedWarpMeasurement")
+       self.dropTableVerbose("ForcedWarpExtended")
+       self.dropTableVerbose("ForcedWarpMeta")
+       self.dropTableVerbose("ForcedWarpToImage")
+       if not self.config.retry:
+       self.scratchDb.dropTable("ForcedWarpMeasurement")
+       self.scratchDb.dropTable("ForcedWarpExtended")
+       self.scratchDb.dropTable("ForcedWarpMeta")
+       self.scratchDb.dropTable("ForcedWarpToImage")
+       self.historyModNum = "0"
+
+       # get a few primary header values. if in test mode, then use defaults
+       if self.safeDictionaryAccessWithDefault(self.header, 'MJD-OBS', "1") == "NULL":
+           self.logger.errorPair("Could not get", "MJD-OBS")
+           raise
+
+
+       if self.safeDictionaryAccessWithDefault(self.header, 'EXPTIME', "1") == "NULL":
+           self.logger.errorPair("Could not get", "EXPTIME")
+           raise
+
+       if self.safeDictionaryAccessWithDefault(self.header, 'FILTERID', "g.0000") == "NULL":
+           self.logger.errorPair("Could not get", "FILTERID")
+           raise
+       
+       # MJD-OBS is the exposure start, EXTIME / 172800 = (EXPTIME sec / 84600 sec/day) / 2
+       self.obsTime = float(self.header['MJD-OBS']) + (float(self.header['EXPTIME']) / 172800.0)
+      
+       # set up some defauts
+       self.calibModNum = 0
+       self.totalNumPhotoRef = 0
+
+
+       self.filter = self.header['FILTERID'][0:1]
+       self.filterID = self.scratchDb.getFilterID(self.filter) 
+
+       # insert what we know about this stack batch into the stack table
+       self.ippToPspsDb.insertDetectionMeta(self.batchID, self.expID, self.filter)
+
+       # dump stuff to log
+       self.logger.infoPair("Cam ID", "%d" % self.id)
+       self.logger.infoPair("Exp ID", "%d" % self.expID)
+       self.logger.infoPair("Exp name", "%s" % self.expName)
+       self.logger.infoPair("Distribution group", "%s" % self.distGroup)
+
+
+
+
+    '''
+    Populates the FrameMeta table, mainly from dictionary values found in IPP FITS header
+    '''
+    def populateForcedWarpMeta(self):
+        self.logger.infoPair("Processing table", "ForcedWarpMeta")
+
+        photoCalID = str(self.scratchDb.getPhotoCalID(forcedWarpID))       
+
+
+        sqlLine = sqlUtility("INSERT INTO ForcedWarpMeta (")
+        #batchID (below)
+        #surveyID (below)
+        #filterdID (below)
+        sqlLine.group("skyCellID",     str(self.skycellID)) 
+        sqlLine.group("photoCalID",    photoCalID) 
+        sqlLine.group("magSat",           self.getKeyFloat(header, "%.8f", 'FSATUR'))
+        sqlLine.group("analVer",           self.analysisVer);
+        sqlLine.group("expTime",               self.getKeyFloat(self.header, "%.8f", 'EXPREQ'));
+        sqlLine.group("completMag",       self.getKeyFloat(header, "%.8f", 'FLIMIT'))
+        sqlLine.group("astroScat",        self.getKeyFloat(header, "%.8f", 'CERROR'))
+        sqlLine.group("photoScat",             self.getKeyFloat(self.header, "%.8f", 'ZPT_ERR'));
+        sqlLine.group("nAstroRef",      self.getKeyValue(header, 'NASTRO'))
+        sqlLine.group("nPhotoRef",      self.getKeyValue(header, 'NASTRO'))
+        sqlLine.group("psfModelID",     self.getKeyValue(header, 'PSFMODEL'))
+        sqlLine.group("psfFwhm_mean",   self.getKeyFloat(header, "%.8f", 'FWHM_MAJ'))
+        sqlLine.group("psfFwhm_max",    self.getKeyFloat(header, "%.8f", 'FW_MJ_UQ'))
+        sqlLine.group("photoZero",        self.getKeyFloat(self.header, "%.8f", 'ZPT_OBS'))
+        #photoColor -- how do I set this? it's also not set in stack meta
+        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.scratchDb.updateAllRows("ForcedWarpMeta", "batchID", str(self.batchID))
+        self.scratchDb.updateAllRows("ForcedWarpMeta", "surveyID", str(self.surveyID))
+        self.scratchDb.updateFilterID("ForcedWarpMeta", self.filter)
+        self.scratchDb.updateAllRows("ForcedWarpMeta", "calibModNum", str(self.calibModNum))
+        self.scratchDb.updateAllRows("ForcedWarpMeta", "dataRelease", str(self.skychunk.dataRelease))
+        self.tablesToExport.append("ForcedWarpMeta")
+       
+    '''
+    Populates the ForcedWarpToImage table
+    '''
+    def populateForcedWarpToImage(self):
+
+        self.logger.infoPair("Procesing table", "ForcedWarpToImage")
+
+        for filter in self.filters:
+            
+            forcedwarpID = self.forcedwarpID
+            if forcedwarpID > 0:
+    
+                imageIDs = self.gpc1Db.getImageIDsForThisForcedWarpID(forcedwarpID)
+
+                for imageID in imageIDs:
+
+                    sql = "INSERT INTO ForcedWarpToImage (forcedwarpID, imageID) \
+                   VALUES (\
+                   " + str(forcedwarpID) + ", " + imageID + ")"
+                    self.scratchDb.execute(sql)
+
+                    # now update StackMeta with correct number of inputs
+
+        self.tablesToExport.append("ForcedWarpToImage")
+
+
+
+
+
+    '''
+    Populates the ForcedWarpMeasurement table
+    '''
+    def populateForcedWarpMeasurementTable(self, results):
+
+        pspsTableName = "ForcedWarpMeasurement"
+        ippTableName = "ForcedWarpMeasurement_psf"
+
+
+        self.logger.infoPair("Procesing table", "ForcedWarpMeasurement")
+
+        # insert the per-object data (IDs, random stack ID, dataRelease, etc)
+        self.selectDvoObjIDs()
+
+        self.generateRandomIDs()
+
+        self.generateUniquePspsIDs()
+           
+        # add indexes ForcedWarpMeasurement
+        self.scratchDb.createIndex("ForcedWarpMeasurement", "objID")
+
+               
+
+
+
+
+
+
+
+        self.tablesToExport.append("ForcedWarpMeasurement")
+
+
+    '''
+    Populates the ForcedWarpExtended table
+    '''
+    def populateForcedWarpExtendedTable(self, results):
+
+        pspsTableName = "ForcedWarpExtended"
+        ippTableName = "ForcedWarpExtended_psf"
+
+    
+        self.tablesToExport.append("ForcedWarpExtended")
+
+           # get dvo ids, 
+        # add in the psps unique ids from stuff
+
+    '''
+    Does the processing, i.e. pulling stuff from IPP tables into PSPS tables
+    '''
+    def populatePspsTables(self):
+        self.logger.infoPair("starting","populatePspsTables");
+        self.skipBatch = False
+        # each of the "populate*" methods below add their tables to the list:
+        self.tablesToExport=[]
+    
+        #this should be done
+        self.populateForcedWarpMeta()          
+
+        self.populateForcedWarpMeasurement()           
+        self.populateForcedWarpExtended()
+        self.populateForcedWarpToImage()
+             
+        self.setMinMaxObjID(["ForcedWarpMeasurement"])
+
+        self.logger.infoPair("finishing","populatePspsTables"); 
+        return True
+          
+
+
+    '''
+    This function reads the cmf/smf file and loads it into the database as its own table
+    '''
+    def importIppTables(self, columns="*", tableRE=""):
+        
+        if self.config.retry: return True
+        self.logger.infoPair("Importing FW tables with table match expression: ", tableRE)
+        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
+        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
+
+
+               
+
+
+
+
+
+
+    def generateRandomIDs(self):
+           sql = "UPDATE ForcedWarpMeasurement set randomWarpID = FLOOR(RAND("+str(self.batchID)+")*9223372036854775807)"
+        try: self.scratchDb.execute(sql)
+        except:
+            self.logger.errorPair('failed sql',sql)
+            return
+
+
+    '''
+    Updates table and generates pspsuniqueids
+    '''
+
+    def generatePspsUniqueIDs(self):
+        sql = "UPDATE ForcedWarpMeasurement join (select @r:=@r+1 rownum, objID from \
+        (select @r:=0) r, ForcedWarpMeasurement t) as foo using (objID) set \
+        uniquePspsFWid = ((" +str(self.batchID)+ "*1000000000 ) + rownum)"
+        try: self.scratchDb.execute(sql)
+        except:
+           self.logger.errorPair('failed sql',sql)
+           return
+            
+
+
+
+
+
+
+    def exportPspsTablesToFits(self, regex="(.*)"):
+       return super(DetectionBatch, self).exportPspsTablesToFits("(ForcedWarp.*)")