Merge branch 'master' into 'eiger_refactor', resolve merge conflicts

# Conflicts:
#   ophyd_devices/epics/devices/eiger9m_csaxs.py
#   ophyd_devices/epics/devices/pilatus_csaxs.py

CHANGELOG.md

@@ -2,6 +2,18 @@
 
 <!--next-version-placeholder-->
 
+## v0.9.1 (2023-11-02)
+
+### Fix
+
+* Fixed complete call for non-otf scans ([`9e6dc2a`](https://gitlab.psi.ch/bec/ophyd_devices/-/commit/9e6dc2a9f72c5615abd8bea1fcdea6719a35f1ad))
+
+## v0.9.0 (2023-10-31)
+
+### Feature
+
+* Added file-based replay for xtreme ([`d25f92c`](https://gitlab.psi.ch/bec/ophyd_devices/-/commit/d25f92c6323cccea6de8471f4445b997cfab85a3))
+
 ## v0.8.1 (2023-09-27)
 
 ### Fix

ophyd_devices/epics/devices/falcon_csaxs.py

@@ -1,37 +1,47 @@
 import enum
 import os
 import time
-from typing import List
-from ophyd import EpicsSignal, EpicsSignalRO, EpicsSignalWithRBV, Component as Cpt, Device
 
+from typing import List
+from bec_lib.core.devicemanager import DeviceStatus
+
+from ophyd import EpicsSignal, EpicsSignalRO, EpicsSignalWithRBV, Component as Cpt, Device
 from ophyd.mca import EpicsMCARecord
-from ophyd.areadetector.plugins import HDF5Plugin_V21, FilePlugin_V22
 
 from bec_lib.core.file_utils import FileWriterMixin
 from bec_lib.core import MessageEndpoints, BECMessage
 from bec_lib.core import bec_logger
-from ophyd_devices.epics.devices.bec_scaninfo_mixin import BecScaninfoMixin
 
+from ophyd_devices.epics.devices.bec_scaninfo_mixin import BecScaninfoMixin
 from ophyd_devices.utils import bec_utils
 
 logger = bec_logger.logger
 
 
 class FalconError(Exception):
+    """Base class for exceptions in this module."""
+
     pass
 
 
 class FalconTimeoutError(Exception):
+    """Raised when the Falcon does not respond in time during unstage."""
+
     pass
 
 
 class DetectorState(int, enum.Enum):
+    """Detector states for Falcon detector"""
+
     DONE = 0
     ACQUIRING = 1
 
 
 class EpicsDXPFalcon(Device):
-    """All high-level DXP parameters for each channel"""
+    """DXP parameters for Falcon detector
+
+    Base class to map EPICS PVs from DXP parameters to ophyd signals.
+    """
 
     elapsed_live_time = Cpt(EpicsSignal, "ElapsedLiveTime")
     elapsed_real_time = Cpt(EpicsSignal, "ElapsedRealTime")
@@ -51,15 +61,17 @@ class EpicsDXPFalcon(Device):
     current_pixel = Cpt(EpicsSignalRO, "CurrentPixel")
 
 
-class FalconHDF5Plugins(Device):  # HDF5Plugin_V21, FilePlugin_V22):
+class FalconHDF5Plugins(Device):
+    """HDF5 parameters for Falcon detector
+
+    Base class to map EPICS PVs from HDF5 Plugin to ophyd signals.
+    """
+
     capture = Cpt(EpicsSignalWithRBV, "Capture")
     enable = Cpt(EpicsSignalWithRBV, "EnableCallbacks", string=True, kind="config")
     xml_file_name = Cpt(EpicsSignalWithRBV, "XMLFileName", string=True, kind="config")
     lazy_open = Cpt(EpicsSignalWithRBV, "LazyOpen", string=True, doc="0='No' 1='Yes'")
     temp_suffix = Cpt(EpicsSignalWithRBV, "TempSuffix", string=True)
-    # file_path = Cpt(
-    #     EpicsSignalWithRBV, "FilePath", string=True, kind="config", path_semantics="posix"
-    # )
     file_path = Cpt(EpicsSignalWithRBV, "FilePath", string=True, kind="config")
     file_name = Cpt(EpicsSignalWithRBV, "FileName", string=True, kind="config")
     file_template = Cpt(EpicsSignalWithRBV, "FileTemplate", string=True, kind="config")
@@ -70,29 +82,40 @@ class FalconHDF5Plugins(Device):  # HDF5Plugin_V21, FilePlugin_V22):
 
 
 class FalconCsaxs(Device):
-    """FalxonX1 with HDF5 writer"""
+    """Falcon Sitoro detector for CSAXS
+
+    Parent class: Device
+    Device classes: EpicsDXPFalcon dxp1:, EpicsMCARecord mca1, FalconHDF5Plugins HDF1:
+
+    Attributes:
+        name str: 'falcon'
+        prefix (str): PV prefix ("X12SA-SITORO:)
+
+    """
+
+    # Specify which functions are revealed to the user in BEC client
+    USER_ACCESS = [
+        "describe",
+    ]
 
     dxp = Cpt(EpicsDXPFalcon, "dxp1:")
     mca = Cpt(EpicsMCARecord, "mca1")
     hdf5 = Cpt(FalconHDF5Plugins, "HDF1:")
 
-    # Control
+    # specify Epics PVs for Falcon
+    # TODO consider moving this outside of this class!
     stop_all = Cpt(EpicsSignal, "StopAll")
     erase_all = Cpt(EpicsSignal, "EraseAll")
     start_all = Cpt(EpicsSignal, "StartAll")
     state = Cpt(EpicsSignal, "Acquiring")
-    # Preset options
     preset_mode = Cpt(EpicsSignal, "PresetMode")  # 0 No preset 1 Real time 2 Events 3 Triggers
     preset_real = Cpt(EpicsSignal, "PresetReal")
     preset_events = Cpt(EpicsSignal, "PresetEvents")
     preset_triggers = Cpt(EpicsSignal, "PresetTriggers")
-    # read-only diagnostics
     triggers = Cpt(EpicsSignalRO, "MaxTriggers", lazy=True)
     events = Cpt(EpicsSignalRO, "MaxEvents", lazy=True)
     input_count_rate = Cpt(EpicsSignalRO, "MaxInputCountRate", lazy=True)
     output_count_rate = Cpt(EpicsSignalRO, "MaxOutputCountRate", lazy=True)
-
-    # Mapping control
     collect_mode = Cpt(EpicsSignal, "CollectMode")  # 0 MCA spectra, 1 MCA mapping
     pixel_advance_mode = Cpt(EpicsSignal, "PixelAdvanceMode")
     ignore_gate = Cpt(EpicsSignal, "IgnoreGate")
@@ -102,8 +125,6 @@ class FalconCsaxs(Device):
     pixels_per_run = Cpt(EpicsSignal, "PixelsPerRun")
     nd_array_mode = Cpt(EpicsSignal, "NDArrayMode")
 
-    # HDF5
-
     def __init__(
         self,
         prefix="",
@@ -117,6 +138,18 @@ class FalconCsaxs(Device):
         sim_mode=False,
         **kwargs,
     ):
+        """Initialize Falcon detector
+        Args:
+        #TODO add here the parameters for kind, read_attrs, configuration_attrs, parent
+            prefix (str): PV prefix ("X12SA-SITORO:)
+            name (str): 'falcon'
+            kind (str):
+            read_attrs (list):
+            configuration_attrs (list):
+            parent (object):
+            device_manager (object): BEC device manager
+            sim_mode (bool): simulation mode to start the detector without BEC, e.g. from ipython shell
+        """
         super().__init__(
             prefix=prefix,
             name=name,
@@ -130,7 +163,8 @@ class FalconCsaxs(Device):
             raise FalconError("Add DeviceManager to initialization or init with sim_mode=True")
         self._stopped = False
         self.name = name
-        self.wait_for_connection()  # Make sure to be connected before talking to PVs
+        self.wait_for_connection()
+        # Spin up connections for simulation or BEC mode
         if not sim_mode:
             from bec_lib.core.bec_service import SERVICE_CONFIG
 
@@ -138,37 +172,60 @@ class FalconCsaxs(Device):
             self._producer = self.device_manager.producer
             self.service_cfg = SERVICE_CONFIG.config["service_config"]["file_writer"]
         else:
+            base_path = f"/sls/X12SA/data/{self.scaninfo.username}/Data10/"
             self._producer = bec_utils.MockProducer()
             self.device_manager = bec_utils.MockDeviceManager()
             self.scaninfo = BecScaninfoMixin(device_manager, sim_mode)
             self.scaninfo.load_scan_metadata()
-            self.service_cfg = {"base_path": f"/sls/X12SA/data/{self.scaninfo.username}/Data10/"}
+            self.service_cfg = {"base_path": base_path}
+
         self.scaninfo = BecScaninfoMixin(device_manager, sim_mode)
         self.scaninfo.load_scan_metadata()
         self.filewriter = FileWriterMixin(self.service_cfg)
+        self._init()
 
-        self.readout = 0.003  # 3 ms
+    def _init(self) -> None:
+        """Initialize detector, filewriter and set default parameters"""
+        self._default_parameter()
+        self._init_detector()
+        self._init_filewriter()
+
+    def _default_parameter(self) -> None:
+        """Set default parameters for Falcon
+        readout (float): readout time in seconds
+        _value_pixel_per_buffer (int): number of spectra in buffer of Falcon Sitoro"""
+        self.readout = 1e-3
         self._value_pixel_per_buffer = 20  # 16
-        self._clean_up()
+        self._stop_det()
-        self._init_hdf5_saving()
+        self._stop_file_writer()
-        self._init_mapping_mode()
 
-    def _clean_up(self) -> None:
+    def _stop_det(self) -> None:
-        """Clean up"""
+        """ "Stop detector"""
-        self.hdf5.capture.put(0)
         self.stop_all.put(1)
         self.erase_all.put(1)
 
-    def _init_hdf5_saving(self) -> None:
+    def _stop_file_writer(self) -> None:
-        """Set up hdf5 save parameters"""
+        """ "Stop the file writer"""
+        self.hdf5.capture.put(0)
+
+    def _init_filewriter(self) -> None:
+        """Initialize file writer for Falcon.
+        This includes setting variables for the HDF5 plugin (EPICS) that is used to write the data.
+        """
         self.hdf5.enable.put(1)  # EnableCallbacks
         self.hdf5.xml_file_name.put("layout.xml")  # Points to hardcopy of HDF5 Layout xml file
-        self.hdf5.lazy_open.put(1)  # Yes -> To be checked how to add FilePlugin_V21+
+        self.hdf5.lazy_open.put(
+            1
+        )  # Potentially not needed, means a temp data file is created first, could be 0
         self.hdf5.temp_suffix.put("")  # -> To be checked how to add FilePlugin_V22+
-        self.hdf5.queue_size.put(2000)
+        self.hdf5.queue_size.put(2000)  # size of queue for spectra in the buffer
 
-    def _init_mapping_mode(self) -> None:
+    def _init_detector(self) -> None:
-        """Set up mapping mode params"""
+        """Initialize Falcon detector.
+        The detector is operated in MCA mapping mode.
+        Parameters here affect the triggering, gating  etc.
+        This includes also the readout chunk size and whether data is segmented into spectra in EPICS.
+        """
         self.collect_mode.put(1)  # 1 MCA Mapping, 0 MCA Spectrum
         self.preset_mode.put(1)  # 1 Realtime
         self.input_logic_polarity.put(0)  # 0 Normal, 1 Inverted
@@ -176,24 +233,47 @@ class FalconCsaxs(Device):
         self.ignore_gate.put(0)  # 1 Yes, 0 No
         self.auto_pixels_per_buffer.put(0)  # 0 Manual 1 Auto
         self.pixels_per_buffer.put(self._value_pixel_per_buffer)  #
-        self.nd_array_mode.put(1)
+        self.nd_array_mode.put(1)  # Segmentation happens in EPICS
+
+    def stage(self) -> List[object]:
+        """Stage command, called from BEC in preparation of a scan.
+        This will iniate the preparation of detector and file writer.
+        The following functuions are called (at least):
+            - _prep_file_writer
+            - _prep_det
+            - _publish_file_location
+        The device returns a List[object] from the Ophyd Device class.
+
+        #TODO make sure this is fullfiled
+
+        Staging not idempotent and should raise
+        :obj:`RedundantStaging` if staged twice without an
+        intermediate :meth:`~BlueskyInterface.unstage`.
+        """
+        self._stopped = False
+        self.scaninfo.load_scan_metadata()
+        self.mokev = self.device_manager.devices.mokev.obj.read()[
+            self.device_manager.devices.mokev.name
+        ]["value"]
+        self._prep_file_writer()
+        self._prep_det()
+        state = False
+        self._publish_file_location(done=state, successful=state)
+        self._arm_acquisition()
+        return super().stage()
 
     def _prep_det(self) -> None:
         """Prepare detector for acquisition"""
         self.collect_mode.put(1)
         self.preset_real.put(self.scaninfo.exp_time)
         self.pixels_per_run.put(int(self.scaninfo.num_points * self.scaninfo.frames_per_trigger))
-        # self.auto_pixels_per_buffer.put(0)
-        # self.pixels_per_buffer.put(self._value_pixel_per_buffer)
 
     def _prep_file_writer(self) -> None:
-        """Prep HDF5 weriting"""
+        """Prepare filewriting from HDF5 plugin"""
-        # TODO creta filename and destination path from filepath
+        self.filepath = self.filewriter.compile_full_filename(
-        self.destination_path = self.filewriter.compile_full_filename(
             self.scaninfo.scan_number, f"{self.name}.h5", 1000, 5, True
         )
-        # self.hdf5.file_path.set(self.destination_path)
+        file_path, file_name = os.path.split(self.filepath)
-        file_path, file_name = os.path.split(self.destination_path)
         self.hdf5.file_path.put(file_path)
         self.hdf5.file_name.put(file_name)
         self.hdf5.file_template.put(f"%s%s")
@@ -202,29 +282,23 @@ class FalconCsaxs(Device):
         self.hdf5.array_counter.put(0)
         self.hdf5.capture.put(1)
 
-    def stage(self) -> List[object]:
+    def _publish_file_location(self, done=False, successful=False) -> None:
-        """stage the detector and file writer"""
+        """Publish the filepath to REDIS
-        # TODO clean up needed?
+        First msg for file writer and the second one for other listeners (e.g. radial integ)
-        self._stopped = False
+        """
-        self.scaninfo.load_scan_metadata()
+        pipe = self._producer.pipeline()
-        self.mokev = self.device_manager.devices.mokev.obj.read()[
+        msg = BECMessage.FileMessage(file_path=self.filepath, done=done, successful=successful)
-            self.device_manager.devices.mokev.name
-        ]["value"]
-
-        logger.info("Waiting for pilatus2 to be armed")
-        self._prep_det()
-        logger.info("Pilatus2 armed")
-        logger.info("Waiting for pilatus2 zmq stream to be ready")
-        self._prep_file_writer()
-        logger.info("Pilatus2 zmq ready")
-
-        msg = BECMessage.FileMessage(file_path=self.destination_path, done=False)
         self._producer.set_and_publish(
-            MessageEndpoints.public_file(self.scaninfo.scanID, self.name),
+            MessageEndpoints.public_file(self.scaninfo.scanID, self.name), msg.dumps(), pipe=pipe
-            msg.dumps(),
         )
-        self.arm_acquisition()
+        self._producer.set_and_publish(
-        logger.info("Waiting for Falcon to be armed")
+            MessageEndpoints.file_event(self.name), msg.dumps(), pip=pipe
+        )
+        pipe.execute()
+
+    def _arm_acquisition(self) -> None:
+        """Arm Falcon detector for acquisition"""
+        self.start_all.put(1)
         while True:
             det_ctrl = self.state.read()[self.state.name]["value"]
             if det_ctrl == int(DetectorState.ACQUIRING):
@@ -232,14 +306,28 @@ class FalconCsaxs(Device):
             if self._stopped == True:
                 break
             time.sleep(0.005)
-        logger.info("Falcon is armed")
-        return super().stage()
 
-    def arm_acquisition(self) -> None:
+    # TODO function for abstract class?
-        self.start_all.put(1)
+    def trigger(self) -> DeviceStatus:
+        """Trigger the detector, called from BEC."""
+        self._on_trigger()
+        return super().trigger()
+
+    # TODO function for abstract class?
+    def _on_trigger(self):
+        """Specify action that should be taken upon trigger signal."""
+        pass
 
     def unstage(self) -> List[object]:
-        logger.info("Waiting for Falcon to return from acquisition")
+        """Unstage the device.
+
+        This method must be idempotent, multiple calls (without a new
+        call to 'stage') have no effect.
+
+        Functions called:
+            - _finished
+            - _publish_file_location
+        """
         old_scanID = self.scaninfo.scanID
         self.scaninfo.load_scan_metadata()
         logger.info(f"Old scanID: {old_scanID}, ")
@@ -247,20 +335,25 @@ class FalconCsaxs(Device):
             self._stopped = True
         if self._stopped:
             return super().unstage()
-        self._falcon_finished()
+        self._finished()
-        self._clean_up()
         state = True
-        msg = BECMessage.FileMessage(file_path=self.destination_path, done=True, successful=state)
+        self._publish_file_location(done=state, successful=state)
-        self._producer.set_and_publish(
-            MessageEndpoints.public_file(self.scaninfo.metadata["scanID"], self.name),
-            msg.dumps(),
-        )
         self._stopped = False
-        logger.info("Falcon done")
         return super().unstage()
 
-    def _falcon_finished(self):
+    def _finished(self):
-        """Function with 10s timeout"""
+        """Check if acquisition is finished.
+
+        This function is called from unstage and stop
+        and will check detector and file backend status.
+        Timeouts after given time
+
+        Functions called:
+            - _stop_det
+            - _stop_file_writer
+        """
+        sleep_time = 0.1
+        timeout = 5
         timer = 0
         while True:
             det_ctrl = self.state.read()[self.state.name]["value"]
@@ -268,25 +361,25 @@ class FalconCsaxs(Device):
             received_frames = self.dxp.current_pixel.get()
             written_frames = self.hdf5.array_counter.get()
             total_frames = int(self.scaninfo.num_points * self.scaninfo.frames_per_trigger)
-            # TODO if no writing was performed before
+            # TODO Could check state of detector (det_ctrl) and file writer (writer_ctrl)
             if total_frames == received_frames and total_frames == written_frames:
                 break
             if self._stopped == True:
                 break
-            time.sleep(0.1)
+            time.sleep(sleep_time)
-            timer += 0.1
+            timer += sleep_time
-            if timer > 5:
+            if timer > timeout:
                 logger.info(
                     f"Falcon missed a trigger: received trigger {received_frames}, send data {written_frames} from total_frames {total_frames}"
                 )
                 break
-                # raise FalconTimeoutError
+        self._stop_det()
-                #     f"Reached timeout with detector state {det_ctrl}, falcon state {writer_ctrl}, received trigger {received_frames} and files written {written_frames}"
+        self._stop_file_writer()
-                # )
 
     def stop(self, *, success=False) -> None:
         """Stop the scan, with camera and file writer"""
-        self._clean_up()
+        self._stop_det()
+        self._stop_file_writer()
         super().stop(success=success)
         self._stopped = True
 

ophyd_devices/epics/devices/pilatus_csaxs.py

@@ -470,6 +470,31 @@ class PilatuscSAXS(DetectorBase):
         """Specify action that should be taken upon trigger signal."""
         pass
 
+    def _publish_file_location(self, done=False, successful=False) -> None:
+        """Publish the filepath to REDIS
+        First msg for file writer and the second one for other listeners (e.g. radial integ)
+        """
+        pipe = self._producer.pipeline()
+        msg = BECMessage.FileMessage(file_path=self.filepath, done=done, successful=successful)
+        self._producer.set_and_publish(
+            MessageEndpoints.public_file(self.scaninfo.scanID, self.name), msg.dumps(), pipe=pipe
+        )
+        self._producer.set_and_publish(
+            MessageEndpoints.file_event(self.name), msg.dumps(), pip=pipe
+        )
+        pipe.execute()
+
+    # TODO function for abstract class?
+    def trigger(self) -> DeviceStatus:
+        """Trigger the detector, called from BEC."""
+        self._on_trigger()
+        return super().trigger()
+
+    # TODO function for abstract class?
+    def _on_trigger(self):
+        """Specify action that should be taken upon trigger signal."""
+        pass
+
     def unstage(self) -> List[object]:
         """Unstage the device.
 
@@ -542,6 +567,10 @@ class PilatuscSAXS(DetectorBase):
         """Stop the detector"""
         self.cam.acquire.put(0)
 
+    def _stop_det(self) -> None:
+        """Stop the detector"""
+        self.cam.acquire.put(0)
+
     def stop(self, *, success=False) -> None:
         """Stop the scan, with camera and file writer"""
         self._stop_det()

ophyd_devices/sim/sim_xtreme.py

@@ -2,6 +2,7 @@ import threading
 import time
 
 import numpy as np
+from bec_lib.core import BECMessage, MessageEndpoints
 from ophyd import Component as Cpt
 from ophyd import Device, Kind, Signal
 from ophyd.flyers import FlyerInterface
@@ -230,6 +231,186 @@ class SynXtremeOtf(FlyerInterface, Device):
         self._run_subs(sub_type=self.SUB_FLYER, value=data)
 
 
+class SynXtremeOtfReplay(FlyerInterface, Device):
+    """
+    PGM on-the-fly scan
+    """
+
+    SUB_VALUE = "value"
+    SUB_FLYER = "flyer"
+    _default_sub = SUB_VALUE
+
+    e1 = Cpt(SynSetpoint, kind=Kind.config)
+    e2 = Cpt(SynSetpoint, kind=Kind.config)
+    time = Cpt(SynSetpoint, kind=Kind.config)
+    folder = Cpt(SynSetpoint, dtype=str, value="", kind=Kind.config)
+    file = Cpt(SynSetpoint, dtype=str, value="", kind=Kind.config)
+    acquire = Cpt(SynSetpoint, auto_monitor=True)
+    edata = Cpt(SynData, kind=Kind.hinted, auto_monitor=True)
+    data = Cpt(SynData, kind=Kind.hinted, auto_monitor=True)
+    idata = Cpt(SynData, kind=Kind.hinted, auto_monitor=True)
+    fdata = Cpt(SynData, kind=Kind.hinted, auto_monitor=True)
+    count = Cpt(SynData, kind=Kind.omitted, auto_monitor=True)
+
+    def __init__(self, *args, device_manager=None, **kwargs):
+        super().__init__(*args, **kwargs)
+        self._start_time = 0
+        self.acquire.subscribe(self._update_status, run=False)
+        self.count.subscribe(self._update_data, run=False)
+        self._data_event = threading.Event()
+        self.precision = 3
+        self._measurement_data = {}
+        self._device_manager = device_manager
+
+    def kickoff(self):
+        self._read_file()
+        self._start_time = time.time()
+        self.acquire.put(1)
+        status = DeviceStatus(self)
+        status.set_finished()
+        return status
+
+    def _read_file(self):
+        data = {}  # Create an empty dictionary to store the data
+
+        # Read the input file
+        with open(self.file.get(), "r") as file:
+            lines = file.readlines()
+
+        # Extract column titles and data types
+        titles = lines[0].strip().split("\t")
+        data_types = lines[1].strip().split("\t")
+
+        # Initialize the dictionary with empty lists for each title
+        for title in titles:
+            data[title] = []
+
+        # Process the data lines and populate the dictionary
+        for line in lines[2:]:
+            values = line.strip().split("\t")
+            for title, value, data_type in zip(titles, values, data_types):
+                if data_type == "java.lang.Double":
+                    data[title].append(float(value))
+                else:
+                    data[title].append(value)
+        self._measurement_data = data
+
+    def complete(self):
+        def check_value(*, old_value, value, **kwargs):
+            return old_value == 1 and value == 0
+
+        if self.acquire.get() == 0:
+            status = DeviceStatus(self)
+            status.set_finished()
+            return status
+        status = SubscriptionStatus(self.acquire, check_value, event_type=self.acquire.SUB_VALUE)
+        return status
+
+    def collect(self):
+        self._update_measurement_data()
+        data = {"time": self._start_time, "data": {}, "timestamps": {}}
+        for attr in ("edata", "data", "idata", "fdata"):
+            obj = getattr(self, attr)
+            if attr == "edata":
+                data["data"][obj.name] = np.asarray(
+                    [float(x) for x in self._measurement_data["#Ecrbk"][7 : 7 + self.count.get()]]
+                )
+            else:
+                data["data"][obj.name] = obj.get()
+            data["timestamps"][obj.name] = obj.timestamp
+
+        return data
+
+    def _update_measurement_data(self):
+        timestamp = time.time()
+        signals = {
+            "signals_s1": {
+                "value": self._measurement_data["CADC1"][self.count.get()],
+                "timestamp": timestamp,
+            },
+            "signals_s2": {
+                "value": self._measurement_data["CADC2"][self.count.get()],
+                "timestamp": timestamp,
+            },
+            "signals_s3": {
+                "value": self._measurement_data["CADC3"][self.count.get()],
+                "timestamp": timestamp,
+            },
+            "signals_s4": {
+                "value": self._measurement_data["CADC4"][self.count.get()],
+                "timestamp": timestamp,
+            },
+            "signals_s5": {
+                "value": self._measurement_data["CADC5"][self.count.get()],
+                "timestamp": timestamp,
+            },
+            "signals_norm_tey": {
+                "value": self._measurement_data["CADC1"][self.count.get()]
+                / self._measurement_data["CADC2"][self.count.get()],
+                "timestamp": timestamp,
+            },
+            "signals_norm_diode": {
+                "value": self._measurement_data["CADC1"][self.count.get()]
+                / self._measurement_data["CADC3"][self.count.get()],
+                "timestamp": timestamp,
+            },
+        }
+        msg = BECMessage.DeviceMessage(
+            signals=signals, metadata=self._device_manager.devices.otf.metadata
+        ).dumps()
+        self._device_manager.producer.set_and_publish(
+            MessageEndpoints.device_readback("signals"), msg
+        )
+
+    def describe_collect(self):
+        desc = {}
+        for attr in ("edata", "data", "idata", "fdata"):
+            desc.update(getattr(self, attr).describe())
+        return desc
+
+    def _update_status(self, *, old_value, value, **kwargs):
+        if old_value == 1 and value == 0:
+            self._done_acquiring()
+            return
+        if old_value == 0 and value == 1:
+            threading.Thread(target=self._start_acquiring, daemon=True).start()
+
+    def _reset_data(self):
+        for entry in ("edata", "data", "idata", "fdata"):
+            getattr(self, entry)._reset_data()
+        self.count._readback = 0
+        self._data_event.clear()
+
+    def _populate_data(self):
+        self._reset_data()
+        while not self._data_event.is_set():
+            for entry in ("edata", "data", "idata", "fdata"):
+                getattr(self, entry).append(np.random.rand())
+            self.count._readback = len(self.edata.get())
+            self.count._run_subs(
+                sub_type="value",
+                old_value=self.count._readback - 1,
+                value=self.count._readback,
+                timestamp=time.time(),
+            )
+            time.sleep(0.1)
+        self._data_event.clear()
+
+    def _start_acquiring(self):
+        threading.Thread(target=self._populate_data, daemon=True).start()
+        timeout_event = threading.Event()
+        flag = timeout_event.wait(self.time.get())
+        if not flag:
+            self._data_event.set()
+            self.acquire.put(0)
+
+    def _update_data(self, value, **kwargs):
+        if value == 0:
+            return
+        data = self.collect()
+        self._run_subs(sub_type=self.SUB_FLYER, value=data)
+
+
 if __name__ == "__main__":
     obj = SynXtremeOtf(name="otf")
     status = obj.time.set(4)

setup.py

@@ -1,6 +1,6 @@
 from setuptools import setup
 
-__version__ = "0.8.1"
+__version__ = "0.9.1"
 
 if __name__ == "__main__":
     setup(