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

wip falcon slim

Browse Source
This commit is contained in:
appel_c
2025-06-24 17:02:33 +02:00
parent 2a8c8b6f1c
commit a7d3bd3ea0
3 changed files with 182 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
superxas_bec/devices/falcon.py
View File

@@ -2,15 +2,16 @@
import enum
import time
import numpy as np
from bec_lib.devicemanager import ScanInfo
from bec_lib.endpoints import MessageEndpoints
from bec_lib.logger import bec_logger
from bec_lib.messages import DeviceMessage
from bec_lib.endpoints import MessageEndpoints
from ophyd_devices import CompareStatus, TransitionStatus
from ophyd import Component as Cpt
from ophyd import DeviceStatus, Kind, Signal, StatusBase, Staged
from ophyd import DeviceStatus, Kind, Signal, Staged, StatusBase
from ophyd.status import SubscriptionStatus
from ophyd_devices import CompareStatus, TransitionStatus
from ophyd_devices.devices.dxp import EpicsDXPFalcon, EpicsMCARecord, Falcon
from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase
@@ -23,7 +24,8 @@ class FalconAcquiringStatus(int, enum.Enum):
DONE = 0
ACQUIRING = 1
def compute_dead_time_corrected_signal(icr:float, ocr:float, roi:float, ert:float):
def compute_dead_time_corrected_signal(icr: float, ocr: float, roi: float, ert: float):
_dead_time = 1.182e-7
if icr == 0 or ocr == 0:
return 0
@@ -89,8 +91,7 @@ class FalconSuperXAS(PSIDeviceBase, FalconControl):
preview = Cpt(
Signal, name="preview", kind=Kind.omitted, doc="Preview signal for Falcon detector"
)
icr = Cpt(
Signal, name="icr", kind=Kind.normal)
icr = Cpt(Signal, name="icr", kind=Kind.normal)
ocr = Cpt(Signal, name="icr", kind=Kind.normal)
elap_real_time = Cpt(Signal, name="icr", kind=Kind.normal)
roi0_count = Cpt(Signal, name="icr", kind=Kind.normal)
@@ -98,14 +99,21 @@ class FalconSuperXAS(PSIDeviceBase, FalconControl):
Signal,
name="dead_cor_roi0_count",
doc="Async dead time corrected ROI 0 count signal",
kind=Kind.normal
kind=Kind.normal,
)
########################################
# Beamline Specific Implementations #
########################################
def __init__(self, name: str, prefix: str = "", scan_info: ScanInfo | None = None, device_manager=None, **kwargs):
def __init__(
self,
name: str,
prefix: str = "",
scan_info: ScanInfo | None = None,
device_manager=None,
**kwargs,
):
"""
Initialize Falcon device.
@@ -115,7 +123,9 @@ class FalconSuperXAS(PSIDeviceBase, FalconControl):
scan_info (ScanInfo): Information about the scan
**kwargs: Additional keyword arguments
"""
super().__init__(name=name, prefix=prefix, scan_info=scan_info, device_manager=device_manager, **kwargs)
super().__init__(
name=name, prefix=prefix, scan_info=scan_info, device_manager=device_manager, **kwargs
)
self.device_manager = device_manager
self.mca1.stage_sigs = {}
self._pv_timeout = 5
@@ -195,14 +205,14 @@ class FalconSuperXAS(PSIDeviceBase, FalconControl):
logger.info(f"Sending data for {self.name} at {time_started}")
icr = self.dxp1.input_count_rate.get()
ocr = self.dxp1.output_count_rate.get()
roi = self.mca1.elapsed_real_time.get()
roi = self.mca1.rois.count.get()
ert = self.mca1.elapsed_real_time.get()
self.icr.put(icr)
logger.info(f"Data to plot {self.icr.get()}")
self.ocr.put(ocr)
self.elap_real_time.put(ert)
self.roi0_count.put(roi)
self.dead_cor_roi0_count.put(compute_dead_time_corrected_signal(icr,ocr,roi,ert))
self.dead_cor_roi0_count.put(compute_dead_time_corrected_signal(icr, ocr, roi, ert))
# self._send_preview_async()
logger.info(f"Data sent for {self.name} at {time.time()- time_started}")

150
superxas_bec/devices/falcon_slim.py Normal file
View File

