from __future__ import annotations
from bec_lib import bec_logger
from bec_lib.endpoints import MessageEndpoints
from bec_qthemes import material_icon
from bec_widgets import BECWidget, SafeProperty, SafeSlot
from bec_widgets.utils.rpc_decorator import rpc_timeout
from bec_widgets.widgets.plots.image.image import Image
from bec_widgets.widgets.plots.waveform.waveform import Waveform
from bec_widgets.widgets.plots.image.setting_widgets.image_roi_tree import ROIPropertyTree
from bec_widgets.widgets.plots.roi.image_roi import BaseROI, CircularROI, RectangularROI
from bec_widgets.widgets.utility.toggle.toggle import ToggleSwitch
from qtpy.QtCore import Qt, QTimer
from qtpy.QtWidgets import (
QFrame,
QGridLayout,
QHBoxLayout,
QLabel,
QLineEdit,
QPushButton,
QSizePolicy,
QSpinBox,
QToolButton,
QVBoxLayout,
QWidget,
QTextEdit,
QTabWidget,
)
logger = bec_logger.logger
CAMERA = ("cam_xeye", "image")
[docs]
class XRayEye2DControl(BECWidget, QWidget):
def __init__(self, parent=None, step_size: int = 100, *arg, **kwargs):
super().__init__(parent=parent, *arg, **kwargs)
self.get_bec_shortcuts()
self._step_size = step_size
self.root_layout = QGridLayout(self)
self.setStyleSheet("""
QToolButton {
border: 1px solid;
border-radius: 4px;
}
""")
# Up
self.move_up_button = QToolButton(parent=self)
self.move_up_button.setIcon(material_icon("keyboard_double_arrow_up"))
self.root_layout.addWidget(self.move_up_button, 0, 2)
# Up tweak button
self.move_up_tweak_button = QToolButton(parent=self)
self.move_up_tweak_button.setIcon(material_icon("keyboard_arrow_up"))
self.root_layout.addWidget(self.move_up_tweak_button, 1, 2)
# Left
self.move_left_button = QToolButton(parent=self)
self.move_left_button.setIcon(material_icon("keyboard_double_arrow_left"))
self.root_layout.addWidget(self.move_left_button, 2, 0)
# Left tweak button
self.move_left_tweak_button = QToolButton(parent=self)
self.move_left_tweak_button.setIcon(material_icon("keyboard_arrow_left"))
self.root_layout.addWidget(self.move_left_tweak_button, 2, 1)
# Right
self.move_right_button = QToolButton(parent=self)
self.move_right_button.setIcon(material_icon("keyboard_double_arrow_right"))
self.root_layout.addWidget(self.move_right_button, 2, 4)
# Right tweak button
self.move_right_tweak_button = QToolButton(parent=self)
self.move_right_tweak_button.setIcon(material_icon("keyboard_arrow_right"))
self.root_layout.addWidget(self.move_right_tweak_button, 2, 3)
# Down
self.move_down_button = QToolButton(parent=self)
self.move_down_button.setIcon(material_icon("keyboard_double_arrow_down"))
self.root_layout.addWidget(self.move_down_button, 4, 2)
# Down tweak button
self.move_down_tweak_button = QToolButton(parent=self)
self.move_down_tweak_button.setIcon(material_icon("keyboard_arrow_down"))
self.root_layout.addWidget(self.move_down_tweak_button, 3, 2)
# Connections
self.move_up_button.clicked.connect(lambda: self.move("up", tweak=False))
self.move_up_tweak_button.clicked.connect(lambda: self.move("up", tweak=True))
self.move_down_button.clicked.connect(lambda: self.move("down", tweak=False))
self.move_down_tweak_button.clicked.connect(lambda: self.move("down", tweak=True))
self.move_left_button.clicked.connect(lambda: self.move("left", tweak=False))
self.move_left_tweak_button.clicked.connect(lambda: self.move("left", tweak=True))
self.move_right_button.clicked.connect(lambda: self.move("right", tweak=False))
self.move_right_tweak_button.clicked.connect(lambda: self.move("right", tweak=True))
@SafeProperty(int)
def step_size(self) -> int:
return self._step_size
@step_size.setter
def step_size(self, step_size: int):
self._step_size = step_size
@SafeSlot(bool)
def enable_controls_hor(self, enable: bool):
self.move_left_button.setEnabled(enable)
self.move_left_tweak_button.setEnabled(enable)
self.move_right_button.setEnabled(enable)
self.move_right_tweak_button.setEnabled(enable)
@SafeSlot(bool)
def enable_controls_ver(self, enable: bool):
self.move_up_button.setEnabled(enable)
self.move_up_tweak_button.setEnabled(enable)
self.move_down_button.setEnabled(enable)
self.move_down_tweak_button.setEnabled(enable)
def move(self, direction: str, tweak: bool = False):
step = self._step_size
if tweak:
step = int(self._step_size / 5)
if direction == "up":
self.dev.omny_xray_gui.mvy.set(step)
elif direction == "down":
self.dev.omny_xray_gui.mvy.set(-step)
elif direction == "left":
self.dev.omny_xray_gui.mvx.set(-step)
elif direction == "right":
self.dev.omny_xray_gui.mvx.set(step)
else:
logger.warning(f"Unknown direction {direction} for move command.")
[docs]
class XRayEye(BECWidget, QWidget):
USER_ACCESS = [
"active_roi",
"user_message",
"user_message.setter",
"on_live_view_enabled",
"on_motors_enable",
"enable_submit_button",
"sample_name",
"sample_name.setter",
"enable_move_buttons",
"enable_move_buttons.setter",
"switch_tab",
"set_dap_params_forwarding",
"submit_fit_array",
]
PLUGIN = True
def __init__(self, parent=None, **kwargs):
super().__init__(parent=parent, **kwargs)
self._connected_motor = None
self._dap_params_forwarding_connected = False
self._queue_busy = False
self._queue_idle_timer = QTimer(self)
self._queue_idle_timer.setSingleShot(True)
self._queue_idle_timer.setInterval(800)
self._queue_idle_timer.timeout.connect(self._release_queue_toggles)
self.get_bec_shortcuts()
self._init_ui()
self._make_connections()
# Connection to redis endpoints
self.bec_dispatcher.connect_slot(
self.getting_shutter_status, MessageEndpoints.device_readback("fsh")
)
self.bec_dispatcher.connect_slot(
self.getting_camera_status, MessageEndpoints.device_read_configuration(CAMERA[0])
)
self.bec_dispatcher.connect_slot(
self.on_queue_status_update, MessageEndpoints.scan_queue_status()
)
self.connect_motors()
self.resize(800, 600)
QTimer.singleShot(0, self._init_queue_status)
QTimer.singleShot(0, self._init_gui_trigger)
def _init_ui(self):
self.root_layout = QVBoxLayout(self)
self.tab_widget = QTabWidget(parent=self)
self.root_layout.addWidget(self.tab_widget)
self.alignment_tab = QWidget(parent=self)
self.core_layout = QHBoxLayout(self.alignment_tab)
self.image = Image(parent=self.alignment_tab)
self.image.color_map = "CET-L2"
self.image.enable_toolbar = (
False # Disable default toolbar to not allow to user set anything
)
self.image.inner_axes = False # Disable inner axes to maximize image area
self.image.enable_full_colorbar = True
self.image.invert_y = True # Invert y axis to match image coordinates
# Control panel on the right: vertical layout inside a fixed-width widget
self.control_panel = QWidget(parent=self.alignment_tab)
self.control_panel_layout = QVBoxLayout(self.control_panel)
self.control_panel_layout.setContentsMargins(0, 0, 0, 0)
self.control_panel_layout.setSpacing(10)
# ROI toolbar + Live toggle (header row)
self.roi_manager = ROIPropertyTree(
parent=self, image_widget=self.image, compact=True, compact_orientation="horizontal"
)
header_row = QHBoxLayout()
header_row.setContentsMargins(0, 0, 0, 0)
header_row.setSpacing(8)
header_row.addWidget(self.roi_manager, 0)
header_row.addStretch()
self.live_preview_label = QLabel("Live Preview", parent=self)
self.live_preview_toggle = ToggleSwitch(parent=self)
self.live_preview_toggle.checked = False
header_row.addWidget(self.live_preview_label, 0, Qt.AlignmentFlag.AlignVCenter)
header_row.addWidget(self.live_preview_toggle, 0, Qt.AlignmentFlag.AlignVCenter)
self.control_panel_layout.addLayout(header_row)
self.switch_row_widget = QWidget(parent=self)
switch_row = QHBoxLayout(self.switch_row_widget)
switch_row.setContentsMargins(0, 0, 0, 0)
switch_row.setSpacing(8)
switch_row.addStretch()
self.camera_running_label = QLabel("Camera running", parent=self)
self.camera_running_toggle = ToggleSwitch(parent=self)
# self.camera_running_toggle.checked = False
self.camera_running_toggle.enabled.connect(self.camera_running_enabled)
self.shutter_label = QLabel("Shutter open", parent=self)
self.shutter_toggle = ToggleSwitch(parent=self)
# self.shutter_toggle.checked = False
self.shutter_toggle.enabled.connect(self.opening_shutter)
switch_row.addWidget(self.shutter_label, 0, Qt.AlignmentFlag.AlignVCenter)
switch_row.addWidget(self.shutter_toggle, 0, Qt.AlignmentFlag.AlignVCenter)
switch_row.addWidget(self.camera_running_label, 0, Qt.AlignmentFlag.AlignVCenter)
switch_row.addWidget(self.camera_running_toggle, 0, Qt.AlignmentFlag.AlignVCenter)
self.control_panel_layout.addWidget(self.switch_row_widget)
# separator
self.control_panel_layout.addWidget(self._create_separator())
# 2D Positioner (fixed size)
self.motor_control_2d = XRayEye2DControl(parent=self)
self.control_panel_layout.addWidget(
self.motor_control_2d, 0, Qt.AlignmentFlag.AlignTop | Qt.AlignmentFlag.AlignCenter
)
# separator
self.control_panel_layout.addWidget(self._create_separator())
# Step size label
step_size_form = QGridLayout()
# General Step size
self.step_size = QSpinBox(parent=self)
self.step_size.setRange(10, 100)
self.step_size.setSingleStep(10)
self.step_size.setValue(100)
# Submit button
self.submit_button = QPushButton("Submit", parent=self)
# Add to layout form
step_size_form.addWidget(QLabel("Step Size", parent=self), 0, 0)
step_size_form.addWidget(self.step_size, 0, 1)
step_size_form.addWidget(self.submit_button, 2, 0, 1, 2)
# Add form to control panel
self.control_panel_layout.addLayout(step_size_form)
# Push form to bottom
self.control_panel_layout.addStretch()
# Sample/Message form (bottom)
form = QGridLayout()
self.sample_name_line_edit = QLineEdit(parent=self)
self.sample_name_line_edit.setReadOnly(True)
form.addWidget(QLabel("Sample", parent=self), 0, 0)
form.addWidget(self.sample_name_line_edit, 0, 1)
self.message_line_edit = QTextEdit(parent=self)
self.message_line_edit.setFixedHeight(60)
self.message_line_edit.setReadOnly(True)
form.addWidget(QLabel("Message", parent=self), 1, 0)
form.addWidget(self.message_line_edit, 1, 1)
self.control_panel_layout.addLayout(form)
# Fix panel width and allow vertical expansion
self.control_panel.adjustSize()
p_hint = self.control_panel.sizeHint()
self.control_panel.setFixedWidth(p_hint.width())
self.control_panel.setSizePolicy(QSizePolicy.Policy.Fixed, QSizePolicy.Policy.Expanding)
# Core Layout: image (expanding) | control panel (fixed)
self.core_layout.addWidget(self.image)
self.core_layout.addWidget(self.control_panel)
self.tab_widget.addTab(self.alignment_tab, "Alignment")
self.fit_tab = QWidget(parent=self)
self.fit_layout = QVBoxLayout(self.fit_tab)
self.waveform_x = Waveform(parent=self.fit_tab)
self.waveform_y = Waveform(parent=self.fit_tab)
self.waveform_x.plot(
x=[0],
y=[1],
label="fit-x",
dap=["SineModel", "LinearModel"],
dap_parameters=[
{"frequency": {"value": 0.0174533, "vary": False, "min": 0.01, "max": 0.02}},
{"slope": {"value": 0, "vary": False, "min": 0.0, "max": 0.02}},
],
dap_oversample=5,
)
self.waveform_y.plot(
x=[0],
y=[2],
label="fit-y",
dap=["SineModel", "LinearModel"],
dap_parameters=[
{"frequency": {"value": 0.0174533, "vary": False, "min": 0.01, "max": 0.02}},
{"slope": {"value": 0, "vary": False, "min": 0.0, "max": 0.02}},
],
dap_oversample=5,
)
self.fit_x = self.waveform_x.curves[0]
self.fit_y = self.waveform_y.curves[0]
for wave in (self.waveform_x, self.waveform_y):
wave.x_label = "Angle (deg)"
wave.x_grid = True
wave.y_grid = True
wave.enable_toolbar = True
self.fit_layout.addWidget(self.waveform_x)
self.fit_layout.addWidget(self.waveform_y)
self.tab_widget.addTab(self.fit_tab, "Fit")
def _make_connections(self):
# Fetch initial state
self.on_live_view_enabled(True)
self.step_size.setValue(self.motor_control_2d.step_size)
# Make connections
self.live_preview_toggle.enabled.connect(self.on_live_view_enabled)
self.step_size.valueChanged.connect(
lambda x: self.motor_control_2d.setProperty("step_size", x)
)
self.submit_button.clicked.connect(self.submit)
def _create_separator(self):
sep = QFrame(parent=self)
sep.setFrameShape(QFrame.Shape.HLine)
sep.setFrameShadow(QFrame.Shadow.Sunken)
sep.setLineWidth(1)
return sep
def _init_gui_trigger(self):
self.dev.omny_xray_gui.read()
self.dev.fsh.read()
################################################################################
# Device Connection logic
################################################################################
[docs]
def connect_motors(self):
"""Checks one of the possible motors for flomni, omny and lamni setup."""
possible_motors = ["osamroy", "lsamrot", "fsamroy"]
for motor in possible_motors:
if motor in self.dev:
self.bec_dispatcher.connect_slot(
self.on_tomo_angle_readback, MessageEndpoints.device_readback(motor)
)
logger.info(f"Successfully connected to {motor}")
self._connected_motor = motor
################################################################################
# Properties ported from the original OmnyAlignment, can be adjusted as needed
################################################################################
@SafeProperty(str)
def user_message(self):
return self.message_line_edit.toPlainText()
@user_message.setter
@rpc_timeout(20)
def user_message(self, message: str):
self.message_line_edit.setText(message)
@SafeProperty(str)
def sample_name(self):
return self.sample_name_line_edit.text()
@sample_name.setter
@rpc_timeout(20)
def sample_name(self, message: str):
self.sample_name_line_edit.setText(message)
@SafeProperty(bool)
def enable_move_buttons(self):
return self.motor_control_2d.isEnabled()
@enable_move_buttons.setter
def enable_move_buttons(self, enabled: bool):
self.motor_control_2d.setEnabled(enabled)
def _queue_guarded_toggles(self) -> tuple[ToggleSwitch, ToggleSwitch, ToggleSwitch]:
return (self.live_preview_toggle, self.shutter_toggle, self.camera_running_toggle)
def _set_queue_toggles_blocked(self, blocked: bool):
if blocked == self._queue_busy:
return
self._queue_busy = blocked
self._refresh_queue_toggle_availability()
def _refresh_queue_toggle_availability(self):
tooltip = "Disabled while scan queue is busy." if self._queue_busy else ""
for toggle in self._queue_guarded_toggles():
toggle.setEnabled(not self._queue_busy)
toggle.setToolTip(tooltip)
def _manual_toggle_blocked_by_queue(self) -> bool:
return self._queue_busy and self.sender() in self._queue_guarded_toggles()
def _update_queue_toggles_from_busy_state(self, busy: bool):
if busy:
self._queue_idle_timer.stop()
self._set_queue_toggles_blocked(True)
return
if self._queue_busy and not self._queue_idle_timer.isActive():
self._queue_idle_timer.start()
def _release_queue_toggles(self):
self._set_queue_toggles_blocked(False)
def _init_queue_status(self):
try:
msg = self.client.connector.get(MessageEndpoints.scan_queue_status())
except Exception as exc:
logger.warning(f"Failed to fetch initial scan queue status for XRayEye: {exc}")
return
if msg is None:
return
self._update_queue_toggles_from_busy_state(self._is_queue_busy(msg.content))
@staticmethod
def _is_queue_busy(msg_content: dict) -> bool:
queues = msg_content.get("queue", {}) if isinstance(msg_content, dict) else {}
primary_queue = queues.get("primary") if isinstance(queues, dict) else None
if primary_queue is None:
return False
queue_info = getattr(primary_queue, "info", None)
if queue_info is None and isinstance(primary_queue, dict):
queue_info = primary_queue.get("info", [])
if not queue_info:
return False
idle_statuses = {"STOPPED", "COMPLETED", "IDLE"}
for item in queue_info:
status = getattr(item, "status", None)
if status is None and isinstance(item, dict):
status = item.get("status")
if str(status).upper() not in idle_statuses:
return True
return False
[docs]
def active_roi(self) -> BaseROI | None:
"""Return the currently active ROI, or None if no ROI is active."""
return self.roi_manager.single_active_roi
################################################################################
# Slots ported from the original OmnyAlignment, can be adjusted as needed
################################################################################
@SafeSlot(str)
@rpc_timeout(20)
def switch_tab(self, tab: str):
if tab == "fit":
self.tab_widget.setCurrentIndex(1)
else:
self.tab_widget.setCurrentIndex(0)
[docs]
@SafeSlot()
def get_roi_coordinates(self) -> dict | None:
"""Get the coordinates of the currently active ROI."""
roi = self.roi_manager.single_active_roi
if roi is None:
logger.warning("No active ROI")
return None
logger.info(f"Active ROI coordinates: {roi.get_coordinates()}")
return roi.get_coordinates()
@SafeSlot(bool)
@rpc_timeout(20)
def on_live_view_enabled(self, enabled: bool):
if self._manual_toggle_blocked_by_queue():
logger.warning("Ignoring live-preview toggle while scan queue is busy.")
return
logger.info(f"Live view is enabled: {enabled}")
self.live_preview_toggle.blockSignals(True)
if enabled:
self.live_preview_toggle.checked = enabled
self.image.image(device=CAMERA[0], signal=CAMERA[1])
self.live_preview_toggle.blockSignals(False)
return
self.image.disconnect_monitor(CAMERA[0], CAMERA[1])
self.live_preview_toggle.checked = enabled
self.live_preview_toggle.blockSignals(False)
@SafeSlot(bool)
def camera_running_enabled(self, enabled: bool):
if self._manual_toggle_blocked_by_queue():
logger.warning("Ignoring camera live-mode toggle while scan queue is busy.")
return
logger.info(f"Camera running: {enabled}")
self.camera_running_toggle.blockSignals(True)
self.dev.get(CAMERA[0]).live_mode_enabled.put(enabled)
self.camera_running_toggle.checked = enabled
self.camera_running_toggle.blockSignals(False)
@SafeSlot(dict, dict)
def getting_camera_status(self, data, meta):
print(f"msg:{data}")
live_mode_enabled = data.get("signals").get(f"{CAMERA[0]}_live_mode_enabled").get("value")
self.camera_running_toggle.blockSignals(True)
self.camera_running_toggle.checked = live_mode_enabled
self.camera_running_toggle.blockSignals(False)
@SafeSlot(bool)
def opening_shutter(self, enabled: bool):
if self._manual_toggle_blocked_by_queue():
logger.warning("Ignoring shutter toggle while scan queue is busy.")
return
logger.info(f"Shutter changed from GUI to: {enabled}")
self.shutter_toggle.blockSignals(True)
if enabled:
self.dev.fsh.fshopen()
else:
self.dev.fsh.fshclose()
# self.shutter_toggle.checked = enabled
self.shutter_toggle.blockSignals(False)
@SafeSlot(dict, dict)
def getting_shutter_status(self, data, meta):
shutter_open = bool(data.get("signals").get("fsh_shutter").get("value"))
self.shutter_toggle.blockSignals(True)
self.shutter_toggle.checked = shutter_open
self.shutter_toggle.blockSignals(False)
@SafeSlot(dict, dict)
def on_queue_status_update(self, data, meta):
_ = meta
self._update_queue_toggles_from_busy_state(self._is_queue_busy(data))
[docs]
@SafeSlot(bool, bool)
@rpc_timeout(20)
def on_motors_enable(self, x_enable: bool, y_enable: bool):
"""
Enable/Disable motor controls
Args:
x_enable(bool): enable x motor controls
y_enable(bool): enable y motor controls
"""
self.motor_control_2d.enable_controls_hor(x_enable)
self.motor_control_2d.enable_controls_ver(y_enable)
[docs]
@SafeSlot(bool)
@rpc_timeout(20)
def set_dap_params_forwarding(self, enabled: bool):
"""
Connect or disconnect DAP fit parameter forwarding to omny_xray_gui.
"""
if enabled == self._dap_params_forwarding_connected:
return
signals = (self.waveform_x.dap_params_update, self.waveform_y.dap_params_update)
if enabled:
for signal in signals:
signal.connect(self.on_dap_params)
self._dap_params_forwarding_connected = True
logger.info("Enabled XRayEye DAP parameter forwarding.")
return
for signal in signals:
try:
signal.disconnect(self.on_dap_params)
except (TypeError, RuntimeError):
pass
self._dap_params_forwarding_connected = False
logger.info("Disabled XRayEye DAP parameter forwarding.")
@SafeSlot(dict, dict)
def on_dap_params(self, data, meta):
print("#######################################")
print("getting dap parameters")
print(f"data: {data}")
print(f"meta: {meta}")
self.waveform_x.auto_range(True)
self.waveform_y.auto_range(True)
# self.bec_dispatcher.disconnect_slot(self.device_updates, MessageEndpoints.device_readback("omny_xray_gui"))
curve_id = meta.get("curve_id")
if curve_id == "fit-x-SineModel+LinearModel":
self.dev.omny_xray_gui.fit_params_x.set(data)
print(f"setting x data to {data}")
else:
self.dev.omny_xray_gui.fit_params_y.set(data)
print(f"setting y data to {data}")
# self.bec_dispatcher.connect_slot(self.device_updates, MessageEndpoints.device_readback("omny_xray_gui"))
@SafeSlot(bool, bool)
def on_tomo_angle_readback(self, data: dict, meta: dict):
# TODO implement if needed
print(f"data: {data}")
print(f"meta: {meta}")
@SafeSlot()
@rpc_timeout(20)
def submit_fit_array(self, fit_array):
self.tab_widget.setCurrentIndex(1)
# self.fix_x.title = " got fit array"
print(f"got fit array {fit_array}")
self.waveform_x.curves[0].set_data(x=fit_array[0], y=fit_array[1])
self.waveform_y.curves[0].set_data(x=fit_array[0], y=fit_array[2])
# self.fit_x.set_data(x=fit_array[0],y=fit_array[1])
# self.fit_y.set_data(x=fit_array[0],y=fit_array[2])
[docs]
@SafeSlot()
def submit(self):
"""Execute submit action by submit button."""
self.submit_button.blockSignals(True)
try:
if self.roi_manager.single_active_roi is None:
logger.warning("No active ROI")
return
roi_coordinates = self.roi_manager.single_active_roi.get_coordinates()
roi_center_x = roi_coordinates["center_x"]
roi_center_y = roi_coordinates["center_y"]
# Case of rectangular ROI
if isinstance(self.roi_manager.single_active_roi, RectangularROI):
roi_width = roi_coordinates["width"]
roi_height = roi_coordinates["height"]
elif isinstance(self.roi_manager.single_active_roi, CircularROI):
roi_width = roi_coordinates["diameter"]
roi_height = roi_coordinates["radius"]
else:
logger.warning("Unsupported ROI type for submit action.")
return
# submit roi coordinates
step = int(self.dev.omny_xray_gui.step.read().get("omny_xray_gui_step").get("value"))
getattr(self.dev.omny_xray_gui, f"xval_x_{step}").set(roi_center_x)
getattr(self.dev.omny_xray_gui, f"yval_y_{step}").set(roi_center_y)
getattr(self.dev.omny_xray_gui, f"width_x_{step}").set(roi_width)
getattr(self.dev.omny_xray_gui, f"width_y_{step}").set(roi_height)
self.dev.omny_xray_gui.submit.set(1)
finally:
self.submit_button.blockSignals(False)
[docs]
def cleanup(self):
"""Cleanup connections on widget close -> disconnect slots and stop live mode of camera."""
self._queue_idle_timer.stop()
if self._connected_motor is not None:
self.bec_dispatcher.disconnect_slot(
self.on_tomo_angle_readback, MessageEndpoints.device_readback(self._connected_motor)
)
self.bec_dispatcher.disconnect_slot(
self.getting_shutter_status, MessageEndpoints.device_readback("fsh")
)
self.bec_dispatcher.disconnect_slot(
self.getting_camera_status, MessageEndpoints.device_read_configuration(CAMERA[0])
)
self.bec_dispatcher.disconnect_slot(
self.on_queue_status_update, MessageEndpoints.scan_queue_status()
)
# getattr(self.dev, CAMERA[0]).stop_live_mode() #TODO temporary disabled
super().cleanup()
if __name__ == "__main__":
import sys
from qtpy.QtWidgets import QApplication
from bec_widgets.utils import BECDispatcher
from bec_widgets.utils.colors import apply_theme
app = QApplication(sys.argv)
apply_theme("light")
dispatcher = BECDispatcher(gui_id="xray")
win = XRayEye()
win.resize(1000, 800)
win.show()
sys.exit(app.exec_())