bec/bec_widgets
0
0
Fork 0
mirror of https://github.com/bec-project/bec_widgets.git synced 2025-07-14 11:41:49 +02:00
Code Activity

cleanup, use dict instead of scan_history message in qt signals/slots

Browse Source
This commit is contained in:
appel_c
2025-07-13 21:25:48 +02:00
parent 27d4b939e1
commit 834a17aeb0
4 changed files with 49 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
bec_widgets/widgets/services/scan_history_browser/components/scan_history_device_viewer.py
View File

@ -86,9 +86,7 @@ class ScanHistoryDeviceViewer(BECWidget, QtWidgets.QWidget):
RPC = False
PLUGIN = False
request_history_plot = QtCore.Signal(
str, str, object
) # (device_name, signal_name, ScanHistoryMessage)
request_history_plot = QtCore.Signal(str, str, str) # (scan_id, device_name, signal_name)
def __init__(
self,
@ -147,16 +145,16 @@ class ScanHistoryDeviceViewer(BECWidget, QtWidgets.QWidget):
main_layout.addWidget(widget)
main_layout.addWidget(self.request_plotting_button)
@SafeSlot(object)
def update_devices_from_scan_history(self, msg: ScanHistoryMessage) -> None:
@SafeSlot(dict, dict)
def update_devices_from_scan_history(self, msg: dict, metadata: dict | None = None) -> None:
"""Update the device combo box with the scan history message.
Args:
msg (ScanHistoryMessage): The scan history message containing device data.
"""
if not isinstance(msg, ScanHistoryMessage):
logger.info(f"Received message of type {type(msg)} instead of ScanHistoryMessage.")
return
msg = ScanHistoryMessage(**msg)
if metadata is not None:
msg.metadata = metadata
# Keep track of current device name
self._last_device_name = self.device_combo.currentText()
@ -182,9 +180,9 @@ class ScanHistoryDeviceViewer(BECWidget, QtWidgets.QWidget):
return
signal_data = self.scan_history_msg.stored_data_info.get(device_name, None)
if signal_data is None:
logger.warning(f"No signal data found for device {device_name}.")
logger.info(f"No signal data found for device {device_name}.")
return
self.signal_model.signals = self.scan_history_msg.stored_data_info[device_name]
self.signal_model.signals = signal_data
if self._last_signal_name is not None:
# Try to restore the last selected signal
index = self.signal_combo.findData(self._last_signal_name, role=QtCore.Qt.UserRole)
@ -211,9 +209,9 @@ class ScanHistoryDeviceViewer(BECWidget, QtWidgets.QWidget):
self.device_combo.model().index(signal_index, 0), QtCore.Qt.UserRole
)
logger.info(
f"Requesting plotting for device: {device_name}, {signal_name} with {self.scan_history_msg} points."
f"Requesting plotting clicked: Scan ID:{self.scan_history_msg.scan_id}, device name: {device_name} with signal name: {signal_name}."
)
self.request_history_plot.emit(device_name, signal_name, self.scan_history_msg)
self.request_history_plot.emit(self.scan_history_msg.scan_id, device_name, signal_name)
if __name__ == "__main__": # pragma: no cover

16
bec_widgets/widgets/services/scan_history_browser/components/scan_history_metadata_viewer.py
View File

@ -1,19 +1,16 @@
from __future__ import annotations
from datetime import datetime
from typing import TYPE_CHECKING
from bec_lib.logger import bec_logger
from bec_lib.messages import ScanHistoryMessage
from bec_qthemes import material_icon
from qtpy import QtCore, QtGui, QtWidgets
from qtpy import QtGui, QtWidgets
from bec_widgets.utils.bec_widget import BECWidget, ConnectionConfig
from bec_widgets.utils.colors import get_theme_palette
from bec_widgets.utils.error_popups import SafeSlot
if TYPE_CHECKING: # pragma: no cover
from bec_lib.messages import ScanHistoryMessage
logger = bec_logger.logger
@ -63,7 +60,7 @@ class ScanHistoryMetadataViewer(BECWidget, QtWidgets.QGroupBox):
self._init_grid_layout()
self.scan_history_msg = scan_history_msg
if scan_history_msg is not None:
self.update_view(self.scan_history_msg)
self.update_view(self.scan_history_msg.content, self.scan_history_msg.metadata)
self.apply_theme()
def apply_theme(self, theme: str | None = None):
@ -113,14 +110,17 @@ class ScanHistoryMetadataViewer(BECWidget, QtWidgets.QGroupBox):
)
layout.addWidget(copy_button, row, 2)
@SafeSlot(object)
def update_view(self, msg: ScanHistoryMessage):
@SafeSlot(dict, dict)
def update_view(self, msg: dict, metadata: dict | None = None) -> None:
"""
Update the view with the given ScanHistoryMessage.
Args:
msg (ScanHistoryMessage): The message containing scan metadata.
"""
msg = ScanHistoryMessage(**msg)
if metadata is not None:
msg.metadata = metadata
if msg == self.scan_history_msg:
return
self.scan_history_msg = msg

