2 """Overlay test helpers
4 Copyright (C) 2002-2025 CERN for the benefit of the ATLAS collaboration
7 from argparse
import ArgumentParser
8 from AthenaCommon.Debugging
import DbgStage
9 from AthenaConfiguration.AutoConfigFlags
import GetFileMD
10 from AthenaConfiguration.Enums
import LHCPeriod
14 """Common overlay test argument parser"""
15 parser = ArgumentParser(prog=prog)
16 parser.add_argument(
"-d",
"--data", default=
False,
17 action=
"store_true", help=
"Run data overlay")
18 parser.add_argument(
"-n",
"--maxEvents", default=3, type=int,
19 help=
"The number of events to run. 0 skips execution")
20 parser.add_argument(
"-t",
"--threads", default=1, type=int,
21 help=
"The number of concurrent threads to run. 0 uses serial Athena.")
22 parser.add_argument(
"-p",
"--processes", default=0, type=int,
23 help=
"The number of concurrent processes to run. 0 uses serial Athena.")
24 parser.add_argument(
"-c",
"--concurrent", default=0, type=int,
25 help=
"The number of concurrent events to run. 0 uses the same as number of threads.")
26 parser.add_argument(
"-V",
"--verboseAccumulators", default=
False, action=
"store_true",
27 help=
"Print full details of the AlgSequence for each accumulator")
28 parser.add_argument(
"-S",
"--verboseStoreGate", default=
False, action=
"store_true",
29 help=
"Dump the StoreGate(s) each event iteration")
30 parser.add_argument(
"-o",
"--output", default=
'', type=str,
31 help=
"Output RDO file")
32 parser.add_argument(
"-s",
"--outputSig", default=
'', type=str,
33 help=
"Output RDO_SGNL file")
34 parser.add_argument(
"-r",
"--run", default=LHCPeriod.Run2,
35 type=LHCPeriod, choices=
list(LHCPeriod))
36 parser.add_argument(
"--disableTruth", default=
False, action=
"store_true",
37 help=
"Disable truth overlay")
38 parser.add_argument(
"--debug", default=
'', type=str,
39 choices=DbgStage.allowed_values,
40 help=
"Debugging flag: " +
','.join (DbgStage.allowed_values))
45 """Fill default overlay flags for testing"""
47 flags.Digitization.EnableTruth =
False
49 from AthenaConfiguration.TestDefaults
import defaultTestFiles, defaultConditionsTags
50 from AthenaConfiguration.Enums
import ProductionStep
51 flags.Common.ProductionStep = ProductionStep.Overlay
53 flags.Overlay.DataOverlay =
True
54 flags.Overlay.DataOverlayConditions =
"OverlayConfiguration.DataOverlayConditions.DataOverlay2023Cfg"
55 flags.Input.isMC =
False
56 flags.Input.Files = defaultTestFiles.RDO_BKG_HI_RUN3_2023
57 flags.Input.SecondaryFiles = defaultTestFiles.HITS_DATA_OVERLAY_HI_RUN3_2023
58 flags.Output.RDOFileName =
"dataOverlayRDO.pool.root"
59 flags.IOVDb.GlobalTag = defaultConditionsTags.RUN3_DATA23
60 flags.IOVDb.DatabaseInstance =
"CONDBR2"
61 flags.Overlay.ByteStream =
False
62 from Campaigns
import DataOverlay2023
65 if args.run
is LHCPeriod.Run2:
66 flags.Input.Files = defaultTestFiles.RDO_BKG_RUN2
67 flags.Input.SecondaryFiles = defaultTestFiles.HITS_RUN2
68 flags.IOVDb.GlobalTag = defaultConditionsTags.RUN2_MC
69 from Campaigns
import MC20e
71 elif args.run
is LHCPeriod.Run3:
72 flags.Input.Files = defaultTestFiles.RDO_BKG_RUN3_2022
73 flags.Input.SecondaryFiles = defaultTestFiles.HITS_RUN3_2022
74 flags.IOVDb.GlobalTag = defaultConditionsTags.RUN3_MC
75 from Campaigns
import MC23a
77 elif args.run
is LHCPeriod.Run4:
78 flags.Input.Files = defaultTestFiles.RDO_BKG_RUN4
79 flags.Input.SecondaryFiles = defaultTestFiles.HITS_RUN4
80 flags.IOVDb.GlobalTag = defaultConditionsTags.RUN4_MC
81 from Campaigns
import PhaseIIPileUp200
84 raise ValueError(
"Run not supported")
85 flags.Input.MCChannelNumber =
GetFileMD(flags.Input.SecondaryFiles, allowEmpty=
False).
get(
"mc_channel_number", 0)
86 flags.Output.RDOFileName =
"mcOverlayRDO.pool.root"
87 flags.Overlay.DataOverlay =
False
90 if args.output ==
'None':
91 flags.Output.RDOFileName =
''
93 flags.Output.RDOFileName = args.output
96 flags.Output.RDO_SGNLFileName = args.outputSig
98 if 'detectors' in args
and args.detectors:
99 detectors = args.detectors
103 from AthenaConfiguration.DetectorConfigFlags
import setupDetectorFlags
108 from OverlayConfiguration.OverlayMetadata
import overlayMetadataCheck
114 """Postprocess and lock config flags for overlay"""
115 flags.Concurrency.NumThreads = args.threads
116 flags.Concurrency.NumProcs = args.processes
118 flags.Scheduler.ShowDataDeps =
True
119 flags.Scheduler.ShowDataFlow =
True
120 flags.Scheduler.ShowControlFlow =
True
121 flags.Concurrency.NumConcurrentEvents = args.concurrent
if args.concurrent > 0
else args.threads
127 """Common debugging and execution for overlay tests"""
129 accessor.printConfig(withDetails=args.verboseAccumulators)
130 if args.verboseStoreGate:
131 accessor.getService(
"StoreGateSvc").Dump =
True
135 accessor.setDebugStage (args.debug)
138 sc = accessor.run(maxEvents=args.maxEvents)
141 return not sc.isSuccess()