|
|
|
@@ -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_())
|