DataModelTestConfig Namespace Reference

Classes
class	LoadReadDicts
class	LoadWriteDicts

Functions
def	DataModelTestFlags (infile=None, evtMax=20, **kw)
def	DataModelTestCfg (flags, testName, loadReadDicts=False, loadWriteDicts=False, EventsPerLB=None, TimeStampInterval=None, readCatalog=None)
def	TestOutputCfg (flags, stream, itemList, typeNames=[], metaItemList=[])
def	LoadWriteDictsCfg (flags)
def	LoadReadDictsCfg (flags)
def	rnt (flags)

Function Documentation

DataModelTestCfg()

def DataModelTestConfig.DataModelTestCfg	(		flags,
			testName,
			loadReadDicts = False,
			loadWriteDicts = False,
			EventsPerLB = None,
			TimeStampInterval = None,
			readCatalog = None
	)

Definition at line 64 of file DataModelTestConfig.py.

def DataModelTestCfg (flags, testName,
                      loadReadDicts = False,
                      loadWriteDicts = False,
                      EventsPerLB = None,
                      TimeStampInterval = None,
                      readCatalog = None):
    from AthenaConfiguration.MainServicesConfig import \
        MainServicesCfg, MessageSvcCfg
    cfg = MainServicesCfg (flags)
    cfg.merge (MessageSvcCfg (flags))
    cfg.getService("MessageSvc").debugLimit = 10000
    cfg.addService (CompFactory.ClassIDSvc (OutputLevel = INFO))
    cfg.addService (CompFactory.ChronoStatSvc (ChronoPrintOutTable = False,
                                               PrintUserTime = False,
                                               StatPrintOutTable = False))
 
    if flags.Input.Files == ['_ATHENA_GENERIC_INPUTFILE_NAME_']:
        # No input file --- configure like an event generator,
        # and make an xAODEventInfo.
        from McEventSelector.McEventSelectorConfig import McEventSelectorCfg
        mckw = {}
        if EventsPerLB is not None:
            mckw['EventsPerLB'] = EventsPerLB
        if TimeStampInterval is not None:
            mckw['TimeStampInterval'] = TimeStampInterval
        cfg.merge (McEventSelectorCfg (flags, **mckw))
 
        from xAODEventInfoCnv.xAODEventInfoCnvConfig import EventInfoCnvAlgCfg
        cfg.merge (EventInfoCnvAlgCfg (flags, disableBeamSpot = True))
    elif not flags.Input.Files[0].endswith ('.bs'):
        # Configure reading.
        from AthenaPoolCnvSvc.PoolReadConfig import PoolReadCfg
        cfg.merge (PoolReadCfg (flags))
 
    # Load dictionaries if requested.
    if loadWriteDicts:
        cfg.merge (LoadWriteDictsCfg (flags))
    if loadReadDicts:
        cfg.merge (LoadReadDictsCfg (flags))
 
    # Prevent races when we run tests in parallel in the same directory.
    if flags.Output.StorageTechnology.EventData == 'ROOTRNTUPLE':
        testName = testName + '_rntup'
    fileCatalog = testName + '_catalog.xml'
    from AthenaPoolCnvSvc.PoolCommonConfig import PoolSvcCfg
    kw = {'WriteCatalog' : 'file:' + fileCatalog}
    if readCatalog:
        kw['ReadCatalog'] = ['file:' + readCatalog]
    cfg.merge (PoolSvcCfg (flags, **kw))
    import os
    try:
        os.remove (fileCatalog)
    except OSError:
        pass
 
 
    return cfg
 
 
#
# Configure an output stream.
#

DataModelTestFlags()

def DataModelTestConfig.DataModelTestFlags	(		infile = None,
			evtMax = 20,
		**	kw
	)

Definition at line 25 of file DataModelTestConfig.py.

