feat(widget): add LogPanel widget

bec_widgets/cli/client.py

@@ -31,6 +31,7 @@ class Widgets(str, enum.Enum):
     DeviceComboBox = "DeviceComboBox"
     DeviceLineEdit = "DeviceLineEdit"
     LMFitDialog = "LMFitDialog"
+    LogPanel = "LogPanel"
     Minesweeper = "Minesweeper"
     PositionIndicator = "PositionIndicator"
     PositionerBox = "PositionerBox"
@@ -3183,6 +3184,26 @@ class LMFitDialog(RPCBase):
         """
 
 
+class LogPanel(RPCBase):
+    @rpc_call
+    def set_plain_text(self, text: str) -> None:
+        """
+        Set the plain text of the widget.
+
+        Args:
+            text (str): The text to set.
+        """
+
+    @rpc_call
+    def set_html_text(self, text: str) -> None:
+        """
+        Set the HTML text of the widget.
+
+        Args:
+            text (str): The text to set.
+        """
+
+
 class Minesweeper(RPCBase): ...
 
 

bec_widgets/widgets/__init__.py

@@ -1 +0,0 @@
-

bec_widgets/widgets/containers/dock/dock_area.py

@@ -30,6 +30,7 @@ from bec_widgets.widgets.plots.waveform.waveform_widget import BECWaveformWidget
 from bec_widgets.widgets.progress.ring_progress_bar.ring_progress_bar import RingProgressBar
 from bec_widgets.widgets.services.bec_queue.bec_queue import BECQueue
 from bec_widgets.widgets.services.bec_status_box.bec_status_box import BECStatusBox
+from bec_widgets.widgets.utility.logpanel.logpanel import LogPanel
 from bec_widgets.widgets.utility.visual.dark_mode_button.dark_mode_button import DarkModeButton
 
 
@@ -139,6 +140,9 @@ class BECDockArea(BECWidget, QWidget):
                             tooltip="Add Circular ProgressBar",
                             filled=True,
                         ),
+                        "log_panel": MaterialIconAction(
+                            icon_name=LogPanel.ICON_NAME, tooltip="Add LogPanel", filled=True
+                        ),
                     },
                 ),
                 "separator_2": SeparatorAction(),
@@ -200,6 +204,9 @@ class BECDockArea(BECWidget, QWidget):
         self.toolbar.widgets["menu_utils"].widgets["progress_bar"].triggered.connect(
             lambda: self.add_dock(widget="RingProgressBar", prefix="progress_bar")
         )
+        self.toolbar.widgets["menu_utils"].widgets["log_panel"].triggered.connect(
+            lambda: self.add_dock(widget="LogPanel", prefix="log_panel")
+        )
 
         # Icons
         self.toolbar.widgets["attach_all"].action.triggered.connect(self.attach_all)

bec_widgets/widgets/editors/text_box/text_box.py

@@ -5,9 +5,9 @@ from html.parser import HTMLParser
 
 from bec_lib.logger import bec_logger
 from pydantic import Field
-from qtpy.QtCore import Property, Slot
 from qtpy.QtWidgets import QTextEdit, QVBoxLayout, QWidget
 
+from bec_widgets.qt_utils.error_popups import SafeProperty, SafeSlot
 from bec_widgets.utils.bec_connector import ConnectionConfig
 from bec_widgets.utils.bec_widget import BECWidget
 
@@ -66,7 +66,7 @@ class TextBox(BECWidget, QWidget):
         else:
             self.set_html_text(DEFAULT_TEXT)
 
-    @Slot(str)
+    @SafeSlot(str)
     def set_plain_text(self, text: str) -> None:
         """Set the plain text of the widget.
 
@@ -77,7 +77,7 @@ class TextBox(BECWidget, QWidget):
         self.config.text = text
         self.config.is_html = False
 
-    @Slot(str)
+    @SafeSlot(str)
     def set_html_text(self, text: str) -> None:
         """Set the HTML text of the widget.
 
@@ -88,7 +88,7 @@ class TextBox(BECWidget, QWidget):
         self.config.text = text
         self.config.is_html = True
 
-    @Property(str)
+    @SafeProperty(str)
     def plain_text(self) -> str:
         """Get the text of the widget.
 
