Collaboration diagram for python.MultipleStreamManager.MultipleStreamManager:

Public Member Functions
def	__init__ (self)

def	NewByteStream (self, StreamName, FileName="default")

def	NewPoolStream (self, StreamName, FileName="default", asAlg=False, noTag=False)

def	NewPoolRootStream (self, StreamName, FileName="default", asAlg=False)

def	NewPoolRootStreamExtension (self, StreamName, Parent="StreamDAOD_PHYS", asAlg=False)

def	NewVirtualStream (self, StreamName, FileName="default", asAlg=False)

def	NewRootStream (self, StreamName, FileName=None, TreeName=None, asAlg=False)

def	NewStream (self, StreamName, FileName="default", type='pool', asAlg=False, TreeName=None, Parent=None, noTag=False)

def	GetStream (self, NameOrIndex)

def	StreamExists (self, StreamName)

def	Print (self)

def	Lock (self)

def	Unlock (self)

def	AddItemToAllStreams (self, item)

def	RemoveItemFromAllStreams (self, item)

def	AddMetaDataItemToAllStreams (self, item)

def	RemoveMetaDataItemFromAllStreams (self, item)

def	RenameAllStreams (self, NameList)

def	WriteSkimDecisionsOfAllStreams (self)

Public Attributes
	StreamList

	nStream

	StreamDict

Private Attributes
	_Locked

Detailed Description

This class helps managing multiple streams.
Normal users only manipulate their own streams with functions like NewStream or GetStream,
while commands like Lock, AddItemToAllStreams or RenameAllStreams are for the real manager
(e.g. RecExCommon or a job transform class)

Definition at line 566 of file MultipleStreamManager.py.

Constructor & Destructor Documentation

◆ init()

def python.MultipleStreamManager.MultipleStreamManager.__init__ ( self )

Definition at line 571 of file MultipleStreamManager.py.

     def __init__(self):
         self.StreamList=[]
         self.nStream=0
         self.StreamDict={}
         self._Locked=False
         return
  

Member Function Documentation

◆ AddItemToAllStreams()

def python.MultipleStreamManager.MultipleStreamManager.AddItemToAllStreams	(	self,
		item
	)

Definition at line 753 of file MultipleStreamManager.py.

     def AddItemToAllStreams(self, item):
         if self._Locked is True:
             raise AssertionError("MSMgr is locked. AddItemToAllStreams cannot be used.")
         for Stream in self.StreamList:
             Stream.AddItem(item)
         return
  

◆ AddMetaDataItemToAllStreams()

def python.MultipleStreamManager.MultipleStreamManager.AddMetaDataItemToAllStreams	(	self,
		item
	)

Definition at line 767 of file MultipleStreamManager.py.

     def AddMetaDataItemToAllStreams(self, item):
         if self._Locked is True:
             raise AssertionError("MSMgr is locked. AddMetaDataItemToAllStreams cannot be used.")
         for Stream in self.StreamList:
             Stream.AddMetaDataItem(item)
         return
  

◆ GetStream()

def python.MultipleStreamManager.MultipleStreamManager.GetStream	(	self,
		NameOrIndex
	)

Definition at line 714 of file MultipleStreamManager.py.

     def GetStream(self, NameOrIndex):
         #If NameOrIndex is an int, treat it as an index
         if isinstance(NameOrIndex, int):
             if NameOrIndex < self.nStream:
                 return self.StreamList[NameOrIndex]
             else:
                 raise IndexError("ERROR: No stream with index %i is defined in MultipleStreamManager."%NameOrIndex)
  
         #else treat NameOrIndex as a name in the Stream Dictionary
         try:
             #Check wheter a stream with the same name already exists
             index=self.StreamDict[NameOrIndex]
         except KeyError:
             raise NameError("Stream %s undefined!"%NameOrIndex)
  
         return self.StreamList[index]
  

◆ Lock()

def python.MultipleStreamManager.MultipleStreamManager.Lock ( self )

