TestWriteFileMetaDataAOD Namespace Reference

Functions
def	writeFileMetaData (flags)
def	testMetaData (file_name)
def	main ()

Function Documentation

main()

def TestWriteFileMetaDataAOD.main

(

)

Run a job writing a file with FileMetaData

Definition at line 52 of file TestWriteFileMetaDataAOD.py.

def main():
    """Run a job writing a file with FileMetaData"""
    msg = Logging.logging.getLogger("TestFileMetaData")
 
    flags = AllConfigFlags.initConfigFlags()
    flags.Exec.OutputLevel = Constants.DEBUG
    flags.Input.Files = TestDefaults.defaultTestFiles.AOD_RUN2_DATA
    flags.Output.AODFileName = "test.pool.root"
    flags.Output.doWriteAOD = True
    flags.Concurrency.NumThreads = 4
    flags.Concurrency.NumConcurrentEvents = 4
    flags.lock()
 
    write = MainServicesConfig.MainServicesCfg(flags)
    write.merge(PoolReadCfg(flags))
    write.merge(writeFileMetaData(flags))
    write.run(100)
 
    try:
        if testMetaData(flags.Output.AODFileName):
            msg.info("File contains xAOD::FileMetaData")
            return 0
        msg.error("File does not contain xAOD::FileMetaData")
    except ReferenceError:
        msg.error("Failed to produce output file")
    except KeyError:
        msg.error("Failed to get metadata item list from file")
    return 1
 
 

testMetaData()

def TestWriteFileMetaDataAOD.testMetaData

(

file_name

)

Check that file metadata is in output

Definition at line 42 of file TestWriteFileMetaDataAOD.py.

def testMetaData(file_name):
    """Check that file metadata is in output"""
    meta = MetaReader.read_metadata([file_name])[file_name]["metadata_items"]
    file_info_items = [
        v for _, v in meta.items() if "FileMetaData" in v
    ]
    print("found file metadata objects:", file_info_items)
    return bool(file_info_items)
 
 

writeFileMetaData()

def TestWriteFileMetaDataAOD.writeFileMetaData

(

flags

)

set up an output stream and xAOD::EventInfo

Definition at line 22 of file TestWriteFileMetaDataAOD.py.

def writeFileMetaData(flags):
    """set up an output stream and xAOD::EventInfo"""
    accumulator = OutputStreamConfig.OutputStreamCfg(flags, streamName="AOD")
    accumulator.merge(SetupMetaDataForStreamCfg(flags, streamName="AOD"))
    # AthAnalysis doesn't have the dictionary for ByteStreamMetadata(Container)
    # Therefore, do not attempt to write it - although this is more of a hack
    if flags.Common.Project == Project.AthAnalysis:
        originalList = accumulator.getEventAlgo('StreamAOD').MetadataItemList
        modifiedList = [s for s in originalList if s != "ByteStreamMetadataContainer#*"]
        accumulator.getEventAlgo('StreamAOD').MetadataItemList = modifiedList
 
    accumulator.merge(
        xAODEventInfoCnvConfig.EventInfoCnvAlgCfg(
            flags=flags, inputKey="", disableBeamSpot=flags.Common.Project is not Enums.Project.Athena
        )
    )
 
    return accumulator