@@ -0,0 +1,150 @@
import enum
import numpy as np
from bec_lib.logger import bec_logger
from ophyd import Component as Cpt
from ophyd import Device, EpicsSignal, EpicsSignalRO, EpicsSignalWithRBV, Kind, Signal
from ophyd_devices import CompareStatus, DeviceStatus, PreviewSignal, StatusBase
from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase
logger = bec_logger.logger
class FalconAcquiringStatus(int, enum.Enum):
"""Status of Falcon"""
DONE = 0
ACQUIRING = 1
class FalconControlSlim(Device):
"""Slim Falcon Control class for SuperXAS. prefix: 'X10DA-SITORO:'"""
start = Cpt(EpicsSignal, "EraseStart", doc="XMAP start signal")
stop = Cpt(EpicsSignal, "StopAll", doc="XMAP stop signal")
acquiring = Cpt(EpicsSignalRO, "Acquiring", auto_monitor=True, doc="XMAP acquiring signal")
icr = Cpt(EpicsSignalRO, "dxp1:InputCountRate", doc="XMAP input count rate")
ocr = Cpt(EpicsSignalRO, "dxp1:OutputCountRate", doc="XMAP output count rate")
ert = Cpt(EpicsSignalRO, "mca1.ERTM", doc="XMAP elapsed real time")
roi = Cpt(EpicsSignalRO, "mca1.R0", kind=Kind.hinted, doc="XMAP ROI signal")
label = Cpt(EpicsSignalRO, "mca1.R0NM", kind=Kind.config, doc="XMAP ROI signal")
spectrum_val = Cpt(EpicsSignalRO, "mca1.VAL", kind=Kind.omitted)
# Preview Signal for Falcon detector
spectrum = Cpt(
PreviewSignal, name="spectrum", ndim=1, doc="Preview signal for Falcon detector spectrum"
)
# Configuration attributes
collect_mode = Cpt(EpicsSignal, "CollectMode", doc="Collect mode signal")
preset_real_time = Cpt(EpicsSignalWithRBV, "PresetRealTime", doc="Preset real time signal")
# Computed signal for dead time corrected counts
dead_cor_roi0_count = Cpt(Signal, "dead_cor_roi0_count", doc="Dead time corrected ROI 0 count")
def compute_dead_time_corrected_signal(icr: float, ocr: float, roi: float, ert: float):
_dead_time = 1.182e-7
if icr == 0 or ocr == 0:
return 0
# Check that relative change is large enough
test = 1e9
test_icr = icr
n = 0
while test > _dead_time and n < 30:
try:
true_icr = icr * np.exp(test_icr * _dead_time)
test = (true_icr - test_icr) / test_icr
test_icr = true_icr
n += 1
except Exception as e: # pylint: disable=broad-except
logger.info(f"Error in computation of deadtime corrected signal, error: {e}")
return 0
# Return corrected roi counts
cor_roi_cnts = 0
if ocr * ert != 0:
cor_roi_cnts = roi * true_icr / (ocr * ert)
return cor_roi_cnts
class FalconSuperXASSlim(PSIDeviceBase, FalconControlSlim):
"""Slim Falcon implementation at SuperXAS. prefix: 'X10DA-SITORO:'"""
def __init__(self, *, name, prefix="", scan_info=None, device_manager=None, **kwargs):
super().__init__(
name=name, prefix=prefix, scan_info=scan_info, device_manager=device_manager, **kwargs
)
self._pv_timeout = 5 # seconds
def on_init(self) -> None:
"""
Called when the device is initialized.
No signals are connected at this point. If you like to
set default values on signals, please use on_connected instead.
"""
def on_connected(self) -> None:
"""
Called after the device is connected and its signals are connected.
Default values for signals should be set here.
"""
def on_stage(self) -> DeviceStatus | StatusBase | None:
"""
Called while staging the device.
Information about the upcoming scan can be accessed from the scan_info (self.scan_info.msg) object.
"""
if self.acquiring.get() != FalconAcquiringStatus.DONE:
logger.info(f"Falcon state was {self.acquiring.get()} during stage. Calling stop_all")
status = CompareStatus(self.acquiring, FalconAcquiringStatus.DONE)
self.cancel_on_stop(status)
self.stop_all.put(1)
status.wait(timeout=self._pv_timeout)
self.collect_mode.set(0).wait(timeout=self._pv_timeout)
self.preset_real_time.set(0).wait(timeout=self._pv_timeout)
def on_unstage(self) -> DeviceStatus | StatusBase | None:
"""Called while unstaging the device."""
if self.acquiring.get() != FalconAcquiringStatus.DONE:
self.stop_all.put(1)
status = CompareStatus(self.acquiring, FalconAcquiringStatus.DONE)
self.cancel_on_stop(status)
return status
def on_pre_scan(self) -> DeviceStatus | StatusBase | None:
"""Called right before the scan starts on all devices automatically."""
def on_trigger(self) -> DeviceStatus | StatusBase | None:
"""Called when the device is triggered."""
def on_complete(self) -> DeviceStatus | StatusBase | None:
"""Called to inquire if a device has completed a scans."""
def on_kickoff(self) -> DeviceStatus | StatusBase | None:
"""Called to kickoff a device for a fly scan. Has to be called explicitly."""
def on_stop(self) -> None:
"""Called when the device is stopped."""
self.stop_all.put(1)
def update_data(self):
"""
Set the dead time corrected signal based on input count rate, output count rate, ROI count, and elapsed real time.
Parameters:
icr (float): Input count rate.
ocr (float): Output count rate.
roi (float): ROI count.
ert (float): Elapsed real time.
"""
dead_time_corrected_signal = compute_dead_time_corrected_signal(
self.icr.get(), self.ocr.get(), self.roi.get(), self.ert.get()
)
self.dead_cor_roi0_count.put(dead_time_corrected_signal)
self.spectrum.put(self.spectrum_val.get())

