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
import urllib.request
from km3dq_common import config_path
###############################################################################
[docs]
def get_detx_caract(det, source="sftp"):
"""Return the number of DOMs (incl base module so far) and PMts"""
try:
if source == "sftp":
with urllib.request.urlopen(
"https://sftp.km3net.de/data/km3dq_lw_db/Common/km3dq_perf_aux.toml"
) as s_f:
toml = tomli.loads(s_f.read().decode("utf-8"))
else:
with open(
"/pbs/home/t/trocme/KM3Net/infrastructure/km3dq_lw_db/Common/km3dq_perf_aux.toml",
mode="rb",
) as fconfig:
toml = tomli.load(fconfig)
except:
print("Problem in km3dq_perf_aux.toml file")
sys.exit()
return toml["hardware_charact"][det]
###############################################################################
[docs]
def configure_var_name(v_prop):
"""Description of JQAQC variables"""
# QAQC variables
v_prop["description"] = {
"run": "Run",
"livetime_s": "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": "Fraction of data without UDP packet-loss",
"WR": "White rabbit status",
"FIFO": "FIFO almost full",
"PMTs": "Number of active PMTs",
"pmts_norm": "Fraction of active PMTs",
"MEAN_Rate_Hz": "Mean rate (Hz)",
"RMS_Rate_Hz": "RMS rate (Hz)",
"hrv_fifo_failures": "Number of hrv/fifo failures",
"duplic_timeslices": "Number of duplicated timeslices",
"out_sync": "Out-of-sync",
"out_usync": "Micro out-of-sync",
"JDAQJTProc": "Triggered != Retriggered",
"Acoustics": "Number of acoustic events",
"Acoustics_per_mn": "Number of acoustic events per minute",
"zero_AHRS": ("Number of DOMs without valid " "AHRS data"),
"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, source="sftp"):
"""
TOMl configuration of the data quality tags
=== Arguments ===
- tag : tag name - [string] - Ex: "default", "calibration"...
=== Output ===
- Dictionnary with all tag characteristics for all detectors
"""
####################################################
def single_property_process(tag, tag_parent, toml_input):
output = {}
for i_tag in (tag_parent, tag):
if i_tag in toml_input.keys():
output = toml_input[i_tag]
return output
####################################################
def all_site_process(tag, tag_parent, toml_input):
output = {}
for i_tag in (tag_parent, tag):
if i_tag == "": # No tag_parent
continue
for i_site in ["ARCA", "ORCA"]:
if i_site in toml_input[i_tag].keys():
output[i_site] = toml_input[i_tag][i_site]
return output
####################################################
def all_dets_process(tag, tag_parent, toml_input, det_list):
# det_list: detector list
output = {}
for i_tag in (tag_parent, tag):
if i_tag == "": # No tag_parent
continue
for i_site in ["ARCA", "ORCA"]:
for i_det in det_list[i_site]:
if i_tag in toml_input.keys():
if i_det in toml_input[i_tag]: # Detector explicitly defined
output[i_det] = toml_input[i_tag][i_det]
elif i_site in toml_input[i_tag]: # Use the site definition
output[i_det] = toml_input[i_tag][i_site]
return output
####################################################
# First retrieve the general caracteristics:
# status, description, dataset name, grl...
try:
if source == "sftp":
with urllib.request.urlopen(
"https://sftp.km3net.de/data/km3dq_lw_db/Common/dataquality_tag.toml"
) as s_f:
# tmp = s_f.read()
toml = tomli.loads(s_f.read().decode("utf-8"))
else:
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]
# Dataset name
dq_t["dataset"] = toml["dataset"][tag]
dq_t["grl"] = toml["grl"][tag]
dq_t["def_tag"] = toml["def_tag"][tag]
except:
print("Problem in dataquality_tag.toml file")
sys.exit()
# Retrieve the dataset description from the dataset.toml file
try:
if source == "sftp":
with urllib.request.urlopen(
"https://sftp.km3net.de/data/km3dq_lw_db/Common/dataset.toml"
) as s_f:
# tmp = s_f.read()
toml_ds = tomli.loads(s_f.read().decode("utf-8"))
else:
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"]]
dq_t["dataset_parent"] = toml_ds["parent"][dq_t["dataset"]]
# Name of processing type / versions
dq_t["proc"] = single_property_process(
dq_t["dataset"], dq_t["dataset_parent"], toml_ds["proc"]
)
# List of detectors per site - Single string
dq_t["det"] = all_site_process(
dq_t["dataset"], dq_t["dataset_parent"], toml_ds["site_all_dets"]
)
for i_site in dq_t["det"]:
if dq_t["det"][i_site] != "":
dq_t["det"][i_site] = dq_t["det"][i_site].split(" ")
else:
dq_t["det"][i_site] = []
# Run type - Detector dependant -> specific processing
# Empty for v16 and v17 as did not contribute to the naming of file
dq_t["run_type"] = all_dets_process(
dq_t["dataset"], dq_t["dataset_parent"], toml_ds["run_type"], dq_t["det"]
)
# QAQC version - Detector dependant -> specific processing
dq_t["qaqc_vers"] = all_dets_process(
dq_t["dataset"], dq_t["dataset_parent"], toml_ds["qaqc_vers"], dq_t["det"]
)
# Compile the run type and qaqc version to get the filename
dq_t["qaqc_name"] = {}
for i_det in dq_t["run_type"]:
dq_t["qaqc_name"][
i_det
] = f"{dq_t['qaqc_vers'][i_det]}_{dq_t['run_type'][i_det]}"
# Run range - Detector dependant -> specific processing
dq_t["run_range"] = all_dets_process(
dq_t["dataset"], dq_t["dataset_parent"], toml_ds["run_range"], dq_t["det"]
)
except:
print("Problem in dataset configuration")
sys.exit()
# Retrieve the processing type/versions from the processing.toml file
# Detector dependant -> specific processing defined in dataset
try:
if source == "sftp":
with urllib.request.urlopen(
"https://sftp.km3net.de/data/km3dq_lw_db/Common/processing.toml"
) as s_f:
tmp = s_f.read()
toml_proc = tomli.loads(tmp.decode("utf-8"))
else:
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"], dq_t["det"]
)
dq_t["proc_version"] = all_dets_process(
dq_t["proc"], "", toml_proc["proc_version"], dq_t["det"]
)
except:
sys.exit()
# Retrieve the good run list definition (name and definition)
# The threshold retrieval is completed in the script configure_var_thresholds
try:
# Open the grl.toml file to retrieve the grl description
if source == "sftp":
with urllib.request.urlopen(
"https://sftp.km3net.de/data/km3dq_lw_db/Common/grl.toml"
) as s_f:
# tmp = s_f.read()
toml_grl = tomli.loads(s_f.read().decode("utf-8"))
else:
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_tag_parent = toml_grl["parent"][dq_t["grl"]]
if grl_tag_parent == "":
dq_t["grl_name"] = toml_grl["grl_name"][dq_t["grl"]]
else:
dq_t["grl_name"] = toml_grl["grl_name"][grl_tag_parent]
for i_name in dq_t["grl_name"].keys():
if i_name in toml_grl["grl_name"][dq_t["grl"]].keys():
dq_t["grl_name"][i_name] = toml_grl["grl_name"][dq_t["grl"]][i_name]
except:
print("Problem in grl configuration")
sys.exit()
return dq_t
###############################################################################
[docs]
def configure_defect(source="sftp"):
"""
TOMl configuration of the defects
"""
if source == "sftp":
with urllib.request.urlopen(
"https://sftp.km3net.de/data/km3dq_lw_db/Common/defect.toml"
) as s_f:
# tmp = s_f.read()
toml = tomli.loads(s_f.read().decode("utf-8"))
else:
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_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(source="sftp"):
"""
TOMl configuration of the facts
"""
if source == "sftp":
with urllib.request.urlopen(
"https://sftp.km3net.de/data/km3dq_lw_db/Common/fact.toml"
) as s_f:
# tmp = s_f.read()
toml = tomli.loads(s_f.read().decode("utf-8"))
else:
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(source="sftp"):
"""
TOMl configuration of the detector facts
"""
if source == "sftp":
with urllib.request.urlopen(
"https://sftp.km3net.de/data/km3dq_lw_db/Common/det_fact.toml"
) as s_f:
# tmp = s_f.read()
toml = tomli.loads(s_f.read().decode("utf-8"))
else:
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_s": 0,
"timestampdiff": 1,
"timestampdiff_r": 2,
"triggerrate": 3,
"HRV": 4,
"DAQ": 5,
"WR": 6,
"FIFO": 7,
"PMTs": 8,
"pmts_norm": 9,
"MEAN_Rate_Hz": 10,
"RMS_Rate_Hz": 11,
"out_sync": 12,
"out_usync": 13,
"JDAQJTProc": 14,
"Acoustics": 15,
"zero_AHRS": 16,
"ahrs_per_mn": 17,
"nb_of_meta": 18,
"hrv_fifo_failures": 19,
"duplic_timeslices": 20,
"def_daq": 23,
"def_operation": 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, source="sftp"):
"""
Configure various variables for the analysis of the TTree derived from
the JQAQC file
TOMl configuration
"""
if source == "sftp":
with urllib.request.urlopen(
"https://sftp.km3net.de/data/km3dq_lw_db/Common/grl.toml"
) as s_f:
toml_grl = tomli.loads(s_f.read().decode("utf-8"))
with urllib.request.urlopen(
"https://sftp.km3net.de/data/km3dq_lw_db/Common/km3dq_perf_aux.toml"
) as s_f:
toml_perf_aux = tomli.loads(s_f.read().decode("utf-8"))
else:
with open(config_path("grl"), mode="rb") as fconfig:
toml = tomli.load(fconfig)
grl_tag_parent = toml_grl["parent"][dqt["grl"]]
if grl_tag_parent == "":
v_prop["veto_thresholds"] = dict(toml_grl["veto_thresholds"][dqt["grl"]])
v_prop["qsco_thresholds"] = dict(toml_grl["qsco_thresholds"][dqt["grl"]])
v_prop["good_thresholds"] = dict(toml_grl["good_thresholds"][dqt["grl"]])
v_prop["poor_thresholds"] = dict(toml_grl["poor_thresholds"][dqt["grl"]])
v_prop["bad_thresholds"] = dict(toml_grl["bad_thresholds"][dqt["grl"]])
v_prop["discard_thresholds"] = dict(toml_grl["discard_thresholds"][dqt["grl"]])
else:
v_prop["veto_thresholds"] = dict(toml_grl["veto_thresholds"][grl_tag_parent])
v_prop["qsco_thresholds"] = dict(toml_grl["qsco_thresholds"][grl_tag_parent])
v_prop["good_thresholds"] = dict(toml_grl["good_thresholds"][grl_tag_parent])
v_prop["poor_thresholds"] = dict(toml_grl["poor_thresholds"][grl_tag_parent])
v_prop["bad_thresholds"] = dict(toml_grl["bad_thresholds"][grl_tag_parent])
v_prop["discard_thresholds"] = dict(
toml_grl["discard_thresholds"][grl_tag_parent]
)
for i_type in (
"veto_thresholds",
"qsco_thresholds",
"good_thresholds",
"poor_thresholds",
"bad_thresholds",
"discard_thresholds",
):
if dqt["grl"] in toml_grl[i_type].keys():
for i_modif in toml_grl[i_type][dqt["grl"]]:
# If empty list, remove the thresholds defined in quasi_online dqt['grl']
if len(toml_grl[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_grl[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
# Detector dependant treatment
try:
# Special cases impossible to handle with TOML
# Variables missing in realease 16
if dqt["qaqc_vers"][det] == "v16.0.3":
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")
# Variables missing in realease 16 and 17
if dqt["qaqc_vers"][det] in ("v16.0.3", "v17.3.2"):
for i_missing in [
"nb_of_meta",
"hrv_fifo_failures",
"duplic_timeslices",
"mean_AHRS",
"zero_AHRS",
]:
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 - Not DQ tag dependant (as of today but could be changed)
v_prop["basic_plots_options"] = dict(
toml_perf_aux["basic_plots_options"]["quasi_online"]
)
v_prop["basic_plots_y_range_display"] = dict(
toml_perf_aux["basic_plots_y_range_display"]["quasi_online"]
)
v_prop["basic_plots_y_range_display"]["PMTs"][1] = (
get_detx_caract(det)["pmts"] * 1.2
)
if "ORCA" in det: # Increased range for bioluminescence
v_prop["basic_plots_y_range_display"]["HRV"] = [-0.1, 0.8]
except KeyError:
print(f"config_library.py: Missing DQ-tag information for the detector {det}")
###############################################################################
[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