diff --git a/tests/tests_devices/test_grashopper_tomcat.py b/tests/tests_devices/test_grashopper_tomcat.py deleted file mode 100644 index 33467f4..0000000 --- a/tests/tests_devices/test_grashopper_tomcat.py +++ /dev/null @@ -1,222 +0,0 @@ -# pylint: skip-file -from unittest import mock - -import pytest - -from tomcat_bec.devices.grashopper_tomcat import ( - COLORMODE, - AutoMode, - DetectorState, - GrashopperError, - GrashopperTimeoutError, - GrashopperTOMCATSetup, - ImageBinning, - ImageMode, - MemoryPolling, - PixelFormat, - TriggerSource, - VideoMode, -) - - -@pytest.fixture(scope="function") -def mock_GrashopperSetup(): - mock_Grashopper = mock.MagicMock() - yield GrashopperTOMCATSetup(parent=mock_Grashopper) - - -# Fixture for scaninfo -@pytest.fixture( - params=[ - { - "scan_id": "1234", - "scan_type": "step", - "num_points": 500, - "frames_per_trigger": 1, - "exp_time": 0.1, - "readout_time": 0.1, - }, - { - "scan_id": "1234", - "scan_type": "step", - "num_points": 500, - "frames_per_trigger": 5, - "exp_time": 0.01, - "readout_time": 0, - }, - { - "scan_id": "1234", - "scan_type": "fly", - "num_points": 500, - "frames_per_trigger": 1, - "exp_time": 1, - "readout_time": 0.2, - }, - { - "scan_id": "1234", - "scan_type": "fly", - "num_points": 500, - "frames_per_trigger": 5, - "exp_time": 0.1, - "readout_time": 0.4, - }, - ] -) -def scaninfo(request): - return request.param - - -def test_initialize_detector(mock_GrashopperSetup): - """Test the initialize_detector method.""" - # Call the function you want to test - mock_GrashopperSetup.initialize_detector() - - # Assert the correct methods are called with the expected arguments - mock_GrashopperSetup.parent.cam.acquire.put.assert_called_once_with(0) - mock_GrashopperSetup.parent.cam.acquire_time_auto.put.assert_called_once_with( - AutoMode.CONTINUOUS - ) - mock_GrashopperSetup.parent.cam.gain_auto.put.assert_called_once_with(AutoMode.CONTINUOUS) - mock_GrashopperSetup.parent.cam.image_mode.put.assert_called_once_with(ImageMode.MULTIPLE) - mock_GrashopperSetup.parent.cam.image_binning.put.assert_called_once_with(ImageBinning.X1) - mock_GrashopperSetup.parent.cam.video_mode.put.assert_called_once_with(VideoMode.MODE0) - mock_GrashopperSetup.parent.cam.pixel_format.put.assert_called_once_with(PixelFormat.MONO16) - mock_GrashopperSetup.parent.cam.trigger_source.put.assert_called_once_with( - TriggerSource.SOFTWARE - ) - mock_GrashopperSetup.parent.cam.memory_polling.put.assert_called_once_with( - MemoryPolling.SECONDS1 - ) - mock_GrashopperSetup.parent.cam.set_image_counter.put.assert_called_once_with(0) - - -def test_initialize_detector_backend(mock_GrashopperSetup): - """Test the initialize_detector_backend method.""" - # Call the function you want to test - mock_GrashopperSetup.initialize_detector_backend() - - # Assert the correct methods are called with the expected arguments - mock_GrashopperSetup.parent.image.queue_size.put.assert_called_once_with(5) - mock_GrashopperSetup.parent.image.array_port.put.assert_called_once_with( - mock_GrashopperSetup.parent.cam.port_name.get() - ) - mock_GrashopperSetup.parent.image.enable_cb.put.assert_called_once_with(1) - mock_GrashopperSetup.parent.image.set_array_counter.put.assert_called_once_with(0) - - -@pytest.mark.parametrize("exposure_time", [1, 0.1, 0.01, 0.001, 0.0001]) -def test_set_exposure_time(mock_GrashopperSetup, exposure_time): - """Test the set_exposure_time method.""" - # Call the function you want to test - frame_rate = 1 / exposure_time - if frame_rate > mock_GrashopperSetup.low_frame_rate: - with pytest.raises(GrashopperError): - mock_GrashopperSetup.set_exposure_time(exposure_time) - else: - mock_GrashopperSetup.set_exposure_time(exposure_time) - mock_GrashopperSetup.parent.cam.frame_rate.put.assert_called_once_with(frame_rate) - - -def test_prepare_detector(mock_GrashopperSetup, scaninfo): - """Test the prepare_detector method.""" - # setup scaninof - for k, v in scaninfo.items(): - setattr(mock_GrashopperSetup.parent.scaninfo, k, v) - - # Call the function you want to test - with ( - mock.patch.object( - mock_GrashopperSetup, "set_acquisition_params" - ) as mock_set_acquisition_params, - mock.patch.object(mock_GrashopperSetup, "set_exposure_time") as mock_set_exposure_time, - ): - mock_GrashopperSetup.prepare_detector() - - # Assert the correct methods are called with the expected arguments - mock_GrashopperSetup.parent.cam.image_mode.put.assert_called_once_with(ImageMode.MULTIPLE) - mock_GrashopperSetup.parent.cam.acquire_time_auto.put.assert_called_once_with( - AutoMode.CONTINUOUS - ) - mock_GrashopperSetup.parent.cam.trigger_source.put.assert_called_once_with( - TriggerSource.SOFTWARE - ) - mock_GrashopperSetup.parent.cam.set_image_counter.put.assert_called_once_with(0) - mock_set_acquisition_params.assert_called_once() - mock_set_exposure_time.assert_called_once_with(scaninfo["exp_time"]) - - -def test_prepare_detector_backend(mock_GrashopperSetup): - """Test the prepare_detector_backend method.""" - # Call the function you want to test - mock_GrashopperSetup.prepare_detector_backend() - mock_GrashopperSetup.parent.image.set_array_counter.put.assert_called_once_with(0) - assert mock_GrashopperSetup.monitor_thread is None - assert not mock_GrashopperSetup.stop_monitor - - -@pytest.mark.parametrize("detector_state", [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10]) -def test_arm_acquisition(mock_GrashopperSetup, detector_state): - """Test the arm_acquisition method.""" - # Call the function you want to test - mock_GrashopperSetup.parent.cam.detector_state.get.return_value = detector_state - mock_GrashopperSetup.parent.stopped = False - mock_GrashopperSetup.parent.TIMEOUT_FOR_SIGNALS = 0.1 - mock_GrashopperSetup.parent.timeout = 0.1 - - if detector_state != DetectorState.WAITING: - with pytest.raises(GrashopperTimeoutError): - mock_GrashopperSetup.arm_acquisition() - assert mock_GrashopperSetup.parent.cam.acquire.put.call_count == 1 - assert mock_GrashopperSetup.parent.cam.acquire.put.call_args == mock.call(1) - else: - mock_GrashopperSetup.arm_acquisition() - assert mock_GrashopperSetup.parent.cam.acquire.put.call_count == 1 - assert mock_GrashopperSetup.parent.cam.acquire.put.call_args == mock.call(1) - - -@pytest.mark.parametrize("trigger_source", [0, 1, 2, 3]) -def test_on_trigger(mock_GrashopperSetup, trigger_source): - """Test the on_trigger method.""" - # Call the function you want to test - mock_GrashopperSetup.parent.cam.trigger_source.get.return_value = trigger_source - with mock.patch.object(mock_GrashopperSetup, "send_data") as mock_send_data: - mock_GrashopperSetup.on_trigger() - - # Assert the correct methods are called with the expected arguments - if trigger_source == TriggerSource.SOFTWARE: - mock_GrashopperSetup.parent.cam.software_trigger_device.put.assert_called_once_with(1) - assert mock_send_data.call_count == 1 - - -def test_set_acquisition_params(mock_GrashopperSetup, scaninfo): - """Test the set_acquisition_params method.""" - # Setup scaninfo - mock_GrashopperSetup.parent.scaninfo.num_points = scaninfo["num_points"] - mock_GrashopperSetup.parent.scaninfo.frames_per_trigger = scaninfo["frames_per_trigger"] - - # Call the function you want to test - mock_GrashopperSetup.set_acquisition_params() - - # Assert the correct methods are called with the expected arguments - expected_num_images = scaninfo["num_points"] * scaninfo["frames_per_trigger"] - mock_GrashopperSetup.parent.cam.num_images.put.assert_called_once_with(expected_num_images) - - -@pytest.mark.parametrize("detector_state", [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10]) -def test_stop_detector(mock_GrashopperSetup, detector_state): - """Test the arm_acquisition method.""" - # Call the function you want to test - mock_GrashopperSetup.parent.cam.detector_state.get.return_value = detector_state - mock_GrashopperSetup.parent.stopped = False - mock_GrashopperSetup.parent.TIMEOUT_FOR_SIGNALS = 0.1 - mock_GrashopperSetup.parent.timeout = 0.1 - - if detector_state != DetectorState.IDLE: - with pytest.raises(GrashopperTimeoutError): - mock_GrashopperSetup.stop_detector() - assert mock_GrashopperSetup.parent.cam.acquire.put.call_count == 2 - assert mock_GrashopperSetup.parent.cam.acquire.put.call_args == mock.call(0) - else: - mock_GrashopperSetup.stop_detector() - assert mock_GrashopperSetup.parent.cam.acquire.put.call_count == 1 - assert mock_GrashopperSetup.parent.cam.acquire.put.call_args == mock.call(0) diff --git a/tomcat_bec/devices/__init__.py b/tomcat_bec/devices/__init__.py index 6590b8d..e69de29 100644 --- a/tomcat_bec/devices/__init__.py +++ b/tomcat_bec/devices/__init__.py @@ -1,12 +0,0 @@ -from .aerotech import ( - aa1AxisDriveDataCollection, - aa1AxisPsoDistance, - aa1Controller, - aa1GlobalVariableBindings, - aa1GlobalVariables, - aa1Tasks, -) -from .gigafrost.gigafrostcamera import GigaFrostCamera -from .grashopper_tomcat import GrashopperTOMCAT -from .pco_edge.pcoedgecamera import PcoEdge5M -from .psimotor import EpicsMotorEC, EpicsMotorMR diff --git a/tomcat_bec/devices/grashopper_tomcat.py b/tomcat_bec/devices/grashopper_tomcat.py deleted file mode 100644 index cbf39db..0000000 --- a/tomcat_bec/devices/grashopper_tomcat.py +++ /dev/null @@ -1,436 +0,0 @@ -import enum -import threading -import time as ttime - -from bec_lib.logger import bec_logger - -# from typing import Any -from ophyd import ADComponent as ADCpt -from ophyd import Component as Cpt -from ophyd import Device, EpicsSignal, EpicsSignalRO, EpicsSignalWithRBV - -# import numpy as np -from ophyd.ophydobj import Kind -from tomcat_bec.devices.unsupported_legacy.psi_detector_base import ( - CustomDetectorMixin, - PSIDetectorBase, -) - -# os.environ["EPICS_CA_AUTO_ADDR_LIST"] = "No" -# os.environ["EPICS_CA_ADDR_LIST"] = "129.129.208.143" - - -logger = bec_logger.logger - - -class GrashopperError(Exception): - """Base class for Grashopper detector errors.""" - - -class GrashopperTimeoutError(GrashopperError): - """Base class for Grashopper detector errors.""" - - -class AutoMode(enum.IntEnum): - """Acquire time auto for Grashopper detector. - - class for acquire_auto and gain_auto - - Off: Gain tap balancing is user controlled using Gain. - Once: Gain tap balancing is automatically adjusted once by the device. - Once it has converged, it automatically returns to the Off state. - Continuous: Gain tap balancing is constantly adjusted by the device. - """ - - OFF = 0 - ONCE = 1 - CONTINUOUS = 2 - - -class ImageMode(enum.IntEnum): - """Image mode for Grashopper detector. - - Single: Acquire a single image, ignores NumImages PV - Multiple: Acquire NumImages images - Continuous: Acquire images continuously - """ - - SINGLE = 0 - MULTIPLE = 1 - CONTINUOUS = 2 - - -class DetectorState(enum.IntEnum): - """Detector states for Grashopper detector""" - - IDLE = 0 - ACQUIRE = 1 - READOUT = 2 - CORRECT = 3 - SAVING = 4 - ABORTING = 5 - ERROR = 6 - WAITING = 7 - INITIALIZING = 8 - DISCONNECTED = 9 - ABORTED = 10 - - -class ImageBinning(enum.IntEnum): - """Image binning for Grashopper detector""" - - X1 = 1 - X2 = 2 - X4 = 4 - - -class VideoMode(enum.IntEnum): - """Video mode for Grashopper detector. - - For details, consult EPICs IOC manual. - """ - - MODE0 = 0 - MODE1 = 1 - MODE2 = 2 - MODE3 = 3 - - -class PixelFormat(enum.IntEnum): - """Pixel format for Grashopper detector.""" - - MONO8 = 0 - MONO12PACKED = 1 - MONO12P = 2 - MONO16 = 3 - - -class COLORMODE(enum.IntEnum): - """Color mode for Grashopper detector. - - Only for readback values from color_mode RO PV. - """ - - MONO = 0 - BAYER = 1 - RGB1 = 2 - RGB2 = 3 - RGB3 = 4 - YUV444 = 5 - YUV422 = 6 - YUV421 = 7 - - -class TriggerSource(enum.IntEnum): - """Trigger signals for Grashopper detector""" - - SOFTWARE = 0 - LINE0 = 1 - LINE2 = 2 - LINE3 = 3 - - -class MemoryPolling(enum.IntEnum): - """Memory polling for Grashopper detector. - - Defines update rate of memory polling for IOC (1s suggested). - """ - - PASSIVE = 0 - EVENT = 1 - IOINTR = 2 - SECONDS10 = 3 - SECONDS5 = 4 - SECONDS2 = 5 - SECONDS1 = 6 - SECONDS05 = 7 - SECONDS02 = 8 - SECONDS01 = 9 - - -class GrashopperTOMCATSetup(CustomDetectorMixin): - """Mixin class to setup TOMCAT specific implementations of the detector. - - This class will be called by the custom_prepare_cls attribute of the detector class. - """ - - def __init__(self, *_args, parent: Device = None, **_kwargs) -> None: - super().__init__(*_args, parent=parent, **_kwargs) - self._rlock = threading.RLock() - - self.image_shape = (self.parent.cam.image_size_y.get(), self.parent.cam.image_size_x.get()) - self.monitor_thread = None - self.stop_monitor = False - self.update_frequency = 1 - self.low_frame_rate = 80 - - def on_init(self) -> None: - """Run on initialization of the detector.""" - self.initialize_detector() - self.initialize_detector_backend() - - def initialize_detector(self) -> None: - """Initialize detector.""" - self.parent.cam.acquire.put(0) - self.parent.cam.acquire_time_auto.put(AutoMode.CONTINUOUS) - self.parent.cam.gain_auto.put(AutoMode.CONTINUOUS) - self.parent.cam.image_mode.put(ImageMode.MULTIPLE) - self.parent.cam.image_binning.put(ImageBinning.X1) - self.parent.cam.video_mode.put(VideoMode.MODE0) - self.parent.cam.pixel_format.put(PixelFormat.MONO16) - self.parent.cam.trigger_source.put(TriggerSource.SOFTWARE) - self.parent.cam.memory_polling.put(MemoryPolling.SECONDS1) - self.parent.cam.set_image_counter.put(0) - - def initialize_detector_backend(self) -> None: - self.parent.image.queue_size.put(5) - self.parent.image.array_port.put(self.parent.cam.port_name.get()) - self.parent.image.enable_cb.put(1) - self.parent.image.set_array_counter.put(0) - - def on_stage(self) -> None: - """Run on stage of the detector.""" - self.prepare_detector() - self.prepare_detector_backend() - self.arm_acquisition() - - def set_exposure_time(self, exposure_time: float) -> None: - """Set the detector framerate. - - Args: - framerate (float): Desired framerate in Hz smallest is 87Hz - """ - framerate = 1 / exposure_time - if framerate > self.low_frame_rate: - raise GrashopperError( - f"Trying to set exposure time to {exposure_time}s, this is below the lowest" - f" possible exposure of {1/self.low_frame_rate}s" - ) - self.parent.cam.frame_rate.put(framerate) - - def prepare_detector(self) -> None: - """Prepare detector for acquisition.""" - self.parent.cam.image_mode.put(ImageMode.MULTIPLE) - self.parent.cam.acquire_time_auto.put(AutoMode.CONTINUOUS) - self.set_exposure_time(self.parent.scaninfo.exp_time) - self.parent.cam.trigger_source.put(TriggerSource.SOFTWARE) - self.parent.cam.set_image_counter.put(0) - self.set_acquisition_params() - - def set_acquisition_params(self) -> None: - """Set acquisition parameters for the detector""" - - # Set number of images and frames (frames is for internal burst of detector) - self.parent.cam.num_images.put( - int(self.parent.scaninfo.num_points * self.parent.scaninfo.frames_per_trigger) - ) - - def prepare_detector_backend(self) -> None: - """Prepare detector backend for acquisition.""" - self.parent.image.set_array_counter.put(0) - self.monitor_thread = None - self.stop_monitor = False - # self.run_monitor() - - def arm_acquisition(self) -> None: - """Arm grashopper detector for acquisition""" - self.parent.cam.acquire.put(1) - signal_conditions = [(self.parent.cam.detector_state.get, DetectorState.WAITING)] - if not self.wait_for_signals( - signal_conditions=signal_conditions, - timeout=self.parent.TIMEOUT_FOR_SIGNALS, - check_stopped=True, - all_signals=False, - ): - raise GrashopperTimeoutError( - f"Failed to arm the acquisition. Detector state {signal_conditions[0][0]}" - ) - - def on_trigger(self) -> None: - """Trigger the detector""" - if self.parent.cam.trigger_source.get() == TriggerSource.SOFTWARE: - self.parent.cam.software_trigger_device.put(1) - ttime.sleep(0.1) - self.send_data() - - def run_monitor(self) -> None: - """ - Run the monitor loop in a separate thread. - """ - self.monitor_thread = threading.Thread(target=self.monitor_loop, daemon=True) - self.monitor_thread.start() - - def monitor_loop(self) -> None: - """ - Monitor the detector status and send data. - """ - while True: - self.send_data() - ttime.sleep(1 / self.update_frequency) - if self.parent.stopped: - break - - def send_data(self) -> None: - """Send data to monitor endpoint in redis.""" - try: - img = self.parent.image.array_data.get().reshape(self.image_shape) - # pylint: disable=protected-access - self.parent._run_subs(sub_type=self.parent.SUB_VALUE, value=img) - except Exception as e: - logger.debug(f"{e} for image with shape {self.parent.image.array_data.get().shape}") - - def on_unstage(self) -> None: - """Run on unstage of the detector.""" - self.stop_monitor = True - - def on_stop(self) -> None: - """Run on stop of the detector.""" - self.stop_detector() - self.stop_detector_backend() - - def stop_detector(self) -> None: - """Stop detector.""" - self.parent.cam.acquire.put(0) - signal_conditions = [(self.parent.cam.detector_state.get, DetectorState.IDLE)] - if not self.wait_for_signals( - signal_conditions=signal_conditions, - timeout=self.parent.TIMEOUT_FOR_SIGNALS - self.parent.TIMEOUT_FOR_SIGNALS // 2, - check_stopped=True, - all_signals=False, - ): - # Retry stop detector and wait for remaining time - self.parent.cam.acquire.put(0) - if not self.wait_for_signals( - signal_conditions=signal_conditions, - timeout=self.parent.TIMEOUT_FOR_SIGNALS - self.parent.TIMEOUT_FOR_SIGNALS // 2, - check_stopped=True, - all_signals=False, - ): - raise GrashopperTimeoutError( - f"Failed to stop detector, detector state {signal_conditions[0][0]}" - ) - - def stop_detector_backend(self) -> None: - """Stop the data backend sending data.""" - self.stop_monitor = True - - -class SLSDetectorCam(Device): - """ - SLS Detector Camera - Grashoppter - - Base class to map EPICS PVs to ophyd signals. - """ - - ## Deprecated PVs, to be checked! - # acquire_time = ADCpt(EpicsSignal, "AcquireTime", kind=Kind.omitted) - # num_exposures = ADCpt(EpicsSignal, "NumExposures", kind=Kind.omitted) - # acquire_period = ADCpt(EpicsSignalWithRBV, "AcquirePeriod", kind=Kind.config) - - # Control PVs - acquire_time_auto = ADCpt(EpicsSignal, "AcquireTimeAuto", kind=Kind.config) - - frame_rate = ADCpt(EpicsSignalWithRBV, "FrameRate", kind=Kind.normal) - num_images = ADCpt(EpicsSignalWithRBV, "NumImages", kind=Kind.normal) - num_images_counter = ADCpt(EpicsSignalRO, "NumImagesCounter_RBV", kind=Kind.normal) - image_mode = ADCpt(EpicsSignalWithRBV, "ImageMode", kind=Kind.config) - acquire = ADCpt(EpicsSignalWithRBV, "Acquire", kind=Kind.config) - detector_state = ADCpt(EpicsSignalRO, "DetectorState_RBV", kind=Kind.normal) - status_message = ADCpt(EpicsSignalRO, "StatusMessage_RBV", string=True, kind=Kind.config) - - set_image_counter = ADCpt(EpicsSignal, "ArrayCounter", kind=Kind.config) - image_counter = ADCpt(EpicsSignal, "ArrayCounter_RBV", kind=Kind.normal) - image_rate = ADCpt(EpicsSignalRO, "ArrayRate_RBV", kind=Kind.omitted) - - # Asyn Port name - port_name = ADCpt(EpicsSignalRO, "PortName_RBV", string=True, kind=Kind.omitted) - - # Readout related PVs - max_image_size_x = ADCpt(EpicsSignalRO, "MaxSizeX_RBV", kind=Kind.config) - max_image_size_y = ADCpt(EpicsSignalRO, "MaxSizeY_RBV", kind=Kind.config) - image_size_x = ADCpt(EpicsSignalRO, "ArraySizeX_RBV", kind=Kind.config) - image_size_y = ADCpt(EpicsSignalRO, "ArraySizeY_RBV", kind=Kind.config) - # Only BinY PV is working, sets both - image_binning = ADCpt(EpicsSignalWithRBV, "BinY", kind=Kind.config) - - gain = ADCpt(EpicsSignalWithRBV, "Gain", kind=Kind.config) - gain_auto = ADCpt(EpicsSignalWithRBV, "GainAuto", kind=Kind.config) - video_mode = ADCpt(EpicsSignalWithRBV, "VideoMode", kind=Kind.config) - pixel_format = ADCpt(EpicsSignalWithRBV, "PixelFormat", kind=Kind.config) - # Desired to set this in future? - color_mode = ADCpt(EpicsSignalRO, "ColorMode_RBV", kind=Kind.config) - - # HW Status PVs - temperature_actual = ADCpt(EpicsSignal, "TemperatureActual", kind=Kind.omitted) - - # Trigger - trigger_mode_active = ADCpt(EpicsSignalWithRBV, "TriggerMode", kind=Kind.config) - trigger_source = ADCpt(EpicsSignalWithRBV, "TriggerSource", kind=Kind.config) - trigger_delay = ADCpt(EpicsSignalWithRBV, "TriggerDelay", kind=Kind.omitted) - exposure_mode = ADCpt(EpicsSignalWithRBV, "ExposureMode", kind=Kind.omitted) - software_trigger_device = ADCpt(EpicsSignal, "SoftwareTrigger", kind=Kind.config) - - # buffer - memory_polling = ADCpt(EpicsSignal, "PoolUsedMem.SCAN", kind=Kind.omitted) - - -class SLSImagePlugin(Device): - """SLS Image Plugin - - Image plugin for SLS detector imitating the behaviour of ImagePlugin from - ophyd's areadetector plugins. - """ - - # Control - array_port = Cpt(EpicsSignal, "NDArrayPort", kind=Kind.omitted, string=True) - enable_cb = Cpt(EpicsSignal, "EnableCallbacks", kind=Kind.config) - queue_size = Cpt(EpicsSignal, "QueueSize", kind=Kind.config) - set_array_counter = Cpt(EpicsSignal, "ArrayCounter", kind=Kind.config) - array_counter = Cpt(EpicsSignal, "ArrayCounter_RBV", kind=Kind.normal) - set_dropped_arrays = Cpt(EpicsSignal, "DroppedArrays", kind=Kind.config) - dropped_arrays = Cpt(EpicsSignal, "DroppedArrays_RBV", kind=Kind.normal) - image_id = Cpt(EpicsSignal, "UniqueId_RBV", kind=Kind.normal) - - # Data - array_data = Cpt(EpicsSignal, "ArrayData", kind=Kind.omitted) - - # Size related PVs from Plugin - array_size_0 = Cpt(EpicsSignalRO, "ArraySize0_RBV", kind=Kind.omitted) - array_size_1 = Cpt(EpicsSignalRO, "ArraySize1_RBV", kind=Kind.omitted) - array_size_2 = Cpt(EpicsSignalRO, "ArraySize2_RBV", kind=Kind.omitted) - array_dimension_size = Cpt(EpicsSignalRO, "NDimensions_RBV", kind=Kind.omitted) - - -class GrashopperTOMCAT(PSIDetectorBase): - """ - Grashopper detector for TOMCAT - - Parent class: PSIDetectorBase - - class attributes: - custom_prepare_cls (GrashopperTOMCATSetup) : Custom detector setup class for TOMCAT, - inherits from CustomDetectorMixin - cam (SLSDetectorCam) : Detector camera - image (SLSImagePlugin) : Image plugin for detector - """ - - # Specify which functions are revealed to the user in BEC client - USER_ACCESS = [] - - SUB_MONITOR = "monitor" - SUB_VALUE = "value" - _default_sub = SUB_VALUE - - # specify Setup class - custom_prepare_cls = GrashopperTOMCATSetup - # specify minimum readout time for detector - MIN_READOUT = 0 - TIMEOUT_FOR_SIGNALS = 5 - # specify class attributes - cam = ADCpt(SLSDetectorCam, "cam1:") - image = ADCpt(SLSImagePlugin, "image1:") - - -if __name__ == "__main__": - hopper = GrashopperTOMCAT(name="hopper", prefix="X02DA-PG-USB:", sim_mode=True) - hopper.wait_for_connection(all_signals=True) diff --git a/tomcat_bec/devices/tomcat_rotation_motors.py b/tomcat_bec/devices/tomcat_rotation_motors.py deleted file mode 100644 index 7085bc2..0000000 --- a/tomcat_bec/devices/tomcat_rotation_motors.py +++ /dev/null @@ -1,181 +0,0 @@ -"""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 ophyd import DeviceStatus -from tomcat_bec.devices.rotation_base.ophyd_rotation_base import EpicsRotationBase -from ophyd_devices.interfaces.protocols.bec_protocols import BECFlyerProtocol, BECScanProtocol - - -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) - - 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""" - with self._lock: - 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)), - )