diff --git a/notebooks/Jungfrau/Jungfrau_dark_analysis_all_gains_burst_mode_NBC.ipynb b/notebooks/Jungfrau/Jungfrau_dark_analysis_all_gains_burst_mode_NBC.ipynb
index 1d35a5d198c94b0b4b83f851bf2aea3b31863dd1..2f1b0b0a48df7ccba136ecc2df98061618617a8c 100644
--- a/notebooks/Jungfrau/Jungfrau_dark_analysis_all_gains_burst_mode_NBC.ipynb
+++ b/notebooks/Jungfrau/Jungfrau_dark_analysis_all_gains_burst_mode_NBC.ipynb
@@ -100,6 +100,7 @@
" write_ccv,\n",
")\n",
"from cal_tools.enums import BadPixels, JungfrauGainMode\n",
+ "from cal_tools.inject import CCVAlreadyInjectedError, inject_ccv\n",
"from cal_tools.jungfrau import jungfraulib\n",
"from cal_tools.restful_config import (\n",
" extra_calibration_client,\n",
diff --git a/src/cal_tools/inject.py b/src/cal_tools/inject.py
index 57ee8d9b7bc068915035f5df4e28f1885953b27a..e7f927a912104b148be4121337bbc8d65a1dd8ae 100644
--- a/src/cal_tools/inject.py
+++ b/src/cal_tools/inject.py
@@ -55,7 +55,7 @@ class ParameterConditionAttribute:
value: str
lower_deviation_value: float = 0
upper_deviation_value: float = 0
- flg_available: str = 'false'
+ flg_available: str = 'true'
description: str = ''
@@ -63,7 +63,7 @@ class ParameterConditionAttribute:
class Condition:
name: str
parameters_conditions_attributes: List[ParameterConditionAttribute] = field(default_factory=list)
- flg_available: str = 'false'
+ flg_available: str = 'true'
event_at: str = str(datetime.today()) # TODO: Why is this needed? it is not written in swagger.
description: str = ''
@@ -79,8 +79,8 @@ class CalibrationConstant:
detector_type_id: int
calibration_id: int
condition_id: int
- flg_auto_approve: str = 'false'
- flg_available: str = 'false'
+ flg_auto_approve: str = 'true'
+ flg_available: str = 'true'
description: str = ""
@@ -88,7 +88,7 @@ class CalibrationConstant:
class Report:
name: str
file_path: str
- flg_available: str = 'false'
+ flg_available: str = 'true'
description: str = ""
@@ -112,7 +112,7 @@ class CalibrationConstantVersion:
description: str = ''
-def generate_unique_name(detector_type, pdu_name, pdu_uuid, cond_params):
+def generate_unique_cond_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())
@@ -152,13 +152,10 @@ def get_raw_data_location(proposal, runs):
def _failed_response(resp):
- # TODO: Add more variant errors.
+ # TODO: Add more errors if needed
if CALIBRATION_CONSTANT_VERSIONS in resp.url.lstrip("/"):
if resp.status_code == 422:
raise CCVAlreadyInjectedError
- warn(
- "Calibration Constant Version has already been injected: "
- f"{dir(resp)}\n{resp.reason}-{resp.status_code}-{resp.text}-{resp.url}", CCVAlreadyInjectedWarning)
elif resp.status_code >= 400:
try:
d = json.loads(resp.content.decode("utf-8"))
@@ -267,8 +264,6 @@ class InjectAPI(CalCatAPIClient):
self,
cond_params,
begin_at,
- proposal,
- runs,
pdu_name,
pdu_uuid,
detector_type,
@@ -277,18 +272,18 @@ class InjectAPI(CalCatAPIClient):
const_filename,
ccv_root,
report_to=None,
-
+ raw_data_location='',
):
"""Inject new CCV into CalCat."""
- cond_name = generate_unique_name(
+ cond_name = generate_unique_cond_name(
detector_type, pdu_name, pdu_uuid, cond_params)
# Create condition table in database, if not available.
resp = self.set_condition(cond_name, list(cond_params.values()))
-
- # Prepare some parameters to set Calibration Constant.
condition_id = resp['id']
+
+ # Prepare some parameters to set Calibration Constant.
cal_id = self.calibration_by_name(calibration)['id']
det_type_id = self.detector_type_by_name(detector_type)['id']
cc_name = create_unique_cc_name(detector_type, calibration, cond_name)
@@ -302,7 +297,8 @@ class InjectAPI(CalCatAPIClient):
report_id = None
if report_to:
report_path = Path(report_to).absolute().with_suffix('.pdf')
- resp = self.get_or_create_report(name=report_path.stem, file_path=str(report_path))
+ resp = self.get_or_create_report(
+ name=report_path.stem, file_path=str(report_path))
report_id = resp["id"]
# Get PDU ID before creating new CCV.
@@ -316,7 +312,7 @@ class InjectAPI(CalCatAPIClient):
data_set_name=ccv_root,
calibration_constant_id=cc_id,
physical_detector_unit_id=pdu_id,
- raw_data_location=get_raw_data_location(proposal, runs),
+ raw_data_location=raw_data_location,
report_id=report_id,
begin_validity_at=begin_at,
end_validity_at='',
@@ -356,43 +352,51 @@ class InjectAPI(CalCatAPIClient):
def set_calibration_constant_version(self, ccv: CalibrationConstantVersion):
return self.create_calibration_constant_version(ccv)
- def get_ccv_info_from_file(self, cfile: Union[str, Path], pdu: str, ccv_root: str):
- def to_string(value):
- """Send only accepted value types to CALCAT."""
- if isinstance(value, bool):
- value = float(value)
- return str(value)
-
- with h5py.File(cfile, 'r') as const_file:
- pdu_group = const_file[pdu]
- 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 = ccv_group.attrs['begin_at']
-
- cond_params = {}
- condition_group = ccv_group['operating_condition']
- # It's really not ideal we're mixing conditionS and condition now.
- # Get parameter data and create a list of `ParameterConditionAttribute`s
- for parameter in condition_group:
- param_dset = condition_group[parameter]
- param_name = param_dset.attrs['database_name']
- cond_params[param_name] = ParameterConditionAttribute(
- parameter_id=self.parameter_by_name(param_name)['id'],
- value=to_string(param_dset[()]),
- lower_deviation_value=param_dset.attrs['lower_deviation'],
- upper_deviation_value=param_dset.attrs['upper_deviation'],
- )
- det_uuid = 'Detector UUID'
- # Add PDU "UUID" to parameters.
- cond_params[det_uuid] = ParameterConditionAttribute(
- value=to_string(unpack('d', pack('q', pdu_uuid))[0]),
- parameter_id=self.parameter_by_name(det_uuid)['id'],
+
+def extract_parameter_conditions(client, ccv_group, pdu_uuid):
+ def _to_string(value):
+ """Send only accepted value types to CALCAT."""
+ if isinstance(value, bool):
+ value = float(value)
+ return str(value)
+
+ cond_params = {}
+ condition_group = ccv_group['operating_condition']
+ # It's really not ideal we're mixing conditionS and condition now.
+ # Get parameter data and create a list of `ParameterConditionAttribute`s
+ for parameter in condition_group:
+ param_dset = condition_group[parameter]
+ param_name = param_dset.attrs['database_name']
+ cond_params[param_name] = ParameterConditionAttribute(
+ parameter_id=client.parameter_by_name(param_name)['id'],
+ value=_to_string(param_dset[()]),
+ lower_deviation_value=param_dset.attrs['lower_deviation'],
+ upper_deviation_value=param_dset.attrs['upper_deviation'],
)
+ det_uuid = 'Detector UUID'
+ # Add PDU "UUID" to parameters.
+ cond_params[det_uuid] = ParameterConditionAttribute(
+ value=_to_string(unpack('d', pack('q', pdu_uuid))[0]),
+ parameter_id=client.parameter_by_name(det_uuid)['id'],
+ )
+
+
+def get_ccv_info_from_file(
+ client: InjectAPI, cfile: Union[str, Path], pdu: str, ccv_root: str):
- return cond_params, begin_at, proposal, runs, pdu_uuid, detector_type
+ with h5py.File(cfile, 'r') as const_file:
+ pdu_group = const_file[pdu]
+ pdu_uuid = pdu_group.attrs['uuid']
+ detector_type = pdu_group.attrs['detector_type']
+ ccv_group = const_file[ccv_root]
+ raw_data_location = get_raw_data_location(
+ ccv_group.attrs['proposal'],
+ ccv_group.attrs['runs']
+ )
+ begin_at = ccv_group.attrs['begin_at']
+ cond_params = extract_parameter_conditions(client, ccv_group, pdu_uuid)
+
+ return cond_params, begin_at, pdu_uuid, detector_type, raw_data_location
def inject_ccv(
@@ -404,11 +408,11 @@ def inject_ccv(
if client is None:
client = extra_calibration_client(inject=True)
- pdu_name, calibration, key = ccv_root.lstrip('/').split('/')
+ pdu_name, calibration, _ = ccv_root.lstrip('/').split('/')
(
- cond_params, begin_at, proposal, runs, pdu_uuid, detector_type
- ) = client.get_ccv_info_from_file(const_src, pdu_name, ccv_root)
+ cond_params, begin_at, pdu_uuid, detector_type, raw_data_location
+ ) = client.get_ccv_info_from_file(client, const_src, pdu_name, ccv_root)
const_rel_path = f'xfel/cal/{detector_type.lower()}/{pdu_name.lower()}'
const_filename = f'cal.{time.time()}.h5'
@@ -416,21 +420,20 @@ def inject_ccv(
const_dest.parent.mkdir(parents=True, exist_ok=True)
copyfile(const_src, const_dest)
- resp = client.create_new_ccv(
- cond_params,
- begin_at,
- proposal,
- runs,
- pdu_name,
- pdu_uuid,
- detector_type,
- calibration,
- const_rel_path,
- const_filename,
- ccv_root,
- report_to,
- )
-
- if not resp['success']:
+ try:
+ client.create_new_ccv(
+ cond_params,
+ begin_at,
+ pdu_name,
+ pdu_uuid,
+ detector_type,
+ calibration,
+ const_rel_path,
+ const_filename,
+ ccv_root,
+ report_to,
+ raw_data_location,
+ )
+ except Exception as e:
const_dest.unlink()
- raise RuntimeError(resp)
+ raise e