diff --git a/src/cal_tools/constants.py b/src/cal_tools/constants.py
index cd4abb00f819ef75f2da7b3315adaf4f863bd3c6..7a0e068d64a021c815ba088d0ccc3cc2d55d7c02 100644
--- a/src/cal_tools/constants.py
+++ b/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)