AthenaPoolExample_Write.py

Go to the documentation of this file.

#!/env/python
 
# Copyright (C) 2002-2025 CERN for the benefit of the ATLAS collaboration
 
 
 
 
 
 
from AthenaConfiguration.AllConfigFlags import initConfigFlags
from AthenaConfiguration.ComponentFactory import CompFactory
from AthenaCommon.Constants import DEBUG
from OutputStreamAthenaPool.OutputStreamConfig import OutputStreamCfg, outputStreamName
 
stream1name = "Stream1"
file1Name = "ROOTTREE:SimplePoolFile1.root"
stream2name = "Stream2"
file2Name = "ROOTTREE:SimplePoolFile2.root"
stream3name = "Stream3"
file3Name =  "ROOTTREE:EmptyPoolFile.root"
noTag = True
 
# Setup flags
flags = initConfigFlags()
flags.Common.MsgSuppression = False
flags.Exec.MaxEvents = 20
flags.Input.Files = []
flags.addFlag(f"Output.{stream1name}FileName", file1Name)
flags.addFlag(f"Output.{stream2name}FileName", file2Name)
flags.addFlag(f"Output.{stream3name}FileName", file3Name)
flags.Exec.DebugMessageComponents = [ outputStreamName(stream1name) , "PoolSvc", "AthenaPoolCnvSvc", "WriteData" ]
flags.lock()
 
# Main services
from AthenaConfiguration.MainServicesConfig import MainServicesCfg
acc = MainServicesCfg( flags )
 
from xAODEventInfoCnv.xAODEventInfoCnvConfig import EventInfoCnvAlgCfg
acc.merge( EventInfoCnvAlgCfg( flags, disableBeamSpot=True ) )
 
# Load CutFlowSvc
from EventBookkeeperTools.EventBookkeeperToolsConfig import CutFlowSvcCfg
acc.merge( CutFlowSvcCfg( flags ) )    #addMetaDataToAllOutputFiles=True ) )
# Ensure proper metadata propagation
from IOVDbSvc.IOVDbSvcConfig import IOVDbSvcCfg
acc.merge( IOVDbSvcCfg( flags ) )
 
# Pool writing
acc.addEventAlgo( CompFactory.AthPoolEx.WriteData("WriteData", OutputLevel = DEBUG) )
 
WriteTag = CompFactory.AthPoolEx.WriteTag("WriteTag", OutputLevel = DEBUG)
WriteTag.Magic = 1
acc.addEventAlgo( WriteTag )
 
MagicWriteTag = CompFactory.AthPoolEx.WriteTag("MagicWriteTag", OutputLevel = DEBUG)
MagicWriteTag.Key = "MagicTag"
MagicWriteTag.Magic = 24
acc.addEventAlgo( MagicWriteTag )
 
# ----------------  Output Stream 1 configuration
from AthenaPoolExampleAlgorithms.AthenaPoolExampleConfig import AthenaPoolExampleWriteCfg
acc.merge( AthenaPoolExampleWriteCfg( flags, stream1name, writeCatalog = "file:Catalog1.xml",
                                      disableEventTag = noTag ) )
 
stream1ca = OutputStreamCfg( flags, stream1name, disableEventTag = noTag,
                             ItemList = [
                                 'EventInfo#*', 'EventStreamInfo#*',
                                 'ExampleHitContainer#MyHits', 'ExampleHitContainer#PetersHits' ] )
stream1 = stream1ca.getEventAlgo( outputStreamName( stream1name ) )
stream1.WritingTool.AttributeListKey = MagicWriteTag.Key
acc.merge( stream1ca )
 
# ----------------  Output Stream 2 configuration
acc.merge( AthenaPoolExampleWriteCfg( flags, stream2name, disableEventTag = noTag ) )
stream2ca = OutputStreamCfg(flags, stream2name, disableEventTag = noTag,
                            ItemList = ['EventInfo#*', 'ExampleHitContainer#MyHits'] )
stream2 = stream2ca.getEventAlgo( outputStreamName( stream2name ) )
stream2.WritingTool.AttributeListKey = "RunEventTag"
acc.merge( stream2ca )
 
# ----------------  Output Stream 3 configuration
filterAlg = CompFactory.AthPoolEx.PassNoneFilter("PassNoneFilter", OutputLevel = DEBUG)
acc.addEventAlgo( filterAlg )
acc.merge( AthenaPoolExampleWriteCfg( flags, stream3name, disableEventTag = noTag ) )
stream3ca = OutputStreamCfg(flags, stream3name, disableEventTag = noTag )
stream3 = stream3ca.getEventAlgo( outputStreamName( stream3name ) )
stream3.RequireAlgs = [ "PassNoneFilter" ]
acc.merge( stream3ca )
 
#--------------------------------------------------------------
# Set output level threshold (2=DEBUG, 3=INFO, 4=WARNING, 5=ERROR, 6=FATAL)
#--------------------------------------------------------------
stream1.WritingTool.OutputLevel = 3
stream1.HelperTools[0].OutputLevel = 3
stream2.WritingTool.OutputLevel = 3
stream2.HelperTools[0].OutputLevel = 3
 
# Run
import sys
sc = acc.run(flags.Exec.MaxEvents)
sys.exit(sc.isFailure())