15from softwaretrigger
import constants, path_utils
16from geometry
import check_components
17from pxd
import add_pxd_percentframe
18from rawdata
import add_unpackers
19from reconstruction
import add_reconstruction, add_cosmics_reconstruction
20from tracking
import add_roiFinder, add_roi_payload_assembler
23def setup_basf2_and_db(event_distribution_mode=constants.EventDistributionModes.ringbuffer):
25 Setup local database usage for HLT
27 parser = argparse.ArgumentParser(description=
'basf2 for online')
29 if event_distribution_mode == constants.EventDistributionModes.ringbuffer:
30 parser.add_argument(
'input_buffer_name', type=str,
31 help=
'Input Ring Buffer names')
32 parser.add_argument(
'output_buffer_name', type=str,
33 help=
'Output Ring Buffer name')
34 parser.add_argument(
'histo_port', type=int,
35 help=
'Port of the HistoManager to connect to')
36 parser.add_argument(
'--input-file', type=str,
37 help=
"Input sroot file, if set no RingBuffer input will be used",
39 parser.add_argument(
'--output-file', type=str,
40 help=
"Filename for SeqRoot output, if set no RingBuffer output will be used",
42 parser.add_argument(
'--histo-output-file', type=str,
43 help=
"Filename for histogram output",
45 parser.add_argument(
'--no-output',
46 help=
"Don't write any output files",
47 action=
"store_true", default=
False)
49 parser.add_argument(
"--input", required=
True, type=str, help=
"ZMQ Address of the distributor process")
50 parser.add_argument(
"--output", required=
True, type=str, help=
"ZMQ Address of the collector process")
51 parser.add_argument(
"--dqm", required=
True, type=str, help=
"ZMQ Address of the histoserver process")
53 parser.add_argument(
'--number-processes', type=int, default=multiprocessing.cpu_count() - 5,
54 help=
'Number of parallel processes to use')
55 parser.add_argument(
'--local-db-path', type=str,
56 help=
"set path to the local payload locations to use for the ConditionDB",
57 default=constants.DEFAULT_DB_FILE_LOCATION)
58 parser.add_argument(
'--local-db-tag', type=str, nargs=
"*",
59 help=
"Use the local db with a specific tag (can be applied multiple times, order is relevant)")
60 parser.add_argument(
'--central-db-tag', type=str, nargs=
"*",
61 help=
"Use the central db with a specific tag (can be applied multiple times, order is relevant)")
62 parser.add_argument(
'--udp-hostname', type=str,
63 help=
"set hostname for UDP logging connection", default=
None)
64 parser.add_argument(
'--udp-port', type=int,
65 help=
"set port number for UDP logging connection", default=
None)
67 args = parser.parse_args()
70 basf2.conditions.override_globaltags()
71 if args.central_db_tag:
72 for central_tag
in args.central_db_tag:
73 basf2.conditions.prepend_globaltag(central_tag)
76 for local_tag
in args.local_db_tag:
77 basf2.conditions.prepend_globaltag(local_tag)
79 basf2.conditions.globaltags = [
"online"]
80 basf2.conditions.metadata_providers = [
"file://" + basf2.find_file(args.local_db_path +
"/metadata.sqlite")]
81 basf2.conditions.payload_locations = [basf2.find_file(args.local_db_path)]
84 basf2.set_nprocesses(args.number_processes)
87 basf2.set_log_level(basf2.LogLevel.ERROR)
90 basf2.logging.enable_escape_newlines =
True
93 if (args.udp_hostname
is not None)
and (args.udp_port
is not None):
94 basf2.logging.add_udp(args.udp_hostname, args.udp_port)
97 basf2.set_realm(
"online")
102def start_path(args, location):
104 Create and return a path used for HLT and ExpressReco running
106 path = basf2.create_path()
108 input_buffer_module_name =
""
109 if location == constants.Location.expressreco:
110 input_buffer_module_name =
"Rbuf2Ds"
111 elif location == constants.Location.hlt:
112 input_buffer_module_name =
"Raw2Ds"
114 basf2.B2FATAL(f
"Does not know location {location}")
117 if not args.input_file:
118 path.add_module(input_buffer_module_name, RingBufferName=args.input_buffer_name)
120 if args.input_file.endswith(
".sroot"):
121 path.add_module(
'SeqRootInput', inputFileName=args.input_file)
123 path.add_module(
'RootInput', inputFileName=args.input_file)
126 if not args.histo_output_file:
127 path.add_module(
'DqmHistoManager', Port=args.histo_port, DumpInterval=1000, workDirName=tempfile.gettempdir()+
"/")
129 workdir = os.path.dirname(args.histo_output_file)
130 filename = os.path.basename(args.histo_output_file)
131 path.add_module(
'HistoManager', histoFileName=filename, workDirName=workdir)
136def start_zmq_path(args, location, event_distribution_mode):
138 reco_path = basf2.Path()
140 if location == constants.Location.expressreco:
141 input_module = path.add_module(
"HLTZMQ2Ds", input=args.input, addExpressRecoObjects=
True)
142 elif location == constants.Location.hlt:
143 input_module = path.add_module(
"HLTZMQ2Ds", input=args.input)
145 basf2.B2FATAL(f
"Does not know location {location}")
148 if event_distribution_mode == constants.EventDistributionModes.zmq:
149 input_module.if_value(
"==0", reco_path, basf2.AfterConditionPath.CONTINUE)
150 reco_path.add_module(
"HLTDQM2ZMQ", output=args.dqm, sendOutInterval=30)
152 elif event_distribution_mode == constants.EventDistributionModes.zmqbasf2:
153 path.add_module(
"HLTDQM2ZMQ", output=args.dqm, sendOutInterval=30)
155 basf2.B2FATAL(f
"Does not know event_distribution_mode {event_distribution_mode}")
157 return path, reco_path
160def add_hlt_processing(path,
161 run_type=constants.RunTypes.beam,
162 softwaretrigger_mode=constants.SoftwareTriggerModes.filter,
165 unpacker_components=None,
166 reco_components=None,
167 create_hlt_unit_histograms=True,
168 switch_off_slow_modules_for_online=True,
171 Add all modules for processing on HLT filter machines
175 if run_type == constants.RunTypes.cosmic:
176 basf2.declare_cosmics()
179 if run_type == constants.RunTypes.beam:
185 path.add_module(
'StatisticsSummary').set_name(
'Sum_Wait')
187 if unpacker_components
is None:
188 unpacker_components = constants.DEFAULT_HLT_COMPONENTS
189 if reco_components
is None:
190 reco_components = constants.DEFAULT_HLT_COMPONENTS
192 check_components(unpacker_components)
193 check_components(reco_components)
199 path.add_module(
"PruneDataStore", matchEntries=constants.HLT_INPUT_OBJECTS)
202 path_utils.add_geometry_if_not_present(path)
203 path.add_module(
'StatisticsSummary').set_name(
'Sum_Initialization')
206 add_unpackers(path, components=unpacker_components, writeKLMDigitRaws=
True)
207 path.add_module(
'StatisticsSummary').set_name(
'Sum_Unpackers')
210 accept_path = basf2.Path()
213 path_utils.add_pre_filter_reconstruction(
216 components=reco_components,
217 switch_off_slow_modules_for_online=switch_off_slow_modules_for_online,
222 path_utils.add_filter_software_trigger(path, store_array_debug_prescale=1)
225 path_utils.add_hlt_dqm(path, run_type=run_type, components=reco_components, dqm_mode=constants.DQMModes.before_filter,
226 create_hlt_unit_histograms=create_hlt_unit_histograms)
229 if softwaretrigger_mode == constants.SoftwareTriggerModes.filter:
231 hlt_filter_module = path_utils.add_filter_module(path)
235 path_utils.hlt_event_abort(hlt_filter_module,
"==0", ROOT.Belle2.EventMetaData.c_HLTDiscard)
237 hlt_filter_module.if_value(
"==1", accept_path, basf2.AfterConditionPath.CONTINUE)
238 elif softwaretrigger_mode == constants.SoftwareTriggerModes.monitor:
240 path.add_path(accept_path)
242 basf2.B2FATAL(f
"The software trigger mode {softwaretrigger_mode} is not supported.")
245 path_utils.add_post_filter_reconstruction(
248 components=reco_components,
249 switch_off_slow_modules_for_online=switch_off_slow_modules_for_online
253 add_roiFinder(accept_path)
254 accept_path.add_module(
'StatisticsSummary').set_name(
'Sum_ROI_Finder')
257 path_utils.add_hlt_dqm(
260 components=reco_components,
261 dqm_mode=constants.DQMModes.filtered,
262 create_hlt_unit_histograms=create_hlt_unit_histograms)
269 pxd_ignores_hlt_decision = (softwaretrigger_mode == constants.SoftwareTriggerModes.monitor)
270 add_roi_payload_assembler(path, ignore_hlt_decision=pxd_ignores_hlt_decision)
271 path.add_module(
'StatisticsSummary').set_name(
'Sum_ROI_Payload_Assembler')
274 path_utils.add_hlt_dqm(path, run_type=run_type, components=reco_components, dqm_mode=constants.DQMModes.all_events,
275 create_hlt_unit_histograms=create_hlt_unit_histograms)
279 path_utils.add_store_only_rawdata_path(path)
280 path.add_module(
'StatisticsSummary').set_name(
'Sum_Close_Event')
283def add_hlt_passthrough(path):
285 Add all modules for operating HLT machines in passthrough mode.
287 add_pxd_percentframe(path, fraction=0.1, random_position=
True)
288 add_roi_payload_assembler(path, ignore_hlt_decision=
True)
291def add_expressreco_processing(path,
292 run_type=constants.RunTypes.beam,
293 select_only_accepted_events=False,
296 unpacker_components=None,
297 reco_components=None,
298 do_reconstruction=True,
299 switch_off_slow_modules_for_online=True,
302 Add all modules for processing on the ExpressReco machines
306 if run_type == constants.RunTypes.cosmic:
307 basf2.declare_cosmics()
310 if run_type == constants.RunTypes.beam:
313 if unpacker_components
is None:
314 unpacker_components = constants.DEFAULT_EXPRESSRECO_COMPONENTS
315 if reco_components
is None:
316 reco_components = constants.DEFAULT_EXPRESSRECO_COMPONENTS
318 check_components(unpacker_components)
319 check_components(reco_components)
324 if select_only_accepted_events:
325 skim_module = path.add_module(
"TriggerSkim", triggerLines=[
"software_trigger_cut&all&total_result"], resultOnMissing=0)
326 skim_module.if_value(
"==0", basf2.Path(), basf2.AfterConditionPath.END)
332 path.add_module(
"PruneDataStore", matchEntries=constants.EXPRESSRECO_INPUT_OBJECTS)
334 path_utils.add_geometry_if_not_present(path)
335 add_unpackers(path, components=unpacker_components, writeKLMDigitRaws=
True)
338 basf2.set_module_parameters(path,
"PXDPostErrorChecker", CriticalErrorMask=0)
340 if do_reconstruction:
341 if run_type == constants.RunTypes.beam:
342 add_reconstruction(path,
343 components=reco_components,
345 skipGeometryAdding=
True,
346 add_trigger_calculation=
False,
347 switch_off_slow_modules_for_online=switch_off_slow_modules_for_online,
349 elif run_type == constants.RunTypes.cosmic:
350 add_cosmics_reconstruction(path, components=reco_components, pruneTracks=
False,
351 skipGeometryAdding=
True, **kwargs)
353 basf2.B2FATAL(f
"Run Type {run_type} not supported.")
355 basf2.set_module_parameters(path,
"SVDTimeGrouping", forceGroupingFromDB=
False,
356 isEnabledIn6Samples=
True, isEnabledIn3Samples=
True)
358 path_utils.add_expressreco_dqm(path, run_type, components=reco_components)
361 path.add_module(
"PruneDataStore", matchEntries=constants.ALWAYS_SAVE_OBJECTS + constants.RAWDATA_OBJECTS +
362 constants.PROCESSED_OBJECTS)
365def finalize_path(path, args, location, show_progress_bar=True):
367 Add the required output modules for expressreco/HLT
369 save_objects = constants.ALWAYS_SAVE_OBJECTS + constants.RAWDATA_OBJECTS
370 if location == constants.Location.expressreco:
371 save_objects += constants.PROCESSED_OBJECTS
373 if show_progress_bar:
374 path.add_module(
"Progress")
377 basf2.set_streamobjs(save_objects)
382 output_buffer_module_name =
""
383 if location == constants.Location.expressreco:
384 output_buffer_module_name =
"Ds2Sample"
385 elif location == constants.Location.hlt:
386 output_buffer_module_name =
"Ds2Rbuf"
388 basf2.B2FATAL(f
"Does not know location {location}")
390 if not args.output_file:
391 path.add_module(output_buffer_module_name, RingBufferName=args.output_buffer_name,
392 saveObjs=save_objects)
394 if args.output_file.endswith(
".sroot"):
395 path.add_module(
"SeqRootOutput", saveObjs=save_objects, outputFileName=args.output_file)
398 path.add_module(
"RootOutput", outputFileName=args.output_file)
401def finalize_zmq_path(path, args, location):
403 Add the required output modules for expressreco/HLT
405 save_objects = constants.ALWAYS_SAVE_OBJECTS + constants.RAWDATA_OBJECTS
406 if location == constants.Location.expressreco:
407 save_objects += constants.PROCESSED_OBJECTS
410 basf2.set_streamobjs(save_objects)
412 if location == constants.Location.expressreco:
413 path.add_module(
"HLTDs2ZMQ", output=args.output, raw=
False, outputConfirmation=
False)
414 elif location == constants.Location.hlt:
415 path.add_module(
"HLTDs2ZMQ", output=args.output, raw=
True, outputConfirmation=
True)
417 basf2.B2FATAL(f
"Does not know location {location}")