ATLAS Offline Software
OverlayTestHelpers.py
Go to the documentation of this file.
1 #!/usr/bin/env python
2 """Overlay test helpers
3 
4 Copyright (C) 2002-2025 CERN for the benefit of the ATLAS collaboration
5 """
6 
7 from argparse import ArgumentParser
8 from AthenaCommon.Debugging import DbgStage
9 from AthenaConfiguration.AutoConfigFlags import GetFileMD
10 from AthenaConfiguration.Enums import LHCPeriod
11 
12 
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 
44 def 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 
113 def 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 
126 def 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()
python.AutoConfigFlags.GetFileMD
def GetFileMD(filenames, allowEmpty=True, maxLevel='peeker')
Definition: AutoConfigFlags.py:65
python.DataOverlayRun3.DataOverlay2023
def DataOverlay2023(flags)
Definition: DataOverlayRun3.py:15
python.MC23.MC23a
def MC23a(flags)
Definition: MC23.py:22
python.OverlayTestHelpers.postprocessAndLockFlags
def postprocessAndLockFlags(flags, args)
Definition: OverlayTestHelpers.py:113
python.OverlayMetadata.overlayMetadataCheck
def overlayMetadataCheck(flags)
Definition: OverlayMetadata.py:176
python.OverlayTestHelpers.printAndRun
def printAndRun(accessor, flags, args)
Definition: OverlayTestHelpers.py:126
histSizes.list
def list(name, path='/')
Definition: histSizes.py:38
python.DetectorConfigFlags.setupDetectorFlags
def setupDetectorFlags(flags, custom_list=None, use_metadata=False, toggle_geometry=False, validate_only=False, keep_beampipe=False)
Definition: DetectorConfigFlags.py:292
python.MC20.MC20e
def MC20e(flags)
Definition: MC20.py:63
python.PhaseII.PhaseIIPileUp200
def PhaseIIPileUp200(flags)
Definition: PhaseII.py:81
python.OverlayTestHelpers.overlayTestFlags
def overlayTestFlags(flags, args)
Definition: OverlayTestHelpers.py:44
python.OverlayTestHelpers.CommonTestArgumentParser
def CommonTestArgumentParser(prog)
Definition: OverlayTestHelpers.py:13
get
T * get(TKey *tobj)
get a TObject* from a TKey* (why can't a TObject be a TKey?)
Definition: hcg.cxx:127