Definition at line 745 of file MultipleStreamManager.py.

     def Lock(self):
         self._Locked=True
         return
  

◆ NewByteStream()

def python.MultipleStreamManager.MultipleStreamManager.NewByteStream	(	self,
		StreamName,
		FileName = `"default"`
	)

Definition at line 578 of file MultipleStreamManager.py.

     def NewByteStream(self,StreamName,FileName="default"):
         if FileName=="default":
             FileName=StreamName
         return self.NewStream(StreamName,FileName,type='bytestream')
  

◆ NewPoolRootStream()

def python.MultipleStreamManager.MultipleStreamManager.NewPoolRootStream	(	self,
		StreamName,
		FileName = `"default"`,
		asAlg = `False`
	)

Definition at line 586 of file MultipleStreamManager.py.

     def NewPoolRootStream(self,StreamName,FileName="default", asAlg=False):
         theStream = self.NewStream(StreamName,FileName,type='pool',asAlg=asAlg)
         from AthenaCommon.AppMgr import theApp
         from AthenaPoolCnvSvc import PoolAttributeHelper as pah
         svcMgr = theApp.serviceMgr()
  
         theStream.Stream.WritingTool.SubLevelBranchName = "<key>"
         # Use ZSTD w/ Level 5 for DAODs
         svcMgr.AthenaPoolCnvSvc.PoolAttributes += [ pah.setFileCompAlg( FileName, "5" ) ]
         svcMgr.AthenaPoolCnvSvc.PoolAttributes += [ pah.setFileCompLvl( FileName, "5" ) ]
         # By default use a maximum basket buffer size of 128k and minimum buffer entries of 10
         svcMgr.AthenaPoolCnvSvc.PoolAttributes += [ pah.setMaxBufferSize( FileName, "131072" ) ]
         svcMgr.AthenaPoolCnvSvc.PoolAttributes += [ pah.setMinBufferEntries( FileName, "10" ) ]
         # By default use 20 MB AutoFlush [or 100 (10) events for DAODs (everything else) for SharedWriter w/ parallel compression]
         # for event data except for a number of select formats (see below)
         TREE_AUTO_FLUSH = -20000000
         from PyUtils.moduleExists import moduleExists
         if moduleExists ('AthenaMP'): # AthenaMP not in AthAnalysis project
             from AthenaMP.AthenaMPFlags import jobproperties as amjp
             if amjp.AthenaMPFlags.UseSharedWriter() and amjp.AthenaMPFlags.UseParallelCompression():
                 TREE_AUTO_FLUSH = 100 if "DAOD_" in StreamName else 10
         # By default use split-level 0 except for DAOD_PHYSLITE which is maximally split
         CONTAINER_SPLITLEVEL = 0
         if StreamName in ["StreamDAOD_PHYSVAL"]:
             TREE_AUTO_FLUSH = 100
         if StreamName in ["StreamDAOD_PHYS"]:
             TREE_AUTO_FLUSH = 500
         if StreamName in ["StreamDAOD_PHYSLITE", "StreamD2AOD_PHYSLITE"]:
             TREE_AUTO_FLUSH = 1000
             CONTAINER_SPLITLEVEL = 1
         svcMgr.AthenaPoolCnvSvc.PoolAttributes += [ pah.setTreeAutoFlush( FileName, "CollectionTree", str(TREE_AUTO_FLUSH) ) ]
         svcMgr.AthenaPoolCnvSvc.PoolAttributes += [ pah.setContainerSplitLevel( FileName, "CollectionTree", str(CONTAINER_SPLITLEVEL) ) ]
         svcMgr.AthenaPoolCnvSvc.PoolAttributes += [ pah.setContainerSplitLevel( FileName, "Aux.", str(CONTAINER_SPLITLEVEL) ) ]
         svcMgr.AthenaPoolCnvSvc.PoolAttributes += [ pah.setContainerSplitLevel( FileName, "Dyn.", "1" ) ]
         return theStream
  
  

◆ NewPoolRootStreamExtension()

