IPP Software Navigation Tools IPP Links Communication Pan-STARRS Links

Changeset 33684 for trunk/ippToPsps


Ignore:
Timestamp:
Apr 3, 2012, 4:41:21 PM (14 years ago)
Author:
rhenders
Message:

added support for OB batches; removed need to pass config params as args to any methods; extended time for 'dead' clients

File:
1 edited

Legend:

Unmodified
Added
Removed

  • trunk/ippToPsps/jython/ipptopspsdb.py

    r33573 r33684  
    2525    Returns a list of merged batch IDs that are merged but not yet deleted
    2626    '''
    27     def getMergedButNotDeletedBatchIDs(self, batchType, epoch, dvoGpc1Label, column):
     27    def getMergedButNotDeletedBatchIDs(self, batchType, column):
    2828
    2929        sql = "SELECT DISTINCT batch_id \
    3030               FROM batch \
    31                WHERE timestamp > '" + epoch + "' \
     31               WHERE timestamp > '" + self.config.epoch + "' \
    3232               AND batch_type = '" + batchType + "' \
    33                AND dvo_db = '" + dvoGpc1Label + "' \
     33               AND dvo_db = '" + self.config.dvoLabel + "' \
    3434               AND merged = 1 \
    3535               AND " + column + " = 0"
    … …  
    5050    Returns a list of batch IDs marked as 'purged' but not yet deleted
    5151    '''
    52     def getPurgedButNotDeletedBatchIDs(self, batchType, epoch, dvoGpc1Label, column):
     52    def getPurgedButNotDeletedBatchIDs(self, batchType, column):
    5353
    5454        sql = "SELECT DISTINCT batch_id \
    5555               FROM batch \
    56                WHERE timestamp > '" + epoch + "' \
     56               WHERE timestamp > '" + self.config.epoch + "' \
    5757               AND batch_type = '" + batchType + "' \
    58                AND dvo_db = '" + dvoGpc1Label + "' \
     58               AND dvo_db = '" + self.config.dvoLabel + "' \
    5959               AND purged = 1 \
    6060               AND " + column + " = 0"
    … …  
    7676    Returns a list of processed batch IDs that have been loaded to the ODM but not yet deleted
    7777    '''
    78     def getLoadedToODMButNotDeletedBatchIDs(self, batchType, epoch, dvoGpc1Label, column):
     78    def getLoadedToODMButNotDeletedBatchIDs(self, batchType, column):
    7979
    8080        sql = "SELECT DISTINCT batch_id \
    8181               FROM batch \
    82                WHERE timestamp > '" + epoch + "' \
     82               WHERE timestamp > '" + self.config.epoch + "' \
    8383               AND batch_type = '" + batchType + "' \
    84                AND dvo_db = '" + dvoGpc1Label + "' \
     84               AND dvo_db = '" + self.config.dvoLabel + "' \
    8585               AND (loaded_to_ODM = -1 OR merge_worthy = 1) \
    8686               AND " + column + " = 0"
    … …  
    101101    Returns a list of merged batch IDs that are not deleted from local disk
    102102    '''
    103     def getMergedButNotDeletedFromLocalDisk(self, batchType, epoch, dvoGpc1Label):
    104         return self.getMergedButNotDeletedBatchIDs(batchType, epoch, dvoGpc1Label, "deleted_local")
     103    def getMergedButNotDeletedFromLocalDisk(self, batchType):
     104        return self.getMergedButNotDeletedBatchIDs(batchType, "deleted_local")
    105105
    106106    '''
    107107    Returns a list of processed batch IDs that are loaded to the ODM, but not deleted from local disk
    108108    '''
    109     def getLoadedToODMButNotDeletedFromLocalDisk(self, batchType, epoch, dvoGpc1Label):
    110         return self.getLoadedToODMButNotDeletedBatchIDs(batchType, epoch, dvoGpc1Label, "deleted_local")
     109    def getLoadedToODMButNotDeletedFromLocalDisk(self, batchType):
     110        return self.getLoadedToODMButNotDeletedBatchIDs(batchType, "deleted_local")
    111111
    112112    '''
    113113    Returns a list of processed batch IDs that are loaded to the ODM, but not deleted from datastore
    114114    '''
    115     def getLoadedToODMButNotDeletedFromDatastore(self, batchType, epoch, dvoGpc1Label):
    116         return self.getLoadedToODMButNotDeletedBatchIDs(batchType, epoch, dvoGpc1Label, "deleted_datastore")
     115    def getLoadedToODMButNotDeletedFromDatastore(self, batchType):
     116        return self.getLoadedToODMButNotDeletedBatchIDs(batchType, "deleted_datastore")
    117117
    118118    '''
    119119    Returns a list of processed batch IDs that are loaded to the ODM, but not deleted from DXLayer
    120120    '''
    121     def getLoadedToODMButNotDeletedFromDXLayer(self, batchType, epoch, dvoGpc1Label):
    122         return self.getLoadedToODMButNotDeletedBatchIDs(batchType, epoch, dvoGpc1Label, "deleted_dxlayer")
     121    def getLoadedToODMButNotDeletedFromDXLayer(self, batchType):
     122        return self.getLoadedToODMButNotDeletedBatchIDs(batchType, "deleted_dxlayer")
    123123
    124124    '''
    125125    Returns a list of purged  batch IDs that not deleted from local disk
    126126    '''
    127     def getPurgedButNotDeletedFromLocalDisk(self, batchType, epoch, dvoGpc1Label):
    128         return self.getPurgedButNotDeletedBatchIDs(batchType, epoch, dvoGpc1Label, "deleted_local")
     127    def getPurgedButNotDeletedFromLocalDisk(self, batchType):
     128        return self.getPurgedButNotDeletedBatchIDs(batchType, "deleted_local")
    129129
    130130    '''
    131131    Returns a list of purged batch IDs that are not deleted from datastore
    132132    '''
    133     def getPurgedButNotDeletedFromDatastore(self, batchType, epoch, dvoGpc1Label):
    134         return self.getPurgedButNotDeletedBatchIDs(batchType, epoch, dvoGpc1Label, "deleted_datastore")
     133    def getPurgedButNotDeletedFromDatastore(self, batchType):
     134        return self.getPurgedButNotDeletedBatchIDs(batchType, "deleted_datastore")
    135135
    136136
    … …  
    144144    publish something simulataneously with the client that it producing it
    145145    '''
    146     def getProcessedButFailedDatastoreBatchIDs(self, epoch, dvoGpc1Label, batchType):
     146    def getProcessedButFailedDatastoreBatchIDs(self, batchType):
    147147
    148148        sql = "SELECT DISTINCT batch_id \
    149149               FROM batch \
    150                WHERE timestamp > '" + epoch + "' \
     150               WHERE timestamp > '" + self.config.epoch + "' \
    151151               AND batch_type = '" + batchType + "' \
    152                AND dvo_db = '" + dvoGpc1Label + "' \
     152               AND dvo_db = '" + self.config.dvoLabel + "' \
    153153               AND processed = 1\
    154154               AND loaded_to_datastore != 1 \
    … …  
    173173    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
    174174    '''
    175     def getUnloadedBatchIDs(self, epoch, dvoGpc1Label, batchType):
     175    def getUnloadedBatchIDs(self, batchType):
    176176
    177177        sql = "SELECT batch_id \
    178178               FROM batch \
    179                WHERE timestamp > '" + epoch + "' \
    180                AND dvo_db = '" + dvoGpc1Label + "' \
     179               WHERE timestamp > '" + self.config.epoch + "' \
     180               AND dvo_db = '" + self.config.dvoLabel + "' \
    181181               AND loaded_to_datastore = 1 \
    182182               AND batch_type = '" + batchType + "' \
    … …  
    191191        except:
    192192            self.logger.exception("Can't query for unloaded batch ids in ipptopsps Db")
    193 
    194         rs.close()
     193        finally:
     194            rs.close()
    195195
    196196        self.logger.debug("Found %d unloaded items" % len(ids))
    … …  
    202202    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
    203203    '''
    204     def getUnmergeWorthyBatchIDs(self, epoch, dvoGpc1Label, batchType):
     204    def getUnmergeWorthyBatchIDs(self, batchType):
    205205
    206206        sql = "SELECT batch_id \
    207207               FROM batch \
    208                WHERE timestamp > '" + epoch + "' \
    209                AND dvo_db = '" + dvoGpc1Label + "' \
     208               WHERE timestamp > '" + self.config.epoch + "' \
     209               AND dvo_db = '" + self.config.dvoLabel + "' \
    210210               AND loaded_to_datastore = 1 \
    211211               AND batch_type = '" + batchType + "' \
    … …  
    221221        except:
    222222            self.logger.exception("Can't query for un-mergeworthy batch ids in ipptopsps Db")
    223 
    224         rs.close()
     223        finally:
     224            rs.close()
    225225
    226226        self.logger.debug("Found %d unloaded items" % len(ids))
    … …  
    231231    Returns a list of processed batch IDs that are merge_worthy, but not yet merged, for this epoch and dvo label
    232232    '''
    233     def getUnmergedBatchIDs(self, epoch, dvoGpc1Label, batchType):
     233    def getUnmergedBatchIDs(self, batchType):
    234234
    235235        sql = "SELECT batch_id \
    236236               FROM batch \
    237                WHERE timestamp > '" + epoch + "' \
    238                AND dvo_db = '" + dvoGpc1Label + "' \
     237               WHERE timestamp > '" + self.config.epoch + "' \
     238               AND dvo_db = '" + self.config.dvoLabel + "' \
    239239               AND merge_worthy = 1 \
    240240               AND batch_type = '" + batchType + "' \
    … …  
    248248        except:
    249249            self.logger.exception("Can't query for unmerged batch ids in ipptopsps Db")
    250 
    251         rs.close()
     250        finally:
     251            rs.close()
    252252
    253253        self.logger.debug("Found %d un-merged items" % len(ids))
    … …  
    258258    Returns a list of batch IDs for this epoch and dvo label for which the specified column is set to value
    259259    '''
    260     def getBatchIDs(self, epoch, dvoGpc1Label, column, value):
     260    def getBatchIDs(self, column, value):
    261261
    262262        sql = "SELECT DISTINCT batch_id \
    263263               FROM batch \
    264                WHERE timestamp > '" + epoch + "' \
    265                AND dvo_db = '" + dvoGpc1Label + "' \
     264               WHERE timestamp > '" + self.config.epoch + "' \
     265               AND dvo_db = '" + self.config.dvoLabel + "' \
    266266               AND " + column + " = " + str(value)
    267267
    … …  
    283283    Returns the total detections published to the datastore for this epoch, dvo label and batch type
    284284    '''
    285     def getTotalDetectionsPublished(self, batchType, epoch, dvoGpc1Label, interval="10 YEAR"):
     285    def getTotalDetectionsPublished(self, batchType, interval="10 YEAR"):
    286286
    287287        sql = "SELECT SUM(total_detections) \
    288288               FROM batch \
    289289               WHERE batch_type = '" + batchType + "' \
    290                AND timestamp > '" + epoch + "' \
    291                AND dvo_db = '" + dvoGpc1Label + "' \
     290               AND timestamp > '" + self.config.epoch + "' \
     291               AND dvo_db = '" + self.config.dvoLabel + "' \
    292292               AND processed = 1 \
    293293               AND loaded_to_datastore = 1 \
    … …  
    300300            total =  rs.getLong(1)
    301301        except:
    302             self.logger.exception("Can't query for total detections published")
     302            self.logger.errorPair("Can't get total detections published", sql)
    303303        finally:
    304304            rs.close()
    305 
    306         self.logger.debug("Found %d detections" % total)
    307305
    308306        return total
    … …  
    312310            cam_ids or stack_ids) for which the specified column is set to value
    313311    '''
    314     def getStageIDs(self, batchType, epoch, dvoGpc1Label, column, value):
     312    def getStageIDs(self, batchType, column, value):
    315313
    316314        sql = "SELECT DISTINCT stage_id \
    317315               FROM batch \
    318316               WHERE batch_type = '" + batchType + "' \
    319                AND timestamp > '" + epoch + "' \
    320                AND dvo_db = '" + dvoGpc1Label + "' \
     317               AND timestamp > '" + self.config.epoch + "' \
     318               AND dvo_db = '" + self.config.dvoLabel + "' \
    321319               AND " + column + " = " + str(value)
    322320
    … …  
    328326        except:
    329327            self.logger.exception("Can't query for ids with " + column + " = " + str(value) + " in ipptopsps Db")
    330 
    331         rs.close()
    332 
    333         self.logger.debug("Found %d batches with %s = 1 for batch type='%s'" % (len(ids), column, batchType))
     328        finally:
     329            rs.close()
    334330
    335331        return ids
    … …  
    338334    Series of getter methods that utilise getBatchIDs() method above
    339335    '''
    340     def getloadedToDatastoreBatchIDs(self, epoch, dvoGpc1Label):
    341         return self.getBatchIDs(epoch, dvoGpc1Label, "loaded_to_datastore", 1)
    342     def getFailedLoadedToODMBatchIDs(self, epoch, dvoGpc1Label):
    343         return self.getBatchIDs(epoch, dvoGpc1Label, "loaded_to_ODM", -1)
     336    def getloadedToDatastoreBatchIDs(self):
     337        return self.getBatchIDs("loaded_to_datastore", 1)
     338    def getFailedLoadedToODMBatchIDs(self):
     339        return self.getBatchIDs("loaded_to_ODM", -1)
    344340
    345341    '''
    … …  
    347343    Series of getter methods that utilise getStageIDs() method above
    348344    '''
    349     def getProcessedIDs(self, batchType, epoch, dvoGpc1Label):
    350         return self.getStageIDs(batchType, epoch, dvoGpc1Label, "processed", 1)
    351     def getFailedProcessedIDs(self, batchType, epoch, dvoGpc1Label):
    352         return self.getStageIDs(batchType, epoch, dvoGpc1Label, "processed", -1)
    353     def getLoadedToDatastoreIDs(self, batchType, epoch, dvoGpc1Label):
    354         return self.getStageIDs(batchType, epoch, dvoGpc1Label, "loaded_to_datastore", 1)
    355     def getFailedLoadedToDatastoreIDs(self, batchType, epoch, dvoGpc1Label):
    356         return self.getStageIDs(batchType, epoch, dvoGpc1Label, "loaded_to_datastore", -1)
    357     def getLoadedToODMIDs(self, batchType, epoch, dvoGpc1Label):
    358         return self.getStageIDs(batchType, epoch, dvoGpc1Label, "loaded_to_ODM", 1)
    359     def getFailedLoadedToODMIDs(self, batchType, epoch, dvoGpc1Label):
    360         return self.getStageIDs(batchType, epoch, dvoGpc1Label, "loaded_to_ODM", -1)
    361     def getMergeWorthyIDs(self, batchType, epoch, dvoGpc1Label):
    362         return self.getStageIDs(batchType, epoch, dvoGpc1Label, "merge_worthy", 1)
    363     def getFailedMergeWorthyIDs(self, batchType, epoch, dvoGpc1Label):
    364         return self.getStageIDs(batchType, epoch, dvoGpc1Label, "merge_worthy", -1)
    365     def getMergedIDs(self, batchType, epoch, dvoGpc1Label):
    366         return self.getStageIDs(batchType, epoch, dvoGpc1Label, "merged", 1)
    367     def getFailedMergedIDs(self, batchType, epoch, dvoGpc1Label):
    368         return self.getStageIDs(batchType, epoch, dvoGpc1Label, "merged", -1)
     345    def getProcessedIDs(self, batchType):
     346        return self.getStageIDs(batchType, "processed", 1)
     347    def getFailedProcessedIDs(self, batchType):
     348        return self.getStageIDs(batchType, "processed", -1)
     349    def getLoadedToDatastoreIDs(self, batchType):
     350        return self.getStageIDs(batchType, "loaded_to_datastore", 1)
     351    def getFailedLoadedToDatastoreIDs(self, batchType):
     352        return self.getStageIDs(batchType, "loaded_to_datastore", -1)
     353    def getLoadedToODMIDs(self, batchType):
     354        return self.getStageIDs(batchType, "loaded_to_ODM", 1)
     355    def getFailedLoadedToODMIDs(self, batchType):
     356        return self.getStageIDs(batchType, "loaded_to_ODM", -1)
     357    def getMergeWorthyIDs(self, batchType):
     358        return self.getStageIDs(batchType, "merge_worthy", 1)
     359    def getFailedMergeWorthyIDs(self, batchType):
     360        return self.getStageIDs(batchType, "merge_worthy", -1)
     361    def getMergedIDs(self, batchType):
     362        return self.getStageIDs(batchType, "merged", 1)
     363    def getFailedMergedIDs(self, batchType):
     364        return self.getStageIDs(batchType, "merged", -1)
    369365
    370366    '''
    … …  
    385381    Returns a count of batches processed in the specified time period.
    386382    '''
    387     def countBatches(self, batchType, epoch, dvoGpc1Label, interval="10 YEAR"):
     383    def countBatches(self, batchType, interval="10 YEAR"):
    388384
    389385        sql = "SELECT COUNT(batch_id) \
    … …  
    391387               WHERE batch_type = '" + batchType + "' \
    392388               AND processed = 1 \
    393                AND timestamp > '" + epoch + "' \
    394                AND dvo_db = '" + dvoGpc1Label + "' \
     389               AND timestamp > '" + self.config.epoch + "' \
     390               AND dvo_db = '" + self.config.dvoLabel + "' \
    395391               AND timestamp BETWEEN now() - INTERVAL " + interval + " AND now()"
    396392
    … …  
    408404    this epoch, dvo label and batch type
    409405    '''
    410     def getTimeOfLastBatchPublished(self, batchType, epoch, dvoGpc1Label):
     406    def getTimeOfLastBatchPublished(self, batchType):
    411407
    412408        seconds = None
    … …  
    416412               WHERE batch_type = '" + batchType + "' \
    417413               AND loaded_to_datastore = 1 \
    418                AND timestamp > '" + epoch + "' \
    419                AND dvo_db = '" + dvoGpc1Label + "' \
     414               AND timestamp > '" + self.config.epoch + "' \
     415               AND dvo_db = '" + self.config.dvoLabel + "' \
    420416               ORDER BY timestamp DESC LIMIT 1"
    421417
    … …  
    601597    Returns a list of stage_ids that have failed the max number of times
    602598    '''
    603     def getConsistentlyFailedIDs(self, batchType, epoch, dvoGpc1Label):
     599    def getConsistentlyFailedIDs(self, batchType):
    604600
    605601        sql = "SELECT DISTINCT stage_id, COUNT(stage_id) AS numoccur \
    606602               FROM batch \
    607603               WHERE batch_type = '" + batchType + "' \
    608                AND timestamp > '" + epoch + "' \
    609                AND dvo_db = '" + dvoGpc1Label + "' \
     604               AND timestamp > '" + self.config.epoch + "' \
     605               AND dvo_db = '" + self.config.dvoLabel + "' \
    610606               AND processed = -1 \
    611607               GROUP BY stage_id HAVING numoccur >= " + str(self.MAX_FAILS)
    … …  
    706702        self.execute(sql)
    707703
     704    '''
     705    Inserts some object metadata for this batch ID
     706    '''
     707    def insertObjectMeta(self, batchID, region):
     708
     709        sql = "INSERT INTO object ( \
     710               batch_id \
     711               ,region \
     712               ) VALUES ( \
     713               " + str(batchID) + " \
     714               ,'" + region + "' \
     715               )"
     716
     717        self.execute(sql)
    708718    '''
    709719    Resets a batch ready for re-loading to the datastore
    … …  
    864874               ,queue_P2 = " + self.config.queuingThisBatchType("P2") + " \
    865875               ,queue_ST = " + self.config.queuingThisBatchType("ST") + " \
     876               ,queue_OB = " + self.config.queuingThisBatchType("OB") + " \
    866877               WHERE name = '" + self.config.name + "'"
    867878
    … …  
    910921        ,queue_P2 \
    911922        ,queue_ST \
     923        ,queue_OB \
    912924        FROM config \
    913925        WHERE name = '" + self.config.name + "'"
    … …  
    942954            if rs.getInt(19) == 1: self.config.batchTypes.append("P2")
    943955            if rs.getInt(20) == 1: self.config.batchTypes.append("ST")
     956            if rs.getInt(21) == 1: self.config.batchTypes.append("OB")
    944957            self.config.force = False # TODO
    945958            self.config.test = False # TODO
    … …  
    10091022
    10101023        for id in ids:
    1011             sql = "INSERT INTO pending \
     1024            sql = "INSERT IGNORE INTO pending \
    10121025               (box_id, batch_type, stage_id) \
    10131026               VALUES \
    … …  
    11321145    Returns ids for pending items for this config
    11331146    '''
    1134     def getPendingIds(self):
    1135 
    1136         sql = "SELECT DISTINCT stage_id FROM pending JOIN box ON (box_id = id) WHERE config = '" + self.config.name + "'"
     1147    def getPendingIds(self, batchType):
     1148
     1149        sql = "SELECT DISTINCT stage_id \
     1150               FROM pending \
     1151               JOIN box ON (box_id = id) \
     1152               WHERE config = '" + self.config.name + "' \
     1153               AND batch_type = '" + batchType + "'"
    11371154
    11381155        ids = []
    … …  
    12821299
    12831300    '''
    1284     Purges old clients that may have die, i.e. those with a timestamp older than n hours
     1301    Purges old clients that may have died, i.e. those with a timestamp older than n hours
    12851302    '''
    12861303    def purgeDeadClients(self):
    1287         self.execute("DELETE FROM clients WHERE timestamp <  now() -  INTERVAL 6 HOUR")
     1304        self.execute("DELETE FROM clients WHERE timestamp < now() - INTERVAL 14 HOUR")
    12881305
    12891306    '''
Note: See TracChangeset for help on using the changeset viewer.