diff --git a/bec_widgets/examples/motor_movement/motor_control_compilations.py b/bec_widgets/examples/motor_movement/motor_control_compilations.py
index 24c94694..0a182ece 100644
--- a/bec_widgets/examples/motor_movement/motor_control_compilations.py
+++ b/bec_widgets/examples/motor_movement/motor_control_compilations.py
@@ -10,7 +10,7 @@ from bec_widgets.widgets import (
MotorControlRelative,
MotorControlSelection,
MotorCoordinateTable,
- MotorMap,
+ # MotorMap,
MotorThread,
)
@@ -58,13 +58,13 @@ class MotorControlApp(QWidget):
# Widgets
self.motor_control_panel = MotorControlPanel(client=self.client, config=self.config)
# Create MotorMap
- self.motion_map = MotorMap(client=self.client, config=self.config)
+ # self.motion_map = MotorMap(client=self.client, config=self.config)
# Create MotorCoordinateTable
self.motor_table = MotorCoordinateTable(client=self.client, config=self.config)
# Create the splitter and add MotorMap and MotorControlPanel
splitter = QSplitter(Qt.Horizontal)
- splitter.addWidget(self.motion_map)
+ # splitter.addWidget(self.motion_map)
splitter.addWidget(self.motor_control_panel)
splitter.addWidget(self.motor_table)
@@ -74,9 +74,9 @@ class MotorControlApp(QWidget):
self.setLayout(layout)
# Connecting signals and slots
- self.motor_control_panel.selection_widget.selected_motors_signal.connect(
- lambda x, y: self.motion_map.change_motors(x, y, 0)
- )
+ # self.motor_control_panel.selection_widget.selected_motors_signal.connect(
+ # lambda x, y: self.motion_map.change_motors(x, y, 0)
+ # )
self.motor_control_panel.absolute_widget.coordinates_signal.connect(
self.motor_table.add_coordinate
)
@@ -87,7 +87,7 @@ class MotorControlApp(QWidget):
self.motor_control_panel.absolute_widget.set_precision
)
- self.motor_table.plot_coordinates_signal.connect(self.motion_map.plot_saved_coordinates)
+ # self.motor_table.plot_coordinates_signal.connect(self.motion_map.plot_saved_coordinates)
class MotorControlMap(QWidget):
@@ -101,11 +101,11 @@ class MotorControlMap(QWidget):
# Widgets
self.motor_control_panel = MotorControlPanel(client=self.client, config=self.config)
# Create MotorMap
- self.motion_map = MotorMap(client=self.client, config=self.config)
+ # self.motion_map = MotorMap(client=self.client, config=self.config)
# Create the splitter and add MotorMap and MotorControlPanel
splitter = QSplitter(Qt.Horizontal)
- splitter.addWidget(self.motion_map)
+ # splitter.addWidget(self.motion_map)
splitter.addWidget(self.motor_control_panel)
# Set the main layout
@@ -114,9 +114,9 @@ class MotorControlMap(QWidget):
self.setLayout(layout)
# Connecting signals and slots
- self.motor_control_panel.selection_widget.selected_motors_signal.connect(
- lambda x, y: self.motion_map.change_motors(x, y, 0)
- )
+ # self.motor_control_panel.selection_widget.selected_motors_signal.connect(
+ # lambda x, y: self.motion_map.change_motors(x, y, 0)
+ # )
class MotorControlPanel(QWidget):
diff --git a/bec_widgets/validation/__init__.py b/bec_widgets/validation/__init__.py
deleted file mode 100644
index b7cd8a37..00000000
--- a/bec_widgets/validation/__init__.py
+++ /dev/null
@@ -1,2 +0,0 @@
-# from .monitor_config import validate_monitor_config, ValidationError
-from .monitor_config_validator import MonitorConfigValidator
diff --git a/bec_widgets/validation/monitor_config_validator.py b/bec_widgets/validation/monitor_config_validator.py
deleted file mode 100644
index 25147032..00000000
--- a/bec_widgets/validation/monitor_config_validator.py
+++ /dev/null
@@ -1,258 +0,0 @@
-from typing import Literal, Optional, Union
-
-from pydantic import BaseModel, Field, ValidationError, field_validator, model_validator
-from pydantic_core import PydanticCustomError
-
-
-class Signal(BaseModel):
- """
- Represents a signal in a plot configuration.
-
- Args:
- name (str): The name of the signal.
- entry (Optional[str]): The entry point of the signal, optional.
- """
-
- name: str
- entry: Optional[str] = Field(None, validate_default=True)
-
- @model_validator(mode="before")
- @classmethod
- def validate_fields(cls, values):
- """Validate the fields of the model.
- First validate the 'name' field, then validate the 'entry' field.
-
- Args:
- values (dict): The values to be validated."""
- devices = MonitorConfigValidator.devices
-
- # Validate 'name'
- name = values.get("name")
-
- # Check if device name provided
- if name is None:
- raise PydanticCustomError(
- "no_device_name", "Device name must be provided", {"wrong_value": name}
- )
- # Check if device exists in BEC
- if name not in devices:
- raise PydanticCustomError(
- "no_device_bec",
- 'Device "{wrong_value}" not found in current BEC session',
- {"wrong_value": name},
- )
-
- device = devices[name] # get the device to check if it has signals
-
- # Get device description
- description = device.describe()
-
- # Validate 'entry'
- entry = values.get("entry")
-
- # Set entry based on hints if not provided
- if entry is None:
- entry = next(iter(device._hints), name) if hasattr(device, "_hints") else name
- if entry not in description:
- raise PydanticCustomError(
- "no_entry_for_device",
- 'Entry "{wrong_value}" not found in device "{device_name}" signals',
- {"wrong_value": entry, "device_name": name},
- )
-
- values["entry"] = entry
- return values
-
-
-class AxisSignal(BaseModel):
- """
- Configuration signal axis for a single plot.
- Attributes:
- x (list): Signal for the X axis.
- y (list): Signals for the Y axis.
- """
-
- x: list[Signal] = Field(default_factory=list)
- y: list[Signal] = Field(default_factory=list)
-
- @field_validator("x")
- @classmethod
- def validate_x_signals(cls, v):
- """Ensure that there is only one signal for x-axis."""
- if len(v) != 1:
- raise PydanticCustomError(
- "x_axis_multiple_signals",
- 'There must be exactly one signal for x axis. Number of x signals: "{wrong_value}"',
- {"wrong_value": v},
- )
-
- return v
-
-
-class SourceHistoryValidator(BaseModel):
- """History source validator
- Attributes:
- type (str): type of source - history
- scan_id (str): Scan ID for history source.
- signals (list): Signal for the source.
- """
-
- type: Literal["history"]
- scan_id: str # TODO can be validated if it is a valid scan_id
- signals: AxisSignal
-
-
-class SourceSegmentValidator(BaseModel):
- """Scan Segment source validator
- Attributes:
- type (str): type of source - scan_segment
- signals (AxisSignal): Signal for the source.
- """
-
- type: Literal["scan_segment"]
- signals: AxisSignal
-
-
-class SourceRedisValidator(BaseModel):
- """Scan Segment source validator
- Attributes:
- type (str): type of source - scan_segment
- endpoint (str): Endpoint reference in redis.
- update (str): Update type.
- """
-
- type: Literal["redis"]
- endpoint: str
- update: str
- signals: dict
-
-
-class Source(BaseModel): # TODO decide if it should stay for general Source validation
- """
- General source validation, includes all Optional arguments of all other sources.
- Attributes:
- type (list): type of source (scan_segment, history)
- scan_id (Optional[str]): Scan ID for history source.
- signals (Optional[AxisSignal]): Signal for the source.
- """
-
- type: Literal["scan_segment", "history", "redis"]
- scan_id: Optional[str] = None
- signals: Optional[dict] = None
-
-
-class PlotConfig(BaseModel):
- """
- Configuration for a single plot.
-
- Attributes:
- plot_name (Optional[str]): Name of the plot.
- x_label (Optional[str]): The label for the x-axis.
- y_label (Optional[str]): The label for the y-axis.
- sources (list): A list of sources to be plotted on this axis.
- """
-
- plot_name: Optional[str] = None
- x_label: Optional[str] = None
- y_label: Optional[str] = None
- sources: list = Field(default_factory=list)
-
- @field_validator("sources")
- @classmethod
- def validate_sources(cls, values):
- """Validate the sources of the plot configuration, based on the type of source."""
- validated_sources = []
- for source in values:
- # Check if source type is supported
- Source(**source)
- source_type = source.get("type", None)
-
- # Validate source based on type
- if source_type == "scan_segment":
- validated_sources.append(SourceSegmentValidator(**source))
- elif source_type == "history":
- validated_sources.append(SourceHistoryValidator(**source))
- elif source_type == "redis":
- validated_sources.append(SourceRedisValidator(**source))
- return validated_sources
-
-
-class PlotSettings(BaseModel):
- """
- Global settings for plotting affecting mostly visuals.
-
- Attributes:
- background_color (str): Color of the plot background. Default is black.
- axis_width (Optional[int]): Width of the plot axes. Default is 2.
- axis_color (Optional[str]): Color of the plot axes. Default is None.
- num_columns (int): Number of columns in the plot layout. Default is 1.
- colormap (str): Colormap to be used. Default is magma.
- scan_types (bool): Indicates if the configuration is for different scan types. Default is False.
- """
-
- background_color: Literal["black", "white"] = "black"
- axis_width: Optional[int] = 2
- axis_color: Optional[str] = None
- num_columns: Optional[int] = 1
- colormap: Optional[str] = "magma"
- scan_types: Optional[bool] = False
-
-
-class DeviceMonitorConfig(BaseModel):
- """
- Configuration model for the device monitor mode.
-
- Attributes:
- plot_settings (PlotSettings): Global settings for plotting.
- plot_data (list[PlotConfig]): List of plot configurations.
- """
-
- plot_settings: PlotSettings
- plot_data: list[PlotConfig]
-
-
-class ScanModeConfig(BaseModel):
- """
- Configuration model for scan mode.
-
- Attributes:
- plot_settings (PlotSettings): Global settings for plotting.
- plot_data (dict[str, list[PlotConfig]]): Dictionary of plot configurations,
- keyed by scan type.
- """
-
- plot_settings: PlotSettings
- plot_data: dict[str, list[PlotConfig]]
-
-
-class MonitorConfigValidator:
- """Validates the configuration data for the BECMonitor."""
-
- devices = None
-
- def __init__(self, devices):
- # self.device_manager = device_manager
- MonitorConfigValidator.devices = devices
-
- def validate_monitor_config(
- self, config_data: dict
- ) -> Union[DeviceMonitorConfig, ScanModeConfig]:
- """
- Validates the configuration data based on the provided schema.
-
- Args:
- config_data (dict): Configuration data to be validated.
-
- Returns:
- Union[DeviceMonitorConfig, ScanModeConfig]: Validated configuration object.
-
- Raises:
- ValidationError: If the configuration data does not conform to the schema.
- """
- config_type = config_data.get("plot_settings", {}).get("scan_types", False)
- if config_type:
- validated_config = ScanModeConfig(**config_data)
- else:
- validated_config = DeviceMonitorConfig(**config_data)
-
- return validated_config
diff --git a/bec_widgets/widgets/__init__.py b/bec_widgets/widgets/__init__.py
index 0fdb16ab..ebe79459 100644
--- a/bec_widgets/widgets/__init__.py
+++ b/bec_widgets/widgets/__init__.py
@@ -1,6 +1,5 @@
from .dock import BECDock, BECDockArea
from .figure import BECFigure, FigureConfig
-from .monitor import BECMonitor
from .motor_control import (
MotorControlAbsolute,
MotorControlRelative,
@@ -8,6 +7,5 @@ from .motor_control import (
MotorCoordinateTable,
MotorThread,
)
-from .motor_map import MotorMap
from .plots import BECCurve, BECMotorMap, BECWaveform
from .scan_control import ScanControl
diff --git a/bec_widgets/widgets/monitor/__init__.py b/bec_widgets/widgets/monitor/__init__.py
deleted file mode 100644
index a91ef3f8..00000000
--- a/bec_widgets/widgets/monitor/__init__.py
+++ /dev/null
@@ -1 +0,0 @@
-from .monitor import BECMonitor
diff --git a/bec_widgets/widgets/monitor/config_dialog.py b/bec_widgets/widgets/monitor/config_dialog.py
deleted file mode 100644
index 271228d7..00000000
--- a/bec_widgets/widgets/monitor/config_dialog.py
+++ /dev/null
@@ -1,574 +0,0 @@
-import os
-
-from pydantic import ValidationError
-from qtpy import uic
-from qtpy.QtCore import Signal as pyqtSignal
-from qtpy.QtWidgets import (
- QApplication,
- QLineEdit,
- QMessageBox,
- QTableWidget,
- QTableWidgetItem,
- QTabWidget,
- QVBoxLayout,
- QWidget,
-)
-
-from bec_widgets.utils.bec_dispatcher import BECDispatcher
-from bec_widgets.utils.yaml_dialog import load_yaml, save_yaml
-from bec_widgets.validation import MonitorConfigValidator
-
-current_path = os.path.dirname(__file__)
-Ui_Form, BaseClass = uic.loadUiType(os.path.join(current_path, "config_dialog.ui"))
-Tab_Ui_Form, Tab_BaseClass = uic.loadUiType(os.path.join(current_path, "tab_template.ui"))
-
-# test configs for demonstration purpose
-
-# Configuration for default mode when only devices are monitored
-CONFIG_DEFAULT = {
- "plot_settings": {
- "background_color": "black",
- "num_columns": 1,
- "colormap": "plasma",
- "scan_types": False,
- },
- "plot_data": [
- {
- "plot_name": "BPM4i plots vs samx",
- "x_label": "Motor Y",
- "y_label": "bpm4i",
- "sources": [
- {
- "type": "scan_segment",
- "signals": {
- "x": [{"name": "samx", "entry": "samx"}],
- "y": [{"name": "bpm4i", "entry": "bpm4i"}],
- },
- }
- ],
- },
- {
- "plot_name": "Gauss plots vs samx",
- "x_label": "Motor X",
- "y_label": "Gauss",
- "sources": [
- {
- "type": "scan_segment",
- "signals": {
- "x": [{"name": "samx", "entry": "samx"}],
- "y": [
- {"name": "gauss_bpm"},
- {"name": "gauss_adc1"},
- {"name": "gauss_adc2"},
- ],
- },
- }
- ],
- },
- ],
-}
-
-# Configuration which is dynamically changing depending on the scan type
-CONFIG_SCAN_MODE = {
- "plot_settings": {
- "background_color": "white",
- "num_columns": 3,
- "colormap": "plasma",
- "scan_types": True,
- },
- "plot_data": {
- "grid_scan": [
- {
- "plot_name": "Grid plot 1",
- "x_label": "Motor X",
- "y_label": "BPM",
- "sources": [
- {
- "type": "scan_segment",
- "signals": {
- "x": [{"name": "samx", "entry": "samx"}],
- "y": [{"name": "gauss_bpm"}],
- },
- }
- ],
- },
- {
- "plot_name": "Grid plot 2",
- "x_label": "Motor X",
- "y_label": "BPM",
- "sources": [
- {
- "type": "scan_segment",
- "signals": {
- "x": [{"name": "samx", "entry": "samx"}],
- "y": [{"name": "gauss_adc1"}],
- },
- }
- ],
- },
- {
- "plot_name": "Grid plot 3",
- "x_label": "Motor X",
- "y_label": "BPM",
- "sources": [
- {
- "type": "scan_segment",
- "signals": {"x": [{"name": "samy"}], "y": [{"name": "gauss_adc2"}]},
- }
- ],
- },
- {
- "plot_name": "Grid plot 4",
- "x_label": "Motor X",
- "y_label": "BPM",
- "sources": [
- {
- "type": "scan_segment",
- "signals": {
- "x": [{"name": "samy", "entry": "samy"}],
- "y": [{"name": "gauss_adc3"}],
- },
- }
- ],
- },
- ],
- "line_scan": [
- {
- "plot_name": "BPM plots vs samx",
- "x_label": "Motor X",
- "y_label": "Gauss",
- "sources": [
- {
- "type": "scan_segment",
- "signals": {
- "x": [{"name": "samx", "entry": "samx"}],
- "y": [{"name": "bpm4i"}],
- },
- }
- ],
- },
- {
- "plot_name": "Gauss plots vs samx",
- "x_label": "Motor X",
- "y_label": "Gauss",
- "sources": [
- {
- "type": "scan_segment",
- "signals": {
- "x": [{"name": "samx", "entry": "samx"}],
- "y": [{"name": "gauss_bpm"}, {"name": "gauss_adc1"}],
- },
- }
- ],
- },
- ],
- },
-}
-
-
-class ConfigDialog(QWidget, Ui_Form):
- config_updated = pyqtSignal(dict)
-
- def __init__(self, client=None, default_config=None, skip_validation: bool = False):
- super(ConfigDialog, self).__init__()
- self.setupUi(self)
-
- # Client
- bec_dispatcher = BECDispatcher()
- self.client = bec_dispatcher.client if client is None else client
- self.dev = self.client.device_manager.devices
-
- # Init validator
- self.skip_validation = skip_validation
- if self.skip_validation is False:
- self.validator = MonitorConfigValidator(self.dev)
-
- # Connect the Ok/Apply/Cancel buttons
- self.pushButton_ok.clicked.connect(self.apply_and_close)
- self.pushButton_apply.clicked.connect(self.apply_config)
- self.pushButton_cancel.clicked.connect(self.close)
-
- # Hook signals top level
- self.pushButton_new_scan_type.clicked.connect(
- lambda: self.generate_empty_scan_tab(
- self.tabWidget_scan_types, self.lineEdit_scan_type.text()
- )
- )
-
- # Load/save yaml file buttons
- self.pushButton_import.clicked.connect(self.load_config_from_yaml)
- self.pushButton_export.clicked.connect(self.save_config_to_yaml)
-
- # Scan Types changed
- self.comboBox_scanTypes.currentIndexChanged.connect(self._init_default)
-
- # Make scan tabs closable
- self.tabWidget_scan_types.tabCloseRequested.connect(self.handle_tab_close_request)
-
- # Init functions to make a default dialog
- if default_config is None:
- self._init_default()
- else:
- self.load_config(default_config)
-
- def _init_default(self):
- """Init default dialog"""
-
- if self.comboBox_scanTypes.currentText() == "Disabled": # Default mode
- self.add_new_scan_tab(self.tabWidget_scan_types, "Default")
- self.add_new_plot_tab(self.tabWidget_scan_types.widget(0))
- self.pushButton_new_scan_type.setEnabled(False)
- self.lineEdit_scan_type.setEnabled(False)
- else: # Scan mode with clear tab
- self.pushButton_new_scan_type.setEnabled(True)
- self.lineEdit_scan_type.setEnabled(True)
- self.tabWidget_scan_types.clear()
-
- def add_new_scan_tab(
- self, parent_tab: QTabWidget, scan_name: str, closable: bool = False
- ) -> QWidget:
- """
- Add a new scan tab to the parent tab widget
-
- Args:
- parent_tab(QTabWidget): Parent tab widget, where to add scan tab
- scan_name(str): Scan name
- closable(bool): If True, the scan tab will be closable
-
- Returns:
- scan_tab(QWidget): Scan tab widget
- """
- # Check for an existing tab with the same name
- for index in range(parent_tab.count()):
- if parent_tab.tabText(index) == scan_name:
- print(f'Scan name "{scan_name}" already exists.')
- return None # or return the existing tab: return parent_tab.widget(index)
-
- # Create a new scan tab
- scan_tab = QWidget()
- scan_tab_layout = QVBoxLayout(scan_tab)
-
- # Set a tab widget for plots
- tabWidget_plots = QTabWidget()
- tabWidget_plots.setObjectName("tabWidget_plots") # TODO decide if needed to give a name
- tabWidget_plots.setTabsClosable(True)
- tabWidget_plots.tabCloseRequested.connect(self.handle_tab_close_request)
- scan_tab_layout.addWidget(tabWidget_plots)
-
- # Add scan tab
- parent_tab.addTab(scan_tab, scan_name)
-
- # Make tabs closable
- if closable:
- parent_tab.setTabsClosable(closable)
-
- return scan_tab
-
- def add_new_plot_tab(self, scan_tab: QWidget) -> QWidget:
- """
- Add a new plot tab to the scan tab
-
- Args:
- scan_tab (QWidget): Scan tab widget
-
- Returns:
- plot_tab (QWidget): Plot tab
- """
-
- # Create a new plot tab from .ui template
- plot_tab = QWidget()
- plot_tab_ui = Tab_Ui_Form()
- plot_tab_ui.setupUi(plot_tab)
- plot_tab.ui = plot_tab_ui
-
- # Add plot to current scan tab
- tabWidget_plots = scan_tab.findChild(
- QTabWidget, "tabWidget_plots"
- ) # TODO decide if putting name is needed
- plot_name = f"Plot {tabWidget_plots.count() + 1}"
- tabWidget_plots.addTab(plot_tab, plot_name)
-
- # Hook signal
- self.hook_plot_tab_signals(scan_tab=scan_tab, plot_tab=plot_tab.ui)
-
- return plot_tab
-
- def hook_plot_tab_signals(self, scan_tab: QTabWidget, plot_tab: Tab_Ui_Form) -> None:
- """
- Hook signals of the plot tab
- Args:
- scan_tab(QTabWidget): Scan tab widget
- plot_tab(Tab_Ui_Form): Plot tab widget
- """
- plot_tab.pushButton_add_new_plot.clicked.connect(
- lambda: self.add_new_plot_tab(scan_tab=scan_tab)
- )
- plot_tab.pushButton_y_new.clicked.connect(
- lambda: self.add_new_signal(plot_tab.tableWidget_y_signals)
- )
-
- def add_new_signal(self, table: QTableWidget) -> None:
- """
- Add a new signal to the table
-
- Args:
- table(QTableWidget): Table widget
- """
-
- row_position = table.rowCount()
- table.insertRow(row_position)
- table.setItem(row_position, 0, QTableWidgetItem(""))
- table.setItem(row_position, 1, QTableWidgetItem(""))
-
- def handle_tab_close_request(self, index: int) -> None:
- """
- Handle tab close request
-
- Args:
- index(int): Index of the tab to be closed
- """
-
- parent_tab = self.sender()
- if parent_tab.count() > 1: # ensure there is at least one tab
- parent_tab.removeTab(index)
-
- def generate_empty_scan_tab(self, parent_tab: QTabWidget, scan_name: str):
- """
- Generate an empty scan tab
-
- Args:
- parent_tab (QTabWidget): Parent tab widget where to add the scan tab
- scan_name(str): name of the scan tab
- """
-
- scan_tab = self.add_new_scan_tab(parent_tab, scan_name, closable=True)
- if scan_tab is not None:
- self.add_new_plot_tab(scan_tab)
-
- def get_plot_config(self, plot_tab: QWidget) -> dict:
- """
- Get plot configuration from the plot tab adn send it as dict
-
- Args:
- plot_tab(QWidget): Plot tab widget
-
- Returns:
- dict: Plot configuration
- """
-
- ui = plot_tab.ui
- table = ui.tableWidget_y_signals
-
- x_signals = [
- {
- "name": self.safe_text(ui.lineEdit_x_name),
- "entry": self.safe_text(ui.lineEdit_x_entry),
- }
- ]
-
- y_signals = [
- {
- "name": self.safe_text(table.item(row, 0)),
- "entry": self.safe_text(table.item(row, 1)),
- }
- for row in range(table.rowCount())
- ]
-
- plot_data = {
- "plot_name": self.safe_text(ui.lineEdit_plot_title),
- "x_label": self.safe_text(ui.lineEdit_x_label),
- "y_label": self.safe_text(ui.lineEdit_y_label),
- "sources": [{"type": "scan_segment", "signals": {"x": x_signals, "y": y_signals}}],
- }
-
- return plot_data
-
- def apply_config(self) -> dict:
- """
- Apply configuration from the whole configuration window
-
- Returns:
- dict: Current configuration
-
- """
-
- # General settings
- config = {
- "plot_settings": {
- "background_color": self.comboBox_appearance.currentText(),
- "num_columns": self.spinBox_n_column.value(),
- "colormap": self.comboBox_colormap.currentText(),
- "scan_types": True if self.comboBox_scanTypes.currentText() == "Enabled" else False,
- },
- "plot_data": {} if self.comboBox_scanTypes.currentText() == "Enabled" else [],
- }
-
- # Iterate through the plot tabs - Device monitor mode
- if config["plot_settings"]["scan_types"] == False:
- plot_tab = self.tabWidget_scan_types.widget(0).findChild(QTabWidget)
- for index in range(plot_tab.count()):
- plot_data = self.get_plot_config(plot_tab.widget(index))
- config["plot_data"].append(plot_data)
-
- # Iterate through the scan tabs - Scan mode
- elif config["plot_settings"]["scan_types"] == True:
- # Iterate through the scan tabs
- for index in range(self.tabWidget_scan_types.count()):
- scan_tab = self.tabWidget_scan_types.widget(index)
- scan_name = self.tabWidget_scan_types.tabText(index)
- plot_tab = scan_tab.findChild(QTabWidget)
- config["plot_data"][scan_name] = []
- # Iterate through the plot tabs
- for index in range(plot_tab.count()):
- plot_data = self.get_plot_config(plot_tab.widget(index))
- config["plot_data"][scan_name].append(plot_data)
-
- return config
-
- def load_config(self, config: dict) -> None:
- """
- Load configuration to the configuration window
-
- Args:
- config(dict): Configuration to be loaded
- """
-
- # Plot setting General box
- plot_settings = config.get("plot_settings", {})
-
- self.comboBox_appearance.setCurrentText(plot_settings.get("background_color", "black"))
- self.spinBox_n_column.setValue(plot_settings.get("num_columns", 1))
- self.comboBox_colormap.setCurrentText(
- plot_settings.get("colormap", "magma")
- ) # TODO make logic to allow also different colormaps -> validation of incoming dict
- self.comboBox_scanTypes.setCurrentText(
- "Enabled" if plot_settings.get("scan_types", False) else "Disabled"
- )
-
- # Clear exiting scan tabs
- self.tabWidget_scan_types.clear()
-
- # Get what mode is active - scan vs default device monitor
- scan_mode = plot_settings.get("scan_types", False)
-
- if scan_mode is False: # default mode:
- plot_data = config.get("plot_data", [])
- self.add_new_scan_tab(self.tabWidget_scan_types, "Default")
- for plot_config in plot_data: # Create plot tab for each plot and populate GUI
- plot = self.add_new_plot_tab(self.tabWidget_scan_types.widget(0))
- self.load_plot_setting(plot, plot_config)
- elif scan_mode is True: # scan mode
- plot_data = config.get("plot_data", {})
- for scan_name, scan_config in plot_data.items():
- scan_tab = self.add_new_scan_tab(self.tabWidget_scan_types, scan_name)
- for plot_config in scan_config:
- plot = self.add_new_plot_tab(scan_tab)
- self.load_plot_setting(plot, plot_config)
-
- def load_plot_setting(self, plot: QWidget, plot_config: dict) -> None:
- """
- Load plot setting from config
-
- Args:
- plot (QWidget): plot tab widget
- plot_config (dict): config for single plot tab
- """
- sources = plot_config.get("sources", [{}])[0]
- x_signals = sources.get("signals", {}).get("x", [{}])[0]
- y_signals = sources.get("signals", {}).get("y", [])
-
- # LabelBox
- plot.ui.lineEdit_plot_title.setText(plot_config.get("plot_name", ""))
- plot.ui.lineEdit_x_label.setText(plot_config.get("x_label", ""))
- plot.ui.lineEdit_y_label.setText(plot_config.get("y_label", ""))
-
- # X axis
- plot.ui.lineEdit_x_name.setText(x_signals.get("name", ""))
- plot.ui.lineEdit_x_entry.setText(x_signals.get("entry", ""))
-
- # Y axis
- for y_signal in y_signals:
- row_position = plot.ui.tableWidget_y_signals.rowCount()
- plot.ui.tableWidget_y_signals.insertRow(row_position)
- plot.ui.tableWidget_y_signals.setItem(
- row_position, 0, QTableWidgetItem(y_signal.get("name", ""))
- )
- plot.ui.tableWidget_y_signals.setItem(
- row_position, 1, QTableWidgetItem(y_signal.get("entry", ""))
- )
-
- def load_config_from_yaml(self):
- """
- Load configuration from yaml file
- """
- config = load_yaml(self)
- self.load_config(config)
-
- def save_config_to_yaml(self):
- """
- Save configuration to yaml file
- """
- config = self.apply_config()
- save_yaml(self, config)
-
- @staticmethod
- def safe_text(line_edit: QLineEdit) -> str:
- """
- Get text from a line edit, if it is None, return empty string
- Args:
- line_edit(QLineEdit): Line edit widget
-
- Returns:
- str: Text from the line edit
- """
- return "" if line_edit is None else line_edit.text()
-
- def apply_and_close(self):
- new_config = self.apply_config()
- if self.skip_validation is True:
- self.config_updated.emit(new_config)
- self.close()
- else:
- try:
- validated_config = self.validator.validate_monitor_config(new_config)
- approved_config = validated_config.model_dump()
- self.config_updated.emit(approved_config)
- self.close()
- except ValidationError as e:
- error_str = str(e)
- formatted_error_message = ConfigDialog.format_validation_error(error_str)
-
- # Display the formatted error message in a popup
- QMessageBox.critical(self, "Configuration Error", formatted_error_message)
-
- @staticmethod
- def format_validation_error(error_str: str) -> str:
- """
- Format the validation error string to be displayed in a popup.
- Args:
- error_str(str): Error string from the validation error.
- """
- error_lines = error_str.split("\n")
- # The first line contains the number of errors.
- error_header = f"{error_lines[0]}
"
-
- formatted_error_message = error_header
- # Skip the first line as it's the header.
- error_details = error_lines[1:]
-
- # Iterate through pairs of lines (each error's two lines).
- for i in range(0, len(error_details), 2):
- location = error_details[i]
- message = error_details[i + 1] if i + 1 < len(error_details) else ""
-
- formatted_error_message += f"{location}
{message}
"
-
- return formatted_error_message
-
-
-if __name__ == "__main__": # pragma: no cover
- app = QApplication([])
- main_app = ConfigDialog()
- main_app.show()
- main_app.load_config(CONFIG_SCAN_MODE)
- app.exec()
diff --git a/bec_widgets/widgets/monitor/config_dialog.ui b/bec_widgets/widgets/monitor/config_dialog.ui
deleted file mode 100644
index 58464f84..00000000
--- a/bec_widgets/widgets/monitor/config_dialog.ui
+++ /dev/null
@@ -1,210 +0,0 @@
-
-
- Form
-
-
-
- 0
- 0
- 597
- 769
-
-
-
- Plot Configuration
-
-
- -
-
-
-
-
-
- Plot Layout Settings
-
-
-
-
-
-
-
-
-
- Number of Columns
-
-
-
- -
-
-
- Scan Types
-
-
-
- -
-
-
- New Scan Type
-
-
-
- -
-
-
-
-
- Disabled
-
-
- -
-
- Enabled
-
-
-
-
- -
-
-
- 1
-
-
-
- -
-
-
-
-
- -
-
-
- Qt::Vertical
-
-
-
- -
-
-
-
-
-
- Appearance
-
-
-
- -
-
-
- false
-
-
-
-
- black
-
-
- -
-
- white
-
-
-
-
- -
-
-
- Default Color Palette
-
-
-
- -
-
-
-
-
- magma
-
-
- -
-
- plasma
-
-
- -
-
- viridis
-
-
- -
-
- reds
-
-
-
-
-
-
-
-
-
- -
-
-
- Configuration
-
-
-
-
-
-
- Import
-
-
-
- -
-
-
- Export
-
-
-
-
-
-
-
-
- -
-
-
- QTabWidget::West
-
-
- QTabWidget::Rounded
-
-
- -1
-
-
-
- -
-
-
-
-
-
- Cancel
-
-
-
- -
-
-
- Apply
-
-
-
- -
-
-
- OK
-
-
-
-
-
-
-
-
-
-
diff --git a/bec_widgets/widgets/monitor/example_configs/config_device.yaml b/bec_widgets/widgets/monitor/example_configs/config_device.yaml
deleted file mode 100644
index 2cce78a4..00000000
--- a/bec_widgets/widgets/monitor/example_configs/config_device.yaml
+++ /dev/null
@@ -1,60 +0,0 @@
-plot_settings:
- background_color: "black"
- num_columns: 2
- colormap: "viridis"
- scan_types: False
-
-plot_data:
- - plot_name: "BPM4i plots vs samy"
- x:
- label: 'Motor Y'
- signals:
- - name: "samy"
- entry: "samy"
- y:
- label: 'bpm4i'
- signals:
- - name: "bpm4i"
- entry: "bpm4i"
-
- - plot_name: "BPM4i plots vs samx"
- x:
- label: 'Motor X'
- signals:
- - name: "samx"
- entry: "samx"
- y:
- label: 'bpm6b'
- signals:
- - name: "bpm6b"
- entry: "bpm6b"
- - name: "samy"
- entry: "samy"
-
- - plot_name: "Multiple Gaussian"
- x:
- label: 'Motor X'
- signals:
- - name: "samx"
- entry: "samx"
- y:
- label: 'Gauss ADC'
- signals:
- - name: "gauss_adc1"
- entry: "gauss_adc1"
- - name: "gauss_adc2"
- entry: "gauss_adc2"
- - name: "gauss_adc3"
- entry: "gauss_adc3"
-
- - plot_name: "Linear Signals"
- x:
- label: 'Motor X'
- signals:
- - name: "samy"
- entry: "samy"
- y:
- label: 'Motor Y'
- signals:
- - name: "samy"
- entry: "samy"
\ No newline at end of file
diff --git a/bec_widgets/widgets/monitor/example_configs/config_scans.yaml b/bec_widgets/widgets/monitor/example_configs/config_scans.yaml
deleted file mode 100644
index 531dae82..00000000
--- a/bec_widgets/widgets/monitor/example_configs/config_scans.yaml
+++ /dev/null
@@ -1,92 +0,0 @@
-plot_settings:
- background_color: "black"
- num_columns: 2
- colormap: "plasma"
- scan_types: True
-
-plot_data:
- line_scan:
-
- - plot_name: "BPM plot"
- x:
- label: 'Motor X'
- signals:
- - name: "samx"
- entry: "samx"
- y:
- label: 'BPM'
- signals:
- - name: "gauss_bpm"
- entry: "gauss_bpm"
- - name: "gauss_adc1"
- entry: "gauss_adc1"
- - name: "gauss_adc2"
- entry: "gauss_adc2"
-
- - plot_name: "Multi"
- x:
- label: 'Motor X'
- signals:
- - name: "samx"
- entry: "samx"
- y:
- label: 'Multi'
- signals:
- - name: "gauss_bpm"
- entry: "gauss_bpm"
- - name: "samx"
- entry: "samx"
-
- grid_scan:
- - plot_name: "Grid plot 1"
- x:
- label: 'Motor X'
- signals:
- - name: "samx"
- entry: "samx"
- y:
- label: 'BPM'
- signals:
- - name: "gauss_bpm"
- entry: "gauss_bpm"
- - name: "gauss_adc1"
- entry: "gauss_adc1"
-
- - plot_name: "Grid plot 2"
- x:
- label: 'Motor X'
- signals:
- - name: "samx"
- entry: "samx"
- y:
- label: 'BPM'
- signals:
- - name: "gauss_bpm"
- entry: "gauss_bpm"
- - name: "gauss_adc1"
- entry: "gauss_adc1"
-
- - plot_name: "Grid plot 3"
- x:
- label: 'Motor Y'
- signals:
- - name: "samy"
- entry: "samy"
- y:
- label: 'BPM'
- signals:
- - name: "gauss_bpm"
- entry: "gauss_bpm"
-
- - plot_name: "Grid plot 4"
- x:
- label: 'Motor Y'
- signals:
- - name: "samy"
- entry: "samy"
- y:
- label: 'BPM'
- signals:
- - name: "gauss_adc3"
- entry: "gauss_adc3"
-
diff --git a/bec_widgets/widgets/monitor/monitor.py b/bec_widgets/widgets/monitor/monitor.py
deleted file mode 100644
index a01f637b..00000000
--- a/bec_widgets/widgets/monitor/monitor.py
+++ /dev/null
@@ -1,845 +0,0 @@
-# pylint: disable = no-name-in-module,missing-module-docstring
-import time
-
-import pyqtgraph as pg
-from bec_lib.endpoints import MessageEndpoints
-from pydantic import ValidationError
-from pyqtgraph import mkBrush, mkPen
-from qtpy import QtCore
-from qtpy.QtCore import Signal as pyqtSignal
-from qtpy.QtCore import Slot as pyqtSlot
-from qtpy.QtWidgets import QApplication, QMessageBox
-
-from bec_widgets.utils import Colors, Crosshair, yaml_dialog
-from bec_widgets.utils.bec_dispatcher import BECDispatcher
-from bec_widgets.validation import MonitorConfigValidator
-from bec_widgets.widgets.monitor.config_dialog import ConfigDialog
-
-# just for demonstration purposes if script run directly
-CONFIG_SCAN_MODE = {
- "plot_settings": {
- "background_color": "white",
- "num_columns": 3,
- "colormap": "plasma",
- "scan_types": True,
- },
- "plot_data": {
- "grid_scan": [
- {
- "plot_name": "Grid plot 1",
- "x_label": "Motor X",
- "y_label": "BPM",
- "sources": [
- {
- "type": "scan_segment",
- "signals": {
- "x": [{"name": "samx", "entry": "samx"}],
- "y": [{"name": "bpm4i"}],
- },
- }
- ],
- },
- {
- "plot_name": "Grid plot 2",
- "x_label": "Motor X",
- "y_label": "BPM",
- "sources": [
- {
- "type": "scan_segment",
- "signals": {
- "x": [{"name": "samx", "entry": "samx"}],
- "y": [{"name": "bpm4i"}],
- },
- }
- ],
- },
- {
- "plot_name": "Grid plot 3",
- "x_label": "Motor X",
- "y_label": "BPM",
- "sources": [
- {
- "type": "scan_segment",
- "signals": {"x": [{"name": "samy"}], "y": [{"name": "bpm4i"}]},
- }
- ],
- },
- {
- "plot_name": "Grid plot 4",
- "x_label": "Motor X",
- "y_label": "BPM",
- "sources": [
- {
- "type": "scan_segment",
- "signals": {
- "x": [{"name": "samy", "entry": "samy"}],
- "y": [{"name": "bpm4i"}],
- },
- }
- ],
- },
- ],
- "line_scan": [
- {
- "plot_name": "BPM plots vs samx",
- "x_label": "Motor X",
- "y_label": "Gauss",
- "sources": [
- {
- "type": "scan_segment",
- "signals": {
- "x": [{"name": "samx", "entry": "samx"}],
- "y": [{"name": "bpm4i"}],
- },
- }
- ],
- },
- {
- "plot_name": "Gauss plots vs samx",
- "x_label": "Motor X",
- "y_label": "Gauss",
- "sources": [
- {
- "type": "scan_segment",
- "signals": {
- "x": [{"name": "samx", "entry": "samx"}],
- "y": [{"name": "bpm4i"}, {"name": "bpm4i"}],
- },
- }
- ],
- },
- ],
- },
-}
-
-
-CONFIG_WRONG = {
- "plot_settings": {
- "background_color": "black",
- "num_columns": 2,
- "colormap": "plasma",
- "scan_types": False,
- },
- "plot_data": [
- {
- "plot_name": "BPM4i plots vs samx",
- "x_label": "Motor Y",
- "y_label": "bpm4i",
- "sources": [
- {
- "type": "non_existing_source",
- "signals": {
- "x": [{"name": "samy"}],
- "y": [{"name": "bpm4i", "entry": "bpm4i"}],
- },
- },
- {
- "type": "history",
- "scan_id": "",
- "signals": {
- "x": [{"name": "samy"}],
- "y": [{"name": "bpm4i", "entry": "bpm4i"}],
- },
- },
- ],
- },
- {
- "plot_name": "Gauss plots vs samx",
- "x_label": "Motor X",
- "y_label": "Gauss",
- "sources": [
- {
- "type": "scan_segment",
- "signals": {
- "x": [{"name": "samx", "entry": "non_sense_entry"}],
- "y": [
- {"name": "non_existing_name"},
- {"name": "samy", "entry": "non_existing_entry"},
- ],
- },
- }
- ],
- },
- {
- "plot_name": "Gauss plots vs samx",
- "x_label": "Motor X",
- "y_label": "Gauss",
- "sources": [
- {
- "signals": {
- "x": [{"name": "samx", "entry": "samx"}],
- "y": [{"name": "samx"}, {"name": "samy", "entry": "samx"}],
- }
- }
- ],
- },
- ],
-}
-
-
-CONFIG_SIMPLE = {
- "plot_settings": {
- "background_color": "black",
- "num_columns": 2,
- "colormap": "plasma",
- "scan_types": False,
- },
- "plot_data": [
- {
- "plot_name": "BPM4i plots vs samx",
- "x_label": "Motor X",
- "y_label": "bpm4i",
- "sources": [
- {
- "type": "scan_segment",
- "signals": {
- "x": [{"name": "samx"}],
- "y": [{"name": "bpm4i", "entry": "bpm4i"}],
- },
- },
- # {
- # "type": "history",
- # "signals": {
- # "x": [{"name": "samx"}],
- # "y": [{"name": "bpm4i", "entry": "bpm4i"}],
- # },
- # },
- # {
- # "type": "dap",
- # 'worker':'some_worker',
- # "signals": {
- # "x": [{"name": "samx"}],
- # "y": [{"name": "bpm4i", "entry": "bpm4i"}],
- # },
- # },
- ],
- },
- {
- "plot_name": "Gauss plots vs samx",
- "x_label": "Motor X",
- "y_label": "Gauss",
- "sources": [
- {
- "type": "scan_segment",
- "signals": {
- "x": [{"name": "samx", "entry": "samx"}],
- "y": [{"name": "bpm4i"}, {"name": "bpm4i"}],
- },
- }
- ],
- },
- ],
-}
-
-CONFIG_REDIS = {
- "plot_settings": {
- "background_color": "white",
- "axis_width": 2,
- "num_columns": 5,
- "colormap": "plasma",
- "scan_types": False,
- },
- "plot_data": [
- {
- "plot_name": "BPM4i plots vs samx",
- "x_label": "Motor Y",
- "y_label": "bpm4i",
- "sources": [
- {
- "type": "scan_segment",
- "signals": {"x": [{"name": "samx"}], "y": [{"name": "gauss_bpm"}]},
- },
- {
- "type": "redis",
- "endpoint": "public/gui/data/6cd5ea3f-a9a9-4736-b4ed-74ab9edfb996",
- "update": "append",
- "signals": {"x": [{"name": "x_default_tag"}], "y": [{"name": "y_default_tag"}]},
- },
- ],
- }
- ],
-}
-
-
-class BECMonitor(pg.GraphicsLayoutWidget):
- update_signal = pyqtSignal()
-
- def __init__(
- self,
- parent=None,
- client=None,
- config: dict = None,
- enable_crosshair: bool = True,
- gui_id=None,
- skip_validation: bool = False,
- ):
- super().__init__(parent=parent)
-
- # Client and device manager from BEC
- self.plot_data = None
- bec_dispatcher = BECDispatcher()
- self.client = bec_dispatcher.client if client is None else client
- self.dev = self.client.device_manager.devices
- self.queue = self.client.queue
-
- self.validator = MonitorConfigValidator(self.dev)
- self.gui_id = gui_id
-
- if self.gui_id is None:
- self.gui_id = self.__class__.__name__ + str(time.time())
-
- # Connect slots dispatcher
- bec_dispatcher.connect_slot(self.on_scan_segment, MessageEndpoints.scan_segment())
- bec_dispatcher.connect_slot(self.on_config_update, MessageEndpoints.gui_config(self.gui_id))
- bec_dispatcher.connect_slot(
- self.on_instruction, MessageEndpoints.gui_instructions(self.gui_id)
- )
- bec_dispatcher.connect_slot(self.on_data_from_redis, MessageEndpoints.gui_data(self.gui_id))
-
- # Current configuration
- self.config = config
- self.skip_validation = skip_validation
-
- # Enable crosshair
- self.enable_crosshair = enable_crosshair
-
- # Displayed Data
- self.database = None
-
- self.crosshairs = None
- self.plots = None
- self.curves_data = None
- self.grid_coordinates = None
- self.scan_id = None
-
- # TODO make colors accessible to users
- self.user_colors = {} # key: (plot_name, y_name, y_entry), value: color
-
- # Connect the update signal to the update plot method
- self.proxy_update_plot = pg.SignalProxy(
- self.update_signal, rateLimit=25, slot=self.update_scan_segment_plot
- )
-
- # Init UI
- if self.config is None:
- print("No initial config found for BECDeviceMonitor")
- else:
- self.on_config_update(self.config)
-
- def _init_config(self):
- """
- Initializes or update the configuration settings for the PlotApp.
- """
-
- # Separate configs
- self.plot_settings = self.config.get("plot_settings", {})
- self.plot_data_config = self.config.get("plot_data", {})
- self.scan_types = self.plot_settings.get("scan_types", False)
-
- if self.scan_types is False: # Device tracking mode
- self.plot_data = self.plot_data_config # TODO logic has to be improved
- else: # without incoming data setup the first configuration to the first scan type sorted alphabetically by name
- self.plot_data = self.plot_data_config[min(list(self.plot_data_config.keys()))]
-
- # Initialize the database
- self.database = self._init_database(self.plot_data)
-
- # Initialize the UI
- self._init_ui(self.plot_settings["num_columns"])
-
- if self.scan_id is not None:
- self.replot_last_scan()
-
- def _init_database(self, plot_data_config: dict, source_type_to_init=None) -> dict:
- """
- Initializes or updates the database for the PlotApp.
- Args:
- plot_data_config(dict): Configuration settings for plots.
- source_type_to_init(str, optional): Specific source type to initialize. If None, initialize all.
- Returns:
- dict: Updated or new database dictionary.
- """
- database = {} if source_type_to_init is None else self.database.copy()
-
- for plot in plot_data_config:
- for source in plot["sources"]:
- source_type = source["type"]
- if source_type_to_init and source_type != source_type_to_init:
- continue # Skip if not the specified source type
-
- if source_type not in database:
- database[source_type] = {}
-
- for axis, signals in source["signals"].items():
- for signal in signals:
- name = signal["name"]
- entry = signal.get("entry", name)
- if name not in database[source_type]:
- database[source_type][name] = {}
- if entry not in database[source_type][name]:
- database[source_type][name][entry] = []
-
- return database
-
- def _init_ui(self, num_columns: int = 3) -> None:
- """
- Initialize the UI components, create plots and store their grid positions.
-
- Args:
- num_columns (int): Number of columns to wrap the layout.
-
- This method initializes a dictionary `self.plots` to store the plot objects
- along with their corresponding x and y signal names. It dynamically arranges
- the plots in a grid layout based on the given number of columns and dynamically
- stretches the last plots to fit the remaining space.
- """
- self.clear()
- self.plots = {}
- self.grid_coordinates = []
-
- num_plots = len(self.plot_data)
-
- # Check if num_columns exceeds the number of plots
- if num_columns >= num_plots:
- num_columns = num_plots
- self.plot_settings["num_columns"] = num_columns # Update the settings
- print(
- "Warning: num_columns in the YAML file was greater than the number of plots."
- f" Resetting num_columns to number of plots:{num_columns}."
- )
- else:
- self.plot_settings["num_columns"] = num_columns # Update the settings
-
- num_rows = num_plots // num_columns
- last_row_cols = num_plots % num_columns
- remaining_space = num_columns - last_row_cols
-
- for i, plot_config in enumerate(self.plot_data):
- row, col = i // num_columns, i % num_columns
- colspan = 1
-
- if row == num_rows and remaining_space > 0:
- if last_row_cols == 1:
- colspan = num_columns
- else:
- colspan = remaining_space // last_row_cols + 1
- remaining_space -= colspan - 1
- last_row_cols -= 1
-
- plot_name = plot_config.get("plot_name", "")
-
- x_label = plot_config.get("x_label", "")
- y_label = plot_config.get("y_label", "")
-
- plot = self.addPlot(row=row, col=col, colspan=colspan, title=plot_name)
- plot.setLabel("bottom", x_label)
- plot.setLabel("left", y_label)
- plot.addLegend()
- self._set_plot_colors(plot, self.plot_settings)
-
- self.plots[plot_name] = plot
- self.grid_coordinates.append((row, col))
-
- # Initialize curves
- self.init_curves()
-
- def _set_plot_colors(self, plot: pg.PlotItem, plot_settings: dict) -> None:
- """
- Set the plot colors based on the plot config.
-
- Args:
- plot (pg.PlotItem): Plot object to set the colors.
- plot_settings (dict): Plot settings dictionary.
- """
- if plot_settings.get("show_grid", False):
- plot.showGrid(x=True, y=True, alpha=0.5)
- pen_width = plot_settings.get("axis_width")
- color = plot_settings.get("axis_color")
- if color is None:
- if plot_settings["background_color"].lower() == "black":
- color = "w"
- self.setBackground("k")
- elif plot_settings["background_color"].lower() == "white":
- color = "k"
- self.setBackground("w")
- else:
- raise ValueError(
- f"Invalid background color {plot_settings['background_color']}. Allowed values"
- " are 'white' or 'black'."
- )
- pen = pg.mkPen(color=color, width=pen_width)
- x_axis = plot.getAxis("bottom") # 'bottom' corresponds to the x-axis
- x_axis.setPen(pen)
- x_axis.setTextPen(pen)
- x_axis.setTickPen(pen)
-
- y_axis = plot.getAxis("left") # 'left' corresponds to the y-axis
- y_axis.setPen(pen)
- y_axis.setTextPen(pen)
- y_axis.setTickPen(pen)
-
- def init_curves(self) -> None:
- """
- Initialize curve data and properties for each plot and data source.
- """
- self.curves_data = {}
-
- for idx, plot_config in enumerate(self.plot_data):
- plot_name = plot_config.get("plot_name", "")
- plot = self.plots[plot_name]
- plot.clear()
-
- for source in plot_config["sources"]:
- source_type = source["type"]
- y_signals = source["signals"].get("y", [])
- colors_ys = Colors.golden_angle_color(
- colormap=self.plot_settings["colormap"], num=len(y_signals)
- )
-
- if source_type not in self.curves_data:
- self.curves_data[source_type] = {}
- if plot_name not in self.curves_data[source_type]:
- self.curves_data[source_type][plot_name] = []
-
- for i, (y_signal, color) in enumerate(zip(y_signals, colors_ys)):
- y_name = y_signal["name"]
- y_entry = y_signal.get("entry", y_name)
- curve_name = f"{y_name} ({y_entry})-{source_type[0].upper()}"
- curve_data = self.create_curve(curve_name, color)
- plot.addItem(curve_data)
- self.curves_data[source_type][plot_name].append((y_name, y_entry, curve_data))
-
- # Render static plot elements
- self.update_plot()
- # # Hook Crosshair #TODO enable later, currently not working
- if self.enable_crosshair is True:
- self.hook_crosshair()
-
- def create_curve(self, curve_name: str, color: str) -> pg.PlotDataItem:
- """
- Create
- Args:
- curve_name: Name of the curve
- color(str): Color of the curve
-
- Returns:
- pg.PlotDataItem: Assigned curve object
- """
- user_color = self.user_colors.get(curve_name, None)
- color_to_use = user_color if user_color else color
- pen_curve = mkPen(color=color_to_use, width=2, style=QtCore.Qt.DashLine)
- brush_curve = mkBrush(color=color_to_use)
-
- return pg.PlotDataItem(
- symbolSize=5,
- symbolBrush=brush_curve,
- pen=pen_curve,
- skipFiniteCheck=True,
- name=curve_name,
- )
-
- def hook_crosshair(self) -> None:
- """Hook the crosshair to all plots."""
- # TODO can be extended to hook crosshair signal for mouse move/clicked
- self.crosshairs = {}
- for plot_name, plot in self.plots.items():
- crosshair = Crosshair(plot, precision=3)
- self.crosshairs[plot_name] = crosshair
-
- def update_scan_segment_plot(self):
- """
- Update the plot with the latest scan segment data.
- """
- self.update_plot(source_type="scan_segment")
-
- def update_plot(self, source_type=None) -> None:
- """
- Update the plot data based on the stored data dictionary.
- Only updates data for the specified source_type if provided.
- """
- for src_type, plots in self.curves_data.items():
- if source_type and src_type != source_type:
- continue
-
- for plot_name, curve_list in plots.items():
- plot_config = next(
- (pc for pc in self.plot_data if pc.get("plot_name") == plot_name), None
- )
- if not plot_config:
- continue
-
- x_name, x_entry = self.extract_x_config(plot_config, src_type)
-
- for y_name, y_entry, curve in curve_list:
- data_x = self.database.get(src_type, {}).get(x_name, {}).get(x_entry, [])
- data_y = self.database.get(src_type, {}).get(y_name, {}).get(y_entry, [])
- curve.setData(data_x, data_y)
-
- def extract_x_config(self, plot_config: dict, source_type: str) -> tuple:
- """Extract the signal configurations for x and y axes from plot_config.
- Args:
- plot_config (dict): Plot configuration.
- Returns:
- tuple: Tuple containing the x name and x entry.
- """
- x_name, x_entry = None, None
-
- for source in plot_config["sources"]:
- if source["type"] == source_type and "x" in source["signals"]:
- x_signal = source["signals"]["x"][0]
- x_name = x_signal.get("name")
- x_entry = x_signal.get("entry", x_name)
- return x_name, x_entry
-
- def get_config(self):
- """Return the current configuration settings."""
- return self.config
-
- def show_config_dialog(self):
- """Show the configuration dialog."""
-
- dialog = ConfigDialog(
- client=self.client, default_config=self.config, skip_validation=self.skip_validation
- )
- dialog.config_updated.connect(self.on_config_update)
- dialog.show()
-
- def update_client(self, client) -> None:
- """Update the client and device manager from BEC.
- Args:
- client: BEC client
- """
- self.client = client
- self.dev = self.client.device_manager.devices
-
- def _close_all_plots(self):
- """Close all plots."""
- for plot in self.plots.values():
- plot.clear()
-
- @pyqtSlot(dict)
- def on_instruction(self, msg_content: dict) -> None:
- """
- Handle instructions sent to the GUI.
- Possible actions are:
- - clear: Clear the plots
- - close: Close the GUI
- - config_dialog: Open the configuration dialog
-
- Args:
- msg_content (dict): Message content with the instruction and parameters.
- """
- action = msg_content.get("action", None)
- parameters = msg_content.get("parameters", None)
-
- if action == "clear":
- self.flush()
- self._close_all_plots()
- elif action == "close":
- self.close()
- elif action == "config_dialog":
- self.show_config_dialog()
- else:
- print(f"Unknown instruction received: {msg_content}")
-
- @pyqtSlot(dict)
- def on_config_update(self, config: dict) -> None:
- """
- Validate and update the configuration settings for the PlotApp.
- Args:
- config(dict): Configuration settings
- """
- # convert config from BEC CLI to correct formatting
- config_tag = config.get("config", None)
- if config_tag is not None:
- config = config["config"]
-
- if self.skip_validation is True:
- self.config = config
- self._init_config()
- else:
- try:
- validated_config = self.validator.validate_monitor_config(config)
- self.config = validated_config.model_dump()
- self._init_config()
- except ValidationError as e:
- error_str = str(e)
- formatted_error_message = BECMonitor.format_validation_error(error_str)
-
- # Display the formatted error message in a popup
- QMessageBox.critical(self, "Configuration Error", formatted_error_message)
-
- @staticmethod
- def format_validation_error(error_str: str) -> str:
- """
- Format the validation error string to be displayed in a popup.
- Args:
- error_str(str): Error string from the validation error.
- """
- error_lines = error_str.split("\n")
- # The first line contains the number of errors.
- error_header = f"{error_lines[0]}
"
-
- formatted_error_message = error_header
- # Skip the first line as it's the header.
- error_details = error_lines[1:]
-
- # Iterate through pairs of lines (each error's two lines).
- for i in range(0, len(error_details), 2):
- location = error_details[i]
- message = error_details[i + 1] if i + 1 < len(error_details) else ""
-
- formatted_error_message += f"{location}
{message}
"
-
- return formatted_error_message
-
- def flush(self, flush_all=False, source_type_to_flush=None) -> None:
- """Update or reset the database to match the current configuration.
-
- Args:
- flush_all (bool): If True, reset the entire database.
- source_type_to_flush (str): Specific source type to reset. Ignored if flush_all is True.
- """
- if flush_all:
- self.database = self._init_database(self.plot_data)
- self.init_curves()
- else:
- if source_type_to_flush in self.database:
- # TODO maybe reinit the database from config again instead of cycle through names/entries
- # Reset only the specified source type
- for name in self.database[source_type_to_flush]:
- for entry in self.database[source_type_to_flush][name]:
- self.database[source_type_to_flush][name][entry] = []
- # Reset curves for the specified source type
- if source_type_to_flush in self.curves_data:
- self.init_curves()
-
- @pyqtSlot(dict, dict)
- def on_scan_segment(self, msg: dict, metadata: dict):
- """
- Handle new scan segments and saves data to a dictionary. Linked through bec_dispatcher.
-
- Args:
- msg (dict): Message received with scan data.
- metadata (dict): Metadata of the scan.
- """
- current_scan_id = msg.get("scan_id", None)
- if current_scan_id is None:
- return
-
- if current_scan_id != self.scan_id:
- if self.scan_types is False:
- self.plot_data = self.plot_data_config
- elif self.scan_types is True:
- current_name = metadata.get("scan_name")
- if current_name is None:
- raise ValueError(
- "Scan name not found in metadata. Please check the scan_name in the YAML"
- " config or in bec configuration."
- )
- self.plot_data = self.plot_data_config.get(current_name, None)
- if not self.plot_data:
- raise ValueError(
- f"Scan name {current_name} not found in the YAML config. Please check the scan_name in the "
- "YAML config or in bec configuration."
- )
-
- # Init UI
- self._init_ui(self.plot_settings["num_columns"])
-
- self.scan_id = current_scan_id
- self.scan_data = self.queue.scan_storage.find_scan_by_ID(self.scan_id)
- if not self.scan_data:
- print(f"No data found for scan_id: {self.scan_id}") # TODO better error
- return
- self.flush(source_type_to_flush="scan_segment")
-
- self.scan_segment_update()
-
- self.update_signal.emit()
-
- def scan_segment_update(self):
- """
- Update the database with data from scan storage based on the provided scan_id.
- """
- scan_data = self.scan_data.data
- for device_name, device_entries in self.database.get("scan_segment", {}).items():
- for entry in device_entries.keys():
- dataset = scan_data[device_name][entry].val
- if dataset:
- self.database["scan_segment"][device_name][entry] = dataset
- else:
- print(f"No data found for {device_name} {entry}")
-
- def replot_last_scan(self):
- """
- Replot the last scan.
- """
- self.scan_segment_update()
- self.update_plot(source_type="scan_segment")
-
- @pyqtSlot(dict)
- def on_data_from_redis(self, msg) -> None:
- """
- Handle new data sent from redis.
- Args:
- msg (dict): Message received with data.
- """
-
- # wait until new config is loaded
- while "redis" not in self.database:
- time.sleep(0.1)
- self._init_database(
- self.plot_data, source_type_to_init="redis"
- ) # add database entry for redis dataset
-
- data = msg.get("data", {})
- x_data = data.get("x", {})
- y_data = data.get("y", {})
-
- # Update x data
- if x_data:
- x_tag = x_data.get("tag")
- self.database["redis"][x_tag][x_tag] = x_data["data"]
-
- # Update y data
- for y_tag, y_info in y_data.items():
- self.database["redis"][y_tag][y_tag] = y_info["data"]
-
- # Trigger plot update
- self.update_plot(source_type="redis")
- print(f"database after: {self.database}")
-
-
-if __name__ == "__main__": # pragma: no cover
- import argparse
- import json
- import sys
-
- parser = argparse.ArgumentParser()
- parser.add_argument("--config_file", help="Path to the config file.")
- parser.add_argument("--config", help="Path to the config file.")
- parser.add_argument("--id", help="GUI ID.")
- args = parser.parse_args()
-
- if args.config is not None:
- # Load config from file
- config = json.loads(args.config)
- elif args.config_file is not None:
- # Load config from file
- config = yaml_dialog.load_yaml(args.config_file)
- else:
- config = CONFIG_SIMPLE
-
- client = BECDispatcher().client
- client.start()
- app = QApplication(sys.argv)
- monitor = BECMonitor(config=config, gui_id=args.id, skip_validation=False)
- monitor.show()
- # just to test redis data
- # redis_data = {
- # "x": {"data": [1, 2, 3], "tag": "x_default_tag"},
- # "y": {"y_default_tag": {"data": [1, 2, 3]}},
- # }
- # monitor.on_data_from_redis({"data": redis_data})
- sys.exit(app.exec())
diff --git a/bec_widgets/widgets/monitor/tab_template.ui b/bec_widgets/widgets/monitor/tab_template.ui
deleted file mode 100644
index 757bf287..00000000
--- a/bec_widgets/widgets/monitor/tab_template.ui
+++ /dev/null
@@ -1,180 +0,0 @@
-
-
- Form
-
-
-
- 0
- 0
- 506
- 592
-
-
-
- Form
-
-
- -
-
-
- General
-
-
-
-
-
-
- X Label
-
-
-
- -
-
-
- Qt::Horizontal
-
-
-
- -
-
-
- -
-
-
- Y Label
-
-
-
- -
-
-
- -
-
-
- -
-
-
- Plot Title
-
-
-
- -
-
-
- Qt::Vertical
-
-
-
-
-
-
- -
-
-
-
-
-
- X Axis
-
-
-
-
-
-
- Name:
-
-
-
- -
-
-
- -
-
-
- Qt::Vertical
-
-
-
- -
-
-
- Entry:
-
-
-
- -
-
-
-
-
-
- -
-
-
- Y Signals
-
-
-
-
-
-
-
- Name
-
-
-
-
- Entries
-
-
-
-
- Color
-
-
-
-
-
-
-
- -
-
-
-
-
-
- Add New Plot
-
-
-
- -
-
-
- Qt::Horizontal
-
-
-
- 40
- 20
-
-
-
-
- -
-
-
- Add New Signal
-
-
-
-
-
-
-
-
-
-
-
- BECTable
- QTableWidget
-
-
-
-
-
-
diff --git a/bec_widgets/widgets/motor_map/__init__.py b/bec_widgets/widgets/motor_map/__init__.py
deleted file mode 100644
index f48fc5eb..00000000
--- a/bec_widgets/widgets/motor_map/__init__.py
+++ /dev/null
@@ -1 +0,0 @@
-from .motor_map import MotorMap
diff --git a/bec_widgets/widgets/motor_map/motor_map.py b/bec_widgets/widgets/motor_map/motor_map.py
deleted file mode 100644
index 168f697b..00000000
--- a/bec_widgets/widgets/motor_map/motor_map.py
+++ /dev/null
@@ -1,594 +0,0 @@
-# pylint: disable = no-name-in-module,missing-module-docstring
-from __future__ import annotations
-
-import time
-from typing import Any, Union
-
-import numpy as np
-import pyqtgraph as pg
-from bec_lib.endpoints import MessageEndpoints
-from qtpy import QtCore, QtGui
-from qtpy.QtCore import Signal as pyqtSignal
-from qtpy.QtCore import Slot as pyqtSlot
-from qtpy.QtWidgets import QApplication
-
-from bec_widgets.utils.bec_dispatcher import BECDispatcher
-from bec_widgets.utils.yaml_dialog import load_yaml
-
-CONFIG_DEFAULT = {
- "plot_settings": {
- "colormap": "Greys",
- "scatter_size": 5,
- "max_points": 1000,
- "num_dim_points": 100,
- "precision": 2,
- "num_columns": 1,
- "background_value": 25,
- },
- "motors": [
- {
- "plot_name": "Motor Map",
- "x_label": "Motor X",
- "y_label": "Motor Y",
- "signals": {
- "x": [{"name": "samx", "entry": "samx"}],
- "y": [{"name": "samy", "entry": "samy"}],
- },
- },
- {
- "plot_name": "Motor Map 2 ",
- "x_label": "Motor X",
- "y_label": "Motor Y",
- "signals": {
- "x": [{"name": "aptrx", "entry": "aptrx"}],
- "y": [{"name": "aptry", "entry": "aptry"}],
- },
- },
- ],
-}
-
-
-class MotorMap(pg.GraphicsLayoutWidget):
- update_signal = pyqtSignal()
-
- def __init__(
- self,
- parent=None,
- client=None,
- config: dict = None,
- gui_id=None,
- skip_validation: bool = True,
- ):
- super().__init__(parent=parent)
-
- # Import BEC related stuff
- bec_dispatcher = BECDispatcher()
- self.client = bec_dispatcher.client if client is None else client
- self.dev = self.client.device_manager.devices
-
- # TODO import validator when prepared
- self.gui_id = gui_id
-
- if self.gui_id is None:
- self.gui_id = self.__class__.__name__ + str(time.time())
-
- # Current configuration
- self.config = config
- self.skip_validation = skip_validation # TODO implement validation when validator is ready
-
- # Connect the update signal to the update plot method
- self.proxy_update_plot = pg.SignalProxy(
- self.update_signal, rateLimit=25, slot=self._update_plots
- )
-
- # Config related variables
- self.plot_data = None
- self.plot_settings = None
- self.max_points = None
- self.num_dim_points = None
- self.scatter_size = None
- self.precision = None
- self.background_value = None
- self.database = {}
- self.device_mapping = {}
- self.plots = {}
- self.grid_coordinates = []
- self.curves_data = {}
-
- # Init UI with config
- if self.config is None:
- print("No initial config found for MotorMap. Using default config.")
- else:
- self.on_config_update(self.config)
-
- @pyqtSlot(dict)
- def on_config_update(self, config: dict) -> None:
- """
- Validate and update the configuration settings for the PlotApp.
- Args:
- config(dict): Configuration settings
- """
- # TODO implement BEC CLI commands similar to BECPlotter
- # convert config from BEC CLI to correct formatting
- config_tag = config.get("config", None)
- if config_tag is not None:
- config = config["config"]
-
- if self.skip_validation is True:
- self.config = config
- self._init_config()
-
- else: # TODO implement validator
- print("Do validation")
-
- @pyqtSlot(str, str, int)
- def change_motors(self, motor_x: str, motor_y: str, subplot: int = 0) -> None:
- """
- Change the active motors for the plot.
- Args:
- motor_x(str): Motor name for the X axis.
- motor_y(str): Motor name for the Y axis.
- subplot(int): Subplot number.
- """
- if subplot >= len(self.plot_data):
- print(f"Invalid subplot index: {subplot}. Available subplots: {len(self.plot_data)}")
- return
-
- # Update the motor names in the plot configuration
- self.config["motors"][subplot]["signals"]["x"][0]["name"] = motor_x
- self.config["motors"][subplot]["signals"]["x"][0]["entry"] = motor_x
- self.config["motors"][subplot]["signals"]["y"][0]["name"] = motor_y
- self.config["motors"][subplot]["signals"]["y"][0]["entry"] = motor_y
-
- # reinitialise the config and UI
- self._init_config()
-
- def _init_config(self):
- """Initiate the configuration."""
-
- # Global widget settings
- self._get_global_settings()
-
- # Motor settings
- self.plot_data = self.config.get("motors", {})
-
- # Include motor limits into the config
- self._add_limits_to_plot_data()
-
- # Initialize the database
- self.database = self._init_database()
-
- # Create device mapping for x/y motor pairs
- self.device_mapping = self._create_device_mapping()
-
- # Initialize the plot UI
- self._init_ui()
-
- # Connect motors to slots
- self._connect_motors_to_slots()
-
- # Render init position of selected motors
- self._update_plots()
-
- def _get_global_settings(self):
- """Get global settings from the config."""
- self.plot_settings = self.config.get("plot_settings", {})
-
- self.max_points = self.plot_settings.get("max_points", 5000)
- self.num_dim_points = self.plot_settings.get("num_dim_points", 100)
- self.scatter_size = self.plot_settings.get("scatter_size", 5)
- self.precision = self.plot_settings.get("precision", 2)
- self.background_value = self.plot_settings.get("background_value", 25)
-
- def _create_device_mapping(self):
- """
- Create a mapping of device names to their corresponding x/y devices.
- """
- mapping = {}
- for motor in self.config.get("motors", []):
- for axis in ["x", "y"]:
- for signal in motor["signals"][axis]:
- other_axis = "y" if axis == "x" else "x"
- corresponding_device = motor["signals"][other_axis][0]["name"]
- mapping[signal["name"]] = corresponding_device
- return mapping
-
- def _connect_motors_to_slots(self):
- """Connect motors to slots."""
-
- # Disconnect all slots before connecting a new ones
- bec_dispatcher = BECDispatcher()
- bec_dispatcher.disconnect_all()
-
- # Get list of all unique motors
- unique_motors = []
- for motor_config in self.plot_data:
- for axis in ["x", "y"]:
- for signal in motor_config["signals"][axis]:
- unique_motors.append(signal["name"])
- unique_motors = list(set(unique_motors))
-
- # Create list of endpoint
- endpoints = []
- for motor in unique_motors:
- endpoints.append(MessageEndpoints.device_readback(motor))
-
- # Connect all topics to a single slot
- bec_dispatcher.connect_slot(self.on_device_readback, endpoints)
-
- def _add_limits_to_plot_data(self):
- """
- Add limits to each motor signal in the plot_data.
- """
- for motor_config in self.plot_data:
- for axis in ["x", "y"]:
- for signal in motor_config["signals"][axis]:
- motor_name = signal["name"]
- motor_limits = self._get_motor_limit(motor_name)
- signal["limits"] = motor_limits
-
- def _get_motor_limit(self, motor: str) -> Union[list | None]:
- """
- Get the motor limit from the config.
- Args:
- motor(str): Motor name.
-
- Returns:
- float: Motor limit.
- """
- try:
- limits = self.dev[motor].limits
- if limits == [0, 0]:
- return None
- return limits
- except AttributeError: # TODO maybe not needed, if no limits it returns [0,0]
- # If the motor doesn't have a 'limits' attribute, return a default value or raise a custom exception
- print(f"The device '{motor}' does not have defined limits.")
- return None
-
- def _init_database(self):
- """Initiate the database according the config."""
- database = {}
-
- for plot in self.plot_data:
- for axis, signals in plot["signals"].items():
- for signal in signals:
- name = signal["name"]
- entry = signal.get("entry", name)
- if name not in database:
- database[name] = {}
- if entry not in database[name]:
- database[name][entry] = [self.get_coordinate(name, entry)]
- return database
-
- def get_coordinate(self, name, entry):
- """Get the initial coordinate value for a motor."""
- try:
- return self.dev[name].read()[entry]["value"]
- except Exception as e:
- print(f"Error getting initial value for {name}: {e}")
- return None
-
- def _init_ui(self, num_columns: int = 3) -> None:
- """
- Initialize the UI components, create plots and store their grid positions.
-
- Args:
- num_columns (int): Number of columns to wrap the layout.
-
- This method initializes a dictionary `self.plots` to store the plot objects
- along with their corresponding x and y signal names. It dynamically arranges
- the plots in a grid layout based on the given number of columns and dynamically
- stretches the last plots to fit the remaining space.
- """
- self.clear()
- self.plots = {}
- self.grid_coordinates = []
- self.curves_data = {} # TODO moved from init_curves
-
- num_plots = len(self.plot_data)
-
- # Check if num_columns exceeds the number of plots
- if num_columns >= num_plots:
- num_columns = num_plots
- self.plot_settings["num_columns"] = num_columns # Update the settings
- print(
- "Warning: num_columns in the YAML file was greater than the number of plots."
- f" Resetting num_columns to number of plots:{num_columns}."
- )
- else:
- self.plot_settings["num_columns"] = num_columns # Update the settings
-
- num_rows = num_plots // num_columns
- last_row_cols = num_plots % num_columns
- remaining_space = num_columns - last_row_cols
-
- for i, plot_config in enumerate(self.plot_data):
- row, col = i // num_columns, i % num_columns
- colspan = 1
-
- if row == num_rows and remaining_space > 0:
- if last_row_cols == 1:
- colspan = num_columns
- else:
- colspan = remaining_space // last_row_cols + 1
- remaining_space -= colspan - 1
- last_row_cols -= 1
-
- if "plot_name" not in plot_config:
- plot_name = f"Plot ({row}, {col})"
- plot_config["plot_name"] = plot_name
- else:
- plot_name = plot_config["plot_name"]
-
- x_label = plot_config.get("x_label", "")
- y_label = plot_config.get("y_label", "")
-
- plot = self.addPlot(row=row, col=col, colspan=colspan, title="Motor position: (X, Y)")
- plot.setLabel("bottom", f"{x_label} ({plot_config['signals']['x'][0]['name']})")
- plot.setLabel("left", f"{y_label} ({plot_config['signals']['y'][0]['name']})")
- plot.addLegend()
- # self._set_plot_colors(plot, self.plot_settings) #TODO implement colors
-
- self.plots[plot_name] = plot
- self.grid_coordinates.append((row, col))
-
- self._init_motor_map(plot_config)
-
- def _init_motor_map(self, plot_config: dict) -> None:
- """
- Initialize the motor map.
- Args:
- plot_config(dict): Plot configuration.
- """
-
- # Get plot name to find appropriate plot
- plot_name = plot_config.get("plot_name", "")
-
- # Reset the curves data
- plot = self.plots[plot_name]
- plot.clear()
-
- limits_x, limits_y = plot_config["signals"]["x"][0].get("limits", None), plot_config[
- "signals"
- ]["y"][0].get("limits", None)
- if limits_x is not None and limits_y is not None:
- self._make_limit_map(plot, [limits_x, limits_y])
-
- # Initiate ScatterPlotItem for motor coordinates
- self.curves_data[plot_name] = {
- "pos": pg.ScatterPlotItem(
- size=self.scatter_size, pen=pg.mkPen(None), brush=pg.mkBrush(255, 255, 255, 255)
- )
- }
-
- # Add the scatter plot to the plot
- plot.addItem(self.curves_data[plot_name]["pos"])
- # Set the point map to be always on the top
- self.curves_data[plot_name]["pos"].setZValue(0)
-
- # Add all layers to the plot
- plot.showGrid(x=True, y=True)
-
- # Add the crosshair for motor coordinates
- init_position_x = self._get_motor_init_position(
- plot_config["signals"]["x"][0]["name"], plot_config["signals"]["x"][0]["entry"]
- )
- init_position_y = self._get_motor_init_position(
- plot_config["signals"]["y"][0]["name"], plot_config["signals"]["y"][0]["entry"]
- )
- self._add_coordinantes_crosshair(plot_name, init_position_x, init_position_y)
-
- def _add_coordinantes_crosshair(self, plot_name: str, x: float, y: float) -> None:
- """
- Add crosshair to the plot to highlight the current position.
- Args:
- plot_name(str): Name of the plot.
- x(float): X coordinate.
- y(float): Y coordinate.
- """
- # find the current plot
- plot = self.plots[plot_name]
-
- # Crosshair to highlight the current position
- highlight_H = pg.InfiniteLine(
- angle=0, movable=False, pen=pg.mkPen(color="r", width=1, style=QtCore.Qt.DashLine)
- )
- highlight_V = pg.InfiniteLine(
- angle=90, movable=False, pen=pg.mkPen(color="r", width=1, style=QtCore.Qt.DashLine)
- )
-
- # Add crosshair to the curve list for future referencing
- self.curves_data[plot_name]["highlight_H"] = highlight_H
- self.curves_data[plot_name]["highlight_V"] = highlight_V
-
- # Add crosshair to the plot
- plot.addItem(highlight_H)
- plot.addItem(highlight_V)
-
- highlight_H.setPos(x)
- highlight_V.setPos(y)
-
- def _make_limit_map(self, plot: pg.PlotItem, limits: list):
- """
- Make a limit map from the limits list.
-
- Args:
- plot(pg.PlotItem): Plot to add the limit map to.
- limits(list): List of limits.
- """
- # Define the size of the image map based on the motor's limits
- limit_x_min, limit_x_max = limits[0]
- limit_y_min, limit_y_max = limits[1]
-
- map_width = int(limit_x_max - limit_x_min + 1)
- map_height = int(limit_y_max - limit_y_min + 1)
-
- limit_map_data = np.full((map_width, map_height), self.background_value, dtype=np.float32)
-
- # Create the image map
- limit_map = pg.ImageItem()
- limit_map.setImage(limit_map_data)
- plot.addItem(limit_map)
-
- # Translate and scale the image item to match the motor coordinates
- tr = QtGui.QTransform()
- tr.translate(limit_x_min, limit_y_min)
- limit_map.setTransform(tr)
-
- def _get_motor_init_position(self, name: str, entry: str) -> float:
- """
- Get the motor initial position from the config.
- Args:
- name(str): Motor name.
- entry(str): Motor entry.
- Returns:
- float: Motor initial position.
- """
- init_position = round(self.dev[name].read()[entry]["value"], self.precision)
- return init_position
-
- def _update_plots(self):
- """Update the motor position on plots."""
- for plot_name, curve_list in self.curves_data.items():
- plot_config = next(
- (pc for pc in self.plot_data if pc.get("plot_name") == plot_name), None
- )
- if not plot_config:
- continue
-
- # Get the motor coordinates
- x_motor_name = plot_config["signals"]["x"][0]["name"]
- x_motor_entry = plot_config["signals"]["x"][0]["entry"]
- y_motor_name = plot_config["signals"]["y"][0]["name"]
- y_motor_entry = plot_config["signals"]["y"][0]["entry"]
-
- # update motor position only if there is data
- if (
- len(self.database[x_motor_name][x_motor_entry]) >= 1
- and len(self.database[y_motor_name][y_motor_entry]) >= 1
- ):
- # Relevant data for the plot
- motor_x_data = self.database[x_motor_name][x_motor_entry]
- motor_y_data = self.database[y_motor_name][y_motor_entry]
-
- # Setup gradient brush for history
- brushes = [pg.mkBrush(50, 50, 50, 255)] * len(motor_x_data)
-
- # Calculate the decrement step based on self.num_dim_points
- decrement_step = (255 - 50) / self.num_dim_points
-
- for i in range(1, min(self.num_dim_points + 1, len(motor_x_data) + 1)):
- brightness = max(60, 255 - decrement_step * (i - 1))
- brushes[-i] = pg.mkBrush(brightness, brightness, brightness, 255)
-
- brushes[-1] = pg.mkBrush(
- 255, 255, 255, 255
- ) # Newest point is always full brightness
-
- # Update the scatter plot
- self.curves_data[plot_name]["pos"].setData(
- x=motor_x_data, y=motor_y_data, brush=brushes, pen=None, size=self.scatter_size
- )
-
- # Get last know position for crosshair
- current_x = motor_x_data[-1]
- current_y = motor_y_data[-1]
-
- # Update plot title
- self.plots[plot_name].setTitle(
- f"Motor position: ({round(current_x,self.precision)}, {round(current_y,self.precision)})"
- )
-
- # Update the crosshair
- self.curves_data[plot_name]["highlight_V"].setPos(current_x)
- self.curves_data[plot_name]["highlight_H"].setPos(current_y)
-
- @pyqtSlot(list, str, str)
- def plot_saved_coordinates(self, coordinates: list, tag: str, color: str):
- """
- Plot saved coordinates on the map.
- Args:
- coordinates(list): List of coordinates to be plotted.
- tag(str): Tag for the coordinates for future reference.
- color(str): Color to plot coordinates in.
- """
- for plot_name in self.plots:
- plot = self.plots[plot_name]
-
- # Clear previous saved points
- if tag in self.curves_data[plot_name]:
- plot.removeItem(self.curves_data[plot_name][tag])
-
- # Filter coordinates to be shown
- visible_coords = [coord[:2] for coord in coordinates if coord[2]]
-
- if visible_coords:
- saved_points = pg.ScatterPlotItem(
- pos=np.array(visible_coords), brush=pg.mkBrush(color)
- )
- plot.addItem(saved_points)
- self.curves_data[plot_name][tag] = saved_points
-
- @pyqtSlot(dict)
- def on_device_readback(self, msg: dict):
- """
- Update the motor coordinates on the plots.
- Args:
- msg (dict): Message received with device readback data.
- """
-
- for device_name, device_info in msg["signals"].items():
- # Check if the device is relevant to our current context
- if device_name in self.device_mapping:
- self._update_device_data(device_name, device_info["value"])
-
- self.update_signal.emit()
-
- def _update_device_data(self, device_name: str, value: float):
- """
- Update the device data.
- Args:
- device_name (str): Device name.
- value (float): Device value.
- """
- if device_name in self.database:
- self.database[device_name][device_name].append(value)
-
- corresponding_device = self.device_mapping.get(device_name)
- if corresponding_device and corresponding_device in self.database:
- last_value = (
- self.database[corresponding_device][corresponding_device][-1]
- if self.database[corresponding_device][corresponding_device]
- else None
- )
- self.database[corresponding_device][corresponding_device].append(last_value)
-
-
-if __name__ == "__main__": # pragma: no cover
- import argparse
- import json
- import sys
-
- parser = argparse.ArgumentParser()
- parser.add_argument("--config_file", help="Path to the config file.")
- parser.add_argument("--config", help="Path to the config file.")
- parser.add_argument("--id", help="GUI ID.")
- args = parser.parse_args()
-
- if args.config is not None:
- # Load config from file
- config = json.loads(args.config)
- elif args.config_file is not None:
- # Load config from file
- config = load_yaml(args.config_file)
- else:
- config = CONFIG_DEFAULT
-
- client = BECDispatcher().client
- client.start()
- app = QApplication(sys.argv)
- motor_map = MotorMap(config=config, gui_id=args.id, skip_validation=True)
- motor_map.show()
-
- sys.exit(app.exec())
diff --git a/tests/unit_tests/test_bec_monitor.py b/tests/unit_tests/test_bec_monitor.py
deleted file mode 100644
index 5feac4e7..00000000
--- a/tests/unit_tests/test_bec_monitor.py
+++ /dev/null
@@ -1,220 +0,0 @@
-# pylint: disable = no-name-in-module,missing-class-docstring, missing-module-docstring
-import os
-from unittest.mock import MagicMock
-
-import pytest
-import yaml
-
-from bec_widgets.widgets import BECMonitor
-
-from .client_mocks import mocked_client
-
-
-def load_test_config(config_name):
- """Helper function to load config from yaml file."""
- config_path = os.path.join(os.path.dirname(__file__), "test_configs", f"{config_name}.yaml")
- with open(config_path, "r") as f:
- config = yaml.safe_load(f)
- return config
-
-
-@pytest.fixture(scope="function")
-def monitor(bec_dispatcher, qtbot, mocked_client):
- # client = MagicMock()
- widget = BECMonitor(client=mocked_client)
- qtbot.addWidget(widget)
- qtbot.waitExposed(widget)
- yield widget
-
-
-@pytest.mark.parametrize(
- "config_name, scan_type, number_of_plots",
- [
- ("config_device", False, 2),
- ("config_device_no_entry", False, 2),
- # ("config_scan", True, 4),
- ],
-)
-def test_initialization_with_device_config(monitor, config_name, scan_type, number_of_plots):
- config = load_test_config(config_name)
- monitor.on_config_update(config)
- assert isinstance(monitor, BECMonitor)
- assert monitor.client is not None
- assert len(monitor.plot_data) == number_of_plots
- assert monitor.scan_types == scan_type
-
-
-@pytest.mark.parametrize(
- "config_initial,config_update",
- [("config_device", "config_scan"), ("config_scan", "config_device")],
-)
-def test_on_config_update(monitor, config_initial, config_update):
- config_initial = load_test_config(config_initial)
- config_update = load_test_config(config_update)
- # validated config has to be compared
- config_initial_validated = monitor.validator.validate_monitor_config(
- config_initial
- ).model_dump()
- config_update_validated = monitor.validator.validate_monitor_config(config_update).model_dump()
- monitor.on_config_update(config_initial)
- assert monitor.config == config_initial_validated
- monitor.on_config_update(config_update)
- assert monitor.config == config_update_validated
-
-
-@pytest.mark.parametrize(
- "config_name, expected_num_columns, expected_plot_names, expected_coordinates",
- [
- ("config_device", 1, ["BPM4i plots vs samx", "Gauss plots vs samx"], [(0, 0), (1, 0)]),
- (
- "config_scan",
- 3,
- ["Grid plot 1", "Grid plot 2", "Grid plot 3", "Grid plot 4"],
- [(0, 0), (0, 1), (0, 2), (1, 0)],
- ),
- ],
-)
-def test_render_initial_plots(
- monitor, config_name, expected_num_columns, expected_plot_names, expected_coordinates
-):
- config = load_test_config(config_name)
- monitor.on_config_update(config)
-
- # Validate number of columns
- assert monitor.plot_settings["num_columns"] == expected_num_columns
-
- # Validate the plots are created correctly
- for expected_name in expected_plot_names:
- assert expected_name in monitor.plots.keys()
-
- # Validate the grid_coordinates
- assert monitor.grid_coordinates == expected_coordinates
-
-
-def mock_getitem(dev_name):
- """Helper function to mock the __getitem__ method of the 'dev'."""
- mock_instance = MagicMock()
- if dev_name == "samx":
- mock_instance._hints = "samx"
- elif dev_name == "bpm4i":
- mock_instance._hints = "bpm4i"
- elif dev_name == "gauss_bpm":
- mock_instance._hints = "gauss_bpm"
-
- return mock_instance
-
-
-def mock_get_scan_storage(scan_id, data):
- """Helper function to mock the __getitem__ method of the 'dev'."""
- mock_instance = MagicMock()
- mock_instance.get_scan_storage.return_value = data
- return mock_instance
-
-
-# mocked messages and metadata
-msg_1 = {
- "data": {
- "samx": {"samx": {"value": 10}},
- "bpm4i": {"bpm4i": {"value": 5}},
- "gauss_bpm": {"gauss_bpm": {"value": 6}},
- "gauss_adc1": {"gauss_adc1": {"value": 8}},
- "gauss_adc2": {"gauss_adc2": {"value": 9}},
- },
- "scan_id": 1,
-}
-metadata_grid = {"scan_name": "grid_scan"}
-metadata_line = {"scan_name": "line_scan"}
-
-
-@pytest.mark.parametrize(
- "config_name, msg, metadata, expected_data",
- [
- # case: msg does not have 'scan_id'
- (
- "config_device",
- {"data": {}},
- {},
- {
- "scan_segment": {
- "bpm4i": {"bpm4i": []},
- "gauss_adc1": {"gauss_adc1": []},
- "gauss_adc2": {"gauss_adc2": []},
- "samx": {"samx": []},
- }
- },
- ),
- # case: scan_types is false, msg contains all valid fields, and entry is present in config
- (
- "config_device",
- msg_1,
- {},
- {
- "scan_segment": {
- "bpm4i": {"bpm4i": [5]},
- "gauss_adc1": {"gauss_adc1": [8]},
- "gauss_adc2": {"gauss_adc2": [9]},
- "samx": {"samx": [10]},
- }
- },
- ),
- # case: scan_types is false, msg contains all valid fields and entry is missing in config, should use hints
- (
- "config_device_no_entry",
- msg_1,
- {},
- {
- "scan_segment": {
- "bpm4i": {"bpm4i": [5]},
- "gauss_bpm": {"gauss_bpm": [6]},
- "samx": {"samx": [10]},
- }
- },
- ),
- # case: scan_types is true, msg contains all valid fields, metadata contains scan "line_scan:"
- (
- "config_scan",
- msg_1,
- metadata_line,
- {
- "scan_segment": {
- "bpm4i": {"bpm4i": [5]},
- "gauss_adc1": {"gauss_adc1": [8]},
- "gauss_adc2": {"gauss_adc2": [9]},
- "gauss_bpm": {"gauss_bpm": [6]},
- "samx": {"samx": [10]},
- }
- },
- ),
- (
- "config_scan",
- msg_1,
- metadata_grid,
- {
- "scan_segment": {
- "bpm4i": {"bpm4i": [5]},
- "gauss_adc1": {"gauss_adc1": [8]},
- "gauss_adc2": {"gauss_adc2": [9]},
- "gauss_bpm": {"gauss_bpm": [6]},
- "samx": {"samx": [10]},
- }
- },
- ),
- ],
-)
-def test_on_scan_segment(monitor, config_name, msg, metadata, expected_data):
- config = load_test_config(config_name)
- monitor.on_config_update(config)
-
- # Mock scan_storage.find_scan_by_ID
- mock_scan_data = MagicMock()
- mock_scan_data.data = {
- device_name: {
- entry: MagicMock(val=[msg["data"][device_name][entry]["value"]])
- for entry in msg["data"][device_name]
- }
- for device_name in msg["data"]
- }
- monitor.queue.scan_storage.find_scan_by_ID.return_value = mock_scan_data
-
- monitor.on_scan_segment(msg, metadata)
- assert monitor.database == expected_data
diff --git a/tests/unit_tests/test_config_dialog.py b/tests/unit_tests/test_config_dialog.py
deleted file mode 100644
index 0724a779..00000000
--- a/tests/unit_tests/test_config_dialog.py
+++ /dev/null
@@ -1,178 +0,0 @@
-# pylint: disable = no-name-in-module,missing-class-docstring, missing-module-docstring
-import os
-from unittest.mock import MagicMock
-
-import pytest
-import yaml
-from qtpy.QtWidgets import QTableWidgetItem, QTabWidget
-
-from bec_widgets.widgets.monitor.config_dialog import ConfigDialog
-
-from .client_mocks import mocked_client
-
-
-def load_test_config(config_name):
- """Helper function to load config from yaml file."""
- config_path = os.path.join(os.path.dirname(__file__), "test_configs", f"{config_name}.yaml")
- with open(config_path, "r") as f:
- config = yaml.safe_load(f)
- return config
-
-
-@pytest.fixture(scope="function")
-def config_dialog(qtbot, mocked_client):
- client = mocked_client
- widget = ConfigDialog(client=client)
- qtbot.addWidget(widget)
- qtbot.waitExposed(widget)
- yield widget
-
-
-@pytest.mark.parametrize("config_name", ["config_device", "config_scan"])
-def test_load_config(config_dialog, config_name):
- config = load_test_config(config_name)
- config_dialog.load_config(config)
-
- assert (
- config_dialog.comboBox_appearance.currentText()
- == config["plot_settings"]["background_color"]
- )
- assert config_dialog.spinBox_n_column.value() == config["plot_settings"]["num_columns"]
- assert config_dialog.comboBox_colormap.currentText() == config["plot_settings"]["colormap"]
-
-
-@pytest.mark.parametrize(
- "config_name, scan_mode",
- [("config_device", False), ("config_scan", True), ("config_device_no_entry", False)],
-)
-def test_initialization(config_dialog, config_name, scan_mode):
- config = load_test_config(config_name)
- config_dialog.load_config(config)
-
- assert isinstance(config_dialog, ConfigDialog)
- assert (
- config_dialog.comboBox_appearance.currentText()
- == config["plot_settings"]["background_color"]
- )
- assert config_dialog.spinBox_n_column.value() == config["plot_settings"]["num_columns"]
- assert (config_dialog.comboBox_scanTypes.currentText() == "Enabled") == scan_mode
- assert (
- config_dialog.tabWidget_scan_types.count() > 0
- ) # Ensures there's at least one tab created
-
- # If there's a need to check the contents of the first tab (there has to be always at least one tab)
- first_tab = config_dialog.tabWidget_scan_types.widget(0)
- if scan_mode:
- assert (
- first_tab.findChild(QTabWidget, "tabWidget_plots") is not None
- ) # Ensures plot tab widget exists in scan mode
- else:
- assert (
- first_tab.findChild(QTabWidget) is not None
- ) # Ensures plot tab widget exists in default mode
-
-
-def test_edit_and_apply_config(config_dialog):
- config_device = load_test_config("config_device")
- config_dialog.load_config(config_device)
-
- config_dialog.comboBox_appearance.setCurrentText("white")
- config_dialog.spinBox_n_column.setValue(2)
- config_dialog.comboBox_colormap.setCurrentText("viridis")
-
- applied_config = config_dialog.apply_config()
-
- assert applied_config["plot_settings"]["background_color"] == "white"
- assert applied_config["plot_settings"]["num_columns"] == 2
- assert applied_config["plot_settings"]["colormap"] == "viridis"
-
-
-def test_edit_and_apply_config_scan_mode(config_dialog):
- config_scan = load_test_config("config_scan")
- config_dialog.load_config(config_scan)
-
- config_dialog.comboBox_appearance.setCurrentText("white")
- config_dialog.spinBox_n_column.setValue(2)
- config_dialog.comboBox_colormap.setCurrentText("viridis")
- config_dialog.comboBox_scanTypes.setCurrentText("Enabled")
-
- applied_config = config_dialog.apply_config()
-
- assert applied_config["plot_settings"]["background_color"] == "white"
- assert applied_config["plot_settings"]["num_columns"] == 2
- assert applied_config["plot_settings"]["colormap"] == "viridis"
- assert applied_config["plot_settings"]["scan_types"] is True
-
-
-def test_add_new_scan(config_dialog):
- # Ensure the tab count is initially 1 (from the default config)
- assert config_dialog.tabWidget_scan_types.count() == 1
-
- # Add a new scan tab
- config_dialog.add_new_scan_tab(config_dialog.tabWidget_scan_types, "Test Scan Tab")
-
- # Ensure the tab count is now 2
- assert config_dialog.tabWidget_scan_types.count() == 2
-
- # Ensure the new tab has the correct name
- assert config_dialog.tabWidget_scan_types.tabText(1) == "Test Scan Tab"
-
-
-def test_add_new_plot_and_modify(config_dialog):
- # Ensure the tab count is initially 1 and it is called "Default"
- assert config_dialog.tabWidget_scan_types.count() == 1
- assert config_dialog.tabWidget_scan_types.tabText(0) == "Default"
-
- # Get the first tab (which should be a scan tab)
- scan_tab = config_dialog.tabWidget_scan_types.widget(0)
-
- # Ensure the plot tab count is initially 1 and it is called "Plot 1"
- tabWidget_plots = scan_tab.findChild(QTabWidget)
- assert tabWidget_plots.count() == 1
- assert tabWidget_plots.tabText(0) == "Plot 1"
-
- # Add a new plot tab
- config_dialog.add_new_plot_tab(scan_tab)
-
- # Ensure the plot tab count is now 2
- assert tabWidget_plots.count() == 2
-
- # Ensure the new tab has the correct name
- assert tabWidget_plots.tabText(1) == "Plot 2"
-
- # Access the new plot tab
- new_plot_tab = tabWidget_plots.widget(1)
-
- # Modify the line edits within the new plot tab
- new_plot_tab.ui.lineEdit_plot_title.setText("Modified Plot Title")
- new_plot_tab.ui.lineEdit_x_label.setText("Modified X Label")
- new_plot_tab.ui.lineEdit_y_label.setText("Modified Y Label")
- new_plot_tab.ui.lineEdit_x_name.setText("Modified X Name")
- new_plot_tab.ui.lineEdit_x_entry.setText("Modified X Entry")
-
- # Modify the table for signals
- config_dialog.add_new_signal(new_plot_tab.ui.tableWidget_y_signals)
-
- table = new_plot_tab.ui.tableWidget_y_signals
- assert table.rowCount() == 1 # Ensure the new row is added
-
- row_position = table.rowCount() - 1
-
- # Modify the first row
- table.setItem(row_position, 0, QTableWidgetItem("New Signal Name"))
- table.setItem(row_position, 1, QTableWidgetItem("New Signal Entry"))
-
- # Apply the configuration
- config = config_dialog.apply_config()
-
- # Check if the modifications are reflected in the configuration
- modified_plot_config = config["plot_data"][1] # Access the second plot in the plot_data list
- sources = modified_plot_config["sources"][0] # Access the first source in the sources list
-
- assert modified_plot_config["plot_name"] == "Modified Plot Title"
- assert modified_plot_config["x_label"] == "Modified X Label"
- assert modified_plot_config["y_label"] == "Modified Y Label"
- assert sources["signals"]["x"][0]["name"] == "Modified X Name"
- assert sources["signals"]["x"][0]["entry"] == "Modified X Entry"
- assert sources["signals"]["y"][0]["name"] == "New Signal Name"
- assert sources["signals"]["y"][0]["entry"] == "New Signal Entry"
diff --git a/tests/unit_tests/test_motor_map.py b/tests/unit_tests/test_motor_map.py
deleted file mode 100644
index 6c6321bb..00000000
--- a/tests/unit_tests/test_motor_map.py
+++ /dev/null
@@ -1,171 +0,0 @@
-# pylint: disable = no-name-in-module,missing-module-docstring, missing-function-docstring
-from unittest.mock import MagicMock
-
-import pytest
-
-from bec_widgets.widgets import MotorMap
-
-from .client_mocks import mocked_client
-
-CONFIG_DEFAULT = {
- "plot_settings": {
- "colormap": "Greys",
- "scatter_size": 5,
- "max_points": 1000,
- "num_dim_points": 100,
- "precision": 2,
- "num_columns": 1,
- "background_value": 25,
- },
- "motors": [
- {
- "plot_name": "Motor Map",
- "x_label": "Motor X",
- "y_label": "Motor Y",
- "signals": {
- "x": [{"name": "samx", "entry": "samx"}],
- "y": [{"name": "samy", "entry": "samy"}],
- },
- },
- {
- "plot_name": "Motor Map 2 ",
- "x_label": "Motor X",
- "y_label": "Motor Y",
- "signals": {
- "x": [{"name": "aptrx", "entry": "aptrx"}],
- "y": [{"name": "aptry", "entry": "aptry"}],
- },
- },
- ],
-}
-
-CONFIG_ONE_DEVICE = {
- "plot_settings": {
- "colormap": "Greys",
- "scatter_size": 5,
- "max_points": 1000,
- "num_dim_points": 100,
- "precision": 2,
- "num_columns": 1,
- "background_value": 25,
- },
- "motors": [
- {
- "plot_name": "Motor Map",
- "x_label": "Motor X",
- "y_label": "Motor Y",
- "signals": {
- "x": [{"name": "samx", "entry": "samx"}],
- "y": [{"name": "samy", "entry": "samy"}],
- },
- }
- ],
-}
-
-
-@pytest.fixture(scope="function")
-def motor_map(qtbot, mocked_client):
- widget = MotorMap(client=mocked_client)
- qtbot.addWidget(widget)
- qtbot.waitExposed(widget)
- yield widget
-
-
-def test_motor_limits_initialization(motor_map):
- # Example test to check if motor limits are correctly initialized
- expected_limits = {"samx": [-10, 10], "samy": [-5, 5]}
- for motor_name, expected_limit in expected_limits.items():
- actual_limit = motor_map._get_motor_limit(motor_name)
- assert actual_limit == expected_limit
-
-
-def test_motor_initial_position(motor_map):
- motor_map.precision = 2
-
- motor_map_dev = motor_map.client.device_manager.devices
-
- # Example test to check if motor initial positions are correctly initialized
- expected_positions = {
- ("samx", "samx"): motor_map_dev["samx"].read()["samx"]["value"],
- ("samy", "samy"): motor_map_dev["samy"].read()["samy"]["value"],
- ("aptrx", "aptrx"): motor_map_dev["aptrx"].read()["aptrx"]["value"],
- ("aptry", "aptry"): motor_map_dev["aptry"].read()["aptry"]["value"],
- }
- for (motor_name, entry), expected_position in expected_positions.items():
- actual_position = motor_map._get_motor_init_position(motor_name, entry)
- assert actual_position == expected_position
-
-
-@pytest.mark.parametrize("config, number_of_plots", [(CONFIG_DEFAULT, 2), (CONFIG_ONE_DEVICE, 1)])
-def test_initialization(motor_map, config, number_of_plots):
- config_load = config
- motor_map.on_config_update(config_load)
- assert isinstance(motor_map, MotorMap)
- assert motor_map.client is not None
- assert motor_map.config == config_load
- assert len(motor_map.plot_data) == number_of_plots
-
-
-def test_motor_movement_updates_position_and_database(motor_map):
- motor_map.on_config_update(CONFIG_DEFAULT)
-
- # Initial positions
- initial_position_samx = 2.0
- initial_position_samy = 3.0
-
- # Set initial positions in the mocked database
- motor_map.database["samx"]["samx"] = [initial_position_samx]
- motor_map.database["samy"]["samy"] = [initial_position_samy]
-
- # Simulate motor movement for 'samx' only
- new_position_samx = 4.0
- motor_map.on_device_readback({"signals": {"samx": {"value": new_position_samx}}})
-
- # Verify database update for 'samx'
- assert motor_map.database["samx"]["samx"] == [initial_position_samx, new_position_samx]
-
- # Verify 'samy' retains its last known position
- assert motor_map.database["samy"]["samy"] == [initial_position_samy, initial_position_samy]
-
-
-def test_scatter_plot_rendering(motor_map):
- motor_map.on_config_update(CONFIG_DEFAULT)
- # Set initial positions
- initial_position_samx = 2.0
- initial_position_samy = 3.0
- motor_map.database["samx"]["samx"] = [initial_position_samx]
- motor_map.database["samy"]["samy"] = [initial_position_samy]
-
- # Simulate motor movement for 'samx' only
- new_position_samx = 4.0
- motor_map.on_device_readback({"signals": {"samx": {"value": new_position_samx}}})
- motor_map._update_plots()
-
- # Get the scatter plot item
- plot_name = "Motor Map" # Update as per your actual plot name
- scatter_plot_item = motor_map.curves_data[plot_name]["pos"]
-
- # Check the scatter plot item properties
- assert len(scatter_plot_item.data) > 0, "Scatter plot data is empty"
- x_data = scatter_plot_item.data["x"]
- y_data = scatter_plot_item.data["y"]
- assert x_data[-1] == new_position_samx, "Scatter plot X data not updated correctly"
- assert (
- y_data[-1] == initial_position_samy
- ), "Scatter plot Y data should retain last known position"
-
-
-def test_plot_visualization_consistency(motor_map):
- motor_map.on_config_update(CONFIG_DEFAULT)
- # Simulate updating the plot with new data
- motor_map.on_device_readback({"signals": {"samx": {"value": 5}}})
- motor_map.on_device_readback({"signals": {"samy": {"value": 9}}})
- motor_map._update_plots()
-
- plot_name = "Motor Map"
- scatter_plot_item = motor_map.curves_data[plot_name]["pos"]
-
- # Check if the scatter plot reflects the new data correctly
- assert (
- scatter_plot_item.data["x"][-1] == 5 and scatter_plot_item.data["y"][-1] == 9
- ), "Plot not updated correctly with new data"
diff --git a/tests/unit_tests/test_validator_errors.py b/tests/unit_tests/test_validator_errors.py
deleted file mode 100644
index a0cce866..00000000
--- a/tests/unit_tests/test_validator_errors.py
+++ /dev/null
@@ -1,110 +0,0 @@
-# pylint: disable = no-name-in-module,missing-class-docstring, missing-module-docstring
-import pytest
-from pydantic import ValidationError
-
-from bec_widgets.validation.monitor_config_validator import (
- AxisSignal,
- MonitorConfigValidator,
- PlotConfig,
- Signal,
-)
-
-from .test_bec_monitor import mocked_client
-
-
-@pytest.fixture(scope="function")
-def setup_devices(mocked_client):
- MonitorConfigValidator.devices = mocked_client.device_manager.devices
-
-
-def test_signal_validation_name_missing(setup_devices):
- with pytest.raises(ValidationError) as excinfo:
- Signal(name=None)
- errors = excinfo.value.errors()
- assert len(errors) == 1
- assert errors[0]["type"] == "no_device_name"
- assert "Device name must be provided" in str(excinfo.value)
-
-
-def test_signal_validation_name_not_in_bec(setup_devices):
- with pytest.raises(ValidationError) as excinfo:
- Signal(name="non_existent_device")
- errors = excinfo.value.errors()
- assert len(errors) == 1
- assert errors[0]["type"] == "no_device_bec"
- assert 'Device "non_existent_device" not found in current BEC session' in str(excinfo.value)
-
-
-def test_signal_validation_entry_not_in_device(setup_devices):
- with pytest.raises(ValidationError) as excinfo:
- Signal(name="samx", entry="non_existent_entry")
-
- errors = excinfo.value.errors()
- assert len(errors) == 1
- assert errors[0]["type"] == "no_entry_for_device"
- assert 'Entry "non_existent_entry" not found in device "samx" signals' in errors[0]["msg"]
-
-
-def test_signal_validation_success(setup_devices):
- signal = Signal(name="samx")
- assert signal.name == "samx"
-
-
-def test_plot_config_x_axis_signal_validation(setup_devices):
- # Setup a valid signal
- valid_signal = Signal(name="samx")
-
- with pytest.raises(ValidationError) as excinfo:
- AxisSignal(x=[valid_signal, valid_signal], y=[valid_signal, valid_signal])
-
- errors = excinfo.value.errors()
- assert len(errors) == 1
- assert errors[0]["type"] == "x_axis_multiple_signals"
- assert "There must be exactly one signal for x axis" in errors[0]["msg"]
-
-
-def test_plot_config_unsupported_source_type(setup_devices):
- with pytest.raises(ValidationError) as excinfo:
- PlotConfig(sources=[{"type": "unsupported_type", "signals": {}}])
-
- errors = excinfo.value.errors()
- print(errors)
- assert len(errors) == 1
- assert errors[0]["type"] == "literal_error"
-
-
-def test_plot_config_no_source_type_provided(setup_devices):
- with pytest.raises(ValidationError) as excinfo:
- PlotConfig(sources=[{"signals": {}}])
-
- errors = excinfo.value.errors()
- assert len(errors) == 1
- assert errors[0]["type"] == "missing"
-
-
-def test_plot_config_history_source_type(setup_devices):
- history_source = {
- "type": "history",
- "scan_id": "valid_scan_id",
- "signals": {"x": [{"name": "samx"}], "y": [{"name": "samx"}]},
- }
-
- plot_config = PlotConfig(sources=[history_source])
-
- assert len(plot_config.sources) == 1
- assert plot_config.sources[0].type == "history"
- assert plot_config.sources[0].scan_id == "valid_scan_id"
-
-
-def test_plot_config_redis_source_type(setup_devices):
- history_source = {
- "type": "redis",
- "endpoint": "valid_endpoint",
- "update": "append",
- "signals": {"x": [{"name": "samx"}], "y": [{"name": "samx"}]},
- }
-
- plot_config = PlotConfig(sources=[history_source])
-
- assert len(plot_config.sources) == 1
- assert plot_config.sources[0].type == "redis"