test_evt_exttrg.py

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

"""Test script for external trigger functionality.
"""
__facility__ = "Online"
__abstract__ = "Test script for external trigger"
__author__   = "Jim Panetta <panetta@slac.stanford.edu> SLAC - GLAST I&T/Online"
__date__     = ("$Date: 2004/11/18 18:31:39 $").split(' ')[1]
__version__  = "$Revision: 1.2 $"
__release__  = "$Name: R04-12-00 $"
__credits__  = "SLAC"

import LATTE.copyright_SLAC

from LATTE.client.gOptions import Options
from LATTE.client.gCmdCli  import CmdCli
from LATTE.client.gEvtCli  import EvtCli
from LATTE.database.gLAT   import GLAT
from LATTE.database.gGLT   import GGLT
from LATTE.database.gOCS   import GOCS

from time import sleep
import gc
import os
import time
import logging as log


from LATTE.tests.apps.support.SimpleGasuExample import *
from LATTE.tests.apps.support.MiniGLTExample    import *

tem = None

# Define a GASU impl of an external trigger engine
class GasuExtTrg(SimpleGasuExample):
  def __init__(self):
    SimpleGasuExample.__init__(self)
    self.__cond = GasuExtTrgCond()
    self.engines().replaceEngine(0, GasuExtTrgEngine())
  def conditions(self): return self.__cond

class GasuExtTrgCond(TrgConditionsValue):
  def __init__(self):
    TrgConditionsValue.__init__(self)
  def external(self): return True

class GasuExtTrgEngine(TrgEngine):
  def __init__(self):
    TrgEngine.__init__(self)
    self.__request = TrgReadout()
    cond = GasuExtTrgCond()
    self.__cv = cond.value()
  def participate(self,conditionsValue):
    if conditionsValue & self.__cv:
      return True
    return False
  def prescale(self):   return 0
  def request(self):    return self.__request

# define a MiniGLT impl of an external trigger
class MiniGltExtTrg(MiniGLTExample):
  def __init__(self):
    MiniGLTExample.__init__(self)
    self.__cond = MiniGltExtTrgCond()
    self.engines().replaceEngine(0, MiniGltTrgEngine())
  def conditions(self): return self.__cond

class MiniGltExtTrgCond(MiniGLTTrgConditionsValue):
  def __init__(self):
    MiniGLTTrgConditionsValue.__init__(self)
  def external(self): return True

class MiniGltTrgEngine(TrgEngine):
  def __init__(self):
    TrgEngine.__init__(self)
    self.__request = TrgReadout()
    cond = MiniGltExtTrgCond()
    self.__cv = cond.value()
  def participate(self,conditionsValue):
    if conditionsValue & self.__cv:
      return True
    return False
  def prescale(self):   return 0
  def request(self):    return self.__request
  

def test(cmd=None, debug=0, disconnectOnExit=1, parse=1, lowlevel=0):
  global tem , glt
  gc.disable()
  
  print "Test of external trigger."
  print "External trigger should be triggering at ~100Hz"
  
  log.getLogger().setLevel(log.INFO)

  options = Options(['server', 'schema'], ['evtdebug'])
  try:
    options.parse()
  except Exception, msg:
    options.usage(str(msg))
    return

  if cmd == None:
    cmd = CmdCli()
    cmd.connect(options.server)
  lat = GLAT(cmd)
  lat.readSchema(options.schema)
  lat.applyConfig()

  try:
    lat.LCB.setEvtEnable(True)
  except Exception, e:
    log.error(e)
    raise

  if lat.GEM is not None:
    trigger = GasuExtTrg()
    trigger.GEM(lat.GEM)
  else:
    glt = GGLT(cmd)
    glt.initialize()
    trigger = MiniGltExtTrg()
    trigger.GLT(glt)
  trigger.commit()

  trigger.disable()

  if options.evtdebug is not None:
    debug = options.evtdebug
  ec = EvtCli(debug)
  ec.connect(options.server)
  ec.socket().settimeout(1)
  
  ocs = GOCS(cmd)
  # ocs.evtHandlerSetVerbosity(0xf)
  ocs.evtHandlerSendAll()

  # Make sure counters are reset
  trigger.triggerCounters().reset()
  
  # Take 1 second of events
  trigger.enable()
  time.sleep(1)
  trigger.disable()
  time.sleep(1)
  
  # get stats
  trgStats = trigger.triggerCounters().stats()
  nSent    = trgStats.sent()
  nDisc    = trgStats.discarded()
  nPre     = trgStats.prescaled()

  
  # loop over "sent"
  nSeen = 0
  for i in range(nSent):
    if lowlevel==1:
      (header, ec.dat) = ec.read()
    else:
      try:
        ec.readEvent()
      except IOError, e:
        pass
        print e
    if parse==1:
      ec.setCurrentContribution(ec.GEM)
      if ec.getContributionStatus() == 0:
        ec.evDumpGEM()
        if (ec.evt.gem.triggerSummary & 0x80):
          nSeen += 1

  if nSent != nSeen:
    print
    print "  !!!!FAILED!!!!"
    print "  nSent (%d) not equal to nSeen (%d)" % ( nSent, nSeen )
    print "  GEM stats: nDiscarded= %d" % (nDisc)
    print "             nPrescaled= %d" % (nPre )    
  else:
    print
    print "  TEST PASSED"    
  
  if disconnectOnExit:
    cmd.disconnect()
  ec.disconnect()

if __name__ == '__main__':
  test()

---