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

fix: motor_config_validation changed to new monitor config structure

Browse Source
This commit is contained in:
wyzula-jan
2023-12-14 14:01:28 +01:00
parent c3f2ad45c3
commit d67bdd2616
2 changed files with 129 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

120
bec_widgets/validation/monitor_config_validator.py
View File

@ -1,6 +1,6 @@
from typing import Optional, Union
from typing import Optional, Union, Literal
from pydantic import BaseModel, Field, field_validator, model_validator
from pydantic import BaseModel, Field, field_validator, model_validator, ValidationError
from pydantic_core import PydanticCustomError
@ -68,37 +68,22 @@ class Signal(BaseModel):
return values
class PlotAxis(BaseModel):
class AxisSignal(BaseModel):
"""
Represents an axis (X or Y) in a plot configuration.
Configuration signal axis for a single plot.
Attributes:
label (Optional[str]): The label for the axis.
signals (list[Signal]): A list of signals to be plotted on this axis.
x (list): Signal for the X axis.
y (list): Signals for the Y 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(...)
x: list[Signal] = Field(default_factory=list)
y: list[Signal] = Field(default_factory=list)
@field_validator("x")
@classmethod
def validate_x_signals(cls, v):
if len(v.signals) != 1:
"""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}"',
@ -108,9 +93,92 @@ class PlotConfig(BaseModel):
return v
class SourceHistoryValidator(BaseModel):
"""History source validator
Attributes:
type (str): type of source - history
scanID (str): Scan ID for history source.
signals (list): Signal for the source.
"""
type: Literal["history"]
scanID: str # TODO can be validated if it is a valid scanID
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 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)
scanID (Optional[str]): Scan ID for history source.
signals (Optional[AxisSignal]): Signal for the source.
"""
type: Literal["scan_segment", "history"]
scanID: Optional[str] = None
signals: Optional[AxisSignal] = 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]
x_label: Optional[str]
y_label: Optional[str]
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:
Source(**source)
source_type = source.get("type", None)
# Check if source type provided
if source_type is None:
raise PydanticCustomError(
"no_source_type", "Source type must be provided", {"wrong_value": source}
)
# Check if source type is supported
if source_type == "scan_segment":
validated_sources.append(SourceSegmentValidator(**source))
elif source_type == "history":
validated_sources.append(SourceHistoryValidator(**source))
else:
raise PydanticCustomError(
"unsupported_source_type",
"Unsupported source type: '{wrong_value}'",
{"wrong_value": source_type},
)
return validated_sources
class PlotSettings(BaseModel):
"""
Global settings for plotting.
Global settings for plotting affecting mostly visuals.
Attributes:
background_color (str): Color of the plot background.

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

@ -98,7 +98,6 @@ CONFIG_SIMPLE = {
"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"}]},
@ -108,7 +107,6 @@ CONFIG_SIMPLE = {
"x": {"label": "Motor X", "signals": [{"name": "samx", "entry": "samx"}]},
"y": {
"label": "Gauss",
# "signals": [{"name": "gauss_bpm", "entry": "gauss_bpm"}],
"signals": [{"name": "gauss_bpm"}, {"name": "samy", "entry": "samy"}],
},
},
@ -202,20 +200,14 @@ CONFIG_SOURCE = {
"y": [{"name": "bpm4i", "entry": "bpm4i"}],
},
},
# {
# "type": "history",
# "scanID": "<scanID>",
# "signals": {
# "y": [{"name": "bpm4i", "entry": "bpm4i_history_entry"}]
# }
# },
# {
# "type": "redis",
# "endpoint": "endpoint1",
# "signals": {
# "y": [{"name": "bpm4i", "entry": "bpm4i_redis_entry"}]
# }
# }
{
"type": "history",
"scanID": "<scanID>",
"signals": {
"x": [{"name": "samy"}],
"y": [{"name": "bpm4i", "entry": "bpm4i"}],
},
},
],
},
{
@ -464,6 +456,7 @@ class BECMonitor(pg.GraphicsLayoutWidget):
plot.clear()
for source in plot_config["sources"]:
source_type = source["type"][0]
y_signals = source["signals"].get("y", [])
colors_ys = Colors.golden_angle_color(
colormap=self.plot_settings["colormap"], num=len(y_signals)
@ -474,20 +467,8 @@ class BECMonitor(pg.GraphicsLayoutWidget):
y_name = y_signal["name"]
y_entry = y_signal.get("entry", y_name)
user_color = self.user_colors.get((plot_name, y_name, y_entry), 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)
curve_data = pg.PlotDataItem(
symbolSize=5,
symbolBrush=brush_curve,
pen=pen_curve,
skipFiniteCheck=True,
name=f"{y_name} ({y_entry})",
)
curve_name = f"{y_name} ({y_entry})-{source_type.upper()}"
curve_data = self.create_curve(curve_name, color)
curve_list.append((y_name, y_entry, curve_data))
plot.addItem(curve_data)
row_labels.append(f"{y_name} ({y_entry}) - {plot_name}")
@ -498,6 +479,29 @@ class BECMonitor(pg.GraphicsLayoutWidget):
if self.enable_crosshair is True:
self.hook_crosshair()
def create_curve(self, curve_name, color):
"""
Create
Args:
curve_name:
color:
Returns:
"""
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
@ -723,7 +727,7 @@ if __name__ == "__main__": # pragma: no cover
monitor = BECMonitor(
config=config,
gui_id=args.id,
skip_validation=True,
skip_validation=False,
)
monitor.show()
sys.exit(app.exec())