From 7a7d4a8e74d9794edfb5080442e498b01c3d7409 Mon Sep 17 00:00:00 2001 From: wakonig_k Date: Thu, 11 Dec 2025 17:22:21 +0100 Subject: [PATCH] feat: pretty print errors on scan history access --- bec_ipython_client/bec_ipython_client/main.py | 42 +---- bec_lib/bec_lib/alarm_handler.py | 41 +---- bec_lib/bec_lib/bec_errors.py | 49 +++--- bec_lib/bec_lib/messages.py | 1 + bec_lib/bec_lib/scan_history.py | 26 ++- bec_lib/bec_lib/utils/error_pretty_print.py | 151 ++++++++++++++++++ bec_lib/tests/test_error_pretty_print.py | 28 ++++ 7 files changed, 239 insertions(+), 99 deletions(-) create mode 100644 bec_lib/bec_lib/utils/error_pretty_print.py create mode 100644 bec_lib/tests/test_error_pretty_print.py diff --git a/bec_ipython_client/bec_ipython_client/main.py b/bec_ipython_client/bec_ipython_client/main.py index ccf549127..633cc022d 100644 --- a/bec_ipython_client/bec_ipython_client/main.py +++ b/bec_ipython_client/bec_ipython_client/main.py @@ -24,7 +24,7 @@ from bec_ipython_client.signals import OperationMode, ScanInterruption, SigintHandler from bec_lib import plugin_helper from bec_lib.alarm_handler import AlarmBase -from bec_lib.bec_errors import DeviceConfigError, ExceptionWithErrorInfo +from bec_lib.bec_errors import BECError, DeviceConfigError, ExceptionWithErrorInfo from bec_lib.bec_service import parse_cmdline_args from bec_lib.callback_handler import EventType from bec_lib.client import BECClient @@ -214,44 +214,18 @@ def show_last_alarm(self, offset: int = 0): except IndexError: print("No alarm has been raised in this session.") return - - console = Console() - - # --- HEADER --- - header = Text() - header.append("Alarm Raised\n", style="bold red") - header.append(f"Severity: {alarm.severity.name}\n", style="bold") - header.append(f"Type: {alarm.alarm_type}\n", style="bold") - if alarm.alarm.info.device: - header.append(f"Device: {alarm.alarm.info.device}\n", style="bold") - - console.print(Panel(header, title="Alarm Info", border_style="red", expand=False)) - - # --- SHOW SUMMARY - if alarm.alarm.info.compact_error_message: - console.print( - Panel( - Text(alarm.alarm.info.compact_error_message, style="yellow"), - title="Summary", - border_style="yellow", - expand=False, - ) - ) - - # --- SHOW FULL TRACEBACK - tb_str = alarm.alarm.info.error_message - if tb_str: - try: - console.print(tb_str) - except Exception: - # fallback in case msg is not a traceback - console.print(Panel(tb_str, title="Message", border_style="cyan")) + if hasattr(alarm, "print_details"): + alarm.print_details() + return + if hasattr(alarm, "pretty_print"): + alarm.pretty_print() + return def _ip_exception_handler( self, etype, evalue, tb, tb_offset=None, parent: BECIPythonClient = None, **kwargs ): - if issubclass(etype, AlarmBase): + if issubclass(etype, (AlarmBase, BECError)): parent._alarm_history.append((etype, evalue, tb, tb_offset)) log_console_error(etype, evalue, tb) print("\x1b[31m BEC alarm:\x1b[0m") diff --git a/bec_lib/bec_lib/alarm_handler.py b/bec_lib/bec_lib/alarm_handler.py index e8e2c1495..9016c13fe 100644 --- a/bec_lib/bec_lib/alarm_handler.py +++ b/bec_lib/bec_lib/alarm_handler.py @@ -9,14 +9,10 @@ from collections import deque from typing import TYPE_CHECKING -from rich.console import Console, Group -from rich.panel import Panel -from rich.syntax import Syntax -from rich.text import Text - from bec_lib.endpoints import MessageEndpoints from bec_lib.logger import bec_logger from bec_lib.utils import threadlocked +from bec_lib.utils.error_pretty_print import AlarmPrettyPrinter if TYPE_CHECKING: # pragma: no cover from bec_lib import messages @@ -42,6 +38,7 @@ def __init__(self, alarm: messages.AlarmMessage, severity: Alarms, handled=False self.severity = severity self.handled = handled self.alarm_type = alarm.info.exception_type + self._pretty_printer = AlarmPrettyPrinter(alarm.info, severity) super().__init__(self.alarm.content) def __str__(self) -> str: @@ -52,38 +49,10 @@ def __str__(self) -> str: ) def pretty_print(self) -> None: - """ - Use Rich to pretty print the alarm message, - following the same logic used in __str__(). - """ - console = Console() + self._pretty_printer.pretty_print() - msg = self.alarm.info.compact_error_message or self.alarm.info.error_message - - text = Text() - text.append(f"{self.alarm_type} | ", style="bold") - text.append(f"Severity {self.severity.name}", style="bold yellow") - if self.alarm.info.device: - text.append(f" | Device {self.alarm.info.device}\n", style="bold") - text.append("\n") - - renderables = [] - # Format message inside a syntax box if it looks like traceback - if "Traceback (most recent call last):" in msg: - renderables.append(Syntax(msg.strip(), "python", word_wrap=True)) - else: - renderables.append(Text(msg.strip())) - - if self.alarm.info.device: - renderables.append( - Text( - f"\n\nThe error is likely unrelated to BEC. Please check the device '{self.alarm.info.device}'.", - style="bold", - ) - ) - body = Group(*renderables) - - console.print(Panel(body, title=text, border_style="red", expand=True)) + def print_details(self) -> None: + self._pretty_printer.print_details() def __eq__(self, other: object) -> bool: if not isinstance(other, AlarmBase): diff --git a/bec_lib/bec_lib/bec_errors.py b/bec_lib/bec_lib/bec_errors.py index 0fe93e199..eaf63c57f 100644 --- a/bec_lib/bec_lib/bec_errors.py +++ b/bec_lib/bec_lib/bec_errors.py @@ -7,10 +7,27 @@ import traceback from typing import TYPE_CHECKING -if TYPE_CHECKING: +from bec_lib.utils.error_pretty_print import ErrorInfoPrettyPrinter + +if TYPE_CHECKING: # pragma: no cover from bec_lib import messages +class BECError(Exception): + """Base class for all BEC exceptions""" + + def __init__(self, message: str, error_info: messages.ErrorInfo) -> None: + super().__init__(message) + self.error_info = error_info + self._pretty_printer = ErrorInfoPrettyPrinter(error_info) + + def pretty_print(self) -> None: + self._pretty_printer.pretty_print() + + def print_details(self) -> None: + self._pretty_printer.print_details() + + class ScanAbortion(Exception): """Scan abortion exception""" @@ -41,38 +58,17 @@ class ExceptionWithErrorInfo(Exception): def __init__(self, error_info: messages.ErrorInfo): super().__init__(error_info.error_message) self.error_info = error_info + self._pretty_printer = ErrorInfoPrettyPrinter(error_info) def __str__(self) -> str: msg = self.error_info.compact_error_message return f"{self.__class__.__name__}: {msg}" if msg else super().__str__() def pretty_print(self) -> None: - """ - Use Rich to pretty print the error message, - following the same logic used in __str__(). - """ - from rich.console import Console, Group - from rich.panel import Panel - from rich.syntax import Syntax - from rich.text import Text - - console = Console() - msg = self.error_info.compact_error_message or self.error_info.error_message - - text = Text() - text.append(f"{self.__class__.__name__}", style="bold") - text.append("\n") + self._pretty_printer.pretty_print() - renderables = [] - # Format message inside a syntax box if it looks like traceback - if "Traceback (most recent call last):" in msg: - renderables.append(Syntax(msg.strip(), "python", word_wrap=True)) - else: - renderables.append(Text(msg.strip())) - - body = Group(*renderables) - - console.print(Panel(body, title=text, border_style="red", expand=True)) + def print_details(self) -> None: + self._pretty_printer.print_details() class ScanInputValidationError(ExceptionWithErrorInfo): @@ -80,7 +76,6 @@ class ScanInputValidationError(ExceptionWithErrorInfo): def __init__(self, error_info: messages.ErrorInfo): super().__init__(error_info) - self.error_info = error_info @classmethod def with_error_info(cls, message: str) -> ScanInputValidationError: diff --git a/bec_lib/bec_lib/messages.py b/bec_lib/bec_lib/messages.py index 845899481..7e86b65c8 100644 --- a/bec_lib/bec_lib/messages.py +++ b/bec_lib/bec_lib/messages.py @@ -560,6 +560,7 @@ class ErrorInfo(BaseModel): compact_error_message: str | None exception_type: str device: str | list[str] | None = None + context: str | None = None class DeviceInstructionResponse(BECMessage): diff --git a/bec_lib/bec_lib/scan_history.py b/bec_lib/bec_lib/scan_history.py index 61fba2848..bbf4e6113 100644 --- a/bec_lib/bec_lib/scan_history.py +++ b/bec_lib/bec_lib/scan_history.py @@ -6,14 +6,16 @@ import os import threading +import traceback from typing import TYPE_CHECKING +from bec_lib import messages +from bec_lib.bec_errors import BECError from bec_lib.callback_handler import EventType from bec_lib.endpoints import MessageEndpoints from bec_lib.scan_data_container import ScanDataContainer if TYPE_CHECKING: # pragma: no cover - from bec_lib import messages from bec_lib.client import BECClient @@ -140,7 +142,27 @@ def __len__(self) -> int: def __getitem__(self, index: int | slice) -> ScanDataContainer | list[ScanDataContainer]: with self._scan_data_lock: if isinstance(index, int): - target_id = self._scan_ids[index] + try: + target_id = self._scan_ids[index] + except IndexError: + if len(self._scan_ids) == 0: + compact_msg = ( + f"ScanHistory is empty. This may be due to no scans being " + f"run yet or the current user {os.getlogin()} not having access to the data files." + ) + else: + compact_msg = ( + f"Index {index} out of range for ScanHistory of length {len(self)}" + ) + error_info = messages.ErrorInfo( + error_message=traceback.format_exc(), + compact_error_message=compact_msg, + exception_type="IndexError", + context="ScanHistory", + device=None, + ) + raise BECError(compact_msg, error_info) + return self.get_by_scan_id(target_id) if isinstance(index, slice): return [self.get_by_scan_id(scan_id) for scan_id in self._scan_ids[index]] diff --git a/bec_lib/bec_lib/utils/error_pretty_print.py b/bec_lib/bec_lib/utils/error_pretty_print.py new file mode 100644 index 000000000..a927fe86d --- /dev/null +++ b/bec_lib/bec_lib/utils/error_pretty_print.py @@ -0,0 +1,151 @@ +""" +Shared Rich pretty-print helpers for BEC errors and alarms. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +if TYPE_CHECKING: # pragma: no cover + from bec_lib import messages + from bec_lib.alarm_handler import Alarms + + +def _lazy_import_rich(): + global Console, Group, Panel, Syntax, Text + from rich.console import Console, Group + from rich.panel import Panel + from rich.syntax import Syntax + from rich.text import Text + + +class ErrorInfoPrettyPrinter: + """Reusable Rich formatter for an ``ErrorInfo`` payload.""" + + def __init__(self, error_info: messages.ErrorInfo): + self.error_info = error_info + + def pretty_print_title(self): + _lazy_import_rich() + + title = Text() + title.append(f"{self.error_info.exception_type}", style="bold") + if self.error_info.context: + title.append(f" | {self.error_info.context}", style="bold") + if self.error_info.device: + title.append(f" | Device {self.error_info.device}", style="bold") + title.append("\n") + return title + + def pretty_print_renderables(self, msg: str) -> list: + _lazy_import_rich() + + renderables = [] + if "Traceback (most recent call last):" in msg: + renderables.append(Syntax(msg.strip(), "python", word_wrap=True)) + else: + renderables.append(Text(msg.strip())) + + if self.error_info.device: + renderables.append( + Text( + f"\n\nThe error is likely unrelated to BEC. Please check the device '{self.error_info.device}'.", + style="bold", + ) + ) + return renderables + + def details_header(self): + _lazy_import_rich() + + header = Text() + header.append("Error Occurred\n", style="bold red") + header.append(f"Type: {self.error_info.exception_type}\n", style="bold") + if self.error_info.context: + header.append(f"Context: {self.error_info.context}\n", style="bold") + if self.error_info.device: + header.append(f"Device: {self.error_info.device}\n", style="bold") + return header + + def details_title(self) -> str: + return "Error Info" + + def summary_text_style(self) -> str | None: + return None + + def pretty_print(self) -> None: + """ + Use Rich to pretty print the compact error message when available. + """ + _lazy_import_rich() + + console = Console() + msg = self.error_info.compact_error_message or self.error_info.error_message + body = Group(*self.pretty_print_renderables(msg)) + console.print(Panel(body, title=self.pretty_print_title(), border_style="red", expand=True)) + + def print_details(self) -> None: + """ + Use Rich to pretty print the full error details, including the full error message. + """ + _lazy_import_rich() + + console = Console() + console.print( + Panel( + self.details_header(), title=self.details_title(), border_style="red", expand=False + ) + ) + + if self.error_info.compact_error_message: + console.print( + Panel( + Text(self.error_info.compact_error_message, style=self.summary_text_style()), + title="Summary", + border_style="yellow", + expand=False, + ) + ) + + tb_str = self.error_info.error_message + if tb_str: + try: + console.print(tb_str) + except Exception: + console.print(Panel(tb_str, title="Message", border_style="cyan")) + + +class AlarmPrettyPrinter(ErrorInfoPrettyPrinter): + """Alarm-specific formatter that adds severity-focused headers.""" + + def __init__(self, error_info: messages.ErrorInfo, severity: Alarms): + super().__init__(error_info) + self.severity = severity + + def pretty_print_title(self): + _lazy_import_rich() + + title = Text() + title.append(f"{self.error_info.exception_type} | ", style="bold") + title.append(f"Severity {self.severity.name}", style="bold yellow") + if self.error_info.device: + title.append(f" | Device {self.error_info.device}\n", style="bold") + title.append("\n") + return title + + def details_header(self): + _lazy_import_rich() + + header = Text() + header.append("Alarm Raised\n", style="bold red") + header.append(f"Severity: {self.severity.name}\n", style="bold") + header.append(f"Type: {self.error_info.exception_type}\n", style="bold") + if self.error_info.device: + header.append(f"Device: {self.error_info.device}\n", style="bold") + return header + + def details_title(self) -> str: + return "Alarm Info" + + def summary_text_style(self) -> str | None: + return "yellow" diff --git a/bec_lib/tests/test_error_pretty_print.py b/bec_lib/tests/test_error_pretty_print.py new file mode 100644 index 000000000..2a8e81d84 --- /dev/null +++ b/bec_lib/tests/test_error_pretty_print.py @@ -0,0 +1,28 @@ +from bec_lib import messages +from bec_lib.utils.error_pretty_print import ErrorInfoPrettyPrinter + + +def test_error_info_pretty_printer_outputs_summary_and_details(capsys): + error_info = messages.ErrorInfo( + error_message='Traceback (most recent call last):\n File "", line 1, in \nException: boom', + compact_error_message="Short summary", + exception_type="TestError", + device="samx", + context="during scan", + ) + printer = ErrorInfoPrettyPrinter(error_info) + + printer.pretty_print() + printer.print_details() + + captured = capsys.readouterr() + + assert "TestError" in captured.out + assert "during scan" in captured.out + assert "Device samx" in captured.out + assert "Short summary" in captured.out + assert "Error Occurred" in captured.out + assert "Type: TestError" in captured.out + assert "Context: during scan" in captured.out + assert "Device: samx" in captured.out + assert "Traceback (most recent call last):" in captured.out