Changeset 33098 for branches/meh_branches/ppsub_test/ippToPsps/jython/metrics.py

Timestamp:

Jan 11, 2012, 11:19:21 PM (15 years ago)

Author:

mhuber

Message:

merged latest r33093 trunk changes to meh_branches/ppsub_test

Location:

branches/meh_branches/ppsub_test

Files:

• . (modified) (1 prop)
• ippToPsps/jython/metrics.py (modified) (5 diffs)

branches/meh_branches/ppsub_test/ippToPsps/jython/metrics.py

@@ -1 +1 @@
 import math
+import time
 import logging
 import smtplib
 import sys
 import os
-
-from java.lang import Math
-from org.apache.commons.math.special import Erf
+import datetime
+
 from xml.etree.ElementTree import ElementTree, Element, tostring
 
 from pslogger import PSLogger
 from ipptopspsdb import IppToPspsDb
+from czardb import CzarDb
 from gpc1db import Gpc1Db
 
@@ -41 +42 @@
 
 '''
-Prepares a string of the format:
-
-n-processed(n-pending)
-
-'''
-def prepareStr(masterList, prevList, thisList, FILE, name):
-
-    unsuccessful = list(set(prevList) - set(thisList))
-
-    countTotal = len(masterList)
-    countSuccessful = len(thisList)
-    countUnsuccessful = len(unsuccessful)
-
-    str = "%d" % countSuccessful
-    if countUnsuccessful: str = str + "(%d)" % countUnsuccessful
-
-    print >> FILE, name, countSuccessful, countUnsuccessful, (countTotal - countSuccessful - countUnsuccessful)
-    #;sys.stdout.softspace=0; 
-    return str
+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(batchType):
-
-    allIDs = gpc1Db.getIDsInThisDVODbForThisStage(DVOLABEL, batchType)
-    processedIDs = ippToPspsDb.getProcessedIDs(batchType, EPOCH, DVOLABEL)
-    datastoreIDs = ippToPspsDb.getloadedToDatastoreIDs(batchType, EPOCH, DVOLABEL)
-    loadedToODMIDs = ippToPspsDb.getLoadedToODMIDs(batchType, EPOCH, DVOLABEL)
-    mergeWorthyIDs = ippToPspsDb.getMergeWothyIDs(batchType, EPOCH, DVOLABEL)
-    mergedIDs = ippToPspsDb.getMergedIDs(batchType, EPOCH, DVOLABEL)
+def printTableRow(stages, batchType):
+
+    # get a master list of IDs in DVO for this batch type
+    if MINRA and MAXRA and MINDEC and MAXDEC:
+        allIDs = gpc1Db.getIDsInThisDVODbForThisStage(DVOLABEL, batchType, MINRA, MAXRA, MINDEC, MAXDEC)
+    else:
+        allIDs = gpc1Db.getIDsInThisDVODbForThisStage(DVOLABEL, batchType)
+    prevList = allIDs
+    numEverything = len(allIDs)
+
+    sys.stdout.write("| %2s |%6d " % (batchType, numEverything))
 
     tempFilename = batchType + '.dat'
     DATFILE = open(tempFilename,'w')
