ATLAS Offline Software
ConfigHelpers.py
Go to the documentation of this file.
1 # Copyright (C) 2002-2024 CERN for the benefit of the ATLAS collaboration
2 
3 """ Helper functions for configuring MET chains"""
4 
5 from __future__ import annotations
6 from typing import Any, Optional
7 
8 from AthenaConfiguration.ComponentAccumulator import ComponentAccumulator
9 from AthenaConfiguration.ComponentFactory import CompFactory
10 from ..Menu.SignatureDicts import METChainParts_Default, METChainParts
11 from ..Config.MenuComponents import (
12  ChainStep,
13  InEventRecoCA,
14  SelectionCA,
15  MenuSequence,
16 )
17 from .StepOutput import StepOutput
18 from copy import copy
19 from ..CommonSequences.FullScanDefs import trkFSRoI
20 from AthenaCommon.Logging import logging
21 from TrigEFMissingET.TrigEFMissingETConfig import getMETMonTool
22 from abc import ABC, abstractmethod
23 from string import ascii_uppercase
24 from TrigMissingETHypo.TrigMissingETHypoConfig import TrigMETHypoToolFromDict
25 import functools
26 
27 def streamer_hypo_tool(flags, chainDict):
28  return CompFactory.TrigStreamerHypoTool(chainDict["chainName"])
29 
30 
31 log = logging.getLogger(__name__)
32 
33 # The keys from the MET chain dict that directly affect reconstruction
34 # The order here is important as it also controls the dict -> string conversion
35 
36 recoKeys = [
37  "EFrecoAlg",
38  "calib",
39  "constitType",
40  "constitmod",
41  "jetCalib",
42  "nSigma",
43  "addInfo",
44 ]
45 metFSRoIs: list[str | None] = ["", None, trkFSRoI]
46 
47 
48 def metRecoDictToString(recoDict: dict[str, str], skipDefaults: bool = True) -> str:
49  """Convert a dictionary containing reconstruction keys to a string
50 
51  Any key (from recoKeys) missing will just be skipped.
52 
53  Parameters
54  ----------
55  recoDict : dict[str, str]
56  The reconstruction dictionary
57  skipDefaults : bool, optional
58  If true, skip any values that match the default, by default True
59 
60  Returns
61  -------
62  str
63  The fixed string representation
64  """
65  return "_".join(
66  recoDict[k]
67  for k in recoKeys
68  if k in recoDict
69  and (not skipDefaults or recoDict[k] != METChainParts_Default[k])
70  )
71 
72 
73 def stringToMETRecoDict(string: str, onlyRecoKeys: bool = False) -> dict[str, Any]:
74  """Convert a string to a MET reco dict. Optionally only keep entries contained in recoKeys.
75  Necessary for correctly configuring algorithms as inputs to NN.
76  """
77  defaults = copy(METChainParts_Default)
78  if onlyRecoKeys:
79  defaults = {key: defaults[key] for key in recoKeys}
80  # Now go through the parts of the string and fill in the dict
81  for part in string.split("_"):
82  for key, values in METChainParts.items():
83  if not isinstance(values, list):
84  continue
85  if part in values:
86  if isinstance(defaults[key], list):
87  defaults[key].append(part) # type: ignore[attr-defined]
88  else:
89  defaults[key] = part
90  return defaults
91 
92 
93 class AlgConfig(ABC):
94  """Base class to describe algorithm configurations
95 
96  Each individual 'EFrecoAlg' should be described by *one* AlgConfig subclass.
97  It must provide its list of required inputs to the constructor and override
98  the make_fex method
99 
100  The name of fexAlg *must* be self.fexName and the METContainerKey property
101  *must* be set to self.outputKey (but this class usually should take care of
102  this).
103 
104  The subclass must also implement the @classmethod 'algType' which returns
105  the EFrecoAlg string that it describes.
106  """
107 
108  @classmethod
109  def algType(cls) -> str:
110  """The algorithm that this object configures - this corresponds to the
111  EFrecoAlg in the METChainParts dictionary
112 
113  Note that no two subclasses of AlgConfig can describe the same algorithm
114  (as identified by this string).
115  """
116  raise NotImplementedError("algType not implemented by subclass!")
117 
118  def __init__(self, **recoDict: str):
119  """Initialise the base class
120 
121  =========
122  Arguments
123  =========
124  recoDict: Pass *all* the keys required for the recoDict
125  """
126 
127  # Make sure that we got *all* the keys (i.e. the subclass didn't
128  # inadvertently steal one of them from us)
129  alg_type = self.algType()
130  assert all(k in recoDict for k in recoKeys), (
131  f"AlgConfig.__init__ for {alg_type} did not receive all the recoKeys"
132  " - this suggests a problem in the subclass __init__ method!"
133  )
134  self.recoDict = copy(recoDict)
135  self._suffix = metRecoDictToString(recoDict)
136 
137  @property
138  def outputKey(self) -> str:
139  """The MET container object produced by this algorithm"""
140  from TrigEDMConfig.TriggerEDM import recordable
141 
142  return recordable("HLT_MET_{}".format(self._suffix))
143 
144  @property
145  def fexName(self) -> str:
146  """The name of the algorithm made by this configuration"""
147  return "EFMET_{}".format(self._suffix)
148 
149  def interpret_reco_dict(self) -> dict[str, Any]:
150  """Return a version of the reco dict with any necessary post-processing"""
151  return self.recoDict
152 
153  def getMonTool(self, flags):
154  """Create the monitoring tool"""
155  return getMETMonTool(flags)
156 
157  def name_step(self, idx) -> str:
158  return f"step{ascii_uppercase[idx]}_{self._suffix}"
159 
160  @abstractmethod
161  def make_reco_algs(self, flags, **recoDict) -> StepOutput:
162  """Create the reconstruction sequences including the FEX
163 
164  Returns a list of CAs split by step
165  """
166 
167  def _append_fex(self, flags, fex, inputs: Optional[StepOutput] = None) -> None:
168  """Append the FEX to the output object
169 
170  Finalizes the FEX by setting the monitoring tool and output name.
171  """
172  fex.MonTool = self.getMonTool(flags)
173  fex.METContainerKey = self.outputKey
174  acc = ComponentAccumulator()
175  acc.addEventAlgo(fex, primary=True)
176  return StepOutput.create(acc, inputs, MET=self.outputKey)
177 
178  def make_reco_ca(self, flags):
179  """Make the reconstruction sequences for the new JO style"""
180  # Retrieve the inputs
181  log.verbose("Create inputs for %s", self._suffix)
182  steps, inputs = self.inputRegistry.build_steps(
183  flags, self._inputs, metFSRoIs, self.recoDict, return_ca=True
184  )
185  # Create the FEX and add it to the last input sequence
186  fex = self.make_fex_accumulator(flags, self.fexName, inputs)
187  fex.MonTool = self.getMonTool(flags)
188  fex.METContainerKey = self.outputKey
189  steps[-1].addEventAlgo(fex)
190  return steps
191 
192  def make_accumulator_steps(self, flags, chainDict):
193  """Make the full accumulator steps"""
194 
195  # Get the reco sequences
196  reco_sequences = self.make_reco_algs(flags, **self.interpret_reco_dict()).steps
197  output_steps = []
198  # build up the output steps
199  # We have to merge together the CAs containing the reconstruction sequences
200  # and the InEventRecoCAs that contain the input makers
201  for step_idx, reco_sequence in enumerate(reco_sequences):
202  reco_acc = self.inputMakerCA(step_idx)
203  sel_acc = None
204  step_name = f"Empty_METStep{step_idx}"
205  if (
206  reco_acc is not None
207  ): # Define reco and hypo algorithms only if reco_acc is not None
208  if step_idx == len(reco_sequences) - 1:
209  # If this is the last step we have to add the hypo alg
210  hypo_alg = self.make_hypo_alg()
211  hypo_tool = TrigMETHypoToolFromDict
212  else:
213  # Otherwise, we add a passthrough hypo
214  hypo_alg = self.make_passthrough_hypo_alg(step_idx)
215  hypo_tool = streamer_hypo_tool
216 
217  # Now merge the reco sequence into the InEventRecoCA
218  reco_acc.mergeReco(reco_sequence)
219 
220  # Create a selection CA
221  sel_acc = SelectionCA("METAthSeq_" + self.name_step(step_idx))
222  # Merge in the reconstruction sequence
223  sel_acc.mergeReco(reco_acc)
224  # Add its hypo alg
225  sel_acc.addHypoAlgo(hypo_alg)
226 
227  # FIXME: We need to mark the CA as merged because during the fast menu generation
228  # the merging is only done once for all chains, which is not compatible with the MET
229  # menu generation code. Remove this once ATR-29211 is fixed.
230  sel_acc.wasMerged()
231 
232  step_name = sel_acc.name
233 
234  # Build the menu sequence and create the actual chainStep
235  output_steps.append(
236  ChainStep(
237  name=step_name,
238  chainDicts=[chainDict],
239  SequenceGens=[]
240  if sel_acc is None
241  else [
242  functools.partial(make_MET_menu_sequenceGenCfg, flags, sel_acc, hypo_tool)
243  ],
244  isEmpty=True if sel_acc is None else False
245  )
246  )
247 
248  return output_steps
249 
250 
251  def make_hypo_alg(self):
252  """The hypo alg used for this configuration"""
253  return CompFactory.TrigMissingETHypoAlg(
254  f"METHypoAlg_{self._suffix}", METContainerKey=self.outputKey
255  )
256 
257  def make_passthrough_hypo_alg(self, step):
258  return CompFactory.TrigStreamerHypoAlg(
259  "METPassThroughHypo_" + self.name_step(step)
260  )
261 
262  def inputMakerCA(self, idx):
263  """Get the InEventRecoCA for the given step index"""
264  from TrigT2CaloCommon.CaloDef import clusterFSInputMaker
265  from AthenaConfiguration.ComponentFactory import CompFactory
266  from ..CommonSequences.FullScanDefs import trkFSRoI
267 
268  name = self.name_step(idx) + "Reco"
269  if idx == 0:
270  return InEventRecoCA(name, inputMaker=clusterFSInputMaker())
271  elif idx == 1:
272  return None
273  elif idx == 2:
274  return InEventRecoCA(
275  name,
276  inputMaker=CompFactory.InputMakerForRoI(
277  "IM_Jet_TrackingStep",
278  RoITool=CompFactory.ViewCreatorInitialROITool(),
279  RoIs=trkFSRoI,
280  ),
281  )
282  else:
283  raise KeyError(f"No input maker for step {idx}")
284 
285  @classmethod
286  def _get_subclasses(cls):
287  """Provides a way to iterate over all subclasses of this class"""
288  for subcls in cls.__subclasses__():
289  for subsubcls in subcls.__subclasses__():
290  yield subsubcls
291  yield subcls
292 
293  @classmethod
294  def fromRecoDict(cls, EFrecoAlg, **recoDict) -> AlgConfig:
295  for subcls in cls._get_subclasses():
296  if subcls.algType() == EFrecoAlg:
297  return subcls(EFrecoAlg=EFrecoAlg, **recoDict)
298 
299  raise ValueError("Unknown EFrecoAlg '{}' requested".format(EFrecoAlg))
300 
301 def make_MET_menu_sequenceGenCfg(flags, sel_acc, hypo_tool):
302  return MenuSequence(flags, selectionCA=sel_acc, HypoToolGen=hypo_tool)
303 
304 # Load all the defined configurations
305 from . import AlgConfigs
306 
307 # Make sure that there is an AlgConfig for every EFrecoAlg
308 AlgConfigs.test_configs()
python.HLT.MET.ConfigHelpers.AlgConfig._append_fex
None _append_fex(self, flags, fex, Optional[StepOutput] inputs=None)
Definition: ConfigHelpers.py:167
python.HLT.MET.ConfigHelpers.streamer_hypo_tool
def streamer_hypo_tool(flags, chainDict)
Definition: ConfigHelpers.py:27
python.JetAnalysisCommon.ComponentAccumulator
ComponentAccumulator
Definition: JetAnalysisCommon.py:302
vtune_athena.format
format
Definition: vtune_athena.py:14
python.HLT.MET.ConfigHelpers.AlgConfig.make_hypo_alg
def make_hypo_alg(self)
Definition: ConfigHelpers.py:251
python.HLT.MET.ConfigHelpers.AlgConfig.make_reco_ca
def make_reco_ca(self, flags)
Definition: ConfigHelpers.py:178
python.HLT.MET.ConfigHelpers.AlgConfig.make_reco_algs
StepOutput make_reco_algs(self, flags, **recoDict)
Definition: ConfigHelpers.py:161
python.HLT.MET.ConfigHelpers.metRecoDictToString
str metRecoDictToString(dict[str, str] recoDict, bool skipDefaults=True)
Definition: ConfigHelpers.py:48
python.HLT.MET.ConfigHelpers.AlgConfig._get_subclasses
def _get_subclasses(cls)
Definition: ConfigHelpers.py:286
python.HLT.MET.ConfigHelpers.AlgConfig.name_step
str name_step(self, idx)
Definition: ConfigHelpers.py:157
dumpHVPathFromNtuple.append
bool append
Definition: dumpHVPathFromNtuple.py:91
CaloDef.clusterFSInputMaker
def clusterFSInputMaker()
cluster maker functions
Definition: CaloDef.py:53
python.HLT.MET.ConfigHelpers.AlgConfig._suffix
_suffix
Definition: ConfigHelpers.py:135
python.HLT.MET.ConfigHelpers.AlgConfig.inputMakerCA
def inputMakerCA(self, idx)
Definition: ConfigHelpers.py:262
python.HLT.MET.ConfigHelpers.AlgConfig.interpret_reco_dict
dict[str, Any] interpret_reco_dict(self)
Definition: ConfigHelpers.py:149
python.HLT.MET.ConfigHelpers.AlgConfig.algType
str algType(cls)
Definition: ConfigHelpers.py:109
TrigEFMissingETConfig.getMETMonTool
def getMETMonTool(flags, name="MonTool")
Definition: TrigEFMissingETConfig.py:5
python.HLT.MET.ConfigHelpers.AlgConfig.__init__
def __init__(self, **str recoDict)
Definition: ConfigHelpers.py:118
python.HLT.MET.ConfigHelpers.AlgConfig
Definition: ConfigHelpers.py:93
python.HLT.MET.ConfigHelpers.AlgConfig.recoDict
recoDict
Definition: ConfigHelpers.py:134
python.HLT.MET.ConfigHelpers.AlgConfig.getMonTool
def getMonTool(self, flags)
Definition: ConfigHelpers.py:153
TCS::join
std::string join(const std::vector< std::string > &v, const char c=',')
Definition: Trigger/TrigT1/L1Topo/L1TopoCommon/Root/StringUtils.cxx:10
python.HLT.MET.ConfigHelpers.AlgConfig.fromRecoDict
AlgConfig fromRecoDict(cls, EFrecoAlg, **recoDict)
Definition: ConfigHelpers.py:294
python.HLT.MET.ConfigHelpers.AlgConfig.outputKey
str outputKey(self)
Definition: ConfigHelpers.py:138
python.HLT.MET.ConfigHelpers.AlgConfig.fexName
str fexName(self)
Definition: ConfigHelpers.py:145
python.HLT.MET.ConfigHelpers.make_MET_menu_sequenceGenCfg
def make_MET_menu_sequenceGenCfg(flags, sel_acc, hypo_tool)
Definition: ConfigHelpers.py:301
python.TriggerEDM.recordable
def recordable(arg, runVersion=3)
Definition: TriggerEDM.py:37
python.HLT.MET.ConfigHelpers.AlgConfig.make_passthrough_hypo_alg
def make_passthrough_hypo_alg(self, step)
Definition: ConfigHelpers.py:257
calibdata.copy
bool copy
Definition: calibdata.py:26
Cut::all
@ all
Definition: SUSYToolsAlg.cxx:67
python.HLT.MET.ConfigHelpers.stringToMETRecoDict
dict[str, Any] stringToMETRecoDict(str string, bool onlyRecoKeys=False)
Definition: ConfigHelpers.py:73
python.HLT.MET.ConfigHelpers.AlgConfig.make_accumulator_steps
def make_accumulator_steps(self, flags, chainDict)
Definition: ConfigHelpers.py:192