def python.MultipleStreamManager.MultipleStreamManager.NewPoolRootStreamExtension	(	self,
		StreamName,
		Parent = `"StreamDAOD_PHYS"`,
		asAlg = `False`
	)

Definition at line 623 of file MultipleStreamManager.py.

     def NewPoolRootStreamExtension(self,StreamName,Parent="StreamDAOD_PHYS", asAlg=False):
         index=0
         FileName="default"
         try:
             #Check wheter a StreamDAOD_PHYS stream exists
             index=MSMgr.StreamDict[Parent]
         except KeyError:
             index=-1
         if index >= 0:
             FileName=self.StreamList[index].Stream.OutputFile
             self.StreamList[index].Stream.WritingTool.SaveDecisions = True
         theStream = self.NewStream(StreamName,FileName,Parent=Parent,type='extension',asAlg=asAlg, noTag=True)
         theStream.Stream.MetadataItemList = [ ]
         theStream.Stream.ItemList = [ ]
         theStream.Stream.WritingTool.SubLevelBranchName = "<key>"
         theStream.Stream.WritingTool.OutputCollection = "POOLContainer_" + StreamName
         theStream.Stream.WritingTool.PoolContainerPrefix = "CollectionTree_" + StreamName
         from AthenaCommon.AppMgr import theApp
         svcMgr = theApp.serviceMgr()
         svcMgr.AthenaPoolCnvSvc.PoolAttributes += [ "DatabaseName = '" + FileName + "'; INDEX_MASTER = 'POOLContainer(DataHeader)'" ]
         # By default use split-level 0 except for DAOD_PHYSLITE which is maximally split
         from AthenaPoolCnvSvc import PoolAttributeHelper as pah
         CONTAINER_SPLITLEVEL = 0
         if Parent in ["StreamDAOD_PHYSVAL"]:
             TREE_AUTO_FLUSH = 100
         if Parent in ["StreamDAOD_PHYS"]:
             TREE_AUTO_FLUSH = 500
         if Parent in ["StreamDAOD_PHYSLITE", "StreamD2AOD_PHYSLITE"]:
             TREE_AUTO_FLUSH = 1000
             CONTAINER_SPLITLEVEL = 1
         svcMgr.AthenaPoolCnvSvc.PoolAttributes += [ pah.setTreeAutoFlush( FileName, "CollectionTree_" + StreamName, str(TREE_AUTO_FLUSH) ) ]
         svcMgr.AthenaPoolCnvSvc.PoolAttributes += [ pah.setContainerSplitLevel( FileName, "CollectionTree_" + StreamName, str(CONTAINER_SPLITLEVEL) ) ]
         svcMgr.AthenaPoolCnvSvc.PoolAttributes += [ pah.setContainerSplitLevel( FileName, "Aux.", str(CONTAINER_SPLITLEVEL) ) ]
         svcMgr.AthenaPoolCnvSvc.PoolAttributes += [ pah.setContainerSplitLevel( FileName, "Dyn.", "1" ) ]
         return theStream
  

◆ NewPoolStream()

def python.MultipleStreamManager.MultipleStreamManager.NewPoolStream	(	self,
		StreamName,
		FileName = `"default"`,
		asAlg = `False`,
		noTag = `False`
	)

Definition at line 583 of file MultipleStreamManager.py.

     def NewPoolStream(self,StreamName,FileName="default", asAlg=False, noTag=False):
         return self.NewStream(StreamName,FileName,type='pool',asAlg=asAlg, noTag=noTag)
  

◆ NewRootStream()

def python.MultipleStreamManager.MultipleStreamManager.NewRootStream	(	self,
		StreamName,
		FileName = `None`,
		TreeName = `None`,
		asAlg = `False`
	)

Function used to create a new D3PD 'stream'. The object returned by
   it behaves both as an AugmentedStreamBase, and as a D3PD::MakerAlg
   configurable.

   Arguments:
      StreamName: Logical name of the D3PD stream. Used also as the D3PD
          TTree name in case a tree name is not specified.
      FileName: Name of the output file into which the D3PD should be
        written. If not specified, '<StreamName>.root' is used.
      TreeName: Name of the TTree in the created file. If not specified,
        StreamName is used for the TTree name as well.
      asAlg: Selects whether the 'stream' should be configured as an
     actual stream, or an algorithm.

