bec/ophyd_devices
1
0
Fork 0
Code Issues 9 Pull Requests 10 Actions Packages Projects Releases 126 Wiki Activity

refactor: clean up code

Browse Source
This commit is contained in:
appel_c 2023-11-17 14:13:35 +01:00
parent 3478435e02
commit 4c86f8cfb2
7 changed files with 73 additions and 82 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
ophyd_devices/epics/devices/eiger9m_csaxs.py
View File

@ -1,4 +1,5 @@
import enum import enum
import threading
import time import time
import numpy as np import numpy as np
import os import os
@ -24,14 +25,10 @@ logger = bec_logger.logger
class EigerError(Exception): class EigerError(Exception):
"""Base class for exceptions in this module.""" """Base class for exceptions in this module."""
pass
class EigerTimeoutError(EigerError): class EigerTimeoutError(EigerError):
"""Raised when the Eiger does not respond in time.""" """Raised when the Eiger does not respond in time."""
pass
class Eiger9MSetup(CustomDetectorMixin): class Eiger9MSetup(CustomDetectorMixin):
"""Eiger setup class """Eiger setup class
@ -40,13 +37,13 @@ class Eiger9MSetup(CustomDetectorMixin):
""" """
def __init__(self, parent: Device = None, *args, **kwargs) -> None: def __init__(self, *args, parent: Device = None, **kwargs) -> None:
super().__init__(parent=parent, *args, **kwargs) super().__init__(*args, parent=parent, **kwargs)
self.std_rest_server_url = ( self.std_rest_server_url = (
kwargs["file_writer_url"] if "file_writer_url" in kwargs else "http://xbl-daq-29:5000" kwargs["file_writer_url"] if "file_writer_url" in kwargs else "http://xbl-daq-29:5000"
) )
self.std_client = None self.std_client = None
self._lock = self.parent._lock self._lock = threading.RLock()
def initialize_default_parameter(self) -> None: def initialize_default_parameter(self) -> None:
"""Set default parameters for Eiger9M detector""" """Set default parameters for Eiger9M detector"""
@ -57,9 +54,9 @@ class Eiger9MSetup(CustomDetectorMixin):
readout_time = ( readout_time = (
self.parent.scaninfo.readout_time self.parent.scaninfo.readout_time
if hasattr(self.parent.scaninfo, "readout_time") if hasattr(self.parent.scaninfo, "readout_time")
else self.parent.get_min_readout() else self.parent.MIN_READOUT
) )
self.parent.readout_time = max(readout_time, self.parent.get_min_readout()) self.parent.readout_time = max(readout_time, self.parent.MIN_READOUT)
def initialize_detector(self) -> None: def initialize_detector(self) -> None:
"""Initialize detector""" """Initialize detector"""
@ -181,7 +178,7 @@ class Eiger9MSetup(CustomDetectorMixin):
# Check if energies are eV or keV, assume keV as the default # Check if energies are eV or keV, assume keV as the default
unit = getattr(self.parent.cam.threshold_energy, "units", None) unit = getattr(self.parent.cam.threshold_energy, "units", None)
if unit != None and unit == "eV": if unit is not None and unit == "eV":
factor = 1000 factor = 1000
# set energy on detector # set energy on detector
@ -246,14 +243,14 @@ class Eiger9MSetup(CustomDetectorMixin):
def filepath_exists(self, filepath: str) -> None: def filepath_exists(self, filepath: str) -> None:
"""Check if filepath exists""" """Check if filepath exists"""
signal_conditions = [(lambda: os.path.exists(os.path.dirname(self.parent.filepath)), True)] signal_conditions = [(lambda: os.path.exists(os.path.dirname(filepath)), True)]
if not self.wait_for_signals( if not self.wait_for_signals(
signal_conditions=signal_conditions, signal_conditions=signal_conditions,
timeout=self.parent.timeout, timeout=self.parent.timeout,
check_stopped=False, check_stopped=False,
all_signals=True, all_signals=True,
): ):
raise EigerError(f"Timeout of 3s reached for filepath {self.parent.filepath}") raise EigerError(f"Timeout of 3s reached for filepath {filepath}")
def arm_acquisition(self) -> None: def arm_acquisition(self) -> None:
"""Arm Eiger detector for acquisition""" """Arm Eiger detector for acquisition"""
@ -282,7 +279,7 @@ class Eiger9MSetup(CustomDetectorMixin):
old_scanID = self.parent.scaninfo.scanID old_scanID = self.parent.scaninfo.scanID
self.parent.scaninfo.load_scan_metadata() self.parent.scaninfo.load_scan_metadata()
if self.parent.scaninfo.scanID != old_scanID: if self.parent.scaninfo.scanID != old_scanID:
self.parent._stopped = True self.parent.stopped is True
def publish_file_location(self, done: bool = False, successful: bool = None) -> None: def publish_file_location(self, done: bool = False, successful: bool = None) -> None:
""" """
@ -296,19 +293,19 @@ class Eiger9MSetup(CustomDetectorMixin):
done (bool): True if scan is finished done (bool): True if scan is finished
successful (bool): True if scan was successful successful (bool): True if scan was successful
""" """
pipe = self.parent._producer.pipeline() pipe = self.parent.producer.pipeline()
if successful is None: if successful is None:
msg = messages.FileMessage(file_path=self.parent.filepath, done=done) msg = messages.FileMessage(file_path=self.parent.filepath, done=done)
else: else:
msg = messages.FileMessage( msg = messages.FileMessage(
file_path=self.parent.filepath, done=done, successful=successful file_path=self.parent.filepath, done=done, successful=successful
) )
self.parent._producer.set_and_publish( self.parent.producer.set_and_publish(
MessageEndpoints.public_file(self.parent.scaninfo.scanID, self.parent.name), MessageEndpoints.public_file(self.parent.scaninfo.scanID, self.parent.name),
msg.dumps(), msg.dumps(),
pipe=pipe, pipe=pipe,
) )
self.parent._producer.set_and_publish( self.parent.producer.set_and_publish(
MessageEndpoints.file_event(self.parent.name), msg.dumps(), pipe=pipe MessageEndpoints.file_event(self.parent.name), msg.dumps(), pipe=pipe
) )
pipe.execute() pipe.execute()
@ -406,7 +403,7 @@ class Eiger9McSAXS(PSIDetectorBase):
# specify Setup class # specify Setup class
custom_prepare_cls = Eiger9MSetup custom_prepare_cls = Eiger9MSetup
# specify minimum readout time for detector # specify minimum readout time for detector
PSIDetectorBase.set_min_readout(3e-3) MIN_READOUT = 3e-3
# specify class attributes # specify class attributes
cam = ADCpt(SLSDetectorCam, "cam1:") cam = ADCpt(SLSDetectorCam, "cam1:")

