bec/debye_bec
0
0 Fork 0
Code Issues 9 Pull Requests 4 Actions Packages Projects Releases Wiki Activity

wip(digital_twin): Move selection button. motion interlock
CI for debye_bec / test (pull_request) Successful in 1m31s
Details
CI for debye_bec / test (push) Successful in 1m32s
Details

Browse Source
This commit is contained in:
hitz_s committed 2026-10-21 13:51:52 +02:00
1 parent a182c3f773
commit 7144e55a7b
6 files changed
+255 -89
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

debye_bec/bec_ipython_client/plugins/digital_twin_core/digital_twin_core.py
+77 -35
View File
@@ -12,7 +12,7 @@ from xrt.backends.raycing.physconsts import AVOGADRO, CHeVcm
from . import parameters as bl
from .beamline import get_beamline_id
from .motion_worker import MotionWorker
from .motion_worker import MotionOutcome, MotionWorker
from .types import BeamlineId, ConfigDict, DataDict, SurfaceDict
H = 6.62606957e-34
@@ -27,21 +27,27 @@ logger = bec_logger.logger
class _ExclusiveGroupManager:
"""Coordinates one exclusive group: only motors from the same
subgroup may be active at once; a different subgroup must wait until
the active one fully finishes."""
def __init__(self):
self._cond = threading.Condition()
self._active_subgroup = None
self._active_count = 0
def acquire(self, subgroup_id):
def acquire(self, subgroup_id, cancelled=None, on_wait=None) -> bool:
"""Block until the subgroup may run. Returns False if cancelled while waiting."""
with self._cond:
notified = False
while self._active_subgroup is not None and self._active_subgroup != subgroup_id:
self._cond.wait()
if cancelled and cancelled():
return False
if on_wait and not notified:
on_wait()
notified = True
self._cond.wait(0.1) # timeout so cancellation is noticed
if cancelled and cancelled():
return False
self._active_subgroup = subgroup_id
self._active_count += 1
return True
def release(self):
with self._cond:
@@ -95,6 +101,8 @@ class DigitalTwinCore:
if load_offsets:
self.load_offsets()
self._group_managers = {gi: _ExclusiveGroupManager() for gi in range(len(EXCLUSIVE_GROUPS))}
def _exclusive_group_key(self, motor: str):
"""Return (group_index, subgroup_index) for `motor`, or (None, None)
if it isn't part of any exclusive group and can move freely."""
@@ -104,60 +112,94 @@ class DigitalTwinCore:
return gi, si
return None, None
def move_with_config(self, config):
def run_exclusive(
self, worker, abort=None, cancel=None, on_waiting=None, on_started=None
) -> str:
"""
Move all motors in `config` to their target positions, all starting at
Run `worker` once its exclusive group allows it. Blocks the calling thread.
abort: shared by a batch of motors; set on the first failure so that
motors of the batch that have not started yet are not started.
cancel: per-motor event (user pressed Stop while waiting).
"""
events = [e for e in (abort, cancel) if e is not None]
cancelled = lambda: any(e.is_set() for e in events)
gi, si = self._exclusive_group_key(
worker.motor
) # before run(): aliases change worker.motor
manager = self._group_managers[gi] if gi is not None else None
if manager is not None and not manager.acquire(si, cancelled, on_waiting):
return MotionOutcome.CANCELLED
try:
if cancelled():
return MotionOutcome.CANCELLED
if on_started:
on_started()
try:
worker.run()
except Exception as e: # raised outside motion()'s own try/except
logger.error(f"Error preparing movement of {worker.motor}: {e}")
worker.last_exception = e
if worker.last_exception is not None:
if abort is not None:
abort.set()
return MotionOutcome.FAILED
return MotionOutcome.DONE
finally:
if manager is not None:
manager.release()
def move_with_config(self, config, motors: list | None = None):
"""
Move all motors according to the "config" to their target positions, all starting at
once. Motors are free to move concurrently unless they belong to the
same exclusive group (see `EXCLUSIVE_GROUPS`) but different subgroups,
in which case one subgroup must fully finish before another starts.
If "motors" is provided, only these motors will be moved
Returns:
dict: {"success": True, "moved": [...]} if every motor completed
without error, else {"success": False, "moved": [...],
"failures": {motor: error}}.
"failures": {motor: error}} or "aborted": {motor}
"""
positions = self.calc_positions(config)
positions = self.apply_offsets(positions, nested_config=True)
managers: dict[int, _ExclusiveGroupManager] = {}
for motor in positions:
gi, _ = self._exclusive_group_key(motor)
if gi is not None:
managers.setdefault(gi, _ExclusiveGroupManager())
if motors is not None:
positions = {m: p for m, p in positions.items() if m in motors}
moved: list[str] = []
abort = threading.Event()
moved, aborted = [], []
failures: dict[str, Exception | str] = {}
results_lock = threading.Lock()
def run_motor(motor: str, target: float):
gi, si = self._exclusive_group_key(motor)
manager = managers.get(gi) if gi is not None else None
if manager:
manager.acquire(si)
try:
worker = MotionWorker(self.beamline, self.dev, motor, target)
worker.run() # blocks this thread until finished or errored
with results_lock:
if worker.last_exception is not None:
failures[motor] = worker.last_exception
else:
moved.append(motor)
finally:
if manager:
manager.release()
worker = MotionWorker(self.beamline, self.dev, motor, target)
outcome = self.run_exclusive(worker, abort=abort)
with results_lock:
if outcome == MotionOutcome.DONE:
moved.append(motor)
elif outcome == MotionOutcome.FAILED:
failures[motor] = worker.last_exception
else:
aborted.append(motor)
threads = [
threading.Thread(target=run_motor, args=(motor, pos["value"]), daemon=True)
for motor, pos in positions.items()
threading.Thread(target=run_motor, args=(m, p["value"]), daemon=True)
for m, p in positions.items()
]
for t in threads:
t.start()
for t in threads:
t.join()
result = {"success": not failures, "moved": moved}
if failures:
return {"success": False, "moved": moved, "failures": failures}
return {"success": True, "moved": moved}
result["failures"] = failures
result["aborted"] = aborted # never started because of the failure
return result
def load_offsets(self):
if self.offsets == {}:
debye_bec/bec_ipython_client/plugins/digital_twin_core/motion_worker.py
+7
View File
@@ -13,6 +13,12 @@ from .types import BeamlineId
logger = bec_logger.logger
class MotionOutcome:
DONE = "done"
FAILED = "failed"
CANCELLED = "cancelled" # stopped by the user, or aborted because another motor failed
class MotionWorker:
"""
Executes motion on the specified motor and includes some safety during
@@ -239,6 +245,7 @@ class MotionWorker:
now = time.time()
if time.time() - last_check < update_interval:
time.sleep(0.01)
continue
last_check = now
if self._stop_flag.is_set():
self.dev[self.motor].stop()
debye_bec/bec_widgets/widgets/digital_twin/digital_twin.py
+14 -3
View File
@@ -4,6 +4,7 @@ Digital Twin: Custom BEC widget to support the beamline alignment.
import json
import sys
import threading
import time
from pathlib import Path
from typing import Literal, cast
@@ -108,7 +109,7 @@ class DigitalTwin(BECWidget, QWidget):
self.plot_layout.addWidget(self.surface_plots, stretch=1)
self.plot_widget.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding)
self.mover = MoverPanel(self.beamline, self.dev)
self.mover = MoverPanel(self.core, self.beamline, self.dev)
self.input_scroll = self._scroll_area(self.input_widget, min_width=320, max_width=360)
self.mover_scroll = self._scroll_area(self.mover, min_width=380, max_width=460)
@@ -155,6 +156,8 @@ class DigitalTwin(BECWidget, QWidget):
self.settings.load_from_config.clicked_connect(self.load_from_config)
self.settings.load_from_scan.clicked_connect(self.load_from_scan)
self.mover.multi_move.clicked_connect(self._multi_move)
self.edge_selector_energy = 0.0
self.edge_selector_element = ""
self.edge_selector_edge = ""
@@ -803,7 +806,7 @@ class DigitalTwin(BECWidget, QWidget):
self.settings.load_offsets.setText("Unload")
self.settings.offsets_status.setText("Loaded and applied")
self.settings.offsets_status.setColor(get_accent_colors().success.name())
self.settings.show_offsets.enable_button(True)
self.settings.show_offsets.setEnabled(True)
else:
# Offsets were unloaded
self.calc_assistant(identifier="init")
@@ -811,7 +814,7 @@ class DigitalTwin(BECWidget, QWidget):
self.settings.load_offsets.setText("Load")
self.settings.offsets_status.setText("No offsets")
self.settings.offsets_status.setColor(get_accent_colors().default.name())
self.settings.show_offsets.enable_button(False)
self.settings.show_offsets.setEnabled(False)
@SafeSlot()
def show_offsets(self, *_):
@@ -1325,6 +1328,14 @@ class DigitalTwin(BECWidget, QWidget):
energy = self.input.energy.value()
self.input.cm_pitch_critical.setValue(-self.core.cm_critical_angle(cm_stripe, energy) * 1e3)
def _multi_move(self):
"""
Move the selected motors at the same time (if permitted)."""
abort = threading.Event() # one failure aborts the not-yet-started motors of this batch
for w in self.mover.mover_widgets:
if w.checkbox.isChecked():
w.start_motion(abort=abort)
if __name__ == "__main__":
app = QApplication(sys.argv)
debye_bec/bec_widgets/widgets/digital_twin/panels/mover_panel.py
+46 -2
View File
@@ -9,22 +9,25 @@ from qtpy.QtWidgets import QVBoxLayout, QWidget
from .....bec_ipython_client.plugins.digital_twin_core.types import BeamlineId
from ..widgets.move_widget import AbsorberWidget, MoveWidget
from ..widgets.qt_widgets import Group
from ..widgets.qt_widgets import Button, Group, QLabel
class MoverPanel(QWidget):
""" "Panel to move an axis to a certain position"""
def __init__(self, beamline: BeamlineId, dev, parent=None):
def __init__(self, core, beamline: BeamlineId, dev, parent=None):
super().__init__(parent)
self._layout = QVBoxLayout(self)
self._layout.setContentsMargins(4, 4, 4, 4)
self._layout.setSpacing(4)
self.core = core
self.mover_widgets = []
# FE Slits
self.sldi_gapx = MoveWidget(
core=self.core,
beamline=beamline,
dev=dev,
motor="sldi_gapx",
@@ -36,6 +39,7 @@ class MoverPanel(QWidget):
self.mover_widgets.append(self.sldi_gapx)
self.sldi_gapy = MoveWidget(
core=self.core,
beamline=beamline,
dev=dev,
motor="sldi_gapy",
@@ -55,6 +59,7 @@ class MoverPanel(QWidget):
# Collimating mirror
self.cm_trx = MoveWidget(
core=self.core,
beamline=beamline,
dev=dev,
motor="cm_trx",
@@ -66,6 +71,7 @@ class MoverPanel(QWidget):
self.mover_widgets.append(self.cm_trx)
self.cm_try = MoveWidget(
core=self.core,
beamline=beamline,
dev=dev,
motor="cm_try",
@@ -77,6 +83,7 @@ class MoverPanel(QWidget):
self.mover_widgets.append(self.cm_try)
self.cm_bnd = MoveWidget(
core=self.core,
beamline=beamline,
dev=dev,
motor="cm_bnd",
@@ -88,6 +95,7 @@ class MoverPanel(QWidget):
self.mover_widgets.append(self.cm_bnd)
self.cm_rotx = MoveWidget(
core=self.core,
beamline=beamline,
dev=dev,
motor="cm_rotx",
@@ -104,6 +112,7 @@ class MoverPanel(QWidget):
# Monochromator
self.mo1_bragg_angle = MoveWidget(
core=self.core,
beamline=beamline,
dev=dev,
motor="mo1_bragg_angle",
@@ -115,6 +124,7 @@ class MoverPanel(QWidget):
self.mover_widgets.append(self.mo1_bragg_angle)
self.mo1_trx = MoveWidget(
core=self.core,
beamline=beamline,
dev=dev,
motor="mo1_trx",
@@ -126,6 +136,7 @@ class MoverPanel(QWidget):
self.mover_widgets.append(self.mo1_trx)
self.mo1_try = MoveWidget(
core=self.core,
beamline=beamline,
dev=dev,
motor="mo1_try",
@@ -138,6 +149,7 @@ class MoverPanel(QWidget):
if beamline == BeamlineId.X01DA:
self.mo1_roty = MoveWidget(
core=self.core,
beamline=beamline,
dev=dev,
motor="mo1_roty",
@@ -158,6 +170,7 @@ class MoverPanel(QWidget):
# OP Slits 1
self.sl1_centery = MoveWidget(
core=self.core,
beamline=beamline,
dev=dev,
motor="sl1_centery",
@@ -169,6 +182,7 @@ class MoverPanel(QWidget):
self.mover_widgets.append(self.sl1_centery)
self.sl1_gapy = MoveWidget(
core=self.core,
beamline=beamline,
dev=dev,
motor="sl1_gapy",
@@ -183,6 +197,7 @@ class MoverPanel(QWidget):
# OP Beam Monitor 1
self.bm1_try = MoveWidget(
core=self.core,
beamline=beamline,
dev=dev,
motor="bm1_try",
@@ -197,6 +212,7 @@ class MoverPanel(QWidget):
# Focusing Mirror
self.fm_trx = MoveWidget(
core=self.core,
beamline=beamline,
dev=dev,
motor="fm_trx",
@@ -208,6 +224,7 @@ class MoverPanel(QWidget):
self.mover_widgets.append(self.fm_trx)
self.fm_try = MoveWidget(
core=self.core,
beamline=beamline,
dev=dev,
motor="fm_try",
@@ -219,6 +236,7 @@ class MoverPanel(QWidget):
self.mover_widgets.append(self.fm_try)
self.fm_bnd = MoveWidget(
core=self.core,
beamline=beamline,
dev=dev,
motor="fm_bnd",
@@ -230,6 +248,7 @@ class MoverPanel(QWidget):
self.mover_widgets.append(self.fm_bnd)
self.fm_rotx = MoveWidget(
core=self.core,
beamline=beamline,
dev=dev,
motor="fm_rotx",
@@ -241,6 +260,7 @@ class MoverPanel(QWidget):
self.mover_widgets.append(self.fm_rotx)
self.fm_roty = MoveWidget(
core=self.core,
beamline=beamline,
dev=dev,
motor="fm_roty",
@@ -252,6 +272,7 @@ class MoverPanel(QWidget):
self.mover_widgets.append(self.fm_roty)
self.fm_rotz = MoveWidget(
core=self.core,
beamline=beamline,
dev=dev,
motor="fm_rotz",
@@ -270,6 +291,7 @@ class MoverPanel(QWidget):
if beamline == "x01da":
# OP Slits 2
self.sl2_centery = MoveWidget(
core=self.core,
beamline=beamline,
dev=dev,
motor="sl2_centery",
@@ -281,6 +303,7 @@ class MoverPanel(QWidget):
self.mover_widgets.append(self.sl2_centery)
self.sl2_gapy = MoveWidget(
core=self.core,
beamline=beamline,
dev=dev,
motor="sl2_gapy",
@@ -295,6 +318,7 @@ class MoverPanel(QWidget):
# OP Beam Monitor 2
self.bm2_try = MoveWidget(
core=self.core,
beamline=beamline,
dev=dev,
motor="bm2_try",
@@ -310,6 +334,7 @@ class MoverPanel(QWidget):
if beamline == "x01da":
# Optical Table
self.ot_try = MoveWidget(
core=self.core,
beamline=beamline,
dev=dev,
motor="ot_try",
@@ -321,6 +346,7 @@ class MoverPanel(QWidget):
self.mover_widgets.append(self.ot_try)
self.ot_rotx = MoveWidget(
core=self.core,
beamline=beamline,
dev=dev,
motor="ot_rotx",
@@ -335,6 +361,7 @@ class MoverPanel(QWidget):
# Experimental Station 0
self.es0wi_try = MoveWidget(
core=self.core,
beamline=beamline,
dev=dev,
motor="es0wi_try",
@@ -350,6 +377,7 @@ class MoverPanel(QWidget):
# Experimental Station 1
if beamline == "x01da":
self.ot_es1_trz = MoveWidget(
core=self.core,
beamline=beamline,
dev=dev,
motor="ot_es1_trz",
@@ -360,6 +388,7 @@ class MoverPanel(QWidget):
)
self.mover_widgets.append(self.ot_es1_trz)
self.es1man_trz = MoveWidget(
core=self.core,
beamline=beamline,
dev=dev,
motor="es1man_trz",
@@ -372,6 +401,7 @@ class MoverPanel(QWidget):
if beamline == "x10da":
self.es1_try = MoveWidget(
core=self.core,
beamline=beamline,
dev=dev,
motor="es1_try",
@@ -383,6 +413,7 @@ class MoverPanel(QWidget):
self.mover_widgets.append(self.es1_try)
self.es1ic0_try = MoveWidget(
core=self.core,
beamline=beamline,
dev=dev,
motor="es1ic0_try",
@@ -394,6 +425,7 @@ class MoverPanel(QWidget):
self.mover_widgets.append(self.es1ic0_try)
self.es1ic1_try = MoveWidget(
core=self.core,
beamline=beamline,
dev=dev,
motor="es1ic1_try",
@@ -405,6 +437,7 @@ class MoverPanel(QWidget):
self.mover_widgets.append(self.es1ic1_try)
self.es1ic2_try = MoveWidget(
core=self.core,
beamline=beamline,
dev=dev,
motor="es1ic2_try",
@@ -425,6 +458,7 @@ class MoverPanel(QWidget):
# Experimental Station 2
if beamline == "x01da":
self.ot_es2_trz = MoveWidget(
core=self.core,
beamline=beamline,
dev=dev,
motor="ot_es2_trz",
@@ -437,6 +471,7 @@ class MoverPanel(QWidget):
if beamline == "x10da":
self.es2_try = MoveWidget(
core=self.core,
beamline=beamline,
dev=dev,
motor="es2_try",
@@ -452,6 +487,13 @@ class MoverPanel(QWidget):
else:
self.es2_mov_group = Group("Experimental Station 2", [self.es2_try])
# Multi-Mover
self.multi_move = Button(label_button="Move Selection", enabled=True, color="success")
self.multi_move_info = QLabel(
"Not all motors can move together at the same time, digital twin will take care of that."
)
self.multi_move_group = Group("Multi Move", [self.multi_move, self.multi_move_info])
# Assemble complete mover group
if beamline == "x01da":
self.mover_group = Group(
@@ -470,6 +512,7 @@ class MoverPanel(QWidget):
self.es0_mov_group,
self.es1_mov_group,
self.es2_mov_group,
self.multi_move_group,
],
)
else:
@@ -487,6 +530,7 @@ class MoverPanel(QWidget):
self.es0_mov_group,
self.es1_mov_group,
self.es2_mov_group,
self.multi_move_group,
],
)
debye_bec/bec_widgets/widgets/digital_twin/widgets/move_widget.py
+89 -36
View File
@@ -12,21 +12,23 @@ from qtpy.QtCore import Property # type: ignore[attr-defined]
from qtpy.QtCore import Signal # type: ignore[attr-defined]
from qtpy.QtCore import QObject, QPropertyAnimation, Qt
from qtpy.QtGui import QTransform
from qtpy.QtWidgets import QApplication, QHBoxLayout, QLabel, QPushButton, QWidget
from qtpy.QtWidgets import QApplication, QCheckBox, QHBoxLayout, QLabel, QPushButton, QWidget
from .....bec_ipython_client.plugins.digital_twin_core.motion_worker import MotionWorker
from .....bec_ipython_client.plugins.digital_twin_core.motion_worker import (
MotionOutcome,
MotionWorker,
)
from .....bec_ipython_client.plugins.digital_twin_core.types import BeamlineId
logger = bec_logger.logger
class Status:
"""Status class for the axis"""
IN_POSITION = "in_position" # green mdi.check-circle
NOT_IN_POSITION = "not_in_position" # orange mdi.close-circle
MOVING = "moving" # blue mdi.loading (spinning)
ERROR = "error" # red mdi.alert-circle
IN_POSITION = "in_position" # green
NOT_IN_POSITION = "not_in_position" # orange
QUEUED = "queued" # grey, waiting for another motor group
MOVING = "moving" # blue
ERROR = "error" # red
class StatusIcon(QWidget):
@@ -42,6 +44,7 @@ class StatusIcon(QWidget):
Status.NOT_IN_POSITION: ("cancel", "#e6d922"),
Status.ERROR: ("warning", "#e74c3c"),
Status.MOVING: ("cycle", "#2980b9"),
Status.QUEUED: ("hourglass_empty", "#7f8c8d"),
}
def __init__(self, parent=None):
@@ -122,10 +125,10 @@ class StatusIcon(QWidget):
class _WorkerSignals(QObject):
"""Tiny QObject just to own signals, since MotionWorker itself is no longer a QObject."""
position_changed = Signal(float)
error = Signal()
waiting = Signal() # must wait for another subgroup
started = Signal() # slot acquired, motion begins
finished = Signal()
@@ -138,10 +141,13 @@ class MoveWidget(QWidget):
- Start / Stop button
"""
def __init__(
self, beamline: BeamlineId, dev, motor, label: str = "", unit=None, decimals=3, deadband=0.0
):
_STATUS_TOOLTIPS = {Status.QUEUED: "Waiting for other motors to finish"}
def __init__(self, core, beamline, dev, motor, label="", unit=None, decimals=3, deadband=0.0):
super().__init__()
self.core = core
self._cancel: threading.Event | None = None
self.fb = 0.0
self.target = 0
self.beamline = beamline
@@ -162,6 +168,11 @@ class MoveWidget(QWidget):
layout.setContentsMargins(4, 0, 4, 0)
layout.setSpacing(4)
# Checkbox
self.checkbox = QCheckBox()
self.checkbox.setChecked(True)
layout.addWidget(self.checkbox)
# Name
self.label = QLabel(label)
self.label.setFixedWidth(76)
@@ -226,6 +237,11 @@ class MoveWidget(QWidget):
+ f"{{background-color: {get_accent_colors().emergency.name()}; color: white;}}"
)
@property
def is_busy(self) -> bool:
"""True from start until the finished signal was handled (moving OR queued)."""
return self._thread is not None
def set_target(self, target):
"""Change the target value in the ui"""
self.target = target
@@ -262,13 +278,13 @@ class MoveWidget(QWidget):
)
def _set_status(self, status: str):
"""Set the current status icon in the ui"""
self.status = status
self.status_icon.set_status(status)
self.status_icon.setToolTip(self._STATUS_TOOLTIPS.get(status, ""))
def _on_target_or_fb_changed(self):
"""Re-evaluate in-position status whenever the target value changes."""
if self.status in (Status.ERROR, Status.MOVING):
if self.status in (Status.ERROR, Status.MOVING, Status.QUEUED):
return
if abs(self.fb - self.target) <= self.deadband:
self._set_status(Status.IN_POSITION)
@@ -276,49 +292,86 @@ class MoveWidget(QWidget):
self._set_status(Status.NOT_IN_POSITION)
def _on_button_clicked(self):
"""Starts or stops motion depending on current situation"""
if self._thread and self._thread.is_alive():
if self.is_busy:
self._stop_motion()
else:
self._start_motion()
self.start_motion()
def _start_motion(self):
"""Start a motion"""
target = self.target
if abs(target - self.fb) <= self.deadband:
def start_motion(self, abort: threading.Event | None = None):
"""Start a motion. `abort` is shared by a batch (Move Selection) so that one failure
prevents the not-yet-started motors of that batch from starting."""
if self.is_busy:
return
if abs(self.target - self.fb) <= self.deadband:
self._set_status(Status.IN_POSITION)
return
self._set_status(Status.MOVING)
self._apply_button_style("stop")
self._cancel = threading.Event()
self._signals = _WorkerSignals()
self._signals.position_changed.connect(self._on_position_changed)
self._signals.error.connect(self._on_error)
self._signals.finished.connect(self._on_motion_finished)
self._signals.finished.connect(self._cleanup_thread)
signals = _WorkerSignals()
signals.waiting.connect(self._on_waiting)
signals.started.connect(self._on_started)
signals.position_changed.connect(self._on_position_changed)
signals.error.connect(self._on_error)
signals.finished.connect(self._on_motion_finished)
signals.finished.connect(self._cleanup_thread)
self._signals = signals
self._worker = MotionWorker(
# No on_finished here: _run emits it exactly once (MotionWorker.run never calls
# it for motors that are "not integrated in digital twin").
worker = MotionWorker(
self.beamline,
self.dev,
self.motor,
target,
on_position_changed=self._signals.position_changed.emit,
on_error=self._signals.error.emit,
on_finished=self._signals.finished.emit,
self.target,
on_position_changed=signals.position_changed.emit,
on_error=signals.error.emit,
)
self._worker = worker
self._thread = threading.Thread(
target=self._run, args=(worker, signals, self._cancel, abort), daemon=True
)
self._thread = threading.Thread(target=self._worker.run, daemon=True)
self._thread.start()
def _run(self, worker, signals, cancel, abort):
try:
outcome = self.core.run_exclusive(...)
if outcome == MotionOutcome.FAILED and worker.last_exception is not None:
signals.error.emit() # harmless if motion() already emitted it
finally:
signals.finished.emit()
try:
self.core.run_exclusive(
worker,
abort=abort,
cancel=cancel,
on_waiting=signals.waiting.emit,
on_started=signals.started.emit,
)
except Exception:
signals.error.emit()
raise
finally:
signals.finished.emit()
def _on_waiting(self):
self._set_status(Status.QUEUED)
def _on_started(self):
self._set_status(Status.MOVING)
def _on_error(self):
"""Called when an error occurs"""
self._set_status(Status.ERROR)
self._apply_button_style("start")
def _stop_motion(self):
"""Attempts to stop the motion"""
"""Stop a running motor, or remove a queued one from the queue."""
if self._cancel:
self._cancel.set() # releases a waiting motor within ~0.1 s
if self._worker:
self._worker.stop()
self._worker.stop() # stops a running motor
def _on_position_changed(self, pos: float):
"""Change the feedback value in the ui"""
debye_bec/bec_widgets/widgets/digital_twin/widgets/qt_widgets.py
+22 -13
View File
@@ -179,7 +179,9 @@ class ComboBox(QWidget):
class Button(QWidget):
def __init__(self, label=None, label_button: str = "", enabled=False):
def __init__(
self, label=None, label_button: str = "", enabled: bool = False, color: str = "default"
):
super().__init__()
layout = QHBoxLayout(self)
layout.setContentsMargins(*ROW_MARGINS)
@@ -189,28 +191,35 @@ class Button(QWidget):
self.label.setFixedWidth(LABEL_WIDTH)
layout.addWidget(self.label)
self.button = QPushButton(label_button)
self.enable_button(enabled)
self.setEnabled(enabled)
layout.addWidget(self.button)
self.color = color
def clicked_connect(self, func):
"""Connect a function to the button press."""
self.button.clicked.connect(func)
def enable_button(self, enable: bool = False):
if enable:
self.button.setStyleSheet(
f"QPushButton {{background-color: {get_accent_colors().default.name()}; color: white;}}"
)
self.button.setEnabled(True)
else: # disabled
self.button.setStyleSheet(
"QPushButton {{background-color: rgb(120, 120, 120); color: white;}}"
)
self.button.setDisabled(True)
def setEnabled(self, enable: bool = True):
self.button.setEnabled(enable)
self.apply_theme()
def setText(self, text):
self.button.setText(text)
def setColor(self, color: str):
self.color = color
self.apply_theme()
def apply_theme(self):
if self.button.isEnabled():
colors = get_accent_colors()
color = getattr(colors, self.color).name()
self.button.setStyleSheet(f"QPushButton {{ background-color: {color}; color: white; }}")
else:
self.button.setStyleSheet(
"QPushButton {{background-color: rgb(120, 120, 120); color: white;}}"
)
class TextIndicator(QWidget):
def __init__(self, label):