@@ -106,7 +106,7 @@ class TextBox(BECWidget, QWidget):
         """
         self.set_plain_text(text)
 
-    @Property(str)
+    @SafeProperty(str)
     def html_text(self) -> str:
         """Get the HTML text of the widget.
 

bec_widgets/widgets/utility/logpanel/__init__.py

@@ -0,0 +1,3 @@
+from bec_widgets.widgets.utility.logpanel.logpanel import LogPanel
+
+__ALL__ = ["LogPanel"]

bec_widgets/widgets/utility/logpanel/_util.py

@@ -0,0 +1,58 @@
+""" Utilities for filtering and formatting in the LogPanel"""
+
+from __future__ import annotations
+
+import re
+from collections import deque
+from typing import Callable, Iterator
+
+from bec_lib.logger import LogLevel
+from bec_lib.messages import LogMessage
+from qtpy.QtCore import QDateTime
+
+LinesHtmlFormatter = Callable[[deque[LogMessage]], Iterator[str]]
+LineFormatter = Callable[[LogMessage], str]
+LineFilter = Callable[[LogMessage], bool] | None
+
+ANSI_ESCAPE_REGEX = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])")
+
+
+def replace_escapes(s: str):
+    s = ANSI_ESCAPE_REGEX.sub("", s)
+    return s.replace(" ", "&nbsp;").replace("\n", "<br />").replace("\t", "    ")
+
+
+def level_filter(msg: LogMessage, thresh: int):
+    return LogLevel[msg.content["log_type"].upper()].value >= thresh
+
+
+def noop_format(line: LogMessage):
+    _textline = line.log_msg if isinstance(line.log_msg, str) else line.log_msg["text"]
+    return replace_escapes(_textline.strip()) + "<br />"
+
+
+def simple_color_format(line: LogMessage, colors: dict[LogLevel, str]):
+    color = colors.get(LogLevel[line.content["log_type"].upper()]) or colors[LogLevel.INFO]
+    return f'<font color="{color}">{noop_format(line)}</font>'
+
+
+def create_formatter(line_format: LineFormatter, line_filter: LineFilter) -> LinesHtmlFormatter:
+    def _formatter(data: deque[LogMessage]):
+        if line_filter is not None:
+            return (line_format(line) for line in data if line_filter(line))
+        else:
+            return (line_format(line) for line in data)
+
+    return _formatter
+
+
+def log_txt(line):
+    return line.log_msg if isinstance(line.log_msg, str) else line.log_msg["text"]
+
+
+def log_time(line):
+    return QDateTime.fromMSecsSinceEpoch(int(line.log_msg["record"]["time"]["timestamp"] * 1000))
+
+
+def log_svc(line):
+    return line.log_msg["service_name"]

bec_widgets/widgets/utility/logpanel/log_panel.pyproject

@@ -0,0 +1 @@
+{'files': ['logpanel.py']}

bec_widgets/widgets/utility/logpanel/log_panel_plugin.py

@@ -0,0 +1,54 @@
+# Copyright (C) 2022 The Qt Company Ltd.
+# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause
+
+from qtpy.QtDesigner import QDesignerCustomWidgetInterface
+
+from bec_widgets.utils.bec_designer import designer_material_icon
+from bec_widgets.widgets.utility.logpanel.logpanel import LogPanel
+
+DOM_XML = """
+<ui language='c++'>
+    <widget class='LogPanel' name='log_panel'>
+    </widget>
+</ui>
+"""
+
+
+class LogPanelPlugin(QDesignerCustomWidgetInterface):  # pragma: no cover
+    def __init__(self):
+        super().__init__()
+        self._form_editor = None
+
+    def createWidget(self, parent):
+        t = LogPanel(parent)
+        return t
+
+    def domXml(self):
+        return DOM_XML
+
+    def group(self):
+        return "BEC Utils"
+
+    def icon(self):
+        return designer_material_icon(LogPanel.ICON_NAME)
+
+    def includeFile(self):
+        return "log_panel"
+
+    def initialize(self, form_editor):
+        self._form_editor = form_editor
+
+    def isContainer(self):
+        return False
+
+    def isInitialized(self):
+        return self._form_editor is not None
+
+    def name(self):
+        return "LogPanel"
+
+    def toolTip(self):
+        return "Displays a log panel"
+
+    def whatsThis(self):
+        return self.toolTip()

bec_widgets/widgets/utility/logpanel/logpanel.py

