IPP Software Navigation Tools IPP Links Communication Pan-STARRS Links

Ignore:
Timestamp:
Mar 30, 2012, 2:49:37 PM (14 years ago)
Author:
eugene
Message:

merge changes from trunk

File:
1 edited

Legend:

Unmodified
Added
Removed

  • branches/eam_branches/ipp-20111122/ippToPsps/jython/ipptopspsdb.py

    r32852 r33638  
    1717    Constructor
    1818    '''
    19     def __init__(self, logger, doc):
    20         super(IppToPspsDb, self).__init__(logger, doc, "ipptopspsdatabase")
    21 
    22         self.MAX_FAILS = 4
     19    def __init__(self, logger, config):
     20        super(IppToPspsDb, self).__init__(logger, config, "ipptopspsdatabase")
     21
     22        self.MAX_FAILS = 5
    2323
    2424    '''
    … …  
    3838        try:
    3939            rs = self.executeQuery(sql)
    40             while (rs.next()):
    41                 ids.append(rs.getInt(1))
     40            while (rs.next()): ids.append(rs.getInt(1))
     41            rs.close()
    4242        except:
    4343            self.logger.exception("Can't query for merged batch ids in ipptopsps Db")
    4444
    45         rs.close()
     45        self.logger.debug("Found %d merged but un-deleted items" % len(ids))
     46
     47        return ids
     48
     49    '''
     50    Returns a list of batch IDs marked as 'purged' but not yet deleted
     51    '''
     52    def getPurgedButNotDeletedBatchIDs(self, batchType, epoch, dvoGpc1Label, column):
     53
     54        sql = "SELECT DISTINCT batch_id \
     55               FROM batch \
     56               WHERE timestamp > '" + epoch + "' \
     57               AND batch_type = '" + batchType + "' \
     58               AND dvo_db = '" + dvoGpc1Label + "' \
     59               AND purged = 1 \
     60               AND " + column + " = 0"
     61
     62        ids = []
     63        try:
     64            rs = self.executeQuery(sql)
     65            while (rs.next()): ids.append(rs.getInt(1))
     66            rs.close()
     67        except:
     68            self.logger.exception("Can't query for merged batch ids in ipptopsps Db")
     69
    4670
    4771        self.logger.debug("Found %d merged but un-deleted items" % len(ids))
    … …  
    6589        try:
    6690            rs = self.executeQuery(sql)
    67             while (rs.next()):
    68                 ids.append(rs.getInt(1))
     91            while (rs.next()): ids.append(rs.getInt(1))
     92            rs.close()
    6993        except:
    7094            self.logger.exception("Can't query for merged batch ids in ipptopsps Db")
    71 
    72         rs.close()
    7395
    7496        self.logger.debug("Found %d merged but un-deleted items" % len(ids))
    … …  
    99121    def getLoadedToODMButNotDeletedFromDXLayer(self, batchType, epoch, dvoGpc1Label):
    100122        return self.getLoadedToODMButNotDeletedBatchIDs(batchType, epoch, dvoGpc1Label, "deleted_dxlayer")
     123
     124    '''
     125    Returns a list of purged  batch IDs that not deleted from local disk
     126    '''
     127    def getPurgedButNotDeletedFromLocalDisk(self, batchType, epoch, dvoGpc1Label):
     128        return self.getPurgedButNotDeletedBatchIDs(batchType, epoch, dvoGpc1Label, "deleted_local")
     129
     130    '''
     131    Returns a list of purged batch IDs that are not deleted from datastore
     132    '''
     133    def getPurgedButNotDeletedFromDatastore(self, batchType, epoch, dvoGpc1Label):
     134        return self.getPurgedButNotDeletedBatchIDs(batchType, epoch, dvoGpc1Label, "deleted_datastore")
    101135
    102136
    … …  
    107141    want to load a test batch to PSPS
    108142
    109     We also check that these batches are more than 4 HOURs old, so that we don't attempt to
     143    We also check that these batches are more than 12 HOURs old, so that we don't attempt to
    110144    publish something simulataneously with the client that it producing it
    111145    '''
    … …  
    120154               AND loaded_to_datastore != 1 \
    121155               AND timestamp > '2011-10-27' \
    122                AND timestamp < now() -  INTERVAL 4 HOUR"
     156               AND timestamp < now() -  INTERVAL 12 HOUR"
    123157
    124158        ids = []
    … …  
    127161            while (rs.next()):
    128162                ids.append(rs.getInt(1))
     163            rs.close()
    129164        except:
    130165            self.logger.exception("Can't query for processed batch ids not loaded to datastore ipptopsps Db")
    131 
    132         rs.close()
    133166
    134167        self.logger.debug("Found %d processed but-not-on-datastore items" % len(ids))
    … …  
    240273        except:
    241274            self.logger.exception("Can't query for batch ids with " + column + " = " + str(value) + " in ipptopsps Db")
    242 
    243         rs.close()
     275        finally:
     276            rs.close()
    244277
    245278        self.logger.debug("Found %d batches with %s = %d" % (len(ids), column, value))
    … …  
    250283    Returns the total detections published to the datastore for this epoch, dvo label and batch type
    251284    '''
    252     def getTotalDetectionsPublished(self, batchType, epoch, dvoGpc1Label):
     285    def getTotalDetectionsPublished(self, batchType, epoch, dvoGpc1Label, interval="10 YEAR"):
    253286
    254287        sql = "SELECT SUM(total_detections) \
    … …  
    258291               AND dvo_db = '" + dvoGpc1Label + "' \
    259292               AND processed = 1 \
    260                AND loaded_to_datastore = 1"
     293               AND loaded_to_datastore = 1 \
     294               AND timestamp BETWEEN now() - INTERVAL " + interval + " AND now()"
    261295
    262296        total = -1
    … …  
    267301        except:
    268302            self.logger.exception("Can't query for total detections published")
    269 
    270         rs.close()
     303        finally:
     304            rs.close()
    271305
    272306        self.logger.debug("Found %d detections" % total)
    … …  
    338372    '''
    339373    def getStages(self):
    340         return ['processed', 'loaded_to_datastore', 'loaded_to_ODM', 'merge_worthy', 'deleted_datastore', 'deleted_dxlayer', 'merged', 'deleted_local' ]
     374        return [
     375        'processed',
     376        'loaded_to_datastore',
     377        'loaded_to_ODM',
     378        'merge_worthy',
     379        'deleted_datastore',
     380        'deleted_dxlayer',
     381        'merged',
     382        'deleted_local' ]
    341383
    342384    '''
    343385    Returns a count of batches processed in the specified time period.
    344386    '''
    345     def countBatchesInLastPeriod(self, batchType, epoch, dvoGpc1Label, interval):
     387    def countBatches(self, batchType, epoch, dvoGpc1Label, interval="10 YEAR"):
    346388
    347389        sql = "SELECT COUNT(batch_id) \
    … …  
    359401            self.logger.exception("Unable to count batches in interval")
    360402            return 0
     403        finally:
     404            rs.close()
    361405
    362406    '''
    … …  
    454498    def updateOdmStatus(self, batchID, odmStatus):
    455499
    456         if odmStatus['LOADFAILED'] == 1: loadedToODM = -1
    457         else: loadedToODM = odmStatus['LOADEDTOODM']
    458500        sql = "UPDATE batch \
    459                SET loaded_to_ODM = " + str(loadedToODM) + ", \
     501               SET loaded_to_ODM = " + str(odmStatus['LOADEDTOODM']) + ", \
    460502               merge_worthy = " + str(odmStatus['MERGEWORTHY']) + ", \
    461503               merged = " + str(odmStatus['MERGED']) + " \
    … …  
    465507
    466508    '''
    467     Is someone processing this item right now? Checks whether this was attempted within the last 2 HOUR
    468     '''
    469     def processingNow(self, batchType, stage_id, epoch, dvoGpc1Label):
     509    Is someone processing this item right now? Checks whether this was attempted within the last 12 HOUR
     510    '''
     511    def processingNow(self, batchType, stage_id):
    470512
    471513        sql = "SELECT COUNT(*) \
    … …  
    473515               WHERE stage_id = " + str(stage_id) + " \
    474516               AND batch_type = '" + batchType + "' \
    475                AND dvo_db = '" + dvoGpc1Label + "' \
    476                AND timestamp BETWEEN now() - INTERVAL 4 HOUR AND now()"
     517               AND dvo_db = '" + self.config.dvoLabel + "' \
     518               AND batch_type != 'IN' \
     519               AND timestamp BETWEEN now() - INTERVAL 12 HOUR AND now()"
    477520
    478521        try:
    … …  
    480523            rs.first()
    481524            if rs.getInt(1) > 0:
    482                 self.logger.errorPair(str(stage_id) + " is already being processed", "skipping")
     525                self.logger.debugPair(str(stage_id) + " is already being processed", "skipping")
    483526                return True
    484527            else:
    … …  
    509552    Have we already processed this stage_id?
    510553    '''
    511     def alreadyProcessed(self, batchType, stage_id, epoch, dvoGpc1Label):
     554    def alreadyProcessed(self, batchType, stage_id):
    512555
    513556        sql = "SELECT COUNT(*) \
    … …  
    515558               WHERE stage_id = " + str(stage_id) + " \
    516559               AND batch_type = '" + batchType + "' \
    517                AND timestamp > '" + epoch + "' \
    518                AND dvo_db = '" + dvoGpc1Label + "' \
     560               AND timestamp > '" + self.config.epoch + "' \
     561               AND dvo_db = '" + self.config.dvoLabel + "' \
     562               AND batch_type != 'IN' \
    519563               AND processed = 1"
    520564
    … …  
    523567            rs.first()
    524568            if rs.getInt(1) > 0:
    525                 self.logger.errorPair(str(stage_id) + " has already been published", "skipping")
     569                self.logger.debugPair(str(stage_id) + " has already been published", "skipping")
    526570                return True
    527571            else:
    … …  
    533577    Has this stage_id consistently failed to process?
    534578    '''
    535     def consistentlyFailed(self, batchType, stage_id, epoch, dvoGpc1Label):
     579    def consistentlyFailed(self, batchType, stage_id):
    536580
    537581        sql = "SELECT COUNT(*) \
    … …  
    539583               WHERE stage_id = " + str(stage_id) + " \
    540584               AND batch_type = '" + batchType + "' \
    541                AND timestamp > '" + epoch + "' \
    542                AND dvo_db = '" + dvoGpc1Label + "' \
     585               AND timestamp > '" + self.config.epoch + "' \
     586               AND dvo_db = '" + self.config.dvoLabel + "' \
    543587               AND processed = -1"
    544588
    … …  
    588632
    589633    '''
    590     Creates a new batch. This is done with a locked table in order that no two clients can start work on the same item
    591     '''
    592     def createNewBatch(self, batchType, stageID, survey, epoch, dvoDb, datastoreProduct, force):
     634    Creates a new batch. This is done with a locked table in order that no two clients can start work on the same item, i.e. this is a crital section
     635    '''
     636    def createNewBatch(self, batchType, stageID):
    593637
    594638        batchID = -1;
    595639
    596         if force or \
    597             (not self.alreadyProcessed(batchType, stageID, epoch, dvoDb) \
    598             and not self.processingNow(batchType, stageID, epoch, dvoDb) \
    599             and not self.consistentlyFailed(batchType, stageID, epoch, dvoDb)):
     640        if self.config.force or \
     641            (not self.alreadyProcessed(batchType, stageID) \
     642            and not self.processingNow(batchType, stageID) \
     643            and not self.consistentlyFailed(batchType, stageID)):
    600644
    601645            sql = "INSERT INTO batch ( \
    … …  
    608652                       '" + batchType + "', \
    609653                       " + str(stageID) + ", \
    610                        '" + survey + "', \
    611                        '" + dvoDb + "', \
    612                        '" + datastoreProduct + "' \
     654                       '" + self.config.survey + "', \
     655                       '" + self.config.dvoLabel + "', \
     656                       '" + self.config.datastoreProduct + "' \
    613657                       )"
    614658
    … …  
    670714               loaded_to_datastore = 0, \
    671715               loaded_to_ODM = 0, \
     716               merge_worthy = 0, \
     717               merged = 0, \
    672718               deleted_datastore = 0, \
    673719               deleted_dxlayer = 0, \
    … …  
    678724
    679725    '''
     726    Pause loading clients
     727    '''
     728    def pauseLoaders(self, bool, ids):
     729        self.setColForLoaders("paused", bool, ids)
     730
     731    '''
     732    Kill loading clients
     733    '''
     734    def killLoaders(self, bool, ids):
     735        self.setColForLoaders("killed", bool, ids)
     736
     737    '''
     738    Sets the config field for a set of loader clients
     739    '''
     740    def setConfigForLoaders(self, config, ids):
     741
     742        for id in ids:
     743            self.execute("UPDATE clients SET config = '" + config + "' \
     744                    WHERE type = 'loader.py' \
     745                    AND id = " + str(id))
     746
     747    '''
     748    Sets the value for a given column for a set of loader clients
     749    '''
     750    def setColForLoaders(self, col, bool, ids):
     751
     752        if bool: value = 1
     753        else: value = 0
     754
     755        for id in ids:
     756            self.execute("UPDATE clients SET " + col + " = " + str(value) + " \
     757                    WHERE type = 'loader.py' \
     758                    AND id = " + str(id))
     759
     760    '''
     761    Remove client by id
     762    '''
     763    def removeClientsByIds(self, ids):
     764        for id in ids:
     765            self.execute("DELETE FROM clients WHERE id = " + str(id))
     766
     767    '''
     768    Remove client
     769    '''
     770    def removeClient(self, host, pid):
     771        self.execute("DELETE FROM clients WHERE host = '" + host + "' AND pid = " + str(pid))
     772
     773    '''
     774    Insert client
     775    '''
     776    def insertClient(self, type, host, pid):
     777        self.execute("INSERT INTO clients (timestamp, type, host, pid, config) VALUES (now(), '" + type + "', '" + host + "', " + str(pid) + ", '" + self.config.name + "')")
     778
     779    '''
     780    Update clients, or inserts it for the first time
     781    '''
     782    def updateClient(self, type, host, pid):
     783
     784        try:
     785            self.insertClient(type, host, pid)
     786        except:
     787            self.execute("UPDATE clients SET timestamp = now() WHERE host = '" + host + "' AND pid = " + str(pid))
     788
     789    '''
     790    Is this client paused?
     791    '''
     792    def isPaused(self, host, pid):
     793        return self.getClientsColumnBoolean("paused", host, pid)
     794
     795    '''
     796    Is this client killed?
     797    '''
     798    def isKilled(self, host, pid):
     799        return self.getClientsColumnBoolean("killed", host, pid)
     800
     801    '''
     802    Returns boolean value for this column in clients table
     803    '''
     804    def getClientsColumnBoolean(self, col, host, pid):
     805
     806        sql = "SELECT " + col + " \
     807               FROM clients \
     808               WHERE host = '" + host + "' AND pid = " + str(pid)
     809
     810        try:
     811            rs = self.executeQuery(sql)
     812            rs.first()
     813            if rs.getInt(1) == 1: return True
     814            else: return False
     815        except:
     816            self.logger.exception("Unable to check whether " + col + " is true or false in clients table")
     817
     818        return False
     819
     820    '''
     821    Returns a list of available configs
     822    '''
     823    def getConfigList(self):
     824       
     825        sql = "SELECT DISTINCT name FROM config"
     826
     827        configs = []
     828        try:
     829            rs = self.executeQuery(sql)
     830            while (rs.next()): configs.append(rs.getString(1))
     831        except:
     832            self.logger.errorPair("Can't get config names", sql)
     833        finally:
     834            rs.close()
     835
     836        return configs
     837
     838
     839    '''
     840    Writes current Config object to Db
     841    '''
     842    def writeConfig(self):
     843
     844        sql = "UPDATE config \
     845               SET \
     846               datastore_product = '" + self.config.datastoreProduct + "' \
     847               ,datastore_type = '" + self.config.datastoreType + "' \
     848               ,datastore_publish = " + self.config.datastorePublishing + " \
     849               ,dvo_label = '" + self.config.dvoLabel + "' \
     850               ,dvo_location = '" + self.config.dvoLocation + "' \
     851               ,min_ra = " + self.config.minRa + " \
     852               ,max_ra = " + self.config.maxRa + " \
     853               ,min_dec = " + self.config.minDec + " \
     854               ,max_dec = " + self.config.maxDec + " \
     855               ,box_size = " + self.config.boxSize + " \
     856               ,base_path = '" + self.config.basePath + "' \
     857               ,data_release = " + self.config.dataRelease + " \
     858               ,delete_local = " + self.config.deleteLocal + " \
     859               ,delete_datastore = " + self.config.deleteDatastore + " \
     860               ,delete_dxlayer = " + self.config.deleteDxLayer + " \
     861               ,epoch = '" + self.config.epoch + "' \
     862               ,survey = '" + self.config.survey + "' \
     863               ,psps_survey = '" + self.config.pspsSurvey + "' \
     864               ,queue_P2 = " + self.config.queuingThisBatchType("P2") + " \
     865               ,queue_ST = " + self.config.queuingThisBatchType("ST") + " \
     866               WHERE name = '" + self.config.name + "'"
     867
     868        self.execute(sql)
     869
     870    '''
     871    Reads config from the database and populates Config object
     872    '''
     873    def readConfig(self, host, pid):
     874
     875        # first get config defined for this client
     876        sql = "SELECT config \
     877               FROM clients \
     878               WHERE host = '" + host + "' \
     879               AND pid = " + str(pid)
     880
     881        try:
     882            rs = self.executeQuery(sql)
     883            rs.first()
     884            self.config.name = rs.getString(1)
     885        except:
     886            self.logger.errorPair("No config set for", "%s (pid=%d)" % (host, pid))
     887            self.config.isLoaded = False
     888            return False
     889
     890        # now load that config
     891        sql = "SELECT \
     892        datastore_product \
     893        ,datastore_type \
     894        ,datastore_publish \
     895        ,dvo_label \
     896        ,dvo_location \
     897        ,min_ra \
     898        ,max_ra \
     899        ,min_dec \
     900        ,max_dec \
     901        ,box_size \
     902        ,base_path \
     903        ,data_release \
     904        ,delete_local \
     905        ,delete_datastore \
     906        ,delete_dxlayer \
     907        ,epoch \
     908        ,survey \
     909        ,psps_survey \
     910        ,queue_P2 \
     911        ,queue_ST \
     912        FROM config \
     913        WHERE name = '" + self.config.name + "'"
     914
     915        self.config.batchTypes = []
     916        try:
     917            rs = self.executeQuery(sql)
     918            rs.first()
     919            self.config.datastoreProduct = rs.getString(1)
     920            self.config.datastoreType = rs.getString(2)
     921            if rs.getInt(3) == 1: self.config.datastorePublishing = True
     922            else: self.config.datastorePublishing = False
     923            self.config.dvoLabel = rs.getString(4)
     924            self.config.dvoLocation = rs.getString(5)
     925            self.config.minRa = rs.getDouble(6)
     926            self.config.maxRa = rs.getDouble(7)
     927            self.config.minDec = rs.getDouble(8)
     928            self.config.maxDec = rs.getDouble(9)
     929
     930            self.config.boxSize = rs.getDouble(10)
     931            self.config.halfBox = self.config.boxSize/2.0
     932            self.config.boxSizeWithBorder = self.config.boxSize + (self.config.BORDER * 2)
     933
     934            self.config.basePath = rs.getString(11)
     935            self.config.dataRelease = rs.getInt(12)
     936            self.config.deleteLocal = rs.getInt(13)
     937            self.config.deleteDatastore = rs.getInt(14)
     938            self.config.deleteDxLayer = rs.getInt(15)
     939            self.config.epoch = rs.getString(16)
     940            self.config.survey = rs.getString(17)
     941            self.config.pspsSurvey = rs.getString(18)
     942            if rs.getInt(19) == 1: self.config.batchTypes.append("P2")
     943            if rs.getInt(20) == 1: self.config.batchTypes.append("ST")
     944            self.config.force = False # TODO
     945            self.config.test = False # TODO
     946            self.config.isLoaded = True
     947        except:
     948            self.logger.errorPair("Could not read config with name", self.config.name)
     949            self.config.isLoaded = False
     950
     951        return self.config.isLoaded
     952
     953    '''
     954    Gets id of box with these coords
     955    '''
     956    def getBoxId(self, ra, dec):
     957
     958        sql = "SELECT id FROM box \
     959               WHERE config  = '" + self.config.name + "' \
     960               AND ra_center = " + str(ra) + " \
     961               AND dec_center = " + str(dec) + " \
     962               AND box_side = " + str(self.config.boxSize)
     963
     964        id = -1
     965        try:
     966            rs = self.executeQuery(sql)
     967            rs.first()
     968            id = rs.getInt(1)
     969        except:
     970            pass
     971
     972        return id
     973
     974    '''
     975    Remove all boxes and associated ids in pending table
     976    '''
     977    def removeAllBoxes(self):
     978        self.execute("DELETE FROM box WHERE config = '" + self.config.name + "'")
     979
     980    '''
     981    Inserts new box into box table. If it already exisits, it returns existing box id
     982    '''
     983    def insertBox(self, ra, dec):
     984
     985        id = self.getBoxId(ra, dec)
     986
     987        # does not already exists so insert
     988        if id < 0:
     989
     990            sql = "INSERT INTO box \
     991                   (config, ra_center, dec_center, box_side) \
     992                   VALUES \
     993                   ('" + self.config.name + "', " + str(ra) + ", " + str(dec) + ", " + str(self.config.boxSize) + ")"
     994
     995            self.execute(sql)
     996            id = self.getBoxId(ra, dec)
     997
     998        return id
     999
     1000    '''
     1001    Inserts new box ids into the pending table
     1002    '''
     1003    def insertPending(self, box_id, batchType, ids):
     1004
     1005        # first delete old pending items
     1006        self.execute("DELETE FROM pending \
     1007                WHERE box_id = " + str(box_id) + " \
     1008                AND batch_type = '" + batchType + "'")
     1009
     1010        for id in ids:
     1011            sql = "INSERT INTO pending \
     1012               (box_id, batch_type, stage_id) \
     1013               VALUES \
     1014               (" + str(box_id) + ", '" + batchType + "', " + str(id) + ")"
     1015
     1016            self.execute(sql)
     1017
     1018    '''
     1019    Returns ids for all boxes with pending items in a stripe of RA not already being processed
     1020    by another client. If none are available, then it uses the stripe already used by another
     1021    client that has the most pending items
     1022    '''
     1023    def getStripeBoxIds(self, host, pid, batchType):
     1024
     1025        self.execute("LOCK TABLES box READ, pending READ, stripe WRITE, clients WRITE")
     1026
     1027        ids = []
     1028        raCenter = -1.0
     1029        sql = "SELECT ra_center \
     1030               FROM box \
     1031               JOIN pending ON (id = box_id) \
     1032               WHERE config = '" + self.config.name + "' \
     1033               AND batch_type = '" + batchType + "' \
     1034               AND ra_center NOT IN \
     1035               (SELECT ra_center \
     1036                FROM stripe \
     1037                WHERE config = '" + self.config.name + "') \
     1038               GROUP BY ra_center LIMIT 1"
     1039
     1040        try:
     1041            rs = self.executeQuery(sql)
     1042            rs.first()
     1043            raCenter = rs.getFloat(1)
     1044            rs.close()
     1045        except:
     1046            self.logger.errorPair("No free stripes", "trying to share with other client...")
     1047            sql = "SELECT ra_center, COUNT(*) AS numPending\
     1048                   FROM box \
     1049                   JOIN pending ON (id = box_id) \
     1050                   WHERE config = '" + self.config.name + "' \
     1051                   AND batch_type = '" + batchType + "' \
     1052                   GROUP BY ra_center \
     1053                   ORDER BY numPending \
     1054                   DESC LIMIT 1"
     1055
     1056            try:
     1057                rs = self.executeQuery(sql)
     1058                rs.first()
     1059                raCenter = rs.getFloat(1)
     1060                rs.close()
     1061            except:
     1062                self.logger.errorPair("No available stripes to share", "no stripe obtained")
     1063
     1064        self.logger.infoPair("Using RA stripe", "%f" % raCenter)
     1065        sql = "SELECT DISTINCT id \
     1066               FROM box \
     1067               JOIN pending ON (id = box_id) \
     1068               WHERE ra_center = " + str(raCenter) + " \
     1069               AND batch_type = '" + batchType + "' \
     1070               AND config = '" + self.config.name + "' \
     1071               ORDER BY dec_center DESC"
     1072
     1073        try:
     1074            rs = self.executeQuery(sql)
     1075            while (rs.next()): ids.append(rs.getInt(1))
     1076            rs.close()
     1077        except:
     1078            self.logger.errorPair("Can't get stripe boxes", sql)
     1079
     1080        self.registerStripe(host, pid, raCenter)
     1081
     1082        self.unlockTables()
     1083
     1084        return ids
     1085
     1086    '''
     1087    Inserts row into strip table
     1088    '''
     1089    def registerStripe(self, host, pid, raCenter):
     1090       
     1091        # first, delete any existing record
     1092        try:
     1093            self.execute("DELETE FROM stripe \
     1094                    WHERE client_id = \
     1095                    (SELECT id FROM clients \
     1096                     WHERE host = '" + host + "' AND pid = " + str(pid) + ")")
     1097        except:
     1098            self.logger.errorPair("Can't delete from stripe table", sql)
     1099
     1100
     1101        # now insert new stripe entry
     1102        sql = "INSERT INTO stripe \
     1103               (client_id, config, ra_center) \
     1104               SELECT clients.id, clients.config, " + str(raCenter) + " \
     1105               FROM clients \
     1106               WHERE host = '" + host + "' \
     1107               AND pid = " + str(pid)
     1108        try:
     1109            self.execute(sql)
     1110        except:
     1111            self.logger.errorPair("Can't insert into stripe", sql)
     1112
     1113
     1114    '''
     1115    Returns dimensions of this box
     1116    '''
     1117    def getBoxDimensions(self, boxId):
     1118
     1119        sql = "SELECT ra_center, dec_center, box_side FROM box WHERE id = " + str(boxId)
     1120
     1121        boxDim = {}
     1122        rs = self.executeQuery(sql)
     1123        rs.first()
     1124
     1125        boxDim['RA'] = rs.getFloat(1)
     1126        boxDim['DEC'] = rs.getFloat(2)
     1127        boxDim['SIDE'] = rs.getFloat(3)
     1128
     1129        return boxDim
     1130
     1131    '''
     1132    Returns ids for pending items for this config
     1133    '''
     1134    def getPendingIds(self):
     1135
     1136        sql = "SELECT DISTINCT stage_id FROM pending JOIN box ON (box_id = id) WHERE config = '" + self.config.name + "'"
     1137
     1138        ids = []
     1139        try:
     1140            rs = self.executeQuery(sql)
     1141            while (rs.next()): ids.append(rs.getInt(1))
     1142            rs.close()
     1143        except:
     1144            self.logger.errorPair("Can't get pending ids for this config", sql)
     1145
     1146        return ids
     1147
     1148    '''
     1149    Returns ids for pending items in this box for this batch type
     1150    '''
     1151    def getPendingIdsForThisBox(self, boxId, batchType):
     1152
     1153        sql = "SELECT stage_id FROM pending WHERE batch_type = '" + batchType + "' AND box_id = " + str(boxId)
     1154
     1155        ids = []
     1156        try:
     1157            rs = self.executeQuery(sql)
     1158            while (rs.next()): ids.append(rs.getInt(1))
     1159            rs.close()
     1160        except:
     1161            self.logger.errorPair("Can't get ids from this box", sql)
     1162
     1163        return ids
     1164
     1165    '''
     1166    Deletes an item from the pending table
     1167    '''
     1168    def deletePendingItem(self, batchType, stageID):
     1169
     1170        sql = "DELETE pending \
     1171               FROM pending \
     1172               JOIN box ON (id = box_id) \
     1173               WHERE stage_id = " + str(stageID) + " \
     1174               AND batch_type = '" + batchType + "' \
     1175               AND config = '" + self.config.name + "'"
     1176
     1177        self.execute(sql)
     1178
     1179    '''
     1180    Writes density-plot data to file for pending stuff over sky for given config
     1181    '''
     1182    def createPendingDensityPlotData(self, batchType, DATFILE):
     1183
     1184        sql = "SELECT id, ra_center, dec_center \
     1185               FROM box \
     1186               WHERE config = '" + self.config.name + "' \
     1187               ORDER BY ra_center, dec_center"
     1188
     1189        maxPending = 0
     1190        try:
     1191            rs = self.executeQuery(sql)
     1192            while (rs.next()):
     1193               
     1194                boxId = rs.getInt(1)
     1195
     1196                sql = "SELECT COUNT(stage_id) \
     1197                       FROM pending \
     1198                       WHERE box_id = " + str(boxId) + " \
     1199                       AND batch_type = '" + batchType + "'"
     1200                rs2 = self.executeQuery(sql)
     1201                rs2.first()
     1202                max = rs2.getInt(1)
     1203                rs2.close()
     1204                print >> DATFILE, rs.getFloat(2), rs.getFloat(3), max
     1205                if max > maxPending: maxPending = max
     1206            rs.close()
     1207        except:
     1208            self.logger.errorPair("Can't get density plot data", sql)
     1209
     1210        return maxPending
     1211
     1212    '''
     1213    Creates a temporary table and shoves it full of stage_ids and ra/dec coords
     1214    '''
     1215    def storeAllItems(self, rows):
     1216
     1217        try:
     1218            self.execute("DROP TABLE all_pending")
     1219        except:
     1220            pass
     1221
     1222        self.execute("CREATE TEMPORARY TABLE all_pending (stage_id bigint(20), ra_bore float, dec_bore float)")
     1223        count = 0
     1224        for row in rows:
     1225
     1226            try:
     1227                sql = "INSERT INTO all_pending (stage_id, ra_bore, dec_bore) \
     1228                       VALUES (%d, %f, %f)" % (row[0], row[1], row[2])
     1229                self.execute(sql)
     1230                count += 1
     1231            except: continue
     1232
     1233        self.logger.infoPair("All items written to Db", "%d" % count)
     1234
     1235    '''
     1236    Gets all items in the all_pending temporary table within the bounds of this box
     1237    '''
     1238    def getItemsInThisThisBox(self, ra, dec):
     1239
     1240        halfSide = self.config.boxSize/2.0
     1241        minRa =  ra-halfSide
     1242        maxRa = ra+halfSide
     1243        minDec = dec-halfSide
     1244        maxDec = dec+halfSide
     1245
     1246        ids = []
     1247
     1248        sql = "SELECT DISTINCT stage_id \
     1249               FROM all_pending \
     1250               WHERE ra_bore BETWEEN " + str(minRa) + " AND " + str(maxRa) + " \
     1251               AND dec_bore BETWEEN " + str(minDec) + " AND " + str(maxDec)
     1252
     1253        try:
     1254            rs = self.executeQuery(sql)
     1255            while (rs.next()): ids.append(rs.getInt(1))
     1256        except:
     1257            self.logger.errorPair("Can't get items in this box", sql)
     1258
     1259        return ids
     1260
     1261    '''
     1262    Gets client info
     1263    '''
     1264    def getClientInfo(self):
     1265
     1266        sql = "SELECT id, host, started, timestamp, ra_center, paused, killed \
     1267               FROM stripe \
     1268               JOIN clients ON (client_id = id) \
     1269               WHERE clients.config = '" + self.config.name + "' \
     1270               ORDER BY host, started"
     1271
     1272        rows = []
     1273        try:
     1274            rs = self.executeQuery(sql)
     1275            while (rs.next()):
     1276                rows.append([rs.getInt(1), rs.getString(2), rs.getString(3), rs.getString(4), rs.getFloat(5), rs.getInt(6), rs.getInt(7)])
     1277            rs.close()
     1278        except:
     1279            self.logger.errorPair("Can't query for client data", sql)
     1280
     1281        return rows
     1282
     1283    '''
     1284    Purges old clients that may have die, i.e. those with a timestamp older than n hours
     1285    '''
     1286    def purgeDeadClients(self):
     1287        self.execute("DELETE FROM clients WHERE timestamp <  now() -  INTERVAL 6 HOUR")
     1288
     1289    '''
    6801290    Destructor
    6811291    '''
    6821292    def __del__(self):
    6831293
    684         self.logger.debug("IppToPspsDb Desstructor")
     1294        self.logger.debug("IppToPspsDb Destructor")
    6851295        self.stmt.close()
    6861296        self.con.close()
Note: See TracChangeset for help on using the changeset viewer.