31
superxas_bec/devices/trigger.py
View File

@@ -10,7 +10,7 @@ from ophyd import Device, DeviceStatus, EpicsSignal, EpicsSignalRO, Kind, Status
from ophyd_devices import CompareStatus, TransitionStatus
from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase
from superxas_bec.devices.falcon import FalconAcquiringStatus, FalconSuperXAS
from superxas_bec.devices.falcon_slim import FalconAcquiringStatus, FalconSuperXASSlim
logger = bec_logger.logger
@@ -119,43 +119,34 @@ class Trigger(PSIDeviceBase, TriggerControl):
and the sampling is done before data is being read from the device.
"""
if self.scan_info.msg.scan_name == "exafs_scan":
exp_time = self.scan_info.msg.scan_parameters['integ_time'][self._trigger_index]
self._trigger_index +=1
exp_time = self.scan_info.msg.scan_parameters["integ_time"][self._trigger_index]
self._trigger_index += 1
self.set_exposure_time(exp_time).wait()
time_started = time.time()
logger.info(f"Triggering device {self.name} at {time_started}")
falcon = self.device_manager.devices.get("falcon", None)
if falcon is not None and falcon.enabled is True:
falcon: FalconSuperXAS
falcon: FalconSuperXASSlim
status = CompareStatus(falcon.acquiring, FalconAcquiringStatus.ACQUIRING)
self.cancel_on_stop(status)
falcon.erase_start.put(1)
print(f"Acquiring state after erase_start: {FalconAcquiringStatus(falcon.acquiring.get())}")
status.wait(timeout=5)
time.sleep(0.4)
status_smpl = TransitionStatus(
self.smpl_done, [SamplingDone.RUNNING, SamplingDone.DONE]
)
status.wait(timeout=self._pv_timeout)
time.sleep(0.4)
status_smpl = TransitionStatus(self.smpl_done, [SamplingDone.RUNNING, SamplingDone.DONE])
logger.info(f"Triggering sampling for {self.name} at {time.time() - time_started}")
self.smpl.put(1)
self.cancel_on_stop(status_smpl)
status_smpl.wait()
status_smpl.wait(timeout=self._pv_timeout)
logger.info(f"Sampling done for {self.name} at {time.time() - time_started}")
if falcon is not None and falcon.enabled is True:
time.sleep(0.4)# Simulate some processing time
time.sleep(0.4) # Simulate some processing time
status = CompareStatus(falcon.acquiring, FalconAcquiringStatus.DONE)
self.cancel_on_stop(status)
falcon.stop_all.put(1)
status.wait(timeout=5)
status.wait(timeout=self._pv_timeout)
time.sleep(0.4) # Simulate some processing time
print(falcon.mca1.rois.roi0.count.get())
print(falcon.mca1.elapsed_live_time.get())
print(falcon.mca1.elapsed_real_time.get())
print(falcon.max_elapsed_live.get())
print(falcon.max_elapsed_real.get())
# falcon.send_data()
time.sleep(2) # Sleep currently needed until Falcon acquiring/readout discussion is resolved
falcon.update_data()
return status_smpl
def on_complete(self) -> DeviceStatus | StatusBase | None: