feat: adapt detector classes and test for psi_detector_base refactoring

csaxs_bec/devices/epics/eiger9m_csaxs.py

@@ -5,8 +5,6 @@ import time
 from typing import Any
 
 import numpy as np
-from bec_lib import messages
-from bec_lib.endpoints import MessageEndpoints
 from bec_lib.logger import bec_logger
 from ophyd import ADComponent as ADCpt
 from ophyd import Device, EpicsSignal, EpicsSignalRO, EpicsSignalWithRBV
@@ -42,6 +40,17 @@ class Eiger9MSetup(CustomDetectorMixin):
         self.std_client = None
         self._lock = threading.RLock()
 
+    def on_init(self) -> None:
+        """Initialize the detector"""
+        self.initialize_default_parameter()
+        self.initialize_detector()
+        self.initialize_detector_backend()
+
+    def initialize_detector(self) -> None:
+        """Initialize detector"""
+        self.stop_detector()
+        self.parent.cam.trigger_mode.put(TriggerSource.GATING)
+
     def initialize_default_parameter(self) -> None:
         """Set default parameters for Eiger9M detector"""
         self.update_readout_time()
@@ -55,29 +64,19 @@ class Eiger9MSetup(CustomDetectorMixin):
         )
         self.parent.readout_time = max(readout_time, self.parent.MIN_READOUT)
 
-    def initialize_detector(self) -> None:
-        """Initialize detector"""
-        # Stops the detector
-        self.stop_detector()
-        # Sets the trigger source to GATING
-        self.parent.set_trigger(TriggerSource.GATING)
-
     def initialize_detector_backend(self) -> None:
         """Initialize detector backend"""
 
-        # Std client
         self.std_client = StdDaqClient(url_base=self.std_rest_server_url)
-
-        # Stop writer
         self.std_client.stop_writer()
