refactor: pilatus changes from stage and minor changes for eiger and falcon

ophyd_devices/epics/devices/eiger9m_csaxs.py

@@ -231,11 +231,10 @@ class Eiger9mCsaxs(DetectorBase):
     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:
+        The following functuions are called (at least):
             - _prep_file_writer
             - _prep_det
             - _publish_file_location
-            - _arm_acquisition
         The device returns a List[object] from the Ophyd Device class.
 
         #TODO make sure this is fullfiled

ophyd_devices/epics/devices/falcon_csaxs.py

@@ -3,6 +3,7 @@ import os
 import time
 
 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
@@ -237,11 +238,10 @@ class FalconCsaxs(Device):
     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:
+        The following functuions are called (at least):
             - _prep_file_writer
             - _prep_det
             - _publish_file_location
-            - _arm_acquisition
         The device returns a List[object] from the Ophyd Device class.
 
         #TODO make sure this is fullfiled

ophyd_devices/epics/devices/pilatus_csaxs.py

@@ -2,6 +2,7 @@ import enum
 import json
 import os
 import time
+from bec_lib.core.devicemanager import DeviceStatus
 import requests
 import numpy as np
 
@@ -23,11 +24,13 @@ logger = bec_logger.logger
 
 class PilatusError(Exception):
     """Base class for exceptions in this module."""
+
     pass
 
 
 class PilatusTimeoutError(Exception):
-    '''Raised when the Pilatus does not respond in time during unstage.'''
+    """Raised when the Pilatus does not respond in time during unstage."""
+
     pass
 
 
@@ -39,11 +42,12 @@ class TriggerSource(int, enum.Enum):
     ALGINMENT = 4
 
 
-class SlsDetectorCam(Device): 
+class SlsDetectorCam(Device):
     """SLS Detector Camera - Pilatus
 
     Base class to map EPICS PVs to ophyd signals.
     """
+
     num_images = ADCpt(EpicsSignalWithRBV, "NumImages")
     num_exposures = ADCpt(EpicsSignalWithRBV, "NumExposures")
     delay_time = ADCpt(EpicsSignalWithRBV, "NumExposures")
