IPP Software Navigation Tools IPP Links Communication Pan-STARRS Links

Ignore:
Timestamp:
Nov 8, 2011, 2:44:12 PM (15 years ago)
Author:
eugene
Message:

merge changes from trunk

Location:
branches/eam_branches/ipp-20110906
Files:
2 edited

Legend:

Unmodified
Added
Removed

  • branches/eam_branches/ipp-20110906

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

    r32362 r32630  
    2020        super(IppToPspsDb, self).__init__(logger, doc, "ipptopspsdatabase")
    2121
    22     '''
    23     Creates a new batch
    24     '''
    25     def createNewBatch(self, batchType, stageID, survey, dvoDb, datastoreProduct):
    26 
    27         sql = "INSERT INTO batch ( \
    28                batch_type, \
    29                stage_id, \
    30                survey, \
    31                dvo_db, \
    32                datastore_product \
    33                ) VALUES ( \
    34                '" + batchType + "', \
    35                " + str(stageID) + ", \
    36                '" + survey + "', \
    37                '" + dvoDb + "', \
    38                '" + datastoreProduct + "' \
    39                )"
    40 
    41         self.execute(sql)
    42 
    43         sql = "SELECT MAX(batch_id) FROM batch"
    44 
    45         batchID = -1;
    46 
    47         try:
    48             rs = self.executeQuery(sql)
    49             rs.first()
    50             batchID = rs.getInt(1)
    51         except:
    52             self.logger.exception("Unable to get batch ID")
    53 
    54         self.logger.debug("Created new batch in ippToPsps database with batchID = %d" % batchID)
    55 
    56         return batchID;
    57 
    58     '''
    59     Returns a list of processed batch IDs that are merged but not yet deleted
     22        self.MAX_FAILS = 4
     23
     24    '''
     25    Returns a list of merged batch IDs that are merged but not yet deleted
     26    '''
     27    def getMergedButNotDeletedBatchIDs(self, batchType, epoch, dvoGpc1Label, column):
     28
     29        sql = "SELECT DISTINCT batch_id \
     30               FROM batch \
     31               WHERE timestamp > '" + epoch + "' \
     32               AND batch_type = '" + batchType + "' \
     33               AND dvo_db = '" + dvoGpc1Label + "' \
     34               AND merged = 1 \
     35               AND " + column + " = 0"
     36
     37        ids = []
     38        try:
     39            rs = self.executeQuery(sql)
     40            while (rs.next()):
     41                ids.append(rs.getInt(1))
     42        except:
     43            self.logger.exception("Can't query for merged batch ids in ipptopsps Db")
     44
     45        rs.close()
     46
     47        self.logger.debug("Found %d merged but un-deleted items" % len(ids))
     48
     49        return ids
     50
     51    '''
     52    Returns a list of processed batch IDs that have been loaded to the ODM but not yet deleted
    6053    '''
    6154    def getLoadedToODMButNotDeletedBatchIDs(self, batchType, epoch, dvoGpc1Label, column):
    … …  
    6659               AND batch_type = '" + batchType + "' \
    6760               AND dvo_db = '" + dvoGpc1Label + "' \
    68                AND loaded_to_ODM != 0 \
     61               AND (loaded_to_ODM = -1 OR merge_worthy = 1) \
    6962               AND " + column + " = 0"
    7063
    … …  
    8477
    8578    '''
    86     Returns a list of processed batch IDs that are merged but not deleted from local disk
     79    Returns a list of merged batch IDs that are not deleted from local disk
     80    '''
     81    def getMergedButNotDeletedFromLocalDisk(self, batchType, epoch, dvoGpc1Label):
     82        return self.getMergedButNotDeletedBatchIDs(batchType, epoch, dvoGpc1Label, "deleted_local")
     83
     84    '''
     85    Returns a list of processed batch IDs that are loaded to the ODM, but not deleted from local disk
    8786    '''
    8887    def getLoadedToODMButNotDeletedFromLocalDisk(self, batchType, epoch, dvoGpc1Label):
    … …  
    9089
    9190    '''
    92     Returns a list of processed batch IDs that are merged but not deleted from datastore
     91    Returns a list of processed batch IDs that are loaded to the ODM, but not deleted from datastore
    9392    '''
    9493    def getLoadedToODMButNotDeletedFromDatastore(self, batchType, epoch, dvoGpc1Label):
    … …  
    9695
    9796    '''
    98     Returns a list of processed batch IDs that are merged but not deleted from DXLayer
     97    Returns a list of processed batch IDs that are loaded to the ODM, but not deleted from DXLayer
    9998    '''
    10099    def getLoadedToODMButNotDeletedFromDXLayer(self, batchType, epoch, dvoGpc1Label):
    … …  
    130129
    131130    '''
    132     Returns a list of processed batch IDs that have not yet been loaded to the ODM
    133     '''
    134     def getBatchIDsUnloadedToODM(self, epoch, dvoGpc1Label):
     131    Returns a list of processed batch IDs that either not yet loaded to the ODM or not merge_worthy or not merged
     132    NOTE we exclude batches already deleted from the datastore, as sometimes we delete stuff prior to loading to ODM if there is a known problem with the batch
     133    '''
     134    def getAllUnfinishedBatchIDs(self, epoch, dvoGpc1Label, batchType):
     135
     136        sql = "SELECT batch_id \
     137               FROM batch \
     138               WHERE timestamp > '" + epoch + "' \
     139               AND dvo_db = '" + dvoGpc1Label + "' \
     140               AND loaded_to_datastore = 1 \
     141               AND batch_type = '" + batchType + "' \
     142               AND deleted_datastore = 0 \
     143               AND loaded_to_ODM != -1 \
     144               AND merge_worthy != -1 \
     145               AND merged = 0"
     146
     147        ids = []
     148        try:
     149            rs = self.executeQuery(sql)
     150            while (rs.next()):
     151                ids.append(rs.getInt(1))
     152        except:
     153            self.logger.exception("Can't query for unloaded batch ids in ipptopsps Db")
     154
     155        rs.close()
     156
     157        self.logger.debug("Found %d unloaded items" % len(ids))
     158
     159        return ids
     160
     161    '''
     162    Returns a list of processed batch IDs that are not yet loaded to the ODM
     163    NOTE we exclude batches already deleted from the datastore, as sometimes we delete stuff prior to loading to ODM if there is a known problem with the batch
     164    '''
     165    def getUnloadedBatchIDs(self, epoch, dvoGpc1Label, batchType):
     166
     167        sql = "SELECT batch_id \
     168               FROM batch \
     169               WHERE timestamp > '" + epoch + "' \
     170               AND dvo_db = '" + dvoGpc1Label + "' \
     171               AND loaded_to_datastore = 1 \
     172               AND batch_type = '" + batchType + "' \
     173               AND loaded_to_ODM = 0 \
     174               AND deleted_datastore = 0"
     175
     176        ids = []
     177        try:
     178            rs = self.executeQuery(sql)
     179            while (rs.next()):
     180                ids.append(rs.getInt(1))
     181        except:
     182            self.logger.exception("Can't query for unloaded batch ids in ipptopsps Db")
     183
     184        rs.close()
     185
     186        self.logger.debug("Found %d unloaded items" % len(ids))
     187
     188        return ids
     189
     190    '''
     191    Returns a list of processed batch IDs that are not yet merge_worthy and that have not failed to load
     192    NOTE we exclude batches already deleted from the datastore, as sometimes we delete stuff prior to loading to ODM if there is a known problem with the batch
     193    '''
     194    def getUnmergeWorthyBatchIDs(self, epoch, dvoGpc1Label, batchType):
     195
     196        sql = "SELECT batch_id \
     197               FROM batch \
     198               WHERE timestamp > '" + epoch + "' \
     199               AND dvo_db = '" + dvoGpc1Label + "' \
     200               AND loaded_to_datastore = 1 \
     201               AND batch_type = '" + batchType + "' \
     202               AND merge_worthy = 0 \
     203               AND loaded_to_ODM != -1 \
     204               AND deleted_datastore = 0"
     205
     206        ids = []
     207        try:
     208            rs = self.executeQuery(sql)
     209            while (rs.next()):
     210                ids.append(rs.getInt(1))
     211        except:
     212            self.logger.exception("Can't query for un-mergeworthy batch ids in ipptopsps Db")
     213
     214        rs.close()
     215
     216        self.logger.debug("Found %d unloaded items" % len(ids))
     217
     218        return ids
     219
     220    '''
     221    Returns a list of processed batch IDs that are merge_worthy, but not yet merged, for this epoch and dvo label
     222    '''
     223    def getUnmergedBatchIDs(self, epoch, dvoGpc1Label, batchType):
     224
     225        sql = "SELECT batch_id \
     226               FROM batch \
     227               WHERE timestamp > '" + epoch + "' \
     228               AND dvo_db = '" + dvoGpc1Label + "' \
     229               AND merge_worthy = 1 \
     230               AND batch_type = '" + batchType + "' \
     231               AND merged != 1"
     232
     233        ids = []
     234        try:
     235            rs = self.executeQuery(sql)
     236            while (rs.next()):
     237                ids.append(rs.getInt(1))
     238        except:
     239            self.logger.exception("Can't query for unmerged batch ids in ipptopsps Db")
     240
     241        rs.close()
     242
     243        self.logger.debug("Found %d un-merged items" % len(ids))
     244
     245        return ids
     246
     247    '''
     248    Returns a list of batch IDs for this epoch and dvo label for which the specified column is set to value
     249    '''
     250    def getBatchIDs(self, epoch, dvoGpc1Label, column, value):
    135251
    136252        sql = "SELECT DISTINCT batch_id \
    … …  
    138254               WHERE timestamp > '" + epoch + "' \
    139255               AND dvo_db = '" + dvoGpc1Label + "' \
    140                AND loaded_to_datastore = 1 \
    141                AND loaded_to_ODM = 0"
    142 
    143 
    144         ids = []
    145         try:
    146             rs = self.executeQuery(sql)
    147             while (rs.next()):
    148                 ids.append(rs.getInt(1))
    149         except:
    150             self.logger.exception("Can't query for unloaded batch ids in ipptopsps Db")
    151 
    152         rs.close()
    153 
    154         self.logger.debug("Found %d unloaded items" % len(ids))
    155 
    156         return ids
    157 
    158     '''
    159     Returns a list of processed batch IDs that have not failed to load and are not yet merged, for this epoch and dvo label
    160     '''
    161     def getUnfinishedBatchIDs(self, epoch, dvoGpc1Label):
    162 
    163         sql = "SELECT DISTINCT batch_id \
    164                FROM batch \
    165                WHERE timestamp > '" + epoch + "' \
    166                AND dvo_db = '" + dvoGpc1Label + "' \
    167                AND loaded_to_datastore = 1 \
    168                AND merged != 1 \
    169                AND loaded_to_ODM != -1"
    170 
    171 
    172         ids = []
    173         try:
    174             rs = self.executeQuery(sql)
    175             while (rs.next()):
    176                 ids.append(rs.getInt(1))
    177         except:
    178             self.logger.exception("Can't query for unfinished batch ids in ipptopsps Db")
    179 
    180         rs.close()
    181 
    182         self.logger.debug("Found %d unfinished items" % len(ids))
    183 
    184         return ids
    185 
    186     '''
    187     Returns a list of batch IDs for this epoch and dvo label for which the specified column is set to value
    188     '''
    189     def getBatchIDs(self, epoch, dvoGpc1Label, column, value):
    190 
    191         sql = "SELECT DISTINCT batch_id \
    192                FROM batch \
    193                WHERE timestamp > '" + epoch + "' \
    194                AND dvo_db = '" + dvoGpc1Label + "' \
    195256               AND " + column + " = " + str(value)
    196257
    … …  
    208269
    209270        return ids
     271
     272    '''
     273    Returns the total detections published to the datastore for this epoch, dvo label and batch type
     274    '''
     275    def getTotalDetectionsPublished(self, batchType, epoch, dvoGpc1Label):
     276
     277        sql = "SELECT SUM(total_detections) \
     278               FROM batch \
     279               WHERE batch_type = '" + batchType + "' \
     280               AND timestamp > '" + epoch + "' \
     281               AND dvo_db = '" + dvoGpc1Label + "' \
     282               AND processed = 1 \
     283               AND loaded_to_datastore = 1"
     284
     285        total = -1
     286        try:
     287            rs = self.executeQuery(sql)
     288            rs.first()
     289            total =  rs.getLong(1)
     290        except:
     291            self.logger.exception("Can't query for total detections published")
     292
     293        rs.close()
     294
     295        self.logger.debug("Found %d detections" % total)
     296
     297        return total
    210298
    211299    '''
    … …  
    273361    '''
    274362    def getStages(self):
    275         return ['processed', 'loaded_to_datastore', 'loaded_to_ODM', 'merge_worthy', 'merged', 'deleted_datastore', 'deleted_dxlayer', 'deleted_local']
    276 
    277     '''
    278     Returns batches per hour. Simply goes back in time for the specified number of hours, so will return 0.0 if not loading
    279     '''
    280     def getBatchesPerHour(self, batchType, epoch, dvoGpc1Label, hours):
    281 
    282         sql = "SELECT COUNT(batch_id)/ " + str(hours) + " \
     363        return ['processed', 'loaded_to_datastore', 'loaded_to_ODM', 'merge_worthy', 'deleted_datastore', 'deleted_dxlayer', 'merged', 'deleted_local' ]
     364
     365    '''
     366    Returns a count of batches processed in the specified time period.
     367    '''
     368    def countBatchesInLastPeriod(self, batchType, epoch, dvoGpc1Label, interval):
     369
     370        sql = "SELECT COUNT(batch_id) \
    283371               FROM batch \
    284372               WHERE batch_type = '" + batchType + "' \
    … …  
    286374               AND timestamp > '" + epoch + "' \
    287375               AND dvo_db = '" + dvoGpc1Label + "' \
    288                AND timestamp > (now() - INTERVAL " +  str(hours) + " HOUR)"
    289 
    290         try:
    291             rs = self.executeQuery(sql)
    292             if rs.first(): return rs.getFloat(1)
    293         except:
    294             self.logger.exception("Unable to get batches per hour")
    295             return 0.0
     376               AND timestamp BETWEEN now() - INTERVAL " + interval + " AND now()"
     377
     378        try:
     379            rs = self.executeQuery(sql)
     380            if rs.first(): return rs.getInt(1)
     381        except:
     382            self.logger.exception("Unable to count batches in interval")
     383            return 0
    296384
    297385    '''
    … …  
    301389    def getTimeOfLastBatchPublished(self, batchType, epoch, dvoGpc1Label):
    302390
    303         minutes = None
    304 
    305         sql = "SELECT TIMESTAMPDIFF(MINUTE, timestamp, now()) \
     391        seconds = None
     392
     393        sql = "SELECT TIMESTAMPDIFF(SECOND, timestamp, now()) \
    306394               FROM batch \
    307395               WHERE batch_type = '" + batchType + "' \
    … …  
    313401        try:
    314402            rs = self.executeQuery(sql)
    315             if rs.first(): minutes = rs.getFloat(1)
     403            if rs.first(): seconds = rs.getFloat(1)
    316404        except:
    317405            self.logger.exception("Unable to get last batch published")
    318406
    319         if not minutes: return "Never"
    320 
     407        if not seconds: return "Never"
     408
     409        minutes = seconds/60.0
    321410        hours = minutes/60.0
    322411        days = hours/24.0
    323412        weeks = days/7.0
    324413
    325         if minutes < 60: return "%.1f mins ago" % minutes
     414        if seconds < 60: return "%.1f secs ago" % seconds
     415        elif minutes < 60: return "%.1f mins ago" % minutes
    326416        elif hours < 48: return "%.1f hours ago" % hours
    327417        elif days < 7: return "%.1f days ago" % days
    … …  
    398488
    399489    '''
     490    Is someone processing this item right now? Checks whether this was attempted within the last 2 HOUR
     491    '''
     492    def processingNow(self, batchType, stage_id, epoch, dvoGpc1Label):
     493
     494        sql = "SELECT COUNT(*) \
     495               FROM batch \
     496               WHERE stage_id = " + str(stage_id) + " \
     497               AND batch_type = '" + batchType + "' \
     498               AND dvo_db = '" + dvoGpc1Label + "' \
     499               AND timestamp BETWEEN now() - INTERVAL 2 HOUR AND now()"
     500
     501        try:
     502            rs = self.executeQuery(sql)
     503            rs.first()
     504            if rs.getInt(1) > 0:
     505                self.logger.errorPair(str(stage_id) + " is already being processed", "skipping")
     506                return True
     507            else:
     508                return False
     509        except:
     510            self.logger.exception("Unable to check whether this batch is being processed")
     511
     512    '''
     513    Gets the batch type of this batch
     514    '''
     515    def getBatchType(self, batch_id):
     516
     517        sql = "SELECT batch_type \
     518               FROM batch \
     519               WHERE batch_id = " + str(batch_id)
     520
     521        batch_type = "NA"
     522        try:
     523            rs = self.executeQuery(sql)
     524            rs.first()
     525            batch_type = rs.getString(1)
     526        except:
     527            self.logger.exception("Unable to check whether this batch has already been processed")
     528 
     529        return batch_type
     530
     531    '''
    400532    Have we already processed and published this batch?
    401533    '''
    … …  
    425557    Has this stage_id consistently failed to process?
    426558    '''
    427     def consistentlyFailed(self, batchType, stage_id, epoch, dvoGpc1Label, count):
     559    def consistentlyFailed(self, batchType, stage_id, epoch, dvoGpc1Label):
    428560
    429561        sql = "SELECT COUNT(*) \
    … …  
    438570            rs = self.executeQuery(sql)
    439571            rs.first()
    440             if rs.getInt(1) >= count:
    441                 self.logger.errorPair(str(stage_id) + " has failed %d times" % count, "skipping")
     572            if rs.getInt(1) >= self.MAX_FAILS:
     573                self.logger.errorPair(str(stage_id) + " has failed %d times" % self.MAX_FAILS, "skipping")
    442574                return True
    443575            else:
    … …  
    446578            self.logger.exception("Unable to check whether this batch has consistently failed")
    447579
     580    '''
     581    Returns a list of stage_ids that have failed the max number of times
     582    '''
     583    def getConsistentlyFailedIDs(self, batchType, epoch, dvoGpc1Label):
     584
     585        sql = "SELECT DISTINCT stage_id, COUNT(stage_id) AS numoccur \
     586               FROM batch \
     587               WHERE batch_type = '" + batchType + "' \
     588               AND timestamp > '" + epoch + "' \
     589               AND dvo_db = '" + dvoGpc1Label + "' \
     590               AND processed = -1 \
     591               GROUP BY stage_id HAVING numoccur >= " + str(self.MAX_FAILS)
     592
     593        ids = []
     594        try:
     595            rs = self.executeQuery(sql)
     596            while (rs.next()):
     597                ids.append(rs.getInt(1))
     598        except:
     599            self.logger.exception("Can't query for consistantly failed ids in ipptopsps Db")
     600
     601        rs.close()
     602
     603        self.logger.debug("Found %d consistantly failed stage_ids for batch type='%s'" % (len(ids), batchType))
     604
     605        return ids
     606
     607    '''
     608    Locks the batch table. This will wait if the lock is held by someone else
     609    '''
     610    def lockBatchTable(self):
     611        self.lockTable("batch")
     612
     613    '''
     614    Creates a new batch. This is done with a locked table in order that no two clients can start work on the same item
     615    '''
     616    def createNewBatch(self, batchType, stageID, survey, epoch, dvoDb, datastoreProduct, force):
     617
     618        batchID = -1;
     619
     620        if force or \
     621            (not self.alreadyProcessed(batchType, stageID, epoch, dvoDb) \
     622            and not self.processingNow(batchType, stageID, epoch, dvoDb) \
     623            and not self.consistentlyFailed(batchType, stageID, epoch, dvoDb)):
     624
     625            sql = "INSERT INTO batch ( \
     626                   batch_type, \
     627                   stage_id, \
     628                   survey, \
     629                   dvo_db, \
     630                   datastore_product \
     631                   ) VALUES ( \
     632                       '" + batchType + "', \
     633                       " + str(stageID) + ", \
     634                       '" + survey + "', \
     635                       '" + dvoDb + "', \
     636                       '" + datastoreProduct + "' \
     637                       )"
     638
     639            self.execute(sql)
     640
     641            sql = "SELECT MAX(batch_id) FROM batch"
     642
     643            try:
     644                rs = self.executeQuery(sql)
     645                rs.first()
     646                batchID = rs.getInt(1)
     647                self.logger.debug("Created new batch in ippToPsps database with batchID = %d" % batchID)
     648            except:
     649                self.logger.exception("Unable to get batch ID")
     650                batchID = -1
     651
     652        return batchID;
    448653     
    449654
Note: See TracChangeset for help on using the changeset viewer.