fix(omny): always route fast-shutter control through dev.fsh; carry owner/status/account across sample transfer

- shutter.py (OMNYFastShutter): drop the "use dev.fsh if configured, else
  fall back to a local dummy signal" branching -- always delegate to the
  'fsh' device. Removed the now-unused MonitorSignal class and the
  dummy shutter Cpt along with it. Rewrote test_omny_shutter.py to assert
  delegation to a mocked 'fsh' device instead of exercising the removed
  dummy fallback.
- omny_sample_transfer_mixin.py: picking a sample up (get_sample, three
  source kinds: shuttle slot, fixed position, sample stage) only carried
  the sample name into set_sample_in_gripper(), silently dropping owner
  (defaulted to "") and measured status/account (reset to "none" by
  set_sample_in_gripper() itself). Same gap in reverse on put_sample.
  All six call sites (three pick, three put) now read owner/status/account
  from the source before clearing it and carry them onto the destination.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018Mupip8x7Hean6E3FFvpbj

csaxs_bec/bec_ipython_client/plugins/omny/omny_sample_transfer_mixin.py

@@ -12,8 +12,6 @@ from typeguard import typechecked
 
 # from functools import wraps
 
-from csaxs_bec.bec_ipython_client.plugins.cSAXS import epics_get, epics_put, fshopen, fshclose
-
 
 if builtins.__dict__.get("bec") is not None:
     bec = builtins.__dict__.get("bec")
@@ -987,7 +985,11 @@ class OMNYSampleTransferMixin:
 
             self.device_manager.devices.rtx.controller.feedback_disable()
 
