From ad1290fec338d458e024d80e024257940c75d970 Mon Sep 17 00:00:00 2001
From: ahmedk <karim.ahmed@xfel.eu>
Date: Fri, 12 May 2023 11:04:18 +0200
Subject: [PATCH] remove exceptions from calcatinterface
---
src/cal_tools/calcat_interface.py | 235 +++++-------------------------
1 file changed, 38 insertions(+), 197 deletions(-)
diff --git a/src/cal_tools/calcat_interface.py b/src/cal_tools/calcat_interface.py
index e0398524f..ab369a8b8 100644
--- a/src/cal_tools/calcat_interface.py
+++ b/src/cal_tools/calcat_interface.py
@@ -1,7 +1,6 @@
"""Interfaces to calibration constant data."""
from datetime import date, datetime, time, timezone
from functools import lru_cache
-from logging import warning
from pathlib import Path
from weakref import WeakKeyDictionary
@@ -57,146 +56,8 @@ class CCVMetadata(dict):
class CalCatError(Exception):
"""CalCat API error."""
- def __init__(self, response, error_msg):
- """
-
- Args:
- response (Dict): CALCAT response.
- error_msg (Str): Error message to attach beside the response.
- """
- super().__init__(f"Got {response['info']}. {error_msg}")
-
-# These error classes are created to classify the errors for each module
-# in calibration client API into a different error handler.
-
-class CCError(CalCatError):
- """Calibration Constant module error.
- https://git.xfel.eu/ITDM/calibration_client/-/blob/master/calibration_client/modules/calibration_constant.py # noqa
- """
-
- def __init__(self, response, error_msg):
- """
-
- Args:
- response (Dict): CALCAT response.
- error_msg (Str): Error message to attach beside the response.
- """
- super().__init__(response, error_msg)
-
-class CalibrationError(CalCatError):
- """Calibration module error.
- https://git.xfel.eu/ITDM/calibration_client/-/blob/master/calibration_client/modules/calibration.py # noqa
- """
-
- def __init__(self, response, error_msg):
- """
-
- Args:
- response (Dict): CALCAT response.
- error_msg (Str): Error message to attach beside the response.
- """
- super().__init__(response, error_msg)
-
-
-class ConditionError(CalCatError):
- """Calibration Constant Version module error.
- https://git.xfel.eu/ITDM/calibration_client/-/blob/master/calibration_client/modules/condition.py # noqa
- """
-
- def __init__(self, response, error_msg):
- """
-
- Args:
- response (Dict): CALCAT response.
- error_msg (Str): Error message to attach beside the response.
- """
- super().__init__(response, error_msg)
-
-class ReportError(CalCatError):
- """Report module error.
- https://git.xfel.eu/ITDM/calibration_client/-/blob/master/calibration_client/modules/report.py # noqa
- """
-
- def __init__(self, response, error_msg):
- """
-
- Args:
- response (Dict): CALCAT response.
- error_msg (Str): Error message to attach beside the response.
- """
- super().__init__(response, error_msg)
-
-
-class DetTypeError(CalCatError):
- """Detector type module error.
- https://git.xfel.eu/ITDM/calibration_client/-/blob/master/calibration_client/modules/detector_type.py # noqa
- """
-
- def __init__(self, response, error_msg):
- """
-
- Args:
- response (Dict): CALCAT response.
- error_msg (Str): Error message to attach beside the response.
- """
- super().__init__(response, error_msg)
-
-
-class ParameterError(CalCatError):
- """Parameter module error.
- https://git.xfel.eu/ITDM/calibration_client/-/blob/master/calibration_client/modules/parameter.py # noqa
- """
-
- def __init__(self, response, error_msg):
- """
-
- Args:
- response (Dict): CALCAT response.
- error_msg (Str): Error message to attach beside the response.
- """
- super().__init__(response, error_msg)
-
-class CCVError(CalCatError):
- """Calibration Constant Version module error.
- https://git.xfel.eu/ITDM/calibration_client/-/blob/master/calibration_client/modules/calibration_constant_version.py # noqa
- """
-
- def __init__(self, response, error_msg):
- """
-
- Args:
- response (Dict): CALCAT response.
- error_msg (Str): Error message to attach beside the response.
- """
- super().__init__(response, error_msg)
-
-class DetectorError(CalCatError):
- """Detector module error.
- https://git.xfel.eu/ITDM/calibration_client/-/blob/master/calibration_client/modules/detector.py # noqa
- """
-
- def __init__(self, response, error_msg):
- """
-
- Args:
- response (Dict): CALCAT response.
- error_msg (Str): Error message to attach beside the response.
- """
- super().__init__(response, error_msg)
-
-class PDUError(CalCatError):
- """Physical Detector Unit module error.
- https://git.xfel.eu/ITDM/calibration_client/-/blob/master/calibration_client/modules/physical_detector_unit.py # noqa
- """
-
- def __init__(self, response, error_msg):
- """
-
- Args:
- response (Dict): CALCAT response.
- error_msg (Str): Error message to attach beside the response.
- """
- super().__init__(response, error_msg)
+ def __init__(self, response):
+ super().__init__(response["info"])
class ClientWrapper(type):
@@ -273,7 +134,7 @@ class CalCatApi(metaclass=ClientWrapper):
resp_detector = Detector.get_by_identifier(self.client, detector_name)
if not resp_detector["success"]:
- raise DetectorError(resp_detector, f"Detector name used is {detector_name}.")
+ raise CalCatError(resp_detector)
return {k: resp_detector["data"][k] for k in self.get_detector_keys}
@@ -291,11 +152,7 @@ class CalCatApi(metaclass=ClientWrapper):
)
if not resp_pdus["success"]:
- raise PDUError(
- resp_pdus,
- f"No PDU found for {detector_id} with "
- f"snapshot at {pdu_snapshot_at}."
- )
+ raise CalCatError(resp_pdus)
# Create dict based on requested keys: karabo_da, module number,
# or QxMx naming convention.
@@ -331,7 +188,7 @@ class CalCatApi(metaclass=ClientWrapper):
)
if not resp_calibration["success"]:
- raise CalibrationError(resp_calibration, f"Calibration name is {calibration_name}.")
+ raise CalCatError(resp_calibration)
return resp_calibration["data"]["id"]
@@ -342,7 +199,7 @@ class CalCatApi(metaclass=ClientWrapper):
resp_parameter = Parameter.get_by_name(self.client, param_name)
if not resp_parameter["success"]:
- raise ParameterError(resp_parameter, f"Parameter name is {param_name}.")
+ raise CalCatError(resp_parameter)
return resp_parameter["data"]["id"]
@@ -438,7 +295,7 @@ class CalCatApi(metaclass=ClientWrapper):
)
if not resp_versions["success"]:
- raise CCVError(resp_versions, f"Calibrations: {calibrations}.")
+ raise CalCatError(resp_versions)
for ccv in resp_versions["data"]:
try:
@@ -721,21 +578,16 @@ class CalibrationData:
"""
metadata = CCVMetadata()
- try:
- self._api.closest_ccv_by_time_by_condition(
- self.detector_name,
- calibrations or self.calibrations,
- self.condition,
- self.modules,
- event_at or self.event_at,
- pdu_snapshot_at or self.pdu_snapshot_at,
- metadata,
- module_naming=self.module_naming,
- )
- except CCVError as e:
- warning(
- f"Failed to retrieve calibration constants. Error: {e}"
- )
+ self._api.closest_ccv_by_time_by_condition(
+ self.detector_name,
+ calibrations or self.calibrations,
+ self.condition,
+ self.modules,
+ event_at or self.event_at,
+ pdu_snapshot_at or self.pdu_snapshot_at,
+ metadata,
+ module_naming=self.module_naming,
+ )
return metadata
def ndarray(
@@ -1022,42 +874,31 @@ class SplitConditionCalibrationData(CalibrationData):
dark_calibrations = sorted(
self.dark_calibrations & set(calibrations))
if dark_calibrations:
- try:
- self._api.closest_ccv_by_time_by_condition(
- self.detector_name,
- dark_calibrations,
- self.dark_condition,
- self.modules,
- event_at or self.event_at,
- pdu_snapshot_at or self.pdu_snapshot_at,
- metadata,
- module_naming=self.module_naming,
- )
- except CCVError as e:
- warning(
- "Failed to retrieve dark calibration constants. "
- f"Error: {e}"
- )
+ self._api.closest_ccv_by_time_by_condition(
+ self.detector_name,
+ dark_calibrations,
+ self.dark_condition,
+ self.modules,
+ event_at or self.event_at,
+ pdu_snapshot_at or self.pdu_snapshot_at,
+ metadata,
+ module_naming=self.module_naming,
+ )
illum_calibrations = sorted(
self.illuminated_calibrations & set(calibrations))
if illum_calibrations:
- try:
- self._api.closest_ccv_by_time_by_condition(
- self.detector_name,
- illum_calibrations,
- self.illuminated_condition,
- self.modules,
- event_at or self.event_at,
- pdu_snapshot_at or self.pdu_snapshot_at,
- metadata,
- module_naming=self.module_naming,
- )
- except CCVError as e:
- warning(
- "Failed to retrieve illuminated calibration constants. "
- f"Error: {e}"
- )
+ self._api.closest_ccv_by_time_by_condition(
+ self.detector_name,
+ illum_calibrations,
+ self.illuminated_condition,
+ self.modules,
+ event_at or self.event_at,
+ pdu_snapshot_at or self.pdu_snapshot_at,
+ metadata,
+ module_naming=self.module_naming,
+ )
+
return metadata
--
GitLab