20
ophyd_devices/epics/devices/falcon_csaxs.py
View File

@ -21,14 +21,10 @@ logger = bec_logger.logger
class FalconError(Exception): class FalconError(Exception):
"""Base class for exceptions in this module.""" """Base class for exceptions in this module."""
pass
class FalconTimeoutError(FalconError): class FalconTimeoutError(FalconError):
"""Raised when the Falcon does not respond in time.""" """Raised when the Falcon does not respond in time."""
pass
class DetectorState(enum.IntEnum): class DetectorState(enum.IntEnum):
"""Detector states for Falcon detector""" """Detector states for Falcon detector"""
@ -114,8 +110,8 @@ class FalconSetup(CustomDetectorMixin):
""" """
def __init__(self, parent: Device = None, *args, **kwargs) -> None: def __init__(self, *args, parent: Device = None, **kwargs) -> None:
super().__init__(parent=parent, *args, **kwargs) super().__init__(*args, parent=parent, **kwargs)
def initialize_default_parameter(self) -> None: def initialize_default_parameter(self) -> None:
""" """
@ -133,9 +129,9 @@ class FalconSetup(CustomDetectorMixin):
readout_time = ( readout_time = (
self.parent.scaninfo.readout_time self.parent.scaninfo.readout_time
if hasattr(self.parent.scaninfo, "readout_time") if hasattr(self.parent.scaninfo, "readout_time")
else self.parent.get_min_readout() else self.parent.MIN_READOUT
) )
self.parent.readout_time = max(readout_time, self.parent.get_min_readout()) self.parent.readout_time = max(readout_time, self.parent.MIN_READOUT)
def initialize_detector(self) -> None: def initialize_detector(self) -> None:
""" """
@ -264,19 +260,19 @@ class FalconSetup(CustomDetectorMixin):
done (bool): True if scan is finished done (bool): True if scan is finished
successful (bool): True if scan was successful successful (bool): True if scan was successful
""" """
pipe = self.parent._producer.pipeline() pipe = self.parent.producer.pipeline()
if successful is None: if successful is None:
msg = messages.FileMessage(file_path=self.parent.filepath, done=done) msg = messages.FileMessage(file_path=self.parent.filepath, done=done)
else: else:
msg = messages.FileMessage( msg = messages.FileMessage(
file_path=self.parent.filepath, done=done, successful=successful file_path=self.parent.filepath, done=done, successful=successful
) )
self.parent._producer.set_and_publish( self.parent.producer.set_and_publish(
MessageEndpoints.public_file(self.parent.scaninfo.scanID, self.parent.name), MessageEndpoints.public_file(self.parent.scaninfo.scanID, self.parent.name),
msg.dumps(), msg.dumps(),
pipe=pipe, pipe=pipe,
) )
self.parent._producer.set_and_publish( self.parent.producer.set_and_publish(
MessageEndpoints.file_event(self.parent.name), msg.dumps(), pipe=pipe MessageEndpoints.file_event(self.parent.name), msg.dumps(), pipe=pipe
) )
pipe.execute() pipe.execute()
@ -339,7 +335,7 @@ class FalconcSAXS(PSIDetectorBase):
# specify Setup class # specify Setup class
custom_prepare_cls = FalconSetup custom_prepare_cls = FalconSetup
# specify minimum readout time for detector # specify minimum readout time for detector
PSIDetectorBase.set_min_readout(3e-3) MIN_READOUT = 3e-3
# specify class attributes # specify class attributes
dxp = Cpt(EpicsDXPFalcon, "dxp1:") dxp = Cpt(EpicsDXPFalcon, "dxp1:")

