bec/bec_widgets
0
0
Fork 0
mirror of https://github.com/bec-project/bec_widgets.git synced 2025-07-13 19:21:50 +02:00
Code Activity

fix(waveform): added support for live_data and data access

Browse Source
This commit is contained in:
wakonig_k
2024-10-18 15:04:04 +02:00
parent cb45527f3e
commit 7469c892c8
4 changed files with 33 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
tests/end-2-end/test_bec_dock_rpc_e2e.py
View File

@ -1,8 +1,6 @@
import time
import numpy as np
import pytest
from bec_lib.client import BECClient
from bec_lib.endpoints import MessageEndpoints
from bec_widgets.cli.auto_updates import AutoUpdates
@ -96,7 +94,8 @@ def test_rpc_add_dock_with_figure_e2e(bec_client_lib, rpc_server_dock):
time.sleep(0.2)
# plot
plt_last_scan_data = queue.scan_storage.storage[-1].data
item = queue.scan_storage.storage[-1]
plt_last_scan_data = item.live_data if hasattr(item, "live_data") else item.data
plt_data = plt.get_all_data()
assert plt_data["bpm4i-bpm4i"]["x"] == plt_last_scan_data["samx"]["samx"].val
assert plt_data["bpm4i-bpm4i"]["y"] == plt_last_scan_data["bpm4i"]["bpm4i"].val
@ -258,7 +257,8 @@ def test_auto_update(bec_client_lib, rpc_server_dock, qtbot):
status = scans.line_scan(dev.samx, -5, 5, steps=10, exp_time=0.05, relative=False)
status.wait()
last_scan_data = queue.scan_storage.storage[-1].data
item = queue.scan_storage.storage[-1]
last_scan_data = item.live_data if hasattr(item, "live_data") else item.data
# get data from curves
widgets = plt.widget_list
@ -284,7 +284,8 @@ def test_auto_update(bec_client_lib, rpc_server_dock, qtbot):
qtbot.waitUntil(lambda: len(plt.widget_list) > 0, timeout=5000)
plt_data = widgets[0].get_all_data()
last_scan_data = queue.scan_storage.storage[-1].data
item = queue.scan_storage.storage[-1]
last_scan_data = item.live_data if hasattr(item, "live_data") else item.data
# check plotted data
assert (

3
tests/end-2-end/test_bec_figure_rpc_e2e.py
View File

@ -94,7 +94,8 @@ def test_rpc_waveform_scan(rpc_server_figure, bec_client_lib):
status = scans.line_scan(dev.samx, -5, 5, steps=10, exp_time=0.05, relative=False)
status.wait()
last_scan_data = queue.scan_storage.storage[-1].data
item = queue.scan_storage.storage[-1]
last_scan_data = item.live_data if hasattr(item, "live_data") else item.data
# get data from curves
plt_data = plt.get_all_data()

18
tests/unit_tests/test_waveform1d.py
View File

@ -3,6 +3,7 @@ from unittest import mock
import numpy as np
import pytest
from bec_lib.scan_items import ScanItem
from bec_widgets.widgets.figure import BECFigure
from bec_widgets.widgets.figure.plots.waveform.waveform_curve import CurveConfig, Signal, SignalData
@ -447,8 +448,8 @@ def test_scan_update(qtbot, mocked_client):
"scan_id": 1,
}
# Mock scan_storage.find_scan_by_ID
mock_scan_data_waveform = mock.MagicMock()
mock_scan_data_waveform.data = {
mock_scan_data_waveform = mock.MagicMock(spec=ScanItem)
mock_scan_data_waveform.live_data = {
device_name: {
entry: mock.MagicMock(val=[msg_waveform["data"][device_name][entry]["value"]])
for entry in msg_waveform["data"][device_name]
@ -477,7 +478,9 @@ def test_scan_history_with_val_access(qtbot, mocked_client):
}
mock_scan_storage = mock.MagicMock()
mock_scan_storage.find_scan_by_ID.return_value = mock.MagicMock(data=mock_scan_data)
scan_item_mock = mock.MagicMock(spec=ScanItem)
scan_item_mock.data = mock_scan_data
mock_scan_storage.find_scan_by_ID.return_value = scan_item_mock
w1.queue.scan_storage = mock_scan_storage
fake_scan_id = "fake_scan_id"
@ -507,8 +510,8 @@ def test_scatter_2d_update(qtbot, mocked_client):
}
msg_metadata = {"scan_name": "line_scan"}
mock_scan_item = mock.MagicMock()
mock_scan_item.data = {
mock_scan_item = mock.MagicMock(spec=ScanItem)
mock_scan_item.live_data = {
device_name: {
entry: mock.MagicMock(val=msg["data"][device_name][entry]["value"])
for entry in msg["data"][device_name]
@ -572,7 +575,7 @@ def test_waveform_set_x_sync(qtbot, mocked_client):
w1.plot("bpm4i")
w1.set_x_label(custom_label)
scan_item_mock = mock.MagicMock()
scan_item_mock = mock.MagicMock(spec=ScanItem)
mock_data = {
"samx": {"samx": mock.MagicMock(val=np.array([1, 2, 3]))},
"samy": {"samy": mock.MagicMock(val=np.array([4, 5, 6]))},
@ -584,7 +587,8 @@ def test_waveform_set_x_sync(qtbot, mocked_client):
},
}
scan_item_mock.data = mock_data
scan_item_mock.live_data = mock_data
scan_item_mock.status_message = mock.MagicMock()
scan_item_mock.status_message.info = {"scan_report_devices": ["samx"]}
w1.queue.scan_storage.find_scan_by_ID.return_value = scan_item_mock