Belle II Software development
__init__.py
1# disable doxygen check for this file
2# @cond
3
4
11import basf2
12from collections import namedtuple
13import json
14
15prompt_script_package = "prompt.calibrations."
16prompt_script_dir = "calibration/scripts/prompt/calibrations"
17
18prompt_validation_script_package = "prompt.validations."
19prompt_validation_script_dir = "calibration/scripts/prompt/validations"
20
21INPUT_DATA_FILTERS = {"Magnet": {"On": "On",
22 "Off": "Off",
23 "Either": "Either"},
24 "Beam Energy": {"No Beam": "No Beam",
25 "4S": "4S",
26 "Continuum": "Continuum",
27 "": "",
28 "Scan": "Scan"},
29 "Run Type": {"beam": "beam",
30 "cosmic": "cosmic",
31 "debug": "debug", "null": "null",
32 "physics": "physics"},
33 "Data Tag": {"bhabha_all_calib": "bhabha_all_calib",
34 "cosmic_calib": "cosmic_calib",
35 "gamma_gamma_calib": "gamma_gamma_calib",
36 "hadron_calib": "hadron_calib",
37 "btocharm_calib": "btocharm_calib",
38 "mumu_tight_or_highm_calib": "mumu_tight_or_highm_calib",
39 "offip_calib": "offip_calib",
40 "radmumu_calib": "radmumu_calib",
41 "random_calib": "random_calib",
42 "delayedbhabha_calib": "delayedbhabha_calib",
43 "single_gamma_mc": "single_gamma_mc"},
44 "Data Quality Tag": {">=30 Minute Run": ">=30 Minute Run",
45 "Bad For Alignment": "Bad For Alignment",
46 "Good": "Good",
47 "Good Shifter": "Good Shifter",
48 "Good For PXD": "Good For PXD",
49 "Good Or Recoverable": "Good Or Recoverable",
50 "Good Or Recoverable Shifter": "Good Or Recoverable Shifter"}
51 }
52
53
54class CalibrationSettings(namedtuple('CalSet_Factory',
55 ["name",
56 "expert_username",
57 "subsystem",
58 "description",
59 "input_data_formats",
60 "input_data_names",
61 "input_data_filters",
62 "depends_on",
63 "expert_config",
64 "produced_payloads"])):
65 """
66 Simple class to hold and display required information for a prompt calibration script (process).
67
68 Parameters:
69 name (str): The unique calibration name, not longer than 64 characters.
70
71 expert_username (str): The GitLab username of the expert to contact about this script.
72 This username will be used to assign the default responsible person for submitting and checking prompt
73 calibration jobs.
74
75 subsystem (str): The name of the subsystem that this calibration is for.
76 e.g. "cdc", "ecl", "klm", "svd", "top", "beam", etc.
77
78 description (str): Long form description of the calibration and what it does. Feel free to make this as long as you need.
79
80 input_data_formats (frozenset(str)): The data formats {'raw', 'cdst', 'mdst', 'udst'} of the input files
81 that should be used as input to the process. Used to figure out if this calibration should occur
82 before the relevant data production e.g. before cDST files are created.
83
84 input_data_names (frozenset(str)): The names that you will use when accessing the input data given to the
85 prompt calibration process i.e. Use these in the ``get_calibrations`` function to access the correct input
86 data files. e.g. input_data_names=["all_events", "offres_photon_events"]
87
88 input_data_filters (dict): The data selection for the data input names, used for automated calibration.
89 The keys should correspond to one of the ``input_data_names`` with the values being a list of the various data
90 filters, e.g. Data Tag, Beam Energy, Run Type, Run Quality Tag and Magnet. All available filters can be found in the
91 input_data_filters dictionary e.g. from prompt import input_data_filters with details about data tags and run quality
92 tags found at: https://calibration.belle2.org/belle2/data_tags/list/.
93 To exclude specific filters, pre-append with *NOT* e.g.
94 {"all_events": ["mumu_tight_or_highm_calib", "hadron_calib", "Good", "On"],
95 "offres_photon_events": ["gamma_gamma_calib", "Good", "NOT On"]}.
96 Not selecting a specific filters (e.g. Magnet) is equivalent to not having any requirements, e.g. (Either)
97
98 depends_on (list(CalibrationSettings)): The settings variables of the other prompt calibrations that you want
99 want to depend on. This will allow the external automatic system to understand the overall ordering of
100 scripts to run. If you encounter an import error when trying to run your prompt calibration script, it is
101 likely that you have introduced a circular dependency.
102
103 expert_config (dict): Default expert configuration for this calibration script. This is an optional dictionary
104 (which must be JSON compliant) of configuration options for your get_calibrations(...) function.
105 This is supposed to be used as a catch-all place to send in options for your calibration setup. For example,
106 you may want to have an optional list of IoV boundaries so that your prompt script knows that it should split the
107 input data between different IoV ranges. Or you might want to send if options like the maximum events per
108 input file to process. The value in your settings object will be the *default*, but you can override the value via
109 the caf_config.json sent into ``b2caf-prompt-run``.
110
111 produced_payloads (list(str)): The names of the payloads that this calibration script produces.
112 """
113
114
117 allowed_data_formats = frozenset({"raw", "cdst", "mdst", "udst"})
118
119 def __new__(cls, name, expert_username, subsystem, description,
120 input_data_formats=None, input_data_names=None, input_data_filters=None,
121 depends_on=None, expert_config=None, produced_payloads=None):
122 """
123 The special method to create the tuple instance. Returning the instance
124 calls the __init__ method
125 """
126 if len(name) > 64:
127 raise ValueError("name cannot be longer than 64 characters!")
128 if not input_data_formats:
129 raise ValueError("You must specify at least one input data format")
130 input_data_formats = frozenset(map(lambda x: x.lower(), input_data_formats))
131 if input_data_formats.difference(cls.allowed_data_formats):
132 raise ValueError("There was a data format that is not in the allowed_data_formats attribute.")
133 if not input_data_names:
134 raise ValueError("You must specify at least one input data name")
135 input_data_names = frozenset(input_data_names)
136
137 # The input data names in the filters MUST correspond to the input data names for the calibration.
138 if input_data_filters:
139 if set(input_data_filters.keys()) != input_data_names:
140 raise ValueError("The 'input_data_filters' keys don't match the 'input_data_names'!")
141 # Requested input data filters MUST exist in the ones we defined in the global dictionary.
142 allowed_filters = {filter_name for category in INPUT_DATA_FILTERS.values() for filter_name in category}
143 requested_filters = {filter_name.replace("NOT", "", 1).lstrip() for filters in input_data_filters.values()
144 for filter_name in filters}
145 if not allowed_filters.issuperset(requested_filters):
146 raise ValueError("The 'input_data_filters' contains unknown filter names:"
147 f"{requested_filters.difference(allowed_filters)}")
148 else:
149 input_data_filters = {}
150
151 # Check that the subsystem is among the allowed ones
152 allowed_subsystems = {"pxd", "svd", "cdc", "ecl", "top", "arich",
153 "klm", "trigger", "tracking", "alignment", "beam", "example"}
154 if subsystem not in allowed_subsystems:
155 raise ValueError(f"subsystem must be one of {allowed_subsystems}, but got '{subsystem}'")
156
157 if expert_config:
158 # Check that it's a dictionary and not some other valid JSON object
159 if not isinstance(expert_config, dict):
160 raise TypeError("expert_config must be a dictionary")
161 # Check if it is JSONable since people might put objects in there by mistake
162 try:
163 json.dumps(expert_config)
164 except TypeError as e:
165 basf2.B2ERROR("expert_config could not be serialised to JSON. "
166 "Most likely you used a non-supported type e.g. datetime.")
167 raise e
168 else:
169 expert_config = {}
170
171 if depends_on:
172 for calibration_settings in depends_on:
173 if not isinstance(calibration_settings, cls):
174 raise TypeError(f"A list of {str(cls)} object is required when setting the 'depends_on' keyword.")
175 else:
176 depends_on = []
177
178 return super().__new__(cls, name, expert_username, subsystem, description,
179 input_data_formats, input_data_names, input_data_filters,
180 depends_on, expert_config, produced_payloads)
181
182 def json_dumps(self):
183 """
184 Returns:
185 str: A valid JSON format string of the attributes.
186 """
187 depends_on_names = [calibration_settings.name for calibration_settings in self.depends_on]
188 return json.dumps({"name": self.name,
189 "expert_username": self.expert_username,
190 "subsystem": self.subsystem,
191 "input_data_formats": list(self.input_data_formats),
192 "input_data_names": list(self.input_data_names),
193 "input_data_filters": self.input_data_filters,
194 "depends_on": list(depends_on_names),
195 "description": self.description,
196 "expert_config": self.expert_config,
197 "produced_payloads": self.produced_payloads
198 })
199
200 def __str__(self):
201 depends_on_names = [calibration_settings.name for calibration_settings in self.depends_on]
202 output_str = str(self.__class__.__name__) + f"(name='{self.name}'):\n"
203 output_str += f" expert_username='{self.expert_username}'\n"
204 output_str += f" subsystem='{self.subsystem}'\n"
205 output_str += f" input_data_formats={list(self.input_data_formats)}\n"
206 output_str += f" input_data_names={list(self.input_data_names)}\n"
207 output_str += f" input_data_filters={list(self.input_data_filters)}\n"
208 output_str += f" depends_on={list(depends_on_names)}\n"
209 output_str += f" description='{self.description}'\n"
210 output_str += f" expert_config={self.expert_config}\n"
211 output_str += f" produced_payloads={self.produced_payloads}"
212 return output_str
213
214
215class ValidationSettings(namedtuple('ValSet_Factory', ["name", "description", "download_files", "expert_config"])):
216 """
217 Simple class to hold and display required information for a validation calibration script (process).
218
219 Parameters:
220 name (str): The unique name that must match the corresponding calibration, not longer than 64 characters.
221
222 description (str): Long form description of the validation and what it does. Feel free to make this as long as you need.
223
224 download_files (list): The names of the files you want downloaded, e.g. mycalibration_stdout. If multiple files of
225 the same name are found, all files are downloaded and appended with the folder they were in.
226
227 expert_config (dict): Default expert configuration for this validation script. This is an optional dictionary
228 (which must be JSON compliant) of configuration options for validation script.
229 This is supposed to be used as a catch-all place to send in options for your calibration setup. For example,
230 you may want to have an optional list of IoV boundaries so that your validation script knows that it should split the
231 input data between different IoV ranges. Or you might want to send if options like the maximum events per
232 input file to process. The value in your settings object will be the *default*, but you can override the value via
233 the caf_config.json sent into ``b2caf-prompt-run``.
234 """
235
236 def __new__(cls, name, description, download_files=None, expert_config=None):
237 """
238 The special method to create the tuple instance. Returning the instance
239 calls the __init__ method
240 """
241 if len(name) > 64:
242 raise ValueError("name cannot be longer than 64 characters!")
243
244 if expert_config:
245 # Check that it's a dictionary and not some other valid JSON object
246 if not isinstance(expert_config, dict):
247 raise TypeError("expert_config must be a dictionary")
248 # Check if it is JSONable since people might put objects in there by mistake
249 try:
250 json.dumps(expert_config)
251 except TypeError as e:
252 basf2.B2ERROR("expert_config could not be serialised to JSON. "
253 "Most likely you used a non-supported type e.g. datetime.")
254 raise e
255 else:
256 expert_config = {}
257
258 return super().__new__(cls, name, description, download_files, expert_config)
259
260 def json_dumps(self):
261 """
262 Returns:
263 str: A valid JSON format string of the attributes.
264 """
265 return json.dumps({"name": self.name,
266 "description": self.description,
267 "download_files": self.download_files,
268 "expert_config": self.expert_config
269 })
270
271 def __str__(self):
272 output_str = str(self.__class__.__name__) + f"(name='{self.name}'):\n"
273 output_str += f" description='{self.description}'\n"
274 output_str += f" download_files='{self.download_files}'\n"
275 output_str += f" expert_config={self.expert_config}"
276 return output_str
277
278# @endcond
STL class.