Index: trunk/ippToPsps/jython/cleanup.py =================================================================== --- trunk/ippToPsps/jython/cleanup.py (revision 33350) +++ trunk/ippToPsps/jython/cleanup.py (revision 33351) @@ -9,116 +9,130 @@ import logging.config -from config import Config -from pslogger import PSLogger -from ipptopspsdb import IppToPspsDb +from ipptopsps import IppToPsps from batch import Batch from datastore import Datastore from dxlayer import DXLayer -''' -Finds stuff to delete for this batch type, then: - -- removes from the datastore -- deletes from local disk -- deletes files from DXLayer ''' -def clean(batchType): +Cleanup class +''' +class Cleanup(IppToPsps): - logger.infoSeparator() - logger.infoTitle("Cleaning up batch type %s" % batchType) + ''' + Constructor + ''' + def __init__(self, argv): + super(Cleanup, self).__init__(argv) - loadedDatastoreIDs = ippToPspsDb.getLoadedToODMButNotDeletedFromDatastore(batchType, config.epoch, config.dvoLabel) - loadedDxlayerIDs = ippToPspsDb.getLoadedToODMButNotDeletedFromDXLayer(batchType, config.epoch, config.dvoLabel) - mergedLocalIDs = ippToPspsDb.getMergedButNotDeletedFromLocalDisk(batchType, config.epoch, config.dvoLabel) + if len(argv) > 2: self.parsePollTimeArg(sys.argv[2]) - purgedDatastoreIDs = ippToPspsDb.getPurgedButNotDeletedFromDatastore(batchType, config.epoch, config.dvoLabel) - purgedLocalIDs = ippToPspsDb.getPurgedButNotDeletedFromLocalDisk(batchType, config.epoch, config.dvoLabel) - logger.infoPair("Loaded to ODM but still on datastore", "%d" % len(loadedDatastoreIDs)) - logger.infoPair("Loaded to ODM but still in DXLayer", "%d" % len(loadedDxlayerIDs)) - logger.infoPair("Merged but still on local disk", "%d" % len(mergedLocalIDs)) - logger.infoPair("Purged but still on datastore", "%d" % len(purgedDatastoreIDs)) - logger.infoPair("Purged but still on local disk", "%d" % len(purgedLocalIDs)) + ''' + Overrides base-class version + ''' + def refreshConfig(self): + ret = super(Cleanup, self).refreshConfig() + self.dxlayer = DXLayer(self.logger) + self.datastore = Datastore(self.logger, self.config, self.ippToPspsDb) + return ret - if config.deleteLocal: - logger.infoSeparator() - count = 0 - ids = mergedLocalIDs + purgedLocalIDs - for id in ids: + ''' + Runs... + ''' + def run(self): + + if self.SECONDS: self.logger.infoPair("Performing clean-up every", "%f hours" % (self.SECONDS/60.0/60.0)) + while True: + + self.config.printDeletionPolicy() + self.clean("P2") + self.clean("ST") + + self.checkClientStatus() + if not self.waitForPollTime(): break + + ''' + Finds stuff to delete for this batch type, then: + + - removes from the datastore + - deletes from local disk + - deletes files from DXLayer + + ''' + def clean(self, batchType): + + self.logger.infoSeparator() + self.logger.infoTitle("Cleaning up batch type %s" % batchType) + + loadedDatastoreIDs = self.ippToPspsDb.getLoadedToODMButNotDeletedFromDatastore(batchType, self.config.epoch, self.config.dvoLabel) + loadedDxlayerIDs = self.ippToPspsDb.getLoadedToODMButNotDeletedFromDXLayer(batchType, self.config.epoch, self.config.dvoLabel) + mergedLocalIDs = self.ippToPspsDb.getMergedButNotDeletedFromLocalDisk(batchType, self.config.epoch, self.config.dvoLabel) + + purgedDatastoreIDs = self.ippToPspsDb.getPurgedButNotDeletedFromDatastore(batchType, self.config.epoch, self.config.dvoLabel) + purgedLocalIDs = self.ippToPspsDb.getPurgedButNotDeletedFromLocalDisk(batchType, self.config.epoch, self.config.dvoLabel) + + self.logger.infoPair("Loaded to ODM but still on datastore", "%d" % len(loadedDatastoreIDs)) + self.logger.infoPair("Loaded to ODM but still in DXLayer", "%d" % len(loadedDxlayerIDs)) + self.logger.infoPair("Merged but still on local disk", "%d" % len(mergedLocalIDs)) + self.logger.infoPair("Purged but still on datastore", "%d" % len(purgedDatastoreIDs)) + self.logger.infoPair("Purged but still on local disk", "%d" % len(purgedLocalIDs)) - if Batch.deleteFromDisk(logger, config.basePath, batchType, config.dvoLabel, id): - ippToPspsDb.updateDeletedLocal(id, 1) - count = count + 1 + if self.config.deleteLocal: + self.logger.infoSeparator() + count = 0 + ids = mergedLocalIDs + purgedLocalIDs + for id in ids: + + if Batch.deleteFromDisk(self.logger, self.config.basePath, batchType, self.config.dvoLabel, id): + self.ippToPspsDb.updateDeletedLocal(id, 1) + count = count + 1 + + self.logger.infoSeparator() + self.logger.infoPair("Removed from local disk", "%d" % count) + self.logger.infoPair("Remaining on local disk", "%d" % (len(ids) - count)) + + + if self.config.deleteDatastore: + self.logger.infoSeparator() + count = 0 + ids = loadedDatastoreIDs + purgedDatastoreIDs + for id in ids: + + batchName = Batch.getNameFromID(id) + + if self.datastore.remove(batchName): + count = count + 1 + + self.logger.infoSeparator() + self.logger.infoPair("Removed from datastore", "%d" % count) + self.logger.infoPair("Remaining on datastore", "%d" % (len(ids) - count)) + + if self.config.deleteDxLayer: + self.logger.infoSeparator() + count = 0 + for id in loadedDxlayerIDs: + + if self.dxlayer.deleteBatch(id): + self.ippToPspsDb.updateDeletedDxlayer(id, 1) + count = count + 1 + + self.logger.infoSeparator() + self.logger.infoPair("Removed from DXLayer", "%d" % count) + self.logger.infoPair("Remaining on DXLayer", "%d" % (len(loadedDxlayerIDs) - count)) + - logger.infoSeparator() - logger.infoPair("Removed from local disk", "%d" % count) - logger.infoPair("Remaining on local disk", "%d" % (len(ids) - count)) - + ''' + Overrides base-class version + ''' + def printUsage(self): + super(Cleanup, self).printUsage() + print " []" - if config.deleteDatastore: - logger.infoSeparator() - count = 0 - ids = loadedDatastoreIDs + purgedDatastoreIDs - for id in ids: - - batchName = Batch.getNameFromID(id) - - if datastore.remove(batchName): - count = count + 1 - - logger.infoSeparator() - logger.infoPair("Removed from datastore", "%d" % count) - logger.infoPair("Remaining on datastore", "%d" % (len(ids) - count)) - - if config.deleteDxLayer: - logger.infoSeparator() - count = 0 - for id in loadedDxlayerIDs: - - if dxlayer.deleteBatch(id): - ippToPspsDb.updateDeletedDxlayer(id, 1) - count = count + 1 - - logger.infoSeparator() - logger.infoPair("Removed from DXLayer", "%d" % count) - logger.infoPair("Remaining on DXLayer", "%d" % (len(loadedDxlayerIDs) - count)) - ''' Start of program ''' - -if len(sys.argv) > 1: CONFIGPATH = sys.argv[1] -else: - print "\n** Usage: " + sys.argv[0] + " [hours]\n" - sys.exit(1) - -if len(sys.argv) > 2: - HOURS = float(sys.argv[2]) - SECONDS = HOURS * 60.0 * 60.0 -else: - SECONDS = None - -# create objects -config = Config(CONFIGPATH) -logger = config.getLogger(sys.argv[0]) -ippToPspsDb = IppToPspsDb(logger, config) -datastore = Datastore(logger, config, ippToPspsDb) -dxlayer = DXLayer(logger) - - -logger.infoTitle("ippToPsps clean-up") -if SECONDS: logger.infoPair("Performing clean-up every", "%f hours" % (SECONDS/60.0/60.0)) -while True: - - config.printDeletionPolicy() - clean("P2") - clean("ST") - - if SECONDS: - logger.infoPair("Waiting for", "%f hours" % HOURS) - time.sleep(SECONDS) - else: break - config.refresh() - +cleanup = Cleanup(sys.argv) +cleanup.run() +cleanup.exitProgram("finished") Index: trunk/ippToPsps/jython/metrics.py =================================================================== --- trunk/ippToPsps/jython/metrics.py (revision 33350) +++ trunk/ippToPsps/jython/metrics.py (revision 33351) @@ -7,183 +7,169 @@ import datetime -from xml.etree.ElementTree import ElementTree, Element, tostring - -from config import Config -from pslogger import PSLogger -from ipptopspsdb import IppToPspsDb +from ipptopsps import IppToPsps from czardb import CzarDb from gpc1db import Gpc1Db ''' -Makes a gnuplot of the data +Metrics class ''' -def plotMe(batchType, file): +class Metrics(IppToPsps): - OUTPUTFILE = "plots/" + config.dvoLabel + "_" + batchType + ".png" - f=os.popen('gnuplot', 'w') + ''' + Constructor + ''' + def __init__(self, argv): + super(Metrics, self).__init__(argv) - if 0: - TERM = "X11" - else: - TERM = "png font \"/usr/share/fonts/corefonts/arial.ttf\" 8" + if len(sys.argv) > 2: self.parsePollTimeArg(sys.argv[2]) + + # create database objects + self.czarDb = CzarDb(self.logger, self.config) + self.gpc1Db = Gpc1Db(self.logger, self.config) + + ''' + Run... + ''' + def run(self): + + while True: + + self.logger.infoTitle("ippToPsps loading summary") + now = datetime.datetime.now() + self.logger.infoPair("Time now", now.strftime("%Y-%m-%d %H:%M:%S")) + self.logger.infoPair("Loading epoch", self.config.epoch) + self.logger.infoPair("DVO label", self.config.dvoLabel) + self.config.printBoxCoords() - print >> f, "set term " + TERM + "; \ - set output \"" + OUTPUTFILE + "\"; \ - set title \"ippToPsps : " + batchType + " loading for " + config.dvoLabel + "\"; \ - set boxwidth; \ - set xtic rotate by -90 scale 0; \ - set style data histogram; \ - set style histogram rowstacked; \ - set style fill solid border -1; \ - set ylabel \"Items processed per loading stage\"; \ - set boxwidth 0.75; \ - plot '" + file + "' using 2:xtic(1) notitle lt 2, '' using 3 notitle lt 1, '' using 4 notitle lt rgb \"#808080\"" + self.logger.info("+----+------------------+---------------+-------------------+------------------+----------------+") + self.logger.info("|Type| batches per hour | last 24 hours | per day this week | total detections | last published |") + self.logger.info("+----+------------------+---------------+-------------------+------------------+----------------+") + rateP2 = self.printStats("P2") + rateST = self.printStats("ST") + self.logger.info("+----+------------------+---------------+-------------------+------------------+----------------+") + + self.logger.info("") + + stages = self.ippToPspsDb.getStages() + colCount = len(stages) + self.writeTableSeparator(colCount) + sys.stdout.write("|Type| DVO ") + + for stage in stages: sys.stdout.write("|%19s" % stage) + sys.stdout.write("|\n") + sys.stdout.write("| | ") + for stage in stages: sys.stdout.write("| Pend Succ Fail ") + sys.stdout.write("|\n") + self.writeTableSeparator(colCount) + pendP2Processed = self.printTableRow(stages, "P2") + pendSTProcessed = self.printTableRow(stages, "ST") + self.writeTableSeparator(colCount) - f.flush() + self.logger.info("") + try: self.logger.infoPair("Estimated time for P2s", "%.1f hours" % (pendP2Processed / rateP2)) + except: pass + try: self.logger.infoPair("Estimated time for STs", "%.1f hours" % (pendSTProcessed / rateST)) + except: pass + + self.checkClientStatus() + if not self.waitForPollTime(): break -''' -Formats an int to a string -''' -def getIntAsString(value): - if value == 0: return "" - return str(value) -''' -Prints (to the log) one row of data (one row being one batch-type) -''' -def printTableRow(stages, batchType): + ''' + Formats an int to a string + ''' + def getIntAsString(self, value): + if value == 0: return "" + return str(value) + + ''' + Prints (to the log) one row of data (one row being one batch-type) + ''' + def printTableRow(self, stages, batchType): + + # get a master list of IDs in DVO for this batch type + #allIDs = self.gpc1Db.getIDsInThisDVODbForThisStage(self.config.dvoLabel, batchType, self.config.minRa, self.config.maxRa, self.config.minDec, self.config.maxDec) + allIDs = self.gpc1Db.getIDsInThisDVODbForThisStage(self.config.dvoLabel, batchType) + prevList = allIDs + numEverything = len(allIDs) + + sys.stdout.write("| %2s |%6d " % (batchType, numEverything)) + + tempFilename = batchType + '.dat' + DATFILE = open(tempFilename,'w') + print >> DATFILE, "DVO", numEverything, 0, 0 + for stage in stages: + + # get lists. Use unions with prev list to make sure the right stuff is included + success = list(set(self.ippToPspsDb.getStageIDs(batchType, self.config.epoch, self.config.dvoLabel, stage, 1)) & set(prevList)) + fail = list(set(self.ippToPspsDb.getStageIDs(batchType, self.config.epoch, self.config.dvoLabel, stage, -1)) & set(prevList) - set(success)) + pending = list(set(prevList) - set(success) - set(fail)) + + # count stuff + numSuccess = len(success) + numFail = len(fail) + numPending = len(pending) + + #if stage == 'loaded_to_datastore' and batchType == "P2": + # for n in fail: print n + + str = "\033[1;34m%5s\033[1;m \033[1;32m%5s\033[1;m \033[1;31m%5s\033[1;m " % (self.getIntAsString(numPending), self.getIntAsString(numSuccess), self.getIntAsString(numFail)) + sys.stdout.write("|%-19s" % str) + + prevList = success + print >> DATFILE, stage, numSuccess, numFail, numPending + + self.czarDb.insertStats(stage, self.config.dvoLabel, batchType, numPending, numSuccess, numFail) + + if stage == 'processed': numPendingProcessed = numPending + + sys.stdout.write("|\n") + DATFILE.close() + + return numPendingProcessed + + ''' + Writes table separator + ''' + def writeTableSeparator(self, colCount): + + sys.stdout.write("+----+-------") + for i in range(0, colCount): sys.stdout.write("+-------------------") + sys.stdout.write("+\n") + + ''' + Prints various timing stats for this batch type + ''' + def printStats(self, batchType): + + rate = self.ippToPspsDb.countBatchesInLastPeriod(batchType, self.config.epoch, self.config.dvoLabel, "1 HOUR") / 1.0 + + self.logger.info("| %2s | %16.1f | %13d | %17.1f | %16d | %14s |" % + (batchType, + rate, + self.ippToPspsDb.countBatchesInLastPeriod(batchType, self.config.epoch, self.config.dvoLabel, "24 HOUR"), + self.ippToPspsDb.countBatchesInLastPeriod(batchType, self.config.epoch, self.config.dvoLabel, "1 WEEK") / 7.0, + self.ippToPspsDb.getTotalDetectionsPublished(batchType, self.config.epoch, self.config.dvoLabel), + self.ippToPspsDb.getTimeOfLastBatchPublished(batchType, self.config.epoch, self.config.dvoLabel))) + + return rate + - # get a master list of IDs in DVO for this batch type - #allIDs = gpc1Db.getIDsInThisDVODbForThisStage(config.dvoLabel, batchType, config.minRa, config.maxRa, config.minDec, config.maxDec) - allIDs = gpc1Db.getIDsInThisDVODbForThisStage(config.dvoLabel, batchType) - prevList = allIDs - numEverything = len(allIDs) + ''' + Overrides base-class version + ''' + def printUsage(self): + super(Metrics, self).printUsage() + print " [hours]" - sys.stdout.write("| %2s |%6d " % (batchType, numEverything)) - tempFilename = batchType + '.dat' - DATFILE = open(tempFilename,'w') - print >> DATFILE, "DVO", numEverything, 0, 0 - for stage in stages: - - # get lists. Use unions with prev list to make sure the right stuff is included - success = list(set(ippToPspsDb.getStageIDs(batchType, config.epoch, config.dvoLabel, stage, 1)) & set(prevList)) - fail = list(set(ippToPspsDb.getStageIDs(batchType, config.epoch, config.dvoLabel, stage, -1)) & set(prevList) - set(success)) - pending = list(set(prevList) - set(success) - set(fail)) - - # count stuff - numSuccess = len(success) - numFail = len(fail) - numPending = len(pending) - - #if stage == 'loaded_to_datastore' and batchType == "P2": - # for n in fail: print n - - str = "\033[1;34m%5s\033[1;m \033[1;32m%5s\033[1;m \033[1;31m%5s\033[1;m " % (getIntAsString(numPending), getIntAsString(numSuccess), getIntAsString(numFail)) - sys.stdout.write("|%-19s" % str) - - prevList = success - print >> DATFILE, stage, numSuccess, numFail, numPending - - czarDb.insertStats(stage, config.dvoLabel, batchType, numPending, numSuccess, numFail) - - if stage == 'processed': numPendingProcessed = numPending - - sys.stdout.write("|\n") - #plotMe(batchType, tempFilename) - DATFILE.close() - - return numPendingProcessed - -''' -Writes table separator -''' -def writetableSeparator(colCount): - - sys.stdout.write("+----+-------") - for i in range(0, colCount): sys.stdout.write("+-------------------") - sys.stdout.write("+\n") - -''' -Prints various timing stats for this batch type -''' -def printStats(batchType): - - rate = ippToPspsDb.countBatchesInLastPeriod(batchType, config.epoch, config.dvoLabel, "1 HOUR") / 1.0 - - logger.info("| %2s | %16.1f | %13d | %17.1f | %16d | %14s |" % - (batchType, - rate, - ippToPspsDb.countBatchesInLastPeriod(batchType, config.epoch, config.dvoLabel, "24 HOUR"), - ippToPspsDb.countBatchesInLastPeriod(batchType, config.epoch, config.dvoLabel, "1 WEEK") / 7.0, - ippToPspsDb.getTotalDetectionsPublished(batchType, config.epoch, config.dvoLabel), - ippToPspsDb.getTimeOfLastBatchPublished(batchType, config.epoch, config.dvoLabel))) - - return rate ''' Start of program. ''' -if len(sys.argv) > 1: CONFIGPATH = sys.argv[1] -else: - print "** Usage: " + sys.argv[0] + " [hours]" - sys.exit(1) +metrics = Metrics(sys.argv) +metrics.run() +metrics.exitProgram("finished") -if len(sys.argv) > 2: - HOURS = float(sys.argv[2]) - SECONDS = HOURS * 60.0 * 60.0 -else: - SECONDS = None -# create objects -config = Config(CONFIGPATH) -logger = config.getLogger(sys.argv[0]) -ippToPspsDb = IppToPspsDb(logger, config) -czarDb = CzarDb(logger, config) -gpc1Db = Gpc1Db(logger, config) -while True: - - logger.infoTitle("ippToPsps loading summary") - now = datetime.datetime.now() - logger.infoPair("Time now", now.strftime("%Y-%m-%d %H:%M:%S")) - logger.infoPair("Loading epoch", config.epoch) - logger.infoPair("DVO label", config.dvoLabel) - config.printBoxCoords() - - logger.info("+----+------------------+---------------+-------------------+------------------+----------------+") - logger.info("|Type| batches per hour | last 24 hours | per day this week | total detections | last published |") - logger.info("+----+------------------+---------------+-------------------+------------------+----------------+") - rateP2 = printStats("P2") - rateST = printStats("ST") - logger.info("+----+------------------+---------------+-------------------+------------------+----------------+") - - logger.info("") - - stages = ippToPspsDb.getStages() - colCount = len(stages) - writetableSeparator(colCount) - sys.stdout.write("|Type| DVO ") - - for stage in stages: sys.stdout.write("|%19s" % stage) - sys.stdout.write("|\n") - sys.stdout.write("| | ") - for stage in stages: sys.stdout.write("| Pend Succ Fail ") - sys.stdout.write("|\n") - writetableSeparator(colCount) - pendP2Processed = printTableRow(stages, "P2") - pendSTProcessed = printTableRow(stages, "ST") - writetableSeparator(colCount) - - logger.info("") - try: logger.infoPair("Estimated time for P2s", "%.1f hours" % (pendP2Processed / rateP2)) - except: pass - try: logger.infoPair("Estimated time for STs", "%.1f hours" % (pendSTProcessed / rateST)) - except: pass - - if SECONDS: - logger.infoPair("Waiting for", "%f hours" % HOURS) - time.sleep(SECONDS) - else: break - config.refresh() - Index: trunk/ippToPsps/jython/pollOdm.py =================================================================== --- trunk/ippToPsps/jython/pollOdm.py (revision 33350) +++ trunk/ippToPsps/jython/pollOdm.py (revision 33351) @@ -8,109 +8,131 @@ import time -import logging.config - +from ipptopsps import IppToPsps from config import Config -from pslogger import PSLogger -from ipptopspsdb import IppToPspsDb from odm import Odm from batch import Batch + ''' -Method to actually poll ODM +PollOdm class ''' -def poll(batchType, stage): - - if stage == "unloaded": - ids = ippToPspsDb.getUnloadedBatchIDs(config.epoch, config.dvoLabel, batchType) - elif stage == "unmergeworthy": - ids = ippToPspsDb.getUnmergeWorthyBatchIDs(config.epoch, config.dvoLabel, batchType) - elif stage == "unmerged": - ids = ippToPspsDb.getUnmergedBatchIDs(config.epoch, config.dvoLabel, batchType) +class PollOdm(IppToPsps): - logger.infoTitle("Checking %d %s %s batches" % (len(ids), stage, batchType)) + ''' + Constructor + ''' + def __init__(self, argv): + super(PollOdm, self).__init__(argv) - results = {} - newLoaded = 0 - newMergeWorthy = 0 - newMerged = 0 + # get batch and stage args + if len(sys.argv) > 3: + self.BATCHTYPE = sys.argv[2] + self.STAGE = sys.argv[3] + else: + self.printUsage() + self.exitProgram("incorrect args") - logger.info("+-----------+----------------+----------------+----------------+----------------+----------------+----------------+") - logger.info("| Batch | Loaded to ODM | Merge worthy | Merged | New loaded |New merge worthy| New merged |") - logger.info("+-----------+----------------+----------------+----------------+----------------+----------------+----------------+") - for id in ids: + # get poll time arg + if len(sys.argv) > 4: self.parsePollTimeArg(sys.argv[4]) - if odm.checkBatch(id, results): - ippToPspsDb.updateOdmStatus(id, results) + if self.STAGE == "all": self.stages = ['unloaded', 'unmergeworthy', 'unmerged'] + else: self.stages = [self.STAGE] - if stage == "unloaded": - if results['LOADEDTOODM'] == 1: newLoaded = newLoaded + 1 - if results['MERGEWORTHY'] == 1: newMergeWorthy = newMergeWorthy + 1 - if results['MERGED'] == 1: newMerged = newMerged + 1 + if self.BATCHTYPE == "all": self.batchTypes = ['P2', 'ST'] + else: self.batchTypes = [self.BATCHTYPE] - elif stage == "unmergeworthy": - if results['MERGEWORTHY'] == 1: newMergeWorthy = newMergeWorthy + 1 - if results['MERGED'] == 1: newMerged = newMerged + 1 + for stage in self.stages: self.logger.infoPair("Checking stage", stage) + for batchType in self.batchTypes: self.logger.infoPair("Checking batch type", batchType) - elif stage == "unmerged": - if results['MERGED'] == 1: newMerged = newMerged + 1 + ''' + Overrides base-class version + ''' + def refreshConfig(self): + ret = super(PollOdm, self).refreshConfig() + self.odm = Odm(self.logger, self.config) + return ret - logger.info("| %9s | %14d | %14d | %14d | %14d | %14d | %14d |" % - (Batch.getNameFromID(id), - results['LOADEDTOODM'], + ''' + Run... + ''' + def run(self): + + while True: + + for stage in self.stages: + for batchType in self.batchTypes: + self.poll(batchType, stage) + + self.checkClientStatus() + if not self.waitForPollTime(): break + + ''' + Method to actually poll ODM + ''' + def poll(self, batchType, stage): + + if stage == "unloaded": + ids = self.ippToPspsDb.getUnloadedBatchIDs(self.config.epoch, self.config.dvoLabel, batchType) + elif stage == "unmergeworthy": + ids = self.ippToPspsDb.getUnmergeWorthyBatchIDs(self.config.epoch, self.config.dvoLabel, batchType) + elif stage == "unmerged": + ids = self.ippToPspsDb.getUnmergedBatchIDs(self.config.epoch, self.config.dvoLabel, batchType) + + self.logger.infoTitle("Checking %d %s %s batches" % (len(ids), stage, batchType)) + + results = {} + newLoaded = 0 + newMergeWorthy = 0 + newMerged = 0 + + self.logger.info("+-----------+----------------+----------------+----------------+----------------+----------------+----------------+") + self.logger.info("| Batch | Loaded to ODM | Merge worthy | Merged | New loaded |New merge worthy| New merged |") + self.logger.info("+-----------+----------------+----------------+----------------+----------------+----------------+----------------+") + for id in ids: + + if self.odm.checkBatch(id, results): + self.ippToPspsDb.updateOdmStatus(id, results) + + if stage == "unloaded": + if results['LOADEDTOODM'] == 1: newLoaded = newLoaded + 1 + if results['MERGEWORTHY'] == 1: newMergeWorthy = newMergeWorthy + 1 + if results['MERGED'] == 1: newMerged = newMerged + 1 + + elif stage == "unmergeworthy": + if results['MERGEWORTHY'] == 1: newMergeWorthy = newMergeWorthy + 1 + if results['MERGED'] == 1: newMerged = newMerged + 1 + + elif stage == "unmerged": + if results['MERGED'] == 1: newMerged = newMerged + 1 + + self.logger.info("| %9s | %14d | %14d | %14d | %14d | %14d | %14d |" % + (Batch.getNameFromID(id), + results['LOADEDTOODM'], results['MERGEWORTHY'], - results['MERGED'], - newLoaded, - newMergeWorthy, - newMerged)) + results['MERGED'], + newLoaded, + newMergeWorthy, + newMerged)) + + + else: + self.logger.info("| %9s | failed to poll ODM |" % Batch.getNameFromID(id)) + + self.logger.info("+-----------+----------------+----------------+----------------+----------------+----------------+----------------+") + - - else: - logger.info("| %9s | failed to poll ODM |" % Batch.getNameFromID(id)) - - logger.info("+-----------+----------------+----------------+----------------+----------------+----------------+----------------+") - - + ''' + Overrides base-class version + ''' + def printUsage(self): + super(PollOdm, self).printUsage() + print " [hours]" + + + ''' Program starts here ''' -if len(sys.argv) > 3: - CONFIGPATH = sys.argv[1] - BATCHTYPE = sys.argv[2] - STAGE = sys.argv[3] -else: - print "\n** Usage: " + sys.argv[0] + " [hours]\n" - sys.exit(1) - -if len(sys.argv) > 4: - HOURS = float(sys.argv[4]) - SECONDS = HOURS * 60.0 * 60.0 -else: - SECONDS = None - -# create objects -config = Config(CONFIGPATH) -logger = config.getLogger(sys.argv[0]) -ippToPspsDb = IppToPspsDb(logger, config) -odm = Odm(logger, config) - -if STAGE == "all": stages = ['unloaded', 'unmergeworthy', 'unmerged'] -else: stages = [STAGE] - -if BATCHTYPE == "all": batchTypes = ['P2', 'ST'] -else: batchTypes = [BATCHTYPE] - -for stage in stages: logger.infoPair("Checking stage", stage) -for batchType in batchTypes: logger.infoPair("Checking batch type", batchType) - -while True: - - for stage in stages: - for batchType in batchTypes: - poll(batchType, stage) - - if SECONDS: - logger.infoPair("Waiting for", "%f hours" % HOURS) - time.sleep(SECONDS) - else: break - - +pollodm = PollOdm(sys.argv) +pollodm.run() +pollodm.exitProgram("finished")