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

refactor: configs for BECMonitor are validated by pydantic outside the main widget

Browse Source
This commit is contained in:
wyzula-jan
2023-11-11 00:24:02 +01:00
parent 92a5325aad
commit 37278e363c
5 changed files with 333 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
bec_widgets/examples/modular_app/modular_app.py
View File

@ -192,7 +192,7 @@ class ModularApp(QMainWindow):
# hook plots, configs and buttons together
for plot, config, button in zip(plots, configs, buttons):
plot.update_config(config)
plot.on_config_update(config)
button.clicked.connect(plot.show_config_dialog)

2
bec_widgets/validation/__init__.py
View File

@ -0,0 +1,2 @@
# from .monitor_config import validate_monitor_config, ValidationError
from .monitor_config_validator import MonitorConfigValidator

106
bec_widgets/validation/monitor_config.py
View File

@ -1,106 +0,0 @@
from pydantic import BaseModel, Field, ValidationError
from typing import List, Dict, Union, Optional
class Signal(BaseModel):
"""
Represents a signal in a plot configuration.
Attributes:
name (str): The name of the signal.
entry (Optional[str]): The entry point of the signal, optional.
"""
name: str
entry: Optional[str]
class PlotAxis(BaseModel):
"""
Represents an axis (X or Y) in a plot configuration.
Attributes:
label (Optional[str]): The label for the axis.
signals (List[Signal]): A list of signals to be plotted on this axis.
"""
label: Optional[str]
signals: List[Signal]
class PlotConfig(BaseModel):
"""
Configuration for a single plot.
Attributes:
plot_name (Optional[str]): Name of the plot.
x (PlotAxis): Configuration for the X axis.
y (PlotAxis): Configuration for the Y axis.
"""
plot_name: Optional[str]
x: PlotAxis
y: PlotAxis
class PlotSettings(BaseModel):
"""
Global settings for plotting.
Attributes:
background_color (str): Color of the plot background.
num_columns (int): Number of columns in the plot layout.
colormap (str): Colormap to be used.
scan_types (bool): Indicates if the configuration is for different scan types.
"""
background_color: str
num_columns: int
colormap: str
scan_types: bool
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]]
def validate_config(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.
"""
if config_data["plot_settings"]["scan_types"]:
return ScanModeConfig(**config_data)
else:
return DeviceMonitorConfig(**config_data)

190
bec_widgets/validation/monitor_config_validator.py Normal file
View File

@ -0,0 +1,190 @@
from typing import List, Dict, Union, Optional
from pydantic import (
BaseModel,
Field,
field_validator,
)
from pydantic_core import PydanticCustomError
class Signal(BaseModel):
"""
Represents a signal in a plot configuration.
Attributes:
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)
@field_validator("name")
@classmethod
def validate_name(cls, v):
device_manager = MonitorConfigValidator.device_manager
# Check if device name provided
if v is None:
raise PydanticCustomError(
"no_device_name",
"Device name must be provided",
dict(wrong_value=v),
)
# Check if device exists in BEC
try:
device = getattr(device_manager, v)
except:
raise PydanticCustomError(
"no_device_bec",
'Device "{wrong_value}" not found in current BEC session',
dict(wrong_value=v),
)
# Check if device have signals #TODO not sure if devices can be defined also without signals
if not hasattr(device, "signals"):
raise PydanticCustomError(
"no_device_signals",
'Device "{wrong_value}" do not have "signals" defined. Check device configuration.',
dict(wrong_value=v),
)
return v
@field_validator("entry")
@classmethod
def set_and_validate_entry(cls, v, values):
device_manager = MonitorConfigValidator.device_manager
device_name = values.data.get("name")
device = getattr(device_manager, device_name, None)
# Set entry based on hints if not provided
if v is None and hasattr(device, "_hints"):
v = next(iter(device._hints), device_name)
elif v is None:
v = device_name
# Validate that the entry exists in device signals
if v not in device.signals:
raise PydanticCustomError(
"no_entry_for_device",
"Entry '{wrong_value}' not found in device '{device_name}' signals",
dict(wrong_value=v, device_name=device_name),
)
return v
class PlotAxis(BaseModel):
"""
Represents an axis (X or Y) in a plot configuration.
Attributes:
label (Optional[str]): The label for the axis.
signals (List[Signal]): A list of signals to be plotted on this axis.
"""
label: Optional[str]
signals: List[Signal] = Field(default_factory=list)
class PlotConfig(BaseModel):
"""
Configuration for a single plot.
Attributes:
plot_name (Optional[str]): Name of the plot.
x (PlotAxis): Configuration for the X axis.
y (PlotAxis): Configuration for the Y axis.
"""
plot_name: Optional[str]
x: PlotAxis = Field(...)
y: PlotAxis = Field(...)
@field_validator("x")
def validate_x_signals(cls, v):
if len(v.signals) != 1:
raise PydanticCustomError(
"no_entry_for_device",
"There must be exactly one signal for x axis. Number of x signals: '{wrong_value}'",
dict(wrong_value=v),
)
return v
class PlotSettings(BaseModel):
"""
Global settings for plotting.
Attributes:
background_color (str): Color of the plot background.
num_columns (int): Number of columns in the plot layout.
colormap (str): Colormap to be used.
scan_types (bool): Indicates if the configuration is for different scan types.
"""
background_color: str
num_columns: int
colormap: str
scan_types: bool
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:
device_manager = None
def __init__(self, device_manager):
# self.device_manager = device_manager
MonitorConfigValidator.device_manager = device_manager
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.
"""
if config_data["plot_settings"]["scan_types"]:
validated_config = ScanModeConfig(**config_data)
else:
validated_config = DeviceMonitorConfig(**config_data)
return validated_config

148
bec_widgets/widgets/monitor/monitor.py
View File

@ -2,18 +2,134 @@ import os
import time
import pyqtgraph as pg
from bec_lib import MessageEndpoints
from pydantic import ValidationError
from bec_lib.core import MessageEndpoints
from PyQt5 import QtCore
from PyQt5.QtCore import pyqtSignal, pyqtSlot
from PyQt5.QtWidgets import QApplication, QTableWidgetItem, QWidget
from PyQt5.QtWidgets import QApplication, QTableWidgetItem, QWidget, QMessageBox
from pyqtgraph import mkPen, mkBrush
from PyQt5 import uic
from bec_widgets.bec_dispatcher import bec_dispatcher
from bec_widgets.qt_utils import Crosshair, Colors
# just for demonstration purposes is script run directly
# from bec_widgets.validation import validate_monitor_config, ValidationError
from bec_widgets.validation import MonitorConfigValidator
# 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", "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": "samx", "entry": "samx"}]},
"y": {
"label": "BPM",
"signals": [{"name": "gauss_bpm", "entry": "gauss_bpm"}],
},
},
{
"plot_name": "Grid plot 4",
"x": {"label": "Motor Y", "signals": [{"name": "samx", "entry": "samx"}]},
"y": {
"label": "BPM",
"signals": [{"name": "gauss_adc3", "entry": "gauss_adc3"}],
},
},
],
"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"},
{"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"},
],
},
},
],
},
}
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 Y",
# "signals": [{"name": "samx", "entry": "samx"}],
"signals": [{"name": "samy"}],
},
"y": {
"label": "bpm4i",
"signals": [{"name": "bpm4i", "entry": "bpm4i"}],
},
},
{
"plot_name": "Gauss plots vs samx",
"x": {
"label": "Motor X",
"signals": [{"name": "samx", "entry": "samx"}],
},
"y": {
"label": "Gauss",
# "signals": [{"name": "gauss_bpm", "entry": "gauss_bpm"}],
"signals": [{"name": "gauss_bpm"}],
},
},
],
}
config_wrong = {
"plot_settings": {
"background_color": "black",
"num_columns": 2,
@ -40,7 +156,7 @@ config_simple = {
},
"y": {
"label": "Gauss",
"signals": [{"name": "gauss_bpm", "entry": "gauss_bpm"}],
"signals": [{"name": "gauss_bpm", "entry": "BS"}],
},
},
],
@ -63,6 +179,9 @@ class BECMonitor(pg.GraphicsLayoutWidget):
# Client and device manager from BEC
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)
if gui_id is None:
self.gui_id = self.__class__.__name__ + str(time.time()) # TODO still in discussion
@ -248,7 +367,9 @@ class BECMonitor(pg.GraphicsLayoutWidget):
)
x_signal_config = x_config["signals"][0]
x_name = x_signal_config.get("name", "")
x_entry = x_signal_config.get("entry", x_name)
x_entry = x_signal_config.get(
"entry", x_name
) # TODO this is buggy if the entry is not specified in the config for x
key = (x_name, x_entry, y_name, y_entry)
data_x = self.data.get(key, {}).get("x", [])
@ -265,7 +386,7 @@ class BECMonitor(pg.GraphicsLayoutWidget):
from .config_dialog import ConfigDialog
dialog = ConfigDialog(default_config=self.config)
dialog.config_updated.connect(self.update_config)
dialog.config_updated.connect(self.on_config_update)
dialog.show()
def update_client(self, client) -> None:
@ -279,12 +400,21 @@ class BECMonitor(pg.GraphicsLayoutWidget):
@pyqtSlot(dict)
def on_config_update(self, config: dict) -> None:
"""
Update the configuration settings for the PlotApp.
Validate and update the configuration settings for the PlotApp.
Args:
config(dict): Configuration settings
"""
self.config = config
# self.config = config
# self._init_config()
try:
validated_config = self.validator.validate_monitor_config(config)
self.config = validated_config.model_dump()
self._init_config()
except ValidationError as e:
error_message = f"Monitor configuration validation error: {e}"
print(error_message)
# QMessageBox.critical(self, "Configuration Error", error_message) #TODO do better error popups
@pyqtSlot(dict, dict)
def on_scan_segment(self, msg, metadata):
@ -402,6 +532,6 @@ if __name__ == "__main__": # pragma: no cover
client.start()
app = QApplication(sys.argv)
monitor = BECMonitor(config=config_simple)
monitor = BECMonitor(config=config_wrong)
monitor.show()
sys.exit(app.exec_())