Source code for km3dq_common.config_library
#! /usr/bin/env python
###############################################################################
# Definition of QAQC variable, defects, data-quality tags...
#
# Developer: Benjamin Trocme (benjamin.trocme at apc.in2p3.fr) - 2023
import tomli
import sys
from km3dq_common import config_path
#####################################
# Definition of the current detectors
###############################################################################
[docs]
def get_detx_caract(det):
""" Return the number of DOMs (incl base module so far) and PMts """
all_caract = {}
all_caract['D0ARCA028'] = {"doms": 504,
"pmts": 14341}
all_caract['D0ARCA021'] = {"doms": 378,
"pmts": 11718}
all_caract['D0ARCA020'] = {"doms": 342,
"pmts": 10602}
all_caract['D0ARCA009'] = {"doms": 162,
"pmts": 5022}
all_caract['D0ORCA023'] = {"doms": 414,
"pmts": 11698}
all_caract['D1ORCA019'] = {"doms": 342,
"pmts": 9643}
all_caract['D0ORCA018'] = {"doms": 324,
"pmts": 10044}
all_caract['D1ORCA015'] = {"doms": 270,
"pmts": 8370}
all_caract['D0ORCA015'] = {"doms": 270,
"pmts": 8370}
all_caract['D1ORCA011'] = {"doms": 198,
"pmts": 6138}
all_caract['D0ORCA011'] = {"doms": 198,
"pmts": 6138}
all_caract['D1ORCA013'] = {"doms": 234,
"pmts": 7254}
all_caract['D0ORCA007'] = {"doms": 126,
"pmts": 3906}
all_caract['D0ORCA010'] = {"doms": 180,
"pmts": 5580}
all_caract['D_ORCA006'] = {"doms": 108,
"pmts": 3348}
return all_caract[det]
###############################################################################
[docs]
def configure_var_name(v_prop):
""" Description of JQAQC variables """
# QAQC variables
v_prop['description'] = {"run": "Run",
"livetime": "Livetime (s)",
"nb_of_meta": "Nb of metadata",
"timestampdiff": "Missing time-slices",
"timestampdiff_r": ("Missing time-slices"
"/Run duration"),
"triggerrate": "Trigger rate(Hz)",
"hrv": "High rate veto bit",
"daq": "UDP packet-loss (DAQ)",
"whiterabbitstatus": "White rabbit status",
"fifo": "FIFO almost full",
"pmts": "Number of PMTs",
"pmts_norm": "Normalized number of PMTs",
"meanrate": "Mean rate (Hz)",
"rmsrate": "RMS rate (Hz)",
"hrv_fifo_failures": "Number of hrv/fifo failures",
"duplic_timeslices": "Number of duplicated timeslices",
"oos": "Out-of-sync",
"out_usync": "Micro out-of-sync",
"JDAQJTProc": "Triggered != Retriggered",
"acoustics": "Acoustic rate",
"ahrs_per_mn": ("Number of AHRS event per minute "
"in > 80% of DOMs"),
"jdaqevent": "Total",
"jtrigger3dshower": "3D shower trigger",
"jtriggermxshower": "MX shower trigger",
"jtrigger3dmuon": "3D Muon trigger "}
###############################################################################
[docs]
def configure_dataquality_tag(tag):
"""
TOMl configuration of the data quality tags
=== Arguments ===
- tag : tag name - [string] - Ex: "default", "neutrino2024"...
=== Output ===
- Dictionnary with all tag characteristics for all detectors
"""
####################################################
def all_dets_process(tag, toml_input, det_list):
# det_list: dictionnary with keys ARCA, ORCA
output = {}
for i_site in ["ARCA", "ORCA"]:
for i_det in det_list[i_site].split(" "):
if tag in toml_input.keys():
if i_det in toml_input[tag]: # Detector explicitly defined
output[i_det] = toml_input[tag][i_det]
else: # Use the site definition
output[i_det] = toml_input[tag][i_site]
else:
output[i_det] = "N/A"
return output
try:
with open(config_path("dataquality_tag"), mode="rb") as fconfig:
toml = tomli.load(fconfig)
dq_t = {}
# The path directory is the name of the data-quality tag
dq_t['dir_path'] = tag
# Retrieve the general descriptions
dq_t['name'] = toml['short_descr'][tag]
dq_t['descr'] = toml['descr'][tag]
dq_t['status'] = toml['status'][tag]
# Retrieve the dataset from the dataset.toml file
dq_t['dataset'] = toml['dataset'][tag]
except:
print("Problem in dataquality_tag.toml file")
sys.exit()
try:
with open(config_path("dataset"), mode="rb") as fconfig_ds:
toml_ds = tomli.load(fconfig_ds)
dq_t['dataset_descr'] = toml_ds['descr'][dq_t['dataset']]
# QAQC version - Detector dependant -> specific processing
dq_t['qaqc_vers'] = all_dets_process(dq_t['dataset'],
toml_ds['qaqc_vers'],
toml_ds['site_all_dets'][dq_t['dataset']])
# Extract the list of available detectors based on the
# definition of the QAQC-file versions and the site_all_dets variable
dq_t['det'] = {"ORCA": [],
"ARCA": []}
for i_det in dq_t['qaqc_vers'].keys():
if "ORCA" in i_det and i_det in toml_ds['site_all_dets'][dq_t['dataset']]['ORCA']:
dq_t['det']['ORCA'].append(i_det)
elif "ARCA" in i_det and i_det in toml_ds['site_all_dets'][dq_t['dataset']]['ARCA']:
dq_t['det']['ARCA'].append(i_det)
# Retrieve the defect tag
dq_t['def_tag'] = toml['def_tag'][tag]
except:
print("Problem in dataset configuration")
sys.exit()
# Retrieve the processing type/versions from the processing.toml file
# Detector dependant -> specific processing
try:
dq_t['proc'] = toml_ds['proc'][dq_t['dataset']]
with open(config_path("processing"), mode="rb") as fconfig_proc:
toml_proc = tomli.load(fconfig_proc)
dq_t['proc_descr'] = toml_proc['descr'][dq_t['proc']]
dq_t['proc_type'] = all_dets_process(dq_t['proc'],
toml_proc['proc_type'],
toml_ds['site_all_dets'][dq_t['dataset']])
dq_t['proc_version'] = all_dets_process(dq_t['proc'],
toml_proc['proc_version'],
toml_ds['site_all_dets'][dq_t['dataset']])
except:
print("Problem in processing configuration")
sys.exit()
# Retrieve the good run list definition
try:
dq_t['grl'] = toml['grl'][tag]
# Open the grl.toml file to retrieve the grl description
with open(config_path("grl"), mode="rb") as fconfig_grl:
toml_grl = tomli.load(fconfig_grl)
dq_t['grl_descr'] = toml_grl['descr'][dq_t['grl']]
# GRL names - If not defined, use the quasi_online one.
if dq_t['grl'] in toml_grl['grl_name'].keys():
dq_t['grl_name'] = toml_grl['grl_name'][dq_t['grl']]
else:
dq_t['grl_name'] = toml_grl['grl_name']['quasi_online']
except:
print("Problem in grl configuration")
sys.exit()
return dq_t
###############################################################################
[docs]
def configure_defect():
"""
TOMl configuration of the defects
"""
with open(config_path("defect"), mode="rb") as fconfig:
toml = tomli.load(fconfig)
defect_0 = {}
defect_0['descr'] = toml['descr']
defect_0['type'] = list(defect_0['descr'].keys())
defect_0['bit'] = toml['bit']
return defect_0
###############################################################################
[docs]
def configure_def_var_name(v_prop):
""" Description of defect variables """
# QAQC variables
v_prop['description']["def_daq"] = "DAQ defects"
v_prop['description']["def_operation"] = "Operation defects"
# v_prop['description']["def_oos"] = "Out-of-sync defects"
v_prop['description']["def_data_corruption"] = "Data corruption defects"
v_prop['description']["def_calibration"] = "Calibration defects"
v_prop['description']["def_data_processing"] = "Data processing defects"
v_prop['description']["def_analysis"] = "Analysis defects"
v_prop['description']["def_high_level_analysis"] = "Neutrino analysis defects"
v_prop['description']["def_signoff"] = "Missing signoff"
###############################################################################
[docs]
def configure_fact():
"""
TOMl configuration of the facts
"""
with open(config_path("fact"), mode="rb") as fconfig:
toml = tomli.load(fconfig)
fact_0 = {}
fact_0['type'] = toml['type']
return fact_0
###############################################################################
[docs]
def configure_det_fact():
"""
TOMl configuration of the detector facts
"""
with open(config_path("det_fact"), mode="rb") as fconfig:
toml = tomli.load(fconfig)
det_fact_0 = {}
det_fact_0['type'] = toml['type']
det_fact_0['status'] = toml['status']
return det_fact_0
###############################################################################
[docs]
def configure_var_unit(v_prop):
""" Description of variables """
v_prop['unit'] = {"livetime": "s",
"timestampdiff": "s",
"triggerrate": "Hz",
"meanrate": "Hz",
"rmsrate": "Hz"}
###############################################################################
[docs]
def configure_var_bit(v_prop):
"""
Bit of variables contributing to the veto//Qscore degradation
used to store in the TTree
"""
v_prop['bit'] = {"livetime": 0,
"timestampdiff": 1,
"timestampdiff_r": 2,
"triggerrate": 3,
"hrv": 4,
"daq": 5,
"whiterabbitstatus": 6,
"fifo": 7,
"pmts": 8,
"pmts_norm": 9,
"meanrate": 10,
"rmsrate": 11,
"oos": 12,
"out_usync": 13,
"JDAQJTProc": 14,
"acoustics": 15,
"ahrs_per_mn": 16,
"nb_of_meta": 17,
"hrv_fifo_failures": 18,
"duplic_timeslices": 19,
"def_daq": 22,
"def_operation": 23,
# "def_oos": 24,
"def_data_corruption": 25,
"def_calibration": 26,
"def_data_processing": 27,
"def_analysis": 28,
"def_high_level_analysis": 29,
"def_signoff": 30}
###############################################################################
[docs]
def configure_var_thresholds(v_prop, det, dqt):
"""
Configure various variables for the analysis of the TTree derived from
the JQAQC file
TOMl configuration
"""
with open(config_path("grl"), mode="rb") as fconfig:
toml = tomli.load(fconfig)
# First fill the thresholds with the quasi_online ones
# It is mandatory as all thresholds are derived from them
v_prop['veto_thresholds'] = dict(toml['veto_thresholds']['quasi_online'])
v_prop['qsco_thresholds'] = dict(toml['qsco_thresholds']['quasi_online'])
v_prop['good_thresholds'] = dict(toml['good_thresholds']['quasi_online'])
v_prop['poor_thresholds'] = dict(toml['poor_thresholds']['quasi_online'])
v_prop['bad_thresholds'] = dict(toml['bad_thresholds']['quasi_online'])
v_prop['discard_thresholds'] = dict(toml['discard_thresholds']['quasi_online'])
v_prop['y_range_display'] = dict(toml['y_range_display']['quasi_online'])
# If non quasi_online GRL, update the thresholds with the config file
for i_type in ("veto_thresholds", "qsco_thresholds",
"good_thresholds", "poor_thresholds",
"bad_thresholds", "discard_thresholds",
"y_range_display"):
if dqt['grl'] != "quasi_online" and dqt['grl'] in toml[i_type].keys():
for i_modif in toml[i_type][dqt['grl']]:
# If empty list, remove the thresholds defined in quasi_online dqt['grl']
if len(toml[i_type][dqt['grl']][i_modif]) == 0:
v_prop[i_type].pop(i_modif)
continue
# Otherwise update them
v_prop[i_type][i_modif] = \
toml[i_type][dqt['grl']][i_modif]
# DUMMY detector used only by the get_grl_html_descr function
# used by km3dq_perf/generatewww.py - No impact on perf
if det == "DUMMY":
return
# Special cases impossible to handle with TOML
if "v16.0.3" in dqt["qaqc_vers"][det]:
if "out_usync" in v_prop['veto_thresholds']:
v_prop['veto_thresholds'].pop("out_usync")
if "out_usync" in v_prop['qsco_thresholds']:
v_prop['qsco_thresholds'].pop("out_usync")
if any(("v16.0.3" in dqt["qaqc_vers"][det],
"v17.3.2" in dqt["qaqc_vers"][det])):
for i_missing in ["nb_of_meta", "hrv_fifo_failures", "duplic_timeslices"]:
if i_missing in v_prop['veto_thresholds']:
v_prop['veto_thresholds'].pop(i_missing)
if i_missing in v_prop['qsco_thresholds']:
v_prop['qsco_thresholds'].pop(i_missing)
# Display options
v_prop['y_range_display']['pmts'][1] = get_detx_caract(det)['pmts']*1.2
if "ORCA" in det: # Increased range for bioluminescence
v_prop['y_range_display']["hrv"] = [-0.1, 0.8]
###############################################################################
[docs]
def get_grl_html_descr(dq_tag_0):
"""
Returns the grl description in an html for the summary webpage
The grl may depend on the detector due to missing variables. This is
by-passed by passing a dummy value to the configure_var_thresholds
function. In this way, no variable is discarded
"""
var_prop = {}
configure_var_thresholds(var_prop, "DUMMY", dq_tag_0)
configure_var_unit(var_prop)
html_descr = {}
for i_key in var_prop.keys():
# One assumes that the QQC variables arrive before the defects
# in the configuration file
html_descr[i_key] = "<b>QAQC variables</b><br>"
defect_found = False
for i_var in var_prop[i_key].keys():
# qaqc variable
if any((isinstance(var_prop[i_key][i_var][0], int),
isinstance(var_prop[i_key][i_var][0], float))):
if var_prop[i_key][i_var][0] == var_prop[i_key][i_var][1]:
thresh = f"{var_prop[i_key][i_var][0]}"
if i_var == "veto":
thresh = f"{var_prop[i_key][i_var][0] == 1}"
else:
thresh = (f"{var_prop[i_key][i_var][0]} — "
f"{var_prop[i_key][i_var][1]}")
if i_var in var_prop['unit'].keys():
thresh += f" {var_prop['unit'][i_var]}"
html_descr[i_key] += (f"{i_var}: {thresh}")
# defect
elif isinstance(var_prop[i_key][i_var][0], str):
if defect_found is False:
html_descr[i_key] += "<hr><b>Defects</b><br>"
defect_found = True
html_descr[i_key] += (f"{i_var.replace('def_', '')}: "
f"{var_prop[i_key][i_var][0]}")
for j in range(1, len(var_prop[i_key][i_var])):
html_descr[i_key] += (f", {var_prop[i_key][i_var][j]}")
html_descr[i_key] += ("<br>")
return html_descr