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

test: test_stream_plot.py basic tests for stream_plot.py, test_basic_plot.py removed

Browse Source
This commit is contained in:
wyzula-jan
2023-10-17 13:40:30 +02:00
parent 7152c5b229
commit 2925a5f20e
3 changed files with 162 additions and 181 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
bec_widgets/examples/stream_plot/stream_plot.py
View File

@ -67,7 +67,6 @@ class StreamPlot(QtWidgets.QWidget):
self.init_ui()
self.init_curves()
self.hook_crosshair()
self.pushButton_generate.clicked.connect(self.generate_data)
def init_ui(self):
"""Setup all ui elements"""
@ -201,34 +200,6 @@ class StreamPlot(QtWidgets.QWidget):
# ROI
self.roi_selector.sigRegionChangeFinished.connect(self.get_roi_region)
def generate_data(self):
def gauss(x, mu, sigma):
return (1 / (sigma * np.sqrt(2 * np.pi))) * np.exp(-0.5 * ((x - mu) / sigma) ** 2)
self.plotter_data_x = np.linspace(0, 10, 1000)
self.plotter_data_y = [
gauss(self.plotter_data_x, 1, 1),
gauss(self.plotter_data_x, 1.5, 3),
np.sin(self.plotter_data_x),
np.cos(self.plotter_data_x),
np.sin(2 * self.plotter_data_x),
] # List of y-values for multiple curves
self.y_value_list = ["Gauss (1,1)", "Gauss (1.5,3)"] # ["Sine"]#, "Cosine", "Sine2x"]
# Curves
color_list = ["#384c6b", "#e28a2b", "#5E3023", "#e41a1c", "#984e83", "#4daf4a"]
self.init_curves()
for ii in range(len(self.y_value_list)):
self.curves[ii].setData(self.plotter_data_x, self.plotter_data_y[ii])
self.data_2D = np.random.random((150, 30))
self.img.setImage(self.data_2D)
if self.roi_selector not in self.plot.items:
self.plot.addItem(self.roi_selector)
def get_roi_region(self):
"""For testing purpose now, get roi region and print it to self.label as tuple"""
region = self.roi_selector.getRegion()

152
tests/test_basic_plot.py
View File

