MuonVertexValidationRun.py

Go to the documentation of this file.

#
# Copyright (C) 2002-2025 CERN for the benefit of the ATLAS collaboration
#
 
from AthenaConfiguration.ComponentAccumulator import ComponentAccumulator
from AthenaConfiguration.ComponentFactory import CompFactory
 
 
def splitOnComma(inputs):
    files = []
    for item in inputs: files.extend(item.split(','))
 
    return files
 
 
def GetArgsFromParser():
    from argparse import ArgumentParser
 
    parser = ArgumentParser()
    parser.add_argument( "-i", "--inputFile", required=True, help="Input files to run on. Files can be comma or space separated", nargs="+")
    parser.add_argument( "-o", "--outputFile", default="MSVtxVal_out.NTUP.root", help="output root file")
    parser.add_argument("--triggers", action="store_true", help="Configure the trigger tools")
    parser.add_argument("--data", action="store_true", help="run on a data file")
    parser.add_argument("--maxEvents", default=-1, type=int, help="How many events shall be run maximally")
    parser.add_argument("--skipEvents", default=0, type=int, help="How many events shall be skipped")
    parser.add_argument("--threads", default=1, type=int, help="number of threads")
    
    args = parser.parse_args()
    args.inputFile = splitOnComma(args.inputFile) # to support comma separated input files
 
    return args
 
def MSVtxValidationCfg(flags, name="MSVertexValidationAlg", outStream="MSVtxValidation", outFile="out.root", **kwargs):
    # outStream defines the steam to place the tree and histograms 
    from TriggerMatchingTool.TriggerMatchingToolConfig import TriggerMatchingToolCfg
    result = ComponentAccumulator()
    # setting algorithm properties here via kwargs.setdefault("<property name>", <property value>)
    alg = CompFactory.MSVtxValidationAlg(name, **kwargs)
    from MuonConfig.MuonConfigUtils import setupHistSvcCfg
    result.merge(setupHistSvcCfg(flags,outFile=outFile, outStream=outStream))
    if kwargs.get("readTriggers", False): result.getPrimaryAndMerge(TriggerMatchingToolCfg(flags, name='R3MatchingTool'))
    result.addEventAlgo(alg)
 
    return result
 
 
def execute(cfg):
    cfg.printConfig(withDetails=True, summariseProps=True)
    if not cfg.run().isSuccess(): exit(1)
 
 
if __name__ == "__main__":
    from AthenaConfiguration.AllConfigFlags import initConfigFlags
    from MuonCondTest.MdtCablingTester import setupServicesCfg
 
    args = GetArgsFromParser()
    flags = initConfigFlags()
    flags.Concurrency.NumThreads = args.threads
    flags.Exec.MaxEvents = args.maxEvents
    flags.Exec.SkipEvents = args.skipEvents
    flags.Concurrency.NumConcurrentEvents = args.threads
    flags.Input.Files = args.inputFile 
    flags.Scheduler.ShowDataDeps = True 
    flags.Scheduler.ShowDataFlow = True
    flags.lock()
    cfg = setupServicesCfg(flags)
    cfg.merge(MSVtxValidationCfg(flags, outFile=args.outputFile, readTriggers=args.triggers, isMC=not args.data))
    execute(cfg)