4
bec_widgets/widgets/services/scan_history_browser/components/scan_history_view.py
View File

@ -18,7 +18,7 @@ class ScanHistoryView(BECWidget, QtWidgets.QTreeWidget):
RPC = False
PLUGIN = False
scan_selected = QtCore.Signal(object)
scan_selected = QtCore.Signal(dict, dict)
no_scan_selected = QtCore.Signal()
def __init__(
@ -97,7 +97,7 @@ class ScanHistoryView(BECWidget, QtWidgets.QTreeWidget):
if not current:
return
index = self.indexOfTopLevelItem(current)
self.scan_selected.emit(self.scan_history[index])
self.scan_selected.emit(self.scan_history[index].content, self.scan_history[index].metadata)
def _start_subscription(self):
"""Subscribe to scan history updates. Starts with oldes scan history first."""

59
tests/unit_tests/test_scan_history_browser.py
View File

@ -116,7 +116,9 @@ def test_scan_history_device_viewer_receive_msg(
assert scan_history_device_viewer.scan_history_msg is None
assert scan_history_device_viewer.signal_model.signals == []
assert scan_history_device_viewer.signal_model.rowCount() == 0
scan_history_device_viewer.update_devices_from_scan_history(scan_history_msg)
scan_history_device_viewer.update_devices_from_scan_history(
scan_history_msg.content, scan_history_msg.metadata
)
assert scan_history_device_viewer.scan_history_msg == scan_history_msg
assert scan_history_device_viewer.device_combo.currentText() == "device2"
assert scan_history_device_viewer.signal_model.signals == [
@ -133,7 +135,7 @@ def test_scan_history_device_viewer_receive_msg(
## Update of second message should not change the device if still available
new_msg = scan_history_msg_2
scan_history_device_viewer.update_devices_from_scan_history(new_msg)
scan_history_device_viewer.update_devices_from_scan_history(new_msg.content, new_msg.metadata)
assert scan_history_device_viewer.scan_history_msg == new_msg
assert scan_history_device_viewer.signal_model.signals == [
("device2_signal4", _StoredDataInfo(shape=(30,))),
@ -152,7 +154,7 @@ def test_scan_history_device_viewer_receive_msg(
def test_scan_history_device_viewer_clear_view(qtbot, scan_history_device_viewer, scan_history_msg):
"""Test clearing the device viewer."""
scan_history_device_viewer.update_devices_from_scan_history(scan_history_msg)
scan_history_device_viewer.update_devices_from_scan_history(scan_history_msg.content)
assert scan_history_device_viewer.scan_history_msg == scan_history_msg
scan_history_device_viewer.clear_view()
assert scan_history_device_viewer.scan_history_msg is None
@ -163,7 +165,7 @@ def test_scan_history_device_viewer_on_request_plotting_clicked(
qtbot, scan_history_device_viewer, scan_history_msg
):
"""Test the request plotting button click."""
scan_history_device_viewer.update_devices_from_scan_history(scan_history_msg)
scan_history_device_viewer.update_devices_from_scan_history(scan_history_msg.content)
plotting_callback_args = []
@ -174,11 +176,14 @@ def test_scan_history_device_viewer_on_request_plotting_clicked(
scan_history_device_viewer.request_history_plot.connect(plotting_callback)
qtbot.mouseClick(scan_history_device_viewer.request_plotting_button, QtCore.Qt.LeftButton)
qtbot.waitUntil(lambda: len(plotting_callback_args) > 0, timeout=5000)
assert plotting_callback_args[0][2] == scan_history_msg
assert plotting_callback_args[0][0] in scan_history_msg.stored_data_info.keys()
# scan_id
assert plotting_callback_args[0][0] == scan_history_msg.scan_id
# device_name
assert plotting_callback_args[0][1] in scan_history_msg.stored_data_info.keys()
# signal_name
assert (
plotting_callback_args[0][1]
in scan_history_msg.stored_data_info[plotting_callback_args[0][0]].keys()
plotting_callback_args[0][2]
in scan_history_msg.stored_data_info[plotting_callback_args[0][1]].keys()
)
@ -188,7 +193,7 @@ def test_scan_history_metadata_viewer_receive_msg(
"""Test the initialization of ScanHistoryMetadataViewer."""
assert scan_history_metadata_viewer.scan_history_msg is None
assert scan_history_metadata_viewer.title() == "No Scan Selected"
scan_history_metadata_viewer.update_view(scan_history_msg)
scan_history_metadata_viewer.update_view(scan_history_msg.content)
assert scan_history_metadata_viewer.scan_history_msg == scan_history_msg
assert scan_history_metadata_viewer.title() == f"Metadata - Scan {scan_history_msg.scan_number}"
for row, k in enumerate(scan_history_metadata_viewer._scan_history_msg_labels.keys()):
@ -204,7 +209,7 @@ def test_scan_history_metadata_viewer_clear_view(
qtbot, scan_history_metadata_viewer, scan_history_msg
):
"""Test clearing the metadata viewer."""
scan_history_metadata_viewer.update_view(scan_history_msg)
scan_history_metadata_viewer.update_view(scan_history_msg.content)
assert scan_history_metadata_viewer.scan_history_msg == scan_history_msg
scan_history_metadata_viewer.clear_view()
assert scan_history_metadata_viewer.scan_history_msg is None
@ -289,17 +294,18 @@ def test_scan_history_browser(qtbot, scan_history_browser, scan_history_msg, sca
scan_history_browser.scan_history_view.topLevelItem(0)
).center(),
)
assert scan_history_browser.scan_history_view.currentIndex().row() == 0
# Both metadata and device viewers should be updated with the first scan
qtbot.waitUntil(
lambda: scan_history_browser.scan_history_metadata_viewer.scan_history_msg
== scan_history_msg_2,
timeout=5000,
timeout=2000,
)
qtbot.waitUntil(
lambda: scan_history_browser.scan_history_device_viewer.scan_history_msg
== scan_history_msg_2,
timeout=5000,
timeout=2000,
)
# Click on second scan item history to select it
@ -310,22 +316,15 @@ def test_scan_history_browser(qtbot, scan_history_browser, scan_history_msg, sca
scan_history_browser.scan_history_view.topLevelItem(1)
).center(),
)
qtbot.wait(2000)
qtbot.waitUntil(
lambda: scan_history_browser.scan_history_view.currentIndex().row() == 1, timeout=5000
)
qtbot.wait(2000) # Wait for cbs to be processed on scan history metadata and device viewers
assert scan_history_browser.scan_history_metadata_viewer.scan_history_msg == scan_history_msg
assert scan_history_browser.scan_history_device_viewer.scan_history_msg == scan_history_msg
# Both metadata and device viewers should be updated with the first scan
# qtbot.waitUntil(
# lambda: scan_history_browser.scan_history_metadata_viewer.scan_history_msg
# == scan_history_msg,
# timeout=5000,
# )
# qtbot.waitUntil(
# lambda: scan_history_browser.scan_history_device_viewer.scan_history_msg
# == scan_history_msg,
# timeout=5000,
# )
callback_args = []
def plotting_callback(device_name, signal_name, msg):
@ -340,9 +339,9 @@ def test_scan_history_browser(qtbot, scan_history_browser, scan_history_msg, sca
QtCore.Qt.LeftButton,
)
qtbot.waitUntil(lambda: len(callback_args) > 0, timeout=5000)
assert callback_args[0][2] == scan_history_msg
device_name = callback_args[0][0]
signal_name = callback_args[0][1]
assert callback_args[0][0] == scan_history_msg.scan_id
device_name = callback_args[0][1]
signal_name = callback_args[0][2]
assert device_name in scan_history_msg.stored_data_info.keys()
assert signal_name in scan_history_msg.stored_data_info[device_name].keys()
@ -355,9 +354,9 @@ def test_scan_history_browser(qtbot, scan_history_browser, scan_history_msg, sca
qtbot.waitUntil(
lambda: scan_history_browser.scan_history_metadata_viewer.scan_history_msg is None,
timeout=5000,
timeout=2000,
)
qtbot.waitUntil(
lambda: scan_history_browser.scan_history_device_viewer.scan_history_msg is None,
timeout=5000,
timeout=2000,
)