ATLAS Offline Software
JetCalibStepsConfig.py
Go to the documentation of this file.
1 # Copyright (C) 2002-2025 CERN for the benefit of the ATLAS collaboration
2 from AthenaCommon import Logging
3 jcslog = Logging.logging.getLogger('JetCalibStepsConfig')
4 
5 from JetToolHelpers.HelperConfig import VarToolCfg, HistoInputCfg
6 from AthenaConfiguration.ComponentFactory import CompFactory
7 from PathResolver import PathResolver
8 
9 def smearingStep(flags, **configDict):
10  """ Configuration of the Smearing step. """
11 
12  # HistoReaderMC and HistoReaderData can be specified simultaneously using a single "HistoReader" YAML block.
13  # This should contain the common configuration for both and histNameMC/histNameData entries for the histogram names.
14  # This cannot be used simultaneously with a HistoReaderMC or HistoReaderData block
15  if "HistoReader" in configDict:
16  if "HistoReaderMC" in configDict:
17  raise JetCalibConfigError("Both HistoReader and HistoReaderMC blocks included in YAML config")
18  if "HistoReaderData" in configDict:
19  raise JetCalibConfigError("Both HistoReader and HistoReaderData blocks included in YAML config")
20 
21  # Build the HistoReaderMC and HistoReaderData blocks from the HistoReader block
22  histoReader = configDict.pop("HistoReader")
23  histNameMC = histoReader.pop("histNameMC")
24  histNameData = histoReader.pop("histNameData")
25  configDict["HistoReaderMC"] = dict(histoReader)
26  configDict["HistoReaderData"] = dict(histoReader)
27  configDict["HistoReaderMC"]["histName"] = histNameMC
28  configDict["HistoReaderData"]["histName"] = histNameData
29 
30  configDict["HistoReaderMC"]["inputFile"] = PathResolver.FindCalibFile(configDict["HistoReaderMC"]["inputFile"])
31  configDict["HistoReaderData"]["inputFile"] = PathResolver.FindCalibFile(configDict["HistoReaderData"]["inputFile"])
32 
33  histToolMC = HistoInputCfg(flags, "HistToolMC", **configDict["HistoReaderMC"])
34  histToolData = HistoInputCfg(flags, "HistToolData", **configDict["HistoReaderData"])
35  configDict["HistoReaderMC"] = histToolMC
36  configDict["HistoReaderData"] = histToolData
37 
38  smearStep = CompFactory.SmearingCalibStep("SmearingCalibStep", **configDict)
39 
40  return smearStep
41 
42 def puresidualStep(flags, **configDict):
43 
44  configDict.setdefault('IsData', not flags.Input.isMC)
45 
46  return CompFactory.Pileup1DResidualCalibStep("PUResid", **configDict)
47 
48 
49 def gscStep(flags, **configDict):
50 
51  defaultFileGSC = PathResolver.FindCalibFile(configDict.pop('fileGSC'))
52 
53  # These HistoInput2D defaults can't be set in the C++, so are set here:
54  defaultHistTools = dict(
55  histTool_EM3 = [dict(varX = "pt", varY = "EM3", histName=f"AntiKt4EMPFlow_EM3_interpolation_resp_eta_{j}", inputFile=defaultFileGSC) for j in range(35)],
56  histTool_CharFrac = [dict(varX = "pt", varY = "ChargedFraction", histName = f"AntiKt4EMPFlow_chargedFraction_interpolation_resp_eta_{j}", inputFile=defaultFileGSC) for j in range(25)],
57  histTool_Tile0 = [dict(varX = "pt", varY = "Tile0", histName=f"AntiKt4EMPFlow_Tile0_interpolation_resp_eta_{j}", inputFile=defaultFileGSC) for j in range(18)],
58  histTool_nTrk=[dict(varX = "pt", varY = dict(Name="nTrk", Type="int",), histName=f"AntiKt4EMPFlow_nTrk_interpolation_resp_eta_{j}", inputFile=defaultFileGSC) for j in range(25)],
59  histTool_trackWIDTH=[dict(varX = "pt", varY = "trackWIDTH", histName=f"AntiKt4EMPFlow_trackWIDTH_interpolation_resp_eta_{j}", inputFile=defaultFileGSC) for j in range(25)],
60  )
61 
62  # Build the hist tools
63  for key in ['histTool_EM3', 'histTool_CharFrac', 'histTool_Tile0', 'histTool_nTrk', 'histTool_trackWIDTH']:
64 
65  # Use defaultHistTools by default
66  if key not in configDict:
67  toolArray = defaultHistTools[key]
68 
69  # In this case the full list of sub-tools has been specified in the YAML
70  # and the defaultDict will be fully overwritten (ie. all parameters must be specified)
71  elif isinstance(configDict[key],list):
72  toolArray = configDict[key]
73  for subDict in toolArray:
74  subDict.setdefault('inputFile',defaultFileGSC)
75 
76  # Functionality to build the arrays of histogram reader from a shorter block in the YAML file
77  # Fall back on defaultHistTools for defaults
78  else:
79  baseDict = dict(configDict[key])
80  N_hist = baseDict.pop('N_hist')
81  histNameBase = baseDict.pop('histNameBase')
82  inputFile = baseDict.pop('inputFile', defaultFileGSC)
83  varX = baseDict.pop('varX',defaultHistTools[key][0]['varX'])
84  varY = baseDict.pop('varY', defaultHistTools[key][0]['varY'])
85  toolArray = [dict(varX = varX, varY = varY, histName=f'{histNameBase}_{j}', inputFile = inputFile) for j in range(N_hist)]
86 
87  # Convert array of properties to HistoInput tools
88  configDict[key] = [HistoInputCfg(flags, Tname=f"{key.split('_')[1]}_{j}", **toolConfig) for j, toolConfig in enumerate(toolArray)]
89 
90  GSCstep = CompFactory.GSCCalibStep("gsccalibstep", **configDict)
91 
92  return GSCstep
93 
94 def etajesStep(flags, **configDic):
95 
96  pVars = configDic.pop("ParametrizedVars")
97 
98  jesstep = CompFactory.EtaMassJESCalibStep("EtaMassJESCalib",
99  VarToolE= VarToolCfg(flags, var=pVars['varE']),
100  VarToolEta= VarToolCfg(flags, var=pVars["varEta"]),
101  **configDic
102  )
103  return jesstep
104 
105 def insituStep(flags, **configDic):
106 
107  histEtaInterCalib = configDic.pop('histEtaInterCalib')
108  histAbsCalib = configDic.pop('histAbsCalib')
109 
110  histoReaderEta_vec, histoReaderAbs_vec = [], []
111 
112  for infile in configDic.pop('fileInsitu'):
113  histoReaderEta_vec.append(dict(inputFile = PathResolver.FindCalibFile(infile), **histEtaInterCalib))
114  histoReaderAbs_vec.append(dict(inputFile = PathResolver.FindCalibFile(infile), **histAbsCalib))
115  configDic['HistoReaderEtaInter'] = [HistoInputCfg(flags, "HistToolEtaInter"+str(j), **etaDic) for j, etaDic in enumerate(histoReaderEta_vec)]
116  configDic['HistoReaderAbs'] = [HistoInputCfg(flags, "HistToolAbs"+str(j), **absDic) for j, absDic in enumerate(histoReaderAbs_vec)]
117 
118  configDic['vartool1'] = VarToolCfg(flags, var=histEtaInterCalib['varX'], Tname="VarTool")
119  configDic['vartool2'] = VarToolCfg(flags, var=histEtaInterCalib['varY'], Tname="VarTool")
120  configDic['isMC'] = flags.Input.isMC
121 
122  insituStep = CompFactory.InSituCalibStep("insitucalibstep", **configDic)
123 
124  return insituStep
125 
126 
127 
128 calibStepDic = dict(
129  JetArea = None,
130  Residual = puresidualStep,
131  EtaJES = etajesStep,
132  GSC = gscStep,
133  Insitu = insituStep,
134  Smear = smearingStep,
135 )
136 
137 def calibConfigToToolList(flags, **configDict):
138  """
139  Returns a list of instantiated tools for each of the calibration steps.
140  The order of the steps is determined by the InScale and OutScale properties given in the config.
141  Tools are instantiated by calling functions declared in the calibStepDic dictionary.
142  """
143 
144  toolDic = {}
145  foundCS = False # check at least one of the steps starts from constituent scale
146  for step in configDict:
147 
148  configDict.get(step).pop('prereqs',{}) # removes the 'prereqs' entry not refined in steps
149  # expert option to skip a step
150  if configDict.get(step).pop('noRun',False):
151  jcslog.warning(f'Expert option: Skipping calib step {step}')
152  continue
153 
154  # Skip Insitu for MC
155  if step=="Insitu" and flags.Input.isMC and not configDict.get("Insitu").get("CalibrateMC",False):
156  jcslog.info('Skipping Insitu for MC')
157  continue
158 
159  calibFunc = calibStepDic.get(step,None)
160  if calibFunc is None:
161  raise NotImplementedError(f'Calibration step {step} is not found in calibStepDic')
162 
163  calibConfig = configDict.get(step)
164 
165  tool = calibFunc(flags, **calibConfig)
166 
167  toolDic[step] = tool
168  if tool.InScale == "JetConstitScaleMomentum":
169  foundCS = True
170 
171  if not foundCS:
172  raise JetCalibConfigError('At least one step must have InScale = JetConstitScaleMomentum')
173 
174  ordered_tools = []
175  ordered_step_names = []
176 
177  def findNextSteps(startScale = "JetConstitScaleMomentum"):
178  ''' Recursively add tools to ordered_tools based on in/out scale '''
179  for step in toolDic:
180  if toolDic[step].InScale == startScale:
181  ordered_tools.append(toolDic[step])
182  ordered_step_names.append(step)
183  findNextSteps(toolDic[step].OutScale)
184 
185  findNextSteps()
186 
187  # Check we've got all the steps
188  for step in toolDic:
189  if step not in ordered_step_names:
190  raise JetCalibConfigError(f'Could not place calib step {step} - have you set InScale and OutScale correctly?')
191 
192  jcslog.info(f'Ordered jet calib steps: {"->".join(ordered_step_names)}')
193 
194  return ordered_tools
195 
196 def calibToolFromConfigFile(flags, configFile, name = "jetcalib"):
197 
198  configDic = load_yaml_cfg(configFile)
199 
200  globalSettings = configDic.pop('Global',{})
201 
202  calibTool = CompFactory.JetCalibTool(name, CalibSteps=calibConfigToToolList(flags, **configDic), **globalSettings)
203  return calibTool
204 
205 def load_yaml_cfg(configFile):
206  from yaml import safe_load
207 
208  path_configFile = PathResolver.FindCalibFile(configFile)
209  configDic = safe_load(open(path_configFile))
210  return configDic
211 
212 
213 class JetCalibConfigError(Exception):
214  """ Exception raised for invalid jet calibration config """
215 
216  def __init__(self, message):
217  super().__init__(message)
218  self.message = message
219 
220  def __str__(self):
221  return self.message
PathResolver::FindCalibFile
static std::string FindCalibFile(const std::string &logical_file_name)
Definition: PathResolver.h:63
JetCalibStepsConfig.JetCalibConfigError.message
message
Definition: JetCalibStepsConfig.py:218
JetCalibStepsConfig.gscStep
def gscStep(flags, **configDict)
Definition: JetCalibStepsConfig.py:49
JetCalibStepsConfig.puresidualStep
def puresidualStep(flags, **configDict)
Definition: JetCalibStepsConfig.py:42
JetCalibStepsConfig.calibToolFromConfigFile
def calibToolFromConfigFile(flags, configFile, name="jetcalib")
Definition: JetCalibStepsConfig.py:196
JetCalibStepsConfig.calibConfigToToolList
def calibConfigToToolList(flags, **configDict)
Definition: JetCalibStepsConfig.py:137
JetCalibStepsConfig.insituStep
def insituStep(flags, **configDic)
Definition: JetCalibStepsConfig.py:105
JetCalibStepsConfig.JetCalibConfigError.__init__
def __init__(self, message)
Definition: JetCalibStepsConfig.py:216
JetCalibStepsConfig.JetCalibConfigError
Definition: JetCalibStepsConfig.py:213
JetCalibStepsConfig.JetCalibConfigError.__str__
def __str__(self)
Definition: JetCalibStepsConfig.py:220
plotBeamSpotVxVal.range
range
Definition: plotBeamSpotVxVal.py:194
HelperConfig.HistoInputCfg
def HistoInputCfg(flags, Tname, inputFile, histName, varX, **kwargs)
Definition: HelperConfig.py:31
HelperConfig.VarToolCfg
def VarToolCfg(flags, var, Tname="VarTool", **kwargs)
Definition: HelperConfig.py:4
Trk::open
@ open
Definition: BinningType.h:40
get
T * get(TKey *tobj)
get a TObject* from a TKey* (why can't a TObject be a TKey?)
Definition: hcg.cxx:127
JetCalibStepsConfig.etajesStep
def etajesStep(flags, **configDic)
Definition: JetCalibStepsConfig.py:94
str
Definition: BTagTrackIpAccessor.cxx:11
JetCalibStepsConfig.smearingStep
def smearingStep(flags, **configDict)
Definition: JetCalibStepsConfig.py:9
JetCalibStepsConfig.load_yaml_cfg
def load_yaml_cfg(configFile)
Definition: JetCalibStepsConfig.py:205