consumer.py

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

__facility__ = "Online"
__abstract__ = "GLAST Online consumer"
__author__   = "Lester Miller"
__date__     = "11/20/2003"
__version__  = "$Revision: 1.5 $"
__credits__  = "SLAC"


import LATTE.copyright_SLAC
import LDF
import LATTE.consumer.standardLDF as standardLDF
import LATTE.tools.DataDistributorPoll as DDP
import LATTE.runcontrol.rcArchiver as rcArchiver
import struct
import time
import sys
import getopt
import array
import logging as myLog

class Consumer(object):
  """\brief Consumer class to consume LDF data
  This class just handles file open/close or server
  open/close
  """
  
  def __init__(self,argv):
    """consumer constructor
    """
    # subclasses should provide iterators in constructor
    self.lci       = standardLDF.LATcomponentIterator()
    self.eei       = standardLDF.EBFeventIterator(self.lci)
    self.lcti      = standardLDF.LATcontributionIterator(self.eei)
    self.ldbi      = LDF.LATdataBufferIterator(self.lcti)

    self.__ds          = None
    self.__file        = None
    self.__archiver    = None
    self.__debug       = 0
    self.eventData     = None
    self.__archiveFile = "consumer.ldf"
    self.__inFileName  = None
    self.__nEvents     = 0
    self.__server      = "1"
    self.__runId       = ''
    self.sleepTime     = 0.1
    self.startRun      = False
    self.endRun        = False
    self.parseArgs(argv)

  def parseEvent(self):
    """Analyse an event
    """
    # subclasses should provide a custom iterator to iterate in this method
    self.ldbi.iterate(self.eventData, len(self.eventData)) 
    self.checkUDF()
    
  def checkUDF(self):
    udfDict = self.lcti.getUdfDict()
    self.startRun = False
    if 'startRun' in udfDict:
      self.setRunId(udfDict['startRun'])
      self.startRun = True
    self.endRun = False
    if 'endRun' in udfDict:
      self.endRun = True
    self.lcti.clearUdfDict()
  
  def getRunId(self):
    return self.__runId
  def setRunId(self,runId):
    self.__runId = runId
  
  def getNextEvent(self):
    """get Next Event into data buffer
    """
    if self.__file is not None:
      evtPtr = self.__file.tell()
      # get datagram
      self.eventData = self.__file.read(8)
      # check for file end
      if len(self.eventData) < 8:
        myLog.info("Consumer.getNextEvent: End of File")
        return 0
      # unpack pair of big-endian 4 byte words
      # this is presumably broken for Sun hardware
      (identity, length) = struct.unpack('!LL', self.eventData)
      self.eventData        += self.__file.read(length - 8)
      if self.__debug:
        myLog.info("got data %d length from file" % (length))
      return length
    if self.__ds is not None:
      self.eventData, sender = self.__ds.receive()
      if self.eventData is None:
        #print "No data from server"
        return 1
      if self.__debug:
        myLog.info("got data %d length from file" % len(self.eventData))
      return 1
    myLog.warn("Consumer.getNextEvent: No data source")
    return 0

  def serverOpen(self, server="1"):
    """  Connect to network server
    """
    myLog.info("Consumer.serverOpen: Connecting to data server %s"%(server))
    # Close previous data sources
    if self.__file is not None:
      self.__fileClose()
    if self.__ds is not None:
      self.serverClose()
    dsName = 'LAT_EBFdata'
    self.__ds = DDP.DataDistributorPoll(dsName)
    if self.__ds is not None:
      self.__ds.connect(server)
      self.__ds.start()
    else:
      myLog.warn("Consumer.serverOpen: Unable to connect to Data Distribution server %s"%(server))
      sys.exit(2)
      
  def serverClose(self):
    """  Close network server
    """
    if self.__ds is not None:
      self.__ds.stop()
      self.__ds.disconnect()
      self.__ds = None

  def fileOpen(self,fn):
    """ Load input data from a file
    """
    # Close previous data sources
    if self.__ds is not None:
      self.serverClose()
    if self.__file is not None:
      self.__fileClose()
    try:
      self.__file = open(fn, 'rb')
    except:
      myLog.warn("Consumer.fileOpen: Fatal: failed to open file %s"%(fn))
      sys.exit(2)
    print myLog.info("Consumer.fileOpen: opening file %s"%(fn))
    
  def fileClose(self):
    """ Close input data file
    """
    if self.__file is not None:
      self.__file.close()
      self.__file = None
    return

  def archiveOpen(self,fn):
    """ Open data archive
    """
    # make a 'being written to' version of the file
    openfn = fn + '_locked'
    # Close previous archive file
    if self.__archiver is not None:
      self.archiveClose(fn)
    myLog.info("Consumer.archiveOpen: opening file %s"%(fn))
    try:
      self.__archiver = rcArchiver.rcArchiver(fileName=openfn)
    except:
      myLog.warn("Consumer.archiveOpen: failed to open archive file %s"%(fn))
      self.__archiver = None
    
  def archiveClose(self,fn):
    """ Close data archive
    """
    if self.__archiver is not None:
      self.__archiver.close()
      self.__archiver = None
      myLog.info("Consumer.archiveClose: closing archive file %s"%(fn))
    # move it to the requested file name if it is closed
    openfn = fn + '_locked'
    import shutil
    shutil.move(openfn,fn)
    return

  def dataSize(self):
    """Return current event data size
    """
    if self.eventData is not None:
      return len(self.eventData)
    else:
      return 0
      
  def usage(self):
    print "\n%s Usage:" % (sys.argv[0])
    print "\t-h or --help for Help"
    print "\t-a or --archive to specify output to archive file (default consumer.ldf)"
    print "\t-n or --nevents to specify the number of events archived (default 0)"
    print "\t-i or --infile to specify an input from file instead of server"
    print "\t-s or --server to specify server to connect to (default 1)"
    print "\t-c or --config to specify configuration file (default None, used by Plotter)"
    print "\t-r or --rate to specify a rate limit in Hz (default 10)"
    

  def parseArgs(self,argv):
    try:
      opts, args = getopt.getopt(argv, "ha:n:i:s:c:r:",
                                 ["help", "archive=", "nevents=", "infile=", "server=","config=","rate="])
    except getopt.GetoptError:
      # print help information and exit:
      self.usage()
      sys.exit(2)
    for o, a in opts:
      if o in ("-h", "--help"):
        self.usage()
        sys.exit()
      if o in ("-a", "--archive"):
        self.__archiveFile = a
      if o in ("-n", "--nevents"):
        self.__nEvents = int(a)
      if o in ("-i", "--infile"):
        self.__inFileName = a
      if o in ("-s", "--server"):
        self.__server = a
      if o in ("-c", "--config"):
        self.configFile = a
      if o in ("-r", "--rate"):
        rate = float(a)
        if rate>0:
          self.sleepTime = 1./rate
                  
  def loop(self):
    if self.__inFileName is not None:
      self.fileOpen(self.__inFileName)
    else:
      self.serverOpen(self.__server)
    eventCount=0
    archivedEventCount=0
    sayWaiting = 0
    timeLastEvent = 0
    while (1):
      self.getNextEvent()
      if self.eventData is not None and self.dataSize()>0:
        timeLastEvent = time.time()
        eventCount+=1
        self.parseEvent()
        # archive
        if self.__archiver is not None and archivedEventCount<self.__nEvents:
          # don't know why, but needs byte swapping?
          outputData = array.array('L',self.eventData)
          outputData.byteswap()
          self.__archiver.write(outputData)
          #self.__archiver.write(self.eventData)
          archivedEventCount+=1
        # open archive if new run seen and none open
        if self.__nEvents>0 and self.startRun:
          self.archiveOpen(self.__archiveFile)
        # close archive if collected enough
        if (self.endRun or archivedEventCount>=self.__nEvents) and self.__archiver is not None:
          self.archiveClose(self.__archiveFile)
          archivedEventCount=0
        if sayWaiting==1:
          sayWaiting=0
          print "resumed acquisition... (eventCount is ",eventCount,")"
        time.sleep(self.sleepTime) # conserve cpu...
      else:
        deltaTime = time.time() - timeLastEvent
        if sayWaiting==0 and deltaTime>1:
          sayWaiting=1
          print "Waiting for data..."
      time.sleep(0.0005) # conserve cpu...
    
    myLog.info("Consumer: Read in %d events" % (eventCount))
    self.serverClose()
    self.fileClose()
    sys.exit()


if __name__ == '__main__':
  consumer = Consumer(sys.argv[1:])
  consumer.loop()
  

---