bec/csaxs_bec
0
0
Fork 1
Code Issues 13 Pull Requests 4 Actions Packages Projects Releases Wiki Activity

refactor: cleanup and formatting

Browse Source
This commit is contained in:
appel_c
2025-07-17 09:41:25 +02:00
parent 62ecdd37ff
commit 7a3ce97e30
4 changed files with 163 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
csaxs_bec/devices/epics/delay_generator_csaxs/ddg_1.py
View File

@@ -28,16 +28,16 @@ DELAY CHANNELS:
- e = d
- f = e + 1us (short pulse to OR gate for MCS triggering)
"""
from __future__ import annotations
from typing import TYPE_CHECKING
import time
from typing import TYPE_CHECKING
from bec_lib.logger import bec_logger
from ophyd import DeviceStatus, StatusBase
from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase
from ophyd_devices import CompareStatus, TransitionStatus
from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase
from csaxs_bec.devices.epics.delay_generator_csaxs.delay_generator_csaxs import (
CHANNELREFERENCE,
@@ -48,11 +48,11 @@ from csaxs_bec.devices.epics.delay_generator_csaxs.delay_generator_csaxs import
ChannelConfig,
DelayGeneratorCSAXS,
)
from csaxs_bec.devices.epics.mcs_card.mcs_card_csaxs import READYTOREAD, ACQUIRING
from csaxs_bec.devices.epics.mcs_card.mcs_card_csaxs import ACQUIRING, READYTOREAD
if TYPE_CHECKING: # pragma: no cover
from bec_lib.devicemanager import DeviceManagerBase, ScanInfo
from csaxs_bec.devices.epics.mcs_card.mcs_card_csaxs import MCSCardCSAXS
logger = bec_logger.logger
@@ -93,16 +93,22 @@ class DDG1(PSIDeviceBase, DelayGeneratorCSAXS):
It is operated in standard mode, not burst mode and will trigger the EXT/EN of DDG2 (channel ab).
It is responsible for opening the shutter (channel cd) and sending an extra trigger to an or gate for the MCS card (channel ef).
"""
def __init__(self, name: str,
def __init__(
self,
name: str,
prefix: str = "",
scan_info: ScanInfo | None = None,
device_manager: DeviceManagerBase | None = None,
**kwargs):
**kwargs,
):
"""
Initialize the MCSCardCSAXS with the given arguments and keyword arguments.
"""
super().__init__(name=name, prefix=prefix, scan_info=scan_info,device_manager=device_manager, **kwargs)
self.device_manager=device_manager
super().__init__(
name=name, prefix=prefix, scan_info=scan_info, device_manager=device_manager, **kwargs
)
self.device_manager = device_manager
# pylint: disable=attribute-defined-outside-init
def on_connected(self) -> None:
@@ -147,22 +153,26 @@ class DDG1(PSIDeviceBase, DelayGeneratorCSAXS):
If we don't then subsequent triggers may reach the DDG too early, and will be ignored. To
avoid this, we've added the option to specify a delay via add_delay, default here is 50ms.
"""
mcs = self.device_manager.devices.get('mcs', None)
mcs = self.device_manager.devices.get("mcs", None)
if mcs is None:
logger.info(f"Did not find mcs card in current session")
else:
mcs :MCSCardCSAXS
mcs: MCSCardCSAXS
status_ready_read = CompareStatus(mcs.ready_to_read, READYTOREAD.DONE)
mcs.stop_all.put(1)
status_acquiring = TransitionStatus(mcs.acquiring, [ACQUIRING.DONE, ACQUIRING.ACQUIRING])
status_acquiring = TransitionStatus(
mcs.acquiring, [ACQUIRING.DONE, ACQUIRING.ACQUIRING]
)
self.cancel_on_stop(status_ready_read)
self.cancel_on_stop(status_acquiring)
status_ready_read.wait(2)
mcs.ready_to_read.put(READYTOREAD.PROCESSING)
mcs.erase_start.put(1)
status_acquiring.wait(timeout=2)# 2 s wait for mcs card to start should be more than enough..
status_acquiring.wait(
timeout=2
) # 2 s wait for mcs card to start should be more than enough..
st = StatusBase()
self.cancel_on_stop(st)
self.trigger_shot.put(1, use_complete=True)

4
csaxs_bec/devices/epics/delay_generator_csaxs/ddg_2.py
View File

@@ -103,13 +103,13 @@ class DDG2(PSIDeviceBase, DelayGeneratorCSAXS):
burst_pulse_width = exp_time - DEFAULT_READOUT_TIMES["ab"]
self.set_delay_pairs(channel="ab", delay=0, width=burst_pulse_width)
self.burst_enable(count=frames_per_trigger, delay=0, period=exp_time)
def on_pre_scan(self):
"""
The delay generator occasionally needs a bit extra time to process all
commands from stage. Therefore, we introduce here a short sleep
"""
# Delay Generator occasionaly needs a bit extra time to process all commands, sleep 50ms
# Delay Generator occasionaly needs a bit extra time to process all commands, sleep 50ms
time.sleep(0.05)
def on_trigger(self) -> DeviceStatus | StatusBase | None:

11
csaxs_bec/devices/epics/mcs_card/mcs_card.py
View File

@@ -19,7 +19,7 @@ from __future__ import annotations
import enum
from ophyd import Component as Cpt
from ophyd import DynamicDeviceComponent, EpicsSignal, EpicsSignalRO, Kind, Device
from ophyd import Device, DynamicDeviceComponent, EpicsSignal, EpicsSignalRO, Kind
class CHANNELADVANCE(int, enum.Enum):
@@ -147,7 +147,7 @@ def _create_mca_channels(num_channels: int) -> dict[str, tuple]:
for i in range(1, num_channels + 1):
mcs_channels[f"mca{i}"] = (
EpicsSignalRO,
f'mca{i}.VAL',
f"mca{i}.VAL",
{"kind": Kind.omitted, "auto_monitor": True, "doc": f"MCA channel {i}."},
)
return mcs_channels
@@ -213,7 +213,12 @@ class MCSCard(Device):
kind=Kind.omitted,
doc="Forces a read of all mca or waveform records from the hardware. This record can be set to periodically process to update the records during acquisition. Note that even if this record has SCAN=Passive the mca or waveform records will always process once when acquisition completes.",
)
read_mode = Cpt(EpicsSignal, "ReadAll.SCAN", kind=Kind.omitted, doc="Readout mode for transferring data from FIFO buffer to mca EPICS scalars.")
read_mode = Cpt(
EpicsSignal,
"ReadAll.SCAN",
kind=Kind.omitted,
doc="Readout mode for transferring data from FIFO buffer to mca EPICS scalars.",
)
num_use_all = Cpt(
EpicsSignal,
"NuseAll",

167
csaxs_bec/devices/epics/mcs_card/mcs_card_csaxs.py
View File

@@ -1,14 +1,17 @@
"""Module for the MCSCard CSAXS implementation."""
from __future__ import annotations
import enum
from threading import RLock
from typing import TYPE_CHECKING
from ophyd import Component as Cpt, Signal, Kind, EpicsSignalRO
from bec_lib.logger import bec_logger
from ophyd import Component as Cpt
from ophyd import Device, EpicsSignalRO, Kind, Signal
from ophyd_devices import CompareStatus, ProgressSignal, TransitionStatus
from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase
from threading import RLock
import enum
from bec_lib.logger import bec_logger
from csaxs_bec.devices.epics.mcs_card.mcs_card import (
ACQUIREMODE,
ACQUIRING,
@@ -20,48 +23,111 @@ from csaxs_bec.devices.epics.mcs_card.mcs_card import (
READMODE,
MCSCard,
)
from csaxs_bec.devices.epics.xbpms import DiffXYSignal, SumSignal
if TYPE_CHECKING: # pragma: no cover
from bec_lib.devicemanager import DeviceManagerBase, ScanInfo
logger =bec_logger.logger
logger = bec_logger.logger
class READYTOREAD(int, enum.Enum):
PROCESSING = 0
DONE = 1
class BPMDevice(Device):
"""Class for BPM device of the MCSCard."""
current1 = Cpt(Signal, name="current1", kind=Kind.normal, doc="Current 1")
current2 = Cpt(Signal, name="current2", kind=Kind.normal, doc="Current 2")
current3 = Cpt(Signal, name="current3", kind=Kind.normal, doc="Current 3")
current4 = Cpt(Signal, name="current4", kind=Kind.normal, doc="Current 4")
sum = Cpt(SumSignal, kind="hinted", doc="Sum of all currents")
x = Cpt(
DiffXYSignal,
sum1=["current1", "current2"],
sum2=["current3", "current4"],
doc="X difference signal",
)
y = Cpt(
DiffXYSignal,
sum1=["current1", "current3"],
sum2=["current2", "current4"],
doc="Y difference signal",
)
diag = Cpt(
DiffXYSignal,
sum1=["current1", "current4"],
sum2=["current2", "current3"],
doc="Diagonal difference signal",
)
class BPMCountNormalized(BPMDevice):
"""Class for BPM device of the MCSCard with normalized currents."""
current1 = Cpt(Signal, name="current1", kind=Kind.normal, doc="Count time normalized current 1")
current2 = Cpt(Signal, name="current2", kind=Kind.normal, doc="Count time normalized current 2")
current3 = Cpt(Signal, name="current3", kind=Kind.normal, doc="Count time normalized current 3")
current4 = Cpt(Signal, name="current4", kind=Kind.normal, doc="Count time normalized current 4")
class MCSCardCSAXS(PSIDeviceBase, MCSCard):
"""
Implementation of the MCSCard SIS3820 for CSAXS, prefix 'X12SA-MCS:'.
The basic functionality is inherited from the MCSCard class.
"""
ready_to_read = Cpt(Signal, name='ready_to_read', kind=Kind.omitted, doc='Signal that indicates if mcs card is ready to be read from after triggers. 0 not ready, 1 ready')
ready_to_read = Cpt(
Signal,
name="ready_to_read",
kind=Kind.omitted,
doc="Signal that indicates if mcs card is ready to be read from after triggers. 0 not ready, 1 ready",
)
progress: ProgressSignal = Cpt(ProgressSignal, name="progress")
count_time = Cpt(Signal, name="count_time", kind=Kind.normal)
bpm_1 = Cpt(Signal, name='bpm_1', kind=Kind.normal)
bpm_2 = Cpt(Signal, name='bpm_2', kind=Kind.normal)
bpm_3 = Cpt(Signal, name='bpm_3', kind=Kind.normal)
bpm_4 = Cpt(Signal, name='bpm_4', kind=Kind.normal)
bpm = Cpt(
BPMDevice, name="bpm", kind=Kind.normal, doc="BPM device for MCSCard, normalized currents"
)
bpm_norm = Cpt(
BPMCountNormalized,
name="bpmnorm",
kind=Kind.normal,
doc="BPM device for MCSCard, normalized currents",
)
bpm_plot = Cpt(
BPMDevice,
name="bpm_plot",
kind=Kind.normal,
doc="subdevice with synchronized readings for plotting with monitored signals",
)
def __init__(self, name: str,
def __init__(
self,
name: str,
prefix: str = "",
scan_info: ScanInfo | None = None,
device_manager: DeviceManagerBase | None = None,
**kwargs):
**kwargs,
):
"""
Initialize the MCSCardCSAXS with the given arguments and 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._mcs_clock = 10e-6 # 10MHz clock
self._pv_timeout = 2
self._rlock = RLock()
self.counter_mapping = {f"{self.counters.name}_mca1" : 'bpm_1',
f"{self.counters.name}_mca2" : 'bpm_2',
f"{self.counters.name}_mca3" : 'bpm_3',
f"{self.counters.name}_mca4" : 'bpm_4',
f"{self.counters.name}_mca5" : 'count_time'
}
self.counter_mapping = {
f"{self.counters.name}_mca1": "current1",
f"{self.counters.name}_mca2": "current2",
f"{self.counters.name}_mca3": "current3",
f"{self.counters.name}_mca4": "current4",
f"{self.counters.name}_mca5": "count_time",
}
self.counter_updated = []
def on_connected(self):
@@ -72,20 +138,17 @@ class MCSCardCSAXS(PSIDeviceBase, MCSCard):
self.stop_all.put(1)
self.channel_advance.set(CHANNELADVANCE.EXTERNAL).wait(timeout=self._pv_timeout)
self.channel_advance.set(CHANNEL1SOURCE.EXTERNAL).wait(
timeout=self._pv_timeout
) # Check if this is correct, or internal clock
self.channel_advance.set(CHANNEL1SOURCE.EXTERNAL).wait(timeout=self._pv_timeout)
self.prescale.set(1).wait(timeout=self._pv_timeout)
# Set the user LED to off
self.user_led.set(0).wait(timeout=self._pv_timeout)
# Only channel 1-5 are connected so far, adjust if more are needed
self.mux_output.set(5).wait(timeout=self._pv_timeout)
# Set the input and output modes & polarities
self.input_mode.set(INPUTMODE.MODE_3).wait(timeout=self._pv_timeout)
self.input_polarity.set(POLARITY.NORMAL).wait(timeout=self._pv_timeout)
self.output_mode.set(OUTPUTMODE.MODE_2).wait(
timeout=self._pv_timeout
) # To be checked due to cabling of time counter
self.output_polarity.set(POLARITY.INVERTED).wait(
timeout=self._pv_timeout
) # To be checked and tested!
self.output_mode.set(OUTPUTMODE.MODE_2).wait(timeout=self._pv_timeout)
self.output_polarity.set(POLARITY.NORMAL).wait(timeout=self._pv_timeout)
self.count_on_start.set(0).wait(timeout=self._pv_timeout)
# Set appropriate read mode
@@ -94,9 +157,12 @@ class MCSCardCSAXS(PSIDeviceBase, MCSCard):
# Subscribe the progress signal
self.current_channel.subscribe(self._progress_update, run=False)
# Set the acquire mode
self.acquire_mode.set(ACQUIREMODE.MCS).wait(timeout=self._pv_timeout)
# Subscribe to the mca updates
for name in self.counters.component_names:
sig:EpicsSignalRO = getattr(self.counters, name)
sig: EpicsSignalRO = getattr(self.counters, name)
sig.subscribe(self._on_counter_update, run=False)
def _on_counter_update(self, value, **kwargs) -> None:
@@ -109,19 +175,43 @@ class MCSCardCSAXS(PSIDeviceBase, MCSCard):
if mapped_signal_name is None:
logger.info(f"Received update from unmapped signal {signal.name}")
return
sig = getattr(self, mapped_signal_name)
if mapped_signal_name == "count_time":
# Count time is not mapped into bpm signals
self.count_time.put(value)
return
# Update self.bpm
sig = getattr(self.bpm, mapped_signal_name)
sig.put(value)
self.counter_updated.append(signal.name)
received_all_updates = (set(self.counter_updated) == set(self.counter_mapping.keys()))
logger.info(f"Value for received_all_updates {received_all_updates}")
received_all_updates = set(self.counter_updated) == set(self.counter_mapping.keys())
if received_all_updates:
self.ready_to_read.put(1) # Reset happens from DDG class!
# Set the normalized currents
count_time = self.count_time.get()
for name in self.counter_mapping.values():
if name == "count_time":
continue
sig = getattr(self.bpm, name)
sig_norm = getattr(self.bpm_norm, name)
if count_time <= 0:
logger.warning(
f"Count time is zero or negative ({count_time}), setting normalized current to 0."
)
sig_norm.put(0.0)
continue
sig_norm.put(v / t for v, t in zip(sig.get(), count_time))
sig_plot = getattr(self.bpm_plot, name)
sig_plot.put(sum(sig_norm.get()) / len(sig_norm.get()))
self.ready_to_read.put(1) # Reset happens from DDG class!
self.counter_updated.clear()
def _progress_update(self, value, **kwargs) -> None:
"""Callback for progress updates from ophyd subscription on current_channel."""
# This logic needs to be further refined as this is currently reporting the progress
# of a single trigger from BEC within a burst scan.
frames_per_trigger = self.scan_info.msg.scan_parameters.get("frames_per_trigger", 1)
self.progress.put(value=value, max_value=frames_per_trigger, done=bool(value == frames_per_trigger))
self.progress.put(
value=value, max_value=frames_per_trigger, done=bool(value == frames_per_trigger)
)
def on_stage(self) -> None:
"""
@@ -139,15 +229,14 @@ class MCSCardCSAXS(PSIDeviceBase, MCSCard):
self.stop_all.put(1)
self.ready_to_read.put(READYTOREAD.DONE)
self.erase_all.set(0).wait(timeout=self._pv_timeout)
def on_trigger(self) -> None:
status = TransitionStatus(self.ready_to_read, stric=True, transitions=[READYTOREAD.PROCESSING, READYTOREAD.DONE])
status = TransitionStatus(
self.ready_to_read, strict=True, transitions=[READYTOREAD.PROCESSING, READYTOREAD.DONE]
)
self.cancel_on_stop(status)
return status
def on_pre_scan(self) -> None:
"""
Called before the scan starts.