run_hypo_mult_from_json.py

Go to the documentation of this file.

#!/usr/bin/env python
# Copyright (C) 2002-2025 CERN for the benefit of the ATLAS collaboration
 
#
# python script to run an Athena job to run the eEmMult Algorithm
# usage: run_hypo_mult.py --evtMax 10
#
import os
 
 
if __name__ == '__main__':
    
    from add_subsystems import add_subsystems
 
    from AthenaConfiguration.AllConfigFlags import initConfigFlags
    flags = initConfigFlags()
    parser = flags.getArgumentParser()
 
    parser.add_argument(
        "-ifex",
        "--doCaloInput",
        action="store_true",
        dest="doCaloInput",
        help="Decoding L1Calo inputs",
        default=False,
        required=False)
 
    parser.add_argument(
        "--dump",
        action="store_true",
        help="Write out dumps",
        default=False)
 
    parser.add_argument(
        "--dumpTerse",
        action="store_true",
        help="Write out dumps: tersely",
        default=False)
 
 
    # from AthenaConfiguration.TestDefaults import defaultTestFiles
    from AthenaConfiguration.TestDefaults import (defaultGeometryTags,
                                                  defaultConditionsTags)
    # flags.Input.Files = defaultTestFiles.RAW_RUN3_DATA24 # this works
    flags.Input.Files = ['/eos/atlas/atlascerngroupdisk/det-l1calo/OfflineSoftware/TestFiles/data23_13p6TeV.DAOD_L1CALO1.with_gFexDataTowers50.root'] # buttinger
    flags.GeoModel.AtlasVersion = defaultGeometryTags.RUN3
    flags.IOVDb.GlobalTag = defaultConditionsTags.RUN3_DATA24
    
    flags.Output.AODFileName = 'AOD.pool.root'
    flags.Concurrency.NumThreads = 1
    flags.Concurrency.NumConcurrentEvents = 1
    flags.Trigger.doLVL1 = True
 
    flags.Scheduler.ShowDataDeps = True
    flags.Scheduler.CheckDependencies = True
    flags.Scheduler.ShowDataFlow = True
    flags.Trigger.EDMVersion = 3
    flags.Trigger.enableL1CaloPhase1 = True
 
    # Enable only calo for this test
    from AthenaConfiguration.DetectorConfigFlags import setupDetectorFlags
 
    setupDetectorFlags(flags, ['LAr','Tile','MBTS'], toggle_geometry=True)
 
    args = flags.fillFromArgs(parser=parser)
    flags.lock()
    flags.dump()
    
    from AthenaConfiguration.MainServicesConfig import MainServicesCfg
    acc = MainServicesCfg(flags)
 
 
    
 
    from AthenaConfiguration.Enums import Format
    if flags.Input.Format == Format.POOL:
        from AthenaPoolCnvSvc.PoolReadConfig import PoolReadCfg
        acc.merge(PoolReadCfg(flags))
    else:
        subsystems = ('eFex',)
        acc.merge(add_subsystems(flags, subsystems, args, OutputLevel=flags.Exec.OutputLevel))
 
        from TriggerJobOpts.TriggerByteStreamConfig import ByteStreamReadCfg
        acc.merge(ByteStreamReadCfg(flags))
 
        from TrigCaloRec.TrigCaloRecConfig import hltCaloCellSeedlessMakerCfg
        acc.merge(hltCaloCellSeedlessMakerCfg(flags, roisKey=''))
 
    # add in the Algorithm to be run
    from GlobalSimAlgCfg_hypo_mult_from_json import GlobalSimulationAlgCfg
    acc.merge(GlobalSimulationAlgCfg(flags, dump='GS_DUMP' in os.environ))
 
    acc.getService("StoreGateSvc").Dump = True
    
    if acc.run().isFailure():
        import sys
        sys.exit(1)