feat(ids-cameras): add IDSCameraSettings widget for multi-camera exposure/gain control
OMNY_XRayEye's exposure/pixel-clock controls are hardcoded to a single camera (cam_xeye). Several beamline configs define more than one IDSCamera at once (e.g. ptycho_omny.yaml's cam200..cam203), none of which were reachable from that widget. Adds a standalone IDSCameraSettings panel: a camera-selection dropdown (filtered by deviceClass, extensible to other camera classes later) plus auto-exposure/exposure-time/pixel-clock controls that re-subscribe to the selected camera's device_read_configuration on every switch, seeded immediately from the retained redis value. Scaffolded via bw-generate-cli --target csaxs_bec (client.py, designer_plugins.py). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TUimPoyFQRvxM6R3njvuVj
This commit is contained in:
x12saandClaude Sonnet 5 committed
1 parent
d11cbbafa9
commit
e1e13cdf17
7 files changed
+611
No files matched your search
@@ -14,6 +14,7 @@ logger = bec_logger.logger
|
||||
|
||||
_Widgets = {
|
||||
"DataViewer": "DataViewer",
|
||||
"IDSCameraSettings": "IDSCameraSettings",
|
||||
"OMNY_SampleStorage": "OMNY_SampleStorage",
|
||||
"OMNY_TomoParams": "OMNY_TomoParams",
|
||||
"OMNY_XRayEye": "OMNY_XRayEye",
|
||||
@@ -47,6 +48,24 @@ class DataViewer(RPCBase):
|
||||
"""
|
||||
|
||||
|
||||
class IDSCameraSettings(RPCBase):
|
||||
"""Pick a configured IDS camera and adjust its exposure time / pixel clock."""
|
||||
|
||||
_IMPORT_MODULE = "csaxs_bec.bec_widgets.widgets.ids_camera_settings.ids_camera_settings"
|
||||
|
||||
@rpc_call
|
||||
def selected_camera():
|
||||
"""
|
||||
Intermediate wrapper used so that the user can optionally chain .setter(...).
|
||||
"""
|
||||
|
||||
@rpc_call
|
||||
def set_camera(self, name: "str"):
|
||||
"""
|
||||
Select a camera by device name, as if chosen from the dropdown.
|
||||
"""
|
||||
|
||||
|
||||
class OMNY_SampleStorage(RPCBase):
|
||||
"""View and correct the FlOMNI sample-storage records."""
|
||||
|
||||
|
||||
@@ -6,6 +6,10 @@ from __future__ import annotations
|
||||
|
||||
designer_plugins = {
|
||||
"DataViewer": ("csaxs_bec.bec_widgets.widgets.data_viewer.data_viewer", "DataViewer"),
|
||||
"IDSCameraSettings": (
|
||||
"csaxs_bec.bec_widgets.widgets.ids_camera_settings.ids_camera_settings",
|
||||
"IDSCameraSettings",
|
||||
),
|
||||
"OMNY_SampleStorage": (
|
||||
"csaxs_bec.bec_widgets.widgets.sample_storage.sample_storage",
|
||||
"OMNY_SampleStorage",
|
||||
@@ -22,6 +26,7 @@ designer_plugins = {
|
||||
|
||||
widget_icons = {
|
||||
"DataViewer": "find_in_page",
|
||||
"IDSCameraSettings": "photo_camera",
|
||||
"OMNY_SampleStorage": "widgets",
|
||||
"OMNY_TomoParams": "widgets",
|
||||
"OMNY_XRayEye": "widgets",
|
||||
|
||||
@@ -0,0 +1,3 @@
|
||||
from csaxs_bec.bec_widgets.widgets.ids_camera_settings.ids_camera_settings import IDSCameraSettings
|
||||
|
||||
__all__ = ["IDSCameraSettings"]
|
||||
@@ -0,0 +1,511 @@
|
||||
"""IDS camera settings widget for cSAXS -- pick a configured IDS camera from
|
||||
a dropdown and adjust its exposure time / pixel clock.
|
||||
|
||||
The exposure/pixel-clock control logic mirrors OMNY_XRayEye
|
||||
(bec_widgets/widgets/xray_eye/x_ray_eye.py), which implements the same
|
||||
controls but hardcoded to a single camera (module-level CAMERA constant).
|
||||
This widget generalizes that to whichever camera is selected, for beamlines
|
||||
that configure more than one IDS camera at once (e.g. ptycho_omny.yaml's
|
||||
cam200..cam203).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
|
||||
from bec_lib import bec_logger
|
||||
from bec_lib.endpoints import MessageEndpoints
|
||||
from bec_widgets import BECWidget, SafeProperty, SafeSlot
|
||||
from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import (
|
||||
DeviceComboBox,
|
||||
)
|
||||
from bec_widgets.widgets.utility.toggle.toggle import ToggleSwitch
|
||||
from qtpy.QtCore import Qt, QTimer
|
||||
from qtpy.QtWidgets import QFrame, QGridLayout, QLabel, QSizePolicy, QSlider, QVBoxLayout, QWidget
|
||||
|
||||
logger = bec_logger.logger
|
||||
|
||||
|
||||
class IDSCameraComboBox(DeviceComboBox):
|
||||
"""Dropdown listing configured IDS cameras.
|
||||
|
||||
DeviceComboBox's device_filter only distinguishes bec_lib.device base
|
||||
classes (Device/Positioner/Signal/ComputedSignal) and can't express
|
||||
"exactly IDSCamera", so this overrides update_devices_from_filters() to
|
||||
filter by each device's deviceClass config instead -- the same technique
|
||||
device_table.py uses to filter/display device classes client-side.
|
||||
|
||||
To support another camera class later (once it grows the same
|
||||
Kind.config exposure/pixel-clock signals this widget depends on), add
|
||||
its bare class name to CAMERA_DEVICE_CLASSES -- no other change should
|
||||
be required here.
|
||||
"""
|
||||
|
||||
PLUGIN = False
|
||||
RPC = False
|
||||
|
||||
CAMERA_DEVICE_CLASSES: tuple[str, ...] = ("IDSCamera",)
|
||||
|
||||
@staticmethod
|
||||
def _device_class_name(device) -> str:
|
||||
"""Bare class name from a device's deviceClass config, e.g.
|
||||
"csaxs_bec.devices.ids_cameras.ids_camera.IDSCamera" -> "IDSCamera"."""
|
||||
device_class = (getattr(device, "_config", None) or {}).get("deviceClass") or ""
|
||||
return device_class.rsplit(".", 1)[-1] if device_class else ""
|
||||
|
||||
@SafeSlot()
|
||||
def update_devices_from_filters(self):
|
||||
if not self.apply_filter:
|
||||
return
|
||||
self.devices = sorted(
|
||||
device.name
|
||||
for device in self.dev.enabled_devices
|
||||
if self._device_class_name(device) in self.CAMERA_DEVICE_CLASSES
|
||||
)
|
||||
|
||||
|
||||
class IDSCameraSettings(BECWidget, QWidget):
|
||||
"""Pick a configured IDS camera and adjust its exposure time / pixel clock."""
|
||||
|
||||
ICON_NAME = "photo_camera"
|
||||
PLUGIN = True
|
||||
USER_ACCESS = ["selected_camera", "set_camera"]
|
||||
|
||||
# exposure_time_slider works in tenths of a ms internally (QSlider is
|
||||
# int-only); divide by this to get ms. Same idiom as OMNY_XRayEye.
|
||||
_EXPOSURE_SLIDER_SCALE = 10
|
||||
|
||||
def __init__(self, parent=None, **kwargs):
|
||||
super().__init__(parent=parent, **kwargs)
|
||||
self.get_bec_shortcuts()
|
||||
|
||||
self._camera_name: str | None = None
|
||||
self._auto_exposure_enabled = True
|
||||
# pixel_clock_slider is index-based over this list (re-fetched once
|
||||
# per camera, see _init_pixel_clock_options()) rather than a raw MHz
|
||||
# range -- many uEye sensors only accept a short discrete list of
|
||||
# pixel clocks, not every value in [min, max] (see OMNY_XRayEye).
|
||||
self._pixel_clock_options: list[int] = []
|
||||
self._last_pixel_clock_mhz: int | None = None
|
||||
|
||||
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_busy)
|
||||
|
||||
self._init_ui()
|
||||
self._make_connections()
|
||||
self._reset_controls_to_placeholder()
|
||||
|
||||
self.bec_dispatcher.connect_slot(
|
||||
self.on_queue_status_update, MessageEndpoints.scan_queue_status()
|
||||
)
|
||||
QTimer.singleShot(0, self._init_queue_status)
|
||||
|
||||
def _init_ui(self):
|
||||
layout = QVBoxLayout(self)
|
||||
|
||||
camera_row = QGridLayout()
|
||||
camera_row.setColumnStretch(1, 1)
|
||||
self.camera_label = QLabel("Camera", parent=self)
|
||||
self.camera_combo = IDSCameraComboBox(parent=self)
|
||||
self.camera_combo.set_first_element_as_empty = True
|
||||
camera_row.addWidget(self.camera_label, 0, 0)
|
||||
camera_row.addWidget(self.camera_combo, 0, 1)
|
||||
layout.addLayout(camera_row)
|
||||
|
||||
self.status_label = QLabel(parent=self)
|
||||
layout.addWidget(self.status_label)
|
||||
|
||||
layout.addWidget(self._create_separator())
|
||||
|
||||
_right_vcenter = Qt.AlignmentFlag.AlignRight | Qt.AlignmentFlag.AlignVCenter
|
||||
|
||||
self.exposure_grid_widget = QWidget(parent=self)
|
||||
exposure_grid = QGridLayout(self.exposure_grid_widget)
|
||||
exposure_grid.setHorizontalSpacing(8)
|
||||
exposure_grid.setVerticalSpacing(4)
|
||||
exposure_grid.setColumnStretch(0, 1)
|
||||
|
||||
# No "Auto gain" control here -- see OMNY_XRayEye, which dropped it
|
||||
# after HW testing found continuous auto-gain isn't useful once
|
||||
# correctly exposed. IDSCamera pulses it on then off internally,
|
||||
# once, at connect; there is no persistent UI for it.
|
||||
self.auto_exposure_label = QLabel("Auto exposure", parent=self)
|
||||
self.auto_exposure_toggle = ToggleSwitch(parent=self)
|
||||
self.auto_exposure_toggle.checked = True
|
||||
self.auto_exposure_toggle.enabled.connect(self.auto_exposure_enabled_changed)
|
||||
exposure_grid.addWidget(self.auto_exposure_label, 0, 1, _right_vcenter)
|
||||
exposure_grid.addWidget(self.auto_exposure_toggle, 0, 2, Qt.AlignmentFlag.AlignVCenter)
|
||||
|
||||
# QSlider only takes ints, so exposure time is tracked in tenths of a
|
||||
# ms internally (_EXPOSURE_SLIDER_SCALE) for one decimal of
|
||||
# resolution on the display label.
|
||||
self.exposure_time_label = QLabel("Exposure time", parent=self)
|
||||
self.exposure_time_slider = QSlider(Qt.Orientation.Horizontal, parent=self)
|
||||
self.exposure_time_slider.setRange(1, 10000) # placeholder; reseeded per camera
|
||||
self.exposure_time_value_label = QLabel("-- ms", parent=self)
|
||||
self.exposure_time_value_label.setMinimumWidth(60)
|
||||
self.exposure_time_slider.valueChanged.connect(self._update_exposure_time_value_label)
|
||||
# sliderReleased (fires once, on mouse-up), not valueChanged (fires on
|
||||
# every tick while dragging) -- don't hammer the device.
|
||||
self.exposure_time_slider.sliderReleased.connect(self.exposure_time_submitted)
|
||||
exposure_grid.addWidget(self.exposure_time_label, 1, 1, _right_vcenter)
|
||||
exposure_grid.addWidget(self.exposure_time_slider, 1, 2, 1, 2)
|
||||
exposure_grid.addWidget(self.exposure_time_value_label, 1, 4, Qt.AlignmentFlag.AlignVCenter)
|
||||
|
||||
# Pixel clock (MHz): lowering it raises the max achievable exposure
|
||||
# time, at the cost of frame rate. Many uEye sensors only accept a
|
||||
# short discrete list of pixel clocks, so this slider's *position* is
|
||||
# an index into self._pixel_clock_options, not a raw MHz value.
|
||||
self.pixel_clock_label = QLabel("Pixel clock", parent=self)
|
||||
self.pixel_clock_slider = QSlider(Qt.Orientation.Horizontal, parent=self)
|
||||
self.pixel_clock_slider.setRange(0, 0) # placeholder; see _set_pixel_clock_options()
|
||||
self.pixel_clock_value_label = QLabel("-- MHz", parent=self)
|
||||
self.pixel_clock_value_label.setMinimumWidth(60)
|
||||
self.pixel_clock_slider.valueChanged.connect(self._update_pixel_clock_value_label)
|
||||
self.pixel_clock_slider.sliderReleased.connect(self.pixel_clock_submitted)
|
||||
exposure_grid.addWidget(self.pixel_clock_label, 2, 1, _right_vcenter)
|
||||
exposure_grid.addWidget(self.pixel_clock_slider, 2, 2, 1, 2)
|
||||
exposure_grid.addWidget(self.pixel_clock_value_label, 2, 4, Qt.AlignmentFlag.AlignVCenter)
|
||||
|
||||
layout.addWidget(self.exposure_grid_widget)
|
||||
layout.addStretch()
|
||||
|
||||
self.setSizePolicy(QSizePolicy.Policy.Preferred, QSizePolicy.Policy.Fixed)
|
||||
|
||||
def _make_connections(self):
|
||||
self.camera_combo.device_selected.connect(self._on_camera_selected)
|
||||
self.camera_combo.device_reset.connect(self._on_camera_reset)
|
||||
|
||||
def _create_separator(self):
|
||||
sep = QFrame(parent=self)
|
||||
sep.setFrameShape(QFrame.Shape.HLine)
|
||||
sep.setFrameShadow(QFrame.Shadow.Sunken)
|
||||
sep.setLineWidth(1)
|
||||
return sep
|
||||
|
||||
################################################################################
|
||||
# Camera selection
|
||||
################################################################################
|
||||
|
||||
@SafeSlot(str)
|
||||
def _on_camera_selected(self, camera_name: str):
|
||||
if not camera_name or camera_name == self._camera_name:
|
||||
return
|
||||
self._switch_camera(camera_name)
|
||||
|
||||
@SafeSlot()
|
||||
def _on_camera_reset(self):
|
||||
if self._camera_name is not None:
|
||||
self._switch_camera(None)
|
||||
|
||||
@SafeProperty(str)
|
||||
def selected_camera(self) -> str:
|
||||
"""Name of the currently selected IDS camera, or an empty string if none."""
|
||||
return self._camera_name or ""
|
||||
|
||||
@SafeSlot(str)
|
||||
def set_camera(self, name: str):
|
||||
"""Select a camera by device name, as if chosen from the dropdown."""
|
||||
self.camera_combo.set_device(name)
|
||||
|
||||
def _switch_camera(self, camera_name: str | None):
|
||||
if self._camera_name is not None:
|
||||
self.bec_dispatcher.disconnect_slot(
|
||||
self.getting_camera_status,
|
||||
MessageEndpoints.device_read_configuration(self._camera_name),
|
||||
)
|
||||
self._camera_name = camera_name
|
||||
self._reset_controls_to_placeholder()
|
||||
|
||||
if camera_name is None:
|
||||
return
|
||||
|
||||
self.bec_dispatcher.connect_slot(
|
||||
self.getting_camera_status, MessageEndpoints.device_read_configuration(camera_name)
|
||||
)
|
||||
# device_read_configuration is retained in redis (set_and_publish), so
|
||||
# seed immediately from the last known value instead of waiting for
|
||||
# the camera to spontaneously republish -- connect_slot alone only
|
||||
# delivers messages published *after* it connects.
|
||||
try:
|
||||
msg = self.client.connector.get(
|
||||
MessageEndpoints.device_read_configuration(camera_name)
|
||||
)
|
||||
except Exception as exc:
|
||||
logger.warning(f"Failed to fetch initial config for {camera_name}: {exc}")
|
||||
msg = None
|
||||
if msg is not None:
|
||||
self.getting_camera_status(msg.content, msg.metadata)
|
||||
|
||||
self._pixel_clock_options = []
|
||||
self._last_pixel_clock_mhz = None
|
||||
QTimer.singleShot(0, self._init_pixel_clock_options)
|
||||
|
||||
def _reset_controls_to_placeholder(self):
|
||||
self._auto_exposure_enabled = True
|
||||
self.auto_exposure_toggle.blockSignals(True)
|
||||
self.auto_exposure_toggle.checked = True
|
||||
self.auto_exposure_toggle.blockSignals(False)
|
||||
|
||||
self.exposure_time_value_label.setText("-- ms")
|
||||
|
||||
self._pixel_clock_options = []
|
||||
self._last_pixel_clock_mhz = None
|
||||
self.pixel_clock_slider.blockSignals(True)
|
||||
self.pixel_clock_slider.setRange(0, 0)
|
||||
self.pixel_clock_slider.blockSignals(False)
|
||||
self.pixel_clock_value_label.setText("-- MHz")
|
||||
|
||||
if self._camera_name is None:
|
||||
if self.camera_combo.devices:
|
||||
self.status_label.setText("No camera selected")
|
||||
else:
|
||||
self.status_label.setText("No IDS cameras configured")
|
||||
else:
|
||||
self.status_label.setText(f"Camera: {self._camera_name}")
|
||||
|
||||
self._refresh_control_availability()
|
||||
|
||||
def _refresh_control_availability(self):
|
||||
have_camera = self._camera_name is not None
|
||||
enabled = have_camera and not self._queue_busy
|
||||
tooltip = "Disabled while scan queue is busy." if have_camera and self._queue_busy else ""
|
||||
|
||||
self.auto_exposure_toggle.setEnabled(enabled)
|
||||
self.auto_exposure_toggle.setToolTip(tooltip)
|
||||
self.exposure_time_slider.setEnabled(enabled and not self._auto_exposure_enabled)
|
||||
self.exposure_time_slider.setToolTip(tooltip)
|
||||
self.pixel_clock_slider.setEnabled(enabled and bool(self._pixel_clock_options))
|
||||
self.pixel_clock_slider.setToolTip(tooltip)
|
||||
|
||||
################################################################################
|
||||
# Exposure / pixel clock
|
||||
################################################################################
|
||||
|
||||
@SafeSlot(dict, dict)
|
||||
def getting_camera_status(self, data, meta):
|
||||
if self._camera_name is None:
|
||||
return
|
||||
signals = data.get("signals") or {}
|
||||
|
||||
auto_exp = signals.get(f"{self._camera_name}_auto_exposure_enabled")
|
||||
if auto_exp is not None:
|
||||
enabled = bool(auto_exp.get("value"))
|
||||
self.auto_exposure_toggle.blockSignals(True)
|
||||
self.auto_exposure_toggle.checked = enabled
|
||||
self.auto_exposure_toggle.blockSignals(False)
|
||||
self._auto_exposure_enabled = enabled
|
||||
self._refresh_control_availability()
|
||||
|
||||
# No auto_gain_enabled handling here -- see the comment above the
|
||||
# (omitted) toggle in _init_ui.
|
||||
|
||||
# Update the pixel-clock slider's position before the exposure ones
|
||||
# below -- changing the pixel clock changes the exposure range, and
|
||||
# both can arrive together in this same message.
|
||||
pixel_clock = signals.get(f"{self._camera_name}_pixel_clock")
|
||||
if pixel_clock is not None:
|
||||
self._set_pixel_clock_display(int(pixel_clock.get("value")))
|
||||
|
||||
# Reseed the exposure slider's bounds next, so a value arriving in
|
||||
# the same message is clamped against up-to-date min/max. Round the
|
||||
# min UP and the max DOWN (never the other way) -- ceil (clamped to
|
||||
# at least 1) keeps the bound physical instead of claiming 0 ms
|
||||
# exposure is settable when it isn't.
|
||||
exposure_min = signals.get(f"{self._camera_name}_exposure_time_min")
|
||||
exposure_max = signals.get(f"{self._camera_name}_exposure_time_max")
|
||||
if exposure_min is not None and exposure_max is not None:
|
||||
lo = max(1, math.ceil(float(exposure_min.get("value")) * self._EXPOSURE_SLIDER_SCALE))
|
||||
hi = math.floor(float(exposure_max.get("value")) * self._EXPOSURE_SLIDER_SCALE)
|
||||
if (lo, hi) != (self.exposure_time_slider.minimum(), self.exposure_time_slider.maximum()):
|
||||
self.exposure_time_slider.setRange(lo, hi)
|
||||
|
||||
exposure_time = signals.get(f"{self._camera_name}_exposure_time")
|
||||
if exposure_time is not None:
|
||||
self.exposure_time_slider.blockSignals(True)
|
||||
self.exposure_time_slider.setValue(
|
||||
round(float(exposure_time.get("value")) * self._EXPOSURE_SLIDER_SCALE)
|
||||
)
|
||||
self.exposure_time_slider.blockSignals(False)
|
||||
self._update_exposure_time_value_label(self.exposure_time_slider.value())
|
||||
|
||||
@SafeSlot(bool)
|
||||
def auto_exposure_enabled_changed(self, enabled: bool):
|
||||
if self._camera_name is None or self._queue_busy:
|
||||
logger.warning("Ignoring auto-exposure toggle: no camera selected or queue busy.")
|
||||
return
|
||||
self.auto_exposure_toggle.blockSignals(True)
|
||||
self.dev.get(self._camera_name).auto_exposure_enabled.put(enabled)
|
||||
self.auto_exposure_toggle.checked = enabled
|
||||
self.auto_exposure_toggle.blockSignals(False)
|
||||
self._auto_exposure_enabled = enabled
|
||||
self._refresh_control_availability()
|
||||
|
||||
def _update_exposure_time_value_label(self, raw_value: int):
|
||||
self.exposure_time_value_label.setText(
|
||||
f"{raw_value / self._EXPOSURE_SLIDER_SCALE:.1f} ms"
|
||||
)
|
||||
|
||||
def exposure_time_submitted(self):
|
||||
if self._camera_name is None or self._queue_busy:
|
||||
return
|
||||
value_ms = self.exposure_time_slider.value() / self._EXPOSURE_SLIDER_SCALE
|
||||
self.dev.get(self._camera_name).exposure_time.put(value_ms)
|
||||
|
||||
def _init_pixel_clock_options(self):
|
||||
"""One-time-per-camera fetch of the actually-supported pixel clocks
|
||||
(see IDSCamera.get_pixel_clock_list()'s docstring for why this can't
|
||||
just be [min, max] from a Kind.config signal) -- not a polling loop,
|
||||
this list is static per camera, so a single RPC call per camera
|
||||
selection is the right cost/benefit trade-off."""
|
||||
camera_name = self._camera_name
|
||||
if camera_name is None:
|
||||
return
|
||||
try:
|
||||
options = self.dev.get(camera_name).get_pixel_clock_list()
|
||||
except Exception as exc:
|
||||
logger.warning(f"Failed to fetch pixel clock options for {camera_name}: {exc}")
|
||||
return
|
||||
if camera_name != self._camera_name:
|
||||
return # camera changed again while this RPC call was in flight
|
||||
self._set_pixel_clock_options(sorted(int(v) for v in options))
|
||||
|
||||
def _set_pixel_clock_options(self, options: list[int]):
|
||||
self._pixel_clock_options = options
|
||||
if not options:
|
||||
self._refresh_control_availability()
|
||||
return
|
||||
self.pixel_clock_slider.blockSignals(True)
|
||||
self.pixel_clock_slider.setRange(0, len(options) - 1)
|
||||
self.pixel_clock_slider.blockSignals(False)
|
||||
if self._last_pixel_clock_mhz is not None:
|
||||
# A status message already arrived before this RPC call
|
||||
# returned -- apply it now that we can actually resolve it to a
|
||||
# slider position.
|
||||
self._set_pixel_clock_display(self._last_pixel_clock_mhz)
|
||||
else:
|
||||
self._update_pixel_clock_value_label(self.pixel_clock_slider.value())
|
||||
self._refresh_control_availability()
|
||||
|
||||
def _set_pixel_clock_display(self, mhz: int):
|
||||
"""Move the slider to the option nearest `mhz` (from hardware, via
|
||||
getting_camera_status()) and update the value label -- never a raw
|
||||
MHz value on the slider itself, only a valid index."""
|
||||
self._last_pixel_clock_mhz = mhz
|
||||
if not self._pixel_clock_options:
|
||||
# Options not fetched yet -- _set_pixel_clock_options() will
|
||||
# call back into this once they arrive.
|
||||
self.pixel_clock_value_label.setText(f"{mhz} MHz")
|
||||
return
|
||||
index = min(
|
||||
range(len(self._pixel_clock_options)),
|
||||
key=lambda i: abs(self._pixel_clock_options[i] - mhz),
|
||||
)
|
||||
self.pixel_clock_slider.blockSignals(True)
|
||||
self.pixel_clock_slider.setValue(index)
|
||||
self.pixel_clock_slider.blockSignals(False)
|
||||
self._update_pixel_clock_value_label(index)
|
||||
|
||||
def _update_pixel_clock_value_label(self, index: int):
|
||||
if 0 <= index < len(self._pixel_clock_options):
|
||||
self.pixel_clock_value_label.setText(f"{self._pixel_clock_options[index]} MHz")
|
||||
else:
|
||||
self.pixel_clock_value_label.setText("-- MHz")
|
||||
|
||||
def pixel_clock_submitted(self):
|
||||
if self._camera_name is None or self._queue_busy:
|
||||
return
|
||||
index = self.pixel_clock_slider.value()
|
||||
if not 0 <= index < len(self._pixel_clock_options):
|
||||
return
|
||||
self.dev.get(self._camera_name).pixel_clock.put(self._pixel_clock_options[index])
|
||||
|
||||
################################################################################
|
||||
# Scan-queue guard
|
||||
################################################################################
|
||||
|
||||
def _update_queue_busy_state(self, busy: bool):
|
||||
if busy:
|
||||
self._queue_idle_timer.stop()
|
||||
self._set_queue_busy(True)
|
||||
return
|
||||
if self._queue_busy and not self._queue_idle_timer.isActive():
|
||||
self._queue_idle_timer.start()
|
||||
|
||||
def _set_queue_busy(self, busy: bool):
|
||||
if busy == self._queue_busy:
|
||||
return
|
||||
self._queue_busy = busy
|
||||
self._refresh_control_availability()
|
||||
|
||||
def _release_queue_busy(self):
|
||||
self._set_queue_busy(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 IDSCameraSettings: {exc}")
|
||||
return
|
||||
if msg is None:
|
||||
return
|
||||
self._update_queue_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
|
||||
|
||||
@SafeSlot(dict, dict)
|
||||
def on_queue_status_update(self, data, meta):
|
||||
_ = meta
|
||||
self._update_queue_busy_state(self._is_queue_busy(data))
|
||||
|
||||
def cleanup(self):
|
||||
self._queue_idle_timer.stop()
|
||||
if self._camera_name is not None:
|
||||
self.bec_dispatcher.disconnect_slot(
|
||||
self.getting_camera_status,
|
||||
MessageEndpoints.device_read_configuration(self._camera_name),
|
||||
)
|
||||
self.bec_dispatcher.disconnect_slot(
|
||||
self.on_queue_status_update, MessageEndpoints.scan_queue_status()
|
||||
)
|
||||
super().cleanup()
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
import sys
|
||||
|
||||
from bec_widgets.utils import BECDispatcher
|
||||
from bec_widgets.utils.colors import apply_theme
|
||||
from qtpy.QtWidgets import QApplication
|
||||
|
||||
app = QApplication(sys.argv)
|
||||
apply_theme("light")
|
||||
dispatcher = BECDispatcher(gui_id="ids_camera_settings")
|
||||
win = IDSCameraSettings()
|
||||
|
||||
win.resize(400, 250)
|
||||
win.show()
|
||||
sys.exit(app.exec_())
|
||||
@@ -0,0 +1 @@
|
||||
{'files': ['ids_camera_settings.py']}
|
||||
@@ -0,0 +1,57 @@
|
||||
# Copyright (C) 2022 The Qt Company Ltd.
|
||||
# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause
|
||||
|
||||
from qtpy.QtDesigner import QDesignerCustomWidgetInterface
|
||||
from qtpy.QtWidgets import QWidget
|
||||
|
||||
from bec_widgets.utils.bec_designer import designer_material_icon
|
||||
from csaxs_bec.bec_widgets.widgets.ids_camera_settings.ids_camera_settings import IDSCameraSettings
|
||||
|
||||
DOM_XML = """
|
||||
<ui language='c++'>
|
||||
<widget class='IDSCameraSettings' name='ids_camera_settings'>
|
||||
</widget>
|
||||
</ui>
|
||||
"""
|
||||
|
||||
|
||||
class IDSCameraSettingsPlugin(QDesignerCustomWidgetInterface): # pragma: no cover
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self._form_editor = None
|
||||
|
||||
def createWidget(self, parent):
|
||||
if parent is None:
|
||||
return QWidget()
|
||||
t = IDSCameraSettings(parent)
|
||||
return t
|
||||
|
||||
def domXml(self):
|
||||
return DOM_XML
|
||||
|
||||
def group(self):
|
||||
return ""
|
||||
|
||||
def icon(self):
|
||||
return designer_material_icon(IDSCameraSettings.ICON_NAME)
|
||||
|
||||
def includeFile(self):
|
||||
return "ids_camera_settings"
|
||||
|
||||
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 "IDSCameraSettings"
|
||||
|
||||
def toolTip(self):
|
||||
return "Pick a configured IDS camera and adjust its exposure time / pixel clock."
|
||||
|
||||
def whatsThis(self):
|
||||
return self.toolTip()
|
||||
@@ -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.ids_camera_settings.ids_camera_settings_plugin import IDSCameraSettingsPlugin
|
||||
|
||||
QPyDesignerCustomWidgetCollection.addCustomWidget(IDSCameraSettingsPlugin())
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
main()
|
||||
Reference in new issue
Block a user