10 Airflow script for automatic CDC dEdx calibration. It is currently for the electron based
11 calibration, where at present only RunGain, injection time, Cosine and WireGain are implimented.
12 The remaining two 2D and 1D will be implimented in the near future.
14 Second part called "Hadron calibration" are not compitable with CAF/AirFlow
15 and will be done offline for a while.
19 from ROOT
import gSystem
20 from ROOT.Belle2
import CDCDedxRunGainAlgorithm, CDCDedxCosineAlgorithm, CDCDedxWireGainAlgorithm
21 from ROOT.Belle2
import CDCDedxCosEdgeAlgorithm, CDCDedxBadWireAlgorithm, CDCDedxInjectTimeAlgorithm
22 from caf.framework
import Calibration
23 from caf.strategies
import SequentialRunByRun, SequentialBoundaries
24 from prompt
import CalibrationSettings, INPUT_DATA_FILTERS
25 import reconstruction
as recon
26 from random
import seed
29 gSystem.Load(
'libreconstruction.so')
30 ROOT.gROOT.SetBatch(
True)
32 settings = CalibrationSettings(
34 expert_username=
"renu",
36 input_data_formats=[
"cdst"],
37 input_data_names=[
"bhabha_all_calib"],
39 "payload_boundaries": [],
40 "calib_datamode":
False,
45 "calibration_procedure": {
"rgtrail0": 0,
"rgpre0": 0,
"rg0": 0}
49 INPUT_DATA_FILTERS[
'Run Type'][
'physics'],
50 INPUT_DATA_FILTERS[
'Data Tag'][
'bhabha_all_calib'],
51 INPUT_DATA_FILTERS[
'Data Quality Tag'][
'Good Or Recoverable'],
52 INPUT_DATA_FILTERS[
'Magnet'][
'On'],
53 INPUT_DATA_FILTERS[
'Beam Energy'][
'4S'],
54 INPUT_DATA_FILTERS[
'Beam Energy'][
'Continuum'],
55 INPUT_DATA_FILTERS[
'Beam Energy'][
'Scan']]},
59 def get_calibrations(input_data, **kwargs):
60 """ REQUIRED FUNCTION used by b2caf-prompt-run tool
61 This function return a list of Calibration
62 objects we assign to the CAF process
66 file_to_iov_physics = input_data[
"bhabha_all_calib"]
68 expert_config = kwargs.get(
"expert_config")
69 calib_mode = expert_config[
"calib_mode"]
71 fulldataMode = expert_config[
"calib_datamode"]
74 input_files_rungain = list(file_to_iov_physics.keys())
75 input_files_coscorr = list(file_to_iov_physics.keys())
76 input_files_wiregain = list(file_to_iov_physics.keys())
80 maxevt_rg = expert_config[
"maxevt_rg"]
81 maxevt_cc = expert_config[
"maxevt_cc"]
82 maxevt_wg = expert_config[
"maxevt_wg"]
84 from prompt.utils import filter_by_max_events_per_run, filter_by_select_max_events_from_files
87 max_files_for_maxevents = maxevt_rg
88 reduced_file_to_iov_rungain = filter_by_max_events_per_run(file_to_iov_physics, max_files_for_maxevents,
True)
89 input_files_rungain = list(reduced_file_to_iov_rungain.keys())
90 basf2.B2INFO(f
"Total number of files used for rungains = {len(input_files_rungain)}")
93 input_files_coscorr = filter_by_select_max_events_from_files(list(file_to_iov_physics.keys()), maxevt_cc)
94 basf2.B2INFO(f
"Total number of files used for cosine = {len(input_files_coscorr)}")
95 if not input_files_coscorr:
97 f
"Cosine: all requested ({maxevt_cc}) events not found")
100 if maxevt_wg == maxevt_cc:
101 input_files_wiregain = input_files_coscorr
103 input_files_wiregain = filter_by_select_max_events_from_files(list(file_to_iov_physics.keys()), maxevt_wg)
105 basf2.B2INFO(f
"Total number of files used for wiregains = {len(input_files_wiregain)}")
106 if not input_files_wiregain:
108 f
"WireGain: all requested ({maxevt_wg}) events not found")
110 requested_iov = kwargs.get(
"requested_iov",
None)
111 from caf.utils
import ExpRun, IoV
112 output_iov = IoV(requested_iov.exp_low, requested_iov.run_low, -1, -1)
114 payload_boundaries = [ExpRun(output_iov.exp_low, output_iov.run_low)]
115 payload_boundaries.extend([ExpRun(*boundary)
for boundary
in expert_config[
"payload_boundaries"]])
116 basf2.B2INFO(f
"Expert set payload boundaries are: {expert_config['payload_boundaries']}")
118 collector_granularity =
'all'
119 if expert_config[
"payload_boundaries"]
is not None:
120 basf2.B2INFO(
'Found payload_boundaries: set collector granularity to run')
121 collector_granularity =
'run'
123 if calib_mode ==
"full":
124 calibration_procedure = {
135 elif calib_mode ==
"quick":
136 calibration_procedure = {
144 elif calib_mode ==
"manual":
145 calibration_procedure = expert_config[
"calibration_procedure"]
147 basf2.B2FATAL(f
"Calibration mode is not defined {calib_mode}, should be quick, full, or manual")
149 calib_keys = list(calibration_procedure)
150 cals = [
None]*len(calib_keys)
151 basf2.B2INFO(f
"Run calibration mode = {calib_mode}:")
154 for i
in range(len(cals)):
155 max_iter = calibration_procedure[calib_keys[i]]
157 data_files = [input_files_rungain, input_files_coscorr, input_files_wiregain]
158 cal_name =
''.join([i
for i
in calib_keys[i]
if not i.isdigit()])
159 if cal_name ==
"rg" or cal_name ==
"rgtrail" or cal_name ==
"rgpre":
161 elif cal_name ==
"cc":
163 elif cal_name ==
"ce":
164 alg = [cosedge_algo()]
165 elif cal_name ==
"tg" or cal_name ==
"tgpre":
167 elif cal_name ==
"bd":
168 alg = [badwire_algo()]
169 elif cal_name ==
"wg":
172 basf2.B2FATAL(f
"The calibration is not defined, check spelling: calib {i}: {calib_keys[i]}")
174 basf2.B2INFO(f
"calibration for {calib_keys[i]} with number of iteration={max_iter}")
176 cals[i] = CDCDedxCalibration(name=cal_name,
178 input_file_dict=data_files,
179 max_iterations=max_iter,
180 collector_granularity=collector_granularity,
181 dependencies=[cals[i-1]]
if i > 0
else None
183 if payload_boundaries:
184 basf2.B2INFO(
"Found payload_boundaries: calibration strategies set to SequentialBoundaries.")
185 if cal_name ==
"rg" or cal_name ==
"rgtrail" or cal_name ==
"rgpre" or cal_name ==
"tg" or cal_name ==
"tgpre":
186 cals[i].strategies = SequentialRunByRun
187 for algorithm
in cals[i].algorithms:
188 algorithm.params = {
"iov_coverage": output_iov}
189 if cal_name ==
"rgtrail" or cal_name ==
"rgpre" or cal_name ==
"tgpre":
190 cals[i].save_payloads =
False
192 cals[i].strategies = SequentialBoundaries
193 for algorithm
in cals[i].algorithms:
194 algorithm.params = {
"iov_coverage": output_iov,
"payload_boundaries": payload_boundaries}
197 for algorithm
in cals[i].algorithms:
198 algorithm.params = {
"apply_iov": output_iov}
204 def pre_collector(name='rg'):
206 Define pre collection.
208 name : name of the calibration
209 rungain rg0 by Default.
211 path : path for pre collection
214 reco_path = basf2.create_path()
215 recon.prepare_cdst_analysis(path=reco_path)
216 if(name ==
"tg" or name ==
"tgpre"):
217 trg_bhabhaskim = reco_path.add_module(
"TriggerSkim", triggerLines=[
"software_trigger_cut&skim&accept_radee"])
218 trg_bhabhaskim.if_value(
"==0", basf2.Path(), basf2.AfterConditionPath.END)
223 trg_bhabhaskim = reco_path.add_module(
226 "software_trigger_cut&skim&accept_bhabha",
227 "software_trigger_cut&filter&ee_flat_90_180",
228 "software_trigger_cut&filter&ee_flat_0_19"])
229 trg_bhabhaskim.if_value(
"==0", basf2.Path(), basf2.AfterConditionPath.END)
231 trg_bhabhaskim = reco_path.add_module(
"TriggerSkim", triggerLines=[
"software_trigger_cut&skim&accept_bhabha"])
232 trg_bhabhaskim.if_value(
"==0", basf2.Path(), basf2.AfterConditionPath.END)
234 reco_path.add_module(
236 relativeCorrections=
False,
249 def collector(granularity='all', name=''):
251 Create a cdcdedx calibration collector
253 name : name of calibration
254 granularity : granularity : all or run
256 collector : collector module
259 from basf2
import register_module
260 col = register_module(
'CDCDedxElectronCollector', cleanupCuts=
True)
261 if name ==
"tg" or name ==
"tgpre":
262 CollParam = {
'isRun':
True,
'isInjTime':
True,
'granularity':
'run'}
264 elif name ==
"cc" or name ==
"ce":
265 CollParam = {
'isCharge':
True,
'isCosth':
True,
'granularity': granularity}
269 CollParam = {
'isWire':
True,
'isDedxhit': isHit,
'isADCcorr':
not isHit,
'granularity': granularity}
272 CollParam = {
'isWire':
True,
'isDedxhit':
True,
'granularity': granularity}
275 CollParam = {
'isRun':
True,
'granularity':
'run'}
285 Create a rungain calibration algorithm.
287 algo : rungain algorithm
289 algo = CDCDedxRunGainAlgorithm()
290 algo.setMonitoringPlots(
True)
298 Create a injection time calibration algorithm.
300 algo : injection time algorithm
302 algo = CDCDedxInjectTimeAlgorithm()
303 algo.setMonitoringPlots(
True)
311 Create a cosine calibration algorithm.
313 algo : cosine algorithm
315 algo = CDCDedxCosineAlgorithm()
316 algo.setMonitoringPlots(
True)
324 Create a cosine edge calibration algorithm.
326 algo : cosine edge algorithm
328 algo = CDCDedxCosEdgeAlgorithm()
329 algo.setMonitoringPlots(
True)
337 Create a badwire calibration algorithm.
339 algo : badwire algorithm
341 algo = CDCDedxBadWireAlgorithm()
343 algo.setHighFracThres(0.2)
344 algo.setMeanThres(0.4)
345 algo.setRMSThres(0.4)
346 algo.setHistPars(150, 0, 5)
347 algo.setMonitoringPlots(
True)
355 Create a wire gain calibration algorithm.
357 algo : wiregain algorithm
359 algo = CDCDedxWireGainAlgorithm()
360 algo.enableExtraPlots(
True)
366 CDCDedxCalibration is a specialized calibration for cdcdedx.
375 collector_granularity='All'):
378 name: name of calibration
379 algorithims: algorithm of calibration
380 input_file_dict: input files list
381 max_iterations: maximum number of iterations
382 dependenices: depends on the previous calibration
383 collector_granularity: granularity : all or run
386 algorithms=algorithms
389 from caf.framework
import Collection
391 if name ==
"bd" or name ==
"wg":
393 input_files=input_file_dict[2],
394 pre_collector_path=pre_collector(name)
396 elif name ==
"cc" or name ==
"ce":
398 input_files=input_file_dict[1],
399 pre_collector_path=pre_collector(name)
403 input_files=input_file_dict[0],
404 pre_collector_path=pre_collector(name)
406 self.add_collection(name=name, collection=collection)
411 if dependencies
is not None:
412 for dep
in dependencies:
max_iterations
maximum iterations
def __init__(self, name, algorithms, input_file_dict, max_iterations=5, dependencies=None, collector_granularity='All')