mirror of
https://github.com/bec-project/bec_widgets.git
synced 2026-10-27 16:01:27 +01:00
feat(scan_progressbar): add QML and QWidget versions with modern UX for comparison
This commit is contained in:
7 files changed
+1329
No files matched your search
@@ -0,0 +1,113 @@
|
||||
"""
|
||||
Side-by-side comparison of the scan progress bar: today's widget, the QML port and the QWidget
|
||||
version with the same UX. A simulated scan runs through every state; no BEC server is needed.
|
||||
|
||||
Run with: python -m bec_widgets.examples.scan_progress_compare [--light]
|
||||
"""
|
||||
|
||||
import sys
|
||||
import time
|
||||
from unittest import mock
|
||||
|
||||
from qtpy.QtCore import QTimer
|
||||
from qtpy.QtWidgets import QApplication, QGridLayout, QLabel, QPushButton, QWidget
|
||||
|
||||
from bec_widgets.utils.colors import apply_theme
|
||||
from bec_widgets.widgets.progress.scan_progressbar.scan_progressbar import ScanProgressBar
|
||||
from bec_widgets.widgets.progress.scan_progressbar.scan_progressbar_modern import (
|
||||
ScanProgressBarModern,
|
||||
)
|
||||
from bec_widgets.widgets.progress.scan_progressbar.scan_progressbar_qml import ScanProgressBarQml
|
||||
|
||||
POINTS = 60
|
||||
|
||||
# (status sent with the progress messages, points to advance, final scan status or None)
|
||||
SCRIPT = [("open", 0, None)] * 3 + [("open", 1, None)] * 25 + [("paused", 0, None)] * 6
|
||||
SCRIPT += [("open", 1, None)] * 20 + [("open", 0, "halted")] * 6 + [("open", 1, None)] * POINTS
|
||||
|
||||
|
||||
class Demo(QWidget):
|
||||
"""Drives all bars with the same simulated scan."""
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.setWindowTitle("Scan progress: today vs QML vs QWidget")
|
||||
client = mock.MagicMock()
|
||||
grid = QGridLayout(self)
|
||||
grid.setHorizontalSpacing(18)
|
||||
self.bars = []
|
||||
for col, (name, cls) in enumerate(
|
||||
[
|
||||
("Today", ScanProgressBar),
|
||||
("QML", ScanProgressBarQml),
|
||||
("QWidget", ScanProgressBarModern),
|
||||
]
|
||||
):
|
||||
grid.addWidget(QLabel(f"<b>{name}</b>"), 0, col)
|
||||
for row, compact in enumerate((False, True)):
|
||||
bar = cls(client=client, one_line_design=compact, rpc_exposed=False)
|
||||
bar.setMinimumWidth(320)
|
||||
grid.addWidget(bar, row + 1, col)
|
||||
self.bars.append(bar)
|
||||
restart = QPushButton("Restart scan")
|
||||
restart.clicked.connect(self.start)
|
||||
grid.addWidget(restart, 3, 0)
|
||||
self.timer = QTimer(self)
|
||||
self.timer.timeout.connect(self.step)
|
||||
self.scan = 0
|
||||
self.start()
|
||||
|
||||
def start(self):
|
||||
self.scan += 1
|
||||
self.value = 0
|
||||
self.index = 0
|
||||
self.scan_id = f"scan-{self.scan}"
|
||||
for bar in self.bars:
|
||||
model = getattr(bar, "model", None)
|
||||
if model is not None:
|
||||
model._on_scan_status(
|
||||
{
|
||||
"scan_id": self.scan_id,
|
||||
"status": "open",
|
||||
"scan_name": "line_scan",
|
||||
"scan_number": self.scan,
|
||||
},
|
||||
{},
|
||||
)
|
||||
self.timer.start(250)
|
||||
|
||||
def step(self):
|
||||
if self.index >= len(SCRIPT):
|
||||
self.timer.stop()
|
||||
return
|
||||
status, advance, final = SCRIPT[self.index]
|
||||
self.index += 1
|
||||
self.value = min(POINTS, self.value + advance)
|
||||
max_value = 0 if self.index <= 3 else POINTS
|
||||
done = self.value >= POINTS
|
||||
for bar in self.bars:
|
||||
if final and getattr(bar, "model", None) is not None:
|
||||
bar.model._on_scan_status({"scan_id": self.scan_id, "status": final}, {})
|
||||
continue
|
||||
bar.progress_tracker.process_progress_message(
|
||||
{"value": self.value, "max_value": max_value, "done": done},
|
||||
{
|
||||
"scan_id": self.scan_id,
|
||||
"RID": self.scan_id,
|
||||
"status": final or status,
|
||||
"scan_number": self.scan,
|
||||
},
|
||||
)
|
||||
if final:
|
||||
self.timer.stop()
|
||||
QTimer.singleShot(2000, self.start)
|
||||
elif done:
|
||||
self.timer.stop()
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
app = QApplication(sys.argv)
|
||||
apply_theme("light" if "--light" in sys.argv else "dark")
|
||||
demo = Demo()
|
||||
demo.show()
|
||||
sys.exit(app.exec())
|
||||
@@ -0,0 +1,196 @@
|
||||
import QtQuick
|
||||
import QtQuick.Layouts
|
||||
|
||||
// Scan progress view. Data comes from `progressModel` (ScanProgressModel), colours from `theme`.
|
||||
Item {
|
||||
id: root
|
||||
|
||||
property bool compact: false
|
||||
property bool showSource: true
|
||||
property bool showElapsed: true
|
||||
property bool showRemaining: true
|
||||
|
||||
readonly property color fg: theme.c["FG"] || "#e0e0e0"
|
||||
readonly property color muted: theme.c["DISABLED_FG"] || "#888888"
|
||||
readonly property color track: theme.dark ? Qt.rgba(1, 1, 1, 0.08) : Qt.rgba(0, 0, 0, 0.07)
|
||||
readonly property color stateColor: theme.c[progressModel.stateColorKey] || "#0a60ff"
|
||||
// chip text needs more contrast than the bar fill on light backgrounds
|
||||
readonly property color stateInk: theme.dark ? stateColor : Qt.darker(stateColor, 1.5)
|
||||
|
||||
implicitHeight: compact ? 22 : 58
|
||||
implicitWidth: compact ? 220 : 320
|
||||
|
||||
|
||||
// ---------- progress track, shared by both layouts ----------
|
||||
component Bar: Rectangle {
|
||||
id: bar
|
||||
radius: height / 2
|
||||
color: root.track
|
||||
clip: true
|
||||
|
||||
Rectangle {
|
||||
id: fill
|
||||
visible: !progressModel.indeterminate
|
||||
height: parent.height
|
||||
radius: parent.radius
|
||||
width: Math.max(progressModel.fraction > 0 ? height : 0, parent.width * progressModel.fraction)
|
||||
color: root.stateColor
|
||||
Behavior on width { NumberAnimation { duration: 250; easing.type: Easing.OutCubic } }
|
||||
Behavior on color { ColorAnimation { duration: 200 } }
|
||||
|
||||
// soft sheen that travels along the fill while the scan is running
|
||||
Rectangle {
|
||||
visible: progressModel.state === "running" && fill.width > 24
|
||||
width: 40; height: parent.height; radius: parent.radius
|
||||
gradient: Gradient {
|
||||
orientation: Gradient.Horizontal
|
||||
GradientStop { position: 0.0; color: "transparent" }
|
||||
GradientStop { position: 0.5; color: Qt.rgba(1, 1, 1, 0.28) }
|
||||
GradientStop { position: 1.0; color: "transparent" }
|
||||
}
|
||||
NumberAnimation on x {
|
||||
from: -40; to: fill.width; duration: 1600; loops: Animation.Infinite
|
||||
running: progressModel.state === "running"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// indeterminate: a segment sliding back and forth until the point count is known
|
||||
Rectangle {
|
||||
visible: progressModel.indeterminate
|
||||
width: parent.width * 0.3; height: parent.height; radius: parent.radius
|
||||
color: root.stateColor
|
||||
SequentialAnimation on x {
|
||||
running: progressModel.indeterminate; loops: Animation.Infinite
|
||||
NumberAnimation { from: 0; to: bar.width * 0.7; duration: 900; easing.type: Easing.InOutQuad }
|
||||
NumberAnimation { from: bar.width * 0.7; to: 0; duration: 900; easing.type: Easing.InOutQuad }
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ---------- compact: one line for status bars ----------
|
||||
RowLayout {
|
||||
visible: root.compact
|
||||
anchors.fill: parent
|
||||
anchors.leftMargin: 4; anchors.rightMargin: 4
|
||||
spacing: 6
|
||||
|
||||
Rectangle {
|
||||
Layout.alignment: Qt.AlignVCenter
|
||||
width: 8; height: 8; radius: 4
|
||||
color: root.stateColor
|
||||
}
|
||||
Text {
|
||||
visible: root.showSource && progressModel.title !== ""
|
||||
text: progressModel.title
|
||||
color: root.fg
|
||||
font.pixelSize: 11
|
||||
elide: Text.ElideRight
|
||||
Layout.maximumWidth: 120
|
||||
}
|
||||
Bar { Layout.fillWidth: true; Layout.preferredHeight: 6; Layout.alignment: Qt.AlignVCenter }
|
||||
Text {
|
||||
text: progressModel.percentText
|
||||
color: root.fg
|
||||
font.pixelSize: 11
|
||||
font.features: { "tnum": 1 }
|
||||
}
|
||||
Text {
|
||||
visible: root.showRemaining && text !== ""
|
||||
text: progressModel.timeText
|
||||
color: root.muted
|
||||
font.pixelSize: 11
|
||||
}
|
||||
}
|
||||
|
||||
// ---------- full layout ----------
|
||||
ColumnLayout {
|
||||
visible: !root.compact
|
||||
anchors.fill: parent
|
||||
anchors.margins: 4
|
||||
spacing: 4
|
||||
|
||||
RowLayout {
|
||||
Layout.fillWidth: true
|
||||
spacing: 8
|
||||
|
||||
// state chip
|
||||
Rectangle {
|
||||
id: chip
|
||||
radius: height / 2
|
||||
implicitHeight: 20
|
||||
implicitWidth: chipRow.implicitWidth + 14
|
||||
color: Qt.rgba(root.stateColor.r, root.stateColor.g, root.stateColor.b, 0.16)
|
||||
Behavior on color { ColorAnimation { duration: 200 } }
|
||||
Row {
|
||||
id: chipRow
|
||||
anchors.centerIn: parent
|
||||
spacing: 4
|
||||
Image {
|
||||
anchors.verticalCenter: parent.verticalCenter
|
||||
width: 14; height: 14
|
||||
sourceSize: Qt.size(28, 28)
|
||||
source: "image://material/" + progressModel.stateIcon + "?filled=1&color=" + encodeURIComponent(root.stateInk)
|
||||
}
|
||||
Text {
|
||||
anchors.verticalCenter: parent.verticalCenter
|
||||
text: progressModel.stateLabel
|
||||
color: root.stateInk
|
||||
font.pixelSize: 11
|
||||
font.weight: Font.DemiBold
|
||||
}
|
||||
}
|
||||
}
|
||||
Text {
|
||||
visible: root.showSource
|
||||
Layout.fillWidth: true
|
||||
text: progressModel.title
|
||||
color: progressModel.state === "idle" ? root.muted : root.fg
|
||||
font.pixelSize: 12
|
||||
elide: Text.ElideRight
|
||||
}
|
||||
Item { visible: !root.showSource; Layout.fillWidth: true }
|
||||
Text {
|
||||
text: progressModel.percentText
|
||||
color: root.fg
|
||||
font.pixelSize: 15
|
||||
font.weight: Font.DemiBold
|
||||
font.features: { "tnum": 1 }
|
||||
}
|
||||
}
|
||||
|
||||
Bar { Layout.fillWidth: true; Layout.preferredHeight: 8 }
|
||||
|
||||
RowLayout {
|
||||
Layout.fillWidth: true
|
||||
spacing: 6
|
||||
Text {
|
||||
text: progressModel.countText !== "" ? progressModel.countText + " points" : ""
|
||||
color: root.muted
|
||||
font.pixelSize: 11
|
||||
font.features: { "tnum": 1 }
|
||||
}
|
||||
Item { Layout.fillWidth: true }
|
||||
Text {
|
||||
visible: root.showElapsed && progressModel.active && progressModel.elapsedText !== ""
|
||||
text: progressModel.elapsedText + " elapsed"
|
||||
color: root.muted
|
||||
font.pixelSize: 11
|
||||
font.features: { "tnum": 1 }
|
||||
}
|
||||
Text {
|
||||
visible: root.showElapsed && root.showRemaining && progressModel.active && progressModel.elapsedText !== ""
|
||||
text: "·"
|
||||
color: root.muted
|
||||
font.pixelSize: 11
|
||||
}
|
||||
Text {
|
||||
visible: root.showRemaining && text !== ""
|
||||
text: progressModel.timeText
|
||||
color: progressModel.state === "running" ? root.fg : root.muted
|
||||
font.pixelSize: 11
|
||||
font.features: { "tnum": 1 }
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,103 @@
|
||||
"""Small helpers to host a QML view inside a QWidget: theme bridge and Material icon provider."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from urllib.parse import unquote
|
||||
|
||||
from bec_qthemes import material_icon
|
||||
from qtpy.QtCore import Property, QObject, QSize, QUrl, Signal
|
||||
from qtpy.QtGui import QColor, QPixmap
|
||||
from qtpy.QtQuick import QQuickImageProvider
|
||||
from qtpy.QtQuickWidgets import QQuickWidget
|
||||
from qtpy.QtWidgets import QApplication, QWidget
|
||||
|
||||
from bec_widgets.utils.colors import get_theme_name
|
||||
from bec_widgets.widgets.progress.scan_progressbar.scan_progress_model import current_theme_colors
|
||||
|
||||
|
||||
class QmlTheme(QObject):
|
||||
"""Exposes the current bec_qthemes palette to QML as ``theme.c.<KEY>`` hex strings."""
|
||||
|
||||
changed = Signal()
|
||||
|
||||
def __init__(self, parent: QObject | None = None):
|
||||
super().__init__(parent=parent)
|
||||
self._colors: dict[str, str] = {}
|
||||
self._dark = True
|
||||
self.refresh()
|
||||
app = QApplication.instance()
|
||||
if hasattr(app, "theme"):
|
||||
app.theme.theme_changed.connect(self.refresh)
|
||||
|
||||
def refresh(self, *_args):
|
||||
"""Re-read the palette from the application theme."""
|
||||
self._dark = get_theme_name() == "dark"
|
||||
self._colors = {k: QColor(v).name() for k, v in current_theme_colors().items()}
|
||||
self.changed.emit()
|
||||
|
||||
@Property("QVariantMap", notify=changed)
|
||||
def c(self) -> dict:
|
||||
return self._colors
|
||||
|
||||
@Property(bool, notify=changed)
|
||||
def dark(self) -> bool:
|
||||
return self._dark
|
||||
|
||||
|
||||
class MaterialIconProvider(QQuickImageProvider):
|
||||
"""Serves ``image://material/<name>?color=<hex>&filled=1`` from bec_qthemes."""
|
||||
|
||||
def __init__(self):
|
||||
super().__init__(QQuickImageProvider.ImageType.Pixmap)
|
||||
|
||||
def requestPixmap(self, icon_id: str, size, requested_size: QSize) -> QPixmap:
|
||||
name, _, query = icon_id.partition("?")
|
||||
if not name or name == "undefined":
|
||||
return QPixmap()
|
||||
params = {k: unquote(v) for k, v in (p.split("=", 1) for p in query.split("&") if "=" in p)}
|
||||
edge = max(requested_size.width(), requested_size.height(), 0) or 24
|
||||
pixmap = material_icon(
|
||||
name,
|
||||
size=(edge, edge),
|
||||
color=params.get("color", "#888888"),
|
||||
filled=params.get("filled") == "1",
|
||||
)
|
||||
if size is not None:
|
||||
size.setWidth(pixmap.width())
|
||||
size.setHeight(pixmap.height())
|
||||
return pixmap
|
||||
|
||||
|
||||
def create_quick_widget(
|
||||
parent: QWidget, qml_file: Path, context: dict[str, QObject]
|
||||
) -> QQuickWidget:
|
||||
"""
|
||||
Create a QQuickWidget that loads ``qml_file`` with the given context properties.
|
||||
|
||||
The theme bridge is always available as ``theme`` and icons under ``image://material/``.
|
||||
|
||||
Args:
|
||||
parent(QWidget): Parent widget.
|
||||
qml_file(Path): QML file to load.
|
||||
context(dict[str, QObject]): Extra context properties.
|
||||
|
||||
Returns:
|
||||
QQuickWidget: The configured widget.
|
||||
"""
|
||||
view = QQuickWidget(parent)
|
||||
view.setResizeMode(QQuickWidget.ResizeMode.SizeRootObjectToView)
|
||||
view.setClearColor(QColor(0, 0, 0, 0))
|
||||
engine = view.engine()
|
||||
# keep Python references, otherwise the wrappers are collected while QML still uses them
|
||||
view._icon_provider = MaterialIconProvider()
|
||||
engine.addImageProvider("material", view._icon_provider)
|
||||
theme = QmlTheme(view)
|
||||
view._qml_theme = theme
|
||||
view.rootContext().setContextProperty("theme", theme)
|
||||
for name, obj in context.items():
|
||||
view.rootContext().setContextProperty(name, obj)
|
||||
view.setSource(QUrl.fromLocalFile(str(qml_file)))
|
||||
for error in view.errors():
|
||||
raise RuntimeError(f"QML error in {qml_file.name}: {error.toString()}")
|
||||
return view
|
||||
@@ -0,0 +1,316 @@
|
||||
"""Presentation model shared by the QML and QWidget versions of the modern scan progress bar."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from bec_lib.endpoints import MessageEndpoints
|
||||
from bec_qthemes._theme import ACCENT_COLORS
|
||||
from qtpy.QtCore import Property, QObject, QTimer, Signal
|
||||
from qtpy.QtGui import QColor, QPalette
|
||||
from qtpy.QtWidgets import QApplication
|
||||
|
||||
from bec_widgets.utils.error_popups import SafeSlot
|
||||
from bec_widgets.widgets.progress.progress_backend import BECProgressTracker, ProgressSnapshot
|
||||
|
||||
# BEC scan status -> display state
|
||||
STATUS_TO_STATE = {
|
||||
"open": "running",
|
||||
"paused": "paused",
|
||||
"aborted": "aborted",
|
||||
"halted": "halted",
|
||||
"closed": "done",
|
||||
"user_completed": "done",
|
||||
}
|
||||
|
||||
STATE_LABELS = {
|
||||
"idle": "Idle",
|
||||
"running": "Running",
|
||||
"paused": "Paused",
|
||||
"aborted": "Aborted",
|
||||
"halted": "Halted",
|
||||
"done": "Done",
|
||||
}
|
||||
|
||||
# Theme color key used for each state (bec_qthemes palette keys)
|
||||
STATE_COLOR_KEYS = {
|
||||
"idle": "DISABLED_FG",
|
||||
"running": "ACCENT_DEFAULT",
|
||||
"paused": "ACCENT_WARNING",
|
||||
"aborted": "ACCENT_HIGHLIGHT",
|
||||
"halted": "ACCENT_EMERGENCY",
|
||||
"done": "ACCENT_SUCCESS",
|
||||
}
|
||||
|
||||
STATE_ICONS = {
|
||||
"idle": "radio_button_unchecked",
|
||||
"running": "play_circle",
|
||||
"paused": "pause_circle",
|
||||
"aborted": "cancel",
|
||||
"halted": "report",
|
||||
"done": "check_circle",
|
||||
}
|
||||
|
||||
|
||||
def current_theme_colors() -> dict[str, QColor]:
|
||||
"""
|
||||
Colours of the active bec_qthemes theme, or a palette-based fallback when no theme is applied.
|
||||
|
||||
Returns:
|
||||
dict[str, QColor]: Theme colours keyed by bec_qthemes palette key.
|
||||
"""
|
||||
app = QApplication.instance()
|
||||
if hasattr(app, "theme"):
|
||||
return dict(app.theme._colors)
|
||||
palette = app.palette()
|
||||
colors = {k: QColor(v) for k, v in ACCENT_COLORS["dark"].items()}
|
||||
colors["FG"] = palette.color(QPalette.ColorRole.WindowText)
|
||||
colors["DISABLED_FG"] = palette.color(QPalette.ColorRole.PlaceholderText)
|
||||
return colors
|
||||
|
||||
|
||||
def format_duration(seconds: float | None) -> str:
|
||||
"""
|
||||
Format a duration compactly: ``m:ss`` below one hour, ``h:mm:ss`` above.
|
||||
|
||||
Args:
|
||||
seconds(float | None): Duration in seconds.
|
||||
|
||||
Returns:
|
||||
str: The formatted duration, or an em dash if unknown.
|
||||
"""
|
||||
if seconds is None:
|
||||
return "—"
|
||||
seconds = max(0, int(round(seconds)))
|
||||
hours, rest = divmod(seconds, 3600)
|
||||
minutes, secs = divmod(rest, 60)
|
||||
if hours:
|
||||
return f"{hours}:{minutes:02}:{secs:02}"
|
||||
return f"{minutes}:{secs:02}"
|
||||
|
||||
|
||||
def format_count(value: float) -> str:
|
||||
"""Format a point count without a trailing ``.0``."""
|
||||
if float(value).is_integer():
|
||||
return f"{int(value)}"
|
||||
return f"{value:.1f}"
|
||||
|
||||
|
||||
class ScanProgressModel(QObject):
|
||||
"""
|
||||
View-agnostic state of the current scan's progress.
|
||||
|
||||
Wraps ``BECProgressTracker`` and the scan status endpoint, and exposes ready-to-display
|
||||
properties. Both the QML view and the QWidget view render from this object only.
|
||||
"""
|
||||
|
||||
changed = Signal()
|
||||
progress_started = Signal()
|
||||
progress_finished = Signal()
|
||||
|
||||
def __init__(self, bec_dispatcher, parent: QObject | None = None):
|
||||
super().__init__(parent=parent)
|
||||
self._dispatcher = bec_dispatcher
|
||||
self._state = "idle"
|
||||
self._value = 0.0
|
||||
self._maximum = 0.0
|
||||
self._scan_number: int | None = None
|
||||
self._scan_name = ""
|
||||
self._reason = ""
|
||||
self._scan_id: str | None = None
|
||||
self._elapsed: float | None = None
|
||||
self._remaining: float | None = None
|
||||
|
||||
self.tracker = BECProgressTracker(bec_dispatcher, parent=self)
|
||||
self.tracker.progress_started.connect(self._on_started)
|
||||
self.tracker.progress_updated.connect(self._on_snapshot)
|
||||
self.tracker.progress_finished.connect(self._on_finished)
|
||||
self.tracker.start()
|
||||
self._dispatcher.connect_slot(self._on_scan_status, MessageEndpoints.scan_status())
|
||||
|
||||
# Elapsed/remaining tick once per second while a scan runs
|
||||
self._tick_timer = QTimer(self)
|
||||
self._tick_timer.setInterval(1000)
|
||||
self._tick_timer.timeout.connect(self._tick)
|
||||
|
||||
################################################################################
|
||||
# Inputs
|
||||
|
||||
@SafeSlot(object)
|
||||
def _on_started(self, _snapshot: ProgressSnapshot):
|
||||
self._tick_timer.start()
|
||||
self.progress_started.emit()
|
||||
|
||||
@SafeSlot(object)
|
||||
def _on_snapshot(self, snapshot: ProgressSnapshot):
|
||||
if snapshot.is_new_scan:
|
||||
self._reason = ""
|
||||
self._elapsed = 0.0
|
||||
self._remaining = None
|
||||
if snapshot.scan_id != self._scan_id:
|
||||
self._scan_name = ""
|
||||
self._scan_id = snapshot.scan_id or self._scan_id
|
||||
self._scan_number = snapshot.scan_number
|
||||
self._value = float(snapshot.value)
|
||||
self._maximum = float(snapshot.max_value)
|
||||
state = STATUS_TO_STATE.get(str(snapshot.status).lower(), "running")
|
||||
if snapshot.done and state == "running":
|
||||
state = "done"
|
||||
# A final scan status (aborted/halted) wins over a late "open" progress message
|
||||
if snapshot.is_new_scan or self._state not in ("aborted", "halted"):
|
||||
self._state = state
|
||||
self._read_task_times()
|
||||
self.changed.emit()
|
||||
|
||||
@SafeSlot(object)
|
||||
def _on_finished(self, _snapshot: ProgressSnapshot):
|
||||
self._tick_timer.stop()
|
||||
if self._state in ("running", "paused"):
|
||||
self._state = "done"
|
||||
self._remaining = 0.0
|
||||
self.changed.emit()
|
||||
self.progress_finished.emit()
|
||||
|
||||
@SafeSlot(dict, dict)
|
||||
def _on_scan_status(self, msg_content: dict, metadata: dict):
|
||||
scan_id = msg_content.get("scan_id")
|
||||
status = msg_content.get("status")
|
||||
if status == "open":
|
||||
if scan_id != self._scan_id:
|
||||
self._scan_id = scan_id
|
||||
self._reason = ""
|
||||
self._scan_name = msg_content.get("scan_name") or self._scan_name
|
||||
if msg_content.get("scan_number") is not None:
|
||||
self._scan_number = msg_content.get("scan_number")
|
||||
self.changed.emit()
|
||||
return
|
||||
if scan_id is None or scan_id != self._scan_id:
|
||||
return
|
||||
new_state = STATUS_TO_STATE.get(status)
|
||||
if new_state is None:
|
||||
return
|
||||
self._state = new_state
|
||||
self._reason = msg_content.get("reason") or ""
|
||||
if new_state in ("aborted", "halted", "done"):
|
||||
self._tick_timer.stop()
|
||||
self.changed.emit()
|
||||
|
||||
@SafeSlot()
|
||||
def _tick(self):
|
||||
self._read_task_times()
|
||||
self.changed.emit()
|
||||
|
||||
def _read_task_times(self):
|
||||
task = self.tracker.task
|
||||
if task is None:
|
||||
return
|
||||
task.update_elapsed_time()
|
||||
self._elapsed = task._elapsed_time
|
||||
if self._state == "paused" or not task.speed or not task.remaining:
|
||||
self._remaining = None if self._state != "done" else 0.0
|
||||
else:
|
||||
self._remaining = task.remaining / task.speed
|
||||
|
||||
################################################################################
|
||||
# Outputs
|
||||
|
||||
@Property(str, notify=changed)
|
||||
def state(self) -> str:
|
||||
"""Display state: idle, running, paused, aborted, halted or done."""
|
||||
return self._state
|
||||
|
||||
@Property(str, notify=changed)
|
||||
def stateLabel(self) -> str:
|
||||
label = STATE_LABELS[self._state]
|
||||
if self._reason == "alarm" and self._state in ("aborted", "halted"):
|
||||
label = f"{label} by alarm"
|
||||
return label
|
||||
|
||||
@Property(str, notify=changed)
|
||||
def stateColorKey(self) -> str:
|
||||
return STATE_COLOR_KEYS[self._state]
|
||||
|
||||
@Property(str, notify=changed)
|
||||
def stateIcon(self) -> str:
|
||||
return STATE_ICONS[self._state]
|
||||
|
||||
@Property(bool, notify=changed)
|
||||
def active(self) -> bool:
|
||||
return self._state in ("running", "paused")
|
||||
|
||||
@Property(str, notify=changed)
|
||||
def title(self) -> str:
|
||||
"""Scan identity, e.g. ``Scan 42 · line_scan``."""
|
||||
if self._state == "idle" and self._scan_number is None:
|
||||
return "No scan yet"
|
||||
parts = [f"Scan {self._scan_number}" if self._scan_number is not None else "Scan"]
|
||||
if self._scan_name:
|
||||
parts.append(self._scan_name)
|
||||
return " · ".join(parts)
|
||||
|
||||
@Property(float, notify=changed)
|
||||
def value(self) -> float:
|
||||
return self._value
|
||||
|
||||
@Property(float, notify=changed)
|
||||
def maximum(self) -> float:
|
||||
return self._maximum
|
||||
|
||||
@Property(float, notify=changed)
|
||||
def fraction(self) -> float:
|
||||
"""Progress between 0 and 1."""
|
||||
if self._maximum <= 0:
|
||||
return 1.0 if self._state == "done" else 0.0
|
||||
return min(1.0, max(0.0, self._value / self._maximum))
|
||||
|
||||
@Property(bool, notify=changed)
|
||||
def indeterminate(self) -> bool:
|
||||
"""True while a scan runs but the number of points is not known yet."""
|
||||
return self._state == "running" and self._maximum <= 0
|
||||
|
||||
@Property(str, notify=changed)
|
||||
def percentText(self) -> str:
|
||||
if self._state == "idle" or self.indeterminate:
|
||||
return ""
|
||||
return f"{int(self.fraction * 100)} %"
|
||||
|
||||
@Property(str, notify=changed)
|
||||
def countText(self) -> str:
|
||||
if self._state == "idle" or self._maximum <= 0:
|
||||
return ""
|
||||
return f"{format_count(self._value)} / {format_count(self._maximum)}"
|
||||
|
||||
@Property(str, notify=changed)
|
||||
def elapsedText(self) -> str:
|
||||
if self._state == "idle" or self._elapsed is None:
|
||||
return ""
|
||||
return format_duration(self._elapsed)
|
||||
|
||||
@Property(str, notify=changed)
|
||||
def timeText(self) -> str:
|
||||
"""The single time hint shown at the right of the bar."""
|
||||
if self._state == "idle":
|
||||
return ""
|
||||
if self._state == "running":
|
||||
if self._remaining is None:
|
||||
return "estimating…"
|
||||
return f"{format_duration(self._remaining)} left"
|
||||
if self._state == "paused":
|
||||
return f"paused at {format_duration(self._elapsed)}"
|
||||
return f"took {format_duration(self._elapsed)}"
|
||||
|
||||
@Property(str, notify=changed)
|
||||
def toolTip(self) -> str:
|
||||
if self._state == "idle":
|
||||
return "No scan has reported progress yet."
|
||||
lines = [f"{self.title} — {self.stateLabel}"]
|
||||
if self.countText:
|
||||
lines.append(f"Points: {self.countText} ({self.percentText})")
|
||||
lines.append(f"Elapsed: {format_duration(self._elapsed)}")
|
||||
if self._state == "running":
|
||||
lines.append(f"Remaining: {format_duration(self._remaining)}")
|
||||
return "\n".join(lines)
|
||||
|
||||
def cleanup(self):
|
||||
"""Disconnect from BEC and stop timers."""
|
||||
self._tick_timer.stop()
|
||||
self.tracker.cleanup()
|
||||
self._dispatcher.disconnect_slot(self._on_scan_status, MessageEndpoints.scan_status())
|
||||
@@ -0,0 +1,374 @@
|
||||
"""Scan progress bar with the modern UX, built from plain QWidgets on top of ScanProgressModel."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from bec_qthemes import material_icon
|
||||
from qtpy.QtCore import Property, QEasingCurve, QPropertyAnimation, QRectF, Qt, QTimer, Signal
|
||||
from qtpy.QtGui import QColor, QFont, QLinearGradient, QPainter, QPainterPath
|
||||
from qtpy.QtWidgets import QHBoxLayout, QLabel, QSizePolicy, QVBoxLayout, QWidget
|
||||
|
||||
from bec_widgets.utils.bec_widget import BECWidget
|
||||
from bec_widgets.utils.colors import get_theme_name
|
||||
from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
|
||||
from bec_widgets.widgets.progress.scan_progressbar.scan_progress_model import (
|
||||
ScanProgressModel,
|
||||
current_theme_colors,
|
||||
)
|
||||
|
||||
|
||||
def _theme_color(key: str, fallback: str = "#888888") -> QColor:
|
||||
return current_theme_colors().get(key, QColor(fallback))
|
||||
|
||||
|
||||
class _ProgressTrack(QWidget):
|
||||
"""Rounded track with an animated fill, a running sheen and an indeterminate mode."""
|
||||
|
||||
def __init__(self, parent=None, height: int = 8):
|
||||
super().__init__(parent)
|
||||
self.setFixedHeight(height)
|
||||
self.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed)
|
||||
self._fraction = 0.0
|
||||
self._color = QColor("#0a60ff")
|
||||
self._running = False
|
||||
self._indeterminate = False
|
||||
self._phase = 0.0
|
||||
|
||||
self._anim = QPropertyAnimation(self, b"fraction", self)
|
||||
self._anim.setDuration(250)
|
||||
self._anim.setEasingCurve(QEasingCurve.Type.OutCubic)
|
||||
|
||||
# one timer drives the sheen and the indeterminate segment (~30 fps, only when needed)
|
||||
self._phase_timer = QTimer(self)
|
||||
self._phase_timer.setInterval(33)
|
||||
self._phase_timer.timeout.connect(self._advance)
|
||||
|
||||
def get_fraction(self) -> float:
|
||||
return self._fraction
|
||||
|
||||
def set_fraction(self, value: float):
|
||||
self._fraction = value
|
||||
self.update()
|
||||
|
||||
fraction = Property(float, get_fraction, set_fraction)
|
||||
|
||||
def set_state(self, fraction: float, color: QColor, running: bool, indeterminate: bool):
|
||||
"""Update target fraction, colour and animation mode."""
|
||||
self._color = color
|
||||
self._running = running
|
||||
self._indeterminate = indeterminate
|
||||
if abs(fraction - self._fraction) > 1e-4:
|
||||
self._anim.stop()
|
||||
self._anim.setStartValue(self._fraction)
|
||||
self._anim.setEndValue(fraction)
|
||||
self._anim.start()
|
||||
animate = (running or indeterminate) and self.isVisible()
|
||||
if animate and not self._phase_timer.isActive():
|
||||
self._phase_timer.start()
|
||||
elif not animate:
|
||||
self._phase_timer.stop()
|
||||
self.update()
|
||||
|
||||
def _advance(self):
|
||||
self._phase = (self._phase + 0.033 / 1.6) % 1.0
|
||||
self.update()
|
||||
|
||||
def hideEvent(self, event):
|
||||
self._phase_timer.stop()
|
||||
super().hideEvent(event)
|
||||
|
||||
def paintEvent(self, _event):
|
||||
painter = QPainter(self)
|
||||
painter.setRenderHint(QPainter.RenderHint.Antialiasing)
|
||||
rect = QRectF(self.rect())
|
||||
radius = rect.height() / 2
|
||||
track = QColor(255, 255, 255, 20) if get_theme_name() == "dark" else QColor(0, 0, 0, 18)
|
||||
clip = QPainterPath()
|
||||
clip.addRoundedRect(rect, radius, radius)
|
||||
painter.setClipPath(clip)
|
||||
painter.fillPath(clip, track)
|
||||
painter.setPen(Qt.PenStyle.NoPen)
|
||||
painter.setBrush(self._color)
|
||||
|
||||
if self._indeterminate:
|
||||
# triangle wave 0..1..0 for the sliding segment
|
||||
t = self._phase * 2 if self._phase < 0.5 else 2 - self._phase * 2
|
||||
seg = rect.width() * 0.3
|
||||
painter.drawRoundedRect(
|
||||
QRectF(t * (rect.width() - seg), 0, seg, rect.height()), radius, radius
|
||||
)
|
||||
return
|
||||
|
||||
if self._fraction <= 0:
|
||||
return
|
||||
width = max(rect.height(), rect.width() * self._fraction)
|
||||
fill = QRectF(0, 0, width, rect.height())
|
||||
painter.drawRoundedRect(fill, radius, radius)
|
||||
|
||||
if self._running and width > 24:
|
||||
x = -40 + self._phase * (width + 40)
|
||||
grad = QLinearGradient(x, 0, x + 40, 0)
|
||||
grad.setColorAt(0.0, QColor(255, 255, 255, 0))
|
||||
grad.setColorAt(0.5, QColor(255, 255, 255, 72))
|
||||
grad.setColorAt(1.0, QColor(255, 255, 255, 0))
|
||||
sheen = QPainterPath()
|
||||
sheen.addRoundedRect(fill, radius, radius)
|
||||
painter.fillPath(sheen, grad)
|
||||
|
||||
|
||||
class _StateChip(QWidget):
|
||||
"""Pill with a filled Material icon and the state label, tinted with the state colour."""
|
||||
|
||||
def __init__(self, parent=None):
|
||||
super().__init__(parent)
|
||||
self._color = QColor("#888888")
|
||||
self._icon = QLabel(self)
|
||||
self._label = QLabel(self)
|
||||
font = self._label.font()
|
||||
font.setPixelSize(11)
|
||||
font.setWeight(QFont.Weight.DemiBold)
|
||||
self._label.setFont(font)
|
||||
layout = QHBoxLayout(self)
|
||||
layout.setContentsMargins(7, 2, 8, 2)
|
||||
layout.setSpacing(4)
|
||||
layout.addWidget(self._icon)
|
||||
layout.addWidget(self._label)
|
||||
self.setFixedHeight(20)
|
||||
self._icon_key = None
|
||||
|
||||
def set_state(self, label: str, icon: str, color: QColor):
|
||||
self._color = color
|
||||
# chip text needs more contrast than the bar fill on light backgrounds
|
||||
ink = color if get_theme_name() == "dark" else color.darker(150)
|
||||
self._label.setText(label)
|
||||
_set_text_color(self._label, ink.name())
|
||||
key = (icon, ink.name())
|
||||
if key != self._icon_key:
|
||||
self._icon.setPixmap(material_icon(icon, size=(14, 14), color=ink, filled=True))
|
||||
self._icon_key = key
|
||||
self.update()
|
||||
|
||||
def paintEvent(self, _event):
|
||||
painter = QPainter(self)
|
||||
painter.setRenderHint(QPainter.RenderHint.Antialiasing)
|
||||
bg = QColor(self._color)
|
||||
bg.setAlphaF(0.16)
|
||||
painter.setPen(Qt.PenStyle.NoPen)
|
||||
painter.setBrush(bg)
|
||||
r = QRectF(self.rect())
|
||||
painter.drawRoundedRect(r, r.height() / 2, r.height() / 2)
|
||||
|
||||
|
||||
class _Dot(QWidget):
|
||||
def __init__(self, parent=None):
|
||||
super().__init__(parent)
|
||||
self.setFixedSize(8, 8)
|
||||
self.color = QColor("#888888")
|
||||
|
||||
def paintEvent(self, _event):
|
||||
painter = QPainter(self)
|
||||
painter.setRenderHint(QPainter.RenderHint.Antialiasing)
|
||||
painter.setPen(Qt.PenStyle.NoPen)
|
||||
painter.setBrush(self.color)
|
||||
painter.drawEllipse(self.rect())
|
||||
|
||||
|
||||
def _set_text_color(label: QLabel, color: str):
|
||||
"""Set the label colour, skipping the costly stylesheet update when it is unchanged."""
|
||||
if label.property("_text_color") != color:
|
||||
label.setProperty("_text_color", color)
|
||||
label.setStyleSheet(f"color: {color};")
|
||||
|
||||
|
||||
def _small_label(parent, size: int = 11, bold: bool = False) -> QLabel:
|
||||
label = QLabel(parent)
|
||||
font = label.font()
|
||||
font.setPixelSize(size)
|
||||
if bold:
|
||||
font.setWeight(QFont.Weight.DemiBold)
|
||||
label.setFont(font)
|
||||
return label
|
||||
|
||||
|
||||
class ScanProgressBarModern(BECWidget, QWidget):
|
||||
"""
|
||||
Scan progress bar with a state chip, smooth bar and time estimate, built from QWidgets.
|
||||
|
||||
Same UX as ``ScanProgressBarQml`` and the same public API as ``ScanProgressBar``.
|
||||
"""
|
||||
|
||||
ICON_NAME = "timelapse"
|
||||
PLUGIN = False
|
||||
RPC = False
|
||||
progress_started = Signal()
|
||||
progress_finished = Signal()
|
||||
|
||||
def __init__(
|
||||
self, parent=None, client=None, config=None, gui_id=None, one_line_design=False, **kwargs
|
||||
):
|
||||
kwargs.pop("enable_dynamic_stylesheet", None)
|
||||
super().__init__(parent=parent, client=client, config=config, gui_id=gui_id, **kwargs)
|
||||
self.get_bec_shortcuts()
|
||||
self._compact = bool(one_line_design)
|
||||
self._show_elapsed = True
|
||||
self._show_remaining = True
|
||||
self._show_source = True
|
||||
|
||||
self.model = ScanProgressModel(self.bec_dispatcher, parent=self)
|
||||
self.model.progress_started.connect(self.progress_started)
|
||||
self.model.progress_finished.connect(self.progress_finished)
|
||||
self.model.changed.connect(self._render)
|
||||
|
||||
if self._compact:
|
||||
self._build_compact()
|
||||
else:
|
||||
self._build_full()
|
||||
self.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed)
|
||||
self._render()
|
||||
|
||||
@property
|
||||
def progress_tracker(self):
|
||||
"""The underlying BECProgressTracker."""
|
||||
return self.model.tracker
|
||||
|
||||
def _build_full(self):
|
||||
layout = QVBoxLayout(self)
|
||||
layout.setContentsMargins(4, 4, 4, 4)
|
||||
layout.setSpacing(4)
|
||||
|
||||
top = QHBoxLayout()
|
||||
top.setSpacing(8)
|
||||
self.chip = _StateChip(self)
|
||||
self.title_label = _small_label(self, 12)
|
||||
self.title_label.setSizePolicy(QSizePolicy.Policy.Ignored, QSizePolicy.Policy.Preferred)
|
||||
self.percent_label = _small_label(self, 15, bold=True)
|
||||
top.addWidget(self.chip)
|
||||
top.addWidget(self.title_label, 1)
|
||||
top.addWidget(self.percent_label)
|
||||
|
||||
self.track = _ProgressTrack(self, height=8)
|
||||
|
||||
bottom = QHBoxLayout()
|
||||
bottom.setSpacing(6)
|
||||
self.count_label = _small_label(self)
|
||||
self.elapsed_label = _small_label(self)
|
||||
self.dot_label = _small_label(self)
|
||||
self.dot_label.setText("·")
|
||||
self.time_label = _small_label(self)
|
||||
bottom.addWidget(self.count_label)
|
||||
bottom.addStretch(1)
|
||||
bottom.addWidget(self.elapsed_label)
|
||||
bottom.addWidget(self.dot_label)
|
||||
bottom.addWidget(self.time_label)
|
||||
|
||||
layout.addLayout(top)
|
||||
layout.addWidget(self.track)
|
||||
layout.addLayout(bottom)
|
||||
self.setFixedHeight(58)
|
||||
|
||||
def _build_compact(self):
|
||||
layout = QHBoxLayout(self)
|
||||
layout.setContentsMargins(4, 0, 4, 0)
|
||||
layout.setSpacing(6)
|
||||
self.dot = _Dot(self)
|
||||
self.title_label = _small_label(self)
|
||||
self.title_label.setMaximumWidth(120)
|
||||
self.track = _ProgressTrack(self, height=6)
|
||||
self.percent_label = _small_label(self)
|
||||
self.time_label = _small_label(self)
|
||||
for w in (self.dot, self.title_label, self.track, self.percent_label, self.time_label):
|
||||
layout.addWidget(w, 1 if w is self.track else 0, Qt.AlignmentFlag.AlignVCenter)
|
||||
self.setFixedHeight(22)
|
||||
|
||||
@SafeSlot()
|
||||
def _render(self):
|
||||
m = self.model
|
||||
color = _theme_color(m.stateColorKey)
|
||||
fg = _theme_color("FG", "#e0e0e0").name()
|
||||
muted = _theme_color("DISABLED_FG").name()
|
||||
self.track.set_state(m.fraction, color, m.state == "running", m.indeterminate)
|
||||
self.title_label.setText(
|
||||
self.title_label.fontMetrics().elidedText(
|
||||
m.title, Qt.TextElideMode.ElideRight, max(40, self.title_label.width())
|
||||
)
|
||||
if not self._compact
|
||||
else m.title
|
||||
)
|
||||
self.title_label.setVisible(self._show_source)
|
||||
self.percent_label.setText(m.percentText)
|
||||
self.time_label.setText(m.timeText)
|
||||
self.time_label.setVisible(self._show_remaining and bool(m.timeText))
|
||||
self.setToolTip(m.toolTip)
|
||||
|
||||
if self._compact:
|
||||
self.dot.color = color
|
||||
self.dot.update()
|
||||
for lbl in (self.title_label, self.percent_label):
|
||||
_set_text_color(lbl, fg)
|
||||
_set_text_color(self.time_label, muted)
|
||||
return
|
||||
|
||||
self.chip.set_state(m.stateLabel, m.stateIcon, color)
|
||||
_set_text_color(self.title_label, muted if m.state == "idle" else fg)
|
||||
_set_text_color(self.percent_label, fg)
|
||||
self.count_label.setText(f"{m.countText} points" if m.countText else "")
|
||||
show_elapsed = self._show_elapsed and m.active and bool(m.elapsedText)
|
||||
self.elapsed_label.setText(f"{m.elapsedText} elapsed")
|
||||
self.elapsed_label.setVisible(show_elapsed)
|
||||
self.dot_label.setVisible(show_elapsed and self.time_label.isVisibleTo(self))
|
||||
for lbl in (self.count_label, self.elapsed_label, self.dot_label):
|
||||
_set_text_color(lbl, muted)
|
||||
_set_text_color(self.time_label, fg if m.state == "running" else muted)
|
||||
|
||||
def resizeEvent(self, event):
|
||||
super().resizeEvent(event)
|
||||
if not self._compact:
|
||||
self._render()
|
||||
|
||||
def apply_theme(self, theme: str):
|
||||
self._render()
|
||||
|
||||
@SafeProperty(bool)
|
||||
def show_elapsed_time(self):
|
||||
return self._show_elapsed
|
||||
|
||||
@show_elapsed_time.setter
|
||||
def show_elapsed_time(self, value):
|
||||
self._show_elapsed = bool(value)
|
||||
self._render()
|
||||
|
||||
@SafeProperty(bool)
|
||||
def show_remaining_time(self):
|
||||
return self._show_remaining
|
||||
|
||||
@show_remaining_time.setter
|
||||
def show_remaining_time(self, value):
|
||||
self._show_remaining = bool(value)
|
||||
self._render()
|
||||
|
||||
@SafeProperty(bool)
|
||||
def show_source_label(self):
|
||||
return self._show_source
|
||||
|
||||
@show_source_label.setter
|
||||
def show_source_label(self, value):
|
||||
self._show_source = bool(value)
|
||||
self._render()
|
||||
|
||||
def update_labels(self):
|
||||
"""Kept for API compatibility; the view updates itself from the model."""
|
||||
self.model._tick()
|
||||
|
||||
def cleanup(self):
|
||||
self.model.cleanup()
|
||||
self.track._phase_timer.stop()
|
||||
self.track._anim.stop()
|
||||
super().cleanup()
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
from qtpy.QtWidgets import QApplication
|
||||
|
||||
app = QApplication([])
|
||||
widget = ScanProgressBarModern()
|
||||
widget.show()
|
||||
app.exec_()
|
||||
@@ -0,0 +1,110 @@
|
||||
"""Scan progress bar rendered with QML (QQuickWidget) on top of ScanProgressModel."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from qtpy.QtCore import Signal
|
||||
from qtpy.QtWidgets import QSizePolicy, QVBoxLayout, QWidget
|
||||
|
||||
from bec_widgets.utils.bec_widget import BECWidget
|
||||
from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
|
||||
from bec_widgets.widgets.progress.scan_progressbar.qml_support import create_quick_widget
|
||||
from bec_widgets.widgets.progress.scan_progressbar.scan_progress_model import ScanProgressModel
|
||||
|
||||
QML_FILE = Path(__file__).parent / "qml" / "ScanProgress.qml"
|
||||
|
||||
|
||||
class ScanProgressBarQml(BECWidget, QWidget):
|
||||
"""
|
||||
Scan progress bar with a state chip, smooth bar and time estimate, drawn in QML.
|
||||
|
||||
Same public API as ``ScanProgressBar``: ``progress_started``/``progress_finished`` signals,
|
||||
``one_line_design`` and the ``show_*`` properties.
|
||||
"""
|
||||
|
||||
ICON_NAME = "timelapse"
|
||||
PLUGIN = False
|
||||
RPC = False
|
||||
progress_started = Signal()
|
||||
progress_finished = Signal()
|
||||
|
||||
def __init__(
|
||||
self, parent=None, client=None, config=None, gui_id=None, one_line_design=False, **kwargs
|
||||
):
|
||||
kwargs.pop("enable_dynamic_stylesheet", None)
|
||||
super().__init__(parent=parent, client=client, config=config, gui_id=gui_id, **kwargs)
|
||||
self.get_bec_shortcuts()
|
||||
self.model = ScanProgressModel(self.bec_dispatcher, parent=self)
|
||||
self.model.progress_started.connect(self.progress_started)
|
||||
self.model.progress_finished.connect(self.progress_finished)
|
||||
self.model.changed.connect(self._update_tooltip)
|
||||
|
||||
self.view = create_quick_widget(self, QML_FILE, {"progressModel": self.model})
|
||||
# the view's root item must be destroyed before the model it binds to
|
||||
self.model.setParent(self.view)
|
||||
self.view.rootObject().setProperty("compact", bool(one_line_design))
|
||||
self.view.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed)
|
||||
self.setFixedHeight(22 if one_line_design else 58)
|
||||
|
||||
layout = QVBoxLayout(self)
|
||||
layout.setContentsMargins(0, 0, 0, 0)
|
||||
layout.addWidget(self.view)
|
||||
self.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed)
|
||||
self._update_tooltip()
|
||||
|
||||
@property
|
||||
def progress_tracker(self):
|
||||
"""The underlying BECProgressTracker."""
|
||||
return self.model.tracker
|
||||
|
||||
@SafeSlot()
|
||||
def _update_tooltip(self):
|
||||
self.view.setToolTip(self.model.toolTip)
|
||||
|
||||
def _root_flag(self, name: str) -> bool:
|
||||
return bool(self.view.rootObject().property(name))
|
||||
|
||||
@SafeProperty(bool)
|
||||
def show_elapsed_time(self):
|
||||
return self._root_flag("showElapsed")
|
||||
|
||||
@show_elapsed_time.setter
|
||||
def show_elapsed_time(self, value):
|
||||
self.view.rootObject().setProperty("showElapsed", bool(value))
|
||||
|
||||
@SafeProperty(bool)
|
||||
def show_remaining_time(self):
|
||||
return self._root_flag("showRemaining")
|
||||
|
||||
@show_remaining_time.setter
|
||||
def show_remaining_time(self, value):
|
||||
self.view.rootObject().setProperty("showRemaining", bool(value))
|
||||
|
||||
@SafeProperty(bool)
|
||||
def show_source_label(self):
|
||||
return self._root_flag("showSource")
|
||||
|
||||
@show_source_label.setter
|
||||
def show_source_label(self, value):
|
||||
self.view.rootObject().setProperty("showSource", bool(value))
|
||||
|
||||
def update_labels(self):
|
||||
"""Kept for API compatibility; the view updates itself from the model."""
|
||||
self.model._tick()
|
||||
|
||||
def cleanup(self):
|
||||
self.model.cleanup()
|
||||
self.view.setSource("") # release QML objects before the context goes away
|
||||
self.view.close()
|
||||
self.view.deleteLater()
|
||||
super().cleanup()
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
from qtpy.QtWidgets import QApplication
|
||||
|
||||
app = QApplication([])
|
||||
widget = ScanProgressBarQml()
|
||||
widget.show()
|
||||
app.exec_()
|
||||
@@ -0,0 +1,117 @@
|
||||
import pytest
|
||||
|
||||
from bec_widgets.tests.client_mocks import mocked_client
|
||||
from bec_widgets.tests.utils import create_widget
|
||||
from bec_widgets.widgets.progress.scan_progressbar.scan_progress_model import (
|
||||
format_count,
|
||||
format_duration,
|
||||
)
|
||||
from bec_widgets.widgets.progress.scan_progressbar.scan_progressbar_modern import (
|
||||
ScanProgressBarModern,
|
||||
)
|
||||
from bec_widgets.widgets.progress.scan_progressbar.scan_progressbar_qml import ScanProgressBarQml
|
||||
|
||||
|
||||
@pytest.fixture(params=[ScanProgressBarModern, ScanProgressBarQml], ids=["qwidget", "qml"])
|
||||
def widget_cls(request):
|
||||
return request.param
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def bar(qtbot, mocked_client, widget_cls):
|
||||
return create_widget(qtbot, widget_cls, client=mocked_client)
|
||||
|
||||
|
||||
def _status(bar, status, scan_id="sid", **extra):
|
||||
bar.model._on_scan_status({"scan_id": scan_id, "status": status, **extra}, {})
|
||||
|
||||
|
||||
def _progress(bar, value, max_value, done=False, status="open", scan_id="sid", scan_number=7):
|
||||
bar.progress_tracker.process_progress_message(
|
||||
{"value": value, "max_value": max_value, "done": done},
|
||||
{"scan_id": scan_id, "RID": f"rid-{scan_id}", "status": status, "scan_number": scan_number},
|
||||
)
|
||||
|
||||
|
||||
def test_format_helpers():
|
||||
assert format_duration(None) == "—"
|
||||
assert format_duration(5) == "0:05"
|
||||
assert format_duration(83) == "1:23"
|
||||
assert format_duration(3725) == "1:02:05"
|
||||
assert format_count(3.0) == "3"
|
||||
assert format_count(2.5) == "2.5"
|
||||
|
||||
|
||||
def test_idle_state(bar):
|
||||
m = bar.model
|
||||
assert m.state == "idle"
|
||||
assert m.title == "No scan yet"
|
||||
assert m.percentText == ""
|
||||
assert m.timeText == ""
|
||||
|
||||
|
||||
def test_running_scan_shows_name_count_and_percent(bar):
|
||||
_status(bar, "open", scan_name="line_scan", scan_number=7)
|
||||
_progress(bar, 3, 12)
|
||||
m = bar.model
|
||||
assert m.state == "running"
|
||||
assert m.title == "Scan 7 · line_scan"
|
||||
assert m.countText == "3 / 12"
|
||||
assert m.percentText == "25 %"
|
||||
assert m.fraction == pytest.approx(0.25)
|
||||
assert m.timeText in ("estimating…",) or m.timeText.endswith("left")
|
||||
|
||||
|
||||
def test_indeterminate_until_points_are_known(bar):
|
||||
_progress(bar, 0, 0)
|
||||
assert bar.model.indeterminate
|
||||
_progress(bar, 1, 10)
|
||||
assert not bar.model.indeterminate
|
||||
|
||||
|
||||
def test_paused_then_resumed(bar):
|
||||
_progress(bar, 2, 10)
|
||||
_progress(bar, 2, 10, status="paused")
|
||||
assert bar.model.state == "paused"
|
||||
assert bar.model.timeText.startswith("paused at")
|
||||
_progress(bar, 3, 10)
|
||||
assert bar.model.state == "running"
|
||||
|
||||
|
||||
def test_done_keeps_result_visible(bar, qtbot):
|
||||
with qtbot.waitSignal(bar.progress_finished, timeout=1000):
|
||||
_progress(bar, 10, 10, done=True)
|
||||
m = bar.model
|
||||
assert m.state == "done"
|
||||
assert m.percentText == "100 %"
|
||||
assert m.timeText.startswith("took")
|
||||
|
||||
|
||||
def test_halted_by_alarm_is_not_overwritten_by_late_progress(bar):
|
||||
_progress(bar, 4, 10)
|
||||
_status(bar, "halted", reason="alarm")
|
||||
assert bar.model.stateLabel == "Halted by alarm"
|
||||
_progress(bar, 4, 10, done=True)
|
||||
assert bar.model.state == "halted"
|
||||
|
||||
|
||||
def test_new_scan_resets_state(bar):
|
||||
_progress(bar, 4, 10)
|
||||
_status(bar, "aborted")
|
||||
_progress(bar, 1, 5, scan_id="next", scan_number=8)
|
||||
assert bar.model.state == "running"
|
||||
assert bar.model.title.startswith("Scan 8")
|
||||
|
||||
|
||||
def test_show_properties(bar):
|
||||
for name in ("show_elapsed_time", "show_remaining_time", "show_source_label"):
|
||||
assert getattr(bar, name) is True
|
||||
setattr(bar, name, False)
|
||||
assert getattr(bar, name) is False
|
||||
|
||||
|
||||
def test_one_line_design(qtbot, mocked_client, widget_cls):
|
||||
widget = create_widget(qtbot, widget_cls, client=mocked_client, one_line_design=True)
|
||||
_progress(widget, 1, 2)
|
||||
assert widget.height() == 22
|
||||
assert widget.model.percentText == "50 %"
|
||||
Reference in new issue
Block a user