[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

@@ -197,131 +197,3 @@ def inject_ccv(const_src, ccv_root, report_to=None):
     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):
-    """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
-    const_input -> doh!
-
-    Recommended way of running this from pycalibration notebooks:
-
-    inject_ccv(
-        in_folder,  # Same as notebook parameter
-        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
-        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
-        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.
-        cond (ConditionsBase): Operating conditions for injected CCV.
-        pdu (dict): PDU information, should be passed as given by CalCat.
-        const_input (str or Path): Path to local calibration constant HDF file.
-        begin_at (datetime): Begin of calibration constant validity.
-
-    Returns:
-        None
-
-    Raises:
-        RuntimeError: If CalCat POST request fails.
-
-    """
-
-    cond_dict = cond.make_dict(cond.calibration_types[calibration])
-    pdu_name = pdu['physical_name']
-    detector_type = pdu['detector_type']['name']
-
-    const_root = f'xfel/cal/{detector_type.lower()}/{pdu_name.lower()}'
-    const_filename = f'cal.{time.time()}.h5'
-    report_path = Path(metadata_folder).with_suffix('.pdf')
-    _, proposal, _ = run_prop_seq_from_path(in_folder)
-    raw_data_location = 'proposal:{} runs: {}'.format(
-        proposal, ' '.join([str(x) for x in runs]))
-
-    # 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))
-
-    for key, value in cond_dict.items():
-        cond_hash.update(str(key).encode())
-        cond_hash.update(str(value).encode())
-
-    unique_name += binascii.b2a_base64(cond_hash.digest()).decode()
-    unique_name = unique_name[:60]
-
-    # Build condition dict for CalCat.
-    cond_params = [
-        {
-            'parameter_name': key,
-            'value': value,
-            'lower_deviation_value': 0.0,
-            'upper_deviation_value': 0.0,
-            'flg_available': True
-        }
-        for key, value in cond_dict.items()
-    ]
-
-    # Add PDU "UUID" to parameters.
-    cond_params.append({
-        'parameter_name': 'Detector UUID',
-        'value': unpack('d', pack('q', pdu['uuid']))[0],
-        'lower_deviation_value': 0.0,
-        'upper_deviation_value': 0.0,
-        'flg_available': True
-    })
-
-    inject_h = {
-        'report': {
-            'name': report_path.stem,
-            'file_path': str(report_path)
-        },
-        'detector_condition': {
-            'name': unique_name,
-            'parameters': cond_params
-        },
-        'calibration_constant': {
-            'calibration_name': calibration,
-            'detector_type_name': detector_type,
-            'flg_auto_approve': True
-        },
-        'calibration_constant_version': {
-            'raw_data_location': raw_data_location,
-            'file_name': const_filename,
-            'path_to_file': const_root,
-            'data_set_name': f'{pdu_name}/{calibration}/0',
-            'start_idx': '0',
-            'end_idx': '0',
-            'begin_validity_at': begin_at.isoformat(),
-            'end_validity_at': '',
-            'begin_at': begin_at.isoformat(),
-            'pdu_physical_name': pdu_name,
-            'flg_good_quality': True
-        }
-    }
-
-    resp = CalibrationClient.inject_new_calibration_constant_version(
-        calibration_client(), inject_h)
-
-    if not resp['success']:
-        raise RuntimeError(resp)
-
-    const_path = _get_default_caldb_root() / const_root / const_filename
-    const_path.parent.mkdir(parents=True, exist_ok=True)
-    copyfile(const_input, const_path)