IPP Software Navigation Tools IPP Links Communication Pan-STARRS Links

Changeset 33351 for trunk/ippToPsps


Ignore:
Timestamp:
Feb 23, 2012, 4:14:32 PM (14 years ago)
Author:
rhenders
Message:

now a subclass of IppToPsps

Location:
trunk/ippToPsps/jython
Files:
3 edited

Legend:

Unmodified
Added
Removed

  • trunk/ippToPsps/jython/cleanup.py

    r33265 r33351  
    99import logging.config
    1010
    11 from config import Config
    12 from pslogger import PSLogger
    13 from ipptopspsdb import IppToPspsDb
     11from ipptopsps import IppToPsps
    1412from batch import Batch
    1513from datastore import Datastore
    1614from dxlayer import DXLayer
    1715
    18 '''
    19 Finds stuff to delete for this batch type, then:
    20 
    21 - removes from the datastore
    22 - deletes from local disk
    23 - deletes files from DXLayer
    2416
    2517'''
    26 def clean(batchType):
     18Cleanup class
     19'''
     20class Cleanup(IppToPsps):
    2721
    28     logger.infoSeparator()
    29     logger.infoTitle("Cleaning up batch type %s" % batchType)
     22    '''
     23    Constructor
     24    '''
     25    def __init__(self, argv):
     26        super(Cleanup, self).__init__(argv)
    3027
    31     loadedDatastoreIDs = ippToPspsDb.getLoadedToODMButNotDeletedFromDatastore(batchType, config.epoch, config.dvoLabel)
    32     loadedDxlayerIDs = ippToPspsDb.getLoadedToODMButNotDeletedFromDXLayer(batchType, config.epoch, config.dvoLabel)
    33     mergedLocalIDs = ippToPspsDb.getMergedButNotDeletedFromLocalDisk(batchType, config.epoch, config.dvoLabel)
     28        if len(argv) > 2: self.parsePollTimeArg(sys.argv[2])
    3429
    35     purgedDatastoreIDs = ippToPspsDb.getPurgedButNotDeletedFromDatastore(batchType, config.epoch, config.dvoLabel)
    36     purgedLocalIDs = ippToPspsDb.getPurgedButNotDeletedFromLocalDisk(batchType, config.epoch, config.dvoLabel)
    3730
    38     logger.infoPair("Loaded to ODM but still on datastore", "%d" % len(loadedDatastoreIDs))
    39     logger.infoPair("Loaded to ODM but still in DXLayer", "%d" % len(loadedDxlayerIDs))
    40     logger.infoPair("Merged but still on local disk", "%d" % len(mergedLocalIDs))
    41     logger.infoPair("Purged but still on datastore", "%d" % len(purgedDatastoreIDs))
    42     logger.infoPair("Purged but still on local disk", "%d" % len(purgedLocalIDs))
     31    '''
     32    Overrides base-class version
     33    '''
     34    def refreshConfig(self):
     35        ret = super(Cleanup, self).refreshConfig()
     36        self.dxlayer = DXLayer(self.logger)
     37        self.datastore = Datastore(self.logger, self.config, self.ippToPspsDb)
     38        return ret
    4339
    44     if config.deleteLocal:
    45         logger.infoSeparator()
    46         count = 0
    47         ids = mergedLocalIDs + purgedLocalIDs
    48         for id in ids:
     40    '''
     41    Runs...
     42    '''
     43    def run(self):
     44
     45        if self.SECONDS: self.logger.infoPair("Performing clean-up every", "%f hours" % (self.SECONDS/60.0/60.0))
     46        while True:
     47
     48            self.config.printDeletionPolicy()
     49            self.clean("P2")
     50            self.clean("ST")
     51
     52            self.checkClientStatus()
     53            if not self.waitForPollTime(): break
     54   
     55    '''
     56    Finds stuff to delete for this batch type, then:
     57
     58    - removes from the datastore
     59    - deletes from local disk
     60    - deletes files from DXLayer
     61
     62    '''
     63    def clean(self, batchType):
     64
     65        self.logger.infoSeparator()
     66        self.logger.infoTitle("Cleaning up batch type %s" % batchType)
     67
     68        loadedDatastoreIDs = self.ippToPspsDb.getLoadedToODMButNotDeletedFromDatastore(batchType, self.config.epoch, self.config.dvoLabel)
     69        loadedDxlayerIDs = self.ippToPspsDb.getLoadedToODMButNotDeletedFromDXLayer(batchType, self.config.epoch, self.config.dvoLabel)
     70        mergedLocalIDs = self.ippToPspsDb.getMergedButNotDeletedFromLocalDisk(batchType, self.config.epoch, self.config.dvoLabel)
     71
     72        purgedDatastoreIDs = self.ippToPspsDb.getPurgedButNotDeletedFromDatastore(batchType, self.config.epoch, self.config.dvoLabel)
     73        purgedLocalIDs = self.ippToPspsDb.getPurgedButNotDeletedFromLocalDisk(batchType, self.config.epoch, self.config.dvoLabel)
     74
     75        self.logger.infoPair("Loaded to ODM but still on datastore", "%d" % len(loadedDatastoreIDs))
     76        self.logger.infoPair("Loaded to ODM but still in DXLayer", "%d" % len(loadedDxlayerIDs))
     77        self.logger.infoPair("Merged but still on local disk", "%d" % len(mergedLocalIDs))
     78        self.logger.infoPair("Purged but still on datastore", "%d" % len(purgedDatastoreIDs))
     79        self.logger.infoPair("Purged but still on local disk", "%d" % len(purgedLocalIDs))
    4980   
    50             if Batch.deleteFromDisk(logger, config.basePath, batchType, config.dvoLabel, id):
    51                 ippToPspsDb.updateDeletedLocal(id, 1)
    52                 count = count + 1
     81        if self.config.deleteLocal:
     82            self.logger.infoSeparator()
     83            count = 0
     84            ids = mergedLocalIDs + purgedLocalIDs
     85            for id in ids:
     86       
     87                if Batch.deleteFromDisk(self.logger, self.config.basePath, batchType, self.config.dvoLabel, id):
     88                    self.ippToPspsDb.updateDeletedLocal(id, 1)
     89                    count = count + 1
     90   
     91            self.logger.infoSeparator()
     92            self.logger.infoPair("Removed from local disk", "%d" % count)
     93            self.logger.infoPair("Remaining on local disk", "%d" % (len(ids) - count))
     94       
     95   
     96        if self.config.deleteDatastore:
     97            self.logger.infoSeparator()
     98            count = 0
     99            ids = loadedDatastoreIDs + purgedDatastoreIDs
     100            for id in ids:
     101       
     102                batchName = Batch.getNameFromID(id)
     103   
     104                if self.datastore.remove(batchName):
     105                    count = count + 1
     106   
     107            self.logger.infoSeparator()
     108            self.logger.infoPair("Removed from datastore", "%d" % count)
     109            self.logger.infoPair("Remaining on datastore", "%d" % (len(ids) - count))
     110       
     111        if self.config.deleteDxLayer:
     112            self.logger.infoSeparator()
     113            count = 0
     114            for id in loadedDxlayerIDs:
     115       
     116                if self.dxlayer.deleteBatch(id):
     117                    self.ippToPspsDb.updateDeletedDxlayer(id, 1)
     118                    count = count + 1
     119   
     120            self.logger.infoSeparator()
     121            self.logger.infoPair("Removed from DXLayer", "%d" % count)
     122            self.logger.infoPair("Remaining on DXLayer", "%d" % (len(loadedDxlayerIDs) - count))
     123       
    53124
    54         logger.infoSeparator()
    55         logger.infoPair("Removed from local disk", "%d" % count)
    56         logger.infoPair("Remaining on local disk", "%d" % (len(ids) - count))
    57    
     125    '''
     126    Overrides base-class version
     127    '''
     128    def printUsage(self):
     129        super(Cleanup, self).printUsage()
     130        print " [<poll time in hours>]"
    58131
    59     if config.deleteDatastore:
    60         logger.infoSeparator()
    61         count = 0
    62         ids = loadedDatastoreIDs + purgedDatastoreIDs
    63         for id in ids:
    64    
    65             batchName = Batch.getNameFromID(id)
    66 
    67             if datastore.remove(batchName):
    68                 count = count + 1
    69 
    70         logger.infoSeparator()
    71         logger.infoPair("Removed from datastore", "%d" % count)
    72         logger.infoPair("Remaining on datastore", "%d" % (len(ids) - count))
    73    
    74     if config.deleteDxLayer:
    75         logger.infoSeparator()
    76         count = 0
    77         for id in loadedDxlayerIDs:
    78    
    79             if dxlayer.deleteBatch(id):
    80                 ippToPspsDb.updateDeletedDxlayer(id, 1)
    81                 count = count + 1
    82 
    83         logger.infoSeparator()
    84         logger.infoPair("Removed from DXLayer", "%d" % count)
    85         logger.infoPair("Remaining on DXLayer", "%d" % (len(loadedDxlayerIDs) - count))
    86    
    87132
    88133'''
    89134Start of program
    90135'''
    91 
    92 if len(sys.argv) > 1: CONFIGPATH  = sys.argv[1]
    93 else:
    94     print "\n** Usage: " + sys.argv[0] + " <configPath> [hours]\n"
    95     sys.exit(1)
    96 
    97 if len(sys.argv) > 2:
    98     HOURS = float(sys.argv[2])
    99     SECONDS = HOURS * 60.0 * 60.0
    100 else:
    101     SECONDS = None
    102 
    103 # create objects
    104 config = Config(CONFIGPATH)
    105 logger = config.getLogger(sys.argv[0])
    106 ippToPspsDb = IppToPspsDb(logger, config)
    107 datastore = Datastore(logger, config, ippToPspsDb)
    108 dxlayer = DXLayer(logger)
    109 
    110 
    111 logger.infoTitle("ippToPsps clean-up")
    112 if SECONDS: logger.infoPair("Performing clean-up every", "%f hours" % (SECONDS/60.0/60.0))
    113 while True:
    114 
    115     config.printDeletionPolicy()
    116     clean("P2")
    117     clean("ST")
    118 
    119     if SECONDS:
    120         logger.infoPair("Waiting for", "%f hours" % HOURS)
    121         time.sleep(SECONDS)
    122     else: break
    123     config.refresh()
    124    
     136cleanup = Cleanup(sys.argv)
     137cleanup.run()
     138cleanup.exitProgram("finished")

  • trunk/ippToPsps/jython/metrics.py

    r33289 r33351  
    77import datetime
    88
    9 from xml.etree.ElementTree import ElementTree, Element, tostring
    10 
    11 from config import Config
    12 from pslogger import PSLogger
    13 from ipptopspsdb import IppToPspsDb
     9from ipptopsps import IppToPsps
    1410from czardb import CzarDb
    1511from gpc1db import Gpc1Db
    1612
    1713'''
    18 Makes a gnuplot of the data
     14Metrics class
    1915'''
    20 def plotMe(batchType, file):
     16class Metrics(IppToPsps):
    2117
    22     OUTPUTFILE = "plots/" + config.dvoLabel + "_" + batchType + ".png"
    23     f=os.popen('gnuplot', 'w')
     18    '''
     19    Constructor
     20    '''
     21    def __init__(self, argv):
     22        super(Metrics, self).__init__(argv)
    2423
    25     if 0:
    26         TERM = "X11"
    27     else:
    28         TERM = "png font \"/usr/share/fonts/corefonts/arial.ttf\" 8"
     24        if len(sys.argv) > 2: self.parsePollTimeArg(sys.argv[2])
     25   
     26        # create database objects
     27        self.czarDb = CzarDb(self.logger, self.config)
     28        self.gpc1Db = Gpc1Db(self.logger, self.config)
     29   
     30    '''
     31    Run...
     32    '''
     33    def run(self):
     34
     35        while True:
     36   
     37            self.logger.infoTitle("ippToPsps loading summary")
     38            now = datetime.datetime.now()
     39            self.logger.infoPair("Time now", now.strftime("%Y-%m-%d %H:%M:%S"))
     40            self.logger.infoPair("Loading epoch", self.config.epoch)
     41            self.logger.infoPair("DVO label", self.config.dvoLabel)
     42            self.config.printBoxCoords()
    2943       
    30     print >> f, "set term " + TERM + "; \
    31               set output \"" + OUTPUTFILE + "\"; \
    32               set title \"ippToPsps : " + batchType + " loading for " + config.dvoLabel + "\"; \
    33               set boxwidth; \
    34               set xtic rotate by -90 scale 0; \
    35               set style data histogram; \
    36               set style histogram rowstacked; \
    37               set style fill solid border -1; \
    38               set ylabel \"Items processed per loading stage\"; \
    39               set boxwidth 0.75; \
    40               plot '" + file + "' using 2:xtic(1) notitle lt 2, '' using 3 notitle lt 1, '' using 4 notitle lt rgb \"#808080\""
     44            self.logger.info("+----+------------------+---------------+-------------------+------------------+----------------+")
     45            self.logger.info("|Type| batches per hour | last 24 hours | per day this week | total detections | last published |")
     46            self.logger.info("+----+------------------+---------------+-------------------+------------------+----------------+")
     47            rateP2 = self.printStats("P2")
     48            rateST = self.printStats("ST")
     49            self.logger.info("+----+------------------+---------------+-------------------+------------------+----------------+")
     50       
     51            self.logger.info("")
     52       
     53            stages = self.ippToPspsDb.getStages()
     54            colCount = len(stages)
     55            self.writeTableSeparator(colCount)
     56            sys.stdout.write("|Type|  DVO  ")
     57       
     58            for stage in stages: sys.stdout.write("|%19s" % stage)
     59            sys.stdout.write("|\n")
     60            sys.stdout.write("|    |       ")
     61            for stage in stages: sys.stdout.write("| Pend  Succ  Fail  ")
     62            sys.stdout.write("|\n")
     63            self.writeTableSeparator(colCount)
     64            pendP2Processed = self.printTableRow(stages, "P2")
     65            pendSTProcessed = self.printTableRow(stages, "ST")
     66            self.writeTableSeparator(colCount)
    4167
    42     f.flush()
     68            self.logger.info("")
     69            try: self.logger.infoPair("Estimated time for P2s", "%.1f hours" % (pendP2Processed / rateP2))
     70            except: pass
     71            try: self.logger.infoPair("Estimated time for STs", "%.1f hours" % (pendSTProcessed / rateST))
     72            except: pass
     73       
     74            self.checkClientStatus()
     75            if not self.waitForPollTime(): break
    4376
    44 '''
    45 Formats an int to a string
    46 '''
    47 def getIntAsString(value):
    48     if value == 0: return ""
    49     return str(value)
    5077
    51 '''
    52 Prints (to the log) one row of data (one row being one batch-type)
    53 '''
    54 def printTableRow(stages, batchType):
     78    '''
     79    Formats an int to a string
     80    '''
     81    def getIntAsString(self, value):
     82        if value == 0: return ""
     83        return str(value)
     84   
     85    '''
     86    Prints (to the log) one row of data (one row being one batch-type)
     87    '''
     88    def printTableRow(self, stages, batchType):
     89   
     90        # get a master list of IDs in DVO for this batch type
     91        #allIDs = self.gpc1Db.getIDsInThisDVODbForThisStage(self.config.dvoLabel, batchType, self.config.minRa, self.config.maxRa, self.config.minDec, self.config.maxDec)
     92        allIDs = self.gpc1Db.getIDsInThisDVODbForThisStage(self.config.dvoLabel, batchType)
     93        prevList = allIDs
     94        numEverything = len(allIDs)
     95   
     96        sys.stdout.write("| %2s |%6d " % (batchType, numEverything))
     97   
     98        tempFilename = batchType + '.dat'
     99        DATFILE = open(tempFilename,'w')
     100        print >> DATFILE, "DVO", numEverything, 0, 0
     101        for stage in stages:
     102   
     103           # get lists. Use unions with prev list to make sure the right stuff is included
     104           success = list(set(self.ippToPspsDb.getStageIDs(batchType, self.config.epoch, self.config.dvoLabel, stage, 1)) & set(prevList))
     105           fail = list(set(self.ippToPspsDb.getStageIDs(batchType, self.config.epoch, self.config.dvoLabel, stage, -1)) & set(prevList) - set(success))
     106           pending = list(set(prevList) - set(success) - set(fail))
     107   
     108           # count stuff
     109           numSuccess = len(success)
     110           numFail = len(fail)
     111           numPending = len(pending)
     112   
     113           #if stage == 'loaded_to_datastore' and batchType == "P2":
     114           #    for n in fail: print n
     115   
     116           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))
     117           sys.stdout.write("|%-19s" % str)
     118   
     119           prevList = success
     120           print >> DATFILE, stage, numSuccess, numFail, numPending
     121   
     122           self.czarDb.insertStats(stage, self.config.dvoLabel, batchType, numPending, numSuccess, numFail)
     123   
     124           if stage == 'processed': numPendingProcessed = numPending
     125   
     126        sys.stdout.write("|\n")
     127        DATFILE.close()
     128   
     129        return numPendingProcessed
     130   
     131    '''
     132    Writes table separator
     133    '''
     134    def writeTableSeparator(self, colCount):
     135   
     136        sys.stdout.write("+----+-------")
     137        for i in range(0, colCount): sys.stdout.write("+-------------------")
     138        sys.stdout.write("+\n")
     139   
     140    '''
     141    Prints various timing stats for this batch type
     142    '''
     143    def printStats(self, batchType):
     144   
     145        rate = self.ippToPspsDb.countBatchesInLastPeriod(batchType, self.config.epoch, self.config.dvoLabel, "1 HOUR") / 1.0
     146   
     147        self.logger.info("| %2s | %16.1f | %13d | %17.1f | %16d | %14s |" %
     148                (batchType,
     149                 rate,
     150                 self.ippToPspsDb.countBatchesInLastPeriod(batchType, self.config.epoch, self.config.dvoLabel, "24 HOUR"),
     151                 self.ippToPspsDb.countBatchesInLastPeriod(batchType, self.config.epoch, self.config.dvoLabel, "1 WEEK") / 7.0,
     152                 self.ippToPspsDb.getTotalDetectionsPublished(batchType, self.config.epoch, self.config.dvoLabel),
     153                 self.ippToPspsDb.getTimeOfLastBatchPublished(batchType, self.config.epoch, self.config.dvoLabel)))
     154   
     155        return rate
     156   
    55157
    56     # get a master list of IDs in DVO for this batch type
    57     #allIDs = gpc1Db.getIDsInThisDVODbForThisStage(config.dvoLabel, batchType, config.minRa, config.maxRa, config.minDec, config.maxDec)
    58     allIDs = gpc1Db.getIDsInThisDVODbForThisStage(config.dvoLabel, batchType)
    59     prevList = allIDs
    60     numEverything = len(allIDs)
     158    '''
     159    Overrides base-class version
     160    '''
     161    def printUsage(self):
     162        super(Metrics, self).printUsage()
     163        print " [hours]"
    61164
    62     sys.stdout.write("| %2s |%6d " % (batchType, numEverything))
    63165
    64     tempFilename = batchType + '.dat'
    65     DATFILE = open(tempFilename,'w')
    66     print >> DATFILE, "DVO", numEverything, 0, 0
    67     for stage in stages:
    68 
    69        # get lists. Use unions with prev list to make sure the right stuff is included
    70        success = list(set(ippToPspsDb.getStageIDs(batchType, config.epoch, config.dvoLabel, stage, 1)) & set(prevList))
    71        fail = list(set(ippToPspsDb.getStageIDs(batchType, config.epoch, config.dvoLabel, stage, -1)) & set(prevList) - set(success))
    72        pending = list(set(prevList) - set(success) - set(fail))
    73 
    74        # count stuff
    75        numSuccess = len(success)
    76        numFail = len(fail)
    77        numPending = len(pending)
    78 
    79        #if stage == 'loaded_to_datastore' and batchType == "P2":
    80        #    for n in fail: print n
    81 
    82        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))
    83        sys.stdout.write("|%-19s" % str)
    84 
    85        prevList = success
    86        print >> DATFILE, stage, numSuccess, numFail, numPending
    87 
    88        czarDb.insertStats(stage, config.dvoLabel, batchType, numPending, numSuccess, numFail)
    89 
    90        if stage == 'processed': numPendingProcessed = numPending
    91 
    92     sys.stdout.write("|\n")
    93     #plotMe(batchType, tempFilename)
    94     DATFILE.close()
    95 
    96     return numPendingProcessed
    97 
    98 '''
    99 Writes table separator
    100 '''
    101 def writetableSeparator(colCount):
    102 
    103     sys.stdout.write("+----+-------")
    104     for i in range(0, colCount): sys.stdout.write("+-------------------")
    105     sys.stdout.write("+\n")
    106 
    107 '''
    108 Prints various timing stats for this batch type
    109 '''
    110 def printStats(batchType):
    111 
    112     rate = ippToPspsDb.countBatchesInLastPeriod(batchType, config.epoch, config.dvoLabel, "1 HOUR") / 1.0
    113 
    114     logger.info("| %2s | %16.1f | %13d | %17.1f | %16d | %14s |" %
    115             (batchType,
    116              rate,
    117              ippToPspsDb.countBatchesInLastPeriod(batchType, config.epoch, config.dvoLabel, "24 HOUR"),
    118              ippToPspsDb.countBatchesInLastPeriod(batchType, config.epoch, config.dvoLabel, "1 WEEK") / 7.0,
    119              ippToPspsDb.getTotalDetectionsPublished(batchType, config.epoch, config.dvoLabel),
    120              ippToPspsDb.getTimeOfLastBatchPublished(batchType, config.epoch, config.dvoLabel)))
    121 
    122     return rate
    123166
    124167'''
    125168Start of program.
    126169'''
    127 if len(sys.argv) > 1: CONFIGPATH  = sys.argv[1]
    128 else:
    129     print "** Usage: " + sys.argv[0] + " <configPath> [hours]"
    130     sys.exit(1)
     170metrics = Metrics(sys.argv)
     171metrics.run()
     172metrics.exitProgram("finished")
    131173
    132 if len(sys.argv) > 2:
    133     HOURS = float(sys.argv[2])
    134     SECONDS = HOURS * 60.0 * 60.0
    135 else:
    136     SECONDS = None
    137174
    138 # create objects
    139 config = Config(CONFIGPATH)
    140 logger = config.getLogger(sys.argv[0])
    141 ippToPspsDb = IppToPspsDb(logger, config)
    142 czarDb = CzarDb(logger, config)
    143 gpc1Db = Gpc1Db(logger, config)
    144175
    145 while True:
    146 
    147     logger.infoTitle("ippToPsps loading summary")
    148     now = datetime.datetime.now()
    149     logger.infoPair("Time now", now.strftime("%Y-%m-%d %H:%M:%S"))
    150     logger.infoPair("Loading epoch", config.epoch)
    151     logger.infoPair("DVO label", config.dvoLabel)
    152     config.printBoxCoords()
    153 
    154     logger.info("+----+------------------+---------------+-------------------+------------------+----------------+")
    155     logger.info("|Type| batches per hour | last 24 hours | per day this week | total detections | last published |")
    156     logger.info("+----+------------------+---------------+-------------------+------------------+----------------+")
    157     rateP2 = printStats("P2")
    158     rateST = printStats("ST")
    159     logger.info("+----+------------------+---------------+-------------------+------------------+----------------+")
    160 
    161     logger.info("")
    162 
    163     stages = ippToPspsDb.getStages()
    164     colCount = len(stages)
    165     writetableSeparator(colCount)
    166     sys.stdout.write("|Type|  DVO  ")
    167 
    168     for stage in stages: sys.stdout.write("|%19s" % stage)
    169     sys.stdout.write("|\n")
    170     sys.stdout.write("|    |       ")
    171     for stage in stages: sys.stdout.write("| Pend  Succ  Fail  ")
    172     sys.stdout.write("|\n")
    173     writetableSeparator(colCount)
    174     pendP2Processed = printTableRow(stages, "P2")
    175     pendSTProcessed = printTableRow(stages, "ST")
    176     writetableSeparator(colCount)
    177 
    178     logger.info("")
    179     try: logger.infoPair("Estimated time for P2s", "%.1f hours" % (pendP2Processed / rateP2))
    180     except: pass
    181     try: logger.infoPair("Estimated time for STs", "%.1f hours" % (pendSTProcessed / rateST))
    182     except: pass
    183 
    184     if SECONDS:
    185         logger.infoPair("Waiting for", "%f hours" % HOURS)
    186         time.sleep(SECONDS)
    187     else: break
    188     config.refresh()
    189 

  • trunk/ippToPsps/jython/pollOdm.py

    r33267 r33351  
    88import time
    99
    10 import logging.config
    11 
     10from ipptopsps import IppToPsps
    1211from config import Config
    13 from pslogger import PSLogger
    14 from ipptopspsdb import IppToPspsDb
    1512from odm import Odm
    1613from batch import Batch
    1714
     15
    1816'''
    19 Method to actually poll ODM
     17PollOdm class
    2018'''
    21 def poll(batchType, stage):
    22  
    23     if stage == "unloaded":
    24         ids = ippToPspsDb.getUnloadedBatchIDs(config.epoch, config.dvoLabel, batchType)
    25     elif stage == "unmergeworthy":
    26         ids = ippToPspsDb.getUnmergeWorthyBatchIDs(config.epoch, config.dvoLabel, batchType)
    27     elif stage == "unmerged":
    28         ids = ippToPspsDb.getUnmergedBatchIDs(config.epoch, config.dvoLabel, batchType)
     19class PollOdm(IppToPsps):
    2920
    30     logger.infoTitle("Checking %d %s %s batches" % (len(ids), stage, batchType))
     21    '''
     22    Constructor
     23    '''
     24    def __init__(self, argv):
     25        super(PollOdm, self).__init__(argv)
    3126
    32     results = {}
    33     newLoaded = 0
    34     newMergeWorthy = 0
    35     newMerged = 0
     27        # get batch and stage args
     28        if len(sys.argv) > 3:
     29            self.BATCHTYPE = sys.argv[2]
     30            self.STAGE = sys.argv[3]
     31        else:
     32            self.printUsage()
     33            self.exitProgram("incorrect args")
    3634
    37     logger.info("+-----------+----------------+----------------+----------------+----------------+----------------+----------------+")
    38     logger.info("|   Batch   | Loaded to ODM  |  Merge worthy  |     Merged     |   New loaded   |New merge worthy|   New merged   |")
    39     logger.info("+-----------+----------------+----------------+----------------+----------------+----------------+----------------+")
    40     for id in ids:
     35        # get poll time arg
     36        if len(sys.argv) > 4: self.parsePollTimeArg(sys.argv[4])
    4137
    42         if odm.checkBatch(id, results):
    43            ippToPspsDb.updateOdmStatus(id, results)
     38        if self.STAGE == "all": self.stages = ['unloaded', 'unmergeworthy', 'unmerged']
     39        else: self.stages = [self.STAGE]
    4440
    45            if stage == "unloaded":
    46                if results['LOADEDTOODM'] == 1: newLoaded = newLoaded + 1
    47                if results['MERGEWORTHY'] == 1: newMergeWorthy = newMergeWorthy + 1
    48                if results['MERGED'] == 1: newMerged = newMerged + 1
     41        if self.BATCHTYPE == "all": self.batchTypes = ['P2', 'ST']
     42        else: self.batchTypes = [self.BATCHTYPE]
    4943
    50            elif stage == "unmergeworthy":
    51                if results['MERGEWORTHY'] == 1: newMergeWorthy = newMergeWorthy + 1
    52                if results['MERGED'] == 1: newMerged = newMerged + 1
     44        for stage in self.stages: self.logger.infoPair("Checking stage", stage)
     45        for batchType in self.batchTypes: self.logger.infoPair("Checking batch type", batchType)
    5346
    54            elif stage == "unmerged":
    55                if results['MERGED'] == 1: newMerged = newMerged + 1
     47    '''
     48    Overrides base-class version
     49    '''
     50    def refreshConfig(self):
     51        ret = super(PollOdm, self).refreshConfig()
     52        self.odm = Odm(self.logger, self.config)
     53        return ret
    5654
    57            logger.info("| %9s | %14d | %14d | %14d | %14d | %14d | %14d |" %
    58                    (Batch.getNameFromID(id),
    59                     results['LOADEDTOODM'],
     55    '''
     56    Run...
     57    '''
     58    def run(self):
     59
     60        while True:
     61
     62            for stage in self.stages:
     63                for batchType in self.batchTypes:
     64                    self.poll(batchType, stage)
     65
     66            self.checkClientStatus()
     67            if not self.waitForPollTime(): break
     68
     69    '''
     70    Method to actually poll ODM
     71    '''
     72    def poll(self, batchType, stage):
     73     
     74        if stage == "unloaded":
     75            ids = self.ippToPspsDb.getUnloadedBatchIDs(self.config.epoch, self.config.dvoLabel, batchType)
     76        elif stage == "unmergeworthy":
     77            ids = self.ippToPspsDb.getUnmergeWorthyBatchIDs(self.config.epoch, self.config.dvoLabel, batchType)
     78        elif stage == "unmerged":
     79            ids = self.ippToPspsDb.getUnmergedBatchIDs(self.config.epoch, self.config.dvoLabel, batchType)
     80   
     81        self.logger.infoTitle("Checking %d %s %s batches" % (len(ids), stage, batchType))
     82   
     83        results = {}
     84        newLoaded = 0
     85        newMergeWorthy = 0
     86        newMerged = 0
     87   
     88        self.logger.info("+-----------+----------------+----------------+----------------+----------------+----------------+----------------+")
     89        self.logger.info("|   Batch   | Loaded to ODM  |  Merge worthy  |     Merged     |   New loaded   |New merge worthy|   New merged   |")
     90        self.logger.info("+-----------+----------------+----------------+----------------+----------------+----------------+----------------+")
     91        for id in ids:
     92   
     93            if self.odm.checkBatch(id, results):
     94               self.ippToPspsDb.updateOdmStatus(id, results)
     95   
     96               if stage == "unloaded":
     97                   if results['LOADEDTOODM'] == 1: newLoaded = newLoaded + 1
     98                   if results['MERGEWORTHY'] == 1: newMergeWorthy = newMergeWorthy + 1
     99                   if results['MERGED'] == 1: newMerged = newMerged + 1
     100   
     101               elif stage == "unmergeworthy":
     102                   if results['MERGEWORTHY'] == 1: newMergeWorthy = newMergeWorthy + 1
     103                   if results['MERGED'] == 1: newMerged = newMerged + 1
     104
     105               elif stage == "unmerged":
     106                   if results['MERGED'] == 1: newMerged = newMerged + 1
     107   
     108               self.logger.info("| %9s | %14d | %14d | %14d | %14d | %14d | %14d |" %
     109                       (Batch.getNameFromID(id),
     110                            results['LOADEDTOODM'],
    60111                    results['MERGEWORTHY'],
    61                     results['MERGED'],
    62                     newLoaded,
    63                     newMergeWorthy,
    64                     newMerged))
     112                        results['MERGED'],
     113                        newLoaded,
     114                        newMergeWorthy,
     115                        newMerged))
     116   
     117   
     118            else:
     119                    self.logger.info("| %9s | failed to poll ODM                                                                                                   |" % Batch.getNameFromID(id))
     120   
     121        self.logger.info("+-----------+----------------+----------------+----------------+----------------+----------------+----------------+")
     122   
    65123
    66 
    67         else:
    68             logger.info("| %9s | failed to poll ODM                                                                                                   |" % Batch.getNameFromID(id))
    69 
    70     logger.info("+-----------+----------------+----------------+----------------+----------------+----------------+----------------+")
    71 
    72 
     124    '''
     125    Overrides base-class version
     126    '''
     127    def printUsage(self):
     128        super(PollOdm, self).printUsage()
     129        print " <P2|ST|all> <unloaded|unmergeworthy|unmerged|all> [hours]"
     130   
     131   
     132   
    73133'''
    74134Program starts here
    75135'''
    76 if len(sys.argv) > 3:
    77     CONFIGPATH  = sys.argv[1]
    78     BATCHTYPE = sys.argv[2]
    79     STAGE = sys.argv[3]
    80 else:
    81     print "\n** Usage: " + sys.argv[0] + " <configPath> <P2|ST|all> <unloaded|unmergeworthy|unmerged|all> [hours]\n"
    82     sys.exit(1)
    83 
    84 if len(sys.argv) > 4: 
    85     HOURS = float(sys.argv[4])
    86     SECONDS = HOURS * 60.0 * 60.0
    87 else:
    88     SECONDS = None
    89 
    90 # create objects
    91 config = Config(CONFIGPATH)
    92 logger = config.getLogger(sys.argv[0])
    93 ippToPspsDb = IppToPspsDb(logger, config)
    94 odm = Odm(logger, config)
    95 
    96 if STAGE == "all": stages = ['unloaded', 'unmergeworthy', 'unmerged']
    97 else: stages = [STAGE]
    98 
    99 if BATCHTYPE == "all": batchTypes = ['P2', 'ST']
    100 else: batchTypes = [BATCHTYPE]
    101 
    102 for stage in stages: logger.infoPair("Checking stage", stage)
    103 for batchType in batchTypes: logger.infoPair("Checking batch type", batchType)
    104 
    105 while True:
    106 
    107     for stage in stages:
    108         for batchType in batchTypes:
    109             poll(batchType, stage)
    110 
    111     if SECONDS:
    112         logger.infoPair("Waiting for", "%f hours" % HOURS)
    113         time.sleep(SECONDS)
    114     else: break
    115 
    116 
     136pollodm = PollOdm(sys.argv)
     137pollodm.run()
     138pollodm.exitProgram("finished")
Note: See TracChangeset for help on using the changeset viewer.