-        fshclose()
+        # dev.fsh (not the raw cSAXS_beamline.fshclose(), which does a bare
+        # epics.caput() straight to X12SA-ES1-TTL:OUT_01 outside BEC entirely
+        # -- unreachable/unmocked in a simulated session) -- same device
+        # omny_optics_mixin.py already uses for shutter control.
+        dev.fsh.fshclose()
 
         if oeyez_current < -20:
             raise OMNYTransferError(
@@ -1294,9 +1296,20 @@ class OMNYSampleTransferMixin:
                         samplename = dev.omny_samples.get_sample_name(
                             installedshuttle, _osampleposition
                         )
-                        # remove sample from slot and put to gripper
+                        owner = dev.omny_samples.get_sample_owner(
+                            installedshuttle, _osampleposition
+                        )
+                        status = dev.omny_samples.get_sample_status(
+                            installedshuttle, _osampleposition
+                        )
+                        account = dev.omny_samples.get_sample_account(
+                            installedshuttle, _osampleposition
+                        )
+                        # remove sample from slot and put to gripper -- owner/status/account
+                        # carry over with it, the same physical sample
                         dev.omny_samples.unset_sample_slot(installedshuttle, _osampleposition)
-                        dev.omny_samples.set_sample_in_gripper(samplename)
+                        dev.omny_samples.set_sample_in_gripper(samplename, owner)
+                        dev.omny_samples.set_sample_status_in_gripper(status, account)
                     return True
                 else:
                     print("There is no sample availabe at that position.")
@@ -1385,8 +1398,12 @@ class OMNYSampleTransferMixin:
             if dev.omny_samples.is_sample_slot_used("O", _osampleposition):
                 if _picked_it:
                     samplename = dev.omny_samples.get_sample_name("O", _osampleposition)
+                    owner = dev.omny_samples.get_sample_owner("O", _osampleposition)
+                    status = dev.omny_samples.get_sample_status("O", _osampleposition)
+                    account = dev.omny_samples.get_sample_account("O", _osampleposition)
                     dev.omny_samples.unset_sample_slot("O", _osampleposition)
-                    dev.omny_samples.set_sample_in_gripper(samplename)
+                    dev.omny_samples.set_sample_in_gripper(samplename, owner)
+                    dev.omny_samples.set_sample_status_in_gripper(status, account)
                     # FZP was picked
                     # if (_osampleposition == 10 || _osampleposition == 11 || _osampleposition == 100)
                     #  _otransfer_modify_sample("omny",110,2,samplename)
@@ -1406,8 +1423,12 @@ class OMNYSampleTransferMixin:
             if dev.omny_samples.is_sample_in_samplestage():
                 if _picked_it:
                     samplename = dev.omny_samples.get_sample_name_in_samplestage()
+                    owner = dev.omny_samples.get_sample_owner_in_samplestage()
+                    status = dev.omny_samples.get_sample_status_in_samplestage()
+                    account = dev.omny_samples.get_sample_account_in_samplestage()
                     dev.omny_samples.unset_sample_in_samplestage()
-                    dev.omny_samples.set_sample_in_gripper(samplename)
+                    dev.omny_samples.set_sample_in_gripper(samplename, owner)
+                    dev.omny_samples.set_sample_status_in_gripper(status, account)
                     # FZP was picked
                     # if (_osampleposition == 10 || _osampleposition == 11 || _osampleposition == 100)
                     #  _otransfer_modify_sample("omny",110,2,samplename)
@@ -1480,9 +1501,15 @@ class OMNYSampleTransferMixin:
                 if not dev.omny_samples.is_sample_slot_used(installedshuttle, _osampleposition):
                     if _put_it:
                         samplename = dev.omny_samples.get_sample_name_in_gripper()
+                        owner = dev.omny_samples.get_sample_owner_in_gripper()
+                        status = dev.omny_samples.get_sample_status_in_gripper()
+                        account = dev.omny_samples.get_sample_account_in_gripper()
                         dev.omny_samples.unset_sample_in_gripper()
                         dev.omny_samples.set_sample_slot(
-                            installedshuttle, _osampleposition, samplename
+                            installedshuttle, _osampleposition, samplename, owner
+                        )
+                        dev.omny_samples.set_sample_status(
+                            installedshuttle, _osampleposition, status, account
                         )
                     return True
                 else:
@@ -1572,7 +1599,11 @@ class OMNYSampleTransferMixin:
             if not dev.omny_samples.is_sample_slot_used("O", _osampleposition):
                 if _put_it:
                     samplename = dev.omny_samples.get_sample_name_in_gripper()
-                    dev.omny_samples.set_sample_slot("O", _osampleposition, samplename)
+                    owner = dev.omny_samples.get_sample_owner_in_gripper()
+                    status = dev.omny_samples.get_sample_status_in_gripper()
+                    account = dev.omny_samples.get_sample_account_in_gripper()
+                    dev.omny_samples.set_sample_slot("O", _osampleposition, samplename, owner)
+                    dev.omny_samples.set_sample_status("O", _osampleposition, status, account)
                     dev.omny_samples.unset_sample_in_gripper()
                 return True
             else:
@@ -1584,7 +1615,11 @@ class OMNYSampleTransferMixin:
             if not dev.omny_samples.is_sample_in_samplestage():
                 if _put_it:
                     samplename = dev.omny_samples.get_sample_name_in_gripper()
-                    dev.omny_samples.set_sample_in_samplestage(samplename)
+                    owner = dev.omny_samples.get_sample_owner_in_gripper()
+                    status = dev.omny_samples.get_sample_status_in_gripper()
+                    account = dev.omny_samples.get_sample_account_in_gripper()
+                    dev.omny_samples.set_sample_in_samplestage(samplename, owner)
+                    dev.omny_samples.set_sample_status_in_samplestage(status, account)
                     dev.omny_samples.unset_sample_in_gripper()
                 return True
             else:

csaxs_bec/devices/omny/shutter.py

@@ -1,95 +1,53 @@
 """
-Fast Shutter control for OMNY setup. If started with a config file in which the device_manager
-has a 'fsh' device (cSAXSFastEpicsShutter), this device will be used as the shutter.
-Otherwise, the device will create a dummy shutter device that will log warnings when shutter
-methods are called, but will not raise exceptions.
+Fast Shutter control for OMNY setup. Delegates directly to the 'fsh' device
+(cSAXSFastEpicsShutter, or its simulated counterpart) via the device manager
+-- 'fsh' must be present and enabled in the running config.
 """
 
 from bec_lib.logger import bec_logger
-from ophyd import Component as Cpt
-from ophyd import Device, Signal
+from ophyd import Device
 from ophyd_devices import PSIDeviceBase
 
 logger = bec_logger.logger
 
 
-class MonitorSignal(Signal):
-    """A simple wrapper around ophyd Signal that automatically monitors the signal for changes."""
-
-    def __init__(self, *, name, auto_monitor=False, **kwargs):
-        super().__init__(name=name, **kwargs)
-        self.auto_monitor = auto_monitor
-
-
 class OMNYFastShutter(PSIDeviceBase, Device):
     """
-    Fast Shutter control for OMNY setup. If started with at the beamline, it will expose
-    the shutter control methods (fshopen, fshclose, fshstatus, fshinfo) from the
-    cSAXSFastEpicsShutter device. The device is identified by the 'fsh' name in the device manager.
-    If the 'fsh' device is not found in the device manager, this device will create a dummy shutter
-    and log warnings when shutter methods are called, but will not raise exceptions.
+    Fast Shutter control for OMNY setup. Exposes the shutter control methods
+    (fshopen, fshclose, fshstatus, fshinfo, fshstatus_readback) by delegating
+    to the 'fsh' device (cSAXSFastEpicsShutter) found in the device manager.
     """
 
     USER_ACCESS = ["fshopen", "fshclose", "fshstatus", "fshinfo", "help", "fshstatus_readback"]
     SUB_VALUE = "value"
     _default_sub = SUB_VALUE
 
-    shutter = Cpt(MonitorSignal, name="shutter", auto_monitor=True)
-
     # -----------------------------------------------------
     # User-facing shutter control functions
     # -----------------------------------------------------
 
-    # pylint: disable=invalid-name
-    def _check_if_cSAXS_shutter_exists_in_config(self) -> bool:
-        """
-        Check on the device manager if the shutter device exists.
-
-        Returns:
-            bool: True if the 'fsh' device exists in the device manager, False otherwise
-        """
-        if self.device_manager.devices.get("fsh", None) is None:
-            logger.warning(f"Fast shutter device not found for {self.name}.")
-            return False
-        return True
-
     def fshopen(self):
         """Open the fast shutter."""
-        if self._check_if_cSAXS_shutter_exists_in_config():
-            return self.device_manager.devices["fsh"].fshopen()
-        else:
-            self.shutter.put(1)
+        return self.device_manager.devices["fsh"].fshopen()
 
     def fshclose(self):
         """Close the fast shutter."""
-        if self._check_if_cSAXS_shutter_exists_in_config():
-            return self.device_manager.devices["fsh"].fshclose()
-        else:
-            self.shutter.put(0)
+        return self.device_manager.devices["fsh"].fshclose()
 
     def fshstatus(self):
         """Return the fast shutter status (0=closed, 1=open)."""
-        if self._check_if_cSAXS_shutter_exists_in_config():
-            return self.device_manager.devices["fsh"].fshstatus()
-        else:
-            return self.shutter.get()
+        return self.device_manager.devices["fsh"].fshstatus()
 
     def fshinfo(self):
         """Print information about which EPICS PV channel is being used."""
-        if self._check_if_cSAXS_shutter_exists_in_config():
-            return self.device_manager.devices["fsh"].fshinfo()
-        else:
-            print("Using dummy fast shutter device. No EPICS channel is connected.")
+        return self.device_manager.devices["fsh"].fshinfo()
+
+    def fshstatus_readback(self):
+        """Return the fast shutter status (0=closed, 1=open) from the readback signal."""
+        return self.device_manager.devices["fsh"].fshstatus_readback()
 
     def help(self):
         """Display available user methods."""
         print("Available methods:")
         for method in self.USER_ACCESS:
             print(f"  - {method}")
-
-    def fshstatus_readback(self):
-        """Return the fast shutter status (0=closed, 1=open) from the readback signal."""
-        if self._check_if_cSAXS_shutter_exists_in_config():
-            return self.device_manager.devices["fsh"].fshstatus_readback()
-        else:
-            self.shutter.get()

tests/tests_devices/test_omny_shutter.py

@@ -1,69 +1,51 @@
+from unittest import mock
+
 import pytest
 from bec_server.device_server.tests.utils import DMMock
 
-from csaxs_bec.devices.omny.shutter import MonitorSignal, OMNYFastShutter
-
-
-@pytest.mark.parametrize("auto_monitor", [False, True])
-def test_monitor_signal_stores_auto_monitor(auto_monitor):
-    signal = MonitorSignal(name="signal", auto_monitor=auto_monitor)
-
-    assert signal.auto_monitor is auto_monitor
-
-
-def test_monitor_signal_put_propagates_value_to_readback_callback():
-    signal = MonitorSignal(name="signal", auto_monitor=True)
-    initial_value = signal.read()[signal.name]["value"]
-    callback_values = []
-    callback_reads = []
-
-    def _test_cb(value, old_value, **kwargs):
-        callback_values.append((value, old_value))
-        callback_reads.append(kwargs["obj"].read())
-
-    signal.subscribe(_test_cb, event_type=signal.SUB_VALUE, run=False)
-
-    signal.put(1)
-
-    assert callback_values == [(1, initial_value)]
-    assert len(callback_reads) == 1
-    assert callback_reads[0][signal.name]["value"] == 1
-    assert signal.read()[signal.name]["value"] == 1
-
-    signal.put(0)
-    assert callback_values == [(1, initial_value), (0, 1)]
-    assert len(callback_reads) == 2
-    assert callback_reads[1][signal.name]["value"] == 0
-    assert signal.read()[signal.name]["value"] == 0
+from csaxs_bec.devices.omny.shutter import OMNYFastShutter
 
 
 @pytest.fixture
 def omny_fast_shutter():
-    shutter = OMNYFastShutter(name="omny_fast_shutter", device_manager=DMMock())
+    device_manager = DMMock()
+    fsh = mock.MagicMock()
+    device_manager.devices["fsh"] = fsh
+    shutter = OMNYFastShutter(name="omny_fast_shutter", device_manager=device_manager)
 
     try:
-        yield shutter
+        yield shutter, fsh
     finally:
         shutter.destroy()
 
 
-def test_omny_fast_shutter_uses_monitor_signal_with_auto_monitor(omny_fast_shutter):
-    assert isinstance(omny_fast_shutter.shutter, MonitorSignal)
-    assert omny_fast_shutter.shutter.auto_monitor is True
+def test_omny_fast_shutter_fshopen_delegates_to_fsh(omny_fast_shutter):
+    shutter, fsh = omny_fast_shutter
+    shutter.fshopen()
+    fsh.fshopen.assert_called_once()
 
 
-def test_omny_fast_shutter_propagates_signal_changes_to_device_readback(omny_fast_shutter):
-    signal_name = omny_fast_shutter.shutter.name
-    callback_reads = []
+def test_omny_fast_shutter_fshclose_delegates_to_fsh(omny_fast_shutter):
+    shutter, fsh = omny_fast_shutter
+    shutter.fshclose()
+    fsh.fshclose.assert_called_once()
 
-    def _test_cb(**kwargs):
-        callback_reads.append(omny_fast_shutter.read())
 
-    omny_fast_shutter.shutter.subscribe(_test_cb, event_type=omny_fast_shutter.shutter.SUB_VALUE, run=False)
+def test_omny_fast_shutter_fshstatus_delegates_to_fsh(omny_fast_shutter):
+    shutter, fsh = omny_fast_shutter
+    fsh.fshstatus.return_value = 1
+    assert shutter.fshstatus() == 1
+    fsh.fshstatus.assert_called_once()
 
-    omny_fast_shutter.shutter.put(1)
 
-    assert len(callback_reads) == 1
-    assert callback_reads[0][signal_name]["value"] == 1
-    assert omny_fast_shutter.read()[signal_name]["value"] == 1
-    assert omny_fast_shutter.fshstatus() == 1
+def test_omny_fast_shutter_fshstatus_readback_delegates_to_fsh(omny_fast_shutter):
+    shutter, fsh = omny_fast_shutter
+    fsh.fshstatus_readback.return_value = 0
+    assert shutter.fshstatus_readback() == 0
+    fsh.fshstatus_readback.assert_called_once()
+
+
+def test_omny_fast_shutter_fshinfo_delegates_to_fsh(omny_fast_shutter):
+    shutter, fsh = omny_fast_shutter
+    shutter.fshinfo()
+    fsh.fshinfo.assert_called_once()