@@ -0,0 +1,516 @@
+""" Module for a LogPanel widget to display BEC log messages """
+
+from __future__ import annotations
+
+import operator
+import os
+import re
+from collections import deque
+from functools import partial, reduce
+from re import Pattern
+from typing import TYPE_CHECKING, Literal
+
+from bec_lib.client import BECClient
+from bec_lib.connector import ConnectorBase
+from bec_lib.endpoints import MessageEndpoints
+from bec_lib.logger import LogLevel, bec_logger
+from bec_lib.messages import LogMessage, StatusMessage
+from qtpy.QtCore import QDateTime, Qt, Signal  # type: ignore
+from qtpy.QtGui import QFont
+from qtpy.QtWidgets import (
+    QApplication,
+    QCheckBox,
+    QComboBox,
+    QDateTimeEdit,
+    QDialog,
+    QGridLayout,
+    QHBoxLayout,
+    QLabel,
+    QLineEdit,
+    QPushButton,
+    QScrollArea,
+    QTextEdit,
+    QVBoxLayout,
+    QWidget,
+)
+
+from bec_widgets.qt_utils.error_popups import SafeSlot
+from bec_widgets.utils.colors import get_theme_palette, set_theme
+from bec_widgets.widgets.editors.text_box.text_box import TextBox
+from bec_widgets.widgets.services.bec_status_box.bec_status_box import BECServiceStatusMixin
+from bec_widgets.widgets.utility.logpanel._util import (
+    LineFilter,
+    LineFormatter,
+    LinesHtmlFormatter,
+    create_formatter,
+    level_filter,
+    log_svc,
+    log_time,
+    log_txt,
+    noop_format,
+    simple_color_format,
+)
+
+if TYPE_CHECKING:
+    from qtpy.QtCore import pyqtBoundSignal  # type: ignore
+
+logger = bec_logger.logger
+
+MODULE_PATH = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
+
+# TODO: improve log color handling
+DEFAULT_LOG_COLORS = {
+    LogLevel.INFO: "#FFFFFF",
+    LogLevel.SUCCESS: "#00FF00",
+    LogLevel.WARNING: "#FFCC00",
+    LogLevel.ERROR: "#FF0000",
+    LogLevel.DEBUG: "#0000CC",
+}
+
+
+class BecLogsQueue:
+    """Manages getting logs from BEC Redis and formatting them for display"""
+
+    def __init__(
+        self,
+        conn: ConnectorBase,
+        new_message_signal: pyqtBoundSignal,
+        maxlen: int = 1000,
+        line_formatter: LineFormatter = noop_format,
+    ) -> None:
+        self._timestamp_start: QDateTime | None = None
+        self._timestamp_end: QDateTime | None = None
+        self._conn = conn
+        self._new_message_signal: pyqtBoundSignal | None = new_message_signal
+        self._max_length = maxlen
+        self._data: deque[LogMessage] = deque([], self._max_length)
+        self._display_queue: deque[str] = deque([], self._max_length)
+        self._log_level: str | None = None
+        self._search_query: Pattern | str | None = None
+        self._selected_services: set[str] | None = None
+        self._set_formatter_and_update_filter(line_formatter)
+        self._conn.register([MessageEndpoints.log()], None, self._process_incoming_log_msg)
+
+    def disconnect(self):
+        self._conn.unregister([MessageEndpoints.log()], None, self._process_incoming_log_msg)
+        self._new_message_signal = None
+
+    def _process_incoming_log_msg(self, msg: dict):
+        _msg: LogMessage = msg["data"]
+        self._data.append(_msg)
+        if self.filter is None or self.filter(_msg):
+            self._display_queue.append(self._line_formatter(_msg))
+            if self._new_message_signal:
+                self._new_message_signal.emit()
+
+    def _set_formatter_and_update_filter(self, line_formatter: LineFormatter = noop_format):
+        self._line_formatter: LineFormatter = line_formatter
+        self._queue_formatter: LinesHtmlFormatter = create_formatter(
+            self._line_formatter, self.filter
+        )
+
+    def _combine_filters(self, *args: LineFilter):
+        return lambda msg: reduce(operator.and_, [filt(msg) for filt in args if filt is not None])
+
+    def _create_re_filter(self) -> LineFilter:
+        if self._search_query is None:
+            return None
+        elif isinstance(self._search_query, str):
+            return lambda line: self._search_query in log_txt(line)
+        return lambda line: self._search_query.match(log_txt(line)) is not None
+
+    def _create_service_filter(self):
+        return (
+            lambda line: self._selected_services is None or log_svc(line) in self._selected_services
+        )
+
+    def _create_timestamp_filter(self) -> LineFilter:
+        s, e = self._timestamp_start, self._timestamp_end
+        if s is e is None:
+            return lambda msg: True
+
+        def _time_filter(msg):
+            msg_time = log_time(msg)
+            if s is None:
+                return msg_time <= e
+            if e is None:
+                return s <= msg_time
+            return s <= msg_time <= e
+
+        return _time_filter
+
+    @property
+    def filter(self) -> LineFilter:
+        thresh = LogLevel[self._log_level].value if self._log_level is not None else 0
+        return self._combine_filters(
+            partial(level_filter, thresh=thresh),
+            self._create_re_filter(),
+            self._create_timestamp_filter(),
+            self._create_service_filter(),
+        )
+
+    def update_level_filter(self, level: str):
+        if level not in [l.name for l in LogLevel]:
+            logger.error(f"Logging level {level} unrecognized for filter!")
+            return
+        self._log_level = level
+        self._set_formatter_and_update_filter(self._line_formatter)
+
+    def update_search_filter(self, search_query: Pattern | str | None = None):
+        self._search_query = search_query
+        self._set_formatter_and_update_filter(self._line_formatter)
+
+    def update_time_filter(self, start: QDateTime | None, end: QDateTime | None):
+        self._timestamp_start = start
+        self._timestamp_end = end
+        self._set_formatter_and_update_filter(self._line_formatter)
+
+    def update_service_filter(self, services: set[str]):
+        self._selected_services = services
+        self._set_formatter_and_update_filter(self._line_formatter)
+
+    def update_line_formatter(self, line_formatter: LineFormatter):
+        self._set_formatter_and_update_filter(line_formatter)
+
+    def display_all(self) -> str:
+        return "\n".join(self._queue_formatter(self._data.copy()))
+
+    def format_new(self):
+        res = "\n".join(self._display_queue)
+        self._display_queue = deque([], self._max_length)
+        return res
+
+    def clear_logs(self):
+        self._data = deque([])
+        self._display_queue = deque([])
+
+    def fetch_history(self):
+        self._data = deque(
+            item["data"]
+            for item in self._conn.xread(
+                MessageEndpoints.log().endpoint, from_start=True, count=self._max_length
+            )
+        )
+
+    def unique_service_names_from_history(self) -> set[str]:
+        return set(msg.log_msg["service_name"] for msg in self._data)
+
+
+class LogPanelToolbar(QWidget):
+
+    services_selected: pyqtBoundSignal = Signal(set)
+
+    def __init__(self, parent: QWidget | None = None) -> None:
+        super().__init__(parent)
+
+        # in unix time
+        self._timestamp_start: QDateTime | None = None
+        self._timestamp_end: QDateTime | None = None
+
+        self._unique_service_names: set[str] = set()
+        self._services_selected: set[str] | None = None
+
+        self.layout = QHBoxLayout(self)  # type: ignore
+
+        self.service_choice_button = QPushButton("Select services", self)
+        self.layout.addWidget(self.service_choice_button)
+        self.service_choice_button.clicked.connect(self._open_service_filter_dialog)
+
+        self.filter_level_dropdown = self._log_level_box()
+        self.layout.addWidget(self.filter_level_dropdown)
+
+        self.clear_button = QPushButton("Clear all", self)
+        self.layout.addWidget(self.clear_button)
+        self.fetch_button = QPushButton("Fetch history", self)
+        self.layout.addWidget(self.fetch_button)
+
+        self._string_search_box()
+
+        self.timerange_button = QPushButton("Set time range", self)
+        self.layout.addWidget(self.timerange_button)
+
+    @property
+    def time_start(self):
+        return self._timestamp_start
+
+    @property
+    def time_end(self):
+        return self._timestamp_end
+
+    def _string_search_box(self):
+        self.layout.addWidget(QLabel("Search: "))
+        self.search_textbox = QLineEdit()
+        self.layout.addWidget(self.search_textbox)
+        self.layout.addWidget(QLabel("Use regex: "))
+        self.regex_enabled = QCheckBox()
+        self.layout.addWidget(self.regex_enabled)
+        self.update_re_button = QPushButton("Update search", self)
+        self.layout.addWidget(self.update_re_button)
+
+    def _log_level_box(self):
+        box = QComboBox()
+        box.setToolTip("Display logs with equal or greater significance to the selected level.")
+        [box.addItem(l.name) for l in LogLevel]
+        return box
+
+    def _current_ts(self, selection_type: Literal["start", "end"]):
+        if selection_type == "start":
+            return self._timestamp_start
+        elif selection_type == "end":
+            return self._timestamp_end
+        else:
+            raise ValueError(f"timestamps can only be for the start or end, not {selection_type}")
+
+    def _open_datetime_dialog(self):
+        """Open dialog window for timestamp filter selection"""
+        self._dt_dialog = QDialog(self)
+        self._dt_dialog.setWindowTitle("Time range selection")
+        layout = QVBoxLayout()
+
+        label_start = QLabel()
+        label_end = QLabel()
+
+        def date_button_set(selection_type: Literal["start", "end"], label: QLabel):
+            dt = self._current_ts(selection_type)
+            _layout = QHBoxLayout()
+            layout.addLayout(_layout)
+            date_button = QPushButton(f"Time {selection_type}")
+            _layout.addWidget(date_button)
+            label.setText(dt.toString() if dt else "not selected")
+            _layout.addWidget(label)
+            date_button.clicked.connect(partial(self._open_cal_dialog, selection_type, label))
+            date_clear_button = QPushButton("clear")
+            date_clear_button.clicked.connect(
+                lambda: (
+                    partial(self._update_time, selection_type)(None),
+                    label.setText("not selected"),
+                )
+            )
+            _layout.addWidget(date_clear_button)
+
+        for v in [("start", label_start), ("end", label_end)]:
+            date_button_set(*v)
+
+        close_button = QPushButton("Close")
+        close_button.clicked.connect(self._dt_dialog.accept)
+        layout.addWidget(close_button)
+        self._dt_dialog.setLayout(layout)
+        self._dt_dialog.exec()
+        self._dt_dialog = None
+
+    def _open_cal_dialog(self, selection_type: Literal["start", "end"], label: QLabel):
+        """Open dialog window for timestamp filter selection"""
+        dt = self._current_ts(selection_type) or QDateTime.currentDateTime()
+        label.setText(dt.toString() if dt else "not selected")
+        if selection_type == "start":
+            self._timestamp_start = dt
+        else:
+            self._timestamp_end = dt
+        self._cal_dialog = QDialog(self)
+        self._cal_dialog.setWindowTitle(f"Select time range {selection_type}")
+        layout = QVBoxLayout()
+        cal = QDateTimeEdit()
+        cal.setCalendarPopup(True)
+        cal.setDateTime(dt)
+        cal.setDisplayFormat("yyyy-MM-dd HH:mm:ss.zzz")
+        cal.dateTimeChanged.connect(partial(self._update_time, selection_type))
+        layout.addWidget(cal)
+        close_button = QPushButton("Close")
+        close_button.clicked.connect(self._cal_dialog.accept)
+        layout.addWidget(close_button)
+        self._cal_dialog.setLayout(layout)
+        self._cal_dialog.exec()
+        self._cal_dialog = None
+
+    def _update_time(self, selection_type: Literal["start", "end"], dt: QDateTime | None):
+        if selection_type == "start":
+            self._timestamp_start = dt
+        else:
+            self._timestamp_end = dt
+
+    @SafeSlot(dict, set)
+    def service_list_update(
+        self, services_info: dict[str, StatusMessage], services_from_history: set[str], *_, **__
+    ):
+        self._unique_service_names = set([s.split("/")[0] for s in services_info.keys()])
+        self._unique_service_names |= services_from_history
+        if self._services_selected is None:
+            self._services_selected = self._unique_service_names
+
+    @SafeSlot()
+    def _open_service_filter_dialog(self):
+        if len(self._unique_service_names) == 0 or self._services_selected is None:
+            return
+        self._svc_dialog = QDialog(self)
+        self._svc_dialog.setWindowTitle(f"Select services to show logs from")
+        layout = QVBoxLayout()
+        service_cb_grid = QGridLayout(parent=self)
+        layout.addLayout(service_cb_grid)
+
+        def check_box(name: str, checked: Qt.CheckState):
+            if checked == Qt.CheckState.Checked:
+                self._services_selected.add(name)
+            else:
+                if name in self._services_selected:
+                    self._services_selected.remove(name)
+            self.services_selected.emit(self._services_selected)
+
+        for i, svc in enumerate(self._unique_service_names):
+            service_cb_grid.addWidget(QLabel(svc), i, 0)
+            cb = QCheckBox()
+            cb.setChecked(svc in self._services_selected)
+            cb.checkStateChanged.connect(partial(check_box, svc))
+            service_cb_grid.addWidget(cb, i, 1)
+
+        close_button = QPushButton("Close")
+        close_button.clicked.connect(self._svc_dialog.accept)
+        layout.addWidget(close_button)
+        self._svc_dialog.setLayout(layout)
+        self._svc_dialog.exec()
+        self._svc_dialog = None
+
+
+class LogPanel(TextBox):
+    """Displays a log panel"""
+
+    ICON_NAME = "terminal"
+    _new_messages = Signal()
+    service_list_update = Signal(dict, set)
+
+    def __init__(self, parent=None, client: BECClient | None = None, **kwargs):
+        """Initialize the LogPanel widget."""
+        super().__init__(parent=parent, client=client, **kwargs)
+        self._update_colors()
+        self._service_status = BECServiceStatusMixin(self, client=self.client)  # type: ignore
+        self._log_manager = BecLogsQueue(
+            self.client.connector,  # type: ignore
+            new_message_signal=self._new_messages,
+            line_formatter=partial(simple_color_format, colors=self._colors),
+        )
+
+        self.toolbar = LogPanelToolbar(parent=parent)
+        self.toolbar_area = QScrollArea()
+        self.toolbar_area.setVerticalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAlwaysOff)
+        self.toolbar_area.setSizeAdjustPolicy(QScrollArea.SizeAdjustPolicy.AdjustToContents)
+        self.toolbar_area.setFixedHeight(int(self.toolbar.clear_button.height() * 2))
+        self.toolbar_area.setWidget(self.toolbar)
+
+        self.layout.addWidget(self.toolbar_area)
+        self.toolbar.clear_button.clicked.connect(self._on_clear)
+        self.toolbar.fetch_button.clicked.connect(self._on_fetch)
+        self.toolbar.update_re_button.clicked.connect(self._on_re_update)
+        self.toolbar.search_textbox.returnPressed.connect(self._on_re_update)
+        self.toolbar.regex_enabled.checkStateChanged.connect(self._on_re_update)
+        self.toolbar.filter_level_dropdown.currentTextChanged.connect(self._set_level_filter)
+        self._new_messages.connect(self._on_append)
+
+        self.toolbar.timerange_button.clicked.connect(self._choose_datetime)
+        self._service_status.services_update.connect(self._update_service_list)
+        self.service_list_update.connect(self.toolbar.service_list_update)
+        self.toolbar.services_selected.connect(self._update_service_filter)
+
+        self.text_box_text_edit.setFont(QFont("monospace", 12))
+        self.text_box_text_edit.setHtml("")
+        self.text_box_text_edit.setLineWrapMode(QTextEdit.LineWrapMode.NoWrap)
+
+        self._connect_to_theme_change()
+
+    @SafeSlot(set)
+    def _update_service_filter(self, services: set[str]):
+        self._log_manager.update_service_filter(services)
+        self._on_redraw()
+
+    @SafeSlot(dict, dict)
+    def _update_service_list(self, services_info: dict[str, StatusMessage], *_, **__):
+        self.service_list_update.emit(
+            services_info, self._log_manager.unique_service_names_from_history()
+        )
+
+    @SafeSlot()
+    def _choose_datetime(self):
+        self.toolbar._open_datetime_dialog()
+        self._set_time_filter()
+
+    def _connect_to_theme_change(self):
+        """Connect to the theme change signal."""
+        qapp = QApplication.instance()
+        if hasattr(qapp, "theme_signal"):
+            qapp.theme_signal.theme_updated.connect(self._on_redraw)  # type: ignore
+
+    def _update_colors(self):
+        self._colors = DEFAULT_LOG_COLORS.copy()
+        self._colors.update({LogLevel.INFO: get_theme_palette().text().color().name()})
+
+    def _cursor_to_end(self):
+        c = self.text_box_text_edit.textCursor()
+        c.movePosition(c.MoveOperation.End)
+        self.text_box_text_edit.setTextCursor(c)
+
+    @SafeSlot()
+    @SafeSlot(str)
+    def _on_redraw(self, *_):
+        self._update_colors()
+        self._log_manager.update_line_formatter(partial(simple_color_format, colors=self._colors))
+        self.set_html_text(self._log_manager.display_all())
+        self._cursor_to_end()
+
+    @SafeSlot()
+    def _on_append(self):
+        self._cursor_to_end()
+        self.text_box_text_edit.insertHtml(self._log_manager.format_new())
+
+    @SafeSlot()
+    def _on_clear(self):
+        self._log_manager.clear_logs()
+        self.set_html_text(self._log_manager.display_all())
+        self._cursor_to_end()
+
+    @SafeSlot()
+    @SafeSlot(Qt.CheckState)
+    def _on_re_update(self, *_):
+        if self.toolbar.regex_enabled.isChecked():
+            try:
+                search_query = re.compile(self.toolbar.search_textbox.text())
+            except Exception as e:
+                logger.warning(f"Failed to compile search regex with error {e}")
+                search_query = None
+            logger.info(f"Setting LogPanel search regex to {search_query}")
+        else:
+            search_query = self.toolbar.search_textbox.text()
+            logger.info(f'Setting LogPanel search string to "{search_query}"')
+        self._log_manager.update_search_filter(search_query)
+        self.set_html_text(self._log_manager.display_all())
+        self._cursor_to_end()
+
+    @SafeSlot()
+    def _on_fetch(self):
+        self._log_manager.fetch_history()
+        self.set_html_text(self._log_manager.display_all())
+        self._cursor_to_end()
+
+    @SafeSlot(str)
+    def _set_level_filter(self, level: str):
+        self._log_manager.update_level_filter(level)
+        self._on_redraw()
+
+    @SafeSlot()
+    def _set_time_filter(self):
+        self._log_manager.update_time_filter(self.toolbar.time_start, self.toolbar.time_end)
+        self._on_redraw()
+
+    def cleanup(self):
+        self._log_manager.disconnect()
+        self._new_messages.disconnect(self._on_append)
+
+
+if __name__ == "__main__":  # pragma: no cover
+    import sys
+
+    from qtpy.QtWidgets import QApplication  # pylint: disable=ungrouped-imports
+
+    app = QApplication(sys.argv)
+    set_theme("dark")
+    widget = LogPanel()
+
+    widget.show()
+    sys.exit(app.exec())