def DataModelTestFlags (infile = None, evtMax = 20, **kw):
    flags = initConfigFlags()
    flags.Exec.MaxEvents = evtMax
    flags.Exec.OutputLevel = INFO
    flags.Common.MsgSourceLength = 18
 
    # Disable FPE auditing.
    flags.Exec.FPE = -2
 
    # Set input/output files.
    if infile:
        flags.Input.Files = [infile]
    for stream, outfile in kw.items():
        flags.addFlag (f'Output.{stream}FileName', outfile)
 
    # Block input file peeking.
    from Campaigns.Utils import Campaign
    flags.Input.RunNumbers = [0]
    flags.Input.TimeStamps = [0]
    flags.Input.ProcessingTags = []
    flags.Input.TypedCollections = []
    flags.Input.isMC = True
    flags.IOVDb.GlobalTag = ''
    flags.Input.MCCampaign = Campaign.Unknown
    flags.fillFromArgs()
 
    if flags.Output.StorageTechnology.EventData == 'ROOTRNTUPLE':
        def to_rntup (s):
            return s.replace ('.root', '.rntup.root')
        flags.Input.Files = [to_rntup(f) for f in flags.Input.Files]
        for stream, outfile in kw.items():
            setattr (flags.Output, stream+'FileName', to_rntup (outfile))
    
    return flags
 
 
#
# Common configuration for tests.
#

LoadReadDictsCfg()

def DataModelTestConfig.LoadReadDictsCfg

(

flags

)

Definition at line 200 of file DataModelTestConfig.py.

def LoadReadDictsCfg (flags):
    acc = ComponentAccumulator()
    acc.addEventAlgo (LoadReadDicts())
    return acc
 
 

LoadWriteDictsCfg()

def DataModelTestConfig.LoadWriteDictsCfg

(

flags

)

Definition at line 173 of file DataModelTestConfig.py.

def LoadWriteDictsCfg (flags):
    acc = ComponentAccumulator()
    acc.addEventAlgo (LoadWriteDicts())
    return acc
 
 
# Arrange to get dictionaries loaded for read tests.
# Do this as an algorithm so we can defer it to initialize().
# In some cases, loading DSOs during initial python processing
# can cause component loading to fail.

rnt()

def DataModelTestConfig.rnt

(

flags

)

Definition at line 206 of file DataModelTestConfig.py.

def rnt (flags):
    is_rntuple = flags.Output.StorageTechnology.EventData == 'ROOTRNTUPLE'
    if is_rntuple:
        return True, lambda k: ''
    return False, lambda k:k
 

TestOutputCfg()

def DataModelTestConfig.TestOutputCfg	(		flags,
			stream,
			itemList,
			typeNames = [],
			metaItemList = []
	)

Definition at line 126 of file DataModelTestConfig.py.

def TestOutputCfg (flags, stream, itemList, typeNames = [], metaItemList = []):
    from OutputStreamAthenaPool.OutputStreamConfig import OutputStreamCfg, outputStreamName
    acc = ComponentAccumulator()
    itemList = ['xAOD::EventInfo#EventInfo',
                'xAOD::EventAuxInfo#EventInfoAux.'] + itemList
    helperTools = []
    metaItemList = ["IOVMetaDataContainer#*"]
    if typeNames:
        helperTools = [ CompFactory.xAODMaker.EventFormatStreamHelperTool(
            f'{stream}_EventFormatStreamHelperTool',
            Key = f'EventFormat{stream}',
            TypeNames = typeNames,
            DataHeaderKey = f'Stream{stream}') ]
        metaItemList = [ f'xAOD::EventFormat#EventFormat{stream}' ] + metaItemList
    acc.merge (OutputStreamCfg (flags, stream,
                                ItemList = itemList,
                                HelperTools = helperTools,
                                MetadataItemList = metaItemList))
    if typeNames:
        alg = acc.getEventAlgo (outputStreamName(stream))
        alg.WritingTool.SubLevelBranchName = '<key>'
        acc.getService ('AthenaPoolCnvSvc').PoolAttributes += ["DEFAULT_SPLITLEVEL='1'"]
    return acc
    
    
 
 
# Arrange to get dictionaries loaded for write tests.
# Do this as an algorithm so we can defer it to initialize().
# In some cases, loading DSOs during initial python processing
# can cause component loading to fail.