Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 8 additions & 34 deletions bec_ipython_client/bec_ipython_client/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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")
Expand Down
41 changes: 5 additions & 36 deletions bec_lib/bec_lib/alarm_handler.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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:
Expand All @@ -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):
Expand Down
49 changes: 22 additions & 27 deletions bec_lib/bec_lib/bec_errors.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"""

Expand Down Expand Up @@ -41,46 +58,24 @@ 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):
"""Scan input validation error"""

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:
Expand Down
1 change: 1 addition & 0 deletions bec_lib/bec_lib/messages.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down
26 changes: 24 additions & 2 deletions bec_lib/bec_lib/scan_history.py
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand Down Expand Up @@ -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]]
Expand Down
Loading
Loading