-
-        # Change e-account
         eacc = self.parent.scaninfo.username
         self.update_std_cfg("writer_user_id", int(eacc.strip(" e")))
 
         signal_conditions = [(lambda: self.std_client.get_status()["state"], "READY")]
         if not self.wait_for_signals(
-            signal_conditions=signal_conditions, timeout=self.parent.timeout, all_signals=True
+            signal_conditions=signal_conditions,
+            timeout=self.parent.TIMEOUT_FOR_SIGNALS,
+            all_signals=True,
         ):
             raise EigerTimeoutError(
                 f"Std client not in READY state, returns: {self.std_client.get_status()}"
@@ -98,7 +97,6 @@ class Eiger9MSetup(CustomDetectorMixin):
 
         """
 
-        # Load config from client and check old value
         cfg = self.std_client.get_config()
         old_value = cfg.get(cfg_key)
         if old_value is None:
@@ -111,55 +109,23 @@ class Eiger9MSetup(CustomDetectorMixin):
                 f" {type(old_value)}:{old_value}"
             )
 
-        # Update config with new value and send back to client
         cfg.update({cfg_key: value})
         logger.debug(cfg)
         self.std_client.set_config(cfg)
         logger.debug(f"Updated std_daq config for key {cfg_key} from {old_value} to {value}")
 
-    def stop_detector(self) -> None:
-        """Stop the detector"""
-
-        # Stop detector
-        self.parent.cam.acquire.put(0)
-
-        # Check if detector returned in idle state
-        signal_conditions = [
-            (
-                lambda: self.parent.cam.detector_state.read()[self.parent.cam.detector_state.name][
-                    "value"
-                ],
-                DetectorState.IDLE,
-            )
-        ]
-
-        if not self.wait_for_signals(
-            signal_conditions=signal_conditions,
-            timeout=self.parent.timeout - self.parent.timeout // 2,
-            check_stopped=True,
-            all_signals=False,
-        ):
-            # Retry stop detector and wait for remaining time
-            self.parent.cam.acquire.put(0)
-            if not self.wait_for_signals(
-                signal_conditions=signal_conditions,
-                timeout=self.parent.timeout - self.parent.timeout // 2,
-                check_stopped=True,
-                all_signals=False,
-            ):
-                raise EigerTimeoutError(
-                    f"Failed to stop detector, detector state {signal_conditions[0][0]}"
-                )
-
-    def stop_detector_backend(self) -> None:
-        """Close file writer"""
-        self.std_client.stop_writer()
+    def on_stage(self) -> None:
+        """Prepare the detector for scan"""
+        self.prepare_detector()
+        self.prepare_data_backend()
+        self.publish_file_location(done=False)
+        self.arm_acquisition()
 
     def prepare_detector(self) -> None:
         """Prepare detector for scan"""
         self.set_detector_threshold()
         self.set_acquisition_params()
-        self.parent.set_trigger(TriggerSource.GATING)
+        self.parent.cam.trigger_mode.put(TriggerSource.GATING)
 
     def set_detector_threshold(self) -> None:
         """
@@ -167,25 +133,20 @@ class Eiger9MSetup(CustomDetectorMixin):
 
         The function sets the detector threshold automatically to 1/2 of the beam energy.
         """
-
-        # get current beam energy from device manageer
         mokev = self.parent.device_manager.devices.mokev.obj.read()[
             self.parent.device_manager.devices.mokev.name
         ]["value"]
         factor = 1
-
-        # Check if energies are eV or keV, assume keV as the default
         unit = getattr(self.parent.cam.threshold_energy, "units", None)
+
         if unit is not None and unit == "eV":
             factor = 1000
-
-        # set energy on detector
         setpoint = int(mokev * factor)
         energy = self.parent.cam.beam_energy.read()[self.parent.cam.beam_energy.name]["value"]
+
         if setpoint != energy:
             self.parent.cam.beam_energy.set(setpoint)
 
-        # set threshold on detector
         threshold = self.parent.cam.threshold_energy.read()[self.parent.cam.threshold_energy.name][
             "value"
         ]
@@ -194,14 +155,10 @@ class Eiger9MSetup(CustomDetectorMixin):
 
     def set_acquisition_params(self) -> None:
         """Set acquisition parameters for the detector"""
-
-        # Set number of images and frames (frames is for internal burst of detector)
         self.parent.cam.num_images.put(
             int(self.parent.scaninfo.num_points * self.parent.scaninfo.frames_per_trigger)
         )
         self.parent.cam.num_frames.put(1)
-
-        # Update the readout time of the detector
         self.update_readout_time()
 
     def prepare_data_backend(self) -> None:
@@ -225,13 +182,12 @@ class Eiger9MSetup(CustomDetectorMixin):
             if self.std_client.get_status()["state"] == "READY":
                 raise EigerTimeoutError(f"Timeout of start_writer_async with {exc}") from exc
 
-        # Check status of std_daq
         signal_conditions = [
             (lambda: self.std_client.get_status()["acquisition"]["state"], "WAITING_IMAGES")
         ]
         if not self.wait_for_signals(
             signal_conditions=signal_conditions,
-            timeout=self.parent.timeout,
+            timeout=self.parent.TIMEOUT_FOR_SIGNALS,
             check_stopped=False,
             all_signals=True,
         ):
@@ -240,12 +196,58 @@ class Eiger9MSetup(CustomDetectorMixin):
                 f" {self.std_client.get_status()}"
             )
 
+    def on_unstage(self) -> None:
+        """Unstage the detector"""
+        self.finished(timeout=self.parent.TIMEOUT_FOR_SIGNALS)
+        self.publish_file_location(done=True, successful=True)
+
+    def on_stop(self) -> None:
+        """Stop the detector"""
+        self.stop_detector()
+        self.stop_detector_backend()
+
+    def stop_detector(self) -> None:
+        """Stop the detector"""
+
+        # Stop detector
+        self.parent.cam.acquire.put(0)
+        signal_conditions = [
+            (
+                lambda: self.parent.cam.detector_state.read()[self.parent.cam.detector_state.name][
+                    "value"
+                ],
+                DetectorState.IDLE,
+            )
+        ]
+
+        if not self.wait_for_signals(
+            signal_conditions=signal_conditions,
+            timeout=self.parent.TIMEOUT_FOR_SIGNALS - self.parent.TIMEOUT_FOR_SIGNALS // 2,
+            check_stopped=True,
+            all_signals=False,
+        ):
+            # Retry stop detector and wait for remaining time
+            self.parent.cam.acquire.put(0)
+            if not self.wait_for_signals(
+                signal_conditions=signal_conditions,
+                timeout=self.parent.TIMEOUT_FOR_SIGNALS - self.parent.TIMEOUT_FOR_SIGNALS // 2,
+                check_stopped=True,
+                all_signals=False,
+            ):
+                raise EigerTimeoutError(
+                    f"Failed to stop detector, detector state {signal_conditions[0][0]}"
+                )
+
+    def stop_detector_backend(self) -> None:
+        """Close file writer"""
+        self.std_client.stop_writer()
+
     def filepath_exists(self, filepath: str) -> None:
         """Check if filepath exists"""
         signal_conditions = [(lambda: os.path.exists(os.path.dirname(filepath)), True)]
         if not self.wait_for_signals(
             signal_conditions=signal_conditions,
-            timeout=self.parent.timeout,
+            timeout=self.parent.TIMEOUT_FOR_SIGNALS,
             check_stopped=False,
             all_signals=True,
         ):
@@ -264,7 +266,7 @@ class Eiger9MSetup(CustomDetectorMixin):
         ]
         if not self.wait_for_signals(
             signal_conditions=signal_conditions,
-            timeout=self.parent.timeout,
+            timeout=self.parent.TIMEOUT_FOR_SIGNALS,
             check_stopped=True,
             all_signals=False,
         ):
@@ -272,43 +274,7 @@ class Eiger9MSetup(CustomDetectorMixin):
                 f"Failed to arm the acquisition. Detector state {signal_conditions[0][0]}"
             )
 
-    def check_scan_id(self) -> None:
-        """Checks if scan_id has changed and stops the scan if it has"""
-        old_scan_id = self.parent.scaninfo.scan_id
-        self.parent.scaninfo.load_scan_metadata()
-        if self.parent.scaninfo.scan_id != old_scan_id:
-            self.parent.stopped = True
-
-    def publish_file_location(self, done: bool = False, successful: bool = None) -> None:
-        """
-        Publish the filepath to REDIS.
-
-        We publish two events here:
-        - file_event: event for the filewriter
-        - public_file: event for any secondary service (e.g. radial integ code)
-
-        Args:
-            done (bool): True if scan is finished
-            successful (bool): True if scan was successful
-        """
-        pipe = self.parent.connector.pipeline()
-        if successful is None:
-            msg = messages.FileMessage(file_path=self.parent.filepath, done=done)
-        else:
-            msg = messages.FileMessage(
-                file_path=self.parent.filepath, done=done, successful=successful
-            )
-        self.parent.connector.set_and_publish(
-            MessageEndpoints.public_file(self.parent.scaninfo.scan_id, self.parent.name),
-            msg,
-            pipe=pipe,
-        )
-        self.parent.connector.set_and_publish(
-            MessageEndpoints.file_event(self.parent.name), msg, pipe=pipe
-        )
-        pipe.execute()
-
-    def finished(self):
+    def finished(self, timeout: int = 5) -> None:
         """Check if acquisition is finished."""
         with self._lock:
             signal_conditions = [
@@ -357,7 +323,7 @@ class SLSDetectorCam(Device):
     detector_state = ADCpt(EpicsSignalRO, "DetectorState_RBV")
 
 
-class TriggerSource(enum.IntEnum):
+class TriggerSource(int, enum.Enum):
     """Trigger signals for Eiger9M detector"""
 
     AUTO = 0
@@ -366,7 +332,7 @@ class TriggerSource(enum.IntEnum):
     BURST_TRIGGER = 3
 
 
-class DetectorState(enum.IntEnum):
+class DetectorState(int, enum.Enum):
     """Detector states for Eiger9M detector"""
 
     IDLE = 0
@@ -396,37 +362,16 @@ class Eiger9McSAXS(PSIDetectorBase):
     """
 
     # Specify which functions are revealed to the user in BEC client
-    USER_ACCESS = ["describe"]
+    USER_ACCESS = []
 
     # specify Setup class
     custom_prepare_cls = Eiger9MSetup
-    # specify minimum readout time for detector
+    # specify minimum readout time for detector and timeout for checks after unstage
     MIN_READOUT = 3e-3
+    TIMEOUT_FOR_SIGNALS = 5
     # specify class attributes
     cam = ADCpt(SLSDetectorCam, "cam1:")
 
-    def set_trigger(self, trigger_source: TriggerSource) -> None:
-        """Set trigger source for the detector.
-        Check the TriggerSource enum for possible values
-
-        Args:
-            trigger_source (TriggerSource): Trigger source for the detector
-
-        """
-        value = trigger_source
-        self.cam.trigger_mode.put(value)
-
-    def stage(self) -> list[object]:
-        """
-        Add functionality to stage, and arm the detector
-
-        Additional call to:
-        - custom_prepare.arm_acquisition()
-        """
-        rtr = super().stage()
-        self.custom_prepare.arm_acquisition()
-        return rtr
-
 
 if __name__ == "__main__":
     eiger = Eiger9McSAXS(name="eiger", prefix="X12SA-ES-EIGER9M:", sim_mode=True)

csaxs_bec/devices/epics/falcon_csaxs.py

@@ -1,8 +1,8 @@
 import enum
 import os
+import threading
 
-from bec_lib import messages
-from bec_lib.endpoints import MessageEndpoints
+from bec_lib import threadlocked
 from bec_lib.logger import bec_logger
 from ophyd import Component as Cpt
 from ophyd import Device, EpicsSignal, EpicsSignalRO, EpicsSignalWithRBV
@@ -99,6 +99,16 @@ class FalconSetup(CustomDetectorMixin):
 
     """
 
+    def __init__(self, *args, parent: Device = None, **kwargs) -> None:
+        super().__init__(*args, parent=parent, **kwargs)
+        self._lock = threading.RLock()
+
+    def on_init(self) -> None:
+        """Initialize Falcon detector"""
+        self.initialize_default_parameter()
+        self.initialize_detector()
+        self.initialize_detector_backend()
+
     def initialize_default_parameter(self) -> None:
         """
         Set default parameters for Falcon
@@ -124,7 +134,7 @@ class FalconSetup(CustomDetectorMixin):
         """Initialize Falcon detector"""
         self.stop_detector()
         self.stop_detector_backend()
-        self.parent.set_trigger(
+        self.set_trigger(
             mapping_mode=MappingSource.MAPPING, trigger_source=TriggerSource.GATE, ignore_gate=0
         )
         # 1 Realtime
@@ -136,30 +146,6 @@ class FalconSetup(CustomDetectorMixin):
         # Sets the number of pixels/spectra in the buffer
         self.parent.pixels_per_buffer.put(self.parent.value_pixel_per_buffer)
 
-    def stop_detector(self) -> None:
-        """Stops detector"""
-
-        self.parent.stop_all.put(1)
-        self.parent.erase_all.put(1)
-
-        signal_conditions = [
-            (lambda: self.parent.state.read()[self.parent.state.name]["value"], DetectorState.DONE)
-        ]
-
-        if not self.wait_for_signals(
-            signal_conditions=signal_conditions,
-            timeout=self.parent.timeout - self.parent.timeout // 2,
-            all_signals=False,
-        ):
-            # Retry stop detector and wait for remaining time
-            raise FalconTimeoutError(
-                f"Failed to stop detector, timeout with state {signal_conditions[0][0]}"
-            )
-
-    def stop_detector_backend(self) -> None:
-        """Stop the detector backend"""
-        self.parent.hdf5.capture.put(0)
-
     def initialize_detector_backend(self) -> None:
         """Initialize the detector backend for Falcon."""
         self.parent.hdf5.enable.put(1)
@@ -173,9 +159,16 @@ class FalconSetup(CustomDetectorMixin):
         # Segmentation into Spectra within EPICS, 1 is activate, 0 is deactivate
         self.parent.nd_array_mode.put(1)
 
+    def on_stage(self) -> None:
+        """Prepare detector and backend for acquisition"""
+        self.prepare_detector()
+        self.prepare_data_backend()
+        self.publish_file_location(done=False)
+        self.arm_acquisition()
+
     def prepare_detector(self) -> None:
         """Prepare detector for acquisition"""
-        self.parent.set_trigger(
+        self.set_trigger(
             mapping_mode=MappingSource.MAPPING, trigger_source=TriggerSource.GATE, ignore_gate=0
         )
         self.parent.preset_real.put(self.parent.scaninfo.exp_time)
@@ -212,7 +205,7 @@ class FalconSetup(CustomDetectorMixin):
         ]
         if not self.wait_for_signals(
             signal_conditions=signal_conditions,
-            timeout=self.parent.timeout,
+            timeout=self.parent.TIMEOUT_FOR_SIGNALS,
             check_stopped=True,
             all_signals=False,
         ):
@@ -220,42 +213,41 @@ class FalconSetup(CustomDetectorMixin):
                 f"Failed to arm the acquisition. Detector state {signal_conditions[0][0]}"
             )
 
-    def check_scan_id(self) -> None:
-        """Checks if scan_id has changed and stops the scan if it has"""
-        old_scan_id = self.parent.scaninfo.scan_id
-        self.parent.scaninfo.load_scan_metadata()
-        if self.parent.scaninfo.scan_id != old_scan_id:
-            self.parent.stopped = True
+    def on_unstage(self) -> None:
+        """Unstage detector and backend"""
+        self.finished()
+        self.publish_file_location(done=True, successful=True)
 
-    def publish_file_location(self, done: bool = False, successful: bool = None) -> None:
-        """
-        Publish the filepath to REDIS.
+    def on_stop(self) -> None:
+        """Stop detector and backend"""
+        self.stop_detector()
+        self.stop_detector_backend()
 
-        We publish two events here:
-        - file_event: event for the filewriter
-        - public_file: event for any secondary service (e.g. radial integ code)
+    def stop_detector(self) -> None:
+        """Stops detector"""
 
-        Args:
-            done (bool): True if scan is finished
-            successful (bool): True if scan was successful
-        """
-        pipe = self.parent.connector.pipeline()
-        if successful is None:
-            msg = messages.FileMessage(file_path=self.parent.filepath, done=done)
-        else:
-            msg = messages.FileMessage(
-                file_path=self.parent.filepath, done=done, successful=successful
+        self.parent.stop_all.put(1)
+        self.parent.erase_all.put(1)
+
+        signal_conditions = [
+            (lambda: self.parent.state.read()[self.parent.state.name]["value"], DetectorState.DONE)
+        ]
+
+        if not self.wait_for_signals(
+            signal_conditions=signal_conditions,
+            timeout=self.parent.TIMEOUT_FOR_SIGNALS - self.parent.TIMEOUT_FOR_SIGNALS // 2,
+            all_signals=False,
+        ):
+            # Retry stop detector and wait for remaining time
+            raise FalconTimeoutError(
+                f"Failed to stop detector, timeout with state {signal_conditions[0][0]}"
             )
-        self.parent.connector.set_and_publish(
-            MessageEndpoints.public_file(self.parent.scaninfo.scan_id, self.parent.name),
-            msg,
-            pipe=pipe,
-        )
-        self.parent.connector.set_and_publish(
-            MessageEndpoints.file_event(self.parent.name), msg, pipe=pipe
-        )
-        pipe.execute()
 
+    def stop_detector_backend(self) -> None:
+        """Stop the detector backend"""
+        self.parent.hdf5.capture.put(0)
+
+    @threadlocked
     def finished(self) -> None:
         """Check if scan finished succesfully"""
         total_frames = int(
@@ -267,7 +259,7 @@ class FalconSetup(CustomDetectorMixin):
         ]
         if not self.wait_for_signals(
             signal_conditions=signal_conditions,
-            timeout=self.parent.timeout,
+            timeout=self.parent.TIMEOUT_FOR_SIGNALS,
             check_stopped=True,
             all_signals=True,
         ):
@@ -279,6 +271,24 @@ class FalconSetup(CustomDetectorMixin):
         self.stop_detector()
         self.stop_detector_backend()
 
+    def set_trigger(
+        self, mapping_mode: MappingSource, trigger_source: TriggerSource, ignore_gate: int = 0
+    ) -> None:
+        """
+        Set triggering mode for detector
+
+        Args:
+            mapping_mode (MappingSource): Mapping mode for the detector
+            trigger_source (TriggerSource): Trigger source for the detector, pixel_advance_signal
+            ignore_gate (int): Ignore gate from TTL signal; defaults to 0
+
+        """
+        mapping = int(mapping_mode)
+        trigger = trigger_source
+        self.parent.collect_mode.put(mapping)
+        self.parent.pixel_advance_mode.put(trigger)
+        self.parent.ignore_gate.put(ignore_gate)
+
 
 class FalconcSAXS(PSIDetectorBase):
     """
@@ -303,6 +313,7 @@ class FalconcSAXS(PSIDetectorBase):
     custom_prepare_cls = FalconSetup
     # specify minimum readout time for detector
     MIN_READOUT = 3e-3
+    TIMEOUT_FOR_SIGNALS = 5
 
     # specify class attributes
     dxp = Cpt(EpicsDXPFalcon, "dxp1:")
@@ -330,30 +341,6 @@ class FalconcSAXS(PSIDetectorBase):
     pixels_per_run = Cpt(EpicsSignal, "PixelsPerRun")
     nd_array_mode = Cpt(EpicsSignal, "NDArrayMode")
 
-    def set_trigger(
-        self, mapping_mode: MappingSource, trigger_source: TriggerSource, ignore_gate: int = 0
-    ) -> None:
-        """
-        Set triggering mode for detector
-
-        Args:
-            mapping_mode (MappingSource): Mapping mode for the detector
-            trigger_source (TriggerSource): Trigger source for the detector, pixel_advance_signal
-            ignore_gate (int): Ignore gate from TTL signal; defaults to 0
-
-        """
-        mapping = int(mapping_mode)
-        trigger = trigger_source
-        self.collect_mode.put(mapping)
-        self.pixel_advance_mode.put(trigger)
-        self.ignore_gate.put(ignore_gate)
-
-    def stage(self) -> list[object]:
-        """Stage"""
-        rtr = super().stage()
-        self.custom_prepare.arm_acquisition()
-        return rtr
-
 
 if __name__ == "__main__":
     falcon = FalconcSAXS(name="falcon", prefix="X12SA-SITORO:", sim_mode=True)

csaxs_bec/devices/epics/mcs_csaxs.py

@@ -74,6 +74,11 @@ class MCSSetup(CustomDetectorMixin):
         ]
         self.mca_data = defaultdict(lambda: [])
 
+    def on_init(self) -> None:
+        """Init sequence for the detector"""
+        self.initialize_detector()
+        self.initialize_detector_backend()
+
     def initialize_detector(self) -> None:
         """Initialize detector"""
         # External trigger for pixel advance
@@ -84,7 +89,7 @@ class MCSSetup(CustomDetectorMixin):
         # Set number of channels to 5
         self.parent.mux_output.set(5)
         # Trigger Mode used for cSAXS
-        self.parent.set_trigger(TriggerSource.MODE3)
+        self.parent.input_mode.set(TriggerSource.MODE3)
         # specify polarity of trigger signals
         self.parent.input_polarity.set(0)
         self.parent.output_polarity.set(1)
@@ -142,10 +147,15 @@ class MCSSetup(CustomDetectorMixin):
             expire=self._stream_ttl,
         )
 
+    def on_stage(self) -> None:
+        """Stage detector"""
+        self.prepare_detector()
+        self.prepare_detector_backend()
+
     def prepare_detector(self) -> None:
         """Prepare detector for scan"""
         self.set_acquisition_params()
-        self.parent.set_trigger(TriggerSource.MODE3)
+        self.parent.input_mode.set(TriggerSource.MODE3)
 
     def set_acquisition_params(self) -> None:
         """Set acquisition parameters for scan"""
@@ -175,6 +185,10 @@ class MCSSetup(CustomDetectorMixin):
         self.counter = 0
         self.parent.erase_start.set(1)
 
+    def on_unstage(self) -> None:
+        """Unstage detector"""
+        self.finished()
+
     def finished(self) -> None:
         """Check if acquisition is finished, if not successful, rais MCSTimeoutError"""
         signal_conditions = [
@@ -183,7 +197,7 @@ class MCSSetup(CustomDetectorMixin):
         ]
         if not self.wait_for_signals(
             signal_conditions=signal_conditions,
-            timeout=self.parent.timeout,
+            timeout=self.parent.TIMEOUT_FOR_SIGNALS,
             check_stopped=True,
             all_signals=True,
         ):
@@ -195,12 +209,15 @@ class MCSSetup(CustomDetectorMixin):
                 f" {total_frames} frames arriving at the mcs card"
             )
 
+    def on_stop(self) -> None:
+        """Stop detector"""
+        self.stop_detector()
+        self.stop_detector_backend()
+
     def stop_detector(self) -> None:
         """Stop detector"""
         self.parent.stop_all.set(1)
 
-        return super().stop_detector()
-
     def stop_detector_backend(self) -> None:
         """Stop acquisition of data"""
         self.acquisition_done = True
@@ -213,7 +230,7 @@ class SIS38XX(Device):
 class MCScSAXS(PSIDetectorBase):
     """MCS card for cSAXS for implementation at cSAXS beamline"""
 
-    USER_ACCESS = ["describe", "_init_mcs"]
+    USER_ACCESS = []
     SUB_PROGRESS = "progress"
     SUB_VALUE = "value"
     _default_sub = SUB_VALUE
@@ -222,6 +239,7 @@ class MCScSAXS(PSIDetectorBase):
     custom_prepare_cls = MCSSetup
     # specify minimum readout time for detector
     MIN_READOUT = 0
+    TIMEOUT_FOR_SIGNALS = 5
 
     # PV access to SISS38XX card
     # Acquisition
@@ -272,11 +290,8 @@ class MCScSAXS(PSIDetectorBase):
         *,
         name,
         kind=None,
-        read_attrs=None,
-        configuration_attrs=None,
         parent=None,
         device_manager=None,
-        sim_mode=False,
         mcs_config=None,
         **kwargs,
     ):
@@ -289,25 +304,11 @@ class MCScSAXS(PSIDetectorBase):
             prefix=prefix,
             name=name,
             kind=kind,
-            read_attrs=read_attrs,
-            configuration_attrs=configuration_attrs,
             parent=parent,
             device_manager=device_manager,
-            sim_mode=sim_mode,
             **kwargs,
         )
 
-    def set_trigger(self, trigger_source: TriggerSource) -> None:
-        """Set trigger mode from TriggerSource"""
-        value = int(trigger_source)
-        self.input_mode.set(value)
-
-    def stage(self) -> list[object]:
-        """stage the detector for upcoming acquisition"""
-        rtr = super().stage()
-        self.custom_prepare.arm_acquisition()
-        return rtr
-
 
 # Automatically connect to test environmenr if directly invoked
 if __name__ == "__main__":

csaxs_bec/devices/epics/pilatus_csaxs.py

@@ -1,12 +1,12 @@
 import enum
 import json
 import os
+import threading
 import time
 
 import numpy as np
 import requests
-from bec_lib import bec_logger, messages
-from bec_lib.endpoints import MessageEndpoints
+from bec_lib import bec_logger
 from ophyd import ADComponent as ADCpt
 from ophyd import Device, EpicsSignal, EpicsSignalRO, EpicsSignalWithRBV, Staged
 from ophyd_devices.interfaces.base_classes.psi_detector_base import (
@@ -16,8 +16,6 @@ from ophyd_devices.interfaces.base_classes.psi_detector_base import (
 
 logger = bec_logger.logger
 
-MIN_READOUT = 3e-3
-
 
 class PilatusError(Exception):
     """Base class for exceptions in this module."""
@@ -72,6 +70,15 @@ class PilatusSetup(CustomDetectorMixin):
 
     """
 
+    def __init__(self, *args, parent: Device = None, **kwargs) -> None:
+        super().__init__(*args, parent=parent, **kwargs)
+        self._lock = threading.RLock()
+
+    def on_init(self) -> None:
+        """Initialize the detector"""
+        self.initialize_default_parameter()
+        self.initialize_detector()
+
     def initialize_default_parameter(self) -> None:
         """Set default parameters for Eiger9M detector"""
         self.update_readout_time()
@@ -90,7 +97,13 @@ class PilatusSetup(CustomDetectorMixin):
         # Stops the detector
         self.stop_detector()
         # Sets the trigger source to GATING
-        self.parent.set_trigger(TriggerSource.EXT_ENABLE)
+        self.parent.cam.trigger_mode.put(TriggerSource.EXT_ENABLE)
+
+    def on_stage(self) -> None:
+        """Stage the detector for scan"""
+        self.prepare_detector()
+        self.prepare_data_backend()
+        self.publish_file_location(done=False, metadata={"input_path": self.parent.filepath_raw})
 
     def prepare_detector(self) -> None:
         """
@@ -101,84 +114,7 @@ class PilatusSetup(CustomDetectorMixin):
         """
         self.set_detector_threshold()
         self.set_acquisition_params()
-        self.parent.set_trigger(TriggerSource.EXT_ENABLE)
-
-    def set_detector_threshold(self) -> None:
-        """
-        Set correct detector threshold to 1/2 of current X-ray energy, allow 5% tolerance
-
-        Threshold might be in ev or keV
-        """
-
-        # get current beam energy from device manageer
-        mokev = self.parent.device_manager.devices.mokev.obj.read()[
-            self.parent.device_manager.devices.mokev.name
-        ]["value"]
-        factor = 1
-
-        # Check if energies are eV or keV, assume keV as the default
-        unit = getattr(self.parent.cam.threshold_energy, "units", None)
-        if unit is not None and unit == "eV":
-            factor = 1000
-
-        # set energy on detector
-        setpoint = int(mokev * factor)
-
-        # set threshold on detector
-        threshold = self.parent.cam.threshold_energy.read()[self.parent.cam.threshold_energy.name][
-            "value"
-        ]
-        if not np.isclose(setpoint / 2, threshold, rtol=0.05):
-            self.parent.cam.threshold_energy.set(setpoint / 2)
-
-    def set_acquisition_params(self) -> None:
-        """Set acquisition parameters for the detector"""
-
-        # Set number of images and frames (frames is for internal burst of detector)
-        self.parent.cam.num_images.put(
-            int(self.parent.scaninfo.num_points * self.parent.scaninfo.frames_per_trigger)
-        )
-        self.parent.cam.num_frames.put(1)
-
-        # Update the readout time of the detector
-        self.update_readout_time()
-
-    def create_directory(self, filepath: str) -> None:
-        """Create directory if it does not exist"""
-        os.makedirs(filepath, exist_ok=True)
-
-    def stop_detector_backend(self) -> None:
-        """Stop the file writer zmq service for pilatus_2"""
-        self.close_file_writer()
-        time.sleep(0.1)
-        self.stop_file_writer()
-        time.sleep(0.1)
-
-    def close_file_writer(self) -> None:
-        """
-        Close the file writer for pilatus_2
-
-        Delete the data from x12sa-pd-2
-
-        """
-        url = "http://x12sa-pd-2:8080/stream/pilatus_2"
-        try:
-            res = self.send_requests_delete(url=url)
-            if not res.ok:
-                res.raise_for_status()
-        except Exception as exc:
-            logger.info(f"Pilatus2 close threw Exception: {exc}")
-
-    def stop_file_writer(self) -> None:
-        """
-        Stop the file writer for pilatus_2
-
-        Runs on xbl-daq-34
-        """
-        url = "http://xbl-daq-34:8091/pilatus_2/stop"
-        res = self.send_requests_put(url=url)
-        if not res.ok:
-            res.raise_for_status()
+        self.parent.cam.trigger_mode.put(TriggerSource.EXT_ENABLE)
 
     def prepare_data_backend(self) -> None:
         """
@@ -275,6 +211,76 @@ class PilatusSetup(CustomDetectorMixin):
         except Exception as exc:
             logger.info(f"Pilatus2 wait threw Exception: {exc}")
 
+    def set_detector_threshold(self) -> None:
+        """
+        Set correct detector threshold to 1/2 of current X-ray energy, allow 5% tolerance
+
+        Threshold might be in ev or keV
+        """
+
+        # get current beam energy from device manageer
+        mokev = self.parent.device_manager.devices.mokev.obj.read()[
+            self.parent.device_manager.devices.mokev.name
+        ]["value"]
+        factor = 1
+
+        # Check if energies are eV or keV, assume keV as the default
+        unit = getattr(self.parent.cam.threshold_energy, "units", None)
+        if unit is not None and unit == "eV":
+            factor = 1000
+
+        # set energy on detector
+        setpoint = int(mokev * factor)
+
+        # set threshold on detector
+        threshold = self.parent.cam.threshold_energy.read()[self.parent.cam.threshold_energy.name][
+            "value"
+        ]
+        if not np.isclose(setpoint / 2, threshold, rtol=0.05):
+            self.parent.cam.threshold_energy.set(setpoint / 2)
+
+    def set_acquisition_params(self) -> None:
+        """Set acquisition parameters for the detector"""
+
+        # Set number of images and frames (frames is for internal burst of detector)
+        self.parent.cam.num_images.put(
+            int(self.parent.scaninfo.num_points * self.parent.scaninfo.frames_per_trigger)
+        )
+        self.parent.cam.num_frames.put(1)
+
+        # Update the readout time of the detector
+        self.update_readout_time()
+
+    def create_directory(self, filepath: str) -> None:
+        """Create directory if it does not exist"""
+        os.makedirs(filepath, exist_ok=True)
+
+    def close_file_writer(self) -> None:
+        """
+        Close the file writer for pilatus_2
+
+        Delete the data from x12sa-pd-2
+
+        """
+        url = "http://x12sa-pd-2:8080/stream/pilatus_2"
+        try:
+            res = self.send_requests_delete(url=url)
+            if not res.ok:
+                res.raise_for_status()
+        except Exception as exc:
+            logger.info(f"Pilatus2 close threw Exception: {exc}")
+
+    def stop_file_writer(self) -> None:
+        """
+        Stop the file writer for pilatus_2
+
+        Runs on xbl-daq-34
+        """
+        url = "http://xbl-daq-34:8091/pilatus_2/stop"
+        res = self.send_requests_put(url=url)
+        if not res.ok:
+            res.raise_for_status()
+
     def send_requests_put(self, url: str, data: list = None, headers: dict = None) -> object:
         """
         Send a put request to the given url
@@ -302,14 +308,8 @@ class PilatusSetup(CustomDetectorMixin):
         """
         return requests.delete(url=url, headers=headers, timeout=5)
 
-    def pre_scan(self) -> None:
-        """
-        Pre_scan function call
-
-        This function is called just before the scan core.
-        Here it is used to arm the detector for the acquisition
-
-        """
+    def on_pre_scan(self) -> None:
+        """Prepare detector for scan"""
         self.arm_acquisition()
 
     def arm_acquisition(self) -> None:
@@ -318,42 +318,14 @@ class PilatusSetup(CustomDetectorMixin):
         # TODO is this sleep needed? to be tested with detector and for how long
         time.sleep(0.5)
 
-    def publish_file_location(self, done: bool = False, successful: bool = None) -> None:
-        """
-        Publish the filepath to REDIS and publish the event for the h5_converter
-
-        We publish two events here:
-        - file_event: event for the filewriter
-        - public_file: event for any secondary service (e.g. radial integ code)
-
-        Args:
-            done (bool): True if scan is finished
-            successful (bool): True if scan was successful
-        """
-        pipe = self.parent.connector.pipeline()
-        if successful is None:
-            msg = messages.FileMessage(
-                file_path=self.parent.filepath,
-                done=done,
-                metadata={"input_path": self.parent.filepath_raw},
-            )
-        else:
-            msg = messages.FileMessage(
-                file_path=self.parent.filepath,
-                done=done,
-                successful=successful,
-                metadata={"input_path": self.parent.filepath_raw},
-            )
-        self.parent.connector.set_and_publish(
-            MessageEndpoints.public_file(self.parent.scaninfo.scan_id, self.parent.name),
-            msg,
-            pipe=pipe,
+    def on_unstage(self) -> None:
+        """Unstage the detector"""
+        self.finished()
+        self.publish_file_location(
+            done=True, successful=True, metadata={"input_path": self.parent.filepath_raw}
         )
-        self.parent.connector.set_and_publish(
-            MessageEndpoints.file_event(self.parent.name), msg, pipe=pipe
-        )
-        pipe.execute()
 
+    @threadlocked
     def finished(self) -> None:
         """Check if acquisition is finished."""
         # pylint: disable=protected-access
@@ -363,7 +335,7 @@ class PilatusSetup(CustomDetectorMixin):
         ]
         if not self.wait_for_signals(
             signal_conditions=signal_conditions,
-            timeout=self.parent.timeout,
+            timeout=self.parent.TIMEOUT_FOR_SIGNALS,
             check_stopped=True,
             all_signals=True,
         ):
@@ -375,16 +347,21 @@ class PilatusSetup(CustomDetectorMixin):
         self.stop_detector()
         self.stop_detector_backend()
 
+    def on_stop(self) -> None:
+        """Stop detector"""
+        self.stop_detector()
+        self.stop_detector_backend()
+
     def stop_detector(self) -> None:
         """Stop detector"""
         self.parent.cam.acquire.put(0)
 
-    def check_scan_id(self) -> None:
-        """Checks if scan_id has changed and stops the scan if it has"""
-        old_scan_id = self.parent.scaninfo.scan_id
-        self.parent.scaninfo.load_scan_metadata()
-        if self.parent.scaninfo.scan_id != old_scan_id:
-            self.parent.stopped = True
+    def stop_detector_backend(self) -> None:
+        """Stop the file writer zmq service for pilatus_2"""
+        self.close_file_writer()
+        time.sleep(0.1)
+        self.stop_file_writer()
+        time.sleep(0.1)
 
 
 class PilatuscSAXS(PSIDetectorBase):
@@ -401,20 +378,16 @@ class PilatuscSAXS(PSIDetectorBase):
     """
 
     # Specify which functions are revealed to the user in BEC client
-    USER_ACCESS = ["describe"]
+    USER_ACCESS = []
 
     # specify Setup class
     custom_prepare_cls = PilatusSetup
     # specify minimum readout time for detector
     MIN_READOUT = 3e-3
+    TIMEOUT_FOR_SIGNALS = 5
     # specify class attributes
     cam = ADCpt(SLSDetectorCam, "cam1:")
 
-    def set_trigger(self, trigger_source: TriggerSource) -> None:
-        """Set trigger source for the detector"""
-        value = trigger_source
-        self.cam.trigger_mode.put(value)
-
 
 if __name__ == "__main__":
     pilatus_2 = PilatuscSAXS(name="pilatus_2", prefix="X12SA-ES-PILATUS300K:", sim_mode=True)

tests/tests_devices/test_eiger9m_csaxs.py

@@ -9,24 +9,14 @@ from bec_lib.endpoints import MessageEndpoints
 from bec_server.device_server.tests.utils import DMMock
 from ophyd_devices.tests.utils import MockPV
 
-from csaxs_bec.devices.epics.eiger9m_csaxs import Eiger9McSAXS
-
-
-def patch_dual_pvs(device):
-    for walk in device.walk_signals():
-        if not hasattr(walk.item, "_read_pv"):
-            continue
-        if not hasattr(walk.item, "_write_pv"):
-            continue
-        if walk.item._read_pv.pvname.endswith("_RBV"):
-            walk.item._read_pv = walk.item._write_pv
+from csaxs_bec.devices.epics.devices.eiger9m_csaxs import Eiger9McSAXS
+from tests.utils import patch_dual_pvs
 
 
 @pytest.fixture(scope="function")
 def mock_det():
     name = "eiger"
     prefix = "X12SA-ES-EIGER9M:"
-    sim_mode = False
     dm = DMMock()
     with mock.patch.object(dm, "connector"):
         with (
@@ -39,10 +29,9 @@ def mock_det():
                 mock_cl.get_pv = MockPV
                 mock_cl.thread_class = threading.Thread
                 with mock.patch.object(Eiger9McSAXS, "_init"):
-                    det = Eiger9McSAXS(
-                        name=name, prefix=prefix, device_manager=dm, sim_mode=sim_mode
-                    )
+                    det = Eiger9McSAXS(name=name, prefix=prefix, device_manager=dm)
                     patch_dual_pvs(det)
+                    det.TIMEOUT_FOR_SIGNALS = 0.1
                     yield det
 
 
@@ -50,7 +39,6 @@ def test_init():
     """Test the _init function:"""
     name = "eiger"
     prefix = "X12SA-ES-EIGER9M:"
-    sim_mode = False
     dm = DMMock()
     with mock.patch.object(dm, "connector"):
         with (
@@ -72,7 +60,7 @@ def test_init():
                         "csaxs_bec.devices.epics.eiger9m_csaxs.Eiger9MSetup.initialize_detector_backend"
                     ) as mock_init_backend,
                 ):
-                    Eiger9McSAXS(name=name, prefix=prefix, device_manager=dm, sim_mode=sim_mode)
+                    Eiger9McSAXS(name=name, prefix=prefix, device_manager=dm)
                     mock_default.assert_called_once()
                     mock_init_det.assert_called_once()
                     mock_init_backend.assert_called_once()
@@ -236,7 +224,7 @@ def test_stage(
         mock_det.cam.beam_energy.put(scaninfo["mokev"])
         mock_det.stopped = stopped
         mock_det.cam.detector_state._read_pv.mock_data = detector_state
-        with mock.patch.object(mock_det.custom_prepare, "prepare_detector_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"]
             if expected_exception:
                 with pytest.raises(Exception):

tests/tests_devices/test_falcon_csaxs.py

@@ -10,24 +10,14 @@ from bec_lib.endpoints import MessageEndpoints
 from bec_server.device_server.tests.utils import DMMock
 from ophyd_devices.tests.utils import MockPV
 
-from csaxs_bec.devices.epics.falcon_csaxs import FalconcSAXS, FalconTimeoutError
-
-
-def patch_dual_pvs(device):
-    for walk in device.walk_signals():
-        if not hasattr(walk.item, "_read_pv"):
-            continue
-        if not hasattr(walk.item, "_write_pv"):
-            continue
-        if walk.item._read_pv.pvname.endswith("_RBV"):
-            walk.item._read_pv = walk.item._write_pv
+from csaxs_bec.devices.epics.devices.falcon_csaxs import FalconcSAXS, FalconTimeoutError
+from tests.utils import patch_dual_pvs
 
 
 @pytest.fixture(scope="function")
 def mock_det():
     name = "falcon"
     prefix = "X12SA-SITORO:"
-    sim_mode = False
     dm = DMMock()
     with mock.patch.object(dm, "connector"):
         with (
@@ -42,10 +32,9 @@ def mock_det():
                 mock_cl.get_pv = MockPV
                 mock_cl.thread_class = threading.Thread
                 with mock.patch.object(FalconcSAXS, "_init"):
-                    det = FalconcSAXS(
-                        name=name, prefix=prefix, device_manager=dm, sim_mode=sim_mode
-                    )
+                    det = FalconcSAXS(name=name, prefix=prefix, device_manager=dm)
                     patch_dual_pvs(det)
+                    det.TIMEOUT_FOR_SIGNALS = 0.1
                     yield det
 
 
@@ -139,9 +128,9 @@ def test_stage(mock_det, scaninfo):
     This includes testing _prep_det
     """
     with (
-        mock.patch.object(mock_det, "set_trigger") as mock_set_trigger,
+        mock.patch.object(mock_det.custom_prepare, "set_trigger") as mock_set_trigger,
         mock.patch.object(
-            mock_det.custom_prepare, "prepare_detector_backend"
+            mock_det.custom_prepare, "prepare_data_backend"
         ) as mock_prep_data_backend,
         mock.patch.object(
             mock_det.custom_prepare, "publish_file_location"

tests/tests_devices/test_mcs_card.py

@@ -16,23 +16,13 @@ from csaxs_bec.devices.epics.mcs_csaxs import (
     ReadoutMode,
     TriggerSource,
 )
-
-
-def patch_dual_pvs(device):
-    for walk in device.walk_signals():
-        if not hasattr(walk.item, "_read_pv"):
-            continue
-        if not hasattr(walk.item, "_write_pv"):
-            continue
-        if walk.item._read_pv.pvname.endswith("_RBV"):
-            walk.item._read_pv = walk.item._write_pv
+from tests.utils import patch_dual_pvs
 
 
 @pytest.fixture(scope="function")
 def mock_det():
     name = "mcs"
     prefix = "X12SA-MCS:"
-    sim_mode = False
     dm = DMMock()
     with mock.patch.object(dm, "connector"):
         with (
@@ -47,8 +37,9 @@ def mock_det():
                 mock_cl.get_pv = MockPV
                 mock_cl.thread_class = threading.Thread
                 with mock.patch.object(MCScSAXS, "_init"):
-                    det = MCScSAXS(name=name, prefix=prefix, device_manager=dm, sim_mode=sim_mode)
+                    det = MCScSAXS(name=name, prefix=prefix, device_manager=dm)
                     patch_dual_pvs(det)
+                    det.TIMEOUT_FOR_SIGNALS = 0.1
                     yield det
 
 
@@ -56,7 +47,6 @@ def test_init():
     """Test the _init function:"""
     name = "eiger"
     prefix = "X12SA-ES-EIGER9M:"
-    sim_mode = False
     dm = DMMock()
     with mock.patch.object(dm, "connector"):
         with (
@@ -68,9 +58,6 @@ def test_init():
             with mock.patch.object(ophyd, "cl") as mock_cl:
                 mock_cl.get_pv = MockPV
                 with (
-                    mock.patch(
-                        "csaxs_bec.devices.epics.mcs_csaxs.MCSSetup.initialize_default_parameter"
-                    ) as mock_default,
                     mock.patch(
                         "csaxs_bec.devices.epics.mcs_csaxs.MCSSetup.initialize_detector"
                     ) as mock_init_det,
@@ -78,8 +65,7 @@ def test_init():
                         "csaxs_bec.devices.epics.mcs_csaxs.MCSSetup.initialize_detector_backend"
                     ) as mock_init_backend,
                 ):
-                    MCScSAXS(name=name, prefix=prefix, device_manager=dm, sim_mode=sim_mode)
-                    mock_default.assert_called_once()
+                    MCScSAXS(name=name, prefix=prefix, device_manager=dm)
                     mock_init_det.assert_called_once()
                     mock_init_backend.assert_called_once()
 
@@ -290,7 +276,6 @@ def test_unstage(mock_det, stopped, expected_exception):
         else:
             mock_det.unstage()
             mock_finished.assert_called_once()
-            mock_publish_file_location.assert_called_with(done=True, successful=True)
             assert mock_det.stopped is False
 
 

tests/tests_devices/test_pilatus_csaxs.py

@@ -10,24 +10,14 @@ from bec_lib.endpoints import MessageEndpoints
 from bec_server.device_server.tests.utils import DMMock
 from ophyd_devices.tests.utils import MockPV
 
-from csaxs_bec.devices.epics.pilatus_csaxs import PilatuscSAXS
-
-
-def patch_dual_pvs(device):
-    for walk in device.walk_signals():
-        if not hasattr(walk.item, "_read_pv"):
-            continue
-        if not hasattr(walk.item, "_write_pv"):
-            continue
-        if walk.item._read_pv.pvname.endswith("_RBV"):
-            walk.item._read_pv = walk.item._write_pv
+from csaxs_bec.devices.epics.devices.pilatus_csaxs import PilatuscSAXS
+from tests.utils import patch_dual_pvs
 
 
 @pytest.fixture(scope="function")
 def mock_det():
     name = "pilatus"
     prefix = "X12SA-ES-PILATUS300K:"
-    sim_mode = False
     dm = DMMock()
     with mock.patch.object(dm, "connector"):
         with (
@@ -40,9 +30,7 @@ def mock_det():
                 mock_cl.get_pv = MockPV
                 mock_cl.thread_class = threading.Thread
                 with mock.patch.object(PilatuscSAXS, "_init"):
-                    det = PilatuscSAXS(
-                        name=name, prefix=prefix, device_manager=dm, sim_mode=sim_mode
-                    )
+                    det = PilatuscSAXS(name=name, prefix=prefix, device_manager=dm)
                     patch_dual_pvs(det)
                     yield det
 
@@ -61,7 +49,7 @@ def test_init_detector(mock_det, trigger_source, detector_state):
     Validation upon setting the correct PVs
 
     """
-    mock_det.custom_prepare.initialize_detector()  # call the method you want to test
+    mock_det.custom_prepare.on_init()  # call the method you want to test
     assert mock_det.cam.acquire.get() == detector_state
     assert mock_det.cam.trigger_mode.get() == trigger_source
 
@@ -96,6 +84,8 @@ def test_init_detector(mock_det, trigger_source, detector_state):
     ],
 )
 def test_stage(mock_det, scaninfo, stopped, expected_exception):
+    path = "tmp"
+    mock_det.filepath_raw = path
     with mock.patch.object(
         mock_det.custom_prepare, "publish_file_location"
     ) as mock_publish_file_location:
@@ -105,9 +95,7 @@ def test_stage(mock_det, scaninfo, stopped, expected_exception):
         mock_det.device_manager.add_device("mokev", value=12.4)
         mock_det.stopped = stopped
         with (
-            mock.patch.object(
-                mock_det.custom_prepare, "prepare_detector_backend"
-            ) as mock_data_backend,
+            mock.patch.object(mock_det.custom_prepare, "prepare_data_backend") as mock_data_backend,
             mock.patch.object(
                 mock_det.custom_prepare, "update_readout_time"
             ) as mock_update_readout_time,
@@ -127,11 +115,13 @@ def test_stage(mock_det, scaninfo, stopped, expected_exception):
                 )
                 assert mock_det.cam.num_frames.get() == 1
 
-                mock_publish_file_location.assert_called_with(done=False)
+                mock_publish_file_location.assert_called_with(
+                    done=False, metadata={"input_path": path}
+                )
 
 
 def test_pre_scan(mock_det):
-    mock_det.custom_prepare.pre_scan()
+    mock_det.custom_prepare.on_pre_scan()
     assert mock_det.cam.acquire.get() == 1
 
 
@@ -185,7 +175,9 @@ def test_publish_file_location(mock_det, scaninfo):
     mock_det.filepath = scaninfo["filepath"]
     mock_det.filepath_raw = scaninfo["filepath_raw"]
     mock_det.custom_prepare.publish_file_location(
-        done=scaninfo["done"], successful=scaninfo["successful"]
+        done=scaninfo["done"],
+        successful=scaninfo["successful"],
+        metadata={"input_path": scaninfo["filepath_raw"]},
     )
     if scaninfo["successful"] is None:
         msg = messages.FileMessage(
@@ -405,6 +397,8 @@ def test_prep_file_writer(mock_det, scaninfo, data_msgs, urls, requests_state, e
 
 @pytest.mark.parametrize("stopped, expected_exception", [(False, False), (True, True)])
 def test_unstage(mock_det, stopped, expected_exception):
+    path = "tmp"
+    mock_det.filepath_raw = path
     with (
         mock.patch.object(mock_det.custom_prepare, "finished") as mock_finished,
         mock.patch.object(
@@ -418,7 +412,9 @@ def test_unstage(mock_det, stopped, expected_exception):
         else:
             mock_det.unstage()
             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, metadata={"input_path": path}
+            )
             assert mock_det.stopped is False
 
 

tests/utils.py

@@ -0,0 +1,9 @@
+def patch_dual_pvs(device):
+    device.wait_for_connection(all_signals=True)
+    for walk in device.walk_signals():
+        if not hasattr(walk.item, "_read_pv"):
+            continue
+        if not hasattr(walk.item, "_write_pv"):
+            continue
+        if walk.item._read_pv.pvname.endswith("_RBV"):
+            walk.item._read_pv = walk.item._write_pv