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-2026 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 from Campaigns import DataOverlay2023
62 DataOverlay2023(flags)
63 else:
64 if args.run is LHCPeriod.Run2:
65 flags.Input.Files = defaultTestFiles.RDO_BKG_RUN2
66 flags.Input.SecondaryFiles = defaultTestFiles.HITS_RUN2
67 flags.IOVDb.GlobalTag = defaultConditionsTags.RUN2_MC
68 from Campaigns import MC20e
69 MC20e(flags)
70 elif args.run is LHCPeriod.Run3:
71 flags.Input.Files = defaultTestFiles.RDO_BKG_RUN3_2022
72 flags.Input.SecondaryFiles = defaultTestFiles.HITS_RUN3_2022
73 flags.IOVDb.GlobalTag = defaultConditionsTags.RUN3_MC
74 from Campaigns import MC23a
75 MC23a(flags)
76 elif args.run is LHCPeriod.Run4:
77 flags.Input.Files = defaultTestFiles.RDO_BKG_RUN4
78 flags.Input.SecondaryFiles = defaultTestFiles.HITS_RUN4
79 flags.IOVDb.GlobalTag = defaultConditionsTags.RUN4_MC
80 from Campaigns import PhaseIIPileUp200
81 PhaseIIPileUp200(flags)
82 else:
83 raise ValueError("Run not supported")
84 flags.Input.MCChannelNumber = GetFileMD(flags.Input.SecondaryFiles, allowEmpty=False).get("mc_channel_number", 0)
85 flags.Output.RDOFileName = "mcOverlayRDO.pool.root"
86 flags.Overlay.DataOverlay = False
87
88 if args.output:
89 if args.output == 'None':
90 flags.Output.RDOFileName = ''
91 else:
92 flags.Output.RDOFileName = args.output
93
94 if args.outputSig:
95 flags.Output.RDO_SGNLFileName = args.outputSig
96
97 if 'detectors' in args and args.detectors:
98 detectors = args.detectors
99 else:
100 detectors = None
101
102 from AthenaConfiguration.DetectorConfigFlags import setupDetectorFlags
103 setupDetectorFlags(flags, detectors, toggle_geometry=True, use_metadata=True)
104
105 # Moving here so that it is ahead of flags being locked. Need to
106 # iterate on exact best position w.r.t. above calls
107 from OverlayConfiguration.OverlayMetadata import overlayMetadataCheck
108 # Handle metadata correctly
109 overlayMetadataCheck(flags)
110
111
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 112 of file OverlayTestHelpers.py.

112def postprocessAndLockFlags(flags, args):
113 """Postprocess and lock config flags for overlay"""
114 flags.Concurrency.NumThreads = args.threads
115 flags.Concurrency.NumProcs = args.processes
116 if args.threads > 0:
117 flags.Scheduler.ShowDataDeps = True
118 flags.Scheduler.ShowDataFlow = True
119 flags.Scheduler.ShowControlFlow = True
120 flags.Concurrency.NumConcurrentEvents = args.concurrent if args.concurrent > 0 else args.threads
121
122 flags.lock()
123
124

◆ printAndRun()

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

Definition at line 125 of file OverlayTestHelpers.py.

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