diff --git a/bec_widgets/cli/client.py b/bec_widgets/cli/client.py index 4ab60ac0..7c752bca 100644 --- a/bec_widgets/cli/client.py +++ b/bec_widgets/cli/client.py @@ -31,7 +31,6 @@ class Widgets(str, enum.Enum): DeviceComboBox = "DeviceComboBox" DeviceLineEdit = "DeviceLineEdit" LMFitDialog = "LMFitDialog" - LogPanel = "LogPanel" Minesweeper = "Minesweeper" PositionIndicator = "PositionIndicator" PositionerBox = "PositionerBox" @@ -3184,26 +3183,6 @@ 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): ... diff --git a/bec_widgets/widgets/__init__.py b/bec_widgets/widgets/__init__.py index e69de29b..8b137891 100644 --- a/bec_widgets/widgets/__init__.py +++ b/bec_widgets/widgets/__init__.py @@ -0,0 +1 @@ + diff --git a/bec_widgets/widgets/containers/dock/dock_area.py b/bec_widgets/widgets/containers/dock/dock_area.py index d9bce695..6d41c0f9 100644 --- a/bec_widgets/widgets/containers/dock/dock_area.py +++ b/bec_widgets/widgets/containers/dock/dock_area.py @@ -30,7 +30,6 @@ 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 @@ -140,9 +139,6 @@ 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(), @@ -204,9 +200,6 @@ 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) diff --git a/bec_widgets/widgets/editors/text_box/text_box.py b/bec_widgets/widgets/editors/text_box/text_box.py index 8b549acb..10316cba 100644 --- a/bec_widgets/widgets/editors/text_box/text_box.py +++ b/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) - @SafeSlot(str) + @Slot(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 - @SafeSlot(str) + @Slot(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 - @SafeProperty(str) + @Property(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) - @SafeProperty(str) + @Property(str) def html_text(self) -> str: """Get the HTML text of the widget. diff --git a/bec_widgets/widgets/utility/logpanel/__init__.py b/bec_widgets/widgets/utility/logpanel/__init__.py deleted file mode 100644 index a90c4978..00000000 --- a/bec_widgets/widgets/utility/logpanel/__init__.py +++ /dev/null @@ -1,3 +0,0 @@ -from bec_widgets.widgets.utility.logpanel.logpanel import LogPanel - -__ALL__ = ["LogPanel"] diff --git a/bec_widgets/widgets/utility/logpanel/_util.py b/bec_widgets/widgets/utility/logpanel/_util.py deleted file mode 100644 index 547ee208..00000000 --- a/bec_widgets/widgets/utility/logpanel/_util.py +++ /dev/null @@ -1,58 +0,0 @@ -""" 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(" ", " ").replace("\n", "
").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()) + "
" - - -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'{noop_format(line)}' - - -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"] diff --git a/bec_widgets/widgets/utility/logpanel/log_panel.pyproject b/bec_widgets/widgets/utility/logpanel/log_panel.pyproject deleted file mode 100644 index 579e3212..00000000 --- a/bec_widgets/widgets/utility/logpanel/log_panel.pyproject +++ /dev/null @@ -1 +0,0 @@ -{'files': ['logpanel.py']} \ No newline at end of file diff --git a/bec_widgets/widgets/utility/logpanel/log_panel_plugin.py b/bec_widgets/widgets/utility/logpanel/log_panel_plugin.py deleted file mode 100644 index 9ee99a3c..00000000 --- a/bec_widgets/widgets/utility/logpanel/log_panel_plugin.py +++ /dev/null @@ -1,54 +0,0 @@ -# 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 = """ - - - - -""" - - -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() diff --git a/bec_widgets/widgets/utility/logpanel/logpanel.py b/bec_widgets/widgets/utility/logpanel/logpanel.py deleted file mode 100644 index ffd68844..00000000 --- a/bec_widgets/widgets/utility/logpanel/logpanel.py +++ /dev/null @@ -1,516 +0,0 @@ -""" 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()) diff --git a/bec_widgets/widgets/utility/logpanel/register_log_panel.py b/bec_widgets/widgets/utility/logpanel/register_log_panel.py deleted file mode 100644 index b4414d04..00000000 --- a/bec_widgets/widgets/utility/logpanel/register_log_panel.py +++ /dev/null @@ -1,15 +0,0 @@ -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() diff --git a/tests/unit_tests/test_logpanel.py b/tests/unit_tests/test_logpanel.py deleted file mode 100644 index b44265b4..00000000 --- a/tests/unit_tests/test_logpanel.py +++ /dev/null @@ -1,117 +0,0 @@ -# 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([])