mx/AareCommon
0
0
Fork 0
Code Issues 1 Pull Requests 2 Actions Packages Projects Releases 6 Wiki Activity

fix: tidy exception classes #12

Open
perl_d wants to merge 1 commits from fix/rationalise_errors into main
pull from: fix/rationalise_errors
merge into: :main
Conversation 0 Commits 1 Files Changed 4
+53 -210
4 changed files with 53 additions and 210 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/aarecommon/errors/exception_handler.py
+45 -202
View File
@@ -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
tests/test_aare_exception.py
+3 -3
View File
@@ -31,7 +31,7 @@ from aarecommon.errors.exception_handler import (
JFJochCommunicationError,
JFJochException,
LoopCenteringFailed,
MagnetPositionSensorErorr,
MagnetPositionSensorError,
MaintenanceStateException,
ManualMountException,
MountingFailed,
@@ -73,7 +73,7 @@ def test_phase3_classes_have_critical_default_true():
MaintenanceStateException,
BeamlineBusyException,
BeamlineBusyTimeoutException,
MagnetPositionSensorErorr,
MagnetPositionSensorError,
SmartMagnetFaultException,
TransformationInvalidException,
):
@@ -149,7 +149,7 @@ def test_automation_errors_are_automation_error():
AareDBCommunicationError,
BeamlineBusyException,
RasterScanException,
MagnetPositionSensorErorr,
MagnetPositionSensorError,
SmartMagnetFaultException,
TransformationInvalidException,
AutoRasterSampleSkipped,
tests/test_error_codes.py
+2 -2
View File
@@ -52,7 +52,7 @@ def test_code_for_each_concrete_exception_is_in_aare_error_code_enum():
DataCollectionException,
JFJochCommunicationError,
LoopCenteringFailed,
MagnetPositionSensorErorr,
MagnetPositionSensorError,
MaintenanceStateException,
ManualMountException,
MountingFailed,
@@ -94,7 +94,7 @@ def test_code_for_each_concrete_exception_is_in_aare_error_code_enum():
AXCFailed,
AutoRasterSampleSkipped,
TransformationInvalidException,
MagnetPositionSensorErorr,
MagnetPositionSensorError,
SmartMagnetFaultException,
ManualMountException,
SampleException,
tests/test_exception_handler.py
+3 -3
View File
@@ -10,7 +10,7 @@ from aarecommon.errors.exception_handler import (
DataCollectionException,
JFJochCommunicationError,
LoopCenteringFailed,
MagnetPositionSensorErorr,
MagnetPositionSensorError,
ManualMountException,
MountingFailed,
RasterScanException,
@@ -152,7 +152,7 @@ def test_aerotech_communication_error():
def test_magnet_position_sensor_error():
exc = MagnetPositionSensorErorr()
exc = MagnetPositionSensorError()
assert "Magnet position sensor error" in str(exc)
@@ -178,7 +178,7 @@ def test_magnet_position_sensor_error():
lambda: JFJochCommunicationError("jce", operation="POST"),
lambda: AareDBCommunicationError("dbce"),
lambda: AerotechCommunicationError("ace"),
lambda: MagnetPositionSensorErorr(),
lambda: MagnetPositionSensorError(),
],
)
def test_exception_construction_emits_no_logs(factory, caplog):