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

feat(utils/entry_validator): possibility to validate add_scan_curve with current BEC session

Browse Source
This commit is contained in:
wyzula-jan
2024-02-21 15:21:14 +01:00
parent 99dce077c4
commit 1db77b969b
4 changed files with 119 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
bec_widgets/utils/__init__.py
View File

@ -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

17
bec_widgets/utils/entry_validator.py Normal file
View File

@ -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

10
bec_widgets/widgets/figure/figure.py
View File

@ -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],

158
bec_widgets/widgets/plots/waveform1d.py
View File

@ -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.