[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

src/cal_tools/constants.py

@@ -137,7 +137,7 @@ def inject_ccv(const_src, ccv_root, report_to):
     unique_name = detector_type[:detector_type.index('-Type')] + ' Def'
     cond_hash = md5(pdu_name.encode())
     cond_hash.update(pdu_uuid.to_bytes(
-        length=8, byteorder='little', signed=False))3
+        length=8, byteorder='little', signed=False))
 
     for param_dict in cond_params:
         cond_hash.update(str(param_dict['parameter_name']).encode())
@@ -184,19 +184,24 @@ def inject_ccv(const_src, ccv_root, report_to):
         }
     }
 
-    resp = CalibrationClient.inject_new_calibration_constant_version(calibration_client(), inject_h)
-
-    if not resp['success']:
-        raise RuntimeError(resp)
-
     const_dest = _get_default_caldb_root() / const_rel_path / const_filename
     const_dest.parent.mkdir(parents=True, exist_ok=True)
     copyfile(const_src, const_dest)
 
+    resp = CalibrationClient.inject_new_calibration_constant_version(
+        calibration_client(), inject_h)
+
+    if not resp['success']:
+        const_dest.unlink()  # Delete already copied CCV file.
+        raise RuntimeError(resp)
+
 
-def inject_ccv_legacy(in_folder, metadata_folder, runs, calibration, cond, pdu, const_input, begin_at):
+def inject_ccv_legacy(in_folder, metadata_folder, runs, calibration, cond,
+                      pdu, const_input, begin_at):
     """Inject new CCV into CalCat.
 
+    LEGACY VERSION - DO NOT USE
+    (from before most details were saved in CCV file itself)
 
     metadata_folder
     calibration -> in file, but pass anyway as there could be multiple
@@ -209,13 +214,15 @@ def inject_ccv_legacy(in_folder, metadata_folder, runs, calibration, cond, pdu,
         metadata_folder,  # Same as notebook parameter
         [run_high, run_mid, run_low],
         'Offset',
-        current_pdu,  # Element of calibration_client.from get_all_phy_det_units_from_detector
+        current_pdu,  # Element of calibration_client.from
+                      # get_all_phy_det_units_from_detector
         out_folder / 'my-constant-file.h5',
         get_dir_creation_date(in_folder, run_high)
     )
 
     Args:
-        in_folder (str or Path): Root for input data, i.e. *not* containing r{run:04d} folder
+        in_folder (str or Path): Root for input data, i.e. *not*
+            containing r{run:04d} folder
         metadata_folder (str or Path): Metadata location
         runs (Iterable of str or int): Run number(s) used for characterization.
         calibration (str): Calibration name, e.g. 'Offset', must exist in CalCat.
@@ -246,7 +253,8 @@ def inject_ccv_legacy(in_folder, metadata_folder, runs, calibration, cond, pdu,
     # Generate condition name.
     unique_name = detector_type[:detector_type.index('-Type')] + ' Def'
     cond_hash = hashlib.md5(pdu_name.encode())
-    cond_hash.update(pdu['uuid'].to_bytes(length=8, byteorder='little', signed=False))
+    cond_hash.update(pdu['uuid'].to_bytes(
+        length=8, byteorder='little', signed=False))
 
     for key, value in cond_dict.items():
         cond_hash.update(str(key).encode())
@@ -305,7 +313,8 @@ def inject_ccv_legacy(in_folder, metadata_folder, runs, calibration, cond, pdu,
         }
     }
 
-    resp = CalibrationClient.inject_new_calibration_constant_version(calibration_client(), inject_h)
+    resp = CalibrationClient.inject_new_calibration_constant_version(
+        calibration_client(), inject_h)
 
     if not resp['success']:
         raise RuntimeError(resp)