feat(saxs-widget): Add SAXS Widget for scanning SAXS and Tensor Tomography measurement setup
CI for csaxs_bec / test (pull_request) Successful in 1m31s
Read the Docs Deploy Trigger / trigger-rtd-webhook (push) Successful in 2s
CI for csaxs_bec / test (push) Successful in 1m34s

This commit was merged in pull request #253.
This commit is contained in:
2026-07-10 16:54:37 +02:00
parent 46a05b48e0
commit 804ffef5b0
16 changed files with 4306 additions and 1 deletions
+77 -1
View File
@@ -4,7 +4,6 @@
from __future__ import annotations
from bec_lib.logger import bec_logger
from bec_widgets.cli.rpc.rpc_base import RPCBase, rpc_call, rpc_timeout
logger = bec_logger.logger
@@ -14,6 +13,7 @@ logger = bec_logger.logger
_Widgets = {
"SampleStorageWidget": "SampleStorageWidget",
"SAXSWidget": "SAXSWidget",
"SlitControlWidget": "SlitControlWidget",
"TomoParamsWidget": "TomoParamsWidget",
"XRayEye": "XRayEye",
@@ -32,6 +32,82 @@ class SampleStorageWidget(RPCBase):
"""
class SAXSWidget(RPCBase):
"""Widget for preparing SAXS measurement tables from an image ROI."""
_IMPORT_MODULE = "csaxs_bec.bec_widgets.widgets.saxs_widget.saxs_widget"
@rpc_call
def add_scanning_row(
self,
sample_name: "str | None" = None,
exposure_time: "float | None" = None,
fast_axis: "str | None" = None,
stepper_axis: "str | None" = None,
fast_step: "float | None" = None,
stepper_step: "float | None" = None,
comment: "str" = "",
) -> "SAXSRow | None":
"""
None
"""
@rpc_call
def add_tensor_rows(
self,
sample_name: "str | None" = None,
exposure_time: "float | None" = None,
fast_axis: "str | None" = None,
stepper_axis: "str | None" = None,
fast_step: "float | None" = None,
stepper_step: "float | None" = None,
comment: "str" = "",
roty: "float | None" = None,
rotx_values: "list[float] | None" = None,
roty_values: "list[float] | None" = None,
angle_pairs: "list[tuple[float, float]] | None" = None,
) -> "list[SAXSRow]":
"""
None
"""
@rpc_call
def clear_rows(self) -> "None":
"""
None
"""
@rpc_call
def delete_selected_rows(self) -> "None":
"""
None
"""
@rpc_call
def export_csv(self, file_name: "str | None" = None, selected: "bool" = False) -> "str | None":
"""
None
"""
@rpc_call
def rows_as_dicts(self) -> "list[dict]":
"""
None
"""
@rpc_call
def set_positioners(self, x_positioner: "str", y_positioner: "str") -> "None":
"""
None
"""
@rpc_call
def mark_scan_result(self, sample_id_or_row, scan_number, success: "bool") -> "None":
"""
None
"""
class SlitControlWidget(RPCBase):
"""Interactive GUI for cSAXS slit center and size control."""
@@ -9,6 +9,7 @@ designer_plugins = {
"csaxs_bec.bec_widgets.widgets.sample_storage.sample_storage",
"SampleStorageWidget",
),
"SAXSWidget": ("csaxs_bec.bec_widgets.widgets.saxs_widget.saxs_widget", "SAXSWidget"),
"SlitControlWidget": (
"csaxs_bec.bec_widgets.widgets.slit_control.slit_control",
"SlitControlWidget",
@@ -22,6 +23,7 @@ designer_plugins = {
widget_icons = {
"SampleStorageWidget": "widgets",
"SAXSWidget": "table_chart",
"SlitControlWidget": "widgets",
"TomoParamsWidget": "widgets",
"XRayEye": "widgets",
@@ -0,0 +1,184 @@
# SAXS Widget
The SAXS widget is a planning widget for preparing scanning SAXS and SAXS tensor tomography
measurements from an image ROI. It combines:
- a live image view with one persistent rectangular ROI,
- X/Y positioner selection and compact positioner controls,
- ROI-to-positioner mapping controls with camera scale per pixel, camera-to-x-ray offset and direction,
- row creation dialogs for scanning SAXS and tensor tomography,
- a measurement table that can be saved, exported as CSV, or submitted to downstream code.
The widget does not run the scan itself. Its main job is to turn the current ROI, mapping
corrections, selected motors and user-entered scan parameters into a flat table payload that a
macro or script can consume.
## Session Payload
Saved sessions are published as `VariableMessage` entries through
`MessageEndpointSAXSWidget.measurement_table(session_name)`, which resolves to:
```text
info/saxs_widget/sessions/{session_name}
```
The saved message has this shape:
```python
VariableMessage(
value=[row_0, row_1, ...],
metadata={
"session_name": session_name,
"timestamp": "2026-07-10T12:34:56Z",
},
)
```
`metadata["timestamp"]` is an ISO-8601 UTC string, not a POSIX timestamp.
The `value` field is a full table snapshot: a list of flat dictionaries, where each dictionary is
one planned scan row. Do not treat one stream message as one scan row; treat it as the saved table
version for that session.
For a session's current plan, select the newest snapshot from its stream using
`metadata["timestamp"]`. Older snapshots are retained as session history (max 50 entries) and should only be used
when a macro intentionally needs to replay an earlier plan.
CSV export uses the same row fields, but additionally prefixes each row with `session_name`.
## Row Fields For Scan Macros
Core scan fields:
| Field | Meaning |
| --- | --- |
| `template` | `scanning` or `tensor`. Tensor rows include rotation angle values. |
| `sample_name` | User-provided sample label. |
| `sample_id` | Stable generated ID for this row. Use this when reporting scan results back. |
| `exposure_time` | Exposure time in seconds. |
| `comment` | Optional free-text note from the user. |
| `fast_axis` | Resolved motor name used as the fast axis. |
| `fast_start` / `fast_stop` | Corrected start and stop positions for the fast axis. |
| `fast_step` | Step size along the fast axis. |
| `stepper_axis` | Resolved motor name used as the stepper axis. |
| `stepper_start` / `stepper_stop` | Corrected start and stop positions for the stepper axis. |
| `stepper_step` | Step size along the stepper axis. |
| `images_per_line` | Number of image points along one fast-axis line. |
| `num_lines` | Number of stepper-axis lines. |
| `total_frames` | Total planned frames, equal to `images_per_line * num_lines`. |
Coordinate and motor context:
| Field | Meaning |
| --- | --- |
| `fast_axis_coordinate` | Coordinate selected as the fast axis: `x` or `y`. |
| `stepper_axis_coordinate` | Coordinate selected as the stepper axis: `x` or `y`. |
| `x_positioner` / `y_positioner` | Motors selected for the ROI X/Y coordinates. |
| `rotx_axis` / `roty_axis` | Rotation motors selected for tensor tomography rows. |
| `rotx` / `roty` | Rotation angles for this row. They are `None` for non-tensor rows. |
| `camera_center_x`, `camera_center_y` | X/Y motor positions that corresponded to the image reference point when the row was created. |
ROI and mapping provenance:
| Field | Meaning |
| --- | --- |
| `raw_roi_x`, `raw_roi_y` | Raw ROI origin in image coordinates, retained for traceability and restoring the camera view. |
| `raw_roi_width`, `raw_roi_height` | Raw ROI size in image coordinates, retained for traceability and restoring the camera view. |
| `correction_camera_scale_pixel_x`, `correction_camera_scale_pixel_y` | Camera pixel-to-motor scale factors used to convert ROI coordinates to positions. |
| `correction_offset_cam_xray_x`, `correction_offset_cam_xray_y` | Calibrated offsets from the camera reference point to the x-ray beam, in motor units. |
| `correction_direction_x`, `correction_direction_y` | Direction signs, usually `1` or `-1`. |
| `correction_reference_x`, `correction_reference_y` | Image coordinates used as the camera reference point. New rows use the image center when image data exists, otherwise the ROI center. |
The mapping reference is reset to the image center only when a first image becomes available
or the image shape changes. While no image data is available, it follows the ROI center.
Coordinate conversion follows this workflow:
```text
camera-space motor position = camera_center + direction * camera_scale_pixel * (raw_roi - reference)
x-ray beam scan position = camera-space motor position + offset_cam_xray
```
`Move to selected` restores the saved raw ROI and moves back to `camera_center_x` /
`camera_center_y`, i.e. the motor positions saved when the row was created. The exported table scan
coordinates include both the camera-scaled ROI displacement and the calibrated camera-to-x-ray offset.
The ROI, camera-center, and correction fields document how the plan was created. A scan macro
should execute the already-corrected `fast_*` and `stepper_*` fields rather than recalculating
positions from the image coordinates.
Bookkeeping and convenience fields:
| Field | Meaning |
| --- | --- |
| `scan_results` | JSON string containing previous scan result records, if any. Not needed to start a new scan. |
| `step_size_x` / `step_size_y` | Coordinate-specific step size aliases. |
## Macro Usage Notes
The fields required to execute a row are `template`, `fast_axis`, `fast_start`, `fast_stop`,
`fast_step`, `stepper_axis`, `stepper_start`, `stepper_stop`, `stepper_step`,
`exposure_time`, and the detector-count fields required by the beamline. Tensor rows additionally
require `rotx_axis`, `roty_axis`, `rotx`, and `roty`.
A scan macro should usually consume the row fields in this order:
1. Validate that the message value is a list of rows and that each row has the required fields and usable motor names.
2. Configure detector/exposure from `exposure_time`, `images_per_line`, `num_lines`,
and `total_frames`.
3. Run the grid scan with `fast_axis`, `fast_start`, `fast_stop`, `fast_step`,
`stepper_axis`, `stepper_start`, `stepper_stop` and `stepper_step`.
4. For `template == "tensor"`, move or include `rotx_axis`/`roty_axis` with `rotx`/`roty`
before running the grid scan.
5. Preserve `sample_name`, `sample_id`, `comment` and ROI/correction fields in metadata so the
scan can be traced back to the planned row.
Minimal pseudo-code:
```python
def run_saxs_table(variable_message):
session_name = variable_message.metadata.get("session_name", "")
rows = variable_message.value
for row in rows:
if not row["fast_axis"] or not row["stepper_axis"]:
raise ValueError("SAXS row does not provide usable fast and stepper motors")
metadata = {
"session_name": session_name,
"sample_name": row["sample_name"],
"sample_id": row["sample_id"],
"comment": row["comment"],
"raw_roi": {
"x": row["raw_roi_x"],
"y": row["raw_roi_y"],
"width": row["raw_roi_width"],
"height": row["raw_roi_height"],
},
"camera_center": {
"x": row["camera_center_x"],
"y": row["camera_center_y"],
},
}
if row["template"] == "tensor":
move(row["rotx_axis"], row["rotx"])
move(row["roty_axis"], row["roty"])
run_grid_scan(
fast_axis=row["fast_axis"],
fast_start=row["fast_start"],
fast_stop=row["fast_stop"],
fast_step=row["fast_step"],
stepper_axis=row["stepper_axis"],
stepper_start=row["stepper_start"],
stepper_stop=row["stepper_stop"],
stepper_step=row["stepper_step"],
exposure_time=row["exposure_time"],
images_per_line=row["images_per_line"],
metadata=metadata,
)
```
The function names above are placeholders. Use the beamline's actual motion and scan APIs, but keep
the field mapping explicit so payload changes are easy to review.
@@ -0,0 +1 @@
@@ -0,0 +1,447 @@
from __future__ import annotations
from bec_qthemes import material_icon
from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import (
BECDeviceFilter,
DeviceComboBox,
)
from bec_widgets.widgets.utility.toggle.toggle import ToggleSwitch
from qtpy.QtCore import Property, QSize, Qt, Signal
from qtpy.QtWidgets import (
QCheckBox,
QComboBox,
QDoubleSpinBox,
QFrame,
QGridLayout,
QGroupBox,
QLabel,
QLineEdit,
QPushButton,
)
from csaxs_bec.bec_widgets.widgets.saxs_widget.models import Correction
TOOLTIP_MAPPING_LOCK = "Lock ROI mapping values to prevent accidental edits during setup."
TOOLTIP_CAMERA_SCALE_PIXEL = (
"Pixel-to-motor scale. This converts an ROI pixel delta from the camera reference "
"point into motor units."
)
TOOLTIP_OFFSET_CAM_XRAY = (
"Calibrated offset from the camera reference point to the x-ray beam, in motor units."
)
TOOLTIP_DIRECTION = (
"Orientation sign. Use + when increasing image pixels should increase the motor "
"coordinate, or - when the camera/motor axis is inverted."
)
TOOLTIP_REFERENCE = (
"Current image reference point in pixels. This is normally the pinned crosshair at "
"the camera center and is read-only here."
)
TOOLTIP_ROI_SIZE = "Target ROI width and height in image pixels."
TOOLTIP_APPLY_ROI_SIZE = (
"Resize the current ROI to the entered width and height while keeping its center."
)
TOOLTIP_APPLY_CORRECTIONS = (
"Apply the current mapping values to selected or all existing table rows."
)
def separator(parent=None) -> QFrame:
line = QFrame(parent=parent)
line.setObjectName("saxs_toolbar_separator")
line.setFrameShape(QFrame.Shape.HLine)
line.setFrameShadow(QFrame.Shadow.Sunken)
return line
def field_label(text: str, parent=None, *, tooltip: str = "") -> QLabel:
label = QLabel(text, parent)
label.setObjectName("saxs_field_label")
if tooltip:
label.setToolTip(tooltip)
return label
def spinbox(
*,
value: float,
parent=None,
minimum: float = -1_000_000.0,
maximum: float = 1_000_000.0,
single_step: float = 0.1,
decimals: int = 4,
) -> QDoubleSpinBox:
spin = QDoubleSpinBox(parent=parent)
spin.setDecimals(decimals)
spin.setRange(minimum, maximum)
spin.setSingleStep(single_step)
spin.setValue(value)
spin.setStepType(QDoubleSpinBox.StepType.AdaptiveDecimalStepType)
return spin
def positive_spinbox(value: float, *, single_step: float = 0.01, parent=None) -> QDoubleSpinBox:
return spinbox(
value=value,
minimum=1e-12,
maximum=1_000_000.0,
single_step=single_step,
decimals=6,
parent=parent,
)
def device_combo(parent=None) -> DeviceComboBox:
combo = DeviceComboBox(
parent=parent, device_filter=BECDeviceFilter.POSITIONER, autocomplete=True
)
combo.set_first_element_as_empty = True
return combo
def direction_combo(parent=None) -> QComboBox:
combo = QComboBox(parent)
combo.addItem("+", 1)
combo.addItem("-", -1)
combo.setItemData(
0,
"Positive image-pixel movement maps to positive motor movement on this axis.",
Qt.ItemDataRole.ToolTipRole,
)
combo.setItemData(
1,
"Positive image-pixel movement maps to negative motor movement on this axis.",
Qt.ItemDataRole.ToolTipRole,
)
return combo
class DeviceSelectionBox(QGroupBox):
"""Device inputs for translation positioners and optional SASTT axes."""
x_positioner_changed = Signal(str)
y_positioner_changed = Signal(str)
rotx_axis_changed = Signal(str)
roty_axis_changed = Signal(str)
lock_changed = Signal(bool)
sastt_enabled_changed = Signal(bool)
device_selection_enabled_changed = Signal(bool)
def __init__(self, parent=None):
super().__init__("Devices", parent)
self.setObjectName("saxs_device_selection_box")
layout = QGridLayout(self)
layout.setContentsMargins(10, 14, 10, 10)
layout.setHorizontalSpacing(8)
layout.setVerticalSpacing(7)
layout.setColumnStretch(1, 1)
layout.setColumnStretch(3, 1)
self.x_positioner_combo = device_combo(self)
self.y_positioner_combo = device_combo(self)
self.rotx_axis_combo = device_combo(self)
self.roty_axis_combo = device_combo(self)
self.device_lock_toggle = ToggleSwitch(self, checked=False)
self.sastt_checkbox = QCheckBox("Enable SASTT", self)
self.sastt_checkbox.setChecked(True)
self.sastt_checkbox.setToolTip(
"Enable tensor tomography mode to use X/Y rotation axes when adding rows."
)
layout.addWidget(field_label("Edit lock", self), 0, 0)
layout.addWidget(self.device_lock_toggle, 0, 1, alignment=Qt.AlignmentFlag.AlignLeft)
layout.addWidget(self.sastt_checkbox, 0, 2, 1, 2)
layout.addWidget(field_label("X axis", self), 1, 0)
layout.addWidget(field_label("Y axis", self), 1, 2)
layout.addWidget(self.x_positioner_combo, 2, 0, 1, 2)
layout.addWidget(self.y_positioner_combo, 2, 2, 1, 2)
layout.addWidget(field_label("X rotation", self), 3, 0)
layout.addWidget(field_label("Y rotation", self), 3, 2)
layout.addWidget(self.rotx_axis_combo, 4, 0, 1, 2)
layout.addWidget(self.roty_axis_combo, 4, 2, 1, 2)
self.x_positioner_combo.currentTextChanged.connect(self.x_positioner_changed.emit)
self.y_positioner_combo.currentTextChanged.connect(self.y_positioner_changed.emit)
self.rotx_axis_combo.currentTextChanged.connect(self.rotx_axis_changed.emit)
self.roty_axis_combo.currentTextChanged.connect(self.roty_axis_changed.emit)
self.device_lock_toggle.enabled.connect(self._on_lock_changed)
self.sastt_checkbox.stateChanged.connect(self._on_sastt_changed)
self._update_enabled_state()
def _get_x_positioner(self) -> str:
return self.x_positioner_combo.currentText()
def _set_x_positioner(self, value: str) -> None:
self.x_positioner_combo.setCurrentText(value or "")
def _get_y_positioner(self) -> str:
return self.y_positioner_combo.currentText()
def _set_y_positioner(self, value: str) -> None:
self.y_positioner_combo.setCurrentText(value or "")
def _get_rotx_axis(self) -> str:
return self.rotx_axis_combo.currentText()
def _set_rotx_axis(self, value: str) -> None:
self.rotx_axis_combo.setCurrentText(value or "")
def _get_roty_axis(self) -> str:
return self.roty_axis_combo.currentText()
def _set_roty_axis(self, value: str) -> None:
self.roty_axis_combo.setCurrentText(value or "")
def _get_sastt_enabled(self) -> bool:
return self.sastt_checkbox.isChecked()
def _set_sastt_enabled(self, enabled: bool) -> None:
self.sastt_checkbox.setChecked(bool(enabled))
def _get_device_selection_enabled(self) -> bool:
return bool(self.device_lock_toggle.checked)
def _set_device_selection_enabled(self, enabled: bool) -> None:
self.device_lock_toggle.checked = bool(enabled)
self._update_enabled_state()
x_positioner = Property(str, _get_x_positioner, _set_x_positioner, notify=x_positioner_changed)
y_positioner = Property(str, _get_y_positioner, _set_y_positioner, notify=y_positioner_changed)
rotx_axis = Property(str, _get_rotx_axis, _set_rotx_axis, notify=rotx_axis_changed)
roty_axis = Property(str, _get_roty_axis, _set_roty_axis, notify=roty_axis_changed)
sastt_enabled = Property(
bool, _get_sastt_enabled, _set_sastt_enabled, notify=sastt_enabled_changed
)
device_selection_enabled = Property(
bool,
_get_device_selection_enabled,
_set_device_selection_enabled,
notify=device_selection_enabled_changed,
)
def is_locked(self) -> bool:
return not self.device_selection_enabled
def is_sastt_enabled(self) -> bool:
return self.sastt_enabled
def set_positioners(self, x_positioner: str, y_positioner: str) -> None:
if x_positioner:
self.x_positioner = x_positioner
if y_positioner:
self.y_positioner = y_positioner
def _on_lock_changed(self, enabled: bool) -> None:
self._update_enabled_state()
self.device_selection_enabled_changed.emit(bool(enabled))
self.lock_changed.emit(not bool(enabled))
def _on_sastt_changed(self, state: int) -> None:
self._update_enabled_state()
self.sastt_enabled_changed.emit(bool(state))
def _update_enabled_state(self) -> None:
is_locked = self.is_locked()
sastt_enabled = self.sastt_enabled
self.x_positioner_combo.setEnabled(is_locked)
self.y_positioner_combo.setEnabled(is_locked)
self.rotx_axis_combo.setEnabled(is_locked and sastt_enabled)
self.roty_axis_combo.setEnabled(is_locked and sastt_enabled)
# Keep SASTT toggle usable even when axis selectors are disabled.
self.sastt_checkbox.setEnabled(True)
class ROIMappingBox(QGroupBox):
"""Controls for converting image ROI coordinates into scan coordinates."""
apply_requested = Signal()
roi_size_requested = Signal(float, float)
def __init__(self, parent=None):
super().__init__("ROI mapping", parent)
self.setObjectName("saxs_roi_mapping_box")
layout = QGridLayout(self)
layout.setContentsMargins(10, 14, 10, 10)
layout.setHorizontalSpacing(8)
layout.setVerticalSpacing(7)
layout.setColumnStretch(1, 1)
layout.setColumnStretch(2, 1)
# An unknown calibration must not silently become a one-motor-unit-per-pixel scan.
self.camera_scale_pixel_x = spinbox(value=0.0, single_step=0.1, parent=self)
self.offset_cam_xray_x = spinbox(value=0.0, single_step=0.1, parent=self)
self.direction_x = direction_combo(self)
self.camera_scale_pixel_y = spinbox(value=0.0, single_step=0.1, parent=self)
self.offset_cam_xray_y = spinbox(value=0.0, single_step=0.1, parent=self)
self.direction_y = direction_combo(self)
self.mapping_lock_toggle = ToggleSwitch(self, checked=False)
self.reference_x = QLineEdit("0", self)
self.reference_y = QLineEdit("0", self)
self.roi_width = positive_spinbox(50.0, single_step=1.0, parent=self)
self.roi_height = positive_spinbox(50.0, single_step=1.0, parent=self)
self._set_field_tooltips()
for reference in [self.reference_x, self.reference_y]:
reference.setObjectName("saxs_reference_field")
reference.setReadOnly(True)
reference.setAlignment(Qt.AlignmentFlag.AlignRight)
self.edit_lock_label = field_label("Edit lock", self, tooltip=TOOLTIP_MAPPING_LOCK)
self.axis_label = field_label("Axis", self)
self.camera_scale_pixel_label = field_label(
"Camera scale / px", self, tooltip=TOOLTIP_CAMERA_SCALE_PIXEL
)
self.offset_cam_xray_label = field_label(
"Cam-xray offset", self, tooltip=TOOLTIP_OFFSET_CAM_XRAY
)
self.direction_label = field_label("Direction", self, tooltip=TOOLTIP_DIRECTION)
self.reference_label = field_label("Reference", self, tooltip=TOOLTIP_REFERENCE)
self.reference_axis_label = field_label("x / y", self, tooltip=TOOLTIP_REFERENCE)
self.roi_size_label = field_label("ROI size", self, tooltip=TOOLTIP_ROI_SIZE)
layout.addWidget(self.edit_lock_label, 0, 2, alignment=Qt.AlignmentFlag.AlignRight)
layout.addWidget(self.mapping_lock_toggle, 0, 3, alignment=Qt.AlignmentFlag.AlignLeft)
layout.addWidget(self.axis_label, 1, 0)
layout.addWidget(self.camera_scale_pixel_label, 1, 1)
layout.addWidget(self.offset_cam_xray_label, 1, 2)
layout.addWidget(self.direction_label, 1, 3)
layout.addWidget(field_label("X", self), 2, 0)
layout.addWidget(self.camera_scale_pixel_x, 2, 1)
layout.addWidget(self.offset_cam_xray_x, 2, 2)
layout.addWidget(self.direction_x, 2, 3)
layout.addWidget(field_label("Y", self), 3, 0)
layout.addWidget(self.camera_scale_pixel_y, 3, 1)
layout.addWidget(self.offset_cam_xray_y, 3, 2)
layout.addWidget(self.direction_y, 3, 3)
layout.addWidget(self.reference_label, 4, 0)
layout.addWidget(self.reference_x, 4, 1)
layout.addWidget(self.reference_y, 4, 2)
layout.addWidget(self.reference_axis_label, 4, 3)
self.apply_roi_size_button = QPushButton("Apply ROI size", self)
self.apply_roi_size_button.setIcon(material_icon("aspect_ratio", size=(16, 16)))
self.apply_roi_size_button.setIconSize(QSize(18, 18))
self.apply_roi_size_button.setMinimumHeight(32)
self.apply_roi_size_button.setCursor(Qt.CursorShape.PointingHandCursor)
self.apply_roi_size_button.setFlat(True)
self.apply_roi_size_button.setProperty("saxsButtonRole", "toolbar")
self.apply_roi_size_button.setToolTip(TOOLTIP_APPLY_ROI_SIZE)
layout.addWidget(self.roi_size_label, 5, 0)
layout.addWidget(self.roi_width, 5, 1)
layout.addWidget(self.roi_height, 5, 2)
layout.addWidget(self.apply_roi_size_button, 5, 3)
self.apply_corrections_button = QPushButton("Apply corrections", self)
self.apply_corrections_button.setIcon(
material_icon("published_with_changes", size=(16, 16))
)
self.apply_corrections_button.setIconSize(QSize(18, 18))
self.apply_corrections_button.setMinimumHeight(32)
self.apply_corrections_button.setCursor(Qt.CursorShape.PointingHandCursor)
self.apply_corrections_button.setFlat(True)
self.apply_corrections_button.setProperty("saxsButtonRole", "toolbar")
self.apply_corrections_button.setToolTip(TOOLTIP_APPLY_CORRECTIONS)
layout.addWidget(self.apply_corrections_button, 6, 0, 1, 4)
self.apply_roi_size_button.clicked.connect(self._emit_roi_size_requested)
self.apply_corrections_button.clicked.connect(self.apply_requested.emit)
self.mapping_lock_toggle.stateChanged.connect(self.set_locked)
self.set_locked(self.mapping_lock_toggle.isChecked())
def _set_field_tooltips(self) -> None:
self.mapping_lock_toggle.setToolTip(TOOLTIP_MAPPING_LOCK)
self.camera_scale_pixel_x.setToolTip(TOOLTIP_CAMERA_SCALE_PIXEL)
self.camera_scale_pixel_y.setToolTip(TOOLTIP_CAMERA_SCALE_PIXEL)
self.offset_cam_xray_x.setToolTip(TOOLTIP_OFFSET_CAM_XRAY)
self.offset_cam_xray_y.setToolTip(TOOLTIP_OFFSET_CAM_XRAY)
self.direction_x.setToolTip(TOOLTIP_DIRECTION)
self.direction_y.setToolTip(TOOLTIP_DIRECTION)
self.reference_x.setToolTip(TOOLTIP_REFERENCE)
self.reference_y.setToolTip(TOOLTIP_REFERENCE)
self.roi_width.setToolTip(TOOLTIP_ROI_SIZE)
self.roi_height.setToolTip(TOOLTIP_ROI_SIZE)
def correction(self) -> Correction:
return Correction(
camera_scale_pixel_x=self.camera_scale_pixel_x.value(),
offset_cam_xray_x=self.offset_cam_xray_x.value(),
direction_x=int(self.direction_x.currentData() or 1),
camera_scale_pixel_y=self.camera_scale_pixel_y.value(),
offset_cam_xray_y=self.offset_cam_xray_y.value(),
direction_y=int(self.direction_y.currentData() or 1),
reference_x=float(self.reference_x.text() or 0.0),
reference_y=float(self.reference_y.text() or 0.0),
)
def set_correction(self, correction: Correction) -> None:
for widget in [
self.camera_scale_pixel_x,
self.offset_cam_xray_x,
self.direction_x,
self.camera_scale_pixel_y,
self.offset_cam_xray_y,
self.direction_y,
]:
widget.blockSignals(True)
try:
self.camera_scale_pixel_x.setValue(correction.camera_scale_pixel_x)
self.offset_cam_xray_x.setValue(correction.offset_cam_xray_x)
self._set_combo_current_data(self.direction_x, correction.direction_x)
self.camera_scale_pixel_y.setValue(correction.camera_scale_pixel_y)
self.offset_cam_xray_y.setValue(correction.offset_cam_xray_y)
self._set_combo_current_data(self.direction_y, correction.direction_y)
finally:
for widget in [
self.camera_scale_pixel_x,
self.offset_cam_xray_x,
self.direction_x,
self.camera_scale_pixel_y,
self.offset_cam_xray_y,
self.direction_y,
]:
widget.blockSignals(False)
self.set_reference_position(correction.reference_x, correction.reference_y)
def set_reference_position(self, x: float, y: float) -> None:
self.reference_x.setText(f"{x:.6g}")
self.reference_y.setText(f"{y:.6g}")
def set_roi_size(self, width: float, height: float) -> None:
self.roi_width.blockSignals(True)
self.roi_height.blockSignals(True)
try:
self.roi_width.setValue(width)
self.roi_height.setValue(height)
finally:
self.roi_width.blockSignals(False)
self.roi_height.blockSignals(False)
@staticmethod
def _set_combo_current_data(combo: QComboBox, value: int) -> None:
for index in range(combo.count()):
if combo.itemData(index) == value:
combo.setCurrentIndex(index)
return
def _emit_roi_size_requested(self) -> None:
self.roi_size_requested.emit(self.roi_width.value(), self.roi_height.value())
def is_locked(self) -> bool:
return self.mapping_lock_toggle.isChecked()
def set_locked(self, locked: bool) -> None:
for widget in [
self.camera_scale_pixel_x,
self.offset_cam_xray_x,
self.direction_x,
self.camera_scale_pixel_y,
self.offset_cam_xray_y,
self.direction_y,
self.roi_width,
self.roi_height,
self.apply_roi_size_button,
self.apply_corrections_button,
]:
widget.setEnabled(not locked)
@@ -0,0 +1,108 @@
from __future__ import annotations
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from csaxs_bec.bec_widgets.widgets.saxs_widget.models import Correction, RawROI
def scaled_delta(
raw_value: float, reference: float, camera_scale_pixel: float, direction: int
) -> float:
"""Convert a raw image coordinate into a motor-space delta."""
return int(direction) * float(camera_scale_pixel) * (float(raw_value) - float(reference))
def mapped_coordinate(
raw_value: float,
*,
reference: float,
camera_scale_pixel: float,
direction: int,
camera_center: float = 0.0,
offset_cam_xray: float = 0.0,
) -> float:
"""Map one image coordinate to motor coordinates."""
return (
float(camera_center)
+ float(offset_cam_xray)
+ scaled_delta(raw_value, reference, camera_scale_pixel, direction)
)
def axis_limits(
raw_start: float,
raw_size: float,
*,
reference: float,
camera_scale_pixel: float,
direction: int,
camera_center: float = 0.0,
offset_cam_xray: float = 0.0,
) -> tuple[float, float]:
"""Return start/stop motor coordinates for one raw ROI axis."""
start = mapped_coordinate(
raw_start,
reference=reference,
camera_scale_pixel=camera_scale_pixel,
direction=direction,
camera_center=camera_center,
offset_cam_xray=offset_cam_xray,
)
stop = mapped_coordinate(
float(raw_start) + float(raw_size),
reference=reference,
camera_scale_pixel=camera_scale_pixel,
direction=direction,
camera_center=camera_center,
offset_cam_xray=offset_cam_xray,
)
return start, stop
def point_position(
image_x: float,
image_y: float,
correction: Correction,
*,
camera_center: tuple[float, float] = (0.0, 0.0),
include_offset_cam_xray: bool = True,
) -> tuple[float, float]:
"""Map one image point into X/Y motor coordinates."""
offset_cam_xray_x = correction.offset_cam_xray_x if include_offset_cam_xray else 0.0
offset_cam_xray_y = correction.offset_cam_xray_y if include_offset_cam_xray else 0.0
return (
mapped_coordinate(
image_x,
reference=correction.reference_x,
camera_scale_pixel=correction.camera_scale_pixel_x,
direction=correction.direction_x,
camera_center=camera_center[0],
offset_cam_xray=offset_cam_xray_x,
),
mapped_coordinate(
image_y,
reference=correction.reference_y,
camera_scale_pixel=correction.camera_scale_pixel_y,
direction=correction.direction_y,
camera_center=camera_center[1],
offset_cam_xray=offset_cam_xray_y,
),
)
def roi_origin_position(
raw_roi: RawROI,
correction: Correction,
*,
camera_center: tuple[float, float] = (0.0, 0.0),
include_offset_cam_xray: bool = True,
) -> tuple[float, float]:
"""Map the raw ROI origin into X/Y motor coordinates."""
return point_position(
raw_roi.x,
raw_roi.y,
correction,
camera_center=camera_center,
include_offset_cam_xray=include_offset_cam_xray,
)
@@ -0,0 +1,336 @@
from __future__ import annotations
from typing import Literal
from qtpy.QtWidgets import (
QCheckBox,
QComboBox,
QDialog,
QDialogButtonBox,
QFileDialog,
QFormLayout,
QGroupBox,
QHBoxLayout,
QLabel,
QLineEdit,
QMessageBox,
QPushButton,
QSpinBox,
QStackedWidget,
QTextEdit,
QVBoxLayout,
QWidget,
)
from csaxs_bec.bec_widgets.widgets.saxs_widget.controls import positive_spinbox, spinbox
from csaxs_bec.bec_widgets.widgets.saxs_widget.models import (
SAXSRow,
linspace_values,
load_angle_file,
pair_angle_lists,
parse_angle_text,
)
class AxisSelectorWidget(QWidget):
"""Simple checkbox-based axis selector."""
def __init__(self, label: str, *, axis_is_x: bool):
super().__init__()
self.axis_checkbox = QCheckBox(f"{label} axis is X", self)
self.axis_checkbox.setChecked(axis_is_x)
layout = QVBoxLayout(self)
layout.setContentsMargins(0, 0, 0, 0)
layout.addWidget(self.axis_checkbox)
def is_x(self) -> bool:
return self.axis_checkbox.isChecked()
def set_is_x(self, axis_is_x: bool) -> None:
if self.axis_checkbox.isChecked() == axis_is_x:
return
self.axis_checkbox.blockSignals(True)
self.axis_checkbox.setChecked(axis_is_x)
self.axis_checkbox.blockSignals(False)
class ScanStepWidget(QWidget):
"""Linked-by-default fast and stepper step size controls."""
def __init__(self, fast_step: float, stepper_step: float | None = None, *, linked: bool = True):
super().__init__()
self.fast_step = positive_spinbox(fast_step, single_step=0.005, parent=self)
self.stepper_step = positive_spinbox(
fast_step if stepper_step is None else stepper_step, single_step=0.005, parent=self
)
self.link_step_sizes = QCheckBox("Link step sizes", self)
self.link_step_sizes.setChecked(linked)
layout = QFormLayout(self)
layout.setContentsMargins(0, 0, 0, 0)
layout.addRow("Fast step", self.fast_step)
layout.addRow("Stepper step", self.stepper_step)
layout.addRow(self.link_step_sizes)
self.fast_step.valueChanged.connect(self._sync_step_sizes)
self.link_step_sizes.toggled.connect(self._update_link_state)
self._update_link_state(self.link_step_sizes.isChecked())
def _sync_step_sizes(self, value: float) -> None:
if self.link_step_sizes.isChecked():
self.stepper_step.blockSignals(True)
self.stepper_step.setValue(value)
self.stepper_step.blockSignals(False)
def _update_link_state(self, linked: bool) -> None:
self.stepper_step.setEnabled(not linked)
if linked:
self.stepper_step.blockSignals(True)
self.stepper_step.setValue(self.fast_step.value())
self.stepper_step.blockSignals(False)
def values(self) -> dict:
return {
"fast_step": self.fast_step.value(),
"stepper_step": self.stepper_step.value(),
"step_sizes_linked": self.link_step_sizes.isChecked(),
}
class ScanningSAXSDialog(QDialog):
"""Dialog for parameters that belong to a SAXS measurement row."""
def __init__(
self,
parent: QWidget | None = None,
*,
exposure_time: float,
fast_axis: Literal["x", "y"] = "x",
stepper_axis: Literal["x", "y"] = "y",
fast_step: float,
stepper_step: float | None = None,
linked: bool = True,
sample_name: str = "",
comment: str = "",
title: str = "Add scanning SAXS",
):
super().__init__(parent)
self.setWindowTitle(title)
self.sample_name = QLineEdit(sample_name, self)
self.exposure_time = positive_spinbox(exposure_time, single_step=0.01, parent=self)
self.fast_axis = AxisSelectorWidget("Fast", axis_is_x=fast_axis == "x")
self.stepper_axis = AxisSelectorWidget("Stepper", axis_is_x=stepper_axis == "x")
self.fast_axis.axis_checkbox.toggled.connect(self._sync_stepper_axis)
self.stepper_axis.axis_checkbox.toggled.connect(self._sync_fast_axis)
self.step_sizes = ScanStepWidget(fast_step, stepper_step, linked=linked)
self.comment = QTextEdit(comment, self)
self.comment.setFixedHeight(70)
form = QFormLayout()
form.addRow("Sample name", self.sample_name)
form.addRow("Exposure time", self.exposure_time)
form.addRow("Fast axis", self.fast_axis)
form.addRow("Stepper axis", self.stepper_axis)
form.addRow("Step sizes", self.step_sizes)
form.addRow("Comment", self.comment)
buttons = QDialogButtonBox(
QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel, parent=self
)
buttons.accepted.connect(self.accept)
buttons.rejected.connect(self.reject)
layout = QVBoxLayout(self)
layout.addLayout(form)
layout.addWidget(buttons)
self._sync_stepper_axis(self.fast_axis.is_x())
def values(self) -> dict:
values = {
"sample_name": self.sample_name.text(),
"exposure_time": self.exposure_time.value(),
"comment": self.comment.toPlainText(),
"fast_axis": "x" if self.fast_axis.is_x() else "y",
"stepper_axis": "x" if self.stepper_axis.is_x() else "y",
}
values.update(self.step_sizes.values())
return values
def _sync_stepper_axis(self, fast_axis_is_x: bool) -> None:
self.stepper_axis.set_is_x(not fast_axis_is_x)
def _sync_fast_axis(self, stepper_axis_is_x: bool) -> None:
self.fast_axis.set_is_x(not stepper_axis_is_x)
class TensorSAXSDialog(ScanningSAXSDialog):
"""Dialog for SAXS tensor tomography row generation."""
def __init__(
self,
parent: QWidget | None = None,
*,
exposure_time: float,
fast_axis: Literal["x", "y"] = "x",
stepper_axis: Literal["x", "y"] = "y",
fast_step: float,
stepper_step: float | None = None,
):
super().__init__(
parent,
exposure_time=exposure_time,
fast_axis=fast_axis,
stepper_axis=stepper_axis,
fast_step=fast_step,
stepper_step=stepper_step,
title="Add SAXS tensor tomography",
)
self.generated_roty = spinbox(value=0.0, single_step=1.0, parent=self)
self.angle_source_combo = QComboBox(self)
self.angle_source_combo.addItems(["Generate range", "Manual lists", "Load file"])
self.rotx_start = spinbox(value=0.0, single_step=1.0, parent=self)
self.rotx_stop = spinbox(value=180.0, single_step=1.0, parent=self)
self.rotx_count = QSpinBox(self)
self.rotx_count.setRange(1, 10_000)
self.rotx_count.setValue(19)
range_widget = QWidget(self)
range_form = QFormLayout(range_widget)
range_form.setContentsMargins(0, 0, 0, 0)
range_form.addRow("RotX start", self.rotx_start)
range_form.addRow("RotX stop", self.rotx_stop)
range_form.addRow("N angles", self.rotx_count)
range_form.addRow("RotY", self.generated_roty)
self.manual_rotx_values = QLineEdit(self)
self.manual_rotx_values.setPlaceholderText("0, 10, 20, 30")
self.manual_roty_values = QLineEdit(self)
self.manual_roty_values.setPlaceholderText("0 or 0, 1, 2, 3")
manual_widget = QWidget(self)
manual_form = QFormLayout(manual_widget)
manual_form.setContentsMargins(0, 0, 0, 0)
manual_form.addRow("RotX values", self.manual_rotx_values)
manual_form.addRow("RotY values", self.manual_roty_values)
file_widget = QWidget(self)
file_layout = QVBoxLayout(file_widget)
file_layout.setContentsMargins(0, 0, 0, 0)
file_row = QHBoxLayout()
self.angle_file_path = QLineEdit(self)
self.angle_file_path.setPlaceholderText("CSV, JSON, YAML")
self.angle_file_button = QPushButton("Browse", self)
file_row.addWidget(self.angle_file_path)
file_row.addWidget(self.angle_file_button)
file_layout.addLayout(file_row)
self.angle_file_preview = QLabel("No angle file selected.", self)
self.angle_file_preview.setWordWrap(True)
file_layout.addWidget(self.angle_file_preview)
self.rotx_stack = QStackedWidget(self)
self.rotx_stack.addWidget(range_widget)
self.rotx_stack.addWidget(manual_widget)
self.rotx_stack.addWidget(file_widget)
self.angle_source_combo.currentIndexChanged.connect(self.rotx_stack.setCurrentIndex)
self.angle_file_button.clicked.connect(self._browse_angle_file)
self.angle_file_path.textChanged.connect(self._update_angle_file_preview)
extra = QGroupBox("Tensor tomography", self)
extra_form = QFormLayout(extra)
extra_form.addRow("Angle source", self.angle_source_combo)
extra_form.addRow(self.rotx_stack)
self.layout().insertWidget(1, extra)
def rotx_values(self) -> list[float]:
return [rotx for rotx, _ in self.angle_pairs()]
def angle_pairs(self) -> list[tuple[float, float]]:
if self.angle_source_combo.currentIndex() == 0:
return pair_angle_lists(
linspace_values(
self.rotx_start.value(), self.rotx_stop.value(), self.rotx_count.value()
),
[self.generated_roty.value()],
)
if self.angle_source_combo.currentIndex() == 1:
rotx_values = parse_angle_text(self.manual_rotx_values.text())
roty_values = parse_angle_text(self.manual_roty_values.text())
return pair_angle_lists(rotx_values, roty_values or [0.0])
if not self.angle_file_path.text().strip():
raise ValueError("Choose an angle file.")
return load_angle_file(self.angle_file_path.text().strip())
def _browse_angle_file(self) -> None:
file_name, _ = QFileDialog.getOpenFileName(
self, "Load tensor angles", "", "Angle files (*.csv *.json *.yaml *.yml);;All files (*)"
)
if file_name:
self.angle_file_path.setText(file_name)
def _update_angle_file_preview(self) -> None:
file_name = self.angle_file_path.text().strip()
if not file_name:
self.angle_file_preview.setText("No angle file selected.")
return
try:
pairs = load_angle_file(file_name)
except Exception as exc: # noqa: BLE001 - user-facing validation preview
self.angle_file_preview.setText(f"Could not load angles: {exc}")
return
first = ", ".join(f"({x:g}, {y:g})" for x, y in pairs[:3])
suffix = "" if len(pairs) <= 3 else " ..."
self.angle_file_preview.setText(f"{len(pairs)} angle pairs: {first}{suffix}")
def values(self) -> dict:
values = super().values()
values["angle_pairs"] = self.angle_pairs()
return values
def accept(self) -> None:
try:
self.angle_pairs()
# File and parser libraries expose several unrelated exception hierarchies.
except Exception as exc: # noqa: BLE001
QMessageBox.warning(self, "Invalid angle values", str(exc))
return
super().accept()
class RowEditDialog(ScanningSAXSDialog):
"""Explicit edit dialog for an existing row."""
def __init__(self, parent: QWidget | None, row: SAXSRow):
super().__init__(
parent,
exposure_time=row.exposure_time,
fast_axis=row.fast_axis,
stepper_axis=row.stepper_axis,
fast_step=row.fast_step,
stepper_step=row.stepper_step,
linked=row.fast_step == row.stepper_step,
sample_name=row.sample_name,
comment=row.comment,
title="Edit SAXS row",
)
self.rotx = spinbox(value=row.rotx or 0.0, single_step=1.0, parent=self)
self.roty = spinbox(value=row.roty or 0.0, single_step=1.0, parent=self)
self.rotx.setEnabled(row.template == "tensor")
self.roty.setEnabled(row.template == "tensor")
extra = QGroupBox("Tensor angles", self)
extra_form = QFormLayout(extra)
extra_form.addRow("RotX", self.rotx)
extra_form.addRow("RotY", self.roty)
self.layout().insertWidget(1, extra)
def values(self) -> dict:
values = super().values()
values["rotx"] = self.rotx.value()
values["roty"] = self.roty.value()
return values
@@ -0,0 +1,538 @@
from __future__ import annotations
import csv
import json
import math
import re
from dataclasses import dataclass, field
from pathlib import Path
from typing import Literal
from bec_lib import messages
from bec_lib.endpoints import EndpointInfo, EndpointType, MessageOp
from csaxs_bec.bec_widgets.widgets.saxs_widget.coordinates import axis_limits
try:
import yaml
except ImportError: # pragma: no cover - depends on the local environment
yaml = None
DEFAULT_EXPOSURE_TIME = 0.05
DEFAULT_STEP_SIZE = 0.025
class MessageEndpointSAXSWidget:
"""Custom endpoint definition for the SAXS widget."""
@staticmethod
def measurement_table(session_name: str) -> EndpointInfo:
"""
Endpoint for SAXS widget measurement table updates. This endpoint is used by the device server to publish
the measurement table using a messages.VariableMessage message.
Args:
session_name (str): Session name, e.g. "session_1".
"""
endpoint = f"{EndpointType.INFO.value}/saxs_widget/sessions/{session_name}"
return EndpointInfo(
endpoint=endpoint, message_type=messages.VariableMessage, message_op=MessageOp.STREAM
)
def linspace_values(start: float, stop: float, count: int) -> list[float]:
"""Return endpoint-inclusive values without depending on numpy in the widget layer."""
if count <= 0:
raise ValueError("count must be positive")
if count == 1:
return [float(start)]
step = (stop - start) / (count - 1)
return [float(start + step * index) for index in range(count)]
def parse_angle_text(text: str) -> list[float]:
"""Parse a comma, semicolon or whitespace separated list of angles."""
tokens = [token for token in re.split(r"[\s,;]+", text.strip()) if token]
if not tokens:
return []
try:
return [float(token) for token in tokens]
except ValueError as exc:
raise ValueError("Angle values must be numeric.") from exc
def pair_angle_lists(rotx_values: list[float], roty_values: list[float] | None = None):
"""Return paired RotX/RotY angles, broadcasting a singleton axis when needed."""
roty_values = roty_values or [0.0]
if not rotx_values:
raise ValueError("At least one RotX angle is required.")
try:
if len(rotx_values) == len(roty_values):
return [(float(x), float(y)) for x, y in zip(rotx_values, roty_values)]
if len(roty_values) == 1:
return [(float(x), float(roty_values[0])) for x in rotx_values]
if len(rotx_values) == 1:
return [(float(rotx_values[0]), float(y)) for y in roty_values]
except (TypeError, ValueError) as exc:
raise ValueError("RotX and RotY angle values must be numeric.") from exc
raise ValueError(
"RotX and RotY angle lists must have the same length, or one list must have one value."
)
def normalize_angle_pairs(angle_pairs) -> list[tuple[float, float]]:
"""Validate explicit RotX/RotY angle pairs from the public API."""
if not angle_pairs:
raise ValueError("At least one angle pair is required.")
try:
pairs = [(float(rotx), float(roty)) for rotx, roty in angle_pairs]
except (TypeError, ValueError) as exc:
raise ValueError("Angle pairs must be a list of two-value RotX/RotY entries.") from exc
if not pairs:
raise ValueError("At least one angle pair is required.")
return pairs
def _coerce_angle_sequence(value) -> list[float]:
if value is None:
return []
if isinstance(value, str):
return parse_angle_text(value)
if isinstance(value, (int, float)):
return [float(value)]
try:
return [float(item) for item in value]
except (TypeError, ValueError) as exc:
raise ValueError("Angle values must be numeric.") from exc
def _mapping_value(mapping: dict, *names: str):
lowered = {str(key).lower().replace("-", "_"): value for key, value in mapping.items()}
for name in names:
if name in lowered:
return lowered[name]
return None
def _angle_pairs_from_records(records: list) -> list[tuple[float, float]]:
if not records:
raise ValueError("No angles found.")
if all(isinstance(record, dict) for record in records):
pairs = []
for record in records:
rotx = _mapping_value(record, "rotx", "rot_x", "x", "x_rotation")
roty = _mapping_value(record, "roty", "rot_y", "y", "y_rotation")
if rotx is None:
raise ValueError("Angle records must contain a RotX column/value.")
pairs.append((float(rotx), 0.0 if roty is None else float(roty)))
return pairs
if all(isinstance(record, (int, float, str)) for record in records):
return pair_angle_lists(_coerce_angle_sequence(records))
pairs = []
for record in records:
if len(record) == 0:
continue
rotx = float(record[0])
roty = float(record[1]) if len(record) > 1 and record[1] != "" else 0.0
pairs.append((rotx, roty))
if not pairs:
raise ValueError("No angles found.")
return pairs
def _angle_pairs_from_mapping(payload: dict) -> list[tuple[float, float]]:
rotx = _mapping_value(payload, "rotx", "rot_x", "x", "x_rotation")
roty = _mapping_value(payload, "roty", "rot_y", "y", "y_rotation")
if rotx is None and "angles" in payload:
return _angle_pairs_from_records(payload["angles"])
if rotx is None:
raise ValueError("Angle file must contain RotX values.")
return pair_angle_lists(_coerce_angle_sequence(rotx), _coerce_angle_sequence(roty))
def _load_csv_angle_file(path: Path) -> list[tuple[float, float]]:
with path.open(newline="", encoding="utf-8") as handle:
sample = handle.read(2048)
handle.seek(0)
has_header = csv.Sniffer().has_header(sample) if sample.strip() else False
if has_header:
return _angle_pairs_from_records(list(csv.DictReader(handle)))
return _angle_pairs_from_records([row for row in csv.reader(handle) if row])
def load_angle_file(file_name: str | Path) -> list[tuple[float, float]]:
"""Load RotX/RotY angle pairs from CSV, JSON or YAML."""
path = Path(file_name)
suffix = path.suffix.lower()
try:
if suffix == ".csv":
return _load_csv_angle_file(path)
if suffix == ".json":
with path.open(encoding="utf-8") as handle:
payload = json.load(handle)
elif suffix in {".yaml", ".yml"}:
if yaml is None:
raise ValueError("YAML angle files require PyYAML to be installed.")
with path.open(encoding="utf-8") as handle:
payload = yaml.safe_load(handle)
else:
raise ValueError("Angle files must be CSV, JSON, YAML or YML.")
except (OSError, csv.Error) as exc:
raise ValueError(f"Could not load angle file '{path}': {exc}") from exc
if isinstance(payload, dict):
return _angle_pairs_from_mapping(payload)
if isinstance(payload, list):
return _angle_pairs_from_records(payload)
raise ValueError("Unsupported angle file format.")
def _normalize_axis(axis: str) -> Literal["x", "y"]:
value = str(axis).strip().lower()
if value not in {"x", "y"}:
raise ValueError("Axis must be 'x' or 'y'.")
return value # type: ignore[return-value]
def cont_grid_point_counts(
fast_start: float,
fast_end: float,
stepper_start: float,
stepper_stop: float,
fast_step: float,
stepper_step: float,
) -> tuple[int, int, int]:
"""Match ``ContGrid.prepare_scan`` point count calculation."""
if fast_step <= 0 or stepper_step <= 0:
raise ValueError("Step sizes must be positive.")
if math.isclose(fast_start, fast_end, abs_tol=fast_step):
raise ValueError("Fast start and stop are too close for the selected step size.")
if math.isclose(stepper_start, stepper_stop, abs_tol=stepper_step):
raise ValueError("Stepper start and stop are too close for the selected step size.")
images_per_line = int(math.ceil(abs(fast_end - fast_start) / fast_step)) + 1
num_lines = int(math.ceil(abs(stepper_stop - stepper_start) / stepper_step)) + 1
return images_per_line, num_lines, images_per_line * num_lines
@dataclass(frozen=True)
class RawROI:
"""Raw ROI coordinates in image pixels, as reported by the ROI controller."""
x: float
y: float
width: float
height: float
@dataclass(frozen=True)
class CameraCenter:
"""Motor coordinates that correspond to the camera reference point."""
x: float = 0.0
y: float = 0.0
@dataclass(frozen=True)
class Correction:
"""Corrections to convert raw ROI coordinates to beam positions."""
camera_scale_pixel_x: float
offset_cam_xray_x: float
direction_x: int
camera_scale_pixel_y: float
offset_cam_xray_y: float
direction_y: int
reference_x: float = 0.0
reference_y: float = 0.0
@dataclass(frozen=True)
class ScanGeometry:
"""Scan geometry in physical units, after applying corrections to the raw ROI."""
fast_axis: Literal["x", "y"]
fast_start: float
fast_stop: float
fast_step: float
stepper_axis: Literal["x", "y"]
stepper_start: float
stepper_stop: float
stepper_step: float
images_per_line: int
num_lines: int
total_frames: int
@property
def grid_label(self) -> str:
return f"{self.images_per_line} x {self.num_lines} = {self.total_frames}"
@property
def x_start(self) -> float:
return self.fast_start if self.fast_axis == "x" else self.stepper_start
@property
def x_stop(self) -> float:
return self.fast_stop if self.fast_axis == "x" else self.stepper_stop
@property
def y_start(self) -> float:
return self.fast_start if self.fast_axis == "y" else self.stepper_start
@property
def y_stop(self) -> float:
return self.fast_stop if self.fast_axis == "y" else self.stepper_stop
@dataclass(frozen=True)
class ScanResult:
"""Result of a scan, including its number and success status."""
scan_number: str
success: bool
@dataclass
class SAXSRow:
"""A row in the SAXS scan table, representing a single scan."""
sample_name: str
sample_id: str
raw_roi: RawROI
correction: Correction
geometry: ScanGeometry
exposure_time: float
camera_center: CameraCenter
comment: str = ""
template: Literal["scanning", "tensor"] = "scanning"
rotx: float | None = None
roty: float | None = None
x_positioner: str = ""
y_positioner: str = ""
rotx_axis: str = ""
roty_axis: str = ""
scan_results: list[ScanResult] = field(default_factory=list)
@property
def fast_axis(self) -> Literal["x", "y"]:
return self.geometry.fast_axis
@property
def stepper_axis(self) -> Literal["x", "y"]:
return self.geometry.stepper_axis
@property
def fast_start(self) -> float:
return self.geometry.fast_start
@property
def fast_stop(self) -> float:
return self.geometry.fast_stop
@property
def fast_step(self) -> float:
return self.geometry.fast_step
@property
def stepper_start(self) -> float:
return self.geometry.stepper_start
@property
def stepper_stop(self) -> float:
return self.geometry.stepper_stop
@property
def stepper_step(self) -> float:
return self.geometry.stepper_step
@property
def step_size_x(self) -> float:
return self.fast_step if self.fast_axis == "x" else self.stepper_step
@property
def step_size_y(self) -> float:
return self.fast_step if self.fast_axis == "y" else self.stepper_step
@property
def fast_positioner(self) -> str:
return self.x_positioner if self.fast_axis == "x" else self.y_positioner
@property
def stepper_positioner(self) -> str:
return self.y_positioner if self.fast_axis == "x" else self.x_positioner
def _flat_payload(self) -> dict:
"""Return one canonical flat payload for CSV and broker transport."""
geometry = self.geometry
scan_results_json = json.dumps(
[
{"scan_number": result.scan_number, "success": result.success}
for result in self.scan_results
],
separators=(",", ":"),
)
return {
"template": self.template,
"sample_name": self.sample_name,
"sample_id": self.sample_id,
"exposure_time": self.exposure_time,
"comment": self.comment,
"fast_axis": self.fast_positioner or geometry.fast_axis,
"fast_start": geometry.fast_start,
"fast_stop": geometry.fast_stop,
"fast_step": geometry.fast_step,
"stepper_axis": self.stepper_positioner or geometry.stepper_axis,
"stepper_start": geometry.stepper_start,
"stepper_stop": geometry.stepper_stop,
"stepper_step": geometry.stepper_step,
"images_per_line": geometry.images_per_line,
"num_lines": geometry.num_lines,
"total_frames": geometry.total_frames,
"fast_axis_coordinate": geometry.fast_axis,
"stepper_axis_coordinate": geometry.stepper_axis,
"x_positioner": self.x_positioner,
"y_positioner": self.y_positioner,
"rotx_axis": self.rotx_axis,
"roty_axis": self.roty_axis,
"rotx": self.rotx,
"roty": self.roty,
"camera_center_x": self.camera_center.x,
"camera_center_y": self.camera_center.y,
"raw_roi_x": self.raw_roi.x,
"raw_roi_y": self.raw_roi.y,
"raw_roi_width": self.raw_roi.width,
"raw_roi_height": self.raw_roi.height,
"correction_camera_scale_pixel_x": self.correction.camera_scale_pixel_x,
"correction_offset_cam_xray_x": self.correction.offset_cam_xray_x,
"correction_direction_x": self.correction.direction_x,
"correction_camera_scale_pixel_y": self.correction.camera_scale_pixel_y,
"correction_offset_cam_xray_y": self.correction.offset_cam_xray_y,
"correction_direction_y": self.correction.direction_y,
"correction_reference_x": self.correction.reference_x,
"correction_reference_y": self.correction.reference_y,
"scan_results": scan_results_json,
# Convenience aliases for scan macro consumers.
"step_size_x": self.step_size_x,
"step_size_y": self.step_size_y,
}
def to_message_dict(self) -> dict:
"""Return a transport-safe flat payload for broker transport."""
return self._flat_payload()
def to_dict(self) -> dict:
return self.to_message_dict()
def to_csv_dict(self) -> dict:
"""Return a flat dict suitable for CSV export."""
return self._flat_payload()
@staticmethod
def help() -> dict[str, str]:
"""Return human-readable documentation for all exported payload fields."""
return {
"template": "Row template type, for example scanning or tensor.",
"sample_name": "User-provided sample label.",
"sample_id": "Generated short unique ID for this row.",
"exposure_time": "Exposure time in seconds.",
"comment": "Optional free-text comment.",
"fast_axis": "Resolved motor name used as fast axis, for example samx or samy.",
"fast_start": "Physical start position of the fast axis.",
"fast_stop": "Physical stop position of the fast axis.",
"fast_step": "Step size used along the fast axis.",
"stepper_axis": "Resolved motor name used as stepper axis.",
"stepper_start": "Physical start position of the stepper axis.",
"stepper_stop": "Physical stop position of the stepper axis.",
"stepper_step": "Step size used along the stepper axis.",
"images_per_line": "Number of images acquired per fast-axis line.",
"num_lines": "Number of stepper-axis lines.",
"total_frames": "Total number of images, images_per_line multiplied by num_lines.",
"fast_axis_coordinate": "Coordinate axis letter selected as fast axis, x or y.",
"stepper_axis_coordinate": "Coordinate axis letter selected as stepper axis, x or y.",
"x_positioner": "Configured motor name representing the X coordinate.",
"y_positioner": "Configured motor name representing the Y coordinate.",
"rotx_axis": "Configured rotation motor name for RotX.",
"roty_axis": "Configured rotation motor name for RotY.",
"rotx": "RotX angle value for this row, null for non-tensor rows.",
"roty": "RotY angle value for this row, null for non-tensor rows.",
"camera_center_x": "X motor position that corresponded to the image reference point when the row was created.",
"camera_center_y": "Y motor position that corresponded to the image reference point when the row was created.",
"raw_roi_x": "Raw ROI X origin in image coordinates.",
"raw_roi_y": "Raw ROI Y origin in image coordinates.",
"raw_roi_width": "Raw ROI width in image coordinates.",
"raw_roi_height": "Raw ROI height in image coordinates.",
"correction_camera_scale_pixel_x": "Camera pixel-to-motor scale applied from ROI X to physical X.",
"correction_offset_cam_xray_x": "Calibrated X offset from the camera reference point to the x-ray beam.",
"correction_direction_x": "Direction sign for X mapping, typically +1 or -1.",
"correction_camera_scale_pixel_y": "Camera pixel-to-motor scale applied from ROI Y to physical Y.",
"correction_offset_cam_xray_y": "Calibrated Y offset from the camera reference point to the x-ray beam.",
"correction_direction_y": "Direction sign for Y mapping, typically +1 or -1.",
"correction_reference_x": "Image coordinate used as the camera reference point for X.",
"correction_reference_y": "Image coordinate used as the camera reference point for Y.",
"scan_results": "JSON string list of scan result objects with scan_number and success.",
"step_size_x": "Convenience alias of step size in X coordinates.",
"step_size_y": "Convenience alias of step size in Y coordinates.",
}
def corrected_geometry(
raw_roi: RawROI,
correction: Correction,
fast_axis: Literal["x", "y"] = "x",
fast_step: float = DEFAULT_STEP_SIZE,
stepper_axis: Literal["x", "y"] = "y",
stepper_step: float = DEFAULT_STEP_SIZE,
*,
camera_center: CameraCenter | tuple[float, float],
) -> ScanGeometry:
fast_axis = _normalize_axis(fast_axis)
stepper_axis = _normalize_axis(stepper_axis)
if fast_axis == stepper_axis:
raise ValueError("Fast axis and stepper axis must be different.")
if isinstance(camera_center, CameraCenter):
center_x, center_y = camera_center.x, camera_center.y
else:
center_x, center_y = camera_center
def roi_axis_limits(axis: Literal["x", "y"]) -> tuple[float, float]:
if axis == "x":
return axis_limits(
raw_roi.x,
raw_roi.width,
reference=correction.reference_x,
camera_scale_pixel=correction.camera_scale_pixel_x,
direction=correction.direction_x,
camera_center=center_x,
offset_cam_xray=correction.offset_cam_xray_x,
)
return axis_limits(
raw_roi.y,
raw_roi.height,
reference=correction.reference_y,
camera_scale_pixel=correction.camera_scale_pixel_y,
direction=correction.direction_y,
camera_center=center_y,
offset_cam_xray=correction.offset_cam_xray_y,
)
fast_start, fast_stop = roi_axis_limits(fast_axis)
stepper_start, stepper_stop = roi_axis_limits(stepper_axis)
images_per_line, num_lines, total_frames = cont_grid_point_counts(
fast_start, fast_stop, stepper_start, stepper_stop, fast_step, stepper_step
)
return ScanGeometry(
fast_axis=fast_axis,
fast_start=fast_start,
fast_stop=fast_stop,
fast_step=fast_step,
stepper_axis=stepper_axis,
stepper_start=stepper_start,
stepper_stop=stepper_stop,
stepper_step=stepper_step,
images_per_line=images_per_line,
num_lines=num_lines,
total_frames=total_frames,
)
@@ -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.saxs_widget.saxs_widget_plugin import SAXSWidgetPlugin
QPyDesignerCustomWidgetCollection.addCustomWidget(SAXSWidgetPlugin())
if __name__ == "__main__": # pragma: no cover
main()
File diff suppressed because it is too large Load Diff
@@ -0,0 +1 @@
{'files': ['saxs_widget.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.saxs_widget.saxs_widget import SAXSWidget
DOM_XML = """
<ui language='c++'>
<widget class='SAXSWidget' name='saxs_widget'>
</widget>
</ui>
"""
class SAXSWidgetPlugin(QDesignerCustomWidgetInterface): # pragma: no cover
def __init__(self):
super().__init__()
self._form_editor = None
def createWidget(self, parent):
if parent is None:
return QWidget()
t = SAXSWidget(parent)
return t
def domXml(self):
return DOM_XML
def group(self):
return ""
def icon(self):
return designer_material_icon(SAXSWidget.ICON_NAME)
def includeFile(self):
return "saxs_widget"
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 "SAXSWidget"
def toolTip(self):
return "Widget for preparing SAXS measurement tables from an image ROI."
def whatsThis(self):
return self.toolTip()
@@ -0,0 +1,393 @@
from __future__ import annotations
import html
from bec_qthemes import material_icon
from qtpy.QtCore import QSize, Qt, Signal
from qtpy.QtGui import QDrag, QKeySequence, QShortcut
from qtpy.QtWidgets import (
QAbstractItemView,
QFrame,
QHBoxLayout,
QHeaderView,
QLabel,
QPushButton,
QTableWidget,
QTableWidgetItem,
QVBoxLayout,
QWidget,
)
from csaxs_bec.bec_widgets.widgets.saxs_widget.models import SAXSRow, ScanResult
class SAXSMeasurementTable(QTableWidget):
"""QTableWidget variant that reports internal row drag/drop as a row order."""
rows_reordered = Signal(list)
def __init__(self, rows: int, columns: int, parent=None):
super().__init__(rows, columns, parent)
self.setDragEnabled(True)
self.setAcceptDrops(True)
self.viewport().setAcceptDrops(True)
self.setDragDropMode(QAbstractItemView.DragDropMode.InternalMove)
self.setDefaultDropAction(Qt.DropAction.MoveAction)
self.setDragDropOverwriteMode(False)
self.setDropIndicatorShown(True)
def dropEvent(self, event) -> None:
order = self.row_order_after_drop(
self.rowCount(), self._selected_row_numbers(), self._drop_row(event)
)
if order is None or order == list(range(self.rowCount())):
event.ignore()
return
self.rows_reordered.emit(order)
event.acceptProposedAction()
def startDrag(self, _supported_actions) -> None:
indexes = self.selectedIndexes()
if not indexes:
return
mime_data = self.model().mimeData(indexes)
if mime_data is None:
return
drag = QDrag(self)
drag.setMimeData(mime_data)
if hasattr(drag, "exec"):
drag.exec(Qt.DropAction.MoveAction)
else: # pragma: no cover - Qt5 compatibility
drag.exec_(Qt.DropAction.MoveAction)
def _selected_row_numbers(self) -> list[int]:
return sorted({index.row() for index in self.selectionModel().selectedRows()})
def _drop_row(self, event) -> int:
point = self._drop_position(event)
row = self.indexAt(point).row()
if row < 0:
return self.rowCount()
if self.dropIndicatorPosition() == QAbstractItemView.DropIndicatorPosition.BelowItem:
return row + 1
if self.dropIndicatorPosition() == QAbstractItemView.DropIndicatorPosition.OnViewport:
return self.rowCount()
return row
@staticmethod
def _drop_position(event):
if hasattr(event, "position"):
position = event.position()
if hasattr(position, "toPoint"):
return position.toPoint()
return position
return event.pos()
@staticmethod
def row_order_after_drop(
row_count: int, source_rows: list[int], drop_row: int
) -> list[int] | None:
source_rows = sorted({row for row in source_rows if 0 <= row < row_count})
if not source_rows:
return None
drop_row = max(0, min(drop_row, row_count))
source_row_set = set(source_rows)
remaining = [row for row in range(row_count) if row not in source_row_set]
insert_at = drop_row - sum(row < drop_row for row in source_rows)
insert_at = max(0, min(insert_at, len(remaining)))
return remaining[:insert_at] + source_rows + remaining[insert_at:]
class SAXSTableWidget(QWidget):
"""Table and row-level actions for prepared SAXS measurements."""
HEADERS = [
"Sample",
"ID",
"Fast start",
"Fast stop",
"Stepper start",
"Stepper stop",
"Exp",
"Fast step",
"Stepper step",
"Images / line",
"RotX",
"RotY",
"Comment",
"Scan results",
]
edit_requested = Signal(int)
clear_selected_requested = Signal()
clear_all_requested = Signal()
save_requested = Signal()
upload_requested = Signal()
store_csv_requested = Signal()
add_scanning_requested = Signal()
add_tensor_requested = Signal()
move_to_selected_requested = Signal(int)
rows_reordered = Signal(list)
def __init__(self, parent=None):
super().__init__(parent)
self._rows: list[SAXSRow] = []
self.table = SAXSMeasurementTable(0, len(self.HEADERS), self)
self.table.setObjectName("saxs_measurement_table")
self.table.setHorizontalHeaderLabels(self.HEADERS)
self.table.setSelectionBehavior(QAbstractItemView.SelectionBehavior.SelectRows)
self.table.setSelectionMode(QAbstractItemView.SelectionMode.ExtendedSelection)
self.table.setEditTriggers(QAbstractItemView.EditTrigger.NoEditTriggers)
self.table.setCornerButtonEnabled(False)
self.table.setShowGrid(False)
self.table.setTextElideMode(Qt.TextElideMode.ElideRight)
self.table.setWordWrap(False)
self.table.verticalHeader().setVisible(False)
self.table.verticalHeader().setDefaultSectionSize(28)
self.table.verticalHeader().setMinimumSectionSize(24)
header = self.table.horizontalHeader()
header.setHighlightSections(False)
header.setMinimumSectionSize(70)
header.setSectionResizeMode(QHeaderView.ResizeMode.ResizeToContents)
header.setSectionResizeMode(13, QHeaderView.ResizeMode.Stretch)
self.table.setAlternatingRowColors(True)
layout = QVBoxLayout(self)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(8)
layout.addWidget(self.table, 1)
layout.addWidget(self._build_controls())
self.table.selectionModel().selectionChanged.connect(self._update_action_state)
self.table.cellDoubleClicked.connect(self._edit_row_from_double_click)
self.table.rows_reordered.connect(self.rows_reordered.emit)
self.delete_shortcut = QShortcut(QKeySequence.StandardKey.Delete, self.table)
self.delete_shortcut.activated.connect(self.clear_selected_requested.emit)
self.delete_shortcut2 = QShortcut(QKeySequence.StandardKey.Backspace, self.table)
self.delete_shortcut2.activated.connect(self.clear_selected_requested.emit)
self._update_action_state()
def set_rows(self, rows: list[SAXSRow]) -> None:
self._rows = rows
self.table.setRowCount(len(rows))
for row_index, row in enumerate(rows):
values = [
row.sample_name,
row.sample_id[:6],
self._format_float(row.fast_start),
self._format_float(row.fast_stop),
self._format_float(row.stepper_start),
self._format_float(row.stepper_stop),
self._format_float(row.exposure_time),
self._format_float(row.fast_step),
self._format_float(row.stepper_step),
row.geometry.grid_label,
"" if row.rotx is None else self._format_float(row.rotx),
"" if row.roty is None else self._format_float(row.roty),
row.comment,
]
for col, value in enumerate(values):
self.table.setItem(row_index, col, self._table_item(value))
self.table.setCellWidget(row_index, 13, self._scan_results_widget(row.scan_results))
self._update_action_state()
def selected_rows(self) -> list[int]:
return sorted({index.row() for index in self.table.selectionModel().selectedRows()})
def selected_model_rows(self) -> list[SAXSRow]:
return [self._rows[index] for index in self.selected_rows()]
def _build_controls(self) -> QWidget:
bar = QFrame(self)
bar.setObjectName("saxs_table_controls")
bar.setFrameShape(QFrame.Shape.NoFrame)
bar.setAttribute(Qt.WidgetAttribute.WA_StyledBackground, True)
layout = QHBoxLayout(bar)
layout.setContentsMargins(10, 7, 10, 7)
layout.setSpacing(8)
add_label = QLabel("Add", bar)
add_label.setObjectName("saxs_toolbar_label")
self.add_scanning_button = QPushButton("Scanning SAXS", bar)
self.add_tensor_button = QPushButton("Tensor tomography", bar)
table_label = QLabel("Table", bar)
table_label.setObjectName("saxs_toolbar_label")
self.row_summary_label = QLabel("0 rows", bar)
self.row_summary_label.setObjectName("saxs_row_summary")
rows_label = QLabel("Rows", bar)
rows_label.setObjectName("saxs_toolbar_label")
output_label = QLabel("Output", bar)
output_label.setObjectName("saxs_toolbar_label")
self.edit_button = QPushButton("Edit", bar)
self.move_to_selected_button = QPushButton("Move to selected", bar)
self.clear_selected_button = QPushButton("Remove", bar)
self.clear_all_button = QPushButton("Clear all", bar)
self.save_button = QPushButton("Save", bar)
self.store_csv_button = QPushButton("Store CSV", bar)
self.status_label = QLabel("Ready", bar)
self.status_label.setObjectName("saxs_table_status")
self.status_label.setAlignment(Qt.AlignmentFlag.AlignRight | Qt.AlignmentFlag.AlignVCenter)
self.status_label.setMinimumWidth(120)
self.add_scanning_button.setToolTip("Create one scanning SAXS row from the current ROI.")
self.add_tensor_button.setToolTip(
"Create tensor tomography rows from the current ROI and angle definition."
)
self.edit_button.setToolTip(
"Edit the selected row.\nHint: double-click a row to open the editor directly."
)
self.move_to_selected_button.setToolTip(
"Move the X/Y positioners to the saved motor positions from when the row "
"was created and restore its saved ROI."
)
self.clear_selected_button.setToolTip(
"Remove all selected rows from the table.\nHint: use Delete or Backspace."
)
self.clear_all_button.setToolTip("Remove every row currently listed in the table.")
self.save_button.setToolTip("Save the current table state for the active session.")
self.store_csv_button.setToolTip(
"Export selected rows or the full table into CSV and save the active session state."
)
self.add_scanning_button.setIcon(material_icon("add", size=(16, 16)))
self.add_tensor_button.setIcon(material_icon("add", size=(16, 16)))
self.edit_button.setIcon(material_icon("edit", size=(16, 16)))
self.move_to_selected_button.setIcon(material_icon("my_location", size=(16, 16)))
self.clear_selected_button.setIcon(material_icon("delete", size=(16, 16)))
self.clear_all_button.setIcon(material_icon("clear_all", size=(16, 16)))
self.save_button.setIcon(material_icon("save", size=(16, 16)))
self.store_csv_button.setIcon(material_icon("download", size=(16, 16)))
self.add_scanning_button.clicked.connect(self.add_scanning_requested.emit)
self.add_tensor_button.clicked.connect(self.add_tensor_requested.emit)
self.edit_button.clicked.connect(self._edit_selected)
self.move_to_selected_button.clicked.connect(self._move_to_selected)
self.clear_selected_button.clicked.connect(self.clear_selected_requested.emit)
self.clear_all_button.clicked.connect(self.clear_all_requested.emit)
self.save_button.clicked.connect(self.save_requested.emit)
self.store_csv_button.clicked.connect(self.store_csv_requested.emit)
for button in [
self.add_scanning_button,
self.add_tensor_button,
self.edit_button,
self.move_to_selected_button,
self.clear_selected_button,
self.clear_all_button,
self.save_button,
self.store_csv_button,
]:
button.setProperty("saxsButtonRole", "toolbar")
button.setMinimumHeight(34)
button.setIconSize(QSize(18, 18))
button.setCursor(Qt.CursorShape.PointingHandCursor)
button.setLayoutDirection(Qt.LayoutDirection.LeftToRight)
button.setFlat(True)
self.clear_selected_button.setProperty("saxsButtonRole", "danger")
self.clear_all_button.setProperty("saxsButtonRole", "danger")
self.add_scanning_button.setMinimumWidth(128)
self.add_tensor_button.setMinimumWidth(150)
for button in [
self.edit_button,
self.move_to_selected_button,
self.clear_selected_button,
self.clear_all_button,
self.save_button,
self.store_csv_button,
]:
button.setMinimumWidth(84)
self.move_to_selected_button.setMinimumWidth(128)
layout.addWidget(add_label)
layout.addWidget(self.add_scanning_button)
layout.addWidget(self.add_tensor_button)
layout.addWidget(self._vertical_separator(bar))
layout.addWidget(table_label)
layout.addWidget(self.row_summary_label)
layout.addStretch()
layout.addWidget(self._vertical_separator(bar))
layout.addWidget(rows_label)
layout.addWidget(self.edit_button)
layout.addWidget(self.move_to_selected_button)
layout.addWidget(self.clear_selected_button)
layout.addWidget(self.clear_all_button)
layout.addWidget(self._vertical_separator(bar))
layout.addWidget(output_label)
layout.addWidget(self.save_button)
layout.addWidget(self.store_csv_button)
layout.addWidget(self.status_label)
return bar
@staticmethod
def _vertical_separator(parent: QWidget) -> QFrame:
separator = QFrame(parent)
separator.setObjectName("saxs_toolbar_separator")
separator.setFrameShape(QFrame.Shape.VLine)
separator.setFrameShadow(QFrame.Shadow.Sunken)
return separator
def _edit_selected(self) -> None:
selected = self.selected_rows()
if len(selected) == 1:
self.edit_requested.emit(selected[0])
def _move_to_selected(self) -> None:
selected = self.selected_rows()
if len(selected) == 1:
self.move_to_selected_requested.emit(selected[0])
def _edit_row_from_double_click(self, row: int, _column: int) -> None:
self.table.selectRow(row)
self.edit_requested.emit(row)
def _update_action_state(self, *_args) -> None:
selected_count = len(self.selected_rows())
row_count = len(self._rows)
if selected_count:
self.row_summary_label.setText(f"{selected_count} selected / {row_count} rows")
else:
self.row_summary_label.setText(f"{row_count} rows")
self.edit_button.setEnabled(selected_count == 1)
self.move_to_selected_button.setEnabled(selected_count == 1)
self.clear_selected_button.setEnabled(selected_count > 0)
self.clear_all_button.setEnabled(row_count > 0)
self.save_button.setEnabled(row_count > 0)
self.store_csv_button.setEnabled(row_count > 0)
@staticmethod
def _table_item(value: str) -> QTableWidgetItem:
item = QTableWidgetItem(str(value))
item.setFlags(
Qt.ItemFlag.ItemIsEnabled
| Qt.ItemFlag.ItemIsSelectable
| Qt.ItemFlag.ItemIsDragEnabled
| Qt.ItemFlag.ItemIsDropEnabled
)
item.setToolTip(str(value))
return item
@staticmethod
def _scan_results_widget(results: list[ScanResult]) -> QLabel:
label = QLabel()
label.setContentsMargins(4, 0, 4, 0)
label.setTextFormat(Qt.TextFormat.RichText)
label.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse)
parts = []
for result in results:
color = "#2e7d32" if result.success else "#c62828"
text = html.escape(str(result.scan_number))
parts.append(f'<span style="color:{color}; font-weight:600;">{text}</span>')
label.setText(", ".join(parts))
return label
@staticmethod
def _format_float(value: float) -> str:
return f"{value:.6g}"
@@ -0,0 +1,30 @@
"""
This macro is used to fetch saved data for a dataset/session created by the SAXS widget.
It returns a Varible message containing 'value' and 'metadata' fields. The 'metadata'
contains some context information, while 'value' contains a list of dictionaries, each
representing a row in the table. The order is kept as exported from the table.
Scan information can be fetched from this dictionary to setup the cont grid scan.
"""
from __future__ import annotations
from csaxs_bec.bec_widgets.widgets.saxs_widget.models import MessageEndpointSAXSWidget
def get_saxs_widget_table_data(session_name: str, count: int = -1):
if not isinstance(count, int):
raise TypeError("count must be an integer stream index.")
all_msgs = []
try:
all_msgs = bec.connector.xread(
MessageEndpointSAXSWidget.measurement_table(session_name), from_start=True
)
msg = all_msgs[count]["value"]
except (IndexError, KeyError, TypeError) as exc:
raise ValueError(
f"No data found for session '{session_name}' and index {count}. Total length of all_msgs: {len(all_msgs)}"
) from exc
print(
f"Retrieved data for session '{msg.metadata.get('session_name')}' with {len(msg.value)} rows; Created at {msg.metadata.get('timestamp')}"
)
return msg
@@ -0,0 +1,133 @@
"""
This macro is used to run a cont_grid scan for a single row of the data provided by the SAXS widget.
These are fetched through the get_saxs_widget_table_data macro, which returns a Variable message containing
'value' and 'metadata' fields. 'value' contains a list of dictionaries, each representing a row in the table.
This dictionary is the input for this macro.
"""
from __future__ import annotations
import numpy as np
from bec_lib.scan_repeat import scan_repeat
# TODO Default True here means that ANY error will be retried up to 3 times. This is dangerous in the sense that
# if the error is due to a misconfiguration of the scan parameters, it will just keep retrying and failing.
# Please contact the BEC team if you want to establish a dedicated error handling for this macro.
@scan_repeat(max_repeats=3, default=True)
def run_cont_grid_scan_for_table_row(
session_name: str, row: dict, phone_numbers: list[str] | None = None
) -> None:
"""
Run a continuous grid scan for a specific row in the SAXS widget table.
Args:
row (dict): A dictionary representing a row in the SAXS widget table. This should contain all necessary
parameters for the continuous grid scan.
phone_numbers (list[str] | None): A list of phone numbers to notify in case of errors. Defaults to None.
"""
metadata = {
"session_name": session_name,
"sample_name": row["sample_name"],
"sample_id": row["sample_id"],
"comment": row["comment"],
"raw_roi": {
"x": row["raw_roi_x"],
"y": row["raw_roi_y"],
"width": row["raw_roi_width"],
"height": row["raw_roi_height"],
},
"camera_center": {"x": row["camera_center_x"], "y": row["camera_center_y"]},
"correction_camera_scale_pixel_x": row["correction_camera_scale_pixel_x"],
"correction_offset_cam_xray_x": row["correction_offset_cam_xray_x"],
"grid_scan_params": {
"fast_axis": row["fast_axis"],
"fast_start": row["fast_start"],
"fast_stop": row["fast_stop"],
"fast_step": row["fast_step"],
"stepper_axis": row["stepper_axis"],
"stepper_start": row["stepper_start"],
"stepper_stop": row["stepper_stop"],
"stepper_step": row["stepper_step"],
"exposure_time": row["exposure_time"],
"images_per_line": row["images_per_line"],
"num_lines": row["num_lines"],
},
}
if row["template"] == "tensor":
metadata.update(
{
"roty_axis": row["roty_axis"],
"roty": row["roty"],
"rotx_axis": row["rotx_axis"],
"rotx": row["rotx"],
}
)
# Move rotation motors if tensor tomography scan template is send.
if row["template"] == "tensor":
try:
# Move motors to the starting position of the scan
move_command = []
roty_motor = dev.devices[row["roty_axis"]]
setpoint_roty = roty_motor.setpoint.get(cached=True)
if not np.isclose(setpoint_roty, row["roty"]):
move_command.extend([row["roty_axis"], row["roty"]])
setpoint_rotx = dev.devices[row["rotx_axis"]].setpoint.get(cached=True)
if not np.isclose(setpoint_rotx, row["rotx"]):
move_command.extend([row["rotx_axis"], row["rotx"]])
# Wait for all motor movements to complete
if move_command:
scans.umv(*move_command, relative=False)
except Exception as e:
info_msg = (
"Error while moving motors to starting position for sample "
f"'{row['sample_name']}' (ID: {row['sample_id']})."
)
print(info_msg)
# Send message to scilog with error tag and text
msg = bec.messaging.scilog.new()
msg.add_tags("error")
msg.add_text(info_msg)
msg.add_text(f"Error details: {e}")
msg.send()
# Notify the user via SMS if phone numbers are provided
if phone_numbers:
msg = bec.messaging.signal.new()
msg.add_text(info_msg)
msg.send(scope=phone_numbers)
raise
# Scan command
try:
scans.cont_grid(
fast_axis=row["fast_axis"],
fast_start=row["fast_start"],
fast_end=row["fast_stop"],
fast_step_size=row["fast_step"],
stepper_axis=row["stepper_axis"],
stepper_start=row["stepper_start"],
stepper_stop=row["stepper_stop"],
stepper_step_size=row["stepper_step"],
exp_time=row["exposure_time"],
relative=False,
metadata=metadata,
)
except Exception as e:
info_msg = (
f"Error while executing cont_grid scan for sample '{row['sample_name']}' "
f"with ID '{row['sample_id']}'."
)
print(info_msg)
# Send message to scilog with error tag and text
msg = bec.messaging.scilog.new()
msg.add_tags("error")
msg.add_text(info_msg)
msg.add_text(f"Error details: {e}")
msg.send()
# Notify the user via SMS if phone numbers are provided
if phone_numbers:
msg = bec.messaging.signal.new()
msg.add_text(info_msg)
msg.send(scope=phone_numbers)
raise
+38
View File
@@ -0,0 +1,38 @@
# List of phone numbers to send signal notifications to, example: ["+1234567890", "+0987654321"].
# If not Swiss numbers, please provide +<country_code> prefix. If empty, no signal notifications will be sent.
#############
### Example script, copy from here and edit to your needs
#############
TELEFON_NUMBERS = []
SESSION_NAME = "test_session"
# count = -1 means latest saved table from the SAXS widget, 0 means first saved table, 1 means second saved table, etc. -1 is DEFAULT
# Provide it as a kwargs if you need to fetch older table data.
msg = get_saxs_widget_table_data(SESSION_NAME)
print(
f"Fetched {len(msg.value)} rows from SAXS widget table for session '{msg.metadata.get('session_name')}'. Table data created at {msg.metadata.get('timestamp')}"
)
total_rows = len(msg.value)
for ii, row in enumerate(msg.value):
print(f"Running continuous grid scan for row {ii + 1}/{len(msg.value)}: {row}")
run_cont_grid_scan_for_table_row(
session_name=SESSION_NAME, row=row, phone_numbers=TELEFON_NUMBERS
)
if ii % 10 == 0:
# Send update to scilog every 10 scans to avoid spamming the log
msg = bec.messaging.scilog.new()
msg.add_text(
f"Completed {ii + 1}/{total_rows} continuous grid scans for session '{SESSION_NAME}'"
)
msg.send()
# Send update to signal if phone numbers are provided
if TELEFON_NUMBERS:
msg = bec.messaging.signal.new()
msg.add_text(
f"Completed {ii + 1}/{total_rows} continuous grid scans for session '{SESSION_NAME}'"
)
msg.send(scope=TELEFON_NUMBERS)