bec/tomcat_bec
0
0 Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

chore: remove legacy devices

Browse Source
This commit is contained in:
wakonig_k committed 2026-09-30 15:41:44 +02:00
1 parent 1b17570095
commit d5cb6ff66f
4 files changed
-851
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/tests_devices/test_grashopper_tomcat.py
-222
View File
@@ -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)
tomcat_bec/devices/__init__.py
-12
View File
@@ -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
tomcat_bec/devices/grashopper_tomcat.py
-436
View File
@@ -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)
tomcat_bec/devices/tomcat_rotation_motors.py
-181
View File
@@ -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)),
)