Definition at line 662 of file MultipleStreamManager.py.

     def NewRootStream(self,StreamName,FileName=None,TreeName=None,asAlg=False):
         """Function used to create a new D3PD 'stream'. The object returned by
            it behaves both as an AugmentedStreamBase, and as a D3PD::MakerAlg
            configurable.
  
            Arguments:
               StreamName: Logical name of the D3PD stream. Used also as the D3PD
                           TTree name in case a tree name is not specified.
               FileName: Name of the output file into which the D3PD should be
                         written. If not specified, '<StreamName>.root' is used.
               TreeName: Name of the TTree in the created file. If not specified,
                         StreamName is used for the TTree name as well.
               asAlg: Selects whether the 'stream' should be configured as an
                      actual stream, or an algorithm.
         """
         # Check if a file name was specified or not:
         if FileName is None:
             FileName = StreamName + ".root"
         # Use the common function for creating the stream:
         return self.NewStream( StreamName, FileName, type='root', asAlg = asAlg,
                                TreeName = TreeName )
  

◆ NewStream()

def python.MultipleStreamManager.MultipleStreamManager.NewStream	(	self,
		StreamName,
		FileName = `"default"`,
		type = `'pool'`,
		asAlg = `False`,
		TreeName = `None`,
		Parent = `None`,
		noTag = `False`
	)

Definition at line 684 of file MultipleStreamManager.py.

     def NewStream(self,StreamName,FileName="default",type='pool',asAlg=False,TreeName=None,Parent=None,noTag=False):
         if FileName=="default":
             FileName=StreamName+".pool.root"
         try:
             #Check wheter a stream with the same name already exists
             index=self.StreamDict[StreamName]
         except KeyError:
             #The stream doesn't already exist. Register it and set it up.
             #(This is expected, not actually an error.)
             index=self.nStream
             if type=='pool':
                 self.StreamList += [ AugmentedPoolStream(StreamName,FileName,asAlg,isVirtual=False,noTag=noTag) ]
             elif type=='bytestream':
                 self.StreamList += [ AugmentedByteStream(StreamName,FileName) ]
             elif type=='virtual':
                 self.StreamList += [ AugmentedPoolStream(StreamName,FileName,asAlg,isVirtual=True) ]
             elif type=='root':
                 self.StreamList += [ AugmentedRootStream(StreamName,FileName,TreeName,asAlg) ]
             elif type=='extension':
                 self.StreamList += [ AugmentedPoolStreamExtension(StreamName,FileName,Parent,asAlg,isVirtual=False,noTag=True) ]
             else:
                 raise RuntimeError("Unknown type '%s'"%type)
  
             self.StreamDict[StreamName]=index
             self.nStream+=1
         else:
             #This is the real error case...
             raise NameError("Stream %s already exists"%StreamName)
         return self.StreamList[index]
  

◆ NewVirtualStream()

def python.MultipleStreamManager.MultipleStreamManager.NewVirtualStream	(	self,
		StreamName,
		FileName = `"default"`,
		asAlg = `False`
	)

Definition at line 659 of file MultipleStreamManager.py.

     def NewVirtualStream(self,StreamName,FileName="default", asAlg=False):
         return self.NewStream(StreamName,FileName,type='virtual',asAlg=asAlg)
  

◆ Print()

def python.MultipleStreamManager.MultipleStreamManager.Print ( self )

Definition at line 734 of file MultipleStreamManager.py.

     def Print(self):
         print("**** MultipleStreamManager INFOS ****")
         print("Number of streams:", self.nStream)
         i=0
         for Stream in self.StreamList:
             print("----------------------- Stream #",i," -----------------------")
             Stream.Print()
             i+=1
         return
  