-    prepareStr(allIDs, allIDs, allIDs, DATFILE, "inDVO"),
-
-    logger.info("|  %2s  |  %-12d  |  %-12s  |  %-12s  |  %-12s  |  %-12s  |  %-12s  |  %-12s  |  %-19s |" % (
-                batchType, 
-                len(allIDs), 
-                prepareStr(allIDs, allIDs, processedIDs, DATFILE, "processed"),
-                prepareStr(allIDs, processedIDs, datastoreIDs, DATFILE, "onDatastore"),
-                prepareStr(allIDs, datastoreIDs, loadedToODMIDs, DATFILE, "loadedToODM"),
-                prepareStr(allIDs, loadedToODMIDs, mergeWorthyIDs, DATFILE, "MergeWorthy"),
-                prepareStr(allIDs, mergeWorthyIDs, mergedIDs, DATFILE, "Merged"),
-                len(list(set(allIDs) - set(mergedIDs))),
-                ippToPspsDb.getTimeOfLastBatchPublished(batchType, EPOCH, DVOLABEL)))
-
+    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, EPOCH, DVOLABEL, stage, 1)) & set(prevList)) 
+       fail = list(set(ippToPspsDb.getStageIDs(batchType, EPOCH, 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, DVOLABEL, batchType, numPending, numSuccess, numFail)
+
+       if stage == 'processed': numPendingProcessed = numPending
+
+    sys.stdout.write("|\n")
     plotMe(batchType, tempFilename)
     DATFILE.close()
-    #os.remove(tempFilename)
+
+    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, EPOCH, DVOLABEL, "1 HOUR") / 1.0 
+
+    logger.info("| %2s | %16.1f | %13d | %17.1f | %16d | %14s |" % 
+            (batchType,
+             rate,
+             ippToPspsDb.countBatchesInLastPeriod(batchType, EPOCH, DVOLABEL, "24 HOUR"),
+             ippToPspsDb.countBatchesInLastPeriod(batchType, EPOCH, DVOLABEL, "1 WEEK") / 7.0, 
+             ippToPspsDb.getTotalDetectionsPublished(batchType, EPOCH, DVOLABEL),
+             ippToPspsDb.getTimeOfLastBatchPublished(batchType, EPOCH, DVOLABEL)))
+
+    return rate
 
 '''
@@ -97 +128 @@
 if len(sys.argv) > 1: CONFIG  = sys.argv[1]
 else:
-    print "** Usage: " + sys.argv[0] + " <configPath>"
+    print "** Usage: " + sys.argv[0] + " <configPath> [hours]"
     sys.exit(1)
+
+if len(sys.argv) > 2:
+    HOURS = float(sys.argv[2])
+    SECONDS = HOURS * 60.0 * 60.0
+else:
+    SECONDS = None
+
+
 
 # open config file
@@ -107 +146 @@
 logger.setup(configDoc, "metrics")
 
+# create database objects
 ippToPspsDb = IppToPspsDb(logger, configDoc)
+czarDb = CzarDb(logger, configDoc)
 gpc1Db = Gpc1Db(logger, configDoc)
 
@@ -113 +154 @@
 EPOCH = configDoc.find("options/epoch").text
 
-logger.infoTitle("ippToPsps loading summary")
-logger.infoPair("Loading epoch", EPOCH)
-logger.infoPair("DVO label", DVOLABEL)
-logger.info("")
-logger.info("+------+----------------+----------------+----------------+----------------+----------------+----------------+----------------+----------------------+")
-logger.info("| Type |  Total in DVO  |   Processed    |  On datastore  | Loaded to ODM  |  Merge-worthy  |     Merged     |  Total missing |    Last published    |")
-logger.info("+------+----------------+----------------+----------------+----------------+----------------+----------------+----------------+----------------------+")
-printTableRow("P2")
-printTableRow("ST")
-logger.info("+------+----------------+----------------+----------------+----------------+----------------+----------------+----------------+----------------------+")
-
-
-
+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", EPOCH)
+    logger.infoPair("DVO label", DVOLABEL)
+
+    # get equatorial coord constraints, if any
+    MINRA = None
+    MAXRA = None
+    MINDEC = None
+    MAXDEC = None
+
+    try:
+        MINRA = float(configDoc.find("dvo/minRA").text)
+        MAXRA = float(configDoc.find("dvo/maxRA").text)
+        MINDEC = float(configDoc.find("dvo/minDec").text)
+        MAXDEC = float(configDoc.find("dvo/maxDec").text)
+        logger.infoPair("Coordinates", "%.1f to %.1f RA, %.1f to %.1f Dec" % (MINRA, MAXRA, MINDEC, MAXDEC))
+    except:
+        pass
+   
+
+    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
+
+