feat(scan_control): add QML and QWidget versions of a modernized scan control for comparison

This commit is contained in:
Claude committed 2026-10-09 20:26:51 +00:00
1 parent 703513b175
commit 5ed57fac38
13 files changed
+3544

No files matched your search

Whitespace-only changes.
@@ -0,0 +1,531 @@
"""Shared behaviour and public API of the modernized scan control widgets.
``ScanControlQml`` and ``ScanControlModern`` derive from ``ModernScanControlBase`` and only add
a view. The public API mirrors the classic ``ScanControl`` so either can replace it.
"""
from __future__ import annotations
from types import NoneType
from bec_lib.endpoints import MessageEndpoints
from bec_lib.logger import bec_logger
from bec_lib.metadata_schema import get_metadata_schema_for_scan
from qtpy.QtCore import QTimer, Signal
from qtpy.QtWidgets import QDialog, QWidget
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
from bec_widgets.utils.scan_arg_metadata import device_units
from bec_widgets.widgets.control.scan_control.modern.form_core import (
MetadataFormState,
ScanFormSpec,
ScanFormState,
device_names_for_scans,
lookup_last_scan_parameters,
)
from bec_widgets.widgets.control.scan_control.scan_control import (
ScanControlConfig,
ScanParameterConfig,
)
from bec_widgets.widgets.control.scan_control.scan_info_adapter import ScanInfoAdapter
from bec_widgets.widgets.control.scan_control.scan_info_dialog import ScanInfoDialog
from bec_widgets.widgets.control.scan_control.scan_selection_dialog import ScanSelectionDialog
logger = bec_logger.logger
class ModernScanControlBase(BECWidget, QWidget):
"""Scan submission form with the same public API as ``ScanControl``.
Subclasses provide the view by overriding ``_build_view`` and reacting to the signals of
``form``, ``metadata`` and ``status_changed``.
"""
USER_ACCESS = ["attach", "detach", "screenshot"]
RPC = False
PLUGIN = False
LAST_SCAN_FETCH_TIMEOUT_MS = 30_000
EMERGENCY_STOP_WINDOW_MS = 3_000
scan_started = Signal()
scan_selected = Signal(str)
device_selected = Signal(str)
scan_args = Signal(list)
#: The scan list, a ``hide_*`` option, or the start/restore availability changed.
status_changed = Signal()
#: Short feedback for the footer, e.g. after submitting or restoring.
notice = Signal(str)
_last_scan_parameters_received = Signal(int, str, object)
def __init__(
self,
parent=None,
client=None,
config: ScanControlConfig | dict | None = None,
gui_id: str | None = None,
allowed_scans: list | None = None,
default_scan: str | None = None,
**kwargs,
):
if config is None:
config = ScanControlConfig(
widget_class=self.__class__.__name__, allowed_scans=allowed_scans
)
super().__init__(parent=parent, client=client, gui_id=gui_id, config=config, **kwargs)
self.get_bec_shortcuts()
if default_scan is not None:
self.config.default_scan = default_scan
if allowed_scans is not None:
self.config.allowed_scans = list(dict.fromkeys(allowed_scans)) or None
self._adapter = ScanInfoAdapter()
self.available_scans: dict = {}
self.visible_scans: list[str] = []
self._selected_scan = ""
self.previous_scan: str | None = None
self._scan_info_dialog: ScanInfoDialog | None = None
self._options = {
"hide_arg_box": False,
"hide_kwarg_boxes": False,
"hide_scan_control_buttons": False,
"hide_metadata": False,
"hide_optional_metadata": False,
"hide_scan_selection_combobox": False,
"hide_scan_selector_settings_button": False,
"hide_add_remove_buttons": False,
}
self.form = ScanFormState(self)
self.form.units_resolver = self._device_units
self.form.values_changed.connect(self._on_form_values_changed)
self.metadata = MetadataFormState(self)
self.metadata.values_changed.connect(self.status_changed)
self._restoring = False
self._fetch_generation = 0
self._fetch_watchdog = QTimer(self)
self._fetch_watchdog.setSingleShot(True)
self._fetch_watchdog.setInterval(self.LAST_SCAN_FETCH_TIMEOUT_MS)
self._fetch_watchdog.timeout.connect(self._on_fetch_timeout)
self._last_scan_parameters_received.connect(self._apply_last_scan_parameters)
self._stop_armed = False
self._stop_timer = QTimer(self)
self._stop_timer.setSingleShot(True)
self._stop_timer.setInterval(self.EMERGENCY_STOP_WINDOW_MS)
self._stop_timer.timeout.connect(self._disarm_stop)
self._last_devices = ""
self._build_view()
self.refresh_devices()
self.populate_scans()
if self.config.default_scan is not None:
self.current_scan = self.config.default_scan
def _build_view(self) -> None:
"""Create the view. Implemented by subclasses."""
raise NotImplementedError
# ------------------------------------------------------------------ scans
def populate_scans(self) -> None:
"""Read the available scans from BEC and refresh the selector."""
msg = self.client.connector.get(MessageEndpoints.available_scans())
self.available_scans = msg.resource if msg is not None else {}
self._update_scan_selector()
def _supported_scan_names(self) -> list[str]:
return [
name
for name, info in self.available_scans.items()
if self._adapter.has_scan_ui_config(info) and not name.startswith("_")
]
def _update_scan_selector(self) -> None:
if self._selected_scan:
self.save_current_scan_parameters()
allowed = self.config.allowed_scans
if allowed is None:
self.visible_scans = self._supported_scan_names()
else:
self.visible_scans = [s for s in allowed if s in self.available_scans]
self.status_changed.emit()
if not self.visible_scans:
self._selected_scan = ""
self.form.load(ScanFormSpec())
elif self._selected_scan not in self.visible_scans:
self._select(self.visible_scans[0])
def scan_docstring(self, scan_name: str) -> str | None:
"""Docstring of ``scan_name`` as published by BEC."""
info = self.available_scans.get(scan_name, {})
doc = info.get("doc") if isinstance(info, dict) else None
return doc if isinstance(doc, str) else None
def is_valid_scan(self, scan_name: str) -> bool:
"""Whether ``scan_name`` is listed in the selector (case-insensitive)."""
return self._resolve_scan_name(scan_name) is not None
def _resolve_scan_name(self, scan_name: str) -> str | None:
lowered = (scan_name or "").lower()
return next((s for s in self.visible_scans if s.lower() == lowered), None)
def _select(self, scan_name: str) -> None:
resolved = self._resolve_scan_name(scan_name)
if resolved is None or resolved == self._selected_scan:
return
if self._selected_scan:
self._save_scan_parameters(self._selected_scan)
self._selected_scan = resolved
info = self.available_scans.get(resolved, {})
ui_config = self._adapter.build_scan_ui_config(info)
self.form.load(
ScanFormSpec.from_ui_config(resolved, self.scan_docstring(resolved), ui_config)
)
self.metadata.load(get_metadata_schema_for_scan(resolved), resolved)
self.restore_scan_parameters(resolved)
self.scan_selected.emit(resolved)
self.status_changed.emit()
@SafeProperty(str)
def current_scan(self) -> str:
"""Name of the selected scan."""
return self._selected_scan
@current_scan.setter
def current_scan(self, scan_name: str):
self._select(scan_name)
@SafeSlot(str)
def set_current_scan(self, scan_name: str):
"""Select ``scan_name`` if it is listed."""
self._select(scan_name)
@SafeProperty("QStringList")
def allowed_scans(self) -> list[str]:
"""Scans shown in the selector; every supported scan when no filter is set."""
allowed = self.config.allowed_scans
return self._supported_scan_names() if allowed is None else list(allowed)
@allowed_scans.setter
def allowed_scans(self, scan_names: list[str] | str | None):
if isinstance(scan_names, str):
scan_names = [scan_names]
if scan_names is not None:
scan_names = list(dict.fromkeys(scan_names))
if not scan_names or scan_names == self._supported_scan_names():
scan_names = None
self.config.allowed_scans = scan_names
self._update_scan_selector()
@SafeSlot()
def show_selected_scan_info(self, *_args) -> None:
"""Show the documentation of the selected scan."""
self.show_scan_info(self._selected_scan)
@SafeSlot(str)
def show_scan_info(self, scan_name: str) -> None:
"""Show the documentation of ``scan_name`` in a non-blocking dialog."""
if self._scan_info_dialog is None:
self._scan_info_dialog = ScanInfoDialog(self)
self._scan_info_dialog.show_scan(scan_name, self.scan_docstring(scan_name))
@SafeSlot()
def show_scan_selector_settings(self, *_args) -> None:
"""Let the user choose which scans the selector offers."""
scan_names = self._supported_scan_names()
allowed = self.config.allowed_scans
if allowed is not None:
scan_names += [s for s in allowed if s not in scan_names]
dialog = ScanSelectionDialog(
scan_names=scan_names,
selected_scans=scan_names if allowed is None else allowed,
scan_docs={name: self.scan_docstring(name) for name in scan_names},
parent=self,
)
try:
accepted = dialog.exec() == QDialog.DialogCode.Accepted
selected = dialog.selected_scans() if accepted else None
finally:
dialog.deleteLater()
if selected is not None:
self.allowed_scans = None if selected == scan_names else selected
# ------------------------------------------------------------------ devices
@SafeSlot()
def refresh_devices(self) -> None:
"""Reload the device names offered by device fields."""
self.form.device_names = device_names_for_scans(self.dev)
def _device_units(self, device_name: str) -> str | None:
device = getattr(self.dev, device_name, None)
return device_units(device) if device is not None else None
def _on_form_values_changed(self) -> None:
devices = " ".join(d for d in self.form.selected_devices() if d in self.form.device_names)
if devices != self._last_devices:
self._last_devices = devices
self.device_selected.emit(devices)
self.status_changed.emit()
# ------------------------------------------------------------------ parameters
def get_scan_parameters(self, bec_object: bool = True) -> tuple[list, dict]:
"""Positional and keyword arguments of the scan as entered in the form.
Args:
bec_object(bool): If True, device fields are returned as device objects.
"""
args = self.form.args_list(self.dev if bec_object else None)
kwargs = self.form.kwargs_dict()
if self.metadata.data is not None:
kwargs["metadata"] = self.metadata.data
return args, kwargs
def save_current_scan_parameters(self) -> None:
"""Remember the values of the selected scan in the widget config."""
if self._selected_scan:
self._save_scan_parameters(self._selected_scan)
def _save_scan_parameters(self, scan_name: str) -> None:
self.previous_scan = scan_name
self.config.scans[scan_name] = ScanParameterConfig(
name=scan_name, args=self.form.args_list(), kwargs=self.form.kwargs_dict()
)
def restore_scan_parameters(self, scan_name: str) -> None:
"""Restore remembered values of ``scan_name``; shared keywords carry over between scans."""
params = self.config.scans.get(scan_name)
if params is None and self.previous_scan is not None:
previous = self.config.scans.get(self.previous_scan)
if previous is not None and previous.kwargs:
self.form.set_kwargs(previous.kwargs)
return
if params is None:
return
if params.args:
self.form.set_args(params.args)
if params.kwargs:
self.form.set_kwargs(params.kwargs)
# ------------------------------------------------------------------ status
def problems(self) -> list[str]:
"""Reasons why Start is disabled, most important first."""
if not self._selected_scan:
return ["No scan selected"]
return self.form.problems() + self.metadata.problems()
def can_start(self) -> bool:
"""Whether the form is complete and valid."""
return not self.problems()
@property
def is_restoring(self) -> bool:
"""Whether the last scan parameters are being fetched."""
return self._restoring
def option(self, name: str) -> bool:
"""Current value of a ``hide_*`` option."""
return self._options[name]
def _set_option(self, name: str, value: bool) -> None:
self._options[name] = bool(value)
self.status_changed.emit()
# ------------------------------------------------------------------ actions
@SafeSlot(popup_error=True)
def run_scan(self, *_args) -> None:
"""Submit the selected scan with the entered parameters."""
if not self.can_start():
self.notice.emit(self.problems()[0])
return
args, kwargs = self.get_scan_parameters()
self.scan_args.emit(args)
scan_function = getattr(self.scans, self._selected_scan)
if callable(scan_function):
self.scan_started.emit()
scan_function(*args, **kwargs)
self.notice.emit(f"Submitted {self._selected_scan}")
@SafeSlot()
def stop_scan(self, *_args) -> None:
"""Abort the running scan; a second click within a few seconds halts it."""
if self._stop_armed:
self.queue.request_scan_halt()
self.notice.emit("Halt requested")
else:
self.queue.request_scan_abortion()
self.notice.emit("Abort requested. Click Stop again to halt immediately.")
self._stop_armed = True
self._stop_timer.start()
self.status_changed.emit()
@property
def stop_armed(self) -> bool:
"""Whether the next Stop click requests a halt."""
return self._stop_armed
def _disarm_stop(self) -> None:
self._stop_armed = False
self.status_changed.emit()
@SafeSlot()
def request_last_executed_scan_parameters(self, *_args) -> None:
"""Fill the form with the parameters of the last run of the selected scan."""
if self._restoring or not self._selected_scan:
return
self._restoring = True
self._fetch_generation += 1
generation = self._fetch_generation
self._fetch_watchdog.start()
self.status_changed.emit()
self.submit_task(
self._fetch_last_parameters,
generation,
self._selected_scan,
on_complete=lambda *_: self._on_fetch_finished(generation),
on_failed=lambda *_: self._on_fetch_finished(generation),
)
def _fetch_last_parameters(self, generation: int, scan_name: str) -> None:
try:
parameters = lookup_last_scan_parameters(self.client, scan_name)
except Exception: # pylint: disable=broad-except
logger.exception(f"Failed to fetch parameters for scan {scan_name}")
return
try:
self._last_scan_parameters_received.emit(generation, scan_name, parameters)
except RuntimeError:
logger.debug("Scan control deleted before the last scan parameters arrived")
@SafeSlot(int, str, object)
def _apply_last_scan_parameters(self, generation: int, scan_name: str, parameters) -> None:
if generation != self._fetch_generation or scan_name != self._selected_scan:
return
if parameters is None:
self.notice.emit(f"No earlier run of {scan_name} found")
return
args, kwargs = parameters
if args:
self.form.set_args(args)
if kwargs:
self.form.set_kwargs(kwargs)
self.notice.emit(f"Restored the last {scan_name} parameters")
def _on_fetch_finished(self, generation: int) -> None:
if generation != self._fetch_generation:
return
self._fetch_watchdog.stop()
self._restoring = False
self.status_changed.emit()
def _on_fetch_timeout(self) -> None:
logger.warning("Timed out while fetching the last executed scan parameters")
self._fetch_generation += 1
self._restoring = False
self.notice.emit("Could not reach the scan history")
self.status_changed.emit()
@SafeSlot(dict)
@SafeSlot(NoneType)
def update_scan_metadata(self, md: dict | None) -> None:
"""Replace the free metadata entries with ``md``."""
self.metadata.set_extras(list((md or {}).items()))
# ------------------------------------------------------------------ hide_* options
@SafeProperty(bool)
def hide_arg_box(self) -> bool:
"""Hide the argument rows."""
return self._options["hide_arg_box"]
@hide_arg_box.setter
def hide_arg_box(self, hide: bool):
self._set_option("hide_arg_box", hide)
@SafeProperty(bool)
def hide_kwarg_boxes(self) -> bool:
"""Hide the keyword argument groups."""
return self._options["hide_kwarg_boxes"]
@hide_kwarg_boxes.setter
def hide_kwarg_boxes(self, hide: bool):
self._set_option("hide_kwarg_boxes", hide)
@SafeProperty(bool)
def hide_scan_control_buttons(self) -> bool:
"""Hide the Start/Stop footer."""
return self._options["hide_scan_control_buttons"]
@hide_scan_control_buttons.setter
def hide_scan_control_buttons(self, hide: bool):
self._set_option("hide_scan_control_buttons", hide)
@SafeSlot(bool)
def show_scan_control_buttons(self, show: bool):
"""Show or hide the Start/Stop footer."""
self._set_option("hide_scan_control_buttons", not show)
@SafeProperty(bool)
def hide_metadata(self) -> bool:
"""Hide the metadata section."""
return self._options["hide_metadata"]
@hide_metadata.setter
def hide_metadata(self, hide: bool):
self._set_option("hide_metadata", hide)
@SafeProperty(bool)
def hide_optional_metadata(self) -> bool:
"""Hide the free key/value metadata."""
return self._options["hide_optional_metadata"]
@hide_optional_metadata.setter
def hide_optional_metadata(self, hide: bool):
self._set_option("hide_optional_metadata", hide)
@SafeProperty(bool)
def hide_scan_selection_combobox(self) -> bool:
"""Hide the scan selector."""
return self._options["hide_scan_selection_combobox"]
@hide_scan_selection_combobox.setter
def hide_scan_selection_combobox(self, hide: bool):
self._set_option("hide_scan_selection_combobox", hide)
@SafeSlot(bool)
def show_scan_selection_combobox(self, show: bool):
"""Show or hide the scan selector."""
self._set_option("hide_scan_selection_combobox", not show)
@SafeProperty(bool)
def hide_scan_selector_settings_button(self) -> bool:
"""Hide the button that filters the scan list."""
return self._options["hide_scan_selector_settings_button"]
@hide_scan_selector_settings_button.setter
def hide_scan_selector_settings_button(self, hide: bool):
self._set_option("hide_scan_selector_settings_button", hide)
@SafeProperty(bool)
def hide_add_remove_buttons(self) -> bool:
"""Hide adding and removing argument rows."""
return self._options["hide_add_remove_buttons"]
@hide_add_remove_buttons.setter
def hide_add_remove_buttons(self, hide: bool):
self._set_option("hide_add_remove_buttons", hide)
def cleanup(self):
"""Stop timers and close dialogs."""
self._fetch_generation += 1
self._fetch_watchdog.stop()
self._stop_timer.stop()
if self._scan_info_dialog is not None:
self._scan_info_dialog.close()
self._scan_info_dialog.deleteLater()
self._scan_info_dialog = None
super().cleanup()
@@ -0,0 +1,576 @@
"""View-independent state for the modernized scan control.
Both the QML and the QWidget versions of the modern scan control render the same
``ScanFormState`` and ``MetadataFormState``. Keeping the form state in plain Python makes the
two views directly comparable: they only differ in how they draw the form, not in what the
form does.
"""
from __future__ import annotations
import inspect
import math
import re
from dataclasses import dataclass, field
from decimal import Decimal
from types import NoneType, UnionType
from typing import Any, Callable, Literal, Union, get_args, get_origin
from bec_lib.endpoints import MessageEndpoints
from bec_lib.logger import bec_logger
from bec_lib.scan_history import ScanHistory
from pydantic import BaseModel, ValidationError
from pydantic_core import PydanticUndefined
from qtpy.QtCore import QObject, Signal
from bec_widgets.utils.scan_arg_metadata import format_display_name
logger = bec_logger.logger
FieldKind = Literal["device", "float", "int", "bool", "str", "choice"]
INT_LIMIT = 2147483647
DEFAULT_DECIMALS = 4
RECENT_SCAN_HISTORY_COUNT = 50
MAX_HISTORY_LOOKBACK = 500
def scan_summary(docstring: str | None) -> str:
"""Return the first paragraph of a scan docstring as one line."""
if not docstring:
return ""
text = inspect.cleandoc(docstring)
first = re.split(r"\n\s*\n", text, maxsplit=1)[0]
return " ".join(line.strip() for line in first.splitlines())
@dataclass
class FieldSpec:
"""One input of the scan form."""
name: str
label: str
kind: FieldKind
default: Any = None
tooltip: str = ""
units: str | None = None
reference_units: str | None = None
minimum: float | None = None
maximum: float | None = None
decimals: int = DEFAULT_DECIMALS
choices: list[str] = field(default_factory=list)
expert: bool = False
@classmethod
def from_input(cls, item: dict) -> FieldSpec | None:
"""Build a field from a ``ScanInfoAdapter`` input config, or None if unsupported."""
item_type = item.get("type")
choices: list[str] = []
if isinstance(item_type, dict) and "Literal" in item_type:
kind = "choice"
# Keep the declared order (the classic widget used a set and shuffled it).
literals = list(dict.fromkeys(item_type["Literal"]))
choices = ["" if value is None else str(value) for value in literals]
elif item_type in ("device", "DeviceBase"):
kind = "device"
elif item_type in ("float", "int", "bool", "str"):
kind = item_type
else:
logger.error(f"Unsupported annotation '{item_type}' for parameter '{item.get('name')}'")
return None
default = item.get("default")
if default == "_empty":
default = None
precision = item.get("precision")
try:
decimals = max(0, int(precision)) if precision is not None else DEFAULT_DECIMALS
except (TypeError, ValueError):
decimals = DEFAULT_DECIMALS
spec = cls(
name=item["name"],
label=item.get("display_name") or format_display_name(item["name"]),
kind=kind,
default=default,
tooltip=item.get("tooltip") or "",
units=item.get("units"),
reference_units=item.get("reference_units"),
decimals=decimals if kind == "float" else 0,
choices=choices,
expert=bool(item.get("expert", False)),
)
spec._apply_limits(item)
return spec
def _apply_limits(self, item: dict) -> None:
if self.kind not in ("int", "float"):
return
step = 1 if self.kind == "int" else 10 ** (-self.decimals)
if item.get("ge") is not None:
self.minimum = float(item["ge"])
if item.get("gt") is not None:
self.minimum = float(item["gt"]) + step
if item.get("le") is not None:
self.maximum = float(item["le"])
if item.get("lt") is not None:
self.maximum = float(item["lt"]) - step
def initial_value(self) -> Any:
"""Value shown before the user edits the field."""
if self.default is not None:
return self.coerce(self.default)
if self.kind in ("int", "float"):
value = 0.0
if self.minimum is not None and value < self.minimum:
value = self.minimum
if self.maximum is not None and value > self.maximum:
value = self.maximum
return int(value) if self.kind == "int" else value
if self.kind == "bool":
return False
if self.kind == "choice":
# An optional literal (None among the choices) starts unset, like the classic widget.
return "" if "" in self.choices or not self.choices else self.choices[0]
return ""
def coerce(self, value: Any) -> Any:
"""Convert a value coming from a view or from BEC to this field's type."""
try:
if self.kind == "int":
return int(value)
if self.kind == "float":
return float(value)
except (TypeError, ValueError):
return 0 if self.kind == "int" else 0.0
if self.kind == "bool":
if isinstance(value, str):
return value.strip().lower() in ("1", "true", "yes", "on")
return bool(value)
if self.kind == "device":
return getattr(value, "name", None) or ("" if value is None else str(value))
return "" if value is None else str(value)
def to_dict(self) -> dict:
"""Plain dict for QML."""
return {
"name": self.name,
"label": self.label,
"kind": self.kind,
"tooltip": self.tooltip,
"minimum": -INT_LIMIT if self.minimum is None else self.minimum,
"maximum": INT_LIMIT if self.maximum is None else self.maximum,
"decimals": self.decimals,
"choices": self.choices,
"expert": self.expert,
}
@dataclass
class FieldGroup:
"""A titled group of keyword fields."""
name: str
fields: list[FieldSpec]
@dataclass
class ScanFormSpec:
"""Everything a view needs to render the form of one scan."""
scan_name: str = ""
summary: str = ""
arg_title: str = ""
arg_fields: list[FieldSpec] = field(default_factory=list)
arg_min: int = 0
arg_max: int | None = None
groups: list[FieldGroup] = field(default_factory=list)
@classmethod
def from_ui_config(cls, scan_name: str, docstring: str | None, ui_config: dict):
"""Build the spec from ``ScanInfoAdapter.build_scan_ui_config`` output."""
spec = cls(scan_name=scan_name, summary=scan_summary(docstring))
arg_group = ui_config.get("arg_group") or {}
if arg_group.get("arg_inputs"):
spec.arg_title = arg_group.get("name") or "Scan Arguments"
spec.arg_fields = [
f for f in map(FieldSpec.from_input, arg_group.get("inputs", [])) if f is not None
]
spec.arg_min = arg_group.get("min") or 1
spec.arg_max = arg_group.get("max")
for group in ui_config.get("kwarg_groups", []):
fields = [f for f in map(FieldSpec.from_input, group.get("inputs", [])) if f]
if fields:
spec.groups.append(FieldGroup(group.get("name", ""), fields))
return spec
@property
def kwarg_fields(self) -> list[FieldSpec]:
"""All keyword fields across groups."""
return [f for group in self.groups for f in group.fields]
class ScanFormState(QObject):
"""Values of the scan form plus their validation.
Signals:
rebuilt: the spec or the number of argument rows changed; views rebuild their layout.
values_changed: one or more values changed; views refresh their editors.
"""
rebuilt = Signal()
values_changed = Signal()
def __init__(self, parent=None):
super().__init__(parent)
self.spec = ScanFormSpec()
self.arg_rows: list[dict[str, Any]] = []
self.kwargs: dict[str, Any] = {}
self.device_names: list[str] = []
self.units_resolver: Callable[[str], str | None] = lambda _name: None
# ------------------------------------------------------------------ structure
def load(self, spec: ScanFormSpec) -> None:
"""Show a new scan with its default values."""
self.spec = spec
self.arg_rows = [self._new_row() for _ in range(spec.arg_min if spec.arg_fields else 0)]
self.kwargs = {f.name: f.initial_value() for f in spec.kwarg_fields}
self.rebuilt.emit()
self.values_changed.emit()
def _new_row(self) -> dict[str, Any]:
return {f.name: f.initial_value() for f in self.spec.arg_fields}
def can_add_row(self) -> bool:
"""Whether one more argument row is allowed."""
if not self.spec.arg_fields:
return False
return self.spec.arg_max is None or len(self.arg_rows) < self.spec.arg_max
def can_remove_row(self) -> bool:
"""Whether an argument row can be removed."""
return len(self.arg_rows) > max(self.spec.arg_min, 0)
def add_row(self) -> None:
"""Append an argument row, copying nothing but defaults."""
if not self.can_add_row():
return
self.arg_rows.append(self._new_row())
self.rebuilt.emit()
self.values_changed.emit()
def remove_row(self, index: int) -> None:
"""Remove the argument row at ``index``."""
if not self.can_remove_row() or not 0 <= index < len(self.arg_rows):
return
del self.arg_rows[index]
self.rebuilt.emit()
self.values_changed.emit()
# ------------------------------------------------------------------ values
def _arg_field(self, name: str) -> FieldSpec | None:
return next((f for f in self.spec.arg_fields if f.name == name), None)
def _kwarg_field(self, name: str) -> FieldSpec | None:
return next((f for f in self.spec.kwarg_fields if f.name == name), None)
def set_arg_value(self, row: int, name: str, value: Any) -> None:
"""Set one value of an argument row."""
spec = self._arg_field(name)
if spec is None or not 0 <= row < len(self.arg_rows):
return
value = spec.coerce(value)
if self.arg_rows[row].get(name) == value:
return
self.arg_rows[row][name] = value
self.values_changed.emit()
def set_kwarg_value(self, name: str, value: Any) -> None:
"""Set one keyword value."""
spec = self._kwarg_field(name)
if spec is None:
return
value = spec.coerce(value)
if self.kwargs.get(name) == value:
return
self.kwargs[name] = value
self.values_changed.emit()
def set_args(self, args: list) -> None:
"""Replace all argument rows from a flat argument list."""
n_fields = len(self.spec.arg_fields)
if not args or n_fields == 0:
return
n_rows = max(math.ceil(len(args) / n_fields), self.spec.arg_min)
row_count_changed = n_rows != len(self.arg_rows)
self.arg_rows = [self._new_row() for _ in range(n_rows)]
for index, value in enumerate(args):
row, column = divmod(index, n_fields)
spec = self.spec.arg_fields[column]
self.arg_rows[row][spec.name] = spec.coerce(value)
if row_count_changed:
self.rebuilt.emit()
self.values_changed.emit()
def set_kwargs(self, kwargs: dict) -> None:
"""Update keyword values that belong to this scan; unknown keys are ignored."""
for spec in self.spec.kwarg_fields:
if spec.name in kwargs:
self.kwargs[spec.name] = spec.coerce(kwargs[spec.name])
self.values_changed.emit()
def args_list(self, dev=None) -> list:
"""Flat positional arguments; device names become device objects when ``dev`` is set."""
args = []
for row in self.arg_rows:
for spec in self.spec.arg_fields:
value = row.get(spec.name)
if spec.kind == "device" and dev is not None:
value = getattr(dev, value, None) or dev[value]
args.append(value)
return args
def kwargs_dict(self) -> dict:
"""Keyword arguments as BEC expects them."""
result = {}
for spec in self.spec.kwarg_fields:
value = self.kwargs.get(spec.name)
if spec.kind == "choice" and value == "":
value = None
result[spec.name] = value
return result
def selected_devices(self) -> list[str]:
"""Device names chosen in the argument rows, in row order."""
return [
row.get(spec.name, "")
for row in self.arg_rows
for spec in self.spec.arg_fields
if spec.kind == "device"
]
# ------------------------------------------------------------------ derived
def units_for(self, spec: FieldSpec, row: int | None = None) -> str:
"""Units of a field, resolving ``reference_units`` through the referenced device."""
if spec.units:
return spec.units
ref = spec.reference_units
if not ref:
return ""
device = None
if self._arg_field(ref) is not None and self.arg_rows:
device = self.arg_rows[row if row is not None else 0].get(ref)
elif ref in self.kwargs:
device = self.kwargs.get(ref)
if device and device in self.device_names:
return self.units_resolver(device) or ""
return ""
def field_error(self, spec: FieldSpec, value: Any) -> str:
"""Message for an invalid value, or an empty string."""
if spec.kind != "device":
return ""
if not value:
return f"Choose a {spec.label.lower()}"
if value not in self.device_names:
return f"Unknown device '{value}'"
return ""
def problems(self) -> list[str]:
"""Human-readable reasons why the scan cannot start yet."""
problems = []
multi_row = len(self.arg_rows) > 1
for index, row in enumerate(self.arg_rows):
for spec in self.spec.arg_fields:
error = self.field_error(spec, row.get(spec.name))
if error:
problems.append(f"Row {index + 1}: {error}" if multi_row else error)
for spec in self.spec.kwarg_fields:
error = self.field_error(spec, self.kwargs.get(spec.name))
if error:
problems.append(f"{spec.label}: {error}")
return problems
@dataclass
class MetadataField:
"""One field of the metadata schema."""
name: str
label: str
kind: Literal["str", "int", "float", "bool"]
required: bool
default: Any
description: str
def placeholder(self, scan_name: str) -> str:
"""Hint shown in the empty field."""
return scan_name if self.name == "scan_name" else self.description
def to_dict(self, scan_name: str = "") -> dict:
"""Plain dict for QML."""
return {
"name": self.name,
"label": self.label,
"kind": self.kind,
"required": self.required,
"description": self.description,
"placeholder": self.placeholder(scan_name),
}
def _metadata_kind(annotation: Any) -> str:
origin = get_origin(annotation)
if origin in (Union, UnionType):
args = [arg for arg in get_args(annotation) if arg is not NoneType]
annotation = args[0] if args else str
if annotation is bool:
return "bool"
if annotation is int:
return "int"
if annotation in (float, Decimal):
return "float"
return "str"
class MetadataFormState(QObject):
"""Values and validation of the scan metadata form, driven by the scan's pydantic schema."""
rebuilt = Signal()
values_changed = Signal()
def __init__(self, parent=None):
super().__init__(parent)
self.schema: type[BaseModel] | None = None
self.fields: list[MetadataField] = []
self.values: dict[str, Any] = {}
self.extras: list[list[str]] = []
self.scan_name = ""
self.errors: dict[str, str] = {}
self.data: dict | None = None
def load(self, schema: type[BaseModel], scan_name: str) -> None:
"""Switch to a new schema, keeping values of fields that still exist."""
old_values = self.values
self.schema = schema
self.scan_name = scan_name
self.fields = []
for name, info in schema.model_fields.items():
default = None if info.default is PydanticUndefined else info.default
self.fields.append(
MetadataField(
name=name,
label=info.title or format_display_name(name),
kind=_metadata_kind(info.annotation),
required=info.is_required(),
default=default,
description=info.description or "",
)
)
self.values = {f.name: old_values.get(f.name, f.default) for f in self.fields}
self.rebuilt.emit()
self.validate()
def set_value(self, name: str, value: Any) -> None:
"""Set a schema field; empty text clears it."""
spec = next((f for f in self.fields if f.name == name), None)
if spec is None:
return
if isinstance(value, str) and value.strip() == "" and spec.kind != "str":
value = None
self.values[name] = value
self.validate()
def set_extras(self, extras: list[list[str]]) -> None:
"""Set the free key/value metadata."""
self.extras = [[str(k), str(v)] for k, v in extras]
self.validate()
def validate(self) -> None:
"""Validate and publish the result in ``data`` and ``errors``."""
self.errors = {}
self.data = None
if self.schema is None:
self.values_changed.emit()
return
data = {k: v for k, v in self.extras if k}
data.update({k: v for k, v in self.values.items() if v is not None})
if "scan_name" in self.values and not data.get("scan_name"):
data["scan_name"] = self.scan_name
try:
self.data = self.schema.model_validate(data).model_dump()
except ValidationError as exc:
for error in exc.errors():
name = str(error["loc"][0]) if error["loc"] else ""
self.errors.setdefault(name, error["msg"])
self.values_changed.emit()
def problems(self) -> list[str]:
"""Human-readable validation problems."""
labels = {f.name: f.label for f in self.fields}
return [f"Metadata {labels.get(k, k)}: {msg}" for k, msg in self.errors.items()]
def summary(self) -> str:
"""Short one-line description of the filled metadata."""
parts = [
f"{f.label}: {self.values[f.name]}"
for f in self.fields
if f.name != "scan_name" and self.values.get(f.name) not in (None, "")
]
parts += [f"{k}: {v}" for k, v in self.extras if k]
return ", ".join(parts)
def device_names_for_scans(dev) -> list[str]:
"""Devices and writable signals the scan form offers, like the classic DeviceComboBox."""
from bec_lib.device import Device # pylint: disable=import-outside-toplevel
try:
enabled = list(dev.enabled_devices)
except Exception: # pylint: disable=broad-except
return []
names = [d.name for d in enabled if isinstance(d, Device)]
seen = set(names)
for d in enabled:
info = getattr(d, "_info", None) or {}
if info.get("write_access") and d.name not in seen:
names.append(d.name)
seen.add(d.name)
return sorted(names)
def lookup_last_scan_parameters(client, scan_name: str) -> tuple[list, dict] | None:
"""Find the newest parameters for ``scan_name`` in the history cache or a bounded window.
Runs off the GUI thread. Mirrors the bounded search of the classic ``ScanControl``.
"""
# pylint: disable=import-outside-toplevel,protected-access
from bec_widgets.widgets.control.scan_control.scan_control import ScanControl
endpoint = MessageEndpoints.scan_history()
def window(count: int) -> list[dict]:
entries = client.connector.get_last(endpoint, count=count)
if isinstance(entries, dict):
entries = [entries]
return entries or []
tip = window(1)
tip_id = getattr(tip[0].get("data"), "scan_id", None) if tip else None
history = getattr(client, "history", None)
if isinstance(history, ScanHistory) and len(history):
newest = getattr(history[len(history) - 1], "_msg", None)
if newest is not None and newest.scan_id == tip_id:
for index in range(len(history) - 1, -1, -1):
msg = getattr(history[index], "_msg", None)
if msg is not None and msg.scan_name == scan_name:
return ScanControl._parameters_from_request_inputs(msg.request_inputs)
entries = window(RECENT_SCAN_HISTORY_COUNT)
found = ScanControl._find_last_scan_parameters(scan_name, entries)
if found is not None or len(entries) < RECENT_SCAN_HISTORY_COUNT:
return found
return ScanControl._find_last_scan_parameters(scan_name, window(MAX_HISTORY_LOOKBACK))
@@ -0,0 +1,53 @@
import QtQuick
import QtQuick.Controls.Basic
import QtQuick.Layouts
// Text button in one of four looks: "primary", "danger", "outlineDanger" or "ghost".
Button {
id: control
property string kind: "primary"
property string iconName
readonly property bool filled: kind === "primary" || kind === "danger"
readonly property color base: kind === "primary" ? theme.primary
: (kind === "danger" || kind === "outlineDanger") ? theme.danger
: theme.fg
readonly property color ink: filled ? theme.onPrimary : base
implicitHeight: 34
leftPadding: 12
rightPadding: 14
hoverEnabled: true
Accessible.name: text
background: Rectangle {
radius: 7
opacity: control.enabled ? 1 : 0.4
color: control.filled
? (control.down ? Qt.darker(control.base, 1.25) : control.hovered ? Qt.lighter(control.base, 1.1) : control.base)
: (control.down ? theme.border : control.hovered ? theme.hover : "transparent")
border.width: control.kind === "outlineDanger" || control.visualFocus ? 1 : 0
border.color: control.visualFocus ? theme.primary : control.base
}
contentItem: RowLayout {
spacing: 6
opacity: control.enabled ? 1 : 0.6
Image {
visible: control.iconName.length > 0
source: visible ? "image://material/" + control.iconName + "?filled=1&color=" + encodeURIComponent(control.ink.toString()) : ""
sourceSize: Qt.size(18, 18)
Layout.alignment: Qt.AlignVCenter
}
Text {
text: control.text
color: control.ink
font.pixelSize: 13
font.weight: Font.DemiBold
elide: Text.ElideRight
Layout.fillWidth: true
Layout.maximumWidth: implicitWidth
verticalAlignment: Text.AlignVCenter
}
}
}
@@ -0,0 +1,70 @@
import QtQuick
import QtQuick.Controls.Basic
import QtQuick.Layouts
// Titled section. With collapsible: true the header toggles the content.
Rectangle {
id: card
property string title
property string subtitle
property bool collapsible: false
property bool expanded: true
property alias trailing: trailingSlot.data
default property alias content: body.data
Layout.fillWidth: true
implicitHeight: column.implicitHeight + 24
radius: 10
color: theme.card
border.color: theme.border
ColumnLayout {
id: column
anchors.fill: parent
anchors.margins: 12
spacing: 10
RowLayout {
Layout.fillWidth: true
spacing: 6
Image {
visible: card.collapsible
source: "image://material/" + (card.expanded ? "expand_more" : "chevron_right") + "?color=" + encodeURIComponent(theme.muted.toString())
sourceSize: Qt.size(18, 18)
}
Text {
text: card.title.toUpperCase()
color: theme.muted
font.pixelSize: 11
font.weight: Font.Bold
font.letterSpacing: 0.8
}
Text {
text: card.subtitle
color: theme.muted
font.pixelSize: 12
elide: Text.ElideRight
Layout.fillWidth: true
}
Row { id: trailingSlot; spacing: 4 }
}
ColumnLayout {
id: body
Layout.fillWidth: true
spacing: 8
visible: card.expanded
}
}
MouseArea {
visible: card.collapsible
anchors { left: parent.left; right: parent.right; top: parent.top }
height: 40
cursorShape: Qt.PointingHandCursor
onClicked: card.expanded = !card.expanded
z: -1
}
}
@@ -0,0 +1,11 @@
import QtQuick
// Shared look of every input: filled box, accent border on focus, red border on error.
Rectangle {
property bool focused: false
property bool invalid: false
radius: 6
color: theme.field
border.width: focused || invalid ? 1.5 : 1
border.color: invalid ? theme.danger : focused ? theme.primary : theme.border
}
@@ -0,0 +1,203 @@
import QtQuick
import QtQuick.Controls.Basic
import QtQuick.Layouts
// One scan input. Picks the editor from spec.kind and reports changes through edited().
Item {
id: editor
property var spec: ({})
property var value
property string units
property string error
property var devices: []
signal edited(var value)
implicitHeight: 32
implicitWidth: 120
Layout.fillWidth: true
ToolTip.visible: hover.hovered && (error.length > 0 || (spec.tooltip || "").length > 0)
ToolTip.text: error.length > 0 ? error : (spec.tooltip || "")
ToolTip.delay: 600
HoverHandler { id: hover }
Loader {
anchors.fill: parent
sourceComponent: {
switch (editor.spec.kind) {
case "device": return deviceEditor
case "int":
case "float": return numberEditor
case "bool": return boolEditor
case "choice": return choiceEditor
default: return textEditor
}
}
}
Component {
id: numberEditor
TextField {
id: field
readonly property bool isInt: editor.spec.kind === "int"
function format(v) {
if (v === undefined || v === null || v === "") return ""
return isInt ? String(Math.round(v)) : String(Number(v))
}
function sync() { if (!activeFocus) text = format(editor.value) }
function commit() {
if (acceptableInput) editor.edited(isInt ? parseInt(text) : parseFloat(text))
else text = format(editor.value)
}
function step(direction) {
var v = (parseFloat(text) || 0) + direction * (isInt ? 1 : Math.pow(10, -Math.min(editor.spec.decimals, 2)))
v = Math.min(Math.max(v, editor.spec.minimum), editor.spec.maximum)
text = format(isInt ? v : Number(v.toFixed(editor.spec.decimals)))
commit()
}
Component.onCompleted: sync()
Connections { target: editor; function onValueChanged() { field.sync() } }
onEditingFinished: commit()
onActiveFocusChanged: if (!activeFocus) commit()
Keys.onUpPressed: step(1)
Keys.onDownPressed: step(-1)
validator: isInt ? intValidator : doubleValidator
IntValidator { id: intValidator; bottom: editor.spec.minimum; top: editor.spec.maximum }
DoubleValidator {
id: doubleValidator
bottom: editor.spec.minimum
top: editor.spec.maximum
decimals: editor.spec.decimals
notation: DoubleValidator.StandardNotation
locale: "C"
}
horizontalAlignment: TextInput.AlignRight
rightPadding: unitText.text.length > 0 ? unitText.implicitWidth + 14 : 8
leftPadding: 8
color: theme.fg
selectionColor: theme.primary
selectedTextColor: theme.onPrimary
font.pixelSize: 13
background: FieldBackground { focused: field.activeFocus; invalid: !field.acceptableInput && field.text.length > 0 }
Text {
id: unitText
text: editor.units
color: theme.muted
font.pixelSize: 12
anchors.right: parent.right
anchors.rightMargin: 8
anchors.verticalCenter: parent.verticalCenter
}
}
}
Component {
id: textEditor
TextField {
id: field
function sync() { if (!activeFocus) text = editor.value === undefined ? "" : String(editor.value) }
Component.onCompleted: sync()
Connections { target: editor; function onValueChanged() { field.sync() } }
onEditingFinished: editor.edited(text)
placeholderText: editor.spec.placeholder || ""
placeholderTextColor: theme.muted
leftPadding: 8
color: theme.fg
selectionColor: theme.primary
selectedTextColor: theme.onPrimary
font.pixelSize: 13
background: FieldBackground { focused: field.activeFocus }
}
}
Component {
id: boolEditor
Item {
Switch {
id: toggle
anchors.left: parent.left
anchors.verticalCenter: parent.verticalCenter
checked: !!editor.value
onToggled: editor.edited(checked)
Accessible.name: editor.spec.label
indicator: Rectangle {
implicitWidth: 36
implicitHeight: 20
x: toggle.leftPadding
y: (toggle.height - height) / 2
radius: 10
color: toggle.checked ? theme.primary : theme.field
border.color: toggle.checked ? theme.primary : theme.border
border.width: toggle.visualFocus ? 2 : 1
Rectangle {
x: toggle.checked ? parent.width - width - 3 : 3
anchors.verticalCenter: parent.verticalCenter
width: 14; height: 14; radius: 7
color: toggle.checked ? theme.onPrimary : theme.muted
Behavior on x { NumberAnimation { duration: 120 } }
}
}
contentItem: Text {
leftPadding: toggle.indicator.width + 8
text: toggle.checked ? "On" : "Off"
color: theme.muted
font.pixelSize: 12
verticalAlignment: Text.AlignVCenter
}
}
}
}
Component {
id: choiceEditor
ComboBox {
id: combo
model: editor.spec.choices
currentIndex: Math.max(0, (editor.spec.choices || []).indexOf(editor.value === null ? "" : String(editor.value)))
displayText: currentText.length > 0 ? currentText : "None"
onActivated: editor.edited(currentText)
font.pixelSize: 13
background: FieldBackground { focused: combo.activeFocus || combo.popup.visible }
contentItem: Text {
leftPadding: 8
text: combo.displayText
color: combo.currentText.length > 0 ? theme.fg : theme.muted
font: combo.font
verticalAlignment: Text.AlignVCenter
elide: Text.ElideRight
}
}
}
Component {
id: deviceEditor
ComboBox {
id: combo
editable: true
model: editor.devices
font.pixelSize: 13
function sync() { if (!contentItem.activeFocus) editText = editor.value || "" }
function commit() { if (editText !== (editor.value || "")) editor.edited(editText) }
Component.onCompleted: sync()
Connections { target: editor; function onValueChanged() { combo.sync() } }
onActivated: commit()
onAccepted: commit()
Connections { target: combo.contentItem; function onActiveFocusChanged() { if (!combo.contentItem.activeFocus) combo.commit() } }
background: FieldBackground { focused: combo.contentItem.activeFocus || combo.popup.visible; invalid: editor.error.length > 0 && editor.value }
contentItem: TextField {
leftPadding: 8
text: combo.editText
placeholderText: "Device"
placeholderTextColor: theme.muted
color: theme.fg
selectionColor: theme.primary
selectedTextColor: theme.onPrimary
font: combo.font
background: null
validator: combo.validator
inputMethodHints: Qt.ImhNoAutoUppercase
verticalAlignment: Text.AlignVCenter
}
}
}
}
@@ -0,0 +1,34 @@
import QtQuick
import QtQuick.Controls.Basic
// Flat, round-cornered icon button with a tooltip.
AbstractButton {
id: control
property string iconName
property color tint: theme.fg
property string tip
implicitWidth: 30
implicitHeight: 30
padding: 6
hoverEnabled: true
focusPolicy: Qt.TabFocus
Accessible.name: tip
ToolTip.visible: hovered && tip.length > 0
ToolTip.text: tip
ToolTip.delay: 500
background: Rectangle {
radius: 6
color: control.down ? theme.border : control.hovered ? theme.hover : "transparent"
border.width: control.visualFocus ? 1 : 0
border.color: theme.primary
}
contentItem: Image {
source: "image://material/" + control.iconName + "?color=" + encodeURIComponent(control.tint.toString())
sourceSize: Qt.size(18, 18)
fillMode: Image.PreserveAspectFit
opacity: control.enabled ? 1 : 0.35
}
}
@@ -0,0 +1,473 @@
import QtQuick
import QtQuick.Controls.Basic
import QtQuick.Layouts
// Root of the QML scan control: scan picker, scrolling form, sticky Start/Stop footer.
Pane {
id: root
readonly property var opts: sc.options
readonly property bool wide: width > 620
property string noticeText
padding: 0
font.pixelSize: 13
background: Rectangle { color: theme.bg }
palette.window: theme.bg
palette.base: theme.field
palette.button: theme.card
palette.text: theme.fg
palette.windowText: theme.fg
palette.buttonText: theme.fg
palette.highlight: theme.primary
palette.highlightedText: theme.onPrimary
palette.mid: theme.border
palette.dark: theme.border
palette.light: theme.card
palette.toolTipBase: theme.card
palette.toolTipText: theme.fg
Connections {
target: sc
function onNotice(text) { root.noticeText = text; noticeTimer.restart() }
}
Timer { id: noticeTimer; interval: 4000; onTriggered: root.noticeText = "" }
ColumnLayout {
anchors.fill: parent
spacing: 0
// ---------------------------------------------------------------- header
Rectangle {
Layout.fillWidth: true
visible: !root.opts.hide_scan_selection_combobox
implicitHeight: header.implicitHeight + 24
color: theme.card
Rectangle { anchors.bottom: parent.bottom; width: parent.width; height: 1; color: theme.border }
ColumnLayout {
id: header
anchors { fill: parent; margins: 12 }
spacing: 6
RowLayout {
Layout.fillWidth: true
spacing: 4
ComboBox {
id: scanPicker
Layout.fillWidth: true
implicitHeight: 36
editable: true
model: sc.scans
font.pixelSize: 15
font.weight: Font.DemiBold
Accessible.name: "Scan"
readonly property bool typedValid: sc.isValidScan(editText)
function sync() { currentIndex = find(sc.currentScan); editText = sc.currentScan }
function confirm() {
if (sc.isValidScan(editText)) sc.selectScan(editText)
sync()
}
Connections { target: sc; function onStructureChanged() { scanPicker.sync() } function onStatusChanged() { if (!scanPicker.contentItem.activeFocus) scanPicker.sync() } }
Component.onCompleted: sync()
onActivated: confirm()
onAccepted: confirm()
Connections { target: scanPicker.contentItem; function onActiveFocusChanged() { if (!scanPicker.contentItem.activeFocus) scanPicker.confirm() } }
background: FieldBackground {
focused: scanPicker.contentItem.activeFocus || scanPicker.popup.visible
invalid: !scanPicker.typedValid && scanPicker.editText.length > 0
}
contentItem: TextField {
leftPadding: 34
text: scanPicker.editText
placeholderText: "Search scans"
placeholderTextColor: theme.muted
color: theme.fg
font: scanPicker.font
selectionColor: theme.primary
selectedTextColor: theme.onPrimary
background: null
verticalAlignment: Text.AlignVCenter
Image {
x: 9
anchors.verticalCenter: parent.verticalCenter
source: "image://material/search?color=" + encodeURIComponent(theme.muted.toString())
sourceSize: Qt.size(18, 18)
}
}
}
IconButton {
iconName: "info"
tip: "Documentation of this scan"
enabled: sc.currentScan.length > 0
onClicked: sc.showInfo()
}
IconButton {
iconName: "filter_list"
tip: "Choose which scans are listed"
visible: !root.opts.hide_scan_selector_settings_button
onClicked: sc.showFilter()
}
}
Text {
Layout.fillWidth: true
visible: text.length > 0
text: sc.summary
color: theme.muted
font.pixelSize: 12
wrapMode: Text.WordWrap
maximumLineCount: 2
elide: Text.ElideRight
}
}
}
// ---------------------------------------------------------------- form
ScrollView {
id: scroll
Layout.fillWidth: true
Layout.fillHeight: true
contentWidth: availableWidth
clip: true
ColumnLayout {
width: scroll.availableWidth
spacing: 10
Item { implicitHeight: 2 }
Text {
Layout.fillWidth: true
Layout.margins: 24
visible: sc.scans.length === 0
text: "No scans available. Check the scan filter or the BEC server."
color: theme.muted
horizontalAlignment: Text.AlignHCenter
wrapMode: Text.WordWrap
}
// Positional argument rows
Card {
id: argCard
Layout.leftMargin: 10
Layout.rightMargin: 10
visible: sc.argFields.length > 0 && !root.opts.hide_arg_box
title: sc.argTitle
subtitle: sc.rowCount > 1 ? sc.rowCount + " rows" : ""
// Column headings
RowLayout {
Layout.fillWidth: true
spacing: 6
Item { implicitWidth: 22 }
Repeater {
model: sc.argFields
Text {
Layout.fillWidth: true
Layout.preferredWidth: modelData.kind === "device" ? 140 : 90
text: modelData.label
color: theme.muted
font.pixelSize: 12
elide: Text.ElideRight
}
}
Item { implicitWidth: 30; visible: !root.opts.hide_add_remove_buttons }
}
Repeater {
model: sc.rowCount
RowLayout {
id: argRow
required property int index
readonly property var row: sc.argRows[index] || ({ values: {}, units: {}, errors: {} })
Layout.fillWidth: true
spacing: 6
Text {
Layout.preferredWidth: 22
text: argRow.index + 1
color: theme.muted
font.pixelSize: 12
horizontalAlignment: Text.AlignHCenter
}
Repeater {
model: sc.argFields
FieldEditor {
Layout.preferredWidth: modelData.kind === "device" ? 140 : 90
spec: modelData
value: argRow.row.values[modelData.name]
units: argRow.row.units[modelData.name] || ""
error: argRow.row.errors[modelData.name] || ""
devices: sc.devices
onEdited: (v) => sc.setArg(argRow.index, modelData.name, v)
}
}
IconButton {
visible: !root.opts.hide_add_remove_buttons
iconName: "close"
tip: "Remove row " + (argRow.index + 1)
tint: theme.muted
// Keep the column while rows cannot be removed, so the fields do not jump.
enabled: sc.canRemoveRow
opacity: enabled ? 1 : 0
onClicked: sc.removeRow(argRow.index)
}
}
}
ActionButton {
visible: !root.opts.hide_add_remove_buttons && sc.canAddRow
kind: "ghost"
iconName: "add"
text: "Add row"
onClicked: sc.addRow()
}
}
// Keyword argument groups
Repeater {
model: root.opts.hide_kwarg_boxes ? [] : sc.groups
Card {
id: groupCard
required property var modelData
property bool showExpert: false
readonly property int expertCount: modelData.fields.filter(f => f.expert).length
Layout.leftMargin: 10
Layout.rightMargin: 10
title: modelData.name
GridLayout {
Layout.fillWidth: true
columns: root.wide ? 4 : 2
columnSpacing: 10
rowSpacing: 8
Repeater {
model: groupCard.modelData.fields.filter(f => !f.expert || groupCard.showExpert)
delegate: Item {
// Wraps label + editor so both land in consecutive grid cells.
required property var modelData
Layout.columnSpan: 2
Layout.fillWidth: true
implicitHeight: 32
RowLayout {
anchors.fill: parent
spacing: 10
Text {
Layout.preferredWidth: 120
Layout.maximumWidth: 140
text: modelData.label
color: theme.fg
font.pixelSize: 13
elide: Text.ElideRight
ToolTip.text: modelData.tooltip || modelData.label
ToolTip.visible: labelHover.hovered
ToolTip.delay: 500
HoverHandler { id: labelHover }
}
FieldEditor {
spec: modelData
value: sc.kwargs.values[modelData.name]
units: sc.kwargs.units[modelData.name] || ""
error: sc.kwargs.errors[modelData.name] || ""
devices: sc.devices
onEdited: (v) => sc.setKwarg(modelData.name, v)
}
}
}
}
}
ActionButton {
visible: groupCard.expertCount > 0
kind: "ghost"
iconName: groupCard.showExpert ? "expand_less" : "expand_more"
text: (groupCard.showExpert ? "Hide" : "Show") + " advanced (" + groupCard.expertCount + ")"
onClicked: groupCard.showExpert = !groupCard.showExpert
}
}
}
// Metadata
Card {
id: metaCard
Layout.leftMargin: 10
Layout.rightMargin: 10
visible: !root.opts.hide_metadata && sc.currentScan.length > 0
collapsible: true
expanded: Object.keys(sc.metaErrors).length > 0
title: "Metadata"
subtitle: Object.keys(sc.metaErrors).length > 0 ? "" : (sc.metaSummary || "optional")
trailing: [
Rectangle {
visible: Object.keys(sc.metaErrors).length > 0
radius: 9
height: 18
width: badge.implicitWidth + 14
color: Qt.rgba(theme.danger.r, theme.danger.g, theme.danger.b, 0.18)
Text {
id: badge
anchors.centerIn: parent
text: Object.keys(sc.metaErrors).length + " to fix"
color: theme.danger
font.pixelSize: 11
font.weight: Font.DemiBold
}
}
]
GridLayout {
Layout.fillWidth: true
columns: 2
columnSpacing: 10
rowSpacing: 8
Repeater {
model: sc.metaFields
delegate: Item {
required property var modelData
Layout.columnSpan: 2
Layout.fillWidth: true
implicitHeight: 32
RowLayout {
anchors.fill: parent
spacing: 10
Text {
Layout.preferredWidth: 120
text: modelData.label + (modelData.required ? " *" : "")
color: theme.fg
font.pixelSize: 13
elide: Text.ElideRight
ToolTip.text: modelData.description || modelData.label
ToolTip.visible: metaHover.hovered
ToolTip.delay: 500
HoverHandler { id: metaHover }
}
FieldEditor {
spec: ({ kind: modelData.kind, label: modelData.label, tooltip: modelData.description, placeholder: modelData.placeholder,
minimum: -2147483647, maximum: 2147483647, decimals: 6 })
value: sc.metaValues[modelData.name]
error: sc.metaErrors[modelData.name] || ""
onEdited: (v) => sc.setMeta(modelData.name, v)
}
}
}
}
}
// Free key/value entries
ColumnLayout {
Layout.fillWidth: true
visible: !root.opts.hide_optional_metadata
spacing: 6
Text {
text: "Additional entries"
color: theme.muted
font.pixelSize: 12
Layout.topMargin: 4
}
Repeater {
id: extrasRepeater
model: sc.metaExtras
RowLayout {
required property int index
required property var modelData
Layout.fillWidth: true
spacing: 6
FieldEditor {
spec: ({ kind: "str", label: "Key", placeholder: "Key" })
value: modelData[0]
onEdited: (v) => { var e = sc.metaExtras; e[index][0] = v; sc.setExtras(e) }
}
FieldEditor {
spec: ({ kind: "str", label: "Value", placeholder: "Value" })
value: modelData[1]
onEdited: (v) => { var e = sc.metaExtras; e[index][1] = v; sc.setExtras(e) }
}
IconButton {
iconName: "close"
tint: theme.muted
tip: "Remove entry"
onClicked: { var e = sc.metaExtras; e.splice(index, 1); sc.setExtras(e) }
}
}
}
ActionButton {
kind: "ghost"
iconName: "add"
text: "Add entry"
onClicked: { var e = sc.metaExtras; e.push(["", ""]); sc.setExtras(e) }
}
}
}
Item { implicitHeight: 6 }
}
}
// ---------------------------------------------------------------- footer
Rectangle {
Layout.fillWidth: true
visible: !root.opts.hide_scan_control_buttons
implicitHeight: footer.implicitHeight + 20
color: theme.card
Rectangle { width: parent.width; height: 1; color: theme.border }
ColumnLayout {
id: footer
anchors { fill: parent; margins: 10 }
spacing: 8
RowLayout {
Layout.fillWidth: true
spacing: 6
readonly property bool showProblem: root.noticeText.length === 0 && sc.problems.length > 0
visible: root.noticeText.length > 0 || sc.problems.length > 0
Image {
source: "image://material/" + (parent.showProblem ? "error" : "check_circle") + "?filled=1&color="
+ encodeURIComponent((parent.showProblem ? theme.warning : theme.success).toString())
sourceSize: Qt.size(16, 16)
}
Text {
Layout.fillWidth: true
text: parent.showProblem
? sc.problems[0] + (sc.problems.length > 1 ? " (+" + (sc.problems.length - 1) + " more)" : "")
: root.noticeText
color: theme.fg
font.pixelSize: 12
elide: Text.ElideRight
}
}
RowLayout {
Layout.fillWidth: true
spacing: 8
ActionButton {
kind: "ghost"
iconName: "history"
text: sc.restoring ? "Restoring…" : "Restore last"
enabled: !sc.restoring && sc.currentScan.length > 0
ToolTip.text: "Fill in the parameters of the last run of this scan"
ToolTip.visible: hovered
ToolTip.delay: 500
onClicked: sc.restoreLast()
}
Item { Layout.fillWidth: true }
ActionButton {
kind: sc.stopArmed ? "danger" : "outlineDanger"
iconName: "stop"
text: sc.stopArmed ? "Halt now" : "Stop"
onClicked: sc.stop()
}
ActionButton {
kind: "primary"
iconName: "play_arrow"
text: "Start " + sc.currentScan
enabled: sc.problems.length === 0
Layout.maximumWidth: 220
onClicked: sc.start()
}
}
}
}
}
}
@@ -0,0 +1,185 @@
"""Small helpers for hosting QML views inside BEC widgets.
Provides a theme object that mirrors the active ``bec_qthemes`` colours, an image provider for
Material icons (``image://material/<name>?color=<hex>&filled=1``) and a factory for a configured
``QQuickWidget``.
"""
from __future__ import annotations
from pathlib import Path
from urllib.parse import parse_qs, urlsplit
from bec_qthemes import material_icon
from qtpy.QtCore import Property, QObject, QSize, QUrl, Signal
from qtpy.QtGui import QColor, QPixmap
from qtpy.QtQuick import QQuickImageProvider
from qtpy.QtQuickWidgets import QQuickWidget
from qtpy.QtWidgets import QApplication, QWidget
_FALLBACK = {
"dark": {
"PRIMARY": "#6089ef",
"ON_PRIMARY": "#ffffff",
"BG": "#1e1f22",
"CARD_BG": "#2b2c2f",
"FIELD_BG": "#26272a",
"FG": "#dddddd",
"BORDER": "#464749",
"ACCENT_WARNING": "#EDA200",
"ACCENT_EMERGENCY": "#E0534B",
"ACCENT_SUCCESS": "#59A869",
},
"light": {
"PRIMARY": "#3b82f6",
"ON_PRIMARY": "#ffffff",
"BG": "#f6f7fb",
"CARD_BG": "#ffffff",
"FIELD_BG": "#ffffff",
"FG": "#151924",
"BORDER": "#d9dde6",
"ACCENT_WARNING": "#EAC435",
"ACCENT_EMERGENCY": "#CC181E",
"ACCENT_SUCCESS": "#2CA58D",
},
}
def _app_theme():
app = QApplication.instance()
return getattr(app, "theme", None) if app is not None else None
def theme_colors() -> tuple[str, dict[str, QColor]]:
"""Name and colours of the active theme, falling back to built-in defaults."""
theme = _app_theme()
name = getattr(theme, "theme", "dark") if theme is not None else "dark"
name = "light" if name == "light" else "dark"
colors = {k: QColor(v) for k, v in _FALLBACK[name].items()}
if theme is not None:
for key in colors:
colors[key] = theme.color(key, colors[key].name())
return name, colors
def _mix(a: QColor, b: QColor, t: float) -> QColor:
return QColor(
round(a.red() + (b.red() - a.red()) * t),
round(a.green() + (b.green() - a.green()) * t),
round(a.blue() + (b.blue() - a.blue()) * t),
)
class QmlTheme(QObject):
"""Theme colours for QML, updated when the BEC theme changes."""
changed = Signal()
def __init__(self, parent=None):
super().__init__(parent)
self._name = "dark"
self._colors: dict[str, QColor] = {}
self.refresh()
theme = _app_theme()
if theme is not None:
theme.theme_changed.connect(self.refresh)
def refresh(self, *_args) -> None:
"""Re-read the colours of the active theme."""
self._name, self._colors = theme_colors()
self.changed.emit()
def _c(self, key: str) -> QColor:
return self._colors[key]
@Property(bool, notify=changed)
def dark(self) -> bool:
return self._name == "dark"
@Property(QColor, notify=changed)
def bg(self) -> QColor:
return self._c("BG")
@Property(QColor, notify=changed)
def card(self) -> QColor:
return self._c("CARD_BG")
@Property(QColor, notify=changed)
def field(self) -> QColor:
return self._c("FIELD_BG")
@Property(QColor, notify=changed)
def fg(self) -> QColor:
return self._c("FG")
@Property(QColor, notify=changed)
def muted(self) -> QColor:
return _mix(self._c("FG"), self._c("CARD_BG"), 0.4)
@Property(QColor, notify=changed)
def border(self) -> QColor:
return self._c("BORDER")
@Property(QColor, notify=changed)
def hover(self) -> QColor:
return _mix(self._c("CARD_BG"), self._c("FG"), 0.08)
@Property(QColor, notify=changed)
def primary(self) -> QColor:
return self._c("PRIMARY")
@Property(QColor, notify=changed)
def onPrimary(self) -> QColor: # pylint: disable=invalid-name
return self._c("ON_PRIMARY")
@Property(QColor, notify=changed)
def success(self) -> QColor:
return self._c("ACCENT_SUCCESS")
@Property(QColor, notify=changed)
def warning(self) -> QColor:
return self._c("ACCENT_WARNING")
@Property(QColor, notify=changed)
def danger(self) -> QColor:
return self._c("ACCENT_EMERGENCY")
class MaterialIconProvider(QQuickImageProvider):
"""Serves ``image://material/<icon>?color=%23rrggbb&filled=1`` from bec_qthemes."""
def __init__(self):
super().__init__(QQuickImageProvider.ImageType.Pixmap)
def requestPixmap(self, icon_id: str, size: QSize, requested_size: QSize) -> QPixmap:
# pylint: disable=invalid-name
parts = urlsplit(icon_id)
query = {k: v[-1] for k, v in parse_qs(parts.query).items()}
side = max(requested_size.width(), requested_size.height(), 0) or 24
pixmap = material_icon(
parts.path,
size=(side, side),
color=query.get("color"),
filled=query.get("filled") == "1",
)
if size is not None:
size.setWidth(pixmap.width())
size.setHeight(pixmap.height())
return pixmap
def create_quick_widget(
parent: QWidget, qml_file: Path, context: dict[str, QObject]
) -> QQuickWidget:
"""Create a ``QQuickWidget`` that sizes its root item to the widget and loads ``qml_file``."""
view = QQuickWidget(parent)
view.setResizeMode(QQuickWidget.ResizeMode.SizeRootObjectToView)
engine = view.engine()
engine.addImageProvider("material", MaterialIconProvider())
for name, obj in context.items():
view.rootContext().setContextProperty(name, obj)
view.setSource(QUrl.fromLocalFile(str(qml_file)))
if view.status() == QQuickWidget.Status.Error:
errors = "\n".join(error.toString() for error in view.errors())
raise RuntimeError(f"Failed to load {qml_file.name}:\n{errors}")
return view
@@ -0,0 +1,250 @@
"""Scan control rendered in QML, hosted in a ``QQuickWidget``."""
from __future__ import annotations
from pathlib import Path
from qtpy.QtCore import Property, QObject, QUrl, Signal, Slot
from qtpy.QtWidgets import QApplication, QVBoxLayout
from bec_widgets.widgets.control.scan_control.modern.base import ModernScanControlBase
from bec_widgets.widgets.control.scan_control.modern.qml_support import (
QmlTheme,
create_quick_widget,
)
QML_DIR = Path(__file__).parent / "qml"
class ScanControlBridge(QObject): # pylint: disable=too-many-public-methods
"""Exposes the scan control state to QML as plain lists and maps.
The accessors below are QML properties and slots named after their QML use.
"""
# pylint: disable=missing-function-docstring
structureChanged = Signal()
valuesChanged = Signal()
statusChanged = Signal()
metadataStructureChanged = Signal()
metadataChanged = Signal()
notice = Signal(str)
def __init__(self, control: ModernScanControlBase):
super().__init__(control)
self._c = control
control.form.rebuilt.connect(self.structureChanged)
control.form.values_changed.connect(self.valuesChanged)
control.status_changed.connect(self.statusChanged)
control.metadata.rebuilt.connect(self.metadataStructureChanged)
control.metadata.values_changed.connect(self.metadataChanged)
control.notice.connect(self.notice)
# ------------------------------------------------------------------ scan list
@Property("QStringList", notify=statusChanged)
def scans(self):
return self._c.visible_scans
@Property(str, notify=structureChanged)
def currentScan(self): # pylint: disable=invalid-name
return self._c.current_scan
@Property(str, notify=structureChanged)
def summary(self):
return self._c.form.spec.summary
@Slot(str)
def selectScan(self, name: str): # pylint: disable=invalid-name
self._c.set_current_scan(name)
@Slot(str, result=bool)
def isValidScan(self, name: str) -> bool: # pylint: disable=invalid-name
return self._c.is_valid_scan(name)
@Slot()
def showInfo(self): # pylint: disable=invalid-name
self._c.show_selected_scan_info()
@Slot()
def showFilter(self): # pylint: disable=invalid-name
self._c.show_scan_selector_settings()
# ------------------------------------------------------------------ arguments
@Property(str, notify=structureChanged)
def argTitle(self): # pylint: disable=invalid-name
return self._c.form.spec.arg_title
@Property("QVariantList", notify=structureChanged)
def argFields(self): # pylint: disable=invalid-name
return [f.to_dict() for f in self._c.form.spec.arg_fields]
@Property(int, notify=structureChanged)
def rowCount(self): # pylint: disable=invalid-name
return len(self._c.form.arg_rows)
@Property("QVariantList", notify=valuesChanged)
def argRows(self): # pylint: disable=invalid-name
form = self._c.form
rows = []
for index, row in enumerate(form.arg_rows):
rows.append(
{
"values": dict(row),
"units": {f.name: form.units_for(f, index) for f in form.spec.arg_fields},
"errors": {
f.name: form.field_error(f, row.get(f.name)) for f in form.spec.arg_fields
},
}
)
return rows
@Property(bool, notify=valuesChanged)
def canAddRow(self): # pylint: disable=invalid-name
return self._c.form.can_add_row()
@Property(bool, notify=valuesChanged)
def canRemoveRow(self): # pylint: disable=invalid-name
return self._c.form.can_remove_row()
@Slot(int, str, "QVariant")
def setArg(self, row: int, name: str, value): # pylint: disable=invalid-name
self._c.form.set_arg_value(row, name, value)
@Slot()
def addRow(self): # pylint: disable=invalid-name
self._c.form.add_row()
@Slot(int)
def removeRow(self, row: int): # pylint: disable=invalid-name
self._c.form.remove_row(row)
# ------------------------------------------------------------------ keyword groups
@Property("QVariantList", notify=structureChanged)
def groups(self):
return [
{"name": g.name, "fields": [f.to_dict() for f in g.fields]}
for g in self._c.form.spec.groups
]
@Property("QVariantMap", notify=valuesChanged)
def kwargs(self):
form = self._c.form
return {
"values": dict(form.kwargs),
"units": {f.name: form.units_for(f) for f in form.spec.kwarg_fields},
"errors": {
f.name: form.field_error(f, form.kwargs.get(f.name)) for f in form.spec.kwarg_fields
},
}
@Slot(str, "QVariant")
def setKwarg(self, name: str, value): # pylint: disable=invalid-name
self._c.form.set_kwarg_value(name, value)
@Property("QStringList", notify=statusChanged)
def devices(self):
return self._c.form.device_names
@Slot()
def refreshDevices(self): # pylint: disable=invalid-name
self._c.refresh_devices()
self.statusChanged.emit()
# ------------------------------------------------------------------ metadata
@Property("QVariantList", notify=metadataStructureChanged)
def metaFields(self): # pylint: disable=invalid-name
return [f.to_dict(self._c.metadata.scan_name) for f in self._c.metadata.fields]
@Property("QVariantMap", notify=metadataChanged)
def metaValues(self): # pylint: disable=invalid-name
return {k: ("" if v is None else v) for k, v in self._c.metadata.values.items()}
@Property("QVariantMap", notify=metadataChanged)
def metaErrors(self): # pylint: disable=invalid-name
return dict(self._c.metadata.errors)
@Property("QVariantList", notify=metadataChanged)
def metaExtras(self): # pylint: disable=invalid-name
return [list(pair) for pair in self._c.metadata.extras]
@Property(str, notify=metadataChanged)
def metaSummary(self): # pylint: disable=invalid-name
return self._c.metadata.summary()
@Slot(str, "QVariant")
def setMeta(self, name: str, value): # pylint: disable=invalid-name
self._c.metadata.set_value(name, value)
@Slot("QVariantList")
def setExtras(self, extras): # pylint: disable=invalid-name
self._c.metadata.set_extras([list(pair) for pair in extras])
# ------------------------------------------------------------------ status and actions
@Property("QStringList", notify=statusChanged)
def problems(self):
return self._c.problems()
@Property(bool, notify=statusChanged)
def restoring(self):
return self._c.is_restoring
@Property(bool, notify=statusChanged)
def stopArmed(self): # pylint: disable=invalid-name
return self._c.stop_armed
@Property("QVariantMap", notify=statusChanged)
def options(self):
return dict(self._c._options) # pylint: disable=protected-access
@Slot()
def start(self):
self._c.run_scan()
@Slot()
def stop(self):
self._c.stop_scan()
@Slot()
def restoreLast(self): # pylint: disable=invalid-name
self._c.request_last_executed_scan_parameters()
class ScanControlQml(ModernScanControlBase):
"""Widget to submit new scans to the queue, drawn in QML."""
ICON_NAME = "tune"
def _build_view(self) -> None:
self.bridge = ScanControlBridge(self)
self.qml_theme = QmlTheme(self)
layout = QVBoxLayout(self)
layout.setContentsMargins(0, 0, 0, 0)
self.view = create_quick_widget(
self, QML_DIR / "ScanControl.qml", {"sc": self.bridge, "theme": self.qml_theme}
)
layout.addWidget(self.view)
self.setMinimumSize(360, 420)
def cleanup(self):
"""Unload the QML scene before the objects it binds to are deleted."""
self.view.setSource(QUrl())
super().cleanup()
if __name__ == "__main__": # pragma: no cover
import sys
from bec_widgets.utils.colors import apply_theme
app = QApplication(sys.argv)
apply_theme("dark")
widget = ScanControlQml()
widget.resize(520, 760)
widget.show()
sys.exit(app.exec())
@@ -0,0 +1,906 @@
"""Scan control built from QWidgets with the same layout and behaviour as the QML version."""
from __future__ import annotations
from bec_qthemes import material_icon
from qtpy.QtCore import QLocale, QSize, Qt, QTimer, Signal
from qtpy.QtGui import QDoubleValidator, QIntValidator
from qtpy.QtWidgets import (
QApplication,
QComboBox,
QCompleter,
QFrame,
QGridLayout,
QHBoxLayout,
QLabel,
QLineEdit,
QPushButton,
QScrollArea,
QSizePolicy,
QToolButton,
QVBoxLayout,
QWidget,
)
from bec_widgets.utils.colors import rgba
from bec_widgets.widgets.control.scan_control.modern.base import ModernScanControlBase
from bec_widgets.widgets.control.scan_control.modern.form_core import FieldSpec
from bec_widgets.widgets.control.scan_control.modern.qml_support import theme_colors
from bec_widgets.widgets.utility.toggle.toggle import ToggleSwitch
WIDE_LAYOUT_WIDTH = 620
def _icon(name: str, color, filled: bool = False):
return material_icon(name, size=(18, 18), color=color, filled=filled, convert_to_pixmap=False)
def _repolish(widget: QWidget) -> None:
widget.style().unpolish(widget)
widget.style().polish(widget)
# ---------------------------------------------------------------------- editors
class _EditorMixin:
"""Common interface of all editors: ``edited`` signal plus value/units/error setters."""
def set_value(self, value) -> None: # pragma: no cover - interface
raise NotImplementedError
def set_units(self, units: str) -> None:
"""Units are only shown by numeric editors."""
def set_error(self, error: str) -> None:
"""Mark the editor invalid and show the reason as tooltip."""
invalid = bool(error)
if self.property("invalid") != invalid:
self.setProperty("invalid", invalid)
_repolish(self)
self.setToolTip(error or getattr(self, "_tooltip", ""))
class NumberEdit(_EditorMixin, QLineEdit):
"""Free-text numeric input with units inside the box and arrow-key stepping."""
edited = Signal(object)
def __init__(self, spec: FieldSpec | dict, parent=None):
super().__init__(parent)
self.setObjectName("scField")
get = spec.get if isinstance(spec, dict) else lambda k, d=None: getattr(spec, k, d)
self._is_int = get("kind") == "int"
self._decimals = get("decimals", 6) or 0
self._min = get("minimum")
self._max = get("maximum")
lo = -2147483647 if self._min is None else self._min
hi = 2147483647 if self._max is None else self._max
if self._is_int:
validator = QIntValidator(int(lo), int(hi), self)
else:
validator = QDoubleValidator(lo, hi, self._decimals, self)
validator.setNotation(QDoubleValidator.Notation.StandardNotation)
validator.setLocale(QLocale.c())
self.setValidator(validator)
self.setAlignment(Qt.AlignmentFlag.AlignRight | Qt.AlignmentFlag.AlignVCenter)
self._tooltip = get("tooltip") or ""
self.setToolTip(self._tooltip)
self._unit = QLabel(self)
self._unit.setObjectName("scMuted")
self._unit.setAttribute(Qt.WidgetAttribute.WA_TransparentForMouseEvents)
self._value = None
self.editingFinished.connect(self._commit)
def _format(self, value) -> str:
if value is None or value == "":
return ""
if self._is_int:
return str(int(value))
return f"{float(value):.{self._decimals}f}".rstrip("0").rstrip(".") or "0"
def set_value(self, value) -> None:
self._value = value
if not self.hasFocus():
self.setText(self._format(value))
def set_units(self, units: str) -> None:
self._unit.setText(units)
self._unit.adjustSize()
self.setTextMargins(0, 0, self._unit.width() + 6 if units else 0, 0)
self._place_unit()
def _place_unit(self) -> None:
self._unit.move(
self.width() - self._unit.width() - 10, (self.height() - self._unit.height()) // 2
)
def resizeEvent(self, event): # pylint: disable=invalid-name
super().resizeEvent(event)
self._place_unit()
def _commit(self) -> None:
if not self.hasAcceptableInput():
self.setText(self._format(self._value))
return
text = self.text()
self.edited.emit(int(text) if self._is_int else float(text))
def keyPressEvent(self, event): # pylint: disable=invalid-name
step = {Qt.Key.Key_Up: 1, Qt.Key.Key_Down: -1}.get(event.key())
if step is None:
super().keyPressEvent(event)
return
try:
value = float(self.text() or 0)
except ValueError:
value = 0.0
value += step * (1 if self._is_int else 10 ** -min(self._decimals, 2))
if self._min is not None:
value = max(value, self._min)
if self._max is not None:
value = min(value, self._max)
self.setText(self._format(round(value, self._decimals)))
self._commit()
class TextEdit(_EditorMixin, QLineEdit):
"""Plain text input."""
edited = Signal(object)
def __init__(self, tooltip: str = "", placeholder: str = "", parent=None):
super().__init__(parent)
self.setObjectName("scField")
self._tooltip = tooltip
self.setToolTip(tooltip)
self.setPlaceholderText(placeholder)
self.editingFinished.connect(lambda: self.edited.emit(self.text()))
def set_value(self, value) -> None:
if not self.hasFocus():
self.setText("" if value is None else str(value))
class DeviceEdit(_EditorMixin, QComboBox):
"""Editable device picker with substring completion."""
edited = Signal(object)
def __init__(self, tooltip: str = "", parent=None):
super().__init__(parent)
self.setObjectName("scField")
self.setEditable(True)
self.setInsertPolicy(QComboBox.InsertPolicy.NoInsert)
self.lineEdit().setPlaceholderText("Device")
completer = QCompleter(self.model(), self)
completer.setCaseSensitivity(Qt.CaseSensitivity.CaseInsensitive)
completer.setFilterMode(Qt.MatchFlag.MatchContains)
self.setCompleter(completer)
self._tooltip = tooltip
self.setToolTip(tooltip)
self._value = ""
self.activated.connect(lambda _i: self._commit())
self.lineEdit().editingFinished.connect(self._commit)
def set_devices(self, devices: list[str]) -> None:
"""Replace the offered devices, keeping the typed text."""
if [self.itemText(i) for i in range(self.count())] == devices:
return
text = self.currentText()
self.blockSignals(True)
self.clear()
self.addItems(devices)
self.setCurrentIndex(-1)
self.setEditText(text)
self.blockSignals(False)
def set_value(self, value) -> None:
self._value = value or ""
if not self.lineEdit().hasFocus():
self.setEditText(self._value)
def set_error(self, error: str) -> None:
# An empty field is explained in the footer; only flag a wrong name in red.
super().set_error(error if self._value else "")
def _commit(self) -> None:
if self.currentText() != self._value:
self.edited.emit(self.currentText())
class ChoiceEdit(_EditorMixin, QComboBox):
"""Drop-down for ``Literal`` arguments; an empty choice is shown as "None"."""
edited = Signal(object)
def __init__(self, choices: list[str], tooltip: str = "", parent=None):
super().__init__(parent)
self.setObjectName("scField")
for choice in choices:
self.addItem(choice or "None", choice)
self._tooltip = tooltip
self.setToolTip(tooltip)
self.activated.connect(lambda i: self.edited.emit(self.itemData(i)))
def set_value(self, value) -> None:
index = self.findData("" if value is None else str(value))
self.setCurrentIndex(max(index, 0))
class BoolEdit(_EditorMixin, QWidget):
"""Toggle switch with an On/Off caption."""
edited = Signal(object)
def __init__(self, tooltip: str = "", parent=None):
super().__init__(parent)
layout = QHBoxLayout(self)
layout.setContentsMargins(0, 0, 0, 0)
self.toggle = ToggleSwitch(self, checked=False)
self.toggle.setFixedSize(38, 22)
self.caption = QLabel("Off", self)
self.caption.setObjectName("scMuted")
layout.addWidget(self.toggle)
layout.addWidget(self.caption)
layout.addStretch(1)
self.setToolTip(tooltip)
self.toggle.stateChanged.connect(self._on_toggled)
def _on_toggled(self, checked: bool) -> None:
self.caption.setText("On" if checked else "Off")
self.edited.emit(checked)
def set_value(self, value) -> None:
self.toggle.blockSignals(True)
self.toggle.setChecked(bool(value))
self.toggle.blockSignals(False)
self.caption.setText("On" if value else "Off")
def make_editor(spec: FieldSpec, parent: QWidget) -> QWidget:
"""Create the editor for one scan field."""
if spec.kind == "device":
return DeviceEdit(spec.tooltip, parent)
if spec.kind in ("int", "float"):
return NumberEdit(spec, parent)
if spec.kind == "bool":
return BoolEdit(spec.tooltip, parent)
if spec.kind == "choice":
return ChoiceEdit(spec.choices, spec.tooltip, parent)
return TextEdit(spec.tooltip, parent=parent)
def _clear_layout(layout) -> None:
while layout.count():
item = layout.takeAt(0)
widget = item.widget()
if widget is not None:
widget.hide()
widget.deleteLater()
elif item.layout() is not None:
_clear_layout(item.layout())
# ---------------------------------------------------------------------- building blocks
class Card(QFrame):
"""Titled section; optionally collapsible by clicking its header."""
def __init__(self, title: str = "", collapsible: bool = False, parent=None):
super().__init__(parent)
self.setObjectName("scCard")
self._collapsible = collapsible
self._expanded = True
outer = QVBoxLayout(self)
outer.setContentsMargins(12, 12, 12, 12)
outer.setSpacing(10)
header = QHBoxLayout()
header.setSpacing(6)
self.chevron = QLabel(self)
self.chevron.setVisible(collapsible)
self.title = QLabel(title.upper(), self)
self.title.setObjectName("scCardTitle")
self.subtitle = QLabel(self)
self.subtitle.setObjectName("scMuted")
self.subtitle.setSizePolicy(QSizePolicy.Policy.Ignored, QSizePolicy.Policy.Preferred)
self.trailing = QHBoxLayout()
header.addWidget(self.chevron)
header.addWidget(self.title)
header.addWidget(self.subtitle, 1)
header.addLayout(self.trailing)
outer.addLayout(header)
self.body_widget = QWidget(self)
self.body = QVBoxLayout(self.body_widget)
self.body.setContentsMargins(0, 0, 0, 0)
self.body.setSpacing(8)
outer.addWidget(self.body_widget)
if collapsible:
self.setCursor(Qt.CursorShape.PointingHandCursor)
self.body_widget.setCursor(Qt.CursorShape.ArrowCursor)
def set_title(self, title: str) -> None:
"""Set the header text."""
self.title.setText(title.upper())
def set_expanded(self, expanded: bool) -> None:
"""Show or hide the body."""
self._expanded = expanded
self.body_widget.setVisible(expanded)
self.update_chevron()
def update_chevron(self) -> None:
"""Redraw the chevron in the current theme colour."""
_, colors = theme_colors()
muted = _muted(colors)
name = "expand_more" if self._expanded else "chevron_right"
self.chevron.setPixmap(_icon(name, muted).pixmap(18, 18))
def mousePressEvent(self, event): # pylint: disable=invalid-name
if self._collapsible and event.position().y() < 40:
self.set_expanded(not self._expanded)
return
super().mousePressEvent(event)
def _muted(colors):
fg, card = colors["FG"], colors["CARD_BG"]
return _mix_hex(fg, card, 0.4)
def _hover(colors):
fg, card = colors["FG"], colors["CARD_BG"]
return _mix_hex(card, fg, 0.08)
def _mix_hex(a, b, t: float) -> str:
r, g, bl = (round(x + (y - x) * t) for x, y in zip(a.getRgb()[:3], b.getRgb()[:3]))
return f"#{r:02x}{g:02x}{bl:02x}"
def _icon_button(parent, tip: str) -> QToolButton:
button = QToolButton(parent)
button.setObjectName("scIcon")
button.setToolTip(tip)
button.setAccessibleName(tip)
button.setIconSize(QSize(18, 18))
button.setCursor(Qt.CursorShape.PointingHandCursor)
return button
def _button(text: str, name: str, parent) -> QPushButton:
button = QPushButton(text, parent)
button.setObjectName(name)
button.setIconSize(QSize(18, 18))
button.setCursor(Qt.CursorShape.PointingHandCursor)
return button
# ---------------------------------------------------------------------- the widget
class ScanControlModern(ModernScanControlBase):
"""Widget to submit new scans to the queue, built from QWidgets."""
# The view is created in _build_view, which ModernScanControlBase.__init__ calls.
# pylint: disable=attribute-defined-outside-init,too-many-instance-attributes
ICON_NAME = "tune"
def _build_view(self) -> None: # pylint: disable=too-many-statements
self.setObjectName("scRoot")
self.setAttribute(Qt.WidgetAttribute.WA_StyledBackground, True)
self._wide = False
self._arg_editors: list[dict[str, QWidget]] = []
self._kwarg_editors: dict[str, QWidget] = {}
self._meta_editors: dict[str, QWidget] = {}
self._expert_shown: dict[str, bool] = {}
self._notice_timer = QTimer(self)
self._notice_timer.setSingleShot(True)
self._notice_timer.setInterval(4000)
self._notice_timer.timeout.connect(self._refresh_status)
self._notice_text = ""
root = QVBoxLayout(self)
root.setContentsMargins(0, 0, 0, 0)
root.setSpacing(0)
root.addWidget(self._build_header())
self.scroll = QScrollArea(self)
self.scroll.setObjectName("scScroll")
self.scroll.setWidgetResizable(True)
self.scroll.setFrameShape(QFrame.Shape.NoFrame)
self.scroll.setHorizontalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAlwaysOff)
content = QWidget()
content.setObjectName("scContent")
self.content_layout = QVBoxLayout(content)
self.content_layout.setContentsMargins(10, 12, 10, 12)
self.content_layout.setSpacing(10)
self.empty_label = QLabel(
"No scans available. Check the scan filter or the BEC server.", content
)
self.empty_label.setObjectName("scMuted")
self.empty_label.setWordWrap(True)
self.empty_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
self.content_layout.addWidget(self.empty_label)
self.arg_card = Card(parent=content)
self.arg_grid = QGridLayout()
self.arg_grid.setHorizontalSpacing(6)
self.arg_grid.setVerticalSpacing(8)
self.arg_card.body.addLayout(self.arg_grid)
self.add_row_button = _button("Add row", "scGhost", self.arg_card)
self.add_row_button.clicked.connect(self.form.add_row)
self.arg_card.body.addWidget(self.add_row_button, 0, Qt.AlignmentFlag.AlignLeft)
self.content_layout.addWidget(self.arg_card)
self.groups_layout = QVBoxLayout()
self.groups_layout.setSpacing(10)
self.content_layout.addLayout(self.groups_layout)
self.content_layout.addWidget(self._build_metadata_card(content))
self.content_layout.addStretch(1)
self.scroll.setWidget(content)
root.addWidget(self.scroll, 1)
root.addWidget(self._build_footer())
self.form.rebuilt.connect(self._rebuild_form)
self.form.values_changed.connect(self._refresh_values)
self.metadata.rebuilt.connect(self._rebuild_metadata)
self.metadata.values_changed.connect(self._refresh_metadata)
self.status_changed.connect(self._refresh_status)
self.notice.connect(self._show_notice)
self.setMinimumSize(360, 420)
self.apply_theme("")
def _build_header(self) -> QWidget:
self.header = QFrame(self)
self.header.setObjectName("scHeader")
layout = QVBoxLayout(self.header)
layout.setContentsMargins(12, 12, 12, 12)
layout.setSpacing(6)
row = QHBoxLayout()
row.setSpacing(4)
self.scan_picker = QComboBox(self.header)
self.scan_picker.setObjectName("scPicker")
self.scan_picker.setEditable(True)
self.scan_picker.setInsertPolicy(QComboBox.InsertPolicy.NoInsert)
self.scan_picker.setAccessibleName("Scan")
self.scan_picker.lineEdit().setPlaceholderText("Search scans")
self._search_action = self.scan_picker.lineEdit().addAction(
_icon("search", "#888888"), QLineEdit.ActionPosition.LeadingPosition
)
completer = QCompleter(self.scan_picker.model(), self.scan_picker)
completer.setCaseSensitivity(Qt.CaseSensitivity.CaseInsensitive)
completer.setFilterMode(Qt.MatchFlag.MatchContains)
self.scan_picker.setCompleter(completer)
self.scan_picker.activated.connect(lambda _i: self._confirm_scan())
self.scan_picker.lineEdit().editingFinished.connect(self._confirm_scan)
self.scan_picker.lineEdit().textEdited.connect(self._on_scan_typed)
self.info_button = _icon_button(self.header, "Documentation of this scan")
self.info_button.clicked.connect(self.show_selected_scan_info)
self.filter_button = _icon_button(self.header, "Choose which scans are listed")
self.filter_button.clicked.connect(self.show_scan_selector_settings)
row.addWidget(self.scan_picker, 1)
row.addWidget(self.info_button)
row.addWidget(self.filter_button)
layout.addLayout(row)
self.summary_label = QLabel(self.header)
self.summary_label.setObjectName("scMuted")
self.summary_label.setWordWrap(True)
layout.addWidget(self.summary_label)
return self.header
def _build_metadata_card(self, parent) -> QWidget:
self.meta_card = Card("Metadata", collapsible=True, parent=parent)
self.meta_badge = QLabel(self.meta_card)
self.meta_badge.setObjectName("scBadge")
self.meta_card.trailing.addWidget(self.meta_badge)
self.meta_grid = QGridLayout()
self.meta_grid.setHorizontalSpacing(10)
self.meta_grid.setVerticalSpacing(8)
self.meta_card.body.addLayout(self.meta_grid)
self.extras_widget = QWidget(self.meta_card)
extras_layout = QVBoxLayout(self.extras_widget)
extras_layout.setContentsMargins(0, 4, 0, 0)
extras_layout.setSpacing(6)
title = QLabel("Additional entries", self.extras_widget)
title.setObjectName("scMuted")
extras_layout.addWidget(title)
self.extras_rows = QVBoxLayout()
self.extras_rows.setSpacing(6)
extras_layout.addLayout(self.extras_rows)
self.add_extra_button = _button("Add entry", "scGhost", self.extras_widget)
self.add_extra_button.clicked.connect(
lambda: self.metadata.set_extras(self.metadata.extras + [["", ""]])
)
extras_layout.addWidget(self.add_extra_button, 0, Qt.AlignmentFlag.AlignLeft)
self.meta_card.body.addWidget(self.extras_widget)
self.meta_card.set_expanded(False)
return self.meta_card
def _build_footer(self) -> QWidget:
self.footer = QFrame(self)
self.footer.setObjectName("scFooter")
layout = QVBoxLayout(self.footer)
layout.setContentsMargins(10, 10, 10, 10)
layout.setSpacing(8)
message = QHBoxLayout()
message.setSpacing(6)
self.message_icon = QLabel(self.footer)
self.message_label = QLabel(self.footer)
self.message_label.setObjectName("scMessage")
self.message_label.setSizePolicy(QSizePolicy.Policy.Ignored, QSizePolicy.Policy.Preferred)
message.addWidget(self.message_icon)
message.addWidget(self.message_label, 1)
layout.addLayout(message)
buttons = QHBoxLayout()
buttons.setSpacing(8)
self.restore_button = _button("Restore last", "scGhost", self.footer)
self.restore_button.setToolTip("Fill in the parameters of the last run of this scan")
self.restore_button.clicked.connect(self.request_last_executed_scan_parameters)
self.stop_button = _button("Stop", "scStop", self.footer)
self.stop_button.clicked.connect(self.stop_scan)
self.start_button = _button("Start", "scPrimary", self.footer)
self.start_button.clicked.connect(self.run_scan)
self.start_button.setMaximumWidth(220)
buttons.addWidget(self.restore_button)
buttons.addStretch(1)
buttons.addWidget(self.stop_button)
buttons.addWidget(self.start_button)
layout.addLayout(buttons)
return self.footer
# ------------------------------------------------------------------ scan picker
def _on_scan_typed(self, text: str) -> None:
self._set_invalid(self.scan_picker, bool(text) and not self.is_valid_scan(text))
def _confirm_scan(self) -> None:
text = self.scan_picker.currentText()
if self.is_valid_scan(text):
self.set_current_scan(text)
self._sync_picker()
def _sync_picker(self) -> None:
items = [self.scan_picker.itemText(i) for i in range(self.scan_picker.count())]
self.scan_picker.blockSignals(True)
if items != self.visible_scans:
self.scan_picker.clear()
self.scan_picker.addItems(self.visible_scans)
self.scan_picker.setCurrentIndex(self.scan_picker.findText(self.current_scan))
self.scan_picker.setEditText(self.current_scan)
self.scan_picker.blockSignals(False)
self._set_invalid(self.scan_picker, False)
@staticmethod
def _set_invalid(widget: QWidget, invalid: bool) -> None:
if widget.property("invalid") != invalid:
widget.setProperty("invalid", invalid)
_repolish(widget)
# ------------------------------------------------------------------ form
def _rebuild_form(self) -> None:
spec = self.form.spec
self.summary_label.setText(spec.summary)
self.summary_label.setVisible(bool(spec.summary))
_clear_layout(self.arg_grid)
self._arg_editors = []
self.arg_card.set_title(spec.arg_title)
rows = len(self.form.arg_rows)
self.arg_card.subtitle.setText(f"{rows} rows" if rows > 1 else "")
for column, field in enumerate(spec.arg_fields, start=1):
heading = QLabel(field.label, self.arg_card)
heading.setObjectName("scMuted")
self.arg_grid.addWidget(heading, 0, column)
self.arg_grid.setColumnStretch(column, 14 if field.kind == "device" else 9)
self.arg_grid.setColumnMinimumWidth(0, 22)
for row in range(rows):
number = QLabel(str(row + 1), self.arg_card)
number.setObjectName("scMuted")
number.setAlignment(Qt.AlignmentFlag.AlignCenter)
self.arg_grid.addWidget(number, row + 1, 0)
editors = {}
for column, field in enumerate(spec.arg_fields, start=1):
editor = make_editor(field, self.arg_card)
editor.edited.connect(
lambda value, r=row, n=field.name: self.form.set_arg_value(r, n, value)
)
self.arg_grid.addWidget(editor, row + 1, column)
editors[field.name] = editor
remove = _icon_button(self.arg_card, f"Remove row {row + 1}")
remove.setIcon(_icon("close", self._colors["muted"]))
remove.clicked.connect(lambda _=False, r=row: self.form.remove_row(r))
remove.setProperty("remove_row", True)
policy = remove.sizePolicy()
policy.setRetainSizeWhenHidden(True)
remove.setSizePolicy(policy)
self.arg_grid.addWidget(remove, row + 1, len(spec.arg_fields) + 1)
self._arg_editors.append(editors)
self._rebuild_groups()
self._refresh_values()
self._refresh_status()
def _rebuild_groups(self) -> None:
_clear_layout(self.groups_layout)
self._kwarg_editors = {}
columns = 2 if self._wide else 1
for group in self.form.spec.groups:
card = Card(group.name, parent=self.scroll.widget())
grid = QGridLayout()
grid.setHorizontalSpacing(10)
grid.setVerticalSpacing(8)
show_expert = self._expert_shown.get(group.name, False)
fields = [f for f in group.fields if not f.expert or show_expert]
for index, field in enumerate(fields):
row, column = divmod(index, columns)
label = QLabel(field.label, card)
label.setToolTip(field.tooltip or field.label)
label.setFixedWidth(120)
editor = make_editor(field, card)
editor.edited.connect(
lambda value, n=field.name: self.form.set_kwarg_value(n, value)
)
grid.addWidget(label, row, column * 2)
grid.addWidget(editor, row, column * 2 + 1)
grid.setColumnStretch(column * 2 + 1, 1)
self._kwarg_editors[field.name] = editor
card.body.addLayout(grid)
expert_count = sum(f.expert for f in group.fields)
if expert_count:
toggle = _button(
f"{'Hide' if show_expert else 'Show'} advanced ({expert_count})",
"scGhost",
card,
)
toggle.setIcon(
_icon("expand_less" if show_expert else "expand_more", self._colors["fg"])
)
toggle.clicked.connect(lambda _=False, g=group.name: self._toggle_expert(g))
card.body.addWidget(toggle, 0, Qt.AlignmentFlag.AlignLeft)
card.setVisible(not self.hide_kwarg_boxes)
self.groups_layout.addWidget(card)
def _toggle_expert(self, group: str) -> None:
self._expert_shown[group] = not self._expert_shown.get(group, False)
self._rebuild_groups()
self._refresh_values()
def _refresh_values(self) -> None:
form = self.form
for row, editors in enumerate(self._arg_editors):
if row >= len(form.arg_rows):
break
values = form.arg_rows[row]
for field in form.spec.arg_fields:
editor = editors[field.name]
if isinstance(editor, DeviceEdit):
editor.set_devices(form.device_names)
editor.set_value(values.get(field.name))
editor.set_units(form.units_for(field, row))
editor.set_error(form.field_error(field, values.get(field.name)))
for field in form.spec.kwarg_fields:
editor = self._kwarg_editors.get(field.name)
if editor is None:
continue
if isinstance(editor, DeviceEdit):
editor.set_devices(form.device_names)
editor.set_value(form.kwargs.get(field.name))
editor.set_units(form.units_for(field))
editor.set_error(form.field_error(field, form.kwargs.get(field.name)))
# ------------------------------------------------------------------ metadata
def _rebuild_metadata(self) -> None:
_clear_layout(self.meta_grid)
self._meta_editors = {}
for index, field in enumerate(self.metadata.fields):
label = QLabel(field.label + (" *" if field.required else ""), self.meta_card)
label.setToolTip(field.description or field.label)
label.setFixedWidth(120)
if field.kind in ("int", "float"):
editor = NumberEdit({"kind": field.kind, "decimals": 6}, self.meta_card)
elif field.kind == "bool":
editor = BoolEdit(field.description, self.meta_card)
else:
editor = TextEdit(
field.description, field.placeholder(self.metadata.scan_name), self.meta_card
)
editor.edited.connect(lambda value, n=field.name: self.metadata.set_value(n, value))
self.meta_grid.addWidget(label, index, 0)
self.meta_grid.addWidget(editor, index, 1)
self._meta_editors[field.name] = editor
self._refresh_metadata()
def _refresh_metadata(self) -> None:
meta = self.metadata
for name, editor in self._meta_editors.items():
editor.set_value(meta.values.get(name))
editor.set_error(meta.errors.get(name, ""))
if len(meta.extras) != self.extras_rows.count():
self._rebuild_extras()
for index, (key, value) in enumerate(meta.extras):
row = self.extras_rows.itemAt(index).widget()
row.key_edit.set_value(key)
row.value_edit.set_value(value)
errors = len(meta.errors)
self.meta_badge.setText(f"{errors} to fix" if errors else "")
self.meta_badge.setVisible(bool(errors))
self.meta_card.subtitle.setText("" if errors else (meta.summary() or "optional"))
if errors:
self.meta_card.set_expanded(True)
def _rebuild_extras(self) -> None:
_clear_layout(self.extras_rows)
for index in range(len(self.metadata.extras)):
row = QWidget(self.extras_widget)
layout = QHBoxLayout(row)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(6)
row.key_edit = TextEdit(placeholder="Key", parent=row)
row.value_edit = TextEdit(placeholder="Value", parent=row)
remove = _icon_button(row, "Remove entry")
remove.setIcon(_icon("close", self._colors["muted"]))
row.key_edit.edited.connect(lambda v, i=index: self._set_extra(i, 0, v))
row.value_edit.edited.connect(lambda v, i=index: self._set_extra(i, 1, v))
remove.clicked.connect(lambda _=False, i=index: self._remove_extra(i))
layout.addWidget(row.key_edit, 1)
layout.addWidget(row.value_edit, 1)
layout.addWidget(remove)
self.extras_rows.addWidget(row)
def _set_extra(self, index: int, column: int, value: str) -> None:
extras = [list(pair) for pair in self.metadata.extras]
extras[index][column] = value
self.metadata.set_extras(extras)
def _remove_extra(self, index: int) -> None:
extras = [list(pair) for pair in self.metadata.extras]
del extras[index]
self.metadata.set_extras(extras)
# ------------------------------------------------------------------ status
def _show_notice(self, text: str) -> None:
self._notice_text = text
self._notice_timer.start()
self._refresh_status()
def _refresh_status(self) -> None:
if not self._notice_timer.isActive():
self._notice_text = ""
problems = self.problems()
if self.scan_picker.lineEdit().hasFocus() is False:
self._sync_picker()
self.header.setVisible(not self.hide_scan_selection_combobox)
self.filter_button.setVisible(not self.hide_scan_selector_settings_button)
self.info_button.setEnabled(bool(self.current_scan))
self.empty_label.setVisible(not self.visible_scans)
self.arg_card.setVisible(bool(self.form.spec.arg_fields) and not self.hide_arg_box)
for index in range(self.groups_layout.count()):
self.groups_layout.itemAt(index).widget().setVisible(not self.hide_kwarg_boxes)
self.meta_card.setVisible(not self.hide_metadata and bool(self.current_scan))
self.extras_widget.setVisible(not self.hide_optional_metadata)
self.footer.setVisible(not self.hide_scan_control_buttons)
editing = not self.hide_add_remove_buttons
self.add_row_button.setVisible(editing and self.form.can_add_row())
for index in range(self.arg_grid.count()):
widget = self.arg_grid.itemAt(index).widget()
if widget is not None and widget.property("remove_row"):
# Keep the column while rows cannot be removed, so the fields do not jump.
widget.setVisible(editing and self.form.can_remove_row())
show_problem = not self._notice_text and bool(problems)
if show_problem:
more = f" (+{len(problems) - 1} more)" if len(problems) > 1 else ""
text, icon, color = problems[0] + more, "error", self._colors["warning"]
else:
text, icon, color = self._notice_text, "check_circle", self._colors["success"]
self.message_label.setText(text)
self.message_icon.setPixmap(_icon(icon, color, filled=True).pixmap(16, 16))
self.message_icon.setVisible(bool(text))
self.message_label.setVisible(bool(text))
self.restore_button.setText("Restoring…" if self.is_restoring else "Restore last")
self.restore_button.setEnabled(not self.is_restoring and bool(self.current_scan))
armed = self.stop_armed
self.stop_button.setText("Halt now" if armed else "Stop")
if self.stop_button.property("armed") != armed:
self.stop_button.setProperty("armed", armed)
_repolish(self.stop_button)
self.stop_button.setIcon(
_icon("stop", self._colors["onPrimary"] if armed else self._colors["danger"], True)
)
metrics = self.start_button.fontMetrics()
self.start_button.setText(
metrics.elidedText(f"Start {self.current_scan}", Qt.TextElideMode.ElideRight, 170)
)
self.start_button.setEnabled(not problems)
def resizeEvent(self, event): # pylint: disable=invalid-name
super().resizeEvent(event)
wide = self.width() > WIDE_LAYOUT_WIDTH
if wide != self._wide:
self._wide = wide
self._rebuild_groups()
self._refresh_values()
# ------------------------------------------------------------------ theme
def apply_theme(self, theme: str):
"""Restyle the widget from the active BEC theme colours."""
_, colors = theme_colors()
c = {k: v.name() for k, v in colors.items()}
self._colors = {
"fg": c["FG"],
"muted": _muted(colors),
"warning": c["ACCENT_WARNING"],
"success": c["ACCENT_SUCCESS"],
"danger": c["ACCENT_EMERGENCY"],
"onPrimary": c["ON_PRIMARY"],
}
hover = _hover(colors)
muted = self._colors["muted"]
# pylint: disable=line-too-long
self.setStyleSheet(f"""
#scRoot, #scContent {{ background: {c['BG']}; }}
#scScroll {{ background: {c['BG']}; border: none; }}
#scHeader {{ background: {c['CARD_BG']}; border-bottom: 1px solid {c['BORDER']}; }}
#scFooter {{ background: {c['CARD_BG']}; border-top: 1px solid {c['BORDER']}; }}
#scCard {{ background: {c['CARD_BG']}; border: 1px solid {c['BORDER']}; border-radius: 10px; }}
#scCard QLabel, #scHeader QLabel, #scFooter QLabel {{ color: {c['FG']}; background: transparent; }}
QLabel#scCardTitle {{ color: {muted}; font-size: 11px; font-weight: 700; letter-spacing: 0.8px; }}
QLabel#scMuted {{ color: {muted}; font-size: 12px; }}
QLabel#scMessage {{ font-size: 12px; }}
QLabel#scBadge {{ color: {c['ACCENT_EMERGENCY']}; background: {rgba(c['ACCENT_EMERGENCY'], 46)};
border-radius: 9px; padding: 1px 7px; font-size: 11px; font-weight: 600; }}
#scField, QComboBox#scPicker {{ background: {c['FIELD_BG']}; color: {c['FG']};
border: 1px solid {c['BORDER']}; border-radius: 6px; padding: 4px 8px; min-height: 22px; }}
QComboBox#scPicker {{ font-size: 15px; font-weight: 600; min-height: 26px; }}
#scField:focus, QComboBox#scPicker:focus {{ border: 1.5px solid {c['PRIMARY']}; }}
#scField[invalid="true"], QComboBox#scPicker[invalid="true"] {{
border: 1.5px solid {c['ACCENT_EMERGENCY']}; }}
QComboBox#scField QLineEdit, QComboBox#scPicker QLineEdit {{ background: transparent; border: none; padding: 0; }}
QComboBox#scField::drop-down, QComboBox#scPicker::drop-down {{ border: none; width: 20px; }}
QToolButton#scIcon {{ border: none; border-radius: 6px; padding: 6px; background: transparent; }}
QToolButton#scIcon:hover {{ background: {hover}; }}
QPushButton#scGhost {{ background: transparent; color: {c['FG']}; border: none;
border-radius: 7px; padding: 7px 12px; font-weight: 600; text-align: left; }}
QPushButton#scGhost:hover {{ background: {hover}; }}
QPushButton#scGhost:disabled {{ color: {muted}; }}
QPushButton#scPrimary {{ background: {c['PRIMARY']}; color: {c['ON_PRIMARY']}; border: none;
border-radius: 7px; padding: 7px 14px; font-weight: 600; }}
QPushButton#scPrimary:hover {{ background: {colors['PRIMARY'].lighter(110).name()}; }}
QPushButton#scPrimary:disabled {{ background: {rgba(c['PRIMARY'], 100)};
color: {rgba(c['ON_PRIMARY'], 170)}; }}
QPushButton#scStop {{ background: transparent; color: {c['ACCENT_EMERGENCY']};
border: 1px solid {c['ACCENT_EMERGENCY']}; border-radius: 7px; padding: 7px 14px; font-weight: 600; }}
QPushButton#scStop:hover {{ background: {hover}; }}
QPushButton#scStop[armed="true"] {{ background: {c['ACCENT_EMERGENCY']}; color: {c['ON_PRIMARY']}; }}
""")
self._search_action.setIcon(_icon("search", muted))
self.info_button.setIcon(_icon("info", c["FG"]))
self.filter_button.setIcon(_icon("filter_list", c["FG"]))
self.add_row_button.setIcon(_icon("add", c["FG"]))
self.add_extra_button.setIcon(_icon("add", c["FG"]))
self.restore_button.setIcon(_icon("history", c["FG"]))
self.start_button.setIcon(_icon("play_arrow", c["ON_PRIMARY"], filled=True))
self.meta_card.update_chevron()
if hasattr(self, "arg_grid"):
self._rebuild_form()
self._rebuild_metadata()
if __name__ == "__main__": # pragma: no cover
import sys
from bec_widgets.utils.colors import apply_theme
app = QApplication(sys.argv)
apply_theme("dark")
demo = ScanControlModern()
demo.resize(520, 760)
demo.show()
sys.exit(app.exec())
@@ -0,0 +1,252 @@
# pylint: disable = no-name-in-module,missing-class-docstring, missing-module-docstring
# pylint: disable=missing-function-docstring,redefined-outer-name,protected-access
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
import pytest
from bec_lib.endpoints import MessageEndpoints
from qtpy.QtCore import Qt
from bec_widgets.widgets.control.scan_control.modern.form_core import FieldSpec, ScanFormSpec
from bec_widgets.widgets.control.scan_control.modern.scan_control_qml import ScanControlQml
from bec_widgets.widgets.control.scan_control.modern.scan_control_widgets import (
DeviceEdit,
NumberEdit,
ScanControlModern,
)
from .test_scan_control import available_scans_message, scan_history
@pytest.fixture(params=[ScanControlQml, ScanControlModern], ids=["qml", "widgets"])
def control(request, qtbot, mocked_client):
mocked_client.connector.set_and_publish(
MessageEndpoints.available_scans(), available_scans_message
)
mocked_client.connector.xadd(
topic=MessageEndpoints.scan_history(), msg_dict={"data": scan_history}
)
widget = request.param(client=mocked_client)
qtbot.addWidget(widget)
widget.resize(520, 720)
widget.show()
qtbot.waitExposed(widget)
yield widget
def fill_devices(control, *devices):
for row, device in enumerate(devices):
control.form.set_arg_value(row, "device", device)
def test_lists_supported_scans_and_selects_first(control):
assert sorted(control.visible_scans) == ["grid_scan", "line_scan"]
assert control.current_scan in control.visible_scans
def test_grid_scan_starts_with_minimum_rows(control):
control.current_scan = "grid_scan"
assert len(control.form.arg_rows) == 2
assert not control.form.can_remove_row()
control.form.add_row()
assert len(control.form.arg_rows) == 3
control.form.remove_row(0)
control.form.remove_row(0)
assert len(control.form.arg_rows) == 2
def test_start_needs_valid_devices(control):
control.current_scan = "line_scan"
assert not control.can_start()
assert control.problems()[0] == "Choose a device"
control.form.set_arg_value(0, "device", "nope")
assert control.problems()[0] == "Unknown device 'nope'"
fill_devices(control, "samx")
assert control.can_start()
def test_get_scan_parameters(control):
control.current_scan = "line_scan"
fill_devices(control, "samx")
control.form.set_arg_value(0, "start", "-1.5")
control.form.set_arg_value(0, "stop", 2)
control.form.set_kwarg_value("steps", 12)
args, kwargs = control.get_scan_parameters(bec_object=False)
assert args == ["samx", -1.5, 2.0]
assert kwargs["steps"] == 12
assert kwargs["relative"] is False
assert kwargs["metadata"]["scan_name"] == "line_scan"
args, _ = control.get_scan_parameters()
assert args[0] is control.dev.samx
def test_optional_literal_defaults_to_none(control):
control.current_scan = "grid_scan"
assert control.get_scan_parameters(False)[1]["optim_trajectory"] is None
control.form.set_kwarg_value("optim_trajectory", "option2")
assert control.get_scan_parameters(False)[1]["optim_trajectory"] == "option2"
def test_run_scan_submits_and_emits(control, qtbot):
control.current_scan = "line_scan"
fill_devices(control, "samx")
scans = SimpleNamespace(line_scan=MagicMock())
with patch.object(control, "scans", scans), qtbot.waitSignal(control.scan_started):
control.run_scan()
scans.line_scan.assert_called_once()
assert scans.line_scan.call_args.args[0] is control.dev.samx
def test_run_scan_refuses_incomplete_form(control):
control.current_scan = "line_scan"
scans = SimpleNamespace(line_scan=MagicMock())
with patch.object(control, "scans", scans):
control.run_scan()
scans.line_scan.assert_not_called()
def test_stop_aborts_then_halts(control):
queue = MagicMock()
with patch.object(control, "queue", queue):
control.stop_scan()
queue.request_scan_abortion.assert_called_once()
assert control.stop_armed
control.stop_scan()
queue.request_scan_halt.assert_called_once()
def test_parameters_are_remembered_per_scan(control):
control.current_scan = "line_scan"
fill_devices(control, "samy")
control.form.set_kwarg_value("exp_time", 0.25)
control.current_scan = "grid_scan"
assert control.form.kwargs["exp_time"] == 0.25 # shared keyword carries over
control.current_scan = "line_scan"
assert control.form.arg_rows[0]["device"] == "samy"
def test_restore_last_scan_parameters(control, qtbot):
control.current_scan = "line_scan"
control.request_last_executed_scan_parameters()
qtbot.waitUntil(lambda: not control.is_restoring)
args, kwargs = control.get_scan_parameters(bec_object=False)
assert args == ["samx", 0.0, 2.0]
assert kwargs["steps"] == 10
def test_device_selected_signal(control, qtbot):
control.current_scan = "grid_scan"
emitted = []
control.device_selected.connect(emitted.append)
fill_devices(control, "samx", "samy")
assert emitted == ["samx", "samx samy"]
def test_allowed_scans_filter(control):
control.allowed_scans = ["grid_scan"]
assert control.visible_scans == ["grid_scan"]
assert control.current_scan == "grid_scan"
control.allowed_scans = []
assert sorted(control.visible_scans) == ["grid_scan", "line_scan"]
def test_invalid_metadata_blocks_start(control):
control.current_scan = "line_scan"
fill_devices(control, "samx")
control.metadata.set_extras([["sample_temp", "300"]])
assert control.get_scan_parameters(False)[1]["metadata"]["sample_temp"] == "300"
control.metadata.errors = {"comment": "broken"}
assert "Metadata Comment: broken" in control.problems()
def test_hide_options(control):
control.hide_scan_control_buttons = True
control.hide_metadata = True
assert control.hide_scan_control_buttons and control.hide_metadata
control.show_scan_control_buttons(True)
assert not control.hide_scan_control_buttons
def test_field_spec_limits_and_choices():
spec = FieldSpec.from_input(
{"name": "n", "type": "int", "gt": 0, "le": 10, "default": None, "display_name": "N"}
)
assert (spec.minimum, spec.maximum, spec.initial_value()) == (1, 10, 1)
choice = FieldSpec.from_input({"name": "c", "type": {"Literal": ["b", "a", None]}})
assert choice.choices == ["b", "a", ""]
assert FieldSpec.from_input({"name": "x", "type": "complex"}) is None
assert ScanFormSpec().kwarg_fields == []
# ---------------------------------------------------------------------- view specific
@pytest.fixture
def qml_control(qtbot, mocked_client):
mocked_client.connector.set_and_publish(
MessageEndpoints.available_scans(), available_scans_message
)
widget = ScanControlQml(client=mocked_client)
qtbot.addWidget(widget)
widget.show()
qtbot.waitExposed(widget)
yield widget
def test_qml_view_loads_and_bridge_writes_back(qml_control):
assert qml_control.view.rootObject() is not None
bridge = qml_control.bridge
bridge.selectScan("grid_scan")
assert bridge.currentScan == "grid_scan"
assert bridge.rowCount == 2
bridge.setArg(1, "device", "samz")
assert bridge.argRows[1]["values"]["device"] == "samz"
bridge.addRow()
assert bridge.rowCount == 3
bridge.setKwarg("exp_time", 0.5)
assert bridge.kwargs["values"]["exp_time"] == 0.5
assert "Row 1: Choose a device" in bridge.problems
@pytest.fixture
def widget_control(qtbot, mocked_client):
mocked_client.connector.set_and_publish(
MessageEndpoints.available_scans(), available_scans_message
)
widget = ScanControlModern(client=mocked_client)
qtbot.addWidget(widget)
widget.show()
qtbot.waitExposed(widget)
yield widget
def test_widget_view_edits_update_form(widget_control, qtbot):
widget_control.current_scan = "line_scan"
device = widget_control._arg_editors[0]["device"]
assert isinstance(device, DeviceEdit)
device.setEditText("samx")
device.lineEdit().editingFinished.emit()
assert widget_control.form.arg_rows[0]["device"] == "samx"
start = widget_control._arg_editors[0]["start"]
assert isinstance(start, NumberEdit)
start.setText("3.25")
start.editingFinished.emit()
assert widget_control.form.arg_rows[0]["start"] == 3.25
assert widget_control.start_button.isEnabled()
widget_control.add_row_button.click()
assert len(widget_control._arg_editors) == 2
def test_widget_view_rejects_unknown_scan_name(widget_control, qtbot):
picker = widget_control.scan_picker
picker.lineEdit().setText("no_such_scan")
picker.lineEdit().editingFinished.emit()
assert picker.currentText() == widget_control.current_scan
def test_widget_number_edit_steps_with_arrow_keys(widget_control, qtbot):
widget_control.current_scan = "line_scan"
steps = widget_control._kwarg_editors["steps"]
steps.setFocus()
qtbot.keyClick(steps, Qt.Key.Key_Up)
assert widget_control.form.kwargs["steps"] == 1