Skip to content
Snippets Groups Projects
Commit ad1290fe authored by Karim Ahmed's avatar Karim Ahmed
Browse files

remove exceptions from calcatinterface

parent 61a4a963
No related branches found
No related tags found
1 merge request!844[PNCCD][CORRECT] Fix: Skip error for missing gain
Hide whitespace changes
Inline Side-by-side
src/cal_tools/calcat_interface.py
+ 38
− 197
View file @ ad1290fe
"""Interfaces to calibration constant data.""" """Interfaces to calibration constant data."""
from datetime import date, datetime, time, timezone from datetime import date, datetime, time, timezone
from functools import lru_cache from functools import lru_cache
from logging import warning
from pathlib import Path from pathlib import Path
from weakref import WeakKeyDictionary from weakref import WeakKeyDictionary
...@@ -57,146 +56,8 @@ class CCVMetadata(dict): ...@@ -57,146 +56,8 @@ class CCVMetadata(dict):
class CalCatError(Exception): class CalCatError(Exception):
"""CalCat API error.""" """CalCat API error."""
def __init__(self, response, error_msg): def __init__(self, response):
""" super().__init__(response["info"])
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)
class ClientWrapper(type): class ClientWrapper(type):
...@@ -273,7 +134,7 @@ class CalCatApi(metaclass=ClientWrapper): ...@@ -273,7 +134,7 @@ class CalCatApi(metaclass=ClientWrapper):
resp_detector = Detector.get_by_identifier(self.client, detector_name) resp_detector = Detector.get_by_identifier(self.client, detector_name)
if not resp_detector["success"]: 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} return {k: resp_detector["data"][k] for k in self.get_detector_keys}
...@@ -291,11 +152,7 @@ class CalCatApi(metaclass=ClientWrapper): ...@@ -291,11 +152,7 @@ class CalCatApi(metaclass=ClientWrapper):
) )
if not resp_pdus["success"]: if not resp_pdus["success"]:
raise PDUError( raise CalCatError(resp_pdus)
resp_pdus,
f"No PDU found for {detector_id} with "
f"snapshot at {pdu_snapshot_at}."
)
# Create dict based on requested keys: karabo_da, module number, # Create dict based on requested keys: karabo_da, module number,
# or QxMx naming convention. # or QxMx naming convention.
...@@ -331,7 +188,7 @@ class CalCatApi(metaclass=ClientWrapper): ...@@ -331,7 +188,7 @@ class CalCatApi(metaclass=ClientWrapper):
) )
if not resp_calibration["success"]: if not resp_calibration["success"]:
raise CalibrationError(resp_calibration, f"Calibration name is {calibration_name}.") raise CalCatError(resp_calibration)
return resp_calibration["data"]["id"] return resp_calibration["data"]["id"]
...@@ -342,7 +199,7 @@ class CalCatApi(metaclass=ClientWrapper): ...@@ -342,7 +199,7 @@ class CalCatApi(metaclass=ClientWrapper):
resp_parameter = Parameter.get_by_name(self.client, param_name) resp_parameter = Parameter.get_by_name(self.client, param_name)
if not resp_parameter["success"]: if not resp_parameter["success"]:
raise ParameterError(resp_parameter, f"Parameter name is {param_name}.") raise CalCatError(resp_parameter)
return resp_parameter["data"]["id"] return resp_parameter["data"]["id"]
...@@ -438,7 +295,7 @@ class CalCatApi(metaclass=ClientWrapper): ...@@ -438,7 +295,7 @@ class CalCatApi(metaclass=ClientWrapper):
) )
if not resp_versions["success"]: if not resp_versions["success"]:
raise CCVError(resp_versions, f"Calibrations: {calibrations}.") raise CalCatError(resp_versions)
for ccv in resp_versions["data"]: for ccv in resp_versions["data"]:
try: try:
...@@ -721,21 +578,16 @@ class CalibrationData: ...@@ -721,21 +578,16 @@ class CalibrationData:
""" """
metadata = CCVMetadata() metadata = CCVMetadata()
try: self._api.closest_ccv_by_time_by_condition(
self._api.closest_ccv_by_time_by_condition( self.detector_name,
self.detector_name, calibrations or self.calibrations,
calibrations or self.calibrations, self.condition,
self.condition, self.modules,
self.modules, event_at or self.event_at,
event_at or self.event_at, pdu_snapshot_at or self.pdu_snapshot_at,
pdu_snapshot_at or self.pdu_snapshot_at, metadata,
metadata, module_naming=self.module_naming,
module_naming=self.module_naming, )
)
except CCVError as e:
warning(
f"Failed to retrieve calibration constants. Error: {e}"
)
return metadata return metadata
def ndarray( def ndarray(
...@@ -1022,42 +874,31 @@ class SplitConditionCalibrationData(CalibrationData): ...@@ -1022,42 +874,31 @@ class SplitConditionCalibrationData(CalibrationData):
dark_calibrations = sorted( dark_calibrations = sorted(
self.dark_calibrations & set(calibrations)) self.dark_calibrations & set(calibrations))
if dark_calibrations: if dark_calibrations:
try: self._api.closest_ccv_by_time_by_condition(
self._api.closest_ccv_by_time_by_condition( self.detector_name,
self.detector_name, dark_calibrations,
dark_calibrations, self.dark_condition,
self.dark_condition, self.modules,
self.modules, event_at or self.event_at,
event_at or self.event_at, pdu_snapshot_at or self.pdu_snapshot_at,
pdu_snapshot_at or self.pdu_snapshot_at, metadata,
metadata, module_naming=self.module_naming,
module_naming=self.module_naming, )
)
except CCVError as e:
warning(
"Failed to retrieve dark calibration constants. "
f"Error: {e}"
)
illum_calibrations = sorted( illum_calibrations = sorted(
self.illuminated_calibrations & set(calibrations)) self.illuminated_calibrations & set(calibrations))
if illum_calibrations: if illum_calibrations:
try: self._api.closest_ccv_by_time_by_condition(
self._api.closest_ccv_by_time_by_condition( self.detector_name,
self.detector_name, illum_calibrations,
illum_calibrations, self.illuminated_condition,
self.illuminated_condition, self.modules,
self.modules, event_at or self.event_at,
event_at or self.event_at, pdu_snapshot_at or self.pdu_snapshot_at,
pdu_snapshot_at or self.pdu_snapshot_at, metadata,
metadata, module_naming=self.module_naming,
module_naming=self.module_naming, )
)
except CCVError as e:
warning(
"Failed to retrieve illuminated calibration constants. "
f"Error: {e}"
)
return metadata return metadata
......
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment

Legal Notice | Contacts