src/aarecommon/config/logger.py

@@ -82,7 +82,7 @@ def get_config_path(env: str = "dev") -> Path:
 
 # TODO fix logging.
 def setup_logger(
-    name="aareDAQ", base_dir: str | None = f"~/tmp/mxlogs/", config_path: str | None = None
+    name="aareDAQ", base_dir: str | None = "~/tmp/mxlogs/", config_path: str | None = None
 ):
     # switch to production mode using: $ APP_ENV=prod python main.py
     env = os.getenv("APP_ENV", "dev")  # default: dev

src/aarecommon/errors/codes.py

@@ -75,9 +75,7 @@ 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"
 

src/aarecommon/errors/exception_handler.py

@@ -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

src/aarecommon/math/find_xtal.py

@@ -12,7 +12,7 @@ logger = setup_logger("aareDAQ")
 def identify_crystal_raster(result, r: RasterGridRequest) -> CenterOfMassModel | None:
     images = result.images
     if images and any(getattr(img, "spots_low_res", 0) for img in images):
-        logger.debug(f"Find image by maximum number of low resolution spots")
+        logger.debug("Find image by maximum number of low resolution spots")
         max_image = max(images, key=lambda img: img.spots_low_res)
         logger.debug(f"Image with maximum spots_low_res: {max_image}")
         logger.debug(f"Maximum spots_low_res value: {max_image.spots_low_res}")

src/aarecommon/models/auth.py

@@ -1,6 +1,5 @@
 from enum import Enum
 from pydantic import BaseModel
-from datetime import datetime
 
 
 class BatonRequestStatus(Enum):
@@ -58,7 +57,8 @@ class BatonStatus(BaseModel):
 
 
 def get_user():
-    import os, getpass
+    import os
+    import getpass
 
     try:
         return os.getlogin()

src/aarecommon/models/models.py

@@ -1,7 +1,6 @@
 import re
 from dataclasses import dataclass
 from enum import Enum
-from pathlib import Path
 from typing import Annotated, List, Literal, Optional, Tuple
 
 from aarecommon.models.beamline import MXBeamline
@@ -102,11 +101,7 @@ class DataCollectionParameters(BaseModel):
             default_value = "{date}/{prefix}"
             return default_value
 
-        # Strip trailing slashes and store original value for comparison
         v = str(v).strip("/")  # Ensure it's a string and no trailing slashes
-        original_value = v
-
-        # Replace spaces with underscores
         v = v.replace(" ", "_")
 
         # Validate directory pattern with macros and allowed characters

tests/test_aare_exception.py

@@ -9,7 +9,6 @@ These cover:
 
 from __future__ import annotations
 
-import pytest
 from aarecommon.errors.exception_handler import (
     AareAuthError,
     AareDBCommunicationError,
@@ -31,7 +30,7 @@ from aarecommon.errors.exception_handler import (
     JFJochCommunicationError,
     JFJochException,
     LoopCenteringFailed,
-    MagnetPositionSensorErorr,
+    MagnetPositionSensorError,
     MaintenanceStateException,
     ManualMountException,
     MountingFailed,
@@ -57,48 +56,50 @@ from aarecommon.errors.exception_handler import (
 
 
 def test_aare_exception_class_critical_default_false():
-    assert AareException.critical is False
+    assert AareException().critical is False
 
 
 def test_phase3_classes_have_critical_default_true():
-    for cls in (
-        TellCommunicationError,
-        TellConnectionException,
-        CriticalTellException,
-        SmargonCommunicationError,
-        AerotechCommunicationError,
-        JFJochCommunicationError,
-        BECCommunicationError,
-        StateTransitionFailed,
-        MaintenanceStateException,
-        BeamlineBusyException,
-        BeamlineBusyTimeoutException,
-        MagnetPositionSensorErorr,
-        SmartMagnetFaultException,
-        TransformationInvalidException,
+    for e in (
+        TellCommunicationError(),
+        TellConnectionException(),
+        CriticalTellException(),
+        SmargonCommunicationError(),
+        AerotechCommunicationError(),
+        JFJochCommunicationError(),
+        BECCommunicationError(),
+        StateTransitionFailed(),
+        MaintenanceStateException(),
+        BeamlineBusyException(),
+        BeamlineBusyTimeoutException(),
+        MagnetPositionSensorError(),
+        SmartMagnetFaultException(),
+        TransformationInvalidException(),
     ):
-        assert cls.critical is True, f"{cls.__name__} should default to critical=True in Phase 3"
+        assert e.critical is True, (
+            f"{e.__class__.__name__} should default to critical=True in Phase 3"
+        )
 
 
 def test_other_core_classes_keep_critical_default_false():
-    for cls in (
-        AutomationError,
-        AareUserError,
-        AareAuthError,
-        TellException,
-        SmargonException,
-        AerotechException,
-        JFJochException,
-        BECException,
-        BeamlineStateException,
-        MountingFailed,
-        UnmountingFailed,
-        LoopCenteringFailed,
-        TellCommandWhileBusyException,
-        WarningTellException,
-        AareDBCommunicationError,
+    for e in (
+        AutomationError(),
+        AareUserError(),
+        AareAuthError(),
+        TellException(),
+        SmargonException(),
+        AerotechException(),
+        JFJochException(),
+        BECException(),
+        BeamlineStateException(),
+        MountingFailed(),
+        UnmountingFailed(),
+        LoopCenteringFailed(),
+        TellCommandWhileBusyException(),
+        WarningTellException(),
+        AareDBCommunicationError(),
     ):
-        assert cls.critical is False, f"{cls.__name__} should remain critical=False"
+        assert e.critical is False, f"{e.__class__.__name__} should remain critical=False"
 
 
 # ---------------------------------------------------------------------------
@@ -110,7 +111,7 @@ def test_instance_critical_override_true():
     exc = MountingFailed("mount fail", critical=True)
     assert exc.critical is True
     # class default unchanged
-    assert MountingFailed.critical is False
+    assert MountingFailed().critical is False
 
 
 def test_instance_critical_override_false():
@@ -149,7 +150,7 @@ def test_automation_errors_are_automation_error():
         AareDBCommunicationError,
         BeamlineBusyException,
         RasterScanException,
-        MagnetPositionSensorErorr,
+        MagnetPositionSensorError,
         SmartMagnetFaultException,
         TransformationInvalidException,
         AutoRasterSampleSkipped,

tests/test_error_codes.py

@@ -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

@@ -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):

tests/test_models_extra.py

@@ -1,4 +1,3 @@
-import pytest
 from aarecommon.models.models import (
     BeamlineStateEnum,
     BeamMarkCoeffModel,