feat(bec_widgets): add SampleStorageWidget for flOMNI sample storage

GUI for ftransfer_show / ftransfer_modify. Displays the sample stage (0), gripper (100) and magazine slots 1-20 in a 4x5 grid matching the CLI layout. Right-click per slot corrects the records (new / rename / clear), each action confirmed by its own dialog and written immediately. Bookkeeping only: writes flomni_samples directly (mirroring flomni_modify_storage_non_interactive), never moves hardware. Guarded on fsamroy; OMNY shows a not-implemented placeholder.
2026-07-07 10:44:43 +02:00
parent 12febe2e3a
commit 02373c47be
9 changed files with 564 additions and 7 deletions
@@ -13,12 +13,25 @@ logger = bec_logger.logger


 _Widgets = {
+    "SampleStorageWidget": "SampleStorageWidget",
    "SlitControlWidget": "SlitControlWidget",
    "TomoParamsWidget": "TomoParamsWidget",
    "XRayEye": "XRayEye",
 }


+class SampleStorageWidget(RPCBase):
+    """View and correct the FlOMNI sample-storage records."""
+
+    _IMPORT_MODULE = "csaxs_bec.bec_widgets.widgets.sample_storage.sample_storage"
+
+    @rpc_call
+    def refresh(self) -> "None":
+        """
+        Re-read every slot from the device and update the cells.
+        """
+
+
 class SlitControlWidget(RPCBase):
    """Interactive GUI for cSAXS slit center and size control."""

@@ -34,31 +47,31 @@ class SlitControlWidget(RPCBase):
    @rpc_call
    def move_center(self, direction: "str"):
        """
-        Move the slit center by one step. direction: up / down / left / right.
+        Move the slit center. direction: up / down / left / right.
        """

    @rpc_call
    def move_size(self, direction: "str"):
        """
-        Adjust the slit size by one step. up/right = grow, down/left = shrink.
+        Adjust slit size. up/right = grow, down/left = shrink.
        """

    @rpc_call
    def set_step(self, value: "float"):
        """
-        Set the shared step size in mm (clamped to 0.005–2.0 mm).
+        Set step size in mm (clamped to 0.005–2.0 mm).
        """

    @rpc_call
    def double_step(self):
        """
-        Double the current step size.
+        None
        """

    @rpc_call
    def halve_step(self):
        """
-        Halve the current step size.
+        None
        """


@@ -70,7 +83,7 @@ class TomoParamsWidget(RPCBase):
    @rpc_call
    def refresh(self) -> "None":
        """
-        Full refresh: params, queue, progress, sample name.
+        Full refresh: params + sample name.
        """

    @rpc_call
@@ -94,7 +107,7 @@ class TomoParamsWidget(RPCBase):
    @rpc_call
    def add_to_queue(self) -> "None":
        """
-        Snapshot the current live global-var parameters and append a new job.
+        Open queue dialog and trigger Add (also accessible via RPC).
        """


