diff --git a/bec_widgets/utils/__init__.py b/bec_widgets/utils/__init__.py index 80a64492..a505a639 100644 --- a/bec_widgets/utils/__init__.py +++ b/bec_widgets/utils/__init__.py @@ -5,3 +5,4 @@ from .bec_table import BECTable from .bec_connector import BECConnector, ConnectionConfig from .bec_dispatcher import BECDispatcher from .rpc_decorator import rpc_public, register_rpc_methods +from .entry_validator import EntryValidator diff --git a/bec_widgets/utils/entry_validator.py b/bec_widgets/utils/entry_validator.py new file mode 100644 index 00000000..98ee1fbf --- /dev/null +++ b/bec_widgets/utils/entry_validator.py @@ -0,0 +1,17 @@ +class EntryValidator: + def __init__(self, devices): + self.devices = devices + + def validate_signal(self, name: str, entry: str = None) -> str: + if name not in self.devices: + raise ValueError(f"Device '{name}' not found in current BEC session") + + device = self.devices[name] + description = device.describe() + + if entry is None: + entry = next(iter(device._hints), name) if hasattr(device, "_hints") else name + if entry not in description: + raise ValueError(f"Entry '{entry}' not found in device '{name}' signals") + + return entry diff --git a/bec_widgets/widgets/figure/figure.py b/bec_widgets/widgets/figure/figure.py index 1d037449..d159a83b 100644 --- a/bec_widgets/widgets/figure/figure.py +++ b/bec_widgets/widgets/figure/figure.py @@ -450,7 +450,7 @@ class DebugWindow(QWidget): self.w4 = self.figure[1, 1] # curves for w1 - self.w1.add_curve_scan("samx", "samx", "bpm4i", "bpm4i", pen_style="dash") + self.w1.add_curve_scan("samx", "bpm4i", pen_style="dash") self.w1.add_curve_custom( x=[1, 2, 3, 4, 5], y=[1, 2, 3, 4, 5], @@ -460,14 +460,14 @@ class DebugWindow(QWidget): ) # curves for w2 - self.w2.add_curve_scan("samx", "samx", "bpm3a", "bpm3a", pen_style="solid") - self.w2.add_curve_scan("samx", "samx", "bpm4d", "bpm4d", pen_style="dot") + self.w2.add_curve_scan("samx", "bpm3a", pen_style="solid") + self.w2.add_curve_scan("samx", "bpm4d", pen_style="dot") self.w2.add_curve_custom( x=[1, 2, 3, 4, 5], y=[5, 4, 3, 2, 1], color="red", pen_style="dashdot" ) # curves for w3 - self.w3.add_curve_scan("samx", "samx", "bpm4i", "bpm4i", pen_style="dash") + self.w3.add_curve_scan("samx", "bpm4i", pen_style="dash") self.w3.add_curve_custom( x=[1, 2, 3, 4, 5], y=[1, 2, 3, 4, 5], @@ -477,7 +477,7 @@ class DebugWindow(QWidget): ) # curves for w4 - self.w4.add_curve_scan("samx", "samx", "bpm4i", "bpm4i", pen_style="dash") + self.w4.add_curve_scan("samx", "bpm4i", pen_style="dash") self.w4.add_curve_custom( x=[1, 2, 3, 4, 5], y=[1, 2, 3, 4, 5], diff --git a/bec_widgets/widgets/plots/waveform1d.py b/bec_widgets/widgets/plots/waveform1d.py index 017ea542..74976e8c 100644 --- a/bec_widgets/widgets/plots/waveform1d.py +++ b/bec_widgets/widgets/plots/waveform1d.py @@ -5,7 +5,7 @@ from typing import Literal, Optional, Any import numpy as np import pyqtgraph as pg -from pydantic import Field, BaseModel +from pydantic import Field, BaseModel, field_validator from pyqtgraph import mkBrush from qtpy import QtCore from qtpy.QtCore import Signal as pyqtSignal @@ -14,7 +14,7 @@ from qtpy.QtWidgets import QWidget from bec_lib import MessageEndpoints from bec_lib.scan_data import ScanData -from bec_widgets.utils import Colors, ConnectionConfig, BECConnector +from bec_widgets.utils import Colors, ConnectionConfig, BECConnector, EntryValidator from bec_widgets.widgets.plots import BECPlotBase, WidgetConfig @@ -90,8 +90,6 @@ class BECCurve(BECConnector, pg.PlotDataItem): super().__init__(config=config, gui_id=gui_id) pg.PlotDataItem.__init__(self, name=name, **kwargs) - # self.config = config - self.apply_config() def apply_config(self): @@ -246,6 +244,8 @@ class BECWaveform1D(BECPlotBase): # Connect dispatcher signals self.bec_dispatcher.connect_slot(self.on_scan_segment, MessageEndpoints.scan_segment()) + self.entry_validator = EntryValidator(self.dev) + self.addLegend() self.apply_config() @@ -321,64 +321,6 @@ class BECWaveform1D(BECPlotBase): "Each identifier must be either an integer (index) or a string (curve_id)." ) - def add_curve_scan( - self, - x_name: str, - x_entry: str, - y_name: str, - y_entry: str, - color: Optional[str] = None, - label: Optional[str] = None, - **kwargs, - ) -> BECCurve: - """ - Add a curve to the plot widget from the scan segment. - Args: - x_name(str): Name of the x signal. - x_entry(str): Entry of the x signal. - y_name(str): Name of the y signal. - y_entry(str): Entry of the y signal. - color(str, optional): Color of the curve. Defaults to None. - label(str, optional): Label of the curve. Defaults to None. - **kwargs: Additional keyword arguments for the curve configuration. - - Returns: - BECCurve: The curve object. - """ - # Check if curve already exists - curve_source = "scan_segment" - label = label or f"{y_name}-{y_entry}" - - curve_exits = self._check_curve_id(label, self.curves_data) - if curve_exits: - raise ValueError(f"Curve with ID '{label}' already exists in widget '{self.gui_id}'.") - return - - color = ( - color - or Colors.golden_angle_color( - colormap=self.config.color_palette, num=len(self.curves) + 1, format="HEX" - )[-1] - ) - - # Create curve by config - curve_config = CurveConfig( - widget_class="BECCurve", - # parent_id=self.config.parent_id, - parent_id=self.gui_id, - label=label, - color=color, - source=curve_source, - signals=Signal( - source=curve_source, - x=SignalData(name=x_name, entry=x_entry), - y=SignalData(name=y_name, entry=y_entry), - ), - **kwargs, - ) - curve = self._add_curve_object(name=label, source=curve_source, config=curve_config) - return curve - def add_curve_custom( self, x: list | np.ndarray, @@ -455,6 +397,98 @@ class BECWaveform1D(BECPlotBase): curve.setData(data[0], data[1]) return curve + def add_curve_scan( + self, + x_name: str, + y_name: str, + x_entry: Optional[str] = None, + y_entry: Optional[str] = None, + color: Optional[str] = None, + label: Optional[str] = None, + validate_bec: bool = True, + **kwargs, + ) -> BECCurve: + """ + Add a curve to the plot widget from the scan segment. + Args: + x_name(str): Name of the x signal. + x_entry(str): Entry of the x signal. + y_name(str): Name of the y signal. + y_entry(str): Entry of the y signal. + color(str, optional): Color of the curve. Defaults to None. + label(str, optional): Label of the curve. Defaults to None. + **kwargs: Additional keyword arguments for the curve configuration. + + Returns: + BECCurve: The curve object. + """ + # Check if curve already exists + curve_source = "scan_segment" + + # Get entry if not provided and validate + x_entry, y_entry = self._validate_signal_entries( + x_name, y_name, x_entry, y_entry, validate_bec + ) + + label = label or f"{y_name}-{y_entry}" + + curve_exits = self._check_curve_id(label, self.curves_data) + if curve_exits: + raise ValueError(f"Curve with ID '{label}' already exists in widget '{self.gui_id}'.") + return + + color = ( + color + or Colors.golden_angle_color( + colormap=self.config.color_palette, num=len(self.curves) + 1, format="HEX" + )[-1] + ) + + # Create curve by config + curve_config = CurveConfig( + widget_class="BECCurve", + # parent_id=self.config.parent_id, + parent_id=self.gui_id, + label=label, + color=color, + source=curve_source, + signals=Signal( + source=curve_source, + x=SignalData(name=x_name, entry=x_entry), + y=SignalData(name=y_name, entry=y_entry), + ), + **kwargs, + ) + curve = self._add_curve_object(name=label, source=curve_source, config=curve_config) + return curve + + def _validate_signal_entries( + self, + x_name: str, + y_name: str, + x_entry: str | None, + y_entry: str | None, + validate_bec: bool = True, + ) -> tuple[str, str]: + """ + Validate the signal name and entry. + Args: + x_name(str): Name of the x signal. + y_name(str): Name of the y signal. + x_entry(str|None): Entry of the x signal. + y_entry(str|None): Entry of the y signal. + validate_bec(bool, optional): If True, validate the signal with BEC. Defaults to True. + Returns: + tuple[str,str]: Validated x and y entries. + """ + if validate_bec: + x_entry = self.entry_validator.validate_signal(x_name, x_entry) + y_entry = self.entry_validator.validate_signal(y_name, y_entry) + else: + x_entry = x_name if x_entry is None else x_entry + y_entry = y_name if y_entry is None else y_entry + return x_entry, y_entry + def _check_curve_id(self, val: Any, dict_to_check: dict) -> bool: """ Check if val is in the values of the dict_to_check or in the values of the nested dictionaries.