feat: add console redis log and error log

bec_ipython_client/bec_ipython_client/main.py

@@ -5,6 +5,7 @@
 import functools
 import os
 import sys
+import traceback
 from typing import Iterable, Literal, Tuple
 
 import IPython
@@ -147,7 +148,7 @@ def _configure_ipython(self):
         self._ip.prompts = BECClientPrompt(ip=self._ip, client=self._client, username="unknown")
         self._refresh_ipython_username()
         self._load_magics()
-        self._ip.events.register("post_run_cell", log_console)
+        self._ip.events.register("pre_run_cell", log_console)
         self._ip.set_custom_exc((Exception,), self._create_exception_handler())
         # represent objects using __str__, if overwritten, otherwise use __repr__
         self._ip.display_formatter.formatters["text/plain"].for_type(
@@ -252,25 +253,31 @@ def _ip_exception_handler(
 ):
     if issubclass(etype, AlarmBase):
         parent._alarm_history.append((etype, evalue, tb, tb_offset))
+        log_console_error(etype, evalue, tb)
         print("\x1b[31m BEC alarm:\x1b[0m")
         evalue.pretty_print()
         print("For more details, use 'bec.show_last_alarm()'")
         return
     if issubclass(etype, ValidationError):
+        log_console_error(etype, evalue, tb)
         pretty_print_pydantic_validation_error(evalue)
         return
     if issubclass(etype, (ScanInterruption, DeviceConfigError)):
+        log_console_error(etype, evalue, tb, f"{evalue.__class__.__name__}: {evalue}")
         print(f"\x1b[31m {evalue.__class__.__name__}:\x1b[0m {evalue}")
         return
     if issubclass(etype, redis.exceptions.NoPermissionError):
         # pylint: disable=protected-access
         msg = f"The current user ({bec._client.username}) does not have the required permissions.\n {evalue}"
+        log_console_error(etype, evalue, tb, f"Unauthorized: {msg}")
         logger.info(f"Unauthorized: {msg}")
         print(f"\x1b[31m Unauthorized:\x1b[0m {msg}")
         return
     if issubclass(etype, ExceptionWithErrorInfo):
+        log_console_error(etype, evalue, tb)
         evalue.pretty_print()
         return
+    log_console_error(etype, evalue, tb)
     self.showtraceback((etype, evalue, tb), tb_offset=None)  # standard IPython's printout
 
 
@@ -322,7 +329,17 @@ def username(self, value):
 
 def log_console(execution_info):
     """log the console input"""
-    logger.log("CONSOLE_LOG", f"{execution_info.info.raw_cell}")
+    logger.log("CONSOLE_LOG", f"{execution_info.raw_cell}")
+
+
+def log_console_error(etype, evalue, tb=None, message: str | None = None):
+    """Log console errors to the shared log stream."""
+    if message is None:
+        if tb is not None:
+            message = "".join(traceback.format_exception(etype, evalue, tb)).rstrip()
+        else:
+            message = f"{etype.__name__}: {evalue}"
+    logger.log("CONSOLE_LOG_ERROR", message)
 
 
 # pylint: disable=wrong-import-position

bec_ipython_client/tests/client_tests/test_bec_client.py

@@ -4,10 +4,12 @@
 
 import IPython
 import pytest
+import redis.exceptions
 
 from bec_ipython_client import BECIPythonClient, main
 from bec_lib import messages
 from bec_lib.alarm_handler import AlarmBase, AlarmHandler, Alarms
+from bec_lib.bec_errors import DeviceConfigError
 from bec_lib.redis_connector import RedisConnector
 from bec_lib.service_config import ServiceConfig
 
@@ -238,3 +240,57 @@ def test_bec_ipython_client_show_last_no_alarm(ipython_client, capsys):
     client.show_last_alarm()
     captured = capsys.readouterr()
     assert "No alarm has been raised in this session." in captured.out
+
+
+def test_ipython_exception_handler_logs_console_error_for_device_config_error():
+    shell = mock.MagicMock()
+    with mock.patch.object(main.logger, "log") as mock_log:
+        main._ip_exception_handler(
+            shell,
+            DeviceConfigError,
+            DeviceConfigError("bad config"),
+            None,
+            parent=mock.MagicMock(),
+        )
+
+    mock_log.assert_called_once_with("CONSOLE_LOG_ERROR", "DeviceConfigError: bad config")
+    shell.showtraceback.assert_not_called()
+
+
+def test_ipython_exception_handler_logs_console_error_for_unhandled_exception():
+    shell = mock.MagicMock()
+    try:
+        raise RuntimeError("boom")
+    except RuntimeError as exc:
+        with mock.patch.object(main.logger, "log") as mock_log:
+            main._ip_exception_handler(shell, RuntimeError, exc, exc.__traceback__, parent=mock.MagicMock())
+
+    mock_log.assert_called_once()
+    assert mock_log.call_args.args[0] == "CONSOLE_LOG_ERROR"
+    assert "RuntimeError: boom" in mock_log.call_args.args[1]
+    shell.showtraceback.assert_called_once()
+
+
+def test_ipython_exception_handler_logs_console_error_for_permission_error():
+    shell = mock.MagicMock()
+    parent = mock.MagicMock()
+    with (
+        mock.patch.object(main.logger, "log") as mock_log,
+        mock.patch.object(main.logger, "info") as mock_info,
+        mock.patch.object(
+            main, "bec", mock.MagicMock(_client=mock.MagicMock(username="alice")), create=True
+        ),
+    ):
+        main._ip_exception_handler(
+            shell,
+            redis.exceptions.NoPermissionError,
+            redis.exceptions.NoPermissionError("denied"),
+            None,
+            parent=parent,
+        )
+
+    mock_log.assert_called_once()
+    assert mock_log.call_args.args[0] == "CONSOLE_LOG_ERROR"
+    assert "Unauthorized:" in mock_log.call_args.args[1]
+    mock_info.assert_called_once()
+    shell.showtraceback.assert_not_called()

bec_lib/bec_lib/logger.py

@@ -42,6 +42,7 @@ class LogLevel(int, enum.Enum):
     ERROR = 40
     CRITICAL = 50
     CONSOLE_LOG = 21
+    CONSOLE_LOG_ERROR = 22
 
 
 class BECLogger:
@@ -201,30 +202,6 @@ def _update_base_path(self, service_config: dict | None = None):
         self._base_path = self.writer_mixin.directory
         self.writer_mixin.create_directory(self._base_path)
 
-    def _logger_callback(self, msg):
-        if not self._configured:
-            return
-        if self.connector is None:
-            return
-        msg = json.loads(msg)
-        msg["service_name"] = self.service_name
-        try:
-            self.connector.xadd(
-                topic=MessageEndpoints.log(),
-                msg_dict={
-                    "data": bec_lib.messages.LogMessage(
-                        log_type=msg["record"]["level"]["name"].lower(), log_msg=msg
-                    )
-                },
-                max_size=10000,
-            )
-        except Exception:
-            # connector disconnected?
-            # just ignore the error here...
-            # Exception is not explicitly specified,
-            # because it depends on the connector
-            pass
-
     def get_format(self, level: LogLevel = None, is_stderr=False, is_container=False) -> str:
         """
         Get the format for a specific log level.
@@ -298,23 +275,38 @@ def filter(self, is_console: bool = False):
 
         Args:
             is_console (bool, optional): Whether the log is for the console. Defaults to False.
-
         Returns:
             function: Filter function.
         """
 
         def _filter(record):
-            if record["name"] in self._disabled_modules:
+            if self._is_disabled_record(record):
                 return False
-            for module in self._disabled_modules:
-                if record["name"].startswith(module):
-                    return False
-            if not is_console and record["level"].no == LogLevel.CONSOLE_LOG:
+            if not is_console and self._is_console_level(record["level"].no):
                 return False
             return True
 
         return _filter
 
+    def _is_disabled_record(self, record) -> bool:
+        if record["name"] in self._disabled_modules:
+            return True
+        return any(record["name"].startswith(module) for module in self._disabled_modules)
+
+    def _is_console_level(self, level_no: int) -> bool:
+        return level_no in (LogLevel.CONSOLE_LOG, LogLevel.CONSOLE_LOG_ERROR)
+
+    def filter_console_redis_log(self, record) -> bool:
+        """
+        Filter function for console redis log messages, which are used to send log messages redis.
+
+        Args:
+            record (dict): Log record.
+        Returns:
+            bool: True if the log message should be sent to the console via redis, False otherwise
+        """
+        return not self._is_disabled_record(record) and self._is_console_level(record["level"].no)
+
     def add_sys_stderr(self, level: LogLevel):
         """
         Add a sink to stderr.
@@ -355,6 +347,7 @@ def add_console_log(self):
         """
         try:
             self.logger.level("CONSOLE_LOG", no=21, color="<yellow>", icon="📣")
+            self.logger.level("CONSOLE_LOG_ERROR", no=22, color="<red>", icon="📣")
         except (TypeError, ValueError):
             # level with same severity already exists: already configured
             pass
@@ -379,6 +372,7 @@ def add_console_log(self):
             opener=self._file_opener,
         )
         self._console_log = True
