From 2086c9e4e6527ced07c10c724ecc3dbf1144b4c9 Mon Sep 17 00:00:00 2001 From: x01dc Date: Fri, 4 Sep 2026 16:51:21 +0200 Subject: [PATCH] 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 Claude-Session: https://claude.ai/code/session_018Mupip8x7Hean6E3FFvpbj --- .../omny/omny_sample_transfer_mixin.py | 55 ++++++++++--- csaxs_bec/devices/omny/shutter.py | 72 ++++------------ tests/tests_devices/test_omny_shutter.py | 82 ++++++++----------- 3 files changed, 92 insertions(+), 117 deletions(-) diff --git a/csaxs_bec/bec_ipython_client/plugins/omny/omny_sample_transfer_mixin.py b/csaxs_bec/bec_ipython_client/plugins/omny/omny_sample_transfer_mixin.py index 7ca98c90..2ae14aa3 100644 --- a/csaxs_bec/bec_ipython_client/plugins/omny/omny_sample_transfer_mixin.py +++ b/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: diff --git a/csaxs_bec/devices/omny/shutter.py b/csaxs_bec/devices/omny/shutter.py index 6e5e4fe8..56e9b219 100644 --- a/csaxs_bec/devices/omny/shutter.py +++ b/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() diff --git a/tests/tests_devices/test_omny_shutter.py b/tests/tests_devices/test_omny_shutter.py index b2bc1c9d..85161b69 100644 --- a/tests/tests_devices/test_omny_shutter.py +++ b/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()