refactor: move protocol and base classes to different directory

ophyd_devices/ophyd_base_devices/__init__.py

@@ -0,0 +1,10 @@
+from .bec_protocols import (
+    BECDeviceProtocol,
+    BECFlyerProtocol,
+    BECMixinProtocol,
+    BECPositionerProtocol,
+    BECRotationProtocol,
+    BECScanProtocol,
+    BECSignalProtocol,
+)
+from .tomcat_rotation_motors import TomcatAerotechRotation

ophyd_devices/ophyd_base_devices/bec_protocols.py

@@ -0,0 +1,482 @@
+""" This module provides a range of protocols that describe the expected interface for different types of devices.
+
+The protocols below can be used as teamplates for functionality to be implemeted by different type of devices.
+They further facilitate runtime checks on devices and provide a minimum set of properties required for a device to be loadable by BEC.
+
+The protocols are:
+- BECDeviceProtocol: Protocol for devices in BEC. All devices must at least implement this protocol.
+- BECSignalProtocol: Protocol for signals.
+- BECScanProtocol: Protocol for the scan interface.
+- BECMixinProtocol: Protocol for utilities in particular relevant for detector implementations.
+- BECPositionerProtocol: Protocol for positioners.
+- BECFlyerProtocol: Protocol with for flyers.
+
+Keep in mind, that a device of type flyer should generally also implement the BECScanProtocol that provides the required functionality for scans.
+Flyers in addition, also implement the BECFlyerProtocol. Similarly, positioners should also implement the BECScanProtocol and BECPositionerProtocol.
+
+"""
+
+from typing import Protocol, runtime_checkable
+
+from bec_lib.file_utils import FileWriterMixin
+from ophyd import Component, DeviceStatus, Kind, Staged
+
+from ophyd_devices.utils import bec_scaninfo_mixin
+
+
+@runtime_checkable
+class BECDeviceProtocol(Protocol):
+    """Protocol for ophyd objects with zero functionality."""
+
+    _destroyed: bool
+
+    @property
+    def name(self) -> str:
+        """name property"""
+
+    @name.setter
+    def name(self, value: str) -> None:
+        """name setter"""
+
+    @property
+    def kind(self) -> Kind:
+        """kind property"""
+
+    @kind.setter
+    def kind(self, value: Kind):
+        """kind setter"""
+
+    @property
+    def parent(self):
+        """Property to find the parent device"""
+
+    @property
+    def root(self):
+        """Property to fint the root device"""
+
+    @property
+    def hints(self) -> dict:
+        """hints property"""
+
+    @property
+    def connected(self) -> bool:
+        """connected property.
+        Check if signals are connected
+
+        Returns:
+            bool: True if connected, False otherwise
+        """
+
+    @connected.setter
+    def connected(self, value: bool):
+        """connected setter"""
+
+    def read(self) -> dict:
+        """read method
+
+        Override by child class with read method
+
+        Returns:
+            dict: Dictionary with nested dictionary of signals with kind.normal or kind.hinted:
+            {'signal_name' : {'value' : .., "timestamp" : ..}, ...}
+        """
+
+    def read_configuration(self) -> dict:
+        """read_configuration method
+
+        Override by child class with read_configuration method
+
+        Returns:
+            dict: Dictionary with nested dictionary of signals with kind.config:
+            {'signal_name' : {'value' : .., "timestamp" : ..}, ...}
+        """
+
+    def describe(self) -> dict:
+        """describe method
+
+        Override by child class with describe method
+
+        Returns:
+            dict: Dictionary with dictionaries with signal descriptions ('source', 'dtype', 'shape')
+        """
+
+    def describe_configuration(self) -> dict:
+        """describe method
+
+        Includes all signals of type Kind.config.
+        Override by child class with describe_configuration method
+
+        Returns:
+            dict: Dictionary with dictionaries with signal descriptions ('source', 'dtype', 'shape')
+        """
+
+    def destroy(self) -> None:
+        """Destroy method.
+
+        _destroyed must be set to True after calling destroy.
+        """
+
+    def trigger(self) -> DeviceStatus:
+        """Trigger method on the device
+
+        Returns ophyd DeviceStatus object, which is used to track the status of the trigger.
+        It can also be blocking until the trigger is completed, and return the status object
+        with set_finished() method called on the DeviceStatus.
+        """
+
+
+@runtime_checkable
+class BECSignalProtocol(Protocol):
+    """Protocol for BEC signals with zero functionality.
+
+    This protocol adds the specific implementation for a signal.
+    Please be aware that a signal must also implement BECDeviceProtocol.
+
+    Note: Currently the implementation of the protocol is not taking into account the
+    event_model from ophyd, i.e. _run_sbus
+    """
+
+    @property
+    def limits(self) -> tuple[float, float]:
+        """Limits property for signals.
+        If low_limit == high_limit, it is equivalent to NO limits!
+
+        Returns:
+            tuple: Tuple with lower and upper limits
+        """
+
+    @property
+    def high_limit(self) -> float:
+        """High limit property for signals.
+
+        Returns:
+            float: Upper limit
+        """
+
+    @property
+    def low_limit(self) -> float:
+        """Low limit property for signals.
+
+        Returns:
+            float: Lower limit
+        """
+
+    @property
+    def write_access(self) -> bool:
+        """Write access method for signals.
+
+        Returns:
+            bool: True if write access is allowed, False otherwise
+        """
+
+    def check_value(self, value: float):
+        """Check whether value is within limits
+
+        Args:
+            value: value to check
+
+        Raises:
+            LimitError in case the requested motion is not inside of limits.
+        """
+
+    def put(self, value: any, force: bool = False, timeout: float = None):
+        """Put method for signals.
+        This method should resolve immediately and not block.
+        If not force, the method checks if the value is within limits using check_value.
+
+
+        Args:
+            value (any)     : value to put
+            force (bool)    : Flag to force the put and ignore limits
+            timeout (float) : Timeout for the put
+        """
+
+    def set(self, value: any, timeout: float = None) -> DeviceStatus:
+        """Set method for signals.
+        This method should be blocking until the set is completed.
+
+        Args:
+            value (any)     : value to set
+            timeout (float) : Timeout for the set
+
+        Returns:
+            DeviceStatus    : DeviceStatus object that will finish upon return
+        """
+
+
+@runtime_checkable
+class BECScanProtocol(BECDeviceProtocol, Protocol):
+    """Protocol for devices offering an Protocol with all relevant functionality for scans.
+
+    In BEC, scans typically follow the order of stage, (pre_scan), trigger, unstage.
+    Stop should be used to interrupt a scan. Be aware that pre_scan is optional and therefor
+    part of the BECMixinProtocol, typically useful for more complex devices such as detectors.
+
+    This protocol allows to perform runtime checks on devices of ophyd.
+    It is the minimum set of properties required for a device to be loadable by BEC.
+    """
+
+    _staged: Staged
+    """Staged property to indicate if the device is staged."""
+
+    def stage(self) -> list[object]:
+        """Stage method to prepare the device for an upcoming acquistion.
+
+        This prepares a device for an upcoming acquisition, i.e. it is the first
+        method for which the scan parameters are known and the device can be configured.
+
+        It can be used to move scan_motors to their start position
+        or also prepare DAQ systems for the upcoming measurement.
+        We can further publish the file location for DAQ systems
+        to BEC and inform BEC's file writer where data will be written to.
+
+        Stagin is not idempoent. If called twice without an unstage it should raise.
+        For ophyd devices, one may used self._staged = True to check if the device is staged.
+
+        Returns:
+            list:   List of objects that were staged, i.e. [self]
+                    For devices with inheritance from ophyd, return
+                    return super().stage() in the child class.
+        """
+
+    def unstage(self) -> list[object]:
+        """Unstage method to cleanup after the acquisition.
+
+        It can also be used to implement checks whether the acquisition was successful,
+        inform BEC that the file has been succesfully written, or raise upon receiving
+        feedback that the scan did not finish successful.
+
+        Unstaging is not idempotent. If called twice without a stage it should raise.
+        It is recommended to return super().unstage() in the child class, if
+        the child class also inherits from ophyd repository.
+        """
+
+    def stop(self, success: bool) -> None:
+        """Stop method to stop the device.
+
+        Args:
+            success: Flag to indicate if the scan was successful or not.
+
+        This method should be called to stop the device. It is recommended to call
+        super().stop(success=success) if class inherits from ophyd repository.
+        """
+
+
+@runtime_checkable
+class BECMixinProtocol(Protocol):
+    """Protocol that offers BEC specific utility functionality for detectors."""
+
+    USER_ACCESS: list[str]
+    """
+    List of methods/properties that will be exposed to the client interface in addition
+    to the the already exposed signals, methods and properties.
+    """
+
+    scaninfo: bec_scaninfo_mixin
+    """ 
+    BEC scan info mixin class that provides an transparent Protocol to scan parameter 
+    as provided by BEC. It is recommended to use this Protocol to retrieve scaninfo from Redis. 
+    """
+
+    stopped: bool
+    """ 
+    Flag to indicate if the device is stopped. 
+    
+    The stop method should set this flag to True, and i.e. stage to set it to False. 
+    """
+
+    filewriter: FileWriterMixin
+    """
+    The file writer mixin main purpose is to unify and centralize the creation of 
+    file paths within BEC. Therefore, we recommend devices to use the same mixin for creation of paths.
+    """
+
+    def pre_scan(self):
+        """Pre-scan method is called from BEC right before executing scancore, thus
+        right before the start of an acquisition.
+
+        It can be used to trigger time critical functions from the device, which
+        are prone to run into timeouts in case called too early.
+        """
+
+
+@runtime_checkable
+class BECPositionerProtocol(Protocol):
+    """Protocol with functionality specific for positioners in BEC."""
+
+    @property
+    def limits(self) -> tuple[float, float]:
+        """Limits property for positioners.
+        For an EpicsMotor, BEC will automatically recover the limits from the IOC.
+
+        If not set, it returns (0,0).
+        Note, low_limit = high_limit is equivalent to NO limits!
+
+        Returns:
+            tuple: Tuple with lower and upper limits
+        """
+
+    @property
+    def low_limit(self) -> float:
+        """Low limit property for positioners.
+
+        Returns:
+            float: Lower limit
+        """
+
+    @property
+    def high_limit(self) -> float:
+        """High limit property for positioners.
+
+        Returns:
+            float: Upper limit
+        """
+
+    def check_value(self, value: float):
+        """Check whether value is within limits
+
+        Args:
+            value: value to check
+
+        Raises:
+            LimitError in case the requested motion is not inside of limits.
+        """
+
+    def move(self, position: float) -> DeviceStatus:
+        """Move method for positioners.
+        The returned DeviceStatus is marked as done once the positioner has reached the target position.
+        DeviceStatus.wait() can be used to block until the move is completed.
+
+        Args:
+            position: position to move to
+
+        Returns:
+            DeviceStatus: DeviceStatus object
+        """
+
+
+@runtime_checkable
+class BECFlyerProtocol(BECScanProtocol, Protocol):
+    """Protocol with functionality specific for flyers in BEC."""
+
+    # def configure(self, d: dict):
+    #     """Configure method of the flyer.
+    #     It is an optional method, but does not need to be implemented by a flyer.
+    #     Instead, stage can be used to prepare time critical operations on the device in preparation of a scan.
+
+    #     Method to configure the flyer in preparation of a scan.
+
+    #     Args:
+    #         d (dict): Dictionary with configuration parameters, i.e. key value pairs of signal_name : value
+    #     """
+
+    def kickoff(self) -> DeviceStatus:
+        """Kickoff method for flyers.
+
+        The returned DeviceStatus is marked as done once the flyer start flying,
+        i.e. is ready to be triggered.
+
+        Returns:
+            DeviceStatus: DeviceStatus object
+        """
+
+    def complete(self) -> DeviceStatus:
+        """Complete method for flyers.
+
+        The returned DeviceStatus is marked as done once the flyer has completed.
+
+        Returns:
+            DeviceStatus: DeviceStatus object
+        """
+
+
+@runtime_checkable
+class BECRotationProtocol(Protocol):
+    """Protocol which defines functionality for a tomography stage for ophyd devices"""
+
+    allow_mod360: Component
+    """Signal to define whether mod360 operations are allowed. """
+
+    @property
+    def has_mod360(self) -> bool:
+        """Property to check if the motor has mod360 option
+
+        Returns:
+            bool: True if mod360 is possible on device, False otherwise
+        """
+
+    @property
+    def has_freerun(self) -> bool:
+        """Property to check if the motor has freerun option
+
+        Returns:
+            bool: True if freerun is allowed, False otherwise
+        """
+
+    @property
+    def valid_rotation_modes(self) -> list[str]:
+        """Method to get the valid rotation modes for the implemented motor.
+
+        Returns:
+            list: List of strings with valid rotation modes
+        """
+
+    def apply_mod360(self) -> None:
+        """Method to apply the modulus 360 operation on the specific device.
+
+        Childrens should override this method
+        """
+
+
+@runtime_checkable
+class BECEventProtocol(Protocol):
+    """Protocol for events in BEC.
+
+    This is a first draft for the event protocol introduced throughout BEC.
+    It needs to be review and extended before it can be used in production.
+    """
+
+    _callbacks: dict[dict]
+
+    @property
+    def event_types(self) -> tuple[str]:
+        """Event types property"""
+
+    def _run_subs(self, sub_type: str, **kwargs):
+        """Run subscriptions for the event.
+
+        Args:
+            sub_type: Subscription type
+            kwargs: Keyword arguments
+        """
+
+    def subscribe(self, callback: callable, event_type: str = None, run: bool = True):
+        """Subscribe to the event.
+
+        Args:
+            callback (callable) :   Callback function
+                                    The expected callback structure is:
+                                    def cb(*args, obj:OphydObject, sub_type:str, **kwargs) -> None:
+                                        pass
+            event_type (str)    :   Event type, if None it defaults to obj._default_sub
+                                    This maps to sub_type in _run_subs
+            run (bool)          :   If true, run the callback directly.
+
+        Returns:
+            cid (int):              Callback id
+        """
+
+    def clear_sub(self, cb: callable, event_type: str = None):
+        """Clear subscription, given the origianl callback fucntion
+
+        Args:
+            cb (callable)   : Callback
+            event_type (str): Event type, if None it will be remove from all event_types
+        """
+
+    def unsubscribe(self, cid: int):
+        """Unsubscribe from the event.
+
+        Args:
+            cid (int): Callback id
+        """

