From 4590b85010fcf2a3984a5fa6187c42443b65921c Mon Sep 17 00:00:00 2001 From: wyzula-jan Date: Wed, 15 Oct 2025 12:42:37 +0200 Subject: [PATCH 1/7] fix(omny_alignment): disabled not working signal connection to on_move_up --- csaxs_bec/bec_widgets/widgets/omny_alignment/omny_alignment.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/csaxs_bec/bec_widgets/widgets/omny_alignment/omny_alignment.py b/csaxs_bec/bec_widgets/widgets/omny_alignment/omny_alignment.py index 5d47042..d39c93e 100644 --- a/csaxs_bec/bec_widgets/widgets/omny_alignment/omny_alignment.py +++ b/csaxs_bec/bec_widgets/widgets/omny_alignment/omny_alignment.py @@ -63,7 +63,7 @@ class OmnyAlignment(BECWidget, QWidget): self.ui.liveViewSwitch.enabled.connect(self.on_live_view_enabled) - self.ui.moveUpButton.clicked.connect(self.on_move_up) + # self.ui.moveUpButton.clicked.connect(self.on_move_up) @property -- 2.49.1 From 5b76c3f769a97c169d7019fca053d9357c0cb928 Mon Sep 17 00:00:00 2001 From: x01dc Date: Thu, 16 Oct 2025 12:11:21 +0200 Subject: [PATCH 2/7] feat(gui_instruction_device): added gui instruction device from epics --- csaxs_bec/device_configs/flomni_config.yaml | 13 +++- csaxs_bec/devices/omny/xray_epics_gui.py | 74 +++++++++++++++++++++ 2 files changed, 86 insertions(+), 1 deletion(-) create mode 100644 csaxs_bec/devices/omny/xray_epics_gui.py diff --git a/csaxs_bec/device_configs/flomni_config.yaml b/csaxs_bec/device_configs/flomni_config.yaml index ab0b16f..fd72916 100644 --- a/csaxs_bec/device_configs/flomni_config.yaml +++ b/csaxs_bec/device_configs/flomni_config.yaml @@ -417,4 +417,15 @@ flomni_temphum: enabled: true onFailure: buffer readOnly: false - readoutPriority: baseline \ No newline at end of file + readoutPriority: baseline +############################################################ +#################### GUI Signals ########################### +############################################################ +omny_xray_gui: + description: Gui Epics signals + deviceClass: csaxs_bec.devices.omny.xray_epics_gui.OMNYXRayEpicsGUI + deviceConfig: {} + enabled: true + onFailure: buffer + readOnly: false + readoutPriority: on_request \ No newline at end of file diff --git a/csaxs_bec/devices/omny/xray_epics_gui.py b/csaxs_bec/devices/omny/xray_epics_gui.py new file mode 100644 index 0000000..7db7bb7 --- /dev/null +++ b/csaxs_bec/devices/omny/xray_epics_gui.py @@ -0,0 +1,74 @@ + +from ophyd import Component as Cpt +from ophyd import Device +from ophyd import DynamicDeviceComponent as Dcpt +from ophyd import EpicsSignal + + + +class OMNYXRayEpicsGUI(Device): + + save_frame = Cpt( + EpicsSignal, name="save_frame", read_pv="XOMNYI-XEYE-SAVFRAME:0",auto_monitor=True + ) + update_frame_acqdone = Cpt( + EpicsSignal, name="update_frame_acqdone", read_pv="XOMNYI-XEYE-ACQDONE:0",auto_monitor=True + ) + update_frame_acq = Cpt( + EpicsSignal, name="update_frame_acq", read_pv="XOMNYI-XEYE-ACQ:0",auto_monitor=True + ) + width_y_dynamic = { + f"width_y_{i}": (EpicsSignal, f"XOMNYI-XEYE-YWIDTH_Y:{i}", {"auto_monitor": True}) for i in range(0, 11) + } + width_y = Dcpt(width_y_dynamic) + width_x_dynamic = { + f"width_x_{i}": (EpicsSignal, f"XOMNYI-XEYE-XWIDTH_X:{i}", {"auto_monitor": True}) for i in range(0, 11) + } + width_x = Dcpt(width_x_dynamic) + enable_mv_x = Cpt( + EpicsSignal, name="enable_mv_x", read_pv="XOMNYI-XEYE-ENAMVX:0",auto_monitor=True + ) + enable_mv_y = Cpt( + EpicsSignal, name="enable_mv_y", read_pv="XOMNYI-XEYE-ENAMVY:0",auto_monitor=True + ) + send_message = Cpt( + EpicsSignal, name="send_message", read_pv="XOMNYI-XEYE-MESSAGE:0.DESC",auto_monitor=True + ) + sample_name = Cpt( + EpicsSignal, name="sample_name", read_pv="XOMNYI-XEYE-SAMPLENAME:0.DESC",auto_monitor=True + ) + angle = Cpt( + EpicsSignal, name="angle", read_pv="XOMNYI-XEYE-ANGLE:0",auto_monitor=True + ) + pixel_size = Cpt( + EpicsSignal, name="pixel_size", read_pv="XOMNYI-XEYE-PIXELSIZE:0",auto_monitor=True + ) + submit = Cpt( + EpicsSignal, name="submit", read_pv="XOMNYI-XEYE-SUBMIT:0",auto_monitor=True + ) + step = Cpt( + EpicsSignal, name="step", read_pv="XOMNYI-XEYE-STEP:0",auto_monitor=True + ) + xval_x_dynamic = { + f"xval_x_{i}": (EpicsSignal, f"XOMNYI-XEYE-XVAL_X:{i}", {"auto_monitor": True}) for i in range(0, 11) + } + xval_x = Dcpt(xval_x_dynamic) + yval_y_dynamic = { + f"yval_y_{i}": (EpicsSignal, f"XOMNYI-XEYE-YVAL_Y:{i}", {"auto_monitor": True}) for i in range(0, 11) + } + yval_y = Dcpt(yval_y_dynamic) + recbg = Cpt( + EpicsSignal, name="recbg", read_pv="XOMNYI-XEYE-RECBG:0",auto_monitor=True + ) + stage_pos_x_dynamic = { + f"stage_pos_x_{i}": (EpicsSignal, f"XOMNYI-XEYE-STAGEPOSX:{i}", {"auto_monitor": True}) for i in range(1, 6) + } + stage_pos_x = Dcpt(stage_pos_x_dynamic) + mvx = Cpt( + EpicsSignal, name="mvx", read_pv="XOMNYI-XEYE-MVX:0",auto_monitor=True + ) + mvy = Cpt( + EpicsSignal, name="mvy", read_pv="XOMNYI-XEYE-MVY:0",auto_monitor=True + ) + + -- 2.49.1 From 4723f6768b8f84415c03eda7b2b52b9fde285ce8 Mon Sep 17 00:00:00 2001 From: wyzula-jan Date: Wed, 15 Oct 2025 12:43:07 +0200 Subject: [PATCH 3/7] feat(xray_eye): add XRayEye widget and plugin for GUI integration --- csaxs_bec/bec_widgets/widgets/client.py | 75 +++- .../bec_widgets/widgets/xray_eye/__init__.py | 0 .../widgets/xray_eye/register_x_ray_eye.py | 15 + .../bec_widgets/widgets/xray_eye/x_ray_eye.py | 423 ++++++++++++++++++ .../widgets/xray_eye/x_ray_eye.pyproject | 1 + .../widgets/xray_eye/x_ray_eye_plugin.py | 57 +++ 6 files changed, 570 insertions(+), 1 deletion(-) create mode 100644 csaxs_bec/bec_widgets/widgets/xray_eye/__init__.py create mode 100644 csaxs_bec/bec_widgets/widgets/xray_eye/register_x_ray_eye.py create mode 100644 csaxs_bec/bec_widgets/widgets/xray_eye/x_ray_eye.py create mode 100644 csaxs_bec/bec_widgets/widgets/xray_eye/x_ray_eye.pyproject create mode 100644 csaxs_bec/bec_widgets/widgets/xray_eye/x_ray_eye_plugin.py diff --git a/csaxs_bec/bec_widgets/widgets/client.py b/csaxs_bec/bec_widgets/widgets/client.py index 0c7f321..ec2d37c 100644 --- a/csaxs_bec/bec_widgets/widgets/client.py +++ b/csaxs_bec/bec_widgets/widgets/client.py @@ -5,7 +5,7 @@ from __future__ import annotations from bec_lib.logger import bec_logger -from bec_widgets.cli.rpc.rpc_base import RPCBase, rpc_call +from bec_widgets.cli.rpc.rpc_base import RPCBase, rpc_call, rpc_timeout logger = bec_logger.logger @@ -14,6 +14,7 @@ logger = bec_logger.logger _Widgets = { "OmnyAlignment": "OmnyAlignment", + "XRayEye": "XRayEye", } @@ -73,3 +74,75 @@ class OmnyAlignment(RPCBase): """ None """ + + +class XRayEye(RPCBase): + @rpc_call + def active_roi(self) -> "BaseROI | None": + """ + Return the currently active ROI, or None if no ROI is active. + """ + + @property + @rpc_call + def enable_live_view(self): + """ + Get or set the live view enabled state. + """ + + @enable_live_view.setter + @rpc_call + def enable_live_view(self): + """ + Get or set the live view enabled state. + """ + + @property + @rpc_call + def user_message(self): + """ + None + """ + + @user_message.setter + @rpc_call + def user_message(self): + """ + None + """ + + @property + @rpc_call + def sample_name(self): + """ + None + """ + + @sample_name.setter + @rpc_call + def sample_name(self): + """ + None + """ + + @property + @rpc_call + def enable_move_buttons(self): + """ + None + """ + + @enable_move_buttons.setter + @rpc_call + def enable_move_buttons(self): + """ + None + """ + + +class XRayEye2DControl(RPCBase): + @rpc_call + def remove(self): + """ + Cleanup the BECConnector + """ diff --git a/csaxs_bec/bec_widgets/widgets/xray_eye/__init__.py b/csaxs_bec/bec_widgets/widgets/xray_eye/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/csaxs_bec/bec_widgets/widgets/xray_eye/register_x_ray_eye.py b/csaxs_bec/bec_widgets/widgets/xray_eye/register_x_ray_eye.py new file mode 100644 index 0000000..8fd0e39 --- /dev/null +++ b/csaxs_bec/bec_widgets/widgets/xray_eye/register_x_ray_eye.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 csaxs_bec.bec_widgets.widgets.xray_eye.x_ray_eye_plugin import XRayEyePlugin + + QPyDesignerCustomWidgetCollection.addCustomWidget(XRayEyePlugin()) + + +if __name__ == "__main__": # pragma: no cover + main() diff --git a/csaxs_bec/bec_widgets/widgets/xray_eye/x_ray_eye.py b/csaxs_bec/bec_widgets/widgets/xray_eye/x_ray_eye.py new file mode 100644 index 0000000..406cc63 --- /dev/null +++ b/csaxs_bec/bec_widgets/widgets/xray_eye/x_ray_eye.py @@ -0,0 +1,423 @@ +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.widgets.plots.image.image import Image +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, +) + +logger = bec_logger.logger +CAMERA = ("cam_xeye_rgb", "image") #TODO here put correct camera + + +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.") + + +class XRayEye(BECWidget, QWidget): + USER_ACCESS = ["active_roi", "enable_live_view", "enable_live_view.setter", "user_message", "user_message.setter", + "sample_name", "sample_name.setter", "enable_move_buttons", "enable_move_buttons.setter"] + PLUGIN = True + + def __init__(self, parent=None, **kwargs): + super().__init__(parent=parent, **kwargs) + self.get_bec_shortcuts() + + self._init_ui() + self._make_connections() + + # Connection to redis endpoints + self.bec_dispatcher.connect_slot(self.device_updates, MessageEndpoints.device_readback("omny_xray_gui")) + self.connect_motors() + self.resize(800, 600) + QTimer.singleShot(0, self._init_gui_trigger) + + def _init_ui(self): + self.core_layout = QHBoxLayout(self) + + self.image = Image(parent=self) + 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.plot_item.vb.invertY(True) # #TODO Invert y axis to match logic of LabView GUI + + # Control panel on the right: vertical layout inside a fixed-width widget + self.control_panel = QWidget(parent=self) + 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.AlignVCenter) + header_row.addWidget(self.live_preview_toggle, 0, Qt.AlignVCenter) + self.control_panel_layout.addLayout(header_row) + + # 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.AlignTop | Qt.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("Horizontal", parent=self), 0, 0) + step_size_form.addWidget(self.step_size, 0, 1) + step_size_form.addWidget(QLabel("Vertical", parent=self), 1, 0) + 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 = QLineEdit(parent=self) + 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.Fixed, QSizePolicy.Expanding) + + # Core Layout: image (expanding) | control panel (fixed) + self.core_layout.addWidget(self.image) + self.core_layout.addWidget(self.control_panel) + + 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.HLine) + sep.setFrameShadow(QFrame.Sunken) + sep.setLineWidth(1) + return sep + + def _init_gui_trigger(self): + self.dev.omny_xray_gui.read() + + ################################################################################ + # Device Connection logic + ################################################################################ + + 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"Succesfully connected to {motor}") + + ################################################################################ + # Properties ported from the original OmnyAlignment, can be adjusted as needed + ################################################################################ + @SafeProperty(bool) + def enable_live_view(self): + """Get or set the live view enabled state.""" + return self.live_preview_toggle.checked + + @enable_live_view.setter + def enable_live_view(self, enable: bool): + self.live_preview_toggle.checked = enable + + @SafeProperty(str) + def user_message(self): + return self.message_line_edit.text() + + @user_message.setter + 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 + 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 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() + 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) + def on_live_view_enabled(self, enabled: bool): + 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(CAMERA) + self.live_preview_toggle.blockSignals(False) + return + + self.image.disconnect_monitor(CAMERA) + self.live_preview_toggle.checked = enabled + self.live_preview_toggle.blockSignals(False) + + @SafeSlot(bool, bool) + def on_motors_enable(self, x_enable: bool, y_enable: bool): + self.motor_control_2d.enable_controls_hor(x_enable) + self.motor_control_2d.enable_controls_ver(y_enable) + + @SafeSlot(str) + def set_message(self, msg: str): + self.message_line_edit.setText(msg) + + @SafeSlot(str) + def set_sample_name(self, msg: str): + self.sample_name_line_edit.setText(msg) + + @SafeSlot(int) + def enable_submit_button(self, enable: int): + """If -1 disable else enable""" + if enable == -1: + self.submit_button.setEnabled(False) + else: + self.submit_button.setEnabled(True) + + @SafeSlot(bool, bool) + def on_tomo_angle_readback(self, data: dict, meta: dict): + print(f"data: {data}") + print(f"meta: {meta}") + + @SafeSlot(dict, dict) + def device_updates(self, data: dict, meta: dict): + + signals = data.get('signals') + enable_live_preview = signals.get("omny_xray_gui_update_frame_acq").get('value') + enable_x_motor = signals.get("omny_xray_gui_enable_mv_x").get('value') + enable_y_motor = signals.get("omny_xray_gui_enable_mv_y").get('value') + self.on_live_view_enabled(bool(enable_live_preview)) + self.on_motors_enable(bool(enable_x_motor), bool(enable_y_motor)) + + # Signals from epics gui device + # send message + send_message = signals.get("omny_xray_gui_send_message").get('value') + self.set_message(send_message) + # sample name + sample_message = signals.get("omny_xray_gui_sample_name").get('value') + self.set_sample_name(sample_message) + # enable frame acquisition + update_frame_acq = signals.get("omny_xray_gui_update_frame_acq").get('value') + self.on_live_view_enabled(bool(update_frame_acq)) + # enable submit button + enable_submit_button = signals.get("omny_xray_gui_submit").get('value') + self.enable_submit_button(enable_submit_button) + + @SafeSlot() + def submit(self): + """Execute submit action by submit button.""" + 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 + + print(f"current roi: {roi_center_x},{roi_center_y}, {roi_width},{roi_height}") + # submit roi coordinates + step = int(self.dev.omny_xray_gui.step.read().get("omny_xray_gui_step").get('value')) + + xval_x = getattr(self.dev.omny_xray_gui.xval_x, f"xval_x_{step}").set(roi_center_x) + xval_y = getattr(self.dev.omny_xray_gui.yval_y, f"yval_y_{step}").set(roi_center_y) + width_x = getattr(self.dev.omny_xray_gui.width_x, f"width_x_{step}").set(roi_width) + width_y = getattr(self.dev.omny_xray_gui.width_y, f"width_y_{step}").set(roi_height) + self.dev.omny_xray_gui.submit.set(1) + + def cleanup(self): + self.bec_dispatcher.disconnect_slot(self.device_updates, MessageEndpoints.device_readback("omny_xray_gui")) + getattr(self.dev,CAMERA).live_mode = False + super().cleanup() + +if __name__ == "__main__": + import sys + + from qtpy.QtWidgets import QApplication + + app = QApplication(sys.argv) + + win = XRayEye() + + win.resize(1000, 800) + win.show() + sys.exit(app.exec_()) diff --git a/csaxs_bec/bec_widgets/widgets/xray_eye/x_ray_eye.pyproject b/csaxs_bec/bec_widgets/widgets/xray_eye/x_ray_eye.pyproject new file mode 100644 index 0000000..b8433bf --- /dev/null +++ b/csaxs_bec/bec_widgets/widgets/xray_eye/x_ray_eye.pyproject @@ -0,0 +1 @@ +{'files': ['x_ray_eye.py']} \ No newline at end of file diff --git a/csaxs_bec/bec_widgets/widgets/xray_eye/x_ray_eye_plugin.py b/csaxs_bec/bec_widgets/widgets/xray_eye/x_ray_eye_plugin.py new file mode 100644 index 0000000..f611e47 --- /dev/null +++ b/csaxs_bec/bec_widgets/widgets/xray_eye/x_ray_eye_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 csaxs_bec.bec_widgets.widgets.xray_eye.x_ray_eye import XRayEye + +DOM_XML = """ + + + + +""" + + +class XRayEyePlugin(QDesignerCustomWidgetInterface): # pragma: no cover + def __init__(self): + super().__init__() + self._form_editor = None + + def createWidget(self, parent): + if parent is None: + return QWidget() + t = XRayEye(parent) + return t + + def domXml(self): + return DOM_XML + + def group(self): + return "" + + def icon(self): + return designer_material_icon(XRayEye.ICON_NAME) + + def includeFile(self): + return "x_ray_eye" + + 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 "XRayEye" + + def toolTip(self): + return "XRayEye" + + def whatsThis(self): + return self.toolTip() -- 2.49.1 From de22611941dfb5fbf0c952529b7cbad919e4b514 Mon Sep 17 00:00:00 2001 From: x01dc Date: Wed, 22 Oct 2025 15:44:41 +0200 Subject: [PATCH 4/7] refactor(ids_camera): old ids_camera deleted, ids_camera_new is now ids_camera --- csaxs_bec/device_configs/flomni_config.yaml | 37 +- csaxs_bec/devices/ids_cameras/__init__.py | 2 +- csaxs_bec/devices/ids_cameras/ids_camera.py | 535 ++++++------------ .../devices/ids_cameras/ids_camera_new.py | 218 ------- tests/tests_devices/test_ids_camera.py | 2 +- 5 files changed, 203 insertions(+), 591 deletions(-) delete mode 100644 csaxs_bec/devices/ids_cameras/ids_camera_new.py diff --git a/csaxs_bec/device_configs/flomni_config.yaml b/csaxs_bec/device_configs/flomni_config.yaml index fd72916..1a3bfd1 100644 --- a/csaxs_bec/device_configs/flomni_config.yaml +++ b/csaxs_bec/device_configs/flomni_config.yaml @@ -393,18 +393,31 @@ cam_flomni_overview: readOnly: false readoutPriority: on_request -# cam_flomni_xeye: -# description: Camera flOMNI Xray eye ID101 -# deviceClass: csaxs_bec.devices.ids_cameras.ids_camera.IDSCamera -# deviceConfig: -# camera_ID: 101 -# bits_per_pixel: 24 -# channels: 3 -# m_n_colormode: 1 -# enabled: true -# onFailure: buffer -# readOnly: false -# readoutPriority: async +cam_xeye_mono: + description: Camera flOMNI Xray eye ID1 + deviceClass: csaxs_bec.devices.ids_cameras.ids_camera.IDSCamera + deviceConfig: + camera_id: 1 + bits_per_pixel: 24 + # channels: 3 + m_n_colormode: 1 + enabled: true + onFailure: buffer + readOnly: false + readoutPriority: async + +cam_xeye_rgb: + description: Camera flOMNI Xray eye ID203 + deviceClass: csaxs_bec.devices.ids_cameras.ids_camera.IDSCamera + deviceConfig: + camera_id: 203 + bits_per_pixel: 24 + # channels: 3 + m_n_colormode: 1 + enabled: true + onFailure: buffer + readOnly: false + readoutPriority: async # ############################################################ diff --git a/csaxs_bec/devices/ids_cameras/__init__.py b/csaxs_bec/devices/ids_cameras/__init__.py index a16d89e..ce0499a 100644 --- a/csaxs_bec/devices/ids_cameras/__init__.py +++ b/csaxs_bec/devices/ids_cameras/__init__.py @@ -1 +1 @@ -from .ids_camera_new import IDSCamera +from .ids_camera import IDSCamera diff --git a/csaxs_bec/devices/ids_cameras/ids_camera.py b/csaxs_bec/devices/ids_cameras/ids_camera.py index 1250927..52c857b 100644 --- a/csaxs_bec/devices/ids_cameras/ids_camera.py +++ b/csaxs_bec/devices/ids_cameras/ids_camera.py @@ -1,403 +1,220 @@ +"""IDS Camera class for cSAXS IDS cameras.""" + +from __future__ import annotations + import threading import time +from typing import TYPE_CHECKING, Literal, Tuple, TypedDict import numpy as np +from bec_lib import messages from bec_lib.logger import bec_logger from ophyd import Component as Cpt -from ophyd import DeviceStatus, Kind, Signal, StatusBase from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase -from ophyd_devices.utils.bec_signals import PreviewSignal +from ophyd_devices.utils.bec_signals import AsyncSignal, PreviewSignal + +from csaxs_bec.devices.ids_cameras.base_integration.camera import Camera + +if TYPE_CHECKING: + from bec_lib.devicemanager import ScanInfo + from pydantic import ValidationInfo + logger = bec_logger.logger -class ROISignal(Signal): - """ - Signal to handle the Region of Interest (ROI) for the IDS camera. - It is a tuple of (x, y, width, height). - """ - - def __init__( - self, - *, - name, - roi: tuple | None = None, - value=0, - dtype=None, - shape=None, - timestamp=None, - parent=None, - labels=None, - kind=Kind.hinted, - tolerance=None, - rtolerance=None, - metadata=None, - cl=None, - attr_name="", - ): - super().__init__( - name=name, - value=value, - dtype=dtype, - shape=shape, - timestamp=timestamp, - parent=parent, - labels=labels, - kind=kind, - tolerance=tolerance, - rtolerance=rtolerance, - metadata=metadata, - cl=cl, - attr_name=attr_name, - ) - self.roi = roi - - def get(self, **kwargs): - image = self.parent.image_data.get().data - if not isinstance(image, np.ndarray): - return -1 # -1 if no valid image is available - - if self.roi is None: - roi = (0, 0, image.shape[1], image.shape[0]) - else: - roi = self.roi - if len(image.shape) > 2: - image = np.sum(image, axis=2) # Convert to grayscale if it's a color image - return np.sum(image[roi[1] : roi[1] + roi[3], roi[0] : roi[0] + roi[2]], (0, 1)) - - class IDSCamera(PSIDeviceBase): - """ " - #--------------------------------------------------------------------------------------------------------------------------------------- + """IDS Camera class for cSAXS. - #Variables - hCam = ueye.HIDS(202) #0: first available camera; 1-254: The camera with the specified camera ID - sInfo = ueye.SENSORINFO() - cInfo = ueye.CAMINFO() - pcImageMemory = ueye.c_mem_p() - MemID = ueye.int() - rectAOI = ueye.IS_RECT() - pitch = ueye.INT() - nBitsPerPixel = ueye.INT(24) #24: bits per pixel for color mode; take 8 bits per pixel for monochrome - channels = 3 #3: channels for color mode(RGB); take 1 channel for monochrome - m_nColorMode = ueye.INT(1) # Y8/RGB16/RGB24/REG32 (1 for our color cameras) - bytes_per_pixel = int(nBitsPerPixel / 8) - - ids_cam - ... + This class inherits from PSIDeviceBase and implements the necessary methods + to interact with the IDS camera using the pyueye library. """ - USER_ACCESS = ["start_live_mode", "stop_live_mode", "set_roi", "width", "height"] + image = Cpt(PreviewSignal, name="image", ndim=2, doc="Preview signal for the camera.") + roi_signal = Cpt( + AsyncSignal, + name="roi_signal", + ndim=0, + max_size=1000, + doc="Signal for the region of interest (ROI).", + async_update={"type": "add", "max_shape": [None]}, + ) - image_data = Cpt(PreviewSignal, ndim=2, kind=Kind.omitted) - # roi_bot_left = Cpt(ROISignal, roi=(400, 525, 118, 105), kind=Kind.normal) - # roi_bot_right = Cpt(ROISignal, roi=(518, 525, 118, 105), kind=Kind.normal) - # roi_top_left = Cpt(ROISignal, roi=(400, 630, 118, 105), kind=Kind.normal) - # roi_top_right = Cpt(ROISignal, roi=(518, 630, 118, 105), kind=Kind.normal) - # roi_signal = Cpt(ROISignal, kind=Kind.normal, doc="Region of Interest signal") + USER_ACCESS = ["live_mode", "mask", "set_rect_roi", "get_last_image"] def __init__( self, - prefix="", *, name: str, - camera_ID: int, - bits_per_pixel: int, - channels: int, - m_n_colormode: int, - kind=None, - device_manager=None, + camera_id: int, + prefix: str = "", + scan_info: ScanInfo | None = None, + m_n_colormode: Literal[0, 1, 2, 3] = 1, + bits_per_pixel: Literal[8, 24] = 24, + live_mode: bool = False, **kwargs, ): + """Initialize the IDS Camera. - super().__init__( - prefix=prefix, name=name, kind=kind, device_manager=device_manager, **kwargs - ) - self.camera_ID = camera_ID - self.bits_per_pixel = bits_per_pixel - self.bytes_per_pixel = None - self.channels = channels - self._m_n_colormode_input = m_n_colormode - self.m_n_colormode = None - self.ueye = ueye - self.h_cam = None - self.s_info = None - self.data_thread = None - self.c_info = None - self.pc_image_memory = None - self.mem_id = None - self.rect_aoi = None - self.pitch = None - self.n_bits_per_pixel = None - self.width = None - self.height = None - self.thread_event = threading.Event() - self.data_thread = None - self._roi: tuple | None = None # x, y, width, height - logger.info( - f"Deprecation warning: The IDSCamera class is deprecated. Use the new IDSCameraNew class instead." + Args: + name (str): Name of the device. + camera_id (int): The ID of the camera device. + prefix (str): Prefix for the device. + scan_info (ScanInfo | None): Scan information for the device. + m_n_colormode (Literal[0, 1, 2, 3]): Color mode for the camera. + bits_per_pixel (Literal[8, 24]): Number of bits per pixel for the camera. + live_mode (bool): Whether to enable live mode for the camera. + """ + super().__init__(name=name, prefix=prefix, scan_info=scan_info, **kwargs) + self._live_mode_thread: threading.Thread | None = None + self._stop_live_mode_event: threading.Event = threading.Event() + self.cam = Camera( + camera_id=camera_id, + m_n_colormode=m_n_colormode, + bits_per_pixel=bits_per_pixel, + connect=False, ) + self._live_mode = False + self._inputs = {"live_mode": live_mode} + self._mask = np.zeros((1, 1), dtype=np.uint8) - def set_roi(self, x: int, y: int, width: int, height: int): - self._roi = (x, y, width, height) + ############## Live Mode Methods ############## - def start_backend(self): - if self.ueye is None: - raise ImportError("The pyueye library is not installed.") - self.h_cam = self.ueye.HIDS( - self.camera_ID - ) # 0: first available camera; 1-254: The camera with the specified camera ID - self.s_info = self.ueye.SENSORINFO() - self.c_info = self.ueye.CAMINFO() - self.pc_image_memory = self.ueye.c_mem_p() - self.mem_id = self.ueye.int() - self.rect_aoi = self.ueye.IS_RECT() - self.pitch = self.ueye.INT() - self.n_bits_per_pixel = self.ueye.INT( - self.bits_per_pixel - ) # 24: bits per pixel for color mode; take 8 bits per pixel for monochrome - self.m_n_colormode = self.ueye.INT( - self._m_n_colormode_input - ) # Y8/RGB16/RGB24/REG32 (1 for our color cameras) - self.bytes_per_pixel = int(self.n_bits_per_pixel / 8) + @property + def mask(self) -> np.ndarray: + """Return the current region of interest (ROI) for the camera.""" + return self._mask - # Starts the driver and establishes the connection to the camera - ret = self.ueye.is_InitCamera(self.h_cam, None) - if ret != self.ueye.IS_SUCCESS: - print("is_InitCamera ERROR") + @mask.setter + def mask(self, value: np.ndarray): + """ + Set the region of interest (ROI) for the camera. - # Reads out the data hard-coded in the non-volatile camera memory and writes it to the data structure that c_info points to - ret = self.ueye.is_GetCameraInfo(self.h_cam, self.c_info) - if ret != self.ueye.IS_SUCCESS: - print("is_GetCameraInfo ERROR") + Args: + value (np.ndarray): The mask to set as the ROI. + """ + if value.ndim != 2: + raise ValueError("ROI mask must be a 2D array.") + img_shape = (self.cam.cam.height.value, self.cam.cam.width.value) + if value.shape[0] != img_shape[0] or value.shape[1] != img_shape[1]: + raise ValueError( + f"ROI mask shape {value.shape} does not match image shape {img_shape}." + ) + self._mask = value - # You can query additional information about the sensor type used in the camera - ret = self.ueye.is_GetSensorInfo(self.h_cam, self.s_info) - if ret != self.ueye.IS_SUCCESS: - print("is_GetSensorInfo ERROR") + @property + def live_mode(self) -> bool: + """Return whether the camera is in live mode.""" + return self._live_mode - ret = self.ueye.is_ResetToDefault(self.h_cam) - if ret != self.ueye.IS_SUCCESS: - print("is_ResetToDefault ERROR") - - # Set display mode to DIB - ret = self.ueye.is_SetDisplayMode(self.h_cam, self.ueye.IS_SET_DM_DIB) - - # Set the right color mode - if ( - int.from_bytes(self.s_info.nColorMode.value, byteorder="big") - == self.ueye.IS_COLORMODE_BAYER - ): - # setup the color depth to the current windows setting - self.ueye.is_GetColorDepth(self.h_cam, self.n_bits_per_pixel, self.m_n_colormode) - bytes_per_pixel = int(self.n_bits_per_pixel / 8) - print("IS_COLORMODE_BAYER: ") - print("\tm_n_colormode: \t\t", self.m_n_colormode) - print("\tn_bits_per_pixel: \t\t", self.n_bits_per_pixel) - print("\tbytes_per_pixel: \t\t", bytes_per_pixel) - print() - - elif ( - int.from_bytes(self.s_info.nColorMode.value, byteorder="big") - == self.ueye.IS_COLORMODE_CBYCRY - ): - # for color camera models use RGB32 mode - m_n_colormode = self.ueye.IS_CM_BGRA8_PACKED - n_bits_per_pixel = self.ueye.INT(32) - bytes_per_pixel = int(self.n_bits_per_pixel / 8) - print("IS_COLORMODE_CBYCRY: ") - print("\tm_n_colormode: \t\t", m_n_colormode) - print("\tn_bits_per_pixel: \t\t", n_bits_per_pixel) - print("\tbytes_per_pixel: \t\t", bytes_per_pixel) - print() - - elif ( - int.from_bytes(self.s_info.nColorMode.value, byteorder="big") - == self.ueye.IS_COLORMODE_MONOCHROME - ): - # for color camera models use RGB32 mode - m_n_colormode = self.ueye.IS_CM_MONO8 - n_bits_per_pixel = self.ueye.INT(8) - bytes_per_pixel = int(n_bits_per_pixel / 8) - print("IS_COLORMODE_MONOCHROME: ") - print("\tm_n_colormode: \t\t", m_n_colormode) - print("\tn_bits_per_pixel: \t\t", n_bits_per_pixel) - print("\tbytes_per_pixel: \t\t", bytes_per_pixel) - print() - - else: - # for monochrome camera models use Y8 mode - m_n_colormode = self.ueye.IS_CM_MONO8 - n_bits_per_pixel = self.ueye.INT(8) - bytes_per_pixel = int(n_bits_per_pixel / 8) - print("else") - - # Can be used to set the size and position of an "area of interest"(AOI) within an image - ret = self.ueye.is_AOI( - self.h_cam, - self.ueye.IS_AOI_IMAGE_GET_AOI, - self.rect_aoi, - self.ueye.sizeof(self.rect_aoi), - ) - if ret != self.ueye.IS_SUCCESS: - print("is_AOI ERROR") - - self.width = self.rect_aoi.s32Width - self.height = self.rect_aoi.s32Height - - # Prints out some information about the camera and the sensor - print("Camera model:\t\t", self.s_info.strSensorName.decode("utf-8")) - print("Camera serial no.:\t", self.c_info.SerNo.decode("utf-8")) - print("Maximum image width:\t", self.width) - print("Maximum image height:\t", self.height) - print() - - # --------------------------------------------------------------------------------------------------------------------------------------- - - # Allocates an image memory for an image having its dimensions defined by width and height and its color depth defined by n_bits_per_pixel - ret = self.ueye.is_AllocImageMem( - self.h_cam, - self.width, - self.height, - self.n_bits_per_pixel, - self.pc_image_memory, - self.mem_id, - ) - if ret != self.ueye.IS_SUCCESS: - print("is_AllocImageMem ERROR") - else: - # Makes the specified image memory the active memory - ret = self.ueye.is_SetImageMem(self.h_cam, self.pc_image_memory, self.mem_id) - if ret != self.ueye.IS_SUCCESS: - print("is_SetImageMem ERROR") + @live_mode.setter + def live_mode(self, value: bool): + """Set the live mode for the camera.""" + if value != self._live_mode: + if self.cam._connected is False: # $ pylint: disable=protected-access + self.cam.on_connect() + self._live_mode = value + if value: + self._start_live() else: - # Set the desired color mode - ret = self.ueye.is_SetColorMode(self.h_cam, self.m_n_colormode) + self._stop_live() - # Activates the camera's live video mode (free run mode) - ret = self.ueye.is_CaptureVideo(self.h_cam, self.ueye.IS_DONT_WAIT) - if ret != self.ueye.IS_SUCCESS: - print("is_CaptureVideo ERROR") + def set_rect_roi(self, x: int, y: int, width: int, height: int): + """Set the rectangular region of interest (ROI) for the camera.""" + if x < 0 or y < 0 or width <= 0 or height <= 0: + raise ValueError("ROI coordinates and dimensions must be positive integers.") + img_shape = (self.cam.cam.height.value, self.cam.cam.width.value) + if x + width > img_shape[1] or y + height > img_shape[0]: + raise ValueError("ROI exceeds camera dimensions.") + mask = np.zeros(img_shape, dtype=np.uint8) + mask[y : y + height, x : x + width] = 1 + self.mask = mask - # Enables the queue mode for existing image memory sequences - ret = self.ueye.is_InquireImageMem( - self.h_cam, - self.pc_image_memory, - self.mem_id, - self.width, - self.height, - self.n_bits_per_pixel, - self.pitch, + def _start_live(self): + """Start the live mode for the camera.""" + if self._live_mode_thread is not None: + logger.info("Live mode thread is already running.") + return + self._stop_live_mode_event.clear() + self._live_mode_thread = threading.Thread( + target=self._live_mode_loop, args=(self._stop_live_mode_event,) ) - if ret != self.ueye.IS_SUCCESS: - print("is_InquireImageMem ERROR") + self._live_mode_thread.start() + + def _stop_live(self): + """Stop the live mode for the camera.""" + if self._live_mode_thread is None: + logger.info("Live mode thread is not running.") + return + self._stop_live_mode_event.set() + self._live_mode_thread.join(timeout=5) + if self._live_mode_thread.is_alive(): + logger.warning("Live mode thread did not stop gracefully.") else: - print("Press q to leave the programm") - # startmeasureframerate = True - # Gain = False + self._live_mode_thread = None + logger.info("Live mode stopped.") - # Start live mode of camera immediately - self.start_live_mode() + def _live_mode_loop(self, stop_event: threading.Event): + """Loop to capture images in live mode.""" + while not stop_event.is_set(): + try: + self.process_data(self.cam.get_image_data()) + except Exception as e: + logger.error(f"Error in live mode loop: {e}") + break + stop_event.wait(0.2) # 5 Hz - def _start_data_thread(self): - self.data_thread = threading.Thread(target=self._receive_data_from_camera, daemon=True) - self.data_thread.start() + def process_data(self, image: np.ndarray | None): + """Process the image data before sending it to the preview signal.""" + if image is None: + return + if len(image.shape)==3 and image.shape[2] == 3: + image = image[:,:,::-1] + self.image.put(image) - def _receive_data_from_camera(self): - while not self.thread_event.is_set(): - if self.ueye is None: - print("pyueye library not available.") + def get_last_image(self) -> np.ndarray: + """Get the last captured image from the camera.""" + image = self.image.get() + if image: + return image.data + + ############## User Interface Methods ############## + + def on_connected(self): + """Connect to the camera.""" + self.cam.on_connect() + self.live_mode = self._inputs.get("live_mode", False) + self.set_rect_roi(0, 0, self.cam.cam.width.value, self.cam.cam.height.value) + + def on_destroy(self): + """Clean up resources when the device is destroyed.""" + self.cam.on_disconnect() + super().on_destroy() + + def on_trigger(self): + """Handle the trigger event.""" + if not self.live_mode: + return + image = self.image.get() + if image is not None: + image: messages.DevicePreviewMessage + if self.mask.shape[0:2] != image.data.shape[0:2]: + logger.info( + f"ROI shape does not match image shape, skipping ROI application for device {self.name}." + ) return - # In order to display the image in an OpenCV window we need to... - # ...extract the data of our image memory - array = self.ueye.get_data( - self.pc_image_memory, - self.width, - self.height, - self.n_bits_per_pixel, - self.pitch, - copy=False, - ) - # ...reshape it in an numpy array... - frame = np.reshape(array, (self.height.value, self.width.value, self.bytes_per_pixel)) - self.image_data.put(frame) - - time.sleep(0.1) - - def wait_for_connection(self, all_signals=False, timeout=10): - if ueye is None: - raise ImportError( - "The pyueye library is not installed or doesn't provide the necessary c libs" - ) - super().wait_for_connection(all_signals, timeout) - - def start_live_mode(self): - if self.data_thread is not None: - self.stop_live_mode() - self._start_data_thread() - - def stop_live_mode(self): - """Stopping the camera live mode.""" - self.thread_event.set() - if self.data_thread is not None: - self.data_thread.join() - self.thread_event.clear() - self.data_thread = None - - ######################################## - # Beamline Specific Implementations # - ######################################## - - def on_init(self) -> None: - """ - Called when the device is initialized. - - No signals are connected at this point. If you like to - set default values on signals, please use on_connected instead. - """ - - def on_connected(self) -> None: - """ - Called after the device is connected and its signals are connected. - Default values for signals should be set here. - """ - self.start_backend() - self.start_live_mode() - - def on_stage(self) -> DeviceStatus | StatusBase | None: - """ - Called while staging the device. - - Information about the upcoming scan can be accessed from the scan_info (self.scan_info.msg) object. - """ - - def on_unstage(self) -> DeviceStatus | StatusBase | None: - """Called while unstaging the device.""" - - def on_pre_scan(self) -> DeviceStatus | StatusBase | None: - """Called right before the scan starts on all devices automatically.""" - - def on_trigger(self) -> DeviceStatus | StatusBase | None: - """Called when the device is triggered.""" - - def on_complete(self) -> DeviceStatus | StatusBase | None: - """Called to inquire if a device has completed a scans.""" - - def on_kickoff(self) -> DeviceStatus | StatusBase | None: - """Called to kickoff a device for a fly scan. Has to be called explicitly.""" - - def on_stop(self) -> None: - """Called when the device is stopped.""" - - def on_destroy(self) -> None: - """Called when the device is destroyed. Cleanup resources here.""" - self.stop_live_mode() + if len(image.data.shape) == 3: + # If the image has multiple channels, apply the mask to each channel + data = image.data * self.mask[:, :, np.newaxis] # Apply mask to the image data + n_channels = 3 + else: + data = image.data * self.mask + n_channels = 1 + self.roi_signal.put(np.sum(data) / (np.sum(self.mask) * n_channels)) if __name__ == "__main__": - # Example usage - camera = IDSCamera(name="camera", camera_ID=201, bits_per_pixel=24, channels=3, m_n_colormode=1) - camera.wait_for_connection() - - camera.on_destroy() + # Example usage of the IDSCamera class + camera = IDSCamera(name="TestCamera", camera_id=201, live_mode=False) + print(f"Camera {camera.name} initialized with ID {camera.cam.camera_id}.") diff --git a/csaxs_bec/devices/ids_cameras/ids_camera_new.py b/csaxs_bec/devices/ids_cameras/ids_camera_new.py deleted file mode 100644 index 3111d61..0000000 --- a/csaxs_bec/devices/ids_cameras/ids_camera_new.py +++ /dev/null @@ -1,218 +0,0 @@ -"""IDS Camera class for cSAXS IDS cameras.""" - -from __future__ import annotations - -import threading -import time -from typing import TYPE_CHECKING, Literal, Tuple, TypedDict - -import numpy as np -from bec_lib import messages -from bec_lib.logger import bec_logger -from ophyd import Component as Cpt -from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase -from ophyd_devices.utils.bec_signals import AsyncSignal, PreviewSignal - -from csaxs_bec.devices.ids_cameras.base_integration.camera import Camera - -if TYPE_CHECKING: - from bec_lib.devicemanager import ScanInfo - from pydantic import ValidationInfo - - -logger = bec_logger.logger - - -class IDSCamera(PSIDeviceBase): - """IDS Camera class for cSAXS. - - This class inherits from PSIDeviceBase and implements the necessary methods - to interact with the IDS camera using the pyueye library. - """ - - image = Cpt(PreviewSignal, name="image", ndim=2, doc="Preview signal for the camera.") - roi_signal = Cpt( - AsyncSignal, - name="roi_signal", - ndim=0, - max_size=1000, - doc="Signal for the region of interest (ROI).", - async_update={"type": "add", "max_shape": [None]}, - ) - - USER_ACCESS = ["live_mode", "mask", "set_rect_roi", "get_last_image"] - - def __init__( - self, - *, - name: str, - camera_id: int, - prefix: str = "", - scan_info: ScanInfo | None = None, - m_n_colormode: Literal[0, 1, 2, 3] = 1, - bits_per_pixel: Literal[8, 24] = 24, - live_mode: bool = False, - **kwargs, - ): - """Initialize the IDS Camera. - - Args: - name (str): Name of the device. - camera_id (int): The ID of the camera device. - prefix (str): Prefix for the device. - scan_info (ScanInfo | None): Scan information for the device. - m_n_colormode (Literal[0, 1, 2, 3]): Color mode for the camera. - bits_per_pixel (Literal[8, 24]): Number of bits per pixel for the camera. - live_mode (bool): Whether to enable live mode for the camera. - """ - super().__init__(name=name, prefix=prefix, scan_info=scan_info, **kwargs) - self._live_mode_thread: threading.Thread | None = None - self._stop_live_mode_event: threading.Event = threading.Event() - self.cam = Camera( - camera_id=camera_id, - m_n_colormode=m_n_colormode, - bits_per_pixel=bits_per_pixel, - connect=False, - ) - self._live_mode = False - self._inputs = {"live_mode": live_mode} - self._mask = np.zeros((1, 1), dtype=np.uint8) - - ############## Live Mode Methods ############## - - @property - def mask(self) -> np.ndarray: - """Return the current region of interest (ROI) for the camera.""" - return self._mask - - @mask.setter - def mask(self, value: np.ndarray): - """ - Set the region of interest (ROI) for the camera. - - Args: - value (np.ndarray): The mask to set as the ROI. - """ - if value.ndim != 2: - raise ValueError("ROI mask must be a 2D array.") - img_shape = (self.cam.cam.height.value, self.cam.cam.width.value) - if value.shape[0] != img_shape[0] or value.shape[1] != img_shape[1]: - raise ValueError( - f"ROI mask shape {value.shape} does not match image shape {img_shape}." - ) - self._mask = value - - @property - def live_mode(self) -> bool: - """Return whether the camera is in live mode.""" - return self._live_mode - - @live_mode.setter - def live_mode(self, value: bool): - """Set the live mode for the camera.""" - if value != self._live_mode: - if self.cam._connected is False: # $ pylint: disable=protected-access - self.cam.on_connect() - self._live_mode = value - if value: - self._start_live() - else: - self._stop_live() - - def set_rect_roi(self, x: int, y: int, width: int, height: int): - """Set the rectangular region of interest (ROI) for the camera.""" - if x < 0 or y < 0 or width <= 0 or height <= 0: - raise ValueError("ROI coordinates and dimensions must be positive integers.") - img_shape = (self.cam.cam.height.value, self.cam.cam.width.value) - if x + width > img_shape[1] or y + height > img_shape[0]: - raise ValueError("ROI exceeds camera dimensions.") - mask = np.zeros(img_shape, dtype=np.uint8) - mask[y : y + height, x : x + width] = 1 - self.mask = mask - - def _start_live(self): - """Start the live mode for the camera.""" - if self._live_mode_thread is not None: - logger.info("Live mode thread is already running.") - return - self._stop_live_mode_event.clear() - self._live_mode_thread = threading.Thread( - target=self._live_mode_loop, args=(self._stop_live_mode_event,) - ) - self._live_mode_thread.start() - - def _stop_live(self): - """Stop the live mode for the camera.""" - if self._live_mode_thread is None: - logger.info("Live mode thread is not running.") - return - self._stop_live_mode_event.set() - self._live_mode_thread.join(timeout=5) - if self._live_mode_thread.is_alive(): - logger.warning("Live mode thread did not stop gracefully.") - else: - self._live_mode_thread = None - logger.info("Live mode stopped.") - - def _live_mode_loop(self, stop_event: threading.Event): - """Loop to capture images in live mode.""" - while not stop_event.is_set(): - try: - self.process_data(self.cam.get_image_data()) - except Exception as e: - logger.error(f"Error in live mode loop: {e}") - break - stop_event.wait(0.2) # 5 Hz - - def process_data(self, image: np.ndarray | None): - """Process the image data before sending it to the preview signal.""" - if image is None: - return - self.image.put(image) - - def get_last_image(self) -> np.ndarray: - """Get the last captured image from the camera.""" - image = self.image.get() - if image: - return image.data - - ############## User Interface Methods ############## - - def on_connected(self): - """Connect to the camera.""" - self.cam.on_connect() - self.live_mode = self._inputs.get("live_mode", False) - self.set_rect_roi(0, 0, self.cam.cam.width.value, self.cam.cam.height.value) - - def on_destroy(self): - """Clean up resources when the device is destroyed.""" - self.cam.on_disconnect() - super().on_destroy() - - def on_trigger(self): - """Handle the trigger event.""" - if not self.live_mode: - return - image = self.image.get() - if image is not None: - image: messages.DevicePreviewMessage - if self.mask.shape[0:2] != image.data.shape[0:2]: - logger.info( - f"ROI shape does not match image shape, skipping ROI application for device {self.name}." - ) - return - - if len(image.data.shape) == 3: - # If the image has multiple channels, apply the mask to each channel - data = image.data * self.mask[:, :, np.newaxis] # Apply mask to the image data - n_channels = 3 - else: - data = image.data * self.mask - n_channels = 1 - self.roi_signal.put(np.sum(data) / (np.sum(self.mask) * n_channels)) - - -if __name__ == "__main__": - # Example usage of the IDSCamera class - camera = IDSCamera(name="TestCamera", camera_id=201, live_mode=False) - print(f"Camera {camera.name} initialized with ID {camera.cam.camera_id}.") diff --git a/tests/tests_devices/test_ids_camera.py b/tests/tests_devices/test_ids_camera.py index a1b51dc..a4c4f8d 100644 --- a/tests/tests_devices/test_ids_camera.py +++ b/tests/tests_devices/test_ids_camera.py @@ -5,7 +5,7 @@ from unittest import mock import numpy as np import pytest -from csaxs_bec.devices.ids_cameras.ids_camera_new import IDSCamera +from csaxs_bec.devices.ids_cameras.ids_camera import IDSCamera @pytest.fixture(scope="function") -- 2.49.1 From 39d2c9724769b471478d9ae65aa9a36ae3dda746 Mon Sep 17 00:00:00 2001 From: x01dc Date: Wed, 22 Oct 2025 15:54:43 +0200 Subject: [PATCH 5/7] fix: xray eye align script changes --- .../plugins/flomni/x_ray_eye_align.py | 39 +++++++++++-------- 1 file changed, 22 insertions(+), 17 deletions(-) diff --git a/csaxs_bec/bec_ipython_client/plugins/flomni/x_ray_eye_align.py b/csaxs_bec/bec_ipython_client/plugins/flomni/x_ray_eye_align.py index fc7b954..f02bf19 100644 --- a/csaxs_bec/bec_ipython_client/plugins/flomni/x_ray_eye_align.py +++ b/csaxs_bec/bec_ipython_client/plugins/flomni/x_ray_eye_align.py @@ -7,7 +7,7 @@ from typing import TYPE_CHECKING from bec_lib import bec_logger -from csaxs_bec.bec_ipython_client.plugins.cSAXS import epics_get, epics_put, fshopen +from csaxs_bec.bec_ipython_client.plugins.cSAXS import epics_get, epics_put, fshopen, fshclose logger = bec_logger.logger # import builtins to avoid linter errors @@ -22,6 +22,7 @@ if TYPE_CHECKING: class XrayEyeAlign: # pixel calibration, multiply to get mm + labview=False PIXEL_CALIBRATION = 0.1 / 113 # .2 with binning def __init__(self, client, flomni: Flomni) -> None: @@ -41,26 +42,29 @@ class XrayEyeAlign: epics_put("XOMNYI-XEYE-SAVFRAME:0", 1) def update_frame(self): - epics_put("XOMNYI-XEYE-ACQDONE:0", 0) + if self.labview: + epics_put("XOMNYI-XEYE-ACQDONE:0", 0) # start live epics_put("XOMNYI-XEYE-ACQ:0", 1) - # wait for start live - while epics_get("XOMNYI-XEYE-ACQDONE:0") == 0: - time.sleep(0.5) - print("waiting for live view to start...") + if self.labview: + # wait for start live + while epics_get("XOMNYI-XEYE-ACQDONE:0") == 0: + time.sleep(0.5) + print("waiting for live view to start...") fshopen() - epics_put("XOMNYI-XEYE-ACQDONE:0", 0) + if self.labview: + epics_put("XOMNYI-XEYE-ACQDONE:0", 0) - while epics_get("XOMNYI-XEYE-ACQDONE:0") == 0: - print("waiting for new frame...") - time.sleep(0.5) + while epics_get("XOMNYI-XEYE-ACQDONE:0") == 0: + print("waiting for new frame...") + time.sleep(0.5) time.sleep(0.5) # stop live view epics_put("XOMNYI-XEYE-ACQ:0", 0) - time.sleep(1) - # fshclose + time.sleep(0.1) + fshclose() print("got new frame") def tomo_rotate(self, val: float): @@ -152,11 +156,12 @@ class XrayEyeAlign: self.flomni.feedback_disable() umv(dev.fsamx, fsamx_in - 0.25) - self.update_frame() - epics_put("XOMNYI-XEYE-RECBG:0", 1) - while epics_get("XOMNYI-XEYE-RECBG:0") == 1: - time.sleep(0.5) - print("waiting for background frame...") + if self.labview: + self.update_frame() + epics_put("XOMNYI-XEYE-RECBG:0", 1) + while epics_get("XOMNYI-XEYE-RECBG:0") == 1: + time.sleep(0.5) + print("waiting for background frame...") umv(dev.fsamx, fsamx_in) time.sleep(0.5) -- 2.49.1 From 94d984b8a29f7f2ac6c086461d7f02e702361565 Mon Sep 17 00:00:00 2001 From: wyzula-jan Date: Thu, 23 Oct 2025 11:11:22 +0200 Subject: [PATCH 6/7] fix(camera): BGR to RGB logic moved to Camera --- .../ids_cameras/base_integration/camera.py | 28 +++++++++++-------- csaxs_bec/devices/ids_cameras/ids_camera.py | 2 -- 2 files changed, 16 insertions(+), 14 deletions(-) diff --git a/csaxs_bec/devices/ids_cameras/base_integration/camera.py b/csaxs_bec/devices/ids_cameras/base_integration/camera.py index bdd27c2..d0cb22b 100644 --- a/csaxs_bec/devices/ids_cameras/base_integration/camera.py +++ b/csaxs_bec/devices/ids_cameras/base_integration/camera.py @@ -66,8 +66,8 @@ class IDSCameraObject: check_error(ueye.is_SetDisplayMode(self.h_cam, ueye.IS_SET_DM_DIB), "IDSCameraObject") if ( - int.from_bytes(self.s_info.nColorMode.value, byteorder="big") - == self.ueye.IS_COLORMODE_BAYER + int.from_bytes(self.s_info.nColorMode.value, byteorder="big") + == self.ueye.IS_COLORMODE_BAYER ): logger.info("Bayer color mode detected.") # setup the color depth to the current windows setting @@ -76,16 +76,16 @@ class IDSCameraObject: ) # TODO This raises an error - maybe check the m_n_colormode value self.bytes_per_pixel = int(self.n_bits_per_pixel / 8) elif ( - int.from_bytes(self.s_info.nColorMode.value, byteorder="big") - == self.ueye.IS_COLORMODE_CBYCRY + int.from_bytes(self.s_info.nColorMode.value, byteorder="big") + == self.ueye.IS_COLORMODE_CBYCRY ): # for color camera models use RGB32 mode self.m_n_colormode = self.ueye.IS_CM_BGRA8_PACKED self.n_bits_per_pixel = self.ueye.INT(32) self.bytes_per_pixel = int(self.n_bits_per_pixel / 8) elif ( - int.from_bytes(self.s_info.nColorMode.value, byteorder="big") - == self.ueye.IS_COLORMODE_MONOCHROME + int.from_bytes(self.s_info.nColorMode.value, byteorder="big") + == self.ueye.IS_COLORMODE_MONOCHROME ): # for color camera models use RGB32 mode self.m_n_colormode = self.ueye.IS_CM_MONO8 @@ -159,11 +159,11 @@ class Camera: """ def __init__( - self, - camera_id: int, - m_n_colormode: Literal[0, 1, 2, 3] = 1, - bits_per_pixel: int = 24, - connect: bool = True, + self, + camera_id: int, + m_n_colormode: Literal[0, 1, 2, 3] = 1, + bits_per_pixel: int = 24, + connect: bool = True, ): self.ueye = ueye self.camera_id = camera_id @@ -263,9 +263,13 @@ class Camera: if array is None: logger.error("Failed to get image data from the camera.") return None - return np.reshape( + img = np.reshape( array, (self.cam.height.value, self.cam.width.value, self.cam.bytes_per_pixel) ) + # If RGB image (H, W, 3), reshuffle channels from BGR → RGB + if img.ndim == 3 and img.shape[2] == 3: + img = img[:, :, ::-1] + return img if __name__ == "__main__": diff --git a/csaxs_bec/devices/ids_cameras/ids_camera.py b/csaxs_bec/devices/ids_cameras/ids_camera.py index 52c857b..3111d61 100644 --- a/csaxs_bec/devices/ids_cameras/ids_camera.py +++ b/csaxs_bec/devices/ids_cameras/ids_camera.py @@ -168,8 +168,6 @@ class IDSCamera(PSIDeviceBase): """Process the image data before sending it to the preview signal.""" if image is None: return - if len(image.shape)==3 and image.shape[2] == 3: - image = image[:,:,::-1] self.image.put(image) def get_last_image(self) -> np.ndarray: -- 2.49.1 From c8c71d466c336b21e9db63f6805470d36639af73 Mon Sep 17 00:00:00 2001 From: wyzula-jan Date: Thu, 23 Oct 2025 11:17:29 +0200 Subject: [PATCH 7/7] refactor(xray_gui): minor cleanup --- .../bec_widgets/widgets/xray_eye/x_ray_eye.py | 49 ++++++++++--------- 1 file changed, 26 insertions(+), 23 deletions(-) diff --git a/csaxs_bec/bec_widgets/widgets/xray_eye/x_ray_eye.py b/csaxs_bec/bec_widgets/widgets/xray_eye/x_ray_eye.py index 406cc63..3b66431 100644 --- a/csaxs_bec/bec_widgets/widgets/xray_eye/x_ray_eye.py +++ b/csaxs_bec/bec_widgets/widgets/xray_eye/x_ray_eye.py @@ -24,7 +24,7 @@ from qtpy.QtWidgets import ( ) logger = bec_logger.logger -CAMERA = ("cam_xeye_rgb", "image") #TODO here put correct camera +CAMERA = ("cam_xeye_rgb", "image") class XRayEye2DControl(BECWidget, QWidget): @@ -259,15 +259,6 @@ class XRayEye(BECWidget, QWidget): ################################################################################ # Properties ported from the original OmnyAlignment, can be adjusted as needed ################################################################################ - @SafeProperty(bool) - def enable_live_view(self): - """Get or set the live view enabled state.""" - return self.live_preview_toggle.checked - - @enable_live_view.setter - def enable_live_view(self, enable: bool): - self.live_preview_toggle.checked = enable - @SafeProperty(str) def user_message(self): return self.message_line_edit.text() @@ -326,20 +317,23 @@ class XRayEye(BECWidget, QWidget): @SafeSlot(bool, bool) 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) - @SafeSlot(str) - def set_message(self, msg: str): - self.message_line_edit.setText(msg) - - @SafeSlot(str) - def set_sample_name(self, msg: str): - self.sample_name_line_edit.setText(msg) - @SafeSlot(int) def enable_submit_button(self, enable: int): - """If -1 disable else enable""" + """ + Enable/disable submit button. + Args: + enable(int): -1 disable else enable + """ if enable == -1: self.submit_button.setEnabled(False) else: @@ -347,11 +341,19 @@ class XRayEye(BECWidget, QWidget): @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(dict, dict) def device_updates(self, data: dict, meta: dict): + """ + Slot to handle device updates from omny_xray_gui device. + + Args: + data(dict): data from device + meta(dict): metadata from device + """ signals = data.get('signals') enable_live_preview = signals.get("omny_xray_gui_update_frame_acq").get('value') @@ -362,11 +364,11 @@ class XRayEye(BECWidget, QWidget): # Signals from epics gui device # send message - send_message = signals.get("omny_xray_gui_send_message").get('value') - self.set_message(send_message) + user_message = signals.get("omny_xray_gui_send_message").get('value') + self.user_message = user_message # sample name sample_message = signals.get("omny_xray_gui_sample_name").get('value') - self.set_sample_name(sample_message) + self.sample_name = sample_message # enable frame acquisition update_frame_acq = signals.get("omny_xray_gui_update_frame_acq").get('value') self.on_live_view_enabled(bool(update_frame_acq)) @@ -394,7 +396,7 @@ class XRayEye(BECWidget, QWidget): logger.warning("Unsupported ROI type for submit action.") return - print(f"current roi: {roi_center_x},{roi_center_y}, {roi_width},{roi_height}") + print(f"current roi: x:{roi_center_x}, y:{roi_center_y}, w:{roi_width},h:{roi_height}") #TODO remove when will be not needed for debugging # submit roi coordinates step = int(self.dev.omny_xray_gui.step.read().get("omny_xray_gui_step").get('value')) @@ -405,6 +407,7 @@ class XRayEye(BECWidget, QWidget): self.dev.omny_xray_gui.submit.set(1) def cleanup(self): + """Cleanup connections on widget close -> disconnect slots and stop live mode of camera.""" self.bec_dispatcher.disconnect_slot(self.device_updates, MessageEndpoints.device_readback("omny_xray_gui")) getattr(self.dev,CAMERA).live_mode = False super().cleanup() -- 2.49.1