ATLAS Offline Software
Loading...
Searching...
No Matches
python.OverlayTestHelpers Namespace Reference

Functions

 CommonTestArgumentParser (prog)
 overlayTestFlags (flags, args)
 postprocessAndLockFlags (flags, args)
 printAndRun (accessor, flags, args)

Detailed Description

Overlay test helpers

Copyright (C) 2002-2025 CERN for the benefit of the ATLAS collaboration

Function Documentation

◆ CommonTestArgumentParser()

python.OverlayTestHelpers.CommonTestArgumentParser ( prog)
Common overlay test argument parser

Definition at line 13 of file OverlayTestHelpers.py.

13def CommonTestArgumentParser(prog):
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))
41 return parser
42
43

◆ overlayTestFlags()

python.OverlayTestHelpers.overlayTestFlags ( flags,
args )
Fill default overlay flags for testing

Definition at line 44 of file OverlayTestHelpers.py.

44def overlayTestFlags(flags, args):
45 """Fill default overlay flags for testing"""
46 if args.disableTruth:
47 flags.Digitization.EnableTruth = False
48
49 from AthenaConfiguration.TestDefaults import defaultTestFiles, defaultConditionsTags
50 from AthenaConfiguration.Enums import ProductionStep
51 flags.Common.ProductionStep = ProductionStep.Overlay
52 if args.data:
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
63 DataOverlay2023(flags)
64 else:
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
70 MC20e(flags)
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
76 MC23a(flags)
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
82 PhaseIIPileUp200(flags)
83 else:
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
88
89 if args.output:
90 if args.output == 'None':
91 flags.Output.RDOFileName = ''
92 else:
93 flags.Output.RDOFileName = args.output
94
95 if args.outputSig:
96 flags.Output.RDO_SGNLFileName = args.outputSig
97
98 if 'detectors' in args and args.detectors:
99 detectors = args.detectors
100 else:
101 detectors = None
102
103 from AthenaConfiguration.DetectorConfigFlags import setupDetectorFlags
104 setupDetectorFlags(flags, detectors, toggle_geometry=True, use_metadata=True)
105
106 # Moving here so that it is ahead of flags being locked. Need to
107 # iterate on exact best position w.r.t. above calls
108 from OverlayConfiguration.OverlayMetadata import overlayMetadataCheck
109 # Handle metadata correctly
110 overlayMetadataCheck(flags)
111
112
T * get(TKey *tobj)
get a TObject* from a TKey* (why can't a TObject be a TKey?)
Definition hcg.cxx:130

◆ postprocessAndLockFlags()

python.OverlayTestHelpers.postprocessAndLockFlags ( flags,
args )
Postprocess and lock config flags for overlay

Definition at line 113 of file OverlayTestHelpers.py.

113def postprocessAndLockFlags(flags, args):
114 """Postprocess and lock config flags for overlay"""
115 flags.Concurrency.NumThreads = args.threads
116 flags.Concurrency.NumProcs = args.processes
117 if args.threads > 0:
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
122
123 flags.lock()
124
125

◆ printAndRun()

python.OverlayTestHelpers.printAndRun ( accessor,
flags,
args )
Common debugging and execution for overlay tests

Definition at line 126 of file OverlayTestHelpers.py.

126def printAndRun(accessor, flags, args):
127 """Common debugging and execution for overlay tests"""
128 # Dump config
129 accessor.printConfig(withDetails=args.verboseAccumulators)
130 if args.verboseStoreGate:
131 accessor.getService("StoreGateSvc").Dump = True
132 flags.dump()
133
134 if args.debug:
135 accessor.setDebugStage (args.debug)
136
137 # Execute and finish
138 sc = accessor.run(maxEvents=args.maxEvents)
139
140 # Success should be 0
141 return not sc.isSuccess()