ATLAS Offline Software
EFTrackingXrtAlgorithmConfig.py
Go to the documentation of this file.
1 # Copyright (C) 2002-2025 CERN for the benefit of the ATLAS collaboration
2 
3 import ROOT
4 
5 def EFTrackingXrtAlgorithmCfg(flags, **kwargs):
6  from AthenaConfiguration.ComponentAccumulator import ComponentAccumulator
8 
9  kwargs.setdefault("bufferSize", 8192)
10  kwargs.setdefault(
11  "xclbinPath",
12  "/eos/project-a/atlas-eftracking/FPGA_compilation/FPGA_compilation_hw/12_road2track_HLS/Road2Track_hw.xclbin"
13  )
14 
15  from json import dumps
16  kwargs.setdefault("kernelDefinitionsJsonString", dumps({
17  "loader": [{"storeGateKey": "inputDataStream",
18  "argumentIndex": 0,
19  "interfaceMode": ROOT.EFTrackingXrtParameters.InterfaceMode.INPUT}],
20  "unloader": [{"storeGateKey": "outputDataStream",
21  "argumentIndex": 1,
22  "interfaceMode": ROOT.EFTrackingXrtParameters.InterfaceMode.OUTPUT}]}))
23 
24  from AthenaConfiguration.ComponentFactory import CompFactory
25  EFTrackingXrtAlgorithm = CompFactory.EFTrackingXrtAlgorithm("EFTrackingXrtAlgorithm", **kwargs)
26 
27  acc.addEventAlgo(EFTrackingXrtAlgorithm)
28 
29  return acc
30 
31 if __name__ == "__main__":
32  from AthenaConfiguration.AllConfigFlags import initConfigFlags
33  flags = initConfigFlags()
34  flags.lock()
35 
36  from AthenaConfiguration.MainServicesConfig import MainServicesCfg
37  acc = MainServicesCfg(flags)
38 
39  from argparse import ArgumentParser
40  argumentParser = ArgumentParser()
41  argumentParser.add_argument("--xclbinPath")
42  argumentParser.add_argument("--bufferSize", type = int, default = 8192)
43 
44  from argparse import Action
45  class JsonToDictAction(Action):
46  def __call__(self, parser, namespace, values, option_string=None):
47  # This is a hack but this is only needed when mapping commandline
48  # arguments into an enum.
49  #
50  # The underlying dictionary can be hardcoded in dedicated Cfg
51  # functions, see defaults in EFTrackingXrtAlgorithmCfg
52  #
53  # Get the members of InterfaceModes that are not present in an
54  # EmptyEnum (i.e. the actual members of InterfaceModes).
55  interfaceModes = [(member, int(getattr(ROOT.EFTrackingXrtParameters.InterfaceMode, member)))
56  for member in dir(ROOT.EFTrackingXrtParameters.InterfaceMode)
57  if member not in dir(ROOT.EFTrackingXrtParameters.EmptyEnum)]
58 
59  for interfaceModeString, interfaceModeValue in interfaceModes:
60  values = values.replace(interfaceModeString, str(interfaceModeValue))
61 
62  from json import loads
63  setattr(namespace, self.dest, loads(values))
64 
65  argumentParser.add_argument("--kernelDefinitions", action=JsonToDictAction)
66  argumentParser.add_argument("--inputCsvPathToSgKeyMap", action=JsonToDictAction)
67  argumentParser.add_argument("--outputCsvPathToSgKeyMap", action=JsonToDictAction)
68 
69  arguments = argumentParser.parse_args()
70 
71  from AthenaConfiguration.ComponentFactory import CompFactory
72  acc.addService(CompFactory.AthXRT.DeviceMgmtSvc(XclbinPathsList = [arguments.xclbinPath]))
73 
74  for inputCsvPath, sgKey in arguments.inputCsvPathToSgKeyMap.items():
75  from EFTrackingFPGAIntegration.EFTrackingDataStreamLoaderAlgorithmConfig import EFTrackingDataStreamLoaderAlgorithmCfg
77  bufferSize = arguments.bufferSize,
78  inputCsvPath = inputCsvPath,
79  inputDataStream = sgKey))
80 
81  from json import dumps
82  acc.merge(EFTrackingXrtAlgorithmCfg(flags,
83  bufferSize = arguments.bufferSize,
84  xclbinPath = arguments.xclbinPath,
85  kernelDefinitionsJsonString = dumps(arguments.kernelDefinitions)))
86 
87  for outputCsvPath, sgKey in arguments.outputCsvPathToSgKeyMap.items():
88  from EFTrackingFPGAIntegration.EFTrackingDataStreamUnloaderAlgorithmConfig import EFTrackingDataStreamUnloaderAlgorithmCfg
90  outputCsvPath = outputCsvPath,
91  outputDataStream = sgKey))
92 
93  acc.run(2)
94 
EFTrackingXrtAlgorithmConfig.int
int
Definition: EFTrackingXrtAlgorithmConfig.py:42
python.JetAnalysisCommon.ComponentAccumulator
ComponentAccumulator
Definition: JetAnalysisCommon.py:302
EFTrackingXrtAlgorithmConfig.JsonToDictAction.__call__
def __call__(self, parser, namespace, values, option_string=None)
Definition: EFTrackingXrtAlgorithmConfig.py:46
EFTrackingXrtAlgorithmConfig.JsonToDictAction
Definition: EFTrackingXrtAlgorithmConfig.py:45
python.MainServicesConfig.MainServicesCfg
def MainServicesCfg(flags, LoopMgr='AthenaEventLoopMgr')
Definition: MainServicesConfig.py:260
beamspotman.dir
string dir
Definition: beamspotman.py:623
EFTrackingDataStreamUnloaderAlgorithmConfig.EFTrackingDataStreamUnloaderAlgorithmCfg
def EFTrackingDataStreamUnloaderAlgorithmCfg(flags, **kwargs)
Definition: EFTrackingDataStreamUnloaderAlgorithmConfig.py:4
EFTrackingXrtAlgorithmConfig.EFTrackingXrtAlgorithmCfg
def EFTrackingXrtAlgorithmCfg(flags, **kwargs)
Definition: EFTrackingXrtAlgorithmConfig.py:5
python.AllConfigFlags.initConfigFlags
def initConfigFlags()
Definition: AllConfigFlags.py:19
EFTrackingDataStreamLoaderAlgorithmConfig.EFTrackingDataStreamLoaderAlgorithmCfg
def EFTrackingDataStreamLoaderAlgorithmCfg(flags, **kwargs)
Definition: EFTrackingDataStreamLoaderAlgorithmConfig.py:4
str
Definition: BTagTrackIpAccessor.cxx:11