[Generic] Injection code for DynamicFF corrections

Description

This MR contains the new machinery to write calibration constant files and inject them into CalCat. I decided to have it in its own MR against !939 (merged) for a cleaner diff and independent discussion.

For writing, for now there's only the low-level function write_ccv expecting all kinds of metadata to be passed explicitly. I've been exploring some more convenient APIs to simplify the boilerplate in various notebooks, but let's give this more time. For now the Shimadzu characterization notebook was explicitly written in a way to have this explicit information handy, so let's make use of it.

For injection, there's the original function used for testing and design inject_ccv_legacy that also takes all data explicitly, and a new version to be used from now on that reads most of it from the CCV file created with write_ccv. This is done to allow the webservice to also perform injection on behalf of a user with (almost) just the files at hand.

How Has This Been Tested?

WIP

Types of changes

• New feature (non-breaking change which adds functionality)

Reviewers

@ahmedk @kluyvert @hammerd

notebooks/DynamicFF/Characterize_DynamicFF_NBC.ipynb

 %% Cell type:markdown id: tags:
 
 # Characterization of dark and flat field for Dynamic Flat Field correction
 
 Author: Egor Sobolev
 
 Computation of dark offsets and flat-field principal components
 
 %% Cell type:code id: tags:
 
 ``` python
 in_folder = "/gpfs/exfel/exp/SPB/202430/p900425/raw"  # input folder, required
 out_folder = '/gpfs/exfel/data/scratch/esobolev/test/shimadzu'  # output folder, required
 metadata_folder = ""  # Directory containing calibration_metadata.yml when run by xfel-calibrate
 run_high = 1 # run number in which dark data was recorded, required
 run_low = 2 # run number in which flat-field data was recorded, required
 operation_mode = "PCA_DynamicFF"  # Detector operation mode, optional (defaults to "PCA_DynamicFF")
 
 # Data files parameters.
 karabo_da = ['-1'] # data aggregators
 karabo_id = "SPB_MIC_HPVX2" # karabo prefix of Shimadzu HPV-X2 devices
 
 # Database access parameters.
 cal_db_interface = "tcp://max-exfl-cal001:8021"  # Unused, calibration DB interface to use
 cal_db_timeout = 30000  # Unused, calibration DB timeout
 db_output = False # if True, the notebook sends dark constants to the calibration database
 local_output = True # if True, the notebook saves dark constants locally
 
 # Calibration constants parameters
 n_components = 50  # Number of principal components of flat-field to compute (default: 50)
 ```
 
 %% Cell type:code id: tags:
 
 ``` python
 import datetime
 import os
 import warnings
 from logging import warning
 from shutil import copyfile
 from tempfile import NamedTemporaryFile
 
 warnings.filterwarnings('ignore')
 
 import time
 import numpy as np
 import matplotlib.pyplot as plt
 from IPython.display import display, Markdown
 
 from extra_data import RunDirectory
 
 %matplotlib inline
 from cal_tools.step_timing import StepTimer
 from cal_tools.tools import (
     get_dir_creation_date,
+    run_prop_seq_from_path,
     save_dict_to_hdf5
 )
 from cal_tools.restful_config import calibration_client, extra_calibration_client
 from cal_tools.shimadzu import ShimadzuHPVX2
+from cal_tools.constants import write_ccv, inject_ccv
 
 import dynflatfield as dffc
 from dynflatfield.draw import plot_images, plot_camera_image
 ```
 
 %% Cell type:code id: tags:
 
 ``` python
 extra_calibration_client()  # Configure CalibrationData.
 
 cc = calibration_client()
 pdus = cc.get_all_phy_det_units_from_detector(
     {"detector_identifier": karabo_id})  # TODO: Use creation_time for snapshot_at
 
 if not pdus["success"]:
     raise ValueError("Failed to retrieve PDUs")
 
 detector_info = pdus['data'][0]['detector']
 detector = ShimadzuHPVX2(detector_info["source_name_pattern"])
 
 print(f"Instrument {detector.instrument}")
 print(f"Detector in use is {karabo_id}")
 
 modules = {}
 for pdu_no, pdu in enumerate(pdus["data"]):
     db_module = pdu["physical_name"]
     module = pdu["module_number"]
     da = pdu["karabo_da"]
     if karabo_da[0] != "-1" and da not in karabo_da:
         continue
 
     instrument_source_name = detector.instrument_source(module)
     print('-', da, db_module, module, instrument_source_name)
 
     modules[da] = dict(
         db_module=db_module,
         module=module,
         raw_source_name=instrument_source_name,
         pdu_no=pdu_no,
     )
 
 constants = {}
 
 step_timer = StepTimer()
 ```
 
 %% Cell type:markdown id: tags:
 
 # Offset map
 
 %% Cell type:code id: tags:
 
 ``` python
 dark_run = run_high
 dark_creation_time = get_dir_creation_date(in_folder, dark_run)
 print(f"Using {dark_creation_time} as creation time of Offset constant.")
 
 for da, meta in modules.items():
     source_name = detector.instrument_source(meta["module"])
     image_key = detector.image_key
 
     display(Markdown(f"## {source_name}"))
 
     # read
     step_timer.start()
     file_da, _, _ = da.partition('/')
     dark_dc = RunDirectory(f"{in_folder}/r{dark_run:04d}",
                            include=f"RAW-R{dark_run:04d}-{file_da}-S*.h5")
 
     if source_name not in dark_dc.all_sources:
         raise ValueError(f"Could not find source {source_name} for module {da} in dark data")
 
     dark_dc = dark_dc.select([(source_name, image_key)])
     conditions = detector.conditions(dark_dc, meta["module"])
 
     key_data = dark_dc[source_name, image_key]
     images_dark = key_data.ndarray()
     ntrain, npulse, ny, nx = images_dark.shape
 
     print(f"N image: {ntrain * npulse} (ntrain: {ntrain}, npulse: {npulse})")
     print(f"Image size: {ny} x {nx} px")
     step_timer.done_step("Read dark images")
 
     # process
     step_timer.start()
     dark = dffc.process_dark(images_dark)  # Amounts to a per-pixel mean right now.
 
     # put results in the dict
     module_constants = constants.setdefault(meta["db_module"], {})
     module_constants["Offset"] = dict(
         conditions=conditions, data=dark, pdu_no=meta["pdu_no"],
-        creation_time=dark_creation_time
+        creation_time=dark_creation_time, dims=['ss', 'fs']
     )
     step_timer.done_step("Process dark images")
     display()
 
     # draw plots
     step_timer.start()
     plot_camera_image(dark)
     plt.show()
     step_timer.done_step("Draw offsets")
 ```
 
 %% Cell type:markdown id: tags:
 
 # Flat-field PCA decomposition
 
 %% Cell type:code id: tags:
 
 ``` python
 flat_run = run_low
 flat_creation_time = get_dir_creation_date(in_folder, flat_run)
 print(f"Using {flat_creation_time} as creation time of DynamicFF constant.")
 
 for da, meta in modules.items():
     source_name = detector.instrument_source(meta["module"])
     image_key = detector.image_key
 
     display(Markdown(f"## {source_name}"))
 
     # read
     step_timer.start()
     file_da, _, _ = da.partition('/')
     flat_dc = RunDirectory(f"{in_folder}/r{flat_run:04d}",
                            include=f"RAW-R{flat_run:04d}-{file_da}-S*.h5")
 
     if source_name not in flat_dc.all_sources:
         raise ValueError(f"Could not find source {source_name} for module {da} in flatfield data")
 
     flat_dc = flat_dc.select([(source_name, image_key)])
     conditions = detector.conditions(flat_dc, meta["module"])
 
     dark = constants[meta["db_module"]]["Offset"]["data"]
     dark_conditions = constants[meta["db_module"]]["Offset"]["conditions"]
 
     if conditions != dark_conditions:
         raise ValueError(f"The conditions for flat-field run {conditions}) do not match "
                          f"the dark run conditions ({dark_conditions}). Skip flat-field characterization.")
 
     key_data = flat_dc[source_name][image_key]
     images_flat = key_data.ndarray()
     ntrain, npulse, ny, nx = images_flat.shape
 
     print(f"N image: {ntrain * npulse} (ntrain: {ntrain}, npulse: {npulse})")
     print(f"Image size: {ny} x {nx} px")
     step_timer.done_step("Read flat-field images")
 
     # process
     step_timer.start()
     flat, components, explained_variance_ratio = dffc.process_flat(
         images_flat, dark, n_components)
     flat_data = np.concatenate([flat[None, ...], components])
 
     # put results in the dict
     conditions = detector.conditions(flat_dc, meta["module"])
     module_constants = constants.setdefault(meta["db_module"], {})
     module_constants["DynamicFF"] = dict(
         conditions=conditions, data=flat_data, pdu_no=meta["pdu_no"],
-        creation_time=flat_creation_time
+        creation_time=flat_creation_time, dims=['component', 'ss', 'fs']
     )
     step_timer.done_step("Process flat-field images")
 
     # draw plots
     step_timer.start()
     display(Markdown("### Average flat-field"))
     plot_camera_image(flat)
     plt.show()
 
     display(Markdown("### Explained variance ratio"))
     fig, ax = plt.subplots(1, 1, figsize=(10,4), tight_layout=True)
     ax.semilogy(explained_variance_ratio, 'o')
     ax.set_xticks(np.arange(len(explained_variance_ratio)))
     ax.set_xlabel("Component no.")
     ax.set_ylabel("Variance fraction")
     plt.show()
 
     display(Markdown("### The first principal components (up to 20)"))
     plot_images(components[:20], figsize=(13, 8))
     plt.show()
 
     step_timer.done_step("Draw flat-field")
 ```
 
 %% Cell type:markdown id: tags:
 
 ## Calibration constants
 
 %% Cell type:code id: tags:
 
 ``` python
 step_timer.start()
 
+_, proposal, _ = run_prop_seq_from_path(in_folder)
+
 # Output Folder Creation:
 if local_output:
     os.makedirs(out_folder, exist_ok=True)
 
-def inject_ccv(in_folder, metadata_folder, runs, calibration, cond, pdu, const_input, begin_at):
-    print("* Send to db:", const_input)
-    print("  - in folder:", in_folder)
-    print("  - metadata folder:", metadata_folder)
-    print("  - runs:", runs)
-    print("  -", calibration)
-    print("  -", cond)
-    print("  -", begin_at)
-
 for db_module, module_constants in constants.items():
     for constant_name, constant in module_constants.items():
         conditions = constant["conditions"]
-        conditions_dict = conditions.make_dict(
-            conditions.calibration_types[constant_name])
-
-        data_to_store = {db_module: {constant_name: {'0': {
-            'conditions': conditions_dict,
-            'data': constant["data"],
-        }}}}
+        pdu = pdus["data"][constant["pdu_no"]]
 
         with NamedTemporaryFile() as tempf:
-            save_dict_to_hdf5(data_to_store, tempf)
+            ccv_root = write_ccv(
+                tempf.name,
+                pdu['physical_name'], pdu['uuid'], pdu['detector_type']['name'],
+                constant_name, conditions, constant['creation_time'],
+                proposal, [dark_run, flat_run],
+                constant["data"], constant['dims'])
 
             if db_output:
-                inject_ccv(
-                    in_folder, metadata_folder, [dark_run, flat_run],
-                    constant_name, conditions, pdus["data"][constant["pdu_no"]],
-                    ofile, constant["creation_time"]
-                )
+                inject_ccv(tempf.name, ccv_root, metadata_folder)
 
             if local_output:
                 ofile = f"{out_folder}/const_{constant_name}_{db_module}.h5"
 
                 if os.path.isfile(ofile):
                     print(f'File {ofile} already exists and will be overwritten')
 
                 copyfile(tempf.name, ofile)
 ```
 
 %% Cell type:code id: tags:
 
 ``` python
 print(f"Total processing time {step_timer.timespan():.01f} s")
 step_timer.print_summary()
 ```