bec/bec_widgets
1
0
Fork 0
mirror of https://github.com/bec-project/bec_widgets.git synced 2026-03-09 02:07:55 +01:00
Code Activity

fix: adjust ring progress bar to ads

Browse Source
This commit is contained in:
wakonig_k
2026-01-20 19:14:55 +01:00
committed by wyzula-jan
parent 2992939b0f
commit 7fd7f67857
13 changed files with 2677 additions and 1172 deletions
Download Patch File Download Diff File Expand all files Collapse all files

278
bec_widgets/cli/client.py
View File

@@ -4713,29 +4713,6 @@ class ResumeButton(RPCBase):
class Ring(RPCBase):
@rpc_call
def _get_all_rpc(self) -> "dict":
"""
Get all registered RPC objects.
"""
@property
@rpc_call
def _rpc_id(self) -> "str":
"""
Get the RPC ID of the widget.
"""
@property
@rpc_call
def _config_dict(self) -> "dict":
"""
Get the configuration of the widget.
Returns:
dict: The configuration of the widget.
"""
@rpc_call
def set_value(self, value: "int | float"):
"""
@@ -4755,14 +4732,24 @@ class Ring(RPCBase):
"""
@rpc_call
def set_background(self, color: "str | tuple"):
def set_background(self, color: "str | tuple | QColor"):
"""
Set the background color for the ring widget
Set the background color for the ring widget. The background color is only used when colors are not linked.
Args:
color(str | tuple): Background color for the ring widget. Can be HEX code or tuple (R, G, B, A).
"""
@rpc_call
def set_colors_linked(self, linked: "bool"):
"""
Set whether the colors are linked for the ring widget.
If colors are linked, changing the main color will also change the background color.
Args:
linked(bool): Whether to link the colors for the ring widget
"""
@rpc_call
def set_line_width(self, width: "int"):
"""
@@ -4785,14 +4772,16 @@ class Ring(RPCBase):
@rpc_call
def set_start_angle(self, start_angle: "int"):
"""
Set the start angle for the ring widget
Set the start angle for the ring widget.
Args:
start_angle(int): Start angle for the ring widget in degrees
"""
@rpc_call
def set_update(self, mode: "Literal['manual', 'scan', 'device']", device: "str" = None):
def set_update(
self, mode: "Literal['manual', 'scan', 'device']", device: "str" = "", signal: "str" = ""
):
"""
Set the update mode for the ring widget.
Modes:
@@ -4803,193 +4792,24 @@ class Ring(RPCBase):
Args:
mode(str): Update mode for the ring widget. Can be "manual", "scan" or "device"
device(str): Device name for the device readback mode, only used when mode is "device"
signal(str): Signal name for the device readback mode, only used when mode is "device"
"""
@rpc_call
def reset_connection(self):
def set_precision(self, precision: "int"):
"""
Reset the connections for the ring widget. Disconnect the current slot and endpoint.
Set the precision for the ring widget.
Args:
precision(int): Precision for the ring widget
"""
class RingProgressBar(RPCBase):
"""Show the progress of devices, scans or custom values in the form of ring progress bars."""
@rpc_call
def _get_all_rpc(self) -> "dict":
def remove(self):
"""
Get all registered RPC objects.
"""
@property
@rpc_call
def _rpc_id(self) -> "str":
"""
Get the RPC ID of the widget.
"""
@property
@rpc_call
def _config_dict(self) -> "dict":
"""
Get the configuration of the widget.
Returns:
dict: The configuration of the widget.
"""
@property
@rpc_call
def rings(self) -> "list[Ring]":
"""
Returns a list of all rings in the progress bar.
"""
@rpc_call
def update_config(self, config: "RingProgressBarConfig | dict"):
"""
Update the configuration of the widget.
Args:
config(SpiralProgressBarConfig|dict): Configuration to update.
"""
@rpc_call
def add_ring(self, **kwargs) -> "Ring":
"""
Add a new progress bar.
Args:
**kwargs: Keyword arguments for the new progress bar.
Returns:
Ring: Ring object.
"""
@rpc_call
def remove_ring(self, index: "int"):
"""
Remove a progress bar by index.
Args:
index(int): Index of the progress bar to remove.
"""
@rpc_call
def set_precision(self, precision: "int", bar_index: "int | None" = None):
"""
Set the precision for the progress bars. If bar_index is not provide, the precision will be set for all progress bars.
Args:
precision(int): Precision for the progress bars.
bar_index(int): Index of the progress bar to set the precision for. If provided, only a single precision can be set.
"""
@rpc_call
def set_min_max_values(
self,
min_values: "int | float | list[int | float]",
max_values: "int | float | list[int | float]",
):
"""
Set the minimum and maximum values for the progress bars.
Args:
min_values(int|float | list[float]): Minimum value(s) for the progress bars. If multiple progress bars are displayed, provide a list of minimum values for each progress bar.
max_values(int|float | list[float]): Maximum value(s) for the progress bars. If multiple progress bars are displayed, provide a list of maximum values for each progress bar.
"""
@rpc_call
def set_number_of_bars(self, num_bars: "int"):
"""
Set the number of progress bars to display.
Args:
num_bars(int): Number of progress bars to display.
"""
@rpc_call
def set_value(self, values: "int | list", ring_index: "int" = None):
"""
Set the values for the progress bars.
Args:
values(int | tuple): Value(s) for the progress bars. If multiple progress bars are displayed, provide a tuple of values for each progress bar.
ring_index(int): Index of the progress bar to set the value for. If provided, only a single value can be set.
Examples:
>>> SpiralProgressBar.set_value(50)
>>> SpiralProgressBar.set_value([30, 40, 50]) # (outer, middle, inner)
>>> SpiralProgressBar.set_value(60, bar_index=1) # Set the value for the middle progress bar.
"""
@rpc_call
def set_colors_from_map(self, colormap, color_format: "Literal['RGB', 'HEX']" = "RGB"):
"""
Set the colors for the progress bars from a colormap.
Args:
colormap(str): Name of the colormap.
color_format(Literal["RGB","HEX"]): Format of the returned colors ('RGB', 'HEX').
"""
@rpc_call
def set_colors_directly(
self, colors: "list[str | tuple] | str | tuple", bar_index: "int" = None
):
"""
Set the colors for the progress bars directly.
Args:
colors(list[str | tuple] | str | tuple): Color(s) for the progress bars. If multiple progress bars are displayed, provide a list of colors for each progress bar.
bar_index(int): Index of the progress bar to set the color for. If provided, only a single color can be set.
"""
@rpc_call
def set_line_widths(self, widths: "int | list[int]", bar_index: "int" = None):
"""
Set the line widths for the progress bars.
Args:
widths(int | list[int]): Line width(s) for the progress bars. If multiple progress bars are displayed, provide a list of line widths for each progress bar.
bar_index(int): Index of the progress bar to set the line width for. If provided, only a single line width can be set.
"""
@rpc_call
def set_gap(self, gap: "int"):
"""
Set the gap between the progress bars.
Args:
gap(int): Gap between the progress bars.
"""
@rpc_call
def set_diameter(self, diameter: "int"):
"""
Set the diameter of the widget.
Args:
diameter(int): Diameter of the widget.
"""
@rpc_call
def reset_diameter(self):
"""
Reset the fixed size of the widget.
"""
@rpc_call
def enable_auto_updates(self, enable: "bool" = True):
"""
Enable or disable updates based on scan status. Overrides manual updates.
The behaviour of the whole progress bar widget will be driven by the scan queue status.
Args:
enable(bool): True or False.
Returns:
bool: True if scan segment updates are enabled.
Cleanup the BECConnector
"""
@rpc_call
@@ -5011,6 +4831,56 @@ class RingProgressBar(RPCBase):
Take a screenshot of the dock area and save it to a file.
"""
@property
@rpc_call
def rings(self) -> list[bec_widgets.widgets.progress.ring_progress_bar.ring.Ring]:
"""
None
"""
@rpc_call
def add_ring(
self, config: dict | None = None
) -> bec_widgets.widgets.progress.ring_progress_bar.ring.Ring:
"""
Add a new ring to the ring progress bar.
Optionally, a configuration dictionary can be provided but the ring
can also be configured later. The config dictionary must provide
the qproperties of the Qt Ring object.
Args:
config(dict | None): Optional configuration dictionary for the ring.
Returns:
Ring: The newly added ring object.
"""
@rpc_call
def remove_ring(self, index: int | None = None):
"""
Remove a ring from the ring progress bar.
Args:
index(int | None): Index of the ring to remove. If None, removes the last ring.
"""
@rpc_call
def set_gap(self, value: int):
"""
Set the gap between rings.
Args:
value(int): Gap value in pixels.
"""
@rpc_call
def set_center_label(self, text: str):
"""
Set the center label text.
Args:
text(str): Text for the center label.
"""
class SBBMonitor(RPCBase):
"""A widget to display the SBB monitor website."""

2
bec_widgets/widgets/containers/advanced_dock_area/advanced_dock_area.py
View File

@@ -77,7 +77,7 @@ from bec_widgets.widgets.plots.motor_map.motor_map import MotorMap
from bec_widgets.widgets.plots.multi_waveform.multi_waveform import MultiWaveform
from bec_widgets.widgets.plots.scatter_waveform.scatter_waveform import ScatterWaveform
from bec_widgets.widgets.plots.waveform.waveform import Waveform
from bec_widgets.widgets.progress.ring_progress_bar import RingProgressBar
from bec_widgets.widgets.progress.ring_progress_bar.ring_progress_bar import RingProgressBar
from bec_widgets.widgets.services.bec_queue.bec_queue import BECQueue
from bec_widgets.widgets.services.bec_status_box.bec_status_box import BECStatusBox
from bec_widgets.widgets.utility.logpanel import LogPanel

1
bec_widgets/widgets/progress/ring_progress_bar/__init__.py
View File

@@ -1 +0,0 @@
from .ring_progress_bar import RingProgressBar

649
bec_widgets/widgets/progress/ring_progress_bar/ring.py
View File

