CvtTesterImpl.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__ = "CVT Tester"
__author__  = "Jim Panetta <panetta@slac.stanford.edu> SLAC - GLAST LAT I&T/Online"
__date__     = ("$Date: 2006/01/25 04:27:12 $").split(' ')[1]
__version__ = "$Revision: 1.8 $"
__release__  = "$Name: HEAD $"
__credits__ = "SLAC"

import LICOS.copyright_SLAC

import sys, asyncore
import threading
from qt import *
import psycopg

from LICOS.core.thread.guiBridges         import GUIbridge
from LICOS.lib.currValTable.CvtTester     import CvtTester
from LICOS.lib.currValTable.CurrValClient import CurrValClient

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


globalApp = QApplication([])
killEvent = threading.Event()
killEvent.clear()

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)

class CvtTesterClient(CurrValClient):
  def __init__(self, host, port, mnemList, tlmdb, gui, guiBridge):
    CurrValClient.__init__(self, host, port, mnemList=mnemList)
    self.__items = {}
    self.__gui = gui
    self.__guiBridge = guiBridge
    self.__tlmdb = tlmdb
    pass

  def update(self, cv):
    # print "in update", cv.name
    self.__guiBridge.execGUImethodNR(self.__gui, self.__update, cv)

  def __update(self, cv):
    if cv.name not in self.__items:
      self.__items[cv.name] = QListViewItem(self.__gui.mnemonicTable)
      self.__items[cv.name].setText(0, cv.name)
    raw = cv.rawValue()
    self.__items[cv.name].setText( 1, str(raw) )

    egu = raw
    mObj = self.__tlmdb.mnem(cv.name)
    if mObj is not None:
      egu = mObj.conversion().egu(raw)
    self.__items[cv.name].setText( 2, str(egu) )
    self.__items[cv.name].setText( 3, str(cv.age()) )
    # self.__items[cv.name].repaint()
    self.__gui.update()


class CvtTesterImpl(CvtTester):
  def __init__(self, cvtHost, cvtPort, mnemList, tlmdb, parent = None, name = None, fl = 0):
    CvtTester.__init__(self, parent, name, fl)
    self.__guiBridge = GUIbridge(threading.currentThread())

    self.__client = CvtTesterClient(cvtHost,cvtPort,mnemList, tlmdb, self, self.__guiBridge)

  def customEvent(self, e):
    """!\brief Post an event to the GUI thread.

    This method overrides the QObject base class's.  There is generally
    no reason to call this method directly.  It is called by the Qt
    infrastructure whenever the custom event is posted, e.g. by the following
    three methods of this class: createGUI, execGUImethodNR and execGUImethod.

    \param e A QEvent object
    """
    self.__guiBridge.handleCustomEvent(e)



def usage():
  print """
where:
  cvtHost       is the hostname of the CVT server
  cvtPort       is the connection port of the CVT server (typically 39800)
  cvList        is a quoted, comma separated list of requested current values
                Ex: --cvList "LHKADAB33V,LHKADABTEMP"
  """

def main():
  options = gOptions.Options(['cvtHost', 'cvtPort','cvList'])
  try:
    options.parse()
  except Exception, msg:
    options.usage(str(msg))
    usage()
    sys.exit()

  cvtHost = options.cvtHost
  cvtPort = int(options.cvtPort)

  mnemList = []

  if options.cvList is not None:
    tmp = options.cvList
    for mnem in tmp.split(','):
      mnemList.append(mnem)

  DSN = "user=trending dbname=trending host=gitow1 password=trending"
  db = psycopg.connect( DSN )
  dbc = db.cursor()
  tlmdb = TlmRdbDb(dbc)

  tlmdb.populate(source=79, build='B0-6-0')



  win = CvtTesterImpl(cvtHost, cvtPort, mnemList,tlmdb)
  globalApp.setMainWidget(win)
  win.show()

  QObject.connect(globalApp, SIGNAL("lastWindowClosed()"), globalApp, SLOT("quit()"))

  # start the asyncore loop.  Termination is set by killEvent
  threading.Thread(target=asyncore_loop).start()
  globalApp.exec_loop()

  killEvent.set()


if __name__ == '__main__':
    main()

---