12
ophyd_devices/epics/devices/pilatus_csaxs.py
View File

@ -86,9 +86,9 @@ class PilatusSetup(CustomDetectorMixin):
readout_time = ( readout_time = (
self.parent.scaninfo.readout_time self.parent.scaninfo.readout_time
if hasattr(self.parent.scaninfo, "readout_time") if hasattr(self.parent.scaninfo, "readout_time")
else self.parent.get_min_readout() else self.parent.MIN_READOUT
) )
self.parent.readout_time = max(readout_time, self.parent.get_min_readout()) self.parent.readout_time = max(readout_time, self.parent.MIN_READOUT)
def initialize_detector(self) -> None: def initialize_detector(self) -> None:
"""Initialize detector""" """Initialize detector"""
@ -336,7 +336,7 @@ class PilatusSetup(CustomDetectorMixin):
done (bool): True if scan is finished done (bool): True if scan is finished
successful (bool): True if scan was successful successful (bool): True if scan was successful
""" """
pipe = self.parent._producer.pipeline() pipe = self.parent.producer.pipeline()
if successful is None: if successful is None:
msg = messages.FileMessage( msg = messages.FileMessage(
file_path=self.parent.filepath, file_path=self.parent.filepath,
@ -350,12 +350,12 @@ class PilatusSetup(CustomDetectorMixin):
successful=successful, successful=successful,
metadata={"input_path": self.parent.filepath_raw}, metadata={"input_path": self.parent.filepath_raw},
) )
self.parent._producer.set_and_publish( self.parent.producer.set_and_publish(
MessageEndpoints.public_file(self.parent.scaninfo.scanID, self.parent.name), MessageEndpoints.public_file(self.parent.scaninfo.scanID, self.parent.name),
msg.dumps(), msg.dumps(),
pipe=pipe, pipe=pipe,
) )
self.parent._producer.set_and_publish( self.parent.producer.set_and_publish(
MessageEndpoints.file_event(self.parent.name), msg.dumps(), pipe=pipe MessageEndpoints.file_event(self.parent.name), msg.dumps(), pipe=pipe
) )
pipe.execute() pipe.execute()
@ -422,7 +422,7 @@ class PilatuscSAXS(PSIDetectorBase):
# specify Setup class # specify Setup class
custom_prepare_cls = PilatusSetup custom_prepare_cls = PilatusSetup
# specify minimum readout time for detector # specify minimum readout time for detector
PSIDetectorBase.set_min_readout(3e-3) MIN_READOUT = 3e-3
# specify class attributes # specify class attributes
cam = ADCpt(SLSDetectorCam, "cam1:") cam = ADCpt(SLSDetectorCam, "cam1:")

