Skip to content
Snippets Groups Projects

[AGIPD][CORRECT] Use calcat_interface and remove precorrection notebook

Merged [AGIPD][CORRECT] Use calcat_interface and remove precorrection notebook
AGIPD_calcat_interface into master
Merged Karim Ahmed requested to merge AGIPD_calcat_interface into master
Viewing commit 1defc617
Show latest version
1 file
+ 58
− 78
Compare changes
  • Side-by-side
  • Inline
notebooks/Gotthard2/Gotthard2_retrieve_constants_precorrection_NBC.ipynb
+ 58
− 78
%% Cell type:markdown id: tags:
# GOTTHARD2 Retrieving Constants Pre-correction #
Author: European XFEL Detector Group, Version: 1.0
Retrieving Required Constants for Offline Calibration of the Gotthard2 Detector
%% Cell type:code id: tags:
``` python
in_folder = "/gpfs/exfel/exp/FXE/202221/p003225/raw" # the folder to read data from, required
out_folder = "/gpfs/exfel/data/scratch/ahmedk/test/gotthard2" # the folder to output to, required
metadata_folder = "" # Directory containing calibration_metadata.yml when run by xfel-calibrate
run = 50 # run to process, required
# Parameters used to access raw data.
karabo_id = "FXE_XAD_G2XES" # karabo prefix of Gotthard-II devices
karabo_da = ["GH201"] # data aggregators
receiver_template = "RECEIVER" # receiver template used to read INSTRUMENT keys.
control_template = "CONTROL" # control template used to read CONTROL keys.
instrument_source_template = "{}/DET/{}:daqOutput" # template for source name (filled with karabo_id & receiver_id). e.g. 'SPB_IRDA_JF4M/DET/JNGFR01:daqOutput'
ctrl_source_template = "{}/DET/{}" # template for control source name (filled with karabo_id_control)
karabo_id_control = "" # Control karabo ID. Set to empty string to use the karabo-id
# Parameters for calibration database.
use_dir_creation_date = True # use the creation data of the input dir for database queries.
cal_db_interface = "tcp://max-exfl017:8017#8025" # the database interface to use.
cal_db_interface = "tcp://max-exfl016:8017#8025" # the database interface to use.
cal_db_timeout = 180000 # timeout on caldb requests.
overwrite_creation_time = "2022-06-28 13:00:00.00" # To overwrite the measured creation_time. Required Format: YYYY-MM-DD HR:MN:SC.00 e.g. "2022-06-28 13:00:00.00"
creation_time = "" # To overwrite the measured creation_time. Required Format: YYYY-MM-DD HR:MN:SC e.g. "2022-06-28 13:00:00"
# Parameters affecting corrected data.
constants_file = ""#/gpfs/exfel/data/scratch/ahmedk/dont_remove/gotthard2/constants/calibration_constants_GH2.h5" # Retrieve constants from local.
constants_file = "" # /gpfs/exfel/data/scratch/ahmedk/dont_remove/gotthard2/constants/calibration_constants_GH2.h5" # Retrieve constants from local.
offset_correction = True # apply offset correction. This can be disabled to only apply LUT or apply LUT and gain correction for non-linear differential results.
gain_correction = True # apply gain correction.
# Parameter conditions.
bias_voltage = -1 # Detector bias voltage, set to -1 to use value in raw file.
exposure_time = -1. # Detector exposure time, set to -1 to use value in raw file.
exposure_period = -1. # Detector exposure period, set to -1 to use value in raw file.
acquisition_rate = 1.1 # Detector acquisition rate (1.1/4.5), set to -1 to use value in raw file.
single_photon = 0 # Detector single photon mode (High/Low CDS), set to -1 to use value in raw file.
acquisition_rate = -1. # Detector acquisition rate (1.1/4.5), set to -1 to use value in raw file.
single_photon = -1 # Detector single photon mode (High/Low CDS), set to -1 to use value in raw file.
if constants_file:
print("Skipping constant retrieval. Specified constants_file is used.")
import sys
sys.exit(0)
```
%% Cell type:code id: tags:
``` python
import datetime
from functools import partial
from logging import warning
import multiprocessing
from extra_data import RunDirectory
from pathlib import Path
from cal_tools.calcat_interface import GOTTHARD2_CalibrationData
from cal_tools.gotthard2 import gotthard2lib
from cal_tools.tools import (
get_dir_creation_date,
calcat_creation_time,
get_from_db,
CalibrationMetadata,
)
from iCalibrationDB import Conditions, Constants
```
%% Cell type:code id: tags:
``` python
in_folder = Path(in_folder)
run_folder = in_folder / f"r{run:04d}"
out_folder = Path(out_folder)
out_folder.mkdir(parents=True, exist_ok=True)
metadata = CalibrationMetadata(metadata_folder or out_folder)
# Constant paths are saved under retrieved-constants in calibration_metadata.yml
retrieved_constants = metadata.setdefault("retrieved-constants", {})
if not karabo_id_control:
karabo_id_control = karabo_id
instrument_src = instrument_source_template.format(karabo_id, receiver_template)
ctrl_src = ctrl_source_template.format(karabo_id_control, control_template)
print(f"Retrieve constants for modules: {karabo_da} for run {run}")
creation_time = None
if overwrite_creation_time:
creation_time = datetime.datetime.strptime(
overwrite_creation_time, "%Y-%m-%d %H:%M:%S.%f"
)
elif use_dir_creation_date:
creation_time = get_dir_creation_date(in_folder, run)
print(f"Using {creation_time} as creation time")
# Run's creation time:
creation_time = calcat_creation_time(in_folder, run, creation_time)
print(f"Creation time: {creation_time}")
```
%% Cell type:code id: tags:
``` python
# Read slow data
run_dc = RunDirectory(run_folder)
g2ctrl = gotthard2lib.Gotthard2Ctrl(run_dc=run_dc, ctrl_src=ctrl_src)
if bias_voltage == -1:
bias_voltage = g2ctrl.get_bias_voltage()
if exposure_time == -1:
exposure_time = g2ctrl.get_exposure_time()
if exposure_period == -1:
exposure_period = g2ctrl.get_exposure_period()
if acquisition_rate == -1:
acquisition_rate = g2ctrl.get_acquisition_rate()
if single_photon == -1:
single_photon = g2ctrl.get_single_photon()
print("Bias Voltage:", bias_voltage)
print("Exposure Time:", exposure_time)
print("Exposure Period:", exposure_period)
print("Acquisition Rate:", acquisition_rate)
print("Single Photon:", single_photon)
```
%% Cell type:code id: tags:
``` python
condition = Conditions.Dark.Gotthard2(
bias_voltage=bias_voltage,
g2_cal = GOTTHARD2_CalibrationData(
detector_name=karabo_id,
sensor_bias_voltage=bias_voltage,
exposure_time=exposure_time,
exposure_period=exposure_period,
single_photon=single_photon,
acquisition_rate=acquisition_rate,
single_photon=single_photon,
event_at=creation_time,
snapshot_at=creation_time,
)
def get_constants_for_module(mod: str):
"""Get calibration constants for given module for Gotthard2."""
retrieval_function = partial(
get_from_db,
karabo_id=karabo_id,
karabo_da=mod,
cal_db_interface=cal_db_interface,
creation_time=creation_time,
timeout=cal_db_timeout,
verbosity=1,
meta_only=True,
load_data=False,
empty_constant=None
)
mdata_dict = dict()
mdata_dict["constants"] = dict()
constants = [
"LUT", "Offset", "BadPixelsDark",
"RelativeGain", "BadPixelsFF",
]
for cname in constants:
mdata_dict["constants"][cname] = dict()
if not gain_correction and cname in ["BadPixelsFF", "RelativeGain"]:
continue
_, mdata = retrieval_function(
condition=condition,
constant=getattr(Constants.Gotthard2, cname)(),
)
mdata_const = mdata.calibration_constant_version
const_mdata = mdata_dict["constants"][cname]
# check if constant was successfully retrieved.
if mdata.comm_db_success:
const_mdata["file-path"] = (
f"{mdata_const.hdf5path}" f"{mdata_const.filename}"
)
const_mdata["dataset-name"] = mdata_const.h5path
const_mdata["creation-time"] = f"{mdata_const.begin_at}"
mdata_dict["physical-detector-unit"] = mdata_const.device_name
else:
const_mdata["file-path"] = None
const_mdata["creation-time"] = None
return mdata_dict, mod
mdata_dict = {"constants": dict()}
with multiprocessing.Pool() as pool:
results = pool.map(get_constants_for_module, karabo_da)
constant_names = ["LUTGotthard2", "OffsetGotthard2", "BadPixelsDarkGotthard2"]
if gain_correction:
constant_names += ["RelativeGainGotthard2", "BadPixelsFFGotthard2"]
# Retrieve metadata for all pnccd constants.
g2_metadata = g2_cal.metadata(constant_names)
# Validate the constants availability and raise/warn correspondingly.
for mod, ccv_dict in g2_metadata.items():
missing_dark_constants = set(
c for c in ["LUTGotthard2", "OffsetGotthard2", "BadPixelsDarkGotthard2"] if c not in ccv_dict.keys())
missing_gain_constants = set(
c for c in ["BadPixelsFF10Hz", "RelativeGain10Hz"] if gain_correction and c not in ccv_dict.keys()) # noqa
if missing_dark_constants:
raise KeyError(
f"Dark constants {missing_dark_constants} are not available for correction. Mod: {mod}")
if gain_correction and missing_gain_constants:
warning(f"Gain constants {missing_gain_constants} are not retrieved. Mod {mod}")
# Add constants metadata in retrieved_constants dict.
for mod, ccv_dict in g2_metadata.items():
mod_dict = retrieved_constants.setdefault(mod, dict())
const_dict = mod_dict.setdefault("constants", dict())
for cname, ccv_metadata in ccv_dict.items():
const_dict[cname] = {
"path": str(g2_cal.caldb_root / ccv_metadata["path"]),
"dataset": ccv_metadata["dataset"],
"creation-time": ccv_metadata["begin_validity_at"],
}
mod_dict["physical-name"] = ccv_metadata["physical_name"]
print(f"Stored retrieved constants in {metadata.filename}")
```
%% Cell type:code id: tags:
``` python
# Constant paths are saved under retrieved-constants in calibration_metadata.yml
retrieved_constants = metadata.setdefault("retrieved-constants", {})
timestamps = dict()
for md_dict, mod in results:
if mod in retrieved_constants:
print(f"Constant for {mod} already in calibration_metadata.yml, won't query again.") # noqa
continue
retrieved_constants[mod] = md_dict
for mod in karabo_da:
module_timestamps = timestamps[mod] = dict()
module_constants = retrieved_constants[mod]
print(f"Module: {mod}:")
for cname, mdata in md_dict["constants"].items():
for cname, mdata in module_constants["constants"].items():
if hasattr(mdata["creation-time"], 'strftime'):
mdata["creation-time"] = mdata["creation-time"].strftime('%y-%m-%d %H:%M')
print(f'{cname:.<12s}', mdata["creation-time"])
for cname in ["Offset", "BadPixelsDark", "RelativeGain", "BadPixelsFF"]:
if cname in md_dict["constants"]:
module_timestamps[cname] = md_dict["constants"][cname]["creation-time"]
if cname in module_constants["constants"]:
module_timestamps[cname] = module_constants["constants"][cname]["creation-time"]
else:
module_timestamps[cname] = "NA"
time_summary = retrieved_constants.setdefault("time-summary", {})
time_summary["SAll"] = timestamps
metadata.save()
```

Legal Notice | Contacts