IPP Software Navigation Tools IPP Links Communication Pan-STARRS Links

Ignore:
Timestamp:
Jun 19, 2012, 5:24:19 PM (14 years ago)
Author:
mhuber
Message:

merging latest r34040 trunk changes to branch

Location:
branches/meh_branches/ppstack_test
Files:
2 edited

Legend:

Unmodified
Added
Removed

  • branches/meh_branches/ppstack_test

  • branches/meh_branches/ppstack_test/ippToPsps/jython/ipptopspsdb.py

    r33415 r34041  
    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
    … …  
    141141    want to load a test batch to PSPS
    142142
    143     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
    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 \
    155155               AND timestamp > '2011-10-27' \
    156                AND timestamp < now() -  INTERVAL 4 HOUR"
     156               AND timestamp < now() -  INTERVAL 12 HOUR"
    157157
    158158        ids = []
    … …  
    161161            while (rs.next()):
    162162                ids.append(rs.getInt(1))
     163            rs.close()
    163164        except:
    164165            self.logger.exception("Can't query for processed batch ids not loaded to datastore ipptopsps Db")
    165 
    166         rs.close()
    167166
    168167        self.logger.debug("Found %d processed but-not-on-datastore items" % len(ids))
    … …  
    174173    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
    175174    '''
    176     def getUnloadedBatchIDs(self, epoch, dvoGpc1Label, batchType):
     175    def getUnloadedBatchIDs(self, batchType):
    177176
    178177        sql = "SELECT batch_id \
    179178               FROM batch \
    180                WHERE timestamp > '" + epoch + "' \
    181                AND dvo_db = '" + dvoGpc1Label + "' \
     179               WHERE timestamp > '" + self.config.epoch + "' \
     180               AND dvo_db = '" + self.config.dvoLabel + "' \
    182181               AND loaded_to_datastore = 1 \
    183182               AND batch_type = '" + batchType + "' \
    … …  
    192191        except:
    193192            self.logger.exception("Can't query for unloaded batch ids in ipptopsps Db")
    194 
    195         rs.close()
     193        finally:
     194            rs.close()
    196195
    197196        self.logger.debug("Found %d unloaded items" % len(ids))
    … …  
    203202    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
    204203    '''
    205     def getUnmergeWorthyBatchIDs(self, epoch, dvoGpc1Label, batchType):
     204    def getUnmergeWorthyBatchIDs(self, batchType):
    206205
    207206        sql = "SELECT batch_id \
    208207               FROM batch \
    209                WHERE timestamp > '" + epoch + "' \
    210                AND dvo_db = '" + dvoGpc1Label + "' \
     208               WHERE timestamp > '" + self.config.epoch + "' \
     209               AND dvo_db = '" + self.config.dvoLabel + "' \
    211210               AND loaded_to_datastore = 1 \
    212211               AND batch_type = '" + batchType + "' \
    … …  
    222221        except:
    223222            self.logger.exception("Can't query for un-mergeworthy batch ids in ipptopsps Db")
    224 
    225         rs.close()
     223        finally:
     224            rs.close()
    226225
    227226        self.logger.debug("Found %d unloaded items" % len(ids))
    … …  
    232231    Returns a list of processed batch IDs that are merge_worthy, but not yet merged, for this epoch and dvo label
    233232    '''
    234     def getUnmergedBatchIDs(self, epoch, dvoGpc1Label, batchType):
     233    def getUnmergedBatchIDs(self, batchType):
    235234
    236235        sql = "SELECT batch_id \
    237236               FROM batch \
    238                WHERE timestamp > '" + epoch + "' \
    239                AND dvo_db = '" + dvoGpc1Label + "' \
     237               WHERE timestamp > '" + self.config.epoch + "' \
     238               AND dvo_db = '" + self.config.dvoLabel + "' \
    240239               AND merge_worthy = 1 \
    241240               AND batch_type = '" + batchType + "' \
    … …  
    249248        except:
    250249            self.logger.exception("Can't query for unmerged batch ids in ipptopsps Db")
    251 
    252         rs.close()
     250        finally:
     251            rs.close()
    253252
    254253        self.logger.debug("Found %d un-merged items" % len(ids))
    … …  
    259258    Returns a list of batch IDs for this epoch and dvo label for which the specified column is set to value
    260259    '''
    261     def getBatchIDs(self, epoch, dvoGpc1Label, column, value):
     260    def getBatchIDs(self, column, value):
    262261
    263262        sql = "SELECT DISTINCT batch_id \
    264263               FROM batch \
    265                WHERE timestamp > '" + epoch + "' \
    266                AND dvo_db = '" + dvoGpc1Label + "' \
     264               WHERE timestamp > '" + self.config.epoch + "' \
     265               AND dvo_db = '" + self.config.dvoLabel + "' \
    267266               AND " + column + " = " + str(value)
    268267
    … …  
    274273        except:
    275274            self.logger.exception("Can't query for batch ids with " + column + " = " + str(value) + " in ipptopsps Db")
    276 
    277         rs.close()
     275        finally:
     276            rs.close()
    278277
    279278        self.logger.debug("Found %d batches with %s = %d" % (len(ids), column, value))
    … …  
    284283    Returns the total detections published to the datastore for this epoch, dvo label and batch type
    285284    '''
    286     def getTotalDetectionsPublished(self, batchType, epoch, dvoGpc1Label):
     285    def getTotalDetectionsPublished(self, batchType, interval="10 YEAR"):
    287286
    288287        sql = "SELECT SUM(total_detections) \
    289288               FROM batch \
    290289               WHERE batch_type = '" + batchType + "' \
    291                AND timestamp > '" + epoch + "' \
    292                AND dvo_db = '" + dvoGpc1Label + "' \
     290               AND timestamp > '" + self.config.epoch + "' \
     291               AND dvo_db = '" + self.config.dvoLabel + "' \
    293292               AND processed = 1 \
    294                AND loaded_to_datastore = 1"
     293               AND loaded_to_datastore = 1 \
     294               AND timestamp BETWEEN now() - INTERVAL " + interval + " AND now()"
    295295
    296296        total = -1
    … …  
    300300            total =  rs.getLong(1)
    301301        except:
    302             self.logger.exception("Can't query for total detections published")
    303 
    304         rs.close()
    305 
    306         self.logger.debug("Found %d detections" % total)
     302            self.logger.errorPair("Can't get total detections published", sql)
     303        finally:
     304            rs.close()
    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    '''
    … …  
    372368    '''
    373369    def getStages(self):
    374         return ['processed', 'loaded_to_datastore', 'loaded_to_ODM', 'merge_worthy', 'deleted_datastore', 'deleted_dxlayer', 'merged', 'deleted_local' ]
     370        return [
     371        'processed',
     372        'loaded_to_datastore',
     373        'loaded_to_ODM',
     374        'merge_worthy',
     375        'deleted_datastore',
     376        'deleted_dxlayer',
     377        'merged',
     378        'deleted_local' ]
    375379
    376380    '''
    377381    Returns a count of batches processed in the specified time period.
    378382    '''
    379     def countBatchesInLastPeriod(self, batchType, epoch, dvoGpc1Label, interval):
     383    def countBatches(self, batchType, interval="10 YEAR"):
    380384
    381385        sql = "SELECT COUNT(batch_id) \
    … …  
    383387               WHERE batch_type = '" + batchType + "' \
    384388               AND processed = 1 \
    385                AND timestamp > '" + epoch + "' \
    386                AND dvo_db = '" + dvoGpc1Label + "' \
     389               AND timestamp > '" + self.config.epoch + "' \
     390               AND dvo_db = '" + self.config.dvoLabel + "' \
    387391               AND timestamp BETWEEN now() - INTERVAL " + interval + " AND now()"
    388392
    … …  
    393397            self.logger.exception("Unable to count batches in interval")
    394398            return 0
     399        finally:
     400            rs.close()
    395401
    396402    '''
    … …  
    398404    this epoch, dvo label and batch type
    399405    '''
    400     def getTimeOfLastBatchPublished(self, batchType, epoch, dvoGpc1Label):
     406    def getTimeOfLastBatchPublished(self, batchType):
    401407
    402408        seconds = None
    … …  
    406412               WHERE batch_type = '" + batchType + "' \
    407413               AND loaded_to_datastore = 1 \
    408                AND timestamp > '" + epoch + "' \
    409                AND dvo_db = '" + dvoGpc1Label + "' \
     414               AND timestamp > '" + self.config.epoch + "' \
     415               AND dvo_db = '" + self.config.dvoLabel + "' \
    410416               ORDER BY timestamp DESC LIMIT 1"
    411417
    … …  
    497503
    498504    '''
    499     Is someone processing this item right now? Checks whether this was attempted within the last 2 HOUR
    500     '''
    501     def processingNow(self, batchType, stage_id, epoch, dvoGpc1Label):
     505    Is someone processing this item right now? Checks whether this was attempted within the last 12 HOUR
     506    '''
     507    def processingNow(self, batchType, stage_id):
    502508
    503509        sql = "SELECT COUNT(*) \
    … …  
    505511               WHERE stage_id = " + str(stage_id) + " \
    506512               AND batch_type = '" + batchType + "' \
    507                AND dvo_db = '" + dvoGpc1Label + "' \
    508                AND timestamp BETWEEN now() - INTERVAL 4 HOUR AND now()"
     513               AND dvo_db = '" + self.config.dvoLabel + "' \
     514               AND batch_type != 'IN' \
     515               AND timestamp BETWEEN now() - INTERVAL 12 HOUR AND now()"
    509516
    510517        try:
    … …  
    541548    Have we already processed this stage_id?
    542549    '''
    543     def alreadyProcessed(self, batchType, stage_id, epoch, dvoGpc1Label):
     550    def alreadyProcessed(self, batchType, stage_id):
    544551
    545552        sql = "SELECT COUNT(*) \
    … …  
    547554               WHERE stage_id = " + str(stage_id) + " \
    548555               AND batch_type = '" + batchType + "' \
    549                AND timestamp > '" + epoch + "' \
    550                AND dvo_db = '" + dvoGpc1Label + "' \
     556               AND timestamp > '" + self.config.epoch + "' \
     557               AND dvo_db = '" + self.config.dvoLabel + "' \
     558               AND batch_type != 'IN' \
    551559               AND processed = 1"
    552560
    … …  
    565573    Has this stage_id consistently failed to process?
    566574    '''
    567     def consistentlyFailed(self, batchType, stage_id, epoch, dvoGpc1Label):
     575    def consistentlyFailed(self, batchType, stage_id):
    568576
    569577        sql = "SELECT COUNT(*) \
    … …  
    571579               WHERE stage_id = " + str(stage_id) + " \
    572580               AND batch_type = '" + batchType + "' \
    573                AND timestamp > '" + epoch + "' \
    574                AND dvo_db = '" + dvoGpc1Label + "' \
     581               AND timestamp > '" + self.config.epoch + "' \
     582               AND dvo_db = '" + self.config.dvoLabel + "' \
    575583               AND processed = -1"
    576584
    … …  
    589597    Returns a list of stage_ids that have failed the max number of times
    590598    '''
    591     def getConsistentlyFailedIDs(self, batchType, epoch, dvoGpc1Label):
     599    def getConsistentlyFailedIDs(self, batchType):
    592600
    593601        sql = "SELECT DISTINCT stage_id, COUNT(stage_id) AS numoccur \
    594602               FROM batch \
    595603               WHERE batch_type = '" + batchType + "' \
    596                AND timestamp > '" + epoch + "' \
    597                AND dvo_db = '" + dvoGpc1Label + "' \
     604               AND timestamp > '" + self.config.epoch + "' \
     605               AND dvo_db = '" + self.config.dvoLabel + "' \
    598606               AND processed = -1 \
    599607               GROUP BY stage_id HAVING numoccur >= " + str(self.MAX_FAILS)
    … …  
    620628
    621629    '''
    622     Creates a new batch. This is done with a locked table in order that no two clients can start work on the same item
    623     '''
    624     def createNewBatch(self, batchType, stageID, config):
     630    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
     631    '''
     632    def createNewBatch(self, batchType, stageID):
    625633
    626634        batchID = -1;
    627635
    628         if config.force or \
    629             (not self.alreadyProcessed(batchType, stageID, config.epoch, config.dvoLabel) \
    630             and not self.processingNow(batchType, stageID, config.epoch, config.dvoLabel) \
    631             and not self.consistentlyFailed(batchType, stageID, config.epoch, config.dvoLabel)):
     636        if self.config.force or \
     637            (not self.alreadyProcessed(batchType, stageID) \
     638            and not self.processingNow(batchType, stageID) \
     639            and not self.consistentlyFailed(batchType, stageID)):
    632640
    633641            sql = "INSERT INTO batch ( \
    … …  
    640648                       '" + batchType + "', \
    641649                       " + str(stageID) + ", \
    642                        '" + config.survey + "', \
    643                        '" + config.dvoLabel + "', \
    644                        '" + config.datastoreProduct + "' \
     650                       '" + self.config.survey + "', \
     651                       '" + self.config.dvoLabel + "', \
     652                       '" + self.config.datastoreProduct + "' \
    645653                       )"
    646654
    … …  
    695703
    696704    '''
     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)
     718    '''
    697719    Resets a batch ready for re-loading to the datastore
    698720    '''
    … …  
    712734
    713735    '''
     736    Pause loading clients
     737    '''
     738    def pauseLoaders(self, bool, ids):
     739        self.setColForLoaders("paused", bool, ids)
     740
     741    '''
     742    Kill loading clients
     743    '''
     744    def killLoaders(self, bool, ids):
     745        self.setColForLoaders("killed", bool, ids)
     746
     747    '''
     748    Sets the config field for this client
     749    '''
     750    def setConfigForThisClient(self, config, host, pid):
     751
     752        self.execute("UPDATE clients SET config = '" + config + "' \
     753                WHERE host = '" + host + "' \
     754                AND pid = " + str(pid))
     755    '''
     756    Sets the config field for a set of loader clients
     757    '''
     758    def setConfigForLoaders(self, config, ids):
     759
     760        for id in ids:
     761            self.execute("UPDATE clients SET config = '" + config + "' \
     762                    WHERE type = 'loader.py' \
     763                    AND id = " + str(id))
     764
     765    '''
     766    Sets the value for a given column for a set of loader clients
     767    '''
     768    def setColForLoaders(self, col, bool, ids):
     769
     770        if bool: value = 1
     771        else: value = 0
     772
     773        for id in ids:
     774            self.execute("UPDATE clients SET " + col + " = " + str(value) + " \
     775                    WHERE type = 'loader.py' \
     776                    AND id = " + str(id))
     777
     778    '''
     779    Remove client by id
     780    '''
     781    def removeClientsByIds(self, ids):
     782        for id in ids:
     783            self.execute("DELETE FROM clients WHERE id = " + str(id))
     784
     785    '''
    714786    Remove client
    715787    '''
    … …  
    720792    Insert client
    721793    '''
    722     def insertClient(self, type, host, pid, config):
    723         self.execute("INSERT INTO clients (timestamp, type, host, pid, config) VALUES (now(), '" + type + "', '" + host + "', " + str(pid) + ", '" + config + "')")
     794    def insertClient(self, type, host, pid):
     795        self.execute("INSERT INTO clients (timestamp, type, host, pid, config) VALUES (now(), '" + type + "', '" + host + "', " + str(pid) + ", '" + self.config.name + "')")
    724796
    725797    '''
    726798    Update clients, or inserts it for the first time
    727799    '''
    728     def updateClient(self, type, host, pid, config):
    729 
    730         try:
    731             self.insertClient(type, host, pid, config)
     800    def updateClient(self, type, host, pid):
     801
     802        try:
     803            self.insertClient(type, host, pid)
    732804        except:
    733805            self.execute("UPDATE clients SET timestamp = now() WHERE host = '" + host + "' AND pid = " + str(pid))
    … …  
    764836        return False
    765837
    766 
    767     '''
    768     Populates Config object
    769     '''
    770     def readConfig(self, host, pid):
    771 
    772         # first get config defined for this client
     838    '''
     839    Returns a list of available configs
     840    '''
     841    def getActiveConfigList(self):
     842       
     843        sql = "SELECT DISTINCT name FROM config WHERE active = 1 ORDER BY name"
     844
     845        configs = []
     846        try:
     847            rs = self.executeQuery(sql)
     848            while (rs.next()): configs.append(rs.getString(1))
     849        except:
     850            self.logger.errorPair("Can't get config names", sql)
     851        finally:
     852            rs.close()
     853
     854        return configs
     855
     856    '''
     857    Prompts user for info about a new or existing config and lets them neter the various details
     858    '''
     859    def editConfig(self):
     860
     861       print " ********************************************************************"
     862       print " *              Config editor"
     863       print " *"
     864       response = raw_input(" * Name for new config, or existing config to edit? ")
     865       if response == "": return
     866       self.config.name = response
     867
     868       # attempt to insert new config, if it already exists then carry on with update
     869       sql = "INSERT INTO config (name) VALUES ('" + self.config.name + "')"
     870       try:       
     871           self.execute(sql)
     872       except: pass
     873
     874       # for each column in the config table
     875       sql = "SHOW COLUMNS FROM config"
     876       rs = self.executeQuery(sql)
     877       while (rs.next()):
     878
     879           field = rs.getString(1)
     880           type = rs.getString(2)
     881           null = rs.getString(3)
     882
     883           # things we don;t want the user to edit
     884           if field == 'timestamp': continue
     885           if field == 'name': continue
     886
     887           # get current value for this field, if any
     888           try:
     889               rs2 = self.executeQuery("SELECT " + field + " FROM config WHERE name = '" + self.config.name + "'")
     890               rs2.first()
     891               default = rs2.getString(1)
     892           except: pass
     893
     894           question = " * " + field + " (" + type + ")"
     895
     896           # there may not be a default value
     897           try: question = question +  " hit return to accept default of: '" + default + "'"
     898           except: pass
     899
     900           # prompt the user for new value or to accept existing value
     901           question = question +  "? "
     902           response = raw_input(question)
     903           if response != "":
     904               if type.find("varchar") != -1 or type.find("timestamp") != -1: quotes = "'"
     905               else: quotes = ""
     906               self.execute("UPDATE config SET " + field + " = " + quotes + response + quotes + " WHERE name = '" + self.config.name + "'")
     907       
     908       print " *"
     909       print " ********************************************************************"
     910
     911    '''
     912    Returns config for this client
     913    '''
     914    def getConfigForThisClient(self, host, pid):
     915
    773916        sql = "SELECT config \
    774917               FROM clients \
    … …  
    785928            return False
    786929
     930        return True
     931
     932    '''
     933    Reads config from the database and populates Config object
     934    '''
     935    def readConfig(self, host, pid):
     936
     937        # first get config defined for this client
     938        if not self.getConfigForThisClient(host, pid): return False
     939
    787940        # now load that config
    788941        sql = "SELECT \
    789         datastore_product, \
    790         datastore_type, \
    791         datastore_publish, \
    792         dvo_label, \
    793         dvo_location, \
    794         min_ra, \
    795         max_ra, \
    796         min_dec, \
    797         max_dec, \
    798         box_size, \
    799         base_path, \
    800         data_release, \
    801         delete_local, \
    802         delete_datastore, \
    803         delete_dxlayer, \
    804         epoch, \
    805         survey, \
    806         psps_survey, \
    807         queue_P2, \
    808         queue_ST \
     942        datastore_product \
     943        ,datastore_type \
     944        ,datastore_publish \
     945        ,dvo_label \
     946        ,dvo_location \
     947        ,min_ra \
     948        ,max_ra \
     949        ,min_dec \
     950        ,max_dec \
     951        ,box_size \
     952        ,base_path \
     953        ,data_release \
     954        ,delete_local \
     955        ,delete_datastore \
     956        ,delete_dxlayer \
     957        ,epoch \
     958        ,survey \
     959        ,psps_survey \
     960        ,queue_P2 \
     961        ,queue_ST \
     962        ,queue_OB \
    809963        FROM config \
    810964        WHERE name = '" + self.config.name + "'"
    … …  
    819973            else: self.config.datastorePublishing = False
    820974            self.config.dvoLabel = rs.getString(4)
     975
     976            # if dvoLabel is null is can break queries, so set to something
     977            if not self.config.dvoLabel: self.config.dvoLabel = "none"
    821978            self.config.dvoLocation = rs.getString(5)
    822979            self.config.minRa = rs.getDouble(6)
    … …  
    839996            if rs.getInt(19) == 1: self.config.batchTypes.append("P2")
    840997            if rs.getInt(20) == 1: self.config.batchTypes.append("ST")
     998            if rs.getInt(21) == 1: self.config.batchTypes.append("OB")
    841999            self.config.force = False # TODO
    8421000            self.config.test = False # TODO
    … …  
    8511009    Gets id of box with these coords
    8521010    '''
    853     def getBoxId(self, config, ra, dec, boxSide):
     1011    def getBoxId(self, ra, dec):
    8541012
    8551013        sql = "SELECT id FROM box \
    856                WHERE config  = '" + config + "' \
     1014               WHERE config  = '" + self.config.name + "' \
    8571015               AND ra_center = " + str(ra) + " \
    8581016               AND dec_center = " + str(dec) + " \
    859                AND box_side = " + str(boxSide)
     1017               AND box_side = " + str(self.config.boxSize)
    8601018
    8611019        id = -1
    … …  
    8691027        return id
    8701028
    871 
    8721029    '''
    8731030    Remove all boxes and associated ids in pending table
    8741031    '''
    875     def removeAllBoxes(self, config):
    876         self.execute("DELETE FROM box WHERE config = '" + config + "'")
    877 
    878     '''
    879     Removes a box and it's associated ids in the pending table
    880     '''
    881     def removeBox(self, config, ra, dec, boxSide):
    882         self.execute("DELETE FROM box \
    883                 WHERE config = '" + config + "' \
    884                 AND ra_center = " + str(ra) + " \
    885                 AND dec_center = " + str(dec) + " \
    886                 AND box_side = " + str(boxSide))
     1032    def removeAllBoxes(self):
     1033        self.execute("DELETE FROM box WHERE config = '" + self.config.name + "'")
    8871034
    8881035    '''
    8891036    Inserts new box into box table. If it already exisits, it returns existing box id
    8901037    '''
    891     def insertBox(self, config, ra, dec, boxSide):
    892 
    893         id = self.getBoxId(config, ra, dec, boxSide)
     1038    def insertBox(self, ra, dec):
     1039
     1040        id = self.getBoxId(ra, dec)
    8941041
    8951042        # does not already exists so insert
    … …  
    8991046                   (config, ra_center, dec_center, box_side) \
    9001047                   VALUES \
    901                    ('" + config + "', " + str(ra) + ", " + str(dec) + ", " + str(boxSide) + ")"
     1048                   ('" + self.config.name + "', " + str(ra) + ", " + str(dec) + ", " + str(self.config.boxSize) + ")"
    9021049
    9031050            self.execute(sql)
    904             id = self.getBoxId(config, ra, dec, boxSide)
     1051            id = self.getBoxId(ra, dec)
    9051052
    9061053        return id
    … …  
    9121059
    9131060        # first delete old pending items
    914         self.execute("DELETE FROM pending WHERE box_id = " + str(box_id))
     1061        self.execute("DELETE FROM pending \
     1062                WHERE box_id = " + str(box_id) + " \
     1063                AND batch_type = '" + batchType + "'")
    9151064
    9161065        for id in ids:
    917             sql = "INSERT INTO pending \
     1066            sql = "INSERT IGNORE INTO pending \
    9181067               (box_id, batch_type, stage_id) \
    9191068               VALUES \
    … …  
    9231072
    9241073    '''
    925     Returns ids for all boxes with pending items
    926     '''
    927     def getBoxIds(self, config):
    928 
     1074    Returns ids for all boxes with pending items in a stripe of RA not already being processed
     1075    by another client. If none are available, then it uses the stripe already used by another
     1076    client that has the most pending items
     1077    '''
     1078    def getStripeBoxIds(self, host, pid, batchType):
     1079
     1080        self.execute("LOCK TABLES box READ, pending READ, stripe WRITE, clients WRITE")
     1081
     1082        ids = []
     1083        raCenter = -1.0
     1084        sql = "SELECT ra_center \
     1085               FROM box \
     1086               JOIN pending ON (id = box_id) \
     1087               WHERE config = '" + self.config.name + "' \
     1088               AND batch_type = '" + batchType + "' \
     1089               AND ra_center NOT IN \
     1090               (SELECT ra_center \
     1091                FROM stripe \
     1092                WHERE config = '" + self.config.name + "') \
     1093               GROUP BY ra_center LIMIT 1"
     1094
     1095        try:
     1096            rs = self.executeQuery(sql)
     1097            rs.first()
     1098            raCenter = rs.getFloat(1)
     1099            rs.close()
     1100        except:
     1101            self.logger.errorPair("No free stripes", "trying to share with other client...")
     1102            sql = "SELECT ra_center, COUNT(*) AS numPending\
     1103                   FROM box \
     1104                   JOIN pending ON (id = box_id) \
     1105                   WHERE config = '" + self.config.name + "' \
     1106                   AND batch_type = '" + batchType + "' \
     1107                   GROUP BY ra_center \
     1108                   ORDER BY numPending \
     1109                   DESC LIMIT 1"
     1110
     1111            try:
     1112                rs = self.executeQuery(sql)
     1113                rs.first()
     1114                raCenter = rs.getFloat(1)
     1115                rs.close()
     1116            except:
     1117                self.logger.errorPair("No available stripes to share", "no stripe obtained")
     1118
     1119        self.logger.infoPair("Using RA stripe", "%f" % raCenter)
    9291120        sql = "SELECT DISTINCT id \
    9301121               FROM box \
    931                JOIN pending ON (id = box_id)  \
    932                WHERE config = '" + config + "' \
    933                ORDER BY ra_center, dec_center"
    934 
    935         ids = []
     1122               JOIN pending ON (id = box_id) \
     1123               WHERE ra_center = " + str(raCenter) + " \
     1124               AND batch_type = '" + batchType + "' \
     1125               AND config = '" + self.config.name + "' \
     1126               ORDER BY dec_center DESC"
     1127
    9361128        try:
    9371129            rs = self.executeQuery(sql)
    … …  
    9391131            rs.close()
    9401132        except:
    941             self.logger.errorPair("Can't get boxes", sql)
     1133            self.logger.errorPair("Can't get stripe boxes", sql)
     1134
     1135        self.registerStripe(host, pid, raCenter)
     1136
     1137        self.unlockTables()
    9421138
    9431139        return ids
     1140
     1141    '''
     1142    Inserts row into strip table
     1143    '''
     1144    def registerStripe(self, host, pid, raCenter):
     1145       
     1146        # first, delete any existing record
     1147        try:
     1148            self.execute("DELETE FROM stripe \
     1149                    WHERE client_id = \
     1150                    (SELECT id FROM clients \
     1151                     WHERE host = '" + host + "' AND pid = " + str(pid) + ")")
     1152        except:
     1153            self.logger.errorPair("Can't delete from stripe table", sql)
     1154
     1155
     1156        # now insert new stripe entry
     1157        sql = "INSERT INTO stripe \
     1158               (client_id, config, ra_center) \
     1159               SELECT clients.id, clients.config, " + str(raCenter) + " \
     1160               FROM clients \
     1161               WHERE host = '" + host + "' \
     1162               AND pid = " + str(pid)
     1163        try:
     1164            self.execute(sql)
     1165        except:
     1166            self.logger.errorPair("Can't insert into stripe", sql)
     1167
    9441168
    9451169    '''
    … …  
    9611185
    9621186    '''
     1187    Returns ids for pending items for this config
     1188    '''
     1189    def getPendingIds(self, batchType):
     1190
     1191        sql = "SELECT DISTINCT stage_id \
     1192               FROM pending \
     1193               JOIN box ON (box_id = id) \
     1194               WHERE config = '" + self.config.name + "' \
     1195               AND batch_type = '" + batchType + "'"
     1196
     1197        ids = []
     1198        try:
     1199            rs = self.executeQuery(sql)
     1200            while (rs.next()): ids.append(rs.getInt(1))
     1201            rs.close()
     1202        except:
     1203            self.logger.errorPair("Can't get pending ids for this config", sql)
     1204
     1205        return ids
     1206
     1207    '''
    9631208    Returns ids for pending items in this box for this batch type
    9641209    '''
    965     def getPendingIds(self, boxId, batchType):
     1210    def getPendingIdsForThisBox(self, boxId, batchType):
    9661211
    9671212        sql = "SELECT stage_id FROM pending WHERE batch_type = '" + batchType + "' AND box_id = " + str(boxId)
    … …  
    9781223
    9791224    '''
    980     Returns max number of pensing items in a box for this config
    981     '''
    982     def getMaxPendingInBox(self, config):
    983 
    984         sql = "SELECT MAX(pending) \
    985                FROM (SELECT COUNT(stage_id) AS pending \
     1225    Deletes an item from the pending table
     1226    '''
     1227    def deletePendingItem(self, batchType, stageID):
     1228
     1229        sql = "DELETE pending \
     1230               FROM pending \
     1231               JOIN box ON (id = box_id) \
     1232               WHERE stage_id = " + str(stageID) + " \
     1233               AND batch_type = '" + batchType + "' \
     1234               AND config = '" + self.config.name + "'"
     1235
     1236        self.execute(sql)
     1237
     1238    '''
     1239    Writes density-plot data to file for pending stuff over sky for given config
     1240    '''
     1241    def createPendingDensityPlotData(self, batchType, DATFILE):
     1242
     1243        sql = "SELECT id, ra_center, dec_center \
     1244               FROM box \
     1245               WHERE config = '" + self.config.name + "' \
     1246               ORDER BY ra_center, dec_center"
     1247
     1248        maxPending = 0
     1249        try:
     1250            rs = self.executeQuery(sql)
     1251            while (rs.next()):
     1252               
     1253                boxId = rs.getInt(1)
     1254
     1255                sql = "SELECT COUNT(stage_id) \
    9861256                       FROM pending \
    987                        JOIN box ON (id = box_id) \
    988                        WHERE config = '" + config + "' \
    989                        GROUP BY box_id) AS t"
    990 
    991         try:
    992             rs = self.executeQuery(sql)
    993             rs.first()
    994             return rs.getInt(1)
    995         except:
    996             self.logger.errorPair("Could not count max pending", sql)
    997             return 100
    998 
    999     '''
    1000     Writes density-plot data to file for pending stuff over sky for given config
    1001     '''
    1002     def createPendingDensityPlotData(self, config, DATFILE):
    1003 
    1004         sql = "SELECT ra_center, dec_center, COUNT(stage_id) \
    1005                FROM box \
    1006                LEFT JOIN pending ON (id = box_id) \
    1007                WHERE config = '" + config + "' \
    1008                GROUP BY id \
    1009                ORDER BY ra_center, dec_center"
    1010 
     1257                       WHERE box_id = " + str(boxId) + " \
     1258                       AND batch_type = '" + batchType + "'"
     1259                rs2 = self.executeQuery(sql)
     1260                rs2.first()
     1261                max = rs2.getInt(1)
     1262                rs2.close()
     1263                print >> DATFILE, rs.getFloat(2), rs.getFloat(3), max
     1264                if max > maxPending: maxPending = max
     1265            rs.close()
     1266        except:
     1267            self.logger.errorPair("Can't get density plot data", sql)
     1268
     1269        return maxPending
     1270
     1271    '''
     1272    Creates a temporary table and shoves it full of stage_ids and ra/dec coords
     1273    '''
     1274    def storeAllItems(self, rows):
     1275
     1276        try:
     1277            self.execute("DROP TABLE all_pending")
     1278        except:
     1279            pass
     1280
     1281        self.execute("CREATE TEMPORARY TABLE all_pending (stage_id bigint(20), ra_bore float, dec_bore float)")
     1282        for row in rows:
     1283
     1284            try:
     1285                sql = "INSERT INTO all_pending (stage_id, ra_bore, dec_bore) \
     1286                       VALUES (%d, %f, %f)" % (row[0], row[1], row[2])
     1287                self.execute(sql)
     1288            except: continue
     1289
     1290        count = self.getRowCount("all_pending")
     1291        self.logger.infoPair("Items written to Db", "%d" % count)
     1292
     1293    '''
     1294    Gets all items in the all_pending temporary table within the bounds of this box
     1295    '''
     1296    def getItemsInThisThisBox(self, ra, dec):
     1297
     1298        halfSide = self.config.boxSize/2.0
     1299        minRa =  ra-halfSide
     1300        maxRa = ra+halfSide
     1301        minDec = dec-halfSide
     1302        maxDec = dec+halfSide
     1303
     1304        ids = []
     1305
     1306        sql = "SELECT DISTINCT stage_id \
     1307               FROM all_pending \
     1308               WHERE ra_bore BETWEEN " + str(minRa) + " AND " + str(maxRa) + " \
     1309               AND dec_bore BETWEEN " + str(minDec) + " AND " + str(maxDec)
     1310
     1311        try:
     1312            rs = self.executeQuery(sql)
     1313            while (rs.next()): ids.append(rs.getInt(1))
     1314        except:
     1315            self.logger.errorPair("Can't get items in this box", sql)
     1316
     1317        return ids
     1318
     1319    '''
     1320    Gets client info
     1321    '''
     1322    def getClientInfo(self):
     1323
     1324        sql = "SELECT id, host, started, timestamp, ra_center, paused, killed \
     1325               FROM stripe \
     1326               JOIN clients ON (client_id = id) \
     1327               WHERE clients.config = '" + self.config.name + "' \
     1328               ORDER BY host, started"
     1329
     1330        rows = []
    10111331        try:
    10121332            rs = self.executeQuery(sql)
    10131333            while (rs.next()):
    1014                 print >> DATFILE, rs.getFloat(1), rs.getFloat(2), rs.getInt(3)
    1015             rs.close()
    1016         except:
    1017             self.logger.errorPair("Can't get density plot data", sql)
     1334                rows.append([rs.getInt(1), rs.getString(2), rs.getString(3), rs.getString(4), rs.getFloat(5), rs.getInt(6), rs.getInt(7)])
     1335            rs.close()
     1336        except:
     1337            self.logger.errorPair("Can't query for client data", sql)
     1338
     1339        return rows
     1340
     1341    '''
     1342    Purges old clients that may have died, i.e. those with a timestamp older than n hours
     1343    '''
     1344    def purgeDeadClients(self):
     1345        self.execute("DELETE FROM clients WHERE timestamp < now() - INTERVAL 14 HOUR")
    10181346
    10191347    '''
Note: See TracChangeset for help on using the changeset viewer.