monitoringDbSyncOut.py

#!/usr/local/bin/python
#
# monitoringDbSyncOut.py
#
#
#
#
#                               Copyright 2005
#                                     by
#                        The Board of Trustees of the
#                     Leland Stanford Junior University.
#                            All rights reserved.
#

__facility__ = "Online"
__abstract__ = "Housekeeping Database Synch Out"
__author__   = "J. Panetta <panetta@SLAC.Stanford.edu> SLAC - GLAST LAT I&T/Online"
__date__     = "11/20/2004"
__version__  = "$Revision: 1.5 $"
__credits__  = "SLAC"

import LATTE.copyright_SLAC

import os, sys, struct, time, gzip
from LATTE.client.gOptions         import Options
import MySQLdb

tlmFieldsList = "MNEMID, MNEM, DESTTYPE, EUCONV, DESCRIP, OPSYMBOL, SUBS, EUSTRING, EVENTFLAG, ARYLENGTH, LIMITS, INIT, NOTASSIGNABLE"
valFieldsList = "USECS, VALUE, MNEMID, SOURCE"
srcFieldsList = "SOURCE, DESCRIP"

noAuto    = "set AUTOCOMMIT=0"
lockMsg   = "LOCK TABLES valfields WRITE;"
unlockMsg = "UNLOCK TABLES;"


valPackFmt    = "!ddLH"
chunkSize     = 1000
nullValue     = 1000000000.0000001    # Standard python does not have IEEE NaN available

def lockTransact(db, msg):
  """Locking database transactor for INNODB databases
  """
  db.execute(noAuto)
  db.execute(lockMsg)
  db.execute(msg)
  rc = db.rowcount
  db.connection.commit()
  db.execute(unlockMsg)
  return rc

def dateMnemSelect(db, start, end):
  """Returns all used mnemonics in a time period
  """
  msg = "select distinct mnemid from valfields where %f<usecs and usecs<%f" % (start, end)
  db.execute(msg)
  m = db.fetchall()
  ret = []
  for item in m:
    ret.append(item[0])
  return ret

def dumpVALfields(db, outFile, start, end):
  # Get the mnemonic space to iterate over
  mnemSet = dateMnemSelect(db, start, end)
  for mnem in mnemSet:
    # print time.ctime(), "Starting select mnemid =", mnem
    selectMsg = "select %s from valfields" % (valFieldsList)
    qualifier = " where mnemid=%d and %f<usecs and usecs<%f and VALIDITY is NULL" %\
           (mnem, start, end)
    db.execute(selectMsg + qualifier)    # doesn't need locking transaction
    m = db.fetchall()
    
    for set in m:
      if set[1] is None:  # Deal with NULL returns
        s = list(set)
        s[1] = nullValue
        set = tuple(s)
      str = struct.pack(valPackFmt, *set)
      outFile.write(str)
    
    updMsg = "update valfields set VALIDITY=1"
    limitMsg  = " LIMIT %d" % chunkSize
    block=0
    # print time.ctime(), "starting update"
    while True:
      # print time.ctime(), "update block", block
      rows = lockTransact(db, updMsg + qualifier + limitMsg)
      block += 1
      if rows < chunkSize:
        break
  
  
#   while True:
#     msg = "select %s from valfields where VALIDITY is NULL AND USECS>%f and USECS<%f limit %d" %\
#       (valFieldsList, start, end, chunkSize)
#     db.execute(msg)
#     rc = db.rowcount
#     if rc == 0: break
#     
#     m = db.fetchmany(chunkSize)
#     for set in m:
#       if set[1] is None:  # Deal with NULL returns
#         s = list(set)
#         s[1] = nullValue
#         set = tuple(s)
#       str = struct.pack(valPackFmt, *set)
#       outFile.write(str)
# 
#     updMsg = "update valfields set VALIDITY=1 where VALIDITY is NULL AND USECS>%f and USECS<%f limit %d" %\
#              ( start, end, chunkSize)
#     db.execute("LOCK TABLES valfields WRITE;")
#     rc = db.execute(updMsg)
#     db.connection.commit()
#     db.execute("UNLOCK TABLES;")
#
  
def findMinTime(db):
  msg = "select min(usecs) from valfields where validity is NULL"
  db.execute(msg)
  m = db.fetchall()   # returns tuple(tuple,)
  return m[0][0]
  
  
def dumpTLMfields(db, outFile):
  msg = "select %s from tlmfields;" % (tlmFieldsList)
  db.execute(msg)
  
  m = db.fetchall()
  
  print >> outFile, tlmFieldsList
  for set in m:
    print >> outFile, set


def usage():
  print """
monitoringDbSyncOut Usage:
$ monitoringDbSyncOut --outdir <dir> [--start <start>] [--end <end>]

Options:
  --outdir    Specify the output directory root.  A subdirectory will be created there
              of the form YYYYMMDD-HHMMSS containing the output files.  The time used is
              from the --end option, below.
  --start     Specify the start time in seconds since Jan 0 1970.
              Default is 0
  --end       Specify the end time in seconds since Jan 0 1970
              Default is the output of time.time() when the program is run.
  """

if __name__ == '__main__':
  
  # Options, first list mandatory, second list optional 
  options = Options(["outdir"],
                    ["start", "end"])
  try:
    options.parse()
  except Exception, msg:
    # options.usage(usage())
    raise Exception, msg
  
  start  = -1
  end    = int(time.time())
  if options.start:
    start = int(options.start)
  if options.end:
    end   = int(options.end)
    
  endTimeStr = time.strftime("%Y%m%d-%H%M%S", time.gmtime(end))
  outDir = os.path.join(options.outdir, endTimeStr)
  
  if not os.path.isdir(outDir):
    os.mkdir(outDir)
  
  sqlDb = MySQLdb.connect(db='itosdb', user='itosdb', host='localhost')
  cursor = sqlDb.cursor()
  
  if start == -1:
    start  = findMinTime(cursor)
  
  valOut = gzip.open(os.path.join(outDir,'val.out'),mode='wb+')
  dumpVALfields(cursor, valOut, start, end)
  valOut.close()
  
  tlmOut = open(os.path.join(outDir,'tlm.out'),mode='wt+')
  dumpTLMfields(cursor, tlmOut)
  tlmOut.close()
  
  done = open(os.path.join(outDir, 'zzz.done'), mode='w')
  done.close()
  
  

  

---