Source code for csaxs_bec.devices.omny.shutter

"""
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.
"""

from bec_lib.logger import bec_logger
from ophyd import Component as Cpt
from ophyd import Device, Signal
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.
    """

    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)




    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)




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




    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.")




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