Belle II Software development
caf_top_pre.py
1
8
9"""
10Airflow script for TOP pre-tracking calibration:
11 channel masks
12"""
13import basf2
14from prompt import CalibrationSettings, INPUT_DATA_FILTERS
15from caf.utils import IoV
16from caf.strategies import SequentialBoundaries
17from top_calibration import channel_mask_calibration
18from prompt.utils import filter_by_max_files_per_run
19
20
21settings = CalibrationSettings(
22 name="TOP pre-tracking calibration",
23 expert_username="kohani",
24 subsystem="top",
25 description=__doc__,
26 input_data_formats=["raw"],
27 input_data_names=["hadron_calib"],
28 input_data_filters={
29 "hadron_calib": [
30 INPUT_DATA_FILTERS["Data Tag"]["hadron_calib"],
31 INPUT_DATA_FILTERS["Run Type"]["physics"],
32 INPUT_DATA_FILTERS["Data Quality Tag"]["Good Or Recoverable"]]},
33 depends_on=[],
34 expert_config={
35 "max_files_per_run": 20,
36 "payload_boundaries": None,
37 "request_memory": "8 GB"
38 },
39 produced_payloads=["TOPCalChannelMask"])
40
41
42# Required function
43def get_calibrations(input_data, **kwargs):
44 '''
45 Returns a list of calibration objects.
46 :input_data (dict): Contains every file name from the 'input_data_names' as a key.
47 :**kwargs: Configuration options to be sent in.
48 '''
49 file_to_iov = input_data["hadron_calib"]
50 expert_config = kwargs.get("expert_config")
51 max_files_per_run = expert_config["max_files_per_run"]
52 min_events_per_file = 1
53 # Applying the min event per file to remove empty root files
54 reduced_file_to_iov = filter_by_max_files_per_run(file_to_iov, max_files_per_run, min_events_per_file, random_select=True)
55 inputFiles = list(reduced_file_to_iov.keys())
56 basf2.B2INFO(f"Total number of files actually used as input = {len(inputFiles)}")
57 requested_iov = kwargs.get("requested_iov", None)
58 output_iov = IoV(requested_iov.exp_low, requested_iov.run_low, -1, -1)
59
60 cal = [channel_mask_calibration(inputFiles)] # this is run-dep
61
62 for c in cal:
63 # If it's a SequentialBoundary calibration, check if there is any boundary in the config file
64 if c.strategies[0] == SequentialBoundaries:
65
66 # Default boundaries. If there are no boundaries in the config file, this calibration will give a single IoV
67 payload_boundaries = [[output_iov.exp_low, output_iov.run_low]]
68
69 # user-defined boundaries are set here.
70 if expert_config["payload_boundaries"] is not None:
71 payload_boundaries = expert_config["payload_boundaries"]
72
73 # Set the actual boundaries.
74 for alg in c.algorithms:
75 alg.params = {"iov_coverage": output_iov, "payload_boundaries": payload_boundaries}
76
77 # If it's not a SequentialBoundary calbration, just set the IoV coverage
78 else:
79 for alg in c.algorithms:
80 alg.params = {"iov_coverage": output_iov}
81
82 return cal