Skip to content

Instantly share code, notes, and snippets.

@mazurov
Last active August 29, 2015 14:07
Show Gist options
  • Select an option

  • Save mazurov/c356d5de6d09305667bc to your computer and use it in GitHub Desktop.

Select an option

Save mazurov/c356d5de6d09305667bc to your computer and use it in GitHub Desktop.
How to test new PpmByteStreamDecoder. Run command from setup.sh file in your lxplus console (ATLAS environment should be set before)
from ByteStreamCnvSvc import ReadByteStream
import struct
# InputFile = "data14_calib.00236011.calibration_L1CaloEnergyScan.daq.RAW._lb0000._SFO-1._0001.data"
# InputFile = "data_test.1412246305.calibration_.daq.RAW._lb0000._ROS-00._0001.data"
# BsFolder = "/afs/cern.ch/user/l/l1ccalib/w0/am"
#InputFile = "data_test.1412873836.calibration_.daq.RAW._lb0000._ROS-00._0001.data"
InputFile = "data_test.1412873981.calibration_.daq.RAW._lb0000._ROS-00._0001.data"
#InputFile = "data_test.1412875644.calibration_.daq.RAW._lb0000._ROS-00._0001.data"
# InputFile = "data_test.1412875805.calibration_.daq.RAW._lb0000._ROS-00._0001.data"
BsFolder = "/afs/cern.ch/work/m/mwessels/public/data/bs_test"
# InputFile = "data_test.1412246305.calibration_.daq.RAW._lb0000._ROS-00._0001.data"
# BsFolder = "/afs/cern.ch/user/a/amazurov/ATLAS"
svcMgr.ByteStreamInputSvc.FullFileName = [
#"/afs/cern.ch/user/l/l1ccalib/w0/am/%s" % InputFile
"%s/%s" % (BsFolder, InputFile)
]
from AthenaCommon.GlobalFlags import globalflags
globalflags.DetGeo = 'atlas'
globalflags.DataSource = 'data'
from AthenaCommon.DetFlags import DetFlags
DetFlags.detdescr.all_setOff()
DetFlags.detdescr.Calo_setOn()
globalflags.DetDescrVersion = "ATLAS-GEO-20-00-02"
include("RecExCond/AllDet_detDescr.py")
svcMgr.IOVDbSvc.GlobalTag = "COMCOND-BLKPA-006-07"
svcMgr.IOVDbSvc.forceRunNumber = 236011
svcMgr.IOVDbSvc.forceLumiblockNumber = 126
include("TrigT1CaloByteStream/ReadLVL1CaloBSRun2_jobOptions.py")
import AthenaCommon.Constants as Lvl
from AthenaCommon.AlgSequence import AlgSequence, AthSequencer
job = AlgSequence()
class PyTriggerTowerRef(PyAthena.Alg):
def __init__(self, name='PyTriggerTowerRef', **kw):
# init base class
kw['name'] = name
self.dbname = kw['db']
super(PyTriggerTowerRef, self).__init__(**kw)
def initialize(self):
self.sg = PyAthena.py_svc('StoreGateSvc')
return PyAthena.StatusCode.Success
def execute(self):
# cpm = self.sg['TriggerTowers']
tt = self.sg['TriggerTowers']
# Loop throw towers
ntower = 1
for n, c in enumerate(tt):
if c.coolId() != 0:
print "#" * 80
print "coolid = 0x%x" % c.coolId()
print "layer = %d" % struct.unpack('<B', c.layer())[0]
print "(eta,phi) = %f, %f" % (c.eta(), c.phi())
print "fadc = %s" % ",".join([str(x) for x in c.adc()])
print "cpET = %d" % struct.unpack('<B', c.cpET())[0]
print "jepET = %d" % struct.unpack('<B', c.jepET())[0]
print "#" * 80
ntower += 1
# self.msg.info("row ==> " + str(row))
self.setFilterPassed(True)
return PyAthena.StatusCode.Success
def finalize(self):
self.db["events"] = self.rows
self.msg.info("finalize ==> total events: %s" % len(self.db["events"]))
self.db.close()
self.msg.info("reference database ==> %s" % self.dbname)
return PyAthena.StatusCode.Success
pass
job += PyTriggerTowerRef("PyTriggerTowerRef", db="%s.db" % InputFile,
OutputLevel=Lvl.INFO)
theApp.EvtMax = 10
TARGET_ATHENA=19.1.3
TARGET_DIR=${TARGET_ATHENA}-test
TEST_SCRIPT=https://gist.githubusercontent.com/mazurov/c356d5de6d09305667bc/raw/335524cb347c8395a1fac197b4b7dee653765321/jobo.py
RED='\e[0;31m'
NC='\e[0m' # No Color
echo -e "${RED}Create and change working directory to ${TARGET_DIR}${NC}"
mkdir $TARGET_DIR
cd $TARGET_DIR
echo -e "${RED}Setup Athena ${TARGET_ATHENA}${NC}"
asetup ${TARGET_ATHENA},here
# Get new packages
echo -e "${RED}Get development source of TrigT1CaloByteStream${NC}"
cmt co -r trunk Trigger/TrigT1/TrigT1CaloByteStream
echo -e "${RED}Get source of xAODTrigL1Calo${NC}"
cmt co -r xAODTrigL1Calo-00-00-11 Event/xAOD/xAODTrigL1Calo
echo -e "${RED}Setup WorkArea${NC}"
setupWorkArea.py
echo -e "Build packages"
cd WorkArea/cmt
cmt br cmt config
cmt br cmt make -j
source setup.sh
cd ..
echo -e "${RED}Change working directory to `pwd`${NC}"
echo -e "${RED}Get test script${NC}"
curl -O ${TEST_SCRIPT}
echo -e "${RED}Run jobo.py options${NC}"
athena -l VERBOSE jobo.py | tee jobo.log
wget -nv -O- https://gist.githubusercontent.com/mazurov/c356d5de6d09305667bc/raw/ppmdecoder.sh | source /dev/stdin
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment