12from collections
import namedtuple
15prompt_script_package =
"prompt.calibrations."
16prompt_script_dir =
"calibration/scripts/prompt/calibrations"
18prompt_validation_script_package =
"prompt.validations."
19prompt_validation_script_dir =
"calibration/scripts/prompt/validations"
21INPUT_DATA_FILTERS = {
"Magnet": {
"On":
"On",
24 "Beam Energy": {
"No Beam":
"No Beam",
26 "Continuum":
"Continuum",
29 "Run Type": {
"beam":
"beam",
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",
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"}
54class CalibrationSettings(namedtuple(
'CalSet_Factory',
64 "produced_payloads"])):
66 Simple class to hold and display required information for a prompt calibration script (process).
69 name (str): The unique calibration name, not longer than 64 characters.
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
75 subsystem (str): The name of the subsystem that this calibration is for.
76 e.g. "cdc", "ecl", "klm", "svd", "top", "beam", etc.
78 description (str): Long form description of the calibration and what it does. Feel free to make this as long as you need.
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.
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"]
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)
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.
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``.
111 produced_payloads (list(str)): The names of the payloads that this calibration script produces.
117 allowed_data_formats = frozenset({
"raw",
"cdst",
"mdst",
"udst"})
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):
123 The special method to create the tuple instance. Returning the instance
124 calls the __init__ method
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)
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'!")
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)}")
149 input_data_filters = {}
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}'")
159 if not isinstance(expert_config, dict):
160 raise TypeError(
"expert_config must be a dictionary")
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.")
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.")
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)
182 def json_dumps(self):
185 str: A valid JSON format string of the attributes.
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
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}"
215class ValidationSettings(namedtuple(
'ValSet_Factory', [
"name",
"description",
"download_files",
"expert_config"])):
217 Simple class to hold and display required information for a validation calibration script (process).
220 name (str): The unique name that must match the corresponding calibration, not longer than 64 characters.
222 description (str): Long form description of the validation and what it does. Feel free to make this as long as you need.
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.
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``.
236 def __new__(cls, name, description, download_files=None, expert_config=None):
238 The special method to create the tuple instance. Returning the instance
239 calls the __init__ method
242 raise ValueError(
"name cannot be longer than 64 characters!")
246 if not isinstance(expert_config, dict):
247 raise TypeError(
"expert_config must be a dictionary")
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.")
258 return super().__new__(cls, name, description, download_files, expert_config)
260 def json_dumps(self):
263 str: A valid JSON format string of the attributes.
265 return json.dumps({
"name": self.name,
266 "description": self.description,
267 "download_files": self.download_files,
268 "expert_config": self.expert_config
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}"