Skip to content
Snippets Groups Projects
Commit ca74aab2 authored by Karim Ahmed's avatar Karim Ahmed Committed by Philipp Schmidt
Browse files

catch CALCATAPIError and create exception classes for two cases in webservice

parent 21173e03
No related branches found
No related tags found
1 merge request!993[Webservice] Enable two runs detectors
......@@ -33,6 +33,16 @@ class MDC:
NOTHING_TO_DO = "Nothing to calibrate for this run, copied raw data only"
class PDUsNotFoundError(Exception):
"""Exception raised when a detector cannot be found."""
pass
class OperationModeNotFoundError(Exception):
"""Exception raised when an operation mode cannot be found for a given detector."""
pass
class Success:
UPLOADED_CONFIG = "SUCCESS: Uploaded config for cycle {}, proposal {}"
START_CORRECTION = "SUCCESS: Started correction: proposal {}, run {}"
......
......@@ -31,10 +31,17 @@ from metadata_client.metadata_client import MetadataClient
try:
from .config import webservice as config
from .messages import MDC, Errors, MigrationError, Success
from .messages import (
MDC,
Errors,
MigrationError,
OperationModeNotFoundError,
PDUsNotFoundError,
Success,
)
except ImportError:
from config import webservice as config
from messages import MDC, Errors, MigrationError, Success
from messages import MDC, Errors, MigrationError, Success, PDUsNotFoundError, OperationModeNotFoundError
from cal_tools.calcat_interface2 import get_client
......@@ -347,23 +354,37 @@ def get_number_of_characterization_runs(karabo_id, operation_mode):
"""
client = get_client()
detector_id = client.detector_by_identifier(karabo_id)['id']
pdus = client.get(
'physical_detector_units/get_all_by_detector?'
f'detector_id={detector_id}')
first_pdu_id = pdus[0]['detector_type']['id']
operation_modes = client.get(
'operation_modes/get_all_by_detector_type?'
f'detector_type_id={first_pdu_id}')
try:
pdus = client.get(
'physical_detector_units/get_all_by_detector?'
f'detector_id={detector_id}')
if len(pdus) == 0:
raise PDUsNotFoundError
except Exception as e:
raise PDUsNotFoundError(
f'Failed to find physical detector units for {karabo_id}: {e}')
det_type = pdus[0]['detector_type']
try:
operation_modes = client.get(
'operation_modes/get_all_by_detector_type?'
f'detector_type_id={det_type["id"]}')
if len(operation_modes) == 0:
raise OperationModeNotFoundError
except Exception as e:
raise OperationModeNotFoundError(
f'Failed to find operation modes for {det_type["name"]}: {e}')
number_of_runs = next(
(
mode['number_of_runs'] for mode in operation_modes
if mode['identifier'] == operation_mode),
None)
if number_of_runs is None:
raise ValueError(
f"{karabo_id} doesn't have this operation mode {operation_mode}.")
raise OperationModeNotFoundError(
f'{karabo_id} does not have this operation mode {operation_mode}.')
return number_of_runs
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment