Skip to content
Snippets Groups Projects
Commit e828b015 authored by Karim Ahmed's avatar Karim Ahmed
Browse files

add LPD_Mini_Inject_calibration_constants_from_h5files notebook

parent c1552168
No related branches found
No related tags found
1 merge request!848[LPDMini] Feat: Inject gain constants notebook
Hide whitespace changes
Inline Side-by-side
notebooks/LPDMini/LPD_Mini_Inject_calibration_constants_from_h5files.ipynb 0 → 100644
+ 261
− 0
View file @ e828b015
%% Cell type:markdown id: tags:
# Injecting LPD-Mini calibration constant data to the database
Author: European XFEL Detector Group, Version: 1.0
Reading h5files of LPD Mini's calibration constants to inject them to the database.
%% Cell type:code id: tags:
``` python
# calibration constant parameters:
constant_names = [""] # calibration constant name, required.
in_folder = "" # calibration constants folder, required.
out_folder = "" # output folder to store report path in case the notebook is executed by CLI, required.
proposal = "" # Add proposal number to be sent to the database as a part of Raw data location.
runs = [""] # Add list of runs to be sent to the database as a part of Raw data location.
# detector parameters:
karabo_id = "FXE_DET_LPD_MINI-1" # detector identifier.
karabo_da = ["all"] # karabo data aggregators. default "all" for all 8 karabo data aggregator names.
# calibration database parameters:
cal_db_interface = "tcp://max-exfl016:8015#8045" # calibration DB zmq address.
# calibration constant conditions:
memory_cells = 512 # Number of memory cells. Used for constant conditions.
bias_voltage_0 = -1 # bias voltage for minis 1, 3, 5, 7; Setting -1 will read the value from files
bias_voltage_1 = -1 # bias voltage for minis 2, 4, 6, 8; Setting -1 will read the value from files
capacitor = 5 # capacitor value. Used for constant conditions.
photon_energy = 9.2 # calibration constant photon energy. Used for constant conditions.
creation_time = '2020-01-20T14:12:06' # creation time for the injected constants. required format '2019-01-20T14:12:06'
```
%% Cell type:code id: tags:
``` python
import multiprocessing
from datetime import datetime
from logging import warning
from pathlib import Path
from typing import List, Tuple
import h5py # noqa
from cal_tools.tools import get_pdu_from_db, get_report, send_to_db
from iCalibrationDB import Conditions, Constants
```
%% Cell type:code id: tags:
``` python
pixels_x = 256
pixels_y = 32
category = 0 # Bad condition.
if karabo_da[0].lower() == "all":
karabo_da = [f"LPDMINI00/{i:02d}" for i in range(1, 9)]
# if proposal or runs are given assign file_loc
# for calibration constant versions metadata.
file_loc = ""
if proposal:
file_loc += f"proposal:{proposal}"
if runs[0] != "":
file_loc += f"runs: {runs}"
if file_loc == "":
print(
"No proposal or runs were given for constant source."
" No \"Raw data location\" will be injected with the constants.\n"
)
```
%% Cell type:code id: tags:
``` python
def validate_input_paths(
in_folder: str,
creation_time: str,
constant_names: List[str],
) -> Tuple[str, datetime]:
# Validate input parameters:
if not (in_folder):
raise ValueError(
"ERROR: \"in_folder\" is not given."
" Please provide the constants input folder."
)
c_folder = Path(in_folder)
if not c_folder.is_dir():
raise ValueError(
f"ERROR: in_folder {in_folder} directory doesn't exist."
)
try:
creation_time = datetime.strptime(creation_time, '%Y-%m-%dT%H:%M:%S')
except ValueError:
raise ValueError(
"Incorrect data format, "
"should be YYYY-MM-DDTHH:MM:SS i.e. 2019-01-20T14:12:06"
)
for constant in constant_names:
if not hasattr(Constants.LPD, constant):
raise ValueError(
f"ERROR: Constant name \"{constant}\" is not a known LPD constant. "
f"Available LPD Constants are {[c for c in dir(Constants.LPD) if c[0] != '_']}"
)
return c_folder, creation_time
```
%% Cell type:code id: tags:
``` python
def inject_constants(
constant_name: str,
mod_da: str,
physical_unit: str,
):
mod_num = int(mod_da.split('/')[-1])
# mod_num is from 1 to 8, so b_v_0 applies to odd numbers
bias_voltage = bias_voltage_0 if mod_num % 2 == 1 else bias_voltage_1
# Calibration constants condition object.
condition = Conditions.Illuminated.LPD(
memory_cells=memory_cells,
bias_voltage=bias_voltage,
photon_energy=photon_energy,
pixels_x=pixels_x,
pixels_y=pixels_y,
capacitor=capacitor,
category=category,
)
constant = getattr(Constants.LPD, constant_name)()
cfile = c_folder / f"{constant_name}_{mod_da.replace('/', '_')}.h5"
if not cfile.exists():
warning(f"Constant file {cfile} doesn't exist.\n")
return
# load constant data.
with h5py.File(cfile, "r") as f:
cdata = f[constant_name][()]
# Validate for only LPD at the moment.
if not cdata.shape == (memory_cells, pixels_y, pixels_x, 3):
raise ValueError(
f"ERROR: {const} constant data shape is not as expected."
f" {cdata.shape} != ({memory_cells}, {pixels_y}, {pixels_x}, 3).\n"
)
constant.data = cdata
send_to_db(
db_module=physical_unit,
karabo_id=karabo_id,
constant=constant,
condition=condition,
file_loc=file_loc,
report_path=report_path,
cal_db_interface=cal_db_interface,
creation_time=creation_time,
)
```
%% Cell type:code id: tags:
``` python
c_folder, creation_time = validate_input_paths(
in_folder,
creation_time,
constant_names,
)
# create a report path for calibration constant versions metadata.
report_name = f"No_report/LPD_{datetime.now().strftime('%Y-%m-%dT%H:%M:%S')}"
report_path = get_report(
out_folder=in_folder,
default_path=report_name
)
# Calibration constants condition object.
condition = Conditions.Illuminated.LPD(
memory_cells=memory_cells,
bias_voltage=bias_voltage_0,
photon_energy=photon_energy,
pixels_x=pixels_x,
pixels_y=pixels_y,
capacitor=capacitor,
category=category,
)
# Retrieve all physical detector units for the given karabo_da list.
physical_units = get_pdu_from_db(
karabo_id=karabo_id,
karabo_da=karabo_da,
constant=getattr(Constants.LPD, constant_names[0])(),
condition=condition,
cal_db_interface=cal_db_interface,
snapshot_at=creation_time
)
mod_mapping = dict(zip(karabo_da, physical_units))
print("Physical detector units retrieved are: ", mod_mapping, "\n")
inp = []
for const in constant_names:
for k_da, pdu in mod_mapping.items():
inp.append((const, k_da, pdu))
with multiprocessing.Pool(processes=5) as pool:
results = pool.starmap(inject_constants, inp)
```
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment

Legal Notice | Contacts