diff --git a/debye_bec/bec_widgets/widgets/client.py b/debye_bec/bec_widgets/widgets/client.py index c974339..0fc46e6 100644 --- a/debye_bec/bec_widgets/widgets/client.py +++ b/debye_bec/bec_widgets/widgets/client.py @@ -69,7 +69,7 @@ class DigitalTwin(RPCBase): class RestartServer(RPCBase): - """Main widget of server restart widget""" + """Main widget of server restart widget.""" _IMPORT_MODULE = "debye_bec.bec_widgets.widgets.restart_server.restart_server" @@ -115,3 +115,102 @@ class ScanControlXAS(RPCBase): """ Take a screenshot of the dock area and save it to a file. """ + + +class Scheduler(RPCBase): + """Schedule, persist and execute a sequence of BEC scan/device commands.""" + + _IMPORT_MODULE = "debye_bec.bec_widgets.widgets.scheduler.scheduler" + + @rpc_call + def run_schedule(self): + """ + Run the schedule, continuing from wherever it last left off. + """ + + @rpc_call + def abort_schedule(self): + """ + Request that the schedule stop after the current item. + """ + + @rpc_call + def reset_schedule(self): + """ + Clear all execution state and start the schedule over from item 1. + """ + + @rpc_call + def get_status(self) -> "dict": + """ + RPC-exposed: current schedule state, e.g. for another widget or a script. + """ + + @rpc_call + def add_item( + self, + command: "str", + index: "int | None" = None, + kind: "str" = "custom", + form_state: "dict | None" = None, + ) -> "str": + """ + RPC-exposed: insert a new, PENDING command into the schedule. + Safe to call while the schedule is running. + + Args: + command: command text, evaluated the same way as the existing + items (against `scans`/`dev`) once the schedule runs. + index: position to insert at (0 = first). Clamped so the item + can never land before something already running or + finished. Defaults to appending at the end. + kind/form_state: optional structured description of how + `command` was built (see `schedule_item.ScheduleItem`), + used to reopen the Edit dialog pre-filled. Leave as + defaults for a plain, hand-typed command. + + Returns: + The new item's item_id. + """ + + @rpc_call + def edit_item( + self, + item_id: "str", + command: "str", + kind: "str" = "custom", + form_state: "dict | None" = None, + ): + """ + RPC-exposed: change the command of an item that has not started + yet. Raises `RuntimeError` for an item that is already + running/finished. + """ + + @rpc_call + def delete_item(self, item_id: "str"): + """ + RPC-exposed: remove an item that has not started yet. Raises + `RuntimeError` for an item that is already running/finished. + """ + + @rpc_call + def move_item(self, item_id: "str", new_index: "int"): + """ + RPC-exposed: move an item that has not started yet to a new + position (0 = first, but never before something already + running/finished). Raises `RuntimeError` for an item that is + already running/finished. + """ + + @rpc_call + def move_item_up(self, item_id: "str"): + """ + RPC-exposed: swap an item with the one directly before it. + """ + + @rpc_call + def move_item_down(self, item_id: "str"): + """ + RPC-exposed: swap an item with the one directly after it. + """ diff --git a/debye_bec/bec_widgets/widgets/scheduler/__init__.py b/debye_bec/bec_widgets/widgets/scheduler/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/debye_bec/bec_widgets/widgets/scheduler/endpoints.py b/debye_bec/bec_widgets/widgets/scheduler/endpoints.py new file mode 100644 index 0000000..06fb111 --- /dev/null +++ b/debye_bec/bec_widgets/widgets/scheduler/endpoints.py @@ -0,0 +1,61 @@ +""" +Redis endpoint(s) for the schedule widget plugin. + +Design notes (why this looks the way it does) +----------------------------------------------- +BEC never touches Redis with raw keys/commands. Every channel is described +by an `EndpointInfo` (endpoint string + message type + allowed operations, +see `bec_lib.endpoints`), and `RedisConnector` enforces both: calling an +operation that isn't in the endpoint's `MessageOp` raises +`IncompatibleRedisOperation`, and passing a message that isn't an instance +of the endpoint's declared `message_type` raises +`IncompatibleMessageForEndpoint`. Plain string topics still work but are +deprecated. So a "conforming" custom endpoint means building a real +`EndpointInfo`, exactly like `bec_lib.endpoints.MessageEndpoints` does +internally. + +Namespace: we use `EndpointType.USER` ("user/...”), the same prefix BEC's +own `MessageEndpoints.scan_queue_schedule()` uses for user-writable, +persisted data (as opposed to `internal/`, `public/`, etc.). + +Message type: we deliberately do NOT declare our own `BECMessage` +subclass. BEC's msgpack codec (`bec_lib.codecs.BECMessageEncoder.decode`) +resolves an incoming message's class by name via +`getattr(bec_lib.messages, type_name)` - i.e. only classes physically +defined inside `bec_lib.messages` are resolvable this way. A message class +declared in plugin code would fail to deserialize unless you monkey-patch +it into that module, which is exactly the kind of side-door this plugin +is meant to avoid. Instead we reuse `bec_lib.messages.VariableMessage`, +a first-class, exported message type built for carrying an arbitrary +(msgpack-serializable) payload under `.value`. +`bec_lib.script_executor.upload_script` persists a script's text in Redis +the same way, for the same reason - it's the established BEC pattern for +"a plugin needs to stash its own structured data in Redis". +""" + +from __future__ import annotations + +from bec_lib.endpoints import EndpointInfo, EndpointType, MessageOp +from bec_lib.messages import VariableMessage + + +def schedule(schedule_name: str) -> EndpointInfo: + """ + Endpoint for one named widget schedule (an ordered list of commands + plus their execution status). The whole schedule is stored as a single + `VariableMessage` document that gets overwritten and republished on + every change (`MessageOp.SET_PUBLISH`), so any other subscriber - + another instance of this widget, a monitoring script, ... - stays in + sync live. + + Args: + schedule_name: a stable, user-chosen name for the schedule. Unlike + a widget's `gui_id` (regenerated every time the widget is + constructed), this name is what lets a widget that was closed + and reopened find its own previously persisted schedule again. + """ + return EndpointInfo( + endpoint=f"{EndpointType.USER.value}/schedule_widget/schedule/{schedule_name}", + message_type=VariableMessage, + message_op=MessageOp.SET_PUBLISH, + ) diff --git a/debye_bec/bec_widgets/widgets/scheduler/guard.py b/debye_bec/bec_widgets/widgets/scheduler/guard.py new file mode 100644 index 0000000..5fd69d5 --- /dev/null +++ b/debye_bec/bec_widgets/widgets/scheduler/guard.py @@ -0,0 +1,160 @@ +""" +A small, self-contained utility for "if signal X drops below A, pause; once +it's back above B, resume" (hysteresis) behavior - e.g. auto-pausing scans +while the beam current is too low. + +This module knows nothing about schedules, scans, or the rest of this +plugin. It only monitors one BEC device's live readback value +(`MessageEndpoints.device_readback`, the same endpoint the device server +publishes for every monitored device) and emits Qt signals when the value +crosses one of the two configured thresholds. `schedule_widget.py` is the +only place that connects those signals to schedule-specific behavior +(aborting/deferring a scan item) - see its module docstring for how. +""" + +from __future__ import annotations + +import threading + +from bec_lib.endpoints import MessageEndpoints +from bec_lib.logger import bec_logger +from pydantic import BaseModel +from qtpy.QtCore import QObject, Signal + +logger = bec_logger.logger + + +class GuardSettings(BaseModel): + """Persisted configuration for one `SignalGuard`.""" + + enabled: bool = False + device_name: str | None = None + pause_below: float | None = None + resume_above: float | None = None + + +class SignalGuard(QObject): + """ + Monitors one device's readback value and reports hysteresis-based + pause/resume crossings. + + - `paused` fires the first time the value drops below `pause_below` + (not again while it stays low). + - `resumed` fires the first time it then climbs back above + `resume_above` (not again while it stays high). + A value sitting between the two thresholds never re-triggers either + signal - that gap is the point of using two thresholds instead of one, + so a value oscillating right at a single cutoff wouldn't cause rapid + pause/resume flapping. + + Callbacks from `RedisConnector.register()` run on a background + (Redis-listener) thread, not the Qt GUI thread; `paused`/`resumed` are + Qt signals, so connecting to them with the default (auto) connection + type safely marshals delivery onto whatever thread the receiver lives + on - no extra locking needed on the receiving end. + """ + + paused = Signal(float) + resumed = Signal(float) + + def __init__(self, connector, parent=None): + super().__init__(parent) + self._connector = connector + self._lock = threading.Lock() + self._clear_event = threading.Event() + self._clear_event.set() # not blocking until configured/proven otherwise + + self.enabled = False + self.device_name: str | None = None + self.pause_below: float | None = None + self.resume_above: float | None = None + self.current_value: float | None = None + + self._subscribed_endpoint = None + + def configure(self, settings: GuardSettings): + """(Re)configure and (re)subscribe. Safe to call repeatedly, e.g. after editing settings.""" + self._unsubscribe() + + self.enabled = settings.enabled + self.device_name = settings.device_name + self.pause_below = settings.pause_below + self.resume_above = settings.resume_above + self.current_value = None + self._clear_event.set() + + if self.enabled and self.device_name and self.pause_below is not None and self.resume_above is not None: + self._subscribed_endpoint = MessageEndpoints.device_readback(self.device_name) + self._connector.register(topics=self._subscribed_endpoint, cb=self._on_readback) + + def _unsubscribe(self): + if self._subscribed_endpoint is not None: + try: + self._connector.unregister(topics=self._subscribed_endpoint, cb=self._on_readback) + except Exception: # pylint: disable=broad-except + logger.exception("Failed to unsubscribe SignalGuard from %s", self._subscribed_endpoint) + self._subscribed_endpoint = None + + def _on_readback(self, msg): + device_msg = getattr(msg, "value", None) + if device_msg is None: + return + value = _extract_value(device_msg, self.device_name) + if value is None: + return + + crossed_pause = crossed_resume = False + with self._lock: + self.current_value = value + if self._clear_event.is_set(): + if value < self.pause_below: + self._clear_event.clear() + crossed_pause = True + elif value > self.resume_above: + self._clear_event.set() + crossed_resume = True + + if crossed_pause: + logger.info("SignalGuard: %s dropped to %s (below %s) - pausing.", self.device_name, value, self.pause_below) + self.paused.emit(value) + elif crossed_resume: + logger.info("SignalGuard: %s recovered to %s (above %s) - resuming.", self.device_name, value, self.resume_above) + self.resumed.emit(value) + + def is_clear(self) -> bool: + """True if not currently blocking (disabled, or value above the resume threshold).""" + return (not self.enabled) or self._clear_event.is_set() + + def wait_until_clear(self, should_abort, poll_interval: float = 0.5) -> bool: + """ + Blocks the calling (non-GUI) thread until `is_clear()` becomes + True, checking `should_abort()` between polls so an operator abort + can still interrupt the wait. Returns True if it cleared, False if + `should_abort()` returned True first. + """ + if self.is_clear(): + return True + while not should_abort(): + if self._clear_event.wait(timeout=poll_interval): + return True + return False + + def cleanup(self): + self._unsubscribe() + + +def _extract_value(device_msg, device_name: str | None): + """ + `DeviceMessage.signals` is keyed by signal name, e.g. {"beam_current": + {"value": ..., "timestamp": ...}, ...}. For a simple scalar device the + primary signal is usually named after the device itself; fall back to + the first signal found if not (e.g. a differently-named primary signal + on a compound device). + """ + signals = getattr(device_msg, "signals", None) or {} + if device_name in signals: + return signals[device_name].get("value") + for signal in signals.values(): + if "value" in signal: + return signal["value"] + return None diff --git a/debye_bec/bec_widgets/widgets/scheduler/guard_dialog.py b/debye_bec/bec_widgets/widgets/scheduler/guard_dialog.py new file mode 100644 index 0000000..3a7654f --- /dev/null +++ b/debye_bec/bec_widgets/widgets/scheduler/guard_dialog.py @@ -0,0 +1,94 @@ +"""Small settings dialog for `guard.GuardSettings` - kept separate from the guard logic itself.""" + +from __future__ import annotations + +from qtpy.QtWidgets import ( + QCheckBox, + QComboBox, + QDialog, + QDialogButtonBox, + QDoubleSpinBox, + QFormLayout, + QLabel, + QMessageBox, + QVBoxLayout, +) + +from .guard import GuardSettings + +_DSPIN_RANGE = (-1e12, 1e12) + + +class GuardSettingsDialog(QDialog): + """Configure (or disable) the auto-pause/resume guard for scan items.""" + + def __init__(self, settings: GuardSettings, device_names: list[str], parent=None): + super().__init__(parent) + self.setWindowTitle("Auto-pause on signal") + + layout = QVBoxLayout(self) + info_label = QLabel( + "If enabled, the currently running scan is aborted as soon as the chosen " + "signal drops below the pause value, and automatically restarted once it " + "rises back above the resume value. Only scan items are affected - device " + "moves and custom/RPC commands are never interrupted by this." + ) + info_label.setWordWrap(True) + layout.addWidget(info_label) + + form = QFormLayout() + layout.addLayout(form) + + self.enabled_check = QCheckBox("Enable auto-pause") + self.enabled_check.setChecked(settings.enabled) + form.addRow("", self.enabled_check) + + self.device_combo = QComboBox() + self.device_combo.addItems(device_names) + if settings.device_name: + idx = self.device_combo.findText(settings.device_name) + if idx >= 0: + self.device_combo.setCurrentIndex(idx) + form.addRow("Signal (device)", self.device_combo) + + self.pause_below_spin = QDoubleSpinBox() + self.pause_below_spin.setDecimals(4) + self.pause_below_spin.setRange(*_DSPIN_RANGE) + if settings.pause_below is not None: + self.pause_below_spin.setValue(settings.pause_below) + form.addRow("Pause below", self.pause_below_spin) + + self.resume_above_spin = QDoubleSpinBox() + self.resume_above_spin.setDecimals(4) + self.resume_above_spin.setRange(*_DSPIN_RANGE) + if settings.resume_above is not None: + self.resume_above_spin.setValue(settings.resume_above) + form.addRow("Resume above", self.resume_above_spin) + + buttons = QDialogButtonBox(QDialogButtonBox.Ok | QDialogButtonBox.Cancel) + buttons.accepted.connect(self._on_accept) + buttons.rejected.connect(self.reject) + layout.addWidget(buttons) + + def _on_accept(self): + if self.enabled_check.isChecked(): + if not self.device_combo.currentText(): + QMessageBox.warning(self, "Missing input", "Select a device to monitor.") + return + if self.resume_above_spin.value() <= self.pause_below_spin.value(): + QMessageBox.warning( + self, + "Invalid thresholds", + "'Resume above' must be greater than 'Pause below' (hysteresis gap).", + ) + return + self.accept() + + def result_settings(self) -> GuardSettings: + """Valid after `exec_()` returns `QDialog.Accepted`.""" + return GuardSettings( + enabled=self.enabled_check.isChecked(), + device_name=self.device_combo.currentText() or None, + pause_below=self.pause_below_spin.value(), + resume_above=self.resume_above_spin.value(), + ) diff --git a/debye_bec/bec_widgets/widgets/scheduler/item_dialog.py b/debye_bec/bec_widgets/widgets/scheduler/item_dialog.py new file mode 100644 index 0000000..001f541 --- /dev/null +++ b/debye_bec/bec_widgets/widgets/scheduler/item_dialog.py @@ -0,0 +1,238 @@ +""" +The dialog behind the schedule widget's "Add..."/"Edit..." buttons. + +Rather than asking the operator to remember and type +`scans.xas_simple_scan(12000, 14000, 2, 10)`-style commands, this presents: + +- a "Scan" tab: BEC's own `bec_widgets` `ScanControl` widget, embedded + as-is - scan selection, its live-generated per-scan argument form, + docs tooltips, metadata, "recall last scan parameters", all of it. + Reusing it instead of a plugin-owned reimplementation means this stays + in sync with BEC's scan capabilities for free, and looks/behaves exactly + like the scan controls an operator already knows from elsewhere in the + GUI. `ScanControl.button_run_scan` ("Start") is hidden here: this dialog + only ever wants the configured scan name/args/kwargs, never an + immediate submission - see `_collect_scan_result`. +- a "Move" tab: pick a device and a target value/relative flag; +- a "Custom" tab: a free-text field, for anything else (including RPC + calls to other widgets), evaluated the same way as the other tabs. + +Whichever tab is used, the dialog's only output is the same kind of plain +command string the executor already knows how to run - this dialog adds a +friendlier way to *build* that string, it doesn't change what happens with +it afterwards. `kind`/`form_state` are carried along purely so "Edit..." +can reopen the dialog pre-filled instead of asking the user to start over. +""" + +from __future__ import annotations + +from bec_widgets.widgets.control.scan_control.scan_control import ScanControl, ScanParameterConfig +from qtpy.QtWidgets import ( + QCheckBox, + QComboBox, + QDialog, + QDialogButtonBox, + QDoubleSpinBox, + QFormLayout, + QLabel, + QLineEdit, + QMessageBox, + QTabWidget, + QVBoxLayout, + QWidget, +) + +from .scan_form import list_movable_device_names + +_DSPIN_RANGE = (-1e12, 1e12) + + +class ScheduleItemDialog(QDialog): + """Add or edit one schedule item, via ScanControl, a move form, or free text.""" + + def __init__(self, scans, dev, parent=None, initial: dict | None = None, client=None): + super().__init__(parent) + self.setWindowTitle("Schedule item") + self.setMinimumSize(520, 480) + + self._scans = scans + self._dev = dev + self._client = client + + layout = QVBoxLayout(self) + self.tabs = QTabWidget() + layout.addWidget(self.tabs) + + self._build_scan_tab() + self._build_move_tab() + self._build_custom_tab() + + buttons = QDialogButtonBox(QDialogButtonBox.Ok | QDialogButtonBox.Cancel) + buttons.accepted.connect(self._on_accept) + buttons.rejected.connect(self.reject) + layout.addWidget(buttons) + + self._apply_initial(initial or {}) + + # ------------------------------------------------------------------ # + # Scan tab - embeds BEC's own ScanControl widget + # ------------------------------------------------------------------ # + def _build_scan_tab(self): + tab = QWidget() + layout = QVBoxLayout(tab) + + # client=None resolves to the same process-wide BEC client + # (bec_dispatcher.client) our own widget uses - no second Redis + # connection is opened. + self.scan_control = ScanControl(parent=tab, client=self._client) + self.scan_control.button_run_scan.hide() + layout.addWidget(self.scan_control) + + hint = QLabel( + "Configure the scan above, then confirm with OK below - it will be " + "added to the schedule, not started immediately." + ) + hint.setWordWrap(True) + hint.setStyleSheet("color: gray;") + layout.addWidget(hint) + + self.tabs.addTab(tab, "Scan") + + def _collect_scan_result(self) -> dict: + # Same call ScanControl.run_scan() makes before actually + # submitting, to resolve a typed-but-unconfirmed scan name. + self.scan_control.validate_scan_selection() + scan_name = self.scan_control.current_scan + if not scan_name: + raise ValueError("No scan selected.") + + # bec_object=False: plain, repr-able values (e.g. a device name + # string rather than the live DeviceBase instance) - needed since + # the result has to survive a round-trip through Redis as text and + # be re-evaluated later, not just used in-process immediately. + args, kwargs = self.scan_control.get_scan_parameters(bec_object=False) + command = _format_scan_call(scan_name, args, kwargs) + return { + "command": command, + "kind": "scan", + "form_state": {"scan_name": scan_name, "args": args, "kwargs": kwargs}, + } + + def _prefill_scan_tab(self, scan_name: str, args: list, kwargs: dict): + # ScanControl restores parameters for a scan from its own config + # cache (see `ScanControl.restore_scan_parameters`); pre-loading + # that cache before switching to the scan reuses that mechanism + # instead of poking at its internal argument widgets directly. + self.scan_control.config.scans[scan_name] = ScanParameterConfig( + name=scan_name, args=args, kwargs=kwargs + ) + self.scan_control.current_scan = scan_name + + # ------------------------------------------------------------------ # + # Move tab + # ------------------------------------------------------------------ # + def _build_move_tab(self): + tab = QWidget() + form = QFormLayout(tab) + + self.move_device_combo = QComboBox() + self.move_device_combo.addItems(list_movable_device_names(self._dev)) + form.addRow("Device", self.move_device_combo) + + self.move_value_spin = QDoubleSpinBox() + self.move_value_spin.setDecimals(6) + self.move_value_spin.setRange(*_DSPIN_RANGE) + form.addRow("Target value", self.move_value_spin) + + self.move_relative_check = QCheckBox("Relative move") + form.addRow("", self.move_relative_check) + + self.tabs.addTab(tab, "Move") + + def _collect_move_result(self) -> dict: + device_name = self.move_device_combo.currentText() + if not device_name: + raise ValueError("No movable device available/selected.") + value = self.move_value_spin.value() + relative = self.move_relative_check.isChecked() + return { + "command": f"dev.{device_name}.move({value!r}, relative={relative!r})", + "kind": "move", + "form_state": {"device_name": device_name, "value": value, "relative": relative}, + } + + # ------------------------------------------------------------------ # + # Custom tab + # ------------------------------------------------------------------ # + def _build_custom_tab(self): + tab = QWidget() + layout = QVBoxLayout(tab) + layout.addWidget( + QLabel( + "Free-form command, evaluated against `scans` and `dev` - use this for " + "anything the Scan/Move tabs don't cover, e.g. an RPC call to another widget." + ) + ) + self.custom_edit = QLineEdit() + self.custom_edit.setPlaceholderText("scans.xas_simple_scan(12000, 14000, 2, 10)") + layout.addWidget(self.custom_edit) + layout.addStretch(1) + self.tabs.addTab(tab, "Custom") + + def _collect_custom_result(self) -> dict: + text = self.custom_edit.text().strip() + if not text: + raise ValueError("Command must not be empty.") + return {"command": text, "kind": "custom", "form_state": {"text": text}} + + # ------------------------------------------------------------------ # + # pre-fill (edit mode) / result extraction + # ------------------------------------------------------------------ # + def _apply_initial(self, initial: dict): + kind = initial.get("kind") + state = initial.get("form_state") or {} + + if kind == "scan" and state.get("scan_name"): + self._prefill_scan_tab( + state["scan_name"], state.get("args") or [], state.get("kwargs") or {} + ) + self.tabs.setCurrentIndex(0) + elif kind == "move" and state.get("device_name"): + idx = self.move_device_combo.findText(state["device_name"]) + if idx >= 0: + self.move_device_combo.setCurrentIndex(idx) + self.move_value_spin.setValue(float(state.get("value", 0.0))) + self.move_relative_check.setChecked(bool(state.get("relative", False))) + self.tabs.setCurrentIndex(1) + elif initial.get("command"): + # "custom" kind, or a legacy/unrecognized item - fall back to + # showing the raw command text as-is. + self.custom_edit.setText(state.get("text", initial["command"])) + self.tabs.setCurrentIndex(2) + + def _on_accept(self): + try: + result = self._collect_result() + except ValueError as exc: + QMessageBox.warning(self, "Missing input", str(exc)) + return + self._result = result + self.accept() + + def _collect_result(self) -> dict: + current = self.tabs.currentIndex() + if current == 0: + return self._collect_scan_result() + if current == 1: + return self._collect_move_result() + return self._collect_custom_result() + + def result(self) -> dict: + """Valid after `exec_()` returns `QDialog.Accepted`.""" + return self._result + + +def _format_scan_call(scan_name: str, args: list, kwargs: dict) -> str: + parts = [repr(a) for a in args] + parts += [f"{name}={value!r}" for name, value in kwargs.items()] + return f"scans.{scan_name}({', '.join(parts)})" diff --git a/debye_bec/bec_widgets/widgets/scheduler/notification_dialog.py b/debye_bec/bec_widgets/widgets/scheduler/notification_dialog.py new file mode 100644 index 0000000..0397c58 --- /dev/null +++ b/debye_bec/bec_widgets/widgets/scheduler/notification_dialog.py @@ -0,0 +1,59 @@ +"""Small settings dialog for `notifications.NotificationSettings` - kept separate from the send logic itself.""" + +from __future__ import annotations + +from qtpy.QtWidgets import QCheckBox, QDialog, QDialogButtonBox, QLabel, QVBoxLayout + +from .notifications import NotificationSettings + + +class NotificationSettingsDialog(QDialog): + """Choose whether, and for which item types, finish/fail notifications are sent.""" + + def __init__(self, settings: NotificationSettings, parent=None): + super().__init__(parent) + self.setWindowTitle("Notifications") + + layout = QVBoxLayout(self) + info_label = QLabel( + "Sends a BEC notification (see MessageEndpoints.notification) whenever a " + "schedule item finishes or fails. Whether it reaches you as a message depends " + "on your facility's notification routing (SciHub + Signal/Teams/SciLog) being " + "configured for these events - ask your beamline admin if you're not sure." + ) + info_label.setWordWrap(True) + layout.addWidget(info_label) + + self.enabled_check = QCheckBox("Send notifications") + self.enabled_check.setChecked(settings.enabled) + layout.addWidget(self.enabled_check) + + self.scan_check = QCheckBox("Scans") + self.scan_check.setChecked(settings.notify_scan) + layout.addWidget(self.scan_check) + + self.move_check = QCheckBox("Movements") + self.move_check.setChecked(settings.notify_move) + layout.addWidget(self.move_check) + + self.rpc_check = QCheckBox("RPC / custom commands") + self.rpc_check.setChecked(settings.notify_rpc) + layout.addWidget(self.rpc_check) + + for check in (self.scan_check, self.move_check, self.rpc_check): + self.enabled_check.toggled.connect(check.setEnabled) + check.setEnabled(settings.enabled) + + buttons = QDialogButtonBox(QDialogButtonBox.Ok | QDialogButtonBox.Cancel) + buttons.accepted.connect(self.accept) + buttons.rejected.connect(self.reject) + layout.addWidget(buttons) + + def result_settings(self) -> NotificationSettings: + """Valid after `exec_()` returns `QDialog.Accepted`.""" + return NotificationSettings( + enabled=self.enabled_check.isChecked(), + notify_scan=self.scan_check.isChecked(), + notify_move=self.move_check.isChecked(), + notify_rpc=self.rpc_check.isChecked(), + ) diff --git a/debye_bec/bec_widgets/widgets/scheduler/notifications.py b/debye_bec/bec_widgets/widgets/scheduler/notifications.py new file mode 100644 index 0000000..b7542e2 --- /dev/null +++ b/debye_bec/bec_widgets/widgets/scheduler/notifications.py @@ -0,0 +1,94 @@ +""" +A small, self-contained "tell the user when something finished/failed" +feature, independent of scheduling/execution logic. + +Sends via BEC's own notification transport: `MessageEndpoints.notification` +carries a `messages.NotificationMessage`, the same message type/endpoint +BEC's own scan lifecycle events (e.g. "scan_completed") use. Reusing it +means schedule-item notifications are just more entries in the same +mechanism, not a plugin-invented side channel - and if your facility runs +SciHub with messaging-service routing configured (Signal/Teams/SciLog), +they can be forwarded out the same way any other BEC notification would +be, by adding a route for the event names below in that routing config. + +Nothing here is scheduling/BECWidget-specific: `notify_item_finished` only +needs a connector, a settings object and a few plain values, so it can be +called from `schedule_widget.py` without either module depending on the +other's internals. +""" + +from __future__ import annotations + +from bec_lib.endpoints import MessageEndpoints +from bec_lib.logger import bec_logger +from bec_lib.messages import MessagingServiceTextContent, NotificationMessage +from pydantic import BaseModel + +logger = bec_logger.logger + +# Namespaced so they can't collide with BEC's own built-in event names +# (e.g. "scan_completed"); an admin who wants these routed to Signal/Teams/ +# SciLog can add a route for exactly these names in their notification +# routing config (`MessageEndpoints.notification_config()`). +_EVENT_NAMES = { + ("scan", True): "schedule_scan_completed", + ("scan", False): "schedule_scan_failed", + ("move", True): "schedule_move_completed", + ("move", False): "schedule_move_failed", + ("custom", True): "schedule_rpc_completed", + ("custom", False): "schedule_rpc_failed", +} + + +class NotificationSettings(BaseModel): + """Persisted per-schedule notification preferences.""" + + enabled: bool = False + notify_scan: bool = True + notify_move: bool = True + # "custom" is what this plugin calls the free-text tab (see + # item_dialog.py) - it covers RPC calls to other widgets as well as + # any hand-typed command, so it's exposed to the user as "RPC". + notify_rpc: bool = True + + +def _enabled_for_kind(settings: NotificationSettings, kind: str) -> bool: + if not settings.enabled: + return False + return { + "scan": settings.notify_scan, + "move": settings.notify_move, + "custom": settings.notify_rpc, + }.get(kind, False) + + +def notify_item_finished( + connector, + schedule_name: str, + kind: str, + command: str, + success: bool, + settings: NotificationSettings, + error: str | None = None, +): + """ + Sends a `NotificationMessage` if notifications are enabled for `kind`. + Never raises - a notification failing to send should not affect + schedule execution. + """ + if not _enabled_for_kind(settings, kind): + return + + event = _EVENT_NAMES.get((kind, success), "schedule_item_finished") + status_word = "completed" if success else "failed" + text = f"[{schedule_name}] {kind} {status_word}: {command}" + if error: + text += f"\n{error.strip().splitlines()[-1]}" + + try: + connector.send( + MessageEndpoints.notification(event), + NotificationMessage(event=event, message=[MessagingServiceTextContent(content=text)]), + ) + except Exception: # pylint: disable=broad-except + logger.exception("Failed to send schedule notification for event '%s'", event) diff --git a/debye_bec/bec_widgets/widgets/scheduler/register_scheduler.py b/debye_bec/bec_widgets/widgets/scheduler/register_scheduler.py new file mode 100644 index 0000000..6b6f1a1 --- /dev/null +++ b/debye_bec/bec_widgets/widgets/scheduler/register_scheduler.py @@ -0,0 +1,15 @@ +def main(): # pragma: no cover + from qtpy import PYSIDE6 + + if not PYSIDE6: + print("PYSIDE6 is not available in the environment. Cannot patch designer.") + return + from PySide6.QtDesigner import QPyDesignerCustomWidgetCollection + + from .scheduler_plugin import SchedulerPlugin + + QPyDesignerCustomWidgetCollection.addCustomWidget(SchedulerPlugin()) + + +if __name__ == "__main__": # pragma: no cover + main() diff --git a/debye_bec/bec_widgets/widgets/scheduler/scan_form.py b/debye_bec/bec_widgets/widgets/scheduler/scan_form.py new file mode 100644 index 0000000..3c77272 --- /dev/null +++ b/debye_bec/bec_widgets/widgets/scheduler/scan_form.py @@ -0,0 +1,16 @@ +""" +Small device-listing helper for the "Move" tab of `item_dialog.py`. + +Scan parameter introspection/forms used to live in this module too, but +that's now handled by embedding BEC's own `bec_widgets` `ScanControl` +widget directly (see `item_dialog.py`) instead of a plugin-owned +reimplementation - one less UI to keep in sync with BEC's scan +capabilities. +""" + +from __future__ import annotations + + +def list_movable_device_names(dev) -> list[str]: + """Device names that expose `.move(...)` (i.e. positioner-like devices).""" + return sorted(name for name, device in dev.items() if hasattr(device, "move")) diff --git a/debye_bec/bec_widgets/widgets/scheduler/schedule_item.py b/debye_bec/bec_widgets/widgets/scheduler/schedule_item.py new file mode 100644 index 0000000..872b2a5 --- /dev/null +++ b/debye_bec/bec_widgets/widgets/scheduler/schedule_item.py @@ -0,0 +1,76 @@ +""" +Plain (non-BECMessage) data model for one schedule entry and the schedule +as a whole. + +These are ordinary pydantic models used only client-side, for validation +and convenience. They are never sent over Redis as their own type - only +ever as the `.value` payload of a `bec_lib.messages.VariableMessage` (see +`endpoints.schedule` for why). +""" + +from __future__ import annotations + +from enum import Enum +from typing import Literal + +from pydantic import BaseModel, Field + +from .guard import GuardSettings +from .notifications import NotificationSettings + + +class ScheduleItemStatus(str, Enum): + PENDING = "pending" + RUNNING = "running" + COMPLETED = "completed" + FAILED = "failed" + ABORTED = "aborted" + + +# Which tab of the Add/Edit dialog built `command`, and therefore how to +# re-open that dialog pre-filled with the same values for editing. "custom" +# covers hand-typed text (including RPC calls to other widgets), which has +# no structured `form_state` to restore beyond the raw text itself. +ScheduleItemKind = Literal["scan", "move", "custom"] + + +class ScheduleItem(BaseModel): + """A single command in the schedule, plus its execution bookkeeping.""" + + item_id: str + command: str + status: ScheduleItemStatus = ScheduleItemStatus.PENDING + + # How `command` was produced, and (for "scan"/"move") the structured + # inputs behind it, so the Add/Edit dialog can be reopened pre-filled + # instead of asking the user to re-type everything. `command` itself + # remains the single source of truth actually evaluated at execution + # time - `kind`/`form_state` only drive the UI. + kind: ScheduleItemKind = "custom" + form_state: dict | None = None + + # Bookkeeping used to reconnect to a submission that is still (or was) + # in flight on the BEC scan/device server, after this widget has been + # closed and reopened. + request_id: str | None = None + scan_id: str | None = None + scan_number: int | None = None + + error: str | None = None + started_at: float | None = None + finished_at: float | None = None + + +class Schedule(BaseModel): + """The full, persisted state of one schedule-widget instance.""" + + schedule_name: str + items: list[ScheduleItem] = Field(default_factory=list) + is_running: bool = False + + # Settings for two independent, optional features (see notifications.py + # and guard.py) - persisted here alongside the schedule itself so they + # survive a widget restart too, but their *logic* lives entirely in + # those separate modules; this is just where their settings are stored. + notifications: NotificationSettings = Field(default_factory=NotificationSettings) + guard: GuardSettings = Field(default_factory=GuardSettings) diff --git a/debye_bec/bec_widgets/widgets/scheduler/schedule_logic.py b/debye_bec/bec_widgets/widgets/scheduler/schedule_logic.py new file mode 100644 index 0000000..29553d9 --- /dev/null +++ b/debye_bec/bec_widgets/widgets/scheduler/schedule_logic.py @@ -0,0 +1,60 @@ +""" +Pure bookkeeping over a list of `ScheduleItem`s - no Qt, no locking, no I/O. +Kept separate from `schedule_widget.py` for the same reason `guard.py` and +`notifications.py` are: it's testable on its own, and the widget should +only be responsible for Qt/orchestration, not figuring out which item runs +next. + +The one invariant everything here assumes and preserves: at any moment, a +schedule's items form a (possibly empty) prefix that is no longer PENDING +(RUNNING - at most one, the one currently executing - COMPLETED, FAILED or +ABORTED), followed by a suffix that is entirely PENDING. `ScheduleWidget` +is responsible for only ever mutating the PENDING suffix (see its module +docstring) and for calling these functions while holding its lock; nothing +here does its own locking. +""" + +from __future__ import annotations + +from typing import Literal + +from .schedule_item import ScheduleItem, ScheduleItemStatus + + +def protected_prefix_length(items: list[ScheduleItem]) -> int: + """How many items, from the start, are no longer PENDING.""" + count = 0 + for item in items: + if item.status != ScheduleItemStatus.PENDING: + count += 1 + else: + break + return count + + +def index_of(items: list[ScheduleItem], item_id: str | None) -> int | None: + if item_id is None: + return None + for i, item in enumerate(items): + if item.item_id == item_id: + return i + return None + + +def pick_next_runnable(items: list[ScheduleItem]) -> ScheduleItem | Literal["stop"] | None: + """ + What the execution loop should do next: always re-derived from scratch + (never a remembered index/object) so edits made to the PENDING suffix + between calls are picked up correctly. + + Returns the next item to run/attach to, the string `"stop"` if an + earlier item failed/was aborted (execution stays parked there until the + operator intervenes), or `None` if every item is COMPLETED. + """ + for item in items: + if item.status == ScheduleItemStatus.COMPLETED: + continue + if item.status in (ScheduleItemStatus.FAILED, ScheduleItemStatus.ABORTED): + return "stop" + return item # PENDING, or RUNNING (reconciled as still active) + return None diff --git a/debye_bec/bec_widgets/widgets/scheduler/scheduler.py b/debye_bec/bec_widgets/widgets/scheduler/scheduler.py new file mode 100644 index 0000000..84a04f1 --- /dev/null +++ b/debye_bec/bec_widgets/widgets/scheduler/scheduler.py @@ -0,0 +1,995 @@ +""" +ScheduleWidget - persists an ordered schedule of BEC commands (scans, +device moves, ...) to Redis, executes them one by one, and survives being +closed and reopened - including while an item is still running +server-side. Items that have not started yet can be added, reordered or +deleted while the schedule is executing; items that have already started +or finished cannot. + +Persistence +----------- +The whole schedule (list of items + status) is stored as one +`VariableMessage` document under a stable, named endpoint +(`endpoints.schedule(schedule_name)`), re-written and republished on every +change. See `endpoints.py` for the reasoning behind that design. + +Building a command +------------------- +The "Add..."/"Edit..." buttons open `item_dialog.ScheduleItemDialog`, +which lets the operator pick a registered scan through BEC's own +`bec_widgets` `ScanControl` widget (embedded as-is, with its "Start" +button hidden so it only ever configures a scan instead of submitting +one) or a device move from a small form, instead of typing a command by +hand - while still allowing every parameter value to be changed. A +"Custom" tab keeps a free-text command for anything else (e.g. an RPC +call to another widget). Whichever tab is used, the result is the same +kind of plain command string described below - the dialog only changes +how that string gets built. + +Execution model +---------------- +Each command (e.g. "scans.xas_simple_scan(12000, 14000, 2, 10)" or +"dev.samx.move(12.3, relative=True)") is evaluated against a small, +fixed namespace of `{"scans": self.scans, "dev": self.dev}`. This mirrors +how `bec_lib.script_executor` runs persisted script text: it does not +invent a second way to talk to devices/scans, it drives the exact same +client objects an operator would use interactively. + +Both `scans.xxx(...)` and `dev..move(...)` (which internally calls +`scans.mv(...)`) submit *asynchronously* to BEC's scan queue and return a +`ScanReport` immediately, with `.request.requestID`, `.wait()` and +`.cancel()`. Submission already happened server-side by the time `.wait()` +returns - closing this widget (or even the whole GUI process) does not +stop a scan or move that has already been submitted; it keeps running on +the scan/device server. + +Reconciliation after being closed and reopened +------------------------------------------------ +If this widget is closed while an item is RUNNING and reopened later, we +must not blindly resume from "pending" (which could re-submit a scan that +is still running) nor blindly trust our last persisted status forever +(the scan may have long since finished). Instead, on load we ask BEC's +*own*, live queue state: `client.queue.queue_storage`, freshly seeded from +`MessageEndpoints.scan_queue_status()`, can look up any earlier +`request_id` via `find_queue_item_by_requestID()`. If it's still +pending/running there, we re-attach and wait; if it's no longer in the +queue, it has left the system (finished, successfully or not) while we +were disconnected, and we mark it completed and move on. + +Editing while running - the "protected prefix" invariant +------------------------------------------------------------ +`run_schedule()` walks the items strictly in order in a background task +(`BECConnector.submit_task`). To let the operator add/reorder/delete +items that haven't started yet *while that task is running*, without +racing it, every mutation goes through `self._lock` and respects one +invariant: at any moment, the schedule's items form a (possibly empty) +prefix of items that are no longer PENDING (RUNNING - at most one, the +one currently executing - COMPLETED/FAILED/ABORTED), followed by a suffix +of items that are all still PENDING. The pure bookkeeping over that +invariant (`protected_prefix_length`, `index_of`, `pick_next_runnable`) +lives in `schedule_logic.py`, kept separate the same way `guard.py` and +`notifications.py` are - no Qt, no locking, independently testable; this +widget only calls it while holding `self._lock`. Edits are only ever +allowed to touch the PENDING suffix: + - `add_item(..., index=...)` clamps `index` up to at least the prefix + length, so a new item can never be inserted before something already + running or finished. + - `move_item`/`delete_item` refuse (raise `RuntimeError`) if the target + item's status isn't PENDING. +The execution task itself never keeps a stale index across its +(potentially long) `report.wait()` calls - each time it needs the next +item, it re-scans `self.schedule.items` from the start under the lock, so +edits made to the PENDING suffix while an earlier item is executing are +picked up correctly. + +Multi-instance note: `_on_remote_update` (another widget instance, or a +script, publishing a new version of this schedule) replaces +`self.schedule` wholesale. If *this* instance is currently running its +own execution task, applying a wholesale replacement here would orphan +the `ScheduleItem` object that task is watching, so incoming remote +updates are ignored while `self.schedule.is_running` locally - this +instance's own state wins until it finishes running, at which point its +next `_persist()` call re-publishes the authoritative state anyway. This +is a simplification: true multi-writer merging (e.g. concurrently editing +the PENDING suffix from two open widgets while a third is executing it) +is out of scope here. + +Notes / things to adapt for your BEC version and deployment +------------------------------------------------------------- +- `eval()` is used deliberately for the "Custom" tab and to actually run + every item, with an empty `__builtins__` and a namespace restricted to + `scans`/`dev`, matching BEC's own trust model for script text (see + `bec_lib.script_executor`). The Scan/Move tabs never eval user text - + values come from typed Qt widgets and are placed into the generated + command with `repr()`. +- `QueueItem.status` is a plain string on the installed bec_lib version + this was written against; the exact set of literal values was not + fully pinned down against a live scan server, so + `_is_queue_item_active()` treats anything containing "PENDING" or + "RUNNING" as still active and everything else as finished. + +Two independent, optional features +------------------------------------- +Both are implemented in their own modules with no dependency on +scheduling/execution internals - `schedule_widget.py` only wires their +signals/functions into the execution loop: + +- **Notifications** (`notifications.py`/`notification_dialog.py`): sends a + BEC `NotificationMessage` whenever an item finishes or fails, if enabled + for that item's kind (scan/move/rpc, independently toggleable). Fires + from `_execute_item()` after an item's final status is known. +- **Auto-pause guard** (`guard.py`/`guard_dialog.py`): a `SignalGuard` + monitors one device's live readback and, on hysteresis-based + pause/resume crossings, emits Qt signals `schedule_widget.py` reacts to. + Applies to scan items only - see "Guard interaction" below. + +Guard interaction with execution +----------------------------------- +When the guard is enabled and about to start a `kind == "scan"` item, +`_execute_item` first calls `self._guard.wait_until_clear(...)`, blocking +(without marking the item RUNNING, so it stays PENDING and visibly +unaffected by the "protected prefix" invariant) until the signal is back +above the resume threshold or the operator aborts. If the guard trips +*while* a scan item is RUNNING, `_on_guard_paused` cancels its +`ScanReport` (the same mechanism `abort_schedule()` uses) and sets +`self._guard_interrupt_requested`; `_execute_item`'s exception handler +checks that flag and, if set, resets the item back to PENDING (clearing +its request/scan bookkeeping) instead of marking it ABORTED - so the +normal execution loop simply picks it up again once the guard clears, +with no special "resume" code path needed. Device moves and custom/RPC +items are never touched by the guard (checked via `item.kind == "scan"` +in both places) - note this means a scan typed into the free-text +"Custom" tab is *not* guard-protected, since we can't tell hand-typed +text apart from a scan call; use the "Scan" tab if you need that. One +more limitation: a scan reconciled as "still running from a previous +session" (see "Reconciliation" above) is only *waited on* via +`_await_running_item`, not submitted through `_execute_item`, so there is +no live `ScanReport` for the guard to cancel if it trips during that +narrow window - it will simply be picked up again once that scan finishes +on its own. +""" + +from __future__ import annotations + +import threading +import time +import traceback +import uuid + +from bec_lib.endpoints import MessageEndpoints +from bec_lib.logger import bec_logger +from bec_lib.messages import VariableMessage +from bec_widgets.utils.bec_connector import ConnectionConfig +from bec_widgets.utils.bec_widget import BECWidget +from bec_widgets.utils.error_popups import SafeSlot +from qtpy.QtCore import Qt, Signal +from qtpy.QtWidgets import ( + QDialog, + QHBoxLayout, + QLabel, + QListWidget, + QListWidgetItem, + QMessageBox, + QPushButton, + QVBoxLayout, + QWidget, +) + +from .endpoints import schedule as schedule_endpoint +from .guard import GuardSettings, SignalGuard +from .guard_dialog import GuardSettingsDialog +from .item_dialog import ScheduleItemDialog +from .notification_dialog import NotificationSettingsDialog +from .notifications import NotificationSettings, notify_item_finished +from .schedule_item import Schedule, ScheduleItem, ScheduleItemStatus +from .schedule_logic import index_of, pick_next_runnable, protected_prefix_length + +logger = bec_logger.logger + +# The static, hard-coded example schedule requested for this plugin. A +# real deployment would more likely let the user build this list purely +# through the UI - the persistence/execution/reconciliation logic below +# does not care where the commands came from. +DEFAULT_COMMANDS = [ + "scans.xas_simple_scan(12000, 14000, 2, 10)", + "dev.samx.move(12.3, relative=True)", + "scans.xas_simple_scan(13000, 17000, 1, 60)", +] + +_ACTIVE_QUEUE_STATES = ("PENDING", "RUNNING") +_ITEM_ID_ROLE = Qt.UserRole + 1 # QListWidgetItem data role used to map a row back to an item_id + +_STATUS_ICON = { + ScheduleItemStatus.PENDING: "\u23f3", # hourglass + ScheduleItemStatus.RUNNING: "\u25b6", # play + ScheduleItemStatus.COMPLETED: "\u2705", # check mark + ScheduleItemStatus.FAILED: "\u274c", # cross mark + ScheduleItemStatus.ABORTED: "\u23f9", # stop +} + + +class ScheduleWidgetConfig(ConnectionConfig): + """ + Per-instance identity/settings for `ScheduleWidget`, following the same + pattern `bec_widgets` itself uses (e.g. `ScanControlConfig`): a + `ConnectionConfig` subclass constructed in `__init__` when none is + passed, carrying whatever small settings distinguish one instance from + another. + + `schedule_name` lives here rather than as a bare constructor kwarg for + that consistency - but note this is *not* where the schedule's actual + contents (items, guard/notification settings) live: `self.config` is + used for RPC identity and optional explicit GUI-layout save/restore, + not auto-persisted across an ordinary widget close/reopen the way this + plugin's Redis-backed `Schedule` document is. Conflating the two would + quietly reintroduce the "closed widget loses its schedule" problem this + plugin exists to solve. + """ + + schedule_name: str = "default_schedule" + + +class Scheduler(BECWidget, QWidget): + """Schedule, persist and execute a sequence of BEC scan/device commands.""" + + ICON_NAME = "assignment_add" + USER_ACCESS = [ + "run_schedule", + "abort_schedule", + "reset_schedule", + "get_status", + "add_item", + "edit_item", + "delete_item", + "move_item", + "move_item_up", + "move_item_down", + ] + + # emitted (thread-safe, may come from the background execution task) + # whenever the schedule state changes and the UI should redraw + schedule_changed = Signal() + + def __init__( + self, + parent=None, + client=None, + config: ScheduleWidgetConfig | dict | None = None, + gui_id: str | None = None, + schedule_name: str | None = None, + **kwargs, + ): + if config is None: + config = ScheduleWidgetConfig( + widget_class=self.__class__.__name__, + schedule_name=schedule_name or "default_schedule", + ) + elif schedule_name is not None: + # Allow overriding even when a config object/dict was supplied, + # mirroring how ScanControl treats its own optional constructor + # args as overrides onto whatever config ends up in self.config. + if isinstance(config, dict): + config = dict(config, schedule_name=schedule_name) + else: + config.schedule_name = schedule_name + super().__init__(parent=parent, client=client, gui_id=gui_id, config=config, **kwargs) + + # A stable name, NOT self.gui_id: gui_id is regenerated every time + # the widget is constructed, so it could never be used to find a + # previously persisted schedule again after the widget is reopened. + self.schedule_name = self.config.schedule_name + + self.get_bec_shortcuts() # -> self.client, self.dev, self.scans, self.queue + self.connector = self.client.connector + + self._endpoint = schedule_endpoint(self.schedule_name) + self._lock = threading.Lock() # guards self.schedule (see module docstring) + self._abort_requested = False + self._guard_interrupt_requested = False # see "Guard interaction" in module docstring + self._current_report = None # ScanReport of the item currently executing, if any + self._current_item_kind: str | None = None # kind of the item _current_report belongs to + self._selected_item_id: str | None = None # survives list repopulation on refresh + + self.schedule_changed.connect(self._refresh_ui) + + self.schedule: Schedule = self._load_or_seed_schedule() + self._reconcile_with_live_state() + + self._guard = SignalGuard(self.connector, parent=self) + self._guard.paused.connect(self._on_guard_paused) + self._guard.resumed.connect(self._on_guard_resumed) + self._guard.configure(self.schedule.guard) + + self._build_ui() + self._refresh_ui() + + # Stay in sync with any other subscriber touching the same + # schedule (another instance of this widget, a script, ...). + self.bec_dispatcher.connect_slot(self._on_remote_update, self._endpoint) + + # ------------------------------------------------------------------ # + # persistence + # ------------------------------------------------------------------ # + def _load_or_seed_schedule(self) -> Schedule: + logger.info("load or seed schedule") + msg: VariableMessage | None = self.connector.get(self._endpoint) + if msg is not None: + logger.info(f"Got msg from endpoint: {msg}") + return Schedule.model_validate(msg.value) + + schedule = Schedule( + schedule_name=self.schedule_name, + items=[ + ScheduleItem(item_id=str(uuid.uuid4()), command=cmd) for cmd in DEFAULT_COMMANDS + ], + ) + self.connector.set_and_publish( + self._endpoint, VariableMessage(value=schedule.model_dump(mode="json")) + ) + return schedule + + def _persist_locked(self): + """Write `self.schedule` to Redis. Caller must hold `self._lock`.""" + self.connector.set_and_publish( + self._endpoint, VariableMessage(value=self.schedule.model_dump(mode="json")) + ) + + def _persist(self): + with self._lock: + self._persist_locked() + + @SafeSlot(dict, dict) + def _on_remote_update(self, msg_content: dict, metadata: dict): + value = msg_content.get("value") if msg_content else None + if not value or value.get("schedule_name") != self.schedule_name: + return + with self._lock: + if self.schedule.is_running: + # This instance's own execution task owns the current + # ScheduleItem objects (and will re-persist its own state + # again shortly); accepting a wholesale replacement here + # would orphan those objects. See module docstring. + logger.info( + "Ignoring remote schedule update for '%s' while a local run is in progress.", + self.schedule_name, + ) + return + self.schedule = Schedule.model_validate(value) + self._apply_guard_settings() + self._refresh_ui() + + # ------------------------------------------------------------------ # + # reconciliation: figure out real state after being closed/reopened + # ------------------------------------------------------------------ # + def _reconcile_with_live_state(self): + """ + For any item last seen as RUNNING, ask BEC's own queue state - + not our local memory, which may be stale or from a previous + process - what actually happened to it while we were gone. + """ + queue_storage = self.client.queue.queue_storage + + # Force a fresh pull of the current queue snapshot from Redis + # rather than relying on pubsub messages we may have missed while + # this widget instance did not exist. + status_msg = self.connector.get(MessageEndpoints.scan_queue_status()) + if status_msg is not None: + queue_storage.update_with_status(status_msg) + + with self._lock: + changed = False + for item in self.schedule.items: + if item.status != ScheduleItemStatus.RUNNING or not item.request_id: + continue + queue_item = queue_storage.find_queue_item_by_requestID(item.request_id) + if queue_item is not None and self._is_queue_item_active(queue_item): + # genuinely still in flight server-side; leave it + # RUNNING - run_schedule() will re-attach to it + # instead of resubmitting. + continue + item.status = ScheduleItemStatus.COMPLETED + item.finished_at = time.time() + changed = True + if changed: + self._persist_locked() + + @staticmethod + def _is_queue_item_active(queue_item) -> bool: + return str(getattr(queue_item, "status", "")).upper() in _ACTIVE_QUEUE_STATES + + # ------------------------------------------------------------------ # + # the "protected prefix" invariant (see module docstring) - the actual + # logic lives in schedule_logic.py, pure and independently testable; + # these thin wrappers just apply it to the current, lock-held state. + # ------------------------------------------------------------------ # + def _protected_prefix_length_locked(self) -> int: + return protected_prefix_length(self.schedule.items) + + def _index_of_locked(self, item_id: str | None) -> int | None: + return index_of(self.schedule.items, item_id) + + # ------------------------------------------------------------------ # + # UI + # ------------------------------------------------------------------ # + def _build_ui(self): + layout = QVBoxLayout(self) + layout.addWidget(QLabel(f"Schedule: {self.schedule_name}")) + + self.list_widget = QListWidget() + self.list_widget.currentRowChanged.connect(self._on_selection_changed) + layout.addWidget(self.list_widget) + + edit_row = QHBoxLayout() + self.add_btn = QPushButton("Add...") + self.edit_btn = QPushButton("Edit...") + self.delete_btn = QPushButton("Delete") + self.move_up_btn = QPushButton("Move Up") + self.move_down_btn = QPushButton("Move Down") + for button in ( + self.add_btn, + self.edit_btn, + self.delete_btn, + self.move_up_btn, + self.move_down_btn, + ): + edit_row.addWidget(button) + layout.addLayout(edit_row) + + self.add_btn.clicked.connect(self._on_add_clicked) + self.edit_btn.clicked.connect(self._on_edit_clicked) + self.delete_btn.clicked.connect(self._on_delete_clicked) + self.move_up_btn.clicked.connect(self._on_move_up_clicked) + self.move_down_btn.clicked.connect(self._on_move_down_clicked) + + run_row = QHBoxLayout() + self.run_btn = QPushButton("Run / Continue") + self.abort_btn = QPushButton("Abort") + self.reset_btn = QPushButton("Reset") + for button in (self.run_btn, self.abort_btn, self.reset_btn): + run_row.addWidget(button) + layout.addLayout(run_row) + + self.run_btn.clicked.connect(self.run_schedule) + self.abort_btn.clicked.connect(self.abort_schedule) + self.reset_btn.clicked.connect(self.reset_schedule) + + settings_row = QHBoxLayout() + self.notifications_btn = QPushButton("Notifications...") + self.guard_btn = QPushButton("Auto-pause...") + settings_row.addWidget(self.notifications_btn) + settings_row.addWidget(self.guard_btn) + layout.addLayout(settings_row) + + self.notifications_btn.clicked.connect(self._on_notifications_clicked) + self.guard_btn.clicked.connect(self._on_guard_clicked) + + self.guard_status_label = QLabel() + self.guard_status_label.setStyleSheet("color: gray;") + layout.addWidget(self.guard_status_label) + + @SafeSlot() + def _refresh_ui(self): + with self._lock: + items = list(self.schedule.items) + is_running = self.schedule.is_running + + self.list_widget.blockSignals(True) + self.list_widget.clear() + selected_row = None + for row, item in enumerate(items): + text = f"{_STATUS_ICON.get(item.status, '?')} {item.command}" + if item.error: + text += f" ({item.error.strip().splitlines()[-1]})" + list_item = QListWidgetItem(text, self.list_widget) + list_item.setData(_ITEM_ID_ROLE, item.item_id) + if item.item_id == self._selected_item_id: + selected_row = row + if selected_row is not None: + self.list_widget.setCurrentRow(selected_row) + else: + self._selected_item_id = None + self.list_widget.blockSignals(False) + + idx = None + selected_item = None + if self._selected_item_id is not None: + for i, item in enumerate(items): + if item.item_id == self._selected_item_id: + idx, selected_item = i, item + break + + # Editable = the selected item hasn't started yet. Thanks to the + # protected-prefix invariant, everything after the first PENDING + # item is guaranteed PENDING too, so a plain index/status check is + # enough here - see module docstring. + can_edit_selected = ( + selected_item is not None and selected_item.status == ScheduleItemStatus.PENDING + ) + + self.run_btn.setEnabled(not is_running) + # Adding is always safe: new items are clamped into the PENDING + # suffix regardless of what's selected (see add_item()). + self.add_btn.setEnabled(True) + self.edit_btn.setEnabled(can_edit_selected) + self.delete_btn.setEnabled(can_edit_selected) + self.move_up_btn.setEnabled(can_edit_selected and idx not in (None, 0)) + self.move_down_btn.setEnabled( + can_edit_selected and idx is not None and idx < len(items) - 1 + ) + + if self._guard.enabled: + state = "OK" if self._guard.is_clear() else "PAUSED - waiting to resume" + value = self._guard.current_value + value_text = f"{value:g}" if value is not None else "?" + self.guard_status_label.setText( + f"Auto-pause: {self._guard.device_name} = {value_text} " + f"(pause < {self._guard.pause_below:g}, resume > {self._guard.resume_above:g}) [{state}]" + ) + self.guard_status_label.show() + else: + self.guard_status_label.hide() + + def _on_selection_changed(self, row: int): + item = self.list_widget.item(row) if row >= 0 else None + self._selected_item_id = item.data(_ITEM_ID_ROLE) if item is not None else None + self._refresh_ui() + + # ---- notifications / auto-pause settings ---- # + @SafeSlot() + def _on_notifications_clicked(self): + dialog = NotificationSettingsDialog(self.schedule.notifications, parent=self) + if dialog.exec_() != QDialog.Accepted: + return + with self._lock: + self.schedule.notifications = dialog.result_settings() + self._persist_locked() + self._refresh_ui() + + @SafeSlot() + def _on_guard_clicked(self): + device_names = sorted(self.dev.keys()) + dialog = GuardSettingsDialog(self.schedule.guard, device_names, parent=self) + if dialog.exec_() != QDialog.Accepted: + return + with self._lock: + self.schedule.guard = dialog.result_settings() + self._persist_locked() + self._guard.configure(self.schedule.guard) + self._refresh_ui() + + def _apply_guard_settings(self): + """Re-subscribe the guard after `self.schedule.guard` changes (e.g. from a remote update).""" + self._guard.configure(self.schedule.guard) + + @SafeSlot(float) + def _on_guard_paused(self, value: float): + logger.info( + "Auto-pause guard tripped for schedule '%s' (value=%s); interrupting the running scan, if any.", + self.schedule_name, + value, + ) + if self._current_report is not None and self._current_item_kind == "scan": + self._guard_interrupt_requested = True + try: + self._current_report.cancel() + except Exception: # pylint: disable=broad-except + logger.exception("Failed to cancel the running scan for the auto-pause guard") + self.schedule_changed.emit() + + @SafeSlot(float) + def _on_guard_resumed(self, value: float): + logger.info( + "Auto-pause guard cleared for schedule '%s' (value=%s); the execution loop will " + "resume the next scan item automatically.", + self.schedule_name, + value, + ) + self.schedule_changed.emit() + + # ---- UI-triggered edit actions ---- # + @SafeSlot() + def _on_add_clicked(self): + dialog = ScheduleItemDialog(self.scans, self.dev, parent=self, client=self.client) + if dialog.exec_() != QDialog.Accepted: + return + result = dialog.result() + with self._lock: + idx = self._index_of_locked(self._selected_item_id) + insert_at = None if idx is None else idx + 1 + new_id = self.add_item( + result["command"], index=insert_at, kind=result["kind"], form_state=result["form_state"] + ) + self._selected_item_id = new_id + self._refresh_ui() + + @SafeSlot() + def _on_edit_clicked(self): + if self._selected_item_id is None: + return + with self._lock: + idx = self._index_of_locked(self._selected_item_id) + item = self.schedule.items[idx] if idx is not None else None + initial = ( + {"kind": item.kind, "command": item.command, "form_state": item.form_state} + if item is not None + else None + ) + if initial is None: + return + + dialog = ScheduleItemDialog( + self.scans, self.dev, parent=self, initial=initial, client=self.client + ) + if dialog.exec_() != QDialog.Accepted: + return + result = dialog.result() + try: + self.edit_item( + self._selected_item_id, + result["command"], + kind=result["kind"], + form_state=result["form_state"], + ) + except RuntimeError as exc: + QMessageBox.warning(self, "Cannot edit item", str(exc)) + return + self._refresh_ui() + + @SafeSlot() + def _on_delete_clicked(self): + if self._selected_item_id is None: + return + confirm = QMessageBox.question( + self, "Delete schedule item", "Remove the selected item from the schedule?" + ) + if confirm != QMessageBox.Yes: + return + try: + self.delete_item(self._selected_item_id) + except RuntimeError as exc: + QMessageBox.warning(self, "Cannot delete item", str(exc)) + return + self._selected_item_id = None + self._refresh_ui() + + @SafeSlot() + def _on_move_up_clicked(self): + if self._selected_item_id is None: + return + try: + self.move_item_up(self._selected_item_id) + except RuntimeError as exc: + QMessageBox.warning(self, "Cannot move item", str(exc)) + + @SafeSlot() + def _on_move_down_clicked(self): + if self._selected_item_id is None: + return + try: + self.move_item_down(self._selected_item_id) + except RuntimeError as exc: + QMessageBox.warning(self, "Cannot move item", str(exc)) + + # ------------------------------------------------------------------ # + # RPC-exposed actions (USER_ACCESS) + # ------------------------------------------------------------------ # + @SafeSlot() + def run_schedule(self): + """Run the schedule, continuing from wherever it last left off.""" + with self._lock: + if self.schedule.is_running: + return + self._abort_requested = False + self.schedule.is_running = True + self._persist_locked() + self.schedule_changed.emit() + self.submit_task( + self._run_all, on_complete=self._on_run_finished, on_failed=self._on_run_failed + ) + + @SafeSlot() + def abort_schedule(self): + """Request that the schedule stop after the current item.""" + self._abort_requested = True + if self._current_report is not None: + try: + self._current_report.cancel() + except Exception: # pylint: disable=broad-except + logger.exception("Failed to cancel the currently running schedule item") + + @SafeSlot() + def reset_schedule(self): + """Clear all execution state and start the schedule over from item 1.""" + with self._lock: + if self.schedule.is_running: + logger.info("Cannot reset schedule, as schedule is running!") + return + for item in self.schedule.items: + logger.info(f"Reset schedule item {item}") + item.status = ScheduleItemStatus.PENDING + item.request_id = None + item.scan_id = None + item.scan_number = None + item.error = None + item.started_at = None + item.finished_at = None + self._persist_locked() + self.schedule_changed.emit() + + def get_status(self) -> dict: + """RPC-exposed: current schedule state, e.g. for another widget or a script.""" + with self._lock: + return self.schedule.model_dump(mode="json") + + # ------------------------------------------------------------------ # + # RPC-exposed schedule editing (add / edit / move / delete) + # + # All of these only ever touch the PENDING suffix of the schedule - + # see the "protected prefix" section of the module docstring. They can + # be called at any time, including while the schedule is running. + # ------------------------------------------------------------------ # + def add_item( + self, + command: str, + index: int | None = None, + kind: str = "custom", + form_state: dict | None = None, + ) -> str: + """ + RPC-exposed: insert a new, PENDING command into the schedule. + Safe to call while the schedule is running. + + Args: + command: command text, evaluated the same way as the existing + items (against `scans`/`dev`) once the schedule runs. + index: position to insert at (0 = first). Clamped so the item + can never land before something already running or + finished. Defaults to appending at the end. + kind/form_state: optional structured description of how + `command` was built (see `schedule_item.ScheduleItem`), + used to reopen the Edit dialog pre-filled. Leave as + defaults for a plain, hand-typed command. + + Returns: + The new item's item_id. + """ + command = command.strip() + if not command: + raise ValueError("command must not be empty") + + with self._lock: + protected = self._protected_prefix_length_locked() + item = ScheduleItem( + item_id=str(uuid.uuid4()), command=command, kind=kind, form_state=form_state + ) + if index is None or index >= len(self.schedule.items): + self.schedule.items.append(item) + else: + self.schedule.items.insert(max(index, protected), item) + self._persist_locked() + self.schedule_changed.emit() + return item.item_id + + def edit_item( + self, item_id: str, command: str, kind: str = "custom", form_state: dict | None = None + ): + """ + RPC-exposed: change the command of an item that has not started + yet. Raises `RuntimeError` for an item that is already + running/finished. + """ + command = command.strip() + if not command: + raise ValueError("command must not be empty") + + with self._lock: + idx = self._index_of_locked(item_id) + if idx is None: + return + item = self.schedule.items[idx] + if item.status != ScheduleItemStatus.PENDING: + raise RuntimeError("Only items that have not started yet (PENDING) can be edited.") + item.command = command + item.kind = kind + item.form_state = form_state + self._persist_locked() + self.schedule_changed.emit() + + def delete_item(self, item_id: str): + """ + RPC-exposed: remove an item that has not started yet. Raises + `RuntimeError` for an item that is already running/finished. + """ + with self._lock: + idx = self._index_of_locked(item_id) + if idx is None: + return + if self.schedule.items[idx].status != ScheduleItemStatus.PENDING: + raise RuntimeError("Only items that have not started yet (PENDING) can be deleted.") + del self.schedule.items[idx] + self._persist_locked() + self.schedule_changed.emit() + + def move_item(self, item_id: str, new_index: int): + """ + RPC-exposed: move an item that has not started yet to a new + position (0 = first, but never before something already + running/finished). Raises `RuntimeError` for an item that is + already running/finished. + """ + with self._lock: + idx = self._index_of_locked(item_id) + if idx is None: + return + if self.schedule.items[idx].status != ScheduleItemStatus.PENDING: + raise RuntimeError("Only items that have not started yet (PENDING) can be moved.") + protected = self._protected_prefix_length_locked() + item = self.schedule.items.pop(idx) + new_index = max(protected, min(new_index, len(self.schedule.items))) + self.schedule.items.insert(new_index, item) + self._persist_locked() + self.schedule_changed.emit() + + def move_item_up(self, item_id: str): + """RPC-exposed: swap an item with the one directly before it.""" + with self._lock: + idx = self._index_of_locked(item_id) + target = idx - 1 if idx is not None and idx > 0 else None + if target is not None: + self.move_item(item_id, target) + + def move_item_down(self, item_id: str): + """RPC-exposed: swap an item with the one directly after it.""" + with self._lock: + idx = self._index_of_locked(item_id) + n = len(self.schedule.items) + target = idx + 1 if idx is not None and idx < n - 1 else None + if target is not None: + self.move_item(item_id, target) + + # ------------------------------------------------------------------ # + # execution - runs in a background task (BECConnector.submit_task, + # backed by a QThreadPool worker thread; UI updates below go through + # the `schedule_changed` Qt signal so they are marshalled back onto + # the GUI thread instead of touching widgets directly) + # ------------------------------------------------------------------ # + def _pick_next_runnable_locked(self): + """Must be called while holding `self._lock`; see `schedule_logic.pick_next_runnable`.""" + return pick_next_runnable(self.schedule.items) + + def _run_all(self): + namespace = {"scans": self.scans, "dev": self.dev} + while not self._abort_requested: + with self._lock: + next_item = self._pick_next_runnable_locked() + if next_item is None or next_item == "stop": + break + if next_item.status == ScheduleItemStatus.RUNNING: + self._await_running_item(next_item) + else: + self._execute_item(next_item, namespace) + + with self._lock: + self.schedule.is_running = False + self._persist_locked() + self.schedule_changed.emit() + + def _execute_item(self, item: ScheduleItem, namespace: dict): + if item.kind == "scan": + # Block here (item stays PENDING - nothing "started" yet) until + # the guard is clear or the operator aborts. See "Guard + # interaction" in the module docstring. + cleared = self._guard.wait_until_clear(should_abort=lambda: self._abort_requested) + self.schedule_changed.emit() + if not cleared: + return + + with self._lock: + item.status = ScheduleItemStatus.RUNNING + item.error = None + item.started_at = time.time() + self._persist_locked() + self.schedule_changed.emit() + + try: + report = eval( + item.command, {"__builtins__": {}}, namespace + ) # noqa: S307 pylint: disable=eval-used + self._current_report = report + self._current_item_kind = item.kind + + request = getattr(report, "request", None) + with self._lock: + item.request_id = getattr(request, "requestID", None) + # persist the request_id right away, before the + # (potentially long) wait below, so a reconnect can find + # it even if this process disappears mid-scan. + self._persist_locked() + + report.wait() + + scan = getattr(report, "scan", None) + with self._lock: + item.scan_id = getattr(scan, "scan_id", None) if scan else None + item.scan_number = getattr(scan, "scan_number", None) if scan else None + item.status = ( + ScheduleItemStatus.ABORTED + if self._abort_requested + else ScheduleItemStatus.COMPLETED + ) + except Exception: # pylint: disable=broad-except + with self._lock: + if self._guard_interrupt_requested: + # Interrupted by the auto-pause guard, not a real + # failure/operator abort - reset to PENDING so the + # normal execution loop retries it once the guard + # clears, instead of leaving it in a terminal state. + item.status = ScheduleItemStatus.PENDING + item.request_id = None + item.scan_id = None + item.scan_number = None + item.error = None + item.started_at = None + self._guard_interrupt_requested = False + elif self._abort_requested: + item.status = ScheduleItemStatus.ABORTED + else: + item.status = ScheduleItemStatus.FAILED + item.error = traceback.format_exc() + logger.error(f"Schedule item failed: {item.command}\n{item.error}") + finally: + with self._lock: + if item.status != ScheduleItemStatus.PENDING: + item.finished_at = time.time() + self._persist_locked() + final_status = item.status + self._current_report = None + self._current_item_kind = None + self.schedule_changed.emit() + + if final_status in (ScheduleItemStatus.COMPLETED, ScheduleItemStatus.FAILED): + notify_item_finished( + self.connector, + self.schedule_name, + item.kind, + item.command, + success=(final_status == ScheduleItemStatus.COMPLETED), + settings=self.schedule.notifications, + error=item.error, + ) + + def _await_running_item(self, item: ScheduleItem): + queue_storage = self.client.queue.queue_storage + while not self._abort_requested: + queue_item = queue_storage.find_queue_item_by_requestID(item.request_id) + if queue_item is None or not self._is_queue_item_active(queue_item): + with self._lock: + item.status = ScheduleItemStatus.COMPLETED + item.finished_at = time.time() + self._persist_locked() + self.schedule_changed.emit() + return + time.sleep(0.5) + + @SafeSlot() + def _on_run_finished(self): + self._refresh_ui() + + @SafeSlot(str) + def _on_run_failed(self, error: str): + logger.error(f"Schedule execution task failed unexpectedly: {error}") + with self._lock: + self.schedule.is_running = False + self._persist_locked() + self._refresh_ui() + + # ------------------------------------------------------------------ # + # cleanup + # ------------------------------------------------------------------ # + def cleanup(self): + # Note: we deliberately do NOT abort a running schedule here. The + # whole point of the Redis-backed design is that closing this + # widget must not interrupt anything already submitted to the + # scan/device server - see the module docstring. + self._guard.cleanup() + super().cleanup() # also disconnects all bec_dispatcher subscriptions for us diff --git a/debye_bec/bec_widgets/widgets/scheduler/scheduler.pyproject b/debye_bec/bec_widgets/widgets/scheduler/scheduler.pyproject new file mode 100644 index 0000000..0b99313 --- /dev/null +++ b/debye_bec/bec_widgets/widgets/scheduler/scheduler.pyproject @@ -0,0 +1 @@ +{'files': ['scheduler.py']} \ No newline at end of file diff --git a/debye_bec/bec_widgets/widgets/scheduler/scheduler_plugin.py b/debye_bec/bec_widgets/widgets/scheduler/scheduler_plugin.py new file mode 100644 index 0000000..50ce14a --- /dev/null +++ b/debye_bec/bec_widgets/widgets/scheduler/scheduler_plugin.py @@ -0,0 +1,57 @@ +# Copyright (C) 2022 The Qt Company Ltd. +# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause + +from bec_widgets.utils.bec_designer import designer_material_icon +from qtpy.QtDesigner import QDesignerCustomWidgetInterface +from qtpy.QtWidgets import QWidget + +from .scheduler import Scheduler + +DOM_XML = """ + + + + +""" + + +class SchedulerPlugin(QDesignerCustomWidgetInterface): # pragma: no cover + def __init__(self): + super().__init__() + self._form_editor = None + + def createWidget(self, parent): + if parent is None: + return QWidget() + t = Scheduler(parent) + return t + + def domXml(self): + return DOM_XML + + def group(self): + return "" + + def icon(self): + return designer_material_icon(Scheduler.ICON_NAME) + + def includeFile(self): + return "scheduler" + + def initialize(self, form_editor): + self._form_editor = form_editor + + def isContainer(self): + return False + + def isInitialized(self): + return self._form_editor is not None + + def name(self): + return "Scheduler" + + def toolTip(self): + return "Scheduler" + + def whatsThis(self): + return self.toolTip()