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

Fix/operational improvements #54

Merged
appel_c merged 10 commits from fix/operational_improvements into main 2025-06-18 14:28:24 +02:00
Conversation 14 Commits 10 Files Changed 9 +876 -26
9 changed files with 876 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
debye_bec/device_configs/x01da_machine.yaml Normal file
View File

@@ -0,0 +1,13 @@
curr:
readoutPriority: baseline
description: SLS ring current
deviceClass: ophyd.EpicsSignalRO
deviceConfig:
auto_monitor: true
read_pv: AGEBD-DBPM3CURR:CURRENT-AVG
deviceTags:
- machine
onFailure: buffer
enabled: true
readOnly: true
softwareTrigger: false

227
debye_bec/device_configs/x01da_optic_slits.yaml Normal file
View File

@@ -0,0 +1,227 @@
## Optics Slits 1 -- Physical positioners
sl1_trxr:
readoutPriority: baseline
description: Optics slits 1 X-translation Ring-edge
deviceClass: ophyd.EpicsMotor
deviceConfig:
prefix: X01DA-OP-SL1:TRXR
onFailure: retry
enabled: true
softwareTrigger: false
deviceTags:
- optics
- slits
sl1_trxw:
readoutPriority: baseline
description: Optics slits 1 X-translation Wall-edge
deviceClass: ophyd.EpicsMotor
deviceConfig:
prefix: X01DA-OP-SL1:TRXW
onFailure: retry
enabled: true
softwareTrigger: false
deviceTags:
- optics
- slits
sl1_tryb:
readoutPriority: baseline
description: Optics slits 1 Y-translation Bottom-edge
deviceClass: ophyd.EpicsMotor
deviceConfig:
prefix: X01DA-OP-SL1:TRYB
onFailure: retry
enabled: true
softwareTrigger: false
deviceTags:
- optics
- slits
sl1_tryt:
readoutPriority: baseline
description: Optics slits 1 X-translation Top-edge
deviceClass: ophyd.EpicsMotor
deviceConfig:
prefix: X01DA-OP-SL1:TRYT
onFailure: retry
enabled: true
softwareTrigger: false
deviceTags:
- optics
- slits
bm1_try:
readoutPriority: baseline
description: Beam Monitor 1 Y-translation
deviceClass: ophyd.EpicsMotor
deviceConfig:
prefix: X01DA-OP-BM1:TRY
onFailure: retry
enabled: true
softwareTrigger: false
deviceTags:
- optics
- slits
## Optics Slits 1 -- Virtual positioners
sl1_centerx:
readoutPriority: baseline
description: Optics slits 1 X-center
deviceClass: ophyd.EpicsMotor
deviceConfig:
prefix: X01DA-OP-SL1:CENTERX
onFailure: retry
enabled: true
softwareTrigger: false
deviceTags:
- optics
- slits
sl1_gapx:
readoutPriority: baseline
description: Optics slits 1 X-gap
deviceClass: ophyd.EpicsMotor
deviceConfig:
prefix: X01DA-OP-SL1:GAPX
onFailure: retry
enabled: true
softwareTrigger: false
deviceTags:
- optics
- slits
sl1_centery:
readoutPriority: baseline
description: Optics slits 1 Y-center
deviceClass: ophyd.EpicsMotor
deviceConfig:
prefix: X01DA-OP-SL1:CENTERY
onFailure: retry
enabled: true
softwareTrigger: false
deviceTags:
- optics
- slits
sl1_gapy:
readoutPriority: baseline
description: Optics slits 1 Y-gap
deviceClass: ophyd.EpicsMotor
deviceConfig:
prefix: X01DA-OP-SL1:GAPY
onFailure: retry
enabled: true
softwareTrigger: false
deviceTags:
- optics
- slits
## Optics Slits 2 -- Physical positioners
sl2_trxr:
readoutPriority: baseline
description: Optics slits 2 X-translation Ring-edge
deviceClass: ophyd.EpicsMotor
deviceConfig:
prefix: X01DA-OP-SL2:TRXR
onFailure: retry
enabled: true
softwareTrigger: false
deviceTags:
- optics
- slits
sl2_trxw:
readoutPriority: baseline
description: Optics slits 2 X-translation Wall-edge
deviceClass: ophyd.EpicsMotor
deviceConfig:
prefix: X01DA-OP-SL2:TRXW
onFailure: retry
enabled: true
softwareTrigger: false
deviceTags:
- optics
- slits
sl2_tryb:
readoutPriority: baseline
description: Optics slits 2 Y-translation Bottom-edge
deviceClass: ophyd.EpicsMotor
deviceConfig:
prefix: X01DA-OP-SL2:TRYB
onFailure: retry
enabled: true
softwareTrigger: false
deviceTags:
- optics
- slits
sl2_tryt:
readoutPriority: baseline
description: Optics slits 2 X-translation Top-edge
deviceClass: ophyd.EpicsMotor
deviceConfig:
prefix: X01DA-OP-SL2:TRYT
onFailure: retry
enabled: true
softwareTrigger: false
deviceTags:
- optics
- slits
bm2_try:
readoutPriority: baseline
description: Beam Monitor 2 Y-translation
deviceClass: ophyd.EpicsMotor
deviceConfig:
prefix: X01DA-OP-BM2:TRY
onFailure: retry
enabled: true
softwareTrigger: false
deviceTags:
- optics
- slits
## Optics Slits 2 -- Virtual positioners
sl2_centerx:
readoutPriority: baseline
description: Optics slits 2 X-center
deviceClass: ophyd.EpicsMotor
deviceConfig:
prefix: X01DA-OP-SL2:CENTERX
onFailure: retry
enabled: true
softwareTrigger: false
deviceTags:
- optics
- slits
sl2_gapx:
readoutPriority: baseline
description: Optics slits 2 X-gap
deviceClass: ophyd.EpicsMotor
deviceConfig:
prefix: X01DA-OP-SL2:GAPX
onFailure: retry
enabled: true
softwareTrigger: false
deviceTags:
- optics
- slits
sl2_centery:
readoutPriority: baseline
description: Optics slits 2 Y-center
deviceClass: ophyd.EpicsMotor
deviceConfig:
prefix: X01DA-OP-SL2:CENTERY
onFailure: retry
enabled: true
softwareTrigger: false
deviceTags:
- optics
- slits
sl2_gapy:
readoutPriority: baseline
description: Optics slits 2 Y-gap
deviceClass: ophyd.EpicsMotor
deviceConfig:
prefix: X01DA-OP-SL2:GAPY
onFailure: retry
enabled: true
softwareTrigger: false
deviceTags:
- optics
- slits

