diff --git a/bec_widgets/applications/main_app.py b/bec_widgets/applications/main_app.py index 56c78730..7f710de6 100644 --- a/bec_widgets/applications/main_app.py +++ b/bec_widgets/applications/main_app.py @@ -14,6 +14,7 @@ from bec_widgets.applications.views.admin_view.admin_view import AdminView from bec_widgets.applications.views.developer_view.developer_view import DeveloperView from bec_widgets.applications.views.device_manager_view.device_manager_view import DeviceManagerView from bec_widgets.applications.views.dock_area_view.dock_area_view import DockAreaView +from bec_widgets.applications.views.sequencer_view.sequencer_view import SequencerView from bec_widgets.applications.views.view import ViewBase, WaveformViewInline, WaveformViewPopup from bec_widgets.utils.colors import apply_theme from bec_widgets.utils.error_popups import SafeSlot @@ -110,6 +111,14 @@ class BECMainApp(BECMainWindow): widget=self.device_manager, mini_text="DM", ) + self.sequencer_view = SequencerView(self) + self.add_view( + icon="account_tree", + title="Sequencer", + view_id="sequencer", + widget=self.sequencer_view, + mini_text="Seq", + ) if is_experimental_features_enabled(): self.developer_view = DeveloperView(self) self.add_view( diff --git a/bec_widgets/applications/views/sequencer_view/__init__.py b/bec_widgets/applications/views/sequencer_view/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/bec_widgets/applications/views/sequencer_view/concepts.py b/bec_widgets/applications/views/sequencer_view/concepts.py new file mode 100644 index 00000000..2dc68ef0 --- /dev/null +++ b/bec_widgets/applications/views/sequencer_view/concepts.py @@ -0,0 +1,5 @@ +"""The sequencer GUI concepts, in the order they are shown.""" + +from bec_widgets.widgets.control.sequencer.recipe.recipe_editor import RecipeEditor + +CONCEPTS = {cls.CONCEPT_ID: cls for cls in (RecipeEditor,)} diff --git a/bec_widgets/applications/views/sequencer_view/sequencer_view.py b/bec_widgets/applications/views/sequencer_view/sequencer_view.py new file mode 100644 index 00000000..7c7d602f --- /dev/null +++ b/bec_widgets/applications/views/sequencer_view/sequencer_view.py @@ -0,0 +1,202 @@ +"""Main-app view that hosts the five sequencer GUI concepts side by side. + +A switcher at the top picks the concept; each concept keeps its own sequence while you +switch, so they can be compared on the same example. Run one concept directly with:: + + python -m bec_widgets.applications.views.sequencer_view.sequencer_view --concept table +""" + +from __future__ import annotations + +import argparse +import os +import sys + +from qtpy.QtCore import Qt +from qtpy.QtWidgets import ( + QApplication, + QButtonGroup, + QHBoxLayout, + QLabel, + QPushButton, + QStackedWidget, + QVBoxLayout, + QWidget, +) + +from bec_widgets.applications.views.sequencer_view.concepts import CONCEPTS +from bec_widgets.applications.views.view import ViewBase +from bec_widgets.utils.colors import rgba +from bec_widgets.utils.error_popups import SafeSlot +from bec_widgets.widgets.control.sequencer.style import heading, hint, theme + + +class SequencerView(ViewBase): + """Sequencer view with a concept switcher. + + Editors are created when first shown, so the main app starts as fast as before. + + Args: + parent (QWidget | None): Parent widget. + concept (str | None): Concept to show right away. Without it, the view waits until it + is opened and then shows ``BEC_SEQUENCER_CONCEPT`` or the first concept. + """ + + RPC = False + + def __init__(self, parent=None, *, concept: str | None = None, **kwargs): + super().__init__(parent=parent, view_id="sequencer", title="Sequencer", **kwargs) + content = QWidget(self) + layout = QVBoxLayout(content) + layout.setContentsMargins(0, 0, 0, 0) + layout.setSpacing(0) + + self.top = QWidget(content) + self.top.setObjectName("SequencerTop") + self.top.setAttribute(Qt.WidgetAttribute.WA_StyledBackground, True) + bar = QHBoxLayout(self.top) + bar.setContentsMargins(16, 10, 16, 10) + bar.setSpacing(14) + bar.addWidget(heading("Sequencer", self.top, 16)) + self.switcher = QButtonGroup(self.top) + self.switcher.setExclusive(True) + segments = QHBoxLayout() + segments.setSpacing(0) + for number, (cid, cls) in enumerate(CONCEPTS.items(), 1): + button = QPushButton(f"{number} {cls.CONCEPT_TITLE}", self.top) + button.setCheckable(True) + button.setProperty("concept", cid) + button.setToolTip(cls.CONCEPT_TAGLINE) + self.switcher.addButton(button) + segments.addWidget(button) + bar.addLayout(segments) + self.tagline = hint("", self.top) + bar.addWidget(self.tagline, 1) + self.prototype_note = QLabel("Prototype · 5 concepts to compare", self.top) + bar.addWidget(self.prototype_note) + layout.addWidget(self.top) + self._style_top() + + self.stack = QStackedWidget(content) + layout.addWidget(self.stack, 1) + self.editors: dict[str, QWidget] = {} + self.switcher.buttonClicked.connect(self._on_switch) + self.set_content(content) + + start = concept or os.environ.get("BEC_SEQUENCER_CONCEPT") or next(iter(CONCEPTS)) + self._start_concept = start if start in CONCEPTS else next(iter(CONCEPTS)) + if concept is not None: + self.show_concept(self._start_concept) + + def _style_top(self) -> None: + self.top.setStyleSheet( + f"#SequencerTop {{ background: {theme('CARD_BG', '#ffffff').name()};" + f" border-bottom: 1px solid {theme('BORDER', '#d0d0d0').name()}; }}" + ) + primary = theme("PRIMARY", "#2f6fdf") + border = theme("BORDER", "#d0d0d0").name() + buttons = self.switcher.buttons() + for number, button in enumerate(buttons, 1): + first, last = number == 1, number == len(buttons) + radius = ( + f"border-top-left-radius: {8 if first else 0}px;" + f" border-bottom-left-radius: {8 if first else 0}px;" + f" border-top-right-radius: {8 if last else 0}px;" + f" border-bottom-right-radius: {8 if last else 0}px;" + ) + button.setStyleSheet( + f"QPushButton {{ padding: 6px 14px; border: 1px solid {border}; {radius}" + " margin-left: -1px; }" + f"QPushButton:checked {{ background: {rgba(primary, 45)};" + f" border: 1px solid {primary.name()}; color: {primary.name()}; font-weight: 600; }}" + ) + self.tagline.setStyleSheet(f"color: {rgba(theme('FG', '#333333'), 150)};") + self.prototype_note.setStyleSheet( + f"color: {theme('ACCENT_WARNING', '#eda200').name()}; font-weight: 600;" + ) + + def apply_theme(self, theme_name: str) -> None: + """Restyle the view and rebuild the editors in the new colors, keeping their work.""" + super().apply_theme(theme_name) + self._style_top() + current = next( + (cid for cid, editor in self.editors.items() if editor is self.stack.currentWidget()), + None, + ) + for cid, old in list(self.editors.items()): + state = old.save_state() + del self.editors[cid] + editor = self._create(cid) + editor.restore_state(state) + self.stack.removeWidget(old) + old.close() + old.deleteLater() + if current is not None: + self.show_concept(current) + + def _create(self, concept: str) -> QWidget: + editor = CONCEPTS[concept](self.stack) + if hasattr(editor, "open_in_recipe"): + editor.open_in_recipe.connect(self._open_in_recipe) + self.editors[concept] = editor + self.stack.addWidget(editor) + return editor + + @SafeSlot() + def on_enter(self) -> None: + """Create the first concept when the view is opened for the first time.""" + if not self.editors: + self.show_concept(self._start_concept) + + def show_concept(self, concept: str) -> QWidget: + """Show a concept, creating its editor on first use. + + Args: + concept (str): Concept id, e.g. ``recipe`` or ``table``. + + Returns: + QWidget: The concept's editor. + """ + editor = self.editors.get(concept) + if editor is None: + editor = self._create(concept) + self.stack.setCurrentWidget(editor) + self.tagline.setText(CONCEPTS[concept].CONCEPT_TAGLINE) + for button in self.switcher.buttons(): + button.setChecked(button.property("concept") == concept) + return editor + + @SafeSlot() + def _on_switch(self, button) -> None: + self.show_concept(button.property("concept")) + + @SafeSlot(dict) + def _open_in_recipe(self, doc: dict) -> None: + recipe = self.show_concept("recipe") + recipe.document.set_document(doc) + + +def main(): # pragma: no cover - manual entry point + """Open the sequencer view in its own window.""" + # pylint: disable=import-outside-toplevel + from bec_widgets.utils.colors import apply_theme + from bec_widgets.widgets.containers.main_window.main_window import BECMainWindow + + parser = argparse.ArgumentParser(description="Try the sequencer GUI concepts") + parser.add_argument("--concept", choices=list(CONCEPTS), default=None) + parser.add_argument("--theme", choices=["light", "dark"], default="dark") + args = parser.parse_args() + + app = QApplication(sys.argv) + apply_theme(args.theme) + window = BECMainWindow() + window.setWindowTitle("BEC Sequencer (prototype)") + view = SequencerView(window, concept=args.concept or next(iter(CONCEPTS))) + window.setCentralWidget(view) + window.resize(1400, 860) + window.show() + sys.exit(app.exec()) + + +if __name__ == "__main__": # pragma: no cover + main() diff --git a/bec_widgets/widgets/control/sequencer/__init__.py b/bec_widgets/widgets/control/sequencer/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/bec_widgets/widgets/control/sequencer/backend.py b/bec_widgets/widgets/control/sequencer/backend.py new file mode 100644 index 00000000..28cb8519 --- /dev/null +++ b/bec_widgets/widgets/control/sequencer/backend.py @@ -0,0 +1,161 @@ +"""Bridge between the sequencer editors and ``bec_lib.scan_sequencer``. + +The backend lives in bec (draft PR bec-project/bec#1128). When the installed bec_lib does not +ship it yet, editing still works and ``available`` is False, so the GUI can say why checking +and running are disabled. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass + +from bec_lib.endpoints import MessageEndpoints +from bec_lib.logger import bec_logger + +logger = bec_logger.logger + +try: # the backend is not released yet + from bec_lib.scan_sequencer import DefinitionLibrary, ScanCatalog, Schedule + from bec_lib.scan_sequencer.core.description import describe as _describe + from bec_lib.scan_sequencer.core.python_export import export_python as _export_python + from bec_lib.scan_sequencer.core.validation import validate as _validate + + BACKEND_AVAILABLE = True +except ImportError: # pragma: no cover - depends on the installed bec_lib + BACKEND_AVAILABLE = False + +# Scans that make no sense as a sequence step (moves have their own step). +_HIDDEN_SCANS = { + "mv", + "umv", + "device_rpc", + "open_scan_def", + "close_scan_def", + "close_scan_group", + "open_interactive_scan", + "close_interactive_scan", + "interactive_scan_trigger", +} + +_NODE_PREFIX = re.compile(r"^Node (?P[^:]+): (?P.*)$", re.S) + + +@dataclass +class Problem: + """A validation problem, attached to a step when the backend names one.""" + + node_id: str + message: str + + +def friendly_message(message: str) -> str: + """Make a backend validation message readable for scientists.""" + message = message.replace("ScanInputValidationError: ", "") + message = re.sub(r"^\d+ validation errors? for \w+\s*", "", message) + if message.startswith("Undefined Reference: "): + name = message.split(": ", 1)[1] + return f"“{name}” is only known inside its “For each {name}” loop" + if message.startswith("Unknown device or signal: "): + return f"There is no device called “{message.split(': ', 1)[1]}”" + if message.startswith("Unknown scan: "): + return f"This beamline has no scan called “{message.split(': ', 1)[1]}”" + return message.strip().splitlines()[0] if message.strip() else "Unknown problem" + + +class SequencerBackend: + """Check, describe, export and run sequence documents for one BEC client. + + Args: + client: BEC client of the hosting widget. + """ + + QUEUE = "sequences" + + def __init__(self, client): + self.client = client + self.definitions: list[dict] = [] + self._helper = None + + @property + def available(self) -> bool: + """Whether bec_lib ships the scan sequencer.""" + return BACKEND_AVAILABLE + + # --- scans ----------------------------------------------------------------------------- + def scan_entries(self) -> dict[str, dict]: + """Advertised scans usable as a sequence step, by name.""" + try: + msg = self.client.connector.get(MessageEndpoints.available_scans()) + except Exception: # pylint: disable=broad-except + return {} + resource = getattr(msg, "resource", None) or {} + return { + name: info + for name, info in resource.items() + if name not in _HIDDEN_SCANS and not info.get("is_internal") + } + + def device_names(self) -> list[str]: + """Names of devices that can be moved or set.""" + try: + return sorted(self.client.device_manager.devices.keys()) + except Exception: # pylint: disable=broad-except + return [] + + # --- checks and exports ---------------------------------------------------------------- + def _tree(self, doc: dict): + schedule = Schedule.model_validate(doc) + tree, _ = DefinitionLibrary(self.definitions).resolve(schedule.root) + return tree + + def _device_manager(self): + return getattr(self.client, "device_manager", None) + + def validate(self, doc: dict) -> list[Problem]: + """Validate a document; returns an empty list when it can run. + + The backend stops at the first problem, so at most one problem is returned. + """ + if not BACKEND_AVAILABLE: + return [] + try: + tree = self._tree(doc) + catalog = ScanCatalog(self.scan_entries(), self._device_manager()) + _validate(tree, catalog, self._device_manager()) + except Exception as exc: # pylint: disable=broad-except + text = str(exc) + match = _NODE_PREFIX.match(text) + if match: + node_id = match.group("path").split("/")[-1] + return [Problem(node_id, friendly_message(match.group("message")))] + return [Problem("", friendly_message(text))] + return [] + + def describe(self, doc: dict) -> str: + """Backend's procedural description.""" + if not BACKEND_AVAILABLE: + return "The scan sequencer backend is not installed." + return _describe(self._tree(doc)) + + def python(self, doc: dict) -> str: + """Equivalent BEC IPython code.""" + if not BACKEND_AVAILABLE: + return "# The scan sequencer backend is not installed." + return _export_python(self._tree(doc), self._device_manager()) + + # --- execution ------------------------------------------------------------------------- + def submit(self, doc: dict): + """Queue the sequence on the procedure worker. + + Returns: + ProcedureStatus: Status of the queued procedure. + """ + # pylint: disable=import-outside-toplevel + from bec_lib.procedures.helper import FrontendProcedureHelper + + if self._helper is None: + self._helper = FrontendProcedureHelper(self.client.connector) + return self._helper.request.procedure( + "run_sequence", ((doc, list(self.definitions)), {}), self.QUEUE + ) diff --git a/bec_widgets/widgets/control/sequencer/document.py b/bec_widgets/widgets/control/sequencer/document.py new file mode 100644 index 00000000..cdf365ac --- /dev/null +++ b/bec_widgets/widgets/control/sequencer/document.py @@ -0,0 +1,179 @@ +"""Editable sequence document shared by an editor and the run bar.""" + +from __future__ import annotations + +import copy + +from qtpy.QtCore import QObject, Signal + +from bec_widgets.widgets.control.sequencer import model + + +class SequenceDocument(QObject): + """Hold one sequence document and announce every change. + + Editors change the tree through these methods (or change node dictionaries directly and + call ``touch``), and listen to ``changed`` to redraw. ``selection_changed`` carries the id + of the selected step, so a list, a canvas and an inspector can follow the same selection. + """ + + changed = Signal() + selection_changed = Signal(str) + + def __init__(self, doc: dict | None = None, parent: QObject | None = None): + super().__init__(parent) + self._doc = model.normalize(copy.deepcopy(doc) if doc else model.empty_document()) + self._selected = "" + self._undo: list[dict] = [] + + # --- access ---------------------------------------------------------------------------- + @property + def doc(self) -> dict: + """The document as plain JSON-compatible data.""" + return self._doc + + @property + def root(self) -> dict: + """The root group.""" + return self._doc["root"] + + @property + def selected(self) -> str: + """Id of the selected step, or an empty string.""" + return self._selected + + def node(self, node_id: str) -> dict | None: + """Return a node by id.""" + return model.find(self.root, node_id)[0] + + # --- changes --------------------------------------------------------------------------- + def set_document(self, doc: dict) -> None: + """Replace the whole document.""" + self._remember() + self._doc = model.normalize(copy.deepcopy(doc)) + self._selected = "" + self.touch() + + def touch(self) -> None: + """Announce that the document changed.""" + self.changed.emit() + + def select(self, node_id: str) -> None: + """Select a step by id (empty string clears).""" + if node_id == self._selected: + return + self._selected = node_id or "" + self.selection_changed.emit(self._selected) + + def add(self, nodes: list[dict], parent_id: str | None = None, index: int = -1) -> None: + """Insert new steps and select the first one.""" + self._remember() + model.insert(self.root, nodes, parent_id, index) + self.touch() + if nodes: + self.select(nodes[0]["id"]) + + def add_next_to_selection(self, nodes: list[dict]) -> None: + """Insert steps inside the selected loop/group, or right after the selected step.""" + node, parent = model.find(self.root, self._selected) if self._selected else (None, None) + if node is not None and model.is_container(node) and node is not self.root: + self.add(nodes, node["id"]) + elif node is not None and parent is not None: + siblings = model.children_of(parent) + self.add(nodes, parent["id"], siblings.index(node) + 1) + else: + self.add(nodes) + + def update(self, node_id: str, **fields) -> None: + """Change fields of a node.""" + node = self.node(node_id) + if node is None: + return + self._remember() + for key, value in fields.items(): + if value is None and key in ("bind_status",): + node.pop(key, None) + else: + node[key] = value + self.touch() + + def delete(self, node_id: str) -> None: + """Remove a step; a "while moving" loop also removes the move it watches.""" + node = self.node(node_id) + if node is None or node is self.root: + return + self._remember() + status = model.while_status(node) if node.get("type") == "while" else None + model.remove(self.root, node_id) + if status: + owner = model.status_owner(self.root, status) + if owner is not None: + model.remove(self.root, owner["id"]) + self._selected = "" + self.touch() + self.selection_changed.emit("") + + def move(self, node_id: str, parent_id: str | None, index: int) -> None: + """Move a step to another position (index counted after removal).""" + node = self.node(node_id) + if node is None or node is self.root: + return + if parent_id and any(a.get("id") == node_id for a in model.ancestors(self.root, parent_id)): + return + if parent_id == node_id: + return + self._remember() + model.remove(self.root, node_id) + model.insert(self.root, [node], parent_id, index) + self.touch() + + def shift(self, node_id: str, offset: int) -> None: + """Move a step up or down among its siblings.""" + node, parent = model.find(self.root, node_id) + if node is None or parent is None: + return + siblings = model.children_of(parent) + index = siblings.index(node) + offset + if 0 <= index < len(siblings): + self._remember() + siblings.remove(node) + siblings.insert(index, node) + self.touch() + + def duplicate(self, node_id: str) -> None: + """Insert a copy right after a step.""" + node, parent = model.find(self.root, node_id) + if node is None or parent is None: + return + siblings = model.children_of(parent) + self.add([model.clone(node)], parent["id"], siblings.index(node) + 1) + + def wrap(self, node_id: str, kind: str) -> None: + """Put a step inside a new repeat or for-each loop.""" + node, parent = model.find(self.root, node_id) + if node is None or parent is None: + return + self._remember() + siblings = model.children_of(parent) + index = siblings.index(node) + loop = model.new_step(kind)[0] + siblings[index] = loop + model.children_of(loop).append(node) + self.touch() + self.select(loop["id"]) + + def undo(self) -> None: + """Restore the previous version.""" + if self._undo: + self._doc = self._undo.pop() + self._selected = "" + self.touch() + self.selection_changed.emit("") + + def can_undo(self) -> bool: + """Return whether there is something to undo.""" + return bool(self._undo) + + def _remember(self) -> None: + self._undo.append(copy.deepcopy(self._doc)) + del self._undo[:-50] diff --git a/bec_widgets/widgets/control/sequencer/editor_base.py b/bec_widgets/widgets/control/sequencer/editor_base.py new file mode 100644 index 00000000..fabae5c8 --- /dev/null +++ b/bec_widgets/widgets/control/sequencer/editor_base.py @@ -0,0 +1,92 @@ +"""Common base for the five sequencer editor concepts.""" + +from __future__ import annotations + +import copy + +from qtpy.QtWidgets import QVBoxLayout, QWidget + +from bec_widgets import BECWidget +from bec_widgets.widgets.control.sequencer import model +from bec_widgets.widgets.control.sequencer.backend import Problem, SequencerBackend +from bec_widgets.widgets.control.sequencer.document import SequenceDocument +from bec_widgets.widgets.control.sequencer.run_bar import RunBar + + +class SequenceEditorBase(BECWidget, QWidget): + """An editor for one sequence document, with the shared run bar underneath. + + Subclasses build their editing surface in ``build_editor`` and redraw in ``refresh``. + ``show_problems`` receives the validation problems so the editor can mark the step. + + Args: + parent (QWidget | None): Parent widget. + client: BEC client. + doc (dict | None): Initial document; defaults to the shared example. + """ + + RPC = False + PLUGIN = False + CONCEPT_ID = "" + CONCEPT_TITLE = "" + CONCEPT_TAGLINE = "" + + def __init__(self, parent=None, client=None, doc: dict | None = None, **kwargs): + super().__init__(parent=parent, client=client, **kwargs) + self.get_bec_shortcuts() + self.backend = SequencerBackend(self.client) + self.document = SequenceDocument(doc if doc is not None else model.example_document(), self) + self.problems: list[Problem] = [] + + layout = QVBoxLayout(self) + layout.setContentsMargins(0, 0, 0, 0) + layout.setSpacing(0) + self.editor = QWidget(self) + layout.addWidget(self.editor, 1) + self.build_editor(self.editor) + self.run_bar = RunBar(self.backend, self.document, self) + layout.addWidget(self.run_bar) + self.run_bar.problems_changed.connect(self._on_problems) + self.document.changed.connect(self.refresh) + self.refresh() + self.run_bar.check() + + def build_editor(self, host: QWidget) -> None: + """Create the editing surface inside ``host``.""" + raise NotImplementedError + + def refresh(self) -> None: + """Redraw after the document changed.""" + + def show_problems(self, problems: list[Problem]) -> None: + """Mark steps that need a fix.""" + + def save_state(self) -> dict: + """What the editor needs to come back after being rebuilt (e.g. on a theme change).""" + return {"doc": copy.deepcopy(self.document.doc), "selected": self.document.selected} + + def restore_state(self, state: dict) -> None: + """Restore ``save_state`` output.""" + self.document.set_document(state["doc"]) + if state.get("selected"): + self.document.select(state["selected"]) + + def _on_problems(self, problems: list) -> None: + self.problems = problems + self.show_problems(problems) + + def problem_for(self, node_id: str) -> str: + """Problem message for a step, or an empty string.""" + return next((p.message for p in self.problems if p.node_id == node_id), "") + + def scan_entries(self) -> dict: + """Advertised scans by name.""" + return self.backend.scan_entries() + + def cleanup(self): + """Stop timers of the shared parts.""" + self.run_bar.cleanup() + inspector = getattr(self, "inspector", None) + if inspector is not None: + inspector.cleanup() + super().cleanup() diff --git a/bec_widgets/widgets/control/sequencer/inspector.py b/bec_widgets/widgets/control/sequencer/inspector.py new file mode 100644 index 00000000..5b1dac6d --- /dev/null +++ b/bec_widgets/widgets/control/sequencer/inspector.py @@ -0,0 +1,467 @@ +"""Side panel that edits the selected step in plain language. + +Used by the Recipe and Flowchart concepts. Scan steps reuse the existing ``ScanControl`` +form, so scientists see the same scan settings they know from the scan control widget. +""" + +from __future__ import annotations + +import numpy as np +from qtpy.QtCore import QTimer +from qtpy.QtWidgets import ( + QAbstractSpinBox, + QCheckBox, + QComboBox, + QDialog, + QDialogButtonBox, + QDoubleSpinBox, + QFormLayout, + QHBoxLayout, + QLabel, + QLineEdit, + QPushButton, + QScrollArea, + QSpinBox, + QStackedWidget, + QVBoxLayout, + QWidget, +) + +from bec_widgets.utils.error_popups import SafeSlot +from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import DeviceComboBox +from bec_widgets.widgets.control.scan_control.scan_control import ScanControl +from bec_widgets.widgets.control.sequencer import model +from bec_widgets.widgets.control.sequencer.document import SequenceDocument +from bec_widgets.widgets.control.sequencer.style import heading, hint, kind_icon + + +def parse_values(text: str) -> list: + """Parse ``1, -1, 2.5`` (or one value per line) into numbers, keeping words as text.""" + values = [] + for part in text.replace("\n", ",").replace(";", ",").split(","): + part = part.strip() + if not part: + continue + try: + number = float(part) + values.append(int(number) if number.is_integer() and "." not in part else number) + except ValueError: + values.append(part) + return values + + +def values_text(values: list) -> str: + """Inverse of ``parse_values``.""" + return ", ".join(model.format_value(v) for v in values) + + +class RangeDialog(QDialog): + """Fill a list of values from start, stop and number of points.""" + + def __init__(self, parent: QWidget | None = None): + super().__init__(parent) + self.setWindowTitle("Fill evenly spaced values") + form = QFormLayout(self) + self.start = QDoubleSpinBox(self, decimals=4, minimum=-1e6, maximum=1e6, value=0) + self.stop = QDoubleSpinBox(self, decimals=4, minimum=-1e6, maximum=1e6, value=10) + self.points = QSpinBox(self, minimum=1, maximum=10000, value=11) + form.addRow("From", self.start) + form.addRow("To", self.stop) + form.addRow("Number of values", self.points) + buttons = QDialogButtonBox( + QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel, self + ) + buttons.accepted.connect(self.accept) + buttons.rejected.connect(self.reject) + form.addRow(buttons) + + def values(self) -> list[float]: + """The generated values.""" + return [ + round(float(v), 6) + for v in np.linspace(self.start.value(), self.stop.value(), self.points.value()) + ] + + +def make_scan_form(parent: QWidget | None, client) -> ScanControl: + """The scan control form without its Start/Stop buttons and metadata.""" + form = ScanControl(parent, client=client) + form.hide_scan_control_buttons = True + form.hide_metadata = True + group = getattr(form, "_metadata_group", None) + if group is not None: + group.setVisible(False) + return form + + +def scan_form_values(form: ScanControl) -> tuple[str, list, dict]: + """Scan name, positional and keyword arguments shown in a scan form.""" + positional, arguments = form.get_scan_parameters(bec_object=False) + arguments.pop("metadata", None) + return form.current_scan, list(positional), arguments + + +def set_scan_form_values(form: ScanControl, node: dict) -> None: + """Show a scan node in a scan form.""" + form.set_current_scan(node.get("name", "")) + positional = [v for v in node.get("positional", []) if not isinstance(v, dict)] + if form.arg_box is not None and positional: + form.arg_box.set_parameters(positional) + arguments = node.get("arguments") if isinstance(node.get("arguments"), dict) else {} + for box in form.kwarg_boxes: + box.set_parameters(arguments) + + +class ScanSettingsDialog(QDialog): + """Edit one scan step in the familiar scan control form.""" + + def __init__(self, node: dict, client, parent: QWidget | None = None): + super().__init__(parent) + self.setWindowTitle("Scan settings") + self.resize(520, 620) + layout = QVBoxLayout(self) + scroll = QScrollArea(self) + scroll.setWidgetResizable(True) + scroll.setFrameShape(QScrollArea.Shape.NoFrame) + self.form = make_scan_form(scroll, client) + scroll.setWidget(self.form) + layout.addWidget(scroll, 1) + set_scan_form_values(self.form, node) + buttons = QDialogButtonBox( + QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel, self + ) + buttons.accepted.connect(self.accept) + buttons.rejected.connect(self.reject) + layout.addWidget(buttons) + + def values(self) -> tuple[str, list, dict]: + """Scan name, positional and keyword arguments.""" + return scan_form_values(self.form) + + +class ValueInput(QWidget): + """Either a fixed number or "the current loop value".""" + + def __init__(self, parent: QWidget | None = None): + super().__init__(parent) + layout = QHBoxLayout(self) + layout.setContentsMargins(0, 0, 0, 0) + self.source = QComboBox(self) + self.number = QDoubleSpinBox(self, decimals=4, minimum=-1e9, maximum=1e9) + self.number.setButtonSymbols(QAbstractSpinBox.ButtonSymbols.NoButtons) + layout.addWidget(self.source, 1) + layout.addWidget(self.number, 1) + self.source.currentIndexChanged.connect(self._sync) + + def set_choices(self, variables: list[str]) -> None: + """Offer the loop variables visible at this step.""" + self.source.blockSignals(True) + self.source.clear() + self.source.addItem("a fixed value", None) + for name in variables: + self.source.addItem(f"each {name} from the loop", name) + self.source.blockSignals(False) + + def set_value(self, value) -> None: + """Show a literal or a reference.""" + if model.is_ref(value) and self.source.findData(value["ref"]) >= 0: + self.source.setCurrentIndex(self.source.findData(value["ref"])) + else: + self.source.setCurrentIndex(0) + self.number.setValue(float(value) if isinstance(value, (int, float)) else 0.0) + self._sync() + + def value(self): + """Literal number or reference expression.""" + name = self.source.currentData() + if name: + return model.ref(name) + number = self.number.value() + return int(number) if number.is_integer() else number + + def _sync(self): + self.number.setVisible(self.source.currentData() is None) + self.source.setVisible(self.source.count() > 1) + + +class StepInspector(QWidget): + """Edit the step selected in a ``SequenceDocument``.""" + + def __init__(self, document: SequenceDocument, parent: QWidget | None = None, client=None): + super().__init__(parent) + self.document = document + self._node_id = "" + self._loading = False + self._commit_timer = QTimer(self, singleShot=True, interval=250) + self._commit_timer.timeout.connect(self._commit) + + layout = QVBoxLayout(self) + layout.setContentsMargins(12, 12, 12, 12) + layout.setSpacing(8) + title_row = QHBoxLayout() + self.icon = QLabel(self) + self.title = heading("Step details", self) + title_row.addWidget(self.icon) + title_row.addWidget(self.title, 1) + layout.addLayout(title_row) + self.subtitle = hint("Select a step to change it.", self) + layout.addWidget(self.subtitle) + + self.pages = QStackedWidget(self) + layout.addWidget(self.pages, 1) + self.pages.addWidget(QWidget(self)) # empty + + # move / set + self.move_page = QWidget(self) + form = QFormLayout(self.move_page) + self.device = DeviceComboBox( + self.move_page, client=client, device_filter=["Positioner"], autocomplete=True + ) + self.device.setEditable(True) + self.move_value = ValueInput(self.move_page) + self.move_wait = QCheckBox("Wait until it has arrived", self.move_page) + form.addRow("Motor", self.device) + form.addRow("Go to", self.move_value) + form.addRow("", self.move_wait) + self.pages.addWidget(self.move_page) + + # scan + self.scan_page = QScrollArea(self) + self.scan_page.setWidgetResizable(True) + self.scan_page.setFrameShape(QScrollArea.Shape.NoFrame) + self.scan_control = make_scan_form(self.scan_page, client) + self.scan_page.setWidget(self.scan_control) + self.pages.addWidget(self.scan_page) + + # repeat + self.repeat_page = QWidget(self) + form = QFormLayout(self.repeat_page) + self.repeat_count = QSpinBox(self.repeat_page, minimum=1, maximum=100000) + form.addRow("How many times", self.repeat_count) + form.addRow(hint("Everything inside the loop runs this many times.", self.repeat_page)) + self.pages.addWidget(self.repeat_page) + + # for each + self.each_page = QWidget(self) + form = QFormLayout(self.each_page) + self.each_name = QLineEdit(self.each_page) + self.each_values = QLineEdit(self.each_page) + self.each_values.setPlaceholderText("e.g. 1, -1, 2.5") + fill = QPushButton("Fill evenly spaced…", self.each_page) + fill.clicked.connect(self._fill_range) + self.each_count = hint("", self.each_page) + form.addRow("Name of the value", self.each_name) + form.addRow("Values (in order)", self.each_values) + form.addRow("", fill) + form.addRow(self.each_count) + form.addRow( + hint( + "Steps inside the loop can use this value, e.g. “Move samx to each position”.", + self.each_page, + ) + ) + self.pages.addWidget(self.each_page) + + # while moving + self.while_page = QWidget(self) + form = QFormLayout(self.while_page) + self.while_device = DeviceComboBox( + self.while_page, client=client, device_filter=["Positioner"], autocomplete=True + ) + self.while_device.setEditable(True) + self.while_target = QDoubleSpinBox(self.while_page, decimals=4, minimum=-1e9, maximum=1e9) + form.addRow("Motor", self.while_device) + form.addRow("Moves to", self.while_target) + form.addRow( + hint( + "The motor starts moving, and the steps inside keep repeating until it arrives.", + self.while_page, + ) + ) + self.pages.addWidget(self.while_page) + + # group / other + self.info_page = QWidget(self) + QVBoxLayout(self.info_page).addWidget( + hint("A group just keeps steps together. Drag steps in or out.", self.info_page) + ) + self.pages.addWidget(self.info_page) + + for widget in (self.move_value.number, self.while_target): + widget.valueChanged.connect(self._schedule) + self.repeat_count.valueChanged.connect(self._schedule) + self.move_value.source.currentIndexChanged.connect(self._schedule) + self.move_wait.toggled.connect(self._schedule) + self.each_name.textChanged.connect(self._schedule) + self.each_values.textChanged.connect(self._schedule) + for combo in (self.device, self.while_device): + combo.currentTextChanged.connect(self._schedule) + self.scan_control.scan_selected.connect(self._on_scan_selected) + + document.selection_changed.connect(self.load) + document.changed.connect(self._on_document_changed) + + # --- loading --------------------------------------------------------------------------- + @SafeSlot(str) + def load(self, node_id: str) -> None: + """Show the editor for a step.""" + self._commit_timer.stop() + self._node_id = node_id + node = self.document.node(node_id) if node_id else None + self._loading = True + try: + if node is None or node is self.document.root: + self.icon.clear() + self.title.setText("Step details") + self.subtitle.setText("Select a step to change it.") + self.pages.setCurrentIndex(0) + return + kind = node["type"] + self.icon.setPixmap(kind_icon(kind, 22).pixmap(22, 22)) + title = model.KIND_TITLES.get(kind, kind) + if kind == "set" and "." not in node.get("device", ""): + title = "Start moving" if node.get("bind_status") else "Move" + self.title.setText(title) + self.subtitle.setText(model.sentence(node, self.document.root, self._entries())) + getattr(self, f"_load_{kind}", self._load_other)(node) + finally: + self._loading = False + + def _entries(self) -> dict: + return getattr(self.scan_control, "available_scans", None) or {} + + def _load_set(self, node: dict) -> None: + self.device.setCurrentText(node.get("device", "")) + self.move_value.set_choices(model.visible_variables(self.document.root, node["id"])) + self.move_value.set_value(node.get("value")) + self.move_wait.setChecked(bool(node.get("wait", True))) + self.move_wait.setEnabled(not node.get("bind_status")) + self.pages.setCurrentWidget(self.move_page) + + def _load_scan_call(self, node: dict) -> None: + self.pages.setCurrentWidget(self.scan_page) + set_scan_form_values(self.scan_control, node) + self._hook_scan_inputs() + + def _load_repeat(self, node: dict) -> None: + self.repeat_count.setValue(int(node.get("count", 1))) + self.pages.setCurrentWidget(self.repeat_page) + + def _load_for_each(self, node: dict) -> None: + self.each_name.setText(node.get("variable", "")) + values = node.get("values") if isinstance(node.get("values"), list) else [] + self.each_values.setText(values_text(values)) + self.each_count.setText(f"{len(values)} value{'s' if len(values) != 1 else ''}") + self.pages.setCurrentWidget(self.each_page) + + def _load_while(self, node: dict) -> None: + status = model.while_status(node) + owner = model.status_owner(self.document.root, status) if status else None + if owner is not None: + self.while_device.setCurrentText(owner.get("device", "")) + value = owner.get("value") + self.while_target.setValue(float(value) if isinstance(value, (int, float)) else 0) + self.pages.setCurrentWidget(self.while_page) + + def _load_other(self, _node: dict) -> None: + self.pages.setCurrentWidget(self.info_page) + + # --- committing ------------------------------------------------------------------------ + def _schedule(self, *_args) -> None: + if not self._loading and self._node_id: + self._commit_timer.start() + + @SafeSlot(str) + def _on_scan_selected(self, _name: str) -> None: + self._hook_scan_inputs() + self._schedule() + + def _hook_scan_inputs(self) -> None: + for widget in self.scan_control.findChildren(QWidget): + if widget.property("_sequencer_hooked"): + continue + widget.setProperty("_sequencer_hooked", True) + if isinstance(widget, (QSpinBox, QDoubleSpinBox)): + widget.valueChanged.connect(self._schedule) + elif ( + isinstance(widget, QComboBox) + and widget is not self.scan_control.comboBox_scan_selection + ): + widget.currentTextChanged.connect(self._schedule) + elif isinstance(widget, QCheckBox): + widget.toggled.connect(self._schedule) + elif isinstance(widget, QLineEdit) and not isinstance(widget.parent(), QComboBox): + widget.textChanged.connect(self._schedule) + + @SafeSlot() + def _commit(self) -> None: + node = self.document.node(self._node_id) + if node is None: + return + kind = node["type"] + if kind == "set": + self.document.update( + node["id"], + device=self.device.currentText().strip() or node.get("device"), + value=self.move_value.value(), + wait=True if node.get("bind_status") else self.move_wait.isChecked(), + ) + elif kind == "scan_call": + name, positional, arguments = scan_form_values(self.scan_control) + if not name: + return + self.document.update( + node["id"], name=name, positional=list(positional), arguments=arguments + ) + elif kind == "repeat": + self.document.update(node["id"], count=self.repeat_count.value()) + elif kind == "for_each": + values = parse_values(self.each_values.text()) + self.each_count.setText(f"{len(values)} value{'s' if len(values) != 1 else ''}") + new_name = self.each_name.text().strip() or node.get("variable") + self._rename_variable(node, node.get("variable"), new_name) + self.document.update(node["id"], variable=new_name, values=values) + elif kind == "while": + status = model.while_status(node) + owner = model.status_owner(self.document.root, status) if status else None + if owner is not None: + target = self.while_target.value() + self.document.update( + owner["id"], + device=self.while_device.currentText().strip() or owner.get("device"), + value=int(target) if target.is_integer() else target, + ) + self.subtitle.setText(model.sentence(node, self.document.root, self._entries())) + + def _rename_variable(self, loop: dict, old: str, new: str) -> None: + if not old or old == new: + return + + def visit(value): + if model.is_ref(value) and value["ref"] == old: + return model.ref(new) + if isinstance(value, list): + return [visit(v) for v in value] + if isinstance(value, dict): + return {k: visit(v) for k, v in value.items()} + return value + + for node, _depth, _parent in model.walk(loop): + for key in ("value", "positional", "arguments", "values"): + if key in node: + node[key] = visit(node[key]) + + @SafeSlot() + def _on_document_changed(self) -> None: + if self._node_id and self.document.node(self._node_id) is None: + self.load("") + + @SafeSlot() + def _fill_range(self) -> None: + dialog = RangeDialog(self) + if dialog.exec(): + self.each_values.setText(values_text(dialog.values())) + + def cleanup(self) -> None: + """Stop the commit timer.""" + self._commit_timer.stop() diff --git a/bec_widgets/widgets/control/sequencer/model.py b/bec_widgets/widgets/control/sequencer/model.py new file mode 100644 index 00000000..a8356dc8 --- /dev/null +++ b/bec_widgets/widgets/control/sequencer/model.py @@ -0,0 +1,545 @@ +"""Plain-data helpers for scan sequencer documents. + +The scan sequencer backend (``bec_lib.scan_sequencer``) receives a JSON document made of +nested nodes. The editors in this package keep that document as plain dictionaries so all +concepts share one data format, and use the helpers here to create nodes, walk the tree, +describe steps in plain language and count what a sequence will do. +""" + +from __future__ import annotations + +import copy +import itertools +import json +from dataclasses import dataclass, field +from typing import Any, Iterator + +LOOP_TYPES = ("repeat", "for_each", "while") +CONTAINER_TYPES = ("sequence", *LOOP_TYPES) + +# Friendly names for the step kinds shown to users. +KIND_TITLES = { + "set": "Move", + "scan_call": "Run scan", + "repeat": "Repeat", + "for_each": "For each value", + "while": "While moving", + "sequence": "Group", + "component_call": "Saved block", +} + + +def empty_document(doc_id: str = "new_sequence") -> dict: + """Return a sequence document with an empty root group. + + Args: + doc_id (str): Document identity. + + Returns: + dict: Sequence document accepted by the backend. + """ + return { + "schema_version": 1, + "type": "sequence", + "id": doc_id, + "revision": 1, + "root": {"id": "root", "type": "sequence", "children": []}, + } + + +def example_document() -> dict: + """Return the shared example: two rounds over two sample positions, one scan at each. + + Returns: + dict: Sequence document. + """ + doc = empty_document("alternating_positions") + loop = for_each_node("position", [1, -1], [move_node("samx", ref("position"))]) + loop["body"]["children"].append( + scan_node("line_scan", ["samy", -2, 2], {"steps": 20, "exp_time": 0.1, "relative": True}) + ) + doc["root"]["children"] = [repeat_node(2, [loop])] + return normalize(doc) + + +# --------------------------------------------------------------------------------------------- +# Node construction +# --------------------------------------------------------------------------------------------- +_counter = itertools.count(1) + + +def _new_id(prefix: str) -> str: + return f"{prefix}_{next(_counter)}" + + +def ref(name: str) -> dict: + """Reference a loop variable. + + Args: + name (str): Loop variable name. + + Returns: + dict: Reference expression. + """ + return {"ref": name} + + +def is_ref(value: Any) -> bool: + """Return whether a value is a loop-variable reference.""" + return isinstance(value, dict) and set(value) == {"ref"} + + +def group_node(children: list[dict] | None = None) -> dict: + """Create a group of steps that run in order.""" + return {"id": _new_id("group"), "type": "sequence", "children": list(children or [])} + + +def move_node(device: str, value: Any, wait: bool = True, bind_status: str | None = None) -> dict: + """Create a Set node for a device or signal.""" + node = {"id": _new_id("move"), "type": "set", "device": device, "value": value, "wait": wait} + if bind_status: + node["bind_status"] = bind_status + return node + + +def scan_node(name: str, positional: list | None = None, arguments: dict | None = None) -> dict: + """Create a scan call.""" + return { + "id": _new_id("scan"), + "type": "scan_call", + "name": name, + "positional": list(positional or []), + "arguments": dict(arguments or {}), + } + + +def repeat_node(count: int, children: list[dict] | None = None) -> dict: + """Create a repeat loop with a group body.""" + return {"id": _new_id("repeat"), "type": "repeat", "count": count, "body": group_node(children)} + + +def for_each_node(variable: str, values: list, children: list[dict] | None = None) -> dict: + """Create a for-each loop with a group body.""" + return { + "id": _new_id("for_each"), + "type": "for_each", + "variable": variable, + "values": list(values), + "body": group_node(children), + } + + +def while_moving_nodes(device: str, value: Any, children: list[dict] | None = None) -> list[dict]: + """Create the two nodes behind "keep doing this while a motor moves". + + The first node starts the move without waiting and binds its status, the second loops + while that status is not done. + + Returns: + list[dict]: The start-move node and the while loop. + """ + status = _new_id("motion") + start = move_node(device, value, wait=False, bind_status=status) + loop = { + "id": _new_id("while"), + "type": "while", + "condition": { + "type": "not", + "operand": {"type": "status_field", "status_ref": status, "field": "done"}, + }, + "body": group_node(children), + } + return [start, loop] + + +def new_step(kind: str, *, device: str = "samx", scan: str = "line_scan") -> list[dict]: + """Create default nodes for a step kind picked in an editor. + + Args: + kind (str): One of the keys of ``KIND_TITLES`` (``while`` creates two nodes). + device (str): Default device for moves. + scan (str): Default scan name. + + Returns: + list[dict]: New nodes to insert. + """ + if kind == "set": + return [move_node(device, 0)] + if kind == "scan_call": + return [scan_node(scan)] + if kind == "repeat": + return [repeat_node(2)] + if kind == "for_each": + return [for_each_node("position", [0, 1])] + if kind == "while": + return while_moving_nodes(device, 5) + if kind == "sequence": + return [group_node()] + raise ValueError(f"Unknown step kind: {kind}") + + +# --------------------------------------------------------------------------------------------- +# Tree access +# --------------------------------------------------------------------------------------------- +def children_of(node: dict) -> list[dict]: + """Return the list holding a node's direct children (empty for leaves).""" + if node.get("type") == "sequence": + return node.setdefault("children", []) + if node.get("type") in LOOP_TYPES: + return children_of(node["body"]) + return [] + + +def is_container(node: dict) -> bool: + """Return whether a node can hold steps.""" + return node.get("type") in CONTAINER_TYPES + + +def normalize(doc: dict) -> dict: + """Wrap every loop body in a group, so editors can always add more than one step. + + Args: + doc (dict): Sequence document; modified in place. + + Returns: + dict: The same document. + """ + + def visit(node: dict): + if node.get("type") in LOOP_TYPES and node["body"].get("type") != "sequence": + node["body"] = group_node([node["body"]]) + for child in children_of(node): + visit(child) + + if doc["root"].get("type") != "sequence": + doc["root"] = {"id": "root", "type": "sequence", "children": [doc["root"]]} + visit(doc["root"]) + return doc + + +def walk(node: dict, depth: int = 0) -> Iterator[tuple[dict, int, dict]]: + """Yield ``(node, depth, parent)`` for every node below ``node`` (not ``node`` itself).""" + for child in children_of(node): + yield child, depth, node + yield from walk(child, depth + 1) + + +def find(root: dict, node_id: str) -> tuple[dict | None, dict | None]: + """Find a node and its container by id. + + Returns: + tuple[dict | None, dict | None]: The node and the node whose children list holds it. + """ + if root.get("id") == node_id: + return root, None + for node, _depth, parent in walk(root): + if node.get("id") == node_id: + return node, parent + return None, None + + +def remove(root: dict, node_id: str) -> dict | None: + """Remove a node by id; return it.""" + node, parent = find(root, node_id) + if node is None or parent is None: + return None + children_of(parent).remove(node) + return node + + +def insert(root: dict, nodes: list[dict], parent_id: str | None = None, index: int = -1) -> None: + """Insert nodes into a container; ``index=-1`` appends.""" + parent, _ = find(root, parent_id) if parent_id else (root, None) + if parent is None or not is_container(parent): + parent = root + target = children_of(parent) + if index < 0 or index > len(target): + index = len(target) + target[index:index] = nodes + + +def ancestors(root: dict, node_id: str) -> list[dict]: + """Return the containers enclosing a node, outermost first (root excluded).""" + path: list[dict] = [] + + def visit(node: dict, trail: list[dict]) -> bool: + for child in children_of(node): + if child.get("id") == node_id: + path.extend(trail) + return True + if visit(child, trail + [child]): + return True + return False + + visit(root, []) + return path + + +def visible_variables(root: dict, node_id: str) -> list[str]: + """Return loop variable names usable at a node.""" + return [n["variable"] for n in ancestors(root, node_id) if n.get("type") == "for_each"] + + +def status_owner(root: dict, status: str) -> dict | None: + """Return the Set node that binds a status name.""" + for node, _depth, _parent in walk(root): + if node.get("type") == "set" and node.get("bind_status") == status: + return node + return None + + +def while_status(node: dict) -> str | None: + """Return the status a "while moving" loop watches, if it uses that pattern.""" + cond = node.get("condition") + if isinstance(cond, dict) and cond.get("type") == "not": + operand = cond.get("operand") + if isinstance(operand, dict) and operand.get("type") == "status_field": + return operand.get("status_ref") + return None + + +def clone(node: dict) -> dict: + """Deep-copy a node and give every copied node a new id.""" + copied = copy.deepcopy(node) + statuses: dict[str, str] = {} + + def visit(n: dict): + n["id"] = _new_id(n["id"].rsplit("_", 1)[0] or "step") + if n.get("bind_status"): + statuses[n["bind_status"]] = _new_id("motion") + n["bind_status"] = statuses[n["bind_status"]] + if n.get("type") in LOOP_TYPES: + n["body"]["id"] = _new_id("group") + for child in children_of(n): + visit(child) + + visit(copied) + return copied + + +# --------------------------------------------------------------------------------------------- +# Plain-language text +# --------------------------------------------------------------------------------------------- +def scan_title(name: str) -> str: + """Turn ``line_scan`` into ``Line scan``.""" + words = name.replace("_", " ").strip() + return words[:1].upper() + words[1:] + + +def format_value(value: Any) -> str: + """Show a value the way a scientist would write it.""" + if is_ref(value): + return f"each {value['ref']}" + if isinstance(value, bool): + return "yes" if value else "no" + if isinstance(value, float): + return f"{value:g}" + if isinstance(value, dict) and value.get("type") == "literal": + return format_value(value.get("value")) + if isinstance(value, (dict, list)): + return json.dumps(value) + return str(value) + + +def format_values(values: Any, limit: int = 6) -> str: + """Show a list of loop values, shortened when long.""" + if not isinstance(values, list): + return format_value(values) + shown = [format_value(v) for v in values[:limit]] + if len(values) > limit: + shown.append(f"… ({len(values)} values)") + return ", ".join(shown) if shown else "no values" + + +def scan_details(node: dict, entries: dict | None = None) -> str: + """Summarize scan arguments, e.g. ``samy -2 → 2 · 20 steps · 0.1 s``.""" + entries = entries or {} + info = entries.get(node.get("name"), {}) + parts = [] + positional = node.get("positional") or [] + bundle = (info.get("arg_bundle_size") or {}).get("bundle") or 0 + if bundle == 3 and len(positional) % 3 == 0: + for i in range(0, len(positional), 3): + dev, start, stop = positional[i : i + 3] + parts.append(f"{format_value(dev)} {format_value(start)} → {format_value(stop)}") + elif positional: + parts.append(", ".join(format_value(v) for v in positional)) + args = node.get("arguments") if isinstance(node.get("arguments"), dict) else {} + for key, value in args.items(): + if key == "metadata" or value in (None, ""): + continue + if key == "steps": + parts.append(f"{format_value(value)} steps") + elif key in ("exp_time", "interval") and isinstance(value, (int, float)): + parts.append(f"{value:g} s {'exposure' if key == 'exp_time' else 'interval'}") + elif key == "relative": + parts.append("relative" if value else "absolute") + elif key in ("points",): + parts.append(f"{format_value(value)} points") + else: + parts.append(f"{key.replace('_', ' ')} {format_value(value)}") + return " · ".join(parts) + + +def sentence(node: dict, root: dict | None = None, entries: dict | None = None) -> str: + """Describe one step in plain language (without its children). + + Args: + node (dict): The step. + root (dict | None): Document root, used to name the motor a "while" waits for. + entries (dict | None): Advertised scan info by scan name. + + Returns: + str: One sentence. + """ + kind = node.get("type") + if kind == "set": + target = node.get("device", "?") + value = format_value(node.get("value")) + verb = "Set" if "." in target else "Move" + if node.get("wait", True): + return f"{verb} {target} to {value}" + return f"Start moving {target} to {value} (don't wait)" + if kind == "scan_call": + details = scan_details(node, entries) + return f"Run {scan_title(node.get('name', 'scan')).lower()}" + ( + f": {details}" if details else "" + ) + if kind == "repeat": + count = node.get("count", 1) + return f"Repeat {count} time{'s' if count != 1 else ''}" + if kind == "for_each": + return f"For each {node.get('variable')}: {format_values(node.get('values'))}" + if kind == "while": + status = while_status(node) + owner = status_owner(root, status) if (root and status) else None + if owner is not None: + return f"While {owner.get('device')} is still moving" + return "While the condition holds" + if kind == "component_call": + definition = node.get("definition", {}) + return f"Use saved block “{definition.get('id')}”" + if kind == "sequence": + return "Group" + return kind or "?" + + +# --------------------------------------------------------------------------------------------- +# Counting and preview +# --------------------------------------------------------------------------------------------- +@dataclass +class Summary: + """What a sequence will do, counted without running it.""" + + scans: int = 0 + moves: int = 0 + open_ended: bool = False + empty_loops: list[str] = field(default_factory=list) + + def text(self) -> str: + """Short summary such as ``8 scans · 4 moves``.""" + if not self.scans and not self.moves and not self.open_ended: + return "Nothing to run yet" + plus = "+" if self.open_ended else "" + parts = [f"{self.scans}{plus} scan{'s' if self.scans != 1 else ''}"] + parts.append(f"{self.moves} move{'s' if self.moves != 1 else ''}") + if self.open_ended: + parts.append("repeats while a motor moves") + return " · ".join(parts) + + +def summarize(root: dict) -> Summary: + """Count scans and moves, multiplying through loops.""" + summary = Summary() + + def visit(node: dict, factor: int): + kind = node.get("type") + if kind == "scan_call": + summary.scans += factor + elif kind == "set": + summary.moves += factor + if kind == "repeat": + factor *= max(int(node.get("count", 1)), 0) + elif kind == "for_each": + values = node.get("values") + factor *= len(values) if isinstance(values, list) else 1 + elif kind == "while": + summary.open_ended = True + if kind in LOOP_TYPES and not children_of(node): + summary.empty_loops.append(node["id"]) + for child in children_of(node): + visit(child, factor) + + for child in children_of(root): + visit(child, 1) + return summary + + +@dataclass +class TimelineItem: + """One concrete action in the order it will run.""" + + node_id: str + kind: str + text: str + context: str + + +def timeline(root: dict, entries: dict | None = None, limit: int = 300) -> list[TimelineItem]: + """Unroll loops into the list of concrete actions, substituting loop values. + + While loops are shown once, marked as repeating. + """ + items: list[TimelineItem] = [] + + def subst(value, env): + if is_ref(value) and value["ref"] in env: + return env[value["ref"]] + if isinstance(value, list): + return [subst(v, env) for v in value] + if isinstance(value, dict) and not value.get("type") == "literal": + return {k: subst(v, env) for k, v in value.items()} + return value + + def visit(node: dict, env: dict, context: list[str]) -> bool: + if len(items) >= limit: + return False + kind = node.get("type") + ctx = " · ".join(context) + if kind in ("set", "scan_call", "component_call"): + concrete = dict(node) + for key in ("value", "positional", "arguments"): + if key in concrete: + concrete[key] = subst(concrete[key], env) + items.append(TimelineItem(node["id"], kind, sentence(concrete, root, entries), ctx)) + return True + if kind == "repeat": + count = int(node.get("count", 1)) + for i in range(count): + for child in children_of(node): + if not visit(child, env, context + [f"round {i + 1}/{count}"]): + return False + return True + if kind == "for_each": + values = node.get("values") if isinstance(node.get("values"), list) else [] + for value in values: + inner = {**env, node["variable"]: subst(value, env)} + label = f"{node['variable']} = {format_value(inner[node['variable']])}" + for child in children_of(node): + if not visit(child, inner, context + [label]): + return False + return True + if kind == "while": + for child in children_of(node): + if not visit(child, env, context + ["repeats while moving"]): + return False + return True + for child in children_of(node): + if not visit(child, env, context): + return False + return True + + for child in children_of(root): + if not visit(child, {}, []): + break + return items diff --git a/bec_widgets/widgets/control/sequencer/recipe/__init__.py b/bec_widgets/widgets/control/sequencer/recipe/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/bec_widgets/widgets/control/sequencer/recipe/recipe_editor.py b/bec_widgets/widgets/control/sequencer/recipe/recipe_editor.py new file mode 100644 index 00000000..7380a320 --- /dev/null +++ b/bec_widgets/widgets/control/sequencer/recipe/recipe_editor.py @@ -0,0 +1,307 @@ +"""Concept 1, Recipe: a numbered list of plain-language steps with a details panel. + +Reads like a lab protocol. Loops are steps that hold indented steps; drag to reorder or to +move steps into a loop. The panel on the right edits the selected step. +""" + +from __future__ import annotations + +from bec_qthemes import material_icon +from qtpy.QtCore import Qt +from qtpy.QtGui import QBrush, QColor +from qtpy.QtWidgets import ( + QAbstractItemView, + QHBoxLayout, + QPushButton, + QSplitter, + QToolButton, + QTreeWidget, + QTreeWidgetItem, + QVBoxLayout, + QWidget, +) + +from bec_widgets.utils.colors import rgba +from bec_widgets.utils.error_popups import SafeSlot +from bec_widgets.widgets.control.sequencer import model +from bec_widgets.widgets.control.sequencer.editor_base import SequenceEditorBase +from bec_widgets.widgets.control.sequencer.inspector import StepInspector +from bec_widgets.widgets.control.sequencer.style import heading, hint, kind_color, kind_icon, theme + +ID_ROLE = Qt.ItemDataRole.UserRole + 1 + +ADD_BUTTONS = [ + ("set", "Move"), + ("scan_call", "Scan"), + ("repeat", "Repeat"), + ("for_each", "For each value"), + ("while", "While moving"), +] + + +class RecipeTree(QTreeWidget): + """Tree whose drag and drop moves steps in the document.""" + + def __init__(self, editor: "RecipeEditor"): + super().__init__(editor) + self.editor = editor + self.setHeaderHidden(True) + self.setColumnCount(2) + self.setIndentation(26) + self.setUniformRowHeights(False) + self.setRootIsDecorated(True) + self.setDragDropMode(QAbstractItemView.DragDropMode.InternalMove) + self.setDefaultDropAction(Qt.DropAction.MoveAction) + self.setSelectionMode(QAbstractItemView.SelectionMode.SingleSelection) + self.setAnimated(True) + self.setExpandsOnDoubleClick(False) + + def dropEvent(self, event): # pylint: disable=invalid-name + """Translate a drop into a document move.""" + source = self.currentItem() + if source is None: + return event.ignore() + node_id = source.data(0, ID_ROLE) + target = self.itemAt(event.position().toPoint()) + indicator = self.dropIndicatorPosition() + doc = self.editor.document + if target is None or indicator == QAbstractItemView.DropIndicatorPosition.OnViewport: + parent_id, index = None, -1 + else: + target_id = target.data(0, ID_ROLE) + target_node, target_parent = model.find(doc.root, target_id) + if indicator == QAbstractItemView.DropIndicatorPosition.OnItem and model.is_container( + target_node + ): + parent_id, index = target_id, -1 + else: + siblings = model.children_of(target_parent) + index = siblings.index(target_node) + if indicator == QAbstractItemView.DropIndicatorPosition.BelowItem or ( + indicator == QAbstractItemView.DropIndicatorPosition.OnItem + ): + index += 1 + source_node, source_parent = model.find(doc.root, node_id) + if source_parent is target_parent and siblings.index(source_node) < index: + index -= 1 + parent_id = target_parent["id"] + event.setDropAction(Qt.DropAction.IgnoreAction) + event.accept() + doc.move(node_id, parent_id, index) + + +class RecipeEditor(SequenceEditorBase): + """Numbered recipe of plain-language steps, with a details panel.""" + + CONCEPT_ID = "recipe" + CONCEPT_TITLE = "Recipe" + CONCEPT_TAGLINE = "A numbered list of plain-language steps, like a lab protocol." + + def build_editor(self, host: QWidget) -> None: + layout = QHBoxLayout(host) + layout.setContentsMargins(0, 0, 0, 0) + splitter = QSplitter(Qt.Orientation.Horizontal, host) + layout.addWidget(splitter) + + left = QWidget(splitter) + left_layout = QVBoxLayout(left) + left_layout.setContentsMargins(16, 14, 8, 8) + left_layout.setSpacing(10) + left_layout.addWidget(heading("Sequence steps", left)) + left_layout.addWidget( + hint("Steps run from top to bottom. Indented steps belong to the loop above.", left) + ) + + add_row = QHBoxLayout() + add_row.setSpacing(6) + for kind, text in ADD_BUTTONS: + btn = QPushButton(f"{text}", left) + btn.setIcon(kind_icon(kind, 16)) + btn.setToolTip(f"Add a “{model.KIND_TITLES[kind]}” step after the selected one") + btn.clicked.connect(lambda _=False, k=kind: self.add_step(k)) + add_row.addWidget(btn) + add_row.addStretch(1) + left_layout.addLayout(add_row) + + self.tree = RecipeTree(self) + self.tree.itemSelectionChanged.connect(self._on_tree_selection) + left_layout.addWidget(self.tree, 1) + + action_row = QHBoxLayout() + action_row.setSpacing(2) + self.action_buttons = {} + for key, icon, tip in ( + ("up", "arrow_upward", "Move up"), + ("down", "arrow_downward", "Move down"), + ("duplicate", "content_copy", "Duplicate"), + ("wrap_repeat", "repeat", "Put inside a Repeat loop"), + ("wrap_each", "format_list_numbered", "Put inside a For each loop"), + ("delete", "delete", "Delete step"), + ("undo", "undo", "Undo"), + ): + btn = QToolButton(left) + btn.setIcon(material_icon(icon, size=(18, 18), convert_to_pixmap=False)) + btn.setToolTip(tip) + btn.setAutoRaise(True) + btn.clicked.connect(lambda _=False, k=key: self._action(k)) + action_row.addWidget(btn) + self.action_buttons[key] = btn + action_row.addStretch(1) + left_layout.addLayout(action_row) + + self.inspector = StepInspector(self.document, splitter, client=self.client) + self.inspector.setMinimumWidth(340) + splitter.addWidget(left) + splitter.addWidget(self.inspector) + splitter.setStretchFactor(0, 3) + splitter.setStretchFactor(1, 2) + self.document.selection_changed.connect(self._select_in_tree) + self._style_tree() + + def _style_tree(self) -> None: + border = theme("BORDER", "#d0d0d0").name() + primary = theme("PRIMARY", "#2f6fdf") + self.tree.setStyleSheet( + f"QTreeWidget {{ border: 1px solid {border}; border-radius: 8px; padding: 4px;" + f" background: {theme('CARD_BG', '#ffffff').name()}; }}" + "QTreeWidget { show-decoration-selected: 0; }" + "QTreeWidget::item { padding: 7px 4px; border-radius: 6px; }" + f"QTreeWidget::item:selected {{ background: {rgba(primary, 50)};" + " color: palette(text); }" + ) + + # --- drawing --------------------------------------------------------------------------- + @SafeSlot() + def refresh(self) -> None: + if not hasattr(self, "tree"): + return + entries = self.scan_entries() + self.tree.blockSignals(True) + self.tree.clear() + root = self.document.root + + def add(node: dict, parent_item, number: str): + text = model.sentence(node, root, entries) + item = QTreeWidgetItem([f"{number} {text}", self._side_text(node)]) + item.setData(0, ID_ROLE, node["id"]) + item.setIcon(0, kind_icon(node["type"], 18)) + flags = item.flags() | Qt.ItemFlag.ItemIsDragEnabled + if model.is_container(node): + flags |= Qt.ItemFlag.ItemIsDropEnabled + font = item.font(0) + font.setBold(True) + item.setFont(0, font) + else: + flags &= ~Qt.ItemFlag.ItemIsDropEnabled + item.setFlags(flags) + item.setForeground(1, QBrush(kind_color(node["type"]))) + item.setTextAlignment(1, Qt.AlignmentFlag.AlignRight | Qt.AlignmentFlag.AlignVCenter) + if parent_item is None: + self.tree.addTopLevelItem(item) + else: + parent_item.addChild(item) + for index, child in enumerate(model.children_of(node), 1): + add(child, item, f"{number}{index}.") + if model.is_container(node) and not model.children_of(node): + empty = QTreeWidgetItem(["Drop or add steps here"]) + empty.setFlags(Qt.ItemFlag.ItemIsEnabled) + empty.setForeground(0, QBrush(theme("DISABLED_FG", "#999999"))) + font = empty.font(0) + font.setItalic(True) + empty.setFont(0, font) + item.addChild(empty) + item.setExpanded(True) + + for index, child in enumerate(model.children_of(root), 1): + add(child, None, f"{index}.") + self.tree.resizeColumnToContents(1) + self.tree.setColumnWidth(1, max(self.tree.columnWidth(1), 110)) + self.tree.header().setStretchLastSection(False) + self.tree.header().setSectionResizeMode(0, self.tree.header().ResizeMode.Stretch) + self.tree.blockSignals(False) + self._select_in_tree(self.document.selected) + self.show_problems(self.problems) + + @staticmethod + def _side_text(node: dict) -> str: + kind = node["type"] + if kind == "repeat": + return f"× {node.get('count', 1)}" + if kind == "for_each": + values = node.get("values") + return f"{len(values)} values" if isinstance(values, list) else "" + if kind == "while": + return "until it arrives" + if kind == "set": + return "waits" if node.get("wait", True) else "keeps going" + return "" + + def show_problems(self, problems) -> None: + bad = {p.node_id: p.message for p in problems} + red = theme("ACCENT_EMERGENCY", "#e0534b") + iterator = [self.tree.topLevelItem(i) for i in range(self.tree.topLevelItemCount())] + while iterator: + item = iterator.pop() + iterator.extend(item.child(i) for i in range(item.childCount())) + node_id = item.data(0, ID_ROLE) + if node_id in bad: + item.setBackground(0, QBrush(QColor(red.red(), red.green(), red.blue(), 45))) + item.setToolTip(0, bad[node_id]) + item.setText(1, "needs a fix") + item.setForeground(1, QBrush(red)) + elif node_id: + item.setBackground(0, QBrush()) + item.setToolTip(0, "") + + # --- interaction ----------------------------------------------------------------------- + def add_step(self, kind: str) -> None: + """Add a new step next to the selection.""" + devices = self.backend.device_names() + device = "samx" if "samx" in devices or not devices else devices[0] + self.document.add_next_to_selection(model.new_step(kind, device=device)) + + @SafeSlot() + def _on_tree_selection(self) -> None: + item = self.tree.currentItem() + node_id = item.data(0, ID_ROLE) if item is not None else "" + if node_id: + self.document.select(node_id) + + @SafeSlot(str) + def _select_in_tree(self, node_id: str) -> None: + iterator = [self.tree.topLevelItem(i) for i in range(self.tree.topLevelItemCount())] + while iterator: + item = iterator.pop(0) + if item.data(0, ID_ROLE) == node_id: + self.tree.blockSignals(True) + self.tree.setCurrentItem(item) + self.tree.blockSignals(False) + return + iterator.extend(item.child(i) for i in range(item.childCount())) + + def _action(self, key: str) -> None: + doc = self.document + if key == "undo": + doc.undo() + return + node_id = doc.selected + if not node_id: + return + if key == "up": + doc.shift(node_id, -1) + elif key == "down": + doc.shift(node_id, 1) + elif key == "duplicate": + doc.duplicate(node_id) + elif key == "wrap_repeat": + doc.wrap(node_id, "repeat") + elif key == "wrap_each": + doc.wrap(node_id, "for_each") + elif key == "delete": + doc.delete(node_id) + + def screenshot_setup(self) -> None: + """Select the scan step (used for screenshots and the demo).""" + scans = [n for n, _d, _p in model.walk(self.document.root) if n["type"] == "scan_call"] + if scans: + self.document.select(scans[0]["id"]) diff --git a/bec_widgets/widgets/control/sequencer/run_bar.py b/bec_widgets/widgets/control/sequencer/run_bar.py new file mode 100644 index 00000000..1b027f70 --- /dev/null +++ b/bec_widgets/widgets/control/sequencer/run_bar.py @@ -0,0 +1,250 @@ +"""Bottom bar shared by all concepts: what will happen, is it valid, preview, save, run.""" + +from __future__ import annotations + +import json +import time +from pathlib import Path + +from bec_qthemes import material_icon +from qtpy.QtCore import Qt, QTimer, Signal +from qtpy.QtGui import QFont +from qtpy.QtWidgets import ( + QDialog, + QFileDialog, + QHBoxLayout, + QLabel, + QListWidget, + QListWidgetItem, + QPlainTextEdit, + QPushButton, + QTabWidget, + QVBoxLayout, + QWidget, +) + +from bec_widgets.utils.error_popups import SafeSlot +from bec_widgets.widgets.control.sequencer import model +from bec_widgets.widgets.control.sequencer.backend import Problem, SequencerBackend +from bec_widgets.widgets.control.sequencer.document import SequenceDocument +from bec_widgets.widgets.control.sequencer.style import Chip, kind_color, kind_icon, theme + + +class PreviewDialog(QDialog): + """Show a sequence step by step, as the backend describes it, and as Python.""" + + def __init__(self, backend: SequencerBackend, doc: dict, parent: QWidget | None = None): + super().__init__(parent) + self.setWindowTitle("Preview sequence") + self.resize(720, 520) + layout = QVBoxLayout(self) + tabs = QTabWidget(self) + layout.addWidget(tabs) + + steps = QListWidget(tabs) + entries = backend.scan_entries() + for index, item in enumerate(model.timeline(doc["root"], entries), 1): + row = QListWidgetItem(kind_icon(item.kind, 16), f"{index:>3}. {item.text}") + if item.context: + row.setToolTip(item.context) + row.setText(f"{row.text()} ({item.context})") + steps.addItem(row) + tabs.addTab(steps, "Step by step") + + mono = QFont("Menlo") + mono.setStyleHint(QFont.StyleHint.Monospace) + for title, producer in (("In words", backend.describe), ("Python", backend.python)): + text = QPlainTextEdit(tabs) + text.setReadOnly(True) + text.setFont(mono) + try: + text.setPlainText(producer(doc)) + except Exception as exc: # pylint: disable=broad-except + text.setPlainText(f"Not available until the problems are fixed:\n{exc}") + tabs.addTab(text, title) + + +class RunBar(QWidget): + """Summary, validation status and the Run/Stop buttons for one document.""" + + problems_changed = Signal(list) + + def __init__( + self, backend: SequencerBackend, document: SequenceDocument, parent: QWidget | None = None + ): + super().__init__(parent) + self.backend = backend + self.document = document + self._status = None + self._started = 0.0 + self.problems: list[Problem] = [] + self.setObjectName("SequencerRunBar") + self.setAttribute(Qt.WidgetAttribute.WA_StyledBackground, True) + + layout = QHBoxLayout(self) + layout.setContentsMargins(14, 8, 14, 8) + layout.setSpacing(10) + self.summary = QLabel(self) + font = self.summary.font() + font.setBold(True) + self.summary.setFont(font) + self.state = Chip("", parent=self) + self.message = QLabel(self) + self.message.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse) + layout.addWidget(self.summary) + layout.addWidget(self.state) + layout.addWidget(self.message, 1) + + def button(text, icon, tip): + btn = QPushButton(text, self) + btn.setIcon(material_icon(icon, size=(18, 18), convert_to_pixmap=False)) + btn.setToolTip(tip) + layout.addWidget(btn) + return btn + + self.open_button = button("Open", "folder_open", "Open a saved sequence") + self.save_button = button("Save", "save", "Save this sequence to a file") + self.preview_button = button("Preview", "visibility", "See every step that will run") + self.run_button = button("Run sequence", "play_arrow", "Start the sequence") + self.run_button.setProperty("variant", "success") + self.stop_button = button("Stop", "stop", "Stop the running sequence") + self.stop_button.setProperty("variant", "danger") + self.stop_button.setVisible(False) + + self.open_button.clicked.connect(self.open_file) + self.save_button.clicked.connect(self.save_file) + self.preview_button.clicked.connect(self.preview) + self.run_button.clicked.connect(self.run) + self.stop_button.clicked.connect(self.stop) + + self._check_timer = QTimer(self, singleShot=True, interval=300) + self._check_timer.timeout.connect(self.check) + self._poll_timer = QTimer(self, interval=500) + self._poll_timer.timeout.connect(self._poll) + document.changed.connect(self._check_timer.start) + self._apply_style() + self.check() + + def _apply_style(self) -> None: + self.setStyleSheet( + f"#SequencerRunBar {{ background: {theme('CARD_BG', '#ffffff').name()};" + f" border-top: 1px solid {theme('BORDER', '#d0d0d0').name()}; }}" + ) + + # --- checking -------------------------------------------------------------------------- + @SafeSlot() + def check(self) -> None: + """Count steps and validate with the backend.""" + summary = model.summarize(self.document.root) + self.summary.setText(summary.text()) + if self._status is not None: + return + if not self.backend.available: + self.problems = [] + self._set_state("Editing only", theme("ACCENT_WARNING", "#eda200")) + self.message.setText("Install bec with the scan sequencer (bec PR #1128) to run.") + self.run_button.setEnabled(False) + self.problems_changed.emit([]) + return + if not summary.scans and not summary.moves: + self.problems = [] + self._set_state("Empty", theme("BORDER", "#999999")) + self.message.setText("Add a step to get started.") + self.run_button.setEnabled(False) + self.problems_changed.emit([]) + return + self.problems = self.backend.validate(self.document.doc) + if summary.empty_loops and not self.problems: + self.problems = [Problem(summary.empty_loops[0], "This loop has no steps inside")] + if self.problems: + problem = self.problems[0] + self._set_state("Needs a fix", theme("ACCENT_EMERGENCY", "#e0534b")) + node = self.document.node(problem.node_id) if problem.node_id else None + where = f"{model.KIND_TITLES.get(node['type'], 'Step')}: " if node else "" + self.message.setText(where + problem.message) + else: + self._set_state("Ready to run", theme("ACCENT_SUCCESS", "#59a869")) + self.message.setText("") + self.run_button.setEnabled(not self.problems) + self.problems_changed.emit(self.problems) + + def _set_state(self, text: str, color) -> None: + self.state.setText(text) + self.state.set_color(color) + + # --- files ----------------------------------------------------------------------------- + @SafeSlot() + def open_file(self) -> None: + """Load a sequence JSON file.""" + path, _ = QFileDialog.getOpenFileName(self, "Open sequence", "", "Sequence (*.json)") + if path: + self.document.set_document(json.loads(Path(path).read_text(encoding="utf-8"))) + + @SafeSlot() + def save_file(self) -> None: + """Save the sequence as JSON.""" + name = f"{self.document.doc.get('id', 'sequence')}.json" + path, _ = QFileDialog.getSaveFileName(self, "Save sequence", name, "Sequence (*.json)") + if path: + Path(path).write_text(json.dumps(self.document.doc, indent=2), encoding="utf-8") + + @SafeSlot() + def preview(self) -> None: + """Open the preview dialog.""" + PreviewDialog(self.backend, self.document.doc, self).exec() + + # --- running --------------------------------------------------------------------------- + @SafeSlot(popup_error=True) + def run(self) -> None: + """Queue the sequence.""" + self.check() + if self.problems: + return + doc = self.document.doc + doc["revision"] = int(doc.get("revision", 1)) + 1 + self._status = self.backend.submit(doc) + self._started = time.monotonic() + self.run_button.setVisible(False) + self.stop_button.setVisible(True) + self._set_state("Queued", kind_color("set")) + self.message.setText("Waiting for the procedure worker…") + self._poll_timer.start() + + @SafeSlot(popup_error=True) + def stop(self) -> None: + """Ask the worker to stop the sequence.""" + if self._status is not None and not self._status.done: + self._status.cancel() + self.message.setText("Stopping… (a move that already started may finish first)") + + @SafeSlot() + def _poll(self) -> None: + status = self._status + if status is None: + self._poll_timer.stop() + return + state = status.state.value + elapsed = int(time.monotonic() - self._started) + if state in ("Running", "Scheduled", "Requested"): + label = {"Running": "Running", "Scheduled": "Queued", "Requested": "Queued"}[state] + self._set_state(label, kind_color("set")) + self.message.setText(f"{elapsed // 60}:{elapsed % 60:02d} elapsed") + return + self._poll_timer.stop() + self._status = None + self.run_button.setVisible(True) + self.stop_button.setVisible(False) + if state == "Success": + self._set_state("Finished", theme("ACCENT_SUCCESS", "#59a869")) + self.message.setText(f"Done after {elapsed // 60}:{elapsed % 60:02d}") + else: + self._set_state( + "Stopped" if "Aborted" in str(status.error) else "Failed", + theme("ACCENT_EMERGENCY", "#e0534b"), + ) + self.message.setText(str(status.error or state).splitlines()[-1][:200]) + + def cleanup(self) -> None: + """Stop timers.""" + self._check_timer.stop() + self._poll_timer.stop() diff --git a/bec_widgets/widgets/control/sequencer/style.py b/bec_widgets/widgets/control/sequencer/style.py new file mode 100644 index 00000000..205d66d0 --- /dev/null +++ b/bec_widgets/widgets/control/sequencer/style.py @@ -0,0 +1,128 @@ +"""Colors, icons and small shared widgets for the sequencer editors.""" + +from __future__ import annotations + +from bec_qthemes import material_icon +from qtpy.QtCore import Qt +from qtpy.QtGui import QColor +from qtpy.QtWidgets import QApplication, QFrame, QLabel, QWidget + +from bec_widgets.utils.colors import get_theme_name, rgba + +# One color per kind of step, used by every concept so they read the same. +KIND_COLORS = { + "light": { + "set": "#2f6fdf", + "scan_call": "#23926f", + "repeat": "#c77a00", + "for_each": "#8a53c9", + "while": "#c2416b", + "sequence": "#6b7280", + "component_call": "#0e8a9a", + }, + "dark": { + "set": "#6c9cf2", + "scan_call": "#4cc09a", + "repeat": "#e8a33a", + "for_each": "#b48ae6", + "while": "#e2729a", + "sequence": "#9aa0aa", + "component_call": "#3fbccc", + }, +} + +KIND_ICONS = { + "set": "open_with", + "scan_call": "query_stats", + "repeat": "repeat", + "for_each": "format_list_numbered", + "while": "hourglass_top", + "sequence": "folder_open", + "component_call": "widgets", +} + + +def is_dark() -> bool: + """Return whether the dark theme is active.""" + return get_theme_name() == "dark" + + +def kind_color(kind: str) -> QColor: + """Color for a step kind in the current theme.""" + colors = KIND_COLORS["dark" if is_dark() else "light"] + return QColor(colors.get(kind, colors["sequence"])) + + +def kind_icon(kind: str, size: int = 18, color: QColor | str | None = None): + """Material icon for a step kind.""" + color = color if color is not None else kind_color(kind) + return material_icon( + KIND_ICONS.get(kind, "radio_button_unchecked"), + size=(size, size), + color=QColor(color).name() if not isinstance(color, str) else color, + convert_to_pixmap=False, + ) + + +def theme(key: str, fallback: str) -> QColor: + """Theme color by key, e.g. ``CARD_BG`` or ``BORDER``.""" + app = QApplication.instance() + if app is not None and hasattr(app, "theme"): + return app.theme.color(key, fallback) + return QColor(fallback) + + +def card_qss(selector: str, radius: int = 8) -> str: + """Style sheet for a card: card background with a thin border.""" + return ( + f"{selector} {{ background: {theme('CARD_BG', '#ffffff').name()};" + f" border: 1px solid {theme('BORDER', '#d0d0d0').name()}; border-radius: {radius}px; }}" + ) + + +def chip_qss(color: QColor, selector: str = "QLabel") -> str: + """Style sheet for a small tinted pill.""" + return ( + f"{selector} {{ background: {rgba(color, 38)}; color: {color.name()};" + " border-radius: 9px; padding: 1px 8px; font-weight: 600; }" + ) + + +class Chip(QLabel): + """A small tinted pill, e.g. "8 scans" or "Ready to run".""" + + def __init__(self, text: str = "", color: QColor | None = None, parent: QWidget | None = None): + super().__init__(text, parent) + self.set_color(color or theme("PRIMARY", "#2f6fdf")) + + def set_color(self, color: QColor) -> None: + """Change the tint.""" + self.setStyleSheet(chip_qss(color)) + + +def hline(parent: QWidget | None = None) -> QFrame: + """A thin horizontal divider.""" + line = QFrame(parent) + line.setFrameShape(QFrame.Shape.HLine) + line.setStyleSheet(f"color: {theme('BORDER', '#d0d0d0').name()};") + return line + + +def heading(text: str, parent: QWidget | None = None, size: int = 15) -> QLabel: + """A bold section heading.""" + label = QLabel(text, parent) + font = label.font() + font.setPointSize(size) + font.setBold(True) + label.setFont(font) + return label + + +def hint(text: str, parent: QWidget | None = None) -> QLabel: + """Muted helper text that wraps.""" + label = QLabel(text, parent) + label.setWordWrap(True) + fg = theme("FG", "#333333") + label.setStyleSheet(f"color: {rgba(fg, 150)};") + label.setTextInteractionFlags(Qt.TextInteractionFlag.NoTextInteraction) + return label