32
ophyd_devices/epics/devices/psi_detector_base.py
View File

@ -1,5 +1,4 @@
import time import time
import threading
from bec_lib.devicemanager import DeviceStatus from bec_lib.devicemanager import DeviceStatus
import os import os
@ -163,7 +162,7 @@ class CustomDetectorMixin:
] ]
if (all_signals and all(checks)) or (not all_signals and any(checks)): if (all_signals and all(checks)) or (not all_signals and any(checks)):
return True return True
if check_stopped == True and self.parent._stopped == True: if check_stopped == True and self.parent.stopped == True:
return False return False
if timer > timeout: if timer > timeout:
return False return False
@ -199,15 +198,15 @@ class PSIDetectorBase(Device):
custom_prepare_cls = CustomDetectorMixin custom_prepare_cls = CustomDetectorMixin
_MIN_READOUT = 1e-3 MIN_READOUT = 1e-3
@classmethod # @classmethod
def get_min_readout(cls): # def get_min_readout(cls):
return cls._MIN_READOUT # return cls._MIN_READOUT
@classmethod # @classmethod
def set_min_readout(cls, value): # def set_min_readout(cls, value):
cls._MIN_READOUT = value # cls._MIN_READOUT = value
# Specify which functions are revealed to the user in BEC client # Specify which functions are revealed to the user in BEC client
USER_ACCESS = [ USER_ACCESS = [
@ -243,8 +242,7 @@ class PSIDetectorBase(Device):
# sim_mode True allows the class to be started without BEC running # sim_mode True allows the class to be started without BEC running
# Init variables # Init variables
self.sim_mode = sim_mode self.sim_mode = sim_mode
self._lock = threading.RLock() self.stopped = False
self._stopped = False
self.name = name self.name = name
self.service_cfg = None self.service_cfg = None
self.std_client = None self.std_client = None
@ -264,7 +262,7 @@ class PSIDetectorBase(Device):
self.device_manager = bec_utils.DMMock() self.device_manager = bec_utils.DMMock()
base_path = kwargs["basepath"] if "basepath" in kwargs else "~/Data10/" base_path = kwargs["basepath"] if "basepath" in kwargs else "~/Data10/"
self.service_cfg = {"base_path": os.path.expanduser(base_path)} self.service_cfg = {"base_path": os.path.expanduser(base_path)}
self._producer = self.device_manager.producer self.producer = self.device_manager.producer
self._update_scaninfo() self._update_scaninfo()
self._update_filewriter() self._update_filewriter()
self._init() self._init()
@ -307,7 +305,7 @@ class PSIDetectorBase(Device):
return super().stage() return super().stage()
# Reset flag for detector stopped # Reset flag for detector stopped
self._stopped = False self.stopped = False
# Load metadata of the scan # Load metadata of the scan
self.scaninfo.load_scan_metadata() self.scaninfo.load_scan_metadata()
# Prepare detector and file writer # Prepare detector and file writer
@ -328,7 +326,7 @@ class PSIDetectorBase(Device):
""" """
Unstage device in preparation for a scan Unstage device in preparation for a scan
Returns directly if self._stopped, Returns directly if self.stopped,
otherwise checks with self._finished otherwise checks with self._finished
if data acquisition on device finished (an was successful) if data acquisition on device finished (an was successful)
@ -341,12 +339,12 @@ class PSIDetectorBase(Device):
List(object): list of objects that were unstaged List(object): list of objects that were unstaged
""" """
self.custom_prepare.check_scanID() self.custom_prepare.check_scanID()
if self._stopped == True: if self.stopped == True:
return super().unstage() return super().unstage()
self.custom_prepare.finished() self.custom_prepare.finished()
state = True state = True
self.custom_prepare.publish_file_location(done=state, successful=state) self.custom_prepare.publish_file_location(done=state, successful=state)
self._stopped = False self.stopped = False
return super().unstage() return super().unstage()
def stop(self, *, success=False) -> None: def stop(self, *, success=False) -> None:
@ -360,4 +358,4 @@ class PSIDetectorBase(Device):
self.custom_prepare.stop_detector() self.custom_prepare.stop_detector()
self.custom_prepare.stop_detector_backend() self.custom_prepare.stop_detector_backend()
super().stop(success=success) super().stop(success=success)
self._stopped = True self.stopped = True

20
tests/test_eiger9m_csaxs.py
View File

@ -278,7 +278,7 @@ def test_stage(
# TODO consider putting energy as variable in scaninfo # TODO consider putting energy as variable in scaninfo
mock_det.device_manager.add_device("mokev", value=12.4) mock_det.device_manager.add_device("mokev", value=12.4)
mock_det.cam.beam_energy.put(scaninfo["mokev"]) mock_det.cam.beam_energy.put(scaninfo["mokev"])
mock_det._stopped = stopped mock_det.stopped = stopped
mock_det.cam.detector_state._read_pv.mock_data = detector_state mock_det.cam.detector_state._read_pv.mock_data = detector_state
with mock.patch.object(mock_det.custom_prepare, "prepare_data_backend") as mock_prep_fw: with mock.patch.object(mock_det.custom_prepare, "prepare_data_backend") as mock_prep_fw:
mock_det.filepath = scaninfo["filepath"] mock_det.filepath = scaninfo["filepath"]
@ -393,15 +393,15 @@ def test_unstage(
with mock.patch.object(mock_det.custom_prepare, "finished") as mock_finished, mock.patch.object( with mock.patch.object(mock_det.custom_prepare, "finished") as mock_finished, mock.patch.object(
mock_det.custom_prepare, "publish_file_location" mock_det.custom_prepare, "publish_file_location"
) as mock_publish_file_location: ) as mock_publish_file_location:
mock_det._stopped = stopped mock_det.stopped = stopped
if expected_exception: if expected_exception:
mock_det.unstage() mock_det.unstage()
assert mock_det._stopped is True assert mock_det.stopped is True
else: else:
mock_det.unstage() mock_det.unstage()
mock_finished.assert_called_once() mock_finished.assert_called_once()
mock_publish_file_location.assert_called_with(done=True, successful=True) mock_publish_file_location.assert_called_with(done=True, successful=True)
assert mock_det._stopped is False assert mock_det.stopped is False
def test_stop_detector_backend(mock_det): def test_stop_detector_backend(mock_det):
@ -436,15 +436,15 @@ def test_publish_file_location(mock_det, scaninfo):
mock.call( mock.call(
MessageEndpoints.public_file(scaninfo["scanID"], mock_det.name), MessageEndpoints.public_file(scaninfo["scanID"], mock_det.name),
msg, msg,
pipe=mock_det._producer.pipeline.return_value, pipe=mock_det.producer.pipeline.return_value,
), ),
mock.call( mock.call(
MessageEndpoints.file_event(mock_det.name), MessageEndpoints.file_event(mock_det.name),
msg, msg,
pipe=mock_det._producer.pipeline.return_value, pipe=mock_det.producer.pipeline.return_value,
), ),
] ]
assert mock_det._producer.set_and_publish.call_args_list == expected_calls assert mock_det.producer.set_and_publish.call_args_list == expected_calls
def test_stop(mock_det): def test_stop(mock_det):
@ -456,7 +456,7 @@ def test_stop(mock_det):
mock_det.stop() mock_det.stop()
mock_stop_det.assert_called_once() mock_stop_det.assert_called_once()
mock_stop_detector_backend.assert_called_once() mock_stop_detector_backend.assert_called_once()
assert mock_det._stopped is True assert mock_det.stopped is True
@pytest.mark.parametrize( @pytest.mark.parametrize(
@ -520,13 +520,13 @@ def test_finished(mock_det, stopped, cam_state, daq_status, scaninfo, expected_e
with pytest.raises(Exception): with pytest.raises(Exception):
mock_det.timeout = 0.1 mock_det.timeout = 0.1
mock_det.custom_prepare.finished() mock_det.custom_prepare.finished()
assert mock_det._stopped is stopped assert mock_det.stopped is stopped
mock_stop_backend.assert_called() mock_stop_backend.assert_called()
mock_stop_det.assert_called_once() mock_stop_det.assert_called_once()
else: else:
mock_det.custom_prepare.finished() mock_det.custom_prepare.finished()
if stopped: if stopped:
assert mock_det._stopped is stopped assert mock_det.stopped is stopped
mock_stop_backend.assert_called() mock_stop_backend.assert_called()
mock_stop_det.assert_called_once() mock_stop_det.assert_called_once()

18
tests/test_falcon_csaxs.py
View File

@ -223,15 +223,15 @@ def test_publish_file_location(mock_det, scaninfo):
mock.call( mock.call(
MessageEndpoints.public_file(scaninfo["scanID"], mock_det.name), MessageEndpoints.public_file(scaninfo["scanID"], mock_det.name),
msg, msg,
pipe=mock_det._producer.pipeline.return_value, pipe=mock_det.producer.pipeline.return_value,
), ),
mock.call( mock.call(
MessageEndpoints.file_event(mock_det.name), MessageEndpoints.file_event(mock_det.name),
msg, msg,
pipe=mock_det._producer.pipeline.return_value, pipe=mock_det.producer.pipeline.return_value,
), ),
] ]
assert mock_det._producer.set_and_publish.call_args_list == expected_calls assert mock_det.producer.set_and_publish.call_args_list == expected_calls
@pytest.mark.parametrize( @pytest.mark.parametrize(
@ -281,16 +281,16 @@ def test_unstage(
with mock.patch.object(mock_det.custom_prepare, "finished") as mock_finished, mock.patch.object( with mock.patch.object(mock_det.custom_prepare, "finished") as mock_finished, mock.patch.object(
mock_det.custom_prepare, "publish_file_location" mock_det.custom_prepare, "publish_file_location"
) as mock_publish_file_location: ) as mock_publish_file_location:
mock_det._stopped = stopped mock_det.stopped = stopped
if expected_abort: if expected_abort:
mock_det.unstage() mock_det.unstage()
assert mock_det._stopped == stopped assert mock_det.stopped is stopped
assert mock_publish_file_location.call_count == 0 assert mock_publish_file_location.call_count == 0
else: else:
mock_det.unstage() mock_det.unstage()
mock_finished.assert_called_once() mock_finished.assert_called_once()
mock_publish_file_location.assert_called_with(done=True, successful=True) mock_publish_file_location.assert_called_with(done=True, successful=True)
assert mock_det._stopped == False assert mock_det.stopped is stopped
def test_stop(mock_det): def test_stop(mock_det):
@ -302,7 +302,7 @@ def test_stop(mock_det):
mock_det.stop() mock_det.stop()
mock_stop_det.assert_called_once() mock_stop_det.assert_called_once()
mock_stop_detector_backend.assert_called_once() mock_stop_detector_backend.assert_called_once()
assert mock_det._stopped == True assert mock_det.stopped is True
@pytest.mark.parametrize( @pytest.mark.parametrize(
@ -318,7 +318,7 @@ def test_finished(mock_det, stopped, scaninfo):
) as mock_stop_det, mock.patch.object( ) as mock_stop_det, mock.patch.object(
mock_det.custom_prepare, "stop_detector_backend" mock_det.custom_prepare, "stop_detector_backend"
) as mock_stop_file_writer: ) as mock_stop_file_writer:
mock_det._stopped = stopped mock_det.stopped = stopped
mock_det.dxp.current_pixel._read_pv.mock_data = int( mock_det.dxp.current_pixel._read_pv.mock_data = int(
scaninfo["num_points"] * scaninfo["frames_per_trigger"] scaninfo["num_points"] * scaninfo["frames_per_trigger"]
) )
@ -328,6 +328,6 @@ def test_finished(mock_det, stopped, scaninfo):
mock_det.scaninfo.frames_per_trigger = scaninfo["frames_per_trigger"] mock_det.scaninfo.frames_per_trigger = scaninfo["frames_per_trigger"]
mock_det.scaninfo.num_points = scaninfo["num_points"] mock_det.scaninfo.num_points = scaninfo["num_points"]
mock_det.custom_prepare.finished() mock_det.custom_prepare.finished()
assert mock_det._stopped == stopped assert mock_det.stopped is stopped
mock_stop_det.assert_called_once() mock_stop_det.assert_called_once()
mock_stop_file_writer.assert_called_once() mock_stop_file_writer.assert_called_once()

22
tests/test_pilatus_csaxs.py
View File

@ -114,7 +114,7 @@ def test_stage(
mock_det.scaninfo.frames_per_trigger = scaninfo["frames_per_trigger"] mock_det.scaninfo.frames_per_trigger = scaninfo["frames_per_trigger"]
mock_det.filewriter.compile_full_filename.return_value = scaninfo["filepath"] mock_det.filewriter.compile_full_filename.return_value = scaninfo["filepath"]
mock_det.device_manager.add_device("mokev", value=12.4) mock_det.device_manager.add_device("mokev", value=12.4)
mock_det._stopped = stopped mock_det.stopped = stopped
with mock.patch.object( with mock.patch.object(
mock_det.custom_prepare, "prepare_data_backend" mock_det.custom_prepare, "prepare_data_backend"
) as mock_data_backend, mock.patch.object( ) as mock_data_backend, mock.patch.object(
@ -218,15 +218,15 @@ def test_publish_file_location(mock_det, scaninfo):
mock.call( mock.call(
MessageEndpoints.public_file(scaninfo["scanID"], mock_det.name), MessageEndpoints.public_file(scaninfo["scanID"], mock_det.name),
msg, msg,
pipe=mock_det._producer.pipeline.return_value, pipe=mock_det.producer.pipeline.return_value,
), ),
mock.call( mock.call(
MessageEndpoints.file_event(mock_det.name), MessageEndpoints.file_event(mock_det.name),
msg, msg,
pipe=mock_det._producer.pipeline.return_value, pipe=mock_det.producer.pipeline.return_value,
), ),
] ]
assert mock_det._producer.set_and_publish.call_args_list == expected_calls assert mock_det.producer.set_and_publish.call_args_list == expected_calls
@pytest.mark.parametrize( @pytest.mark.parametrize(
@ -451,15 +451,15 @@ def test_unstage(
with mock.patch.object(mock_det.custom_prepare, "finished") as mock_finished, mock.patch.object( with mock.patch.object(mock_det.custom_prepare, "finished") as mock_finished, mock.patch.object(
mock_det.custom_prepare, "publish_file_location" mock_det.custom_prepare, "publish_file_location"
) as mock_publish_file_location: ) as mock_publish_file_location:
mock_det._stopped = stopped mock_det.stopped = stopped
if expected_exception: if expected_exception:
mock_det.unstage() mock_det.unstage()
assert mock_det._stopped == True assert mock_det.stopped is True
else: else:
mock_det.unstage() mock_det.unstage()
mock_finished.assert_called_once() mock_finished.assert_called_once()
mock_publish_file_location.assert_called_with(done=True, successful=True) mock_publish_file_location.assert_called_with(done=True, successful=True)
assert mock_det._stopped == False assert mock_det.stopped is False
def test_stop(mock_det): def test_stop(mock_det):
@ -474,7 +474,7 @@ def test_stop(mock_det):
mock_stop_det.assert_called_once() mock_stop_det.assert_called_once()
mock_stop_file_writer.assert_called_once() mock_stop_file_writer.assert_called_once()
mock_close_file_writer.assert_called_once() mock_close_file_writer.assert_called_once()
assert mock_det._stopped == True assert mock_det.stopped is True
@pytest.mark.parametrize( @pytest.mark.parametrize(
@ -506,19 +506,19 @@ def test_finished(mock_det, stopped, mcs_stage_state, expected_exception):
mock_det.custom_prepare, "close_file_writer" mock_det.custom_prepare, "close_file_writer"
) as mock_close_file_writer: ) as mock_close_file_writer:
mock_dm.devices.mcs.obj._staged = mcs_stage_state mock_dm.devices.mcs.obj._staged = mcs_stage_state
mock_det._stopped = stopped mock_det.stopped = stopped
if expected_exception: if expected_exception:
with pytest.raises(Exception): with pytest.raises(Exception):
mock_det.timeout = 0.1 mock_det.timeout = 0.1
mock_det.custom_prepare.finished() mock_det.custom_prepare.finished()
assert mock_det._stopped == stopped assert mock_det.stopped is stopped
mock_stop_file_friter.assert_called() mock_stop_file_friter.assert_called()
mock_stop_det.assert_called_once() mock_stop_det.assert_called_once()
mock_close_file_writer.assert_called_once() mock_close_file_writer.assert_called_once()
else: else:
mock_det.custom_prepare.finished() mock_det.custom_prepare.finished()
if stopped: if stopped:
assert mock_det._stopped == stopped assert mock_det.stopped is stopped
mock_stop_file_friter.assert_called() mock_stop_file_friter.assert_called()
mock_stop_det.assert_called_once() mock_stop_det.assert_called_once()