feat(sequencer): add sequencer view with the recipe concept

This commit is contained in:
wyzula_j committed 2026-10-23 23:32:58 +00:00
1 parent e5e1bb1386
commit b4d870099e
14 files changed
+2345

No files matched your search

+9
View File
@@ -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(
Whitespace-only changes.
@@ -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,)}
@@ -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()
Whitespace-only changes.
@@ -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<path>[^:]+): (?P<message>.*)$", 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
)
@@ -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]
@@ -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()
@@ -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()
@@ -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
Whitespace-only changes.
@@ -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"])
@@ -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()
@@ -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