177
debye_bec/device_configs/x01da_test_config.yaml
View File

@@ -1,4 +1,7 @@
optic_slit_config:
- !include ./x01da_optic_slits.yaml
machine_config:
- !include ./x01da_machine.yaml
## Slit Diaphragm -- Physical positioners
sldi_trxr:
readoutPriority: baseline
@@ -210,15 +213,6 @@ mo1_bragg:
onFailure: retry
enabled: true
softwareTrigger: false
dummy_pv:
readoutPriority: monitored
description: Heartbeat of Bragg
deviceClass: ophyd.EpicsSignalRO
deviceConfig:
read_pv: "X01DA-OP-MO1:BRAGG:heartbeat_RBV"
onFailure: retry
enabled: true
softwareTrigger: false
mo1_bragg_angle:
readoutPriority: baseline
description: Positioner for the Monochromator
@@ -256,7 +250,7 @@ mo_trx:
description: Monochromator X Translation
deviceClass: ophyd.EpicsMotor
deviceConfig:
prefix: X01DA-OP-MO1:TRY
prefix: X01DA-OP-MO1:TRX
onFailure: retry
enabled: true
softwareTrigger: false
@@ -270,6 +264,120 @@ mo_roty:
enabled: true
softwareTrigger: false
## Focusing Mirror -- Physical Positioners
fm_trxu:
readoutPriority: baseline
description: Focusing Mirror X-translation upstream
deviceClass: ophyd.EpicsMotor
deviceConfig:
prefix: X01DA-OP-FM:TRXU
onFailure: retry
enabled: true
softwareTrigger: false
fm_trxd:
readoutPriority: baseline
description: Focusing Mirror X-translation downstream
deviceClass: ophyd.EpicsMotor
deviceConfig:
prefix: X01DA-OP-FM:TRXD
onFailure: retry
enabled: true
softwareTrigger: false
fm_tryd:
readoutPriority: baseline
description: Focusing Mirror Y-translation downstream
deviceClass: ophyd.EpicsMotor
deviceConfig:
prefix: X01DA-OP-FM:TRYD
onFailure: retry
enabled: true
softwareTrigger: false
fm_tryur:
readoutPriority: baseline
description: Focusing Mirror Y-translation upstream ring
deviceClass: ophyd.EpicsMotor
deviceConfig:
prefix: X01DA-OP-FM:TRYUR
onFailure: retry
enabled: true
softwareTrigger: false
fm_tryuw:
readoutPriority: baseline
description: Focusing Mirror Y-translation upstream wall
deviceClass: ophyd.EpicsMotor
deviceConfig:
prefix: X01DA-OP-FM:TRYUW
onFailure: retry
enabled: true
softwareTrigger: false
fm_bnd:
readoutPriority: baseline
description: Focusing Mirror bender
deviceClass: ophyd.EpicsMotor
deviceConfig:
prefix: X01DA-OP-FM:BND
onFailure: retry
enabled: true
softwareTrigger: false
## Focusing Mirror -- Virtual Positioners
fm_rotx:
readoutPriority: baseline
description: Focusing Morror Pitch
deviceClass: ophyd.EpicsMotor
deviceConfig:
prefix: X01DA-OP-FM:ROTX
onFailure: retry
enabled: true
softwareTrigger: false
fm_roty:
readoutPriority: baseline
description: Focusing Morror Yaw
deviceClass: ophyd.EpicsMotor
deviceConfig:
prefix: X01DA-OP-FM:ROTY
onFailure: retry
enabled: true
softwareTrigger: false
fm_rotz:
readoutPriority: baseline
description: Focusing Morror Roll
deviceClass: ophyd.EpicsMotor
deviceConfig:
prefix: X01DA-OP-FM:ROTZ
onFailure: retry
enabled: true
softwareTrigger: false
fm_xctp:
readoutPriority: baseline
description: Focusing Morror Center Point X
deviceClass: ophyd.EpicsMotor
deviceConfig:
prefix: X01DA-OP-FM:XTCP
onFailure: retry
enabled: true
softwareTrigger: false
fm_ytcp:
readoutPriority: baseline
description: Focusing Morror Center Point Y
deviceClass: ophyd.EpicsMotor
deviceConfig:
prefix: X01DA-OP-FM:YTCP
onFailure: retry
enabled: true
softwareTrigger: false
fm_ztcp:
readoutPriority: baseline
description: Focusing Morror Center Point Z
deviceClass: ophyd.EpicsMotor
deviceConfig:
prefix: X01DA-OP-FM:ZTCP
onFailure: retry
enabled: true
softwareTrigger: false
# Ionization Chambers
ic0:
@@ -480,4 +588,49 @@ es1_alignment_laser:
onFailure: retry
enabled: true
softwareTrigger: false
## Pinhole alignment stages -- Physical Positioners
pin1_trx:
readoutPriority: baseline
description: Pinhole X-translation
deviceClass: ophyd.EpicsMotor
deviceConfig:
prefix: X01DA-ES1-PIN1:TRX
onFailure: retry
enabled: true
softwareTrigger: false
tags: Endstation
pin1_try:
readoutPriority: baseline
description: Pinhole Y-translation
deviceClass: ophyd.EpicsMotor
deviceConfig:
prefix: X01DA-ES1-PIN1:TRY
onFailure: retry
enabled: true
softwareTrigger: false
tags: Endstation
pin1_rotx:
readoutPriority: baseline
description: Pinhole X-rotation
deviceClass: ophyd.EpicsMotor
deviceConfig:
prefix: X01DA-ES1-PIN1:ROTX
onFailure: retry
enabled: true
softwareTrigger: false
tags: Endstation
pin1_roty:
readoutPriority: baseline
description: Pinhole Y-rotation
deviceClass: ophyd.EpicsMotor
deviceConfig:
prefix: X01DA-ES1-PIN1:ROTY
onFailure: retry
enabled: true
softwareTrigger: false
tags: Endstation