ophyd_devices/ophyd_base_devices/ophyd_rotation_base.py

@@ -0,0 +1,105 @@
+from abc import ABC, abstractmethod
+
+from bec_lib import bec_logger
+from ophyd import Component as Cpt
+from ophyd import EpicsMotor
+from typeguard import typechecked
+
+from ophyd_devices.ophyd_base_devices.bec_protocols import BECRotationProtocol
+from ophyd_devices.utils.bec_utils import ConfigSignal
+
+logger = bec_logger.logger
+
+
+class OphtyRotationBaseError(Exception):
+    """Exception specific for implmenetation of rotation stages."""
+
+
+class OphydRotationBase(BECRotationProtocol, ABC):
+
+    allow_mod360 = Cpt(ConfigSignal, name="allow_mod360", value=False, kind="config")
+
+    def __init__(self, *args, **kwargs):
+        """
+        Base class to implement functionality specific for rotation devices.
+
+        Childrens should override the instance attributes:
+        - has_mod360
+        - has_freerun
+        - valid_rotation_modes
+
+        """
+        # pylint: disable=protected-access
+        self._has_mod360 = False
+        self._has_freerun = False
+        self._valid_rotation_modes = []
+        if "allow_mod360" in kwargs:
+            if not isinstance(kwargs["allow_mod360"], bool):
+                raise ValueError("allow_mod360 must be a boolean")
+            self.allow_mod360.put(kwargs["allow_mod360"])
+        super().__init__(*args, **kwargs)
+
+    @abstractmethod
+    def apply_mod360(self) -> None:
+        """Method to apply the modulus 360 operation on the specific device.
+
+        Childrens should override this method
+        """
+
+    @property
+    def has_mod360(self) -> bool:
+        """Property to check if the device has mod360 operation.
+
+        ReadOnly property, childrens should override this method.
+        """
+        return self._has_mod360
+
+    @property
+    def has_freerun(self) -> bool:
+        """Property to check if the device has freerun operation.
+
+        ReadOnly property, childrens should override this method.
+        """
+        return self._has_freerun
+
+    @property
+    def valid_rotation_modes(self) -> list:
+        """Method to get the valid rotation modes for the specific device."""
+        return self._valid_rotation_modes
+
+    @typechecked
+    @valid_rotation_modes.setter
+    def valid_rotation_modes(self, value: list[str]):
+        """Method to set the valid rotation modes for the specific device."""
+        self._valid_rotation_modes = value
+        return self._valid_rotation_modes
+
+
+# pylint: disable=too-many-ancestors
+class EpicsRotationBase(OphydRotationBase, EpicsMotor):
+    """Class for Epics rotation devices."""
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self._has_freerun = True
+        self._has_freerun = True
+        self._valid_rotation_modes = ["target", "radiography"]
+
+    def apply_mod360(self) -> None:
+        """Apply modulos 360 operation for EpicsMotorRecord.
+
+        EpicsMotor has the function "set_current_position" which can be used for this purpose.
+        In addition, there is a check if mod360 is allowed and available.
+        """
+        if self.has_mod360 and self.allow_mod360.get():
+            cur_val = self.user_readback.get()
+            new_val = cur_val % 360
+            try:
+                self.set_current_position(new_val)
+            except Exception as exc:
+                error_msg = f"Failed to set new position {new_val} from {cur_val} on device {self.name} with error {exc}"
+                raise OphtyRotationBaseError(error_msg) from exc
+            return
+        logger.info(
+            f"Did not apply mod360 for device {self.name} with has_mod={self.has_mod360} and allow_mod={self.allow_mod360.get()}"
+        )