+        self.add_console_redis_log()
 
     def add_redis_log(self, level: LogLevel):
         """
@@ -388,13 +382,58 @@ def add_redis_log(self, level: LogLevel):
             level (LogLevel): Log level.
         """
         self.logger.add(
-            self._logger_callback,
+            self._publish_log_message,
             serialize=True,
             level=level,
             format=self.formatting(),
             filter=self.filter(),
         )
 
+    def add_console_redis_log(self):
+        """
+        Add a sink to the console redis log.
+        It deviates from the regular redis log in that it only includes messages with
+        level CONSOLE_LOG and CONSOLE_LOG_ERROR.
+        """
+        self.logger.add(
+            self._console_redis_logger_callback,
+            serialize=True,
+            level=LogLevel.CONSOLE_LOG,
+            format=self.formatting(is_stderr=True),
+            filter=self.filter_console_redis_log,
+        )
+
+    def _console_redis_logger_callback(self, msg):
+        if not self._configured or self.connector is None:
+            return
+        self._publish_log_message(msg, service_name=f"{self.service_name}_CONSOLE")
+
+    def _decode_log_payload(self, msg: str | dict) -> dict:
+        return json.loads(msg) if isinstance(msg, str) else dict(msg)
+
+    def _publish_log_message(self, msg: str | dict, service_name: str | None = None) -> bool:
+        if not self._configured or self.connector is None:
+            return False
+        payload = self._decode_log_payload(msg)
+        payload["service_name"] = self.service_name if service_name is None else service_name
+        try:
+            self.connector.xadd(
+                topic=MessageEndpoints.log(),
+                msg_dict={
+                    "data": bec_lib.messages.LogMessage(
+                        log_type=payload["record"]["level"]["name"].lower(), log_msg=payload
+                    )
+                },
+                max_size=10000,
+            )
+            return True
+        except Exception:
+            # connector disconnected?
+            # just ignore the error here...
+            # Exception is not explicitly specified,
+            # because it depends on the connector
+            return False
+
     @property
     def disabled_modules(self) -> set[str]:
         """

bec_lib/bec_lib/messages.py

@@ -1008,15 +1008,23 @@ class LogMessage(BECMessage):
     """Log message
 
     Args:
