10Airflow script for TOP pre-tracking calibration:
14from prompt
import CalibrationSettings, INPUT_DATA_FILTERS
15from caf.utils
import IoV
16from caf.strategies
import SequentialBoundaries
17from top_calibration
import channel_mask_calibration
21settings = CalibrationSettings(
22 name=
"TOP pre-tracking calibration",
23 expert_username=
"kohani",
26 input_data_formats=[
"raw"],
27 input_data_names=[
"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"]]},
35 "max_files_per_run": 20,
36 "payload_boundaries":
None,
37 "request_memory":
"8 GB"
39 produced_payloads=[
"TOPCalChannelMask"])
43def get_calibrations(input_data, **kwargs):
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.
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
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)
60 cal = [channel_mask_calibration(inputFiles)]
64 if c.strategies[0] == SequentialBoundaries:
67 payload_boundaries = [[output_iov.exp_low, output_iov.run_low]]
70 if expert_config[
"payload_boundaries"]
is not None:
71 payload_boundaries = expert_config[
"payload_boundaries"]
74 for alg
in c.algorithms:
75 alg.params = {
"iov_coverage": output_iov,
"payload_boundaries": payload_boundaries}
79 for alg
in c.algorithms:
80 alg.params = {
"iov_coverage": output_iov}