bec/superxas_bec
0
0
Fork 0
Code Issues 2 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

refactor(falcon): cleanup and add tests

Browse Source
This commit is contained in:
appel_c
2025-03-22 11:23:26 +01:00
parent 5a0702f918
commit cd91375f34
2 changed files with 180 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

114
superxas_bec/devices/falcon.py
View File

@@ -1,76 +1,88 @@
from ophyd_devices.devices.dxp import Falcon, EpicsMCARecord, EpicsDXPFalcon
from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase
from ophyd import DeviceStatus, StatusBase, EpicsSignalRO, Kind, Signal, Component as Cpt
"""FALCON device implementation for SuperXAS"""
import enum
import numpy as np
from bec_lib.devicemanager import ScanInfo
from bec_lib.logger import bec_logger
from ophyd import Component as Cpt
from ophyd import DeviceStatus, EpicsSignalRO, Kind, Signal, StatusBase
from ophyd.device import DynamicDeviceComponent as DCpt
from ophyd.mca import add_rois
from ophyd.status import SubscriptionStatus
from bec_lib.devicemanager import ScanInfo
import enum
import math
import numpy as np
from bec_lib.logger import bec_logger
from ophyd_devices.devices.dxp import EpicsDXPFalcon, EpicsMCARecord, Falcon
from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase
logger = bec_logger.logger
class FalconAcquiringStatus(int, enum.Enum):
""" Status of Falcon"""
"""Status of Falcon"""
DONE = 0
ACQUIRING = 1
class DeadTimeCorrectedCounts(Signal):
"""Signal to calculate dead time corrected counts"""
def __init__(self, name:str, channel:int, **kwargs):
def __init__(self, name: str, channel: int, **kwargs):
super().__init__(name=name, **kwargs)
self._channel = channel
self._dead_time = 1.182e-7
# pylint: disable=arguments-differ
def get(self) -> float:
dxp:EpicsDXPFalconSuperXAS = getattr(self.parent, f"dxp{self._channel}")
mca:EpicsMCARecordSuperXAS = getattr(self.parent, f"mca{self._channel}")
"""Get dead time corrected counts base on signals from dxp and mca of Falcon"""
dxp: EpicsDXPFalconSuperXAS = getattr(self.parent, f"dxp{self._channel}")
mca: EpicsMCARecordSuperXAS = getattr(self.parent, f"mca{self._channel}")
icr = dxp.input_count_rate.get()
ocr = dxp.output_count_rate.get()
roi = mca.rois.roi0.count.get()
ert = mca.elapsed_real_time.get()
print(icr,ocr,roi,ert)
print(icr, ocr, roi, ert)
if icr == 0 or ocr ==0:
if icr == 0 or ocr == 0:
return 0
# Check that relative change is large enough
test = 1e9
test_icr = icr
n = 0
while test > self._dead_time and n < 30:
try:
true_icr = icr*np.exp(test_icr * self._dead_time)
test = (true_icr - test_icr) / test_icr
true_icr = icr * np.exp(test_icr * self._dead_time)
test = (true_icr - test_icr) / test_icr
test_icr = true_icr
n +=1
except Exception as e:
logger.info(f"Error in computation of signal {self.name}")
n += 1
except Exception as e: # pylint: disable=broad-except
logger.info(f"Error in computation of signal {self.name}, error: {e}")
return 0
# Return corrected roi counts
cor_roi_cnts = 0
if ocr *ert != 0:
if ocr * ert != 0:
cor_roi_cnts = roi * true_icr / (ocr * ert)
return cor_roi_cnts
class EpicsDXPFalconSuperXAS(EpicsDXPFalcon):
"""DXPFalcon class wrapper for SuperXAS."""
_default_read_attrs = ['input_count_rate', 'output_count_rate']
_default_read_attrs = [
"input_count_rate",
"output_count_rate",
] # add exposable signals/subdevices here
input_count_rate = Cpt(EpicsSignalRO, "InputCountRate", kind=Kind.normal, auto_monitor=True)
output_count_rate = Cpt(EpicsSignalRO, "OutputCountRate", kind=Kind.normal, auto_monitor=True)
class EpicsMCARecordSuperXAS(EpicsMCARecord):
_default_read_attrs = ['rois']
class EpicsMCARecordSuperXAS(EpicsMCARecord):
"""MCARecord class wrapper for SuperXAS."""
_default_read_attrs = ["rois"] # add exposable signals/subdevices here
elapsed_real_time = Cpt(EpicsSignalRO, ".ERTM", kind=Kind.normal, auto_monitor=True)
rois = DCpt(add_rois(range(0, 3), kind=Kind.normal), kind=Kind.normal)
@@ -78,9 +90,13 @@ class EpicsMCARecordSuperXAS(EpicsMCARecord):
class FalconControl(Falcon):
""" Falcon Control class at SuperXAS. prefix: 'X10DA-SITORO:'"""
"""Falcon Control class at SuperXAS. prefix: 'X10DA-SITORO:'"""
_default_read_attrs = ['mca1', 'dxp1', "dead_time_cor_cnts1"]
_default_read_attrs = [
"mca1",
"dxp1",
"dead_time_cor_cnts1",
] # add exposable signals/subdevices here
# DXP parameters
dxp1 = Cpt(EpicsDXPFalconSuperXAS, "dxp1:")
@@ -89,7 +105,6 @@ class FalconControl(Falcon):
# dxp4 = Cpt(EpicsDXPFalconSuperXAS, "dxp4:")
# dxp5 = Cpt(EpicsDXPFalconSuperXAS, "dxp5:")
# MCA record with spectrum data
mca1 = Cpt(EpicsMCARecordSuperXAS, "mca1")
# mca2 = Cpt(EpicsMCARecord, "mca2")
@@ -97,21 +112,33 @@ class FalconControl(Falcon):
# mca4 = Cpt(EpicsMCARecord, "mca4")
# mca5 = Cpt(EpicsMCARecord, "mca5")
#Norm Signal
dead_time_cor_cnts1 = Cpt(DeadTimeCorrectedCounts, name='dead_time_cor_cnts', channel=1, kind=Kind.hinted)
# Norm Signal
dead_time_cor_cnts1 = Cpt(
DeadTimeCorrectedCounts, name="dead_time_cor_cnts", channel=1, kind=Kind.hinted
)
# dead_time_cor_cnts2 = Cpt(DeadTimeCorrectedCounts, name='dead_time_cor_cnts', channel=2, kind=Kind.normal)
# dead_time_cor_cnts3 = Cpt(DeadTimeCorrectedCounts, name='dead_time_cor_cnts', channel=3, kind=Kind.normal)
# dead_time_cor_cnts4 = Cpt(DeadTimeCorrectedCounts, name='dead_time_cor_cnts', channel=4, kind=Kind.normal)
# dead_time_cor_cnts5 = Cpt(DeadTimeCorrectedCounts, name='dead_time_cor_cnts', channel=5, kind=Kind.normal)
class FalconSuperXAS(PSIDeviceBase, FalconControl):
""" Falcon implementierung at SuperXAS. prefix: 'X10DA-SITORO:'"""
"""Falcon implementierung at SuperXAS. prefix: 'X10DA-SITORO:'"""
########################################
# Beamline Specific Implementations #
########################################
def __init__(self, name: str, prefix:str='',scan_info: ScanInfo | None = None, **kwargs):
def __init__(self, name: str, prefix: str = "", scan_info: ScanInfo | None = None, **kwargs):
"""
Initialize Falcon device.
Args:
name (str): Name of the device
prefix (str): Prefix of the device
scan_info (ScanInfo): Information about the scan
**kwargs: Additional keyword arguments
"""
super().__init__(name=name, prefix=prefix, scan_info=scan_info, **kwargs)
self._pv_timeout = 1
@@ -138,13 +165,25 @@ class FalconSuperXAS(PSIDeviceBase, FalconControl):
self.collect_mode.set(0).wait()
self.preset_real_time.set(0).wait()
self.stop_all.put(1)
self.wait_for_condition(lambda: self.acquiring.get() == FalconAcquiringStatus.DONE, timeout=self._pv_timeout)
if (
self.wait_for_condition(
lambda: self.acquiring.get() == FalconAcquiringStatus.DONE, timeout=self._pv_timeout
)
is False
):
raise TimeoutError("Timeout on Falcon stage")
def on_unstage(self) -> DeviceStatus | StatusBase | None:
"""Called while unstaging the device."""
self.stop_all.put(1)
self.erase_all.put(1)
self.wait_for_condition(lambda: self.acquiring.get() == FalconAcquiringStatus.DONE, timeout=self._pv_timeout)
if (
self.wait_for_condition(
lambda: self.acquiring.get() == FalconAcquiringStatus.DONE, timeout=self._pv_timeout
)
is False
):
raise TimeoutError("Timeout on Falcon unstage")
def on_pre_scan(self) -> DeviceStatus | StatusBase | None:
"""Called right before the scan starts on all devices automatically."""
@@ -164,7 +203,7 @@ class FalconSuperXAS(PSIDeviceBase, FalconControl):
def _stop_erase_and_wait_for_acquiring(self) -> DeviceStatus:
"""Method called from the Trigger card to reset counts on the Falcon"""
if self.acquiring.get() != FalconAcquiringStatus.DONE:
self.stop_all.put(1)
@@ -172,8 +211,9 @@ class FalconSuperXAS(PSIDeviceBase, FalconControl):
if old_value == FalconAcquiringStatus.DONE and value == FalconAcquiringStatus.ACQUIRING:
return True
return False
status = SubscriptionStatus(self.acquiring, _check_acquiriting)
logger.info("Triggering Falcon")
self.erase_start.put(1)
return status
return status

