diff --git a/src/sophys/common/devices/shutters.py b/src/sophys/common/devices/shutters.py new file mode 100644 index 00000000..a77586c2 --- /dev/null +++ b/src/sophys/common/devices/shutters.py @@ -0,0 +1,198 @@ +from ophyd import EpicsSignal, EpicsSignalRO, FormattedComponent, Device +from ophyd.pv_positioner import PVPositionerComparator +from ..utils.status import PremadeStatus +from ophyd.status import AndStatus, SubscriptionStatus + + +class ShutterToggle(PVPositionerComparator): + """ + Abstraction layer for shutters with one actuation PV (OPENCLOSE) and one readback PV (PG_STATUS). There's an optional parameter for a permission PV. + + Parameters + ---------- + prefix: str + Prefix for the shutter's PVs. + + setpoint_suffix: str + Suffix for the actuation PV. NOTE: This should be place/location of the shutter, e.g. OEA/FOE. + The PV will be formatted as "{prefix}{setpoint_suffix}OPENCLOSE". + + readback_suffix: str + Suffix for the readback PV, e.g. PG_STATUS + + permission_suffix: str, optional + Permission PV string, if it exists. + + NOTE + ---- + This implemantation considers that the value of the `readback` signal is 0 for an open shutter and 1 for a closed shutter. + This is not so intuitive, so the `set` method considers that 1 is for opening and 0 for closing the shutter. + + Usage Example + ------------- + >>> shutter = ShutterOpenClose(prefix="prefix", setpoint_suffix="setpoint_suffix", readback="readback_suffix", name="shutter") + >>> shutter.set(0).wait() # for closing + >>> shutter.set(1).wait() # for opening + """ + + real_setpoint = None + setpoint = FormattedComponent( + EpicsSignal, "{prefix}{setpoint_suffix}OPENCLOSE", kind="config" + ) + readback = FormattedComponent( + EpicsSignalRO, "{prefix}{readback_suffix}", kind="hinted" + ) + + def __init__( + self, + *args, + setpoint_suffix: str, + readback_suffix: str, + permission_pv: str = None, + **kwargs, + ): + self.setpoint_suffix = setpoint_suffix + self.readback_suffix = readback_suffix + self._permission_pv_name = permission_pv + super().__init__(*args, **kwargs) + if self._permission_pv_name is not None: + self.permission_signal = EpicsSignalRO( + f"{self._permission_pv_name}", name="permission" + ) + + def set(self, value, *args, **kwargs): + if hasattr(self, "permission_signal"): + permission_status = self.permission_signal.get( + connection_timeout=2, **kwargs + ) + if not permission_status: + raise PremadeStatus( + success=False, + exception=PermissionError( + f"Shutter open permission is denied: {self.permission_signal.pvname} {permission_status}." + ), + ) + + if ( + value == self.readback.get() + ): # Since we're swapping the readback values (0 for closing and 1 for opening), we actuate when value == readback + self.real_setpoint = 1 if value == 0 else 0 + return super().set(1, *args, **kwargs) + else: + return PremadeStatus(success=True) + + def done_comparator(self, readback, setpoint): + return self.real_setpoint == readback + + +class ShutterOpenClose(Device): + """ + Abstraction layer for shutters with two actuation PV (OPEN and CLOSE) and two readback PV (PS_STATUS and GS_STATUS). There's an optional parameter for a permission PV. + + Parameters + ---------- + prefix: str + Prefix for the shutter's PVs. + + shutter_suffix: str + Suffix for the OPEN and CLOSE PVs. NOTE: This should be place/location of the shutter, e.g. OEA/FOE. + The PVs will be formatted as "{prefix}{shutter_suffix}OPEN" and "{prefix}{shutter_suffix}CLOSE". + + ps_suffix: str + Suffix for one readback PVs, e.g. PS_STATUS + + gs_suffix: str + Suffix for the second readback PV, e.g. GS_STATUS + + permission_pv: str, optional + Permission PV string, if it exists. + + NOTES + ----- + This implemantation considers that the value of the `readback` signal is 0 for an open shutter and 1 for a closed shutter. + This is not so intuitive, so the `set` method considers that 1 is for opening and 0 for closing the shutter. + + The `return` of the `set` method is an `AndStatus` with both `readback` signals. + + There's a `done_comparator` method that returns the state of the shutter, based in the two `readback` PVs. This method is + used as the `callback` for both `readback` signals. + + Usage Example + ------------- + >>> shutter = ShutterToggle(prefix="prefix", open_suffix="open_suffix", close_suffix="close_suffix", ps_suffix="ps_suffix", gs_suffix="gs_suffix", name="shutter") + >>> shutter.set(0).wait() # for closing + >>> shutter.set(1).wait() # for opening + """ + + setpoint = None + photon_status = FormattedComponent( + EpicsSignalRO, "{prefix}{ps_suffix}", kind="hinted" + ) + gamma_status = FormattedComponent( + EpicsSignalRO, "{prefix}{gs_suffix}", kind="hinted" + ) + open = FormattedComponent( + EpicsSignal, "{prefix}{shutter_suffix}OPEN", kind="config" + ) + close = FormattedComponent( + EpicsSignal, "{prefix}{shutter_suffix}CLOSE", kind="config" + ) + + def __init__( + self, + *args, + shutter_suffix: str, + ps_suffix: str, + gs_suffix: str, + permission_pv: str = None, + **kwargs, + ): + self.shutter_suffix = shutter_suffix + self.ps_suffix = ps_suffix + self.gs_suffix = gs_suffix + self._permission_pv_name = permission_pv + super().__init__(*args, **kwargs) + if self._permission_pv_name is not None: + self.permission_signal = EpicsSignalRO( + f"{self._permission_pv_name}", name="permission" + ) + + def set(self, value, *args, **kwargs): + if hasattr(self, "permission_signal"): + permission_status = self.permission_signal.get( + connection_timeout=2, **kwargs + ) + if not permission_status: + raise PremadeStatus( + success=False, + exception=PermissionError( + f"Shutter open permission is denied: {self.permission_signal.pvname} {permission_status}" + ), + ) + + if value == 0 and not self._is_closed(): + self.close.set(1, *args, **kwargs).wait() + + elif value == 1 and self._is_closed(): + self.open.set(1, *args, **kwargs).wait() + + else: + return PremadeStatus(success=True) + + self.setpoint = value + + return AndStatus( + SubscriptionStatus(self.photon_status, self.done_comparator, settle_time=3), + SubscriptionStatus(self.gamma_status, self.done_comparator, settle_time=3), + timeout=15, + ) + + def _is_closed(self): + """Check wheter the shutter is open or closed given the photon and gamma status PVs.""" + return (self.photon_status.get() == 1) and ( + self.gamma_status.get() == 1 + ) # NOTE: if one of the status is equal to zero, the shutter can be partially open + + def done_comparator(self, value, **kwargs): + is_closed = self._is_closed() + return is_closed if self.setpoint == 0 else not is_closed