IPP Software Navigation Tools IPP Links Communication Pan-STARRS Links

Ignore:
Timestamp:
May 4, 2011, 3:20:38 PM (15 years ago)
Author:
watersc1
Message:

Merging trunk into this branch

Location:
branches/czw_branch/20110406
Files:
2 edited

Legend:

Unmodified
Added
Removed

  • branches/czw_branch/20110406

  • branches/czw_branch/20110406/ippToPsps/jython/batch.py

    r31253 r31434  
    1212
    1313from datastore import Datastore
     14from scratchdb import ScratchDb
    1415from gpc1db import Gpc1Db
    1516from ipptopspsdb import IppToPspsDb
    … …  
    2425class Batch(object):
    2526
    26     driverName="com.mysql.jdbc.Driver"
    27 
    2827    '''
    2928    Constructor
    … …  
    3332    "../config/2/tables.vot"
    3433    '''
    35     def __init__(self, logger, batchType, inputFitsPath="", survey=""):
     34    def __init__(self, logger, batchType, inputFitsPath="", survey="", useFullTables=False):
    3635
    3736        # set up logging
    3837        self.logger = logger
     38        self.logger.info("-------------------------------------------------------------------------------")
    3939        self.logger.debug("Batch class constructor")
    4040
    … …  
    4444        self.inputFitsPath = inputFitsPath
    4545        self.survey = survey
     46        self.useFullTables = useFullTables
     47
     48        # TODO
     49        self.tablesToExport = []
    4650
    4751        # open config
    4852        doc = ElementTree(file="config.xml")
    4953
    50         # set up JDBC connection to local Db
    51         dbName = doc.find("localdatabase/name").text
    52         dbHost = doc.find("localdatabase/host").text
    53         dbUser = doc.find("localdatabase/user").text
    54         dbPass = doc.find("localdatabase/password").text
    55         self.localUrl = "jdbc:mysql://"+dbHost+"/"+dbName+"?user="+dbUser+"&password="+dbPass
    56         self.localCon = DriverManager.getConnection(self.localUrl)
    57         self.localStmt = self.localCon.createStatement()
    58 
    5954        # create Gpc1Db object
    6055        self.gpc1Db = Gpc1Db(self.logger)
    61 
    62         # get survey ID from init table
    63         sql = "SELECT surveyID from Survey WHERE name = '" + survey + "'"
    64         try:
    65             rs = self.localStmt.executeQuery(sql) 
    66             rs.first()
    67             self.surveyID = rs.getInt(1)
    68         except:
    69             self.logger.exception("No survey ID found for this survey: '" + survey + "'")
    70             self.surveyID = -1;
     56        self.ippToPspsDb = IppToPspsDb(logger)
     57        self.scratchDb = ScratchDb(logger, self.useFullTables)
     58
     59        if self.survey != "":
     60            self.surveyID = self.scratchDb.getSurveyID(self.survey)
    7161   
    72         # get dvo info from config
    73         if survey != "":
    74             dvoName = doc.find(survey+"dvo/name").text
    75             dvoLocation = doc.find(survey+"dvo/location").text
     62            # get dvo info from config
     63            dvoName = doc.find("dvo_" + self.survey + "/name").text
     64            self.dvoLocation = doc.find("dvo_" + self.survey + "/location").text
    7665        else:
    7766            dvoName = ""
    78             dvoLocation = ""
     67            self.dvoLocation = ""
     68            self.surveyID = -1;
    7969         
    8070        # get datastore info from config
    8171        self.datastore = Datastore(self.logger)
    8272
    83         # create IppToPspsDb object and create a new batch
    84         self.ippToPspsDb = IppToPspsDb(logger)
    85         self.batchID = self.ippToPspsDb.createNewBatch(66,
    86                 survey,
     73        # create a new batch
     74        self.batchID = self.ippToPspsDb.createNewBatch(
    8775                self.getPspsBatchType(),
     76                survey,
    8877                dvoName,
    8978                self.datastore.product)
    … …  
    9584        if not os.path.exists(self.localOutPath): os.makedirs(self.localOutPath)
    9685
    97 
    9886        # store today's date
    9987        now = datetime.datetime.now();
    10088        self.dateStr = now.strftime("%Y-%m-%d")
    10189
    102         if self.inputFitsPath != "": self.parseFitsHeader()
    103 
    104         # create DVO table
    105         self.createDvoTable()
     90        if self.inputFitsPath != "":
     91            file = open(self.inputFitsPath)
     92            self.header = self.parseFitsHeader(file)
     93            self.logger.info("Read primary and found " + str(len(self.header)) + " header cards")
     94            # TODO close file?
     95
     96        # create DVO tables if accessing DVO directly
     97        if not self.useFullTables: self.scratchDb.createDvoTables()
    10698
    10799    '''
    … …  
    111103
    112104        self.logger.debug("Batch destructor")
    113         self.localStmt.close()
    114         self.localCon.close()
     105
     106
     107    '''
     108    Returns the value from this dictinary or else NULL
     109    '''
     110    def safeDictionaryAccess(self, header, key):
     111
     112         if key in header: return header[key]
     113         else: return "NULL"
     114
     115    '''
     116    Finds and reads a header extension
     117    '''
     118    def findAndReadFITSHeader(self, name, file):
     119
     120        found = False
     121       
     122        while True:
     123           
     124            index = file.tell()
     125
     126            record = file.read(80)
     127            if not record: break;
     128
     129            # quit when we reach 'END'
     130            if record.startswith("XTENSION= 'IMAGE"):
     131
     132                header = self.parseFitsHeader(file)
     133                if header['EXTNAME'] == name:
     134                    found = True
     135                    file.seek(index + 2880, 0)
     136                    break
     137
     138            file.seek(index + 2880, 0)
     139           
     140        if found != True: self.logger.error("...could not find extension '" + name + "'")
     141        else: self.logger.info("...read header at '" + name + "' and found " + str(len(header)) + " header cards")
     142
     143        return header
     144
    115145
    116146    '''
    … …  
    127157        root.attrib['name'] = self.batchName
    128158        root.attrib['type'] = self.getPspsBatchType()
    129         root.attrib['survey'] = self.getPspsSurveyType()
    130159        root.attrib['timestamp'] = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    131         root.attrib['minObjId'] = str(self.minObjID)
    132         root.attrib['maxObjId'] = str(self.maxObjID)
     160        if self.survey != "":
     161            root.attrib['survey'] = self.getBatchFriendlySurveyType()
     162        try: self.minObjID
     163        except: pass
     164        else: root.attrib['minObjId'] = str(self.minObjID)
     165        try: self.maxObjID
     166        except: pass
     167        else: root.attrib['maxObjId'] = str(self.maxObjID)
    133168
    134169        # get md5sum
    … …  
    190225    def publishToDatastore(self):
    191226
    192         self.datastore.publish(self.batchName, self.subDir, self.tarballFile, "tgz")
    193         # TODO update ippToPsps Db here
     227        if self.datastore.publish(self.batchName, self.subDir, self.tarballFile, "tgz"):
     228            self.ippToPspsDb.updateLoadedToDatastore(self.batchID, 1)
    194229
    195230    '''
    196231    Gets PSPS-friendly survey type
    197232    '''
    198     def getPspsSurveyType(self):
    199 
    200         if self.survey == "ThreePi": return "3PI"
    201         elif self.survey == "MD04": return "MD04"
    202         else: self.logger.error("Don't know this survey: " + self.survey)
     233    def getBatchFriendlySurveyType(self):
     234
     235        return "SCR" # TODO
     236
     237        try:
     238            self.survey
     239        except:
     240            return "NA"
     241
     242        if self.survey == "3PI": return "3PI"
     243        elif self.survey == "MD04": return "MD4"
     244        else:
     245            self.logger.error("Don't know this survey: '" + self.survey + "'")
     246            return "NA"
    203247
    204248    '''
    … …  
    212256        else: self.logger.error("Don't know this batch type: " + self.survey)
    213257
    214 
    215     '''
    216     Prints a log message with the current time
    217     '''
    218     def log(self, msg):
    219 
    220         print datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") + " | " + msg
    221 
    222     '''
    223     Sets min and max obj ID using the provided table
    224     '''
    225     def setMinMaxObjID(self, table):
    226 
    227         sql = "SELECT MIN(objID), MAX(objID) FROM " + table
    228         print
    229         rs = self.localStmt.executeQuery(sql)
    230         rs.first()
    231         self.minObjID = rs.getLong(1)
    232         self.maxObjID = rs.getLong(2)
    233 
    234         print "MIN mAX = %d %d" % (self.minObjID,  self.maxObjID)
    235 
    236     '''
    237     Updates a table with surveyID
    238     '''
    239     def updateSurveyID(self, table):
    240 
    241         sql = "UPDATE " + table + "  SET surveyID=%d" % self.surveyID
    242         self.localStmt.execute(sql)
    243 
    244     '''
    245     Updates a table with filterID grabbed from Filter init table
    246     '''
    247     def updateFilterID(self, table):
    248 
    249         sql = "UPDATE "+table+" AS a, Filter AS b SET a.filterID=b.filterID WHERE b.filterType = '" + self.filter + "'"
    250         self.localStmt.execute(sql)
     258    '''
     259    Sets min and max obj ID using the provided table, or list of tables
     260    '''
     261    def setMinMaxObjID(self, tables):
     262
     263        first = True
     264        for table in tables:
     265
     266            sql = "SELECT MIN(objID), MAX(objID) FROM " + table
     267            rs = self.scratchDb.stmt.executeQuery(sql)
     268            rs.first()
     269
     270            if first:
     271                self.minObjID = rs.getLong(1)
     272                self.maxObjID = rs.getLong(2)
     273            else:
     274                if rs.getLong(1) < self.minObjID: self.minObjID = rs.getLong(1)
     275                if rs.getLong(2) > self.maxObjID: self.maxObjID = rs.getLong(2)
     276
     277            first = False
     278
     279        self.ippToPspsDb.updateMinMaxObjID(self.batchID, self.minObjID, self.maxObjID)
    251280
    252281    '''
    253282    Reads FITS header and stores all fields in a dictionary object
    254283    '''
    255     def parseFitsHeader(self):
    256 
    257         fitsFile = open(self.inputFitsPath)
    258 
    259         self.header = {}
     284    def parseFitsHeader(self, fitsFile):
     285
     286        header = {}
    260287
    261288        while (True):
     289
    262290           record = fitsFile.read(80)
    263291
    264292           # quit when we reach 'END'
    265            if record.startswith("END"): break
    266 
    267            # ignore comments
    268            if record.startswith("COMMENT"): continue
    269            match = re.match('(.*)=(.*)', record)
     293           if re.match('END\s+', record): break
     294
     295           # this regex will get param/value pairs for all header cards, ignoring comments and parsing out 'HIERACH' prefixes
     296           match = re.match('^(HIERARCH )*([a-zA-Z0-9-_\.]+)\s*=\s+\'*([a-zA-Z0-9-_\.:\s@#]+)\'*\\/*', record)
    270297           if match:
    271298
    272                # remove HIERARCH prefix
    273                param = match.group(1).replace("HIERARCH", "")
    274                param = param.strip()
    275 
    276                value = match.group(2)
    277                # remove trailing comment after / char, if there is one
    278                index = value.find("/")
    279                if index != -1: value = value[0:index]
    280 
    281                # remove ' chars around content
    282                value = value.replace("'", "")
    283 
    284                # remove leading and trailing whitespace
    285                value = value.strip()
    286 
    287                # store in out dictionary object
    288                self.header[param] = value
     299               param = match.group(2)
     300               value = match.group(3).strip()
     301               if value == "NaN": value = "NULL"
     302               header[param] = value
    289303
    290304               #print param + "|" + value + "|"
     305
     306        return header
    291307
    292308    '''
    … …  
    298314         for table in self.pspsTables:
    299315             self.logger.info("Creating PSPS table: " + table.name)
    300              table.write(self.localUrl + '#' + table.name)
    301 
    302          self.indexPspsTables()
     316             table.write(self.scratchDb.url + '#' + table.name)
     317             self.tablesToExport.append(table.name)
     318
     319         self.alterPspsTables();
    303320
    304321    '''
    … …  
    310327
    311328    '''
    312     Adds an index to the supplied table and column
    313     '''
    314     def createIndex(self, table, column):
    315 
    316         self.logger.info("Creating index on column '"+column+"' for table '"+table+"'")
    317 
    318         sql = "CREATE INDEX "+table+"_index ON "+table+" ("+column+")"
    319         try:
    320             self.localStmt.execute(sql)
    321         except:
    322             self.logger.warn("Index already in place on '" + column + "' for table '" + table + "'")
    323 
    324 
    325     '''
    326     Subclass should implement this to index PSPS tables
     329    Alter PSPS tables
    327330    '''   
    328     def indexPspsTables(self):
    329         self.logger.warn("indexPspsTables not implemented")
     331    def alterPspsTables(self):
     332        self.logger.warn("alterPspsTables not implemented")
    330333
    331334    '''
    332335    Imports IPP tables from FITS file
    333336
    334     Accepts a regular expression filter so not all tabls need to be imported
     337    Accepts a regular expression filter so not all tables need to be imported
    335338    '''
    336339    def importIppTables(self, filter):
    337340
     341      self.logger.info("Attempting to import tables from input FITS file")
    338342      tables = stilts.treads(self.inputFitsPath)
    339343
    … …  
    343347          match = re.match(filter, table.name)
    344348          if not match: continue
    345           self.logger.info("Creating IPP table " + table.name)
     349          self.logger.info("   Reading IPP table " + table.name + " from FITS file")
    346350          table = stilts.tpipe(table, cmd='explodeall')
     351
     352          # drop any previous tables before import
     353          self.scratchDb.dropTable(table.name)
     354
     355          # IPP FITS files are littered with infinities, so remove these
     356          self.logger.info("   Removing Infinity values from all columns")
     357          table = stilts.tpipe(table, cmd='replaceval -Infinity null *')
     358          table = stilts.tpipe(table, cmd='replaceval Infinity null *')
     359
    347360          try:
    348               table.write(self.localUrl + '#' + table.name)
     361              table.write(self.scratchDb.url + '#' + table.name)
    349362          except:
    350               self.logger.exception("Problem writing table '" + table.name + "' to the database")
    351 
     363              self.logger.exception("   Problem writing table '" + table.name + "' to the database")
    352364          count = count + 1
    353365
    354       self.logger.info("Imported %d tables from '%s' " % (count, self.inputFitsPath))
     366      self.logger.info("Done. Imported %d tables" % count)
    355367
    356368      self.indexIppTables()
    357369
    358370    '''
    359     Exports PSPS tables from the database to FITS format
     371    Exports PSPS tables from the database to FITS format. Optional regex if you want to alter table names prior to export
    360372    '''   
    361     def exportPspsTablesToFits(self):
    362 
    363         self.logger.info("Exporting all PSPS tables to FITS")
     373    def exportPspsTablesToFits(self, regex="(.*)"):
     374
     375        self.logger.info("Replacing NULLs with -999 then exporting all PSPS tables to FITS")
    364376        _tables = []
    365377
    366378        self.logger.info("    Selecting database tables")
    367         for table in self.pspsTables:
    368            _table = stilts.tread(self.localUrl + '#SELECT * FROM ' + table.name)
    369            _table = stilts.tpipe(_table, cmd='tablename ' + table.name)
     379        for table in self.tablesToExport:
     380
     381           # check for an empty table
     382           if self.scratchDb.getRowCount(table) < 1: continue
     383
     384           # get everything from table
     385           _table = stilts.tread(self.scratchDb.url + '#SELECT * FROM ' + table)
     386
     387           # replace nulls and empty fields with weird PSPS -999 pseudo-null
     388           _table = stilts.tpipe(_table, cmd='replaceval "" -999 *')
     389
     390           match = re.match(regex, table)
     391           newTableName = match.group(1)
     392
     393           # change table names
     394           _table = stilts.tpipe(_table, cmd='tablename ' + newTableName)
    370395           _tables.append(_table)
    371396
    … …  
    373398        stilts.twrites(_tables, self.outputFitsPath, fmt='fits')
    374399        self.logger.info("    ...done")
    375 
    376     '''
    377     Returns a list of column names for this table
    378     '''
    379     def getColumnNames(self, tableName):
    380 
    381        sql = "SHOW COLUMNS FROM " + tableName
    382        rs = self.localStmt.executeQuery(sql)
    383        columns = []
    384        while (rs.next()): columns.append(rs.getString(1))
    385        rs.close()
    386        
    387        return columns
    388 
    389     '''
    390     Replaces all NULL values in the provided table with the prvoded substitute
    391     '''
    392     def replaceNulls(self, tableName, sub):
    393 
    394        # get list of columns
    395        columns = self.getColumnNames(tableName)
    396 
    397        # now loop through all columns and replace all NULLs with sub
    398        for column in columns:
    399          
    400           sql = "UPDATE " + tableName + " SET " + column + " = " + sub + " WHERE " + column + " IS NULL"
    401           self.localStmt.execute(sql)
    402 
    403 
    404     '''
    405     Searches a table and reports the columns that are either partially or completely populated with NULLs
    406     '''
    407     def reportNulls(self, tableName, showPartials):
    408 
    409        # first, count rows
    410        sql = "SELECT COUNT(*) FROM " + tableName
    411        rs = self.localStmt.executeQuery(sql)
    412        rs.first()
    413        numRows = rs.getInt(1)
    414 
    415        # get list of columns
    416        columns = self.getColumnNames(tableName)
    417 
    418        print "+----------------------+---------------+"
    419        print "|  %25s           |" % tableName
    420        print "+----------------------+---------------+"
    421 
    422        # now see which columns are full of NULLS, with are partially NULL
    423        for column in columns:
    424          
    425           sql = "SELECT COUNT(*) FROM " + tableName + " WHERE " + column + " IS NULL"
    426           rs = self.localStmt.executeQuery(sql)
    427           rs.first()
    428           if rs.getInt(1) == numRows:
    429               print "| %20s | all NULL      |" % column
    430           elif showPartials and rs.getInt(1) > 0:
    431               print "| %20s | partial NULL  |" % column
    432        rs.close()
    433        print "+----------------------+---------------+"
    434 
     400        self.ippToPspsDb.updateProcessed(self.batchID, 1)
    435401
    436402    '''
    … …  
    440406
    441407        for table in self.pspsTables:
    442             self.reportNulls(table.name, showPartials)
     408            self.scratchDb.reportNulls(table.name, showPartials)
    443409
    444410    '''
    … …  
    449415        self.logger.info("Replacing all NULL values in PSPS tables with '" + sub + "'...")
    450416        for table in self.pspsTables:
    451             self.replaceNulls(table.name, sub)
     417            self.scratchDb.replaceNulls(table.name, sub)
    452418        self.logger.info("...done")
    453419
    … …  
    459425
    460426    '''
    461     Updates provided table with DVO IDs from DVO table
    462     '''
    463     def updateDvoIDs(self, table):
    464 
    465         self.logger.info("Updating table '" + table + "' with DVO IDs...")
    466         sql = "UPDATE " + table + " AS a, dvo AS b SET \
    467                a.ippObjID = b.ippObjID, \
    468                a.objID = b.objID \
    469                WHERE a.ippDetectID = b.ippDetectID"
    470         self.localStmt.execute(sql)
     427    Calls DVO program to 'query' DVO database and populate results to local MySQL Db table
     428    '''
     429    def getIDsFromDVO(self):
     430
     431        # TODO path to DVO prog hardcoded temporarily
     432        cmd = "../src/dvograbber " + self.dvoLocation
     433        self.logger.info("Running: '" + cmd + "'...")
     434        p = Popen(cmd, shell=True, stdout=PIPE)
     435        p.wait()
     436        #        out = p.stdout.read()
    471437        self.logger.info("...done")
    472438
    473 
    474     '''
    475     Creates a table for for ID matching
    476     '''
    477     def createDvoTable(self):
    478 
    479         self.logger.info("Creating DVO table for ID matching")
    480         sql = "DROP TABLE dvo"
    481         self.localStmt.execute(sql)
    482         sql = "CREATE TABLE dvo (ippDetectID BIGINT PRIMARY KEY, ippObjID BIGINT, objID BIGINT)"
    483         self.localStmt.execute(sql)
    484 
     439        if self.scratchDb.getRowCount("dvoDetection") < 1:
     440            self.logger.error("No DVO IDs found")
     441            return False
     442           
     443        return True
     444
     445    '''
     446    Checks whether this batch has already been processed and published. To be implemented by all subclasses
     447    '''
     448    def alreadyProcessed(self):
     449           self.logger.info("Not implemented")
     450
     451
     452
Note: See TracChangeset for help on using the changeset viewer.