ophyd_devices/ophyd_base_devices/tomcat_rotation_motors.py

@@ -0,0 +1,191 @@
+""" Module for Tomcat rotation motors.
+
+The following classes implement the rotation motors for:
+
+- AerotechAutomation1 (Tomcat), based on EpicsMotorIOC.
+
+"""
+
+import threading
+import time
+
+import numpy as np
+from bec_lib import threadlocked
+from ophyd import DeviceStatus
+
+from ophyd_devices.ophyd_base_devices.bec_protocols import (
+    BECFlyerProtocol,
+    BECScanProtocol,
+)
+from ophyd_devices.ophyd_base_devices.ophyd_rotation_base import EpicsRotationBase
+
+
+class TomcatAerotechRotation(EpicsRotationBase, BECFlyerProtocol, BECScanProtocol):
+    """Special motor class that provides flyer interface and progress bar."""
+
+    SUB_PROGRESS = "progress"
+
+    def __init__(
+        self,
+        prefix="",
+        *,
+        name,
+        kind=None,
+        read_attrs=None,
+        configuration_attrs=None,
+        parent=None,
+        **kwargs,
+    ):
+        """Implementation of the Tomcat AerotechAutomation 1 rotation motor class.
+
+        This motor class is based on EpicsRotationBase and provides in addition the flyer interface for BEC
+        and a progress update.
+        """
+        super().__init__(
+            prefix=prefix,
+            name=name,
+            kind=kind,
+            read_attrs=read_attrs,
+            configuration_attrs=configuration_attrs,
+            parent=parent,
+            **kwargs,
+        )
+        self._start_position = None
+        self._target_position = None
+        self._stopped = False
+        self._rlock = threading.RLock()
+        self.subscribe(self._progress_update, run=False)
+
+    # ------------------ alternative to using configure method --------------------- #
+    @property
+    def start_position(self) -> float:
+        """Get the start position."""
+        return self._start_position
+
+    @start_position.setter
+    def start_position(self, value: float) -> None:
+        """Set the start position."""
+        self._start_position = value
+
+    @property
+    def target_position(self) -> float:
+        """Get the start position."""
+        return self._target_position
+
+    @target_position.setter
+    def target_position(self, value: float) -> None:
+        """Set the start position."""
+        self._target_position = value
+
+    # ------------------ alternative to using configure method --------------------- #
+    # configure method is optional for flyers within BEC, you can use stage method or pre_scan method to
+    # set relevant parameters on the device.
+
+    # def configure(self, d: dict) -> dict:
+    #     """Configure method from the device.
+
+    #     This method is usually used to set configuration parameters for the device.
+
+    #     Args:
+    #         d (dict): Dictionary with configuration parameters.
+
+    #     """
+    #     if "target" in d:
+    #         self._target_position = d["target"]
+    #         del d["target"]
+    #     if "position" in d:
+    #         self._target_position = d["position"]
+    #         del d["position"]
+    #     return super().configure(d)
+
+    def pre_scan(self):
+        """Perform pre-scan operation, e.g. move to start position."""
+        if self._start_position:
+            self.move(self._start_position, wait=True)
+
+    def kickoff(self) -> DeviceStatus:
+        """Kickoff the scan.
+
+        The kickoff method should return a status object that is set to finish once the flyer flys, and is ready for the next actions.
+        I would consider the following implementation.
+        """
+        self._start_position = float(self.position)
+        self.move(self._target_position, wait=False)
+        status = DeviceStatus(self)
+        status.set_finished()
+        return status
+
+    def complete(self) -> DeviceStatus:
+        """Complete method of the scan.
+
+        This will be called in a fly scan after the kickoff, thus, the stage will be moving to it's target position.
+        It should
+
+        The complete method should return a status object that is set to finish once the flyer is done and the scan is complete.
+        I would consider the following implementation.
+        """
+        threading.Thread(target=self._is_motor_moving, daemon=True).start()
+        status = DeviceStatus(self)
+        self.subscribe(status.set_finished, event_type=self.SUB_DONE, run=False)
+        return status
+
+    def stage(self) -> list[object]:
+        """Stage the scan.
+
+        We add here in addition the setting of the _stopped flag to False for the thread.
+        """
+        self._stopped = False
+        return super().stage()
+
+    def stop(self, success: bool = False) -> None:
+        """Stop the scan.
+
+        If the device is stopped, the _stopped flag is set to True.
+        """
+        self._stopped = True
+        super().stop(success=success)
+
+    @threadlocked
+    def _is_motor_moving(self):
+        """Function to check if the motor is moving.
+
+        This function is used in a thread to check if the motor is moving.
+        It resolves by running"""
+        while self.motor_done_move.get():
+            if self._stopped:
+                self._done_moving(success=False)
+                return
+            time.sleep(0.1)
+        self._done_moving(success=True)
+
+    # TODO This logic could be refined to be more robust for various scan types, i.e. at the moment it just takes
+    # the start and target position and calculates the progress based on the current position.
+    def _progress_update(self, value, **kwargs) -> None:
+        """Progress update on the scan.
+
+        Runs the progress update on the device progress during the scan.
+        It uses the SUB_PROGRESS event from ophyd to update BEC about the progress.
+        Scans need to be aware which device progress is relevant for the scan.
+
+        Args:
+            value (float): The value of the motor position.
+        """
+        if (self._start_position is None) or (self._target_position is None) or (not self.moving):
+            self._run_subs(
+                sub_type=self.SUB_PROGRESS,
+                value=1,
+                max_value=1,
+                done=1,
+            )
+            return
+
+        progress = np.abs(
+            (value - self._start_position) / (self._target_position - self._start_position)
+        )
+        max_value = 100
+        self._run_subs(
+            sub_type=self.SUB_PROGRESS,
+            value=int(100 * progress),
+            max_value=max_value,
+            done=int(np.isclose(max_value, progress, 1e-3)),
+        )