bec_widgets/widgets/utility/logpanel/register_log_panel.py

@@ -0,0 +1,15 @@
+def main():  # pragma: no cover
+    from qtpy import PYSIDE6
+
+    if not PYSIDE6:
+        print("PYSIDE6 is not available in the environment. Cannot patch designer.")
+        return
+    from PySide6.QtDesigner import QPyDesignerCustomWidgetCollection
+
+    from bec_widgets.widgets.utility.logpanel.log_panel_plugin import LogPanelPlugin
+
+    QPyDesignerCustomWidgetCollection.addCustomWidget(LogPanelPlugin())
+
+
+if __name__ == "__main__":  # pragma: no cover
+    main()

tests/unit_tests/test_logpanel.py

@@ -0,0 +1,117 @@
+# pylint: disable=missing-function-docstring, missing-module-docstring, unused-import
+
+from collections import deque
+from unittest.mock import MagicMock
+
+import pytest
+from bec_lib.messages import LogMessage
+
+from bec_widgets.widgets.utility.logpanel._util import replace_escapes, simple_color_format
+from bec_widgets.widgets.utility.logpanel.logpanel import DEFAULT_LOG_COLORS, LogPanel
+
+from .client_mocks import mocked_client
+
+TEST_TABLE_STRING = "2025-01-15 15:57:18 | bec_server.scan_server.scan_queue | [INFO] | \n \x1b[3m              primary queue / ScanQueueStatus.RUNNING              \x1b[0m\n┏━━━━━━━━━━━━━┳━━━━━━━━━┳━━━━━━━━━┳━━━━━━┳━━━━━━━━━━━━┳━━━━━━━━━━━┓\n┃\x1b[1m \x1b[0m\x1b[1m queue_id  \x1b[0m\x1b[1m \x1b[0m┃\x1b[1m \x1b[0m\x1b[1mscan_id\x1b[0m\x1b[1m \x1b[0m┃\x1b[1m \x1b[0m\x1b[1mis_scan\x1b[0m\x1b[1m \x1b[0m┃\x1b[1m \x1b[0m\x1b[1mtype\x1b[0m\x1b[1m \x1b[0m┃\x1b[1m \x1b[0m\x1b[1mscan_numb…\x1b[0m\x1b[1m \x1b[0m┃\x1b[1m \x1b[0m\x1b[1mIQ status\x1b[0m\x1b[1m \x1b[0m┃\n┡━━━━━━━━━━━━━╇━━━━━━━━━╇━━━━━━━━━╇━━━━━━╇━━━━━━━━━━━━╇━━━━━━━━━━━┩\n│ bbe50c82-6… │  None   │  False  │  mv  │    None    │  PENDING  │\n└─────────────┴─────────┴─────────┴──────┴────────────┴───────────┘\n\n"
+
+TEST_LOG_MESSAGES = [
+    LogMessage(
+        metadata={},
+        log_type="debug",
+        log_msg={
+            "text": "datetime | debug | test log message",
+            "record": {},
+            "service_name": "ScanServer",
+        },
+    ),
+    LogMessage(
+        metadata={},
+        log_type="info",
+        log_msg={
+            "text": "datetime | info | test log message",
+            "record": {},
+            "service_name": "ScanServer",
+        },
+    ),
+    LogMessage(
+        metadata={},
+        log_type="success",
+        log_msg={
+            "text": "datetime | success | test log message",
+            "record": {},
+            "service_name": "ScanServer",
+        },
+    ),
+]
+
+TEST_COMBINED_PLAINTEXT = "datetime | debug | test log message\ndatetime | info | test log message\ndatetime | success | test log message\n"
+
+
+@pytest.fixture
+def raw_queue():
+    yield deque(TEST_LOG_MESSAGES, maxlen=100)
+
+
+@pytest.fixture
+def log_panel(qtbot, mocked_client: MagicMock):
+    widget = LogPanel(client=mocked_client)
+    qtbot.addWidget(widget)
+    qtbot.waitExposed(widget)
+    yield widget
+
+
+def test_log_panel_init(log_panel: LogPanel):
+    assert log_panel.plain_text == ""
+
+
+def test_table_string_processing():
+    assert "\x1b" in TEST_TABLE_STRING
+    sanitized = replace_escapes(TEST_TABLE_STRING)
+    assert "\x1b" not in sanitized
+    assert "  " not in sanitized
+    assert "\n" not in sanitized
+
+
+@pytest.mark.parametrize(
+    ["msg", "color"], zip(TEST_LOG_MESSAGES, ["#0000CC", "#FFFFFF", "#00FF00"])
+)
+def test_color_format(msg: LogMessage, color: str):
+    assert color in simple_color_format(msg, DEFAULT_LOG_COLORS)
+
+
+def test_logpanel_output(qtbot, log_panel: LogPanel):
+    log_panel._log_manager._data = deque(TEST_LOG_MESSAGES)
+    log_panel._on_redraw()
+    assert log_panel.plain_text == TEST_COMBINED_PLAINTEXT
+
+    def display_queue_empty():
+        return len(log_panel._log_manager._display_queue) == 0
+
+    next_text = "datetime | error | test log message"
+    log_panel._log_manager._process_incoming_log_msg(
+        {
+            "data": LogMessage(
+                metadata={},
+                log_type="error",
+                log_msg={"text": next_text, "record": {}, "service_name": "ScanServer"},
+            )
+        }
+    )
+
+    qtbot.waitUntil(display_queue_empty)
+    assert log_panel.plain_text == TEST_COMBINED_PLAINTEXT + next_text + "\n"
+
+
+def test_level_filter(log_panel: LogPanel):
+    log_panel._log_manager._data = deque(TEST_LOG_MESSAGES)
+    log_panel._log_manager.update_level_filter("INFO")
+    log_panel._on_redraw()
+    assert (
+        log_panel.plain_text
+        == "datetime | info | test log message\ndatetime | success | test log message\n"
+    )
+
+
+def test_clear_button(log_panel: LogPanel):
+    log_panel._log_manager._data = deque(TEST_LOG_MESSAGES)
+    log_panel.toolbar.clear_button.click()
+    assert log_panel._log_manager._data == deque([])