mirror of
https://github.com/ivan-usov-org/bec.git
synced 2025-04-22 02:20:02 +02:00
feat(scan_server): added support for additional gui config
This commit is contained in:
parent
3108c3d830
commit
c6987b6ec2
@ -96,7 +96,7 @@ def deserialize_dtype(dtype: Any) -> type:
|
|||||||
return ScanItem
|
return ScanItem
|
||||||
|
|
||||||
|
|
||||||
def signature_to_dict(func: Callable, include_class_obj=False) -> dict:
|
def signature_to_dict(func: Callable, include_class_obj=False) -> list[dict]:
|
||||||
"""
|
"""
|
||||||
Convert a function signature to a dictionary.
|
Convert a function signature to a dictionary.
|
||||||
The dictionary can be used to reconstruct the signature using dict_to_signature.
|
The dictionary can be used to reconstruct the signature using dict_to_signature.
|
||||||
@ -105,7 +105,7 @@ def signature_to_dict(func: Callable, include_class_obj=False) -> dict:
|
|||||||
func (Callable): Function to be converted
|
func (Callable): Function to be converted
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
dict: Dictionary representation of the function signature
|
list[dict]: List of dictionaries representing the function signature
|
||||||
"""
|
"""
|
||||||
out = []
|
out = []
|
||||||
params = inspect.signature(func).parameters
|
params = inspect.signature(func).parameters
|
||||||
|
239
bec_server/bec_server/scan_server/scan_gui_models.py
Normal file
239
bec_server/bec_server/scan_server/scan_gui_models.py
Normal file
@ -0,0 +1,239 @@
|
|||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import re
|
||||||
|
from contextvars import ContextVar
|
||||||
|
from typing import Any, Literal, Optional, Type
|
||||||
|
|
||||||
|
from pydantic import BaseModel, Field, field_validator
|
||||||
|
from pydantic_core import PydanticCustomError
|
||||||
|
|
||||||
|
from bec_lib.signature_serializer import signature_to_dict
|
||||||
|
from bec_server.scan_server.scans import ScanBase
|
||||||
|
|
||||||
|
context_signature = ContextVar("context_signature")
|
||||||
|
context_docstring = ContextVar("context_docstring")
|
||||||
|
|
||||||
|
|
||||||
|
class GUIInput(BaseModel):
|
||||||
|
"""
|
||||||
|
InputConfig is a data model for the input configuration of a scan.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
name (str): name of the input, has to be same as in the function signature.
|
||||||
|
"""
|
||||||
|
|
||||||
|
arg: bool = Field(False)
|
||||||
|
name: str = Field(None, validate_default=True)
|
||||||
|
type: Optional[
|
||||||
|
Literal["DeviceBase", "device", "float", "int", "bool", "str", "list", "dict"]
|
||||||
|
] = Field(None, validate_default=True)
|
||||||
|
display_name: Optional[str] = Field(None, validate_default=True)
|
||||||
|
tooltip: Optional[str] = Field(None, validate_default=True)
|
||||||
|
default: Optional[Any] = Field(None, validate_default=True)
|
||||||
|
expert: Optional[bool] = Field(False) # TODO decide later how to implement
|
||||||
|
|
||||||
|
@field_validator("name")
|
||||||
|
@classmethod
|
||||||
|
def validate_name(cls, v, values):
|
||||||
|
# args cannot be validated with the current implementation of signature of scans
|
||||||
|
if values.data["arg"]:
|
||||||
|
return v
|
||||||
|
signature = context_signature.get()
|
||||||
|
available_args = [entry["name"] for entry in signature]
|
||||||
|
if v not in available_args:
|
||||||
|
raise PydanticCustomError(
|
||||||
|
"wrong argument name",
|
||||||
|
"The argument name is not available in the function signature",
|
||||||
|
{"wrong_value": v},
|
||||||
|
)
|
||||||
|
return v
|
||||||
|
|
||||||
|
@field_validator("type")
|
||||||
|
@classmethod
|
||||||
|
def validate_field(cls, v, values):
|
||||||
|
# args cannot be validated with the current implementation of signature of scans
|
||||||
|
if values.data["arg"]:
|
||||||
|
return v
|
||||||
|
signature = context_signature.get()
|
||||||
|
if v is None:
|
||||||
|
name = values.data.get("name", None)
|
||||||
|
if name is None:
|
||||||
|
raise PydanticCustomError(
|
||||||
|
"missing argument name",
|
||||||
|
"The argument name is required to infer the type",
|
||||||
|
{"wrong_value": v},
|
||||||
|
)
|
||||||
|
for entry in signature:
|
||||||
|
if entry["name"] == name:
|
||||||
|
v = entry["annotation"]
|
||||||
|
return v
|
||||||
|
|
||||||
|
@field_validator("tooltip")
|
||||||
|
@classmethod
|
||||||
|
def validate_tooltip(cls, v, values):
|
||||||
|
# args cannot be validated with the current implementation of signature of scans
|
||||||
|
if values.data["arg"]:
|
||||||
|
return v
|
||||||
|
if v is not None:
|
||||||
|
return v
|
||||||
|
|
||||||
|
docstring = context_docstring.get()
|
||||||
|
name = values.data.get("name", None)
|
||||||
|
if name is None:
|
||||||
|
raise PydanticCustomError(
|
||||||
|
"missing argument name",
|
||||||
|
"The argument name is required to infer the tooltip",
|
||||||
|
{"wrong_value": v},
|
||||||
|
)
|
||||||
|
|
||||||
|
try:
|
||||||
|
args_part = docstring.split("Args:")[1].split("Returns:")[0]
|
||||||
|
except IndexError:
|
||||||
|
return None
|
||||||
|
|
||||||
|
pattern = re.compile(r"\s*" + re.escape(name) + r" \(([^)]+)\): (.+?)(?:\.|\n|$)")
|
||||||
|
match = pattern.search(args_part)
|
||||||
|
|
||||||
|
if match:
|
||||||
|
description = match.group(2)
|
||||||
|
first_sentence = description.split(".")[0].strip()
|
||||||
|
if first_sentence:
|
||||||
|
v = first_sentence[0].upper() + first_sentence[1:]
|
||||||
|
return v
|
||||||
|
return None
|
||||||
|
|
||||||
|
@field_validator("display_name")
|
||||||
|
@classmethod
|
||||||
|
def validate_display_name(cls, v, values):
|
||||||
|
if v is not None:
|
||||||
|
return v
|
||||||
|
name = values.data.get("name", None)
|
||||||
|
if name is None:
|
||||||
|
raise PydanticCustomError(
|
||||||
|
"missing argument name",
|
||||||
|
"The argument name is required to infer the display name",
|
||||||
|
{"wrong_value": v},
|
||||||
|
)
|
||||||
|
parts = re.split("(_|\d+)", name)
|
||||||
|
formatted_parts = []
|
||||||
|
for part in parts:
|
||||||
|
if part.isdigit():
|
||||||
|
formatted_parts.append("" + part)
|
||||||
|
elif part.isalpha():
|
||||||
|
formatted_parts.append(part.capitalize())
|
||||||
|
v = " ".join(formatted_parts).strip()
|
||||||
|
return v
|
||||||
|
|
||||||
|
@field_validator("default")
|
||||||
|
@classmethod
|
||||||
|
def validate_default(cls, v, values):
|
||||||
|
# args cannot be validated with the current implementation of signature of scans
|
||||||
|
if values.data["arg"]:
|
||||||
|
return v
|
||||||
|
if v is not None:
|
||||||
|
return v
|
||||||
|
signature = context_signature.get()
|
||||||
|
name = values.data.get("name", None)
|
||||||
|
if name is None:
|
||||||
|
raise PydanticCustomError(
|
||||||
|
"missing argument name",
|
||||||
|
"The argument name is required to infer the type",
|
||||||
|
{"wrong_value": v},
|
||||||
|
)
|
||||||
|
for entry in signature:
|
||||||
|
if entry["name"] == name:
|
||||||
|
v = entry["default"]
|
||||||
|
return v
|
||||||
|
|
||||||
|
|
||||||
|
class GUIGroup(BaseModel):
|
||||||
|
"""
|
||||||
|
GUIGroup is a data model for the GUI group configuration of a scan.
|
||||||
|
"""
|
||||||
|
|
||||||
|
name: str
|
||||||
|
inputs: list[GUIInput]
|
||||||
|
|
||||||
|
|
||||||
|
class GUIArgGroup(BaseModel):
|
||||||
|
"""
|
||||||
|
GUIArgGroup is a data model for the GUI group configuration of a scan.
|
||||||
|
"""
|
||||||
|
|
||||||
|
name: str = "Scan Arguments"
|
||||||
|
bundle: int = Field(None)
|
||||||
|
arg_inputs: dict
|
||||||
|
inputs: Optional[list[GUIInput]] = Field(None, validate_default=True)
|
||||||
|
min: Optional[int] = Field(None)
|
||||||
|
max: Optional[int] = Field(None)
|
||||||
|
|
||||||
|
@field_validator("inputs")
|
||||||
|
@classmethod
|
||||||
|
def validate_inputs(cls, v, values):
|
||||||
|
if v is not None:
|
||||||
|
return v
|
||||||
|
arg_inputs = values.data["arg_inputs"]
|
||||||
|
arg_inputs_str = {key: value.value for key, value in arg_inputs.items()}
|
||||||
|
v = []
|
||||||
|
for name, type_ in arg_inputs_str.items():
|
||||||
|
v.append(GUIInput(name=name, type=type_, arg=True))
|
||||||
|
return v
|
||||||
|
|
||||||
|
|
||||||
|
class GUIConfig(BaseModel):
|
||||||
|
"""
|
||||||
|
GUIConfig is a data model for the GUI configuration of a scan.
|
||||||
|
"""
|
||||||
|
|
||||||
|
scan_class_name: str
|
||||||
|
arg_group: Optional[GUIArgGroup] = Field(None)
|
||||||
|
kwarg_groups: list[GUIGroup] = Field(None)
|
||||||
|
signature: list[dict] = Field(..., exclude=True)
|
||||||
|
docstring: str = Field(..., exclude=True)
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def from_dict(cls, scan_cls: Type[ScanBase]) -> GUIConfig:
|
||||||
|
"""
|
||||||
|
Create a GUIConfig object from a scan class.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
scan_cls(Type[ScanBase]): scan class
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
GUIConfig: GUIConfig object
|
||||||
|
"""
|
||||||
|
|
||||||
|
groups = []
|
||||||
|
config = scan_cls.gui_config
|
||||||
|
signature = signature_to_dict(scan_cls.__init__)
|
||||||
|
signature_token = context_signature.set(signature)
|
||||||
|
docstring = scan_cls.__doc__ or scan_cls.__init__.__doc__
|
||||||
|
docstring_token = context_docstring.set(docstring)
|
||||||
|
# kwargs from gui config
|
||||||
|
for group_name, input_names in config.items():
|
||||||
|
inputs = [GUIInput(name=name, arg=False) for name in input_names]
|
||||||
|
group = GUIGroup(name=group_name, inputs=inputs)
|
||||||
|
groups.append(group)
|
||||||
|
# args from arg_input if available
|
||||||
|
arg_group = None
|
||||||
|
if hasattr(scan_cls, "arg_input"):
|
||||||
|
arg_input = scan_cls.arg_input
|
||||||
|
if hasattr(scan_cls, "arg_bundle_size"):
|
||||||
|
arg_group = GUIArgGroup(
|
||||||
|
bundle=scan_cls.arg_bundle_size.get("bundle"),
|
||||||
|
min=scan_cls.arg_bundle_size.get("min"),
|
||||||
|
max=scan_cls.arg_bundle_size.get("max"),
|
||||||
|
arg_inputs=arg_input,
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
arg_group = GUIArgGroup(inputs=scan_cls.arg_input)
|
||||||
|
|
||||||
|
context_signature.reset(signature_token)
|
||||||
|
context_docstring.reset(docstring_token)
|
||||||
|
return cls(
|
||||||
|
scan_class_name=scan_cls.__name__,
|
||||||
|
signature=signature,
|
||||||
|
docstring=docstring,
|
||||||
|
kwarg_groups=groups,
|
||||||
|
arg_group=arg_group,
|
||||||
|
)
|
@ -10,6 +10,7 @@ from bec_lib.endpoints import MessageEndpoints
|
|||||||
from bec_lib.logger import bec_logger
|
from bec_lib.logger import bec_logger
|
||||||
from bec_lib.messages import AvailableResourceMessage
|
from bec_lib.messages import AvailableResourceMessage
|
||||||
from bec_lib.signature_serializer import signature_to_dict
|
from bec_lib.signature_serializer import signature_to_dict
|
||||||
|
from bec_server.scan_server.scan_gui_models import GUIConfig
|
||||||
|
|
||||||
from . import scans as ScanServerScans
|
from . import scans as ScanServerScans
|
||||||
|
|
||||||
@ -79,16 +80,43 @@ class ScanManager:
|
|||||||
if issubclass(scan_cls, report_cls):
|
if issubclass(scan_cls, report_cls):
|
||||||
base_cls = report_cls.__name__
|
base_cls = report_cls.__name__
|
||||||
self.scan_dict[scan_cls.__name__] = scan_cls
|
self.scan_dict[scan_cls.__name__] = scan_cls
|
||||||
|
gui_config = self.validate_gui_config(scan_cls)
|
||||||
self.available_scans[scan_cls.scan_name] = {
|
self.available_scans[scan_cls.scan_name] = {
|
||||||
"class": scan_cls.__name__,
|
"class": scan_cls.__name__,
|
||||||
"base_class": base_cls,
|
"base_class": base_cls,
|
||||||
"arg_input": self.convert_arg_input(scan_cls.arg_input),
|
"arg_input": self.convert_arg_input(scan_cls.arg_input),
|
||||||
|
"gui_config": gui_config,
|
||||||
"required_kwargs": scan_cls.required_kwargs,
|
"required_kwargs": scan_cls.required_kwargs,
|
||||||
"arg_bundle_size": scan_cls.arg_bundle_size,
|
"arg_bundle_size": scan_cls.arg_bundle_size,
|
||||||
"doc": scan_cls.__doc__ or scan_cls.__init__.__doc__,
|
"doc": scan_cls.__doc__ or scan_cls.__init__.__doc__,
|
||||||
"signature": signature_to_dict(scan_cls.__init__),
|
"signature": signature_to_dict(scan_cls.__init__),
|
||||||
}
|
}
|
||||||
|
|
||||||
|
def validate_gui_config(self, scan_cls) -> dict:
|
||||||
|
"""
|
||||||
|
Validate the gui_config of the scan class
|
||||||
|
|
||||||
|
Args:
|
||||||
|
scan_cls: class
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: gui_config
|
||||||
|
"""
|
||||||
|
|
||||||
|
if not hasattr(scan_cls, "gui_config"):
|
||||||
|
return {}
|
||||||
|
if not isinstance(scan_cls.gui_config, GUIConfig) and not isinstance(
|
||||||
|
scan_cls.gui_config, dict
|
||||||
|
):
|
||||||
|
logger.error(
|
||||||
|
f"Invalid gui_config for {scan_cls.scan_name}. gui_config must be of type GUIConfig or dict."
|
||||||
|
)
|
||||||
|
return {}
|
||||||
|
gui_config = scan_cls.gui_config
|
||||||
|
if isinstance(scan_cls.gui_config, dict):
|
||||||
|
gui_config = GUIConfig.from_dict(scan_cls)
|
||||||
|
return gui_config.model_dump()
|
||||||
|
|
||||||
def convert_arg_input(self, arg_input) -> dict:
|
def convert_arg_input(self, arg_input) -> dict:
|
||||||
"""
|
"""
|
||||||
Convert the arg_input to supported data types
|
Convert the arg_input to supported data types
|
||||||
|
@ -1,3 +1,5 @@
|
|||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
import ast
|
import ast
|
||||||
import enum
|
import enum
|
||||||
import threading
|
import threading
|
||||||
@ -184,6 +186,7 @@ class RequestBase(ABC):
|
|||||||
scan_name = ""
|
scan_name = ""
|
||||||
arg_input = {}
|
arg_input = {}
|
||||||
arg_bundle_size = {"bundle": len(arg_input), "min": None, "max": None}
|
arg_bundle_size = {"bundle": len(arg_input), "min": None, "max": None}
|
||||||
|
gui_args = {}
|
||||||
required_kwargs = []
|
required_kwargs = []
|
||||||
return_to_start_after_abort = False
|
return_to_start_after_abort = False
|
||||||
use_scan_progress_report = False
|
use_scan_progress_report = False
|
||||||
@ -779,6 +782,9 @@ class Scan(ScanBase):
|
|||||||
}
|
}
|
||||||
arg_bundle_size = {"bundle": len(arg_input), "min": 2, "max": None}
|
arg_bundle_size = {"bundle": len(arg_input), "min": 2, "max": None}
|
||||||
required_kwargs = ["relative"]
|
required_kwargs = ["relative"]
|
||||||
|
gui_config = {
|
||||||
|
"Scan Parameters": ["exp_time", "settling_time", "burst_at_each_point", "relative"]
|
||||||
|
}
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
@ -827,6 +833,12 @@ class Scan(ScanBase):
|
|||||||
class FermatSpiralScan(ScanBase):
|
class FermatSpiralScan(ScanBase):
|
||||||
scan_name = "fermat_scan"
|
scan_name = "fermat_scan"
|
||||||
required_kwargs = ["step", "relative"]
|
required_kwargs = ["step", "relative"]
|
||||||
|
gui_config = {
|
||||||
|
"Device 1": ["motor1", "start_motor1", "stop_motor1"],
|
||||||
|
"Device 2": ["motor2", "start_motor2", "stop_motor2"],
|
||||||
|
"Movement Parameters": ["step", "relative"],
|
||||||
|
"Acquisition Parameters": ["exp_time", "settling_time", "burst_at_each_point"],
|
||||||
|
}
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
@ -854,6 +866,7 @@ class FermatSpiralScan(ScanBase):
|
|||||||
stop_motor1 (float): end position motor 1
|
stop_motor1 (float): end position motor 1
|
||||||
motor2 (DeviceBase): second motor
|
motor2 (DeviceBase): second motor
|
||||||
start_motor2 (float): start position motor 2
|
start_motor2 (float): start position motor 2
|
||||||
|
stop_motor2 (float): end position motor 2
|
||||||
step (float): step size in motor units. Default is 0.1.
|
step (float): step size in motor units. Default is 0.1.
|
||||||
exp_time (float): exposure time in seconds. Default is 0.
|
exp_time (float): exposure time in seconds. Default is 0.
|
||||||
settling_time (float): settling time in seconds. Default is 0.
|
settling_time (float): settling time in seconds. Default is 0.
|
||||||
@ -901,11 +914,16 @@ class FermatSpiralScan(ScanBase):
|
|||||||
class RoundScan(ScanBase):
|
class RoundScan(ScanBase):
|
||||||
scan_name = "round_scan"
|
scan_name = "round_scan"
|
||||||
required_kwargs = ["relative"]
|
required_kwargs = ["relative"]
|
||||||
|
gui_config = {
|
||||||
|
"Motors": ["motor_1", "motor_2"],
|
||||||
|
"Ring Parameters": ["inner_ring", "outer_ring", "number_of_rings", "pos_in_first_ring"],
|
||||||
|
"Scan Parameters": ["relative", "burst_at_each_point"],
|
||||||
|
}
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
motor_1: DeviceBase,
|
motor_1: DeviceBase,
|
||||||
motor2: DeviceBase,
|
motor_2: DeviceBase,
|
||||||
inner_ring: float,
|
inner_ring: float,
|
||||||
outer_ring: float,
|
outer_ring: float,
|
||||||
number_of_rings: int,
|
number_of_rings: int,
|
||||||
@ -919,7 +937,7 @@ class RoundScan(ScanBase):
|
|||||||
|
|
||||||
Args:
|
Args:
|
||||||
motor_1 (DeviceBase): first motor
|
motor_1 (DeviceBase): first motor
|
||||||
motor2 (DeviceBase): second motor
|
motor_2 (DeviceBase): second motor
|
||||||
inner_ring (float): inner radius
|
inner_ring (float): inner radius
|
||||||
outer_ring (float): outer radius
|
outer_ring (float): outer radius
|
||||||
number_of_rings (int): number of rings
|
number_of_rings (int): number of rings
|
||||||
@ -937,7 +955,7 @@ class RoundScan(ScanBase):
|
|||||||
super().__init__(relative=relative, burst_at_each_point=burst_at_each_point, **kwargs)
|
super().__init__(relative=relative, burst_at_each_point=burst_at_each_point, **kwargs)
|
||||||
self.axis = []
|
self.axis = []
|
||||||
self.motor_1 = motor_1
|
self.motor_1 = motor_1
|
||||||
self.motor_2 = motor2
|
self.motor_2 = motor_2
|
||||||
self.inner_ring = inner_ring
|
self.inner_ring = inner_ring
|
||||||
self.outer_ring = outer_ring
|
self.outer_ring = outer_ring
|
||||||
self.number_of_rings = number_of_rings
|
self.number_of_rings = number_of_rings
|
||||||
@ -960,6 +978,11 @@ class ContLineScan(ScanBase):
|
|||||||
scan_name = "cont_line_scan"
|
scan_name = "cont_line_scan"
|
||||||
required_kwargs = ["steps", "relative"]
|
required_kwargs = ["steps", "relative"]
|
||||||
scan_type = "step"
|
scan_type = "step"
|
||||||
|
gui_config = {
|
||||||
|
"Device": ["device", "start", "stop"],
|
||||||
|
"Movement Parameters": ["steps", "relative", "offset"],
|
||||||
|
"Acquisition Parameters": ["exp_time", "burst_at_each_point"],
|
||||||
|
}
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
@ -1039,6 +1062,7 @@ class ContLineFlyScan(AsyncFlyScanBase):
|
|||||||
scan_name = "cont_line_fly_scan"
|
scan_name = "cont_line_fly_scan"
|
||||||
required_kwargs = []
|
required_kwargs = []
|
||||||
use_scan_progress_report = False
|
use_scan_progress_report = False
|
||||||
|
gui_config = {"Device": ["motor", "start", "stop"], "Scan Parameters": ["exp_time", "relative"]}
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
@ -1122,6 +1146,10 @@ class RoundScanFlySim(SyncFlyScanBase):
|
|||||||
scan_type = "fly"
|
scan_type = "fly"
|
||||||
pre_move = False
|
pre_move = False
|
||||||
required_kwargs = ["relative"]
|
required_kwargs = ["relative"]
|
||||||
|
gui_config = {
|
||||||
|
"Fly Parameters": ["flyer", "relative"],
|
||||||
|
"Ring Parameters": ["inner_ring", "outer_ring", "number_of_rings", "number_pos"],
|
||||||
|
}
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
@ -1130,6 +1158,7 @@ class RoundScanFlySim(SyncFlyScanBase):
|
|||||||
outer_ring: float,
|
outer_ring: float,
|
||||||
number_of_rings: int,
|
number_of_rings: int,
|
||||||
number_pos: int,
|
number_pos: int,
|
||||||
|
relative: bool = False,
|
||||||
**kwargs,
|
**kwargs,
|
||||||
):
|
):
|
||||||
"""
|
"""
|
||||||
@ -1210,6 +1239,12 @@ class RoundScanFlySim(SyncFlyScanBase):
|
|||||||
class RoundROIScan(ScanBase):
|
class RoundROIScan(ScanBase):
|
||||||
scan_name = "round_roi_scan"
|
scan_name = "round_roi_scan"
|
||||||
required_kwargs = ["dr", "nth", "relative"]
|
required_kwargs = ["dr", "nth", "relative"]
|
||||||
|
gui_config = {
|
||||||
|
"Motor 1": ["motor_1", "width_1"],
|
||||||
|
"Motor 2": ["motor_2", "width_2"],
|
||||||
|
"Shell Parametes": ["dr", "nth"],
|
||||||
|
"Acquisition Parameters": ["exp_time", "relative", "burst_at_each_point"],
|
||||||
|
}
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
@ -1295,6 +1330,7 @@ class ListScan(ScanBase):
|
|||||||
class TimeScan(ScanBase):
|
class TimeScan(ScanBase):
|
||||||
scan_name = "time_scan"
|
scan_name = "time_scan"
|
||||||
required_kwargs = ["points", "interval"]
|
required_kwargs = ["points", "interval"]
|
||||||
|
gui_config = {"Scan Parameters": ["points", "interval", "exp_time", "burst_at_each_point"]}
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
@ -1357,6 +1393,7 @@ class MonitorScan(ScanBase):
|
|||||||
scan_name = "monitor_scan"
|
scan_name = "monitor_scan"
|
||||||
required_kwargs = ["relative"]
|
required_kwargs = ["relative"]
|
||||||
scan_type = "fly"
|
scan_type = "fly"
|
||||||
|
gui_config = {"Device": ["device", "start", "stop"], "Scan Parameters": ["relative"]}
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
self, device: DeviceBase, start: float, stop: float, relative: bool = False, **kwargs
|
self, device: DeviceBase, start: float, stop: float, relative: bool = False, **kwargs
|
||||||
@ -1368,6 +1405,7 @@ class MonitorScan(ScanBase):
|
|||||||
device (Device): monitored device
|
device (Device): monitored device
|
||||||
start (float): start position of the monitored device
|
start (float): start position of the monitored device
|
||||||
stop (float): stop position of the monitored device
|
stop (float): stop position of the monitored device
|
||||||
|
relative (bool): if True, the motor will be moved relative to its current position. Default is False.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
ScanReport
|
ScanReport
|
||||||
@ -1441,12 +1479,14 @@ class MonitorScan(ScanBase):
|
|||||||
class Acquire(ScanBase):
|
class Acquire(ScanBase):
|
||||||
scan_name = "acquire"
|
scan_name = "acquire"
|
||||||
required_kwargs = []
|
required_kwargs = []
|
||||||
|
gui_config = {"Scan Parameters": ["exp_time", "burst_at_each_point"]}
|
||||||
|
|
||||||
def __init__(self, *args, exp_time: float = 0, burst_at_each_point: int = 1, **kwargs):
|
def __init__(self, *args, exp_time: float = 0, burst_at_each_point: int = 1, **kwargs):
|
||||||
"""
|
"""
|
||||||
A simple acquisition at the current position.
|
A simple acquisition at the current position.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
|
exp_time (float): exposure time in s
|
||||||
burst: number of acquisition per point
|
burst: number of acquisition per point
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
@ -1503,6 +1543,10 @@ class LineScan(ScanBase):
|
|||||||
"stop": ScanArgType.FLOAT,
|
"stop": ScanArgType.FLOAT,
|
||||||
}
|
}
|
||||||
arg_bundle_size = {"bundle": len(arg_input), "min": 1, "max": None}
|
arg_bundle_size = {"bundle": len(arg_input), "min": 1, "max": None}
|
||||||
|
gui_config = {
|
||||||
|
"Movement Parameters": ["steps", "relative"],
|
||||||
|
"Acquisition Parameters": ["exp_time", "burst_at_each_point"],
|
||||||
|
}
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
|
244
bec_server/tests/tests_scan_server/test_scan_gui_models.py
Normal file
244
bec_server/tests/tests_scan_server/test_scan_gui_models.py
Normal file
@ -0,0 +1,244 @@
|
|||||||
|
import pytest
|
||||||
|
from pydantic import ValidationError
|
||||||
|
|
||||||
|
from bec_server.scan_server.scan_gui_models import GUIConfig
|
||||||
|
from bec_server.scan_server.scans import ScanArgType, ScanBase
|
||||||
|
|
||||||
|
|
||||||
|
class GoodScan(ScanBase): # pragma: no cover
|
||||||
|
scan_name = "good_scan"
|
||||||
|
required_kwargs = ["steps", "relative"]
|
||||||
|
arg_input = {
|
||||||
|
"device": ScanArgType.DEVICE,
|
||||||
|
"start": ScanArgType.FLOAT,
|
||||||
|
"stop": ScanArgType.FLOAT,
|
||||||
|
}
|
||||||
|
arg_bundle_size = {"bundle": len(arg_input), "min": 1, "max": None}
|
||||||
|
gui_config = {"Scan Parameters": ["steps", "exp_time", "relative", "burst_at_each_point"]}
|
||||||
|
|
||||||
|
def __init__(
|
||||||
|
self,
|
||||||
|
*args,
|
||||||
|
exp_time: float = 0,
|
||||||
|
steps: int = None,
|
||||||
|
relative: bool = False,
|
||||||
|
burst_at_each_point: int = 1,
|
||||||
|
**kwargs,
|
||||||
|
):
|
||||||
|
"""
|
||||||
|
A good scan for one or more motors.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
*args (Device, float, float): pairs of device / start position / end position
|
||||||
|
exp_time (float): exposure time in s. Default: 0
|
||||||
|
steps (int): number of steps. Default: 10
|
||||||
|
relative (bool): if True, the start and end positions are relative to the current position. Default: False
|
||||||
|
burst_at_each_point (int): number of acquisition per point. Default: 1
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
ScanReport
|
||||||
|
|
||||||
|
|
||||||
|
"""
|
||||||
|
super().__init__(
|
||||||
|
exp_time=exp_time, relative=relative, burst_at_each_point=burst_at_each_point, **kwargs
|
||||||
|
)
|
||||||
|
self.steps = steps
|
||||||
|
|
||||||
|
def doing_something_good(self):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
class ExtraKwarg(ScanBase): # pragma: no cover
|
||||||
|
scan_name = "wrong_name"
|
||||||
|
required_kwargs = ["steps", "relative"]
|
||||||
|
|
||||||
|
gui_config = {"Device 1": ["motor1", "start_motor1", "stop_motor1"]}
|
||||||
|
|
||||||
|
def __init__(self, motor1: str, stop_motor1: float, **kwargs):
|
||||||
|
"""
|
||||||
|
A scan following Fermat's spiral.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
motor1 (DeviceBase): first motor
|
||||||
|
start_motor1 (float): start position motor 1
|
||||||
|
stop_motor1 (float): end position motor 1
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
ScanReport
|
||||||
|
"""
|
||||||
|
|
||||||
|
print("I am a wrong scan")
|
||||||
|
|
||||||
|
|
||||||
|
class WrongDocs(ScanBase): # pragma: no cover
|
||||||
|
scan_name = "wrong_name"
|
||||||
|
required_kwargs = ["steps", "relative"]
|
||||||
|
|
||||||
|
gui_config = {"Device 1": ["motor1", "start_motor1", "stop_motor1"]}
|
||||||
|
|
||||||
|
def __init__(self, motor1: str, start_motor1: float, stop_motor1: float, **kwargs):
|
||||||
|
"""
|
||||||
|
A scan following Fermat's spiral.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
motor1 (DeviceBase): first motor
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
ScanReport
|
||||||
|
"""
|
||||||
|
|
||||||
|
print("I am a scan with wrong docs.")
|
||||||
|
|
||||||
|
|
||||||
|
def test_gui_config_good_scan_dump():
|
||||||
|
gui_config = GUIConfig.from_dict(GoodScan)
|
||||||
|
expected_config = {
|
||||||
|
"scan_class_name": "GoodScan",
|
||||||
|
"arg_group": {
|
||||||
|
"name": "Scan Arguments",
|
||||||
|
"bundle": 3,
|
||||||
|
"arg_inputs": {
|
||||||
|
"device": ScanArgType.DEVICE,
|
||||||
|
"start": ScanArgType.FLOAT,
|
||||||
|
"stop": ScanArgType.FLOAT,
|
||||||
|
},
|
||||||
|
"inputs": [
|
||||||
|
{
|
||||||
|
"arg": True,
|
||||||
|
"name": "device",
|
||||||
|
"display_name": "Device",
|
||||||
|
"type": "device",
|
||||||
|
"tooltip": None,
|
||||||
|
"default": None,
|
||||||
|
"expert": False,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"arg": True,
|
||||||
|
"name": "start",
|
||||||
|
"display_name": "Start",
|
||||||
|
"type": "float",
|
||||||
|
"tooltip": None,
|
||||||
|
"default": None,
|
||||||
|
"expert": False,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"arg": True,
|
||||||
|
"name": "stop",
|
||||||
|
"display_name": "Stop",
|
||||||
|
"type": "float",
|
||||||
|
"tooltip": None,
|
||||||
|
"default": None,
|
||||||
|
"expert": False,
|
||||||
|
},
|
||||||
|
],
|
||||||
|
"min": 1,
|
||||||
|
"max": None,
|
||||||
|
},
|
||||||
|
"kwarg_groups": [
|
||||||
|
{
|
||||||
|
"name": "Scan Parameters",
|
||||||
|
"inputs": [
|
||||||
|
{
|
||||||
|
"arg": False,
|
||||||
|
"name": "steps",
|
||||||
|
"display_name": "Steps",
|
||||||
|
"type": "int",
|
||||||
|
"tooltip": "Number of steps",
|
||||||
|
"default": None,
|
||||||
|
"expert": False,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"arg": False,
|
||||||
|
"name": "exp_time",
|
||||||
|
"display_name": "Exp Time",
|
||||||
|
"type": "float",
|
||||||
|
"tooltip": "Exposure time in s",
|
||||||
|
"default": 0,
|
||||||
|
"expert": False,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"arg": False,
|
||||||
|
"name": "relative",
|
||||||
|
"display_name": "Relative",
|
||||||
|
"type": "bool",
|
||||||
|
"tooltip": "If True, the start and end positions are relative to the current position",
|
||||||
|
"default": False,
|
||||||
|
"expert": False,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"arg": False,
|
||||||
|
"name": "burst_at_each_point",
|
||||||
|
"display_name": "Burst At Each Point",
|
||||||
|
"type": "int",
|
||||||
|
"tooltip": "Number of acquisition per point",
|
||||||
|
"default": 1,
|
||||||
|
"expert": False,
|
||||||
|
},
|
||||||
|
],
|
||||||
|
}
|
||||||
|
],
|
||||||
|
}
|
||||||
|
assert gui_config.model_dump() == expected_config
|
||||||
|
|
||||||
|
|
||||||
|
def test_gui_config_extra_kwarg():
|
||||||
|
with pytest.raises(ValidationError) as excinfo:
|
||||||
|
GUIConfig.from_dict(ExtraKwarg)
|
||||||
|
errors = excinfo.value.errors()
|
||||||
|
assert len(errors) == 5
|
||||||
|
assert errors[0]["type"] == ("wrong argument name")
|
||||||
|
assert errors[1]["type"] == ("missing argument name")
|
||||||
|
assert errors[2]["type"] == ("missing argument name")
|
||||||
|
assert errors[3]["type"] == ("missing argument name")
|
||||||
|
assert errors[4]["type"] == ("missing argument name")
|
||||||
|
|
||||||
|
|
||||||
|
def test_gui_config_wrong_docs():
|
||||||
|
gui_config = GUIConfig.from_dict(WrongDocs)
|
||||||
|
expected = {
|
||||||
|
"scan_class_name": "WrongDocs",
|
||||||
|
"arg_group": {
|
||||||
|
"name": "Scan Arguments",
|
||||||
|
"bundle": 0,
|
||||||
|
"arg_inputs": {},
|
||||||
|
"inputs": [],
|
||||||
|
"min": None,
|
||||||
|
"max": None,
|
||||||
|
},
|
||||||
|
"kwarg_groups": [
|
||||||
|
{
|
||||||
|
"name": "Device 1",
|
||||||
|
"inputs": [
|
||||||
|
{
|
||||||
|
"arg": False,
|
||||||
|
"name": "motor1",
|
||||||
|
"type": "str",
|
||||||
|
"display_name": "Motor 1",
|
||||||
|
"tooltip": "First motor",
|
||||||
|
"default": "_empty",
|
||||||
|
"expert": False,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"arg": False,
|
||||||
|
"name": "start_motor1",
|
||||||
|
"type": "float",
|
||||||
|
"display_name": "Start Motor 1",
|
||||||
|
"tooltip": None,
|
||||||
|
"default": "_empty",
|
||||||
|
"expert": False,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"arg": False,
|
||||||
|
"name": "stop_motor1",
|
||||||
|
"type": "float",
|
||||||
|
"display_name": "Stop Motor 1",
|
||||||
|
"tooltip": None,
|
||||||
|
"default": "_empty",
|
||||||
|
"expert": False,
|
||||||
|
},
|
||||||
|
],
|
||||||
|
}
|
||||||
|
],
|
||||||
|
}
|
||||||
|
assert gui_config.model_dump() == expected
|
BIN
docs/source/assets/scan_GUI_example.png
Normal file
BIN
docs/source/assets/scan_GUI_example.png
Normal file
Binary file not shown.
After Width: | Height: | Size: 108 KiB |
136
docs/source/developer/scans/scan_gui_config.md
Normal file
136
docs/source/developer/scans/scan_gui_config.md
Normal file
@ -0,0 +1,136 @@
|
|||||||
|
# Tutorial - Automatic Scan GUI Generation
|
||||||
|
|
||||||
|
The gui_config feature is an optional addition for users who want to control their scans using the automatically
|
||||||
|
generated ScanControl GUI from `bec_widgets`. This configuration helps in organizing scan parameters into specific
|
||||||
|
groups, making the GUI more user-friendly and intuitive.
|
||||||
|
|
||||||
|
## Overview
|
||||||
|
|
||||||
|
The `gui_config` attribute in a scan class specifies how the parameters should be grouped in the GUI. This
|
||||||
|
configuration:
|
||||||
|
|
||||||
|
- Requires minimal user input.
|
||||||
|
- Focuses on grouping parameters into specific categories.
|
||||||
|
- Uses Pydantic validators to ensure integrity and completeness.
|
||||||
|
|
||||||
|
## Step-by-Step Guide
|
||||||
|
|
||||||
|
### Step 1: Add `gui_config` Attribute
|
||||||
|
|
||||||
|
Add the `gui_config` attribute to your scan class. This attribute is a dictionary where keys represent the group names
|
||||||
|
and values are lists of parameter names. These groups dictate how the parameters are organized in the GUI.
|
||||||
|
|
||||||
|
```python
|
||||||
|
class FermatSpiralScan(ScanBase):
|
||||||
|
scan_name = "fermat_scan"
|
||||||
|
required_kwargs = ["step", "relative"]
|
||||||
|
gui_config = {
|
||||||
|
"Device 1": ["motor1", "start_motor1", "stop_motor1"],
|
||||||
|
"Device 2": ["motor2", "start_motor2", "stop_motor2"],
|
||||||
|
"Movement Parameters": ["step", "relative"],
|
||||||
|
"Acquisition Parameters": ["exp_time", "settling_time", "burst_at_each_point"],
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
### Step 2: Ensure Complete Signatures and Docstrings
|
||||||
|
|
||||||
|
Make sure that the signatures of all parameters in your scan class are complete with types and detailed docstrings. This
|
||||||
|
ensures that Pydantic can validate and process the configuration without errors.
|
||||||
|
|
||||||
|
Example of a detailed `__init__` method:
|
||||||
|
|
||||||
|
```python
|
||||||
|
class FermatSpiralScan(ScanBase):
|
||||||
|
scan_name = "fermat_scan"
|
||||||
|
required_kwargs = ["step", "relative"]
|
||||||
|
gui_config = {
|
||||||
|
"Device 1": ["motor1", "start_motor1", "stop_motor1"],
|
||||||
|
"Device 2": ["motor2", "start_motor2", "stop_motor2"],
|
||||||
|
"Movement Parameters": ["step", "relative"],
|
||||||
|
"Acquisition Parameters": ["exp_time", "settling_time", "burst_at_each_point"],
|
||||||
|
}
|
||||||
|
|
||||||
|
def __init__(
|
||||||
|
self,
|
||||||
|
motor1: DeviceBase,
|
||||||
|
start_motor1: float,
|
||||||
|
stop_motor1: float,
|
||||||
|
motor2: DeviceBase,
|
||||||
|
start_motor2: float,
|
||||||
|
stop_motor2: float,
|
||||||
|
step: float = 0.1,
|
||||||
|
exp_time: float = 0,
|
||||||
|
settling_time: float = 0,
|
||||||
|
relative: bool = False,
|
||||||
|
burst_at_each_point: int = 1,
|
||||||
|
spiral_type: float = 0,
|
||||||
|
optim_trajectory: Literal["corridor", None] = None,
|
||||||
|
**kwargs,
|
||||||
|
):
|
||||||
|
"""
|
||||||
|
A scan following Fermat's spiral.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
motor1 (DeviceBase): first motor
|
||||||
|
start_motor1 (float): start position motor 1
|
||||||
|
stop_motor1 (float): end position motor 1
|
||||||
|
motor2 (DeviceBase): second motor
|
||||||
|
start_motor2 (float): start position motor 2
|
||||||
|
stop_motor2 (float): end position motor 2
|
||||||
|
step (float): step size in motor units. Default is 0.1.
|
||||||
|
exp_time (float): exposure time in seconds. Default is 0.
|
||||||
|
settling_time (float): settling time in seconds. Default is 0.
|
||||||
|
relative (bool): if True, the motors will be moved relative to their current position. Default is False.
|
||||||
|
burst_at_each_point (int): number of exposures at each point. Default is 1.
|
||||||
|
spiral_type (float): type of spiral to use. Default is 0.
|
||||||
|
optim_trajectory (str): trajectory optimization method. Default is None. Options are "corridor" and "none".
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
ScanReport
|
||||||
|
|
||||||
|
Examples:
|
||||||
|
>>> scans.fermat_scan(dev.motor1, -5, 5, dev.motor2, -5, 5, step=0.5, exp_time=0.1, relative=True, optim_trajectory="corridor")
|
||||||
|
"""
|
||||||
|
```
|
||||||
|
|
||||||
|
Note that you can omit certain parameters from the `gui_config` if they are not required to be displayed in the GUI or
|
||||||
|
not expose them to the user.
|
||||||
|
|
||||||
|
### Step 3: Utilize Pydantic Validators
|
||||||
|
|
||||||
|
Pydantic validators are used to enhance the `gui_config` by:
|
||||||
|
|
||||||
|
- **Validating the arguments**: Ensuring that each argument specified in the `gui_config` exists within the scan class.
|
||||||
|
- **Formatting display names**: Creating user-friendly display names for labels based on parameter names.
|
||||||
|
- **Extracting tooltips**: Deriving tooltips from the first sentence of each parameter's docstring.
|
||||||
|
- **Retrieving default values**: Obtaining default values for each parameter.
|
||||||
|
|
||||||
|
The Pydantic is automatically applied when the `bec-scan-server` starts up, and it will raise an error if
|
||||||
|
the `gui_config` is incomplete or incorrect.
|
||||||
|
|
||||||
|
```{note}
|
||||||
|
Note that if the signature or docstring of a parameter is incomplete, Pydantic will raise an error during a `bec-scan-server` startup!
|
||||||
|
```
|
||||||
|
|
||||||
|
## Example of a Complete Scan Class with `gui_config`
|
||||||
|
|
||||||
|
Here is the complete example of the `FermatSpiralScan` class with the `gui_config` implemented:
|
||||||
|
|
||||||
|
````{dropdown} View code: ScanBase class
|
||||||
|
:icon: code-square
|
||||||
|
:animate: fade-in-slide-down
|
||||||
|
|
||||||
|
```{literalinclude} ../../../../bec_server/bec_server/scan_server/scans.py
|
||||||
|
:language: python
|
||||||
|
:pyobject: FermatSpiralScan
|
||||||
|
```
|
||||||
|
````
|
||||||
|
|
||||||
|
By following these steps, you can easily configure the GUI for your scans, making them more user-friendly and intuitive
|
||||||
|
for users who want to use the ScanControl GUI from `bec_widgets`.
|
||||||
|
|
||||||
|
The resulting GUI will display the parameters in the specified groups, making it easier for users to understand and
|
||||||
|
interact with the scan settings:
|
||||||
|
|
||||||
|
```{figure} ../assets/scan_GUI_example.png
|
||||||
|
```
|
Loading…
x
Reference in New Issue
Block a user