{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# Injecting calibration constant data to the database #\n",
"\n",
"Author: European XFEL Detector Group, Version: 1.0\n",
"\n",
"Reading h5files of calibration constants to inject them to the database. Used for LPD"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# calibration constant parameters:\n",
"constant_names = [\"\"] # calibration constant name, required.\n",
"in_folder = \"\" # calibration constants folder, required.\n",
"out_folder = \"\" # output folder to store report path in case the notebook is executed by CLI, required.\n",
"proposal = \"\" # Add proposal number to be sent to the database as a part of Raw data location.\n",
"runs = [\"\"] # Add list of runs to be sent to the database as a part of Raw data location.\n",
"\n",
"# detector parameters:\n",
"karabo_id = \"FXE_DET_LPD1M-1\" # detector identifier.\n",
"karabo_da = [\"all\"] # karabo data aggregators. default \"all\" for all 16 karabo data aggregator names.\n",
"\n",
"# calibration database parameters:\n",
"cal_db_interface = \"tcp://max-exfl-cal001:8015#8045\" # calibration DB zmq address.\n",
"\n",
"# calibration constant conditions:\n",
"memory_cells = 512 # Number of memory cells. Used for constant conditions.\n",
"bias_voltage = 250 # bias voltage value. Used for constant conditions.\n",
"capacitor = 5 # capacitor value. Used for constant conditions.\n",
"category = 0 # calibration constant source category, 0 for European XFEl and 1 for RAL. Used for constant conditions.\n",
"photon_energy = 9.2 # calibration constant photon energy. Used for constant conditions.\n",
"creation_time = '2020-01-20T14:12:06' # creation time for the injected constants. required format '2019-01-20T14:12:06'"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"import multiprocessing # noqa\n",
"from datetime import datetime # noqa\n",
"from pathlib import Path # noqa\n",
"from typing import List, Tuple # noqa\n",
"\n",
"import h5py # noqa \n",
"from cal_tools.tools import get_pdu_from_db, get_report, send_to_db # noqa\n",
"from iCalibrationDB import Conditions, Constants # noqa"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"pixels_x = pixels_y = 256\n",
"\n",
"if karabo_da[0].lower() == \"all\":\n",
" karabo_da = [f\"LPD{i:02d}\" for i in range(16)]\n",
"\n",
"# if proposal or runs are given assign file_loc\n",
"# for calibration constant versions metadata.\n",
"file_loc = \"\"\n",
"if proposal:\n",
" file_loc += f\"proposal:{proposal}\"\n",
"if runs[0] != \"\":\n",
" file_loc += f\"runs: {runs}\"\n",
"\n",
"if file_loc == \"\":\n",
" print(\n",
" \"No proposal or runs were given for constant source.\"\n",
" \" No \\\"Raw data location\\\" will be injected with the constants.\\n\"\n",
" )"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"def validate_input_paths(\n",
" in_folder: str,\n",
" creation_time: str,\n",
" constant_names: List[str],\n",
") -> Tuple[str, datetime]:\n",
"\n",
" # Validate input parameters:\n",
" if not (in_folder):\n",
" raise ValueError(\n",
" \"ERROR: \\\"in_folder\\\" is not given.\"\n",
" \" Please provide the constants input folder.\"\n",
" )\n",
"\n",
" c_folder = Path(in_folder)\n",
"\n",
" if not c_folder.is_dir():\n",
" raise ValueError(\n",
" f\"ERROR: in_folder {in_folder} directory doesn't exist.\"\n",
" )\n",
"\n",
" try:\n",
" creation_time = datetime.strptime(creation_time, '%Y-%m-%dT%H:%M:%S')\n",
" except ValueError:\n",
" raise ValueError(\n",
" \"Incorrect data format, \"\n",
" \"should be YYYY-MM-DDTHH:MM:SS i.e. 2019-01-20T14:12:06\"\n",
" )\n",
"\n",
" for constant in constant_names:\n",
" if not hasattr(Constants.LPD, constant):\n",
" raise ValueError(\n",
" f\"ERROR: Constant name \\\"{constant}\\\" is not a known LPD constant. \"\n",
" f\"Available LPD Constants are {[c for c in dir(Constants.LPD) if c[0] != '_']}\"\n",
" )\n",
" return c_folder, creation_time"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"def inject_constants(\n",
" constant_name: str,\n",
" karabo_da: str,\n",
" physical_unit: str,\n",
"):\n",
" constant = getattr(Constants.LPD, constant_name)()\n",
"\n",
" cfile = c_folder / f\"{constant_name}_{karabo_da}.h5\"\n",
"\n",
" if not cfile.exists():\n",
" raise ValueError(f\"ERROR: constant file {cfile} doesn't exists.\\n\")\n",
"\n",
" # load constant data.\n",
" with h5py.File(cfile, \"r\") as f:\n",
" cdata = f[constant_name][()]\n",
"\n",
" # Validate for only LPD at the moment.\n",
" if not cdata.shape == (pixels_x, pixels_y, memory_cells, 3):\n",
" raise ValueError(\n",
" f\"ERROR: {const} constant data shape is not as expected.\"\n",
" f\" {cdata.shape} != ({pixels_x}, {pixels_y}, {memory_cells}, 3).\\n\"\n",
" )\n",
"\n",
" constant.data = cdata\n",
"\n",
" send_to_db(\n",
" db_module=physical_unit,\n",
" karabo_id=karabo_id,\n",
" constant=constant,\n",
" condition=condition,\n",
" file_loc=file_loc,\n",
" report_path=report_path,\n",
" cal_db_interface=cal_db_interface,\n",
" creation_time=creation_time,\n",
" )"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"c_folder, creation_time = validate_input_paths(\n",
" in_folder,\n",
" creation_time,\n",
" constant_names,\n",
")\n",
"\n",
"# create a report path for calibration constant versions metadata.\n",
"report_name = f\"No_report/LPD_{datetime.now().strftime('%Y-%m-%dT%H:%M:%S')}\"\n",
"report_path = get_report(\n",
" out_folder=in_folder,\n",
" default_path=report_name\n",
")\n",
"\n",
"# Calibration constants condition object.\n",
"condition = Conditions.Illuminated.LPD(\n",
" memory_cells=memory_cells,\n",
" bias_voltage=bias_voltage,\n",
" photon_energy=photon_energy,\n",
" pixels_x=pixels_x,\n",
" pixels_y=pixels_y,\n",
" capacitor=capacitor,\n",
" category=category,\n",
")\n",
"\n",
"# Retrieve all physical detector units for the given karabo_da list.\n",
"physical_units = get_pdu_from_db(\n",
" karabo_id=karabo_id,\n",
" karabo_da=karabo_da,\n",
" constant=getattr(Constants.LPD, constant_names[0])(),\n",
" condition=condition,\n",
" cal_db_interface=cal_db_interface,\n",
" snapshot_at=creation_time\n",
")\n",
"\n",
"mod_mapping = dict(zip(karabo_da, physical_units))\n",
"print(\"Physical detector units retrieved are: \", mod_mapping, \"\\n\")\n",
"\n",
"inp = []\n",
"for const in constant_names:\n",
" for k_da, pdu in mod_mapping.items():\n",
" inp.append((const, k_da, pdu))\n",
"\n",
"with multiprocessing.Pool(processes=5) as pool:\n",
" results = pool.starmap(inject_constants, inp)"
]
}
],
"metadata": {
"kernelspec": {
"display_name": "Python 3",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.6.7"
}
},
"nbformat": 4,
"nbformat_minor": 4
}