Collector.py

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

__facility__ = "Online"
__abstract__ = "GLAST LAT Housekeeping server infrastructure"
__author__   = "Jim Panetta <panetta@slac.stanford.edu> SLAC - GLAST I&T"
__date__     = "07/28/04"
__version__  = "$Revision: 1.6 $"
__release__  = "$Name: R04-12-00 $"
__credits__  = "SLAC"

import LATTE.copyright_SLAC

import logging
import threading
import time

class Collector(object):
  def __init__(self, *args):
    self.__loopStop   = 0
    self.__monDelay   = 0
    self.__monThread  = None
    self.__retrievers = {}
    self.__recorders  = []
    self.__delayUnits = 0.01  # seconds
    self.__log        = logging.getLogger('collecting.Collector')
    self.setDelay(10)
  
  def addRetriever(self,name,retriever):
    self.__retrievers[name]=retriever
    retriever.startup()
  
  def delRetriever(self,name):
    if self.__retrievers.has_key(name):
      del self.__retrievers[name]
  
  def addRecorder(self,recorder):
    self.__recorders.append(recorder)
  
  def setDelay(self,delay):
    """Set the delay in units of seconds  If the delay < 0, delay set to 0. 
    """
    if delay < 0:
      delay = 0
    acq = self.isAcquiring()
    self.setAcquiring(0)
    self.__monDelay = int(delay / self.__delayUnits)
    self.setAcquiring(acq)
  
  def getDelay(self):
    return self.__monDelay * self.__delayUnits
  
  def isAcquiring(self):
    if self.__monThread:
      return 1
    return 0
  
  def setAcquiring(self,bool):
    if bool:
      self.__startLoop()
    else:
      self.__stopLoop()
  
  def collectData(self):
    for r in self.__retrievers.values():
      data = r.retrievePacket()
      # Process the data
      for l in self.__recorders:
        l.updateTrends(data)
  
  def shutdown(self):
    for r in self.__retrievers.values() + self.__recorders:
      r.shutdown()
    self.__stopLoop()

  def __startLoop(self):
    if self.__monThread:
      self.__log.warning("There is already a Collector Loop")
    else:
      self.__monThread = threading.Thread(None,
                                          self.__monLoop,
                                          'Collector',
                                          (self.__monDelay,))
      self.__monThread.start()

  def __stopLoop(self):
    if self.__monThread:
      self.__log.info("Shutting down Collector Loop")
      self.__loopStop  = 1
      self.__monThread.join()
      self.__monThread = None
      self.__loopStop  = 0

  def __monLoop(self,delay):
    self.__log.info("Starting Collector Loop")
    while (not self.__loopStop):
      self.collectData()
      for i in xrange(delay):
        if (self.__loopStop):
          break;
        time.sleep(self.__delayUnits)
    self.__log.info("Collector Loop Stopped")

def testCollector():
  logging.basicConfig()
  logging.root.setLevel(logging.DEBUG)

  udpServerName = 'testTrendingServer'
  udpServerAddr = '54320@239.255.1.14'
  from Retriever import DummyRetriever
  from Recorder  import UDPRecorder, DummyRecorder

  c   = Collector()
  rec = UDPRecorder()
  rec.startUDPServer(udpServerName,udpServerAddr)
  c.addRetriever('dummy',DummyRetriever())
  c.addRecorder(rec)
  c.addRecorder(DummyRecorder())
  c.setAcquiring(True)
  
  try:
    while 1:
      time.sleep(.01)
  except Exception, e:
    c.shutdown()
  
if __name__ == '__main__':
  testCollector()

---