diff --git a/bec_widgets/validation/monitor_config_validator.py b/bec_widgets/validation/monitor_config_validator.py index 20527087..3b73788b 100644 --- a/bec_widgets/validation/monitor_config_validator.py +++ b/bec_widgets/validation/monitor_config_validator.py @@ -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. diff --git a/bec_widgets/widgets/monitor/monitor.py b/bec_widgets/widgets/monitor/monitor.py index ab2c06a1..873626d2 100644 --- a/bec_widgets/widgets/monitor/monitor.py +++ b/bec_widgets/widgets/monitor/monitor.py @@ -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": "", - # "signals": { - # "y": [{"name": "bpm4i", "entry": "bpm4i_history_entry"}] - # } - # }, - # { - # "type": "redis", - # "endpoint": "endpoint1", - # "signals": { - # "y": [{"name": "bpm4i", "entry": "bpm4i_redis_entry"}] - # } - # } + { + "type": "history", + "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())