bec/bec_widgets
1
0
Fork 0
mirror of https://github.com/bec-project/bec_widgets.git synced 2026-06-05 04:48:40 +02:00
Code Activity

fix(bec_progress_bar): replace the custom paint event progressbar with native QProgressBar

Browse Source
This commit is contained in:
wyzula_j
2026-05-26 11:32:39 +02:00
parent af125e2222
commit cd150c09a9
7 changed files with 580 additions and 158 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bec_widgets/cli/client.py
+1 -1
View File
@@ -427,7 +427,7 @@ class BECMainWindow(RPCBase):
class BECProgressBar(RPCBase):
"""A custom progress bar with smooth transitions. The displayed text can be customized using a template."""
"""A BEC progress bar backed by Qt's native QProgressBar."""
_IMPORT_MODULE = "bec_widgets.widgets.progress.bec_progressbar.bec_progressbar"
bec_widgets/widgets/containers/main_window/main_window.py
+8 -4
View File
@@ -46,8 +46,8 @@ logger = bec_logger.logger
class BECMainWindow(BECWidget, QMainWindow):
RPC = True
PLUGIN = True
SCAN_PROGRESS_WIDTH = 100 # px
SCAN_PROGRESS_HEIGHT = 12 # px
SCAN_PROGRESS_WIDTH = 120 # px
SCAN_PROGRESS_HEIGHT = 20 # px
def __init__(self, parent=None, window_title: str = "BEC", **kwargs):
super().__init__(parent=parent, **kwargs)
@@ -197,7 +197,11 @@ class BECMainWindow(BECWidget, QMainWindow):
# Setting HoverWidget for the scan progress bar - minimal and full version
self._scan_progress_bar_simple = ScanProgressBar(
self, one_line_design=True, rpc_exposed=False, rpc_passthrough_children=False
self,
one_line_design=True,
rpc_exposed=False,
rpc_passthrough_children=False,
enable_dynamic_stylesheet=True,
)
self._scan_progress_bar_simple.show_elapsed_time = False
self._scan_progress_bar_simple.show_remaining_time = False
@@ -206,7 +210,7 @@ class BECMainWindow(BECWidget, QMainWindow):
self._scan_progress_bar_simple.progressbar.setFixedHeight(self.SCAN_PROGRESS_HEIGHT)
self._scan_progress_bar_simple.progressbar.setFixedWidth(self.SCAN_PROGRESS_WIDTH)
self._scan_progress_bar_full = ScanProgressBar(
self, rpc_exposed=False, rpc_passthrough_children=False
self, rpc_exposed=False, rpc_passthrough_children=False, enable_dynamic_stylesheet=False
)
self._scan_progress_hover = HoverWidget(
self, simple=self._scan_progress_bar_simple, full=self._scan_progress_bar_full
bec_widgets/widgets/progress/bec_progressbar/bec_progressbar.py
+160 -130
View File
@@ -2,8 +2,13 @@ import sys
from enum import Enum
from string import Template
from qtpy.QtCore import QEasingCurve, QPropertyAnimation, QRectF, Qt, QTimer
from qtpy.QtGui import QColor, QPainter, QPainterPath
from qtpy.QtCore import QTimer
from qtpy.QtGui import QPalette
from qtpy.QtWidgets import QApplication, QProgressBar, QSizePolicy, QVBoxLayout, QWidget
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.colors import get_accent_colors
from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
class ProgressState(Enum):
@@ -29,23 +34,12 @@ class ProgressState(Enum):
return mapping.get(status.lower(), cls.NORMAL)
PROGRESS_STATE_COLORS = {
ProgressState.NORMAL: QColor("#2979ff"), # blue – normal progress
ProgressState.PAUSED: QColor("#ffca28"), # orange/amber – paused
ProgressState.INTERRUPTED: QColor("#ff5252"), # red – interrupted
ProgressState.COMPLETED: QColor("#00e676"), # green – finished
}
from qtpy.QtWidgets import QApplication, QLabel, QVBoxLayout, QWidget
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.colors import get_accent_colors
from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
class BECProgressBar(BECWidget, QWidget):
"""
A custom progress bar with smooth transitions. The displayed text can be customized using a template.
A BEC progress bar backed by Qt's native QProgressBar.
The displayed text can be customized using a template with $value, $maximum,
and $percentage placeholders.
"""
PLUGIN = True
@@ -61,7 +55,15 @@ class BECProgressBar(BECWidget, QWidget):
]
ICON_NAME = "page_control"
def __init__(self, parent=None, client=None, config=None, gui_id=None, **kwargs):
def __init__(
self,
parent=None,
client=None,
config=None,
gui_id=None,
enable_dynamic_stylesheet: bool = True,
**kwargs,
):
super().__init__(
parent=parent, client=client, gui_id=gui_id, config=config, theme_update=True, **kwargs
)
@@ -71,7 +73,6 @@ class BECProgressBar(BECWidget, QWidget):
# internal values
self._oversampling_factor = 50
self._value = 0
self._target_value = 0
self._maximum = 100 * self._oversampling_factor
# User values
@@ -80,14 +81,7 @@ class BECProgressBar(BECWidget, QWidget):
self._user_maximum = 100
self._label_template = "$value / $maximum - $percentage %"
# Color settings
self._background_color = QColor(30, 30, 30)
self._progress_color = accent_colors.highlight
self._completed_color = accent_colors.success
self._border_color = QColor(50, 50, 50)
# Corner‑rounding: base radius in pixels (auto‑reduced if bar is small)
self._corner_radius = 10
self._corner_radius = 8
# Progress‑bar state handling
self._state = ProgressState.NORMAL
@@ -101,25 +95,28 @@ class BECProgressBar(BECWidget, QWidget):
# layout settings
self._padding_left_right = 10
self._value_animation = QPropertyAnimation(self, b"_progressbar_value")
self._value_animation.setDuration(200)
self._value_animation.setEasingCurve(QEasingCurve.Type.OutCubic)
self._chunk_radius = None
self._enable_dynamic_stylesheet = enable_dynamic_stylesheet
# label on top of the progress bar
self.center_label = QLabel(self)
self.center_label.setAlignment(Qt.AlignHCenter)
self.center_label.setMinimumSize(0, 0)
self.center_label.setStyleSheet("background: transparent; color: white;")
self.progressbar = QProgressBar(self)
self.progressbar.setTextVisible(True)
self.progressbar.setRange(0, self._maximum)
self.progressbar.setMinimumHeight(0)
self.progressbar.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Ignored)
layout = QVBoxLayout(self)
layout.setContentsMargins(10, 0, 10, 0)
layout.setSpacing(0)
layout.addWidget(self.center_label)
layout.setAlignment(self.center_label, Qt.AlignCenter)
self.setLayout(layout)
# Backwards-compatible alias used by existing tests and downstream code.
self.center_label = self.progressbar
self.update()
self._adjust_label_width()
self._layout = QVBoxLayout(self)
self._layout.setContentsMargins(self._padding_left_right, 0, self._padding_left_right, 0)
self._layout.setSpacing(0)
self._layout.addWidget(self.progressbar)
self.setLayout(self._layout)
self._setup_style_sheet(chunk_radius=self._initial_chunk_radius())
self._sync_progressbar()
self._apply_state_style()
self._update_chunk_radius(force=True)
@SafeProperty(
str, doc="The template for the center label. Use $value, $maximum, and $percentage."
@@ -144,13 +141,12 @@ class BECProgressBar(BECWidget, QWidget):
ProgressState.INTERRUPTED: accent_colors.emergency,
ProgressState.COMPLETED: accent_colors.success,
}
self._apply_state_style()
@label_template.setter
def label_template(self, template):
self._label_template = template
self._adjust_label_width()
self.set_value(self._user_value)
self.update()
self._sync_progressbar()
@SafeProperty(float, designable=False)
def _progressbar_value(self):
@@ -162,28 +158,16 @@ class BECProgressBar(BECWidget, QWidget):
@_progressbar_value.setter
def _progressbar_value(self, val):
self._value = val
self.update()
self.progressbar.setValue(int(round(val)))
def _update_template(self):
template = Template(self._label_template)
return template.safe_substitute(
value=self._user_value,
maximum=self._user_maximum,
percentage=int((self.map_value(self._user_value) / self._maximum) * 100),
percentage=int(self._percentage(self._user_value)),
)
def _adjust_label_width(self):
"""
Reserve enough horizontal space for the center label so the widget
doesn't resize as the text grows during progress.
"""
template = Template(self._label_template)
sample_text = template.safe_substitute(
value=self._user_maximum, maximum=self._user_maximum, percentage=100
)
width = self.center_label.fontMetrics().horizontalAdvance(sample_text)
self.center_label.setFixedWidth(width)
@SafeSlot(float)
@SafeSlot(int)
def set_value(self, value):
@@ -193,13 +177,12 @@ class BECProgressBar(BECWidget, QWidget):
Args:
value (float): The value to set.
"""
if value > self._user_maximum:
value = self._user_maximum
elif value < self._user_minimum:
value = self._user_minimum
self._target_value = self.map_value(value)
self._user_value = value
self.center_label.setText(self._update_template())
previous_visual_state = self._current_visual_state()
previous_value = self._value
self._user_value = self._clamp_value(value)
self._value = self.map_value(self._user_value)
if self._value < previous_value:
self._chunk_radius = None
# Update state automatically unless paused or interrupted
if self._state not in (ProgressState.PAUSED, ProgressState.INTERRUPTED):
self._state = (
@@ -207,7 +190,12 @@ class BECProgressBar(BECWidget, QWidget):
if self._user_value >= self._user_maximum
else ProgressState.NORMAL
)
self.animate_progress()
self._sync_progressbar()
target_radius = self._target_chunk_radius()
if self._enable_dynamic_stylesheet and self._chunk_radius != target_radius:
self._update_chunk_radius()
if self._current_visual_state() is not previous_visual_state:
self._apply_state_style()
@SafeProperty(object, doc="Current visual state of the progress bar.")
def state(self):
@@ -226,7 +214,7 @@ class BECProgressBar(BECWidget, QWidget):
if not isinstance(state, ProgressState):
raise ValueError("state must be a ProgressState or its value")
self._state = state
self.update()
self._apply_state_style()
@SafeProperty(float, doc="Base corner radius in pixels (auto‑scaled down on small bars).")
def corner_radius(self) -> float:
@@ -235,7 +223,18 @@ class BECProgressBar(BECWidget, QWidget):
@corner_radius.setter
def corner_radius(self, radius: float):
self._corner_radius = max(0.0, radius)
self.update()
self._chunk_radius = None
self._update_chunk_radius(force=True)
@SafeProperty(bool)
def enable_dynamic_stylesheet(self) -> bool:
return self._enable_dynamic_stylesheet
@enable_dynamic_stylesheet.setter
def enable_dynamic_stylesheet(self, enabled: bool):
self._enable_dynamic_stylesheet = bool(enabled)
self._chunk_radius = None
self._update_chunk_radius(force=True)
@SafeProperty(float)
def padding_left_right(self) -> float:
@@ -244,60 +243,12 @@ class BECProgressBar(BECWidget, QWidget):
@padding_left_right.setter
def padding_left_right(self, padding: float):
self._padding_left_right = padding
self.update()
self._layout.setContentsMargins(int(round(padding)), 0, int(round(padding)), 0)
def paintEvent(self, event):
painter = QPainter(self)
painter.setRenderHint(QPainter.Antialiasing)
rect = self.rect().adjusted(self._padding_left_right, 0, -self._padding_left_right, -1)
# Corner radius adapts to widget height so it never exceeds half the bar’s thickness
radius = min(self._corner_radius, rect.height() / 2)
# Draw background
painter.setBrush(self._background_color)
painter.setPen(Qt.NoPen)
painter.drawRoundedRect(rect, radius, radius) # Rounded corners
# Draw border
painter.setBrush(Qt.NoBrush)
painter.setPen(self._border_color)
painter.drawRoundedRect(rect, radius, radius)
# Determine progress colour based on current state
if self._state == ProgressState.PAUSED:
current_color = self._state_colors[ProgressState.PAUSED]
elif self._state == ProgressState.INTERRUPTED:
current_color = self._state_colors[ProgressState.INTERRUPTED]
elif self._state == ProgressState.COMPLETED or self._value >= self._maximum:
current_color = self._state_colors[ProgressState.COMPLETED]
else:
current_color = self._state_colors[ProgressState.NORMAL]
# Set clipping region to preserve the background's rounded corners
progress_rect = rect.adjusted(
0, 0, int(-rect.width() + (self._value / self._maximum) * rect.width()), 0
)
clip_path = QPainterPath()
clip_path.addRoundedRect(
QRectF(rect), radius, radius
) # Clip to the background's rounded corners
painter.setClipPath(clip_path)
# Draw progress bar
painter.setBrush(current_color)
painter.drawRect(progress_rect) # Less rounded, no additional rounding
painter.end()
def animate_progress(self):
"""
Animate the progress bar from the current value to the target value.
"""
self._value_animation.stop()
self._value_animation.setStartValue(self._value)
self._value_animation.setEndValue(self._target_value)
self._value_animation.start()
def resizeEvent(self, event):
super().resizeEvent(event)
self._chunk_radius = None
self._update_chunk_radius(force=True)
@SafeProperty(float)
def maximum(self):
@@ -343,10 +294,11 @@ class BECProgressBar(BECWidget, QWidget):
Args:
maximum (float): The maximum value.
"""
previous_maximum = self._user_maximum
self._user_maximum = maximum
self._adjust_label_width()
if maximum != previous_maximum:
self._chunk_radius = None
self.set_value(self._user_value) # Update the value to fit the new range
self.update()
@SafeSlot(float)
def set_minimum(self, minimum: float):
@@ -356,21 +308,99 @@ class BECProgressBar(BECWidget, QWidget):
Args:
minimum (float): The minimum value.
"""
previous_minimum = self._user_minimum
self._user_minimum = minimum
if minimum != previous_minimum:
self._chunk_radius = None
self.set_value(self._user_value) # Update the value to fit the new range
self.update()
def map_value(self, value: float):
"""
Map the user value to the range [0, 100*self._oversampling_factor] for the progress
"""
return (
(value - self._user_minimum) / (self._user_maximum - self._user_minimum) * self._maximum
)
span = self._user_maximum - self._user_minimum
if span <= 0:
return float(self._maximum if value >= self._user_maximum else 0)
mapped_value = (value - self._user_minimum) / span * self._maximum
return min(float(self._maximum), max(0.0, mapped_value))
def _percentage(self, value: float) -> float:
return (self.map_value(value) / self._maximum) * 100 if self._maximum else 0.0
def _clamp_value(self, value: float) -> float:
if self._user_maximum <= self._user_minimum:
return self._user_maximum
return min(self._user_maximum, max(self._user_minimum, value))
def _sync_progressbar(self) -> None:
self.progressbar.setRange(0, int(self._maximum))
self.progressbar.setValue(int(round(self._value)))
self.progressbar.setFormat(self._update_template())
def _setup_style_sheet(self, *, chunk_radius: int) -> None:
radius = int(round(self._corner_radius))
self.progressbar.setStyleSheet(f"""
QProgressBar {{
background-color: palette(mid);
border: none;
border-radius: {radius}px;
color: palette(text);
text-align: center;
}}
QProgressBar::chunk {{
background-color: palette(highlight);
border-radius: {chunk_radius}px;
}}
""")
def _update_chunk_radius(self, *, force: bool = False) -> None:
target_radius = self._target_chunk_radius()
if not self._enable_dynamic_stylesheet:
if not force and self._chunk_radius == target_radius:
return
self._chunk_radius = target_radius
self._setup_style_sheet(chunk_radius=target_radius)
return
if not force and self._chunk_radius == target_radius:
return
chunk_radius = self._calculate_chunk_radius(target_radius)
if not force and chunk_radius == self._chunk_radius:
return
self._chunk_radius = chunk_radius
self._setup_style_sheet(chunk_radius=chunk_radius)
def _target_chunk_radius(self) -> int:
radius = int(round(self._corner_radius))
return max(0, radius - 1)
def _initial_chunk_radius(self) -> int:
return 0 if self._enable_dynamic_stylesheet else self._target_chunk_radius()
def _calculate_chunk_radius(self, target_radius: int) -> int:
if target_radius <= 0 or self._maximum <= 0:
return 0
fill_width = self.progressbar.width() * min(1.0, max(0.0, self._value / self._maximum))
if fill_width <= 0:
return 0
return min(target_radius, max(1, int(fill_width / 2)))
def _apply_state_style(self) -> None:
color = self._state_colors[self._current_visual_state()]
palette = self.progressbar.palette()
palette.setColor(QPalette.ColorRole.Highlight, color)
palette.setColor(QPalette.ColorRole.HighlightedText, palette.color(QPalette.ColorRole.Text))
self.progressbar.setPalette(palette)
def _current_visual_state(self) -> ProgressState:
if self._state in (ProgressState.PAUSED, ProgressState.INTERRUPTED):
return self._state
if self._state == ProgressState.COMPLETED or self._value >= self._maximum:
return ProgressState.COMPLETED
return ProgressState.NORMAL
def _get_label(self) -> str:
"""Return the label text. mostly used for testing rpc."""
return self.center_label.text()
return self.progressbar.text()
if __name__ == "__main__": # pragma: no cover
bec_widgets/widgets/progress/scan_progressbar/scan_progressbar.py
+57 -15
View File
@@ -37,7 +37,7 @@ class ProgressTask(QObject):
def __init__(self, parent: QWidget, value: float = 0, max_value: float = 0, done: bool = False):
super().__init__(parent=parent)
self.start_time = time.time()
self.start_time = time.monotonic()
self.done = done
self.value = value
self.max_value = max_value
@@ -45,7 +45,7 @@ class ProgressTask(QObject):
self.timer = QTimer(self)
self.timer.timeout.connect(self.update_elapsed_time)
self.timer.start(100) # update the elapsed time every 100 ms
self.timer.start(1000)
def update(self, value: float, max_value: float, done: bool = False):
"""
@@ -59,9 +59,9 @@ class ProgressTask(QObject):
def update_elapsed_time(self):
"""
Update the time estimates. This is called every 100 ms by a QTimer.
Update the time estimates. This is called every second by a QTimer.
"""
self._elapsed_time += 0.1
self._elapsed_time = max(0.0, time.monotonic() - self.start_time)
@property
def percentage(self) -> float:
@@ -130,7 +130,14 @@ class ScanProgressBar(BECWidget, QWidget):
progress_finished = Signal()
def __init__(
self, parent=None, client=None, config=None, gui_id=None, one_line_design=False, **kwargs
self,
parent=None,
client=None,
config=None,
gui_id=None,
one_line_design=False,
enable_dynamic_stylesheet: bool = True,
**kwargs,
):
super().__init__(parent=parent, client=client, config=config, gui_id=gui_id, **kwargs)
@@ -146,15 +153,17 @@ class ScanProgressBar(BECWidget, QWidget):
self.layout.addWidget(self.ui)
self.setLayout(self.layout)
self.progressbar = self.ui.progressbar
self.progressbar.enable_dynamic_stylesheet = enable_dynamic_stylesheet
self._show_elapsed_time = self.ui.elapsed_time_label.isVisible()
self._show_remaining_time = self.ui.remaining_time_label.isVisible()
self._show_source_label = self.ui.source_label.isVisible()
self.connect_to_queue()
self._progress_source = None
self._progress_device = None
self.task = None
self.scan_number = None
self._active_scan_id = None
self.connect_to_queue()
def connect_to_queue(self):
"""
@@ -191,9 +200,31 @@ class ScanProgressBar(BECWidget, QWidget):
self.update_source_label(source, device=device)
# self.progress_started.emit()
def _start_task(self, scan_id: str | None) -> None:
if self.task is not None:
self.task.timer.stop()
self.task.deleteLater()
self.task = ProgressTask(parent=self)
self.task.timer.timeout.connect(self.update_labels)
self._active_scan_id = scan_id
self.progress_started.emit()
def _clear_task(self, *, emit_finished: bool = True) -> None:
if self.task is None:
self._active_scan_id = None
return
self.task.timer.stop()
self.task.deleteLater()
self.task = None
self._active_scan_id = None
if emit_finished:
self.progress_finished.emit()
def update_source_label(self, source: ProgressSource, device=None):
scan_text = f"Scan {self.scan_number}" if self.scan_number is not None else "Scan"
text = scan_text if source == ProgressSource.SCAN_PROGRESS else f"Device {device}"
if self.ui.source_label.text() == text:
return
logger.info(f"Set progress source to {text}")
self.ui.source_label.setText(text)
@@ -220,8 +251,7 @@ class ScanProgressBar(BECWidget, QWidget):
self.progressbar.set_value(self.task.value)
if done:
self.task = None
self.progress_finished.emit()
self._clear_task()
return
@SafeProperty(bool)
@@ -271,27 +301,40 @@ class ScanProgressBar(BECWidget, QWidget):
Update the progress bar based on the queue status.
"""
if not "queue" in msg_content:
self._clear_task()
return
if "primary" not in msg_content["queue"]:
self._clear_task()
return
if (primary_queue := msg_content.get("queue").get("primary")) is None:
self._clear_task()
return
if not isinstance(primary_queue, messages.ScanQueueStatus):
self._clear_task()
return
primary_queue_info = primary_queue.info
if len(primary_queue_info) == 0:
self._clear_task()
return
scan_info = primary_queue_info[0]
if scan_info is None:
self._clear_task()
return
if scan_info.status.lower() == "running" and self.task is None:
self.task = ProgressTask(parent=self)
self.progress_started.emit()
active_request_block = scan_info.active_request_block
if active_request_block is None:
self._clear_task()
return
status = scan_info.status.lower()
if status != "running":
self._clear_task()
return
scan_id = active_request_block.scan_id or str(active_request_block.scan_number)
if self.task is None or self._active_scan_id != scan_id:
self._start_task(scan_id)
self.scan_number = active_request_block.scan_number
report_instructions = active_request_block.report_instructions
if not report_instructions:
@@ -303,14 +346,13 @@ class ScanProgressBar(BECWidget, QWidget):
if "scan_progress" in instruction:
self.set_progress_source(ProgressSource.SCAN_PROGRESS)
elif "device_progress" in instruction:
if not instruction["device_progress"]:
return
device = instruction["device_progress"][0]
self.set_progress_source(ProgressSource.DEVICE_PROGRESS, device=device)
def cleanup(self):
if self.task is not None:
self.task.timer.stop()
self.close()
self.deleteLater()
self._clear_task(emit_finished=False)
if self._progress_source is not None:
self.bec_dispatcher.disconnect_slot(
self.on_progress_update,
tests/unit_tests/test_bec_progressbar.py
+134 -7
View File
@@ -1,5 +1,8 @@
import numpy as np
from unittest import mock
import pytest
from qtpy.QtGui import QPalette
from qtpy.QtWidgets import QProgressBar
from bec_widgets.widgets.progress.bec_progressbar.bec_progressbar import (
BECProgressBar,
@@ -15,6 +18,14 @@ def progressbar(qtbot):
yield widget
@pytest.fixture
def static_progressbar(qtbot):
widget = BECProgressBar(enable_dynamic_stylesheet=False)
qtbot.addWidget(widget)
qtbot.waitExposed(widget)
yield widget
def test_progressbar(progressbar):
progressbar.update()
@@ -23,21 +34,137 @@ def test_progressbar_set_value(qtbot, progressbar):
progressbar.set_minimum(0)
progressbar.set_maximum(100)
progressbar.set_value(50)
progressbar.paintEvent(None)
qtbot.waitUntil(
lambda: np.isclose(
progressbar._value, progressbar._user_value * progressbar._oversampling_factor
)
)
assert isinstance(progressbar.progressbar, QProgressBar)
assert progressbar._value == progressbar._user_value * progressbar._oversampling_factor
assert progressbar.progressbar.value() == 50 * progressbar._oversampling_factor
def test_progressbar_label(progressbar):
progressbar.label_template = "Test: $value"
progressbar.set_value(50)
assert progressbar._get_label() == "Test: 50"
assert progressbar.center_label.text() == "Test: 50"
def test_progressbar_equal_minimum_and_maximum_does_not_raise(progressbar):
progressbar.set_minimum(0)
progressbar.set_maximum(0)
progressbar.set_value(0)
assert progressbar._get_label() == "0 / 0 - 100 %"
assert progressbar.progressbar.value() == progressbar.progressbar.maximum()
def test_progressbar_uses_static_stylesheet_with_palette_state_color(progressbar):
progressbar.progressbar.resize(100, 20)
progressbar.set_value(50)
progressbar.state = ProgressState.PAUSED
style_sheet = progressbar.progressbar.styleSheet()
assert "QProgressBar::chunk" in style_sheet
assert "background-color: palette(highlight);" in style_sheet
assert "background-color: palette(mid);" in style_sheet
assert "border-radius: 7px;" in style_sheet
assert (
progressbar.progressbar.palette().color(QPalette.ColorRole.Highlight)
== progressbar._state_colors[ProgressState.PAUSED]
)
def test_progressbar_value_updates_do_not_rebuild_stylesheet_within_same_chunk_mode(progressbar):
progressbar.progressbar.resize(100, 20)
progressbar.set_value(30)
with mock.patch.object(
progressbar, "_setup_style_sheet", wraps=progressbar._setup_style_sheet
) as setup_style_sheet:
progressbar.set_value(35)
progressbar.set_value(42)
progressbar.set_value(50)
setup_style_sheet.assert_not_called()
def test_progressbar_value_updates_skip_chunk_radius_after_target_reached(progressbar):
progressbar.progressbar.resize(100, 20)
progressbar.set_value(30)
assert progressbar._chunk_radius == progressbar._target_chunk_radius()
with mock.patch.object(
progressbar, "_update_chunk_radius", wraps=progressbar._update_chunk_radius
) as update_chunk_radius:
progressbar.set_value(35)
progressbar.set_value(42)
progressbar.set_value(50)
update_chunk_radius.assert_not_called()
def test_progressbar_repeated_same_maximum_does_not_reset_chunk_radius(progressbar):
progressbar.progressbar.resize(100, 20)
progressbar.set_maximum(100)
progressbar.set_value(30)
assert progressbar._chunk_radius == progressbar._target_chunk_radius()
with mock.patch.object(
progressbar, "_update_chunk_radius", wraps=progressbar._update_chunk_radius
) as update_chunk_radius:
progressbar.set_maximum(100)
progressbar.set_value(40)
update_chunk_radius.assert_not_called()
def test_progressbar_can_disable_dynamic_stylesheet(static_progressbar):
static_progressbar.progressbar.resize(100, 20)
assert static_progressbar.enable_dynamic_stylesheet is False
assert static_progressbar._chunk_radius == static_progressbar._target_chunk_radius()
with mock.patch.object(
static_progressbar, "_setup_style_sheet", wraps=static_progressbar._setup_style_sheet
) as setup_style_sheet:
static_progressbar.set_value(1)
static_progressbar.set_value(2)
static_progressbar.set_value(3)
setup_style_sheet.assert_not_called()
assert "border-radius: 7px;" in static_progressbar.progressbar.styleSheet()
def test_progressbar_dynamic_stylesheet_can_be_toggled(progressbar):
progressbar.enable_dynamic_stylesheet = False
assert progressbar.enable_dynamic_stylesheet is False
assert progressbar._chunk_radius == progressbar._target_chunk_radius()
assert "border-radius: 7px;" in progressbar.progressbar.styleSheet()
def test_progressbar_rebuilds_stylesheet_until_chunk_radius_reaches_target(progressbar):
progressbar.progressbar.resize(100, 20)
progressbar.set_value(9)
with mock.patch.object(
progressbar, "_setup_style_sheet", wraps=progressbar._setup_style_sheet
) as setup_style_sheet:
progressbar.set_value(12)
progressbar.set_value(25)
progressbar.set_value(30)
assert setup_style_sheet.call_count == 2
assert "border-radius: 7px;" in progressbar.progressbar.styleSheet()
def test_progressbar_resets_chunk_radius_when_value_goes_backwards(progressbar):
progressbar.progressbar.resize(100, 20)
progressbar.set_value(30)
assert "border-radius: 7px;" in progressbar.progressbar.styleSheet()
progressbar.set_value(4)
assert "border-radius: 2px;" in progressbar.progressbar.styleSheet()
def test_progress_state_from_bec_status():
"""ProgressState.from_bec_status() maps BEC literals correctly."""
mapping = {
tests/unit_tests/test_main_widnow.py
+7
View File
@@ -117,6 +117,13 @@ def test_hidden_scan_progress_parent_blocks_children_namespace(bec_main_window):
assert nested_progress.parent_id == hidden_progress.gui_id
def test_compact_scan_progress_bar_uses_status_bar_sizing(bec_main_window):
progressbar = bec_main_window._scan_progress_bar_simple.progressbar
assert progressbar.height() == bec_main_window.SCAN_PROGRESS_HEIGHT
assert progressbar.progressbar.minimumHeight() == 0
#################################################################
# Tests for BECMainWindow Addons
#################################################################
tests/unit_tests/test_scan_progress_bar.py
+213 -1
View File
@@ -71,11 +71,33 @@ def test_progress_task_basic():
assert task.time_elapsed == "00:00:10"
def test_progress_task_elapsed_time_uses_monotonic_clock(monkeypatch):
times = iter([100.0, 102.5])
monkeypatch.setattr(
"bec_widgets.widgets.progress.scan_progressbar.scan_progressbar.time.monotonic",
lambda: next(times),
)
task = ProgressTask(parent=None)
task.timer.stop()
task.update_elapsed_time()
assert task._elapsed_time == 2.5
assert task.time_elapsed == "00:00:02"
def test_scan_progressbar_initialization(scan_progressbar):
assert isinstance(scan_progressbar, ScanProgressBar)
assert isinstance(scan_progressbar.progressbar, BECProgressBar)
def test_scan_progressbar_passes_dynamic_stylesheet_setting(qtbot, mocked_client):
widget = ScanProgressBar(client=mocked_client, enable_dynamic_stylesheet=False)
qtbot.addWidget(widget)
assert widget.progressbar.enable_dynamic_stylesheet is False
def test_update_labels_content(scan_progressbar):
"""update_labels() reflects ProgressTask time strings on the UI."""
# fabricate a task with known timings
@@ -148,6 +170,19 @@ def test_source_label_updates(scan_progressbar):
assert scan_progressbar.ui.source_label.text() == "Scan 5"
def test_source_label_update_logs_only_on_text_change(scan_progressbar):
scan_progressbar.scan_number = 5
with mock.patch(
"bec_widgets.widgets.progress.scan_progressbar.scan_progressbar.logger.info"
) as mock_info:
scan_progressbar.set_progress_source(ProgressSource.SCAN_PROGRESS)
scan_progressbar.set_progress_source(ProgressSource.SCAN_PROGRESS)
scan_progressbar.set_progress_source(ProgressSource.SCAN_PROGRESS)
mock_info.assert_called_once_with("Set progress source to Scan 5")
def test_set_progress_source_connections(scan_progressbar, monkeypatch):
""" """
@@ -241,7 +276,13 @@ def test_cleanup_disconnects_active_device_subscription(scan_progressbar, monkey
monkeypatch.setattr(BECWidget, "cleanup", lambda self: None)
scan_progressbar.set_progress_source(ProgressSource.DEVICE_PROGRESS, device="motor1")
ScanProgressBar.cleanup(scan_progressbar)
with (
mock.patch.object(scan_progressbar, "close", wraps=scan_progressbar.close) as close_mock,
mock.patch.object(
scan_progressbar, "deleteLater", wraps=scan_progressbar.deleteLater
) as delete_later_mock,
):
ScanProgressBar.cleanup(scan_progressbar)
assert disconnect_calls == [
MessageEndpoints.device_progress(device="motor1"),
@@ -249,6 +290,21 @@ def test_cleanup_disconnects_active_device_subscription(scan_progressbar, monkey
]
assert scan_progressbar._progress_source is None
assert scan_progressbar._progress_device is None
close_mock.assert_not_called()
delete_later_mock.assert_not_called()
def test_cleanup_stops_active_task(scan_progressbar, monkeypatch):
monkeypatch.setattr(BECWidget, "cleanup", lambda self: None)
scan_progressbar.task = ProgressTask(parent=scan_progressbar)
scan_progressbar._active_scan_id = "scan-1"
timer = scan_progressbar.task.timer
ScanProgressBar.cleanup(scan_progressbar)
assert not timer.isActive()
assert scan_progressbar.task is None
assert scan_progressbar._active_scan_id is None
def test_progressbar_queue_update(scan_progressbar):
@@ -265,6 +321,70 @@ def test_progressbar_queue_update(scan_progressbar):
mock_set_source.assert_not_called()
def test_progressbar_queue_update_clears_task_when_queue_is_empty(scan_progressbar):
scan_progressbar.task = ProgressTask(parent=scan_progressbar)
scan_progressbar._active_scan_id = "scan-1"
timer = scan_progressbar.task.timer
msg = messages.ScanQueueStatusMessage(
queue={"primary": messages.ScanQueueStatus(info=[], status="RUNNING")}
)
scan_progressbar.on_queue_update(
msg.content, msg.metadata, _override_slot_params={"verify_sender": False}
)
assert not timer.isActive()
assert scan_progressbar.task is None
assert scan_progressbar._active_scan_id is None
def test_progressbar_queue_update_clears_task_when_scan_is_not_running(
scan_progressbar, scan_message
):
scan_progressbar.task = ProgressTask(parent=scan_progressbar)
scan_progressbar._active_scan_id = "scan-1"
timer = scan_progressbar.task.timer
request_block = messages.RequestBlock(
msg=scan_message,
RID="some-rid",
scan_motors=["samx"],
readout_priority={"monitored": ["samx"]},
is_scan=True,
scan_number=1,
scan_id="scan-1",
report_instructions=[{"scan_progress": 20}],
)
msg = messages.ScanQueueStatusMessage(
metadata={},
queue={
"primary": messages.ScanQueueStatus(
info=[
messages.QueueInfoEntry(
queue_id="queue-1",
scan_id=["scan-1"],
status="completed",
active_request_block=request_block,
is_scan=[True],
request_blocks=[request_block],
scan_number=[1],
)
],
status="RUNNING",
)
},
)
with mock.patch.object(scan_progressbar, "set_progress_source") as mock_set_source:
scan_progressbar.on_queue_update(
msg.content, msg.metadata, _override_slot_params={"verify_sender": False}
)
assert not timer.isActive()
assert scan_progressbar.task is None
assert scan_progressbar._active_scan_id is None
mock_set_source.assert_not_called()
def test_progressbar_queue_update_with_scan(scan_progressbar, scan_message):
"""
Test that a queue update with a scan changes the progress source to SCAN_PROGRESS.
@@ -306,6 +426,60 @@ def test_progressbar_queue_update_with_scan(scan_progressbar, scan_message):
mock_set_source.assert_called_once_with(ProgressSource.SCAN_PROGRESS)
def test_progressbar_queue_update_starts_new_task_for_new_scan(scan_progressbar, scan_message):
started = mock.Mock()
scan_progressbar.progress_started.connect(started)
def queue_msg(scan_id: str, scan_number: int):
request_block = messages.RequestBlock(
msg=scan_message,
RID=f"rid-{scan_number}",
scan_motors=["samx"],
readout_priority={"monitored": ["samx"]},
is_scan=True,
scan_number=scan_number,
scan_id=scan_id,
report_instructions=[{"scan_progress": 20}],
)
return messages.ScanQueueStatusMessage(
metadata={},
queue={
"primary": messages.ScanQueueStatus(
info=[
messages.QueueInfoEntry(
queue_id=f"queue-{scan_number}",
scan_id=[scan_id],
status="RUNNING",
active_request_block=request_block,
is_scan=[True],
request_blocks=[request_block],
scan_number=[scan_number],
)
],
status="RUNNING",
)
},
)
first_msg = queue_msg("scan-1", 1)
scan_progressbar.on_queue_update(
first_msg.content, first_msg.metadata, _override_slot_params={"verify_sender": False}
)
first_task = scan_progressbar.task
assert first_task is not None
assert first_task.timer.isActive()
second_msg = queue_msg("scan-2", 2)
scan_progressbar.on_queue_update(
second_msg.content, second_msg.metadata, _override_slot_params={"verify_sender": False}
)
assert started.call_count == 2
assert not first_task.timer.isActive()
assert scan_progressbar.task is not first_task
assert scan_progressbar._active_scan_id == "scan-2"
def test_progressbar_queue_update_with_device(scan_progressbar, scan_message):
"""
Test that a queue update with a device changes the progress source to DEVICE_PROGRESS.
@@ -347,6 +521,44 @@ def test_progressbar_queue_update_with_device(scan_progressbar, scan_message):
mock_set_source.assert_called_once_with(ProgressSource.DEVICE_PROGRESS, device="samx")
def test_progressbar_queue_update_ignores_empty_device_progress(scan_progressbar, scan_message):
request_block = messages.RequestBlock(
msg=scan_message,
RID="some-rid",
scan_motors=["samx"],
readout_priority={"monitored": ["samx"]},
is_scan=True,
scan_number=1,
scan_id="e3f50794-852c-4bb1-965e-41c585ab0aa9",
report_instructions=[{"device_progress": []}],
)
msg = messages.ScanQueueStatusMessage(
metadata={},
queue={
"primary": messages.ScanQueueStatus(
info=[
messages.QueueInfoEntry(
queue_id="40831e2c-fbd1-4432-8072-ad168a7ad964",
scan_id=["e3f50794-852c-4bb1-965e-41c585ab0aa9"],
status="RUNNING",
active_request_block=request_block,
is_scan=[True],
request_blocks=[request_block],
scan_number=[1],
)
],
status="RUNNING",
)
},
)
with mock.patch.object(scan_progressbar, "set_progress_source") as mock_set_source:
scan_progressbar.on_queue_update(
msg.content, msg.metadata, _override_slot_params={"verify_sender": False}
)
mock_set_source.assert_not_called()
def test_progressbar_queue_update_with_no_scan_or_device(scan_progressbar, scan_message):
"""
Test that a queue update with neither scan nor device does not change the progress source.