@@ -1,130 +1,88 @@
from __future__ import annotations
from typing import Literal, Optional
from typing import TYPE_CHECKING, Callable, Literal
from bec_lib.endpoints import EndpointInfo, MessageEndpoints
from pydantic import BaseModel, Field, field_validator
from pydantic_core import PydanticCustomError
from qtpy import QtGui
from qtpy.QtCore import QObject
from bec_lib.logger import bec_logger
from pydantic import Field
from qtpy import QtCore, QtGui
from qtpy.QtGui import QColor
from qtpy.QtWidgets import QWidget
from bec_widgets.utils import BECConnector, ConnectionConfig
from bec_widgets.utils.bec_connector import BECConnector, ConnectionConfig
from bec_widgets.utils.colors import Colors
from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
class ProgressbarConnections(BaseModel):
slot: Literal["on_scan_progress", "on_device_readback", None] = None
endpoint: EndpointInfo | str | None = None
model_config: dict = {"validate_assignment": True}
@field_validator("endpoint")
@classmethod
def validate_endpoint(cls, v, values):
slot = values.data["slot"]
v = v.endpoint if isinstance(v, EndpointInfo) else v
if slot == "on_scan_progress":
if v != MessageEndpoints.scan_progress().endpoint:
raise PydanticCustomError(
"unsupported endpoint",
"For slot 'on_scan_progress', endpoint must be MessageEndpoint.scan_progress or 'scans/scan_progress'.",
{"wrong_value": v},
)
elif slot == "on_device_readback":
if not v.startswith(MessageEndpoints.device_readback("").endpoint):
raise PydanticCustomError(
"unsupported endpoint",
"For slot 'on_device_readback', endpoint must be MessageEndpoint.device_readback(device) or 'internal/devices/readback/{device}'.",
{"wrong_value": v},
)
return v
logger = bec_logger.logger
if TYPE_CHECKING:
from bec_widgets.widgets.progress.ring_progress_bar.ring_progress_bar import (
RingProgressContainerWidget,
)
class ProgressbarConfig(ConnectionConfig):
value: int | float | None = Field(0, description="Value for the progress bars.")
direction: int | None = Field(
value: int | float = Field(0, description="Value for the progress bars.")
direction: int = Field(
-1, description="Direction of the progress bars. -1 for clockwise, 1 for counter-clockwise."
)
color: str | tuple | None = Field(
color: str | tuple = Field(
(0, 159, 227, 255),
description="Color for the progress bars. Can be tuple (R, G, B, A) or string HEX Code.",
)
background_color: str | tuple | None = Field(
background_color: str | tuple = Field(
(200, 200, 200, 50),
description="Background color for the progress bars. Can be tuple (R, G, B, A) or string HEX Code.",
)
index: int | None = Field(0, description="Index of the progress bar. 0 is outer ring.")
line_width: int | None = Field(10, description="Line widths for the progress bars.")
start_position: int | None = Field(
link_colors: bool = Field(
True,
description="Whether to link the background color to the main color. If True, changing the main color will also change the background color.",
)
line_width: int = Field(20, description="Line widths for the progress bars.")
start_position: int = Field(
90,
description="Start position for the progress bars in degrees. Default is 90 degrees - corespons to "
"the top of the ring.",
)
min_value: int | float | None = Field(0, description="Minimum value for the progress bars.")
max_value: int | float | None = Field(100, description="Maximum value for the progress bars.")
precision: int | None = Field(3, description="Precision for the progress bars.")
update_behaviour: Literal["manual", "auto"] | None = Field(
"auto", description="Update behaviour for the progress bars."
min_value: int | float = Field(0, description="Minimum value for the progress bars.")
max_value: int | float = Field(100, description="Maximum value for the progress bars.")
precision: int = Field(3, description="Precision for the progress bars.")
mode: Literal["manual", "scan", "device"] = Field(
"manual", description="Update mode for the progress bars."
)
connections: ProgressbarConnections | None = Field(
default_factory=ProgressbarConnections, description="Connections for the progress bars."
device: str | None = Field(
None,
description="Device name for the device readback mode, only used when mode is 'device'.",
)
signal: str | None = Field(
None,
description="Signal name for the device readback mode, only used when mode is 'device'.",
)
class RingConfig(ProgressbarConfig):
index: int | None = Field(0, description="Index of the progress bar. 0 is outer ring.")
start_position: int | None = Field(
90,
description="Start position for the progress bars in degrees. Default is 90 degrees - corespons to "
"the top of the ring.",
)
class Ring(BECConnector, QObject):
class Ring(BECConnector, QWidget):
USER_ACCESS = [
"_get_all_rpc",
"_rpc_id",
"_config_dict",
"set_value",
"set_color",
"set_background",
"set_colors_linked",
"set_line_width",
"set_min_max_values",
"set_start_angle",
"set_update",
"reset_connection",
"set_precision",
]
RPC = True
def __init__(
self,
parent=None,
config: RingConfig | dict | None = None,
client=None,
gui_id: Optional[str] = None,
**kwargs,
):
if config is None:
config = RingConfig(widget_class=self.__class__.__name__)
self.config = config
else:
if isinstance(config, dict):
config = RingConfig(**config)
self.config = config
super().__init__(parent=parent, client=client, config=config, gui_id=gui_id, **kwargs)
self.parent_progress_widget = parent
self.color = None
self.background_color = None
self.start_position = None
self.config = config
def __init__(self, parent: RingProgressContainerWidget | None = None, client=None, **kwargs):
self.progress_container = parent
self.config: ProgressbarConfig = ProgressbarConfig(widget_class=self.__class__.__name__) # type: ignore
super().__init__(parent=parent, client=client, config=self.config, **kwargs)
self._color: QColor = self.convert_color(self.config.color)
self._background_color: QColor = self.convert_color(self.config.background_color)
self.registered_slot: tuple[Callable, str | EndpointInfo] | None = None
self.RID = None
self._init_config_params()
def _init_config_params(self):
self.color = self.convert_color(self.config.color)
self.background_color = self.convert_color(self.config.background_color)
self._gap = 5
self.set_start_angle(self.config.start_position)
if self.config.connections:
self.set_connections(self.config.connections.slot, self.config.connections.endpoint)
def set_value(self, value: int | float):
"""
@@ -133,11 +91,7 @@ class Ring(BECConnector, QObject):
Args:
value(int | float): Value for the ring widget
"""
self.config.value = round(
float(max(self.config.min_value, min(self.config.max_value, value))),
self.config.precision,
)
self.parent_progress_widget.update()
self.value = value
def set_color(self, color: str | tuple):
"""
@@ -146,20 +100,53 @@ class Ring(BECConnector, QObject):
Args:
color(str | tuple): Color for the ring widget. Can be HEX code or tuple (R, G, B, A).
"""
self.config.color = color
self.color = self.convert_color(color)
self.parent_progress_widget.update()
self._color = self.convert_color(color)
self.config.color = self._color.name()
def set_background(self, color: str | tuple):
# Automatically set background color
if self.config.link_colors:
self._auto_set_background_color()
self.update()
def set_background(self, color: str | tuple | QColor):
"""
Set the background color for the ring widget
Set the background color for the ring widget. The background color is only used when colors are not linked.
Args:
color(str | tuple): Background color for the ring widget. Can be HEX code or tuple (R, G, B, A).
"""
self.config.background_color = color
self.color = self.convert_color(color)
self.parent_progress_widget.update()
# Only allow manual background color changes when colors are not linked
if self.config.link_colors:
return
self._background_color = self.convert_color(color)
self.config.background_color = self._background_color.name()
self.update()
def _auto_set_background_color(self):
"""
Automatically set the background color based on the main color and the current theme.
"""
palette = self.palette()
bg = palette.color(QtGui.QPalette.ColorRole.Window)
bg_color = Colors.subtle_background_color(self._color, bg)
self.config.background_color = bg_color.name()
self._background_color = bg_color
self.update()
def set_colors_linked(self, linked: bool):
"""
Set whether the colors are linked for the ring widget.
If colors are linked, changing the main color will also change the background color.
Args:
linked(bool): Whether to link the colors for the ring widget
"""
self.config.link_colors = linked
if linked:
self._auto_set_background_color()
self.update()
def set_line_width(self, width: int):
"""
@@ -169,7 +156,7 @@ class Ring(BECConnector, QObject):
width(int): Line width for the ring widget
"""
self.config.line_width = width
self.parent_progress_widget.update()
self.update()
def set_min_max_values(self, min_value: int | float, max_value: int | float):
"""
@@ -181,35 +168,21 @@ class Ring(BECConnector, QObject):
"""
self.config.min_value = min_value
self.config.max_value = max_value
self.parent_progress_widget.update()
self.update()
def set_start_angle(self, start_angle: int):
"""
Set the start angle for the ring widget
Set the start angle for the ring widget.
Args:
start_angle(int): Start angle for the ring widget in degrees
"""
self.config.start_position = start_angle
self.start_position = start_angle * 16
self.parent_progress_widget.update()
self.update()
@staticmethod
def convert_color(color):
"""
Convert the color to QColor
Args:
color(str | tuple): Color for the ring widget. Can be HEX code or tuple (R, G, B, A).
"""
converted_color = None
if isinstance(color, str):
converted_color = QtGui.QColor(color)
elif isinstance(color, tuple):
converted_color = QtGui.QColor(*color)
return converted_color
def set_update(self, mode: Literal["manual", "scan", "device"], device: str = None):
def set_update(
self, mode: Literal["manual", "scan", "device"], device: str = "", signal: str = ""
):
"""
Set the update mode for the ring widget.
Modes:
@@ -220,47 +193,167 @@ class Ring(BECConnector, QObject):
Args:
mode(str): Update mode for the ring widget. Can be "manual", "scan" or "device"
device(str): Device name for the device readback mode, only used when mode is "device"
signal(str): Signal name for the device readback mode, only used when mode is "device"
"""
if mode == "manual":
if self.config.connections.slot is not None:
self.bec_dispatcher.disconnect_slot(
getattr(self, self.config.connections.slot), self.config.connections.endpoint
match mode:
case "manual":
if self.config.mode == "manual":
return
if self.registered_slot is not None:
self.bec_dispatcher.disconnect_slot(*self.registered_slot)
self.config.mode = "manual"
self.registered_slot = None
case "scan":
if self.config.mode == "scan":
return
if self.registered_slot is not None:
self.bec_dispatcher.disconnect_slot(*self.registered_slot)
self.config.mode = "scan"
self.bec_dispatcher.connect_slot(
self.on_scan_progress, MessageEndpoints.scan_progress()
)
self.config.connections.slot = None
self.config.connections.endpoint = None
elif mode == "scan":
self.set_connections("on_scan_progress", MessageEndpoints.scan_progress())
elif mode == "device":
self.set_connections("on_device_readback", MessageEndpoints.device_readback(device))
self.registered_slot = (self.on_scan_progress, MessageEndpoints.scan_progress())
case "device":
if self.registered_slot is not None:
self.bec_dispatcher.disconnect_slot(*self.registered_slot)
self.config.mode = "device"
if device == "":
self.registered_slot = None
return
self.config.device = device
# self.config.signal = self._get_signal_from_device(device, signal)
signal = self._update_device_connection(device, signal)
self.config.signal = signal
self.parent_progress_widget.enable_auto_updates(False)
case _:
raise ValueError(f"Unsupported mode: {mode}")
def set_connections(self, slot: str, endpoint: str | EndpointInfo):
def set_precision(self, precision: int):
"""
Set the connections for the ring widget
Set the precision for the ring widget.
Args:
slot(str): Slot for the ring widget update. Can be "on_scan_progress" or "on_device_readback".
endpoint(str | EndpointInfo): Endpoint for the ring widget update. Endpoint has to match the slot type.
precision(int): Precision for the ring widget
"""
if self.config.connections.endpoint == endpoint and self.config.connections.slot == slot:
return
if self.config.connections.slot is not None:
self.bec_dispatcher.disconnect_slot(
getattr(self, self.config.connections.slot), self.config.connections.endpoint
)
self.config.connections = ProgressbarConnections(slot=slot, endpoint=endpoint)
self.bec_dispatcher.connect_slot(getattr(self, slot), endpoint)
self.config.precision = precision
self.update()
def reset_connection(self):
def set_direction(self, direction: int):
"""
Reset the connections for the ring widget. Disconnect the current slot and endpoint.
"""
self.bec_dispatcher.disconnect_slot(
self.config.connections.slot, self.config.connections.endpoint
)
self.config.connections = ProgressbarConnections()
Set the direction for the ring widget.
Args:
direction(int): Direction for the ring widget. -1 for clockwise, 1 for counter-clockwise.
"""
self.config.direction = direction
self.update()
def _get_signals_for_device(self, device: str) -> dict[str, list[str]]:
"""
Get the signals for the device.
Args:
device(str): Device name for the device
Returns:
dict[str, list[str]]: Dictionary with the signals for the device
"""
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.")
progress_signals = [
obj["component_name"]
for obj in dev_obj._info["signals"].values()
if obj["signal_class"] == "ProgressSignal"
]
hinted_signals = [
obj["obj_name"]
for obj in dev_obj._info["signals"].values()
if obj["kind_str"] == "hinted"
and obj["signal_class"]
not in ["ProgressSignal", "AyncSignal", "AsyncMultiSignal", "DynamicSignal"]
]
normal_signals = [
obj["component_name"]
for obj in dev_obj._info["signals"].values()
if obj["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:
"""
Update the device connection for the ring widget.
In general, we support two modes here:
- If signal is provided, we use that directly.
- If signal is not provided, we try to get the signal from the device manager.
We first check for progress signals, then for hinted signals, and finally for normal signals.
Depending on what type of signal we get (progress or hinted/normal), we subscribe to different endpoints.
Args:
device(str): Device name for the device mode
signal(str): Signal name for the device mode
Returns:
str: The selected signal name for the device mode
"""
logger.info(f"Updating device connection for device '{device}' and signal '{signal}'")
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:
return ""
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 signal is not provided, we try to get it from the device manager
if len(progress_signals) > 0:
signal = progress_signals[0]
logger.info(
f"Using progress signal '{signal}' for device '{device}' in ring progress bar."
)
elif len(hinted_signals) > 0:
signal = hinted_signals[0]
logger.info(
f"Using hinted signal '{signal}' for device '{device}' in ring progress bar."
)
elif len(normal_signals) > 0:
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
@SafeSlot(dict, dict)
def on_scan_progress(self, msg, meta):
"""
Update the ring widget with the scan progress.
@@ -273,8 +366,9 @@ class Ring(BECConnector, QObject):
if current_RID != self.RID:
self.set_min_max_values(0, msg.get("max_value", 100))
self.set_value(msg.get("value", 0))
self.parent_progress_widget.update()
self.update()
@SafeSlot(dict, dict)
def on_device_readback(self, msg, meta):
"""
Update the ring widget with the device readback.
@@ -283,11 +377,242 @@ class Ring(BECConnector, QObject):
msg(dict): Message with the device readback
meta(dict): Metadata for the message
"""
if isinstance(self.config.connections.endpoint, EndpointInfo):
endpoint = self.config.connections.endpoint.endpoint
else:
endpoint = self.config.connections.endpoint
device = endpoint.split("/")[-1]
value = msg.get("signals").get(device).get("value")
device = self.config.device
if device is None:
return
signal = self.config.signal or device
value = msg.get("signals", {}).get(signal, {}).get("value", None)
if value is None:
return
self.set_value(value)
self.parent_progress_widget.update()
self.update()
@SafeSlot(dict, dict)
def on_device_progress(self, msg, meta):
"""
Update the ring widget with the device progress.
Args:
msg(dict): Message with the device progress
meta(dict): Metadata for the message
"""
device = self.config.device
if device is None:
return
max_val = msg.get("max_value", 100)
self.set_min_max_values(0, max_val)
value = msg.get("value", 0)
if msg.get("done"):
value = max_val
self.set_value(value)
self.update()
def paintEvent(self, event):
if not self.progress_container:
return
painter = QtGui.QPainter(self)
painter.setRenderHint(QtGui.QPainter.RenderHint.Antialiasing)
size = min(self.width(), self.height())
# Center the ring
x_offset = (self.width() - size) // 2
y_offset = (self.height() - size) // 2
max_ring_size = self.progress_container.get_max_ring_size()
rect = QtCore.QRect(x_offset, y_offset, size, size)
rect.adjust(max_ring_size, max_ring_size, -max_ring_size, -max_ring_size)
# Background arc
painter.setPen(
QtGui.QPen(self._background_color, self.config.line_width, QtCore.Qt.PenStyle.SolidLine)
)
gap: int = self.gap # type: ignore
# Important: Qt uses a 16th of a degree for angles. start_position is therefore multiplied by 16.
start_position: float = self.config.start_position * 16 # type: ignore
adjusted_rect = QtCore.QRect(
rect.left() + gap, rect.top() + gap, rect.width() - 2 * gap, rect.height() - 2 * gap
)
painter.drawArc(adjusted_rect, start_position, 360 * 16)
# Foreground arc
pen = QtGui.QPen(self.color, self.config.line_width, QtCore.Qt.PenStyle.SolidLine)
pen.setCapStyle(QtCore.Qt.PenCapStyle.RoundCap)
painter.setPen(pen)
proportion = (self.config.value - self.config.min_value) / (
(self.config.max_value - self.config.min_value) + 1e-3
)
angle = int(proportion * 360 * 16 * self.config.direction)
painter.drawArc(adjusted_rect, start_position, angle)
painter.end()
def convert_color(self, color: str | tuple | QColor) -> QColor:
"""
Convert the color to QColor
Args:
color(str | tuple | QColor): Color for the ring widget. Can be HEX code or tuple (R, G, B, A) or QColor.
"""
if isinstance(color, QColor):
return color
if isinstance(color, str):
return QtGui.QColor(color)
if isinstance(color, (tuple, list)):
return QtGui.QColor(*color)
raise ValueError(f"Unsupported color format: {color}")
def cleanup(self):
"""
Cleanup the ring widget.
Disconnect any registered slots.
"""
if self.registered_slot is not None:
self.bec_dispatcher.disconnect_slot(*self.registered_slot)
self.registered_slot = None
###############################################
####### QProperties ###########################
###############################################
@SafeProperty(int)
def gap(self) -> int:
return self._gap
@gap.setter
def gap(self, value: int):
self._gap = value
self.update()
@SafeProperty(bool)
def link_colors(self) -> bool:
return self.config.link_colors
@link_colors.setter
def link_colors(self, value: bool):
logger.info(f"Setting link_colors to {value}")
self.set_colors_linked(value)
@SafeProperty(QColor)
def color(self) -> QColor:
return self._color
@color.setter
def color(self, value: QColor):
self.set_color(value)
@SafeProperty(QColor)
def background_color(self) -> QColor:
return self._background_color
@background_color.setter
def background_color(self, value: QColor):
self.set_background(value)
@SafeProperty(float)
def value(self) -> float:
return self.config.value
@value.setter
def value(self, value: float):
self.config.value = round(
float(max(self.config.min_value, min(self.config.max_value, value))),
self.config.precision,
)
self.update()
@SafeProperty(float)
def min_value(self) -> float:
return self.config.min_value
@min_value.setter
def min_value(self, value: float):
self.config.min_value = value
self.update()
@SafeProperty(float)
def max_value(self) -> float:
return self.config.max_value
@max_value.setter
def max_value(self, value: float):
self.config.max_value = value
self.update()
@SafeProperty(str)
def mode(self) -> str:
return self.config.mode
@mode.setter
def mode(self, value: str):
self.set_update(value)
@SafeProperty(str)
def device(self) -> str:
return self.config.device or ""
@device.setter
def device(self, value: str):
self.config.device = value
@SafeProperty(str)
def signal(self) -> str:
return self.config.signal or ""
@signal.setter
def signal(self, value: str):
self.config.signal = value
@SafeProperty(int)
def line_width(self) -> int:
return self.config.line_width
@line_width.setter
def line_width(self, value: int):
self.config.line_width = value
self.update()
@SafeProperty(int)
def start_position(self) -> int:
return self.config.start_position
@start_position.setter
def start_position(self, value: int):
self.config.start_position = value
self.update()
@SafeProperty(int)
def precision(self) -> int:
return self.config.precision
@precision.setter
def precision(self, value: int):
self.config.precision = value
self.update()
@SafeProperty(int)
def direction(self) -> int:
return self.config.direction
@direction.setter
def direction(self, value: int):
self.config.direction = value
self.update()
if __name__ == "__main__": # pragma: no cover
import sys
from qtpy.QtWidgets import QApplication
from bec_widgets.utils.colors import apply_theme
app = QApplication(sys.argv)
apply_theme("dark")
ring = Ring()
ring.export_settings()
ring.show()
sys.exit(app.exec())

892
bec_widgets/widgets/progress/ring_progress_bar/ring_progress_bar.py
View File

@@ -1,348 +1,154 @@
from __future__ import annotations
from typing import Literal, Optional
import json
from typing import Literal
import pyqtgraph as pg
from bec_lib.endpoints import MessageEndpoints
from bec_lib.logger import bec_logger
from pydantic import Field, field_validator
from pydantic_core import PydanticCustomError
from qtpy import QtCore, QtGui
from qtpy.QtCore import QSize, Slot
from qtpy.QtWidgets import QSizePolicy, QWidget
from qtpy.QtCore import QSize, Qt
from qtpy.QtWidgets import QHBoxLayout, QLabel, QVBoxLayout, QWidget
from bec_widgets.utils import Colors, ConnectionConfig, EntryValidator
from bec_widgets.utils import Colors
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.widgets.progress.ring_progress_bar.ring import Ring, RingConfig
from bec_widgets.utils.error_popups import SafeProperty
from bec_widgets.utils.settings_dialog import SettingsDialog
from bec_widgets.utils.toolbars.actions import MaterialIconAction
from bec_widgets.utils.toolbars.toolbar import ModularToolBar
from bec_widgets.widgets.progress.ring_progress_bar.ring import Ring
from bec_widgets.widgets.progress.ring_progress_bar.ring_progress_settings_cards import RingSettings
logger = bec_logger.logger
class RingProgressBarConfig(ConnectionConfig):
color_map: Optional[str] = Field(
"plasma", description="Color scheme for the progress bars.", validate_default=True
)
min_number_of_bars: int = Field(1, description="Minimum number of progress bars to display.")
max_number_of_bars: int = Field(10, description="Maximum number of progress bars to display.")
num_bars: int = Field(1, description="Number of progress bars to display.")
gap: int | None = Field(20, description="Gap between progress bars.")
auto_updates: bool | None = Field(
True, description="Enable or disable updates based on scan queue status."
)
rings: list[RingConfig] | None = Field([], description="List of ring configurations.")
@field_validator("num_bars")
@classmethod
def validate_num_bars(cls, v, values):
min_number_of_bars = values.data.get("min_number_of_bars", None)
max_number_of_bars = values.data.get("max_number_of_bars", None)
if min_number_of_bars is not None and max_number_of_bars is not None:
logger.info(
f"Number of bars adjusted to be between defined min:{min_number_of_bars} and max:{max_number_of_bars} number of bars."
)
v = max(min_number_of_bars, min(v, max_number_of_bars))
return v
@field_validator("rings")
@classmethod
def validate_rings(cls, v, values):
if v is not None and v is not []:
num_bars = values.data.get("num_bars", None)
if len(v) != num_bars:
raise PydanticCustomError(
"different number of configs",
f"Length of rings configuration ({len(v)}) does not match the number of bars ({num_bars}).",
{"wrong_value": len(v)},
)
indices = [ring.index for ring in v]
if sorted(indices) != list(range(len(indices))):
raise PydanticCustomError(
"wrong indices",
f"Indices of ring configurations must be unique and in order from 0 to num_bars {num_bars}.",
{"wrong_value": indices},
)
return v
_validate_colormap = field_validator("color_map")(Colors.validate_color_map)
class RingProgressBar(BECWidget, QWidget):
class RingProgressContainerWidget(QWidget):
"""
Show the progress of devices, scans or custom values in the form of ring progress bars.
A container widget for the Ring Progress Bar widget.
It holds the rings and manages their layout and painting.
"""
PLUGIN = True
ICON_NAME = "track_changes"
USER_ACCESS = [
"_get_all_rpc",
"_rpc_id",
"_config_dict",
"rings",
"update_config",
"add_ring",
"remove_ring",
"set_precision",
"set_min_max_values",
"set_number_of_bars",
"set_value",
"set_colors_from_map",
"set_colors_directly",
"set_line_widths",
"set_gap",
"set_diameter",
"reset_diameter",
"enable_auto_updates",
"attach",
"detach",
"screenshot",
]
def __init__(
self,
parent=None,
config: RingProgressBarConfig | dict | None = None,
client=None,
gui_id: str | None = None,
num_bars: int | None = None,
**kwargs,
):
if config is None:
config = RingProgressBarConfig(widget_class=self.__class__.__name__)
self.config = config
else:
if isinstance(config, dict):
config = RingProgressBarConfig(**config, widget_class=self.__class__.__name__)
self.config = config
super().__init__(parent=parent, client=client, gui_id=gui_id, config=config, **kwargs)
self.get_bec_shortcuts()
self.entry_validator = EntryValidator(self.dev)
self.RID = None
# For updating bar behaviour
self._auto_updates = True
self._rings = []
if num_bars is not None:
self.config.num_bars = max(
self.config.min_number_of_bars, min(num_bars, self.config.max_number_of_bars)
)
def __init__(self, parent: QWidget | None = None, **kwargs):
super().__init__(parent=parent, **kwargs)
self.rings: list[Ring] = []
self.gap = 20 # Gap between rings
self.color_map: str = "turbo"
self.setLayout(QHBoxLayout())
self.initialize_bars()
self.enable_auto_updates(self.config.auto_updates)
self.initialize_center_label()
@property
def rings(self) -> list[Ring]:
"""Returns a list of all rings in the progress bar."""
return self._rings
@rings.setter
def rings(self, value: list[Ring]):
self._rings = value
def update_config(self, config: RingProgressBarConfig | dict):
"""
Update the configuration of the widget.
Args:
config(SpiralProgressBarConfig|dict): Configuration to update.
"""
if isinstance(config, dict):
config = RingProgressBarConfig(**config, widget_class=self.__class__.__name__)
self.config = config
self.clear_all()
def num_bars(self) -> int:
return len(self.rings)
def initialize_bars(self):
"""
Initialize the progress bars.
"""
start_positions = [90 * 16] * self.config.num_bars
directions = [-1] * self.config.num_bars
for _ in range(self.num_bars):
self.add_ring()
self.config.rings = [
RingConfig(
widget_class="Ring",
index=i,
start_positions=start_positions[i],
directions=directions[i],
)
for i in range(self.config.num_bars)
]
self._rings = [Ring(parent=self, config=config) for config in self.config.rings]
if self.color_map:
self.set_colors_from_map(self.color_map)
if self.config.color_map:
self.set_colors_from_map(self.config.color_map)
min_size = self._calculate_minimum_size()
self.setMinimumSize(min_size)
# Set outer ring to listen to scan progress
self.rings[0].set_update(mode="scan")
self.update()
def add_ring(self, **kwargs) -> Ring:
def add_ring(self, config: dict | None = None) -> Ring:
"""
Add a new progress bar.
Add a new ring to the container.
Args:
**kwargs: Keyword arguments for the new progress bar.
config(dict | None): Optional configuration dictionary for the ring.
Returns:
Ring: Ring object.
Ring: The newly added ring object.
"""
if self.config.num_bars < self.config.max_number_of_bars:
ring_index = self.config.num_bars
ring_config = RingConfig(
widget_class="Ring",
index=ring_index,
start_positions=90 * 16,
directions=-1,
**kwargs,
)
ring = Ring(parent=self, config=ring_config)
self.config.num_bars += 1
self._rings.append(ring)
self.config.rings.append(ring.config)
if self.config.color_map:
self.set_colors_from_map(self.config.color_map)
base_line_width = self._rings[ring.config.index].config.line_width
self.set_line_widths(base_line_width, ring.config.index)
self.update()
return ring
ring = Ring(parent=self)
ring.setGeometry(self.rect())
ring.gap = self.gap * len(self.rings)
ring.set_value(0)
self.rings.append(ring)
if config:
# We have to first get the link_colors property before loading the settings
# While this is an ugly hack, we do not have control over the order of properties
# being set when loading.
ring.link_colors = config.pop("link_colors", True)
ring.load_settings(config)
if self.color_map:
self.set_colors_from_map(self.color_map)
ring.show()
ring.raise_()
self.update()
return ring
def remove_ring(self, index: int):
def remove_ring(self, index: int | None = None):
"""
Remove a progress bar by index.
Remove a ring from the container.
Args:
index(int): Index of the progress bar to remove.
index(int | None): Index of the ring to remove. If None, removes the last ring.
"""
ring = self._find_ring_by_index(index)
self._cleanup_ring(ring)
self.update()
def _cleanup_ring(self, ring: Ring) -> None:
ring.reset_connection()
self._rings.remove(ring)
self.config.rings.remove(ring.config)
self.config.num_bars -= 1
self._reindex_rings()
if self.config.color_map:
self.set_colors_from_map(self.config.color_map)
# Remove ring from rpc, afterwards call close event.
ring.rpc_register.remove_rpc(ring)
if self.num_bars == 0:
return
if index is None:
index = self.num_bars - 1
index = self._validate_index(index)
ring = self.rings[index]
ring.cleanup()
ring.close()
ring.deleteLater()
# del ring
def _reindex_rings(self):
"""
Reindex the progress bars.
"""
for i, ring in enumerate(self._rings):
ring.config.index = i
def set_precision(self, precision: int, bar_index: int | None = None):
"""
Set the precision for the progress bars. If bar_index is not provide, the precision will be set for all progress bars.
Args:
precision(int): Precision for the progress bars.
bar_index(int): Index of the progress bar to set the precision for. If provided, only a single precision can be set.
"""
if bar_index is not None:
bar_index = self._bar_index_check(bar_index)
ring = self._find_ring_by_index(bar_index)
ring.config.precision = precision
else:
for ring in self._rings:
ring.config.precision = precision
self.rings.pop(index)
# Update gaps for remaining rings
for i, r in enumerate(self.rings):
r.gap = self.gap * i
self.update()
def set_min_max_values(
self,
min_values: int | float | list[int | float],
max_values: int | float | list[int | float],
):
def initialize_center_label(self):
"""
Set the minimum and maximum values for the progress bars.
Args:
min_values(int|float | list[float]): Minimum value(s) for the progress bars. If multiple progress bars are displayed, provide a list of minimum values for each progress bar.
max_values(int|float | list[float]): Maximum value(s) for the progress bars. If multiple progress bars are displayed, provide a list of maximum values for each progress bar.
Initialize the center label.
"""
if isinstance(min_values, (int, float)):
min_values = [min_values]
if isinstance(max_values, (int, float)):
max_values = [max_values]
min_values = self._adjust_list_to_bars(min_values)
max_values = self._adjust_list_to_bars(max_values)
for ring, min_value, max_value in zip(self._rings, min_values, max_values):
ring.set_min_max_values(min_value, max_value)
self.update()
layout = self.layout()
layout.setContentsMargins(0, 0, 0, 0)
def set_number_of_bars(self, num_bars: int):
self.center_label = QLabel("", parent=self)
self.center_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
layout.addWidget(self.center_label)
def _calculate_minimum_size(self):
"""
Set the number of progress bars to display.
Args:
num_bars(int): Number of progress bars to display.
Calculate the minimum size of the widget.
"""
num_bars = max(
self.config.min_number_of_bars, min(num_bars, self.config.max_number_of_bars)
)
current_num_bars = self.config.num_bars
if not self.rings:
return QSize(10, 10)
ring_widths = self.get_ring_line_widths()
total_width = sum(ring_widths) + self.gap * (self.num_bars - 1)
diameter = max(total_width * 2, 50)
if num_bars > current_num_bars:
for i in range(current_num_bars, num_bars):
new_ring_config = RingConfig(
widget_class="Ring", index=i, start_positions=90 * 16, directions=-1
)
self.config.rings.append(new_ring_config)
new_ring = Ring(parent=self, config=new_ring_config)
self._rings.append(new_ring)
return QSize(diameter, diameter)
elif num_bars < current_num_bars:
for i in range(current_num_bars - 1, num_bars - 1, -1):
self.remove_ring(i)
self.config.num_bars = num_bars
if self.config.color_map:
self.set_colors_from_map(self.config.color_map)
base_line_width = self._rings[0].config.line_width
self.set_line_widths(base_line_width)
self.update()
def set_value(self, values: int | list, ring_index: int = None):
def get_ring_line_widths(self):
"""
Set the values for the progress bars.
Args:
values(int | tuple): Value(s) for the progress bars. If multiple progress bars are displayed, provide a tuple of values for each progress bar.
ring_index(int): Index of the progress bar to set the value for. If provided, only a single value can be set.
Examples:
>>> SpiralProgressBar.set_value(50)
>>> SpiralProgressBar.set_value([30, 40, 50]) # (outer, middle, inner)
>>> SpiralProgressBar.set_value(60, bar_index=1) # Set the value for the middle progress bar.
Get the line widths of the rings.
"""
if ring_index is not None:
ring = self._find_ring_by_index(ring_index)
if isinstance(values, list):
values = values[0]
logger.warning(
f"Warning: Only a single value can be set for a single progress bar. Using the first value in the list {values}"
)
ring.set_value(values)
else:
if isinstance(values, int):
values = [values]
values = self._adjust_list_to_bars(values)
for ring, value in zip(self._rings, values):
ring.set_value(value)
self.update()
if not self.rings:
return [10]
ring_widths = [ring.config.line_width for ring in self.rings]
return ring_widths
def get_max_ring_size(self) -> int:
"""
Get the size of the rings.
"""
if not self.rings:
return 10
ring_widths = self.get_ring_line_widths()
return max(ring_widths)
def sizeHint(self):
min_size = self._calculate_minimum_size()
return min_size
def resizeEvent(self, event):
"""
Handle resize events to update ring geometries.
"""
super().resizeEvent(event)
for ring in self.rings:
ring.setGeometry(self.rect())
def set_colors_from_map(self, colormap, color_format: Literal["RGB", "HEX"] = "RGB"):
"""
@@ -356,12 +162,14 @@ class RingProgressBar(BECWidget, QWidget):
raise ValueError(
f"Colormap '{colormap}' not found in the current installation of pyqtgraph"
)
colors = Colors.golden_angle_color(colormap, self.config.num_bars, color_format)
colors = Colors.golden_angle_color(colormap, self.num_bars, color_format)
self.set_colors_directly(colors)
self.config.color_map = colormap
self.color_map = colormap
self.update()
def set_colors_directly(self, colors: list[str | tuple] | str | tuple, bar_index: int = None):
def set_colors_directly(
self, colors: list[str | tuple] | str | tuple, bar_index: int | None = None
):
"""
Set the colors for the progress bars directly.
@@ -370,170 +178,16 @@ class RingProgressBar(BECWidget, QWidget):
bar_index(int): Index of the progress bar to set the color for. If provided, only a single color can be set.
"""
if bar_index is not None and isinstance(colors, (str, tuple)):
bar_index = self._bar_index_check(bar_index)
ring = self._find_ring_by_index(bar_index)
ring.set_color(colors)
bar_index = self._validate_index(bar_index)
self.rings[bar_index].set_color(colors)
else:
if isinstance(colors, (str, tuple)):
colors = [colors]
colors = self._adjust_list_to_bars(colors)
for ring, color in zip(self._rings, colors):
for ring, color in zip(self.rings, colors):
ring.set_color(color)
self.update()
def set_line_widths(self, widths: int | list[int], bar_index: int = None):
"""
Set the line widths for the progress bars.
Args:
widths(int | list[int]): Line width(s) for the progress bars. If multiple progress bars are displayed, provide a list of line widths for each progress bar.
bar_index(int): Index of the progress bar to set the line width for. If provided, only a single line width can be set.
"""
if bar_index is not None:
bar_index = self._bar_index_check(bar_index)
ring = self._find_ring_by_index(bar_index)
if isinstance(widths, list):
widths = widths[0]
logger.warning(
f"Warning: Only a single line width can be set for a single progress bar. Using the first value in the list {widths}"
)
ring.set_line_width(widths)
else:
if isinstance(widths, int):
widths = [widths]
widths = self._adjust_list_to_bars(widths)
self.config.gap = max(widths) * 2
for ring, width in zip(self._rings, widths):
ring.set_line_width(width)
min_size = self._calculate_minimum_size()
self.setMinimumSize(min_size)
self.update()
def set_gap(self, gap: int):
"""
Set the gap between the progress bars.
Args:
gap(int): Gap between the progress bars.
"""
self.config.gap = gap
self.update()
def set_diameter(self, diameter: int):
"""
Set the diameter of the widget.
Args:
diameter(int): Diameter of the widget.
"""
size = QSize(diameter, diameter)
self.resize(size)
self.setFixedSize(size)
def _find_ring_by_index(self, index: int) -> Ring:
"""
Find the ring by index.
Args:
index(int): Index of the ring.
Returns:
Ring: Ring object.
"""
for ring in self._rings:
if ring.config.index == index:
return ring
raise ValueError(f"Ring with index {index} not found.")
def enable_auto_updates(self, enable: bool = True):
"""
Enable or disable updates based on scan status. Overrides manual updates.
The behaviour of the whole progress bar widget will be driven by the scan queue status.
Args:
enable(bool): True or False.
Returns:
bool: True if scan segment updates are enabled.
"""
self._auto_updates = enable
if enable is True:
self.bec_dispatcher.connect_slot(
self.on_scan_queue_status, MessageEndpoints.scan_queue_status()
)
else:
self.bec_dispatcher.disconnect_slot(
self.on_scan_queue_status, MessageEndpoints.scan_queue_status()
)
return self._auto_updates
@Slot(dict, dict)
def on_scan_queue_status(self, msg, meta):
"""
Slot to handle scan queue status messages. Decides what update to perform based on the scan queue status.
Args:
msg(dict): Message from the BEC.
meta(dict): Metadata from the BEC.
"""
primary_queue = msg.get("queue").get("primary")
info = primary_queue.get("info", None)
if not info:
return
active_request_block = info[0].get("active_request_block", None)
if not active_request_block:
return
report_instructions = active_request_block.get("report_instructions", None)
if not report_instructions:
return
instruction_type = list(report_instructions[0].keys())[0]
if instruction_type == "scan_progress":
self._hook_scan_progress(ring_index=0)
elif instruction_type == "readback":
devices = report_instructions[0].get("readback").get("devices")
start = report_instructions[0].get("readback").get("start")
end = report_instructions[0].get("readback").get("end")
if self.config.num_bars != len(devices):
self.set_number_of_bars(len(devices))
for index, device in enumerate(devices):
self._hook_readback(index, device, start[index], end[index])
else:
logger.error(f"{instruction_type} not supported yet.")
def _hook_scan_progress(self, ring_index: int | None = None):
"""
Hook the scan progress to the progress bars.
Args:
ring_index(int): Index of the progress bar to hook the scan progress to.
"""
if ring_index is not None:
ring = self._find_ring_by_index(ring_index)
else:
ring = self._rings[0]
if ring.config.connections.slot == "on_scan_progress":
return
ring.set_connections("on_scan_progress", MessageEndpoints.scan_progress())
def _hook_readback(self, bar_index: int, device: str, min: float | int, max: float | int):
"""
Hook the readback values to the progress bars.
Args:
bar_index(int): Index of the progress bar to hook the readback values to.
device(str): Device to readback values from.
min(float|int): Minimum value for the progress bar.
max(float|int): Maximum value for the progress bar.
"""
ring = self._find_ring_by_index(bar_index)
ring.set_min_max_values(min, max)
endpoint = MessageEndpoints.device_readback(device)
ring.set_connections("on_device_readback", endpoint)
def _adjust_list_to_bars(self, items: list) -> list:
"""
Utility method to adjust the list of parameters to match the number of progress bars.
@@ -550,101 +204,249 @@ class RingProgressBar(BECWidget, QWidget):
)
if not isinstance(items, list):
items = [items]
if len(items) < self.config.num_bars:
if len(items) < self.num_bars:
last_item = items[-1]
items.extend([last_item] * (self.config.num_bars - len(items)))
elif len(items) > self.config.num_bars:
items = items[: self.config.num_bars]
items.extend([last_item] * (self.num_bars - len(items)))
elif len(items) > self.num_bars:
items = items[: self.num_bars]
return items
def _bar_index_check(self, bar_index: int):
def _validate_index(self, index: int) -> int:
"""
Utility method to check if the bar index is within the range of the number of progress bars.
Check if the provided index is valid for the number of bars.
Args:
bar_index(int): Index of the progress bar to set the value for.
index(int): Index to check.
Returns:
int: Validated index.
"""
if not (0 <= bar_index < self.config.num_bars):
raise ValueError(
f"bar_index {bar_index} out of range of number of bars {self.config.num_bars}."
)
return bar_index
def paintEvent(self, event):
if not self._rings:
return
painter = QtGui.QPainter(self)
painter.setRenderHint(QtGui.QPainter.Antialiasing)
size = min(self.width(), self.height())
rect = QtCore.QRect(0, 0, size, size)
rect.adjust(
max(ring.config.line_width for ring in self._rings),
max(ring.config.line_width for ring in self._rings),
-max(ring.config.line_width for ring in self._rings),
-max(ring.config.line_width for ring in self._rings),
)
for i, ring in enumerate(self._rings):
# Background arc
painter.setPen(
QtGui.QPen(ring.background_color, ring.config.line_width, QtCore.Qt.SolidLine)
)
offset = self.config.gap * i
adjusted_rect = QtCore.QRect(
rect.left() + offset,
rect.top() + offset,
rect.width() - 2 * offset,
rect.height() - 2 * offset,
)
painter.drawArc(adjusted_rect, ring.config.start_position, 360 * 16)
# Foreground arc
pen = QtGui.QPen(ring.color, ring.config.line_width, QtCore.Qt.SolidLine)
pen.setCapStyle(QtCore.Qt.RoundCap)
painter.setPen(pen)
proportion = (ring.config.value - ring.config.min_value) / (
(ring.config.max_value - ring.config.min_value) + 1e-3
)
angle = int(proportion * 360 * 16 * ring.config.direction)
painter.drawArc(adjusted_rect, ring.start_position, angle)
def reset_diameter(self):
"""
Reset the fixed size of the widget.
"""
self.setSizePolicy(QSizePolicy.Preferred, QSizePolicy.Preferred)
self.setMinimumSize(self._calculate_minimum_size())
self.setMaximumSize(16777215, 16777215)
def _calculate_minimum_size(self):
"""
Calculate the minimum size of the widget.
"""
if not self.config.rings:
logger.warning("no rings to get size from setting size to 10x10")
return QSize(10, 10)
ring_widths = [self.config.rings[i].line_width for i in range(self.config.num_bars)]
total_width = sum(ring_widths) + self.config.gap * (self.config.num_bars - 1)
diameter = max(total_width * 2, 50)
return QSize(diameter, diameter)
def sizeHint(self):
min_size = self._calculate_minimum_size()
return min_size
try:
self.rings[index]
except IndexError:
raise IndexError(f"Index {index} is out of range for {self.num_bars} rings.")
return index
def clear_all(self):
for ring in self._rings:
ring.reset_connection()
self._rings.clear()
"""
Clear all rings from the widget.
"""
for ring in self.rings:
ring.close()
ring.deleteLater()
self.rings = []
self.update()
self.initialize_bars()
class RingProgressBar(BECWidget, QWidget):
ICON_NAME = "track_changes"
PLUGIN = True
RPC = True
USER_ACCESS = [
*BECWidget.USER_ACCESS,
"screenshot",
"rings",
"add_ring",
"remove_ring",
"set_gap",
"set_center_label",
]
def __init__(self, parent: QWidget | None = None, client=None, **kwargs):
super().__init__(parent=parent, client=client, theme_update=True, **kwargs)
self.setWindowTitle("Ring Progress Bar")
self.layout = QVBoxLayout(self)
self.layout.setContentsMargins(0, 0, 0, 0)
self.setLayout(self.layout)
self.toolbar = ModularToolBar(self)
self._init_toolbar()
self.layout.addWidget(self.toolbar)
# Placeholder for the actual ring progress bar widget
self.ring_progress_bar = RingProgressContainerWidget(self)
self.layout.addWidget(self.ring_progress_bar)
self.settings_dialog = None
self.toolbar.show_bundles(["rpb_settings"])
def apply_theme(self, theme: str):
super().apply_theme(theme)
if self.ring_progress_bar.color_map:
self.ring_progress_bar.set_colors_from_map(self.ring_progress_bar.color_map)
def _init_toolbar(self):
settings_action = MaterialIconAction(
icon_name="settings",
tooltip="Show Ring Progress Bar Settings",
checkable=True,
parent=self,
)
self.toolbar.add_action("rpb_settings", settings_action)
settings_action.action.triggered.connect(self._open_settings_dialog)
def _open_settings_dialog(self):
""" "
Open the settings dialog for the ring progress bar.
"""
settings_action = self.toolbar.components.get_action("rpb_settings").action
if self.settings_dialog is None or not self.settings_dialog.isVisible():
settings = RingSettings(parent=self, target_widget=self, popup=True)
self.settings_dialog = SettingsDialog(
self,
settings_widget=settings,
window_title="Ring Progress Bar Settings",
modal=False,
)
self.settings_dialog.resize(900, 500)
self.settings_dialog.finished.connect(self._settings_dialog_closed)
self.settings_dialog.show()
settings_action.setChecked(True)
else:
# Dialog is already open, raise it
self.settings_dialog.raise_()
self.settings_dialog.activateWindow()
settings_action.setChecked(True)
def _settings_dialog_closed(self):
"""
Handle the settings dialog being closed.
"""
settings_action = self.toolbar.components.get_action("rpb_settings").action
settings_action.setChecked(False)
self.settings_dialog = None
#################################################
###### RPC User Access Methods ##################
#################################################
def add_ring(self, config: dict | None = None) -> Ring:
"""
Add a new ring to the ring progress bar.
Optionally, a configuration dictionary can be provided but the ring
can also be configured later. The config dictionary must provide
the qproperties of the Qt Ring object.
Args:
config(dict | None): Optional configuration dictionary for the ring.
Returns:
Ring: The newly added ring object.
"""
return self.ring_progress_bar.add_ring(config=config)
def remove_ring(self, index: int | None = None):
"""
Remove a ring from the ring progress bar.
Args:
index(int | None): Index of the ring to remove. If None, removes the last ring.
"""
if self.ring_progress_bar.num_bars == 0:
return
self.ring_progress_bar.remove_ring(index=index)
def set_gap(self, value: int):
"""
Set the gap between rings.
Args:
value(int): Gap value in pixels.
"""
self.gap = value
def set_center_label(self, text: str):
"""
Set the center label text.
Args:
text(str): Text for the center label.
"""
self.center_label = text
@property
def rings(self) -> list[Ring]:
return self.ring_progress_bar.rings
###############################################
####### QProperties ###########################
###############################################
@SafeProperty(int)
def gap(self) -> int:
return self.ring_progress_bar.gap
@gap.setter
def gap(self, value: int):
self.ring_progress_bar.gap = value
self.ring_progress_bar.update()
@SafeProperty(str)
def color_map(self) -> str:
return self.ring_progress_bar.color_map or ""
@color_map.setter
def color_map(self, colormap: str):
if colormap == "":
self.ring_progress_bar.color_map = ""
return
if colormap not in pg.colormap.listMaps():
return
self.ring_progress_bar.set_colors_from_map(colormap)
self.ring_progress_bar.color_map = colormap
@SafeProperty(str)
def center_label(self) -> str:
return self.ring_progress_bar.center_label.text()
@center_label.setter
def center_label(self, text: str):
self.ring_progress_bar.center_label.setText(text)
@SafeProperty(str, designable=False, popup_error=True)
def ring_json(self) -> str:
"""
A JSON string property that serializes all ring pydantic configs.
"""
raw_list = []
for ring in self.rings:
cfg_dict = ring.config.model_dump()
raw_list.append(cfg_dict)
return json.dumps(raw_list, indent=2)
@ring_json.setter
def ring_json(self, json_data: str):
"""
Load rings from a JSON string and add them to the ring progress bar.
"""
try:
ring_configs = json.loads(json_data)
self.ring_progress_bar.clear_all()
for cfg_dict in ring_configs:
self.add_ring(config=cfg_dict)
except json.JSONDecodeError as e:
logger.error(f"Failed to decode JSON: {e}")
def cleanup(self):
self.bec_dispatcher.disconnect_slot(
self.on_scan_queue_status, MessageEndpoints.scan_queue_status()
)
for ring in self._rings:
self._cleanup_ring(ring)
self._rings.clear()
self.ring_progress_bar.clear_all()
self.ring_progress_bar.close()
self.ring_progress_bar.deleteLater()
super().cleanup()
if __name__ == "__main__": # pragma: no cover
import sys
from qtpy.QtWidgets import QApplication
from bec_widgets.utils.colors import apply_theme
app = QApplication(sys.argv)
apply_theme("dark")
widget = RingProgressBar()
widget.show()
sys.exit(app.exec_())

2
bec_widgets/widgets/progress/ring_progress_bar/ring_progress_bar_plugin.py
View File

@@ -51,7 +51,7 @@ class RingProgressBarPlugin(QDesignerCustomWidgetInterface): # pragma: no cover
return "RingProgressBar"
def toolTip(self):
return ""
return "RingProgressBar"
def whatsThis(self):
return self.toolTip()

509
bec_widgets/widgets/progress/ring_progress_bar/ring_progress_settings_cards.py Normal file
View File

@@ -0,0 +1,509 @@
from __future__ import annotations
import os
from typing import TYPE_CHECKING
from bec_qthemes._icon.material_icons import material_icon
from qtpy.QtCore import QSize
from qtpy.QtGui import QColor
from qtpy.QtWidgets import (
QApplication,
QComboBox,
QFrame,
QHBoxLayout,
QLabel,
QLineEdit,
QPushButton,
QScrollArea,
QVBoxLayout,
QWidget,
)
from bec_widgets.utils import UILoader
from bec_widgets.utils.error_popups import SafeSlot
from bec_widgets.utils.settings_dialog import SettingWidget
from bec_widgets.widgets.progress.ring_progress_bar.ring import Ring
from bec_widgets.widgets.utility.visual.colormap_widget.colormap_widget import BECColorMapWidget
if TYPE_CHECKING: # pragma: no cover
from bec_widgets.widgets.progress.ring_progress_bar.ring_progress_bar import (
RingProgressBar,
RingProgressContainerWidget,
)
class RingCardWidget(QFrame):
def __init__(self, ring: Ring, container: RingProgressContainerWidget, parent=None):
super().__init__(parent)
self.ring = ring
self.container = container
self.details_visible = False
self.setProperty("skip_settings", True)
self.setFrameShape(QFrame.Shape.StyledPanel)
self.setObjectName("RingCardWidget")
bg = self._get_theme_color("BORDER")
self.setStyleSheet(
f"""
#RingCardWidget {{
border: 1px solid {bg.name() if bg else '#CCCCCC'};
border-radius: 4px;
}}
"""
)
layout = QVBoxLayout(self)
layout.setContentsMargins(8, 8, 8, 8)
layout.setSpacing(6)
self._init_header(layout)
self._init_details(layout)
self._init_values()
self._connect_signals()
self.mode_combo.setCurrentText(self._get_display_mode_string(self.ring.config.mode))
self._set_widget_mode_enabled(self.ring.config.mode)
def _get_theme_color(self, color_name: str) -> QColor | None:
app = QApplication.instance()
if not app:
return
if not app.theme:
return
return app.theme.color(color_name)
def _init_header(self, parent_layout: QVBoxLayout):
"""Create the collapsible header with basic controls"""
header = QWidget()
layout = QHBoxLayout(header)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(8)
self.expand_btn = QPushButton("▶")
self.expand_btn.setFixedWidth(24)
self.expand_btn.clicked.connect(self.toggle_details)
self.mode_combo = QComboBox()
self.mode_combo.addItems(["Manual", "Scan Progress", "Device Readback"])
self.mode_combo.currentTextChanged.connect(self._update_mode)
delete_btn = QPushButton(material_icon("delete"), "")
color = self._get_theme_color("ACCENT_HIGHLIGHT")
delete_btn.setStyleSheet(f"background-color: {color.name() if color else '#CC181E'}")
delete_btn.clicked.connect(self._delete_self)
layout.addWidget(self.expand_btn)
layout.addWidget(QLabel("Mode"))
layout.addWidget(self.mode_combo)
layout.addStretch()
layout.addWidget(delete_btn)
parent_layout.addWidget(header)
def _init_details(self, parent_layout: QVBoxLayout):
"""Create the collapsible details area with the UI file"""
self.details = QWidget()
self.details.setVisible(False)
details_layout = QVBoxLayout(self.details)
details_layout.setContentsMargins(0, 0, 0, 0)
# Load UI file into details area
current_path = os.path.dirname(__file__)
self.ui = UILoader().load_ui(os.path.join(current_path, "ring_settings.ui"), self.details)
details_layout.addWidget(self.ui)
parent_layout.addWidget(self.details)
def toggle_details(self):
"""Toggle visibility of the details area"""
self.details_visible = not self.details_visible
self.details.setVisible(self.details_visible)
self.expand_btn.setText("▼" if self.details_visible else "▶")
# --------------------------------------------------------
def _connect_signals(self):
"""Connect UI signals to ring methods"""
# Data connections
self.ui.value_spin_box.valueChanged.connect(self.ring.set_value)
self.ui.min_spin_box.valueChanged.connect(self._update_min_max)
self.ui.max_spin_box.valueChanged.connect(self._update_min_max)
# Config connections
self.ui.start_angle_spin_box.valueChanged.connect(self.ring.set_start_angle)
self.ui.direction_combo_box.currentIndexChanged.connect(self._update_direction)
self.ui.line_width_spin_box.valueChanged.connect(self.ring.set_line_width)
self.ui.background_color_button.color_changed.connect(self.ring.set_background)
self.ui.ring_color_button.color_changed.connect(self._on_ring_color_changed)
self.ui.device_combo_box.device_selected.connect(self._on_device_changed)
self.ui.signal_combo_box.device_signal_changed.connect(self._on_signal_changed)
def _init_values(self):
"""Initialize UI values from ring config"""
# Data values
self.ui.value_spin_box.setRange(-1e6, 1e6)
self.ui.value_spin_box.setValue(self.ring.config.value)
self.ui.min_spin_box.setRange(-1e6, 1e6)
self.ui.min_spin_box.setValue(self.ring.config.min_value)
self.ui.max_spin_box.setRange(-1e6, 1e6)
self.ui.max_spin_box.setValue(self.ring.config.max_value)
self._update_min_max()
self.ui.device_combo_box.setEditable(True)
self.ui.signal_combo_box.setEditable(True)
device, signal = self.ring.config.device, self.ring.config.signal
if device:
self.ui.device_combo_box.set_device(device)
if signal:
for i in range(self.ui.signal_combo_box.count()):
data_item = self.ui.signal_combo_box.itemData(i)
if data_item and data_item.get("obj_name") == signal:
self.ui.signal_combo_box.setCurrentIndex(i)
break
# Config values
self.ui.start_angle_spin_box.setValue(self.ring.config.start_position)
self.ui.direction_combo_box.setCurrentIndex(0 if self.ring.config.direction == -1 else 1)
self.ui.line_width_spin_box.setRange(1, 100)
self.ui.line_width_spin_box.setValue(self.ring.config.line_width)
# Colors
self.ui.ring_color_button.set_color(self.ring.color)
self.ui.color_sync_button.setCheckable(True)
self.ui.color_sync_button.setChecked(self.ring.config.link_colors)
# Set initial button state based on link_colors
if self.ring.config.link_colors:
self.ui.color_sync_button.setIcon(material_icon("link"))
self.ui.color_sync_button.setToolTip(
"Colors are linked - background derives from main color"
)
self.ui.background_color_button.setEnabled(False)
self.ui.background_color_label.setEnabled(False)
# Trigger sync to ensure background color is derived from main color
self.ring.set_color(self.ring.config.color)
self.ui.background_color_button.set_color(self.ring.background_color)
else:
self.ui.color_sync_button.setIcon(material_icon("link_off"))
self.ui.color_sync_button.setToolTip(
"Colors are unlinked - set background independently"
)
self.ui.background_color_button.setEnabled(True)
self.ui.background_color_label.setEnabled(True)
self.ui.background_color_button.set_color(self.ring.background_color)
self.ui.color_sync_button.toggled.connect(self._toggle_color_link)
# --------------------------------------------------------
def _toggle_color_link(self, checked: bool):
"""Toggle the color linking between main and background color"""
self.ring.config.link_colors = checked
# Update button icon and tooltip based on state
if checked:
self.ui.color_sync_button.setIcon(material_icon("link"))
self.ui.color_sync_button.setToolTip(
"Colors are linked - background derives from main color"
)
# Trigger background color update by calling set_color
self.ring.set_color(self.ring.config.color)
# Update UI to show the new background color
self.ui.background_color_button.set_color(self.ring.background_color)
else:
self.ui.color_sync_button.setIcon(material_icon("link_off"))
self.ui.color_sync_button.setToolTip(
"Colors are unlinked - set background independently"
)
# Enable/disable background color controls based on link state
self.ui.background_color_button.setEnabled(not checked)
self.ui.background_color_label.setEnabled(not checked)
def _on_ring_color_changed(self, color: QColor):
"""Handle ring color changes and update background if colors are linked"""
self.ring.set_color(color)
# If colors are linked, update the background color button to show the new derived color
if self.ring.config.link_colors:
self.ui.background_color_button.set_color(self.ring.background_color)
def _update_min_max(self):
self.ui.value_spin_box.setRange(self.ui.min_spin_box.value(), self.ui.max_spin_box.value())
self.ring.set_min_max_values(self.ui.min_spin_box.value(), self.ui.max_spin_box.value())
def _update_direction(self, index: int):
self.ring.config.direction = -1 if index == 0 else 1
self.ring.update()
@SafeSlot(str)
def _on_device_changed(self, device: str):
signal = self.ui.signal_combo_box.get_signal_name()
self.ring.set_update("device", device=device, signal=signal)
self.ring.config.device = device
@SafeSlot(str)
def _on_signal_changed(self, signal: str):
device = self.ui.device_combo_box.currentText()
signal = self.ui.signal_combo_box.get_signal_name()
if not device or device not in self.container.bec_dispatcher.client.device_manager.devices:
return
self.ring.set_update("device", device=device, signal=signal)
self.ring.config.signal = signal
def _unify_mode_string(self, mode: str) -> str:
"""Convert mode string to a unified format"""
mode = mode.lower()
if mode == "scan progress":
return "scan"
if mode == "device readback":
return "device"
return mode
def _get_display_mode_string(self, mode: str) -> str:
"""Convert mode string to display format"""
match mode:
case "manual":
return "Manual"
case "scan":
return "Scan Progress"
case "device":
return "Device Readback"
return mode.capitalize()
def _update_mode(self, mode: str):
"""Update the ring's mode based on combo box selection"""
mode = self._unify_mode_string(mode)
match mode:
case "manual":
self.ring.set_update("manual")
case "scan":
self.ring.set_update("scan")
case "device":
self.ring.set_update("device", device=self.ui.device_combo_box.currentText())
self._set_widget_mode_enabled(mode)
def _set_widget_mode_enabled(self, mode: str):
"""Show/hide controls based on the current mode"""
mode = self._unify_mode_string(mode)
self.ui.device_combo_box.setEnabled(mode == "device")
self.ui.signal_combo_box.setEnabled(mode == "device")
self.ui.device_label.setEnabled(mode == "device")
self.ui.signal_label.setEnabled(mode == "device")
self.ui.min_label.setEnabled(mode in ["manual", "device"])
self.ui.max_label.setEnabled(mode in ["manual", "device"])
self.ui.value_label.setEnabled(mode == "manual")
self.ui.value_spin_box.setEnabled(mode == "manual")
self.ui.min_spin_box.setEnabled(mode in ["manual", "device"])
self.ui.max_spin_box.setEnabled(mode in ["manual", "device"])
def _delete_self(self):
"""Delete this ring from the container"""
if self.ring in self.container.rings:
self.container.rings.remove(self.ring)
self.ring.deleteLater()
self.cleanup()
def cleanup(self):
"""Cleanup the card widget"""
self.ui.device_combo_box.close()
self.ui.device_combo_box.deleteLater()
self.ui.signal_combo_box.close()
self.ui.signal_combo_box.deleteLater()
self.close()
self.deleteLater()
# ============================================================
# Ring settings widget
# ============================================================
class RingSettings(SettingWidget):
def __init__(
self, parent=None, target_widget: RingProgressBar | None = None, popup=False, **kwargs
):
super().__init__(parent=parent, **kwargs)
self.setProperty("skip_settings", True)
self.target_widget = target_widget
self.popup = popup
if not target_widget:
return
self.container: RingProgressContainerWidget = target_widget.ring_progress_bar
self.original_num_bars = len(self.container.rings)
self.original_configs = [ring.config.model_dump() for ring in self.container.rings]
layout = QVBoxLayout(self)
layout.setContentsMargins(10, 10, 10, 10)
layout.setSpacing(10)
add_button = QPushButton(material_icon("add"), "Add Ring")
add_button.clicked.connect(self.add_ring)
self.center_label_edit = QLineEdit(self.container.center_label.text())
self.center_label_edit.setPlaceholderText("Center Label")
self.center_label_edit.textChanged.connect(self._update_center_label)
self.colormap_toggle = QPushButton()
self.colormap_toggle.setCheckable(True)
self.colormap_toggle.setIcon(material_icon("palette"))
self.colormap_toggle.setToolTip(
f"Colormap mode is {'enabled' if self.container.color_map else 'disabled'}"
)
self.colormap_toggle.toggled.connect(self._toggle_colormap_mode)
self.colormap_button = BECColorMapWidget(parent=self)
self.colormap_button.setToolTip("Set a global colormap for all rings")
self.colormap_button.colormap_changed_signal.connect(self._set_global_colormap)
toolbar = QHBoxLayout()
toolbar.addWidget(add_button)
toolbar.addWidget(self.center_label_edit)
toolbar.addStretch()
toolbar.addWidget(self.colormap_toggle)
toolbar.addWidget(self.colormap_button)
layout.addLayout(toolbar)
self.scroll = QScrollArea()
self.scroll.setWidgetResizable(True)
self.cards_container = QWidget()
self.cards_layout = QVBoxLayout(self.cards_container)
self.cards_layout.setSpacing(10)
self.cards_layout.addStretch()
self.scroll.setWidget(self.cards_container)
layout.addWidget(self.scroll)
self.refresh_from_container()
self.original_label = self.container.center_label.text()
def sizeHint(self) -> QSize:
return QSize(720, 520)
def refresh_from_container(self):
if not self.container:
return
for ring in self.container.rings:
card = RingCardWidget(ring, self.container)
self.cards_layout.insertWidget(self.cards_layout.count() - 1, card)
if self.container.color_map:
self.colormap_button.colormap = self.container.color_map
self.colormap_toggle.setChecked(bool(self.container.color_map))
@SafeSlot()
def add_ring(self):
if not self.container:
return
self.container.add_ring()
ring = self.container.rings[len(self.container.rings) - 1]
if ring:
card = RingCardWidget(ring, self.container)
self.cards_layout.insertWidget(self.cards_layout.count() - 1, card)
# If a global colormap is set, apply it
if self.container.color_map:
self._toggle_colormap_mode(bool(self.container.color_map))
@SafeSlot(str)
def _update_center_label(self, text: str):
if not self.container:
return
self.container.center_label.setText(text)
@SafeSlot(bool)
def _toggle_colormap_mode(self, enabled: bool):
self.colormap_toggle.setToolTip(f"Colormap mode is {'enabled' if enabled else 'disabled'}")
if enabled:
colormap = self.colormap_button.colormap
self._set_global_colormap(colormap)
else:
self.container.color_map = ""
for i in range(self.cards_layout.count() - 1): # -1 to exclude the stretch
widget = self.cards_layout.itemAt(i).widget()
if not isinstance(widget, RingCardWidget):
continue
widget.ui.ring_color_button.setEnabled(not enabled)
widget.ui.ring_color_button.setToolTip(
"Disabled in colormap mode" if enabled else "Set the ring color"
)
widget.ui.ring_color_label.setEnabled(not enabled)
widget.ui.background_color_button.setEnabled(
not enabled and not widget.ring.config.link_colors
)
widget.ui.color_sync_button.setEnabled(not enabled)
@SafeSlot(str)
def _set_global_colormap(self, colormap: str):
if not self.container:
return
self.container.set_colors_from_map(colormap)
# Update all ring card color buttons to reflect the new colors
for i in range(self.cards_layout.count() - 1): # -1 to exclude the stretch
widget = self.cards_layout.itemAt(i).widget()
if not isinstance(widget, RingCardWidget):
continue
widget.ui.ring_color_button.set_color(widget.ring.color)
if widget.ring.config.link_colors:
widget.ui.background_color_button.set_color(widget.ring.background_color)
@SafeSlot()
def accept_changes(self):
if not self.container:
return
self.original_configs = [ring.config.model_dump() for ring in self.container.rings]
for i, ring in enumerate(self.container.rings):
ring.setGeometry(self.container.rect())
ring.gap = self.container.gap * i
ring.show() # Ensure ring is visible
ring.raise_() # Bring ring to front
self.container.center_label.setText(self.center_label_edit.text())
self.original_label = self.container.center_label.text()
self.original_num_bars = len(self.container.rings)
self.container.update()
def cleanup(self):
"""
Cleanup the settings widget.
"""
# Remove any rings that were added but not applied
if not self.container:
return
if len(self.container.rings) > self.original_num_bars:
remove_rings = self.container.rings[self.original_num_bars :]
for ring in remove_rings:
self.container.rings.remove(ring)
ring.deleteLater()
rings_to_add = max(0, self.original_num_bars - len(self.container.rings))
for _ in range(rings_to_add):
self.container.add_ring()
# apply original configs to all rings
for i, ring in enumerate(self.container.rings):
ring.config = ring.config.model_validate(self.original_configs[i])
for i in range(self.cards_layout.count()):
item = self.cards_layout.itemAt(i)
if not item or not item.widget():
continue
widget: RingCardWidget = item.widget()
widget.cleanup()
self.container.update()
self.container.center_label.setText(self.original_label)

235
bec_widgets/widgets/progress/ring_progress_bar/ring_settings.ui Normal file
View File

@@ -0,0 +1,235 @@
<?xml version="1.0" encoding="UTF-8"?>
<ui version="4.0">
<class>Form</class>
<widget class="QWidget" name="Form">
<property name="geometry">
<rect>
<x>0</x>
<y>0</y>
<width>731</width>
<height>199</height>
</rect>
</property>
<property name="windowTitle">
<string>Form</string>
</property>
<layout class="QGridLayout" name="gridLayout">
<item row="0" column="0">
<widget class="QGroupBox" name="data_group_box">
<property name="title">
<string>Data</string>
</property>
<layout class="QGridLayout" name="gridLayout_2">
<item row="1" column="0">
<widget class="QDoubleSpinBox" name="value_spin_box"/>
</item>
<item row="3" column="1" colspan="2">
<widget class="DeviceComboBox" name="device_combo_box"/>
</item>
<item row="0" column="0">
<widget class="QLabel" name="value_label">
<property name="text">
<string>Value</string>
</property>
</widget>
</item>
<item row="0" column="1">
<widget class="QLabel" name="min_label">
<property name="text">
<string>Min</string>
</property>
</widget>
</item>
<item row="0" column="2">
<widget class="QLabel" name="max_label">
<property name="text">
<string>Max</string>
</property>
</widget>
</item>
<item row="1" column="2">
<widget class="QDoubleSpinBox" name="max_spin_box"/>
</item>
<item row="4" column="0">
<widget class="QLabel" name="signal_label">
<property name="text">
<string>Signal</string>
</property>
</widget>
</item>
<item row="1" column="1">
<widget class="QDoubleSpinBox" name="min_spin_box"/>
</item>
<item row="4" column="1" colspan="2">
<widget class="SignalComboBox" name="signal_combo_box"/>
</item>
<item row="3" column="0">
<widget class="QLabel" name="device_label">
<property name="text">
<string>Device</string>
</property>
</widget>
</item>
<item row="2" column="0" colspan="3">
<widget class="Line" name="data_hor_line">
<property name="orientation">
<enum>Qt::Orientation::Horizontal</enum>
</property>
</widget>
</item>
</layout>
</widget>
</item>
<item row="0" column="1">
<widget class="QGroupBox" name="config_group_box">
<property name="title">
<string>Config</string>
</property>
<layout class="QGridLayout" name="gridLayout_3">
<item row="2" column="0" colspan="3">
<widget class="Line" name="config_hor_line">
<property name="orientation">
<enum>Qt::Orientation::Horizontal</enum>
</property>
</widget>
</item>
<item row="1" column="0">
<widget class="QSpinBox" name="start_angle_spin_box">
<property name="suffix">
<string>°</string>
</property>
<property name="maximum">
<number>360</number>
</property>
<property name="value">
<number>90</number>
</property>
</widget>
</item>
<item row="0" column="2">
<widget class="QLabel" name="line_width_label">
<property name="text">
<string>Line Width</string>
</property>
</widget>
</item>
<item row="1" column="1">
<widget class="QComboBox" name="direction_combo_box">
<item>
<property name="text">
<string>Clockwise</string>
</property>
</item>
<item>
<property name="text">
<string>Counter-clockwise</string>
</property>
</item>
</widget>
</item>
<item row="0" column="1">
<widget class="QLabel" name="direction_label">
<property name="text">
<string>Direction</string>
</property>
</widget>
</item>
<item row="1" column="2">
<widget class="QSpinBox" name="line_width_spin_box">
<property name="value">
<number>12</number>
</property>
</widget>
</item>
<item row="0" column="0">
<widget class="QLabel" name="start_angle_label">
<property name="text">
<string>Start Angle</string>
</property>
</widget>
</item>
<item row="5" column="1">
<widget class="ColorButtonNative" name="background_color_button"/>
</item>
<item row="4" column="1">
<widget class="ColorButtonNative" name="ring_color_button"/>
</item>
<item row="5" column="0">
<widget class="QLabel" name="background_color_label">
<property name="text">
<string>Background Color</string>
</property>
</widget>
</item>
<item row="4" column="0">
<widget class="QLabel" name="ring_color_label">
<property name="text">
<string>Ring Color</string>
</property>
</widget>
</item>
<item row="5" column="2">
<widget class="QToolButton" name="color_sync_button">
<property name="text">
<string>...</string>
</property>
</widget>
</item>
</layout>
</widget>
</item>
</layout>
</widget>
<customwidgets>
<customwidget>
<class>ColorButtonNative</class>
<extends></extends>
<header>color_button_native</header>
</customwidget>
<customwidget>
<class>DeviceComboBox</class>
<extends></extends>
<header>device_combo_box</header>
</customwidget>
<customwidget>
<class>SignalComboBox</class>
<extends></extends>
<header>signal_combo_box</header>
</customwidget>
</customwidgets>
<resources/>
<connections>
<connection>
<sender>device_combo_box</sender>
<signal>currentTextChanged(QString)</signal>
<receiver>signal_combo_box</receiver>
<slot>set_device(QString)</slot>
<hints>
<hint type="sourcelabel">
<x>209</x>
<y>133</y>
</hint>
<hint type="destinationlabel">
<x>213</x>
<y>153</y>
</hint>
</hints>
</connection>
<connection>
<sender>device_combo_box</sender>
<signal>device_reset()</signal>
<receiver>signal_combo_box</receiver>
<slot>reset_selection()</slot>
<hints>
<hint type="sourcelabel">
<x>248</x>
<y>135</y>
</hint>
<hint type="destinationlabel">
<x>250</x>
<y>147</y>
</hint>
</hints>
</connection>
</connections>
</ui>

51
docs/user/widgets/progress_bar/ring_progress_bar.md
View File

@@ -24,11 +24,13 @@ In this example, we demonstrate how to add a `RingProgressBar` widget to a `BECD
```python
# Add a new dock with a RingProgressBar widget
dock_area = gui.new('my_new_dock_area') # Create a new dock area
progress = dock_area.new().new(gui.available_widgets.RingProgressBar)
dock_area = gui.new() # Create a new dock area
progress = dock_area.new(gui.available_widgets.RingProgressBar)
# Customize the size of the progress ring
progress.set_line_widths(20)
# Add a ring to the RingProgressBar
progress.add_ring()
ring = progress.rings[0]
ring.set_value(50) # Set the progress value to 50
```
## Example 2 - Adding Multiple Rings to Track Parallel Tasks
@@ -40,8 +42,7 @@ By default, the `RingProgressBar` widget displays a single ring. You can add add
progress.add_ring()
# Customize the rings
progress.rings[0].set_line_widths(20) # Set the width of the first ring
progress.rings[1].set_line_widths(10) # Set the width of the second ring
progress.rings[1].set_value(30) # Set the second ring to 30
```
## Example 3 - Integrating with Device Readback and Scans
@@ -56,44 +57,6 @@ progress.rings[0].set_update("scan")
progress.rings[1].set_update("device", "samx")
```
## Example 4 - Customizing Visual Elements of the Rings
The `RingProgressBar` widget offers various customization options, such as changing colors, line widths, and the gap between rings.
```python
# Set the color of the first ring to blue
progress.rings[0].set_color("blue")
# Set the background color of the second ring
progress.rings[1].set_background("gray")
# Adjust the gap between the rings
progress.set_gap(5)
# Set the diameter of the progress bar
progress.set_diameter(150)
```
## Example 5 - Manual Updates and Precision Control
While the `RingProgressBar` supports automatic updates, you can also manually control the progress and set the precision for each ring.
```python
# Disable automatic updates and manually set the progress value
progress.enable_auto_updates(False)
progress.rings[0].set_value(75) # Set the first ring to 75%
# Set precision for the progress display
progress.set_precision(2) # Display progress with two decimal places
# Setting multiple rigns with different values
progress.set_number_of_bars(3)
# Set the values of the rings to 50, 75, and 25 from outer to inner ring
progress.set_value([50, 75, 25])
```
````
````{tab} API

50
tests/end-2-end/user_interaction/test_user_interaction_e2e.py
View File

@@ -537,30 +537,32 @@ def test_widgets_e2e_positioner_control_line(
# TODO passes locally, fails on CI for some reason... -> issue #1003
# @pytest.mark.timeout(PYTEST_TIMEOUT)
# def test_widgets_e2e_ring_progress_bar(qtbot, connected_client_gui_obj, random_generator_from_seed):
# """Test the RingProgressBar widget"""
# gui = connected_client_gui_obj
# bec = gui._client
# # Create dock_area and widget
# widget = create_widget(qtbot, gui, gui.available_widgets.RingProgressBar)
# widget: client.RingProgressBar
#
# widget.set_number_of_bars(3)
# widget.rings[0].set_update("manual")
# widget.rings[0].set_value(30)
# widget.rings[0].set_min_max_values(0, 100)
# widget.rings[1].set_update("scan")
# widget.rings[2].set_update("device", device="samx")
#
# # Test rpc calls
# dev = bec.device_manager.devices
# scans = bec.scans
# # Do a scan
# scans.line_scan(dev.samx, -3, 3, steps=50, exp_time=0.01, relative=False).wait()
#
# # Test removing the widget, or leaving it open for the next test
# maybe_remove_dock_area(qtbot, gui=gui, random_int_gen=random_generator_from_seed)
@pytest.mark.timeout(PYTEST_TIMEOUT)
def test_widgets_e2e_ring_progress_bar(qtbot, connected_client_gui_obj, random_generator_from_seed):
"""Test the RingProgressBar widget"""
gui = connected_client_gui_obj
bec = gui._client
# Create dock_area and widget
widget = create_widget(qtbot, gui, gui.available_widgets.RingProgressBar)
widget: client.RingProgressBar
widget.add_ring()
widget.add_ring()
widget.add_ring()
widget.rings[0].set_update("manual")
widget.rings[0].set_value(30)
widget.rings[0].set_min_max_values(0, 100)
widget.rings[1].set_update("scan")
widget.rings[2].set_update("device", device="samx")
# Test rpc calls
dev = bec.device_manager.devices
scans = bec.scans
# Do a scan
scans.line_scan(dev.samx, -3, 3, steps=50, exp_time=0.01, relative=False).wait()
# Test removing the widget, or leaving it open for the next test
maybe_remove_dock_area(qtbot, gui=gui, random_int_gen=random_generator_from_seed)
@pytest.mark.timeout(PYTEST_TIMEOUT)

512
tests/unit_tests/test_ring_progress_bar.py
View File

@@ -1,13 +1,14 @@
# pylint: disable=missing-function-docstring, missing-module-docstring, unused-import
import json
import pytest
from bec_lib.endpoints import MessageEndpoints
from pydantic import ValidationError
from qtpy.QtGui import QColor
from bec_widgets.utils import Colors
from bec_widgets.widgets.progress.ring_progress_bar import RingProgressBar
from bec_widgets.widgets.progress.ring_progress_bar.ring import ProgressbarConnections, RingConfig
from bec_widgets.widgets.progress.ring_progress_bar.ring_progress_bar import RingProgressBarConfig
from bec_widgets.widgets.progress.ring_progress_bar.ring_progress_bar import RingProgressBar
from .client_mocks import mocked_client
@@ -29,176 +30,117 @@ def test_bar_init(ring_progress_bar):
assert ring_progress_bar.gui_id == ring_progress_bar.config.gui_id
def test_config_validation_num_of_bars():
config = RingProgressBarConfig(num_bars=100, min_num_bars=1, max_num_bars=10)
assert config.num_bars == 10
def test_rpb_center_label(ring_progress_bar):
test_text = "Center Label"
ring_progress_bar.set_center_label(test_text)
assert ring_progress_bar.center_label == test_text
assert ring_progress_bar.ring_progress_bar.center_label.text() == test_text
def test_config_validation_num_of_ring_error():
ring_config_0 = RingConfig(index=0)
ring_config_1 = RingConfig(index=1)
with pytest.raises(ValidationError) as excinfo:
RingProgressBarConfig(rings=[ring_config_0, ring_config_1], num_bars=1)
errors = excinfo.value.errors()
assert len(errors) == 1
assert errors[0]["type"] == "different number of configs"
assert "Length of rings configuration (2) does not match the number of bars (1)." in str(
excinfo.value
)
def test_config_validation_ring_indices_wrong_order():
ring_config_0 = RingConfig(index=2)
ring_config_1 = RingConfig(index=5)
with pytest.raises(ValidationError) as excinfo:
RingProgressBarConfig(rings=[ring_config_0, ring_config_1], num_bars=2)
errors = excinfo.value.errors()
assert len(errors) == 1
assert errors[0]["type"] == "wrong indices"
assert (
"Indices of ring configurations must be unique and in order from 0 to num_bars 2."
in str(excinfo.value)
)
def test_config_validation_ring_same_indices():
ring_config_0 = RingConfig(index=0)
ring_config_1 = RingConfig(index=0)
with pytest.raises(ValidationError) as excinfo:
RingProgressBarConfig(rings=[ring_config_0, ring_config_1], num_bars=2)
errors = excinfo.value.errors()
assert len(errors) == 1
assert errors[0]["type"] == "wrong indices"
assert (
"Indices of ring configurations must be unique and in order from 0 to num_bars 2."
in str(excinfo.value)
)
def test_config_validation_invalid_colormap():
with pytest.raises(ValueError) as excinfo:
RingProgressBarConfig(color_map="crazy_colors")
errors = excinfo.value.errors()
assert len(errors) == 1
assert errors[0]["type"] == "unsupported colormap"
assert "Colormap 'crazy_colors' not found in the current installation of pyqtgraph" in str(
excinfo.value
)
def test_ring_connection_endpoint_validation():
with pytest.raises(ValueError) as excinfo:
ProgressbarConnections(slot="on_scan_progress", endpoint="non_existing")
errors = excinfo.value.errors()
assert len(errors) == 1
assert errors[0]["type"] == "unsupported endpoint"
assert (
"For slot 'on_scan_progress', endpoint must be MessageEndpoint.scan_progress or 'scans/scan_progress'."
in str(excinfo.value)
)
with pytest.raises(ValueError) as excinfo:
ProgressbarConnections(slot="on_device_readback", endpoint="non_existing")
errors = excinfo.value.errors()
assert len(errors) == 1
assert errors[0]["type"] == "unsupported endpoint"
assert (
"For slot 'on_device_readback', endpoint must be MessageEndpoint.device_readback(device) or 'internal/devices/readback/{device}'."
in str(excinfo.value)
)
def test_bar_add_number_of_bars(ring_progress_bar):
assert ring_progress_bar.config.num_bars == 1
ring_progress_bar.set_number_of_bars(5)
assert ring_progress_bar.config.num_bars == 5
ring_progress_bar.set_number_of_bars(2)
assert ring_progress_bar.config.num_bars == 2
def test_add_remove_bars_individually(ring_progress_bar):
def test_add_ring(qtbot, ring_progress_bar):
ring_progress_bar.show()
initial_num_bars = ring_progress_bar.ring_progress_bar.num_bars
assert initial_num_bars == len(ring_progress_bar.rings)
ring_progress_bar.add_ring()
assert ring_progress_bar.ring_progress_bar.num_bars == initial_num_bars + 1
assert len(ring_progress_bar.rings) == initial_num_bars + 1
qtbot.wait(200)
def test_remove_ring(ring_progress_bar):
ring_progress_bar.add_ring()
initial_num_bars = ring_progress_bar.ring_progress_bar.num_bars
assert initial_num_bars == len(ring_progress_bar.rings)
ring_progress_bar.remove_ring()
assert ring_progress_bar.ring_progress_bar.num_bars == initial_num_bars - 1
assert len(ring_progress_bar.rings) == initial_num_bars - 1
assert ring_progress_bar.config.num_bars == 3
assert len(ring_progress_bar.config.rings) == 3
ring_progress_bar.remove_ring(1)
assert ring_progress_bar.config.num_bars == 2
assert len(ring_progress_bar.config.rings) == 2
assert ring_progress_bar.rings[0].config.index == 0
assert ring_progress_bar.rings[1].config.index == 1
def test_remove_ring_no_bars(ring_progress_bar):
# Remove all rings first
while ring_progress_bar.ring_progress_bar.num_bars > 0:
ring_progress_bar.remove_ring()
initial_num_bars = ring_progress_bar.ring_progress_bar.num_bars
assert initial_num_bars == 0
# Attempt to remove a ring when there are none
ring_progress_bar.remove_ring()
assert ring_progress_bar.ring_progress_bar.num_bars == initial_num_bars
assert len(ring_progress_bar.rings) == initial_num_bars
def test_bar_set_value(ring_progress_bar):
ring_progress_bar.set_number_of_bars(5)
ring_progress_bar.add_ring()
ring_progress_bar.add_ring()
assert ring_progress_bar.config.num_bars == 5
assert len(ring_progress_bar.config.rings) == 5
assert len(ring_progress_bar.rings) == 5
ring_progress_bar.set_value([10, 20, 30, 40, 50])
ring_progress_bar.rings[0].set_value(10)
ring_progress_bar.rings[1].set_value(20)
ring_values = [ring.config.value for ring in ring_progress_bar.rings]
assert ring_values == [10, 20, 30, 40, 50]
# update just one bar
ring_progress_bar.set_value(90, 1)
ring_values = [ring.config.value for ring in ring_progress_bar.rings]
assert ring_values == [10, 90, 30, 40, 50]
assert ring_values == [10, 20]
def test_bar_set_precision(ring_progress_bar):
ring_progress_bar.set_number_of_bars(3)
# Add 3 rings
ring_progress_bar.add_ring()
ring_progress_bar.add_ring()
ring_progress_bar.add_ring()
assert ring_progress_bar.config.num_bars == 3
assert len(ring_progress_bar.config.rings) == 3
assert ring_progress_bar.ring_progress_bar.num_bars == 3
assert len(ring_progress_bar.rings) == 3
ring_progress_bar.set_precision(2)
# Set precision for all rings
for ring in ring_progress_bar.rings:
ring.set_precision(2)
ring_precision = [ring.config.precision for ring in ring_progress_bar.rings]
assert ring_precision == [2, 2, 2]
ring_progress_bar.set_value([10.1234, 20.1234, 30.1234])
# Set values
for i, ring in enumerate(ring_progress_bar.rings):
ring.set_value([10.1234, 20.1234, 30.1234][i])
ring_values = [ring.config.value for ring in ring_progress_bar.rings]
assert ring_values == [10.12, 20.12, 30.12]
ring_progress_bar.set_precision(4, 1)
# Set precision for ring at index 1
ring_progress_bar.rings[1].set_precision(4)
ring_precision = [ring.config.precision for ring in ring_progress_bar.rings]
assert ring_precision == [2, 4, 2]
ring_progress_bar.set_value([10.1234, 20.1234, 30.1234])
# Set values again
for i, ring in enumerate(ring_progress_bar.rings):
ring.set_value([10.1234, 20.1234, 30.1234][i])
ring_values = [ring.config.value for ring in ring_progress_bar.rings]
assert ring_values == [10.12, 20.1234, 30.12]
def test_set_min_max_value(ring_progress_bar):
ring_progress_bar.set_number_of_bars(2)
# Add 2 rings
ring_progress_bar.add_ring()
ring_progress_bar.add_ring()
ring_progress_bar.set_min_max_values(0, 10)
# Set min/max values for all rings
for ring in ring_progress_bar.rings:
ring.set_min_max_values(0, 10)
ring_min_values = [ring.config.min_value for ring in ring_progress_bar.rings]
ring_max_values = [ring.config.max_value for ring in ring_progress_bar.rings]
assert ring_min_values == [0, 0]
assert ring_max_values == [10, 10]
ring_progress_bar.set_value([5, 15])
# Set values
ring_progress_bar.rings[0].set_value(5)
ring_progress_bar.rings[1].set_value(15)
ring_values = [ring.config.value for ring in ring_progress_bar.rings]
assert ring_values == [5, 10]
def test_setup_colors_from_colormap(ring_progress_bar):
ring_progress_bar.set_number_of_bars(5)
ring_progress_bar.set_colors_from_map("viridis", "RGB")
# Add 5 rings
for _ in range(5):
ring_progress_bar.add_ring()
ring_progress_bar.ring_progress_bar.set_colors_from_map("viridis", "RGB")
expected_colors = Colors.golden_angle_color("viridis", 5, "RGB")
converted_colors = [ring.color.getRgb() for ring in ring_progress_bar.rings]
ring_config_colors = [ring.config.color for ring in ring_progress_bar.rings]
ring_config_colors = [QColor(ring.config.color).getRgb() for ring in ring_progress_bar.rings]
assert expected_colors == converted_colors
assert ring_config_colors == expected_colors
@@ -206,13 +148,15 @@ def test_setup_colors_from_colormap(ring_progress_bar):
def get_colors_from_rings(rings):
converted_colors = [ring.color.getRgb() for ring in rings]
ring_config_colors = [ring.config.color for ring in rings]
ring_config_colors = [QColor(ring.config.color).getRgb() for ring in rings]
return converted_colors, ring_config_colors
def test_set_colors_from_colormap_and_change_num_of_bars(ring_progress_bar):
ring_progress_bar.set_number_of_bars(2)
ring_progress_bar.set_colors_from_map("viridis", "RGB")
# Add 2 rings
ring_progress_bar.add_ring()
ring_progress_bar.add_ring()
ring_progress_bar.ring_progress_bar.set_colors_from_map("viridis", "RGB")
expected_colors = Colors.golden_angle_color("viridis", 2, "RGB")
converted_colors, ring_config_colors = get_colors_from_rings(ring_progress_bar.rings)
@@ -221,7 +165,9 @@ def test_set_colors_from_colormap_and_change_num_of_bars(ring_progress_bar):
assert ring_config_colors == expected_colors
# increase the number of bars to 6
ring_progress_bar.set_number_of_bars(6)
for _ in range(4):
ring_progress_bar.add_ring()
ring_progress_bar.ring_progress_bar.set_colors_from_map("viridis", "RGB")
expected_colors = Colors.golden_angle_color("viridis", 6, "RGB")
converted_colors, ring_config_colors = get_colors_from_rings(ring_progress_bar.rings)
@@ -229,7 +175,9 @@ def test_set_colors_from_colormap_and_change_num_of_bars(ring_progress_bar):
assert ring_config_colors == expected_colors
# decrease the number of bars to 3
ring_progress_bar.set_number_of_bars(3)
for _ in range(3):
ring_progress_bar.remove_ring()
ring_progress_bar.ring_progress_bar.set_colors_from_map("viridis", "RGB")
expected_colors = Colors.golden_angle_color("viridis", 3, "RGB")
converted_colors, ring_config_colors = get_colors_from_rings(ring_progress_bar.rings)
@@ -238,100 +186,284 @@ def test_set_colors_from_colormap_and_change_num_of_bars(ring_progress_bar):
def test_set_colors_directly(ring_progress_bar):
ring_progress_bar.set_number_of_bars(3)
# Add 3 rings
for _ in range(3):
ring_progress_bar.add_ring()
# setting as a list of rgb tuples
colors = [(255, 0, 0, 255), (0, 255, 0, 255), (0, 0, 255, 255)]
ring_progress_bar.set_colors_directly(colors)
ring_progress_bar.ring_progress_bar.set_colors_directly(colors)
converted_colors = get_colors_from_rings(ring_progress_bar.rings)[0]
assert colors == converted_colors
ring_progress_bar.set_colors_directly((255, 0, 0, 255), 1)
ring_progress_bar.ring_progress_bar.set_colors_directly((255, 0, 0, 255), 1)
converted_colors = get_colors_from_rings(ring_progress_bar.rings)[0]
assert converted_colors == [(255, 0, 0, 255), (255, 0, 0, 255), (0, 0, 255, 255)]
def test_set_line_width(ring_progress_bar):
ring_progress_bar.set_number_of_bars(3)
# Add 3 rings
for _ in range(3):
ring_progress_bar.add_ring()
ring_progress_bar.set_line_widths(5)
# Set line width for all rings
for ring in ring_progress_bar.rings:
ring.set_line_width(5)
line_widths = [ring.config.line_width for ring in ring_progress_bar.rings]
assert line_widths == [5, 5, 5]
ring_progress_bar.set_line_widths([10, 20, 30])
# Set individual line widths
for i, ring in enumerate(ring_progress_bar.rings):
ring.set_line_width([10, 20, 30][i])
line_widths = [ring.config.line_width for ring in ring_progress_bar.rings]
assert line_widths == [10, 20, 30]
ring_progress_bar.set_line_widths(15, 1)
# Set line width for ring at index 1
ring_progress_bar.rings[1].set_line_width(15)
line_widths = [ring.config.line_width for ring in ring_progress_bar.rings]
assert line_widths == [10, 15, 30]
def test_set_gap(ring_progress_bar):
ring_progress_bar.set_number_of_bars(3)
ring_progress_bar.add_ring()
ring_progress_bar.set_gap(20)
assert ring_progress_bar.config.gap == 20
assert ring_progress_bar.ring_progress_bar.gap == 20 == ring_progress_bar.gap
def test_auto_update(ring_progress_bar):
ring_progress_bar.enable_auto_updates(True)
def test_remove_ring_by_index(ring_progress_bar):
# Add 5 rings
for _ in range(5):
ring_progress_bar.add_ring()
scan_queue_status_scan_progress = {
"queue": {
"primary": {
"info": [{"active_request_block": {"report_instructions": [{"scan_progress": 10}]}}]
}
}
assert ring_progress_bar.ring_progress_bar.num_bars == 5
# Store the ring at index 2 before removal
ring_at_3 = ring_progress_bar.rings[3]
# Remove ring at index 2 (middle ring)
ring_progress_bar.remove_ring(index=2)
assert ring_progress_bar.ring_progress_bar.num_bars == 4
# Ring that was at index 3 is now at index 2
assert ring_progress_bar.rings[2] == ring_at_3
def test_remove_ring_updates_gaps(ring_progress_bar):
# Add 3 rings with default gap
for _ in range(3):
ring_progress_bar.add_ring()
initial_gap = ring_progress_bar.gap
# Gaps should be: 0, gap, 2*gap
expected_gaps = [0, initial_gap, 2 * initial_gap]
actual_gaps = [ring.gap for ring in ring_progress_bar.rings]
assert actual_gaps == expected_gaps
# Remove middle ring
ring_progress_bar.remove_ring(index=1)
# Gaps should now be: 0, gap (for the remaining 2 rings)
expected_gaps = [0, initial_gap]
actual_gaps = [ring.gap for ring in ring_progress_bar.rings]
assert actual_gaps == expected_gaps
def test_center_label_property(ring_progress_bar):
test_text = "Test Label"
ring_progress_bar.center_label = test_text
assert ring_progress_bar.center_label == test_text
assert ring_progress_bar.ring_progress_bar.center_label.text() == test_text
def test_color_map_property(ring_progress_bar):
# Add some rings
for _ in range(3):
ring_progress_bar.add_ring()
# Set colormap via property
ring_progress_bar.color_map = "plasma"
assert ring_progress_bar.color_map == "plasma"
assert ring_progress_bar.ring_progress_bar.color_map == "plasma"
# Verify colors were applied
expected_colors = Colors.golden_angle_color("plasma", 3, "RGB")
actual_colors = [ring.color.getRgb() for ring in ring_progress_bar.rings]
assert actual_colors == expected_colors
def test_color_map_property_invalid_colormap(ring_progress_bar):
# Make sure that invalid colormaps do not crash the application
ring_progress_bar.color_map = "plasma"
ring_progress_bar.color_map = "invalid_colormap_name"
assert ring_progress_bar.color_map == "plasma" # Should remain unchanged
def test_ring_json_serialization(ring_progress_bar):
# Add rings with specific configurations
ring_progress_bar.add_ring()
ring_progress_bar.add_ring()
ring_progress_bar.add_ring()
# Configure rings
ring_progress_bar.rings[0].set_value(25)
ring_progress_bar.rings[0].set_color((255, 0, 0, 255))
ring_progress_bar.rings[1].set_value(50)
ring_progress_bar.rings[1].set_line_width(15)
ring_progress_bar.rings[2].set_value(75)
ring_progress_bar.rings[2].set_precision(4)
# Get JSON
json_str = ring_progress_bar.ring_json
# Verify it's valid JSON
ring_configs = json.loads(json_str)
assert isinstance(ring_configs, list)
assert len(ring_configs) == 3
# Check some values
assert ring_configs[0]["value"] == 25
assert ring_configs[1]["value"] == 50
assert ring_configs[1]["line_width"] == 15
assert ring_configs[2]["precision"] == 4
def test_ring_json_deserialization(ring_progress_bar):
# Create JSON config
ring_configs = [
{"value": 10, "color": (100, 150, 200, 255), "line_width": 8},
{"value": 20, "precision": 2, "min_value": 0, "max_value": 50},
{"value": 30, "direction": 1},
]
json_str = json.dumps(ring_configs)
# Load via property
ring_progress_bar.ring_json = json_str
# Verify rings were created
assert len(ring_progress_bar.rings) == 3
# Verify configurations
assert ring_progress_bar.rings[0].config.value == 10
assert ring_progress_bar.rings[0].config.line_width == 8
assert ring_progress_bar.rings[1].config.precision == 2
assert ring_progress_bar.rings[1].config.max_value == 50
assert ring_progress_bar.rings[2].config.direction == 1
def test_ring_json_replaces_existing_rings(ring_progress_bar):
# Add some initial rings
for _ in range(5):
ring_progress_bar.add_ring()
assert len(ring_progress_bar.rings) == 5
# Load new config with only 2 rings
ring_configs = [{"value": 10}, {"value": 20}]
ring_progress_bar.ring_json = json.dumps(ring_configs)
# Should have replaced all rings
assert len(ring_progress_bar.rings) == 2
assert ring_progress_bar.rings[0].config.value == 10
assert ring_progress_bar.rings[1].config.value == 20
def test_add_ring_with_config(ring_progress_bar):
config = {
"value": 42,
"color": (128, 64, 192, 255),
"line_width": 12,
"precision": 1,
"min_value": 0,
"max_value": 100,
}
meta = {}
ring_progress_bar.color_map = ""
ring_progress_bar.add_ring(config=config)
ring_progress_bar.on_scan_queue_status(scan_queue_status_scan_progress, meta)
assert len(ring_progress_bar.rings) == 1
ring = ring_progress_bar.rings[0]
assert ring_progress_bar._auto_updates is True
assert len(ring_progress_bar._rings) == 1
assert ring_progress_bar._rings[0].config.connections == ProgressbarConnections(
slot="on_scan_progress", endpoint=MessageEndpoints.scan_progress()
)
assert ring.config.value == 42
assert ring.config.line_width == 12
assert ring.config.precision == 1
assert ring.config.max_value == 100
assert ring.config.color == "#8040c0" # Hex representation of (128, 64, 192)
scan_queue_status_device_readback = {
"queue": {
"primary": {
"info": [
{
"active_request_block": {
"report_instructions": [
{
"readback": {
"devices": ["samx", "samy"],
"start": [1, 2],
"end": [10, 20],
}
}
]
}
}
]
}
}
}
ring_progress_bar.on_scan_queue_status(scan_queue_status_device_readback, meta)
assert ring_progress_bar._auto_updates is True
assert len(ring_progress_bar._rings) == 2
assert ring_progress_bar._rings[0].config.connections == ProgressbarConnections(
slot="on_device_readback", endpoint=MessageEndpoints.device_readback("samx")
)
assert ring_progress_bar._rings[1].config.connections == ProgressbarConnections(
slot="on_device_readback", endpoint=MessageEndpoints.device_readback("samy")
)
def test_set_colors_directly_single_color_extends_to_all(ring_progress_bar):
# Add 4 rings
for _ in range(4):
ring_progress_bar.add_ring()
assert ring_progress_bar._rings[0].config.min_value == 1
assert ring_progress_bar._rings[0].config.max_value == 10
assert ring_progress_bar._rings[1].config.min_value == 2
assert ring_progress_bar._rings[1].config.max_value == 20
# Set a single color, should extend to all rings
single_color = (200, 100, 50, 255)
ring_progress_bar.ring_progress_bar.set_colors_directly(single_color)
colors = [ring.color.getRgb() for ring in ring_progress_bar.rings]
assert all(color == single_color for color in colors)
def test_set_colors_directly_list_too_short(ring_progress_bar):
# Add 5 rings
for _ in range(5):
ring_progress_bar.add_ring()
# Provide only 2 colors
colors = [(255, 0, 0, 255), (0, 255, 0, 255)]
ring_progress_bar.ring_progress_bar.set_colors_directly(colors)
# Last color should be extended to remaining rings
actual_colors = [ring.color.getRgb() for ring in ring_progress_bar.rings]
assert actual_colors[0] == (255, 0, 0, 255)
assert actual_colors[1] == (0, 255, 0, 255)
assert all(color == (0, 255, 0, 255) for color in actual_colors[2:])
def test_gap_affects_ring_positioning(ring_progress_bar):
# Add 3 rings
for _ in range(3):
ring_progress_bar.add_ring()
initial_gap = ring_progress_bar.gap
# Change gap
new_gap = 30
ring_progress_bar.set_gap(new_gap)
# Verify gaps are updated but update method is needed for visual changes
assert ring_progress_bar.gap == new_gap
def test_clear_all_rings(ring_progress_bar):
# Add multiple rings
for _ in range(5):
ring_progress_bar.add_ring()
assert len(ring_progress_bar.rings) == 5
# Clear all
ring_progress_bar.ring_progress_bar.clear_all()
assert len(ring_progress_bar.rings) == 0
assert ring_progress_bar.ring_progress_bar.num_bars == 0
def test_rings_property_returns_correct_list(ring_progress_bar):
# Add some rings
for _ in range(3):
ring_progress_bar.add_ring()
rings_via_property = ring_progress_bar.rings
rings_direct = ring_progress_bar.ring_progress_bar.rings
# Should return the same list
assert rings_via_property is rings_direct
assert len(rings_via_property) == 3

602
tests/unit_tests/test_ring_progress_bar_ring.py Normal file
View File

@@ -0,0 +1,602 @@
# pylint: disable=missing-function-docstring, missing-module-docstring
from unittest.mock import MagicMock
import pytest
from qtpy.QtGui import QColor
from bec_widgets.tests.utils import FakeDevice
from bec_widgets.widgets.progress.ring_progress_bar.ring import ProgressbarConfig, Ring
from bec_widgets.widgets.progress.ring_progress_bar.ring_progress_bar import (
RingProgressContainerWidget,
)
from .client_mocks import mocked_client
@pytest.fixture
def ring_container(qtbot, mocked_client):
container = RingProgressContainerWidget()
qtbot.addWidget(container)
yield container
@pytest.fixture
def ring_widget(qtbot, ring_container, mocked_client):
ring = Ring(parent=ring_container, client=mocked_client)
qtbot.addWidget(ring)
qtbot.waitExposed(ring)
yield ring
@pytest.fixture
def ring_widget_with_device(ring_widget):
mock_device = FakeDevice(name="samx")
ring_widget.bec_dispatcher.client.device_manager.devices["samx"] = mock_device
yield ring_widget
def test_ring_initialization(ring_widget):
assert ring_widget is not None
assert isinstance(ring_widget.config, ProgressbarConfig)
assert ring_widget.config.mode == "manual"
assert ring_widget.config.value == 0
assert ring_widget.registered_slot is None
def test_ring_has_default_config_values(ring_widget):
assert ring_widget.config.direction == -1
assert ring_widget.config.line_width == 20
assert ring_widget.config.start_position == 90
assert ring_widget.config.min_value == 0
assert ring_widget.config.max_value == 100
assert ring_widget.config.precision == 3
###################################
# set_update method tests
###################################
def test_set_update_to_manual(ring_widget):
# Start in manual mode
assert ring_widget.config.mode == "manual"
# Set to manual again (should return early)
ring_widget.set_update("manual")
assert ring_widget.config.mode == "manual"
assert ring_widget.registered_slot is None
def test_set_update_to_scan(ring_widget):
# Mock the dispatcher to avoid actual connections
ring_widget.bec_dispatcher.connect_slot = MagicMock()
# Set to scan mode
ring_widget.set_update("scan")
assert ring_widget.config.mode == "scan"
# Verify that connect_slot was called
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_scan_progress
assert "scan_progress" in str(call_args[0][1])
def test_set_update_from_scan_to_manual(ring_widget):
# Mock the dispatcher
ring_widget.bec_dispatcher.connect_slot = MagicMock()
ring_widget.bec_dispatcher.disconnect_slot = MagicMock()
# Set to scan mode first
ring_widget.set_update("scan")
assert ring_widget.config.mode == "scan"
# Now switch back to manual
ring_widget.set_update("manual")
assert ring_widget.config.mode == "manual"
assert ring_widget.registered_slot is None
def test_set_update_to_device(ring_widget_with_device):
ring_widget = ring_widget_with_device
# Mock the dispatcher
ring_widget.bec_dispatcher.connect_slot = MagicMock()
# Set to device mode
test_device = "samx"
ring_widget.set_update("device", device=test_device)
assert ring_widget.config.mode == "device"
assert ring_widget.config.device == test_device
assert ring_widget.config.signal == "samx"
ring_widget.bec_dispatcher.connect_slot.assert_called_once()
def test_set_update_from_device_to_manual(ring_widget_with_device):
ring_widget = ring_widget_with_device
# Mock the dispatcher
ring_widget.bec_dispatcher.connect_slot = MagicMock()
ring_widget.bec_dispatcher.disconnect_slot = MagicMock()
# Set to device mode first
ring_widget.set_update("device", device="samx")
assert ring_widget.config.mode == "device"
# Switch to manual
ring_widget.set_update("manual")
assert ring_widget.config.mode == "manual"
assert ring_widget.registered_slot is None
def test_set_update_scan_to_device(ring_widget_with_device):
ring_widget = ring_widget_with_device
# Mock the dispatcher
ring_widget.bec_dispatcher.connect_slot = MagicMock()
ring_widget.bec_dispatcher.disconnect_slot = MagicMock()
# Set to scan mode first
ring_widget.set_update("scan")
assert ring_widget.config.mode == "scan"
# Switch to device mode
ring_widget.set_update("device", device="samx")
assert ring_widget.config.mode == "device"
assert ring_widget.config.device == "samx"
def test_set_update_device_to_scan(ring_widget_with_device):
ring_widget = ring_widget_with_device
# Mock the dispatcher
ring_widget.bec_dispatcher.connect_slot = MagicMock()
ring_widget.bec_dispatcher.disconnect_slot = MagicMock()
# Set to device mode first
ring_widget.set_update("device", device="samx")
assert ring_widget.config.mode == "device"
# Switch to scan mode
ring_widget.set_update("scan")
assert ring_widget.config.mode == "scan"
def test_set_update_same_device_resubscribes(ring_widget_with_device):
ring_widget = ring_widget_with_device
# Mock the dispatcher
ring_widget.bec_dispatcher.connect_slot = MagicMock()
ring_widget.bec_dispatcher.disconnect_slot = MagicMock()
# Set to device mode
test_device = "samx"
ring_widget.set_update("device", device=test_device)
# Reset mocks
ring_widget.bec_dispatcher.connect_slot.reset_mock()
ring_widget.bec_dispatcher.disconnect_slot.reset_mock()
# Set to same device mode again (should resubscribe)
ring_widget.set_update("device", device=test_device)
# Should disconnect and reconnect
ring_widget.bec_dispatcher.disconnect_slot.assert_called_once()
ring_widget.bec_dispatcher.connect_slot.assert_called_once()
def test_set_update_invalid_mode(ring_widget):
with pytest.raises(ValueError) as excinfo:
ring_widget.set_update("invalid_mode")
assert "Unsupported mode: invalid_mode" in str(excinfo.value)
###################################
# Value and property tests
###################################
def test_set_value(ring_widget):
ring_widget.set_value(42.5)
assert ring_widget.config.value == 42.5
def test_set_value_with_min_max_clamping(ring_widget):
ring_widget.set_min_max_values(0, 100)
# Set value above max
ring_widget.set_value(150)
assert ring_widget.config.value == 100
# Set value below min
ring_widget.set_value(-10)
assert ring_widget.config.value == 0
def test_set_precision(ring_widget):
ring_widget.set_precision(2)
assert ring_widget.config.precision == 2
ring_widget.set_value(10.12345)
assert ring_widget.config.value == 10.12
def test_set_min_max_values(ring_widget):
ring_widget.set_min_max_values(10, 90)
assert ring_widget.config.min_value == 10
assert ring_widget.config.max_value == 90
def test_set_line_width(ring_widget):
ring_widget.set_line_width(25)
assert ring_widget.config.line_width == 25
def test_set_start_angle(ring_widget):
ring_widget.set_start_angle(180)
assert ring_widget.config.start_position == 180
###################################
# Color management tests
###################################
def test_set_color(ring_widget):
test_color = (255, 128, 64, 255)
ring_widget.set_color(test_color)
# Color is stored as hex string internally
assert ring_widget.color.getRgb() == test_color
def test_set_color_with_link_colors_updates_background(ring_widget):
# Enable color linking
ring_widget.config.link_colors = True
# Store original background
original_bg = ring_widget.background_color.getRgb()
test_color = (255, 100, 50, 255)
ring_widget.set_color(test_color)
# Background should be derived using subtle_background_color
bg_color = ring_widget.background_color
# Background should have changed
assert bg_color.getRgb() != original_bg
# Background should be different from the main color
assert bg_color.getRgb() != test_color
def test_set_background_when_colors_unlinked(ring_widget):
# Disable color linking
ring_widget.config.link_colors = False
test_bg = (100, 100, 100, 128)
ring_widget.set_background(test_bg)
assert ring_widget.background_color.getRgb() == test_bg
def test_set_background_when_colors_linked_does_nothing(ring_widget):
# Enable color linking
ring_widget.config.link_colors = True
original_bg = ring_widget.background_color.getRgb()
test_bg = (100, 100, 100, 128)
ring_widget.set_background(test_bg)
# Background should not change when colors are linked
assert ring_widget.background_color.getRgb() == original_bg
def test_color_link_derives_background(ring_widget):
ring_widget.config.link_colors = True
bright_color = QColor(255, 255, 0, 255) # Bright yellow
original_bg = ring_widget.background_color.getRgb()
ring_widget.set_color(bright_color.getRgb())
# Get the derived background color
bg_color = ring_widget.background_color
# Background should have changed
assert bg_color.getRgb() != original_bg
# Background should be a subtle blend, not the same as the main color
assert bg_color.getRgb() != bright_color.getRgb()
def test_convert_color_from_tuple(ring_widget):
color_tuple = (200, 150, 100, 255)
qcolor = ring_widget.convert_color(color_tuple)
assert isinstance(qcolor, QColor)
assert qcolor.getRgb() == color_tuple
def test_convert_color_from_hex_string(ring_widget):
hex_color = "#FF8040FF"
qcolor = ring_widget.convert_color(hex_color)
assert isinstance(qcolor, QColor)
assert qcolor.isValid()
###################################
# Gap property tests
###################################
def test_gap_property(ring_widget):
ring_widget.gap = 15
assert ring_widget.gap == 15
###################################
# Config validation tests
###################################
def test_config_default_values():
config = ProgressbarConfig()
assert config.value == 0
assert config.direction == -1
assert config.line_width == 20
assert config.start_position == 90
assert config.min_value == 0
assert config.max_value == 100
assert config.precision == 3
assert config.mode == "manual"
assert config.link_colors is True
def test_config_with_custom_values():
config = ProgressbarConfig(
value=50, direction=1, line_width=20, min_value=10, max_value=90, precision=2, mode="scan"
)
assert config.value == 50
assert config.direction == 1
assert config.line_width == 20
assert config.min_value == 10
assert config.max_value == 90
assert config.precision == 2
assert config.mode == "scan"
###################################
# set_direction tests
###################################
def test_set_direction_clockwise(ring_widget):
ring_widget.set_direction(-1)
assert ring_widget.config.direction == -1
def test_set_direction_counter_clockwise(ring_widget):
ring_widget.set_direction(1)
assert ring_widget.config.direction == 1
###################################
# _update_device_connection tests
###################################
def test_update_device_connection_with_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()
ring_widget._update_device_connection("samx", "progress")
# Should connect to device_progress endpoint
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
def test_update_device_connection_with_hinted_signal(ring_widget):
mock_device = FakeDevice(name="samx")
mock_device._info = {
"signals": {
"samx": {"obj_name": "samx", "signal_class": "SomeOtherSignal", "kind_str": "hinted"}
}
}
ring_widget.bec_dispatcher.client.device_manager.devices["samx"] = mock_device
ring_widget.bec_dispatcher.connect_slot = MagicMock()
ring_widget._update_device_connection("samx", "samx")
# Should connect to device_readback endpoint
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
def test_update_device_connection_no_device_manager(ring_widget):
ring_widget.bec_dispatcher.client.device_manager = None
with pytest.raises(ValueError) as excinfo:
ring_widget._update_device_connection("samx", "signal")
assert "Device manager is not available" in str(excinfo.value)
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")
###################################
# on_scan_progress tests
###################################
def test_on_scan_progress_updates_value(ring_widget):
msg = {"value": 42, "max_value": 100}
meta = {"RID": "test_rid_123"}
ring_widget.on_scan_progress(msg, meta)
assert ring_widget.config.value == 42
def test_on_scan_progress_updates_min_max_on_new_rid(ring_widget):
msg = {"value": 50, "max_value": 200}
meta = {"RID": "new_rid"}
ring_widget.RID = "old_rid"
ring_widget.on_scan_progress(msg, meta)
assert ring_widget.config.min_value == 0
assert ring_widget.config.max_value == 200
assert ring_widget.config.value == 50
def test_on_scan_progress_same_rid_no_min_max_update(ring_widget):
msg = {"value": 75, "max_value": 300}
meta = {"RID": "same_rid"}
ring_widget.RID = "same_rid"
ring_widget.set_min_max_values(0, 100)
ring_widget.on_scan_progress(msg, meta)
# Max value should not be updated when RID is the same
assert ring_widget.config.max_value == 100
assert ring_widget.config.value == 75
###################################
# on_device_readback tests
###################################
def test_on_device_readback_updates_value(ring_widget):
ring_widget.config.device = "samx"
ring_widget.config.signal = "readback"
msg = {"signals": {"readback": {"value": 12.34}}}
meta = {}
ring_widget.on_device_readback(msg, meta)
assert ring_widget.config.value == 12.34
def test_on_device_readback_uses_device_name_when_no_signal(ring_widget):
ring_widget.config.device = "samy"
ring_widget.config.signal = None
msg = {"signals": {"samy": {"value": 56.78}}}
meta = {}
ring_widget.on_device_readback(msg, meta)
assert ring_widget.config.value == 56.78
def test_on_device_readback_no_device_returns_early(ring_widget):
ring_widget.config.device = None
msg = {"signals": {"samx": {"value": 99.99}}}
meta = {}
initial_value = ring_widget.config.value
ring_widget.on_device_readback(msg, meta)
# Value should not change
assert ring_widget.config.value == initial_value
def test_on_device_readback_missing_signal_data(ring_widget):
ring_widget.config.device = "samx"
ring_widget.config.signal = "missing_signal"
msg = {"signals": {"other_signal": {"value": 11.11}}}
meta = {}
initial_value = ring_widget.config.value
ring_widget.on_device_readback(msg, meta)
# 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

66
tests/unit_tests/test_ring_progress_settings.py Normal file
View File

@@ -0,0 +1,66 @@
import pytest
from bec_widgets.utils.settings_dialog import SettingsDialog
from bec_widgets.widgets.progress.ring_progress_bar.ring_progress_bar import RingProgressBar
from bec_widgets.widgets.progress.ring_progress_bar.ring_progress_settings_cards import RingSettings
from tests.unit_tests.client_mocks import mocked_client
@pytest.fixture
def ring_progress_bar_widget(qtbot, mocked_client):
widget = RingProgressBar(client=mocked_client)
qtbot.addWidget(widget)
qtbot.waitExposed(widget)
yield widget
@pytest.fixture
def rpb_settings_dialog(qtbot, ring_progress_bar_widget):
settings = RingSettings(
parent=ring_progress_bar_widget, target_widget=ring_progress_bar_widget, popup=True
)
dialog = SettingsDialog(
ring_progress_bar_widget,
settings_widget=settings,
window_title="Ring Progress Bar Settings",
modal=False,
)
qtbot.addWidget(dialog)
qtbot.waitExposed(dialog)
yield dialog
@pytest.fixture
def rpb_settings_dialog_with_rings(qtbot, ring_progress_bar_widget):
ring_progress_bar_widget.add_ring()
ring_progress_bar_widget.add_ring()
settings = RingSettings(
parent=ring_progress_bar_widget, target_widget=ring_progress_bar_widget, popup=True
)
dialog = SettingsDialog(
ring_progress_bar_widget,
settings_widget=settings,
window_title="Ring Progress Bar Settings",
modal=False,
)
qtbot.addWidget(dialog)
qtbot.waitExposed(dialog)
yield dialog
def test_ring_progress_settings_dialog_opens(rpb_settings_dialog):
"""Test that the Ring Progress Bar settings dialog opens correctly."""
dialog = rpb_settings_dialog
dialog.show()
assert dialog.isVisible()
assert dialog.windowTitle() == "Ring Progress Bar Settings"
dialog.accept()
def test_ring_progress_settings_dialog_with_rings(rpb_settings_dialog_with_rings):
"""Test that the Ring Progress Bar settings dialog opens correctly with rings."""
dialog = rpb_settings_dialog_with_rings
dialog.show()
assert dialog.isVisible()
assert dialog.windowTitle() == "Ring Progress Bar Settings"
dialog.accept() # Close the dialog