-        log_type (Literal["trace", "debug", "info", "success", "warning", "error", "critical", "console_log"]) : Log type.
+        log_type (Literal["trace", "debug", "info", "success", "warning", "error", "critical", "console_log", "console_log_error"]) : Log type.
         log_msg (dict or str): Log message.
         metadata (dict, optional): Additional metadata.
 
     """
 
     msg_type: ClassVar[str] = "log_message"
     log_type: Literal[
-        "trace", "debug", "info", "success", "warning", "error", "critical", "console_log"
+        "trace",
+        "debug",
+        "info",
+        "success",
+        "warning",
+        "error",
+        "critical",
+        "console_log",
+        "console_log_error",
     ]
     log_msg: dict | str
 

bec_lib/tests/test_bec_logger.py

@@ -1,3 +1,4 @@
+import json
 import os
 from pathlib import Path
 from unittest import mock
@@ -98,3 +99,33 @@ def test_set_log_level(logger, log_level, sink, expected_level):
     logger.set_log_level(log_level, sink)
     for key, value in expected_level.items():
         assert getattr(logger, key) == value
+
+
+def test_console_redis_callback_publishes_to_log_endpoint_with_console_service_name(logger):
+    logger._configured = True
+    logger.service_name = "test"
+    logger.connector = mock.MagicMock(spec=RedisConnector)
+
+    logger._console_redis_logger_callback(
+        json.dumps({"record": {"level": {"name": "CONSOLE_LOG"}}, "text": "hello"})
+    )
+
+    logger.connector.xadd.assert_called_once()
+    kwargs = logger.connector.xadd.call_args.kwargs
+    assert kwargs["topic"] == logger.connector.xadd.call_args.kwargs["topic"]
+    assert kwargs["topic"].endpoint == "info/log"
+    assert kwargs["msg_dict"]["data"].log_type == "console_log"
+    assert kwargs["msg_dict"]["data"].log_msg["service_name"] == "test_CONSOLE"
+
+
+def test_console_redis_callback_ignores_publish_failures(logger):
+    logger._configured = True
+    logger.service_name = "test"
+    logger.connector = mock.MagicMock(spec=RedisConnector)
+    logger.connector.xadd.side_effect = RuntimeError("redis unavailable")
+
+    logger._console_redis_logger_callback(
+        json.dumps({"record": {"level": {"name": "CONSOLE_LOG_ERROR"}}, "text": "oops"})
+    )
+
+    logger.connector.xadd.assert_called_once()