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

refactor(plot/Waveform1D,plot/BECCurve): BECCurve inherits from BECConnector and can refer to parent_id (Waveform1D) and has its own gui_id

This commit is contained in:
wyzula-jan
2024-02-21 14:26:46 +01:00
parent 402adc44e8
commit 99dce077c4
10 changed files with 511 additions and 207 deletions

View File

@ -4,9 +4,108 @@ from bec_widgets.cli.client_utils import rpc_call, RPCBase, BECFigureClientMixin
from typing import Literal, Optional, overload
class BECPlotBase(RPCBase):
@rpc_call
def set(self, **kwargs) -> "None":
"""
Set the properties of the plot widget.
Args:
**kwargs: Keyword arguments for the properties to be set.
Possible properties:
- title: str
- x_label: str
- y_label: str
- x_scale: Literal["linear", "log"]
- y_scale: Literal["linear", "log"]
- x_lim: tuple
- y_lim: tuple
"""
@rpc_call
def set_title(self, title: "str"):
"""
Set the title of the plot widget.
Args:
title(str): Title of the plot widget.
"""
@rpc_call
def set_x_label(self, label: "str"):
"""
Set the label of the x-axis.
Args:
label(str): Label of the x-axis.
"""
@rpc_call
def set_y_label(self, label: "str"):
"""
Set the label of the y-axis.
Args:
label(str): Label of the y-axis.
"""
@rpc_call
def set_x_scale(self, scale: "Literal['linear', 'log']" = "linear"):
"""
Set the scale of the x-axis.
Args:
scale(Literal["linear", "log"]): Scale of the x-axis.
"""
@rpc_call
def set_y_scale(self, scale: "Literal['linear', 'log']" = "linear"):
"""
Set the scale of the y-axis.
Args:
scale(Literal["linear", "log"]): Scale of the y-axis.
"""
@rpc_call
def set_x_lim(self, x_lim: "tuple") -> "None":
"""
Set the limits of the x-axis.
Args:
x_lim(tuple): Limits of the x-axis.
"""
@rpc_call
def set_y_lim(self, y_lim: "tuple") -> "None":
"""
Set the limits of the y-axis.
Args:
y_lim(tuple): Limits of the y-axis.
"""
@rpc_call
def set_grid(self, x: "bool" = False, y: "bool" = False):
"""
Set the grid of the plot widget.
Args:
x(bool): Show grid on the x-axis.
y(bool): Show grid on the y-axis.
"""
@rpc_call
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.
Args:
data_x(list|np.ndarray): x-axis data
data_y(list|np.ndarray): y-axis data
**kwargs: Keyword arguments for the plot.
"""
@rpc_call
def remove(self):
"""
Remove the plot widget from the figure.
"""
class BECWaveform1D(RPCBase):
@rpc_call
def add_scan(
def add_curve_scan(
self,
x_name: "str",
x_entry: "str",
@ -15,22 +114,42 @@ class BECWaveform1D(RPCBase):
color: "Optional[str]" = None,
label: "Optional[str]" = None,
**kwargs
):
) -> "BECCurve":
"""
None
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.
"""
@rpc_call
def add_curve(
def add_curve_custom(
self,
x: "list | np.ndarray",
y: "list | np.ndarray",
label: "str" = None,
color: "str" = None,
**kwargs
):
) -> "BECCurve":
"""
None
Add a custom data curve to the plot widget.
Args:
x(list|np.ndarray): X data of the curve.
y(list|np.ndarray): Y data of the curve.
label(str, optional): Label of the curve. Defaults to None.
color(str, optional): Color of the curve. Defaults to None.
**kwargs: Additional keyword arguments for the curve configuration.
Returns:
BECCurve: The curve object.
"""
@rpc_call
@ -42,7 +161,7 @@ class BECWaveform1D(RPCBase):
"""
@rpc_call
def update_scan_curve_history(self, scanID: "str" = None, scan_index: "int" = None):
def scan_history(self, scan_index: "int" = None, scanID: "str" = None):
"""
Update the scan curves with the data from the scan storage.
Provide only one of scanID or scan_index.
@ -67,6 +186,26 @@ class BECWaveform1D(RPCBase):
dict: Dictionary of curves data.
"""
@rpc_call
def get_curve(self, identifier) -> "BECCurve":
"""
Get the curve by its index or ID.
Args:
identifier(int|str): Identifier of the curve. Can be either an integer (index) or a string (curve_id).
Returns:
BECCurve: The curve object.
"""
@rpc_call
def get_curve_config(self, curve_id: "str", dict_output: "bool" = True) -> "CurveConfig | dict":
"""
Get the configuration of a curve by its ID.
Args:
curve_id(str): ID of the curve.
Returns:
CurveConfig|dict: Configuration of the curve.
"""
class BECFigure(RPCBase, BECFigureClientMixin):
@rpc_call
@ -79,7 +218,13 @@ class BECFigure(RPCBase, BECFigureClientMixin):
**axis_kwargs
) -> "BECWaveform1D":
"""
None
Add a Waveform1D plot to the figure at the specified position.
Args:
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.
config(dict): Additional configuration for the widget.
**axis_kwargs(dict): Additional axis properties to set on the widget after creation.
"""
@rpc_call
@ -98,3 +243,86 @@ class BECFigure(RPCBase, BECFigureClientMixin):
widget_id(str): The unique identifier of the widget to remove.
coordinates(tuple[int, int], optional): The coordinates of the widget to remove.
"""
@rpc_call
def change_layout(self, max_columns=None, max_rows=None):
"""
Reshuffle the layout of the figure to adjust to a new number of max_columns or max_rows.
If both max_columns and max_rows are provided, max_rows is ignored.
Args:
max_columns (Optional[int]): The new maximum number of columns in the figure.
max_rows (Optional[int]): The new maximum number of rows in the figure.
"""
class BECCurve(RPCBase):
@rpc_call
def set(self, **kwargs):
"""
Set the properties of the curve.
Args:
**kwargs: Keyword arguments for the properties to be set.
Possible properties:
- color: str
- symbol: str
- symbol_color: str
- symbol_size: int
- pen_width: int
- pen_style: Literal["solid", "dash", "dot", "dashdot"]
"""
@rpc_call
def set_data(self, x, y):
"""
None
"""
@rpc_call
def set_color(self, color: "str", symbol_color: "Optional[str]" = None):
"""
Change the color of the curve.
Args:
color(str): Color of the curve.
symbol_color(str, optional): Color of the symbol. Defaults to None.
"""
@rpc_call
def set_symbol(self, symbol: "str"):
"""
Change the symbol of the curve.
Args:
symbol(str): Symbol of the curve.
"""
@rpc_call
def set_symbol_color(self, symbol_color: "str"):
"""
Change the symbol color of the curve.
Args:
symbol_color(str): Color of the symbol.
"""
@rpc_call
def set_symbol_size(self, symbol_size: "int"):
"""
Change the symbol size of the curve.
Args:
symbol_size(int): Size of the symbol.
"""
@rpc_call
def set_pen_width(self, pen_width: "int"):
"""
Change the pen width of the curve.
Args:
pen_width(int): Width of the pen.
"""
@rpc_call
def set_pen_style(self, pen_style: "Literal['solid', 'dash', 'dot', 'dashdot']"):
"""
Change the pen style of the curve.
Args:
pen_style(Literal["solid", "dash", "dot", "dashdot"]): Style of the pen.
"""

View File

@ -88,7 +88,6 @@ class BECFigureClientMixin:
class RPCBase:
def __init__(self, gui_id: str = None, config: dict = None, **kwargs) -> None:
self._client = BECDispatcher().client
self._config = config if config is not None else {}
@ -116,7 +115,7 @@ class RPCBase:
metadata={"request_id": request_id},
)
print(f"RPCBase: {rpc_msg}")
receiver = self._config.get("parent_figure_id", self._gui_id)
receiver = self._config.get("parent_id", self._gui_id)
self._client.producer.send(MessageEndpoints.gui_instructions(receiver), rpc_msg)
if not wait_for_rpc_response:

View File

@ -3,7 +3,6 @@ import typing
class ClientGenerator:
def __init__(self):
self.header = """# This file was automatically generated by generate_cli.py\n
from bec_widgets.cli.client_utils import rpc_call, RPCBase, BECFigureClientMixin
@ -75,10 +74,14 @@ class {class_name}(RPCBase):"""
if __name__ == "__main__":
import os
from bec_widgets.widgets.figure import BECFigure
from bec_widgets.widgets.plots import BECWaveform1D
from bec_widgets.widgets.plots import BECWaveform1D, BECPlotBase # ,BECCurve
from bec_widgets.widgets.plots.waveform1d import BECCurve
clss = [BECWaveform1D, BECFigure]
current_path = os.path.dirname(__file__)
client_path = os.path.join(current_path, "client.py")
clss = [BECPlotBase, BECWaveform1D, BECFigure, BECCurve]
generator = ClientGenerator()
generator.generate_client(clss)
generator.write("bec_widgets/cli/client.py")
generator.write(client_path)

View File

@ -4,11 +4,11 @@ from bec_lib import MessageEndpoints, messages
from bec_widgets.utils import BECDispatcher
from bec_widgets.widgets.figure import BECFigure
from bec_widgets.widgets.plots import BECPlotBase, BECWaveform1D
from bec_widgets.widgets.plots import BECPlotBase, BECWaveform1D, BECCurve
class BECWidgetsCLIServer:
WIDGETS = [BECWaveform1D, BECFigure]
WIDGETS = [BECWaveform1D, BECFigure, BECCurve]
def __init__(self, gui_id: str = None) -> None:
self.dispatcher = BECDispatcher()
@ -82,3 +82,4 @@ if __name__ == "__main__":
args = parser.parse_args()
server = BECWidgetsCLIServer(gui_id=args.id)
# server = BECWidgetsCLIServer(gui_id="test")

View File

@ -93,9 +93,15 @@ class BECConnector:
self.config = config
def get_config(self):
return self.config
# connector = BECConnector()
# print(connector.config)
def get_config(self, dict_output: bool = True) -> dict | BaseModel:
"""
Get the configuration of the widget.
Args:
dict_output(bool): If True, return the configuration as a dictionary. If False, return the configuration as a pydantic model.
Returns:
dict: The configuration of the plot widget.
"""
if dict_output:
return self.config.model_dump()
else:
return self.config

View File

@ -11,3 +11,5 @@ from .motor_control import (
MotorThread,
MotorCoordinateTable,
)
from .figure import FigureConfig, BECFigure
from .plots import BECWaveform1D, BECCurve, BECPlotBase

View File

@ -3,19 +3,15 @@ from __future__ import annotations
import itertools
import os
import sys
from typing import Literal, Optional, overload
from collections import defaultdict
from dataclasses import dataclass
from typing import Literal, Optional
import numpy as np
import pyqtgraph as pg
from bec_lib.utils import user_access
from qtpy.QtWidgets import QVBoxLayout, QMainWindow
from pydantic import Field
from pyqtgraph.Qt import uic
from qtpy.QtWidgets import QApplication, QWidget
from qtpy.QtWidgets import QVBoxLayout, QMainWindow
from bec_widgets.utils import BECConnector, BECDispatcher, ConnectionConfig
from bec_widgets.widgets.plots import BECPlotBase, BECWaveform1D, Waveform1DConfig, WidgetConfig
@ -46,7 +42,7 @@ class WidgetHandler:
widget_type: str,
widget_id: str,
parent_figure,
parent_figure_id: str,
parent_id: str,
config: dict = None,
**axis_kwargs,
) -> BECPlotBase:
@ -56,7 +52,7 @@ class WidgetHandler:
Args:
widget_type (str): The type of the widget to create.
widget_id (str): Unique identifier for the widget.
parent_figure_id (str): Identifier of the parent figure.
parent_id (str): Identifier of the parent figure.
config (dict, optional): Additional configuration for the widget.
**axis_kwargs: Additional axis properties to set on the widget after creation.
@ -70,7 +66,7 @@ class WidgetHandler:
widget_class, config_class = entry
widget_config_dict = {
"widget_class": widget_class.__name__,
"parent_figure_id": parent_figure_id,
"parent_id": parent_id,
"gui_id": widget_id,
**(config if config is not None else {}),
}
@ -84,7 +80,7 @@ class WidgetHandler:
class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
USER_ACCESS = ["add_widget", "remove"]
USER_ACCESS = ["add_plot", "remove", "change_layout"]
def __init__(
self,
@ -98,95 +94,36 @@ class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
else:
self.config = config
super().__init__(client=client, config=config, gui_id=gui_id)
pg.GraphicsLayoutWidget.__init__(self, parent) # in case of inheritance
pg.GraphicsLayoutWidget.__init__(self, parent)
self.widget_handler = WidgetHandler()
# Widget container to reference widgets by 'widget_id'
self.widgets = defaultdict(dict)
# Container to keep track of the grid
self.grid = []
def change_grid(self, widget_id: str, row: int, col: int):
def add_plot(
self, widget_id: str = None, row: int = None, col: int = None, config=None, **axis_kwargs
) -> BECWaveform1D:
"""
Change the grid to reflect the new position of the widget.
Add a Waveform1D plot to the figure at the specified position.
Args:
widget_id(str): The unique identifier of the widget.
row(int): The new row coordinate of the widget in the figure.
col(int): The new column coordinate of the widget in the figure.
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.
config(dict): Additional configuration for the widget.
**axis_kwargs(dict): Additional axis properties to set on the widget after creation.
"""
while len(self.grid) <= row:
self.grid.append([])
row = self.grid[row]
while len(row) <= col:
row.append(None)
row[col] = widget_id
def reindex_grid(self):
"""Reindex the grid to remove empty rows and columns."""
print(f"old grid: {self.grid}")
new_grid = []
for row in self.grid:
new_row = [widget for widget in row if widget is not None]
if new_row:
new_grid.append(new_row)
#
# Update the config of each object to reflect its new position
for row_idx, row in enumerate(new_grid):
for col_idx, widget in enumerate(row):
self.widgets[widget].config.row, self.widgets[widget].config.col = row_idx, col_idx
self.grid = new_grid
self.replot_layout()
def replot_layout(self):
"""Replot the layout based on the current grid configuration."""
self.clear()
for row_idx, row in enumerate(self.grid):
for col_idx, widget in enumerate(row):
self.addItem(self.widgets[widget], row=row_idx, col=col_idx)
@overload
def add_widget(
self,
widget_type: Literal["Waveform1D"] = "Waveform1D",
widget_id: str = ...,
row: int = ...,
col: int = ...,
config: dict = ...,
**axis_kwargs,
) -> BECWaveform1D: ...
@overload
def add_widget(
self,
widget_type: Literal["PlotBase"] = "PlotBase",
widget_id: str = ...,
row: int = ...,
col: int = ...,
config: dict = ...,
**axis_kwargs,
) -> BECPlotBase: ...
# @overload
# def add_widget(
# self,
# widget_type: Literal["Waveform1D"] = "Waveform1D",
# widget_id: str = None,
# row: int = None,
# col: int = None,
# config: dict = None,
# **axis_kwargs,
# ) -> BECWaveform1D: ...
# @overload
# def add_widget(
# self,
# widget_type: Literal["PlotBase"] = "PlotBase",
# widget_id: str = None,
# row: int = None,
# col: int = None,
# config: dict = None,
# **axis_kwargs,
# ) -> BECPlotBase: ...
return self.add_widget(
widget_type="Waveform1D",
widget_id=widget_id,
row=row,
col=col,
config=config,
**axis_kwargs,
)
def add_widget(
self,
@ -216,7 +153,7 @@ class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
widget_type=widget_type,
widget_id=widget_id,
parent_figure=self,
parent_figure_id=self.gui_id,
parent_id=self.gui_id,
config=config,
**axis_kwargs,
)
@ -239,21 +176,17 @@ class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
# Add widget to the figure
self.addItem(widget, row=row, col=col)
#
# TODO decide if needed
# Update num_columns and num_rows based on the added widget
self.config.num_rows = max(self.config.num_rows, row + 1)
self.config.num_columns = max(self.config.num_columns, col + 1)
# By default, set the title of the widget to its unique identifier #TODO will be removed after debugging
widget.set_title(f"{widget_id}")
# Saving config for future referencing
self.config.widgets[widget_id] = widget.config
self.widgets[widget_id] = widget
# Reflect the grid coordinates
self.change_grid(widget_id, row, col)
self._change_grid(widget_id, row, col)
return widget
@ -308,7 +241,7 @@ class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
widget = self.widgets.pop(widget_id)
self.removeItem(widget)
self.grid[widget.config.row][widget.config.col] = None
self.reindex_grid()
self._reindex_grid()
if widget_id in self.config.widgets:
self.config.widgets.pop(widget_id)
print(f"Removed widget {widget_id}.")
@ -358,7 +291,85 @@ class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
if widget_id not in existing_ids:
return widget_id
def _change_grid(self, widget_id: str, row: int, col: int):
"""
Change the grid to reflect the new position of the widget.
Args:
widget_id(str): The unique identifier of the widget.
row(int): The new row coordinate of the widget in the figure.
col(int): The new column coordinate of the widget in the figure.
"""
while len(self.grid) <= row:
self.grid.append([])
row = self.grid[row]
while len(row) <= col:
row.append(None)
row[col] = widget_id
def _reindex_grid(self):
"""Reindex the grid to remove empty rows and columns."""
print(f"old grid: {self.grid}")
new_grid = []
for row in self.grid:
new_row = [widget for widget in row if widget is not None]
if new_row:
new_grid.append(new_row)
#
# Update the config of each object to reflect its new position
for row_idx, row in enumerate(new_grid):
for col_idx, widget in enumerate(row):
self.widgets[widget].config.row, self.widgets[widget].config.col = row_idx, col_idx
self.grid = new_grid
self._replot_layout()
def _replot_layout(self):
"""Replot the layout based on the current grid configuration."""
self.clear()
for row_idx, row in enumerate(self.grid):
for col_idx, widget in enumerate(row):
self.addItem(self.widgets[widget], row=row_idx, col=col_idx)
def change_layout(self, max_columns=None, max_rows=None):
"""
Reshuffle the layout of the figure to adjust to a new number of max_columns or max_rows.
If both max_columns and max_rows are provided, max_rows is ignored.
Args:
max_columns (Optional[int]): The new maximum number of columns in the figure.
max_rows (Optional[int]): The new maximum number of rows in the figure.
"""
# Calculate total number of widgets
total_widgets = len(self.widgets)
if max_columns:
# Calculate the required number of rows based on max_columns
required_rows = (total_widgets + max_columns - 1) // max_columns
new_grid = [[None for _ in range(max_columns)] for _ in range(required_rows)]
elif max_rows:
# Calculate the required number of columns based on max_rows
required_columns = (total_widgets + max_rows - 1) // max_rows
new_grid = [[None for _ in range(required_columns)] for _ in range(max_rows)]
else:
# If neither max_columns nor max_rows is specified, just return without changing the layout
return
# Populate the new grid with widgets' IDs
current_idx = 0
for widget_id, widget in self.widgets.items():
row = current_idx // len(new_grid[0])
col = current_idx % len(new_grid[0])
new_grid[row][col] = widget_id
current_idx += 1
# Update widgets' positions and replot them according to the new grid
self.grid = new_grid
self._reindex_grid() # This method should be updated to handle reshuffling correctly
self._replot_layout() # Assumes this method re-adds widgets to the layout based on self.grid
def start(self):
import sys
app = QApplication(sys.argv)
win = QMainWindow()
win.setCentralWidget(self)
@ -439,8 +450,8 @@ class DebugWindow(QWidget):
self.w4 = self.figure[1, 1]
# curves for w1
self.w1.add_scan("samx", "samx", "bpm4i", "bpm4i", pen_style="dash")
self.w1.add_curve(
self.w1.add_curve_scan("samx", "samx", "bpm4i", "bpm4i", pen_style="dash")
self.w1.add_curve_custom(
x=[1, 2, 3, 4, 5],
y=[1, 2, 3, 4, 5],
label="curve-custom",
@ -449,13 +460,15 @@ class DebugWindow(QWidget):
)
# curves for w2
self.w2.add_scan("samx", "samx", "bpm3a", "bpm3a", pen_style="solid")
self.w2.add_scan("samx", "samx", "bpm4d", "bpm4d", pen_style="dot")
self.w2.add_curve(x=[1, 2, 3, 4, 5], y=[5, 4, 3, 2, 1], color="red", pen_style="dashdot")
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_custom(
x=[1, 2, 3, 4, 5], y=[5, 4, 3, 2, 1], color="red", pen_style="dashdot"
)
# curves for w3
self.w3.add_scan("samx", "samx", "bpm4i", "bpm4i", pen_style="dash")
self.w3.add_curve(
self.w3.add_curve_scan("samx", "samx", "bpm4i", "bpm4i", pen_style="dash")
self.w3.add_curve_custom(
x=[1, 2, 3, 4, 5],
y=[1, 2, 3, 4, 5],
label="curve-custom",
@ -464,8 +477,8 @@ class DebugWindow(QWidget):
)
# curves for w4
self.w4.add_scan("samx", "samx", "bpm4i", "bpm4i", pen_style="dash")
self.w4.add_curve(
self.w4.add_curve_scan("samx", "samx", "bpm4i", "bpm4i", pen_style="dash")
self.w4.add_curve_custom(
x=[1, 2, 3, 4, 5],
y=[1, 2, 3, 4, 5],
label="curve-custom",

View File

@ -1,2 +1,2 @@
from .plot_base import AxisConfig, WidgetConfig, BECPlotBase
from .waveform1d import Waveform1DConfig, BECWaveform1D
from .waveform1d import Waveform1DConfig, BECWaveform1D, BECCurve

View File

@ -4,7 +4,7 @@ from typing import Literal, Optional
import numpy as np
import pyqtgraph as pg
from bec_lib.utils import user_access
from pydantic import BaseModel, Field
from qtpy.QtWidgets import QWidget
@ -24,7 +24,7 @@ class AxisConfig(BaseModel):
class WidgetConfig(ConnectionConfig):
parent_figure_id: Optional[str] = Field(None, description="The parent figure of the plot.")
parent_id: Optional[str] = Field(None, description="The parent figure of the plot.")
# Coordinates in the figure
row: int = Field(0, description="The row coordinate in the figure.")
@ -37,6 +37,20 @@ class WidgetConfig(ConnectionConfig):
class BECPlotBase(BECConnector, pg.PlotItem):
USER_ACCESS = [
"set",
"set_title",
"set_x_label",
"set_y_label",
"set_x_scale",
"set_y_scale",
"set_x_lim",
"set_y_lim",
"set_grid",
"plot_data",
"remove",
]
def __init__(
self,
parent: Optional[QWidget] = None, # TODO decide if needed for this class
@ -68,8 +82,6 @@ class BECPlotBase(BECConnector, pg.PlotItem):
- x_lim: tuple
- y_lim: tuple
"""
# TODO check functionality
# Mapping of keywords to setter methods
method_map = {
"title": self.set_title,
@ -88,7 +100,6 @@ 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,
@ -110,7 +121,6 @@ class BECPlotBase(BECConnector, pg.PlotItem):
self.setTitle(title)
self.config.axis.title = title
@user_access
def set_x_label(self, label: str):
"""
Set the label of the x-axis.
@ -120,7 +130,6 @@ class BECPlotBase(BECConnector, pg.PlotItem):
self.setLabel("bottom", label)
self.config.axis.x_label = label
@user_access
def set_y_label(self, label: str):
"""
Set the label of the y-axis.
@ -130,7 +139,6 @@ class BECPlotBase(BECConnector, pg.PlotItem):
self.setLabel("left", label)
self.config.axis.y_label = label
@user_access
def set_x_scale(self, scale: Literal["linear", "log"] = "linear"):
"""
Set the scale of the x-axis.
@ -140,7 +148,6 @@ class BECPlotBase(BECConnector, pg.PlotItem):
self.setLogMode(x=(scale == "log"))
self.config.axis.x_scale = scale
@user_access
def set_y_scale(self, scale: Literal["linear", "log"] = "linear"):
"""
Set the scale of the y-axis.
@ -150,7 +157,6 @@ class BECPlotBase(BECConnector, pg.PlotItem):
self.setLogMode(y=(scale == "log"))
self.config.axis.y_scale = scale
@user_access
def set_x_lim(self, x_lim: tuple) -> None:
"""
Set the limits of the x-axis.
@ -160,7 +166,6 @@ class BECPlotBase(BECConnector, pg.PlotItem):
self.setXRange(x_lim[0], x_lim[1])
self.config.axis.x_lim = x_lim
@user_access
def set_y_lim(self, y_lim: tuple) -> None:
"""
Set the limits of the y-axis.
@ -170,7 +175,6 @@ class BECPlotBase(BECConnector, pg.PlotItem):
self.setYRange(y_lim[0], y_lim[1])
self.config.axis.y_lim = y_lim
@user_access
def set_grid(self, x: bool = False, y: bool = False):
"""
Set the grid of the plot widget.
@ -185,7 +189,6 @@ class BECPlotBase(BECConnector, pg.PlotItem):
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.
@ -198,7 +201,6 @@ class BECPlotBase(BECConnector, pg.PlotItem):
# TODO decide name of the method
self.plot(data_x, data_y, **kwargs)
@user_access
def remove(self):
"""Remove the plot widget from the figure."""
if self.figure is not None:

View File

@ -14,8 +14,7 @@ from qtpy.QtWidgets import QWidget
from bec_lib import MessageEndpoints
from bec_lib.scan_data import ScanData
from bec_lib.utils import user_access
from bec_widgets.utils import Colors
from bec_widgets.utils import Colors, ConnectionConfig, BECConnector
from bec_widgets.widgets.plots import BECPlotBase, WidgetConfig
@ -37,7 +36,8 @@ class Signal(BaseModel):
y: SignalData
class CurveConfig(BaseModel):
class CurveConfig(ConnectionConfig):
parent_id: Optional[str] = Field(None, description="The parent plot of the curve.")
label: Optional[str] = Field(None, description="The label of the curve.")
color: Optional[Any] = Field(None, description="The color of the curve.")
symbol: Optional[str] = Field("o", description="The symbol of the curve.")
@ -62,17 +62,35 @@ class Waveform1DConfig(WidgetConfig):
)
class BECCurve(pg.PlotDataItem): # TODO decide what will be accessible from the parent
class BECCurve(BECConnector, pg.PlotDataItem):
USER_ACCESS = [
"set",
"set_data",
"set_color",
"set_symbol",
"set_symbol_color",
"set_symbol_size",
"set_pen_width",
"set_pen_style",
]
def __init__(
self,
name: Optional[str] = None,
config: Optional[CurveConfig] = None,
gui_id: Optional[str] = None,
**kwargs,
):
super().__init__(name=name, **kwargs)
if config is None:
config = CurveConfig(label=name, widget_class=self.__class__.__name__)
self.config = config
self.config = config
else:
self.config = config
# config.widget_class = self.__class__.__name__
super().__init__(config=config, gui_id=gui_id)
pg.PlotDataItem.__init__(self, name=name, **kwargs)
# self.config = config
self.apply_config()
@ -188,6 +206,16 @@ class BECCurve(pg.PlotDataItem): # TODO decide what will be accessible from the
class BECWaveform1D(BECPlotBase):
USER_ACCESS = [
"add_curve_scan",
"add_curve_custom",
"remove_curve",
"scan_history",
"curves",
"curves_data",
"get_curve",
"get_curve_config",
]
scan_signal_update = pyqtSignal()
def __init__(
@ -207,6 +235,7 @@ class BECWaveform1D(BECPlotBase):
self.curves_data = defaultdict(dict)
self.scanID = None
# Scan segment update proxy
self.proxy_update_plot = pg.SignalProxy(
self.scan_signal_update, rateLimit=25, slot=self._update_scan_segment_plot
)
@ -221,75 +250,44 @@ class BECWaveform1D(BECPlotBase):
self.apply_config()
# # TODO decide if to use class methods or not
# wid = BECWaveform1D.from_config()
#
# @classmethod
# def from_config(
# cls,
# parent: Optional[QWidget],
# config: Waveform1DConfig,
# client=None,
# gui_id: Optional[str] = None,
# replot_last_scan: bool = False,
# ):
# """
# Class method to create an instance of BECWaveform1D from a configuration object.
#
# Args:
# parent: The parent widget.
# config: Configuration object for the Waveform1D widget.
# client: Client for communication with backend services.
# gui_id: Optional unique identifier for the GUI component.
#
# Returns:
# An instance of BECWaveform1D configured according to the provided config.
# """
# # Initialize the widget with the provided config
# widget = cls(parent=parent, config=config, client=client, gui_id=gui_id)
# widget.apply_axis_config()
#
# # Reconstruct curves based on the config
# for curve_id, curve_config in config.curves.items():
# widget.add_curve_by_config(curve_config)
# if replot_last_scan:
# widget.update_scan_curve_history(-1)
#
# return widget
# TODO check the functionality of config generator
def apply_config(self, replot_last_scan: bool = False):
self.apply_axis_config()
for curve_id, curve_config in self.config.curves.items():
self.add_curve_by_config(curve_config)
if replot_last_scan:
self.update_scan_curve_history(-1)
self.scan_history(scanID=-1)
def add_curve_by_config(self, curve_config: CurveConfig | dict):
def add_curve_by_config(self, curve_config: CurveConfig | dict) -> BECCurve:
"""
Add a curve to the plot widget by its configuration.
Args:
curve_config(CurveConfig|dict): Configuration of the curve to be added.
Returns:
BECCurve: The curve object.
"""
if isinstance(curve_config, dict):
curve_config = CurveConfig(**curve_config)
self._add_curve_object(
curve = self._add_curve_object(
name=curve_config.label, source=curve_config.source, config=curve_config
)
return curve
def get_curve_config(self, curve_id: str) -> CurveConfig:
def get_curve_config(self, curve_id: str, dict_output: bool = True) -> CurveConfig | dict:
"""
Get the configuration of a curve by its ID.
Args:
curve_id(str): ID of the curve.
Returns:
CurveConfig: Configuration of the curve.
CurveConfig|dict: Configuration of the curve.
"""
for source, curves in self.curves_data.items():
if curve_id in curves:
return curves[curve_id].config
if dict_output:
return curves[curve_id].config.model_dump()
else:
return curves[curve_id].config
@user_access
def curves(self) -> list: # TODO discuss if it should be marked as @property for RPC
"""
Get the curves of the plot widget as a list
@ -298,7 +296,6 @@ class BECWaveform1D(BECPlotBase):
"""
return self.curves
@user_access
def curves_data(self) -> dict: # TODO discuss if it should be marked as @property for RPC
"""
Get the curves data of the plot widget as a dictionary
@ -307,8 +304,24 @@ class BECWaveform1D(BECPlotBase):
"""
return self.curves_data
@user_access
def add_scan(
def get_curve(self, identifier) -> BECCurve:
"""
Get the curve by its index or ID.
Args:
identifier(int|str): Identifier of the curve. Can be either an integer (index) or a string (curve_id).
Returns:
BECCurve: The curve object.
"""
if isinstance(identifier, int):
return self.curves[identifier]
elif isinstance(identifier, str):
return self.curves_data[identifier]
else:
raise ValueError(
"Each identifier must be either an integer (index) or a string (curve_id)."
)
def add_curve_scan(
self,
x_name: str,
x_entry: str,
@ -317,7 +330,21 @@ class BECWaveform1D(BECPlotBase):
color: Optional[str] = None,
label: Optional[str] = None,
**kwargs,
): # add_scan_curve
) -> 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}"
@ -336,6 +363,9 @@ class BECWaveform1D(BECPlotBase):
# 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,
@ -346,17 +376,29 @@ class BECWaveform1D(BECPlotBase):
),
**kwargs,
)
self._add_curve_object(name=label, source=curve_source, config=curve_config)
curve = self._add_curve_object(name=label, source=curve_source, config=curve_config)
return curve
@user_access
def add_curve(
def add_curve_custom(
self,
x: list | np.ndarray,
y: list | np.ndarray,
label: str = None,
color: str = None,
**kwargs,
):
) -> BECCurve:
"""
Add a custom data curve to the plot widget.
Args:
x(list|np.ndarray): X data of the curve.
y(list|np.ndarray): Y data of the curve.
label(str, optional): Label of the curve. Defaults to None.
color(str, optional): Color of the curve. Defaults to None.
**kwargs: Additional keyword arguments for the curve configuration.
Returns:
BECCurve: The curve object.
"""
curve_source = "custom"
curve_id = label or f"Curve {len(self.curves) + 1}"
@ -369,21 +411,24 @@ class BECWaveform1D(BECPlotBase):
color = (
color
or Colors.golden_angle_color(
colormap=self.config.color_palette, num=len(self.curves) + 1
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.gui_id,
label=curve_id,
color=color,
source=curve_source,
**kwargs,
)
self._add_curve_object(name=curve_id, source=curve_source, config=curve_config, data=(x, y))
# self.crosshair = Crosshair(self, precision=3)
curve = self._add_curve_object(
name=curve_id, source=curve_source, config=curve_config, data=(x, y)
)
return curve
def _add_curve_object(
self,
@ -391,7 +436,7 @@ class BECWaveform1D(BECPlotBase):
source: str,
config: CurveConfig,
data: tuple[list | np.ndarray, list | np.ndarray] = None,
):
) -> BECCurve:
"""
Add a curve object to the plot widget.
Args:
@ -399,6 +444,8 @@ class BECWaveform1D(BECPlotBase):
source(str): Source of the curve.
config(CurveConfig): Configuration of the curve.
data(tuple[list|np.ndarray,list|np.ndarray], optional): Data (x,y) to be plotted. Defaults to None.
Returns:
BECCurve: The curve object.
"""
curve = BECCurve(config=config, name=name)
self.curves_data[source][name] = curve
@ -406,6 +453,7 @@ class BECWaveform1D(BECPlotBase):
self.config.curves[name] = curve.config
if data is not None:
curve.setData(data[0], data[1])
return curve
def _check_curve_id(self, val: Any, dict_to_check: dict) -> bool:
"""
@ -490,7 +538,6 @@ class BECWaveform1D(BECPlotBase):
return
if current_scanID != self.scanID:
# self.clear() #TODO check if this is the right way to clear the plot
self.scanID = current_scanID
self.scan_segment_data = self.queue.scan_storage.find_scan_by_ID(self.scanID)
@ -521,8 +568,11 @@ class BECWaveform1D(BECPlotBase):
curve.setData(data_x, data_y)
@user_access
def update_scan_curve_history(self, scanID: str = None, scan_index: int = None):
def scan_history(
self,
scan_index: int = None,
scanID: str = None,
):
"""
Update the scan curves with the data from the scan storage.
Provide only one of scanID or scan_index.