◆ RemoveItemFromAllStreams()

def python.MultipleStreamManager.MultipleStreamManager.RemoveItemFromAllStreams	(	self,
		item
	)

Definition at line 760 of file MultipleStreamManager.py.

     def RemoveItemFromAllStreams(self, item):
         if self._Locked is True:
             raise AssertionError("MSMgr is locked. RemoveItemFromAllStreams cannot be used.")
         for Stream in self.StreamList:
             Stream.RemoveItem(item)
         return
  

◆ RemoveMetaDataItemFromAllStreams()

def python.MultipleStreamManager.MultipleStreamManager.RemoveMetaDataItemFromAllStreams	(	self,
		item
	)

Definition at line 774 of file MultipleStreamManager.py.

     def RemoveMetaDataItemFromAllStreams(self, item):
         if self._Locked is True:
             raise AssertionError("MSMgr is locked. AddMetaDataItemFromAllStreams cannot be used.")
         for Stream in self.StreamList:
             Stream.RemoveMetaDataItem(item)
         return
  

◆ RenameAllStreams()

def python.MultipleStreamManager.MultipleStreamManager.RenameAllStreams	(	self,
		NameList
	)

Definition at line 781 of file MultipleStreamManager.py.

     def RenameAllStreams(self, NameList):
         if self._Locked is True:
             raise AssertionError("MSMgr is locked. RenameAllStreams cannot be used.")
         if not isinstance(NameList, list):
             raise TypeError("RenameAllStreams does not accep arguments of type %s"%type(NameList))
         if len(NameList) != self.nStream:
             raise IndexError("NameList needs to have the same length as self.StreamList.")
  
         i=0
         while i<self.nStream:
             self.StreamList[i].SetOutputFileName(NameList[i])
             i+=1
         return
  

◆ StreamExists()

def python.MultipleStreamManager.MultipleStreamManager.StreamExists	(	self,
		StreamName
	)

Definition at line 731 of file MultipleStreamManager.py.

     def StreamExists(self, StreamName):
         return StreamName in self.StreamDict
  

◆ Unlock()

def python.MultipleStreamManager.MultipleStreamManager.Unlock ( self )

Definition at line 749 of file MultipleStreamManager.py.

     def Unlock(self):
         self._Locked=False
         return
  

◆ WriteSkimDecisionsOfAllStreams()

def python.MultipleStreamManager.MultipleStreamManager.WriteSkimDecisionsOfAllStreams ( self )

Definition at line 795 of file MultipleStreamManager.py.

     def WriteSkimDecisionsOfAllStreams(self):
         if self._Locked:
             raise AssertionError("MSMgr is locked. WriteSkimDecisionsOfAllStreams cannot be used.")
  
         from AthenaCommon.AlgSequence import AlgSequence
         topSequence = AlgSequence()
         for Stream in self.StreamList:
             if Stream.GetAcceptAlgs() or Stream.GetOtherAlgsToBookkeep() or Stream.GetRequireAlgs() or Stream.GetVetoAlgs():
                 sdw=Stream.GetSkimDecisionsWriter()
                 topSequence+=sdw
                 if isinstance(Stream,AugmentedPoolStream):
                     Stream.AddItem("SkimDecisionCollection#"+sdw.SkimDecisionsContainerName)
         return
  
  

Member Data Documentation

◆ _Locked

python.MultipleStreamManager.MultipleStreamManager._Locked

private

Definition at line 575 of file MultipleStreamManager.py.

◆ nStream

python.MultipleStreamManager.MultipleStreamManager.nStream

Definition at line 573 of file MultipleStreamManager.py.

◆ StreamDict

python.MultipleStreamManager.MultipleStreamManager.StreamDict

Definition at line 574 of file MultipleStreamManager.py.

◆ StreamList

python.MultipleStreamManager.MultipleStreamManager.StreamList

Definition at line 572 of file MultipleStreamManager.py.

The documentation for this class was generated from the following file:

MultipleStreamManager.py

Public Member Functions

Public Attributes

Private Attributes