103
tests/tests_devices/test_devices_falcon.py Normal file
View File

@@ -0,0 +1,103 @@
"""Tests for Falcon device."""
import threading
from unittest import mock
import ophyd
import pytest
from ophyd import DeviceStatus
from ophyd_devices.tests.utils import MockPV, patch_dual_pvs
from superxas_bec.devices.falcon import FalconAcquiringStatus, FalconSuperXAS
# pylint: disable=protected-access
@pytest.fixture(scope="function")
def falcon():
"""Trigger device with mocked EPICS PVs."""
name = "falcon"
prefix = "X10DA-SITORO:"
with mock.patch.object(ophyd, "cl") as mock_cl:
mock_cl.get_pv = MockPV
mock_cl.thread_class = threading.Thread
dev = FalconSuperXAS(name=name, prefix=prefix)
patch_dual_pvs(dev)
yield dev
def test_devices_falcon(falcon):
"""Test init and on_connected methods of Falcon device"""
assert falcon.prefix == "X10DA-SITORO:"
assert falcon.name == "falcon"
assert falcon._pv_timeout == 1
falcon.on_connected()
def test_devices_falcon_stage(falcon):
"""Test on_stage method of Falcon device"""
falcon.collect_mode.put(1)
falcon.preset_real_time.put(1)
falcon.stop_all.put(0)
falcon.acquiring.put(FalconAcquiringStatus.DONE)
# Should resolve with that status
falcon.on_stage()
assert falcon.collect_mode.get() == 0
assert falcon.preset_real_time.get() == 0
assert falcon.stop_all.get() == 1
# Should timeout
falcon.acquiring.put(FalconAcquiringStatus.ACQUIRING)
falcon._pv_timeout = 0.1
with pytest.raises(TimeoutError):
falcon.on_stage()
def test_devices_falcon_unstage(falcon):
"""Test on_unstage method of Falcon device"""
falcon.stop_all.put(0)
falcon.erase_all.put(0)
falcon.acquiring.put(FalconAcquiringStatus.DONE)
# Should resolve with that status
falcon.on_unstage()
assert falcon.stop_all.get() == 1
assert falcon.erase_all.get() == 1
# Should timeout
falcon.acquiring.put(FalconAcquiringStatus.ACQUIRING)
falcon._pv_timeout = 0.1
with pytest.raises(TimeoutError):
falcon.on_unstage()
def test_devices_falcon_stop(falcon):
"""Test on_stop method of Falcon device"""
assert falcon.stopped is False
falcon.stop_all.put(0)
falcon.stop()
assert falcon.stopped is True
assert falcon.stop_all.get() == 1
def test_devices_falcon_stop_erase_and_wait_for_acquiring(falcon):
"""
Test _stop_erase_and_wait_for_acquiring method of Falcon device.
This method is called by the trigger card when the Falcon needs to be reset, and
placed in a state where it can receive a trigger again
"""
# Set initial values to different states
falcon.stop_all.put(0)
falcon.erase_start.put(0)
falcon.acquiring.put(FalconAcquiringStatus.ACQUIRING)
# If falcon status is acquiring, it should call stop_all
status = falcon._stop_erase_and_wait_for_acquiring()
assert falcon.stop_all.get() == 1
assert falcon.erase_start.get() == 1
# The status resolved once it sees acquiring change from DONE to ACQUIRING
assert status.done is False
falcon.acquiring.put(FalconAcquiringStatus.DONE)
falcon.acquiring.set(FalconAcquiringStatus.ACQUIRING).wait()
assert status.done is True