9"""A simple example calibration that takes one input data list from raw data and performs
10a single calibration."""
12from prompt
import CalibrationSettings, INPUT_DATA_FILTERS
15settings = CalibrationSettings(
16 name=
"ARICH channel masks",
17 expert_username=
"kristof.spenko",
20 input_data_formats=[
"raw"],
21 input_data_names=[
"bhabha_all_calib"],
24 INPUT_DATA_FILTERS[
"Data Tag"][
"bhabha_all_calib"],
25 INPUT_DATA_FILTERS[
"Run Type"][
"physics"],
26 INPUT_DATA_FILTERS[
"Data Quality Tag"][
"Good Or Recoverable"]]},
28 produced_payloads=[
'ARICHChannelMask'])
33def get_calibrations(input_data, **kwargs):
36 input_data (dict): Should contain every name from the 'input_data_names' variable as a key.
37 Each value is a dictionary with {"/path/to/file_e1_r5.root": IoV(1,5,1,5), ...}. Useful for
38 assigning to calibration.files_to_iov
40 **kwargs: Configuration options to be sent in. Since this may change we use kwargs as a way to help prevent
41 backwards compatibility problems. But you could use the correct arguments in b2caf-prompt-run for this
42 release explicitly if you want to.
44 Currently only kwargs["output_iov"] is used. This is the output IoV range that your payloads should
45 correspond to. Generally your highest ExpRun payload should be open ended e.g. IoV(3,4,-1,-1)
48 list(caf.framework.Calibration): All of the calibration objects we want to assign to the CAF process
55 file_to_iov_physics = input_data[
"bhabha_all_calib"]
60 max_files_per_run = 100
66 reduced_file_to_iov_physics = filter_by_max_files_per_run(file_to_iov_physics, max_files_per_run)
67 input_files_physics = list(reduced_file_to_iov_physics.keys())
68 basf2.B2INFO(f
"Total number of files actually used as input = {len(input_files_physics)}")
72 requested_iov = kwargs.get(
"requested_iov",
None)
74 from caf.utils
import IoV
76 output_iov = IoV(requested_iov.exp_low, requested_iov.run_low, -1, -1)
81 from basf2
import create_path
82 from ROOT
import Belle2
83 from ROOT.Belle2
import ARICHChannelMaskMaker
85 alg_arich = ARICHChannelMaskMaker()
90 from caf.framework
import Calibration
91 from caf.strategies
import SequentialRunByRun
94 rec_path_1 = create_path()
95 rec_path_1.add_module(
'ARICHUnpacker')
97 cal_test = Calibration(
"ARICHChannelMasks",
98 collector=
"ARICHChannelMask",
99 algorithms=[alg_arich],
100 input_files=input_files_physics,
101 pre_collector_path=rec_path_1
104 cal_test.strategies = SequentialRunByRun
108 for algorithm
in cal_test.algorithms:
109 algorithm.params = {
"iov_coverage": output_iov}