@ -1,152 +0,0 @@
# from unittest import mock
#
# import numpy as np
# from pytestqt import qtbot
#
# from bec_widgets import line_plot
#
#
# def test_line_plot_emits_no_signal(qtbot):
# """Test LinePlot emits no signal when only one data entry is present."""
#
# y_value_list = ["y1", "y2"]
# plot = line_plot.BasicPlot(y_value_list=y_value_list)
# data = {
# "data": {
# "x": {"x": {"value": 1}},
# "y1": {"y1": {"value": 1}},
# "y2": {"y2": {"value": 3}},
# }
# }
# metadata = {"scanID": "test", "scan_number": 1, "scan_report_devices": ["x"]}
# with mock.patch("bec_widgets.line_plot.client") as mock_client:
# with mock.patch.object(plot, "update_signal") as mock_update_signal:
# plot.on_scan_segment(data=data, metadata=metadata)
# mock_update_signal.emit.assert_not_called()
#
#
# def test_line_plot_emits_signal(qtbot):
# """Test LinePlot emits signal."""
#
# y_value_list = ["y1", "y2"]
# plot = line_plot.BasicPlot(y_value_list=y_value_list)
# data = {
# "data": {
# "x": {"x": {"value": 1}},
# "y1": {"y1": {"value": 1}},
# "y2": {"y2": {"value": 3}},
# }
# }
# plotter_data_y = [[1, 1], [3, 3]]
# metadata = {"scanID": "test", "scan_number": 1, "scan_report_devices": ["x"]}
# with mock.patch("bec_widgets.line_plot.client") as mock_client:
# # mock_client.device_manager.devices.keys.return_value = ["y1"]
# with mock.patch.object(plot, "update_signal") as mock_update_signal:
# mock_update_signal.emit()
# plot.on_scan_segment(data=data, metadata=metadata)
# plot.on_scan_segment(data=data, metadata=metadata)
# mock_update_signal.emit.assert_called()
# # TODO allow mock_client to create return values for device_manager_devices
# # assert plot.plotter_data_y == plotter_data_y
#
#
# def test_line_plot_raise_warning_wrong_signal_request(qtbot):
# """Test LinePlot raises warning and skips signal when entry not present in data."""
#
# y_value_list = ["y1", "y22"]
# plot = line_plot.BasicPlot(y_value_list=y_value_list)
# data = {
# "data": {
# "x": {"x": {"value": [1, 2, 3, 4, 5]}},
# "y1": {"y1": {"value": [1, 2, 3, 4, 5]}},
# "y2": {"y2": {"value": [1, 2, 3, 4, 5]}},
# }
# }
# metadata = {"scanID": "test", "scan_number": 1, "scan_report_devices": ["x"]}
# with mock.patch("bec_widgets.line_plot.client") as mock_client:
# # TODO fix mock_client
# mock_dict = {"y1": [1, 2]}
# mock_client.device_manager.devices.__contains__.side_effect = mock_dict.__contains__
#
# # = {"y1": [1, 2]}
# with mock.patch.object(plot, "update_signal") as mock_update_signal:
# mock_update_signal.emit()
# plot.on_scan_segment(data=data, metadata=metadata)
# assert plot.y_value_list == ["y1"]
# def test_line_plot_update(qtbot):
# """Test LinePlot update."""
# y_value_list = ["y1", "y2"]
# plot = line_plot.BasicPlot(y_value_list=y_value_list)
# plot.label_bottom = "x"
# plot.label_left = f"{', '.join(y_value_list)}"
# plot.plotter_data_x = [1, 2, 3, 4, 5]
# plot.plotter_data_y = [[1, 2, 3, 4, 5], [3, 4, 5, 6, 7]]
# plot.update()
# assert all(plot.curves[0].getData()[0] == np.array([1, 2, 3, 4, 5]))
# assert all(plot.curves[0].getData()[1] == np.array([1, 2, 3, 4, 5]))
# assert all(plot.curves[1].getData()[1] == np.array([3, 4, 5, 6, 7]))
# # TODO Outputting the wrong data, e.g. motor is not in list of devices
# def test_line_plot_update(qtbot):
# """Test LinePlot update."""
# y_value_list = ["y1", "y2"]
# plot = line_plot.BasicPlot(y_value_list=y_value_list)
# plot.label_bottom = "x"
# plot.label_left = f"{', '.join(y_value_list)}"
# plot.plotter_data_x = [1, 2, 3, 4, 5]
# plot.plotter_data_y = [[1, 2, 3, 4, 5], [3, 4, 5, 6, 7]]
# plot.update()
# assert all(plot.curves[0].getData()[0] == np.array([1, 2, 3, 4, 5]))
# assert all(plot.curves[0].getData()[1] == np.array([1, 2, 3, 4, 5]))
# assert all(plot.curves[1].getData()[1] == np.array([3, 4, 5, 6, 7]))
# def test_line_plot_mouse_moved(qtbot):
# """Test LinePlot mouse_moved."""
# y_value_list = ["y1", "y2"]
# plot = line_plot.BasicPlot(y_value_list=y_value_list)
# plot.plotter_data_x = [1, 2, 3, 4, 5]
# plot.plotter_data_y = [[1, 2, 3, 4, 5], [3, 4, 5, 6, 7]]
# plot.precision = 3
# string_cap = 10
# x_data = f"{3:.{plot.precision}f}"
# y_data = f"{3:.{plot.precision}f}"
# output_string = "".join(
# [
# "Mouse cursor",
# "\n",
# f"{y_value_list[0]}",
# "\n",
# f"X_data: {x_data:>{string_cap}}",
# "\n",
# f"Y_data: {y_data:>{string_cap}}",
# ]
# )
# x_data = f"{3:.{plot.precision}f}"
# y_data = f"{5:.{plot.precision}f}"
# output_string = "".join(
# [
# output_string,
# "\n",
# f"{y_value_list[1]}",
# "\n",
# f"X_data: {x_data:>{string_cap}}",
# "\n",
# f"Y_data: {y_data:>{string_cap}}",
# ]
# )
# with mock.patch.object(
# plot, "plot"
# ) as mock_plot: # TODO change test to simulate QTable instead of QLabel
# mock_plot.sceneBoundingRect.contains.return_value = True
# mock_plot.vb.mapSceneToView((20, 10)).x.return_value = 2.8
# plot.mouse_moved((20, 10))
# assert plot.mouse_box_data.text() == output_string

162
tests/test_stream_plot.py Normal file
View File

