fix: refactor write_ccv into smaller functios and enable adding lower and upper deviations

src/cal_tools/constants.py

@@ -20,7 +20,7 @@ def write_ccv(
     const_path,
     pdu_name, pdu_uuid, detector_type,
     calibration, conditions, created_at, proposal, runs,
-    data, dims, key='0'
+    data, dims, key='0', lower_deviations={}, upper_deviations={},
 ):
     """Write CCV data file.
 
@@ -73,8 +73,8 @@ def write_ccv(
             key = db_name.lower().replace(' ', '_')
             dset = opcond_group.create_dataset(key, data=value,
                                                dtype=np.float64)
-            dset.attrs['lower_deviation'] = 0.0
-            dset.attrs['upper_deviation'] = 0.0
+            dset.attrs['lower_deviation'] = lower_deviations.get(db_name, 0.0)
+            dset.attrs['upper_deviation'] = upper_deviations.get(db_name, 0.0)
             dset.attrs['database_name'] = db_name
 
         dset = ccv_group.create_dataset('data', data=data)
@@ -83,56 +83,27 @@ def write_ccv(
     return ccv_group_name
 
 
-def inject_ccv(const_src, ccv_root, report_to=None):
-    """Inject new CCV into CalCat.
+def get_ccv_attributes(ccv_group):
+    return ccv_group.attrs['proposal'], ccv_group.attrs['runs'], ccv_group.attrs['begin_at']
 
-    Args:
-        const_path (str or Path): Path to CCV data file.
-        ccv_root (str): CCV HDF group name.
-        report_to (str): Metadata location.
 
-    Returns:
-        None
+def condition_parameters(condition_group):
+    params = []
 
-    Raises:
-        RuntimeError: If CalCat POST request fails.
-    """
+    # It's really not ideal we're mixing conditionS and condition now.
+    for parameter in condition_group:
+        param_dset = condition_group[parameter]
+        params.append({
+            'parameter_name': param_dset.attrs['database_name'],
+            'value': float(param_dset[()]),
+            'lower_deviation_value': param_dset.attrs['lower_deviation'],
+            'upper_deviation_value': param_dset.attrs['upper_deviation'],
+            'flg_available': True
+        })
+    return params
 
-    pdu_name, calibration, key = ccv_root.lstrip('/').split('/')
-
-    with h5py.File(const_src, 'r') as const_file:
-        pdu_group = const_file[pdu_name]
-        pdu_uuid = pdu_group.attrs['uuid']
-        detector_type = pdu_group.attrs['detector_type']
-
-        ccv_group = const_file[ccv_root]
-        proposal, runs = ccv_group.attrs['proposal'], ccv_group.attrs['runs']
-        begin_at_str = ccv_group.attrs['begin_at']
-
-        condition_group = ccv_group['operating_condition']
-
-        cond_params = []
-
-        # It's really not ideal we're mixing conditionS and condition now.
-        for parameter in condition_group:
-            param_dset = condition_group[parameter]
-            cond_params.append({
-                'parameter_name': param_dset.attrs['database_name'],
-                'value': float(param_dset[()]),
-                'lower_deviation_value': param_dset.attrs['lower_deviation'],
-                'upper_deviation_value': param_dset.attrs['upper_deviation'],
-                'flg_available': True
-            })
-
-    const_rel_path = f'xfel/cal/{detector_type.lower()}/{pdu_name.lower()}'
-    const_filename = f'cal.{time.time()}.h5'
-
-    if proposal and len(runs) > 0:
-        raw_data_location = 'proposal:{} runs: {}'.format(
-            proposal, ' '.join([str(x) for x in runs]))
-    else:
-        pass  # Fallback for non-run based constants
 
+def generate_unique_name(detector_type, pdu_name, pdu_uuid, cond_params):
     # Generate condition name.
     unique_name = detector_type[:detector_type.index('-Type')] + ' Def'
     cond_hash = md5(pdu_name.encode())
@@ -144,7 +115,26 @@ def inject_ccv(const_src, ccv_root, report_to=None):
         cond_hash.update(str(param_dict['value']).encode())
 
     unique_name += binascii.b2a_base64(cond_hash.digest()).decode()
-    unique_name = unique_name[:60]
+    return unique_name[:60]
+
+
+def prepare_injection_payload(
+    ccv_attrs,
+    pdu_attrs,
+    cond_params,
+    const_filename,
+    const_rel_path,
+    unique_name,
+    report_to
+):
+    proposal, runs, begin_at_str = ccv_attrs
+    pdu_name, calibration, detector_type, pdu_uuid = pdu_attrs
+
+    if proposal and len(runs) > 0:
+        raw_data_location = (
+            f'proposal:{proposal} runs: {" ".join([str(x) for x in runs])}')
+    else:
+        pass  # Fallback for non-run based constants
 
     # Add PDU "UUID" to parameters.
     cond_params.append({
@@ -187,6 +177,55 @@ def inject_ccv(const_src, ccv_root, report_to=None):
             'file_path': str(report_path)
         }
 
+
+def inject_ccv(const_src, ccv_root, report_to=None):
+    """Inject new CCV into CalCat.
+
+    Args:
+        const_path (str or Path): Path to CCV data file.
+        ccv_root (str): CCV HDF group name.
+        report_to (str): Metadata location.
+
+    Returns:
+        None
+
+    Raises:
+        RuntimeError: If CalCat POST request fails.
+    """
+
+    pdu_name, calibration, key = ccv_root.lstrip('/').split('/')
+
+    with h5py.File(const_src, 'r') as const_file:
+        pdu_group = const_file[pdu_name]
+        pdu_uuid = pdu_group.attrs['uuid']
+        detector_type = pdu_group.attrs['detector_type']
+
+        ccv_group = const_file[ccv_root]
+        ccv_attrs = get_ccv_attributes(ccv_group)
+        proposal, runs = ccv_group.attrs['proposal'], ccv_group.attrs['runs']
+        begin_at_str = ccv_group.attrs['begin_at']
+
+        condition_group = ccv_group['operating_condition']
+
+        cond_params = condition_parameters(condition_group)
+
+    const_rel_path = f'xfel/cal/{detector_type.lower()}/{pdu_name.lower()}'
+    const_filename = f'cal.{time.time()}.h5'
+
+    unique_name = generate_unique_name(
+        detector_type, pdu_name, pdu_uuid, cond_params)
+
+    pdu_attrs = (pdu_name, calibration, detector_type, pdu_uuid)
+    inject_h = prepare_injection_payload(
+        ccv_attrs,
+        pdu_attrs,
+        cond_params,
+        const_filename,
+        const_rel_path,
+        unique_name,
+        report_to
+    )
+
     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)