CvtDbPopulator.py

Go to the documentation of this file.

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

__facility__ = "Online"
__abstract__ = "Current Value Table Database Populator"
__author__  = "Jim Panetta <panetta@slac.stanford.edu> SLAC - GLAST LAT I&T/Online"
__date__     = ("$Date: 2006/04/26 01:40:08 $").split(' ')[1]
__version__ = "$Revision: 1.9 $"
__release__  = "$Name: HEAD $"
__credits__ = "SLAC"

import sys, asyncore, time
import threading
import psycopg
from ConfigParser import *

from LICOS.lib.currValTable.CurrValClient import CurrValClient
from LICOS.lib.LATconstants               import LAT_EPOCH
from LICOS.tools.proxy.VscProxyPorts import VscProxyPorts

import LICOS.util.gOptions             as gOptions
from ISOC.TlmUtils.TlmRdbInterface     import TlmRdbDb


killEvent = threading.Event()
killEvent.clear()

multiplyDefinedMnems = ['LHKSPARE10', 'LHKSPARE08', 'LHKSPARE13', 'LHKSPARE16']

TIMER_GRANULARITY = 1  # second

def asyncore_loop(timeout=1.0, use_poll=False, map=None):
  """replace the asyncore central loop and run it its own thread.
     Valid for Python 2.3.x  
     Python 2.4 has an extra agument for loop, count.
  """
  if map is None:
      map = asyncore.socket_map

  if use_poll:
    if hasattr(select, 'poll'):
      poll_fun = asyncore.poll3
    else:
      poll_fun = asyncore.poll2
  else:
    poll_fun = asyncore.poll

  while map and not killEvent.isSet():
    poll_fun(timeout, map)

  print map, killEvent.isSet()

class CvtDbPopulator(CurrValClient):
  def __init__(self, host, port, db, source, build):
    dbc = db.cursor()
    tlmdb = TlmRdbDb(dbc)

    print "populating...", time.asctime()
    tlmdb.populate(source=source, build=build)
    print "populated", time.asctime()

    mnemList = []
    for item in tlmdb.getAllMnems():
      # *don't* add the PBC telemetry...
      if item.name()[0:4] == "LPBC":
        continue
      if item in multiplyDefinedMnems:
        continue
      mnemList.append(item.name())
    
    CurrValClient.__init__(self, host, port, mnemList=mnemList)

    self.__tlmDb = tlmdb
    self.__source = source
    self.__dbc   = dbc

    self.__cvListLock = threading.Lock()
    self.__updateLock = threading.Lock()
    self.__cvList = []
    self.__timer  = threading.Timer(TIMER_GRANULARITY, self.multiUpdate)
    self.__timer.start()
    self.__nUpdates = 0
    pass
    
  def update(self, cv):
    #print "in update", cv.name
    tlm = self.__tlmDb.getMnemList([cv.name])[0]
    if cv.name in multiplyDefinedMnems:
      return

    dispatch = False
    self.__cvListLock.acquire()
    self.__cvList.append(cv)
    self.__cvListLock.release()


  def multiUpdate(self):
    self.__updateLock.acquire()
    self.__cvListLock.acquire()
    
    nCv = len(self.__cvList)
    dispatchList = self.__cvList[0:nCv]
    self.__cvList = self.__cvList[nCv:]

    self.__cvListLock.release()
    self.__updateLock.release()

    sendList = []
    for item in dispatchList:
      sendList.append( (item.name, self.__source, item.age() + LAT_EPOCH, item.rawValue()) )

    while sendList != []:
      end = min(100, len(sendList))
      try:
        #print "inserting %d items into the database" % end, time.time()
        self.__tlmDb.insertTelemetryCollection(sendList[:end])
        sendList = sendList[end:]
      except Exception, e:
        msg = "Error in inserting telemetry.  Data follows:"
        msg += "mnem, source, age, value: \n" + str(sendList[:end])
        print e, msg

    del(sendList)
    # reanalyze the db as per LCS-96
    if not self.__nUpdates % 5000:   # do the first time and every ~5000 seconds hence
      query = "ANALYZE v3hkvalfields;"
      print time.time(), query
      self.__dbc.execute(query)
    self.__nUpdates += 1
    
    sys.stdout.flush()
    self.__timer  = threading.Timer(TIMER_GRANULARITY, self.multiUpdate)
    self.__timer.start()
    
    
   
def usage():
  print """
where:
  cvtHost       is the hostname of the CVT server
  config        is the catchall configuration file
  """
  
def main():
  options = gOptions.Options(['cvtHost', 'config'])
  try:
    options.parse()
  except Exception, msg:
    options.usage(str(msg))
    usage()
    sys.exit()
    
  cvtHost    = options.cvtHost
  configFile = options.config
  config = ConfigParser()
  config.read(configFile)

  # DSN = "user=panetta dbname=panetta host=lat-hobbit4 password=panetta"
  db = psycopg.connect( config.get('postgres','dsn').strip() )
  
  ports = VscProxyPorts(config.getint('vsc', 'proxyPortBase'))
  CvtDbPopulator( cvtHost, ports.cvtOut(), db, source=config.getint('tlmdb','source'),build=config.get('tlmdb', 'build'))

  # start the asyncore loop.  Termination is set by killEvent
  asyncore_loop()
  
  killEvent.set()


if __name__ == '__main__':
    main()

---