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

fix(ring): ProgressSignal fetch logic back

Browse Source
This commit is contained in:
wyzula_j
2026-06-06 10:54:47 +02:00
committed by Jan Wyzula
parent e8e67f68a2
commit e8bd80377e
2 changed files with 271 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bec_widgets/widgets/progress/ring_progress_bar/ring.py
+100 -14
View File
@@ -12,7 +12,6 @@ from qtpy.QtWidgets import QWidget
from bec_widgets import BECWidget
from bec_widgets.utils.bec_connector import ConnectionConfig
from bec_widgets.utils.colors import Colors
from bec_widgets.utils.entry_validator import EntryValidator
from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
from bec_widgets.widgets.progress.progress_backend import BECProgressTracker, ProgressSnapshot
@@ -269,19 +268,59 @@ class Ring(BECWidget, QWidget):
self.config.direction = direction
self._request_update()
def _update_device_connection(self, device: str, signal: str | None) -> str:
def _get_signals_for_device(self, device: str) -> dict[str, list[str]]:
"""
Update the device connection for the ring widget.
Device mode always subscribes to the device readback endpoint. If no signal is provided,
the signal is resolved from the device hints, matching the plot widgets.
Get the appropriate signals for the device to be used in the ring widget, based on the signal infos from the device manager.
Args:
device(str): Device name for the device mode
signal(str): Signal name for the device mode
device(str): Device name for the device readback mode
Returns:
str: The selected signal name for the device mode
dict[str, list[str]]: Signal infos for the device to be used in the ring widget
"""
dm = self.bec_dispatcher.client.device_manager
if not dm:
raise ValueError("Device manager is not available in the BEC client.")
dev_obj = dm.devices.get(device)
if dev_obj is None:
raise ValueError(f"Device '{device}' not found in device manager.")
signal_infos = getattr(dev_obj, "_info", {}).get("signals", {})
progress_signals = [
obj["component_name"]
for obj in signal_infos.values()
if obj.get("signal_class") == "ProgressSignal"
]
hinted_signals = [
obj["obj_name"]
for obj in signal_infos.values()
if obj.get("kind_str") == "hinted"
and obj.get("signal_class")
not in ["ProgressSignal", "AsyncSignal", "AsyncMultiSignal", "DynamicSignal"]
]
normal_signals = [
obj["component_name"]
for obj in signal_infos.values()
if obj.get("kind_str") == "normal"
]
return {
"progress_signals": progress_signals,
"hinted_signals": hinted_signals,
"normal_signals": normal_signals,
}
def _update_device_connection(self, device: str, signal: str | None) -> str:
"""
Subscribe device mode to the endpoint matching the selected signal.
When no signal is provided, the ring selects the first available progress
signal, then the first hinted readback signal, then the first normal
readback signal. Progress signals use the device_progress endpoint;
readback signals use the device_readback endpoint.
Returns:
The selected signal name, or an empty string if the device is not known.
"""
logger.info(f"Updating device connection for device '{device}' and signal '{signal}'")
dm = self.bec_dispatcher.client.device_manager
@@ -291,11 +330,48 @@ class Ring(BECWidget, QWidget):
if dev_obj is None:
return ""
signal = EntryValidator(dm.devices).validate_signal(device, signal or None)
endpoint = MessageEndpoints.device_readback(device)
self.bec_dispatcher.connect_slot(self.on_device_readback, endpoint)
self.registered_slot = (self.on_device_readback, endpoint)
return signal
signals = self._get_signals_for_device(device)
progress_signals = signals["progress_signals"]
hinted_signals = signals["hinted_signals"]
normal_signals = signals["normal_signals"]
if not signal:
if progress_signals:
signal = progress_signals[0]
logger.info(
f"Using progress signal '{signal}' for device '{device}' in ring progress bar."
)
elif hinted_signals:
signal = hinted_signals[0]
logger.info(
f"Using hinted signal '{signal}' for device '{device}' in ring progress bar."
)
elif normal_signals:
signal = normal_signals[0]
logger.info(
f"Using normal signal '{signal}' for device '{device}' in ring progress bar."
)
else:
logger.warning(f"No signals found for device '{device}' in ring progress bar.")
return ""
if signal in progress_signals:
endpoint = MessageEndpoints.device_progress(device)
self.bec_dispatcher.connect_slot(self.on_device_progress, endpoint)
self.registered_slot = (self.on_device_progress, endpoint)
return signal
if signal in hinted_signals or signal in normal_signals:
endpoint = MessageEndpoints.device_readback(device)
self.bec_dispatcher.connect_slot(self.on_device_readback, endpoint)
self.registered_slot = (self.on_device_readback, endpoint)
return signal
raise ValueError(
f"Signal '{signal}' is not usable for ring progress device mode. "
f"Available progress signals: {progress_signals}; "
f"available readback signals: {hinted_signals + normal_signals}."
)
@SafeSlot(dict, dict)
def on_device_readback(self, msg, meta):
@@ -316,6 +392,16 @@ class Ring(BECWidget, QWidget):
self.set_value(value)
self.update()
@SafeSlot(dict, dict)
def on_device_progress(self, msg, meta):
device = self.config.device
if device is None:
return
max_val = msg.get("max_value", 100)
self.set_min_max_values(0, max_val)
self.set_value(max_val if msg.get("done") else msg.get("value", 0))
self.update()
def _on_progress_snapshot(self, snapshot: ProgressSnapshot):
if snapshot.is_new_scan:
self.set_min_max_values(0, snapshot.max_value)
tests/unit_tests/test_ring_progress_bar_ring.py
+171 -8
View File
@@ -3,6 +3,7 @@
from unittest.mock import MagicMock
import pytest
from bec_lib.endpoints import MessageEndpoints
from qtpy.QtGui import QColor
from bec_widgets.tests.utils import FakeDevice
@@ -97,6 +98,10 @@ def test_set_update_from_scan_to_manual(ring_widget):
assert ring_widget.config.mode == "manual"
assert ring_widget.registered_slot is None
ring_widget.bec_dispatcher.disconnect_slot.assert_called_once()
call_args = ring_widget.bec_dispatcher.disconnect_slot.call_args
assert call_args[0][0] == ring_widget.progress_tracker.process_progress_message
assert call_args[0][1] == MessageEndpoints.scan_progress()
def test_set_update_to_device(ring_widget_with_device):
@@ -420,7 +425,28 @@ def test_set_direction_counter_clockwise(ring_widget):
###################################
def test_update_device_connection_with_progress_signal(ring_widget_with_device):
def test_update_device_connection_prefers_progress_signal(ring_widget_with_device):
ring_widget = ring_widget_with_device
samx = ring_widget.bec_dispatcher.client.device_manager.devices.samx
samx._info["signals"]["progress"] = {
"obj_name": "samx_progress",
"component_name": "progress",
"signal_class": "ProgressSignal",
"kind_str": "hinted",
}
ring_widget.bec_dispatcher.connect_slot = MagicMock()
signal = ring_widget._update_device_connection("samx", "")
assert signal == "progress"
ring_widget.bec_dispatcher.connect_slot.assert_called_once()
call_args = ring_widget.bec_dispatcher.connect_slot.call_args
assert call_args[0][0] == ring_widget.on_device_progress
assert call_args[0][1] == MessageEndpoints.device_progress("samx")
def test_update_device_connection_accepts_explicit_progress_signal(ring_widget_with_device):
ring_widget = ring_widget_with_device
samx = ring_widget.bec_dispatcher.client.device_manager.devices.samx
samx._info["signals"]["progress"] = {
@@ -434,14 +460,91 @@ def test_update_device_connection_with_progress_signal(ring_widget_with_device):
signal = ring_widget._update_device_connection("samx", "progress")
# Device mode always connects to device_readback, even if the explicit signal is a ProgressSignal.
assert signal == "samx_progress"
assert signal == "progress"
ring_widget.bec_dispatcher.connect_slot.assert_called_once()
call_args = ring_widget.bec_dispatcher.connect_slot.call_args
assert call_args[0][0] == ring_widget.on_device_progress
assert call_args[0][1] == MessageEndpoints.device_progress("samx")
def test_update_device_connection_resolves_component_name_to_readback_signal(
ring_widget_with_device,
):
ring_widget = ring_widget_with_device
samx = ring_widget.bec_dispatcher.client.device_manager.devices.samx
samx._info["signals"]["setpoint"] = {
"obj_name": "samx_setpoint",
"component_name": "setpoint",
"signal_class": "Signal",
"kind_str": "normal",
}
ring_widget.bec_dispatcher.connect_slot = MagicMock()
signal = ring_widget._update_device_connection("samx", "setpoint")
assert signal == "setpoint"
ring_widget.bec_dispatcher.connect_slot.assert_called_once()
call_args = ring_widget.bec_dispatcher.connect_slot.call_args
assert call_args[0][0] == ring_widget.on_device_readback
assert call_args[0][1] == MessageEndpoints.device_readback("samx")
def test_update_device_connection_with_hinted_signal(ring_widget):
def test_update_device_connection_falls_back_to_hinted_signal(ring_widget_with_device):
ring_widget = ring_widget_with_device
ring_widget.bec_dispatcher.connect_slot = MagicMock()
signal = ring_widget._update_device_connection("samx", "")
assert signal == "samx"
ring_widget.bec_dispatcher.connect_slot.assert_called_once()
call_args = ring_widget.bec_dispatcher.connect_slot.call_args
assert call_args[0][0] == ring_widget.on_device_readback
assert call_args[0][1] == MessageEndpoints.device_readback("samx")
def test_update_device_connection_falls_back_to_normal_signal(ring_widget_with_device):
ring_widget = ring_widget_with_device
samx = ring_widget.bec_dispatcher.client.device_manager.devices.samx
samx._info["signals"] = {
"setpoint": {
"obj_name": "samx_setpoint",
"component_name": "setpoint",
"signal_class": "Signal",
"kind_str": "normal",
}
}
ring_widget.bec_dispatcher.connect_slot = MagicMock()
signal = ring_widget._update_device_connection("samx", "")
assert signal == "setpoint"
ring_widget.bec_dispatcher.connect_slot.assert_called_once()
call_args = ring_widget.bec_dispatcher.connect_slot.call_args
assert call_args[0][0] == ring_widget.on_device_readback
assert call_args[0][1] == MessageEndpoints.device_readback("samx")
def test_update_device_connection_rejects_unusable_signal(ring_widget_with_device):
ring_widget = ring_widget_with_device
samx = ring_widget.bec_dispatcher.client.device_manager.devices.samx
samx._info["signals"]["async_signal"] = {
"obj_name": "samx_async",
"component_name": "async_signal",
"signal_class": "AsyncSignal",
"kind_str": "hinted",
}
ring_widget.bec_dispatcher.connect_slot = MagicMock()
with pytest.raises(ValueError, match="not usable for ring progress device mode"):
ring_widget._update_device_connection("samx", "samx_async")
ring_widget.bec_dispatcher.connect_slot.assert_not_called()
def test_update_device_connection_accepts_explicit_hinted_signal(ring_widget):
mock_device = FakeDevice(name="samx")
mock_device._info = {
"signals": {
@@ -453,12 +556,13 @@ def test_update_device_connection_with_hinted_signal(ring_widget):
ring_widget.bec_dispatcher.connect_slot = MagicMock()
ring_widget._update_device_connection("samx", "samx")
signal = ring_widget._update_device_connection("samx", "samx")
# Should connect to device_readback endpoint
assert signal == "samx"
ring_widget.bec_dispatcher.connect_slot.assert_called_once()
call_args = ring_widget.bec_dispatcher.connect_slot.call_args
assert call_args[0][0] == ring_widget.on_device_readback
assert call_args[0][1] == MessageEndpoints.device_readback("samx")
def test_update_device_connection_no_device_manager(ring_widget):
@@ -473,8 +577,7 @@ def test_update_device_connection_device_not_found(ring_widget):
mock_device = FakeDevice(name="samx")
ring_widget.bec_dispatcher.client.device_manager.devices["samx"] = mock_device
# Should return without raising an error
ring_widget._update_device_connection("nonexistent", "signal")
assert ring_widget._update_device_connection("nonexistent", "signal") == ""
###################################
@@ -567,3 +670,63 @@ def test_on_device_readback_missing_signal_data(ring_widget):
# Value should not change when signal is missing
assert ring_widget.config.value == initial_value
###################################
# on_device_progress tests
###################################
def test_on_device_progress_updates_value_and_max(ring_widget):
ring_widget.config.device = "samx"
msg = {"value": 30, "max_value": 150, "done": False}
meta = {}
ring_widget.on_device_progress(msg, meta)
assert ring_widget.config.value == 30
assert ring_widget.config.max_value == 150
def test_on_device_progress_done_sets_to_max(ring_widget):
ring_widget.config.device = "samx"
msg = {"value": 80, "max_value": 100, "done": True}
meta = {}
ring_widget.on_device_progress(msg, meta)
# When done is True, value should be set to max_value
assert ring_widget.config.value == 100
assert ring_widget.config.max_value == 100
def test_on_device_progress_no_device_returns_early(ring_widget):
ring_widget.config.device = None
msg = {"value": 50, "max_value": 100, "done": False}
meta = {}
initial_value = ring_widget.config.value
initial_max = ring_widget.config.max_value
ring_widget.on_device_progress(msg, meta)
# Nothing should change
assert ring_widget.config.value == initial_value
assert ring_widget.config.max_value == initial_max
def test_on_device_progress_default_values(ring_widget):
ring_widget.config.device = "samx"
# Message without value and max_value
msg = {}
meta = {}
ring_widget.on_device_progress(msg, meta)
# Should use defaults: value=0, max_value=100
assert ring_widget.config.value == 0
assert ring_widget.config.max_value == 100