@@ -103,13 +107,13 @@ class PilatusCsaxs(DetectorBase):
         #TODO add here the parameters for kind, read_attrs, configuration_attrs, parent
             prefix (str): PV prefix ("X12SA-ES-PILATUS300K:)
             name (str): 'pilatus_2'
-            kind (str): 
+            kind (str):
-            read_attrs (list): 
+            read_attrs (list):
-            configuration_attrs (list): 
+            configuration_attrs (list):
-            parent (object): 
+            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,
@@ -145,8 +149,7 @@ class PilatusCsaxs(DetectorBase):
         self._init()
 
     def _init(self) -> None:
-        """Initialize detector, filewriter and set default parameters 
+        """Initialize detector, filewriter and set default parameters"""
-        """
         self._default_parameter()
         self._init_detector()
         self._init_filewriter()
@@ -159,12 +162,12 @@ class PilatusCsaxs(DetectorBase):
 
     def _init_detector(self) -> None:
         """Initialize the detector"""
-        #TODO add check if detector is running
+        # TODO add check if detector is running
         pass
 
     def _init_filewriter(self) -> None:
         """Initialize the file writer"""
-        #TODO in case the data backend is rewritten, add check if it is ready!
+        # TODO in case the data backend is rewritten, add check if it is ready!
         pass
 
     def _prep_det(self) -> None:
@@ -216,7 +219,7 @@ class PilatusCsaxs(DetectorBase):
         self._stop_file_writer()
         time.sleep(0.1)
 
-        self.filepath_h5 = self.filewriter.compile_full_filename(
+        self.filepath_raw = self.filewriter.compile_full_filename(
             self.scaninfo.scan_number, "pilatus_2.h5", 1000, 5, True
         )
         self.cam.file_path.put(f"/dev/shm/zmq/")
@@ -228,19 +231,19 @@ class PilatusCsaxs(DetectorBase):
 
         # compile filename
         basepath = f"/sls/X12SA/data/{self.scaninfo.username}/Data10/pilatus_2/"
-        self.destination_path = os.path.join(
+        self.filepath = os.path.join(
             basepath,
             self.filewriter.get_scan_directory(self.scaninfo.scan_number, 1000, 5),
         )
         # Make directory if needed
-        os.makedirs(self.destination_path, exist_ok=True)
+        os.makedirs(self.filepath, exist_ok=True)
 
         data_msg = {
             "source": [
                 {
                     "searchPath": "/",
                     "searchPattern": "glob:*.cbf",
-                    "destinationPath": self.destination_path,
+                    "destinationPath": self.filepath,
                 }
             ]
         }
@@ -335,32 +338,76 @@ class PilatusCsaxs(DetectorBase):
             res.raise_for_status()
 
     def stage(self) -> List[object]:
-        """stage the detector and file writer"""
+        """Stage command, called from BEC in preparation of a scan.
-        self._acquisition_done = False
+        This will iniate the preparation of detector and file writer.
+        The following functuions are called:
+            - _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"]
-
+        # TODO refactor logger.info to DEBUG mode?
-        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")
+        self._prep_det()
-        msg = BECMessage.FileMessage(
+        state = False
-            file_path=self.filepath_h5, done=False, metadata={"input_path": self.destination_path}
+        self._publish_file_location(done=state, successful=state)
-        )
         return super().stage()
 
+    # TODO might be useful for base class
     def pre_scan(self) -> None:
+        """ " Pre_scan gets executed right before"""
+        self._arm_acquisition()
+
+    def _arm_acquisition(self) -> None:
         self.acquire()
 
+    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 detector and file writer"""
+        """Unstage the device.
-        # Reset to software trigger
+
-        logger.info("Waiting for Pilatus to return from acquisition")
+        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}, ")
@@ -368,27 +415,37 @@ class PilatusCsaxs(DetectorBase):
             self._stopped = True
         if self._stopped:
             return super().unstage()
-        self._pilatus_finished()
+        self._finished()
-        msg = BECMessage.FileMessage(
+        state = True
-            file_path=self.filepath_h5, done=True, metadata={"input_path": self.destination_path}
+        self._publish_file_location(done=state, successful=state)
-        )
+        self._start_h5converter(done=state)
-        self._producer.set_and_publish(
-            MessageEndpoints.public_file(self.scaninfo.scanID, self.name),
-            msg.dumps(),
-        )
-        self._producer.set_and_publish(
-            MessageEndpoints.file_event(self.name),
-            msg.dumps(),
-        )
-        logger.info("Pilatus2 done")
         return super().unstage()
 
-    def _pilatus_finished(self) -> None:
+    def _start_h5converter(self, done=False) -> None:
-        # time.sleep(2)
+        """Start the h5converter"""
+        msg = BECMessage.FileMessage(
+            file_path=self.filepath_raw, done=done, metadata={"input_path": self.filepath}
+        )
+        self._producer.set_and_publish(
+            MessageEndpoints.public_file(self.scaninfo.scanID, self.name), msg.dumps()
+        )
+
+    def _finished(self) -> None:
+        """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
+        """
         while True:
             if self.device_manager.devices.mcs.obj._staged != Staged.yes:
                 break
             time.sleep(0.1)
+        # TODO implement a waiting function or not
         # time.sleep(2)
         # timer = 0
         # while True:
@@ -408,7 +465,9 @@ class PilatusCsaxs(DetectorBase):
         #         #     f"Pilatus timeout with detector state {self.cam.acquire.get()} and camserver return status: {rtr} "
         #         # )
 
+        self._stop_det()
         self._stop_file_writer()
+        # TODO explore if sleep is needed
         time.sleep(0.5)
         self._close_file_writer()
 
@@ -417,15 +476,18 @@ class PilatusCsaxs(DetectorBase):
         or arm the detector in hardware of the detector
         """
         self.cam.acquire.put(1)
+        # TODO check if sleep of 1s is needed, could be that less is enough
         time.sleep(1)
 
+    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.cam.acquire.put(0)
+        self._stop_det()
         self._stop_file_writer()
-        # TODO maybe needed
         self._close_file_writer()
-        # self.unstage()
         super().stop(success=success)
         self._stopped = True