reduce processes

a7b702c4 · Karim Ahmed · 965d39ef · a7b702c4
Commit a7b702c4 authored 3 years ago by Karim Ahmed
--- a/notebooks/LPD/Inject_calibration_constants_from_h5files.ipynb
+++ b/notebooks/LPD/Inject_calibration_constants_from_h5files.ipynb
@@ -221,7 +221,7 @@
    "    for k_da, pdu in mod_mapping.items():\n",
    "        inp.append((const, k_da, pdu))\n",
    "\n",
-    "with multiprocessing.Pool(processes=16) as pool:\n",
+    "with multiprocessing.Pool(processes=5) as pool:\n",
    "    results = pool.starmap(inject_constants, inp)"
   ]
  }

 %% Cell type:markdown id: tags:

 # Injecting calibration constant data to the database #

 Author: European XFEL Detector Group, Version: 1.0

 Reading h5files of calibration constants to inject them to the database. Used for LPD

 %% 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_LPD1M-1"  # detector identifier.
 karabo_da = ["all"]  # karabo data aggregators. default "all" for all 16 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 = 250  # bias voltage value. Used for constant conditions.
 capacitor = 5  # capacitor value. Used for constant conditions.
 category = 0 # calibration constant source category, 0 for European XFEl and 1 for RAL. 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 pathlib import Path
 from typing import List, Tuple

 import h5py

 from iCalibrationDB import (
    Constants,
    Conditions,
 )
 from cal_tools.tools import (
    get_from_db,
    get_pdu_from_db,
    get_report,
    send_to_db,
 )
 ```

 %% Cell type:code id: tags:

 ``` python
 pixels_x = pixels_y = 256

 if karabo_da[0].lower() == "all":
    karabo_da = [f"LPD{i:02d}" for i in range(16)]

 # 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,
    karabo_da: str,
    physical_unit: str,
 ):
    constant = getattr(Constants.LPD, constant_name)()

    cfile = c_folder / f"{constant_name}_{karabo_da}.h5"

    if not cfile.exists():
        raise ValueError(f"ERROR: constant file {cfile} doesn't exists.\n")

    # 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 == (pixels_x, pixels_y, memory_cells, 3):
        raise ValueError(
            f"ERROR: {const} constant data shape is not as expected."
            f" {cdata.shape} != ({pixels_x}, {pixels_y}, {memory_cells}, 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_path = get_report(
    out_folder=in_folder,
    default_path=f"No_report/LPD_{datetime.now().strftime('%Y-%m-%dT%H:%M:%S')}"
 )

 # 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,
 )

 # 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=16) as pool:
+with multiprocessing.Pool(processes=5) as pool:
    results = pool.starmap(inject_constants, inp)
 ```