@ -0,0 +1,162 @@
from unittest import mock
import numpy as np
import pytest
from bec_lib.core import BECMessage
from pytestqt import qtbot
import threading
from bec_lib.core import RedisConnector
from bec_widgets.examples.stream_plot.stream_plot import StreamPlot
@pytest.fixture
def stream_app(qtbot):
"""Helper function to set up the StreamPlot widget."""
client = mock.MagicMock()
widget = StreamPlot(client=client)
qtbot.addWidget(widget)
qtbot.waitExposed(widget)
return widget
def test_roi_signals_emitted(qtbot, stream_app):
region = (0.1, 0.9)
with qtbot.waitSignal(stream_app.roi_signal, timeout=1000) as blocker:
stream_app.roi_signal.emit(region)
assert blocker.signal_triggered
assert blocker.args == [region]
def test_update_signals_emitted(qtbot, stream_app):
# Mimic data coming from the data stream
stream_app.plotter_data_x = [list(range(10))] # Replace with the actual x data
stream_app.plotter_data_y = [list(range(10))] # Replace with the actual y data
# Initialize curves
stream_app.init_curves()
with qtbot.waitSignal(stream_app.update_signal, timeout=1000) as blocker:
stream_app.update_signal.emit()
assert blocker.signal_triggered
def test_ui_initialization(qtbot, stream_app):
"""Checking the UI creation."""
# Check if UI elements are initialized correctly
assert stream_app.label_plot is not None
assert stream_app.label_plot_moved is not None
assert stream_app.label_plot_clicked is not None
assert stream_app.label_image_moved is not None
assert stream_app.label_image_clicked is not None
# Check if plots are initialized correctly
assert stream_app.plot is not None
assert stream_app.plot_image is not None
# Check if ROI selector is initialized correctly
assert stream_app.roi_selector is not None
def test_1d_plotting_data(qtbot, stream_app):
# Set up some mock data
x_data = [list(range(10))]
y_data = [list(range(10))]
# Manually set the data attributes
stream_app.plotter_data_x = x_data
stream_app.plotter_data_y = y_data
stream_app.y_value_list = ["Curve 1"]
# Initialize curves and update the plot
stream_app.init_curves()
stream_app.update() # This should update the plot with the new data
# Check the data on the plot
for idx, curve in enumerate(stream_app.curves):
np.testing.assert_array_equal(curve.xData, x_data[0]) # Access the first list of x_data
np.testing.assert_array_equal(
curve.yData, y_data[idx]
) # Access the list of y_data for each curve without additional indexing
def test_flip_even_rows(qtbot, stream_app):
# Create a numpy array with some known data
original_array = np.array(
[
[1, 2, 3, 4, 5],
[6, 7, 8, 9, 10],
[11, 12, 13, 14, 15],
[16, 17, 18, 19, 20],
]
)
# Call flip_even_rows on the original array
flipped_array = stream_app.flip_even_rows(original_array)
# Expected array flipped along the rows with even indices
expected_array = np.array(
[
[1, 2, 3, 4, 5],
[10, 9, 8, 7, 6],
[11, 12, 13, 14, 15],
[20, 19, 18, 17, 16],
]
)
# Check that flip_even_rows returned the expected result
np.testing.assert_array_equal(flipped_array, expected_array)
def test_on_dap_update(qtbot, stream_app):
"""2D image rendering by dap update"""
# Create some mock data to be "received" by the slot
data_dict = {"data": {"z": np.random.rand(10, 10)}}
metadata_dict = {}
# Trigger the slot
stream_app.on_dap_update(data_dict, metadata_dict)
# Apply the same transformation to the test data
expected_data = stream_app.flip_even_rows(data_dict["data"]["z"])
# Now check the state of the StreamPlot object
# For example, check the data of the image plot:
np.testing.assert_array_equal(stream_app.img.image, expected_data)
# def test_new_proj(qtbot, stream_app): #TODO this test is not working, does it make sense testing even?
# # Create some mock content to be "received" by the slot
# content_dict = {"signals": {"proj_nr": 1}}
# metadata_dict = {}
#
# # Manually create some mock data that new_proj would work with
# # This step may need to be adjusted to fit the actual behavior of new_proj
# mock_data = {
# "q": np.array([1, 2, 3, 4, 5]),
# "norm_sum": np.array([6, 7, 8, 9, 10]),
# "metadata": "some_metadata",
# }
#
# # Assume the RedisConnector client would return this data when new_proj is called
# mock_message = mock.MagicMock(spec=BECMessage.DeviceMessage)
# mock_message.__getitem__.side_effect = lambda key: mock_data[key]
# stream_app.client.producer.get = mock.MagicMock(return_value=mock_message.dumps())
#
# # Trigger the slot
# stream_app.new_proj(content_dict, metadata_dict)
#
# # Now check the state of the StreamPlot object
# # For example, check that the plotter_data_x attribute was updated correctly:
# np.testing.assert_array_equal(stream_app.plotter_data_x, [mock_data["q"]])
# assert stream_app._current_proj == 1
# assert stream_app._current_q == mock_data["q"]
# assert stream_app._current_norm == mock_data["norm_sum"]
# assert stream_app._current_metadata == mock_data["metadata"]
# def test_connection_creation(qtbot, stream_app): #TODO maybe test connections in a different way?
# assert isinstance(stream_app.producer, RedisConnector)
# assert isinstance(stream_app.data_retriever, threading.Thread)
# assert stream_app.data_retriever.is_alive()