Source code for mafw.steering_gui.utils.exception_handling

#  Copyright 2026 European Union
#  Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
#  SPDX-License-Identifier: EUPL-1.2
"""GUI exception handling utilities for the steering application.

:Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
:Description: Provide a reusable PySide6-compatible exception handler that
    surfaces errors from the GUI and worker threads via dialog boxes.
"""

from __future__ import annotations

import sys
import threading
import traceback
from types import TracebackType
from typing import Any, Callable

from PySide6.QtCore import QObject, Qt, Signal
from PySide6.QtWidgets import QApplication, QMessageBox

_DisplayCallback = Callable[[type[BaseException] | None, BaseException | None, TracebackType | None], Any]
"""Type alias for exception display callbacks."""


[docs] class ThreadExceptionHandler(QObject): """Handle exceptions raised in non-GUI threads and forward them to the GUI.""" exception_caught: Signal = Signal(object, object, object) def __init__(self, display_callback: _DisplayCallback | None = None) -> None: """Initialize the handler and connect its signal to a display callback. :param display_callback: Optional callable to show the exception details. :type display_callback: Callable[[type[BaseException] | None, BaseException | None, TracebackType | None], Any] """ super().__init__() callback = display_callback or show_exception_dialog self.exception_caught.connect(callback, Qt.ConnectionType.QueuedConnection)
[docs] def handle_exception( self, exc_type: type[BaseException] | None, exc_value: BaseException | None, exc_traceback: TracebackType | None, ) -> None: """Emit an exception signal so it can be shown on the GUI thread. :param exc_type: Type of the exception. :type exc_type: Type[BaseException] | None :param exc_value: The exception instance. :type exc_value: BaseException | None :param exc_traceback: Traceback object. :type exc_traceback: TracebackType | None """ self.exception_caught.emit(exc_type, exc_value, exc_traceback)
[docs] def show_exception_dialog( exception_type: type[BaseException] | None, exception_value: BaseException | None, exception_traceback: TracebackType | None, ) -> None: """Display an exception in a message box dialog. :param exception_type: Type of the exception. :type exception_type: Type[BaseException] | None :param exception_value: The exception instance. :type exception_value: BaseException | None :param exception_traceback: Traceback object. :type exception_traceback: TracebackType | None """ error_msg = ''.join(traceback.format_exception(exception_type, exception_value, exception_traceback)) # if the QApplication is not yet available, drop the error message on the stderr and return if QApplication.instance() is None: print(error_msg, file=sys.stderr) return try: error_box = QMessageBox() error_box.setIcon(QMessageBox.Icon.Critical) error_box.setWindowTitle('Error') message = str(exception_value) if exception_value is not None else 'Unexpected error.' error_box.setText(f'An error occurred: {message}') error_box.setDetailedText(error_msg) error_box.setStandardButtons(QMessageBox.StandardButton.Ok) error_box.exec() finally: print(error_msg, file=sys.stderr)
_thread_exception_handler: ThreadExceptionHandler | None = None """Global instance of the thread exception handler."""
[docs] def get_thread_exception_handler() -> ThreadExceptionHandler: """Get or create the global thread exception handler. :return: The global thread exception handler instance. :rtype: ThreadExceptionHandler """ global _thread_exception_handler if _thread_exception_handler is None: _thread_exception_handler = ThreadExceptionHandler() return _thread_exception_handler
[docs] def install_exception_handler() -> Callable[[type[BaseException], BaseException, TracebackType | None], Any]: """Install a global exception handler to show errors in a dialog box. :return: The original sys.excepthook implementation. :rtype: Callable[[Type[BaseException], BaseException, TracebackType | None], Any] """ original_sys_excepthook = sys.excepthook original_thread_excepthook = getattr(threading, 'excepthook', None) def _sys_excepthook( exc_type: type[BaseException], exc_value: BaseException, exc_traceback: TracebackType | None, ) -> None: if exc_type in (KeyboardInterrupt, SystemExit): original_sys_excepthook(exc_type, exc_value, exc_traceback) return show_exception_dialog(exc_type, exc_value, exc_traceback) sys.excepthook = _sys_excepthook handler = get_thread_exception_handler() if original_thread_excepthook is not None: def _thread_excepthook(args: threading.ExceptHookArgs) -> None: if args.exc_type in (KeyboardInterrupt, SystemExit): original_thread_excepthook(args) return handler.handle_exception(args.exc_type, args.exc_value, args.exc_traceback) threading.excepthook = _thread_excepthook return original_sys_excepthook