fix: tidy exception classes
This commit is contained in:
@@ -75,8 +75,8 @@ class AareErrorCode(StrEnum):
|
||||
AXC_FAILED = "AXC_FAILED"
|
||||
AUTO_RASTER_SAMPLE_SKIPPED = "AUTO_RASTER_SAMPLE_SKIPPED"
|
||||
TRANSFORMATION_INVALID_EXCEPTION = "TRANSFORMATION_INVALID_EXCEPTION"
|
||||
MAGNET_POSITION_SENSOR_ERORR = (
|
||||
"MAGNET_POSITION_SENSOR_ERORR" # NOTE: class name "Erorr" has a typo; preserved
|
||||
MAGNET_POSITION_SENSOR_ERROR = (
|
||||
"MAGNET_POSITION_SENSOR_ERROR"
|
||||
)
|
||||
SMART_MAGNET_FAULT_EXCEPTION = "SMART_MAGNET_FAULT_EXCEPTION"
|
||||
DOOR_SAFETY_ERROR = "DOOR_SAFETY_ERROR"
|
||||
|
||||
@@ -1,6 +1,5 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
from typing import ClassVar
|
||||
|
||||
from aarecommon.errors.codes import AuthErrorCode
|
||||
@@ -10,19 +9,18 @@ logger = setup_logger("aareDAQ")
|
||||
|
||||
|
||||
class AareException(Exception):
|
||||
"""Root of the aare exception hierarchy.
|
||||
"""Root of the aare exception hierarchy."""
|
||||
|
||||
Class-level ``critical`` is the default; pass ``critical=...`` to the
|
||||
constructor of an "optionally critical" subclass to override on a single
|
||||
raise site.
|
||||
"""
|
||||
DEFAULT_CRITICALITY: ClassVar[bool] = False
|
||||
DEFAULT_MESSAGE: ClassVar[str] = "Base Aare Exception"
|
||||
|
||||
critical: ClassVar[bool] = False
|
||||
def __init__(self, message: str | None = None, *, critical: bool | None = None, **kwargs):
|
||||
super().__init__(message, **kwargs)
|
||||
self.critical = critical if critical is not None else self.__class__.DEFAULT_CRITICALITY
|
||||
self.message = message if message is not None else self.__class__.DEFAULT_MESSAGE
|
||||
|
||||
def __init__(self, *args, critical: bool | None = None, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
if critical is not None:
|
||||
self.critical = critical
|
||||
def __str__(self) -> str:
|
||||
return self.message
|
||||
|
||||
|
||||
class AutomationError(AareException):
|
||||
@@ -76,62 +74,31 @@ class BeamlineStateException(AutomationError):
|
||||
|
||||
|
||||
class TransformationInvalidException(AutomationError):
|
||||
critical: ClassVar[bool] = True
|
||||
|
||||
def __init__(
|
||||
self, message: str = "Transformation is not implemented", *, critical: bool | None = None
|
||||
):
|
||||
super().__init__(message, critical=critical)
|
||||
self.message = message
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.message
|
||||
DEFAULT_CRITICALITY = True
|
||||
DEFAULT_MESSAGE = "Transformation is not implemented"
|
||||
|
||||
|
||||
class StateTransitionFailed(BeamlineStateException):
|
||||
critical: ClassVar[bool] = True
|
||||
DEFAULT_CRITICALITY = True
|
||||
|
||||
def __init__(
|
||||
self, message: str = "Beamline state transition failed", *, critical: bool | None = None
|
||||
):
|
||||
def __init__(self, message: str = "Beamline state transition failed", *, critical: bool = True):
|
||||
super().__init__(message, critical=critical)
|
||||
self.message = message
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.message
|
||||
|
||||
|
||||
class MaintenanceStateException(BeamlineStateException):
|
||||
critical: ClassVar[bool] = True
|
||||
|
||||
def __init__(
|
||||
self, message: str = "Beamline is in Maintenance state", *, critical: bool | None = None
|
||||
):
|
||||
super().__init__(message, critical=critical)
|
||||
self.message = message
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.message
|
||||
DEFAULT_CRITICALITY = True
|
||||
DEFAULT_MESSAGE = "Beamline is in Maintenance state"
|
||||
|
||||
|
||||
class DataCollectionException(AutomationError):
|
||||
"""Group parent for data-collection-time exceptions; also raisable directly."""
|
||||
|
||||
def __init__(self, message: str = "Data collection failed", *, critical: bool | None = None):
|
||||
super().__init__(message, critical=critical)
|
||||
self.message = message
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.message
|
||||
DEFAULT_MESSAGE = "Data collection failed"
|
||||
|
||||
|
||||
class RasterScanException(DataCollectionException):
|
||||
def __init__(self, message: str = "Data collection failed", *, critical: bool | None = None):
|
||||
super().__init__(message, critical=critical)
|
||||
self.message = message
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.message
|
||||
DEFAULT_MESSAGE = "Data collection failed"
|
||||
|
||||
|
||||
class AutoRasterSampleSkipped(AutomationError):
|
||||
@@ -139,65 +106,32 @@ class AutoRasterSampleSkipped(AutomationError):
|
||||
|
||||
|
||||
class LoopCenteringFailed(AutomationError):
|
||||
def __init__(
|
||||
self,
|
||||
message: str = "Loop Centering did not detect a sample",
|
||||
*,
|
||||
critical: bool | None = None,
|
||||
):
|
||||
super().__init__(message, critical=critical)
|
||||
self.message = message
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.message
|
||||
DEFAULT_MESSAGE = "Loop Centering did not detect a sample"
|
||||
|
||||
|
||||
class UnmountingFailed(TellException):
|
||||
"""Sample failed to unmount. Optionally critical via instance flag."""
|
||||
|
||||
def __init__(
|
||||
self, message: str = "A sample was not unmounted", *, critical: bool | None = None
|
||||
):
|
||||
super().__init__(message, critical=critical)
|
||||
self.message = message
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.message
|
||||
DEFAULT_MESSAGE = "A sample was not unmounted"
|
||||
|
||||
|
||||
class MountingFailed(TellException):
|
||||
"""Sample failed to mount. Optionally critical via instance flag."""
|
||||
|
||||
def __init__(self, message: str = "A sample was not mounted", *, critical: bool | None = None):
|
||||
super().__init__(message, critical=critical)
|
||||
self.message = message
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.message
|
||||
DEFAULT_MESSAGE = "A sample was not mounted"
|
||||
|
||||
|
||||
class ManualMountException(AareUserError):
|
||||
"""Custom exception for manual mounting"""
|
||||
|
||||
def __init__(self, message: str = "Manual mounting failed", *, critical: bool | None = None):
|
||||
super().__init__(message, critical=critical)
|
||||
self.message = message
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.message
|
||||
DEFAULT_MESSAGE = "Manual mounting failed"
|
||||
|
||||
|
||||
class SmartMagnetFaultException(AutomationError):
|
||||
"""Custom exception for smart magnet fault"""
|
||||
|
||||
critical: ClassVar[bool] = True
|
||||
|
||||
def __init__(self, message: str = "Smart magnet fault", *, critical: bool | None = None):
|
||||
super().__init__(message, critical=critical)
|
||||
self.message = message
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.message
|
||||
DEFAULT_CRITICALITY = True
|
||||
DEFAULT_MESSAGE = "Smart magnet fault"
|
||||
|
||||
|
||||
class DoorSafetyError(AutomationError):
|
||||
@@ -209,98 +143,44 @@ class DoorSafetyError(AutomationError):
|
||||
critical so automation halts and the GUI shows a pop-up.
|
||||
"""
|
||||
|
||||
critical: ClassVar[bool] = True
|
||||
|
||||
def __init__(
|
||||
self, message: str = "Door safety could not be activated", *, critical: bool | None = None
|
||||
):
|
||||
super().__init__(message, critical=critical)
|
||||
self.message = message
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.message
|
||||
DEFAULT_CRITICALITY = True
|
||||
DEFAULT_MESSAGE = "Door safety could not be activated"
|
||||
|
||||
|
||||
class TellCommandWhileBusyException(TellException):
|
||||
"""Custom exception for trying to move Tell when it is busy"""
|
||||
|
||||
def __init__(self, message: str = "Tell is busy", *, critical: bool | None = None):
|
||||
super().__init__(message, critical=critical)
|
||||
self.message = message
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.message
|
||||
DEFAULT_MESSAGE = "Tell is busy"
|
||||
|
||||
|
||||
class TellConnectionException(TellException):
|
||||
"""Custom exception for connection problems"""
|
||||
|
||||
critical: ClassVar[bool] = True
|
||||
|
||||
def __init__(self, message: str = "Lost connection to Tell", *, critical: bool | None = None):
|
||||
super().__init__(message, critical=critical)
|
||||
self.message = message
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.message
|
||||
DEFAULT_CRITICALITY = True
|
||||
DEFAULT_MESSAGE = "Lost connection to Tell"
|
||||
|
||||
|
||||
class WarningTellException(TellException):
|
||||
def __init__(self, message: str = "Warning error in TELL", *, critical: bool | None = None):
|
||||
super().__init__(message, critical=critical)
|
||||
self.message = message
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.message
|
||||
DEFAULT_MESSAGE = "Warning error in TELL"
|
||||
|
||||
|
||||
class CriticalTellException(TellException):
|
||||
critical: ClassVar[bool] = True
|
||||
|
||||
def __init__(self, message: str = "Critical error in TELL", *, critical: bool | None = None):
|
||||
super().__init__(message, critical=critical)
|
||||
self.message = message
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.message
|
||||
DEFAULT_CRITICALITY = True
|
||||
DEFAULT_MESSAGE = "Critical error in TELL"
|
||||
|
||||
|
||||
class AXCFailed(AutomationError):
|
||||
def __init__(
|
||||
self, message: str = "Auto X-ray centering failed", *, critical: bool | None = None
|
||||
):
|
||||
super().__init__(message, critical=critical)
|
||||
self.message = message
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.message
|
||||
DEFAULT_MESSAGE = "Auto X-ray centering failed"
|
||||
|
||||
|
||||
class BeamlineBusyException(BeamlineStateException):
|
||||
critical: ClassVar[bool] = True
|
||||
|
||||
def __init__(self, message: str = "Beamline is in busy state", *, critical: bool | None = None):
|
||||
super().__init__(message, critical=critical)
|
||||
self.message = message
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.message
|
||||
DEFAULT_CRITICALITY = True
|
||||
DEFAULT_MESSAGE = "Beamline is in busy state"
|
||||
|
||||
|
||||
class BeamlineBusyTimeoutException(BeamlineStateException):
|
||||
critical: ClassVar[bool] = True
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
message: str = "Beamline busy state expired during operation",
|
||||
*,
|
||||
critical: bool | None = None,
|
||||
):
|
||||
super().__init__(message, critical=critical)
|
||||
self.message = message
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.message
|
||||
DEFAULT_CRITICALITY = True
|
||||
DEFAULT_MESSAGE = "Beamline busy state expired during operation"
|
||||
|
||||
|
||||
class SampleException(AareUserError):
|
||||
@@ -310,12 +190,7 @@ class SampleException(AareUserError):
|
||||
AareUserError pending confirmation -- "sample not found" reads as bad
|
||||
input rather than a runtime failure."""
|
||||
|
||||
def __init__(self, message: str = "Sample not found", *, critical: bool | None = None):
|
||||
super().__init__(message, critical=critical)
|
||||
self.message = message
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.message
|
||||
DEFAULT_MESSAGE = "Sample not found"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -339,9 +214,6 @@ class AuthenticationException(AareAuthError):
|
||||
self.headers = headers
|
||||
self.code = code
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.message
|
||||
|
||||
|
||||
class UserRightsException(AareAuthError):
|
||||
_last_log_ts_by_message: dict[str, float] = {}
|
||||
@@ -362,9 +234,6 @@ class UserRightsException(AareAuthError):
|
||||
self.headers = headers
|
||||
self.code = code
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.message
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Device communication exceptions
|
||||
@@ -377,7 +246,7 @@ class SmargonCommunicationError(SmargonException):
|
||||
Keep the original exception in `__cause__` by using `raise ... from e`.
|
||||
"""
|
||||
|
||||
critical: ClassVar[bool] = True
|
||||
DEFAULT_CRITICALITY = True
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
@@ -396,9 +265,6 @@ class SmargonCommunicationError(SmargonException):
|
||||
self.operation = operation
|
||||
self.status_code = status_code
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.message
|
||||
|
||||
|
||||
class TellCommunicationError(TellException):
|
||||
"""
|
||||
@@ -406,7 +272,7 @@ class TellCommunicationError(TellException):
|
||||
Intended to be caught centrally by FastAPI exception handlers.
|
||||
"""
|
||||
|
||||
critical: ClassVar[bool] = True
|
||||
DEFAULT_CRITICALITY = True
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
@@ -423,9 +289,6 @@ class TellCommunicationError(TellException):
|
||||
self.base_url = base_url
|
||||
self.operation = operation
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.message
|
||||
|
||||
|
||||
class JFJochCommunicationError(JFJochException):
|
||||
"""
|
||||
@@ -433,7 +296,7 @@ class JFJochCommunicationError(JFJochException):
|
||||
Intended for scan-time fallbacks and GUI-visible alerts.
|
||||
"""
|
||||
|
||||
critical: ClassVar[bool] = True
|
||||
DEFAULT_CRITICALITY = True
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
@@ -452,9 +315,6 @@ class JFJochCommunicationError(JFJochException):
|
||||
self.base_url = base_url
|
||||
self.status_code = status_code
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.message
|
||||
|
||||
|
||||
class AareDBCommunicationError(AutomationError):
|
||||
"""Raised when AareDB HTTPS communication fails.
|
||||
@@ -479,9 +339,6 @@ class AareDBCommunicationError(AutomationError):
|
||||
self.base_url = base_url
|
||||
self.status_code = status_code
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.message
|
||||
|
||||
|
||||
class AerotechCommunicationError(AerotechException):
|
||||
"""
|
||||
@@ -489,7 +346,7 @@ class AerotechCommunicationError(AerotechException):
|
||||
Keep the original exception in `__cause__` by using `raise ... from e`.
|
||||
"""
|
||||
|
||||
critical: ClassVar[bool] = True
|
||||
DEFAULT_CRITICALITY = True
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
@@ -508,25 +365,14 @@ class AerotechCommunicationError(AerotechException):
|
||||
self.operation = operation
|
||||
self.status_code = status_code
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.message
|
||||
|
||||
|
||||
class MagnetPositionSensorErorr(AutomationError):
|
||||
critical: ClassVar[bool] = True
|
||||
|
||||
def __init__(
|
||||
self, message: str = "Magnet position sensor error", *, critical: bool | None = None
|
||||
):
|
||||
super().__init__(message, critical=critical)
|
||||
self.message = message
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.message
|
||||
class MagnetPositionSensorError(AutomationError):
|
||||
DEFAULT_CRITICALITY = True
|
||||
DEFAULT_MESSAGE = "Magnet position sensor error"
|
||||
|
||||
|
||||
class BECCommunicationError(BECException):
|
||||
critical: ClassVar[bool] = True
|
||||
DEFAULT_CRITICALITY = True
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
@@ -544,6 +390,3 @@ class BECCommunicationError(BECException):
|
||||
self.exception = exception
|
||||
self.endpoint = endpoint
|
||||
self.base_url = base_url
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.message
|
||||
|
||||
Reference in New Issue
Block a user