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

feat: added @user_access from bec_lib.utils

Browse Source
This commit is contained in:
wyzula-jan
2024-02-15 11:21:21 +01:00
parent 60d150a411
commit b827e9eaa7
3 changed files with 165 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@ -10,12 +10,12 @@ from pydantic import Field
from pyqtgraph.Qt import uic
from qtpy.QtWidgets import QApplication, QWidget
from bec_lib.utils import user_access
from bec_widgets.utils import (
BECDispatcher,
BECConnector,
ConnectionConfig,
register_rpc_methods,
rpc_public,
)
from bec_widgets.widgets.plots import WidgetConfig, BECPlotBase, Waveform1DConfig, BECWaveform1D
@ -81,7 +81,6 @@ class WidgetHandler:
return widget
@register_rpc_methods
class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
def __init__(
self,
@ -106,6 +105,9 @@ class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
np.linspace(0, 10, 100), np.sin(np.linspace(0, 10, 100)), label="sin(x)"
)
# TODO debug 1dwaveform
self.add_widget(widget_type="Waveform1D", widget_id="widget_2", row=1, col=0)
# def show(self): # TODO check if useful for anything
# self.window = QMainWindow()
# self.window.setCentralWidget(self)
@ -115,10 +117,10 @@ class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
# if hasattr(self, "window"):
# self.window.close()
@rpc_public
@user_access
def add_widget(
self,
widget_type: str = "PlotBase",
widget_type: Literal["PlotBase", "Waveform1D"] = "PlotBase",
widget_id: str = None,
row: int = None,
col: int = None,
@ -128,7 +130,7 @@ class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
"""
Add a widget to the figure at the specified position.
Args:
widget_type(str): The type of the widget to add.
widget_type(Literal["PlotBase","Waveform1D"]): The type of the widget to add.
widget_id(str): The unique identifier of the widget. If not provided, a unique ID will be generated.
row(int): The row coordinate of the widget in the figure. If not provided, the next empty row will be used.
col(int): The column coordinate of the widget in the figure. If not provided, the next empty column will be used.
@ -171,7 +173,10 @@ class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
self.config.widgets[widget_id] = widget.config
self.widgets[widget_id] = widget
@rpc_public
# TODO rpc debug
print(f"Added widget {widget_id} at position ({row}, {col}).")
@user_access
def remove(
self,
row: int = None,
@ -319,8 +324,6 @@ class DebugWindow(QWidget):
self.figure = BECFigure(parent=self) # Create a new BECDeviceMonitor
self.glw_1_layout.addWidget(self.figure) # Add BECDeviceMonitor to the layout
print(f"USER_ACCESS for BECFigure: {self.figure.USER_ACCESS}")
self.console_layout = QVBoxLayout(self.widget_console)
self.console = JupyterConsoleWidget()
self.console_layout.addWidget(self.console)

35
bec_widgets/widgets/plots/plot_base.py
View File

@ -5,7 +5,8 @@ import numpy as np
from pydantic import BaseModel, Field
from qtpy.QtWidgets import QWidget
from bec_widgets.utils import BECConnector, ConnectionConfig, register_rpc_methods, rpc_public
from bec_lib.utils import user_access
from bec_widgets.utils import BECConnector, ConnectionConfig
class AxisConfig(BaseModel):
@ -33,7 +34,6 @@ class WidgetConfig(ConnectionConfig):
)
@register_rpc_methods
class BECPlotBase(BECConnector, pg.PlotItem):
def __init__(
self,
@ -50,7 +50,9 @@ class BECPlotBase(BECConnector, pg.PlotItem):
self.figure = parent_figure
@rpc_public
self.add_legend()
@user_access
def set(self, **kwargs) -> None:
"""
Set the properties of the plot widget.
@ -65,6 +67,7 @@ class BECPlotBase(BECConnector, pg.PlotItem):
- x_lim: tuple
- y_lim: tuple
"""
# TODO check functionality
# Mapping of keywords to setter methods
method_map = {
@ -84,17 +87,20 @@ class BECPlotBase(BECConnector, pg.PlotItem):
def apply_axis_config(self):
"""Apply the axis configuration to the plot widget."""
# TODO check functionality
config_mappings = {
"title": self.config.axis.title,
"x_label": self.config.axis.x_label,
"y_label": self.config.axis.y_label,
"x_scale": self.config.axis.x_scale,
"y_scale": self.config.axis.y_scale,
"x_lim": self.config.axis.x_lim,
"y_lim": self.config.axis.y_lim,
}
self.set(**{k: v for k, v in config_mappings.items() if v is not None})
@rpc_public
@user_access
def set_title(self, title: str):
"""
Set the title of the plot widget.
@ -104,7 +110,7 @@ class BECPlotBase(BECConnector, pg.PlotItem):
self.setTitle(title)
self.config.axis.title = title
@rpc_public
@user_access
def set_x_label(self, label: str):
"""
Set the label of the x-axis.
@ -114,7 +120,7 @@ class BECPlotBase(BECConnector, pg.PlotItem):
self.setLabel("bottom", label)
self.config.axis.x_label = label
@rpc_public
@user_access
def set_y_label(self, label: str):
"""
Set the label of the y-axis.
@ -124,7 +130,7 @@ class BECPlotBase(BECConnector, pg.PlotItem):
self.setLabel("left", label)
self.config.axis.y_label = label
@rpc_public
@user_access
def set_x_scale(self, scale: Literal["linear", "log"] = "linear"):
"""
Set the scale of the x-axis.
@ -134,7 +140,7 @@ class BECPlotBase(BECConnector, pg.PlotItem):
self.setLogMode(x=(scale == "log"))
self.config.axis.x_scale = scale
@rpc_public
@user_access
def set_y_scale(self, scale: Literal["linear", "log"] = "linear"):
"""
Set the scale of the y-axis.
@ -144,7 +150,7 @@ class BECPlotBase(BECConnector, pg.PlotItem):
self.setLogMode(y=(scale == "log"))
self.config.axis.y_scale = scale
@rpc_public
@user_access
def set_x_lim(self, x_lim: tuple) -> None:
"""
Set the limits of the x-axis.
@ -154,7 +160,7 @@ class BECPlotBase(BECConnector, pg.PlotItem):
self.setXRange(x_lim[0], x_lim[1])
self.config.axis.x_lim = x_lim
@rpc_public
@user_access
def set_y_lim(self, y_lim: tuple) -> None:
"""
Set the limits of the y-axis.
@ -164,7 +170,7 @@ class BECPlotBase(BECConnector, pg.PlotItem):
self.setYRange(y_lim[0], y_lim[1])
self.config.axis.y_lim = y_lim
@rpc_public
@user_access
def set_grid(self, x: bool = False, y: bool = False):
"""
Set the grid of the plot widget.
@ -176,7 +182,10 @@ class BECPlotBase(BECConnector, pg.PlotItem):
self.config.axis.x_grid = x
self.config.axis.y_grid = y
@rpc_public
def add_legend(self):
self.addLegend()
@user_access
def plot_data(self, data_x: list | np.ndarray, data_y: list | np.ndarray, **kwargs):
"""
Plot custom data on the plot widget. These data are not saved in config.
@ -189,7 +198,7 @@ class BECPlotBase(BECConnector, pg.PlotItem):
# TODO decide name of the method
self.plot(data_x, data_y, **kwargs)
@rpc_public
@user_access
def remove(self):
"""Remove the plot widget from the figure."""
if self.figure is not None:

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

@ -1,22 +1,29 @@
from typing import Literal, Optional
from collections import defaultdict
from typing import Literal, Optional, Any
import pyqtgraph as pg
from pydantic import Field, BaseModel
from pyqtgraph import mkBrush
from qtpy.QtWidgets import QWidget
from qtpy.QtCore import Slot as pyqtSlot
from qtpy.QtCore import Signal as pyqtSignal
from qtpy.QtGui import QColor
from bec_lib.utils import user_access
from bec_lib import MessageEndpoints
from bec_widgets.utils import Colors
from bec_widgets.widgets.plots import BECPlotBase, WidgetConfig
class SignalData(BaseModel):
"""The data configuration of a signal in the 1D waveform widget for x and y axis."""
# TODO add validator on name and entry
name: str
entry: str
unit: Optional[str] # todo implement later
modifier: Optional[str] # todo implement later
unit: Optional[str] = None # todo implement later
modifier: Optional[str] = None # todo implement later
class Signal(BaseModel):
@ -29,12 +36,12 @@ class Signal(BaseModel):
class CurveConfig(BaseModel):
label: Optional[str] = Field(None, description="The label of the curve.")
color: Optional[str] = Field(None, description="The color of the curve.")
color: Optional[Any] = Field(None, description="The color of the curve.")
symbol: Optional[str] = Field(None, description="The symbol of the curve.")
symbol_size: Optional[int] = Field(None, description="The size of the symbol of the curve.")
pen_width: Optional[int] = Field(None, description="The width of the pen of the curve.")
symbol_size: Optional[int] = Field(5, description="The size of the symbol of the curve.")
pen_width: Optional[int] = Field(2, description="The width of the pen of the curve.")
pen_style: Optional[Literal["solid", "dash", "dot", "dashdot"]] = Field(
None, description="The style of the pen of the curve."
"solid", description="The style of the pen of the curve."
) # TODO check if valid
source: Optional[str] = Field(
None, description="The source of the curve."
@ -51,33 +58,51 @@ class Waveform1DConfig(WidgetConfig):
) # todo maybe dict??
class BECCurve(pg.PlotDataItem):
class BECCurve(pg.PlotDataItem): # TODO decide what will be accessible from the parent
def __init__(
self,
config: Optional[CurveConfig] = None,
**kwargs,
):
# if config is None: #TODO custom later
# config = CurveConfig(widget_class=self.__class__.__name__)
super().__init__(**kwargs)
if config is None:
config = CurveConfig(widget_class=self.__class__.__name__)
self.config = config
self.apply_config()
def apply_config(self):
self.setPen(self.config.color)
self.setSymbol(self.config.symbol)
# self.setSymbolSize(self.config.symbol_size)
# self.setSymbolBrush(self.config.color)
# self.setPenWidth(self.config.pen_width)
class BECWaveform1D(BECPlotBase):
scan_signal_update = pyqtSignal()
def __init__(
self,
parent: Optional[QWidget] = None,
parent_figure=None,
config: Optional[Waveform1DConfig] = None,
client=None,
gui_id: Optional[str] = None,
):
if config is None:
config = Waveform1DConfig(widget_class=self.__class__.__name__)
super().__init__(parent=parent, config=config, client=client, gui_id=gui_id)
super().__init__(
parent=parent, parent_figure=parent_figure, config=config, client=client, gui_id=gui_id
)
self.curves = {}
# self.curves = {}
self.curve_data = defaultdict(dict)
self.scanID = None
# TODO add proxy later when update function is ready
self.proxy_update_plot = pg.SignalProxy(
self.update_signal, rateLimit=25, slot=self.update_scan_segment_plot
self.scan_signal_update, rateLimit=25, slot=self.update_scan_segment_plot
)
# Get bec shortcuts dev, scans, queue, scan_storage, dap
@ -86,19 +111,91 @@ class BECWaveform1D(BECPlotBase):
# Connect dispatcher signals
self.bec_dispatcher.connect_slot(self.on_scan_segment, MessageEndpoints.scan_segment())
# TODO DEbug
# self.add_curve("bpm4i")
self.add_scan("samx", "samx", "bpm4i", "bpm4i")
self.addLegend()
def add_curve_by_config(self, curve_config: CurveConfig):
# TODO something like this
curve = BECCurve()
self.curves[curve_config.label] = curve
self.addItem(curve)
def add_curve(self, curve_id: str):
curve = BECCurve()
self.curves[curve_id] = curve
def save_curve_config(self): ...
@user_access
def add_scan(
self,
x_name: str,
x_entry: str,
y_name: str,
y_entry: str,
color: Optional[str] = None,
label: Optional[str] = None,
symbol: Optional[str] = None,
symbol_size: Optional[int] = None,
symbol_color: Optional[str] = None,
pen_width: Optional[int] = None,
pen_style: Optional[Literal["solid", "dash", "dot", "dashdot"]] = None,
):
# Check if curve already exists
curve_source = "scan_segment"
curve_id = (x_name, x_entry, y_name, y_entry)
if curve_id in self.curve_data[curve_source]:
raise ValueError(f"Curve with ID {curve_id} already exists in widget {self.gui_id}.")
# Generate curve properties if not given
if label is None:
label = f"{y_name}-{y_entry}"
if color is None:
color = Colors.golden_angle_color(
colormap=self.config.color_palette, num=len(self.curves) + 1
)[-1]
# color_brush = mkBrush(color)
if symbol_color is None:
symbol_color = color
# Create curve by config
curve_config = CurveConfig(
label=label,
color=color,
symbol=symbol,
symbol_size=symbol_size,
symbol_color=symbol_color,
pen_width=pen_width,
pen_style=pen_style,
source=curve_source,
signals=Signal(
source=curve_source,
x=SignalData(name=x_name, entry=x_entry),
y=SignalData(name=y_name, entry=y_entry),
),
)
curve = BECCurve(config=curve_config, name=label)
self.curve_data[curve_source][curve_id] = curve
self.addItem(curve)
def update_curve(self, curve_id: str, x, y):
self.curves[curve_id].setData(x, y)
# def _create_bec_curve(self, curve_config: CurveConfig, source: str = "scan_segment"):
# curve = BECCurve(config=curve_config)
# #TODO add checkign if curve already exists
# self.curve_data[source][curve_config.label] = curve
# return curve
def add_source(self, source: str):
# TODO general function to add different sources
# self.curve_data[source]
pass
def add_curve(self, curve_id: str, source: str = None):
# curve = BECCurve()
# curve = pg.PlotDataItem(name=curve_id)
curve = BECCurve(name=curve_id)
# self.curves[curve_id] = curve
self.addItem(curve)
def update_curve(self, source: str, curve_id: tuple, x, y): ...
@pyqtSlot(dict, dict)
def on_scan_segment(self, msg: dict, metadata: dict):
@ -118,8 +215,20 @@ class BECWaveform1D(BECPlotBase):
self.scanID = current_scanID
self.scan_data = self.queue.scan_storage.find_scan_by_ID(self.scanID)
scan_data_current = self.scan_data.data
data_x = scan_data_current["samx"]["samx"]
data_y = scan_data_current["bpm4i"]["bpm4i"]
# self.scan_signal_update.emit()
# self.update_from_storage(data=self.scan_data)
self.update_scan_segment_plot()
self.update_curve("bpm4i", data_x, data_y)
def update_scan_segment_plot(self):
data = self.scan_data.data
for curve_id, curve in self.curve_data["scan_segment"].items():
x_name = curve_id[0]
x_entry = curve_id[1]
y_name = curve_id[2]
y_entry = curve_id[3]
data_x = data[x_name][x_entry].val
data_y = data[y_name][y_entry].val
curve.setData(data_x, data_y)