@@ -5,6 +5,10 @@ from __future__ import annotations
 # pylint: skip-file

 designer_plugins = {
+    "SampleStorageWidget": (
+        "csaxs_bec.bec_widgets.widgets.sample_storage.sample_storage",
+        "SampleStorageWidget",
+    ),
    "SlitControlWidget": (
        "csaxs_bec.bec_widgets.widgets.slit_control.slit_control",
        "SlitControlWidget",
@@ -17,6 +21,7 @@ designer_plugins = {
 }

 widget_icons = {
+    "SampleStorageWidget": "widgets",
    "SlitControlWidget": "widgets",
    "TomoParamsWidget": "widgets",
    "XRayEye": "widgets",
@@ -0,0 +1,3 @@
+from .sample_storage import SampleStorageWidget
+
+__all__ = ["SampleStorageWidget"]
@@ -0,0 +1,18 @@
+def main():  # pragma: no cover
+    """Register SampleStorageWidget with the Qt Designer.
+
+    Mirrors register_tomo_params.py — invoked by the Qt Designer's plugin
+    discovery mechanism.
+    """
+    from qtpy import QtDesigner
+
+    from csaxs_bec.bec_widgets.widgets.sample_storage.sample_storage_plugin import (
+        SampleStorageWidgetPlugin,
+    )
+
+    QPyDesignerCustomWidgetCollection = QtDesigner.QPyDesignerCustomWidgetCollection
+    QPyDesignerCustomWidgetCollection.addCustomWidget(SampleStorageWidgetPlugin())
+
+
+if __name__ == "__main__":  # pragma: no cover
+    main()
@@ -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.sample_storage.sample_storage_widget_plugin import SampleStorageWidgetPlugin
+
+    QPyDesignerCustomWidgetCollection.addCustomWidget(SampleStorageWidgetPlugin())
+
+
+if __name__ == "__main__":  # pragma: no cover
+    main()
@@ -0,0 +1,392 @@
+"""
+SampleStorageWidget – BEC widget for viewing and correcting the FlOMNI
+sample-storage bookkeeping.
+
+This widget is a GUI replacement for the ``ftransfer_show`` /
+``ftransfer_modify`` CLI commands.  It shows the sample magazine (slots
+1–20), the sample stage (slot 0) and the gripper (slot 100), and lets the
+user correct the stored records via a right-click context menu on each slot.
+
+IMPORTANT – bookkeeping only
+----------------------------
+This widget never moves hardware.  It only reads and writes the sample
+*records* held on the ``flomni_samples`` device — exactly the same records
+that ``flomni_modify_storage_non_interactive()`` writes.  It does NOT perform
+sample transfers; those remain CLI-only (``flomni.ftransfer_sample_change``
+etc.), because they are blocking hardware sequences with interactive
+confirmations.  Editing a record here changes only what BEC *believes* is in
+a slot, so entries must be kept in sync with physical reality by the operator.
+
+Because a transfer and a manual record edit would both happen at the same
+workstation, there is deliberately no software interlock against editing
+during a transfer — operator discipline covers it (nobody edits the storage
+while a transfer they started is running).
+
+Backend
+-------
+All state lives on ``dev.flomni_samples`` (not BEC global vars), which is the
+single source of truth also consulted by the transfer state machine.  The
+widget reads/writes it directly, mirroring
+``Flomni.flomni_modify_storage_non_interactive()``:
+
+  slots 0–20 : sample_placed.sample{N}.set(used)
+               sample_names.sample{N}.set(name)
+  gripper    : sample_in_gripper.set(used)
+               sample_in_gripper_name.set(name)
+
+Polling
+-------
+BEC publishes no push notifications for these signals, so a ``QTimer`` polls
+every 2 s to reflect CLI-driven changes.  Writes are immediate (on the
+affirmative button of each context-menu dialog); there is no batched
+edit/submit — every mutating action is individually confirmed instead.
+"""
+
+from __future__ import annotations
+
+from typing import Optional
+
+from bec_lib import bec_logger
+from bec_widgets import BECWidget, SafeSlot
+from qtpy.QtCore import Qt, QTimer
+from qtpy.QtWidgets import (
+    QFrame,
+    QGridLayout,
+    QHBoxLayout,
+    QInputDialog,
+    QLabel,
+    QMenu,
+    QMessageBox,
+    QSizePolicy,
+    QVBoxLayout,
+    QWidget,
+)
+
+logger = bec_logger.logger
+
+# ── constants ────────────────────────────────────────────────────────────────
+
+# Magazine slot numbers, laid out to match the CLI printout:
+#   1  2  3  4  5
+#   6  7  8  9 10
+#  11 12 13 14 15
+#  16 17 18 19 20
+STORAGE_ROWS = 4
+STORAGE_COLS = 5
+STORAGE_SLOTS = tuple(range(1, STORAGE_ROWS * STORAGE_COLS + 1))  # 1..20
+
+STAGE_SLOT = 0  # sample stage (in the beam)
+GRIPPER_SLOT = 100  # gripper pseudo-slot
+
+EMPTY_NAME = "-"  # backend convention for an empty slot's name
+EMPTY_LABEL = "— empty —"  # what an empty cell shows in the GUI
+
+# colours
+COLOR_OCCUPIED = "#1b1b1b"
+COLOR_EMPTY = "#9e9e9e"
+COLOR_STAGE_BORDER = "#2196F3"
+COLOR_GRIPPER_BORDER = "#FF9800"
+COLOR_SLOT_BORDER = "#c0c0c0"
+
+
+# ── slot cell widget ─────────────────────────────────────────────────────────
+
+
+class _SlotCell(QFrame):
+    """A single storage slot: shows its number and either the sample name or
+    an "empty" placeholder.  Right-click opens a context menu whose actions
+    depend on whether the slot is currently occupied.
+
+    The cell does not talk to the device itself; it calls back into the
+    parent widget's mutation methods, which own all backend access.
+    """
+
+    def __init__(self, slot: int, owner: "SampleStorageWidget", border_color: str):
+        super().__init__()
+        self._slot = slot
+        self._owner = owner
+        self._occupied = False
+        self._name = EMPTY_NAME
+
+        self.setFrameShape(QFrame.Shape.Box)
+        self.setLineWidth(1)
+        self.setMinimumSize(120, 60)
+        self.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding)
+        self.setStyleSheet(f"_SlotCell {{ border: 1px solid {border_color}; border-radius: 4px; }}")
+
+        layout = QVBoxLayout(self)
+        layout.setContentsMargins(6, 4, 6, 4)
+        layout.setSpacing(2)
+
+        self._lbl_num = QLabel(self._slot_caption())
+        self._lbl_num.setStyleSheet("color: #888888; font-size: 10px;")
+        self._lbl_num.setAlignment(Qt.AlignmentFlag.AlignLeft | Qt.AlignmentFlag.AlignTop)
+
+        self._lbl_name = QLabel(EMPTY_LABEL)
+        self._lbl_name.setAlignment(Qt.AlignmentFlag.AlignCenter)
+        self._lbl_name.setWordWrap(True)
+
+        layout.addWidget(self._lbl_num)
+        layout.addWidget(self._lbl_name, stretch=1)
+
+        self.setContextMenuPolicy(Qt.ContextMenuPolicy.CustomContextMenu)
+        self.customContextMenuRequested.connect(self._show_menu)
+
+    # ── labels ────────────────────────────────────────────────────────────────
+
+    def _slot_caption(self) -> str:
+        if self._slot == STAGE_SLOT:
+            return "Stage (0)"
+        if self._slot == GRIPPER_SLOT:
+            return "Gripper (100)"
+        return str(self._slot)
+
+    def set_state(self, occupied: bool, name: str) -> None:
+        """Update the cell's displayed state (called by the poll/refresh)."""
+        self._occupied = occupied
+        self._name = name if name else EMPTY_NAME
+        if occupied:
+            self._lbl_name.setText(self._name)
+            self._lbl_name.setStyleSheet(f"color: {COLOR_OCCUPIED};")
+        else:
+            self._lbl_name.setText(EMPTY_LABEL)
+            self._lbl_name.setStyleSheet(f"color: {COLOR_EMPTY}; font-style: italic;")
+
+    # ── context menu ──────────────────────────────────────────────────────────
+
+    def _show_menu(self, pos) -> None:
+        menu = QMenu(self)
+        if self._occupied:
+            act_rename = menu.addAction("Change name…")
+            act_clear = menu.addAction("Clear (set empty)")
+            chosen = menu.exec_(self.mapToGlobal(pos))
+            if chosen == act_rename:
+                self._owner.action_change_name(self._slot, self._name)
+            elif chosen == act_clear:
+                self._owner.action_clear(self._slot)
+        else:
+            act_new = menu.addAction("New sample…")
+            chosen = menu.exec_(self.mapToGlobal(pos))
+            if chosen == act_new:
+                self._owner.action_new_sample(self._slot)
+
+
+# ── main widget ──────────────────────────────────────────────────────────────
+
+
+class SampleStorageWidget(BECWidget, QWidget):
+    """
+    View and correct the FlOMNI sample-storage records.
+
+    Layout
+    ------
+      - Header: Stage (slot 0) and Gripper (slot 100), right-click correctable
+      - 4×5 grid: magazine slots 1–20, right-click correctable
+      - Caveat line: records only, not hardware
+
+    All mutations go through a right-click context menu; each mutating action
+    is confirmed by its own dialog and written to the device immediately.
+    """
+
+    PLUGIN = True
+
+    USER_ACCESS = ["refresh"]
+
+    _POLL_INTERVAL_MS = 2000
+
+    def __init__(self, parent=None, **kwargs):
+        super().__init__(parent=parent, **kwargs)
+        self.get_bec_shortcuts()
+        self._cells: dict[int, _SlotCell] = {}
+        self._flomni_available = self._check_flomni_available()
+        self._build_ui()
+        self._poll_timer = QTimer(self)
+        self._poll_timer.setInterval(self._POLL_INTERVAL_MS)
+        self._poll_timer.timeout.connect(self.refresh)
+        if self._flomni_available:
+            self._poll_timer.start()
+            self.refresh()
+
+    # ── setup detection ──────────────────────────────────────────────────────
+
+    def _check_flomni_available(self) -> bool:
+        """
+        Return True if the FlOMNI setup is active in this BEC session.
+        Uses ``fsamroy`` (the FlOMNI-specific rotation stage) as the
+        discriminator — it is absent in OMNY and LamNI sessions.
+        """
+        try:
+            return getattr(self.dev, "fsamroy", None) is not None
+        except Exception:
+            return False
+
+    # ── device access ────────────────────────────────────────────────────────
+
+    @property
+    def _samples(self):
+        """The flomni_samples device (single source of truth)."""
+        return self.dev.flomni_samples
+
+    def _read_slot(self, slot: int) -> tuple[bool, str]:
+        """Return (occupied, name) for a storage slot 0–20 or the gripper."""
+        try:
+            if slot == GRIPPER_SLOT:
+                occupied = bool(self._samples.is_sample_in_gripper())
+                name = self._samples.sample_in_gripper_name.get()
+            else:
+                occupied = bool(self._samples.is_sample_slot_used(slot))
+                name = getattr(self._samples.sample_names, f"sample{slot}").get()
+            return occupied, (name if name else EMPTY_NAME)
+        except Exception as exc:
+            logger.warning(f"SampleStorageWidget: read of slot {slot} failed: {exc}")
+            return False, EMPTY_NAME
+
+    def _write_slot(self, slot: int, used: int, name: str) -> bool:
+        """
+        Write (used, name) to a storage slot 0–20 or the gripper, mirroring
+        ``Flomni.flomni_modify_storage_non_interactive()`` exactly.
+        """
+        try:
+            if slot == GRIPPER_SLOT:
+                self._samples.sample_in_gripper.set(used)
+                self._samples.sample_in_gripper_name.set(name)
+            else:
+                getattr(self._samples.sample_placed, f"sample{slot}").set(used)
+                getattr(self._samples.sample_names, f"sample{slot}").set(name)
+            return True
+        except Exception as exc:
+            logger.warning(f"SampleStorageWidget: write of slot {slot} failed: {exc}")
+            QMessageBox.critical(
+                self,
+                "Write failed",
+                f"Could not write slot {self._slot_title(slot)} to the device:\n{exc}",
+            )
+            return False
+
+    @staticmethod
+    def _slot_title(slot: int) -> str:
+        if slot == STAGE_SLOT:
+            return "stage (0)"
+        if slot == GRIPPER_SLOT:
+            return "gripper (100)"
+        return str(slot)
+
+    # ── UI construction ───────────────────────────────────────────────────────
+
+    def _build_ui(self):
+        root = QVBoxLayout(self)
+        root.setSpacing(8)
+
+        if not self._flomni_available:
+            lbl = QLabel(
+                "⚠ FlOMNI setup not detected in this BEC session.\n\n"
+                "This widget is only meaningful when the FlOMNI\n"
+                "BEC session is active (fsamroy device present).\n\n"
+                "OMNY support is not implemented yet."
+            )
+            lbl.setAlignment(Qt.AlignmentFlag.AlignCenter)
+            lbl.setWordWrap(True)
+            root.addStretch()
+            root.addWidget(lbl)
+            root.addStretch()
+            return
+
+        # header: stage + gripper
+        header = QHBoxLayout()
+        header.setSpacing(8)
+        self._cells[STAGE_SLOT] = _SlotCell(STAGE_SLOT, self, COLOR_STAGE_BORDER)
+        self._cells[GRIPPER_SLOT] = _SlotCell(GRIPPER_SLOT, self, COLOR_GRIPPER_BORDER)
+        header.addWidget(self._cells[STAGE_SLOT])
+        header.addWidget(self._cells[GRIPPER_SLOT])
+        root.addLayout(header)
+
+        # magazine grid 1..20
+        grid_box = QFrame()
+        grid = QGridLayout(grid_box)
+        grid.setSpacing(6)
+        grid.setContentsMargins(0, 0, 0, 0)
+        for idx, slot in enumerate(STORAGE_SLOTS):
+            r, c = divmod(idx, STORAGE_COLS)
+            cell = _SlotCell(slot, self, COLOR_SLOT_BORDER)
+            self._cells[slot] = cell
+            grid.addWidget(cell, r, c)
+        root.addWidget(grid_box, stretch=1)
+
+        # caveat line
+        caveat = QLabel(
+            "Records only — this does not move any hardware. "
+            "Ensure the entries here match physical reality."
+        )
+        caveat.setWordWrap(True)
+        caveat.setStyleSheet("color: #FF9800; font-size: 11px;")
+        root.addWidget(caveat)
+
+    # ── refresh / poll ────────────────────────────────────────────────────────
+
+    @SafeSlot()
+    def refresh(self) -> None:
+        """Re-read every slot from the device and update the cells."""
+        if not self._flomni_available:
+            return
+        for slot, cell in self._cells.items():
+            occupied, name = self._read_slot(slot)
+            cell.set_state(occupied, name)
+
+    # ── mutation actions (called from _SlotCell context menu) ─────────────────
+
+    def action_change_name(self, slot: int, current_name: str) -> None:
+        """Rename an occupied slot (stays occupied)."""
+        name, ok = QInputDialog.getText(
+            self,
+            "Change sample name",
+            f"New name for slot {self._slot_title(slot)}:",
+            text=current_name,
+        )
+        if not ok:
+            return
+        name = name.strip()
+        if not name or name == EMPTY_NAME:
+            QMessageBox.warning(
+                self,
+                "Invalid name",
+                "An occupied slot needs a non-empty name.\n"
+                "Use “Clear (set empty)” to empty the slot instead.",
+            )
+            return
+        if self._write_slot(slot, 1, name):
+            self.refresh()
+
+    def action_new_sample(self, slot: int) -> None:
+        """Mark an empty slot occupied and give it a name."""
+        name, ok = QInputDialog.getText(
+            self, "New sample", f"Sample name for slot {self._slot_title(slot)}:"
+        )
+        if not ok:
+            return
+        name = name.strip()
+        if not name or name == EMPTY_NAME:
+            QMessageBox.warning(self, "Invalid name", "Please enter a non-empty sample name.")
+            return
+        if self._write_slot(slot, 1, name):
+            self.refresh()
+
+    def action_clear(self, slot: int) -> None:
+        """Mark a slot empty (destructive; confirmed)."""
+        reply = QMessageBox.question(
+            self,
+            "Clear slot",
+            f"Set slot {self._slot_title(slot)} to empty?\n\n"
+            "Confirm the sample has been physically removed — this only "
+            "changes the record, not the hardware.",
+            QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.Cancel,
+        )
+        if reply != QMessageBox.StandardButton.Yes:
+            return
+        if self._write_slot(slot, 0, EMPTY_NAME):
+            self.refresh()
+
+    # ── cleanup ───────────────────────────────────────────────────────────────
+
+    def cleanup(self) -> None:
+        self._poll_timer.stop()
+        super().cleanup()
@@ -0,0 +1,53 @@
+# pylint: disable=missing-module-docstring, missing-function-docstring
+# This file is auto-generated style boilerplate for the Qt Designer plugin,
+# mirroring the pattern used by the other csaxs_bec widget plugins
+# (e.g. tomo_params_plugin.py).  It registers SampleStorageWidget with the
+# Qt Designer so it can be dropped onto a form.
+
+from qtpy.QtDesigner import QDesignerCustomWidgetInterface
+
+from csaxs_bec.bec_widgets.widgets.sample_storage.sample_storage import SampleStorageWidget
+
+DOM_XML = """
+<ui language='c++'>
+    <widget class='SampleStorageWidget' name='sample_storage'>
+    </widget>
+</ui>
+"""
+
+
+class SampleStorageWidgetPlugin(QDesignerCustomWidgetInterface):  # pragma: no cover
+    def __init__(self):
+        super().__init__()
+        self._form_editor = None
+
+    def initialize(self, form_editor):
+        self._form_editor = form_editor
+
+    def isInitialized(self):
+        return self._form_editor is not None
+
+    def createWidget(self, parent):
+        t = SampleStorageWidget(parent)
+        return t
+
+    def name(self):
+        return "SampleStorageWidget"
+
+    def group(self):
+        return "cSAXS FlOMNI"
+
+    def toolTip(self):
+        return "View and correct the FlOMNI sample-storage records (bookkeeping only)."
+
+    def whatsThis(self):
+        return self.toolTip()
+
+    def isContainer(self):
+        return False
+
+    def domXml(self):
+        return DOM_XML
+
+    def includeFile(self):
+        return "csaxs_bec.bec_widgets.widgets.sample_storage.sample_storage"
@@ -0,0 +1 @@
+{'files': ['sample_storage.py']}
@@ -0,0 +1,57 @@
+# Copyright (C) 2022 The Qt Company Ltd.
+# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause
+
+from qtpy.QtDesigner import QDesignerCustomWidgetInterface
+from qtpy.QtWidgets import QWidget
+
+from bec_widgets.utils.bec_designer import designer_material_icon
+from csaxs_bec.bec_widgets.widgets.sample_storage.sample_storage import SampleStorageWidget
+
+DOM_XML = """
+<ui language='c++'>
+    <widget class='SampleStorageWidget' name='sample_storage_widget'>
+    </widget>
+</ui>
+"""
+
+
+class SampleStorageWidgetPlugin(QDesignerCustomWidgetInterface):  # pragma: no cover
+    def __init__(self):
+        super().__init__()
+        self._form_editor = None
+
+    def createWidget(self, parent):
+        if parent is None:
+            return QWidget()
+        t = SampleStorageWidget(parent)
+        return t
+
+    def domXml(self):
+        return DOM_XML
+
+    def group(self):
+        return ""
+
+    def icon(self):
+        return designer_material_icon(SampleStorageWidget.ICON_NAME)
+
+    def includeFile(self):
+        return "sample_storage_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 "SampleStorageWidget"
+
+    def toolTip(self):
+        return ""
+
+    def whatsThis(self):
+        return self.toolTip()