2
debye_bec/devices/mo1_bragg/mo1_bragg.py
View File

@@ -204,7 +204,7 @@ class Mo1Bragg(PSIDeviceBase, Mo1BraggPositioner):
self.wait_for_signal(
self.scan_control.scan_msg,
ScanControlLoadMessage.SUCCESS,
timeout=self.timeout_for_pvwait,
timeout=2 * self.timeout_for_pvwait,
)
return None

104
debye_bec/devices/nidaq/nidaq.py
View File

@@ -1,10 +1,12 @@
from __future__ import annotations
import time
from typing import TYPE_CHECKING, Literal, cast
from bec_lib.logger import bec_logger
from ophyd import Component as Cpt
from ophyd import Device, DeviceStatus, EpicsSignal, EpicsSignalRO, Kind, StatusBase
from ophyd.status import SubscriptionStatus, WaitTimeoutError
from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase
from ophyd_devices.sim.sim_signals import SetableSignal
@@ -189,6 +191,14 @@ class NidaqControl(Device):
auto_monitor=True,
)
energy_epics = Cpt(
EpicsSignalRO,
suffix="NIDAQ-ENERGY",
kind=Kind.normal,
doc="EPICS Energy reading",
auto_monitor=True,
)
### Readback for BEC emitter ###
ai0_mean = Cpt(
@@ -298,6 +308,7 @@ class NidaqControl(Device):
di4_max = Cpt(SetableSignal, value=0, kind=Kind.normal, doc="NIDAQ stream digital input 4, MAX")
enc = Cpt(SetableSignal, value=0, kind=Kind.normal)
energy = Cpt(SetableSignal, value=0, kind=Kind.normal)
### Control PVs ###
@@ -313,6 +324,9 @@ class NidaqControl(Device):
readout_range = Cpt(EpicsSignal, suffix="NIDAQ-ReadoutRange", kind=Kind.config)
encoder_type = Cpt(EpicsSignal, suffix="NIDAQ-EncoderType", kind=Kind.config)
stop_call = Cpt(EpicsSignal, suffix="NIDAQ-Stop", kind=Kind.config)
power = Cpt(EpicsSignal, suffix="NIDAQ-Power", kind=Kind.config)
heartbeat = Cpt(EpicsSignal, suffix="NIDAQ-Heartbeat", kind=Kind.config, auto_monitor=True)
time_left = Cpt(EpicsSignalRO, suffix="NIDAQ-TimeLeft", kind=Kind.config, auto_monitor=True)
ai_chans = Cpt(EpicsSignal, suffix="NIDAQ-AIChans", kind=Kind.config)
ci_chans = Cpt(EpicsSignal, suffix="NIDAQ-CIChans6614", kind=Kind.config)
@@ -332,6 +346,7 @@ class Nidaq(PSIDeviceBase, NidaqControl):
def __init__(self, prefix: str = "", *, name: str, scan_info: ScanInfo = None, **kwargs):
super().__init__(name=name, prefix=prefix, scan_info=scan_info, **kwargs)
self.scan_info: ScanInfo
self.timeout_wait_for_signal = 5 # put 5s firsts
self._timeout_wait_for_pv = 3 # 3s timeout for pv calls
self.valid_scan_names = [
@@ -339,6 +354,7 @@ class Nidaq(PSIDeviceBase, NidaqControl):
"xas_simple_scan_with_xrd",
"xas_advanced_scan",
"xas_advanced_scan_with_xrd",
"nidaq_continuous_scan",
]
########################################
@@ -472,6 +488,18 @@ class Nidaq(PSIDeviceBase, NidaqControl):
Called after the device is connected and its signals are connected.
Default values for signals should be set here.
"""
def heartbeat_callback(*, old_value, value, **kwargs):
return ((old_value) == 0 and (value == 1)) or ((old_value) == 1 and (value == 0))
status = SubscriptionStatus(self.heartbeat, callback=heartbeat_callback)
try:
status.wait(timeout=self.timeout_wait_for_signal) # Raises if timeout is reached
except WaitTimeoutError:
self.power.put(1)
status.wait(timeout=self.timeout_wait_for_signal)
if not self.wait_for_condition(
condition=lambda: self.state.get() == NidaqState.STANDBY,
timeout=self.timeout_wait_for_signal,
@@ -481,6 +509,7 @@ class Nidaq(PSIDeviceBase, NidaqControl):
f"Device {self.name} has not been reached in state STANDBY, current state {NidaqState(self.state.get())}"
)
self.scan_duration.set(0).wait(timeout=self._timeout_wait_for_pv)
self.time_left.subscribe(self._progress_update, run=False)
def on_stage(self) -> DeviceStatus | StatusBase | None:
"""
@@ -500,8 +529,20 @@ class Nidaq(PSIDeviceBase, NidaqControl):
raise NidaqError(
f"Device {self.name} has not been reached in state STANDBY, current state {NidaqState(self.state.get())}"
)
self.scan_type.set(ScanType.TRIGGERED).wait(timeout=self._timeout_wait_for_pv)
self.scan_duration.set(0).wait(timeout=self._timeout_wait_for_pv)
# If scan is not part of the valid_scan_names,
if self.scan_info.msg.scan_name != "nidaq_continuous_scan":
self.scan_type.set(ScanType.TRIGGERED).wait(timeout=self._timeout_wait_for_pv)
self.scan_duration.set(0).wait(timeout=self._timeout_wait_for_pv)
self.enable_compression.set(1).wait(timeout=self._timeout_wait_for_pv)
else:
self.scan_type.set(ScanType.CONTINUOUS).wait(timeout=self._timeout_wait_for_pv)
self.scan_duration.set(self.scan_info.msg.scan_parameters["scan_duration"]).wait(
timeout=self._timeout_wait_for_pv
)
self.enable_compression.set(self.scan_info.msg.scan_parameters["compression"]).wait(
timeout=self._timeout_wait_for_pv
)
self.stage_call.set(1).wait(timeout=self._timeout_wait_for_pv)
if not self.wait_for_condition(
@@ -512,21 +553,30 @@ class Nidaq(PSIDeviceBase, NidaqControl):
raise NidaqError(
f"Device {self.name} has not been reached in state STAGE, current state {NidaqState(self.state.get())}"
)
self.kickoff_call.set(1).wait(timeout=self._timeout_wait_for_pv)
if self.scan_info.msg.scan_name != "nidaq_continuous_scan":
status = self.on_kickoff()
status.wait(timeout=self._timeout_wait_for_pv)
logger.info(f"Device {self.name} was staged: {NidaqState(self.state.get())}")
def on_kickoff(self) -> DeviceStatus | StatusBase:
"""Kickoff the Nidaq"""
status = self.kickoff_call.set(1)
return status
def on_unstage(self) -> DeviceStatus | StatusBase | None:
"""Called while unstaging the device. Check that the Nidaq goes into Standby"""
def _get_state():
return self.state.get() == NidaqState.STANDBY
# TODO We need to wait longer if rle is disabled
if not self.wait_for_condition(
condition=_get_state, timeout=self.timeout_wait_for_signal, check_stopped=False
):
raise NidaqError(
f"Device {self.name} has not been reached in state STANDBY, current state {NidaqState(self.state.get())}"
)
self.enable_compression.set(1).wait(self._timeout_wait_for_pv)
logger.info(f"Device {self.name} was unstaged: {NidaqState(self.state.get())}")
def on_pre_scan(self) -> DeviceStatus | StatusBase | None:
@@ -540,6 +590,10 @@ class Nidaq(PSIDeviceBase, NidaqControl):
if not self._check_if_scan_name_is_valid():
return None
if self.scan_info.msg.scan_name == "nidaq_continuous_scan":
logger.info(f"Device {self.name} ready to be kicked off for nidaq_continuous_scan")
return None
def _wait_for_state():
return self.state.get() == NidaqState.KICKOFF
@@ -565,18 +619,42 @@ class Nidaq(PSIDeviceBase, NidaqControl):
"""
if not self._check_if_scan_name_is_valid():
return None
self.on_stop()
# TODO check if this wait can be removed. We are waiting in unstage anyways which will always be called afterwards
# Wait for device to be stopped
status = self.wait_for_condition(
condition=lambda: self.state.get() == NidaqState.STANDBY,
check_stopped=True,
timeout=self.timeout_wait_for_signal,
)
def _check_state(self) -> bool:
while True:
if self.stopped is True:
raise NidaqError(f"Device {self.name} was stopped")
if self.state.get() == NidaqState.STANDBY:
return
# if time.time() > timeout_time:
# raise TimeoutError(f"Device {self.name} ran into timeout")
time.sleep(0.1)
if self.scan_info.msg.scan_name != "nidaq_continuous_scan":
self.on_stop()
status = self.task_handler.submit_task(task=_check_state, task_args=(self,))
else:
status = self.task_handler.submit_task(task=_check_state, task_args=(self,))
return status
def on_kickoff(self) -> DeviceStatus | StatusBase | None:
"""Called to kickoff a device for a fly scan. Has to be called explicitly."""
def _progress_update(self, value, **kwargs) -> None:
"""Callback method to update the scan progress, runs a callback
to SUB_PROGRESS subscribers, i.e. BEC.
Args:
value (int) : current progress value
"""
scan_duration = self.scan_info.msg.scan_parameters.get("scan_duration", None)
if not isinstance(scan_duration, (int, float)):
return
value = scan_duration - value
max_value = scan_duration
self._run_subs(
sub_type=self.SUB_PROGRESS,
value=value,
max_value=max_value,
done=bool(value == max_value),
)
def on_stop(self) -> None:
"""Called when the device is stopped."""

1
debye_bec/scans/__init__.py
View File

@@ -4,3 +4,4 @@ from .mono_bragg_scans import (
XASSimpleScan,
XASSimpleScanWithXRD,
)
from .nidaq_cont_scan import NIDAQContinuousScan

84
debye_bec/scans/nidaq_cont_scan.py Normal file
View File

@@ -0,0 +1,84 @@
"""This module contains the scan class for the nidaq of the Debye beamline for use in continuous mode."""
import time
from typing import Literal
import numpy as np
from bec_lib.device import DeviceBase
from bec_lib.logger import bec_logger
from bec_server.scan_server.scans import AsyncFlyScanBase
logger = bec_logger.logger
class NIDAQContinuousScan(AsyncFlyScanBase):
"""Class for the nidaq continuous scan (without mono)"""
scan_name = "nidaq_continuous_scan"
scan_type = "fly"
scan_report_hint = "device_progress"
required_kwargs = []
use_scan_progress_report = False
pre_move = False
gui_config = {"Scan Parameters": ["scan_duration"], "Data Compression": ["compression"]}
def __init__(
self, scan_duration: float, daq: DeviceBase = "nidaq", compression: bool = False, **kwargs
):
"""The NIDAQ continuous scan is used to measure with the NIDAQ without moving the
monochromator or any other motor. The NIDAQ thus runs in continuous mode, with a
set scan_duration.
Args:
scan_duration (float): Duration of the scan.
daq (DeviceBase, optional): DAQ device to be used for the scan.
Defaults to "nidaq".
Examples:
>>> scans.nidaq_continuous_scan(scan_duration=10)
"""
super().__init__(**kwargs)
self.scan_duration = scan_duration
self.daq = daq
self.start_time = 0
self.primary_readout_cycle = 1
self.scan_parameters["scan_duration"] = scan_duration
self.scan_parameters["compression"] = compression
def update_readout_priority(self):
"""Ensure that NIDAQ is not monitored for any quick EXAFS."""
super().update_readout_priority()
self.readout_priority["async"].append("nidaq")
def prepare_positions(self):
"""Prepare the positions for the scan."""
yield None
def pre_scan(self):
"""Pre Scan action."""
self.start_time = time.time()
# Ensure parent class pre_scan actions to be called.
yield from super().pre_scan()
def scan_report_instructions(self):
"""
Return the instructions for the scan report.
"""
yield from self.stubs.scan_report_instruction({"device_progress": [self.daq]})
def scan_core(self):
"""Run the scan core.
Kickoff the acquisition of the NIDAQ wait for the completion of the scan.
"""
kickoff_status = yield from self.stubs.kickoff(device=self.daq)
kickoff_status.wait(timeout=5) # wait for proper kickoff of device
complete_status = yield from self.stubs.complete(device=self.daq, wait=False)
while not complete_status.done:
# Readout monitored devices
yield from self.stubs.read(group="monitored", point_id=self.point_id)
time.sleep(self.primary_readout_cycle)
self.point_id += 1
self.num_pos = self.point_id

167
tests/tests_devices/test_nidaq.py Normal file
View File

@@ -0,0 +1,167 @@
# pylint: skip-file
import threading
from typing import Generator
from unittest import mock
import ophyd
import pytest
from bec_server.scan_server.scan_worker import ScanWorker
from ophyd.status import WaitTimeoutError
from ophyd_devices.interfaces.base_classes.psi_device_base import DeviceStoppedError
from ophyd_devices.tests.utils import MockPV
# from bec_server.device_server.tests.utils import DMMock
from debye_bec.devices.nidaq.nidaq import Nidaq, NidaqError
# TODO move this function to ophyd_devices, it is duplicated in csaxs_bec and needed for other pluging repositories
from debye_bec.devices.test_utils.utils import patch_dual_pvs
@pytest.fixture(scope="function")
def scan_worker_mock(scan_server_mock):
"""Scan worker fixture, utility to generate scan_info for a given scan name."""
scan_server_mock.device_manager.connector = mock.MagicMock()
scan_worker = ScanWorker(parent=scan_server_mock)
yield scan_worker
@pytest.fixture(scope="function")
def mock_nidaq() -> Generator[Nidaq, None, None]:
"""Fixture for the Nidaq device."""
name = "nidaq"
prefix = "nidaq:prefix_test:"
with mock.patch.object(ophyd, "cl") as mock_cl:
mock_cl.get_pv = MockPV
mock_cl.thread_class = threading.Thread
dev = Nidaq(name=name, prefix=prefix)
patch_dual_pvs(dev)
yield dev
def test_init(mock_nidaq):
"""Test the initialization of the Nidaq device."""
dev = mock_nidaq
assert dev.name == "nidaq"
assert dev.prefix == "nidaq:prefix_test:"
assert dev.valid_scan_names == [
"xas_simple_scan",
"xas_simple_scan_with_xrd",
"xas_advanced_scan",
"xas_advanced_scan_with_xrd",
"nidaq_continuous_scan",
]
def test_check_if_scan_name_is_valid(mock_nidaq):
"""Test the check_if_scan_name_is_valid method."""
dev = mock_nidaq
dev.scan_info.msg.scan_name = "xas_simple_scan"
assert dev._check_if_scan_name_is_valid()
dev.scan_info.msg.scan_name = "invalid_scan_name"
assert not dev._check_if_scan_name_is_valid()
def test_set_config(mock_nidaq):
dev = mock_nidaq
# TODO #21 Add test logic for set_config, issue created #
def test_on_connected(mock_nidaq):
"""Test the on_connected method of the Nidaq device."""
dev = mock_nidaq
dev.power.put(0)
dev.heartbeat._read_pv.mock_data = 0
# First scenario, raise timeout error
# This will raise a WaitTimeoutError error as we currently do not support callbacks in the MockPV
dev.timeout_wait_for_signal = 0.1
# To check that it raised, we check that dev.power PV is set to 1
# Set state PV to 0, 1 is expected value
dev.state._read_pv.mock_data = 0
with pytest.raises(WaitTimeoutError):
dev.on_connected()
assert dev.power.get() == 1
# TODO, once the MOCKPv supports callbacks, we can test the rest of the logic issue #22
# def test_on_stage(mock_nidaq):
# dev = mock_nidaq
# #TODO Add once MockPV supports callbacks #22
def test_on_kickoff(mock_nidaq):
"""Test the on_kickoff method of the Nidaq device."""
dev = mock_nidaq
dev.kickoff_call.put(0)
dev.kickoff()
assert dev.kickoff_call.get() == 1
def test_on_unstage(mock_nidaq):
"""Test the on_unstage method of the Nidaq device."""
dev = mock_nidaq
dev.state._read_pv.mock_data = 0 # Set state to 0, 1 is Standby
dev._timeout_wait_for_pv = 0.1 # Set a short timeout for testing
dev.enable_compression._read_pv.mock_data = 0 # Compression enabled
with pytest.raises(NidaqError):
dev.on_unstage()
dev.state._read_pv.mock_data = 1
dev.on_unstage()
assert dev.enable_compression.get() == 1
@pytest.mark.parametrize(
["scan_name", "raise_error", "nidaq_state"],
[
("line_scan", False, 0),
("xas_simple_scan", False, 3),
("xas_simple_scan", True, 0),
("nidaq_continuous_scan", False, 0),
],
)
def test_on_pre_scan(mock_nidaq, scan_name, raise_error, nidaq_state):
"""Test the on_pre_scan method of the Nidaq device."""
dev = mock_nidaq
dev.state.put(nidaq_state)
dev.scan_info.msg.scan_name = scan_name
dev._timeout_wait_for_pv = 0.1 # Set a short timeout for testing
if not raise_error:
dev.pre_scan()
else:
with pytest.raises(NidaqError):
dev.pre_scan()
def test_on_complete(mock_nidaq):
"""Test the on_complete method of the Nidaq device."""
dev = mock_nidaq
# Check for nidaq_continuous_scan
dev.scan_info.msg.scan_name = "nidaq_continuous_scan"
dev.state.put(0) # Set state to DISABLED
status = dev.complete()
assert status.done is False
dev.state.put(1)
# Should resolve now
status.wait(timeout=5) # Wait for the status to complete
assert status.done is True
# Check for XAS simple scan
dev.scan_info.msg.scan_name = "xas_simple_scan"
dev.state.put(0) # Set state to ACQUIRE
dev.stop_call.put(0)
dev._timeout_wait_for_pv = 5
status = dev.on_complete()
assert status.done is False
assert dev.stop_call.get() == 1 # Should have called stop
dev.state.put(1) # Set state to STANDBY
# Should resolve now
status.wait(timeout=5) # Wait for the status to complete
assert status.done is True
# Test that it resolves if device is stopped
dev.state.put(0) # Set state to DISABLED
dev.stopped = True # Reset stopped state
status = dev.on_complete()
with pytest.raises(NidaqError):
status.wait(timeout=5)
assert status.done is True

127
tests/tests_scans/test_nidaq_continous_scan.py Normal file
View File

@@ -0,0 +1,127 @@
# pylint: skip-file
from unittest import mock
from bec_lib.messages import DeviceInstructionMessage
from bec_server.device_server.tests.utils import DMMock
from debye_bec.scans import NIDAQContinuousScan
def get_instructions(request, ScanStubStatusMock):
request.metadata["RID"] = "my_test_request_id"
def fake_done():
"""
Fake done function for ScanStubStatusMock. Upon each call, it returns the next value from the generator.
This is used to simulate the completion of the scan.
"""
yield False
yield False
yield True
def fake_complete(*args, **kwargs):
yield "fake_complete"
return ScanStubStatusMock(done_func=fake_done)
with (
mock.patch.object(request.stubs, "complete", side_effect=fake_complete),
mock.patch.object(request.stubs, "_get_result_from_status", return_value=None),
):
reference_commands = list(request.run())
for cmd in reference_commands:
if not cmd or isinstance(cmd, str):
continue
if "RID" in cmd.metadata:
cmd.metadata["RID"] = "my_test_request_id"
if "rpc_id" in cmd.parameter:
cmd.parameter["rpc_id"] = "my_test_rpc_id"
cmd.metadata.pop("device_instr_id", None)
return reference_commands
def test_xas_simple_scan(scan_assembler, ScanStubStatusMock):
request = scan_assembler(NIDAQContinuousScan, scan_duration=10)
request.device_manager.add_device("nidaq")
reference_commands = get_instructions(request, ScanStubStatusMock)
assert reference_commands == [
None,
None,
DeviceInstructionMessage(
metadata={"readout_priority": "monitored", "RID": "my_test_request_id"},
device=None,
action="scan_report_instruction",
parameter={"device_progress": ["nidaq"]},
),
DeviceInstructionMessage(
metadata={"readout_priority": "monitored", "RID": "my_test_request_id"},
device=None,
action="open_scan",
parameter={
"scan_motors": [],
"readout_priority": {
"monitored": [],
"baseline": [],
"on_request": [],
"async": ["nidaq"],
},
"num_points": 0,
"positions": [],
"scan_name": "nidaq_continuous_scan",
"scan_type": "fly",
},
),
DeviceInstructionMessage(metadata={}, device="nidaq", action="stage", parameter={}),
DeviceInstructionMessage(
metadata={},
device=["bpm4i", "eiger", "mo1_bragg", "samx"],
action="stage",
parameter={},
),
DeviceInstructionMessage(
metadata={"readout_priority": "baseline", "RID": "my_test_request_id"},
device=["samx"],
action="read",
parameter={},
),
DeviceInstructionMessage(
metadata={"readout_priority": "monitored", "RID": "my_test_request_id"},
device=["bpm4i", "eiger", "mo1_bragg", "nidaq", "samx"],
action="pre_scan",
parameter={},
),
DeviceInstructionMessage(
metadata={"readout_priority": "monitored", "RID": "my_test_request_id"},
device="nidaq",
action="kickoff",
parameter={"configure": {}},
),
"fake_complete",
DeviceInstructionMessage(
metadata={"readout_priority": "monitored", "RID": "my_test_request_id", "point_id": 0},
device=["bpm4i", "eiger", "mo1_bragg"],
action="read",
parameter={"group": "monitored"},
),
DeviceInstructionMessage(
metadata={"readout_priority": "monitored", "RID": "my_test_request_id", "point_id": 1},
device=["bpm4i", "eiger", "mo1_bragg"],
action="read",
parameter={"group": "monitored"},
),
"fake_complete",
DeviceInstructionMessage(
metadata={},
device=["bpm4i", "eiger", "mo1_bragg", "nidaq", "samx"],
action="unstage",
parameter={},
),
DeviceInstructionMessage(
metadata={"readout_priority": "monitored", "RID": "my_test_request_id"},
device=None,
action="close_scan",
parameter={},
),
]