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

remove unneeded get_pdu_from_db and add more documentation

parent c283a2f4
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
+ 10
− 26
View file @ 1431053d
%% 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.
This notebook is used to read HDF5 files with LPDMINI constants to inject into the CALCAT database.
The notebook expects an explicit filename format: <calibration_constant_name>_<karabo_da>.h5 for example `RelativeGain_LPDMINI00_2.h5`
- <calibration_constant_name>: The calibration name as it is named in CALCAT.
- <karabo_da>: The data aggregator name. As LPDMini aggregators have `/` we replace it with `_`.
%% 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" # detector identifier.
karabo_da = ["all"] # karabo data aggregators. default "all" for all 8 karabo data aggregator names.
karabo_da = [""] # 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 = 250 # bias voltage for minis 1, 3, 5, 7
bias_voltage_1 = 250 # bias voltage for minis 2, 4, 6, 8
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'
creation_time = '2023-05-07T15:10:07' # 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
from cal_tools.calcat_interface import CalCatApi
from cal_tools.restful_config import calibration_client
from cal_tools.tools import get_pdu_from_db, get_report, send_to_db
from cal_tools.tools import get_report, send_to_db
from iCalibrationDB import Conditions, Constants
```
%% Cell type:code id: tags:
``` python
pixels_x = 256
pixels_y = 32
calcat_client = calibration_client()
calcat = CalCatApi(client=calcat_client)
# Look up PDUs
detector_id = calcat.detector(karabo_id)['id']
pdus_by_da = calcat.physical_detector_units(detector_id, pdu_snapshot_at=creation_time)
if not karabo_da or karabo_da == ['']:
if not karabo_da or karabo_da == [""]:
karabo_da = sorted(pdus_by_da.keys())
else:
karabo_da = karabo_da
# 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=None,
)
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=None,
)
# 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))
mod_mapping = {mod: pdus_by_da[mod]["physical_name"] for mod in karabo_da}
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