CurrValServer.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 Server"
__author__  = "Jim Panetta <panetta@slac.stanford.edu> SLAC - GLAST LAT I&T/Online"
__date__     = ("$Date: 2006/01/28 05:35:26 $").split(' ')[1]
__version__ = "$Revision: 1.13 $"
__release__  = "$Name: HEAD $"
__credits__ = "SLAC"

import LICOS.copyright_SLAC

import struct, time, threading, copy
import cPickle as pickle

import asyncore, socket

from CurrValTable import *
from LICOS.lib.currValTable.StringTcpProtocol import Int16StringProtocol
import LICOS.util.gOptions as gOptions

from LICOS.lib.cmdTlmDb.LCATtlmDb  import LICOS_TlmPacketFactory

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

class CurrentValueProtocol(Int16StringProtocol):
  TIMEOUT   = 10    # seconds
  def __init__(self, port, cvt):
    super(CurrentValueProtocol, self).__init__(port)
    self.__cvt  = cvt
    self.__mnemFlags    = {}
    self.sendString("ok ")

  def handle_close(self):
    # print "CVP: handle_close"
    self.__cvt.removeObserver(self)
    self.close()


  def stringReceived(self, line):
    """ here's where the full connection protocol between the clients and
        server is defined.

        observer               # client wishes to be an observer.

        register <mnemonic> <flags>  # client wishes to be an observer of <mnemonic> with flags <flags>  (space separated set of flags)
        unregister <mnemonic>  # client wishes *not* to be an observer of <mnemonic>

        drop                   # client wishes to be no longer an observer
                               #   *REVISIT* -- how to disconnect from mnemonics?
        request <mnem> <flags> # Client wants to request an observable ASAP.
    """
    fields = line.split()
    query = fields[0]

    cvt = self.__cvt
    egu = False
    #print fields

    if   query == 'observer':  # register an observer
      resp = 'ob '
      self.sendString(resp)
      pass
    elif query == 'drop':
      pass
    elif query == 'register':
      mnem = fields[1]
      flags = fields[2:]
      self.register(mnem, flags)
    elif query == 'unregister':
      mnem = fields[1]
      self.unregister(mnem)
    elif query == 'request':
      mnem = fields[1]
      flags = fields[2:]
      self.request(mnem, flags)
    else:
      print "bad query:", line

    # print line

  def register(self, mnem, flags):
    cvt = self.__cvt
    if mnem == 'everythingunderthesun':
      self.__registerAll(flags)
    else:
      # print cvt.fullTable().keys()
      if cvt.hasValue(mnem):
        cv = cvt.getValue(mnem)
        cv.addObserver(self)
        #print "register:", mnem, " flags:", hex(cv.flags())
        if not(cv.flags() & cv.FLAG_STALE):
          self.update(cv)
      else:
        cvt.setValue(mnem, None, time.time())
        cv = cvt.getValue(mnem)
        cv.addObserver(self)
        cv.setFlags(cv.FLAG_STALE)
        # Don't call self.update()  The value is poor.

      self.__mnemFlags[mnem] = flags
    # ****revisit******    Add the update here.  (to seed the observer)

  def __registerAll(self, flags):
    for mnem, cv in self.__cvt.fullTable().items():
      cv.addObserver(self)
      self.__mnemFlags[mnem] = flags

  def unregister(self, mnem):
    cvt = self.__cvt
    if cvt.hasValue(mnem):
      cvt.getValue(mnem).delObserver(self)

  def request(self, mnem, flags):
    cvt = self.__cvt
    replyStr = "rq "
    # print "request", mnem, flags
    if cvt.hasValue(mnem):
      cv = cvt.getValue(mnem)
      replyStr += encodeCurrentValue(cv)

    else:  # mnem doesn't exist, pass None across
      replyStr += pickle.dumps(None,2)

    #print "rq:", len(replyStr)
    self.sendString(replyStr)

  def update(self, cv):
    """update method
    \param cv  A CurrentValue object
    """

    replyStr =  "up "
    replyStr += encodeCurrentValue(cv)

    # print "up:", len(replyStr)
    self.sendString(replyStr)

  def __encode(self, cv):
    # To reduce the size  and complexity of the transferred object, make
    # a copy, and perform a sticky <reinterpret_cast> back
    # to the base class.  Then delete the observers list
    # This list is only useful on the server side, and increases
    # the size of the pickled object by a large factor if full.

    
    cvCop = copy.copy(cv)
    cvCop.__class__ = CurrentValueBase
    del(cvCop.observers)

    # pickle the object into a binary blob
    return pickle.dumps(cvCop,2)

class CurrentValueFactory(asyncore.dispatcher, object):
  def __init__(self, port, cvt):
    super(CurrentValueFactory, self).__init__()
    self.__cvt = cvt

    # build the main connection socket
    self.create_socket(socket.AF_INET,socket.SOCK_STREAM)
    self.set_reuse_addr()
    self.bind(("", port))
    self.listen(100)

  def handle_accept(self):
    """This is called once per *connection*
    """
    clientPort, addr = self.accept()
    print "Adding client connection: ", addr
    return CurrentValueProtocol(clientPort, self.__cvt)

def readLoop(cvt, packetFile):

  pFh = open(packetFile, 'rb')
  count = 0
  delta = 0
  delta2 = 0
  while not killEvent.isSet():
    packet = getPacket(pFh)
    if packet is None:
      pFh.seek(0)
      continue

    apIdPkt = LICOS_TlmPacketFactory(packet)
    aTime = apIdPkt.secs + apIdPkt.usecs/1E6
    print "dispatching apid 0x%04x, time %12d" %( apIdPkt.apid, aTime )
    apIdPkt.decode_payload()
    # toss each mnemonic in the cvt and then...
    for mnem in apIdPkt.get_payload_list():
      cvt.setValue(mnem, apIdPkt.get_payload(mnem), aTime)
    # add in the packet as well
    cvt.setValue(apIdPkt.name.upper(), packet, aTime)
    count += 1
    time.sleep(0.25)
    if count%100 == 0:
      print count, time.time()

def getPacket(fileHandle):
  try:
    packet = ""
    packet += fileHandle.read(6)
    pktHdr = struct.unpack('!HHH', packet)
    apId = ( pktHdr[0] & 0x07FF )
    pLen = pktHdr[2] +1
    #print "returning apId %x, length %d" % (apId, pLen)
    packet += fileHandle.read(pLen)
    return packet

  except Exception, e:
    # print e
    return None

if __name__ == '__main__':
  options = gOptions.Options(['pktFile', 'cvtPort'])
  try:
    options.parse()
  except Exception, msg:
    options.usage(str(msg))
    exit

  packetFile = options.pktFile
  cvtPort = int(options.cvtPort)

  cvt = CurrentValueTable()
  cvs = CurrentValueFactory(cvtPort, cvt)


  readThread = threading.Thread(name="readLoop", target=readLoop, args= (cvt, packetFile))
  # readThread.setDaemon(True)
  readThread.start()

  # print "reactor.run"
  try:
    asyncore.loop(1)
  except:
    pass
  print "doneloop"


  killEvent.set()
  readThread.join()








#eof

---